billing: ledger spool, budget holds at admission, abandoned streams billed from the upstream's usage

Cargo.toml

@@ -35,6 +35,7 @@ undocumented_unsafe_blocks = "deny"
 [workspace.dependencies]
 # --- async runtime ---
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "time", "sync", "net", "signal", "io-util"] }
+tokio-util = { version = "0.7", features = ["rt"] }
 async-trait = "0.1"
 futures = "0.3"
 # --- web (server) ---

crates/config/src/lib.rs

@@ -580,6 +580,9 @@ pub struct StorageConf {
     /// Prune rolled billing records beyond this many; 0 = unlimited.
     #[serde(default)]
     pub ledger_max_rows: u64,
+    /// Local file holding ledger rows the store refuses until they replay; empty drops them after the retries.
+    #[serde(default)]
+    pub ledger_spool_path: String,
     /// Postgres pool size for the store and key-table pools; 0 = their built-in defaults.
     #[serde(default)]
     pub postgres_max_connections: u32,

crates/dag/src/context.rs

@@ -8,7 +8,7 @@ use std::sync::Arc;
 use gw_config::GatewayConfig;
 use gw_engines::{EngineOutcome, SharedTransport};
 use gw_models::{GResult, GatewayError, GatewayRequest, ModelParamV2};
-use gw_state::admission::TpmReserve;
+use gw_state::admission::{BudgetHold, TpmReserve};
 use gw_state::{AkInfo, GatewayState, UserBudget};
 
 pub struct DagContext {
@@ -28,6 +28,7 @@ pub struct DagContext {
     /// A fallback model retries this attempt's upstream fault, so its failure is not the outcome.
     pub fallback_ahead: bool,
     pub(crate) request_limits_admitted: bool,
+    pub(crate) reserve_parts: (i64, i64),
     /// This request's cache key (computed by cache_lookup, reused by cache_store).
     pub cache_key: Option<String>,
     /// Governance key for the (AK, model) daily counter; set only when a cap is configured.
@@ -40,8 +41,11 @@ pub struct DagContext {
     pub tpm_reserved: Option<TpmReserve>,
     /// The user's budget override resolved at admission, charged at settlement.
     pub user_budget: Option<UserBudget>,
+    /// Day-budget amounts held at admission (same lifecycle as `quota_reserved`).
+    pub budget_hold: BudgetHold,
     /// Outbound DLP buffered this stream; billing waits for the view's delivery result.
     pub billing_deferred: bool,
+    pub served_at: Option<std::time::Instant>,
 }
 
 impl DagContext {
@@ -63,13 +67,16 @@ impl DagContext {
             cache_hit: false,
             fallback_ahead: false,
             request_limits_admitted: false,
+            reserve_parts: (0, 0),
             cache_key: None,
             model_quota_key: None,
             quota_reserved: None,
             quota_at: 0,
             tpm_reserved: None,
             user_budget: None,
+            budget_hold: BudgetHold::default(),
             billing_deferred: false,
+            served_at: None,
         }
     }
 
@@ -85,6 +92,19 @@ impl DagContext {
             .ok_or_else(|| GatewayError::internal("model param missing after resolve_model"))
     }
 
+    pub async fn refund_reserves(&mut self) {
+        let gov = self.state.governance.as_ref();
+        tokio::join!(
+            gov.refund_reserves(
+                &self.ak.ak_id,
+                self.quota_reserved.take().unwrap_or(0),
+                self.tpm_reserved.take(),
+                self.quota_at,
+            ),
+            std::mem::take(&mut self.budget_hold).refund(gov),
+        );
+    }
+
     /// The effective end user: the key's `owner`, else request metadata, else `""`.
     pub fn effective_user_id(&self) -> &str {
         self.ak