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
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
//! Boot restore, owed checkpoints and unloadable conversations of the production participant handler, moved out of `handler.rs` unchanged
//! so that file stays within the 500-line bound.
use super::{
Arc, ConversationAuthority, ConversationId, LogAppender, OperationBatch, OperationLog,
OperationLogError, OutboxLog, ParticipantSemanticError, ProductionParticipantHandler, RefCell,
UnloadableConversation, UnloadableConversationRecord, block_on, bridge_error,
checkpoint_conversation_route, log_error, restore_error, state_error,
};
impl ProductionParticipantHandler {
/// Restores exactly one conversation, and fails as exactly one
/// conversation.
///
/// Every propagation point in here belongs to a single conversation, so
/// its `Err` is that conversation's answer and nobody else's. The caller
/// is what makes that true — this function is deliberately allowed to be
/// fallible, and deliberately not allowed to be fatal.
///
/// Answers the number of operation-log rows the restored owner holds.
/// Restores one conversation and answers its log rows and whether it
/// still owes the checkpoint its rebuild could not publish (DUR-020 R6):
/// the restore unit publishes it once the incarnation's repair has run.
pub(in super::super) fn restore_one_conversation(
&self,
conversation_id: ConversationId,
) -> Result<(u64, bool), ParticipantSemanticError> {
let mut rows = 0_u64;
let mut still_owed = false;
{
let cell = self.cell(conversation_id)?;
let mut owner = cell
.lock()
.map_err(|_| ParticipantSemanticError::Internal {
message: format!(
"participant conversation {conversation_id} owner lock is poisoned"
),
})?;
if owner.is_none() {
let log = OperationLog::new(Arc::clone(&self.store), conversation_id);
let outbox_log = OutboxLog::new(Arc::clone(&self.store), conversation_id);
let (mut replayed, publication) = block_on(
super::super::restore_checkpoint_rebuild::restore_at_boot_reporting(
self.store.as_ref(),
&log,
&outbox_log,
&self.config,
self.outbox_limits,
),
)
.map_err(|error| bridge_error(&error))?
.map_err(|error| restore_error(&error))?;
self.repair_loaded_fates(conversation_id, &mut replayed, &log, &outbox_log)?;
let witnesses = replayed.observer_progress_witnesses();
if !replayed.tokens.is_empty() || replayed.is_closed() {
self.reconcile_observer_progress(
conversation_id,
witnesses,
replayed.observer_progress_pruned_through(),
replayed.observer_progress,
)?;
replayed
.prune_durable_observer_witnesses()
.map_err(|error| state_error(&error))?;
} else if !witnesses.is_empty() {
return Err(ParticipantSemanticError::Internal {
message: format!(
"unenrolled conversation {conversation_id} projected observer progress"
),
});
}
let now = self
.now_ms()
.map_err(|error| ParticipantSemanticError::Internal {
message: format!("participant clock read failed: {error}"),
})?;
replayed.prune_expired_provenance(u128::from(now));
let contribution = replayed
.capacity_contribution(u128::from(now))
.map_err(|error| state_error(&error))?;
self.capacity
.fold_conversation(conversation_id, contribution)
.map_err(|error| state_error(&error))?;
let durably_empty = replayed.next_log_sequence == 0;
if durably_empty {
drop(owner);
self.evict_uncommitted(conversation_id, &cell)?;
return Ok((0, false));
}
// R-BOOT-DRAIN (F8B §6.2): empty this conversation's
// immutable-candidate lane before any retained connection-fate
// `Open` replays, and before the owner is installed. The drain
// lives HERE and not in `replay_and_repair`, which is also the
// load a request runs for a conversation with no live owner.
let before_drain = replayed.next_log_sequence;
let verdict =
self.drain_restored_candidate_lane(conversation_id, &mut replayed, &log);
let drained = verdict.drained_any();
verdict.observe()?;
// DUR-020 R6: a rebuild whose owner could not be captured
// published nothing, and left every later load a full rebuild.
// Once the drain has finished the fate that refused the
// capture, this unit publishes the checkpoint it owes.
let unpublished = publication
== super::super::restore_checkpoint_rebuild::BootCheckpoint::Unpublished;
// Owed only when the whole checkpoint can be framed: its
// positions as well as its lifecycle. The owed crossing is
// forced, and a forced crossing refused at its positions
// fails the restore unit instead of serving on (11 October
// 2026: a drained Detached terminal moves observer progress
// past the outbox's highest delivery).
let owe_checkpoint = unpublished
&& replayed.checkpoint_lifecycle_capturable().is_ok()
&& replayed.checkpoint_positions_framable().is_ok();
still_owed = unpublished && !owe_checkpoint;
if drained || owe_checkpoint {
self.commit_boot_drain(
conversation_id,
&mut replayed,
&log,
before_drain,
owe_checkpoint,
)?;
}
// DUR-027 R5/R6: a drain that sealed the conversation, or a
// conversation found closed and unarmed on its first restore,
// is retired here, once; boot never visits it again.
if self.retire_if_closed(conversation_id, &replayed) {
drop(owner);
self.evict_uncommitted(conversation_id, &cell)?;
return Ok((0, false));
}
rows = replayed.next_log_sequence;
*owner = Some(replayed);
}
drop(owner);
}
Ok((rows, still_owed))
}
/// DUR-020 R6, after the restore unit's repair: a rebuild that could not
/// be captured before the repair publishes its checkpoint now if the
/// repair finished what refused the capture. Still uncapturable, it
/// publishes nothing here and the next crossing publishes, said by name.
/// Answers whether the checkpoint was published (or handed to the
/// publisher); `false` when the owner is absent or still uncapturable.
pub(in super::super) fn publish_owed_checkpoint(
&self,
conversation_id: ConversationId,
) -> Result<bool, ParticipantSemanticError> {
let cell = self.cell(conversation_id)?;
let mut owner = cell
.lock()
.map_err(|_| ParticipantSemanticError::Internal {
message: format!(
"participant conversation {conversation_id} owner lock is poisoned"
),
})?;
let Some(authority) = owner.as_mut() else {
return Ok(false);
};
if let Err(refusal) = authority
.checkpoint_lifecycle_capturable()
.and_then(|()| authority.checkpoint_positions_framable())
{
tracing::info!(
step = "participant restore",
conversation_id,
refusal = %refusal,
"rebuilt conversation still uncapturable; it publishes after its repair or at the next crossing"
);
return Ok(false);
}
let log = OperationLog::new(Arc::clone(&self.store), conversation_id);
let before = authority.next_log_sequence;
self.commit_boot_drain(conversation_id, authority, &log, before, true)?;
drop(owner);
Ok(true)
}
/// The commit tail a boot drain owes once its heads are committed: the
/// observer reconciliation and capacity re-fold every live commit runs,
/// then the checkpoint the crossing owes.
pub(in super::super) fn commit_boot_drain(
&self,
conversation_id: ConversationId,
replayed: &mut ConversationAuthority,
log: &OperationLog,
before_drain: u64,
owe_checkpoint: bool,
) -> Result<(), ParticipantSemanticError> {
// The drain appended durable rows and completed each one's
// extension row in place, exactly as a live commit does, so
// the restored owner is its log's current state. What the
// drain still owes is the commit tail every live commit
// runs: the observer reconciliation for the projections it
// recorded and the capacity re-fold for the binding slot it
// released.
let outbox_log = OutboxLog::new(Arc::clone(&self.store), conversation_id);
let batch = RefCell::new(Some(OperationBatch::new(&checkpoint_conversation_route(
conversation_id,
))));
#[cfg(test)]
let flat_legacy = super::super::log::legacy_rows::armed();
#[cfg(not(test))]
let flat_legacy = false;
let appender = LogAppender::new(
log,
&self.registry,
&outbox_log,
Some(super::super::prune::effective_keep(self.config.pruning)),
(!flat_legacy).then_some(&batch),
0,
&self.publisher,
);
self.complete_live_commit(conversation_id, replayed, &appender)?;
// The commit tail every live commit runs. The drain's own
// rows were committed head by head, so nothing staged is
// not "nothing owed": when the drain crossed a period or
// sealed the conversation, the checkpoint that crossing
// owes (`checkpoint_due`, the closed marker included) is
// captured, committed and installed exactly as a live
// seal's is through `finish_operation`.
let due = owe_checkpoint
|| OperationBatch::checkpoint_due(
before_drain,
replayed.next_log_sequence,
replayed.is_closed(),
)
.map_err(|error| log_error(&OperationLogError::from(error)))?;
if due {
appender
.commit_with_checkpoint(replayed, before_drain, &self.config, owe_checkpoint)
.map_err(|error| state_error(&error))?;
} else if appender.has_staged().map_err(|error| log_error(&error))? {
appender
.commit_staged()
.map_err(|error| log_error(&error))?;
}
Ok(())
}
/// Records one conversation as unloadable and renders the NAMED refusal
/// every later request on it is answered with.
///
/// This is the attribution half of containment, and it is deliberately the
/// same call on both sides: the boot loop records and continues, the
/// request path records and refuses, and both produce a refusal carrying
/// the conversation id. An operator therefore never has to correlate a
/// bare decode failure against a log to learn which conversation stopped.
pub(in super::super) fn record_unloadable(
&self,
conversation_id: ConversationId,
error: &ParticipantSemanticError,
) -> ParticipantSemanticError {
// A retired conversation is sealed, not unloadable: nothing is
// recorded and the sealed refusal is answered as it is (DUR-027 R4).
if let ParticipantSemanticError::ConversationSealed { .. } = error {
return error.clone();
}
let reason = error.to_string();
let class = error.class();
// The operator's first tell, and it carries the class as its own field:
// the refusal text of a failed load ("expected value at line 1 column
// 1") names no class at all, so a reader with only the message has to
// discriminate on a substring or not at all.
tracing::error!(
conversation_id,
class,
reason = %reason,
"CONTAINMENT: participant conversation is unloadable and is refused on its own; every \
other conversation keeps being served"
);
if !self.unloadable.record(UnloadableConversation {
conversation_id,
class,
reason: reason.clone(),
}) {
// Never a silent drop: a refusal that is reported but not retained
// is a different, weaker state than one that was retained, and an
// operator reading `GET /unloadable-conversations` must not be told
// a shorter story than the log tells.
tracing::error!(
conversation_id,
"the unloadable-conversation record is poisoned; this refusal is reported but not \
retained"
);
}
ParticipantSemanticError::ConversationUnloadable {
conversation_id,
reason,
}
}
/// Retires an unloadable record once the conversation has actually loaded,
/// so the map never keeps reporting a conversation that recovered.
pub(super) fn clear_unloadable(&self, conversation_id: ConversationId) {
if !self.unloadable.retire(conversation_id) {
tracing::error!(
conversation_id,
"the unloadable-conversation record is poisoned; a conversation that recovered may \
still be reported as unloadable"
);
}
}
/// The shared refused-load record, for publication onto the operator read
/// surface.
///
/// The clone shares the record rather than copying it, so a refusal
/// recorded after publication is still the answer the surface gives. This
/// is the only way out of the handler for the containment record, and the
/// server's startup path (`server/runtime.rs`) is its one production
/// caller.
pub(crate) fn unloadable_record(&self) -> UnloadableConversationRecord {
self.unloadable.clone()
}
}