1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
//! WAL replay on broker startup.
//!
//! Reads uncommitted WAL entries and:
//! - Reserved entries → cancel (refund credits in-memory, queue for API flush)
//! - Executed entries → commit (refund diff in-memory) + publish transaction via API
use super::ledger::Ledger;
use super::wal::{Wal, WalStatus};
/// Replay uncommitted WAL entries on startup
pub fn replay_wal(wal: &Wal, ledger: &Ledger) {
let uncommitted = match wal.read_uncommitted() {
Ok(entries) => entries,
Err(e) => {
eprintln!(" [WAL] Failed to read uncommitted entries: {}", e);
return;
}
};
if uncommitted.is_empty() {
return;
}
println!(
" [WAL] Replaying {} uncommitted entries...",
uncommitted.len()
);
for entry in &uncommitted {
match entry.status {
WalStatus::Reserved => {
// Worker never returned — refund the reserved amount in-memory
println!(
" [WAL] Cancelling stale reservation for user {} (req: {}, cost: {:.6})",
entry.user_id, entry.request_id, entry.estimated_cost
);
if let Err(e) = ledger.cancel_from_wal(&entry.user_id, entry.estimated_cost) {
eprintln!(" [WAL] Failed to cancel for {}: {}", entry.request_id, e);
}
let _ = wal.update_status(
&entry.request_id,
WalStatus::Failed,
None,
None,
);
}
WalStatus::Executed => {
// Worker completed but commit didn't happen — commit now
let actual_cost = entry.actual_cost.unwrap_or(entry.estimated_cost);
println!(
" [WAL] Recovering executed request {} for user {} (cost: {:.6})",
entry.request_id, entry.user_id, actual_cost
);
// Publish transaction via dashboard API
ledger.publish_transaction(
&entry.request_id,
&entry.user_id,
"commit",
actual_cost,
0.0,
&entry.worker_id,
entry.duration_ms.unwrap_or(0.0),
None,
);
// Refund difference (reserved - actual) in-memory
match ledger.commit_from_wal(&entry.user_id, entry.estimated_cost, actual_cost) {
Ok(balance) => {
println!(
" [WAL] Recovered: user {} balance now {:.4}",
entry.user_id, balance
);
}
Err(e) => {
eprintln!(
" [WAL] Failed to commit for {}: {}",
entry.request_id, e
);
}
}
let _ = wal.update_status(
&entry.request_id,
WalStatus::Committed,
entry.actual_cost,
entry.duration_ms,
);
}
// Committed/Failed are filtered out by read_uncommitted
_ => {}
}
}
// Compact after replay
if let Err(e) = wal.compact() {
eprintln!(" [WAL] Failed to compact after replay: {}", e);
}
println!(" [WAL] Replay complete");
}