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
182
183
184
185
186
187
188
189
190
191
192
193
//! The observer-recovery row, its stored codec and its fold (split from
//! [`super`] by a pure move under the 500-line lens, DUR-027 R1).
use liminal_protocol::lifecycle::{
ObserverProgressAdvanceDecision, ObserverProgressRetireDecision, ObserverProgressTrackDecision,
ObserverRecoveryAggregate,
};
use liminal_protocol::wire::{ConversationId, DeliverySeq, ObserverEpoch};
use serde::{Deserialize, Serialize};
use super::super::log::OperationLogError;
use super::OBSERVER_SNAPSHOT_EVERY;
/// Stored-row schema version of the operation rows.
const SCHEMA_VERSION: u8 = 1;
/// Stored-row schema version of a snapshot row.
///
/// Downgrade across a snapshot row is unsupported. A binary from before
/// snapshots parses the whole row before it checks the version, so it fails
/// on the unknown `snapshot` variant (a JSON error), not by version name. The
/// supported path is upgrade: a stream with no snapshot row is folded whole
/// once, and a snapshot is then taken (see [`ObserverLog::restore`]).
const SNAPSHOT_SCHEMA_VERSION: u8 = 2;
/// Stored-row schema version of a retire row (DUR-027 R3).
///
/// As with the snapshot row, downgrade across it is unsupported: a binary
/// from before retirement fails on the unknown `retire` variant (a JSON error)
/// before it reads the version, so it never folds the stream as if the
/// conversation were still tracked.
const RETIRE_SCHEMA_VERSION: u8 = 3;
/// The schema version each row kind is stored at, and the only one it is read at.
const fn version_of(row: &ObserverRow) -> u8 {
match row {
ObserverRow::Snapshot { .. } => SNAPSHOT_SCHEMA_VERSION,
ObserverRow::Retire { .. } => RETIRE_SCHEMA_VERSION,
ObserverRow::Track { .. } | ObserverRow::Advance { .. } | ObserverRow::Arms { .. } => {
SCHEMA_VERSION
}
}
}
/// One durable observer-recovery row.
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "snake_case", tag = "row")]
pub(in super::super) enum ObserverRow {
/// A conversation's observer progress became tracked.
Track {
/// Newly tracked conversation.
conversation_id: ConversationId,
/// Initial durable observer progress.
observer_progress: DeliverySeq,
},
/// A tracked conversation's observer progress advanced.
Advance {
/// Advancing conversation.
conversation_id: ConversationId,
/// New strictly greater progress.
observer_progress: DeliverySeq,
},
/// One complete equal-epoch arm plan was installed atomically.
Arms {
/// Every arm of the plan, in conversation order.
arms: Vec<(ConversationId, ObserverEpoch)>,
},
/// A closed conversation left the aggregate (DUR-027): its progress row
/// is removed, and nothing for it is folded after this row.
Retire {
/// Retired conversation.
conversation_id: ConversationId,
},
/// The whole aggregate as folded through the row before this one.
Snapshot {
/// Every progress row, in conversation order.
progress: Vec<(ConversationId, DeliverySeq)>,
/// Every installed arm, in conversation order.
armed: Vec<(ConversationId, ObserverEpoch)>,
},
}
impl ObserverRow {
/// The snapshot row of `aggregate`.
pub(in super::super) fn snapshot_of(aggregate: &ObserverRecoveryAggregate) -> Self {
Self::Snapshot {
progress: aggregate.progress_rows(),
armed: aggregate.armed_rows(),
}
}
}
/// Whether a snapshot row is owed at `head`: the snapshot then takes `head`
/// and the row being appended takes the sequence after it.
pub(in super::super) const fn snapshot_due(head: u64) -> bool {
head % (OBSERVER_SNAPSHOT_EVERY + 1) == OBSERVER_SNAPSHOT_EVERY
}
/// The snapshot row owed before a row appended at `head`, taken from the
/// aggregate the row applies to; `None` when none is due. Callers take it
/// before the aggregate moves into its transaction.
pub(in super::super) fn snapshot_before(
head: u64,
aggregate: &ObserverRecoveryAggregate,
) -> Option<ObserverRow> {
snapshot_due(head).then(|| ObserverRow::snapshot_of(aggregate))
}
/// The head after appending `rows` operation rows at `head`, counting the
/// snapshot rows the appender writes on the way.
pub(in super::super) fn head_after(head: u64, rows: u64) -> Option<u64> {
let mut next = head;
for _ in 0..rows {
next = next.checked_add(if snapshot_due(next) { 2 } else { 1 })?;
}
Some(next)
}
#[derive(Clone, Debug, Deserialize, Serialize)]
struct StoredRow {
schema_version: u8,
row: ObserverRow,
}
/// Encodes one row at its schema version.
pub(super) fn encode(row: &ObserverRow) -> Result<Vec<u8>, OperationLogError> {
Ok(serde_json::to_vec(&StoredRow {
schema_version: version_of(row),
row: row.clone(),
})?)
}
/// Encodes one row exactly as the appender stores it, for fixtures that write
/// a stream the way an older binary did (operation rows only, schema 1).
#[cfg(test)]
pub(in super::super) fn encode_for_tests(row: &ObserverRow) -> Result<Vec<u8>, OperationLogError> {
encode(row)
}
/// Decodes one row, refusing a schema version its kind does not carry.
pub(super) fn decode(payload: &[u8]) -> Result<ObserverRow, OperationLogError> {
let stored: StoredRow = serde_json::from_slice(payload)?;
if stored.schema_version != version_of(&stored.row) {
return Err(OperationLogError::SchemaVersion(stored.schema_version));
}
Ok(stored.row)
}
/// Applies one durable row through the aggregate's own barriered mutations.
pub(super) fn fold_row(
aggregate: ObserverRecoveryAggregate,
row: ObserverRow,
sequence: u64,
) -> Result<ObserverRecoveryAggregate, OperationLogError> {
match row {
ObserverRow::Track {
conversation_id,
observer_progress,
} => match aggregate.decide_track(conversation_id, observer_progress) {
ObserverProgressTrackDecision::Commit(transaction) => Ok(transaction.commit()),
ObserverProgressTrackDecision::Refuse { .. } => Err(refused(sequence)),
},
ObserverRow::Advance {
conversation_id,
observer_progress,
} => match aggregate.decide_progress_advance(conversation_id, observer_progress) {
ObserverProgressAdvanceDecision::Commit(transaction) => {
let (aggregate, _fired) = transaction.commit();
Ok(aggregate)
}
ObserverProgressAdvanceDecision::Refuse { .. } => Err(refused(sequence)),
},
ObserverRow::Retire { conversation_id } => match aggregate.decide_retire(conversation_id) {
ObserverProgressRetireDecision::Commit(transaction) => Ok(transaction.commit()),
ObserverProgressRetireDecision::Refuse { .. } => Err(refused(sequence)),
},
ObserverRow::Arms { arms } => aggregate
.install_restored_arms(&arms)
.map_err(|_| refused(sequence)),
ObserverRow::Snapshot { progress, armed } => {
// A snapshot restates the fold so far; one that disagrees is a
// corrupt stream.
if aggregate.progress_rows() == progress && aggregate.armed_rows() == armed {
Ok(aggregate)
} else {
Err(refused(sequence))
}
}
}
}
/// A durable row the aggregate itself refuses is a corrupt stream.
pub(super) const fn refused(sequence: u64) -> OperationLogError {
OperationLogError::CorruptRow { sequence }
}