1use std::collections::BTreeMap;
43use std::fmt;
44use std::fs::{File, OpenOptions};
45use std::io::{Read, Seek, SeekFrom, Write};
46use std::path::{Path, PathBuf};
47use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
48use std::sync::{Arc, Mutex};
49
50use rustev_contract::execution::CostPolicy;
51use rustev_contract::run::Charge;
52use rustev_runtime::Ledger;
53use serde::{Deserialize, Serialize};
54
55pub const JOURNAL_SCHEMA: &str = "rustev-jev.spend-journal/1";
56pub const TESTING_CAP_UNITS: u64 = 10_000_000_000;
59pub const SUPERSEDED_TESTING_CAP_UNITS: u64 = 5_000_000_000;
62pub const PRODUCTION_CAP_UNITS: u64 = 25_000_000_000;
64
65#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
67#[serde(rename_all = "snake_case")]
68pub enum BudgetKind {
69 Testing,
71 Production,
73}
74
75impl BudgetKind {
76 pub const fn cap(self) -> u64 {
77 match self {
78 BudgetKind::Testing => TESTING_CAP_UNITS,
79 BudgetKind::Production => PRODUCTION_CAP_UNITS,
80 }
81 }
82
83 pub fn period(self, now_ms: i64) -> String {
86 match self {
87 BudgetKind::Testing => "total".to_string(),
88 BudgetKind::Production => utc_month(now_ms),
89 }
90 }
91}
92
93pub fn utc_month(ms: i64) -> String {
95 let days = ms.div_euclid(86_400_000);
96 let z = days + 719_468;
98 let era = z.div_euclid(146_097);
99 let doe = z - era * 146_097;
100 let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365;
101 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
102 let mp = (5 * doy + 2) / 153;
103 let m = if mp < 10 { mp + 3 } else { mp - 9 };
104 let y = yoe + era * 400 + i64::from(m <= 2);
105 format!("{y:04}-{m:02}")
106}
107
108pub trait SpendClock: Send + Sync {
111 fn now_ms(&self) -> i64;
112}
113
114pub struct SystemClock;
116
117impl SpendClock for SystemClock {
118 fn now_ms(&self) -> i64 {
119 std::time::SystemTime::now()
120 .duration_since(std::time::UNIX_EPOCH)
121 .map(|d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX))
122 .unwrap_or(0)
123 }
124}
125
126#[derive(Debug, Default)]
128pub struct SetClock(AtomicI64);
129
130impl SetClock {
131 pub fn new(ms: i64) -> Self {
132 SetClock(AtomicI64::new(ms))
133 }
134 pub fn set(&self, ms: i64) {
135 self.0.store(ms, Ordering::SeqCst);
136 }
137}
138
139impl SpendClock for SetClock {
140 fn now_ms(&self) -> i64 {
141 self.0.load(Ordering::SeqCst)
142 }
143}
144
145#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
146#[serde(deny_unknown_fields)]
147struct Header {
148 schema: String,
149 budget: BudgetKind,
150 cap: u64,
151}
152
153#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
154#[serde(tag = "op", rename_all = "snake_case")]
155enum Entry {
156 Reserve {
157 seq: u64,
158 period: String,
159 attempt: String,
160 units: u64,
161 at_ms: i64,
162 },
163 Settle {
164 seq: u64,
165 charge: Charge,
166 },
167 Reconcile {
168 seq: u64,
169 units: u64,
170 },
171 Release {
172 seq: u64,
173 },
174}
175
176#[derive(Debug, Clone, Copy, PartialEq, Eq)]
177enum ResState {
178 Outstanding,
179 Observed(u64),
180 Estimated(u64),
181 Liability,
182 Unknown,
183 Released,
184}
185
186#[derive(Debug, Clone)]
187struct Res {
188 attempt: String,
189 period: String,
190 units: u64,
191 state: ResState,
192}
193
194#[derive(Debug, Clone, Copy, PartialEq, Eq)]
196pub struct Reservation {
197 pub seq: u64,
198 pub units: u64,
199}
200
201#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
203#[serde(deny_unknown_fields)]
204pub struct RunSpend {
205 pub observed: u64,
206 pub estimated: u64,
207 pub liability: u64,
208 pub unknown: u64,
209}
210
211#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
213#[serde(deny_unknown_fields)]
214pub struct SpendSummary {
215 pub budget: BudgetKind,
216 pub period: String,
217 pub cap: u64,
218 pub observed: u64,
219 pub estimated: u64,
220 pub liability: u64,
223 pub outstanding: u64,
225 pub headroom: u64,
228}
229
230impl SpendSummary {
231 pub fn committed(&self) -> u64 {
232 self.observed
233 .saturating_add(self.estimated)
234 .saturating_add(self.liability)
235 .saturating_add(self.outstanding)
236 }
237}
238
239#[derive(Debug)]
241pub enum JournalError {
242 Io(String),
243 Locked(PathBuf),
245 Mismatch(String),
247 Corrupt {
249 line: usize,
250 detail: String,
251 },
252}
253
254impl fmt::Display for JournalError {
255 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
256 match self {
257 JournalError::Io(e) => write!(f, "spend journal: {e}"),
258 JournalError::Locked(p) => write!(
259 f,
260 "spend journal: {} exists; another process holds the journal, or one ended \
261 without releasing it (remove it only after checking)",
262 p.display()
263 ),
264 JournalError::Mismatch(e) => write!(f, "spend journal: {e}"),
265 JournalError::Corrupt { line, detail } => {
266 write!(f, "spend journal: line {line}: {detail}")
267 }
268 }
269 }
270}
271
272impl std::error::Error for JournalError {}
273
274#[derive(Debug, Clone, PartialEq, Eq)]
276pub enum SpendRefusal {
277 RunCap,
279 OverCap {
281 budget: BudgetKind,
282 period: String,
283 cap: u64,
284 committed: u64,
285 requested: u64,
286 },
287 Journal(String),
289}
290
291impl fmt::Display for SpendRefusal {
292 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
293 match self {
294 SpendRefusal::RunCap => f.write_str("run-cap-exceeded"),
295 SpendRefusal::OverCap {
296 budget,
297 period,
298 cap,
299 committed,
300 requested,
301 } => write!(
302 f,
303 "spend cap: {budget:?} {period} has {committed} of {cap} units committed; \
304 {requested} more would pass it; not sent"
305 ),
306 SpendRefusal::Journal(e) => write!(f, "spend journal unavailable ({e}); not sent"),
307 }
308 }
309}
310
311struct State {
312 file: File,
313 next_seq: u64,
314 res: BTreeMap<u64, Res>,
315}
316
317struct Lock(PathBuf);
318
319impl Drop for Lock {
320 fn drop(&mut self) {
321 let _ = std::fs::remove_file(&self.0);
322 }
323}
324
325struct Inner {
326 path: PathBuf,
327 kind: BudgetKind,
328 clock: Arc<dyn SpendClock>,
329 state: Mutex<State>,
330 write_failures: AtomicU64,
331 _lock: Lock,
333}
334
335#[derive(Clone)]
337pub struct SpendJournal {
338 inner: Arc<Inner>,
339}
340
341impl fmt::Debug for SpendJournal {
342 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
343 f.debug_struct("SpendJournal")
344 .field("path", &self.inner.path)
345 .field("budget", &self.inner.kind)
346 .finish()
347 }
348}
349
350fn io(e: std::io::Error) -> JournalError {
351 JournalError::Io(e.to_string())
352}
353
354fn lock_path(path: &Path) -> PathBuf {
355 let mut s = path.as_os_str().to_owned();
356 s.push(".lock");
357 PathBuf::from(s)
358}
359
360fn append(file: &mut File, line: &[u8]) -> std::io::Result<()> {
361 let mut buf = Vec::with_capacity(line.len() + 1);
362 buf.extend_from_slice(line);
363 buf.push(b'\n');
364 #[cfg(feature = "fault-injection")]
365 let entry: serde_json::Value = serde_json::from_slice(line).expect("journal entry");
366 #[cfg(feature = "fault-injection")]
367 let name = entry["op"].as_str().unwrap_or("header");
368 #[cfg(feature = "fault-injection")]
369 crate::fault_injection::point("journal", name, "before-write")?;
370 file.write_all(&buf)?;
371 #[cfg(feature = "fault-injection")]
372 crate::fault_injection::point("journal", name, "before-sync")?;
373 file.sync_data()?;
374 #[cfg(feature = "fault-injection")]
375 crate::fault_injection::point("journal", name, "after-sync")?;
376 Ok(())
377}
378
379impl SpendJournal {
380 pub fn open(
386 path: impl AsRef<Path>,
387 kind: BudgetKind,
388 clock: Arc<dyn SpendClock>,
389 ) -> Result<Self, JournalError> {
390 let path = path.as_ref().to_path_buf();
391 let lp = lock_path(&path);
392 let mut lf = match OpenOptions::new().write(true).create_new(true).open(&lp) {
393 Ok(f) => f,
394 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
395 return Err(JournalError::Locked(lp));
396 }
397 Err(e) => return Err(io(e)),
398 };
399 let lock = Lock(lp);
400 let _ = writeln!(lf, "{}", std::process::id());
401 let _ = lf.sync_all();
402
403 let created = !path.exists();
404 let mut file = OpenOptions::new()
405 .read(true)
406 .append(true)
407 .create(true)
408 .open(&path)
409 .map_err(io)?;
410 let header = Header {
411 schema: JOURNAL_SCHEMA.to_string(),
412 budget: kind,
413 cap: kind.cap(),
414 };
415 let mut text = String::new();
416 file.read_to_string(&mut text).map_err(io)?;
417 if created || text.is_empty() {
418 let line = serde_json::to_vec(&header).map_err(|e| JournalError::Io(e.to_string()))?;
419 append(&mut file, &line).map_err(io)?;
420 file.sync_all().map_err(io)?;
421 if let Some(dir) = path.parent().filter(|d| !d.as_os_str().is_empty())
422 && let Ok(d) = File::open(dir)
423 {
424 let _ = d.sync_all();
425 }
426 text = String::from_utf8(line).unwrap_or_default() + "\n";
427 }
428 if !text.ends_with('\n') {
430 let keep = text.rfind('\n').map(|i| i + 1).unwrap_or(0);
431 file.set_len(keep as u64).map_err(io)?;
432 file.sync_all().map_err(io)?;
433 text.truncate(keep);
434 }
435 file.seek(SeekFrom::End(0)).map_err(io)?;
436 let first = text.lines().next().unwrap_or_default();
437 let found: Header = serde_json::from_str(first).map_err(|e| JournalError::Corrupt {
438 line: 1,
439 detail: format!("header: {e}"),
440 })?;
441 let superseded = Header {
442 cap: SUPERSEDED_TESTING_CAP_UNITS,
443 ..header.clone()
444 };
445 let migrate = kind == BudgetKind::Testing && found == superseded;
446 if !migrate && found != header {
447 return Err(JournalError::Mismatch(format!(
448 "the journal holds {:?} with cap {} under {}, not {:?} with cap {}",
449 found.budget,
450 found.cap,
451 found.schema,
452 kind,
453 kind.cap()
454 )));
455 }
456 let mut res = BTreeMap::new();
457 let mut next_seq = 1;
458 for (i, l) in text.lines().enumerate().skip(1) {
459 let corrupt = |detail: String| JournalError::Corrupt {
460 line: i + 1,
461 detail,
462 };
463 let e: Entry = serde_json::from_str(l).map_err(|e| corrupt(e.to_string()))?;
464 match e {
465 Entry::Reserve {
466 seq,
467 period,
468 attempt,
469 units,
470 ..
471 } => {
472 if res.contains_key(&seq) {
473 return Err(corrupt(format!("reservation {seq} made twice")));
474 }
475 next_seq = next_seq.max(seq + 1);
476 res.insert(
477 seq,
478 Res {
479 attempt,
480 period,
481 units,
482 state: ResState::Outstanding,
483 },
484 );
485 }
486 Entry::Settle { seq, charge } => {
487 let r = res
488 .get_mut(&seq)
489 .ok_or_else(|| corrupt(format!("no reservation {seq}")))?;
490 r.state = settled(charge);
491 }
492 Entry::Reconcile { seq, units } => {
493 let r = res
494 .get_mut(&seq)
495 .ok_or_else(|| corrupt(format!("no reservation {seq}")))?;
496 r.state = ResState::Observed(units);
497 }
498 Entry::Release { seq } => {
499 let r = res
500 .get_mut(&seq)
501 .ok_or_else(|| corrupt(format!("no reservation {seq}")))?;
502 r.state = ResState::Released;
503 }
504 }
505 }
506 if migrate {
509 let first_len = text.find('\n').map(|i| i + 1).unwrap_or(text.len());
510 let mut bytes =
511 serde_json::to_vec(&header).map_err(|e| JournalError::Io(e.to_string()))?;
512 bytes.push(b'\n');
513 bytes.extend_from_slice(&text.as_bytes()[first_len..]);
514 drop(file);
518 file = migrate_header(&path, &bytes).map_err(io)?;
519 }
520 for r in res.values_mut() {
523 if r.state == ResState::Outstanding {
524 r.state = ResState::Liability;
525 }
526 }
527 Ok(SpendJournal {
528 inner: Arc::new(Inner {
529 path,
530 kind,
531 clock,
532 state: Mutex::new(State {
533 file,
534 next_seq,
535 res,
536 }),
537 write_failures: AtomicU64::new(0),
538 _lock: lock,
539 }),
540 })
541 }
542
543 fn lock(&self) -> std::sync::MutexGuard<'_, State> {
544 self.inner.state.lock().unwrap_or_else(|e| e.into_inner())
545 }
546
547 pub fn budget(&self) -> BudgetKind {
548 self.inner.kind
549 }
550
551 pub fn path(&self) -> &Path {
552 &self.inner.path
553 }
554
555 pub fn write_failures(&self) -> u64 {
559 self.inner.write_failures.load(Ordering::SeqCst)
560 }
561
562 pub fn reserve(&self, attempt_id: &str, units: u64) -> Result<Reservation, SpendRefusal> {
565 self.reserve_checked(attempt_id, units, None)
566 }
567
568 pub(crate) fn reserve_checked(
569 &self,
570 attempt_id: &str,
571 units: u64,
572 run: Option<(&str, u64)>,
573 ) -> Result<Reservation, SpendRefusal> {
574 let now = self.inner.clock.now_ms();
575 let kind = self.inner.kind;
576 let period = kind.period(now);
577 let mut s = self.lock();
578 if let Some((run_id, cap)) = run {
579 let committed = run_total(&s.res, run_id);
580 if committed > cap || units > cap.saturating_sub(committed) {
581 return Err(SpendRefusal::RunCap);
582 }
583 }
584 let t = totals(&s.res, kind, &period);
585 let committed = t.committed();
586 if committed.saturating_add(units) > kind.cap() {
587 return Err(SpendRefusal::OverCap {
588 budget: kind,
589 period,
590 cap: kind.cap(),
591 committed,
592 requested: units,
593 });
594 }
595 let seq = s.next_seq;
596 let line = serde_json::to_vec(&Entry::Reserve {
597 seq,
598 period: period.clone(),
599 attempt: attempt_id.to_string(),
600 units,
601 at_ms: now,
602 })
603 .map_err(|e| SpendRefusal::Journal(e.to_string()))?;
604 append(&mut s.file, &line).map_err(|e| SpendRefusal::Journal(e.to_string()))?;
605 s.next_seq += 1;
606 s.res.insert(
607 seq,
608 Res {
609 attempt: attempt_id.to_string(),
610 period,
611 units,
612 state: ResState::Outstanding,
613 },
614 );
615 Ok(Reservation { seq, units })
616 }
617
618 fn write(&self, s: &mut State, e: &Entry) -> bool {
619 let ok = serde_json::to_vec(e)
620 .ok()
621 .is_some_and(|line| append(&mut s.file, &line).is_ok());
622 if !ok {
623 self.inner.write_failures.fetch_add(1, Ordering::SeqCst);
624 }
625 ok
626 }
627
628 pub fn settle(&self, r: Reservation, charge: Charge) {
631 let mut s = self.lock();
632 let Some(state) = s.res.get(&r.seq).map(|x| x.state) else {
633 return;
634 };
635 if state != ResState::Outstanding {
636 return;
637 }
638 if !self.write(&mut s, &Entry::Settle { seq: r.seq, charge }) {
639 return;
640 }
641 if let Some(x) = s.res.get_mut(&r.seq) {
642 x.state = settled(charge);
643 }
644 }
645
646 pub fn release(&self, r: Reservation) {
648 let mut s = self.lock();
649 if s.res.get(&r.seq).map(|x| x.state) != Some(ResState::Outstanding) {
650 return;
651 }
652 if !self.write(&mut s, &Entry::Release { seq: r.seq }) {
653 return;
654 }
655 if let Some(x) = s.res.get_mut(&r.seq) {
656 x.state = ResState::Released;
657 }
658 }
659
660 pub fn reconcile(&self, seq: u64, observed_units: u64) -> bool {
663 if self.budget() == BudgetKind::Production {
665 return false;
666 }
667 let mut s = self.lock();
668 if s.res
669 .get(&seq)
670 .map(|x| x.state)
671 .is_none_or(|s| !matches!(s, ResState::Liability | ResState::Unknown))
672 {
673 return false;
674 }
675 if !self.write(
676 &mut s,
677 &Entry::Reconcile {
678 seq,
679 units: observed_units,
680 },
681 ) {
682 return false;
683 }
684 if let Some(x) = s.res.get_mut(&seq) {
685 x.state = ResState::Observed(observed_units);
686 }
687 true
688 }
689
690 pub fn reconcile_billing(
695 &self,
696 journal_credential_reference: &str,
697 billing: &BillingSettlement,
698 ) -> Result<(), Unresolved> {
699 if self.budget() != BudgetKind::Production
700 || journal_credential_reference.trim().is_empty()
701 || journal_credential_reference != billing.credential_reference
702 || billing.provider_record.trim().is_empty()
703 {
704 return Err(Unresolved(
705 "billing record does not identify this production journal",
706 ));
707 }
708 let mut state = self.lock();
709 let reservation = state
710 .res
711 .get(&billing.reservation)
712 .ok_or(Unresolved("billing reservation is missing"))?;
713 if reservation.period != billing.period
714 || !matches!(reservation.state, ResState::Liability | ResState::Unknown)
715 {
716 return Err(Unresolved(
717 "billing month conflicts or reservation is not liability",
718 ));
719 }
720 let line = serde_json::to_vec(&Entry::Reconcile {
721 seq: billing.reservation,
722 units: billing.observed_units,
723 })
724 .map_err(|_| Unresolved("billing settlement could not be serialized"))?;
725 if append(&mut state.file, &line).is_err() {
726 self.inner.write_failures.fetch_add(1, Ordering::SeqCst);
727 return Err(Unresolved("billing settlement could not be recorded"));
728 }
729 state
730 .res
731 .get_mut(&billing.reservation)
732 .expect("checked reservation")
733 .state = ResState::Observed(billing.observed_units);
734 Ok(())
735 }
736
737 pub fn summary(&self) -> SpendSummary {
739 let kind = self.inner.kind;
740 let period = kind.period(self.inner.clock.now_ms());
741 totals(&self.lock().res, kind, &period)
742 }
743
744 pub fn run_spend(&self, run_id: &str) -> u64 {
749 run_total(&self.lock().res, run_id)
750 }
751
752 pub fn reservations_for(&self, decision_id: &str) -> Vec<String> {
754 let prefix = format!(
755 "{decision_id}/{}",
756 rustev_contract::live::attempt_prefix(decision_id)
757 );
758 self.lock()
759 .res
760 .values()
761 .filter(|r| r.attempt.starts_with(&prefix))
762 .map(|r| r.attempt[decision_id.len() + 1..].to_string())
763 .collect()
764 }
765
766 pub(crate) fn attempts_with_prefix(&self, prefix: &str) -> Vec<String> {
769 self.lock()
770 .res
771 .values()
772 .filter(|r| r.attempt.starts_with(prefix))
773 .map(|r| r.attempt.clone())
774 .collect()
775 }
776
777 pub fn reservations_detail(&self, decision_id: &str) -> Vec<crate::documents::CaseSpend> {
780 use crate::documents::{CaseSettlement, CaseSpend};
781 let prefix = format!(
782 "{decision_id}/{}",
783 rustev_contract::live::attempt_prefix(decision_id)
784 );
785 self.lock()
786 .res
787 .iter()
788 .filter(|(_, r)| r.attempt.starts_with(&prefix))
789 .map(|(seq, r)| CaseSpend::Reserved {
790 attempt_id: r.attempt[decision_id.len() + 1..].to_string(),
791 seq: *seq,
792 reserved_units: r.units,
793 settlement: match r.state {
794 ResState::Observed(units) => CaseSettlement::Observed { units },
795 ResState::Estimated(units) => CaseSettlement::Estimated { units },
796 ResState::Released => CaseSettlement::Released,
797 ResState::Unknown | ResState::Outstanding | ResState::Liability => {
798 CaseSettlement::Unknown
799 }
800 },
801 })
802 .collect()
803 }
804
805 pub fn run_split(&self, run_id: &str) -> RunSpend {
806 let prefix = format!("live/{run_id}/");
807 let mut split = RunSpend::default();
808 for r in self
809 .lock()
810 .res
811 .values()
812 .filter(|r| r.attempt.starts_with(&prefix))
813 {
814 let (total, units) = match r.state {
815 ResState::Observed(u) => (&mut split.observed, u),
816 ResState::Estimated(u) => (&mut split.estimated, u),
817 ResState::Unknown => (&mut split.unknown, r.units),
818 ResState::Outstanding | ResState::Liability => (&mut split.liability, r.units),
819 ResState::Released => continue,
820 };
821 *total = total.saturating_add(units);
822 }
823 split
824 }
825
826 pub fn testing_header_digest() -> rustev_contract::ids::ContentDigest {
828 header_digest(TESTING_CAP_UNITS)
829 }
830
831 pub fn superseded_testing_header_digest() -> rustev_contract::ids::ContentDigest {
834 header_digest(SUPERSEDED_TESTING_CAP_UNITS)
835 }
836
837 pub fn headroom(&self) -> u64 {
839 self.summary().headroom
840 }
841
842 pub fn runtime_cost_policy(&self) -> CostPolicy {
848 CostPolicy::Estimated {
849 max_units: self.headroom(),
850 }
851 }
852
853 pub fn shared_runtime_ledger(&self) -> Arc<Ledger> {
856 Arc::new(Ledger::new(self.runtime_cost_policy()))
857 }
858}
859
860fn header_digest(cap: u64) -> rustev_contract::ids::ContentDigest {
861 let header = Header {
862 schema: JOURNAL_SCHEMA.into(),
863 budget: BudgetKind::Testing,
864 cap,
865 };
866 let mut bytes = serde_json::to_vec(&header).expect("header serializes");
867 bytes.push(b'\n');
868 rustev_contract::ids::ContentDigest::parse(&rustev_contract::canonical::tagged_digest(
869 JOURNAL_SCHEMA,
870 &bytes,
871 ))
872 .expect("computed digest")
873}
874
875fn migrate_header(path: &Path, bytes: &[u8]) -> std::io::Result<File> {
881 let target = std::fs::canonicalize(path)?;
882 let mut tmp = target.as_os_str().to_owned();
883 tmp.push(".migrate");
884 let tmp = PathBuf::from(tmp);
885 let permissions = std::fs::metadata(&target)?.permissions();
886 {
887 let mut f = OpenOptions::new()
888 .write(true)
889 .create(true)
890 .truncate(true)
891 .open(&tmp)?;
892 f.set_permissions(permissions)?;
893 f.write_all(bytes)?;
894 f.sync_all()?;
895 }
896 std::fs::rename(&tmp, &target)?;
897 if let Some(dir) = target.parent()
898 && let Ok(d) = File::open(dir)
899 {
900 d.sync_all()?;
901 }
902 let mut file = OpenOptions::new().read(true).append(true).open(&target)?;
903 file.seek(SeekFrom::End(0))?;
904 Ok(file)
905}
906
907fn run_total(res: &BTreeMap<u64, Res>, run_id: &str) -> u64 {
908 let prefix = format!("live/{run_id}/");
909 res.values()
910 .filter(|r| r.attempt.starts_with(&prefix))
911 .fold(0_u64, |total, r| {
912 let units = match r.state {
913 ResState::Observed(u) | ResState::Estimated(u) => u,
914 ResState::Outstanding | ResState::Liability | ResState::Unknown => r.units,
915 ResState::Released => 0,
916 };
917 total.saturating_add(units)
918 })
919}
920
921fn settled(charge: Charge) -> ResState {
922 match charge {
923 Charge::Observed { units } => ResState::Observed(units),
924 Charge::Estimated { units } => ResState::Estimated(units),
925 Charge::Unknown => ResState::Unknown,
926 }
927}
928
929fn totals(res: &BTreeMap<u64, Res>, kind: BudgetKind, period: &str) -> SpendSummary {
930 let mut t = SpendSummary {
931 budget: kind,
932 period: period.to_string(),
933 cap: kind.cap(),
934 observed: 0,
935 estimated: 0,
936 liability: 0,
937 outstanding: 0,
938 headroom: 0,
939 };
940 for r in res.values().filter(|r| r.period == period) {
941 match r.state {
942 ResState::Outstanding => t.outstanding = t.outstanding.saturating_add(r.units),
943 ResState::Observed(u) => t.observed = t.observed.saturating_add(u),
944 ResState::Estimated(u) => t.estimated = t.estimated.saturating_add(u),
945 ResState::Liability | ResState::Unknown => {
946 t.liability = t.liability.saturating_add(r.units)
947 }
948 ResState::Released => {}
949 }
950 }
951 t.headroom = t.cap.saturating_sub(t.committed());
952 t
953}
954
955#[cfg(test)]
956mod tests {
957 use super::*;
958
959 #[test]
960 fn failed_settlement_and_release_keep_the_reservation() {
961 let dir = std::env::temp_dir().join(format!(
962 "rustev-journal-write-failure-{}",
963 std::process::id()
964 ));
965 std::fs::create_dir(&dir).unwrap();
966 let path = dir.join("journal");
967 let journal =
968 SpendJournal::open(&path, BudgetKind::Testing, Arc::new(SetClock::new(0))).unwrap();
969 let ticket = journal.reserve("live/run/attempt", 10).unwrap();
970 journal.lock().file = File::open(&path).unwrap();
972 journal.settle(ticket, Charge::Observed { units: 1 });
973 assert_eq!(journal.run_spend("run"), 10);
974 journal.release(ticket);
975 assert_eq!(journal.run_spend("run"), 10);
976 assert_eq!(journal.write_failures(), 2);
977 drop(journal);
978 let reopened =
979 SpendJournal::open(&path, BudgetKind::Testing, Arc::new(SetClock::new(0))).unwrap();
980 assert_eq!(reopened.summary().liability, 10);
981 drop(reopened);
982 std::fs::remove_dir_all(dir).unwrap();
983 }
984
985 #[test]
986 fn utc_months() {
987 assert_eq!(utc_month(0), "1970-01");
988 assert_eq!(utc_month(1_790_812_799_999), "2026-09");
990 assert_eq!(utc_month(1_790_812_800_000), "2026-10");
991 assert_eq!(utc_month(951_782_400_000), "2000-02"); assert_eq!(utc_month(-1), "1969-12");
993 }
994}
995
996#[derive(Debug, Clone, PartialEq, Eq)]
1000pub struct Unresolved(pub &'static str);
1001impl std::fmt::Display for Unresolved {
1002 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1003 write!(f, "Unresolved: {}", self.0)
1004 }
1005}
1006impl std::error::Error for Unresolved {}
1007
1008#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
1012#[serde(deny_unknown_fields)]
1013pub struct BillingSettlement {
1014 pub credential_reference: String,
1015 pub provider_record: String,
1016 pub period: String,
1017 pub reservation: u64,
1018 pub observed_units: u64,
1019}