fix(pm-tape): survive tick-level load without restart-looping

deployment/aliyun/polymarket-market-tape.service

@@ -25,9 +25,9 @@ ProtectHome=true
 ReadWritePaths=/data/monday/spool/polymarket
 RestrictAddressFamilies=AF_INET AF_INET6
 UMask=0027
-CPUQuota=30%
-MemoryHigh=384M
-MemoryMax=512M
+CPUQuota=100%
+MemoryHigh=1G
+MemoryMax=1536M
 
 [Install]
 WantedBy=multi-user.target

deployment/aliyun/polymarket-market-tape.toml

@@ -9,6 +9,8 @@ record_market_updates_rotate_seconds = 3600
 record_market_updates_include_kinds = ["quote", "event_discovered", "event_expired", "reference_price", "spot_price", "agg_trade", "l2"]
 record_market_updates_quote_sample_ms = 0
 record_market_updates_event_scoped_quotes = true
+feed_broadcast_capacity = 32768
+feed_lag_policy = "skip_and_continue"
 skip_settlement_exits = true
 
 [strategy]

rust_hft/prediction-markets/crates/ploy-strategy-bundles/src/config.rs

@@ -16,7 +16,7 @@ use std::path::PathBuf;
 
 use crate::engine::{RuntimeConfig, RuntimeMode};
 use crate::executor::SimulatedExecutorConfig;
-use crate::feed::{RecordingKind, RecordingLimits, RecordingPolicy};
+use crate::feed::{LagPolicy, RecordingKind, RecordingLimits, RecordingPolicy};
 use crate::strategies::directional::DirectionalConfig;
 
 /// Top-level config deserialized from a TOML file.
@@ -94,6 +94,13 @@ pub struct RuntimeSection {
     /// Restrict persisted quotes to tokens belonging to an active discovered event.
     #[serde(default)]
     pub record_market_updates_event_scoped_quotes: bool,
+    /// Broadcast channel capacity between feed producers and the runtime
+    /// consumer. Defaults to 8192 when unset.
+    pub feed_broadcast_capacity: Option<usize>,
+    /// How the live feed reacts to broadcast lag. Defaults to `fail_closed`
+    /// for trading runtimes; pure recorders may opt into `skip_and_continue`.
+    #[serde(default)]
+    pub feed_lag_policy: LagPolicy,
     /// Source boundary for live/dry-run market data.
     ///
     /// Defaults to `local_db`, where strategy runners consume collector-persisted
@@ -437,6 +444,34 @@ impl FullConfig {
         self.runtime.replay_market_updates_from.as_deref()
     }
 
+    /// Broadcast channel capacity between feed producers and the runtime consumer.
+    pub fn feed_broadcast_capacity(&self) -> usize {
+        self.runtime.feed_broadcast_capacity.unwrap_or(8192)
+    }
+
+    /// Validated broadcast capacity; rejects zero before the channel is built
+    /// so a malformed config fails fast instead of panicking in tokio.
+    pub fn validated_feed_broadcast_capacity(&self) -> Result<usize, String> {
+        let capacity = self.feed_broadcast_capacity();
+        if capacity == 0 {
+            return Err("feed_broadcast_capacity must be greater than zero".to_string());
+        }
+        Ok(capacity)
+    }
+
+    /// Lag policy for the live feed.
+    pub fn feed_lag_policy(&self) -> LagPolicy {
+        self.runtime.feed_lag_policy
+    }
+
+    /// Trading runtimes must remain fail-closed on feed lag; only a pure noop
+    /// dry-run recorder may skip lagged updates, because there a restart loses
+    /// more tape than a bounded, warn-logged gap.
+    pub fn feed_lag_policy_allowed(&self, mode: RuntimeMode) -> bool {
+        self.runtime.feed_lag_policy != LagPolicy::SkipAndContinue
+            || (mode == RuntimeMode::DryRun && self.runtime.strategy_variant == "noop")
+    }
+
     /// Build SimulatedExecutorConfig from the parsed config.
     pub fn sim_executor_config(&self) -> SimulatedExecutorConfig {
         let e = &self.execution;
@@ -844,6 +879,63 @@ replay_market_updates_from = "captures/dryrun.ndjson"
         assert!(sec.enable_market_impact);
     }
 
+    #[test]
+    fn feed_lag_policy_defaults_to_fail_closed_with_default_capacity() {
+        let minimal = r#"
+[runtime]
+mode = "dryrun"
+
+[strategy]
+"#;
+        let config = FullConfig::from_toml(minimal).unwrap();
+        assert_eq!(config.feed_lag_policy(), LagPolicy::FailClosed);
+        assert_eq!(config.feed_broadcast_capacity(), 8192);
+        assert!(config.feed_lag_policy_allowed(RuntimeMode::Live));
+        assert!(config.feed_lag_policy_allowed(RuntimeMode::DryRun));
+    }
+
+    #[test]
+    fn skip_and_continue_is_only_allowed_for_noop_dry_run_recorders() {
+        let recorder = r#"
+[runtime]
+mode = "dryrun"
+strategy_variant = "noop"
+feed_lag_policy = "skip_and_continue"
+
+[strategy]
+"#;
+        let config = FullConfig::from_toml(recorder).unwrap();
+        assert_eq!(config.feed_lag_policy(), LagPolicy::SkipAndContinue);
+        assert!(config.feed_lag_policy_allowed(RuntimeMode::DryRun));
+        assert!(!config.feed_lag_policy_allowed(RuntimeMode::Live));
+        assert!(!config.feed_lag_policy_allowed(RuntimeMode::Backtest));
+
+        let trading = recorder.replace(
+            "strategy_variant = \"noop\"",
+            "strategy_variant = \"delta_neutral\"",
+        );
+        let config = FullConfig::from_toml(&trading).unwrap();
+        assert!(!config.feed_lag_policy_allowed(RuntimeMode::DryRun));
+    }
+
+    #[test]
+    fn feed_broadcast_capacity_passes_through_configured_value() {
+        let toml = r#"
+[runtime]
+mode = "dryrun"
+feed_broadcast_capacity = 65536
+
+[strategy]
+"#;
+        let config = FullConfig::from_toml(toml).unwrap();
+        assert_eq!(config.feed_broadcast_capacity(), 65536);
+        assert_eq!(config.validated_feed_broadcast_capacity(), Ok(65536));
+
+        let zero = toml.replace("65536", "0");
+        let config = FullConfig::from_toml(&zero).unwrap();
+        assert!(config.validated_feed_broadcast_capacity().is_err());
+    }
+
     #[test]
     fn defaults_work_with_minimal_config() {
         let minimal = r#"

rust_hft/prediction-markets/crates/ploy-strategy-bundles/src/feed/live.rs

@@ -5,63 +5,134 @@
 //! adapters.
 
 use async_trait::async_trait;
+use serde::{Deserialize, Serialize};
 use tokio::sync::broadcast;
 use tracing::warn;
 
 use crate::traits::{Feed, MarketUpdate};
 
+/// How the feed reacts when the broadcast receiver lags behind producers.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum LagPolicy {
+    /// Close the feed on any lag so live/dry-run trading runtimes fail closed
+    /// instead of evaluating against a market state with missing deltas.
+    #[default]
+    FailClosed,
+    /// Skip the missed updates and keep consuming. Intended for pure data
+    /// recorders, where a process restart loses more tape than a bounded gap.
+    SkipAndContinue,
+}
+
 /// Live feed that consumes market updates from a broadcast channel.
 ///
 /// Multiple strategies can subscribe to the same broadcast sender.
-/// Any lag closes the feed so live/dry-run runtimes fail closed instead of
-/// evaluating against a market state with missing deltas.
+/// Lag handling is controlled by [`LagPolicy`]; the default closes the feed so
+/// live/dry-run runtimes fail closed instead of evaluating against a market
+/// state with missing deltas.
 pub struct LiveFeed {
     rx: broadcast::Receiver<MarketUpdate>,
+    lag_policy: LagPolicy,
 }
 
 impl LiveFeed {
     /// Create a live feed from a broadcast receiver.
     pub fn new(rx: broadcast::Receiver<MarketUpdate>) -> Self {
-        Self { rx }
+        Self::with_lag_policy(rx, LagPolicy::default())
+    }
+
+    /// Create a live feed with an explicit lag policy.
+    pub fn with_lag_policy(rx: broadcast::Receiver<MarketUpdate>, lag_policy: LagPolicy) -> Self {
+        Self { rx, lag_policy }
     }
 }
 
 #[async_trait]
 impl Feed for LiveFeed {
     async fn next(&mut self) -> Option<MarketUpdate> {
-        match self.rx.recv().await {
-            Ok(update) => Some(update),
-            Err(broadcast::error::RecvError::Lagged(n)) => {
-                warn!(skipped = n, "LiveFeed lagged; closing feed fail-closed");
-                None
+        loop {
+            match self.rx.recv().await {
+                Ok(update) => return Some(update),
+                Err(broadcast::error::RecvError::Lagged(n)) => match self.lag_policy {
+                    LagPolicy::FailClosed => {
+                        warn!(skipped = n, "LiveFeed lagged; closing feed fail-closed");
+                        return None;
+                    }
+                    LagPolicy::SkipAndContinue => {
+                        warn!(
+                            skipped = n,
+                            "LiveFeed lagged; skipping missed updates and continuing"
+                        );
+                    }
+                },
+                Err(broadcast::error::RecvError::Closed) => return None,
             }
-            Err(broadcast::error::RecvError::Closed) => None,
         }
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use super::LiveFeed;
+    use super::{LagPolicy, LiveFeed};
     use crate::traits::{Feed, MarketUpdate};
     use chrono::Utc;
     use rust_decimal::Decimal;
     use std::sync::Arc;
     use tokio::sync::broadcast;
 
+    fn update(price: Decimal) -> MarketUpdate {
+        MarketUpdate::SpotPrice {
+            symbol: Arc::from("BTCUSDT"),
+            price,
+            ts: Utc::now(),
+        }
+    }
+
     #[tokio::test]
     async fn lagged_live_feed_closes_fail_closed() {
         let (tx, rx) = broadcast::channel(1);
         let mut feed = LiveFeed::new(rx);
-        let update = |price| MarketUpdate::SpotPrice {
-            symbol: Arc::from("BTCUSDT"),
-            price,
-            ts: Utc::now(),
-        };
 
         tx.send(update(Decimal::ONE)).unwrap();
         tx.send(update(Decimal::from(2))).unwrap();
 
         assert!(feed.next().await.is_none());
     }
+
+    #[tokio::test]
+    async fn lagged_live_feed_skip_and_continue_survives() {
+        let (tx, rx) = broadcast::channel(1);
+        let mut feed = LiveFeed::with_lag_policy(rx, LagPolicy::SkipAndContinue);
+
+        tx.send(update(Decimal::ONE)).unwrap();
+        tx.send(update(Decimal::from(2))).unwrap();
+        tx.send(update(Decimal::from(3))).unwrap();
+
+        // The oldest updates were overwritten; the feed must deliver the newest
+        // one instead of closing.
+        let Some(MarketUpdate::SpotPrice { price, .. }) = feed.next().await else {
+            panic!("skip-and-continue feed must keep delivering after lag");
+        };
+        assert_eq!(price, Decimal::from(3));
+
+        tx.send(update(Decimal::from(4))).unwrap();
+        assert!(feed.next().await.is_some());
+    }
+
+    #[test]
+    fn lag_policy_defaults_to_fail_closed() {
+        assert_eq!(LagPolicy::default(), LagPolicy::FailClosed);
+    }
+
+    #[test]
+    fn lag_policy_deserializes_from_config_names() {
+        assert_eq!(
+            serde_json::from_str::<LagPolicy>("\"fail_closed\"").unwrap(),
+            LagPolicy::FailClosed
+        );
+        assert_eq!(
+            serde_json::from_str::<LagPolicy>("\"skip_and_continue\"").unwrap(),
+            LagPolicy::SkipAndContinue
+        );
+    }
 }

rust_hft/prediction-markets/crates/ploy-strategy-bundles/src/feed/mod.rs

@@ -10,7 +10,7 @@ pub mod parquet_stream;
 mod recorded;
 
 pub use historical::HistoricalFeed;
-pub use live::LiveFeed;
+pub use live::{LagPolicy, LiveFeed};
 pub use options::HistoricalLoadOptions;
 #[cfg(feature = "parquet-feed")]
 pub use parquet_stream::StreamingParquetFeed;

rust_hft/prediction-markets/crates/ploy-strategy-runtime/src/live.rs

@@ -572,7 +572,19 @@ async fn run_live_or_dry_run(
         }
     };
 
-    let (tx, rx) = broadcast::channel(8192);
+    let feed_capacity = config
+        .validated_feed_broadcast_capacity()
+        .unwrap_or_else(|message| {
+            eprintln!("{message}");
+            std::process::exit(1);
+        });
+    if !config.feed_lag_policy_allowed(runtime_config.mode) {
+        eprintln!(
+            "feed_lag_policy = \"skip_and_continue\" is only allowed for a pure noop dry-run recorder"
+        );
+        std::process::exit(1);
+    }
+    let (tx, rx) = broadcast::channel(feed_capacity);
     let tx = Arc::new(tx);
     let reference_prices = new_reference_price_registry();
     let market_data_source = config.runtime.market_data_source;
@@ -644,7 +656,7 @@ async fn run_live_or_dry_run(
     let feed: Box<dyn Feed> = if let Some(record_path) = config.record_market_updates_path() {
         Box::new(
             RecordingFeed::with_policy(
-                LiveFeed::new(rx),
+                LiveFeed::with_lag_policy(rx, config.feed_lag_policy()),
                 record_path,
                 config.record_market_updates_policy(),
             )
@@ -657,7 +669,7 @@ async fn run_live_or_dry_run(
             }),
         )
     } else {
-        Box::new(LiveFeed::new(rx))
+        Box::new(LiveFeed::with_lag_policy(rx, config.feed_lag_policy()))
     };
 
     let recorder = build_signal_recorder(db_pool.clone(), runtime_config.mode);

rust_hft/prediction-markets/tests/workspace_runtime_retirement.rs

@@ -1679,6 +1679,8 @@ fn monday_polymarket_data_service_is_read_only_and_fail_closed() {
         "record_market_updates_include_kinds = [\"quote\", \"event_discovered\", \"event_expired\", \"reference_price\", \"spot_price\", \"agg_trade\", \"l2\"]",
         "record_market_updates_quote_sample_ms = 0",
         "record_market_updates_event_scoped_quotes = true",
+        "feed_broadcast_capacity = 32768",
+        "feed_lag_policy = \"skip_and_continue\"",
         "symbols = [\"BTCUSDT\", \"ETHUSDT\", \"SOLUSDT\", \"XRPUSDT\", \"DOGEUSDT\", \"HYPEUSDT\", \"BNBUSDT\"]",
     ] {
         assert!(config.contains(required), "config missing {required}");
@@ -1689,6 +1691,9 @@ fn monday_polymarket_data_service_is_read_only_and_fail_closed() {
         "NoNewPrivileges=true",
         "ProtectSystem=strict",
         "ReadWritePaths=/data/monday/spool/polymarket",
+        "CPUQuota=100%",
+        "MemoryHigh=1G",
+        "MemoryMax=1536M",
     ] {
         assert!(service.contains(required), "service missing {required}");
     }