liminal_sdk/remote/participant/
recovery.rs

1use liminal_protocol::client::{
2    DetachReplayRefusalReason, DetachTransportAttemptDecision, DetachTransportFate,
3    DetachTransportFateDecision, ExplicitReconnectAction, LostAuthorityKind,
4    LostOperationAuthorityDecision, LostReconnectAuthorityDecision, ProvedOnlineTransition,
5    ReconnectAttemptDecision, ReconnectAttemptFate, ReconnectAttemptFateDecision,
6    ReconnectAttemptFateRefusalReason, ReconnectAttemptRefusalReason, ReconnectPermitDecision,
7    RecoveredExpectedOperationDecision, RecoveredReconnectPermitDecision, record_attempt_fate,
8    record_explicit_reconnect, record_online_transition, recover_expected_operation,
9    recover_reconnect_permit, redeem_attempt, resolve_lost_operation_authority,
10    resolve_lost_reconnect_authority, transport_attempt_started, transport_fate,
11};
12use liminal_protocol::outcome::ReconnectState;
13use liminal_protocol::wire::{ClientRequest, DetachRequest};
14
15use super::{
16    OperationDurability, ParticipantResumeStore, RemoteOperationTransportFate,
17    RemoteParticipantError, RemoteParticipantHandle, RemoteParticipantOperation,
18    RemoteParticipantSendOutcome, RemoteReconnectPermit, RemoteReconnectPermitOutcome, persist,
19    record_connection_fate, record_operation_transport_fate, take_aggregate,
20};
21
22/// Result of releasing a committed cold-restored operation.
23#[derive(Debug)]
24pub enum RemoteExpectedOperationRecovery {
25    /// One unissued operation authority was recovered.
26    Recovered(RemoteParticipantOperation),
27    /// No recoverable operation exists.
28    NotAvailable {
29        /// Whether the retained operation had already been issued.
30        already_issued: bool,
31    },
32}
33
34/// Result of consuming operation-domain crash testimony.
35#[derive(Debug, PartialEq, Eq)]
36pub enum RemoteLostOperationResolution {
37    /// A non-detach operation was terminalized by serialized testimony.
38    Recorded {
39        /// Exact request whose authority was destroyed.
40        request: ClientRequest,
41        /// Closed testimony kind consumed by the crate.
42        testimony: LostAuthorityKind,
43    },
44    /// A detach was returned to parked replay by serialized testimony.
45    DetachParked {
46        /// Exact detach request retained for replay.
47        request: ClientRequest,
48        /// Closed testimony kind consumed by the crate.
49        testimony: LostAuthorityKind,
50    },
51    /// No operation-domain testimony was pending.
52    Refused {
53        /// Closed crate refusal reason.
54        reason: liminal_protocol::client::LostAuthorityResolutionRefusalReason,
55    },
56}
57
58/// Result of consuming reconnect-domain crash testimony.
59#[derive(Debug, PartialEq, Eq)]
60pub enum RemoteLostReconnectResolution {
61    /// Testimony parked reconnect state without minting replacement authority.
62    Recorded {
63        /// Closed testimony kind consumed by the crate.
64        testimony: LostAuthorityKind,
65    },
66    /// No reconnect-domain testimony was pending.
67    Refused {
68        /// Closed crate refusal reason.
69        reason: liminal_protocol::client::LostAuthorityResolutionRefusalReason,
70    },
71}
72
73/// Result of releasing a committed cold-restored reconnect permit.
74#[derive(Debug)]
75pub enum RemoteReconnectPermitRecovery {
76    /// One unissued permit was recovered.
77    Recovered(RemoteReconnectPermit),
78    /// No recoverable permit exists.
79    NotAvailable {
80        /// Current crate reconnect state.
81        state: ReconnectState,
82    },
83}
84
85/// Result of a real connection attempt redeemed from one crate permit.
86#[derive(Debug)]
87pub enum RemoteReconnectAttemptOutcome {
88    /// The real attempt connected and the crate recorded online state.
89    Connected {
90        /// Provenance assigned to the new established socket.
91        provenance: super::ParticipantResponseProvenance,
92    },
93    /// The real attempt failed and the crate parked without timer authority.
94    Failed {
95        /// Concrete socket failure.
96        error: crate::SdkError,
97    },
98    /// The crate refused permit redemption and returned it unchanged.
99    Refused {
100        /// Reusable unchanged permit.
101        permit: RemoteReconnectPermit,
102        /// Closed crate refusal reason.
103        reason: ReconnectAttemptRefusalReason,
104    },
105    /// The transport ran, but the crate retained the in-progress fate authority.
106    FateRefused {
107        /// Closed crate refusal reason.
108        reason: ReconnectAttemptFateRefusalReason,
109        /// Socket failure when the attempted fate was `Failed`.
110        error: Option<crate::SdkError>,
111        /// New socket provenance when the attempted fate was `Connected`.
112        provenance: Option<super::ParticipantResponseProvenance>,
113    },
114}
115
116/// Result of starting and sending a parked detach replay.
117#[derive(Debug)]
118pub enum RemoteDetachReplayOutcome {
119    /// A real transport send was attempted.
120    Send(RemoteParticipantSendOutcome),
121    /// The crate refused replay start without changing state.
122    Refused {
123        /// Closed crate replay refusal reason.
124        reason: DetachReplayRefusalReason,
125    },
126}
127
128/// Result of an explicit replay apply seam delegated to the crate.
129#[derive(Debug, PartialEq, Eq)]
130pub enum RemoteReplayApplyOutcome<T> {
131    /// The crate applied the transition.
132    Applied,
133    /// The crate retained state, correlation, and exact input.
134    Refused {
135        /// Exact refused input.
136        input: T,
137        /// Closed crate refusal reason.
138        reason: DetachReplayRefusalReason,
139    },
140}
141
142/// Combined typed consequence of an established connection loss.
143#[derive(Debug)]
144pub struct RemoteTransportLossOutcome {
145    /// Operation-domain fate selected by crate rules.
146    pub operation_fate: RemoteOperationTransportFate,
147    /// Event-driven reconnect decision selected by crate rules.
148    pub reconnect: RemoteReconnectPermitOutcome,
149}
150
151impl<S: ParticipantResumeStore> RemoteParticipantHandle<S> {
152    /// Releases one unissued operation from committed cold-restored state.
153    ///
154    /// # Errors
155    ///
156    /// Returns [`RemoteParticipantError::StateUnavailable`] after a prior fatal
157    /// durability failure.
158    pub fn recover_expected_operation(
159        &self,
160    ) -> Result<RemoteExpectedOperationRecovery, RemoteParticipantError> {
161        let mut state = self.state.lock();
162        let aggregate = take_aggregate(&mut state)?;
163        match recover_expected_operation(aggregate) {
164            RecoveredExpectedOperationDecision::Recovered {
165                aggregate,
166                operation,
167            } => {
168                state.aggregate = Some(aggregate);
169                Ok(RemoteExpectedOperationRecovery::Recovered(
170                    RemoteParticipantOperation {
171                        operation,
172                        durability: OperationDurability::WriteAhead,
173                    },
174                ))
175            }
176            RecoveredExpectedOperationDecision::NotAvailable {
177                aggregate,
178                already_issued,
179            } => {
180                state.aggregate = Some(aggregate);
181                Ok(RemoteExpectedOperationRecovery::NotAvailable { already_issued })
182            }
183        }
184    }
185
186    /// Consumes operation-domain lost-authority testimony exactly once.
187    ///
188    /// # Errors
189    ///
190    /// Returns LPCR encode or storage failures while checkpointing the decision.
191    pub fn resolve_lost_operation_authority(
192        &self,
193    ) -> Result<RemoteLostOperationResolution, RemoteParticipantError> {
194        let mut state = self.state.lock();
195        let aggregate = take_aggregate(&mut state)?;
196        let outcome = match resolve_lost_operation_authority(aggregate) {
197            LostOperationAuthorityDecision::Recorded {
198                aggregate,
199                request,
200                testimony,
201            } => {
202                state.aggregate = Some(aggregate);
203                RemoteLostOperationResolution::Recorded {
204                    request,
205                    testimony: testimony.kind(),
206                }
207            }
208            LostOperationAuthorityDecision::DetachParked {
209                aggregate,
210                request,
211                testimony,
212            } => {
213                state.aggregate = Some(aggregate);
214                RemoteLostOperationResolution::DetachParked {
215                    request,
216                    testimony: testimony.kind(),
217                }
218            }
219            LostOperationAuthorityDecision::Refused { aggregate, reason } => {
220                state.aggregate = Some(aggregate);
221                RemoteLostOperationResolution::Refused { reason }
222            }
223        };
224        checkpoint_state(&mut state)?;
225        Ok(outcome)
226    }
227
228    /// Takes a durable tokenless abandonment so its exact request can be re-recorded.
229    ///
230    /// # Errors
231    ///
232    /// Returns LPCR encode or storage failures while durably recording the take.
233    pub fn take_restored_operation_abandonment(
234        &self,
235    ) -> Result<
236        Option<liminal_protocol::client::RestoredExpectedOperationAbandonment>,
237        RemoteParticipantError,
238    > {
239        let mut state = self.state.lock();
240        let mut aggregate = take_aggregate(&mut state)?;
241        let abandonment = aggregate.take_restored_operation_abandonment();
242        if abandonment.is_some() {
243            persist(&mut state.store, &aggregate)?;
244        }
245        state.aggregate = Some(aggregate);
246        Ok(abandonment)
247    }
248
249    /// Records established-connection fate and returns at most one reconnect permit.
250    ///
251    /// # Errors
252    ///
253    /// Returns LPCR encode or storage failures while checkpointing the event.
254    pub fn record_transport_fate(
255        &self,
256    ) -> Result<RemoteReconnectPermitOutcome, RemoteParticipantError> {
257        let mut state = self.state.lock();
258        record_connection_fate(&mut state)
259    }
260
261    /// Records a proved online transition as a crate fresh event.
262    ///
263    /// # Errors
264    ///
265    /// Returns LPCR encode or storage failures while checkpointing issued authority.
266    pub fn record_online_transition(
267        &self,
268    ) -> Result<RemoteReconnectPermitOutcome, RemoteParticipantError> {
269        self.record_fresh_reconnect(|aggregate| {
270            record_online_transition(aggregate, ProvedOnlineTransition::ProvedOnline)
271        })
272    }
273
274    /// Records explicit caller action as a crate fresh event, with no timer arm.
275    ///
276    /// # Errors
277    ///
278    /// Returns LPCR encode or storage failures while checkpointing issued authority.
279    pub fn record_explicit_reconnect(
280        &self,
281    ) -> Result<RemoteReconnectPermitOutcome, RemoteParticipantError> {
282        self.record_fresh_reconnect(|aggregate| {
283            record_explicit_reconnect(aggregate, ExplicitReconnectAction::ReconnectNow)
284        })
285    }
286
287    fn record_fresh_reconnect(
288        &self,
289        decide: impl FnOnce(
290            liminal_protocol::client::ClientParticipantAggregate,
291        ) -> ReconnectPermitDecision,
292    ) -> Result<RemoteReconnectPermitOutcome, RemoteParticipantError> {
293        let mut state = self.state.lock();
294        let aggregate = take_aggregate(&mut state)?;
295        let outcome = match decide(aggregate) {
296            ReconnectPermitDecision::Permitted {
297                aggregate,
298                permit,
299                result,
300            } => {
301                state.aggregate = Some(aggregate);
302                RemoteReconnectPermitOutcome::Permitted {
303                    permit: RemoteReconnectPermit { permit },
304                    result,
305                }
306            }
307            ReconnectPermitDecision::Refused(refusal) => {
308                let reason = refusal.reason();
309                let result = refusal.result();
310                let (aggregate, _) = refusal.into_parts();
311                state.aggregate = Some(aggregate);
312                RemoteReconnectPermitOutcome::Refused { reason, result }
313            }
314        };
315        checkpoint_state(&mut state)?;
316        Ok(outcome)
317    }
318
319    /// Releases one unissued reconnect permit from committed cold-restored state.
320    ///
321    /// # Errors
322    ///
323    /// Returns [`RemoteParticipantError::StateUnavailable`] after a prior fatal failure.
324    pub fn recover_reconnect_permit(
325        &self,
326    ) -> Result<RemoteReconnectPermitRecovery, RemoteParticipantError> {
327        let mut state = self.state.lock();
328        let aggregate = take_aggregate(&mut state)?;
329        match recover_reconnect_permit(aggregate) {
330            RecoveredReconnectPermitDecision::Recovered { aggregate, permit } => {
331                state.aggregate = Some(aggregate);
332                Ok(RemoteReconnectPermitRecovery::Recovered(
333                    RemoteReconnectPermit { permit },
334                ))
335            }
336            RecoveredReconnectPermitDecision::NotAvailable {
337                aggregate,
338                state: value,
339            } => {
340                state.aggregate = Some(aggregate);
341                Ok(RemoteReconnectPermitRecovery::NotAvailable { state: value })
342            }
343        }
344    }
345
346    /// Consumes reconnect-domain lost-authority testimony exactly once.
347    ///
348    /// # Errors
349    ///
350    /// Returns LPCR encode or storage failures while checkpointing the resolution.
351    pub fn resolve_lost_reconnect_authority(
352        &self,
353    ) -> Result<RemoteLostReconnectResolution, RemoteParticipantError> {
354        let mut state = self.state.lock();
355        let aggregate = take_aggregate(&mut state)?;
356        let outcome = match resolve_lost_reconnect_authority(aggregate) {
357            LostReconnectAuthorityDecision::Recorded {
358                aggregate,
359                testimony,
360            } => {
361                state.aggregate = Some(aggregate);
362                RemoteLostReconnectResolution::Recorded {
363                    testimony: testimony.kind(),
364                }
365            }
366            LostReconnectAuthorityDecision::Refused { aggregate, reason } => {
367                state.aggregate = Some(aggregate);
368                RemoteLostReconnectResolution::Refused { reason }
369            }
370        };
371        checkpoint_state(&mut state)?;
372        Ok(outcome)
373    }
374
375    /// Redeems one permit before opening one real transport connection.
376    ///
377    /// # Errors
378    ///
379    /// Returns LPCR encode or storage failures before or after the real attempt.
380    pub fn reconnect(
381        &self,
382        permit: RemoteReconnectPermit,
383    ) -> Result<RemoteReconnectAttemptOutcome, RemoteParticipantError> {
384        let mut state = self.state.lock();
385        let aggregate = take_aggregate(&mut state)?;
386        let (aggregate, attempt) = match redeem_attempt(aggregate, permit.permit) {
387            ReconnectAttemptDecision::Started { aggregate, attempt } => (aggregate, attempt),
388            ReconnectAttemptDecision::Refused {
389                aggregate,
390                permit,
391                reason,
392            } => {
393                state.aggregate = Some(aggregate);
394                return Ok(RemoteReconnectAttemptOutcome::Refused {
395                    permit: RemoteReconnectPermit { permit },
396                    reason,
397                });
398            }
399        };
400        persist(&mut state.store, &aggregate)?;
401        state.aggregate = Some(aggregate);
402        drop(state);
403
404        let transport_result = self.transport.reconnect_participant(&self.server_address);
405        let fate = if transport_result.is_ok() {
406            ReconnectAttemptFate::Connected
407        } else {
408            ReconnectAttemptFate::Failed
409        };
410
411        let mut state = self.state.lock();
412        let aggregate = take_aggregate(&mut state)?;
413        match record_attempt_fate(aggregate, attempt, fate) {
414            ReconnectAttemptFateDecision::Recorded(aggregate) => {
415                persist(&mut state.store, &aggregate)?;
416                state.aggregate = Some(aggregate);
417                match transport_result {
418                    Ok(provenance) => Ok(RemoteReconnectAttemptOutcome::Connected { provenance }),
419                    Err(error) => Ok(RemoteReconnectAttemptOutcome::Failed { error }),
420                }
421            }
422            ReconnectAttemptFateDecision::Refused {
423                aggregate,
424                attempt,
425                reason,
426                ..
427            } => {
428                state.aggregate = Some(aggregate);
429                state.reconnect_attempt = Some(attempt);
430                let (provenance, error) = match transport_result {
431                    Ok(value) => (Some(value), None),
432                    Err(value) => (None, Some(value)),
433                };
434                Ok(RemoteReconnectAttemptOutcome::FateRefused {
435                    reason,
436                    error,
437                    provenance,
438                })
439            }
440        }
441    }
442
443    /// Records response and connection fates after an established transport loss.
444    ///
445    /// # Errors
446    ///
447    /// Returns LPCR encode or storage failures while checkpointing both decisions.
448    pub fn record_established_transport_loss(
449        &self,
450    ) -> Result<RemoteTransportLossOutcome, RemoteParticipantError> {
451        let mut state = self.state.lock();
452        let operation_fate = if let Some(correlation) = state.correlation.take() {
453            let aggregate = take_aggregate(&mut state)?;
454            record_operation_transport_fate(&mut state, aggregate, correlation)
455        } else {
456            RemoteOperationTransportFate::NotOutstanding
457        };
458        let reconnect = record_connection_fate(&mut state)?;
459        Ok(RemoteTransportLossOutcome {
460            operation_fate,
461            reconnect,
462        })
463    }
464
465    /// Starts and sends the exact parked detach replay selected by the crate.
466    ///
467    /// # Errors
468    ///
469    /// Returns LPCR, storage, or state failures. Socket failure is a typed send outcome.
470    pub fn replay_detach(&self) -> Result<RemoteDetachReplayOutcome, RemoteParticipantError> {
471        let mut state = self.state.lock();
472        let aggregate = take_aggregate(&mut state)?;
473        let (aggregate, attempt) = match transport_attempt_started(aggregate) {
474            DetachTransportAttemptDecision::Started { aggregate, attempt } => (aggregate, attempt),
475            DetachTransportAttemptDecision::Refused(refusal) => {
476                let reason = refusal.reason();
477                let (aggregate, ()) = refusal.into_parts();
478                state.aggregate = Some(aggregate);
479                return Ok(RemoteDetachReplayOutcome::Refused { reason });
480            }
481        };
482        persist(&mut state.store, &aggregate)?;
483        let (request, correlation) = attempt.into_request();
484        let request = ClientRequest::Detach(DetachRequest {
485            conversation_id: request.conversation_id,
486            participant_id: request.participant_id,
487            capability_generation: request.capability_generation,
488            detach_attempt_token: request.detach_attempt_token,
489        });
490        match self
491            .transport
492            .send_participant(&self.server_address, &request)
493        {
494            Ok(provenance) => {
495                state.aggregate = Some(aggregate);
496                state.correlation = Some(correlation);
497                Ok(RemoteDetachReplayOutcome::Send(
498                    RemoteParticipantSendOutcome::Sent { provenance },
499                ))
500            }
501            Err(error) => {
502                let operation_fate = match transport_fate(
503                    aggregate,
504                    correlation,
505                    DetachTransportFate::ResponseUnavailable,
506                ) {
507                    DetachTransportFateDecision::Parked(applied) => {
508                        state.aggregate = Some(applied.into_aggregate());
509                        RemoteOperationTransportFate::DetachParked
510                    }
511                    DetachTransportFateDecision::Refused(refusal) => {
512                        let (aggregate, (correlation, _)) = refusal.into_parts();
513                        state.aggregate = Some(aggregate);
514                        state.correlation = Some(correlation);
515                        RemoteOperationTransportFate::Refused {
516                            reason: liminal_protocol::client::ExpectedOperationFateRefusalReason::DetachUsesReplayFate,
517                        }
518                    }
519                };
520                let reconnect = record_connection_fate(&mut state)?;
521                Ok(RemoteDetachReplayOutcome::Send(
522                    RemoteParticipantSendOutcome::TransportLost {
523                        error,
524                        operation_fate,
525                        reconnect,
526                    },
527                ))
528            }
529        }
530    }
531}
532
533fn checkpoint_state<S: ParticipantResumeStore>(
534    state: &mut super::RemoteParticipantState<S>,
535) -> Result<(), RemoteParticipantError> {
536    let aggregate = take_aggregate(state)?;
537    persist(&mut state.store, &aggregate)?;
538    state.aggregate = Some(aggregate);
539    Ok(())
540}
liminal_sdk/remote/participant/recovery.rs

liminal_sdk/remote/participant/
recovery.rs
