From d70e9ce14e0666d80b734c549c2caf7c23e6bbf9 Mon Sep 17 00:00:00 2001 From: JuaniRios Date: Sat, 10 Oct 2026 10:35:24 -0300 Subject: [PATCH] fix: hold the liq_ inventory series when a snapshot event fails to apply --- SPEC.md | 9 +- src/conductor.rs | 175 ++++++++++++++++++++++++++++++++++---- src/inventory/snapshot.rs | 118 +++++++++++++++++++++++-- 3 files changed, 275 insertions(+), 27 deletions(-) diff --git a/SPEC.md b/SPEC.md index 4f83512f12..d6c4bb8193 100644 --- a/SPEC.md +++ b/SPEC.md @@ -1775,9 +1775,12 @@ copy of the balances taken while the lock is held. Each copy is numbered in write order, and the store ignores a copy older than the one it holds, so a slow publisher never puts older balances back. Publishing starts once boot has restored the inventory from its snapshots: before that the view is empty, and -its series would read as zero balances. If boot fails to read a snapshot, the -inventory series stay unpublished until the next start and an error is logged. A -task republishes the inventory every 60 seconds, which covers quiet periods. +its series would read as zero balances. If boot fails to read a snapshot, or one +of its events fails to apply, the inventory series stay unpublished for that run +and an error is logged. They return only after a start where every snapshot +reads and applies; a restart does not help while the failing snapshot is +unchanged, because boot rebuilds the same event from it. A task republishes the +inventory every 60 seconds, which covers quiet periods. Every 60 seconds the same task publishes `liq_position_last_price_usd{symbol}` and `liq_equity_exposure_usd{symbol}` (net position times that price) for each diff --git a/src/conductor.rs b/src/conductor.rs index ce3eaea2b9..2df9e985f6 100644 --- a/src/conductor.rs +++ b/src/conductor.rs @@ -704,21 +704,13 @@ async fn finish_startup_recovery(deps: StartupRecoveryDeps<'_>) -> anyhow::Resul // post-restart poll may emit no events (unchanged values are // deduplicated), leaving the view empty and potentially causing // incorrect rebalancing. - let hydration = restore_inventory_at_boot( + restore_inventory_at_boot( deps.pool, deps.inventory, deps.rebalancing_service, deps.position_projection, ) .await?; - match hydration { - Hydration::Complete => deps.inventory.start_publishing_liq_metrics().await, - // A partly restored view would publish missing balances as zero. - Hydration::Incomplete => error!( - "Inventory hydration was incomplete; the liq_ inventory series stay \ - unpublished until the next start" - ), - } deps.rebalancing_service .enqueue_recovery_for_current_wallet_balances() .await; @@ -2048,12 +2040,16 @@ async fn compact_inventory_snapshot_events(pool: &SqlitePool) -> Result, rebalancing_service: &Arc, position_projection: &Projection, -) -> Result> { +) -> Result<(), ProjectionError> { inventory .write_without_broadcast() .await @@ -2065,11 +2061,21 @@ pub(crate) async fn restore_inventory_at_boot( .recover_pending_offchain_orders(position_projection) .await?; - Ok(hydration) + match hydration { + Hydration::Complete => inventory.start_publishing_liq_metrics().await, + Hydration::Incomplete => error!( + "Inventory hydration was incomplete; the liq_ inventory series stay \ + unpublished for this run. A restart does not bring them back while the \ + failing snapshot is unchanged" + ), + } + + Ok(()) } -/// Whether boot read every persisted inventory snapshot. Boot goes on -/// either way; only the `liq_*` inventory series wait for a complete read. +/// Whether boot read and applied every persisted inventory snapshot. Boot +/// goes on either way; only the `liq_*` inventory series wait for a complete +/// hydration. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum Hydration { Complete, @@ -2111,10 +2117,25 @@ async fn hydrate_single_snapshot( } }; - let event_count = snapshot.hydrate_inventory(inventory).await; - if event_count > 0 { - info!(%id, event_count, "Hydrated InventoryView from persisted snapshot"); + let report = snapshot.hydrate_inventory(inventory).await; + if report.events > 0 { + info!( + %id, + events = report.events, + failed = report.failed, + "Hydrated InventoryView from persisted snapshot" + ); + } + + if report.failed > 0 { + warn!( + %id, + failed = report.failed, + "Some InventorySnapshot events failed to apply during hydration" + ); + return Hydration::Incomplete; } + Hydration::Complete } @@ -7936,8 +7957,12 @@ mod tests { use crate::conductor::builder::CqrsFrameworks; use crate::dashboard::{DashboardTradeDeliveryCtx, DashboardTradeDeliveryJobQueue}; use crate::equity_redemption::{EquityRedemptionCommand, redemption_aggregate_id}; + use crate::inventory::snapshot::{InventorySnapshotCommand, InventorySnapshotId}; use crate::inventory::view::Operator; use crate::inventory::{ImbalanceThreshold, Inventory, InventoryView, Venue}; + use crate::metrics::liquidity::LiqFamilies; + use crate::metrics::liquidity::inventory::tests::{leaked_families, publisher, rendered_store}; + use crate::metrics::liquidity::tests::series; use crate::mint_authorization::{ MintAuthorizationError, MockMintAuthorizer, StubVaultModeReader, VaultModeCheckError, }; @@ -8971,6 +8996,124 @@ mod tests { ); } + /// Runs the boot seam over `pool` into a view that publishes the `liq_*` + /// inventory series, and returns the families it publishes into. + async fn boot_publishing_inventory(pool: &SqlitePool) -> &'static LiqFamilies { + let families = leaked_families(); + let (event_sender, _) = broadcast::channel::(16); + let inventory = Arc::new( + BroadcastingInventory::new(InventoryView::default(), event_sender) + .publishing_liq_metrics(publisher(families)), + ); + let projection = Projection::::sqlite(pool.clone()); + projection.catch_up().await.unwrap(); + + restore_inventory_at_boot( + pool, + &inventory, + &freeze_guard_test_service().await, + &projection, + ) + .await + .unwrap(); + + families + } + + fn snapshot_id(owner: Address) -> InventorySnapshotId { + InventorySnapshotId { + orderbook: Address::repeat_byte(0x11), + owner, + } + } + + async fn persist_offchain_aapl(pool: &SqlitePool, id: &InventorySnapshotId) { + test_store::(pool.clone(), ()) + .send( + id, + InventorySnapshotCommand::OffchainEquity { + positions: BTreeMap::from([( + Symbol::new("AAPL").unwrap(), + FractionalShares::new(float!(90)), + )]), + fetched_at: Utc::now(), + }, + ) + .await + .unwrap(); + } + + fn offchain_aapl(families: &LiqFamilies) -> Option { + rendered_store(families) + .get(&series( + "liq_equity_offchain_available", + &[("symbol", "AAPL")], + )) + .copied() + } + + /// Boot starts the `liq_*` inventory series once every snapshot was + /// read and applied, with the restored balances. + #[tokio::test] + async fn boot_publishes_the_inventory_series_after_a_complete_hydration() { + let pool = setup_test_db().await; + persist_offchain_aapl(&pool, &snapshot_id(Address::repeat_byte(0x22))).await; + + let families = boot_publishing_inventory(&pool).await; + + assert_eq!(offchain_aapl(families), Some(90.0)); + } + + /// One snapshot that cannot be read keeps every inventory series off, + /// even though another snapshot restored its balances: a partly + /// restored view would publish the missing balances as zero. + #[tokio::test] + async fn boot_keeps_the_inventory_series_off_when_a_snapshot_read_fails() { + let pool = setup_test_db().await; + persist_offchain_aapl(&pool, &snapshot_id(Address::repeat_byte(0x22))).await; + sqlx::query( + "INSERT INTO events (aggregate_type, aggregate_id, sequence, \ + event_type, event_version, payload, metadata) \ + VALUES (?1, ?2, 1, 'InventorySnapshotEvent::OffchainEquity', '1.0', \ + '{not-json', '{}')", + ) + .bind(InventorySnapshot::AGGREGATE_TYPE) + .bind(snapshot_id(Address::repeat_byte(0x33)).to_string()) + .execute(&pool) + .await + .unwrap(); + + let families = boot_publishing_inventory(&pool).await; + + assert_eq!(rendered_store(families), BTreeMap::new()); + } + + /// A snapshot event the view refuses also keeps the series off. + #[tokio::test] + async fn boot_keeps_the_inventory_series_off_when_a_snapshot_event_fails_to_apply() { + let pool = setup_test_db().await; + let id = snapshot_id(Address::repeat_byte(0x22)); + persist_offchain_aapl(&pool, &id).await; + test_store::(pool.clone(), ()) + .send( + &id, + InventorySnapshotCommand::InflightEquity { + mints: BTreeMap::from([( + Symbol::new("AAPL").unwrap(), + FractionalShares::new(float!(-3)), + )]), + redemptions: BTreeMap::new(), + fetched_at: Utc::now(), + }, + ) + .await + .unwrap(); + + let families = boot_publishing_inventory(&pool).await; + + assert_eq!(rendered_store(families), BTreeMap::new()); + } + async fn freeze_guard_test_service() -> Arc { let (pool, apalis_pool) = setup_test_pools().await; let (event_sender, _) = broadcast::channel::(16); diff --git a/src/inventory/snapshot.rs b/src/inventory/snapshot.rs index cac3f03eab..1d746b88ff 100644 --- a/src/inventory/snapshot.rs +++ b/src/inventory/snapshot.rs @@ -13,7 +13,7 @@ use std::collections::{BTreeMap, BTreeSet}; use std::str::FromStr; use std::sync::Arc; use thiserror::Error; -use tracing::error; +use tracing::{error, warn}; use super::{BroadcastingInventory, divergence::ReconciliationGeneration}; @@ -99,6 +99,14 @@ fn reconciled_usdc_events( }] } +/// What [`InventorySnapshot::hydrate_inventory`] did with one snapshot: the +/// events it read, and how many of them failed to apply. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub(crate) struct HydrationReport { + pub(crate) events: usize, + pub(crate) failed: usize, +} + /// State tracking the latest inventory snapshots. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub(crate) struct InventorySnapshot { @@ -772,25 +780,40 @@ impl InventorySnapshot { /// /// Each field is applied as it is emitted -- no intermediate event /// buffer -- so startup hydration matches the persisted snapshot even - /// when deduplication suppresses the first post-restart poll. - pub(crate) async fn hydrate_inventory(&self, inventory: &Arc) -> usize { + /// when deduplication suppresses the first post-restart poll. An event + /// that fails to apply is skipped with a warning and counted in the + /// report, and hydration goes on with the rest. + pub(crate) async fn hydrate_inventory( + &self, + inventory: &Arc, + ) -> HydrationReport { let now = Utc::now(); let mut view = inventory.write().await; - let mut event_count = 0usize; + let mut report = HydrationReport::default(); let legacy_redemptions = !self.base_redemptions_chain_scoped; self.each_hydration_event(|event| { - event_count += 1; - if let Ok(updated) = view.clone().apply_snapshot_hydration_event_with_format( + report.events += 1; + match view.clone().apply_snapshot_hydration_event_with_format( &event, now, legacy_redemptions, ) { - *view = updated; + Ok(updated) => *view = updated, + Err(error) => { + report.failed += 1; + // The event type only: the event carries balances. + warn!( + target: "inventory", + event = event.event_type(), + ?error, + "Skipped a persisted inventory snapshot event that failed to apply" + ); + } } }); - event_count + report } /// Produce events representing the full persisted state. @@ -3292,6 +3315,85 @@ mod tests { ); } + /// A persisted event the view refuses (here a negative inflight) is + /// skipped and counted, and the events before and after it still + /// hydrate: the chain redemptions come after the inflight equity. + #[tokio::test] + async fn hydration_counts_an_event_that_fails_to_apply_and_keeps_going() { + let symbol = test_symbol("AAPL"); + let fetched_at = Utc::now(); + let snapshot = replay::(vec![ + InventorySnapshotEvent::OffchainEquity { + positions: BTreeMap::from([(symbol.clone(), test_shares(10))]), + fetched_at, + }, + InventorySnapshotEvent::InflightEquity { + mints: BTreeMap::from([(symbol.clone(), test_shares(-3))]), + redemptions: BTreeMap::new(), + fetched_at, + base_redemptions_chain_scoped: true, + }, + InventorySnapshotEvent::ChainInflightRedemptions { + chain: Chain::Base, + redemptions: BTreeMap::from([(symbol.clone(), test_shares(2))]), + fetched_at, + }, + ]) + .unwrap() + .unwrap(); + let (sender, _) = tokio::sync::broadcast::channel(16); + let inventory = Arc::new(BroadcastingInventory::new( + crate::inventory::InventoryView::default(), + sender, + )); + + let report = snapshot.hydrate_inventory(&inventory).await; + + assert_eq!( + report, + HydrationReport { + events: 3, + failed: 1 + } + ); + let view = inventory.read().await; + assert_eq!( + view.equity_available(&symbol, crate::inventory::Venue::Hedging), + Some(test_shares(10)) + ); + assert_eq!( + view.equity_inflight_at(&symbol, crate::inventory::Venue::MarketMaking, Chain::Base), + Some(test_shares(2)) + ); + drop(view); + } + + #[tokio::test] + async fn a_snapshot_that_applies_cleanly_reports_no_failure() { + let symbol = test_symbol("AAPL"); + let snapshot = replay::(vec![InventorySnapshotEvent::OffchainEquity { + positions: BTreeMap::from([(symbol, test_shares(10))]), + fetched_at: Utc::now(), + }]) + .unwrap() + .unwrap(); + let (sender, _) = tokio::sync::broadcast::channel(16); + let inventory = Arc::new(BroadcastingInventory::new( + crate::inventory::InventoryView::default(), + sender, + )); + + let report = snapshot.hydrate_inventory(&inventory).await; + + assert_eq!( + report, + HydrationReport { + events: 1, + failed: 0 + } + ); + } + #[tokio::test] async fn first_chain_scoped_poll_persists_the_format_even_when_empty() { let now = Utc::now();