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
//! 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 let ParticipantSemanticError::ConversationSealed {
conversation_id: retired,
} = &error
{
// DUR-027 R5: a retired conversation was closed with no
// live identity, so no slot of it is left for this fate
// to depart; its step is done, never a fatal.
tracing::info!(
open_sequence = work_item.open_sequence,
conversation_id = retired,
"connection-fate step for a retired conversation has nothing to depart"
);
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),
}
}