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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
//! 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::flush::{BufferedTransaction, TransactionBuffer};
use super::ledger::Ledger;
use super::wal::{Wal, WalStatus};
/// Replay uncommitted WAL entries on startup.
///
/// `tx_buffer` and `node_name` are used to re-deliver `Earned` entries (peer
/// earns that were credited in-memory but never confirmed flushed to the
/// dashboard) via the normal flush path — see the `WalStatus::Earned` arm
/// below.
pub fn replay_wal(
wal: &Wal,
ledger: &Ledger,
tx_buffer: &TransactionBuffer,
node_name: Option<&str>,
) {
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,
);
}
WalStatus::Earned => {
// Worker earn was credited in-memory but never confirmed
// flushed to the dashboard (crash before/while flushing).
//
// We must NOT re-apply the in-memory credit here. The dashboard
// is authoritative for spendable balance: `authoritative_balances`
// re-seeds lazily from the dashboard (via load_balance_if_needed)
// the next time this user is touched, and on the crash-after-
// dashboard-write window the dashboard ALREADY reflects this
// earn. Calling local_add_credits would seed-from-dashboard and
// then add `amount` on top, inflating the in-memory spendable
// balance by `amount` until a clean restart (bounded over-spend).
//
// So replay ONLY re-delivers the earn to the dashboard to make it
// durable there: re-buffer the same `earn-{request_id}` credit tx
// (idempotent via dashboard request_id dedup) and leave the WAL
// entry `Earned`, so flush_to_api marks it Committed once the
// dashboard confirms it applied/duplicate. balance_after is only
// metadata on the tx row; use the ledger's current view without
// mutating it.
let amount = entry.actual_cost.unwrap_or(entry.estimated_cost);
println!(
" [WAL] Re-delivering unflushed earn {} for user {} (amount: {:.6})",
entry.request_id, entry.user_id, amount
);
let balance_after = ledger.get_balance(&entry.user_id);
tx_buffer.push_transaction(BufferedTransaction {
request_id: entry.request_id.clone(),
user_id: entry.user_id.clone(),
tx_type: "credit".to_string(),
amount,
balance_after,
worker_id: entry.worker_id.clone(),
duration_ms: entry.duration_ms.unwrap_or(0.0),
source_node: node_name.map(|s| s.to_string()),
worker_name: Some(entry.worker_id.clone()),
worker_uri: None,
price_per_hour: 0.0,
});
}
// 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");
}