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
//! Clean-stop candidate-lane drain for parked connection fates.
//!
//! A connection fate refused by candidate-lane `Precedence` is parked until a
//! drain removes the lane head that refused it. While the server runs, that
//! drain is the next publish's `DrainFirst`. A clean stop has no next publish,
//! and a fate still parked when the process exits leaves its durable Open for
//! the next boot to reconcile — which is exactly what a clean stop promises
//! not to do.
//!
//! The next boot would empty the lane first anyway: `boot_drain.rs` drains
//! every restored conversation's lane before any retained Open replays. This
//! module runs that same drain at the stop, for the conversations a parked
//! fate names and no others, through the same `persist_drain_first` a publish
//! uses and under the live conversation lock, so the drain's commit tail is the
//! live one. Each drain records itself on the impact, and the installed
//! service's tell retries the parked fate.
//!
//! The refusals mirror the boot drain's: a binding-terminal head under an
//! armed fenced-attach recovery block cannot drain here either, and a drain
//! that does not shorten the lane is refused rather than looped on.
use liminal_protocol::lifecycle::ImmutableSequenceCandidate;
use liminal_protocol::wire::ConversationId;
use crate::server::participant::dispatch_impact::DispatchImpactAccumulator;
use crate::server::participant::{ParticipantSemanticError, ParticipantSemanticOutcome};
use super::handler::ProductionParticipantHandler;
use super::state::{ConversationAuthority, DurableAppend, StateError};
impl ProductionParticipantHandler {
/// Drains `conversation_id`'s whole candidate lane under its live owner.
pub(super) fn drain_candidate_lane_for_stop(
&self,
conversation_id: ConversationId,
) -> ParticipantSemanticOutcome<()> {
let mut impact = DispatchImpactAccumulator::new();
if let Err(error) = self.ensure_service_live() {
return ParticipantSemanticOutcome::new(Err(error), impact.finish(conversation_id));
}
let result = self.with_conversation_fate_source(
conversation_id,
Some(&mut impact),
|authority, appender, impact| {
let impact = impact.ok_or_else(|| {
StateError::invariant("stop lane drain lost its impact owner")
})?;
authority.drain_candidate_lane(appender, impact)
},
);
// DUR-027 R5: a retired conversation's lane went with its streams, so
// there is nothing for a stop to drain.
let result = match result {
Err(ParticipantSemanticError::ConversationSealed { .. }) => Ok(()),
other => other,
};
ParticipantSemanticOutcome::new(result, impact.finish(conversation_id))
}
}
impl ConversationAuthority {
/// Removes every head of this conversation's immutable-candidate lane.
fn drain_candidate_lane(
&mut self,
appender: &dyn DurableAppend,
impact: &mut DispatchImpactAccumulator,
) -> Result<(), StateError> {
while let Some((candidate, recovery_armed, remaining)) = self.candidate_lane_head() {
if recovery_armed
&& matches!(
candidate,
ImmutableSequenceCandidate::BindingTerminal { .. }
)
{
return Err(StateError::invariant(format!(
"the candidate-lane head {candidate:?} rests under an armed fenced-attach \
recovery block, which only a live fenced attach can consume"
)));
}
let owner = self.take_frontier()?;
self.persist_drain_first(candidate, owner, appender, impact)?;
if self
.candidate_lane_head()
.is_some_and(|(_, _, next)| next >= remaining)
{
return Err(StateError::invariant(format!(
"the drained head {candidate:?} did not leave the immutable-candidate lane"
)));
}
}
Ok(())
}
/// The lane head, whether a recovery block is armed, and the lane length.
fn candidate_lane_head(&self) -> Option<(ImmutableSequenceCandidate, bool, usize)> {
let frontiers = self.frontier()?.frontiers();
let candidates = frontiers.sequence().immutable_candidates();
Some((
*candidates.first()?,
frontiers.sequence().recovery().is_some() || frontiers.order().recovery().is_some(),
candidates.len(),
))
}
}