acyclic_fs/kernel/
live.rs

1//! Total direct-live publication and safe-rebase retry state machine.
2
3use super::{RebaseConflict, RebaseDecision};
4use crate::foundation::{Digest, Epoch, GenerationId, Head};
5use thiserror::Error;
6
7/// One authority publication observation consumed by live retry orchestration.
8#[derive(Clone, Copy, Debug, Eq, PartialEq)]
9pub enum LivePublicationObservation {
10    /// This attempt durably published the candidate.
11    Committed {
12        /// Published generation.
13        generation_id: GenerationId,
14        /// Durable authority head.
15        head: Head,
16    },
17    /// The same operation and candidate were already durable.
18    AlreadyCommitted {
19        /// Previously published generation.
20        generation_id: GenerationId,
21        /// Original durable authority head.
22        head: Head,
23    },
24    /// Another operation advanced authority first.
25    Conflict {
26        /// Latest linearizable authority head.
27        actual: Head,
28    },
29    /// The writer epoch is stale.
30    Fenced {
31        /// Active authority epoch.
32        actual_epoch: Epoch,
33    },
34    /// The operation identity belongs to another candidate fingerprint.
35    IdempotencyConflict {
36        /// Existing durable fingerprint.
37        committed_fingerprint: Digest,
38    },
39}
40
41/// Terminal outcome of one direct-live mutation publication.
42#[derive(Clone, Debug, Eq, PartialEq)]
43pub enum LiveMutationOutcome {
44    /// This call durably published the candidate generation.
45    Committed {
46        /// Published content-addressed generation.
47        generation_id: GenerationId,
48        /// Authority head produced by the durable append.
49        head: Head,
50    },
51    /// The same operation identity and fingerprint was already durable.
52    AlreadyCommitted {
53        /// Previously published content-addressed generation.
54        generation_id: GenerationId,
55        /// Authority head at the original durable append.
56        head: Head,
57    },
58    /// An exact observed or mutated region changed concurrently.
59    Conflicted {
60        /// Bounded region-specific conflicts.
61        conflicts: Vec<RebaseConflict>,
62        /// Whether additional conflicts were omitted by the caller's bound.
63        truncated: bool,
64    },
65    /// Safe retries were exhausted because the authority kept advancing.
66    RetryLimit {
67        /// Latest linearizable authority head observed by publication.
68        actual: Head,
69    },
70    /// The checkout's writer epoch is stale.
71    Fenced {
72        /// Active authority epoch.
73        actual_epoch: Epoch,
74    },
75    /// The operation identity was already bound to another fingerprint.
76    IdempotencyConflict {
77        /// Fingerprint durably bound to the reused operation identity.
78        committed_fingerprint: Digest,
79    },
80}
81
82/// Next exact action selected by live retry orchestration.
83#[derive(Clone, Debug, Eq, PartialEq)]
84pub enum LiveRetryAction {
85    /// Rebase the retained candidate against the current authority head.
86    Rebase,
87    /// Publish the safely rebased candidate with the same operation identity.
88    Publish,
89    /// Stop with one typed terminal outcome.
90    Complete(LiveMutationOutcome),
91}
92
93#[derive(Clone, Copy, Debug, Eq, PartialEq)]
94enum Phase {
95    Publication,
96    Rebase,
97    Complete,
98}
99
100/// Operation-local total retry controller.
101#[derive(Clone, Copy, Debug, Eq, PartialEq)]
102pub struct LiveRetryState {
103    attempts_remaining: u32,
104    phase: Phase,
105}
106
107impl LiveRetryState {
108    /// Starts before the first publication attempt.
109    ///
110    /// # Errors
111    ///
112    /// Rejects a zero-attempt controller.
113    pub const fn new(maximum_attempts: u32) -> Result<Self, LiveRetryError> {
114        if maximum_attempts == 0 {
115            return Err(LiveRetryError::ZeroAttempts);
116        }
117        Ok(Self {
118            attempts_remaining: maximum_attempts,
119            phase: Phase::Publication,
120        })
121    }
122
123    /// Consumes one authority publication observation.
124    ///
125    /// # Errors
126    ///
127    /// Rejects observations outside the publication phase.
128    pub fn observe_publication(
129        &mut self,
130        observation: LivePublicationObservation,
131    ) -> Result<LiveRetryAction, LiveRetryError> {
132        if self.phase != Phase::Publication {
133            return Err(LiveRetryError::InvalidTransition);
134        }
135        let terminal = match observation {
136            LivePublicationObservation::Committed {
137                generation_id,
138                head,
139            } => Some(LiveMutationOutcome::Committed {
140                generation_id,
141                head,
142            }),
143            LivePublicationObservation::AlreadyCommitted {
144                generation_id,
145                head,
146            } => Some(LiveMutationOutcome::AlreadyCommitted {
147                generation_id,
148                head,
149            }),
150            LivePublicationObservation::Fenced { actual_epoch } => {
151                Some(LiveMutationOutcome::Fenced { actual_epoch })
152            }
153            LivePublicationObservation::IdempotencyConflict {
154                committed_fingerprint,
155            } => Some(LiveMutationOutcome::IdempotencyConflict {
156                committed_fingerprint,
157            }),
158            LivePublicationObservation::Conflict { actual } => {
159                self.attempts_remaining -= 1;
160                if self.attempts_remaining == 0 {
161                    Some(LiveMutationOutcome::RetryLimit { actual })
162                } else {
163                    self.phase = Phase::Rebase;
164                    return Ok(LiveRetryAction::Rebase);
165                }
166            }
167        };
168        self.phase = Phase::Complete;
169        Ok(LiveRetryAction::Complete(
170            terminal.ok_or(LiveRetryError::InvalidTransition)?,
171        ))
172    }
173
174    /// Consumes one exact rebase classification.
175    ///
176    /// # Errors
177    ///
178    /// Rejects classifications outside the rebase phase.
179    pub fn observe_rebase(
180        &mut self,
181        decision: RebaseDecision,
182    ) -> Result<LiveRetryAction, LiveRetryError> {
183        if self.phase != Phase::Rebase {
184            return Err(LiveRetryError::InvalidTransition);
185        }
186        match decision {
187            RebaseDecision::Safe { .. } => {
188                self.phase = Phase::Publication;
189                Ok(LiveRetryAction::Publish)
190            }
191            RebaseDecision::Conflicted {
192                conflicts,
193                truncated,
194            } => {
195                self.phase = Phase::Complete;
196                Ok(LiveRetryAction::Complete(LiveMutationOutcome::Conflicted {
197                    conflicts,
198                    truncated,
199                }))
200            }
201        }
202    }
203}
204
205/// Invalid retry-controller construction or event order.
206#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
207pub enum LiveRetryError {
208    /// No first publication attempt was admitted.
209    #[error("live mutation retry bound must be positive")]
210    ZeroAttempts,
211    /// Driver supplied an observation outside its required phase.
212    #[error("live mutation retry transition is invalid")]
213    InvalidTransition,
214}
215
216#[cfg(test)]
217#[path = "tests/live.rs"]
218mod tests;
acyclic_fs/kernel/live.rs

acyclic_fs/kernel/
live.rs
