avoid loopback event

Cargo.toml

@@ -73,9 +73,9 @@ silver_network = {path = "crates/network" }
 silver_peer = {path = "crates/peer" }
 silver_storage = { path = "crates/storage" }
 silver_engine = { path = "crates/engine"}
-flux = { git = "https://github.com/gattaca-com/flux", rev = "ceddf228cdcda60bd134ddb49cc0a8375a09b853"}
-flux-utils = { git = "https://github.com/gattaca-com/flux", rev = "ceddf228cdcda60bd134ddb49cc0a8375a09b853", features = ["bytes"]}
-flux-profiler = { git = "https://github.com/gattaca-com/flux", rev = "ceddf228cdcda60bd134ddb49cc0a8375a09b853"}
+flux = { git = "https://github.com/gattaca-com/flux", rev = "2d1b4691a93842dd0cc602cf668ad8b9f71f3904"}
+flux-utils = { git = "https://github.com/gattaca-com/flux", rev = "2d1b4691a93842dd0cc602cf668ad8b9f71f3904", features = ["bytes"]}
+flux-profiler = { git = "https://github.com/gattaca-com/flux", rev = "2d1b4691a93842dd0cc602cf668ad8b9f71f3904"}
 
 # External
 backtrace = "0.3"

crates/beacon_state/tile/src/tile/block.rs

@@ -60,6 +60,8 @@ impl BeaconStateTile {
             ?source,
             head_slot = self.head_state_slot(),
             block_slot,
+            wall_slot = self.ticker.current_slot(),
+            time_into_slot = ?self.ticker.slot_time_elapsed(),
             "applied block: {:?}",
             f
         );

crates/bin/src/main.rs

@@ -68,6 +68,8 @@ fn main() -> Result<(), Box<dyn Error>> {
         incoming_rpc_producer.cache_ref().random_access("ds_persist_incoming_rpc", true)?;
     let incoming_rpc_consumer_eng =
         incoming_rpc_producer.cache_ref().random_access("eng_incoming_rpc", true)?;
+    let incoming_rpc_consumer_ctl =
+        incoming_rpc_producer.cache_ref().random_access("ctl_incoming_rpc", true)?;
     let incoming_engine_resp_producer = TCache::producer(
         "incoming_engine_resp",
         config.engine_config().incoming_engine_resp_tcache_size,
@@ -185,6 +187,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         gossip_handler,
         outgoing_rpc_producer.clone(),
         spec.clone(),
+        incoming_rpc_consumer_ctl,
     );
 
     // A finalized checkpoint state is mandatory (no genesis or runtime sync):

crates/common/src/gossip.rs

@@ -73,12 +73,96 @@ impl From<GossipTopic> for String {
     }
 }
 
+pub const ATTESTATION_SUBNETS: usize = 64;
+pub const SYNC_COMMITTEE_SUBNETS: usize = 4;
+
+const ATTESTATION_BASE: usize = 2;
+const VOLUNTARY_EXIT_SLOT: usize = ATTESTATION_BASE + ATTESTATION_SUBNETS;
+const SYNC_CONTRIB_SLOT: usize = VOLUNTARY_EXIT_SLOT + 3;
+const SYNC_COMMITTEE_BASE: usize = SYNC_CONTRIB_SLOT + 1;
+const LC_FINALITY_SLOT: usize = SYNC_COMMITTEE_BASE + SYNC_COMMITTEE_SUBNETS;
+const DATA_COLUMN_BASE: usize = LC_FINALITY_SLOT + 3;
+const EXECUTION_BID_SLOT: usize = DATA_COLUMN_BASE + crate::NUMBER_OF_CUSTODY_GROUPS as usize;
+
+/// Dense per-topic slot space for counters/telemetry: subnet topics get one
+/// slot per subnet, everything else one slot. The layout is fixed so
+/// observers label by the same arithmetic (`gossip_topic_for_counter_slot`).
+pub const GOSSIP_TOPIC_COUNTER_SLOTS: usize = EXECUTION_BID_SLOT + 4;
+
+/// Inverse of `GossipTopic::counter_slot`, for observer-side labelling.
+pub fn gossip_topic_for_counter_slot(slot: usize) -> Option<GossipTopic> {
+    Some(match slot {
+        0 => GossipTopic::BeaconBlock,
+        1 => GossipTopic::BeaconAggregateAndProof,
+        s if s < VOLUNTARY_EXIT_SLOT => {
+            GossipTopic::BeaconAttestation((s - ATTESTATION_BASE) as u64)
+        }
+        s if s == VOLUNTARY_EXIT_SLOT => GossipTopic::VoluntaryExit,
+        s if s == VOLUNTARY_EXIT_SLOT + 1 => GossipTopic::ProposerSlashing,
+        s if s == VOLUNTARY_EXIT_SLOT + 2 => GossipTopic::AttesterSlashing,
+        s if s == SYNC_CONTRIB_SLOT => GossipTopic::SyncCommitteeContributionAndProof,
+        s if s < LC_FINALITY_SLOT => GossipTopic::SyncCommittee((s - SYNC_COMMITTEE_BASE) as u64),
+        s if s == LC_FINALITY_SLOT => GossipTopic::LightClientFinalityUpdate,
+        s if s == LC_FINALITY_SLOT + 1 => GossipTopic::LightClientOptimisticUpdate,
+        s if s == LC_FINALITY_SLOT + 2 => GossipTopic::BlsToExecutionChange,
+        s if s < EXECUTION_BID_SLOT => {
+            GossipTopic::DataColumnSidecar((s - DATA_COLUMN_BASE) as u64)
+        }
+        s if s == EXECUTION_BID_SLOT => GossipTopic::ExecutionPayloadBid,
+        s if s == EXECUTION_BID_SLOT + 1 => GossipTopic::ExecutionPayload,
+        s if s == EXECUTION_BID_SLOT + 2 => GossipTopic::PayloadAttestationMessage,
+        s if s == EXECUTION_BID_SLOT + 3 => GossipTopic::ProposerPreferences,
+        _ => return None,
+    })
+}
+
 impl GossipTopic {
     /// Format the full wire topic `/eth2/{fork_digest_hex}/{name}/ssz_snappy`.
     pub fn to_wire(&self, fork_digest_hex: &str) -> String {
         format!("/eth2/{fork_digest_hex}/{self}/ssz_snappy")
     }
 
+    /// Slot in the dense counter space; out-of-range subnet ids clamp to
+    /// their kind's last slot.
+    pub fn counter_slot(self) -> usize {
+        fn subnet(base: usize, id: u64, count: usize) -> usize {
+            base + (id as usize).min(count - 1)
+        }
+        match self {
+            Self::BeaconBlock => 0,
+            Self::BeaconAggregateAndProof => 1,
+            Self::BeaconAttestation(id) => subnet(ATTESTATION_BASE, id, ATTESTATION_SUBNETS),
+            Self::VoluntaryExit => VOLUNTARY_EXIT_SLOT,
+            Self::ProposerSlashing => VOLUNTARY_EXIT_SLOT + 1,
+            Self::AttesterSlashing => VOLUNTARY_EXIT_SLOT + 2,
+            Self::SyncCommitteeContributionAndProof => SYNC_CONTRIB_SLOT,
+            Self::SyncCommittee(id) => subnet(SYNC_COMMITTEE_BASE, id, SYNC_COMMITTEE_SUBNETS),
+            Self::LightClientFinalityUpdate => LC_FINALITY_SLOT,
+            Self::LightClientOptimisticUpdate => LC_FINALITY_SLOT + 1,
+            Self::BlsToExecutionChange => LC_FINALITY_SLOT + 2,
+            Self::DataColumnSidecar(id) => {
+                subnet(DATA_COLUMN_BASE, id, crate::NUMBER_OF_CUSTODY_GROUPS as usize)
+            }
+            Self::ExecutionPayloadBid => EXECUTION_BID_SLOT,
+            Self::ExecutionPayload => EXECUTION_BID_SLOT + 1,
+            Self::PayloadAttestationMessage => EXECUTION_BID_SLOT + 2,
+            Self::ProposerPreferences => EXECUTION_BID_SLOT + 3,
+        }
+    }
+
+    pub fn p3_scored(&self) -> bool {
+        matches!(
+            self,
+            Self::BeaconBlock |
+                Self::BeaconAggregateAndProof |
+                Self::BeaconAttestation(_) |
+                Self::SyncCommitteeContributionAndProof |
+                Self::SyncCommittee(_) |
+                Self::ExecutionPayload |
+                Self::PayloadAttestationMessage
+        )
+    }
+
     /// Parse the full wire topic `/eth2/{fork_digest_hex}/{name}/ssz_snappy`.
     /// Verifies the envelope and that the fork digest matches
     /// `fork_digest_hex`.
@@ -245,4 +329,20 @@ mod tests {
             "fork digest mismatch must fail"
         );
     }
+
+    #[test]
+    fn counter_slots_round_trip() {
+        for slot in 0..GOSSIP_TOPIC_COUNTER_SLOTS {
+            let topic = gossip_topic_for_counter_slot(slot)
+                .unwrap_or_else(|| panic!("slot {slot} has no topic"));
+            assert_eq!(topic.counter_slot(), slot, "{topic:?}");
+        }
+        assert!(gossip_topic_for_counter_slot(GOSSIP_TOPIC_COUNTER_SLOTS).is_none());
+
+        // Out-of-range subnets clamp to their kind's last slot.
+        assert_eq!(
+            GossipTopic::BeaconAttestation(9999).counter_slot(),
+            GossipTopic::BeaconAttestation(ATTESTATION_SUBNETS as u64 - 1).counter_slot(),
+        );
+    }
 }

crates/common/src/identity.rs

@@ -208,6 +208,10 @@ impl Identify {
         (self, keypair).into()
     }
 
+    pub fn user_agent(&self) -> &str {
+        str::from_utf8(&self.user_agent[..self.user_agent_len]).unwrap_or("unknown")
+    }
+
     /// Walk the four eth2-shaped address fields, yielding one `Eth2Addr`
     /// per populated slot.
     fn address_iter(&self) -> impl Iterator<Item = Eth2Addr> + '_ {

crates/common/src/lib.rs

@@ -3,8 +3,8 @@ extern crate self as silver_common;
 pub use crate::{
     error::Error,
     gossip::{
-        GossipTopic, MESSAGE_ID_LEN, MessageId, MessageIdHasher, msg_id_invalid_snappy,
-        msg_id_valid_snappy,
+        GOSSIP_TOPIC_COUNTER_SLOTS, GossipTopic, MESSAGE_ID_LEN, MessageId, MessageIdHasher,
+        gossip_topic_for_counter_slot, msg_id_invalid_snappy, msg_id_valid_snappy,
     },
     id::{Keypair, PeerId, decode_protobuf_pubkey, encode_secp256k1_protobuf},
     identity::{

crates/common/src/rpc_rate_limit.rs

@@ -202,6 +202,10 @@ mod tests {
         );
         assert_eq!(
             outbound.peek_outbound(protocol, 128, now + Duration::from_secs(30)),
+            RpcRateLimit::TooSoon
+        );
+        assert_eq!(
+            outbound.peek_outbound(protocol, 128, now + Duration::from_secs(60)),
             RpcRateLimit::Allowed
         );
     }

crates/common/src/spine/messages.rs

@@ -379,6 +379,16 @@ pub enum PeerEvent {
         p2p_peer: usize,
         iwant: TCacheRead,
     },
+    /// A data column sidecar validated from a non-gossip source (RPC
+    /// by-root / EL blobs). Control re-publishes it on its subnet: the
+    /// gossip handler wraps the SSZ (a ref into `incoming_rpc`) as
+    /// protobuf and PM fans it out to the topic mesh, excluding
+    /// `originator` (the peer that served it to us).
+    PublishDataColumn {
+        originator: P2pStreamId,
+        topic: GossipTopic,
+        ssz: TCacheRead,
+    },
     /// Emitted in order to trigger sending of a gossip message.
     /// Peer manager will generate select peers to send to.
     SendGossip {

crates/common/src/spine/stream_protocol.rs

@@ -108,10 +108,10 @@ impl StreamProtocol {
             Self::BeaconBlocksByRoot |
             Self::ExecutionPayloadEnvelopesByRange |
             Self::ExecutionPayloadEnvelopesByRoot => {
-                Some(RpcQuota::n_every(MAX_BLOCK_RATE_LIMIT_TOKENS, 30)) // 30s appears to be the default period of Prysm. Lighthouse is 10s.
+                Some(RpcQuota::n_every(MAX_BLOCK_RATE_LIMIT_TOKENS, 60)) // TODO lighthouse permits per 10s
             }
             Self::DataColumnSidecarsByRange | Self::DataColumnSidecarsByRoot => {
-                Some(RpcQuota::n_every(MAX_SIDECAR_RATE_LIMIT_TOKENS, 15))
+                Some(RpcQuota::n_every(MAX_SIDECAR_RATE_LIMIT_TOKENS, 120)) // TODO, lighthouse permits per 30s
             }
         }
     }

crates/common/src/ticker.rs

@@ -101,6 +101,10 @@ impl SlotTicker {
         self.anchor_genesis_ms + self.anchor.elapsed().as_millis() as u64
     }
 
+    pub fn slot_time_elapsed(&self) -> Duration {
+        Duration::from_millis(self.millis_since_genesis() % self.slot_ms)
+    }
+
     pub fn current_slot(&self) -> u64 {
         self.millis_since_genesis() / self.slot_ms
     }