fix: harden internal runtime and transport paths — reduce fail-open behavior and churn

Cargo.toml

@@ -4,13 +4,15 @@ members = [
     "crates/sof-gossip-tuning",
     "crates/sof-observer",
     "crates/sof-solana-compat",
+    "crates/sof-support",
     "crates/sof-tx",
     "crates/sof-types",
 ]
 exclude = ["crates/sof-solana-gossip"]
 
 [patch.crates-io]
 sof-solana-gossip = { path = "crates/sof-solana-gossip" }
+helius-laserstream = { path = "vendor/helius-laserstream" }
 
 [workspace.package]
 edition = "2024"

crates/sof-gossip-tuning/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "sof-gossip-tuning"
-version = "0.18.1"
+version = "0.18.2"
 edition.workspace = true
 description = "Typed gossip and ingest tuning presets for SOF hosts"
 license = "Apache-2.0 OR MIT"

crates/sof-observer/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "sof"
-version = "0.18.1"
+version = "0.18.2"
 edition.workspace = true
 description = "Solana Observer Framework for low-latency shred ingestion and plugin-driven transaction observation"
 license = "Apache-2.0 OR MIT"
@@ -32,8 +32,9 @@ provider-websocket = []
 
 [dependencies]
 agave-transaction-view = { version = "3.1.11", features = ["agave-unstable-api"] }
-sof-gossip-tuning = { version = "0.18.1", path = "../sof-gossip-tuning" }
-sof-types = { version = "0.18.1", path = "../sof-types", features = ["solana-compat"] }
+sof-gossip-tuning = { version = "0.18.2", path = "../sof-gossip-tuning" }
+sof-support = { version = "0.18.2", path = "../sof-support" }
+sof-types = { version = "0.18.2", path = "../sof-types", features = ["solana-compat"] }
 solana-gossip = { package = "sof-solana-gossip", version = "3.1.11-sof.9", optional = true, features = ["agave-unstable-api"] }
 solana-entry = { version = "3.1.11", features = ["agave-unstable-api"] }
 solana-epoch-schedule = "3.0.0"
@@ -50,6 +51,7 @@ solana-signer = "3.0.0"
 solana-streamer = "3.1.11"
 solana-transaction = { version = "3.0.2", features = ["serde"] }
 solana-packet = "3.0.0"
+solana-system-interface = "3.1.0"
 solana-vote = "3.1.11"
 solana-vote-program = { version = "3.1.11", features = ["agave-unstable-api"] }
 reed-solomon-erasure = { version = "6.0.0", features = ["simd-accel"] }
@@ -81,6 +83,7 @@ futures-channel = "0.3.32"
 tokio-tungstenite = { version = "0.28", default-features = false, features = ["connect", "handshake", "rustls-tls-webpki-roots"] }
 signal-hook = "0.3"
 smallvec = "1.15"
+ahash = "0.8.12"
 xdp = { version = "0.7.3", optional = true }
 yellowstone-grpc-client = { version = "12.2.0", optional = true }
 yellowstone-grpc-proto = { version = "12.1.0", optional = true }

crates/sof-observer/README.md

@@ -581,7 +581,7 @@ cargo add sof
 Optional gossip bootstrap support at compile time:
 
 ```toml
-sof = { version = "0.18.1", features = ["gossip-bootstrap"] }
+sof = { version = "0.18.2", features = ["gossip-bootstrap"] }
 ```
 
 `gossip-bootstrap` uses the vendored `sof-solana-gossip` backend, but it no longer exact-pins the
@@ -590,7 +590,7 @@ Solana `3.1.11` patch line. Downstream crates can resolve newer compatible `3.1.
 Optional external `kernel-bypass` ingress support:
 
 ```toml
-sof = { version = "0.18.1", features = ["kernel-bypass"] }
+sof = { version = "0.18.2", features = ["kernel-bypass"] }
 ```
 
 The bundled `sof-solana-gossip` backend defaults to SOF's lightweight in-memory duplicate/conflict
@@ -1088,6 +1088,7 @@ Design references:
 - Queue pressure drops hook events instead of stalling ingest.
 - Typed host tuning is available through `sof-gossip-tuning` and `RuntimeSetup::with_gossip_tuning_profile(...)`.
 - `RuntimeExtension` WebSocket connectors support full `ws://` / `wss://` handshake + frame decoding.
+- Runtime extensions require non-empty names and resource metadata; startup rejects empty `resource_id` / shared tags and bounds `read_buffer_bytes`.
 - WebSocket close frames emit `RuntimePacketEventClass::ConnectionClosed` in `on_packet_received`.
 - WebSocket packet events expose `websocket_frame_type` (`Text`/`Binary`/`Ping`/`Pong`) for startup-time filtering and runtime routing.
 - In gossip mode, SOF runs as an active bounded relay client by default (UDP relay + repair serve), not as an observer-only passive consumer.

crates/sof-observer/fuzz/fuzz_targets/shred_fec_recover.rs

@@ -1,5 +1,7 @@
 #![no_main]
 
+use std::sync::Arc;
+
 use libfuzzer_sys::fuzz_target;
 use sof::{
     protocol::shred_wire::{
@@ -230,6 +232,7 @@ fuzz_target!(|bytes: &[u8]| {
             }),
             ParsedShred::Code(code) => ParsedShredHeader::Code(code),
         };
+        let packet = Arc::<[u8]>::from(packet);
         let recovered = recoverer.ingest_packet(&packet, &parsed_header);
         assert!(recoverer.tracked_sets() <= max_tracked_sets);
 

crates/sof-observer/src/app/config/common.rs

@@ -1,4 +1,4 @@
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use std::time::Duration;
 
 pub(crate) use crate::runtime_env::read_env_var;
 
@@ -16,11 +16,3 @@ pub fn duration_to_ms_u64(duration: Duration) -> u64 {
         millis as u64
     }
 }
-
-pub fn current_unix_ms() -> u64 {
-    SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .map_or(0, |duration| {
-            duration.as_millis().min(u128::from(u64::MAX)) as u64
-        })
-}

crates/sof-observer/src/app/runtime/bootstrap/gossip/handoff.rs

@@ -1,6 +1,8 @@
 #[cfg(feature = "gossip-bootstrap")]
 use super::*;
 #[cfg(feature = "gossip-bootstrap")]
+use sof_support::time_support::current_unix_ms;
+#[cfg(feature = "gossip-bootstrap")]
 use thiserror::Error;
 
 #[cfg(feature = "gossip-bootstrap")]

crates/sof-observer/src/app/runtime/bootstrap/repair.rs

@@ -125,6 +125,15 @@ pub(crate) struct RepairSourceHintBuffer {
 #[cfg(feature = "gossip-bootstrap")]
 impl RepairSourceHintBuffer {
     pub(crate) fn new(capacity: usize) -> Self {
+        let capacity = capacity.max(1);
+        Self {
+            counts: HashMap::with_capacity(capacity),
+            capacity,
+        }
+    }
+
+    #[cfg(test)]
+    pub(crate) fn new_baseline(capacity: usize) -> Self {
         Self {
             counts: HashMap::new(),
             capacity: capacity.max(1),
@@ -335,6 +344,90 @@ pub(crate) fn spawn_repair_driver(
     Ok((command_tx, result_rx, peer_snapshot, driver_handle))
 }
 
+#[cfg(all(test, feature = "gossip-bootstrap"))]
+mod tests {
+    use std::{
+        env,
+        net::{IpAddr, Ipv4Addr, SocketAddr},
+        time::Instant,
+    };
+
+    use super::RepairSourceHintBuffer;
+
+    #[test]
+    #[ignore = "profiling fixture for repair source hint buffer allocation"]
+    fn repair_source_hint_buffer_profile_fixture() {
+        let iterations = env::var("SOF_REPAIR_SOURCE_HINT_PROFILE_ITERS")
+            .ok()
+            .and_then(|raw| raw.parse::<usize>().ok())
+            .filter(|value| *value > 0)
+            .unwrap_or(20_000);
+        let capacity = env::var("SOF_REPAIR_SOURCE_HINT_PROFILE_CAPACITY")
+            .ok()
+            .and_then(|raw| raw.parse::<usize>().ok())
+            .filter(|value| *value > 0)
+            .unwrap_or(256);
+        let batch_size = env::var("SOF_REPAIR_SOURCE_HINT_PROFILE_BATCH")
+            .ok()
+            .and_then(|raw| raw.parse::<usize>().ok())
+            .filter(|value| *value > 0)
+            .unwrap_or(capacity / 2);
+        let addresses = (0..capacity)
+            .map(|index| {
+                SocketAddr::new(
+                    IpAddr::V4(Ipv4Addr::new(
+                        127,
+                        0,
+                        u8::try_from((index / 255) % 255).unwrap_or(0),
+                        u8::try_from((index % 255) + 1).unwrap_or(u8::MAX),
+                    )),
+                    u16::try_from((10_000 + index) % usize::from(u16::MAX)).unwrap_or(u16::MAX),
+                )
+            })
+            .collect::<Vec<_>>();
+        assert!(!addresses.is_empty());
+
+        let baseline_started_at = Instant::now();
+        for _ in 0..iterations {
+            let mut buffer = RepairSourceHintBuffer::new_baseline(capacity);
+            for addr in addresses.iter().copied() {
+                assert!(buffer.record(addr).is_ok());
+            }
+            let drained = buffer.drain_batch(batch_size);
+            assert!(!drained.is_empty());
+        }
+        let baseline_elapsed = baseline_started_at.elapsed();
+
+        let optimized_started_at = Instant::now();
+        for _ in 0..iterations {
+            let mut buffer = RepairSourceHintBuffer::new(capacity);
+            for addr in addresses.iter().copied() {
+                assert!(buffer.record(addr).is_ok());
+            }
+            let drained = buffer.drain_batch(batch_size);
+            assert!(!drained.is_empty());
+        }
+        let optimized_elapsed = optimized_started_at.elapsed();
+
+        let baseline_avg_ns =
+            baseline_elapsed.as_nanos() / u128::try_from(iterations).unwrap_or(u128::MAX);
+        let optimized_avg_ns =
+            optimized_elapsed.as_nanos() / u128::try_from(iterations).unwrap_or(u128::MAX);
+        let baseline_avg_us = baseline_avg_ns as f64 / 1_000.0;
+        let optimized_avg_us = optimized_avg_ns as f64 / 1_000.0;
+        println!(
+            "repair_source_hint_buffer_profile_fixture iterations={} baseline_us={} optimized_us={} baseline_avg_ns_per_iteration={} optimized_avg_ns_per_iteration={} baseline_avg_us_per_iteration={:.6} optimized_avg_us_per_iteration={:.6}",
+            iterations,
+            baseline_elapsed.as_micros(),
+            optimized_elapsed.as_micros(),
+            baseline_avg_ns,
+            optimized_avg_ns,
+            baseline_avg_us,
+            optimized_avg_us
+        );
+    }
+}
+
 #[cfg(feature = "gossip-bootstrap")]
 pub(crate) fn replace_repair_driver(
     repair_client: crate::repair::GossipRepairClient,

crates/sof-observer/src/app/runtime/dataset/process.rs

@@ -16,6 +16,9 @@ use crate::{
 };
 use agave_transaction_view::transaction_view::SanitizedTransactionView;
 use core::mem::size_of;
+use sof_support::short_vec::{
+    ShortVecDecodeError as PartialParseError, decode_short_u16_len, decode_short_u16_len_partial,
+};
 use solana_hash::Hash;
 use solana_packet::PACKET_DATA_SIZE;
 use solana_pubkey::Pubkey;
@@ -1833,50 +1836,6 @@ fn read_u64_le_partial(payload: &[u8], offset: &mut usize) -> Result<u64, Partia
     Ok(u64::from_le_bytes(raw))
 }
 
-fn decode_short_u16_len(payload: &[u8], offset: &mut usize) -> Option<usize> {
-    let mut value = 0_usize;
-    let mut shift = 0_u32;
-    for byte_index in 0..3 {
-        let byte = usize::from(*payload.get(*offset)?);
-        *offset = (*offset).saturating_add(1);
-        value |= (byte & 0x7f) << shift;
-        if byte & 0x80 == 0 {
-            return Some(value);
-        }
-        shift = shift.saturating_add(7);
-        if byte_index == 2 {
-            return None;
-        }
-    }
-    None
-}
-
-fn decode_short_u16_len_partial(
-    payload: &[u8],
-    offset: &mut usize,
-) -> Result<usize, PartialParseError> {
-    let mut value = 0_usize;
-    let mut shift = 0_u32;
-    for byte_index in 0..3 {
-        let byte = usize::from(*payload.get(*offset).ok_or(PartialParseError::Incomplete)?);
-        *offset = (*offset).saturating_add(1);
-        value |= (byte & 0x7f) << shift;
-        if byte & 0x80 == 0 {
-            return Ok(value);
-        }
-        shift = shift.saturating_add(7);
-        if byte_index == 2 {
-            return Err(PartialParseError::Invalid);
-        }
-    }
-    Err(PartialParseError::Invalid)
-}
-
-enum PartialParseError {
-    Incomplete,
-    Invalid,
-}
-
 fn join_payload_fragments_into(
     buffer: &mut Vec<u8>,
     fragments: &[SharedPayloadFragment],
@@ -1955,11 +1914,13 @@ mod tests {
     use solana_signer::Signer as _;
     use solana_transaction::Transaction;
     use std::{
+        env,
         net::{Ipv4Addr, SocketAddr, SocketAddrV4},
         sync::{
             Arc,
             atomic::{AtomicUsize, Ordering},
         },
+        thread,
         time::{Duration, Instant},
     };
     use wincode::{
@@ -2366,7 +2327,7 @@ mod tests {
     #[test]
     #[ignore = "profiling fixture for perf"]
     fn multi_hook_profile_fixture() {
-        let iterations = std::env::var("SOF_MULTI_HOOK_PROFILE_ITERS")
+        let iterations = env::var("SOF_MULTI_HOOK_PROFILE_ITERS")
             .ok()
             .and_then(|raw| raw.parse::<usize>().ok())
             .filter(|value| *value > 0)
@@ -2442,7 +2403,7 @@ mod tests {
             );
             assert!(matches!(outcome, DatasetProcessOutcome::Decoded));
         }
-        std::thread::sleep(Duration::from_millis(250));
+        thread::sleep(Duration::from_millis(250));
         assert_eq!(dataset_decode_fail_count.load(Ordering::Relaxed), 0);
         assert_eq!(tx_event_drop_count.load(Ordering::Relaxed), 0);
         assert_eq!(plugin_host.dropped_event_count(), 0);
@@ -2474,12 +2435,12 @@ mod tests {
     #[test]
     #[ignore = "profiling fixture for completed-dataset prefilter decode skip A/B"]
     fn completed_dataset_prefilter_profile_fixture() {
-        let iterations = std::env::var("SOF_COMPLETED_DATASET_PREFILTER_PROFILE_ITERS")
+        let iterations = env::var("SOF_COMPLETED_DATASET_PREFILTER_PROFILE_ITERS")
             .ok()
             .and_then(|raw| raw.parse::<usize>().ok())
             .filter(|value| *value > 0)
             .unwrap_or(20_000);
-        let mode = std::env::var("SOF_COMPLETED_DATASET_PREFILTER_PROFILE_MODE")
+        let mode = env::var("SOF_COMPLETED_DATASET_PREFILTER_PROFILE_MODE")
             .unwrap_or_else(|_| "manual".to_owned());
         let ignored_account = Pubkey::new_unique();
         let payload = build_profile_payload(PROFILE_ENTRY_COUNT);
@@ -2529,7 +2490,7 @@ mod tests {
             );
             assert!(matches!(outcome, DatasetProcessOutcome::Decoded));
         }
-        std::thread::sleep(Duration::from_millis(100));
+        thread::sleep(Duration::from_millis(100));
         assert_eq!(dataset_decode_fail_count.load(Ordering::Relaxed), 0);
         assert_eq!(tx_event_drop_count.load(Ordering::Relaxed), 0);
         assert_eq!(plugin_host.dropped_event_count(), 0);
@@ -2697,7 +2658,7 @@ mod tests {
         );
 
         assert!(matches!(outcome, DatasetProcessOutcome::Decoded));
-        std::thread::sleep(Duration::from_millis(50));
+        thread::sleep(Duration::from_millis(50));
         assert_eq!(handled.load(Ordering::Relaxed), 1);
         assert_eq!(dataset_decode_fail_count.load(Ordering::Relaxed), 0);
         assert_eq!(tx_event_drop_count.load(Ordering::Relaxed), 0);