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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
//! §Fase 74.c — the durable event outbox.
//!
//! §74.a wired `emit` → a daemon `listen` through the §13 `TypedEventBus`;
//! §74.b made delivery at-least-once WHILE THE PROCESS LIVES (retry over
//! the ephemeral mpsc queue). §74.c adds the DURABLE substrate: an `emit`
//! to a `persistent_axonstore` channel APPENDS to an outbox — an
//! append-only log keyed by channel, with a per-channel processed cursor.
//! The receive driver DRAINS the unprocessed tail, delivers each event
//! (at-least-once, §74.b), and MARKS it processed. An event stays
//! redeliverable until acked + marked — so a consumer that was DOWN when
//! the event was emitted picks it up when it returns. This is the Q3
//! guarantee over a persisted log rather than an ephemeral queue.
//!
//! **Scope (honest).** This OSS module is the ABSTRACTION + an in-memory
//! reference. The two stronger guarantees layer on top:
//!
//! - **Crash durability** — the in-memory reference survives the *consumer*
//! being down, but not a *process restart*. The per-tenant **Postgres**
//! outbox that survives a crash is §74.f (the [`EventOutbox`] trait is
//! the seam the enterprise `PgEventOutbox` implements).
//! - **Transactional atomicity** — the outbox row committed in the SAME
//! store transaction as the producing flow's `mutate`/`persist`, so the
//! event is written IFF the state change committed. The §35/§37 store
//! ops are per-op transactions today (no flow-level tx), so this is a
//! documented refinement: it closes the small lost-/phantom-event window
//! between the two separate commits. Until it lands, `emit` appends to
//! the outbox as its own durable op — durable, but with that bounded
//! atomicity gap. We do not call this layer "transactional" until the
//! flow-level transaction is real (`delivery_is_a_kept_promise`).
use std::collections::HashMap;
use std::sync::Mutex;
/// One event in the outbox: its monotonic per-channel offset + payload.
#[derive(Debug, Clone, PartialEq)]
pub struct OutboxEntry {
pub offset: u64,
pub channel: String,
pub payload: serde_json::Value,
}
/// The durable event outbox seam. `append` records an emitted event;
/// `unprocessed` returns the not-yet-acked tail for a channel (in offset
/// order — the redelivery set); `mark_processed` acks an offset so it is
/// never redelivered. The enterprise `PgEventOutbox` (§74.f) implements
/// this over a per-tenant Postgres table; the in-memory reference below is
/// the OSS default + the test double.
pub trait EventOutbox: Send + Sync {
/// Append an event to `channel`'s log; returns its monotonic offset.
fn append(&self, channel: &str, payload: serde_json::Value) -> u64;
/// The unprocessed (not-yet-acked) events for `channel`, offset-ordered.
fn unprocessed(&self, channel: &str) -> Vec<OutboxEntry>;
/// Ack `offset` on `channel` — it will not be redelivered.
fn mark_processed(&self, channel: &str, offset: u64);
}
#[derive(Default)]
struct ChannelLog {
/// (offset, payload) in append order.
entries: Vec<(u64, serde_json::Value)>,
/// Acked offsets.
processed: std::collections::HashSet<u64>,
/// Next offset to assign (monotonic per channel).
next_offset: u64,
}
/// In-memory reference [`EventOutbox`]. Durable WITHIN the process (an
/// event survives the consumer being down → it stays unprocessed +
/// redelivers); NOT across a process restart (that is the §74.f Postgres
/// outbox). Thread-safe (one `Mutex` over the per-channel logs).
#[derive(Default)]
pub struct InMemoryEventOutbox {
channels: Mutex<HashMap<String, ChannelLog>>,
}
impl InMemoryEventOutbox {
pub fn new() -> Self {
Self::default()
}
/// Total unprocessed count across all channels (introspection / tests).
pub fn pending_total(&self) -> usize {
self.channels
.lock()
.unwrap()
.values()
.map(|log| log.entries.iter().filter(|(o, _)| !log.processed.contains(o)).count())
.sum()
}
}
impl EventOutbox for InMemoryEventOutbox {
fn append(&self, channel: &str, payload: serde_json::Value) -> u64 {
let mut chans = self.channels.lock().unwrap();
let log = chans.entry(channel.to_string()).or_default();
let offset = log.next_offset;
log.next_offset += 1;
log.entries.push((offset, payload));
offset
}
fn unprocessed(&self, channel: &str) -> Vec<OutboxEntry> {
let chans = self.channels.lock().unwrap();
match chans.get(channel) {
None => Vec::new(),
Some(log) => log
.entries
.iter()
.filter(|(o, _)| !log.processed.contains(o))
.map(|(o, p)| OutboxEntry {
offset: *o,
channel: channel.to_string(),
payload: p.clone(),
})
.collect(),
}
}
fn mark_processed(&self, channel: &str, offset: u64) {
let mut chans = self.channels.lock().unwrap();
if let Some(log) = chans.get_mut(channel) {
log.processed.insert(offset);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn append_assigns_monotonic_offsets_per_channel() {
let ob = InMemoryEventOutbox::new();
assert_eq!(ob.append("A", json!(1)), 0);
assert_eq!(ob.append("A", json!(2)), 1);
// Offsets are per-channel — B starts at 0.
assert_eq!(ob.append("B", json!(9)), 0);
}
#[test]
fn unprocessed_returns_the_redelivery_tail_in_order() {
let ob = InMemoryEventOutbox::new();
ob.append("A", json!("x"));
ob.append("A", json!("y"));
let tail = ob.unprocessed("A");
assert_eq!(tail.len(), 2);
assert_eq!(tail[0].offset, 0);
assert_eq!(tail[0].payload, json!("x"));
assert_eq!(tail[1].offset, 1);
}
#[test]
fn mark_processed_removes_from_the_redelivery_tail() {
// The headline: an event stays redeliverable until acked. A consumer
// that was down sees it on `unprocessed`; once it acks, it is gone.
let ob = InMemoryEventOutbox::new();
let off = ob.append("A", json!("hib"));
assert_eq!(ob.unprocessed("A").len(), 1, "redeliverable until acked");
ob.mark_processed("A", off);
assert!(ob.unprocessed("A").is_empty(), "acked → not redelivered");
assert_eq!(ob.pending_total(), 0);
}
#[test]
fn unknown_channel_has_no_unprocessed_events() {
let ob = InMemoryEventOutbox::new();
assert!(ob.unprocessed("never").is_empty());
}
#[test]
fn pending_total_counts_only_unacked() {
let ob = InMemoryEventOutbox::new();
ob.append("A", json!(1));
let o2 = ob.append("A", json!(2));
ob.append("B", json!(3));
assert_eq!(ob.pending_total(), 3);
ob.mark_processed("A", o2);
assert_eq!(ob.pending_total(), 2);
}
}