hypercall/rsm/unified_engine/

journaling.rs

//! Journal and idempotency orchestration for `UnifiedEngine`.

use super::*;

/// Strategy for persisting journal entries after order processing.
enum JournalStrategy {
    /// Push entry to batch channel for async write, ACK immediately
    Batched(crate::journal::JournalBatchSender),
    /// Write synchronously to journal, panic on failure
    Sync(crate::journal::SharedEngineJournalWriter),
}

impl JournalStrategy {
    /// Select the appropriate journal strategy based on available writers.
    /// Prefers batched mode if a batch sender is configured.
    fn select(
        batch_sender: &Option<crate::journal::JournalBatchSender>,
        sync_writer: crate::journal::SharedEngineJournalWriter,
    ) -> Self {
        if let Some(sender) = batch_sender.clone() {
            JournalStrategy::Batched(sender)
        } else {
            JournalStrategy::Sync(sync_writer)
        }
    }
}

impl UnifiedEngine {
    /// Process an order with durable journaling.
    ///
    /// This method provides:
    /// - Idempotency: duplicate request_id returns cached response
    /// - Restart-safe ACK: response only sent after DB commit
    /// - Full event capture for replay
    ///
    /// The flow is:
    /// 1. Check journal for existing request_id (idempotency)
    /// 2. If found: return cached response
    /// 3. If new: process order, journal transition, then ACK
    ///
    /// When batch_sender is set, journaling is async (push to channel, ACK immediately).
    /// When batch_sender is not set, falls back to synchronous journaling.
    pub(super) async fn process_order_journaled(&mut self, request: UnifiedEngineRequest) {
        let start = Instant::now();

        let request_id = request.message.request_id.clone();

        // Extract request_id - non-journaled fallback is allowed only in tests.
        let Some(ref req_id) = request_id else {
            counter!("ht_engine_journal_fallback_total", "reason" => "missing_request_id")
                .increment(1);
            if cfg!(test) {
                warn!("Order missing request_id, falling back to non-journaled processing");
                self.process_order(request).await;
                return;
            }
            panic!("JOURNAL_FATAL: request_id missing; refusing to process without idempotency.");
        };
        let req_uuid = uuid::Uuid::parse_str(req_id).unwrap_or_else(|e| {
            panic!(
                "JOURNAL_FATAL: request_id '{}' is not a valid UUID: {}. \
                 Refusing to process without a canonical idempotency key.",
                req_id, e
            )
        });

        let queue_wait = request.enqueued_at.elapsed().as_secs_f64();
        histogram!("ht_engine_queue_wait_seconds").record(queue_wait);
        decrement_pending_requests();
        gauge!("ht_engine_pending_requests").set(get_pending_requests() as f64);

        // Get journal writer for idempotency check
        let journal_writer = match &self.journal_writer {
            Some(w) => w.clone(),
            None => {
                // No journal writer - if we have batch sender, we can still proceed
                // (idempotency won't work but batching will)
                if self.journal_batch_sender.is_none() {
                    // No journal configured - this should never happen as builder validates
                    error!(
                        "Journal unavailable: no writer or batch_sender configured. \
                           This is a configuration error - use .with_mock_journal() for tests."
                    );
                    counter!("ht_engine_journal_error_total", "reason" => "missing_writer")
                        .increment(1);
                    panic!(
                        "JOURNAL_FATAL: no writer or batch_sender configured for request_id={}. \
                         Refusing to process without durability.",
                        req_id
                    );
                }
                // Proceed without idempotency check - add to known set after processing
                warn!("No journal_writer for idempotency check, proceeding with batch-only mode");
                let sender = self
                    .journal_batch_sender
                    .as_ref()
                    .expect("batch sender missing in batch-only journal mode")
                    .clone();
                self.process_order_journaled_impl(
                    request,
                    req_id,
                    req_uuid,
                    start,
                    JournalStrategy::Batched(sender),
                )
                .await;
                self.known_request_ids.insert(req_uuid);
                return;
            }
        };

        // Fast path: check in-memory cache first.
        // Cache entries are loaded from a configured lookback window and the cache is bounded.
        // A miss here means request_id is either new, or older than the cache window and
        // assumed not retried, so we skip the DB lookup.
        if !self.known_request_ids.contains(&req_uuid) {
            // New request - no need to check DB, proceed directly
            counter!("ht_engine_journal_cache_miss_total").increment(1);

            // Process the order (batched or sync path)
            let strategy = JournalStrategy::select(&self.journal_batch_sender, journal_writer);
            self.process_order_journaled_impl(request, req_id, req_uuid, start, strategy)
                .await;

            // Add to known set so future duplicates (in this session) will be caught
            self.known_request_ids.insert(req_uuid);
            return;
        }

        // Slow path: request_id is in known set (either from startup load or this session).
        // Query DB to check if it's actually a duplicate with a cached response.
        counter!("ht_engine_journal_cache_hit_total").increment(1);

        let idempotency_span = info_span!("journal_idempotency_check");
        let req_id_clone = req_uuid;
        let journal_clone = journal_writer.clone();
        let idempotency_result = async {
            tokio::task::spawn_blocking(move || journal_clone.get_by_request_id(&req_id_clone))
                .await
        }
        .instrument(idempotency_span)
        .await;

        match idempotency_result {
            Ok(Ok(Some(existing))) => {
                // Idempotent hit - return cached response
                debug!("Journal idempotency hit for request_id: {}", req_id);
                counter!("ht_engine_journal_idempotency_hit_total").increment(1);

                if let Some(cached_response) = existing.decode_response() {
                    let _ = request.response_tx.send(cached_response).await;
                    return;
                }
                panic!(
                    "JOURNAL_FATAL: no response_data in cached journal record \
                     for request_id={}",
                    req_id
                );
            }
            Ok(Ok(None)) => {
                // request_id was in cache but not in DB - cache may have been stale
                // or this is a re-used request_id within this session (shouldn't happen with UUIDs)
                debug!(
                    "Journal miss for request_id: {} (was in cache) - processing",
                    req_id
                );
            }
            Ok(Err(e)) => {
                error!("Journal query failed for request_id: {}: {}", req_id, e);
                counter!("ht_engine_journal_error_total", "reason" => "query_error").increment(1);
                panic!(
                    "JOURNAL_FATAL: journal query failed for request_id={}: {}. \
                     Cannot guarantee idempotency, restart required.",
                    req_id, e
                );
            }
            Err(e) => {
                error!("spawn_blocking failed for idempotency check: {}", e);
                counter!("ht_engine_journal_error_total", "reason" => "spawn_error").increment(1);
                panic!(
                    "JOURNAL_FATAL: spawn_blocking failed for idempotency check: {}. \
                     Runtime is degraded, restart required.",
                    e
                );
            }
        }

        // Slow path fallthrough: request_id was in cache but not in DB.
        // Process the order (batched or sync path).
        let strategy = JournalStrategy::select(&self.journal_batch_sender, journal_writer);
        self.process_order_journaled_impl(request, req_id, req_uuid, start, strategy)
            .await;

        // request_id is already in known_request_ids (that's why we hit the slow path),
        // but we processed it so no need to insert again.
    }

    /// Process order with journaling using the specified strategy.
    async fn process_order_journaled_impl(
        &mut self,
        request: UnifiedEngineRequest,
        req_id: &str,
        req_uuid: uuid::Uuid,
        start: Instant,
        strategy: JournalStrategy,
    ) {
        let engine_span = info_span!("engine_step");
        let timestamp = request.message.timestamp;

        use crate::observability::command_trace::EngineStateDigest;
        let pre_digest: EngineStateDigest = Default::default();

        let original_response_tx = request.response_tx;

        let env = crate::rsm::apply::CommandEnvelope::new(
            timestamp,
            crate::rsm::apply::EngineCommand::OrderAction(request.message.clone()),
        );
        let apply_output = match engine_span.in_scope(|| self.apply(env)) {
            Ok(output) => output,
            Err(crate::rsm::unified_engine::apply_interface::EngineError::Rejected(reason)) => {
                warn!(
                    request_id = %req_id,
                    reason = %reason,
                    "Order rejected by engine (nonce or admission)"
                );
                counter!("ht_engine_order_rejected_total", "reason" => "nonce").increment(1);
                let rejection = OrderUpdateMessage {
                    timestamp,
                    info: request.message.info.clone(),
                    status: hypercall_types::OrderUpdateStatus::Rejected,
                    reason: Some(reason),
                    filled_size: rust_decimal::Decimal::ZERO,
                    order_id: None,
                    wallet_address: request.message.wallet,
                    mmp_triggered: false,
                    request_id: request.message.request_id.clone(),
                };
                let _ = original_response_tx.send(rejection).await;
                self.known_request_ids.insert(req_uuid);
                return;
            }
            Err(e) => {
                panic!(
                    "JOURNAL_FATAL: apply() failed for request_id={}: {}",
                    req_id, e
                )
            }
        };
        let engine_duration = start.elapsed();
        histogram!("ht_engine_step_seconds").record(engine_duration.as_secs_f64());

        #[cfg(feature = "rsm-state")]
        let identity_hash = apply_output.command_identity_hash;
        let captured_balance_updates = apply_output.balance_updates;
        let captured_events = apply_output.events;
        let response = apply_output
            .order_responses
            .into_iter()
            .last()
            .unwrap_or_else(|| {
                error!(
                    "No response received from apply() for request_id: {}",
                    req_id
                );
                counter!("ht_engine_journal_error_total", "reason" => "no_response").increment(1);
                panic!(
                    "JOURNAL_FATAL: apply() emitted no response for request_id={}. \
                     This violates engine response invariants.",
                    req_id
                );
            });

        // For the journal, we need the FINAL order status, not just the preliminary ACK.
        // `allocate_and_ack` sends an Acked response to response_tx, but `finalize_order`
        // sends the final status (Filled/Open/PartiallyFilled/Canceled) only to event_sender.
        // The proxy channel thus contains Acked, but the real final status is in the captured
        // events. Without this fix, all create orders would be journaled with Acked status,
        // causing replay to add them all to the book (including fully-filled takers and
        // MMP-cancelled orders), leading to state regressions after SIGKILL.
        let journal_response = if request.message.action == OrderAction::CreateOrder
            || request.message.action == OrderAction::ReplaceOrder
        {
            // Scan captured events for the LAST OrderUpdate matching this order_id.
            // This captures the final status from finalize_order.
            captured_events
                .iter()
                .rev()
                .find_map(|e| {
                    if let EngineMessage::OrderUpdate(update) = e {
                        if update.order_id == response.order_id {
                            Some(update.clone())
                        } else {
                            None
                        }
                    } else {
                        None
                    }
                })
                .unwrap_or_else(|| {
                    if response.status == OrderUpdateStatus::Acked {
                        panic!(
                            "JOURNAL_FATAL: no OrderUpdate captured for order_id={:?}, \
                             request_id={}: no OrderUpdate found in {} captured events. \
                             finalize_order did not emit a terminal update; refusing to journal Acked.",
                            response.order_id,
                            req_id,
                            captured_events.len()
                        );
                    }
                    response.clone()
                })
        } else {
            // Cancel orders send the final response directly to response_tx
            response.clone()
        };

        let post_digest: EngineStateDigest = Default::default();
        #[cfg(feature = "rsm-state")]
        let post_rsm_state_digest = crate::rsm::engine_snapshot::EngineStateDigest::from_ctx(
            &self.ctx,
            self.ctx
                .l2_update_seq
                .load(std::sync::atomic::Ordering::SeqCst),
        );
        let duration_ms = start.elapsed().as_millis() as u64;

        // Serialize command and response for journal (wire format:
        // [version byte][msgpack payload]).
        // Patch the command with the engine-assigned order_id from the response
        // so journal replay can reconstruct orders correctly.
        // For ReplaceOrder, info.order_id is the cancel target - don't overwrite it.
        // The new order's ID is available in the journal response for replay.
        let mut command_for_journal = request.message.clone();
        if request.message.action != OrderAction::ReplaceOrder {
            if let Some(order_id) = journal_response.order_id {
                command_for_journal.info.order_id = Some(order_id);
            }
        }
        let command_data = hypercall_types::serialize_to_wire_bytes(&command_for_journal);

        let response_data = Some(hypercall_types::serialize_to_wire_bytes(&journal_response));
        let order_id_i64 = journal_response.order_id.map(|id| {
            i64::try_from(id).unwrap_or_else(|_| {
                panic!(
                    "JOURNAL_FATAL: order_id {} exceeds i64 range for request_id={}",
                    id, req_id
                )
            })
        });
        let command_type_enum = Some(match request.message.action {
            OrderAction::CreateOrder => hypercall_db_diesel::engine_enums::CommandType::CreateOrder,
            OrderAction::CancelOrder => hypercall_db_diesel::engine_enums::CommandType::CancelOrder,
            OrderAction::ReplaceOrder => {
                hypercall_db_diesel::engine_enums::CommandType::ReplaceOrder
            }
        });

        // Serialize events for batched journal path.
        let event_payloads = self.serialize_events_for_journal(&captured_events);
        let portfolio_projection_guard =
            if let Some(portfolio_cache) = &self.ctx.deps.portfolio_cache {
                Some(portfolio_cache.lock_projection_barrier().await)
            } else {
                None
            };
        let fill_side_effects = fill_side_effects_from_events(&captured_events);

        let req_uuid = hypercall_db_diesel::engine_enums::DbUuid(req_uuid);

        // Execute strategy-specific journaling
        match strategy {
            JournalStrategy::Batched(sender) => {
                // Create commit_ack channel so we wait for DB commit before emitting
                // events or ACKing the client. This prevents SIGKILL from causing
                // state regressions (client saw cancel, but journal didn't persist it).
                let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();

                let entry = crate::journal::JournalEntry {
                    received_ts_ms: timestamp,
                    command_data: command_data.clone(),
                    response_data: response_data.clone(),
                    order_id: order_id_i64,
                    pre_digest: pre_digest.clone(),
                    post_digest: post_digest.clone(),
                    duration_ms,
                    events: event_payloads,
                    outbox_appends: Vec::new(),
                    fill_side_effects: fill_side_effects.clone(),
                    cash_withdrawal_side_effect: None,
                    balance_updates: captured_balance_updates.clone(),
                    created_at: Instant::now(),
                    commit_ack: Some(ack_tx),
                    request_uuid: req_uuid,
                    command_type_enum,
                    #[cfg(feature = "rsm-state")]
                    command_identity_hash: identity_hash,
                    #[cfg(feature = "rsm-state")]
                    rsm_state_digest: Some(post_rsm_state_digest.clone()),
                };

                // Push to batch channel (non-blocking unless channel is full)
                let push_span = info_span!("journal_batch_push");

                async {
                    let msg = crate::journal::JournalMessage::Entry(entry);
                    match sender.try_send(msg) {
                        Ok(()) => {
                            counter!("ht_journal_batch_pushed_total").increment(1);
                            debug!("Pushed journal entry for request_id: {}", req_id);
                        }
                        Err(mpsc::error::TrySendError::Full(msg)) => {
                            // Channel full - block until space available (backpressure)
                            warn!("Journal batch channel full, applying backpressure");
                            counter!("ht_journal_batch_backpressure_total").increment(1);
                            gauge!("ht_journal_batch_channel_full").set(1.0);

                            if let Err(e) = sender.send(msg).await {
                                panic!(
                                    "CRITICAL_FAILURE: Failed to send journal entry for request_id {}: {}. \
                                     Journal channel closed - idempotency will be broken. Restart required.",
                                    req_id, e
                                );
                            }
                            gauge!("ht_journal_batch_channel_full").set(0.0);
                        }
                        Err(mpsc::error::TrySendError::Closed(_)) => {
                            panic!(
                                "CRITICAL_FAILURE: Journal batch channel closed for request_id {}. \
                                 Journal system is dead - idempotency will be broken. Restart required.",
                                req_id
                            );
                        }
                    }
                }
                .instrument(push_span)
                .await;

                // Wait for the WAL-first journal batcher to durably fsync this entry.
                // Only THEN emit events and ACK the client - this guarantees that
                // anything the client sees is durable and will survive SIGKILL.
                match ack_rx.await {
                    Ok(()) => {}
                    Err(_) => {
                        panic!(
                            "CRITICAL_FAILURE: Journal commit_ack dropped for request_id {}. \
                             Durability boundary is unknown after state mutation. \
                             Engine must stop to avoid running with non-durable state.",
                            req_id
                        );
                    }
                }

                // NATS publish after WAL durability confirmed — standby only
                // sees commands that survived fsync.
                if let Some(ref publisher) = self.nats_publisher {
                    publisher
                        .publish(crate::nats::CommandType::Order, &command_data)
                        .await;
                }
                self.publish_balance_updates_to_nats(&captured_balance_updates)
                    .await;

                // Keep portfolio state in sync before emitting events.
                let pending_notifications = self
                    .apply_portfolio_projection_events_sync(&captured_events, req_id)
                    .await;

                // NOW emit events and ACK - data is durable
                for event in &captured_events {
                    self.ctx.deps.emit_event(event);
                }
                drop(portfolio_projection_guard);

                self.send_portfolio_notifications(pending_notifications)
                    .await;

                self.publish_snapshot();
                let _ = original_response_tx.send(response).await;
                counter!("ht_engine_journal_commits_total").increment(1);
                histogram!("ht_engine_journaled_seconds").record(start.elapsed().as_secs_f64());
            }

            JournalStrategy::Sync(journal_writer) => {
                // Synchronous write with spawn_blocking
                let postgres_span = info_span!("journal_postgres_insert");
                let req_uuid_for_journal = req_uuid;
                let command_data_for_journal = command_data.clone();
                let response_data_for_journal = response_data.clone();
                let order_id_for_journal = order_id_i64;
                let pre_digest_for_journal = pre_digest.clone();
                let post_digest_for_journal = post_digest.clone();
                let events_for_journal = captured_events.clone();
                let fill_side_effects_for_journal = fill_side_effects.clone();
                let balance_updates_for_journal = captured_balance_updates.clone();
                let command_type_enum_for_journal = command_type_enum;
                let append_result = async {
                    let pg_start = Instant::now();
                    let result = tokio::task::spawn_blocking(move || {
                        journal_writer.append_transition_with_fill_side_effects(
                            timestamp,
                            &command_data_for_journal,
                            response_data_for_journal.as_deref(),
                            order_id_for_journal,
                            &pre_digest_for_journal,
                            &post_digest_for_journal,
                            duration_ms,
                            &events_for_journal,
                            &fill_side_effects_for_journal,
                            &balance_updates_for_journal,
                            req_uuid_for_journal,
                            command_type_enum_for_journal,
                        )
                    })
                    .await;
                    histogram!("ht_journal_postgres_insert_seconds")
                        .record(pg_start.elapsed().as_secs_f64());
                    result
                }
                .instrument(postgres_span)
                .await;

                match append_result {
                    Ok(Ok(result)) => {
                        let pending_notifications = if result.was_new_insert {
                            debug!(
                                "Journaled request_id: {} with command_id: {}",
                                req_id, result.command_id
                            );
                            counter!("ht_engine_journal_commits_total").increment(1);

                            if let Some(ref publisher) = self.nats_publisher {
                                publisher
                                    .publish(crate::nats::CommandType::Order, &command_data)
                                    .await;
                            }
                            self.publish_balance_updates_to_nats(&captured_balance_updates)
                                .await;

                            Some(
                                self.apply_portfolio_projection_events_sync(
                                    &captured_events,
                                    req_id,
                                )
                                .await,
                            )
                        } else {
                            debug!(
                                "Journal race: request_id: {} already committed by command_id: {}",
                                req_id, result.command_id
                            );
                            counter!("ht_engine_journal_race_total").increment(1);
                            None
                        };

                        // Emit events and send response - journal is durable
                        for event in &captured_events {
                            self.ctx.deps.emit_event(event);
                        }
                        drop(portfolio_projection_guard);

                        if let Some(pending_notifications) = pending_notifications {
                            self.send_portfolio_notifications(pending_notifications)
                                .await;
                        }

                        self.publish_snapshot();
                        let _ = original_response_tx.send(response).await;
                    }
                    Ok(Err(e)) => {
                        // CRITICAL: State was mutated but journal append failed.
                        // We MUST NOT continue running with mutated-but-unlogged state.
                        error!(
                            "FATAL: Journal append failed for request_id: {} after state mutation: {}",
                            req_id, e
                        );
                        counter!("ht_engine_journal_error_total", "reason" => "append_failed")
                            .increment(1);
                        panic!(
                            "JOURNAL_FATAL: Journal append failed after state mutation for request_id={}: {}. \
                            Engine state is now inconsistent and unrecoverable. \
                            Restart required to restore from last consistent snapshot.",
                            req_id, e
                        );
                    }
                    Err(e) => {
                        // CRITICAL: spawn_blocking failed
                        error!(
                            "FATAL: spawn_blocking failed for journal write, request_id: {}: {}",
                            req_id, e
                        );
                        counter!("ht_engine_journal_error_total", "reason" => "spawn_error")
                            .increment(1);
                        panic!(
                            "JOURNAL_FATAL: spawn_blocking failed for journal write, request_id={}: {}. \
                            Engine state is now inconsistent and unrecoverable. \
                            Restart required to restore from last consistent snapshot.",
                            req_id, e
                        );
                    }
                }

                histogram!("ht_engine_order_processing_journaled_seconds")
                    .record(start.elapsed().as_secs_f64());
            }
        };
    }

    /// Serialize events for journal.
    /// Produces wire-format bytes (version byte + msgpack) along with topic, key, and l2_sequence.
    /// These bytes are stored directly in the DB, avoiding double deserialization.
    fn serialize_events_for_journal(
        &self,
        events: &[EngineMessage],
    ) -> Vec<crate::journal::engine_journal_batcher::EventPayload> {
        use crate::journal::engine_journal_batcher::EventPayload;
        events
            .iter()
            .map(|e| {
                let event_data = e
                    .serialize_inner()
                    .expect("Failed to serialize EngineMessage - this is a bug");
                let event_topic = e.topic().to_string();
                let event_key = e.partition_key();
                let l2_sequence = match e {
                    EngineMessage::L2Update(l2) => l2.sequence,
                    _ => None,
                };
                EventPayload {
                    event_type_enum:
                        hypercall_db_diesel::engine_enums::EventType::from_engine_message(e),
                    event_topic,
                    event_key,
                    event_data,
                    l2_sequence,
                }
            })
            .collect()
    }

    /// Apply captured fill events to PortfolioCache synchronously.
    ///
    /// This guarantees that the next margin check sees post-fill portfolio state
    /// without waiting for event-bus round trips.
    pub(crate) async fn apply_portfolio_projection_events_sync(
        &self,
        events: &[EngineMessage],
        req_id: &str,
    ) -> Vec<crate::read_cache::portfolio::PendingNotifications> {
        let Some(portfolio_cache) = &self.ctx.deps.portfolio_cache else {
            return Vec::new();
        };

        let mut pending_notifications = Vec::new();

        for event in events {
            match event {
                EngineMessage::OrderFilled { fill, .. } => {
                    let seq = i64::try_from(fill.trade_id).unwrap_or_else(|_| {
                        panic!(
                            "CRITICAL_FAILURE: trade_id {} does not fit into i64 for synchronous portfolio update",
                            fill.trade_id
                        )
                    });
                    portfolio_cache
                        .handle_order_filled_under_barrier(fill, seq)
                        .await;
                    pending_notifications.push(
                        crate::read_cache::portfolio::PendingNotifications::Fill {
                            fill: fill.clone(),
                        },
                    );
                    debug!(
                        "Applied synchronous portfolio fill update: request_id={}, trade_id={}, symbol={}",
                        req_id, fill.trade_id, fill.symbol
                    );
                }
                EngineMessage::PositionExpired(expiry_msg) => {
                    pending_notifications.push(
                        portfolio_cache
                            .handle_position_expired_under_barrier(expiry_msg, None)
                            .await,
                    );
                    debug!(
                        "Applied synchronous portfolio expiry update: request_id={}, wallet={}, symbol={}",
                        req_id, expiry_msg.wallet_address, expiry_msg.symbol
                    );
                }
                _ => {}
            }
        }

        pending_notifications
    }

    /// Send portfolio notifications after the projection barrier has been
    /// released. This keeps WebSocket sends off the hot
    /// mutation path.
    async fn send_portfolio_notifications(
        &self,
        pending_notifications: Vec<crate::read_cache::portfolio::PendingNotifications>,
    ) {
        let Some(portfolio_cache) = &self.ctx.deps.portfolio_cache else {
            return;
        };

        for pending in pending_notifications {
            portfolio_cache.send_pending_notifications(pending).await;
        }
    }

    pub(crate) async fn apply_replayed_events_sync(
        &mut self,
        events: &[EngineMessage],
        req_id: &str,
    ) {
        if events.is_empty() {
            return;
        }

        let portfolio_projection_guard =
            if let Some(portfolio_cache) = &self.ctx.deps.portfolio_cache {
                Some(portfolio_cache.lock_projection_barrier().await)
            } else {
                None
            };
        let pending_notifications = self
            .apply_portfolio_projection_events_sync(events, req_id)
            .await;
        for event in events {
            self.ctx.deps.emit_event(event);
        }
        drop(portfolio_projection_guard);
        self.send_portfolio_notifications(pending_notifications)
            .await;
    }

    pub(crate) async fn apply_startup_replayed_events_sync(
        &mut self,
        events: &[EngineMessage],
        req_id: &str,
    ) {
        if events.is_empty() {
            return;
        }

        let Some(portfolio_cache) = self.ctx.deps.portfolio_cache.clone() else {
            for event in events {
                self.ctx.deps.emit_event(event);
            }
            return;
        };

        let portfolio_projection_guard = portfolio_cache.lock_projection_barrier().await;
        let mut pending_notifications = Vec::new();

        for event in events {
            if let EngineMessage::PositionExpired(expiry_msg) = event {
                pending_notifications.push(
                    portfolio_cache
                        .handle_replayed_position_expired_projection_under_barrier(expiry_msg)
                        .await,
                );
                debug!(
                    "Applied startup replay expiry projection: request_id={}, wallet={}, symbol={}",
                    req_id, expiry_msg.wallet_address, expiry_msg.symbol
                );
            }
        }

        for event in events {
            self.ctx.deps.emit_event(event);
        }
        drop(portfolio_projection_guard);
        self.send_portfolio_notifications(pending_notifications)
            .await;
    }

    /// Process an RFQ execution with durable journaling.
    ///
    /// Mirrors `process_order_journaled` but for `RfqExecuteCommand`. The
    /// caller (`handle_rfq_execute`) has already produced the planned
    /// `Vec<EngineMessage>` (per-leg `OrderFilled`s + summary `RfqFilled`)
    /// via the pure `plan_rfq_execution` function — this method is
    /// responsible for:
    ///
    /// 1. Translating engine-emitted `FillAccounting` into journal fill side effects.
    /// 2. Serializing the command + events as a journal entry tagged
    ///    `CommandType::RfqExecute`.
    /// 3. Pushing to the journal batcher (or sync writer) and waiting for
    ///    the durable commit ACK.
    /// 4. Applying the portfolio projection under barrier (uses
    ///    `apply_portfolio_projection_events_sync`, the same function the
    ///    orderbook path uses).
    /// 5. Emitting the events to the WS bus AFTER the journal commit, so
    ///    nothing visible to the client/UI happens before durability.
    /// 6. Firing post-journal MMP updates (which are intentionally
    ///    non-durable side effects).
    /// 7. Sending the `RfqExecuteResult::Success { fill_id }` response.
    ///
    /// On any journal failure this panics rather than returning, matching
    /// the orderbook journaling semantics — there is no safe way to
    /// continue running with mutated-but-unlogged state.
    pub(super) async fn process_rfq_journaled(
        &mut self,
        request: hypercall_runtime_api::RfqExecuteRequest,
        fill_id: String,
        captured_events: Vec<hypercall_types::EngineMessage>,
        captured_balance_updates: Vec<hypercall_types::BalanceUpdate>,
        mmp_updates: Vec<crate::rsm::apply::MmpFillUpdate>,
    ) {
        let start = Instant::now();
        let cmd = &request.command;

        // Idempotency cache lookup happens in `handle_rfq_execute` before
        // we even reach planning, so this function only ever sees
        // first-time-or-cache-stale commands. We still derive a stable
        // request_uuid here so the journal entry is keyed correctly.
        let req_id = if cmd.request_id.is_empty() {
            warn!(
                "RFQ command missing request_id (rfq_id={}, quote_id={}); \
                 idempotency tracking disabled for this request",
                cmd.rfq_id, cmd.quote_id
            );
            counter!("ht_engine_journal_fallback_total", "reason" => "missing_request_id")
                .increment(1);
            // Synthesize a deterministic UUID from (rfq_id, quote_id) so
            // the journal still has *something* keyed to this command.
            // We don't have the `uuid` crate's v5 feature enabled, so
            // build the UUID by hashing the pair via sha3 and slicing
            // the first 16 bytes — same effective property (deterministic
            // for the same input, well-distributed) without pulling in
            // another feature flag.
            use sha3::{Digest, Sha3_256};
            let mut hasher = Sha3_256::new();
            hasher.update(cmd.rfq_id.as_bytes());
            hasher.update(b":");
            hasher.update(cmd.quote_id.as_bytes());
            let digest = hasher.finalize();
            let mut bytes = [0u8; 16];
            bytes.copy_from_slice(&digest[..16]);
            // Mark as v8 (custom) per RFC 4122 §5.8.
            bytes[6] = (bytes[6] & 0x0F) | 0x80;
            bytes[8] = (bytes[8] & 0x3F) | 0x80;
            uuid::Uuid::from_bytes(bytes).to_string()
        } else {
            cmd.request_id.clone()
        };

        let req_uuid = uuid::Uuid::parse_str(&req_id).unwrap_or_else(|e| {
            panic!(
                "JOURNAL_FATAL: rfq command request_id '{}' is not a valid UUID: {}",
                req_id, e
            )
        });

        // Serialize command for journal. Wire format:
        // [version_byte][msgpack payload]. We use the existing
        // `serialize_to_wire_bytes` helper if it exists for RFQ, else
        // build it inline.
        let command_data = {
            let mut buf = Vec::with_capacity(256);
            // Version byte 1, matching the orderbook command format.
            buf.push(1u8);
            rmp_serde::encode::write_named(&mut buf, cmd).unwrap_or_else(|e| {
                panic!(
                    "JOURNAL_FATAL: failed to serialize RfqExecuteCommand for rfq_id={}: {}",
                    cmd.rfq_id, e
                )
            });
            buf
        };

        // RFQ doesn't have an OrderUpdateMessage equivalent, but we
        // still want to cache the planned fill_id so duplicate dispatches
        // get the SAME fill_id back instead of a fresh plan. Encode the
        // fill_id as the response_data so `lookup_cached_rfq_response`
        // can deserialize it.
        let response_data: Option<Vec<u8>> =
            Some(super::UnifiedEngine::encode_rfq_response_data(&fill_id));

        let event_payloads = self.serialize_events_for_journal(&captured_events);
        let portfolio_projection_guard =
            if let Some(portfolio_cache) = &self.ctx.deps.portfolio_cache {
                Some(portfolio_cache.lock_projection_barrier().await)
            } else {
                None
            };
        let fill_side_effects = fill_side_effects_from_events(&captured_events);

        let timestamp = get_timestamp_millis();
        use crate::observability::command_trace::EngineStateDigest;
        let pre_digest: EngineStateDigest = Default::default();
        let post_digest: EngineStateDigest = Default::default();
        #[cfg(feature = "rsm-state")]
        let post_rsm_state_digest = crate::rsm::engine_snapshot::EngineStateDigest::from_ctx(
            &self.ctx,
            self.ctx
                .l2_update_seq
                .load(std::sync::atomic::Ordering::SeqCst),
        );
        let duration_ms = start.elapsed().as_millis() as u64;
        let req_uuid_db = hypercall_db_diesel::engine_enums::DbUuid(req_uuid);
        let command_type_enum = Some(hypercall_db_diesel::engine_enums::CommandType::RfqExecute);

        let journal_writer = match &self.journal_writer {
            Some(w) => w.clone(),
            None => {
                error!(
                    "Journal unavailable for RFQ: no writer configured. \
                     Use .with_mock_journal() for tests."
                );
                panic!(
                    "JOURNAL_FATAL: no journal_writer configured for RFQ rfq_id={}. \
                     Refusing to process without durability.",
                    cmd.rfq_id
                );
            }
        };

        let strategy = JournalStrategy::select(&self.journal_batch_sender, journal_writer);

        match strategy {
            JournalStrategy::Batched(sender) => {
                let (ack_tx, ack_rx) = tokio::sync::oneshot::channel();

                let entry = crate::journal::JournalEntry {
                    received_ts_ms: timestamp,
                    command_data: command_data.clone(),
                    response_data: response_data.clone(),
                    order_id: None,
                    pre_digest: pre_digest.clone(),
                    post_digest: post_digest.clone(),
                    duration_ms,
                    events: event_payloads,
                    outbox_appends: Vec::new(),
                    fill_side_effects: fill_side_effects.clone(),
                    cash_withdrawal_side_effect: None,
                    balance_updates: captured_balance_updates.clone(),
                    created_at: Instant::now(),
                    commit_ack: Some(ack_tx),
                    request_uuid: req_uuid_db,
                    command_type_enum,
                    #[cfg(feature = "rsm-state")]
                    command_identity_hash: crate::rsm::apply::EngineCommand::RfqExecute(
                        cmd.clone(),
                    )
                    .identity_hash(),
                    #[cfg(feature = "rsm-state")]
                    rsm_state_digest: Some(post_rsm_state_digest.clone()),
                };

                let push_span = info_span!("journal_batch_push_rfq");
                async {
                    let msg = crate::journal::JournalMessage::Entry(entry);
                    match sender.try_send(msg) {
                        Ok(()) => {
                            counter!("ht_journal_batch_pushed_total", "command" => "RfqExecute")
                                .increment(1);
                        }
                        Err(mpsc::error::TrySendError::Full(msg)) => {
                            warn!("Journal batch channel full (RFQ), applying backpressure");
                            counter!(
                                "ht_journal_batch_backpressure_total",
                                "command" => "RfqExecute"
                            )
                            .increment(1);
                            if let Err(e) = sender.send(msg).await {
                                panic!(
                                    "CRITICAL_FAILURE: Failed to send RFQ journal entry for rfq_id {}: {}. \
                                     Journal channel closed - idempotency will be broken. Restart required.",
                                    cmd.rfq_id, e
                                );
                            }
                        }
                        Err(mpsc::error::TrySendError::Closed(_)) => {
                            panic!(
                                "CRITICAL_FAILURE: Journal batch channel closed for RFQ rfq_id {}. \
                                 Journal system is dead - idempotency will be broken. Restart required.",
                                cmd.rfq_id
                            );
                        }
                    }
                }
                .instrument(push_span)
                .await;

                // Wait for the WAL-first journal batcher to durably fsync
                // this entry. ONLY THEN apply the projection / emit events
                // / ACK — this is the durability boundary.
                let wal_wait_start = Instant::now();
                match ack_rx.await {
                    Ok(()) => {}
                    Err(_) => {
                        panic!(
                            "CRITICAL_FAILURE: RFQ journal commit_ack dropped for rfq_id {}. \
                             Durability boundary unknown after planning; engine must stop.",
                            cmd.rfq_id
                        );
                    }
                }
                histogram!("ht_rfq_wal_wait_seconds")
                    .record(wal_wait_start.elapsed().as_secs_f64());

                let nats_start = Instant::now();
                if let Some(ref publisher) = self.nats_publisher {
                    publisher
                        .publish(crate::nats::CommandType::RfqExecute, &command_data)
                        .await;
                }
                self.publish_balance_updates_to_nats(&captured_balance_updates)
                    .await;
                histogram!("ht_rfq_nats_publish_seconds")
                    .record(nats_start.elapsed().as_secs_f64());

                let projection_start = Instant::now();
                let pending_notifications = self
                    .apply_portfolio_projection_events_sync(&captured_events, &req_id)
                    .await;
                histogram!("ht_rfq_projection_seconds")
                    .record(projection_start.elapsed().as_secs_f64());

                for event in &captured_events {
                    self.ctx.deps.emit_event(event);
                }
                drop(portfolio_projection_guard);

                let notify_start = Instant::now();
                self.send_portfolio_notifications(pending_notifications)
                    .await;
                histogram!("ht_rfq_notify_seconds").record(notify_start.elapsed().as_secs_f64());

                counter!("ht_engine_journal_commits_total", "command" => "RfqExecute").increment(1);
                histogram!("ht_engine_journaled_seconds", "command" => "RfqExecute")
                    .record(start.elapsed().as_secs_f64());
            }

            JournalStrategy::Sync(journal_writer) => {
                let postgres_span = info_span!("journal_postgres_insert_rfq");
                let req_uuid_for_journal = req_uuid_db;
                let command_data_for_journal = command_data.clone();
                let response_data_for_journal = response_data.clone();
                let pre_digest_for_journal = pre_digest.clone();
                let post_digest_for_journal = post_digest.clone();
                let events_for_journal = captured_events.clone();
                let fill_side_effects_for_journal = fill_side_effects.clone();
                let balance_updates_for_journal = captured_balance_updates.clone();
                let command_type_enum_for_journal = command_type_enum;

                let append_result = async {
                    tokio::task::spawn_blocking(move || {
                        journal_writer.append_transition_with_fill_side_effects(
                            timestamp,
                            &command_data_for_journal,
                            response_data_for_journal.as_deref(),
                            None,
                            &pre_digest_for_journal,
                            &post_digest_for_journal,
                            duration_ms,
                            &events_for_journal,
                            &fill_side_effects_for_journal,
                            &balance_updates_for_journal,
                            req_uuid_for_journal,
                            command_type_enum_for_journal,
                        )
                    })
                    .await
                }
                .instrument(postgres_span)
                .await;

                match append_result {
                    Ok(Ok(result)) => {
                        let pending_notifications = if result.was_new_insert {
                            counter!("ht_engine_journal_commits_total", "command" => "RfqExecute")
                                .increment(1);

                            if let Some(ref publisher) = self.nats_publisher {
                                publisher
                                    .publish(crate::nats::CommandType::RfqExecute, &command_data)
                                    .await;
                            }
                            self.publish_balance_updates_to_nats(&captured_balance_updates)
                                .await;

                            Some(
                                self.apply_portfolio_projection_events_sync(
                                    &captured_events,
                                    &req_id,
                                )
                                .await,
                            )
                        } else {
                            counter!("ht_engine_journal_race_total", "command" => "RfqExecute")
                                .increment(1);
                            None
                        };

                        for event in &captured_events {
                            self.ctx.deps.emit_event(event);
                        }
                        drop(portfolio_projection_guard);

                        if let Some(pending_notifications) = pending_notifications {
                            self.send_portfolio_notifications(pending_notifications)
                                .await;
                        }
                    }
                    Ok(Err(e)) => {
                        error!(
                            "FATAL: RFQ journal append failed for rfq_id={}: {}",
                            cmd.rfq_id, e
                        );
                        counter!(
                            "ht_engine_journal_error_total",
                            "reason" => "append_failed",
                            "command" => "RfqExecute"
                        )
                        .increment(1);
                        panic!(
                            "JOURNAL_FATAL: RFQ journal append failed for rfq_id={}: {}",
                            cmd.rfq_id, e
                        );
                    }
                    Err(e) => {
                        error!("FATAL: spawn_blocking failed for RFQ journal write: {}", e);
                        panic!(
                            "JOURNAL_FATAL: spawn_blocking failed for RFQ journal write: {}",
                            e
                        );
                    }
                }
            }
        }

        // Add to known set so future duplicates (within this session) hit
        // the fast path.
        if !cmd.request_id.is_empty() {
            self.known_request_ids.insert(req_uuid);
        }

        // Post-journal side effects: MMP fill updates. These are
        // intentionally non-durable — losing them on a crash is
        // acceptable, the next quote refresh from the QP will repopulate
        // MMP state. Running them after journal commit ensures the
        // visible-to-client durability boundary is unaffected.
        if let Some(ref mmp_cache) = self.ctx.deps.mmp_cache {
            for update in &mmp_updates {
                let _ = mmp_cache
                    .process_fill(
                        &update.qp_wallet,
                        &update.underlying,
                        &update.fill,
                        update.timestamp_ms,
                    )
                    .await;
            }
        }

        self.publish_snapshot();

        // ACK the caller with the planned fill_id.
        let _ = request
            .response_tx
            .send(hypercall_runtime_api::RfqExecuteResult::Success { fill_id });
    }
}

fn fill_side_effects_from_events(
    events: &[EngineMessage],
) -> Vec<crate::journal::JournalFillSideEffect> {
    events
        .iter()
        .enumerate()
        .filter_map(|(event_idx, event)| match event {
            EngineMessage::OrderFilled { fill, accounting } => {
                assert_eq!(
                    accounting.trade_id, fill.trade_id,
                    "CRITICAL: fill accounting trade_id mismatch for fill {}",
                    fill.trade_id
                );
                let mut side_effect = crate::journal::JournalFillSideEffect::from_fill_accounting(
                    event_idx as i32,
                    accounting,
                );
                side_effect.underlying_notional = fill.underlying_notional;
                Some(side_effect)
            }
            _ => None,
        })
        .collect()
}