hypercall_competition/

recorder.rs

//! Downstream consumer that records completed fills into the competition tables.
//!
//! This is the ingestion side of competitions, decoupled from the websocket
//! broadcast path. The recorder consumes a stream of [`EngineMessage`] and, for
//! each `OrderFilled`, persists a competition fill.
//!
//! ## Source-agnostic by design
//!
//! `run` takes a plain `mpsc::UnboundedReceiver<EngineMessage>`, so it does not
//! care where the fill stream comes from. Today the runtime feeds it from the
//! in-process engine EventBus (`TOPIC_FILLS` subscription). When the engine runs
//! out-of-process (see the transport-evolution roadmap), the same recorder is fed
//! by a NATS JetStream consumer instead, with no change here. Other downstream
//! services (analytics, auditors) subscribe to the same stream the same way.
//!
//! ## Failure handling
//!
//! A competition write failure is logged and metered but does NOT crash the
//! engine node. The previous implementation recorded fills inline in the
//! websocket forwarder and triggered a full process shutdown on failure, so a
//! transient leaderboard DB error could take down trading. Fill recording is
//! idempotent (`record_competition_fill` is `ON CONFLICT (trade_id, wallet) DO
//! NOTHING`), so a dropped fill is safely re-recordable.
//!
//! ## Durability caveat (not yet a durable follower)
//!
//! This consumes an in-memory event stream. Fills already written to
//! `competition_fill_events` survive restart, but a fill emitted but not yet
//! recorded when the node dies is NOT recovered: engine replay rebuilds engine
//! state without re-emitting fills to the event bus. Closing that crash window
//! requires making competition a durable external-follower projection that
//! replays the fill stream from the journal/NATS at a recorded offset (tracked
//! separately).

use std::sync::Arc;

use async_trait::async_trait;
use hypercall_runtime_api::ShutdownRx;
use hypercall_types::{EngineMessage, Fill};
use tokio::sync::mpsc::UnboundedReceiver;

use crate::CompetitionService;

/// Minimal sink the recorder depends on: persist one competition fill.
///
/// Implemented by [`CompetitionService`]; abstracted so the recorder's only
/// dependency is "record a fill", which keeps it decoupled and unit-testable.
#[async_trait]
pub trait CompetitionFillSink: Send + Sync {
    async fn record_fill(&self, fill: &Fill) -> anyhow::Result<()>;
}

#[async_trait]
impl CompetitionFillSink for CompetitionService {
    async fn record_fill(&self, fill: &Fill) -> anyhow::Result<()> {
        CompetitionService::record_fill(self, fill)
            .await
            .map_err(anyhow::Error::new)
    }
}

/// Consumes the engine fill stream and records competition fills.
pub struct CompetitionFillRecorder {
    sink: Arc<dyn CompetitionFillSink>,
}

impl CompetitionFillRecorder {
    pub fn new(sink: Arc<dyn CompetitionFillSink>) -> Self {
        Self { sink }
    }

    /// Record every `OrderFilled` until the event stream closes or shutdown fires.
    ///
    /// Intended to be spawned in the runtime task group. Returns when either the
    /// sender side of `events` is dropped or the `shutdown` signal is triggered.
    /// Selecting on `shutdown` is required: the event-bus subscription is not
    /// closed during graceful shutdown, so without it the task group would block
    /// on this task until its shutdown timeout elapsed.
    pub async fn run(self, mut events: UnboundedReceiver<EngineMessage>, mut shutdown: ShutdownRx) {
        tracing::info!("CompetitionFillRecorder started");
        loop {
            let event = tokio::select! {
                _ = shutdown.recv() => {
                    tracing::info!("CompetitionFillRecorder received shutdown signal");
                    break;
                }
                maybe_event = events.recv() => match maybe_event {
                    Some(event) => event,
                    None => {
                        tracing::info!(
                            "CompetitionFillRecorder stopped: engine event stream closed"
                        );
                        break;
                    }
                },
            };
            let EngineMessage::OrderFilled { fill, .. } = &event else {
                continue;
            };
            match self.sink.record_fill(fill).await {
                Ok(()) => {
                    metrics::counter!("competition_fills_recorded_total").increment(1);
                }
                Err(error) => {
                    tracing::error!(
                        trade_id = fill.trade_id,
                        %error,
                        "failed to record competition fill; continuing (recording is idempotent)"
                    );
                    metrics::counter!("competition_fill_record_failures_total").increment(1);
                }
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use hypercall_runtime_api::Shutdown;
    use hypercall_types::{FillAccounting, FillSource, Side, WalletAddress};
    use rust_decimal_macros::dec;
    use std::sync::atomic::{AtomicU64, Ordering};
    use tokio::sync::mpsc;

    struct MockSink {
        recorded: AtomicU64,
        fail: bool,
    }

    impl MockSink {
        fn new(fail: bool) -> Arc<Self> {
            Arc::new(Self {
                recorded: AtomicU64::new(0),
                fail,
            })
        }
        fn count(&self) -> u64 {
            self.recorded.load(Ordering::SeqCst)
        }
    }

    #[async_trait]
    impl CompetitionFillSink for MockSink {
        async fn record_fill(&self, _fill: &Fill) -> anyhow::Result<()> {
            self.recorded.fetch_add(1, Ordering::SeqCst);
            if self.fail {
                anyhow::bail!("simulated competition DB failure");
            }
            Ok(())
        }
    }

    fn order_filled(trade_id: u64) -> EngineMessage {
        EngineMessage::OrderFilled {
            fill: Fill {
                trade_id,
                taker_order_id: 1,
                maker_order_id: 2,
                symbol: "BTC-20260331-100000-C".to_string(),
                price: dec!(100),
                size: dec!(1),
                taker_side: Side::Buy,
                taker_wallet_address: WalletAddress::from([1u8; 20]),
                maker_wallet_address: WalletAddress::from([2u8; 20]),
                fee: dec!(0),
                is_taker: true,
                timestamp: 0,
                builder_code_address: None,
                builder_code_fee: None,
                source: FillSource::default(),
                taker_realized_pnl: None,
                maker_realized_pnl: None,
                underlying_notional: None,
            },
            accounting: FillAccounting::default(),
        }
    }

    #[tokio::test]
    async fn records_each_order_filled() {
        let sink = MockSink::new(false);
        let (tx, rx) = mpsc::unbounded_channel();
        tx.send(order_filled(1)).unwrap();
        tx.send(order_filled(2)).unwrap();
        tx.send(order_filled(3)).unwrap();
        drop(tx); // close the stream so run() returns

        let shutdown = Shutdown::new();
        CompetitionFillRecorder::new(sink.clone())
            .run(rx, shutdown.subscribe())
            .await;

        assert_eq!(sink.count(), 3, "every OrderFilled should be recorded");
    }

    #[tokio::test]
    async fn record_failure_does_not_stop_the_loop() {
        // The whole point of the refactor: a competition DB failure must not
        // crash or stall the recorder; every subsequent fill is still attempted.
        let sink = MockSink::new(true);
        let (tx, rx) = mpsc::unbounded_channel();
        tx.send(order_filled(1)).unwrap();
        tx.send(order_filled(2)).unwrap();
        drop(tx);

        // Must complete without panicking despite every record_fill erroring.
        let shutdown = Shutdown::new();
        CompetitionFillRecorder::new(sink.clone())
            .run(rx, shutdown.subscribe())
            .await;

        assert_eq!(
            sink.count(),
            2,
            "both fills attempted even though recording errored"
        );
    }

    #[tokio::test]
    async fn shutdown_signal_stops_the_loop_with_stream_open() {
        // Regression: the runtime spawns this in a task group and never closes
        // the event-bus subscription on shutdown, so `run` must return when the
        // shutdown signal fires even though `tx` is still alive. Without that,
        // graceful shutdown blocks until the task group's timeout (the failure
        // mode that took down the heavy E2E shutdown assertion).
        let sink = MockSink::new(false);
        let (tx, rx) = mpsc::unbounded_channel();
        let shutdown = Shutdown::new();
        let handle =
            tokio::spawn(CompetitionFillRecorder::new(sink.clone()).run(rx, shutdown.subscribe()));

        shutdown.trigger();

        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("recorder did not exit on shutdown signal")
            .expect("recorder task panicked");

        // tx is still open here: the loop exited via shutdown, not stream close.
        drop(tx);
    }
}