1use super::*;
2
3impl BudgetStore for SqliteBudgetStore {
4 fn try_increment(
5 &self,
6 capability_id: &str,
7 grant_index: usize,
8 max_invocations: Option<u32>,
9 ) -> Result<bool, BudgetStoreError> {
10 self.try_increment_with_event_id(capability_id, grant_index, max_invocations, None)
11 }
12
13 fn authorize_budget_hold(
14 &self,
15 request: BudgetAuthorizeHoldRequest,
16 ) -> Result<BudgetAuthorizeHoldDecision, BudgetStoreError> {
17 request.validate()?;
18 if !request.invocation_quotas.is_empty()
19 || request.cumulative_approval.is_some()
20 || request.admission_binding.is_some()
21 {
22 return self.authorize_composite_hold(request);
23 }
24 self.require_standalone_mutation("unbound authorization")?;
25 self.authorize_budget_hold_atomic(&request)
26 }
27
28 fn capture_invocation_reservations(
29 &self,
30 request: BudgetCaptureInvocationRequest,
31 ) -> Result<BudgetInvocationCaptureDecision, BudgetStoreError> {
32 validate_budget_grant_index(request.grant_index)?;
33 request.validate()?;
34 if self.is_structured_hold(&request.hold_id)? {
35 return self.capture_composite_invocation(request);
36 }
37 self.require_standalone_mutation("unbound invocation capture")?;
38 if request.trusted_time.is_some() {
39 return Err(BudgetStoreError::Invariant(
40 "trusted capture time is not supported by the sqlite budget store".to_string(),
41 ));
42 }
43 if let Some(authority) = request.authority.as_ref() {
44 budget_u64_to_sqlite(authority.lease_epoch, "lease_epoch")?;
45 }
46 let mut connection = self.connection()?;
47 let transaction = self.begin_write(&mut connection)?;
48 Self::reject_structured_hold_from_legacy_writer(
49 &transaction,
50 Some(&request.hold_id),
51 "invocation capture",
52 )?;
53
54 let existing_event =
55 SqliteBudgetStore::load_mutation_event(&transaction, &request.event_id)?;
56 let existing = SqliteBudgetStore::existing_event_allowed(
57 &transaction,
58 Some(&request.event_id),
59 BudgetMutationKind::CaptureInvocation,
60 &request.capability_id,
61 request.grant_index,
62 Some(&request.hold_id),
63 request.authority.as_ref(),
64 existing_event
65 .as_ref()
66 .map_or(0, |event| event.exposure_units),
67 0,
68 None,
69 None,
70 None,
71 )?;
72 let usage = transaction
73 .query_row(
74 r#"
75 SELECT seq, invocation_count, total_cost_exposed, total_cost_realized_spend
76 FROM capability_grant_budgets
77 WHERE capability_id = ?1 AND grant_index = ?2
78 "#,
79 params![&request.capability_id, request.grant_index as i64],
80 |row| {
81 Ok((
82 budget_u64_from_row(row, 0, "seq")?,
83 budget_u32_from_row(row, 1, "invocation_count")?,
84 budget_u64_from_row(row, 2, "total_cost_exposed")?,
85 budget_u64_from_row(row, 3, "total_cost_realized_spend")?,
86 ))
87 },
88 )
89 .optional()?
90 .ok_or_else(|| BudgetStoreError::Invariant("missing charged budget row".to_string()))?;
91 let mutation = |event_id: String,
92 event_seq,
93 authority: Option<BudgetEventAuthority>,
94 invocation_count_after,
95 total_cost_exposed_after,
96 total_cost_realized_spend_after,
97 exposure_units|
98 -> Result<BudgetHoldMutationDecision, BudgetStoreError> {
99 Ok(BudgetHoldMutationDecision {
100 hold_id: Some(request.hold_id.clone()),
101 admission_binding: None,
102 exposure_units,
103 realized_spend_units: 0,
104 committed_cost_units_after: checked_committed_cost_units(
105 total_cost_exposed_after,
106 total_cost_realized_spend_after,
107 )?,
108 invocation_count_after,
109 invocation_quota_usages: Vec::new(),
110 cumulative_approval: None,
111 invocation_state: BudgetInvocationState::Captured,
112 monetary_state: if exposure_units == 0 {
113 BudgetMonetaryState::None
114 } else {
115 BudgetMonetaryState::Exposed
116 },
117 metadata: BudgetCommitMetadata {
118 authority,
119 guarantee_level: self.budget_guarantee_level(),
120 budget_profile: self.budget_authority_profile(),
121 metering_profile: self.budget_metering_profile(),
122 budget_commit_index: Some(event_seq),
123 event_id: Some(event_id),
124 recorded_at_unix_seconds: None,
125 },
126 })
127 };
128
129 if existing.is_some() {
130 let hold =
131 SqliteBudgetStore::load_hold(&transaction, &request.hold_id)?.ok_or_else(|| {
132 BudgetStoreError::Invariant(format!(
133 "captured budget hold `{}` disappeared",
134 request.hold_id
135 ))
136 })?;
137 if !hold.invocation_captured {
138 transaction.rollback()?;
139 return Err(BudgetStoreError::Invariant(format!(
140 "budget hold `{}` invocation capture is no longer current",
141 request.hold_id
142 )));
143 }
144 let original =
145 SqliteBudgetStore::load_current_capture_event(&transaction, &request.hold_id)?;
146 if original.event_id != request.event_id {
147 transaction.rollback()?;
148 return Err(BudgetStoreError::Invariant(format!(
149 "budget hold `{}` invocation capture is no longer current",
150 request.hold_id
151 )));
152 }
153 transaction.rollback()?;
154 return Ok(BudgetInvocationCaptureDecision::AlreadyCaptured(mutation(
155 original.event_id,
156 original.event_seq,
157 original.authority,
158 original.invocation_count_after,
159 original.total_cost_exposed_after,
160 original.total_cost_realized_spend_after,
161 original.exposure_units,
162 )?));
163 }
164
165 let hold = SqliteBudgetStore::ensure_open_hold(
166 &transaction,
167 &request.hold_id,
168 &request.capability_id,
169 request.grant_index,
170 )?;
171 SqliteBudgetStore::validate_hold_authority(
172 &request.hold_id,
173 hold.authority.as_ref(),
174 request.authority.as_ref(),
175 )?;
176 if !hold.invocation_count_debited {
177 transaction.rollback()?;
178 return Err(BudgetStoreError::Invariant(format!(
179 "budget hold `{}` has no invocation reservation to capture",
180 request.hold_id
181 )));
182 }
183 if hold.invocation_captured {
184 let original =
185 SqliteBudgetStore::load_current_capture_event(&transaction, &request.hold_id)?;
186 if original.event_id != request.event_id {
187 transaction.rollback()?;
188 return Err(BudgetStoreError::Invariant(format!(
189 "budget hold `{}` invocation was captured by a different event",
190 request.hold_id
191 )));
192 }
193 transaction.rollback()?;
194 return Ok(BudgetInvocationCaptureDecision::AlreadyCaptured(mutation(
195 original.event_id,
196 original.event_seq,
197 original.authority,
198 original.invocation_count_after,
199 original.total_cost_exposed_after,
200 original.total_cost_realized_spend_after,
201 original.exposure_units,
202 )?));
203 }
204
205 let event_seq = allocate_budget_replication_seq(&transaction)?;
206 let changed = transaction.execute(
207 r#"
208 UPDATE budget_authorization_holds
209 SET invocation_captured = 1,
210 disposition = CASE
211 WHEN remaining_exposure_units = 0 THEN 'reconciled'
212 ELSE disposition
213 END,
214 updated_at = ?2
215 WHERE hold_id = ?1
216 AND invocation_count_debited = 1
217 AND invocation_captured = 0
218 "#,
219 params![&request.hold_id, unix_now()],
220 )?;
221 if changed != 1 {
222 transaction.rollback()?;
223 return Err(BudgetStoreError::Invariant(format!(
224 "budget hold `{}` invocation capture compare-and-set failed",
225 request.hold_id
226 )));
227 }
228
229 SqliteBudgetStore::append_mutation_event(
230 &transaction,
231 Some(&request.event_id),
232 Some(&request.hold_id),
233 request.authority.as_ref(),
234 &request.capability_id,
235 request.grant_index,
236 BudgetMutationKind::CaptureInvocation,
237 Some(true),
238 event_seq,
239 Some(usage.0),
240 hold.remaining_exposure_units,
241 0,
242 None,
243 None,
244 None,
245 usage.1,
246 usage.2,
247 usage.3,
248 )?;
249 transaction.commit()?;
250 Ok(BudgetInvocationCaptureDecision::Captured(mutation(
251 request.event_id,
252 event_seq,
253 request.authority,
254 usage.1,
255 usage.2,
256 usage.3,
257 hold.remaining_exposure_units,
258 )?))
259 }
260
261 fn authorize_cumulative_approval(
262 &self,
263 request: BudgetAuthorizeCumulativeApprovalRequest,
264 ) -> Result<BudgetCumulativeApprovalAuthorizationDecision, BudgetStoreError> {
265 self.authorize_composite_cumulative_approval(request)
266 }
267
268 fn cancel_captured_before_dispatch(
269 &self,
270 request: BudgetCancelCapturedBeforeDispatchRequest,
271 ) -> Result<BudgetCapturedBeforeDispatchCancellationDecision, BudgetStoreError> {
272 validate_budget_grant_index(request.grant_index)?;
273 request.validate()?;
274 if self.is_structured_hold(&request.hold_id)? {
275 return self.cancel_captured_composite_hold(request);
276 }
277 self.require_standalone_mutation("captured invocation cancellation")?;
278 if let Some(authority) = request.authority.as_ref() {
279 budget_u64_to_sqlite(authority.lease_epoch, "lease_epoch")?;
280 }
281 let mut connection = self.connection()?;
282 let transaction = self.begin_write(&mut connection)?;
283 Self::reject_structured_hold_from_legacy_writer(
284 &transaction,
285 Some(&request.hold_id),
286 "captured invocation cancellation",
287 )?;
288
289 if let Some(existing) =
290 SqliteBudgetStore::load_mutation_event(&transaction, &request.event_id)?
291 {
292 SqliteBudgetStore::validate_replay_authority(
293 &request.event_id,
294 existing.authority.as_ref(),
295 request.authority.as_ref(),
296 )?;
297 let matches = existing.kind == BudgetMutationKind::CancelCapturedBeforeDispatch
298 && existing.hold_id.as_deref() == Some(request.hold_id.as_str())
299 && existing.capability_id == request.capability_id
300 && existing.grant_index == request.grant_index as u32
301 && existing.allowed == Some(true)
302 && existing.realized_spend_units == 0
303 && existing.max_invocations.is_none()
304 && existing.max_cost_per_invocation.is_none()
305 && existing.max_total_cost_units.is_none();
306 if !matches {
307 transaction.rollback()?;
308 return Err(BudgetStoreError::Invariant(format!(
309 "budget event_id `{}` was reused for a different mutation",
310 request.event_id
311 )));
312 }
313 let mutation = BudgetHoldMutationDecision {
314 hold_id: Some(request.hold_id),
315 admission_binding: None,
316 exposure_units: existing.exposure_units,
317 realized_spend_units: 0,
318 committed_cost_units_after: checked_committed_cost_units(
319 existing.total_cost_exposed_after,
320 existing.total_cost_realized_spend_after,
321 )?,
322 invocation_count_after: existing.invocation_count_after,
323 invocation_quota_usages: Vec::new(),
324 cumulative_approval: None,
325 invocation_state: BudgetInvocationState::Reversed,
326 monetary_state: if existing.exposure_units == 0 {
327 BudgetMonetaryState::None
328 } else {
329 BudgetMonetaryState::Reversed
330 },
331 metadata: BudgetCommitMetadata {
332 authority: existing.authority,
333 guarantee_level: self.budget_guarantee_level(),
334 budget_profile: self.budget_authority_profile(),
335 metering_profile: self.budget_metering_profile(),
336 budget_commit_index: Some(existing.event_seq),
337 event_id: Some(existing.event_id),
338 recorded_at_unix_seconds: u64::try_from(existing.recorded_at).ok(),
339 },
340 };
341 transaction.rollback()?;
342 return Ok(
343 BudgetCapturedBeforeDispatchCancellationDecision::AlreadyCancelled(mutation),
344 );
345 }
346
347 let hold = SqliteBudgetStore::ensure_open_hold(
348 &transaction,
349 &request.hold_id,
350 &request.capability_id,
351 request.grant_index,
352 )?;
353 SqliteBudgetStore::validate_hold_authority(
354 &request.hold_id,
355 hold.authority.as_ref(),
356 request.authority.as_ref(),
357 )?;
358 if !hold.invocation_count_debited || !hold.invocation_captured {
359 transaction.rollback()?;
360 return Err(BudgetStoreError::Invariant(format!(
361 "budget hold `{}` has no captured invocation to cancel before dispatch",
362 request.hold_id
363 )));
364 }
365 let capture =
366 SqliteBudgetStore::load_current_capture_event(&transaction, &request.hold_id)?;
367 if capture.capability_id != request.capability_id
368 || capture.grant_index != request.grant_index as u32
369 || capture.allowed != Some(true)
370 {
371 transaction.rollback()?;
372 return Err(BudgetStoreError::Invariant(format!(
373 "captured budget hold `{}` has mismatched capture evidence",
374 request.hold_id
375 )));
376 }
377
378 let current = transaction
379 .query_row(
380 r#"
381 SELECT invocation_count, total_cost_exposed, total_cost_realized_spend
382 FROM capability_grant_budgets
383 WHERE capability_id = ?1 AND grant_index = ?2
384 "#,
385 params![&request.capability_id, request.grant_index as i64],
386 |row| {
387 Ok((
388 budget_u32_from_row(row, 0, "invocation_count")?,
389 budget_u64_from_row(row, 1, "total_cost_exposed")?,
390 budget_u64_from_row(row, 2, "total_cost_realized_spend")?,
391 ))
392 },
393 )
394 .optional()?
395 .ok_or_else(|| BudgetStoreError::Invariant("missing charged budget row".to_string()))?;
396 if hold.remaining_exposure_units != hold.authorized_exposure_units
397 || current.0 == 0
398 || current.1 < hold.authorized_exposure_units
399 {
400 transaction.rollback()?;
401 return Err(BudgetStoreError::Invariant(format!(
402 "budget hold `{}` does not match captured-before-dispatch cancellation",
403 request.hold_id
404 )));
405 }
406
407 let total_cost_exposed_after = current.1 - hold.authorized_exposure_units;
408 let event_seq = allocate_budget_replication_seq(&transaction)?;
409 transaction.execute(
410 r#"
411 UPDATE capability_grant_budgets
412 SET invocation_count = ?3,
413 updated_at = ?4,
414 seq = ?5,
415 total_cost_exposed = ?6
416 WHERE capability_id = ?1 AND grant_index = ?2
417 "#,
418 params![
419 &request.capability_id,
420 request.grant_index as i64,
421 current.0 - 1,
422 unix_now(),
423 budget_u64_to_sqlite(event_seq, "seq")?,
424 budget_u64_to_sqlite(total_cost_exposed_after, "total_cost_exposed")?,
425 ],
426 )?;
427 SqliteBudgetStore::update_hold(
428 &transaction,
429 &request.hold_id,
430 0,
431 HoldDisposition::Reversed,
432 request.authority.as_ref().or(hold.authority.as_ref()),
433 )?;
434 transaction.execute(
435 "UPDATE budget_authorization_holds SET invocation_captured = 0 WHERE hold_id = ?1",
436 params![&request.hold_id],
437 )?;
438 SqliteBudgetStore::append_mutation_event(
439 &transaction,
440 Some(&request.event_id),
441 Some(&request.hold_id),
442 request.authority.as_ref(),
443 &request.capability_id,
444 request.grant_index,
445 BudgetMutationKind::CancelCapturedBeforeDispatch,
446 Some(true),
447 event_seq,
448 Some(event_seq),
449 hold.authorized_exposure_units,
450 0,
451 None,
452 None,
453 None,
454 current.0 - 1,
455 total_cost_exposed_after,
456 current.2,
457 )?;
458 transaction.commit()?;
459
460 Ok(BudgetCapturedBeforeDispatchCancellationDecision::Cancelled(
461 BudgetHoldMutationDecision {
462 hold_id: Some(request.hold_id),
463 admission_binding: None,
464 exposure_units: hold.authorized_exposure_units,
465 realized_spend_units: 0,
466 committed_cost_units_after: checked_committed_cost_units(
467 total_cost_exposed_after,
468 current.2,
469 )?,
470 invocation_count_after: current.0 - 1,
471 invocation_quota_usages: Vec::new(),
472 cumulative_approval: None,
473 invocation_state: BudgetInvocationState::Reversed,
474 monetary_state: if hold.authorized_exposure_units == 0 {
475 BudgetMonetaryState::None
476 } else {
477 BudgetMonetaryState::Reversed
478 },
479 metadata: BudgetCommitMetadata {
480 authority: request.authority,
481 guarantee_level: self.budget_guarantee_level(),
482 budget_profile: self.budget_authority_profile(),
483 metering_profile: self.budget_metering_profile(),
484 budget_commit_index: Some(event_seq),
485 event_id: Some(request.event_id),
486 recorded_at_unix_seconds: None,
487 },
488 },
489 ))
490 }
491
492 fn release_budget_hold(
493 &self,
494 request: BudgetReleaseHoldRequest,
495 ) -> Result<BudgetReleaseHoldDecision, BudgetStoreError> {
496 request.validate()?;
497 let hold_id = request.hold_id.as_deref().ok_or_else(|| {
498 BudgetStoreError::Invariant(
499 "sqlite rich budget release requires a durable hold identity".to_string(),
500 )
501 })?;
502 let event_id = request.event_id.as_deref().ok_or_else(|| {
503 BudgetStoreError::Invariant(
504 "sqlite rich budget release requires a durable event identity".to_string(),
505 )
506 })?;
507 if self.is_structured_hold(hold_id)? {
508 return self.release_composite_hold(request);
509 }
510 self.require_standalone_mutation("legacy exposure release")?;
511 self.reduce_charge_cost_with_ids_and_authority(
512 &request.capability_id,
513 request.grant_index,
514 request.released_exposure_units,
515 Some(hold_id),
516 Some(event_id),
517 request.authority.as_ref(),
518 )?;
519 recorded_sqlite_hold_mutation(self, hold_id, event_id, BudgetMutationKind::ReleaseExposure)
520 }
521
522 fn reverse_budget_hold(
523 &self,
524 request: BudgetReverseHoldRequest,
525 ) -> Result<BudgetReverseHoldDecision, BudgetStoreError> {
526 request.validate()?;
527 let hold_id = request.hold_id.as_deref().ok_or_else(|| {
528 BudgetStoreError::Invariant(
529 "sqlite rich budget reversal requires a durable hold identity".to_string(),
530 )
531 })?;
532 let event_id = request.event_id.as_deref().ok_or_else(|| {
533 BudgetStoreError::Invariant(
534 "sqlite rich budget reversal requires a durable event identity".to_string(),
535 )
536 })?;
537 if self.is_structured_hold(hold_id)? {
538 return self.reverse_composite_hold(request);
539 }
540 self.require_standalone_mutation("legacy hold reversal")?;
541 if request.expected_cumulative_approval_state.is_some() {
542 return Err(BudgetStoreError::Invariant(
543 "legacy sqlite budget holds do not support cumulative approval state fencing"
544 .to_string(),
545 ));
546 }
547 self.reverse_charge_cost_with_ids_and_authority(
548 &request.capability_id,
549 request.grant_index,
550 request.reversed_exposure_units,
551 Some(hold_id),
552 Some(event_id),
553 request.authority.as_ref(),
554 )?;
555 recorded_sqlite_hold_mutation(self, hold_id, event_id, BudgetMutationKind::ReverseExposure)
556 }
557
558 fn reconcile_budget_hold(
559 &self,
560 request: BudgetReconcileHoldRequest,
561 ) -> Result<BudgetReconcileHoldDecision, BudgetStoreError> {
562 request.validate()?;
563 let hold_id = request.hold_id.as_deref().ok_or_else(|| {
564 BudgetStoreError::Invariant(
565 "sqlite rich budget reconciliation requires a durable hold identity".to_string(),
566 )
567 })?;
568 let event_id = request.event_id.as_deref().ok_or_else(|| {
569 BudgetStoreError::Invariant(
570 "sqlite rich budget reconciliation requires a durable event identity".to_string(),
571 )
572 })?;
573 if self.is_structured_hold(hold_id)? {
574 return self.reconcile_composite_hold(request);
575 }
576 self.require_standalone_mutation("legacy spend reconciliation")?;
577 self.settle_charge_cost_with_ids_and_authority(
578 &request.capability_id,
579 request.grant_index,
580 request.exposed_cost_units,
581 request.realized_spend_units,
582 Some(hold_id),
583 Some(event_id),
584 request.authority.as_ref(),
585 )?;
586 recorded_sqlite_hold_mutation(self, hold_id, event_id, BudgetMutationKind::ReconcileSpend)
587 }
588
589 fn capture_budget_hold(
590 &self,
591 request: BudgetCaptureHoldRequest,
592 ) -> Result<BudgetCaptureHoldDecision, BudgetStoreError> {
593 request.validate()?;
594 let hold_id = request.hold_id.as_deref().ok_or_else(|| {
595 BudgetStoreError::Invariant(
596 "sqlite monetary capture requires a durable hold identity".to_string(),
597 )
598 })?;
599 if !self.is_structured_hold(hold_id)? {
600 return Err(BudgetStoreError::Invariant(
601 "legacy sqlite budget holds do not support distinct monetary capture".to_string(),
602 ));
603 }
604 self.capture_composite_hold(request)
605 }
606
607 fn try_charge_cost(
608 &self,
609 capability_id: &str,
610 grant_index: usize,
611 max_invocations: Option<u32>,
612 cost_units: u64,
613 max_cost_per_invocation: Option<u64>,
614 max_total_cost_units: Option<u64>,
615 ) -> Result<bool, BudgetStoreError> {
616 self.try_charge_cost_with_ids(
617 capability_id,
618 grant_index,
619 max_invocations,
620 cost_units,
621 max_cost_per_invocation,
622 max_total_cost_units,
623 None,
624 None,
625 )
626 }
627
628 fn try_charge_cost_with_ids(
629 &self,
630 capability_id: &str,
631 grant_index: usize,
632 max_invocations: Option<u32>,
633 cost_units: u64,
634 max_cost_per_invocation: Option<u64>,
635 max_total_cost_units: Option<u64>,
636 hold_id: Option<&str>,
637 event_id: Option<&str>,
638 ) -> Result<bool, BudgetStoreError> {
639 self.try_charge_cost_with_ids_and_authority(
640 capability_id,
641 grant_index,
642 max_invocations,
643 cost_units,
644 max_cost_per_invocation,
645 max_total_cost_units,
646 hold_id,
647 event_id,
648 None,
649 )
650 }
651
652 fn try_charge_cost_with_ids_and_authority(
653 &self,
654 capability_id: &str,
655 grant_index: usize,
656 max_invocations: Option<u32>,
657 cost_units: u64,
658 max_cost_per_invocation: Option<u64>,
659 max_total_cost_units: Option<u64>,
660 hold_id: Option<&str>,
661 event_id: Option<&str>,
662 authority: Option<&BudgetEventAuthority>,
663 ) -> Result<bool, BudgetStoreError> {
664 self.require_standalone_mutation("unbound charge")?;
665 validate_budget_grant_index(grant_index)?;
666 let request = BudgetAuthorizeHoldRequest {
667 capability_id: capability_id.to_string(),
668 grant_index,
669 max_invocations,
670 invocation_quotas: Vec::new(),
671 cumulative_approval: None,
672 admission_binding: None,
673 requested_exposure_units: cost_units,
674 max_cost_per_invocation,
675 max_total_cost_units,
676 hold_id: hold_id.map(ToOwned::to_owned),
677 event_id: event_id.map(ToOwned::to_owned),
678 authority: authority.cloned(),
679 };
680 Ok(matches!(
681 self.authorize_budget_hold_atomic(&request)?,
682 BudgetAuthorizeHoldDecision::Authorized(_)
683 ))
684 }
685
686 fn reverse_charge_cost(
687 &self,
688 capability_id: &str,
689 grant_index: usize,
690 cost_units: u64,
691 ) -> Result<(), BudgetStoreError> {
692 self.reverse_charge_cost_with_ids(capability_id, grant_index, cost_units, None, None)
693 }
694
695 fn reverse_charge_cost_with_ids(
696 &self,
697 capability_id: &str,
698 grant_index: usize,
699 cost_units: u64,
700 hold_id: Option<&str>,
701 event_id: Option<&str>,
702 ) -> Result<(), BudgetStoreError> {
703 self.reverse_charge_cost_with_ids_and_authority(
704 capability_id,
705 grant_index,
706 cost_units,
707 hold_id,
708 event_id,
709 None,
710 )
711 }
712
713 fn reverse_charge_cost_with_ids_and_authority(
714 &self,
715 capability_id: &str,
716 grant_index: usize,
717 cost_units: u64,
718 hold_id: Option<&str>,
719 event_id: Option<&str>,
720 authority: Option<&BudgetEventAuthority>,
721 ) -> Result<(), BudgetStoreError> {
722 self.require_standalone_mutation("legacy charge reversal")?;
723 validate_budget_grant_index(grant_index)?;
724 budget_u64_to_sqlite(cost_units, "exposure_units")?;
725 if let Some(authority) = authority {
726 budget_u64_to_sqlite(authority.lease_epoch, "lease_epoch")?;
727 }
728 let mut connection = self.connection()?;
729 let transaction = self.begin_write(&mut connection)?;
730 Self::reject_structured_hold_from_legacy_writer(&transaction, hold_id, "charge reversal")?;
731
732 if SqliteBudgetStore::existing_event_allowed(
733 &transaction,
734 event_id,
735 BudgetMutationKind::ReverseExposure,
736 capability_id,
737 grant_index,
738 hold_id,
739 authority,
740 cost_units,
741 0,
742 None,
743 None,
744 None,
745 )?
746 .is_some()
747 {
748 transaction.rollback()?;
749 return Ok(());
750 }
751 if hold_id.is_none()
752 && SqliteBudgetStore::has_live_hold(&transaction, capability_id, grant_index)?
753 {
754 transaction.rollback()?;
755 return Err(BudgetStoreError::Invariant(
756 "live budget hold blocks generic reverse".to_string(),
757 ));
758 }
759 if let Some(hold_id) = hold_id {
760 let hold = SqliteBudgetStore::ensure_open_hold(
761 &transaction,
762 hold_id,
763 capability_id,
764 grant_index,
765 )?;
766 if hold.remaining_exposure_units != cost_units
767 || !hold.invocation_count_debited
768 || hold.invocation_captured
769 {
770 transaction.rollback()?;
771 return Err(BudgetStoreError::Invariant(format!(
772 "budget hold `{hold_id}` does not match reverse amount"
773 )));
774 }
775 SqliteBudgetStore::validate_hold_authority(
776 hold_id,
777 hold.authority.as_ref(),
778 authority,
779 )?;
780 }
781
782 let current = transaction
783 .query_row(
784 r#"
785 SELECT invocation_count, total_cost_exposed, total_cost_realized_spend
786 FROM capability_grant_budgets
787 WHERE capability_id = ?1 AND grant_index = ?2
788 "#,
789 params![capability_id, grant_index as i64],
790 |row| {
791 Ok((
792 budget_u32_from_row(row, 0, "invocation_count")?,
793 budget_u64_from_row(row, 1, "total_cost_exposed")?,
794 budget_u64_from_row(row, 2, "total_cost_realized_spend")?,
795 ))
796 },
797 )
798 .optional()?;
799
800 let Some((invocation_count, total_cost_exposed, total_cost_realized_spend)) = current
801 else {
802 transaction.rollback()?;
803 return Err(BudgetStoreError::Invariant(
804 "missing charged budget row".to_string(),
805 ));
806 };
807
808 if invocation_count == 0 {
809 transaction.rollback()?;
810 return Err(BudgetStoreError::Invariant(
811 "cannot reverse charge with zero invocation_count".to_string(),
812 ));
813 }
814 if total_cost_exposed < cost_units {
815 transaction.rollback()?;
816 return Err(BudgetStoreError::Invariant(
817 "cannot reverse charge larger than total_cost_exposed".to_string(),
818 ));
819 }
820
821 let new_total_cost_exposed = total_cost_exposed - cost_units;
822 let seq = allocate_budget_replication_seq(&transaction)?;
823 transaction.execute(
824 r#"
825 UPDATE capability_grant_budgets
826 SET invocation_count = ?3,
827 updated_at = ?4,
828 seq = ?5,
829 total_cost_exposed = ?6
830 WHERE capability_id = ?1 AND grant_index = ?2
831 "#,
832 params![
833 capability_id,
834 grant_index as i64,
835 invocation_count - 1,
836 unix_now(),
837 budget_u64_to_sqlite(seq, "seq")?,
838 budget_u64_to_sqlite(new_total_cost_exposed, "total_cost_exposed")?,
839 ],
840 )?;
841 if let Some(hold_id) = hold_id {
842 let next_authority = SqliteBudgetStore::validate_hold_authority(
843 hold_id,
844 SqliteBudgetStore::ensure_open_hold(
845 &transaction,
846 hold_id,
847 capability_id,
848 grant_index,
849 )?
850 .authority
851 .as_ref(),
852 authority,
853 )?;
854 SqliteBudgetStore::update_hold(
855 &transaction,
856 hold_id,
857 0,
858 HoldDisposition::Reversed,
859 next_authority.as_ref(),
860 )?;
861 }
862 SqliteBudgetStore::append_mutation_event(
863 &transaction,
864 event_id,
865 hold_id,
866 authority,
867 capability_id,
868 grant_index,
869 BudgetMutationKind::ReverseExposure,
870 None,
871 seq,
872 Some(seq),
873 cost_units,
874 0,
875 None,
876 None,
877 None,
878 invocation_count - 1,
879 new_total_cost_exposed,
880 total_cost_realized_spend,
881 )?;
882 transaction.commit()?;
883 Ok(())
884 }
885
886 fn reduce_charge_cost(
887 &self,
888 capability_id: &str,
889 grant_index: usize,
890 cost_units: u64,
891 ) -> Result<(), BudgetStoreError> {
892 self.reduce_charge_cost_with_ids(capability_id, grant_index, cost_units, None, None)
893 }
894
895 fn reduce_charge_cost_with_ids(
896 &self,
897 capability_id: &str,
898 grant_index: usize,
899 cost_units: u64,
900 hold_id: Option<&str>,
901 event_id: Option<&str>,
902 ) -> Result<(), BudgetStoreError> {
903 self.reduce_charge_cost_with_ids_and_authority(
904 capability_id,
905 grant_index,
906 cost_units,
907 hold_id,
908 event_id,
909 None,
910 )
911 }
912
913 fn reduce_charge_cost_with_ids_and_authority(
914 &self,
915 capability_id: &str,
916 grant_index: usize,
917 cost_units: u64,
918 hold_id: Option<&str>,
919 event_id: Option<&str>,
920 authority: Option<&BudgetEventAuthority>,
921 ) -> Result<(), BudgetStoreError> {
922 self.require_standalone_mutation("legacy charge reduction")?;
923 validate_budget_grant_index(grant_index)?;
924 budget_u64_to_sqlite(cost_units, "exposure_units")?;
925 if let Some(authority) = authority {
926 budget_u64_to_sqlite(authority.lease_epoch, "lease_epoch")?;
927 }
928 let mut connection = self.connection()?;
929 let transaction = self.begin_write(&mut connection)?;
930 Self::reject_structured_hold_from_legacy_writer(&transaction, hold_id, "charge reduction")?;
931
932 if SqliteBudgetStore::existing_event_allowed(
933 &transaction,
934 event_id,
935 BudgetMutationKind::ReleaseExposure,
936 capability_id,
937 grant_index,
938 hold_id,
939 authority,
940 cost_units,
941 0,
942 None,
943 None,
944 None,
945 )?
946 .is_some()
947 {
948 transaction.rollback()?;
949 return Ok(());
950 }
951 if hold_id.is_none()
952 && SqliteBudgetStore::has_live_hold(&transaction, capability_id, grant_index)?
953 {
954 transaction.rollback()?;
955 return Err(BudgetStoreError::Invariant(
956 "live budget hold blocks generic release".to_string(),
957 ));
958 }
959 if let Some(hold_id) = hold_id {
960 let hold = SqliteBudgetStore::ensure_open_hold(
961 &transaction,
962 hold_id,
963 capability_id,
964 grant_index,
965 )?;
966 if hold.invocation_captured {
967 transaction.rollback()?;
968 return Err(BudgetStoreError::Invariant(format!(
969 "budget hold `{hold_id}` invocation was already captured"
970 )));
971 }
972 if hold.remaining_exposure_units < cost_units {
973 transaction.rollback()?;
974 return Err(BudgetStoreError::Invariant(format!(
975 "budget hold `{hold_id}` cannot release more than remaining exposure"
976 )));
977 }
978 SqliteBudgetStore::validate_hold_authority(
979 hold_id,
980 hold.authority.as_ref(),
981 authority,
982 )?;
983 }
984
985 let current = transaction
986 .query_row(
987 r#"
988 SELECT invocation_count, total_cost_exposed, total_cost_realized_spend
989 FROM capability_grant_budgets
990 WHERE capability_id = ?1 AND grant_index = ?2
991 "#,
992 params![capability_id, grant_index as i64],
993 |row| {
994 Ok((
995 budget_u32_from_row(row, 0, "invocation_count")?,
996 budget_u64_from_row(row, 1, "total_cost_exposed")?,
997 budget_u64_from_row(row, 2, "total_cost_realized_spend")?,
998 ))
999 },
1000 )
1001 .optional()?;
1002
1003 let Some((invocation_count, total_cost_exposed, total_cost_realized_spend)) = current
1004 else {
1005 transaction.rollback()?;
1006 return Err(BudgetStoreError::Invariant(
1007 "missing charged budget row".to_string(),
1008 ));
1009 };
1010
1011 if total_cost_exposed < cost_units {
1012 transaction.rollback()?;
1013 return Err(BudgetStoreError::Invariant(
1014 "cannot reduce charge larger than total_cost_exposed".to_string(),
1015 ));
1016 }
1017
1018 let new_total_cost_exposed = total_cost_exposed - cost_units;
1019 let seq = allocate_budget_replication_seq(&transaction)?;
1020 transaction.execute(
1021 r#"
1022 UPDATE capability_grant_budgets
1023 SET updated_at = ?3,
1024 seq = ?4,
1025 total_cost_exposed = ?5
1026 WHERE capability_id = ?1 AND grant_index = ?2
1027 "#,
1028 params![
1029 capability_id,
1030 grant_index as i64,
1031 unix_now(),
1032 budget_u64_to_sqlite(seq, "seq")?,
1033 budget_u64_to_sqlite(new_total_cost_exposed, "total_cost_exposed")?,
1034 ],
1035 )?;
1036 if let Some(hold_id) = hold_id {
1037 let hold = SqliteBudgetStore::ensure_open_hold(
1038 &transaction,
1039 hold_id,
1040 capability_id,
1041 grant_index,
1042 )?;
1043 let next_authority = SqliteBudgetStore::validate_hold_authority(
1044 hold_id,
1045 hold.authority.as_ref(),
1046 authority,
1047 )?;
1048 let remaining = hold.remaining_exposure_units - cost_units;
1049 let disposition = if remaining == 0 {
1050 HoldDisposition::Released
1051 } else {
1052 HoldDisposition::Open
1053 };
1054 SqliteBudgetStore::update_hold(
1055 &transaction,
1056 hold_id,
1057 remaining,
1058 disposition,
1059 next_authority.as_ref(),
1060 )?;
1061 }
1062 SqliteBudgetStore::append_mutation_event(
1063 &transaction,
1064 event_id,
1065 hold_id,
1066 authority,
1067 capability_id,
1068 grant_index,
1069 BudgetMutationKind::ReleaseExposure,
1070 None,
1071 seq,
1072 Some(seq),
1073 cost_units,
1074 0,
1075 None,
1076 None,
1077 None,
1078 invocation_count,
1079 new_total_cost_exposed,
1080 total_cost_realized_spend,
1081 )?;
1082 transaction.commit()?;
1083 Ok(())
1084 }
1085
1086 fn settle_charge_cost(
1087 &self,
1088 capability_id: &str,
1089 grant_index: usize,
1090 exposed_cost_units: u64,
1091 realized_cost_units: u64,
1092 ) -> Result<(), BudgetStoreError> {
1093 self.settle_charge_cost_with_ids(
1094 capability_id,
1095 grant_index,
1096 exposed_cost_units,
1097 realized_cost_units,
1098 None,
1099 None,
1100 )
1101 }
1102
1103 fn settle_charge_cost_with_ids(
1104 &self,
1105 capability_id: &str,
1106 grant_index: usize,
1107 exposed_cost_units: u64,
1108 realized_cost_units: u64,
1109 hold_id: Option<&str>,
1110 event_id: Option<&str>,
1111 ) -> Result<(), BudgetStoreError> {
1112 self.settle_charge_cost_with_ids_and_authority(
1113 capability_id,
1114 grant_index,
1115 exposed_cost_units,
1116 realized_cost_units,
1117 hold_id,
1118 event_id,
1119 None,
1120 )
1121 }
1122
1123 fn settle_charge_cost_with_ids_and_authority(
1124 &self,
1125 capability_id: &str,
1126 grant_index: usize,
1127 exposed_cost_units: u64,
1128 realized_cost_units: u64,
1129 hold_id: Option<&str>,
1130 event_id: Option<&str>,
1131 authority: Option<&BudgetEventAuthority>,
1132 ) -> Result<(), BudgetStoreError> {
1133 self.require_standalone_mutation("legacy charge settlement")?;
1134 validate_budget_grant_index(grant_index)?;
1135 if realized_cost_units > exposed_cost_units {
1136 return Err(BudgetStoreError::Invariant(
1137 "cannot realize spend larger than exposed cost".to_string(),
1138 ));
1139 }
1140 budget_u64_to_sqlite(exposed_cost_units, "exposure_units")?;
1141 budget_u64_to_sqlite(realized_cost_units, "realized_spend_units")?;
1142 if let Some(authority) = authority {
1143 budget_u64_to_sqlite(authority.lease_epoch, "lease_epoch")?;
1144 }
1145
1146 let mut connection = self.connection()?;
1147 let transaction = self.begin_write(&mut connection)?;
1148 Self::reject_structured_hold_from_legacy_writer(
1149 &transaction,
1150 hold_id,
1151 "charge settlement",
1152 )?;
1153
1154 if SqliteBudgetStore::existing_event_allowed(
1155 &transaction,
1156 event_id,
1157 BudgetMutationKind::ReconcileSpend,
1158 capability_id,
1159 grant_index,
1160 hold_id,
1161 authority,
1162 exposed_cost_units,
1163 realized_cost_units,
1164 None,
1165 None,
1166 None,
1167 )?
1168 .is_some()
1169 {
1170 transaction.rollback()?;
1171 return Ok(());
1172 }
1173 if hold_id.is_none()
1174 && SqliteBudgetStore::has_live_hold(&transaction, capability_id, grant_index)?
1175 {
1176 transaction.rollback()?;
1177 return Err(BudgetStoreError::Invariant(
1178 "live budget hold blocks generic reconciliation".to_string(),
1179 ));
1180 }
1181 if let Some(hold_id) = hold_id {
1182 let hold = SqliteBudgetStore::ensure_open_hold(
1183 &transaction,
1184 hold_id,
1185 capability_id,
1186 grant_index,
1187 )?;
1188 if !hold.invocation_captured {
1189 transaction.rollback()?;
1190 return Err(BudgetStoreError::Invariant(format!(
1191 "budget hold `{hold_id}` invocation was not captured before reconciliation"
1192 )));
1193 }
1194 if hold.remaining_exposure_units != exposed_cost_units {
1195 transaction.rollback()?;
1196 return Err(BudgetStoreError::Invariant(format!(
1197 "budget hold `{hold_id}` does not match reconciled exposure"
1198 )));
1199 }
1200 SqliteBudgetStore::validate_hold_authority(
1201 hold_id,
1202 hold.authority.as_ref(),
1203 authority,
1204 )?;
1205 }
1206
1207 let current = transaction
1208 .query_row(
1209 r#"
1210 SELECT invocation_count, total_cost_exposed, total_cost_realized_spend
1211 FROM capability_grant_budgets
1212 WHERE capability_id = ?1 AND grant_index = ?2
1213 "#,
1214 params![capability_id, grant_index as i64],
1215 |row| {
1216 Ok((
1217 budget_u32_from_row(row, 0, "invocation_count")?,
1218 budget_u64_from_row(row, 1, "total_cost_exposed")?,
1219 budget_u64_from_row(row, 2, "total_cost_realized_spend")?,
1220 ))
1221 },
1222 )
1223 .optional()?;
1224
1225 let Some((invocation_count, total_cost_exposed, total_cost_realized_spend)) = current
1226 else {
1227 transaction.rollback()?;
1228 return Err(BudgetStoreError::Invariant(
1229 "missing charged budget row".to_string(),
1230 ));
1231 };
1232
1233 if invocation_count == 0 {
1234 transaction.rollback()?;
1235 return Err(BudgetStoreError::Invariant(
1236 "cannot settle charge with zero invocation_count".to_string(),
1237 ));
1238 }
1239 if total_cost_exposed < exposed_cost_units {
1240 transaction.rollback()?;
1241 return Err(BudgetStoreError::Invariant(
1242 "cannot settle more exposure than total_cost_exposed".to_string(),
1243 ));
1244 }
1245
1246 let new_total_cost_exposed = total_cost_exposed - exposed_cost_units;
1247 let new_total_cost_realized_spend = total_cost_realized_spend
1248 .checked_add(realized_cost_units)
1249 .ok_or_else(|| {
1250 BudgetStoreError::Overflow(
1251 "total_cost_realized_spend + realized_cost_units overflowed u64".to_string(),
1252 )
1253 })?;
1254 budget_u64_to_sqlite(new_total_cost_realized_spend, "total_cost_realized_spend")?;
1255
1256 let seq = allocate_budget_replication_seq(&transaction)?;
1257 transaction.execute(
1258 r#"
1259 UPDATE capability_grant_budgets
1260 SET updated_at = ?3,
1261 seq = ?4,
1262 total_cost_exposed = ?5,
1263 total_cost_realized_spend = ?6
1264 WHERE capability_id = ?1 AND grant_index = ?2
1265 "#,
1266 params![
1267 capability_id,
1268 grant_index as i64,
1269 unix_now(),
1270 budget_u64_to_sqlite(seq, "seq")?,
1271 budget_u64_to_sqlite(new_total_cost_exposed, "total_cost_exposed")?,
1272 budget_u64_to_sqlite(new_total_cost_realized_spend, "total_cost_realized_spend",)?,
1273 ],
1274 )?;
1275 if let Some(hold_id) = hold_id {
1276 let next_authority = SqliteBudgetStore::validate_hold_authority(
1277 hold_id,
1278 SqliteBudgetStore::ensure_open_hold(
1279 &transaction,
1280 hold_id,
1281 capability_id,
1282 grant_index,
1283 )?
1284 .authority
1285 .as_ref(),
1286 authority,
1287 )?;
1288 SqliteBudgetStore::update_hold(
1289 &transaction,
1290 hold_id,
1291 0,
1292 HoldDisposition::Reconciled,
1293 next_authority.as_ref(),
1294 )?;
1295 }
1296 SqliteBudgetStore::append_mutation_event(
1297 &transaction,
1298 event_id,
1299 hold_id,
1300 authority,
1301 capability_id,
1302 grant_index,
1303 BudgetMutationKind::ReconcileSpend,
1304 None,
1305 seq,
1306 Some(seq),
1307 exposed_cost_units,
1308 realized_cost_units,
1309 None,
1310 None,
1311 None,
1312 invocation_count,
1313 new_total_cost_exposed,
1314 new_total_cost_realized_spend,
1315 )?;
1316 transaction.commit()?;
1317 Ok(())
1318 }
1319
1320 fn list_usages(
1321 &self,
1322 limit: usize,
1323 capability_id: Option<&str>,
1324 ) -> Result<Vec<BudgetUsageRecord>, BudgetStoreError> {
1325 let mut connection = self.connection()?;
1326 let transaction = self.begin_read(&mut connection)?;
1327 let mut statement = transaction.prepare(
1328 r#"
1329 SELECT
1330 capability_id,
1331 grant_index,
1332 invocation_count,
1333 updated_at,
1334 seq,
1335 total_cost_exposed,
1336 total_cost_realized_spend
1337 FROM capability_grant_budgets
1338 WHERE (?1 IS NULL OR capability_id = ?1)
1339 ORDER BY updated_at DESC, capability_id ASC, grant_index ASC
1340 LIMIT ?2
1341 "#,
1342 )?;
1343 let rows = statement.query_map(params![capability_id, limit as i64], record_from_row)?;
1344 let rows = rows.collect::<Result<Vec<_>, _>>()?;
1345 drop(statement);
1346 transaction.rollback()?;
1347 Ok(rows)
1348 }
1349
1350 fn get_usage(
1351 &self,
1352 capability_id: &str,
1353 grant_index: usize,
1354 ) -> Result<Option<BudgetUsageRecord>, BudgetStoreError> {
1355 validate_budget_grant_index(grant_index)?;
1356 let mut connection = self.connection()?;
1357 let transaction = self.begin_read(&mut connection)?;
1358 let row = transaction
1359 .query_row(
1360 r#"
1361 SELECT
1362 capability_id,
1363 grant_index,
1364 invocation_count,
1365 updated_at,
1366 seq,
1367 total_cost_exposed,
1368 total_cost_realized_spend
1369 FROM capability_grant_budgets
1370 WHERE capability_id = ?1 AND grant_index = ?2
1371 "#,
1372 params![capability_id, grant_index as i64],
1373 record_from_row,
1374 )
1375 .optional()?;
1376 transaction.rollback()?;
1377 Ok(row)
1378 }
1379
1380 fn get_invocation_quota_usage(
1381 &self,
1382 key: &BudgetQuotaKey,
1383 ) -> Result<Option<BudgetInvocationQuotaUsage>, BudgetStoreError> {
1384 self.composite_quota_usage(key)
1385 }
1386
1387 fn get_cumulative_approval_account_usage(
1388 &self,
1389 key: &BudgetCumulativeApprovalAccountKey,
1390 ) -> Result<Option<BudgetCumulativeApprovalAccountUsage>, BudgetStoreError> {
1391 let mut connection = self.connection()?;
1392 let transaction = self.begin_read(&mut connection)?;
1393 let row = transaction
1394 .query_row(
1395 r#"
1396 SELECT root_grant_hash, delegation_root_id, root_binding_digest,
1397 currency, authority_threshold_units,
1398 reserved_authorized_units, captured_authorized_units, version
1399 FROM budget_cumulative_approval_accounts
1400 WHERE authority_id = ?1 AND owner_id = ?2
1401 AND approval_budget_id = ?3 AND approval_budget_epoch = ?4
1402 "#,
1403 params![
1404 &key.authority_id,
1405 &key.owner_id,
1406 &key.approval_budget_id,
1407 budget_u64_to_sqlite(key.approval_budget_epoch, "approval_budget_epoch")?,
1408 ],
1409 |row| {
1410 Ok((
1411 row.get::<_, String>(0)?,
1412 row.get::<_, Option<String>>(1)?,
1413 row.get::<_, Option<String>>(2)?,
1414 row.get::<_, String>(3)?,
1415 budget_u64_from_row(row, 4, "authority_threshold_units")?,
1416 budget_u64_from_row(row, 5, "reserved_authorized_units")?,
1417 budget_u64_from_row(row, 6, "captured_authorized_units")?,
1418 budget_u64_from_row(row, 7, "version")?,
1419 ))
1420 },
1421 )
1422 .optional()?;
1423 let Some(row) = row else {
1424 transaction.rollback()?;
1425 return Ok(None);
1426 };
1427 if row.0 != key.root_grant_hash
1428 || row.1 != key.delegation_root_id
1429 || row.2 != key.root_binding_digest
1430 || row.3 != key.currency
1431 {
1432 return Err(BudgetStoreError::Invariant(
1433 "cumulative approval account immutable identity changed".to_string(),
1434 ));
1435 }
1436 let amount = |units| MonetaryAmount {
1437 units,
1438 currency: key.currency.clone(),
1439 };
1440 let usage = BudgetCumulativeApprovalAccountUsage {
1441 account_key: key.clone(),
1442 authority_threshold: amount(row.4),
1443 reserved_authorized: amount(row.5),
1444 captured_authorized: amount(row.6),
1445 version: row.7,
1446 };
1447 transaction.rollback()?;
1448 Ok(Some(usage))
1449 }
1450
1451 fn get_cumulative_approval_operation_usage(
1452 &self,
1453 operation_id: &str,
1454 ) -> Result<Option<BudgetCumulativeApprovalUsage>, BudgetStoreError> {
1455 self.composite_cumulative_operation_usage(operation_id)
1456 }
1457
1458 fn list_mutation_events(
1459 &self,
1460 limit: usize,
1461 capability_id: Option<&str>,
1462 grant_index: Option<usize>,
1463 ) -> Result<Vec<BudgetMutationRecord>, BudgetStoreError> {
1464 if let Some(grant_index) = grant_index {
1465 validate_budget_grant_index(grant_index)?;
1466 }
1467 let mut connection = self.connection()?;
1468 let transaction = self.begin_read(&mut connection)?;
1469 let mut statement = transaction.prepare(
1470 r#"
1471 SELECT event_id
1472 FROM budget_mutation_events
1473 WHERE (?1 IS NULL OR capability_id = ?1)
1474 AND (?2 IS NULL OR grant_index = ?2)
1475 ORDER BY event_seq ASC
1476 LIMIT ?3
1477 "#,
1478 )?;
1479 let event_ids = statement
1480 .query_map(
1481 params![
1482 capability_id,
1483 grant_index.map(|value| value as i64),
1484 limit as i64
1485 ],
1486 |row| row.get::<_, String>(0),
1487 )?
1488 .collect::<Result<Vec<_>, _>>()?;
1489 drop(statement);
1490 let events = event_ids
1491 .iter()
1492 .map(|event_id| {
1493 Self::load_projected_mutation_event(&transaction, event_id)?.ok_or_else(|| {
1494 BudgetStoreError::Invariant(format!(
1495 "budget mutation event `{event_id}` disappeared while listing"
1496 ))
1497 })
1498 })
1499 .collect::<Result<Vec<_>, _>>()?;
1500 transaction.rollback()?;
1501 Ok(events)
1502 }
1503
1504 fn reap_orphaned_holds(
1505 &self,
1506 realized_by_hold: &std::collections::HashMap<String, u64>,
1507 ) -> Result<(usize, usize), BudgetStoreError> {
1508 let summary = self.reap_holds_by_map(realized_by_hold)?;
1509 Ok((summary.reconciled, summary.reversed))
1510 }
1511
1512 fn count_open_holds(&self) -> Result<usize, BudgetStoreError> {
1513 Ok(self.list_open_holds()?.len())
1514 }
1515
1516 fn list_open_delegated_reserved_hold_ids(
1517 &self,
1518 ) -> Result<Option<Vec<String>>, BudgetStoreError> {
1519 Ok(Some(self.list_open_delegated_reserved_holds()?))
1520 }
1521
1522 fn request_id_has_reserved_hold(
1523 &self,
1524 request_id: &str,
1525 ) -> Result<Option<bool>, BudgetStoreError> {
1526 Ok(Some(self.hold_exists_for_request_id(request_id)?))
1527 }
1528
1529 fn get_budget_hold(
1530 &self,
1531 hold_id: &str,
1532 ) -> Result<Option<BudgetHoldSnapshot>, BudgetStoreError> {
1533 self.budget_hold_snapshot(hold_id)
1534 }
1535
1536 fn mark_hold_reserved(
1537 &self,
1538 hold_id: &str,
1539 reserved_until_unix_secs: i64,
1540 currency: &str,
1541 payment_reference: Option<&str>,
1542 envelope: &ReservedHoldEnvelope,
1543 ) -> Result<(), BudgetStoreError> {
1544 self.mark_hold_reserved_until(
1545 hold_id,
1546 reserved_until_unix_secs,
1547 currency,
1548 payment_reference,
1549 envelope,
1550 )
1551 }
1552
1553 fn reserve_invocation_hold(
1554 &self,
1555 hold_id: &str,
1556 capability_id: &str,
1557 grant_index: usize,
1558 reserved_until_unix_secs: i64,
1559 envelope: &ReservedHoldEnvelope,
1560 ) -> Result<(), BudgetStoreError> {
1561 SqliteBudgetStore::reserve_invocation_hold(
1562 self,
1563 hold_id,
1564 capability_id,
1565 grant_index,
1566 reserved_until_unix_secs,
1567 envelope,
1568 )
1569 }
1570
1571 fn reap_expired_reserved_holds(&self, now_unix_secs: i64) -> Result<usize, BudgetStoreError> {
1572 SqliteBudgetStore::reap_expired_reserved_holds(self, now_unix_secs)
1573 }
1574}
1575
1576fn recorded_sqlite_hold_mutation(
1577 store: &SqliteBudgetStore,
1578 hold_id: &str,
1579 event_id: &str,
1580 expected_kind: BudgetMutationKind,
1581) -> Result<BudgetHoldMutationDecision, BudgetStoreError> {
1582 let mut connection = store.connection()?;
1583 let transaction = store.begin_read(&mut connection)?;
1584 let event =
1585 SqliteBudgetStore::load_mutation_event(&transaction, event_id)?.ok_or_else(|| {
1586 BudgetStoreError::Invariant(format!(
1587 "sqlite budget mutation event `{event_id}` disappeared"
1588 ))
1589 })?;
1590 if event.kind != expected_kind || event.hold_id.as_deref() != Some(hold_id) {
1591 return Err(BudgetStoreError::Invariant(format!(
1592 "sqlite budget mutation event `{event_id}` changed identity"
1593 )));
1594 }
1595 let (authorization_seq, authorized_exposure_units) = transaction
1596 .query_row(
1597 r#"
1598 SELECT event_seq, exposure_units
1599 FROM budget_mutation_events
1600 WHERE hold_id = ?1
1601 AND kind = ?2
1602 AND allowed = 1
1603 AND event_seq < ?3
1604 ORDER BY event_seq DESC LIMIT 1
1605 "#,
1606 params![
1607 hold_id,
1608 BudgetMutationKind::AuthorizeExposure.as_str(),
1609 budget_u64_to_sqlite(event.event_seq, "event_seq")?,
1610 ],
1611 |row| {
1612 Ok((
1613 budget_u64_from_row(row, 0, "event_seq")?,
1614 budget_u64_from_row(row, 1, "exposure_units")?,
1615 ))
1616 },
1617 )
1618 .optional()?
1619 .ok_or_else(|| {
1620 BudgetStoreError::Invariant(format!(
1621 "sqlite budget hold `{hold_id}` has no authorization generation"
1622 ))
1623 })?;
1624 let invocation_captured = transaction.query_row(
1625 r#"
1626 SELECT EXISTS(
1627 SELECT 1 FROM budget_mutation_events
1628 WHERE hold_id = ?1
1629 AND kind = ?2
1630 AND event_seq > ?3
1631 AND event_seq <= ?4
1632 )
1633 "#,
1634 params![
1635 hold_id,
1636 BudgetMutationKind::CaptureInvocation.as_str(),
1637 budget_u64_to_sqlite(authorization_seq, "authorization_seq")?,
1638 budget_u64_to_sqlite(event.event_seq, "event_seq")?,
1639 ],
1640 |row| Ok(row.get::<_, i64>(0)? != 0),
1641 )?;
1642 let invocation_state = if expected_kind == BudgetMutationKind::ReverseExposure {
1643 BudgetInvocationState::Reversed
1644 } else if invocation_captured {
1645 BudgetInvocationState::Captured
1646 } else {
1647 BudgetInvocationState::Authorized
1648 };
1649 let monetary_state = if expected_kind == BudgetMutationKind::ReverseExposure {
1650 if event.exposure_units == 0 {
1651 BudgetMonetaryState::None
1652 } else {
1653 BudgetMonetaryState::Reversed
1654 }
1655 } else if expected_kind == BudgetMutationKind::ReleaseExposure {
1656 let released_exposure_units = transaction.query_row(
1657 r#"
1658 SELECT COALESCE(SUM(exposure_units), 0)
1659 FROM budget_mutation_events
1660 WHERE hold_id = ?1
1661 AND kind = ?2
1662 AND event_seq > ?3
1663 AND event_seq <= ?4
1664 "#,
1665 params![
1666 hold_id,
1667 BudgetMutationKind::ReleaseExposure.as_str(),
1668 budget_u64_to_sqlite(authorization_seq, "authorization_seq")?,
1669 budget_u64_to_sqlite(event.event_seq, "event_seq")?,
1670 ],
1671 |row| budget_u64_from_row(row, 0, "released_exposure_units"),
1672 )?;
1673 let remaining_exposure_units = authorized_exposure_units
1674 .checked_sub(released_exposure_units)
1675 .ok_or_else(|| {
1676 BudgetStoreError::Invariant(format!(
1677 "sqlite budget hold `{hold_id}` release history exceeds authorization"
1678 ))
1679 })?;
1680 if authorized_exposure_units == 0 {
1681 BudgetMonetaryState::None
1682 } else if remaining_exposure_units == 0 {
1683 BudgetMonetaryState::Released
1684 } else {
1685 BudgetMonetaryState::Exposed
1686 }
1687 } else if event.exposure_units == 0 && event.realized_spend_units == 0 {
1688 BudgetMonetaryState::None
1689 } else {
1690 BudgetMonetaryState::Reconciled
1691 };
1692 let decision = BudgetHoldMutationDecision {
1693 hold_id: Some(hold_id.to_string()),
1694 admission_binding: None,
1695 exposure_units: event.exposure_units,
1696 realized_spend_units: event.realized_spend_units,
1697 committed_cost_units_after: checked_committed_cost_units(
1698 event.total_cost_exposed_after,
1699 event.total_cost_realized_spend_after,
1700 )?,
1701 invocation_count_after: event.invocation_count_after,
1702 invocation_quota_usages: Vec::new(),
1703 cumulative_approval: None,
1704 invocation_state,
1705 monetary_state,
1706 metadata: BudgetCommitMetadata {
1707 authority: event.authority,
1708 guarantee_level: store.budget_guarantee_level(),
1709 budget_profile: store.budget_authority_profile(),
1710 metering_profile: store.budget_metering_profile(),
1711 budget_commit_index: Some(event.event_seq),
1712 event_id: Some(event.event_id),
1713 recorded_at_unix_seconds: u64::try_from(event.recorded_at).ok(),
1714 },
1715 };
1716 transaction.rollback()?;
1717 Ok(decision)
1718}