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
//! Shared aggregate-barrier plumbing for the production operation arms.
//!
//! Every committing arm (enrollment, attach, detach) resolves its pending A3
//! aggregate barrier through [`commit_through_barrier`]: the live mode
//! appends the operation's stored inputs plus canonical event bytes at the
//! optimistic head before the shell advances; the replay mode re-mints the
//! canonical bytes and cross-checks them against the stored entry, so byte
//! drift between live and replayed decisions fails loudly.
use liminal_protocol::lifecycle::{
CapacityCounter, ConnectionConversationCapacityCommit, ConnectionConversationTracking,
ReceiptDeadlines, SemanticConnectionCapacityDecision, select_semantic_connection_capacity,
};
use liminal_protocol::wire::{ConnectionIncarnation, ServerValue};
use crate::config::types::ParticipantConfig;
use super::log::StoredOperation;
use super::state::{DurableAppend, StateError};
/// Connection-scoped and configured facts supplied to each operation.
#[derive(Clone, Copy, Debug)]
pub(super) struct OperationFacts {
/// Durable incarnation of the receiving connection.
pub(super) receiving_incarnation: ConnectionIncarnation,
/// Admitted wall-clock read for deadline derivation and receipt phases.
pub(super) now_ms: u64,
/// Configured per-conversation identity limit `I` (the contract's
/// half-open `0..=I` bound on permanent participant ordinals).
pub(super) identity_slots: u64,
/// Configured secret-bearing receipt TTL.
pub(super) attach_receipt_ttl_ms: u64,
/// Configured non-secret provenance TTL.
pub(super) receipt_provenance_ttl_ms: u64,
/// Signed R-D1 stage-8 identity/receipt capacity limits.
pub(super) receipt_limits: ReceiptCapacityLimits,
/// Whether the receiving connection already tracks this conversation.
pub(super) connection_tracking: ConnectionConversationTracking,
/// Signed connection-conversation limit with current occupancy.
pub(super) connection_capacity: CapacityCounter,
}
/// Signed stage-8 capacity numbers, straight from validated configuration.
///
/// # Lane p0-39: three kinds of number, and only one of them refuses
///
/// * `identity_server` is a GATE. Identity capacity is out of the hybrid's
/// scope and behaves exactly as it always has.
/// * The two `*_window` values are BOUNDS ON RETENTION. At a full window the
/// participant's own oldest entry is displaced and the arrival lands; the
/// number never refuses.
/// * [`SharedPoolTripwires`] are REPORTING THRESHOLDS. They gate nothing at
/// all — they decide only when the server counts and warns.
#[derive(Clone, Copy, Debug)]
pub(super) struct ReceiptCapacityLimits {
/// Server-wide identity-slot limit — the one receipt-family neighbour
/// that is still an admission gate.
pub(super) identity_server: u64,
/// Per-participant live-receipt window size.
pub(super) live_receipt_participant_window: u64,
/// Per-participant provenance-fingerprint window size.
pub(super) provenance_participant_window: u64,
/// Reporting thresholds for the three shared pools.
pub(super) shared_pool_tripwires: SharedPoolTripwires,
}
/// Occupancy at which each shared pool is reported as running away.
///
/// Crossing one of these refuses NOTHING. It counts an observation and, on the
/// rising edge, emits one warning naming the pool, its occupancy, and this
/// threshold — the disclosure that replaces the wall these pools used to be.
#[derive(Clone, Copy, Debug)]
pub(super) struct SharedPoolTripwires {
/// Server-wide live-receipt reporting threshold.
pub(super) live_receipt_server: u64,
/// Server-wide provenance reporting threshold.
pub(super) provenance_server: u64,
/// Per-conversation provenance reporting threshold.
pub(super) provenance_conversation: u64,
}
impl ReceiptCapacityLimits {
/// Derives every stage-8 number from one validated participant config.
///
/// One constructor, used by the live operation path, the replay path, and
/// every fixture — so a window size and its tripwire threshold can never
/// be read from configuration two different ways.
pub(super) const fn from_config(config: &ParticipantConfig) -> Self {
Self {
identity_server: config.max_retired_identity_slots_server,
live_receipt_participant_window: config.max_live_attach_receipts_per_participant,
provenance_participant_window: config.max_receipt_provenance_per_participant,
shared_pool_tripwires: SharedPoolTripwires {
live_receipt_server: config.live_receipt_server_report_threshold,
provenance_server: config.receipt_provenance_server_report_threshold,
provenance_conversation: config
.receipt_provenance_per_conversation_report_threshold,
},
}
}
}
impl OperationFacts {
/// Derives the receipt/provenance deadline pair from the admitted clock.
pub(super) fn deadlines(&self) -> Result<ReceiptDeadlines, StateError> {
ReceiptDeadlines::try_from_ttls(
self.now_ms,
self.attach_receipt_ttl_ms,
self.receipt_provenance_ttl_ms,
)
.map_err(|error| {
StateError::invariant(format!("validated TTL configuration rejected: {error:?}"))
})
}
/// Runs the crate's stage-6 semantic connection-conversation capacity
/// selector for this operation's connection facts.
pub(super) const fn semantic_connection_capacity(&self) -> SemanticConnectionCapacityDecision {
select_semantic_connection_capacity(self.connection_tracking, self.connection_capacity)
}
}
/// One operation arm's response paired with its connection-tracking effect.
///
/// `newly_tracked` is `true` exactly when the operation COMMITTED and its
/// stage-6 capacity commit reserved a new connection-conversation slot; every
/// refusal and replay carries `false`, mirroring the crate's rule that a
/// refused operation leaves the connection counter unchanged.
#[derive(Debug)]
pub(super) struct ArmOutcome {
/// Protocol-owned response value.
pub(super) value: ServerValue,
/// Whether the caller must install this conversation's connection slot.
pub(super) newly_tracked: bool,
}
impl ArmOutcome {
/// A refusal or replay: the connection's dispatch map is unchanged.
pub(super) const fn respond(value: ServerValue) -> Self {
Self {
value,
newly_tracked: false,
}
}
/// A committed operation carrying its stage-6 capacity commit.
pub(super) const fn committed(
value: ServerValue,
capacity: ConnectionConversationCapacityCommit,
) -> Self {
Self {
value,
newly_tracked: capacity.newly_tracked(),
}
}
}
/// One barrier resolution mode: live durable append or replay byte-check.
#[derive(Clone, Copy)]
pub(super) enum CommitMode<'a> {
/// Append the operation at the optimistic head, then commit.
Live(&'a dyn DurableAppend),
/// Cross-check re-minted canonical bytes against the stored entry.
Replay {
/// Canonical event bytes read from the durable entry.
stored_event: &'a [u8],
/// Durable log sequence of the entry (for drift diagnostics).
sequence: u64,
},
}
/// Resolves one pending aggregate barrier through the selected mode.
pub(super) fn commit_through_barrier<T>(
barrier: liminal_protocol::lifecycle::AggregateOperationCommit<T>,
mode: CommitMode<'_>,
next_log_sequence: u64,
make_operation: &dyn Fn(Vec<u8>) -> StoredOperation,
) -> Result<(liminal_protocol::lifecycle::ParticipantConversation, T), StateError> {
let event = barrier.event().encode_canonical();
match mode {
CommitMode::Live(appender) => {
let operation = make_operation(event);
appender.append(&operation, next_log_sequence)?;
}
CommitMode::Replay {
stored_event,
sequence,
} => {
if event != stored_event {
return Err(StateError::ReplayedEventDrift { sequence });
}
}
}
Ok(barrier.commit())
}