chio_store_sqlite/budget_store/
reaper.rs

1use super::*;
2
3use std::collections::HashMap;
4
5use chio_kernel::budget_store::{
6    BudgetCaptureInvocationRequest, BudgetReconcileHoldRequest, BudgetReverseHoldRequest,
7};
8
9/// Outcome of a startup reap pass over orphaned open holds.
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub struct ReapSummary {
12    pub reconciled: usize,
13    pub reversed: usize,
14}
15
16/// An open reserved hold past its TTL deadline:
17/// `(hold_id, capability_id, grant_index, remaining_exposure_units,
18/// invocation_captured, authority)`.
19type ExpiredReservedHold = (String, String, u32, u64, bool, Option<BudgetEventAuthority>);
20
21/// A hold still `open` at startup:
22/// `(hold_id, capability_id, grant_index, remaining_exposure_units,
23/// invocation_captured, authority)`.
24type OpenHold = (String, String, u32, u64, bool, Option<BudgetEventAuthority>);
25
26impl SqliteBudgetStore {
27    fn sqlite_like_prefix_pattern(prefix: &str) -> String {
28        let mut pattern = String::with_capacity(prefix.len() + 1);
29        for ch in prefix.chars() {
30            match ch {
31                '%' | '_' | '\\' => {
32                    pattern.push('\\');
33                    pattern.push(ch);
34                }
35                _ => pattern.push(ch),
36            }
37        }
38        pattern.push('%');
39        pattern
40    }
41
42    /// Reconcile or reverse every hold still `open` at startup. Holds present in
43    /// `realized_by_hold` (arbitrated by the ADR-0013 durable receipt log) are
44    /// reconciled to their realized spend; holds absent from it (never durably
45    /// admitted) are reversed. This is fail-closed against double-spend: a naive
46    /// blanket release is never used.
47    ///
48    /// Called by the `BudgetStore` trait implementation of `reap_orphaned_holds`.
49    pub fn reap_holds_by_map(
50        &self,
51        realized_by_hold: &HashMap<String, u64>,
52    ) -> Result<ReapSummary, BudgetStoreError> {
53        let open_holds = self.list_open_holds()?;
54        let mut summary = ReapSummary {
55            reconciled: 0,
56            reversed: 0,
57        };
58        let mut first_error = None;
59        for (hold_id, capability_id, grant_index, exposure, captured, authority) in open_holds {
60            // Kernel-authored holds carry a BudgetEventAuthority lease that the
61            // reconcile/reverse authority check enforces. Present each hold's stored
62            // authority so orphaned kernel holds are reclaimed rather than rejected;
63            // a hold whose authority columns cannot be loaded fails closed in
64            // `list_open_holds` rather than silently reaping with no authority.
65            let result = (|| match realized_by_hold.get(&hold_id) {
66                Some(&realized) => {
67                    if !captured {
68                        self.capture_invocation_reservations(BudgetCaptureInvocationRequest {
69                            capability_id: capability_id.clone(),
70                            grant_index: grant_index as usize,
71                            hold_id: hold_id.clone(),
72                            event_id: format!("{hold_id}:reap-capture-invocation"),
73                            trusted_time: None,
74                            authority: authority.clone(),
75                        })?;
76                    }
77                    if exposure > 0 {
78                        self.reconcile_budget_hold(BudgetReconcileHoldRequest {
79                            capability_id: capability_id.clone(),
80                            grant_index: grant_index as usize,
81                            exposed_cost_units: exposure,
82                            realized_spend_units: realized.min(exposure),
83                            hold_id: Some(hold_id.clone()),
84                            event_id: Some(format!("{hold_id}:reap-reconcile")),
85                            authority,
86                        })?;
87                    }
88                    Ok(true)
89                }
90                None if captured => {
91                    if exposure > 0 {
92                        self.reconcile_budget_hold(BudgetReconcileHoldRequest {
93                            capability_id: capability_id.clone(),
94                            grant_index: grant_index as usize,
95                            exposed_cost_units: exposure,
96                            realized_spend_units: exposure,
97                            hold_id: Some(hold_id.clone()),
98                            event_id: Some(format!("{hold_id}:reap-unknown-outcome")),
99                            authority,
100                        })?;
101                    }
102                    Ok(true)
103                }
104                None => {
105                    self.reverse_budget_hold(BudgetReverseHoldRequest {
106                        capability_id: capability_id.clone(),
107                        grant_index: grant_index as usize,
108                        reversed_exposure_units: exposure,
109                        hold_id: Some(hold_id.clone()),
110                        event_id: Some(format!("{hold_id}:reap-reverse")),
111                        authority,
112                        expected_cumulative_approval_state: None,
113                    })?;
114                    Ok(false)
115                }
116            })();
117            match result {
118                Ok(true) => summary.reconciled += 1,
119                Ok(false) => summary.reversed += 1,
120                Err(error) => {
121                    tracing::warn!(
122                        hold_id,
123                        error = %error,
124                        "orphan budget hold cleanup failed"
125                    );
126                    if first_error.is_none() {
127                        first_error = Some(error);
128                    }
129                }
130            }
131        }
132        first_error.map_or(Ok(summary), Err)
133    }
134
135    /// Settle every reserved hold that is still `open` and whose `reserved_until`
136    /// deadline is at or before `now_unix_secs` at its reserved worst-case,
137    /// forfeiting the reserved amount to realized spend. In the two-phase
138    /// reserve/reconcile flow the only evidence a spend occurred is the caller's
139    /// reconcile; an expired-and-unreconciled hold may correspond to a call that
140    /// executed and spent, so releasing it (realized 0) would under-count real
141    /// spend and fail open for a cumulative cap. The abandoning caller instead
142    /// forfeits the worst-case (must reconcile before expiry to reclaim the
143    /// difference). Fail-closed: only holds explicitly marked reserved (a non-NULL
144    /// `reserved_until`) and past expiry are touched; a not-yet-expired reserved
145    /// hold and any non-open hold are left alone. Idempotent (a settled hold is no
146    /// longer open). Returns the number of holds settled.
147    pub fn reap_expired_reserved_holds(
148        &self,
149        now_unix_secs: i64,
150    ) -> Result<usize, BudgetStoreError> {
151        let expired = self.list_expired_reserved_holds(now_unix_secs)?;
152        let mut settled = 0usize;
153        for (hold_id, capability_id, grant_index, remaining, captured, authority) in expired {
154            if !captured {
155                self.capture_invocation_reservations(BudgetCaptureInvocationRequest {
156                    capability_id: capability_id.clone(),
157                    grant_index: grant_index as usize,
158                    hold_id: hold_id.clone(),
159                    event_id: format!("{hold_id}:ttl-reap-capture-invocation"),
160                    trusted_time: None,
161                    authority: authority.clone(),
162                })?;
163            }
164            if remaining > 0 {
165                self.reconcile_budget_hold(BudgetReconcileHoldRequest {
166                    capability_id,
167                    grant_index: grant_index as usize,
168                    exposed_cost_units: remaining,
169                    realized_spend_units: remaining,
170                    hold_id: Some(hold_id.clone()),
171                    event_id: Some(format!("{hold_id}:ttl-reap-settle")),
172                    authority,
173                })?;
174            }
175            settled += 1;
176        }
177        Ok(settled)
178    }
179
180    /// Open reserved holds past their expiry:
181    /// `(hold_id, capability_id, grant_index, remaining_exposure_units,
182    /// invocation_captured, authority)`.
183    fn list_expired_reserved_holds(
184        &self,
185        now_unix_secs: i64,
186    ) -> Result<Vec<ExpiredReservedHold>, BudgetStoreError> {
187        let connection = self
188            .connection
189            .lock()
190            .map_err(|_| BudgetStoreError::Invariant("budget store mutex poisoned".to_string()))?;
191        let mut statement = connection.prepare(
192            "SELECT hold_id, capability_id, grant_index, remaining_exposure_units, \
193             invocation_captured, authority_id, lease_id, lease_epoch \
194             FROM budget_authorization_holds \
195             WHERE disposition = 'open' AND reserved_until IS NOT NULL AND reserved_until <= ?1",
196        )?;
197        let rows = statement.query_map([now_unix_secs], |row| {
198            let authority = sqlite_budget_event_authority(row.get(5)?, row.get(6)?, row.get(7)?)?;
199            Ok((
200                row.get::<_, String>(0)?,
201                row.get::<_, String>(1)?,
202                row.get::<_, i64>(2)? as u32,
203                row.get::<_, i64>(3)? as u64,
204                row.get::<_, i64>(4)? > 0,
205                authority,
206            ))
207        })?;
208        let mut holds = Vec::new();
209        for row in rows {
210            holds.push(row?);
211        }
212        Ok(holds)
213    }
214
215    /// Stamp an open hold with a TTL reaper deadline, the grant currency, and the
216    /// rail transaction id of a prepaid MustPrepay reservation (`None` when the
217    /// reserve carried no prepayment). Errors fail-closed when the hold is missing
218    /// or is no longer open.
219    pub fn mark_hold_reserved_until(
220        &self,
221        hold_id: &str,
222        reserved_until_unix_secs: i64,
223        currency: &str,
224        payment_reference: Option<&str>,
225        envelope: &ReservedHoldEnvelope,
226    ) -> Result<(), BudgetStoreError> {
227        let connection = self
228            .connection
229            .lock()
230            .map_err(|_| BudgetStoreError::Invariant("budget store mutex poisoned".to_string()))?;
231        let affected = connection.execute(
232            "UPDATE budget_authorization_holds \
233             SET reserved_until = ?2, reserved_currency = ?3, reserved_payment_reference = ?4, \
234                 reserved_budget_total = ?5, reserved_delegation_depth = ?6, \
235                 reserved_root_budget_holder = ?7 \
236             WHERE hold_id = ?1 AND disposition = 'open'",
237            params![
238                hold_id,
239                reserved_until_unix_secs,
240                currency,
241                payment_reference,
242                envelope.budget_total.map(|value| value as i64),
243                envelope.delegation_depth as i64,
244                envelope.root_budget_holder,
245            ],
246        )?;
247        if affected == 0 {
248            return Err(BudgetStoreError::Invariant(format!(
249                "cannot mark budget hold `{hold_id}` reserved: missing or not open"
250            )));
251        }
252        Ok(())
253    }
254
255    /// Adopt an already-debited invocation into a durable zero-exposure reserved
256    /// hold, stamped with the TTL deadline and no currency. The invocation was
257    /// already counted by `try_increment`, so this only records the open hold
258    /// (never touching the invocation count); reversing it by hold id returns the
259    /// invocation, while reconciling or reaping it keeps the invocation consumed.
260    /// Fails closed when a hold already exists under the id.
261    pub fn reserve_invocation_hold(
262        &self,
263        hold_id: &str,
264        capability_id: &str,
265        grant_index: usize,
266        reserved_until_unix_secs: i64,
267        envelope: &ReservedHoldEnvelope,
268    ) -> Result<(), BudgetStoreError> {
269        let connection = self
270            .connection
271            .lock()
272            .map_err(|_| BudgetStoreError::Invariant("budget store mutex poisoned".to_string()))?;
273        let updated = connection.execute(
274            "UPDATE budget_authorization_holds \
275             SET reserved_until = ?4, reserved_currency = NULL, \
276                 reserved_payment_reference = NULL, reserved_budget_total = ?5, \
277                 reserved_delegation_depth = ?6, reserved_root_budget_holder = ?7, \
278                 updated_at = ?8 \
279             WHERE hold_id = ?1 AND capability_id = ?2 AND grant_index = ?3 \
280                 AND authorized_exposure_units = 0 AND remaining_exposure_units = 0 \
281                 AND invocation_count_debited = 1 AND invocation_captured = 0 \
282                 AND disposition = 'open'",
283            params![
284                hold_id,
285                capability_id,
286                grant_index as i64,
287                reserved_until_unix_secs,
288                envelope.budget_total.map(|value| value as i64),
289                envelope.delegation_depth as i64,
290                envelope.root_budget_holder,
291                unix_now(),
292            ],
293        )?;
294        if updated == 0 {
295            return Err(BudgetStoreError::Invariant(format!(
296                "budget hold `{hold_id}` is not an open invocation authorization"
297            )));
298        }
299        Ok(())
300    }
301
302    /// Project a single hold by id, including its reserved-until deadline.
303    pub fn budget_hold_snapshot(
304        &self,
305        hold_id: &str,
306    ) -> Result<Option<BudgetHoldSnapshot>, BudgetStoreError> {
307        let connection = self
308            .connection
309            .lock()
310            .map_err(|_| BudgetStoreError::Invariant("budget store mutex poisoned".to_string()))?;
311        connection
312            .query_row(
313                "SELECT hold_id, capability_id, grant_index, authorized_exposure_units, \
314                 remaining_exposure_units, disposition, reserved_until, \
315                 authority_id, lease_id, lease_epoch, reserved_currency, \
316                 reserved_payment_reference, reserved_budget_total, \
317                 reserved_delegation_depth, reserved_root_budget_holder \
318                 FROM budget_authorization_holds WHERE hold_id = ?1",
319                params![hold_id],
320                |row| {
321                    let disposition = row.get::<_, String>(5)?;
322                    let disposition = HoldDisposition::parse(&disposition)
323                        .map(|value| match value {
324                            HoldDisposition::Open => BudgetHoldDispositionView::Open,
325                            HoldDisposition::Released => BudgetHoldDispositionView::Released,
326                            HoldDisposition::Reversed => BudgetHoldDispositionView::Reversed,
327                            HoldDisposition::Reconciled => BudgetHoldDispositionView::Reconciled,
328                        })
329                        .ok_or_else(|| {
330                            rusqlite::Error::FromSqlConversionFailure(
331                                5,
332                                rusqlite::types::Type::Text,
333                                Box::new(std::io::Error::new(
334                                    std::io::ErrorKind::InvalidData,
335                                    format!("unknown hold disposition `{disposition}`"),
336                                )),
337                            )
338                        })?;
339                    let authority =
340                        sqlite_budget_event_authority(row.get(7)?, row.get(8)?, row.get(9)?)?;
341                    Ok(BudgetHoldSnapshot {
342                        hold_id: row.get::<_, String>(0)?,
343                        capability_id: row.get::<_, String>(1)?,
344                        grant_index: row.get::<_, i64>(2)? as usize,
345                        authorized_exposure_units: row.get::<_, i64>(3)? as u64,
346                        remaining_exposure_units: row.get::<_, i64>(4)? as u64,
347                        disposition,
348                        reserved_until: row.get::<_, Option<i64>>(6)?,
349                        reserved_currency: row.get::<_, Option<String>>(10)?,
350                        reserved_payment_reference: row.get::<_, Option<String>>(11)?,
351                        reserved_budget_total: row
352                            .get::<_, Option<i64>>(12)?
353                            .map(|value| value as u64),
354                        reserved_delegation_depth: row
355                            .get::<_, Option<i64>>(13)?
356                            .map(|value| value as u32),
357                        reserved_root_budget_holder: row.get::<_, Option<String>>(14)?,
358                        authority,
359                    })
360                },
361            )
362            .optional()
363            .map_err(Into::into)
364    }
365
366    /// Ids of holds still `open` that were stamped as delegated
367    /// reserve-for-caller reservations: marked reserved (a non-NULL
368    /// `reserved_until`) with a delegation depth of at least one. Such a hold
369    /// keeps its delegated child's sibling-sum share admitted against the parent
370    /// for as long as it stays open, so a freshly built mediation kernel drains
371    /// exactly these holds before resuming delegated admission after a restart. A
372    /// depth-zero or unstamped reserved hold carries no such share and is
373    /// excluded, as is any non-open hold. Errors fail-closed rather than reporting
374    /// a partial set, so the kernel aborts admission on a store read error.
375    pub(super) fn list_open_delegated_reserved_holds(
376        &self,
377    ) -> Result<Vec<String>, BudgetStoreError> {
378        let connection = self
379            .connection
380            .lock()
381            .map_err(|_| BudgetStoreError::Invariant("budget store mutex poisoned".to_string()))?;
382        let mut statement = connection.prepare(
383            "SELECT hold_id FROM budget_authorization_holds \
384             WHERE disposition = 'open' AND reserved_until IS NOT NULL \
385               AND reserved_delegation_depth IS NOT NULL AND reserved_delegation_depth >= 1",
386        )?;
387        let rows = statement.query_map([], |row| row.get::<_, String>(0))?;
388        let mut ids = Vec::new();
389        for row in rows {
390            ids.push(row?);
391        }
392        Ok(ids)
393    }
394
395    /// Whether any hold id begins with `budget-hold:{request_id}:`, regardless of
396    /// disposition or the capability id and grant index encoded after it. The
397    /// mediated pre-execution gate derives each hold id from
398    /// (request_id, capability id, grant index), so a replay of one request_id
399    /// under a different capability token must still be seen as taken. `request_id`
400    /// is caller-supplied, so its LIKE metacharacters (`%`, `_`, and the escape
401    /// char) are escaped and the pattern is bound with an explicit ESCAPE clause,
402    /// so a request_id containing `%` or `_` can neither widen nor narrow the match
403    /// onto another caller's reservation.
404    pub(super) fn hold_exists_for_request_id(
405        &self,
406        request_id: &str,
407    ) -> Result<bool, BudgetStoreError> {
408        let pattern = Self::sqlite_like_prefix_pattern(&format!("budget-hold:{request_id}:"));
409        let nonce_pattern =
410            Self::sqlite_like_prefix_pattern(&format!("nonce-preflight-budget-hold:{request_id}:"));
411        let connection = self
412            .connection
413            .lock()
414            .map_err(|_| BudgetStoreError::Invariant("budget store mutex poisoned".to_string()))?;
415        Ok(connection
416            .query_row(
417                "SELECT 1 FROM budget_authorization_holds \
418                 WHERE hold_id LIKE ?1 ESCAPE '\\' OR hold_id LIKE ?2 ESCAPE '\\' LIMIT 1",
419                params![pattern, nonce_pattern],
420                |_| Ok(()),
421            )
422            .optional()?
423            .is_some())
424    }
425
426    /// Rows still `open`:
427    /// `(hold_id, capability_id, grant_index, remaining_exposure_units, authority)`.
428    /// A hold with inconsistent authority lease columns is rejected fail-closed.
429    pub(super) fn list_open_holds(&self) -> Result<Vec<OpenHold>, BudgetStoreError> {
430        let connection = self
431            .connection
432            .lock()
433            .map_err(|_| BudgetStoreError::Invariant("budget store mutex poisoned".to_string()))?;
434        let mut statement = connection.prepare(
435            "SELECT hold_id, capability_id, grant_index, remaining_exposure_units, \
436             invocation_captured, authority_id, lease_id, lease_epoch \
437             FROM budget_authorization_holds WHERE disposition = 'open'",
438        )?;
439        let rows = statement.query_map([], |row| {
440            let authority = sqlite_budget_event_authority(row.get(5)?, row.get(6)?, row.get(7)?)?;
441            Ok((
442                row.get::<_, String>(0)?,
443                row.get::<_, String>(1)?,
444                row.get::<_, i64>(2)? as u32,
445                row.get::<_, i64>(3)? as u64,
446                row.get::<_, i64>(4)? != 0,
447                authority,
448            ))
449        })?;
450        let mut holds = Vec::new();
451        for row in rows {
452            holds.push(row?);
453        }
454        Ok(holds)
455    }
456}
457
458#[cfg(test)]
459#[allow(clippy::unwrap_used, clippy::expect_used)]
460mod tests {
461    use super::*;
462    use chio_kernel::budget_store::{
463        BudgetAuthorizeHoldDecision, BudgetAuthorizeHoldRequest, BudgetHoldDispositionView,
464        BudgetStore,
465    };
466    use std::collections::HashMap;
467
468    fn open_temp_store() -> SqliteBudgetStore {
469        let dir = std::env::temp_dir().join(format!("chio-reaper-{}", uuid::Uuid::now_v7()));
470        std::fs::create_dir_all(&dir).unwrap();
471        SqliteBudgetStore::open(dir.join("budget.sqlite")).unwrap()
472    }
473
474    fn authorize(store: &SqliteBudgetStore, hold_id: &str, cap: &str) {
475        authorize_with_authority(store, hold_id, cap, None);
476    }
477
478    fn authorize_invocation(store: &SqliteBudgetStore, hold_id: &str, cap: &str) {
479        let decision = store
480            .authorize_budget_hold(BudgetAuthorizeHoldRequest {
481                capability_id: cap.to_string(),
482                grant_index: 0,
483                max_invocations: Some(1),
484                requested_exposure_units: 0,
485                max_cost_per_invocation: None,
486                max_total_cost_units: None,
487                hold_id: Some(hold_id.to_string()),
488                event_id: Some(format!("{hold_id}:authorize")),
489                authority: None,
490                invocation_quotas: Vec::new(),
491                cumulative_approval: None,
492                admission_binding: None,
493            })
494            .unwrap();
495        assert!(matches!(
496            decision,
497            BudgetAuthorizeHoldDecision::Authorized(_)
498        ));
499    }
500
501    fn capture_invocation(store: &SqliteBudgetStore, hold_id: &str, cap: &str) {
502        store
503            .capture_invocation_reservations(BudgetCaptureInvocationRequest {
504                capability_id: cap.to_string(),
505                grant_index: 0,
506                hold_id: hold_id.to_string(),
507                event_id: format!("{hold_id}:capture-invocation"),
508                trusted_time: None,
509                authority: None,
510            })
511            .unwrap();
512    }
513
514    fn authorize_with_authority(
515        store: &SqliteBudgetStore,
516        hold_id: &str,
517        cap: &str,
518        authority: Option<BudgetEventAuthority>,
519    ) {
520        let decision = store
521            .authorize_budget_hold(BudgetAuthorizeHoldRequest {
522                capability_id: cap.to_string(),
523                grant_index: 0,
524                max_invocations: Some(10),
525                requested_exposure_units: 100,
526                max_cost_per_invocation: Some(100),
527                max_total_cost_units: Some(1000),
528                hold_id: Some(hold_id.to_string()),
529                event_id: Some(format!("{hold_id}:authorize")),
530                authority,
531                invocation_quotas: Vec::new(),
532                cumulative_approval: None,
533                admission_binding: None,
534            })
535            .unwrap();
536        assert!(matches!(
537            decision,
538            BudgetAuthorizeHoldDecision::Authorized(_)
539        ));
540    }
541
542    #[test]
543    fn ttl_reaper_settles_expired_unreconciled_reserved_holds_at_worst_case() {
544        use chio_kernel::budget_store::{BudgetHoldDispositionView, BudgetReconcileHoldRequest};
545
546        let store = open_temp_store();
547        // Expired reserved hold.
548        authorize(&store, "hold-expired", "cap-a");
549        store
550            .mark_hold_reserved_until(
551                "hold-expired",
552                100,
553                "USD",
554                None,
555                &ReservedHoldEnvelope::default(),
556            )
557            .unwrap();
558        // Not-yet-expired reserved hold.
559        authorize(&store, "hold-fresh", "cap-b");
560        store
561            .mark_hold_reserved_until(
562                "hold-fresh",
563                5_000,
564                "USD",
565                None,
566                &ReservedHoldEnvelope::default(),
567            )
568            .unwrap();
569        // Reconciled reserved hold.
570        authorize(&store, "hold-done", "cap-c");
571        store
572            .mark_hold_reserved_until(
573                "hold-done",
574                100,
575                "USD",
576                None,
577                &ReservedHoldEnvelope::default(),
578            )
579            .unwrap();
580        capture_invocation(&store, "hold-done", "cap-c");
581        store
582            .reconcile_budget_hold(BudgetReconcileHoldRequest {
583                capability_id: "cap-c".to_string(),
584                grant_index: 0,
585                exposed_cost_units: 100,
586                realized_spend_units: 40,
587                hold_id: Some("hold-done".to_string()),
588                event_id: Some("hold-done:reconcile".to_string()),
589                authority: None,
590            })
591            .unwrap();
592
593        let settled = store.reap_expired_reserved_holds(1_000).unwrap();
594        assert_eq!(settled, 1, "only the expired reserved hold is settled");
595
596        // cap-a expired reserved hold SETTLED at worst-case: the reserved amount is
597        // forfeited to realized spend (committed stays 100), not released to 0.
598        let cap_a = store.get_usage("cap-a", 0).unwrap().unwrap();
599        assert_eq!(
600            cap_a.total_cost_realized_spend, 100,
601            "the forfeited worst-case becomes realized spend"
602        );
603        assert_eq!(
604            cap_a.committed_cost_units().unwrap(),
605            100,
606            "the reserved worst-case stays consumed, the freed difference is gone"
607        );
608        // A second reap is idempotent: the settled hold is no longer open.
609        assert_eq!(store.reap_expired_reserved_holds(1_000).unwrap(), 0);
610        assert_eq!(
611            store
612                .budget_hold_snapshot("hold-expired")
613                .unwrap()
614                .unwrap()
615                .disposition,
616            BudgetHoldDispositionView::Reconciled
617        );
618        // cap-b not-yet-expired reserved hold untouched.
619        assert_eq!(
620            store
621                .get_usage("cap-b", 0)
622                .unwrap()
623                .unwrap()
624                .committed_cost_units()
625                .unwrap(),
626            100
627        );
628        assert_eq!(
629            store
630                .budget_hold_snapshot("hold-fresh")
631                .unwrap()
632                .unwrap()
633                .disposition,
634            BudgetHoldDispositionView::Open
635        );
636        // cap-c reconciled hold untouched (realized 40).
637        assert_eq!(
638            store
639                .get_usage("cap-c", 0)
640                .unwrap()
641                .unwrap()
642                .committed_cost_units()
643                .unwrap(),
644            40
645        );
646        assert_eq!(
647            store
648                .budget_hold_snapshot("hold-done")
649                .unwrap()
650                .unwrap()
651                .disposition,
652            BudgetHoldDispositionView::Reconciled
653        );
654    }
655
656    #[test]
657    fn ttl_reaper_settles_expired_hold_after_invocation_capture() {
658        use chio_kernel::budget_store::BudgetHoldDispositionView;
659
660        let store = open_temp_store();
661        authorize(&store, "hold-captured-expired", "cap-captured-expired");
662        store
663            .mark_hold_reserved_until(
664                "hold-captured-expired",
665                100,
666                "USD",
667                None,
668                &ReservedHoldEnvelope::default(),
669            )
670            .unwrap();
671        capture_invocation(&store, "hold-captured-expired", "cap-captured-expired");
672
673        let settled = store.reap_expired_reserved_holds(1_000).unwrap();
674        assert_eq!(settled, 1);
675        let hold = store
676            .budget_hold_snapshot("hold-captured-expired")
677            .unwrap()
678            .unwrap();
679        assert_eq!(hold.disposition, BudgetHoldDispositionView::Reconciled);
680        assert_eq!(
681            store
682                .get_usage("cap-captured-expired", 0)
683                .unwrap()
684                .unwrap()
685                .total_cost_realized_spend,
686            100
687        );
688        assert_eq!(store.reap_expired_reserved_holds(1_000).unwrap(), 0);
689    }
690
691    #[test]
692    fn budget_hold_snapshot_projects_reserved_hold() {
693        use chio_kernel::budget_store::BudgetHoldDispositionView;
694
695        let store = open_temp_store();
696        authorize(&store, "hold-snap", "cap-snap");
697        assert!(store
698            .budget_hold_snapshot("hold-missing")
699            .unwrap()
700            .is_none());
701        store
702            .mark_hold_reserved_until(
703                "hold-snap",
704                4_242,
705                "USD",
706                Some("rail_txn_ref"),
707                &ReservedHoldEnvelope {
708                    budget_total: Some(1_000),
709                    delegation_depth: 2,
710                    root_budget_holder: "root-holder".to_string(),
711                },
712            )
713            .unwrap();
714        let snapshot = store.budget_hold_snapshot("hold-snap").unwrap().unwrap();
715        assert_eq!(snapshot.capability_id, "cap-snap");
716        assert_eq!(snapshot.remaining_exposure_units, 100);
717        assert_eq!(snapshot.disposition, BudgetHoldDispositionView::Open);
718        assert_eq!(snapshot.reserved_until, Some(4_242));
719        assert_eq!(snapshot.reserved_currency.as_deref(), Some("USD"));
720        assert_eq!(
721            snapshot.reserved_payment_reference.as_deref(),
722            Some("rail_txn_ref"),
723            "a prepaid reservation records its rail transaction id durably"
724        );
725        assert_eq!(
726            snapshot.reserved_budget_total,
727            Some(1_000),
728            "the grant ceiling is recorded durably on the reserved hold"
729        );
730        assert_eq!(snapshot.reserved_delegation_depth, Some(2));
731        assert_eq!(
732            snapshot.reserved_root_budget_holder.as_deref(),
733            Some("root-holder"),
734            "the delegation root is recorded durably on the reserved hold"
735        );
736    }
737
738    #[test]
739    fn mark_hold_reserved_on_missing_hold_fails_closed() {
740        let store = open_temp_store();
741        assert!(store
742            .mark_hold_reserved_until("nope", 100, "USD", None, &ReservedHoldEnvelope::default())
743            .is_err());
744    }
745
746    #[test]
747    fn request_id_has_reserved_hold_matches_by_prefix_and_escapes_like_metacharacters() {
748        let store = open_temp_store();
749        // A reservation under capability A binds request_id `axc` to a hold whose
750        // id embeds A. The reuse guard must report `axc` taken regardless of the
751        // capability that opened it.
752        authorize(&store, "budget-hold:axc:cap-a:0", "cap-a");
753
754        assert_eq!(
755            store.request_id_has_reserved_hold("axc").unwrap(),
756            Some(true),
757            "the request_id that backs the hold is reported taken"
758        );
759        assert_eq!(
760            store.request_id_has_reserved_hold("other").unwrap(),
761            Some(false),
762            "a request_id with no hold is reported free"
763        );
764        // request_id is caller-supplied, so LIKE metacharacters in it must be
765        // escaped: `_` (any single char) and `%` (any run) must not widen the
766        // prefix onto the different stored request_id `axc`.
767        assert_eq!(
768            store.request_id_has_reserved_hold("a_c").unwrap(),
769            Some(false),
770            "an underscore in request_id must not match a different stored id"
771        );
772        assert_eq!(
773            store.request_id_has_reserved_hold("a%c").unwrap(),
774            Some(false),
775            "a percent in request_id must not match a different stored id"
776        );
777    }
778
779    #[test]
780    fn reaper_reconciles_admitted_hold_and_reverses_orphan() {
781        // SIGKILL after authorize commits but before reconcile. A naive
782        // "release Open on restart" would enable double-spend; instead the
783        // durable receipt log arbitrates.
784        let store = open_temp_store();
785        authorize(&store, "hold-admitted", "cap-a"); // durably admitted, realized 40
786        authorize(&store, "hold-orphan", "cap-b"); // never admitted downstream
787                                                   // Before reap both holds inflate committed_cost by their worst-case 100.
788        assert_eq!(
789            store
790                .get_usage("cap-a", 0)
791                .unwrap()
792                .unwrap()
793                .committed_cost_units()
794                .unwrap(),
795            100
796        );
797
798        let mut realized = HashMap::new();
799        realized.insert("hold-admitted".to_string(), 40u64);
800        let summary = store.reap_holds_by_map(&realized).unwrap();
801        assert_eq!(summary.reconciled, 1);
802        assert_eq!(summary.reversed, 1);
803
804        // cap-a reconciled down to realized 40; cap-b reversed back to 0.
805        assert_eq!(
806            store
807                .get_usage("cap-a", 0)
808                .unwrap()
809                .unwrap()
810                .committed_cost_units()
811                .unwrap(),
812            40
813        );
814        assert_eq!(
815            store
816                .get_usage("cap-b", 0)
817                .unwrap()
818                .unwrap()
819                .committed_cost_units()
820                .unwrap(),
821            0
822        );
823    }
824
825    #[test]
826    fn reaper_forfeits_captured_orphan_at_worst_case() {
827        let store = open_temp_store();
828        authorize(&store, "hold-captured", "cap-captured");
829        capture_invocation(&store, "hold-captured", "cap-captured");
830
831        let summary = store.reap_holds_by_map(&HashMap::new()).unwrap();
832
833        assert_eq!(summary.reconciled, 1);
834        assert_eq!(summary.reversed, 0);
835        assert_eq!(
836            store
837                .get_usage("cap-captured", 0)
838                .unwrap()
839                .unwrap()
840                .committed_cost_units()
841                .unwrap(),
842            100
843        );
844        assert_eq!(
845            store
846                .budget_hold_snapshot("hold-captured")
847                .unwrap()
848                .unwrap()
849                .disposition,
850            BudgetHoldDispositionView::Reconciled
851        );
852    }
853
854    #[test]
855    fn reserve_invocation_hold_is_reversible_and_returns_the_invocation() {
856        use chio_kernel::budget_store::{
857            BudgetHoldDispositionView, BudgetReverseHoldRequest, BudgetStore,
858        };
859
860        let store = open_temp_store();
861        // Debit the single invocation exactly as the reserve path does, then adopt
862        // it into a durable zero-exposure reserved hold.
863        authorize_invocation(&store, "hold-inv", "cap-inv");
864        store
865            .reserve_invocation_hold(
866                "hold-inv",
867                "cap-inv",
868                0,
869                4_242,
870                &ReservedHoldEnvelope {
871                    budget_total: None,
872                    delegation_depth: 1,
873                    root_budget_holder: "inv-root".to_string(),
874                },
875            )
876            .unwrap();
877
878        let snapshot = store.budget_hold_snapshot("hold-inv").unwrap().unwrap();
879        assert_eq!(snapshot.authorized_exposure_units, 0);
880        assert_eq!(snapshot.remaining_exposure_units, 0);
881        assert_eq!(snapshot.disposition, BudgetHoldDispositionView::Open);
882        assert_eq!(snapshot.reserved_until, Some(4_242));
883        assert_eq!(
884            snapshot.reserved_currency, None,
885            "an invocation reservation records no currency"
886        );
887        assert_eq!(
888            snapshot.reserved_budget_total, None,
889            "an invocation reservation carries no monetary ceiling"
890        );
891        assert_eq!(snapshot.reserved_delegation_depth, Some(1));
892        assert_eq!(
893            snapshot.reserved_root_budget_holder.as_deref(),
894            Some("inv-root"),
895            "an invocation reservation still records its delegation root"
896        );
897        assert_eq!(
898            store
899                .get_usage("cap-inv", 0)
900                .unwrap()
901                .unwrap()
902                .invocation_count,
903            1
904        );
905
906        // Reversing the hold returns the invocation to the grant.
907        store
908            .reverse_budget_hold(BudgetReverseHoldRequest {
909                capability_id: "cap-inv".to_string(),
910                grant_index: 0,
911                reversed_exposure_units: snapshot.remaining_exposure_units,
912                hold_id: Some("hold-inv".to_string()),
913                event_id: Some("hold-inv:reverse".to_string()),
914                authority: snapshot.authority,
915                expected_cumulative_approval_state: None,
916            })
917            .unwrap();
918        assert_eq!(
919            store
920                .get_usage("cap-inv", 0)
921                .unwrap()
922                .unwrap()
923                .invocation_count,
924            0,
925            "reversing an invocation reservation returns the debited invocation"
926        );
927        assert_eq!(
928            store
929                .budget_hold_snapshot("hold-inv")
930                .unwrap()
931                .unwrap()
932                .disposition,
933            BudgetHoldDispositionView::Reversed
934        );
935
936        // A duplicate reserve under the same id fails closed.
937        assert!(store
938            .reserve_invocation_hold(
939                "hold-inv",
940                "cap-inv",
941                0,
942                9_000,
943                &ReservedHoldEnvelope::default()
944            )
945            .is_err());
946    }
947
948    #[test]
949    fn reaper_forfeits_expired_invocation_reserve_keeping_it_consumed() {
950        use chio_kernel::budget_store::{BudgetHoldDispositionView, BudgetStore};
951
952        let store = open_temp_store();
953        authorize_invocation(&store, "hold-inv", "cap-inv");
954        store
955            .reserve_invocation_hold(
956                "hold-inv",
957                "cap-inv",
958                0,
959                100,
960                &ReservedHoldEnvelope::default(),
961            )
962            .unwrap();
963
964        let settled = store.reap_expired_reserved_holds(1_000).unwrap();
965        assert_eq!(settled, 1, "the expired invocation reservation is settled");
966        assert_eq!(
967            store
968                .get_usage("cap-inv", 0)
969                .unwrap()
970                .unwrap()
971                .invocation_count,
972            1,
973            "reaping forfeits the invocation (stays consumed), matching monetary reap"
974        );
975        assert_eq!(
976            store
977                .budget_hold_snapshot("hold-inv")
978                .unwrap()
979                .unwrap()
980                .disposition,
981            BudgetHoldDispositionView::Reconciled
982        );
983        // Idempotent: a settled hold is no longer open.
984        assert_eq!(store.reap_expired_reserved_holds(1_000).unwrap(), 0);
985    }
986
987    #[test]
988    fn reaper_reclaims_kernel_authored_holds_bearing_authority() {
989        // Kernel-authored holds carry a BudgetEventAuthority lease. A crash after
990        // authorize leaves them open; the reaper must load and present each hold's
991        // stored authority so the store's authority check passes and the orphaned
992        // holds are reclaimed rather than left reserved indefinitely.
993        let store = open_temp_store();
994        let authority = BudgetEventAuthority {
995            authority_id: "kernel-authority".to_string(),
996            lease_id: "lease-1".to_string(),
997            lease_epoch: 0,
998        };
999        authorize_with_authority(&store, "hold-admitted", "cap-a", Some(authority.clone()));
1000        authorize_with_authority(&store, "hold-orphan", "cap-b", Some(authority));
1001
1002        let mut realized = HashMap::new();
1003        realized.insert("hold-admitted".to_string(), 40u64);
1004        let summary = store.reap_holds_by_map(&realized).unwrap();
1005        assert_eq!(summary.reconciled, 1);
1006        assert_eq!(summary.reversed, 1);
1007
1008        // cap-a reconciled to realized 40; cap-b orphan reversed to 0.
1009        assert_eq!(
1010            store
1011                .get_usage("cap-a", 0)
1012                .unwrap()
1013                .unwrap()
1014                .committed_cost_units()
1015                .unwrap(),
1016            40
1017        );
1018        assert_eq!(
1019            store
1020                .get_usage("cap-b", 0)
1021                .unwrap()
1022                .unwrap()
1023                .committed_cost_units()
1024                .unwrap(),
1025            0
1026        );
1027    }
1028
1029    #[test]
1030    fn list_open_delegated_reserved_hold_ids_enumerates_only_open_delegated_reserved() {
1031        use chio_kernel::budget_store::BudgetReconcileHoldRequest;
1032
1033        let store = open_temp_store();
1034
1035        // (a) Open delegated reserve-for-caller hold: reserved with delegation
1036        // depth one, so its child's sibling-sum share stays admitted against the
1037        // parent until it closes. The restart gate must drain exactly this hold.
1038        authorize(&store, "hold-a-delegated", "cap-a");
1039        store
1040            .mark_hold_reserved_until(
1041                "hold-a-delegated",
1042                4_242,
1043                "USD",
1044                None,
1045                &ReservedHoldEnvelope {
1046                    budget_total: None,
1047                    delegation_depth: 1,
1048                    root_budget_holder: "root-a".to_string(),
1049                },
1050            )
1051            .unwrap();
1052
1053        // (b) Open reserved hold at delegation depth zero: reserved but not
1054        // delegated, so it holds no sibling-sum share and must be excluded.
1055        authorize(&store, "hold-b-nondelegated", "cap-b");
1056        store
1057            .mark_hold_reserved_until(
1058                "hold-b-nondelegated",
1059                4_242,
1060                "USD",
1061                None,
1062                &ReservedHoldEnvelope {
1063                    budget_total: None,
1064                    delegation_depth: 0,
1065                    root_budget_holder: "root-b".to_string(),
1066                },
1067            )
1068            .unwrap();
1069
1070        // (c) Delegated hold that has since closed: reconciled, so no longer open
1071        // and no longer holds a share, and must be excluded.
1072        authorize(&store, "hold-c-closed", "cap-c");
1073        store
1074            .mark_hold_reserved_until(
1075                "hold-c-closed",
1076                4_242,
1077                "USD",
1078                None,
1079                &ReservedHoldEnvelope {
1080                    budget_total: None,
1081                    delegation_depth: 1,
1082                    root_budget_holder: "root-c".to_string(),
1083                },
1084            )
1085            .unwrap();
1086        capture_invocation(&store, "hold-c-closed", "cap-c");
1087        store
1088            .reconcile_budget_hold(BudgetReconcileHoldRequest {
1089                capability_id: "cap-c".to_string(),
1090                grant_index: 0,
1091                exposed_cost_units: 100,
1092                realized_spend_units: 40,
1093                hold_id: Some("hold-c-closed".to_string()),
1094                event_id: Some("hold-c-closed:reconcile".to_string()),
1095                authority: None,
1096            })
1097            .unwrap();
1098
1099        // Some(..) switches the kernel onto the precise restart gate; the set
1100        // enumerates only the open delegated reserved hold.
1101        let ids = store.list_open_delegated_reserved_hold_ids().unwrap();
1102        assert_eq!(ids, Some(vec!["hold-a-delegated".to_string()]));
1103    }
1104}
chio_store_sqlite/budget_store/reaper.rs

chio_store_sqlite/budget_store/
reaper.rs
