From c6a68fbdda7d09a5a73df727338e9288e1009eb5 Mon Sep 17 00:00:00 2001 From: guyverino Date: Fri, 16 Oct 2026 15:34:08 +0200 Subject: [PATCH] fix(station): stop unlistable cores from holding tape retention open --- .../src/market/tape_recorder/catchup.rs | 38 +++++++++++++++++++ .../src/market/tape_recorder/catchup/state.rs | 11 ++++++ .../tape_recorder/catchup/state/tests.rs | 31 +++++++++++++++ .../src/market/tape_recorder/catchup/tests.rs | 13 +++++++ crates/moon-core/src/session/lifecycle.rs | 19 ++++++++-- 5 files changed, 109 insertions(+), 3 deletions(-) diff --git a/crates/moon-core/src/market/tape_recorder/catchup.rs b/crates/moon-core/src/market/tape_recorder/catchup.rs index bf69203b..9a1cbf0f 100644 --- a/crates/moon-core/src/market/tape_recorder/catchup.rs +++ b/crates/moon-core/src/market/tape_recorder/catchup.rs @@ -121,6 +121,29 @@ pub(super) fn listings_complete(generation: u64) -> bool { .clone(); cores.is_some_and(|cores| all_listed(&cores, generation)) } + +/// Required cores whose listing has not completed for `generation` — what the station log names +/// when retention gives up waiting, since that barrier decides whether a pass deletes anything. +/// `None` while the session has not published its cores yet, which holds the barrier as well. +pub(super) fn unlisted(generation: u64) -> Option> { + let cores = REQUIRED_CORES + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone()?; + let listings = LISTINGS + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + Some( + cores + .into_iter() + .filter(|core| { + listings + .get(core) + .is_none_or(|entry| entry.1 != Some(generation)) + }) + .collect(), + ) +} static STATUS: Mutex = Mutex::new(TapeCatchup { running: false, auto: None, @@ -267,6 +290,21 @@ pub fn listed(core: u64, generation: u64) { super::send(Cmd::Listed { core, generation }); } +/// Settle at once the listing of a core that can attribute nothing: its platform is not in the +/// exchange directory, so no report row can name a venue and the listing would protect no +/// market. Left unlisted, such a core held [`listings_complete`] open forever, and every purge +/// then kept all tape of every market — the age limit and the emulator choice alike. Applied +/// directly: the queued [`listed`] only orders completion behind submitted pages, and there are +/// none. +/// +/// The cost, chosen by LinKvo on 2026-10-16: tape on such a core's own markets goes by age +/// alone. Its trades are never known, so one straddling the retention edge loses its older part +/// and its emulator tape stays until it ages out — as for a live close on an unknown venue. +pub fn listed_empty(core: u64, generation: u64) { + listing_started(core, generation); + listing_applied(core, generation); +} + /// Apply a successful completion only when its retention generation is still current. pub(super) fn listing_applied(core: u64, generation: u64) { let _status = STATUS diff --git a/crates/moon-core/src/market/tape_recorder/catchup/state.rs b/crates/moon-core/src/market/tape_recorder/catchup/state.rs index 61f0f7f0..e169acf0 100644 --- a/crates/moon-core/src/market/tape_recorder/catchup/state.rs +++ b/crates/moon-core/src/market/tape_recorder/catchup/state.rs @@ -141,6 +141,17 @@ impl RecorderCatchup { }; let complete = super::listings_complete(generation); if !self.active && (complete || now.saturating_sub(since) >= 600_000) { + if !complete { + // This pass keeps all tape of every market; say which cores held it. + let cores = super::unlisted(generation).map_or_else( + || "core list not received yet".to_string(), + |cores| format!("cores not listed: {cores:?}"), + ); + super::super::line(&format!( + "retention: listings incomplete after 10 min, nothing is deleted \ + this pass; {cores}" + )); + } self.retention.hour = Some(now.div_euclid(HOUR_MS)); status = Some(TapePurge::Running { done: 0, total: 0 }); Phase::Inventory { diff --git a/crates/moon-core/src/market/tape_recorder/catchup/state/tests.rs b/crates/moon-core/src/market/tape_recorder/catchup/state/tests.rs index 0302b474..2cb9ceed 100644 --- a/crates/moon-core/src/market/tape_recorder/catchup/state/tests.rs +++ b/crates/moon-core/src/market/tape_recorder/catchup/state/tests.rs @@ -464,6 +464,37 @@ fn retention_drops_tape_that_ages_out() { ); } +/// A required core on a platform the exchange directory does not know can never list; once its +/// empty listing is settled, the barrier closes and aged tape goes. Before, every pass fell back +/// to keeping all tape after ten minutes, and a station's file never shrank. +#[test] +fn a_venue_less_core_does_not_hold_aged_tape() { + let _lock = SWITCH_LOCK.lock().unwrap(); + let _restore = Restore::new(); + let _settings = RetentionSettings::new(); + catchup::set_keep(1, true); + let core = 70_001; + catchup::require_listings(vec![core]); + catchup::listed_empty(core, catchup::relist_gen()); + let cache = TradeCache::open_evictable(":memory:".into()).unwrap(); + cache.insert( + "synthetic", + "SYNUSDT", + NOW - 3 * DAY, + NOW - 2 * DAY, + vec![], + crate::market::trade_replay::tick_tiles::TileSource::Core, + ); + let mut state = RecorderCatchup::default(); + settle(&mut state, &cache, NOW); + assert!( + cache + .held_spans("synthetic", "SYNUSDT", i64::MIN, i64::MAX) + .unwrap() + .is_empty() + ); +} + /// Without a deferred request, forever retention would never revisit an emulator purge vetoed by an offline report. #[test] fn incomplete_listing_defers_emulator_purge() { diff --git a/crates/moon-core/src/market/tape_recorder/catchup/tests.rs b/crates/moon-core/src/market/tape_recorder/catchup/tests.rs index d3b17357..c397b0a2 100644 --- a/crates/moon-core/src/market/tape_recorder/catchup/tests.rs +++ b/crates/moon-core/src/market/tape_recorder/catchup/tests.rs @@ -309,6 +309,19 @@ fn two_reloads_during_one_listing_count_only_the_last() { take_purge_wanted(); } +/// A core whose platform is outside the exchange directory lists nothing; its listing must close +/// the barrier at once, or every purge keeps all tape of every market forever. +#[test] +fn an_empty_listing_settles_the_barrier_at_once() { + let _guard = SWITCH_LOCK.lock().unwrap(); + let core = 9_000_011; + let generation = relist_gen(); + assert!(!all_listed(&[core], generation)); + listed_empty(core, generation); + assert!(all_listed(&[core], generation)); + assert_eq!(listing_generation(core), Some(generation)); +} + /// Synthetic closed positions share no production core or market identity. fn trade(id: i64, brand: Brand, kind: MarketKind, age_ms: i64) -> CatchupTrade { let close_ms = NOW - age_ms; diff --git a/crates/moon-core/src/session/lifecycle.rs b/crates/moon-core/src/session/lifecycle.rs index 3250a105..7f8d9ab2 100644 --- a/crates/moon-core/src/session/lifecycle.rs +++ b/crates/moon-core/src/session/lifecycle.rs @@ -796,7 +796,15 @@ impl SessionManager { } } if crate::market::tape_recorder::catchup::enabled() { - crate::market::tape_recorder::catchup::require_listings(self.config_order.clone()); + // Every session, connected or not — but not a server switched off in the config: it + // has no session, could never list, and would hold the retention barrier forever. + // The cost, chosen by LinKvo on 2026-10-16: its trades are not known to retention, + // so tape under them goes by age alone — one straddling the edge loses its older + // part, and on a shared market it gets no shelter from another core's emulator or + // tuner-unusable subtraction. + crate::market::tape_recorder::catchup::require_listings( + self.sessions.iter().map(|sess| sess.id).collect(), + ); let now = Instant::now(); let generation = crate::market::tape_recorder::catchup::relist_gen(); let cores: Vec<_> = self @@ -831,9 +839,14 @@ impl SessionManager { return false; }; // The catch-up fetches from a venue's public REST: a venue the directory does not know - // has no route, so its core is not listed — as before its address resolved at all. + // has no route, so its core is not listed. Its listing still counts as complete — it + // could protect nothing — or the retention barrier would wait on it forever. let Some(venue) = address.venue else { - return false; + crate::market::tape_recorder::catchup::listed_empty( + core, + crate::market::tape_recorder::catchup::relist_gen(), + ); + return true; }; let catalog_ready = self.market_source.tradable_coins(core).is_some(); if !catalog_ready {