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
//! Lossless live connection-fate impact orchestration.
use liminal_protocol::lifecycle::BindingTerminalAdmitError;
use crate::server::participant::dispatch_impact::{DispatchImpact, DispatchImpactAccumulator};
use crate::server::participant::{
ConnectionFateWorkItem, ParticipantConnectionFateOutcome, ParticipantSemanticError,
};
use super::handler::ProductionParticipantHandler;
use super::state::StateError;
impl ProductionParticipantHandler {
pub(super) fn apply_connection_fate_with_impacts(
&self,
work_item: &ConnectionFateWorkItem,
) -> ParticipantConnectionFateOutcome {
let mut impacts = Vec::new();
if let Err(error) = self.ensure_service_live() {
return ParticipantConnectionFateOutcome::new(Err(error), impacts);
}
for conversation_id in work_item.tracked_conversations.iter().copied() {
// DUR-024 (11 Oct 2026): a conversation not yet restored is not
// restored here. Its fate is held and applied first in its own
// restore unit, after the listener binds, so boot recovery never
// replays a conversation before the door answers.
match self
.pending
.defer_fate_if_pending(conversation_id, work_item)
{
Ok(true) => {
tracing::info!(
open_sequence = work_item.open_sequence,
conversation_id,
"connection-fate step held for a conversation not yet restored; it applies in that conversation's restore"
);
continue;
}
Ok(false) => {}
Err(error) => return ParticipantConnectionFateOutcome::new(Err(error), impacts),
}
let mut impact = DispatchImpactAccumulator::new();
let result = self.with_conversation_fate_source(
conversation_id,
Some(&mut impact),
|authority, appender, impact| {
let impact = impact.ok_or_else(|| {
StateError::invariant("live connection fate lost its impact owner")
})?;
let transaction = authority.prepare_connection_fate_transaction(work_item);
transaction.complete_with_impact(authority, appender, impact)
},
);
if let Some(changed) = changed_impact(impact.finish(conversation_id)) {
impacts.push(changed);
}
if let Err(error) = result {
if let ParticipantSemanticError::ConversationUnloadable {
conversation_id: contained,
reason,
} = &error
{
// DUR-024 R3: a contained conversation does not fail the
// boot. Its step of this Open is recorded refused, by
// name, and the Open finishes for every other
// conversation it tracks; the process-wide fatal is for a
// fate the node could not apply, not for a conversation
// the node already refuses on its own. The slot this
// fate would have departed is repaired by the
// unclean-restart repair the conversation runs inside its
// own restore once it loads again.
tracing::warn!(
open_sequence = work_item.open_sequence,
conversation_id = contained,
refusal = %reason,
"connection-fate step refused for a contained conversation; the Open \
finishes for every other conversation it tracks"
);
continue;
}
if matches!(
error,
ParticipantSemanticError::BindingTerminalAdmissionRefused {
error: BindingTerminalAdmitError::Precedence,
}
) {
// The conversation's immutable-candidate lane is occupied.
// The already-durable Open is the retry token, so the whole
// work item is replayed later — by boot recovery at
// connection/incarnation.rs:88, and once the lane drains at
// its sole production emptier, ops_frontier.rs:142 on the
// next successful publish. Latching the process-wide fatal
// here would refuse that publish too, so the lane could
// never drain and the replay could never converge. The
// conversations already committed above are a no-op on
// replay: their slots left Bound, so the re-run's prepared
// target set for this connection is empty and appends
// nothing. The typed refusal travels on unchanged.
return ParticipantConnectionFateOutcome::new(Err(error), impacts);
}
let fatal = match self
.latch_connection_fate_fatal(work_item.open_sequence, conversation_id)
{
Ok(fatal) => fatal,
Err(latch_error) => {
return ParticipantConnectionFateOutcome::new(Err(latch_error), impacts);
}
};
let error = ParticipantSemanticError::Internal {
message: format!("{fatal}: {error}"),
};
return ParticipantConnectionFateOutcome::new(Err(error), impacts);
}
}
ParticipantConnectionFateOutcome::new(Ok(()), impacts)
}
}
fn changed_impact(impact: DispatchImpact) -> Option<DispatchImpact> {
match impact {
DispatchImpact::Unchanged => None,
changed @ DispatchImpact::Changed { .. } => Some(changed),
}
}