diff --git a/src/api.rs b/src/api.rs index ca5382595..c15b68e43 100644 --- a/src/api.rs +++ b/src/api.rs @@ -3,7 +3,6 @@ use std::num::NonZeroUsize; use std::str::FromStr; use std::sync::Arc; -use std::time::Duration; use alloy::primitives::{TxHash, U256}; use axum::Json; @@ -33,7 +32,6 @@ use st0x_event_sorcery::{ send_command, }; use st0x_evm::Chain; -use st0x_execution::alpaca_broker_api::AccountActivitiesQuery; use st0x_execution::{AlpacaWalletError, Symbol}; use st0x_finance::{FractionalShares, Positive}; use st0x_float_serde::format_float; @@ -43,8 +41,7 @@ use st0x_tokenization::IssuerRequestId; use crate::AppState; use crate::cctp_burn::{CctpBurnOperation, CctpSourceChain}; use crate::dashboard::pnl::{ - PnlError, PnlQuery, PnlResponse, acquire_pnl_report_permit, build_pnl_report_with_permit, - validate_pnl_snapshot_rowid, + PnlError, PnlQuery, PnlReportDeps, PnlReportError, PnlResponse, run_pnl_report, }; use crate::dashboard::transfer_loader::{ InvalidTransferKind, TransferHistoryQuery, TransferKind, query_transfer_history, @@ -276,60 +273,38 @@ async fn health(State(state): State) -> (StatusCode, Json, Query(query): Query, ) -> Result, (StatusCode, String)> { - let after = query - .activity_after() - .map_err(|error| (StatusCode::BAD_REQUEST, error.to_string()))?; - let until = query - .activity_until() - .map_err(|error| (StatusCode::BAD_REQUEST, error.to_string()))?; - query - .symbol_filter(&mut Vec::new()) - .map_err(|error| (StatusCode::BAD_REQUEST, error.to_string()))?; - let head = tokio::time::timeout(PNL_CATCH_UP_TIMEOUT, state.pnl_ledger.catch_up()) - .await - .map_err(|_elapsed| { - warn!("PnL ledger catch-up exceeded its request deadline"); - ( - StatusCode::SERVICE_UNAVAILABLE, - "PnL ledger catch-up timed out".to_string(), - ) - })? - .map_err(|error| pnl_error_response(PnlError::Ledger(error)))?; - validate_pnl_snapshot_rowid(head, &query).map_err(pnl_error_response)?; - let permit = - acquire_pnl_report_permit(&state.pnl_report_admission).map_err(pnl_error_response)?; + let BrokerCtx::AlpacaBrokerApi(broker) = &state.ctx.broker; + let deps = PnlReportDeps { + pool: &state.pool, + ledger: &state.pnl_ledger, + broker, + }; - let BrokerCtx::AlpacaBrokerApi(alpaca_auth) = &state.ctx.broker; - let activities = alpaca_auth - .fetch_account_activities(&AccountActivitiesQuery::pnl(after, until)) + run_pnl_report(&deps, &query, &state.pnl_report_admission) .await - .map_err(|error| { + .map(Json) + .map_err(pnl_report_error_response) +} + +fn pnl_report_error_response(error: PnlReportError) -> (StatusCode, String) { + match error { + PnlReportError::Report(error) => pnl_error_response(error), + PnlReportError::CatchUpTimeout => ( + StatusCode::SERVICE_UNAVAILABLE, + "PnL ledger catch-up timed out".to_string(), + ), + PnlReportError::Activities(error) => { error!(%error, "Failed to fetch Alpaca account activities for PnL"); ( StatusCode::BAD_GATEWAY, "Failed to fetch Alpaca account activities".to_string(), ) - })?; - - build_pnl_report_with_permit(&state.pool, &query, activities, Utc::now(), permit, head) - .await - .map(Json) - .map_err(pnl_error_response) + } + } } fn pnl_error_response(error: PnlError) -> (StatusCode, String) { @@ -4758,6 +4733,7 @@ mod tests { use std::net::SocketAddr; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; + use std::time::Duration; use alloy::consensus::{SignableTransaction as _, TxEip1559, TxEnvelope}; use alloy::eips::eip2718::{EIP1559_TX_TYPE_ID, Encodable2718 as _}; @@ -4808,6 +4784,7 @@ mod tests { use super::*; use crate::bindings::IRaindexV6::{SignedContextV1, TakeOrderConfigV4, TakeOrderV3}; use crate::dashboard; + use crate::dashboard::pnl::{PnlReportAdmission, acquire_pnl_report_permit}; use crate::equity_redemption::redemption_aggregate_id; use crate::inventory::{ self, BroadcastingInventory, PortfolioAsset, PortfolioBalanceRow, PortfolioLocation, @@ -6084,6 +6061,78 @@ mod tests { .unwrap(); } + /// The state and the Alpaca mock it reads, which serves until dropped. + async fn pnl_test_state() -> (AppState, AlpacaBrokerMock) { + let broker_mock = AlpacaBrokerMock::start() + .symbol_fill_prices(vec![]) + .symbol_positions(vec![]) + .call() + .await; + let mut ctx = create_test_ctx_with_order_owner(Address::ZERO); + ctx.broker = crate::test_utils::mock_alpaca_broker_ctx(broker_mock.base_url()); + + (empty_app_state(ctx).await, broker_mock) + } + + async fn shared_pnl_report(state: &AppState, query: &PnlQuery) -> PnlResponse { + let BrokerCtx::AlpacaBrokerApi(broker) = &state.ctx.broker; + let deps = PnlReportDeps { + pool: &state.pool, + ledger: &state.pnl_ledger, + broker, + }; + + run_pnl_report(&deps, query, &PnlReportAdmission::with_permits(1)) + .await + .unwrap() + } + + #[tokio::test] + async fn pnl_route_returns_what_the_shared_report_path_builds() { + let (state, _broker_mock) = pnl_test_state().await; + seed_position_pnl_fill(&state.pool, &Symbol::new("RKLB").unwrap()).await; + let query = PnlQuery { + limit: Some(1), + from_date: Some("2026-03-01".to_string()), + to_date: Some("2026-03-09".to_string()), + ..PnlQuery::default() + }; + + let shared = serde_json::to_value(shared_pnl_report(&state, &query).await).unwrap(); + let response = build_app(state) + .oneshot( + Request::builder() + .uri("/pnl?limit=1&fromDate=2026-03-01&toDate=2026-03-09") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + let routed: serde_json::Value = + serde_json::from_str(&body_to_string(response).await).unwrap(); + assert_eq!(routed, shared); + assert_eq!(routed["summary"]["matchedLotCount"], serde_json::json!(1)); + } + + #[tokio::test] + async fn each_shared_report_sees_the_fills_recorded_before_it() { + let (state, _broker_mock) = pnl_test_state().await; + let query = PnlQuery { + limit: Some(1), + ..PnlQuery::default() + }; + + seed_position_pnl_fill(&state.pool, &Symbol::new("RKLB").unwrap()).await; + let first = shared_pnl_report(&state, &query).await; + seed_position_pnl_fill(&state.pool, &Symbol::new("TSLA").unwrap()).await; + let second = shared_pnl_report(&state, &query).await; + + assert_eq!(first.summary.matched_lot_count, 1); + assert_eq!(second.summary.matched_lot_count, 2); + } + /// End-to-end `/pnl` coverage for capital/return-on-capital figures: three /// portfolio_snapshot days are captured through the real /// aggregate (`Store::send`, never raw SQL), each `captured_at` built diff --git a/src/dashboard/pnl/mod.rs b/src/dashboard/pnl/mod.rs index 67ec62b24..f26c29354 100644 --- a/src/dashboard/pnl/mod.rs +++ b/src/dashboard/pnl/mod.rs @@ -24,10 +24,12 @@ pub(crate) use ledger::{LedgerHead, PnlLedger, PnlLedgerReactor}; pub(crate) use query::{PnlError, PnlQuery}; pub(crate) use response::PnlResponse; #[cfg(test)] -pub(crate) use source::{MAX_CONCURRENT_PNL_REPORTS, build_pnl_report}; pub(crate) use source::{ - PnlReportAdmission, acquire_pnl_report_permit, build_pnl_report_with_permit, - pnl_report_admission, validate_pnl_snapshot_rowid, + MAX_CONCURRENT_PNL_REPORTS, acquire_pnl_report_permit, build_pnl_report, + validate_pnl_snapshot_rowid, +}; +pub(crate) use source::{ + PnlReportAdmission, PnlReportDeps, PnlReportError, pnl_report_admission, run_pnl_report, }; const ATTRIBUTION_METHOD: &str = "backend_position_fill_replay_fifo"; diff --git a/src/dashboard/pnl/source.rs b/src/dashboard/pnl/source.rs index 99fd7ba4c..02dca5343 100644 --- a/src/dashboard/pnl/source.rs +++ b/src/dashboard/pnl/source.rs @@ -11,10 +11,13 @@ use rain_math_float::Float; use sqlx::{QueryBuilder, Sqlite, SqlitePool}; use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; +use std::time::Duration; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use tokio::task; +use tracing::warn; -use st0x_execution::alpaca_broker_api::AccountActivity; +use st0x_execution::alpaca_broker_api::{AccountActivitiesQuery, AccountActivity}; +use st0x_execution::{AlpacaBrokerApiCtx, AlpacaBrokerApiError}; use st0x_float_serde::format_float; use crate::portfolio_snapshot::{ @@ -23,7 +26,8 @@ use crate::portfolio_snapshot::{ use super::builder::build_pnl_response_from_rows; use super::ledger::{ - CCTP_FEE_SOURCE, DIRECTION_BUY_TEXT, DIRECTION_SELL_TEXT, LedgerHead, TOKENIZATION_FEE_SOURCE, + CCTP_FEE_SOURCE, DIRECTION_BUY_TEXT, DIRECTION_SELL_TEXT, LedgerHead, PnlLedger, + TOKENIZATION_FEE_SOURCE, }; use super::query::{PnlError, PnlQuery}; use super::response::{PnlCapitalSummary, PnlResponse}; @@ -43,7 +47,13 @@ pub(crate) struct PnlReportAdmission(Arc); impl PnlReportAdmission { fn new() -> Self { - Self(Arc::new(Semaphore::new(MAX_CONCURRENT_PNL_REPORTS))) + Self::with_permits(MAX_CONCURRENT_PNL_REPORTS) + } + + /// An admission of its own, so a caller that is not a live `/pnl` + /// request never takes one of the live request permits. + pub(crate) fn with_permits(permits: usize) -> Self { + Self(Arc::new(Semaphore::new(permits))) } fn try_acquire(&self) -> Result { @@ -87,6 +97,70 @@ where .map(|(result, _permit)| result) } +/// Deadline for bringing the PnL ledger current before one report. +/// Catch-up work is proportional to the un-ingested backlog, not to the +/// request, and concurrent callers serialize on the ledger's internal mutex +/// before any admission control -- so past this deadline the report sheds +/// instead of queueing behind ingestion. The boot-path catch-up stays +/// unbounded: first-deploy backfill may legitimately exceed any request +/// deadline. Cancellation is safe because each ingest batch commits its rows +/// and checkpoint atomically; an elapsed deadline only rolls back the +/// in-flight batch. +pub(crate) const PNL_CATCH_UP_TIMEOUT: Duration = Duration::from_secs(10); + +/// What one report reads besides the query. +pub(crate) struct PnlReportDeps<'a> { + pub(crate) pool: &'a SqlitePool, + pub(crate) ledger: &'a PnlLedger, + pub(crate) broker: &'a AlpacaBrokerApiCtx, +} + +#[derive(Debug, thiserror::Error)] +pub(crate) enum PnlReportError { + #[error(transparent)] + Report(#[from] PnlError), + #[error("PnL ledger catch-up exceeded its {:?} deadline", PNL_CATCH_UP_TIMEOUT)] + CatchUpTimeout, + #[error("failed to fetch Alpaca account activities")] + Activities(#[source] Box), +} + +/// One PnL report, the way `GET /pnl` builds it: validate the query, bring +/// the ledger current within [`PNL_CATCH_UP_TIMEOUT`], check `asOfRowid` +/// against the head, take a permit from `admission`, fetch the Alpaca +/// activities for the window and replay. Every caller runs its own catch-up, +/// so each report sees the events ingested before it. +pub(crate) async fn run_pnl_report( + deps: &PnlReportDeps<'_>, + query: &PnlQuery, + admission: &PnlReportAdmission, +) -> Result { + let after = query.activity_after()?; + let until = query.activity_until()?; + query.symbol_filter(&mut Vec::new())?; + + let head = tokio::time::timeout(PNL_CATCH_UP_TIMEOUT, deps.ledger.catch_up()) + .await + .map_err(|_elapsed| { + warn!("PnL ledger catch-up exceeded its deadline"); + PnlReportError::CatchUpTimeout + })? + .map_err(PnlError::Ledger)?; + validate_pnl_snapshot_rowid(head, query)?; + let permit = acquire_pnl_report_permit(admission)?; + + let activities = deps + .broker + .fetch_account_activities(&AccountActivitiesQuery::pnl(after, until)) + .await + .map_err(|error| PnlReportError::Activities(Box::new(error)))?; + + Ok( + build_pnl_report_with_permit(deps.pool, query, activities, Utc::now(), permit, head) + .await?, + ) +} + #[cfg(test)] pub(crate) async fn build_pnl_report( pool: &SqlitePool,