oxide_batch/
fault_state.rs

1//! Retry-key derivation, bounded fault-state reservation, and runtime bundle.
2//!
3//! The chunk runtime reserves a retry ordinal through [`FaultStateStore`]
4//! *after* a known rollback and *before* backoff, so a process that stops
5//! between reservation and re-invocation has still consumed the ordinal. This
6//! module owns the framework side of that boundary: the opaque retry key, the
7//! compare-and-swap reservation contract, and a bounded in-memory
8//! implementation.
9//!
10//! [`InMemoryFaultState`] keeps reservations for one process only, which the
11//! contract permits because a restart may invoke fewer retries than were
12//! reserved, never more. [`FaultStateEnvelope`] is the durable format a
13//! repository adapter persists and validates; `PostgresFaultState` implements
14//! the same ordering against schema 2.
15
16use std::collections::BTreeMap;
17use std::error::Error;
18use std::fmt;
19use std::sync::{Arc, Mutex, PoisonError};
20
21use serde_json::{Map, Number, Value};
22use sha2::{Digest, Sha256};
23
24use crate::{
25    BackoffSleeper, BoxFuture, ChunkDeliveryMode, ChunkTransactionContext, ClassifierRevision,
26    FailureCategory, FaultPhase, FaultPolicy, RetryLimit, RetryOrdinal, RetryStateLimit,
27    SkipCounts, StepName,
28};
29
30/// The domain separator that keeps retry keys distinct from other digests.
31const RETRY_KEY_DOMAIN: &[u8] = b"oxide-batch/retry-key/1";
32
33/// An opaque framework digest identifying one retryable unit of work.
34///
35/// The key is a SHA-256 digest over the definition fingerprint, step logical
36/// ID, failure phase, committed checkpoint identity, and the stable item or
37/// output ordinal. It contains no item value, and it is never a telemetry
38/// field: [`Debug`] redacts it and durable state sorts keys by digest.
39#[derive(Clone, Copy, Eq, Hash, Ord, PartialEq, PartialOrd)]
40pub struct RetryKey([u8; 32]);
41
42impl RetryKey {
43    /// Derives the key for one failed unit of work.
44    #[must_use]
45    pub(crate) fn derive(
46        definition_digest: &[u8; 32],
47        step_name: &StepName,
48        phase: FaultPhase,
49        checkpoint_digest: &[u8; 32],
50        ordinal: u64,
51    ) -> Self {
52        let mut hasher = Sha256::new();
53        hasher.update(RETRY_KEY_DOMAIN);
54        hasher.update(definition_digest);
55        hasher.update((step_name.as_str().len() as u64).to_be_bytes());
56        hasher.update(step_name.as_str().as_bytes());
57        hasher.update(phase.as_str().as_bytes());
58        hasher.update([0]);
59        hasher.update(checkpoint_digest);
60        hasher.update(ordinal.to_be_bytes());
61        Self(hasher.finalize().into())
62    }
63
64    /// Restores a key an authorized durable-state adapter persisted.
65    ///
66    /// Only a store that round-trips [`Self::as_bytes`] may call this. The
67    /// runtime always derives keys from framework inputs.
68    #[must_use]
69    pub const fn from_bytes(digest: [u8; 32]) -> Self {
70        Self(digest)
71    }
72
73    /// Borrows the digest for an authorized durable-state adapter.
74    ///
75    /// The digest is restart-relevant persistence input. It must not be logged,
76    /// exported as telemetry, or used as a metric label.
77    #[must_use]
78    pub const fn as_bytes(&self) -> &[u8; 32] {
79        &self.0
80    }
81}
82
83impl fmt::Debug for RetryKey {
84    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
85        formatter
86            .debug_struct("RetryKey")
87            .field("digest", &"<redacted>")
88            .finish()
89    }
90}
91
92/// One durable retry reservation for a single key.
93///
94/// The reservation records the phase and stable category that produced it, so
95/// exhaustion preserves the last typed category without retaining error text.
96#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
97pub struct RetryReservation {
98    key: RetryKey,
99    phase: FaultPhase,
100    category: FailureCategory,
101    ordinal: RetryOrdinal,
102}
103
104impl RetryReservation {
105    /// Constructs the reservation the runtime asks the store to commit.
106    #[must_use]
107    pub const fn new(
108        key: RetryKey,
109        phase: FaultPhase,
110        category: FailureCategory,
111        ordinal: RetryOrdinal,
112    ) -> Self {
113        Self {
114            key,
115            phase,
116            category,
117            ordinal,
118        }
119    }
120
121    /// Returns the opaque retry key.
122    #[must_use]
123    pub const fn key(self) -> RetryKey {
124        self.key
125    }
126
127    /// Returns the phase that produced the fault.
128    #[must_use]
129    pub const fn phase(self) -> FaultPhase {
130        self.phase
131    }
132
133    /// Returns the stable category preserved for exhaustion.
134    #[must_use]
135    pub const fn category(self) -> FailureCategory {
136        self.category
137    }
138
139    /// Returns the reserved retry ordinal.
140    #[must_use]
141    pub const fn ordinal(self) -> RetryOrdinal {
142        self.ordinal
143    }
144}
145
146/// A value-redacted fault-state reservation failure.
147#[derive(Clone, Copy, Debug, Eq, PartialEq)]
148#[non_exhaustive]
149pub enum FaultStateError {
150    /// The step already retains its maximum unresolved retry keys.
151    CapacityExhausted {
152        /// The configured unresolved-key capacity.
153        max: u32,
154    },
155    /// The supplied ordinal did not follow the persisted one.
156    ///
157    /// A stale or concurrent writer loses rather than spending the same
158    /// ordinal twice.
159    StaleReservation,
160    /// Durable fault state could not be interpreted and no work may begin.
161    Corrupt(FaultStateFormatError),
162    /// A durable store was used before the runtime bound its step execution.
163    Unbound,
164    /// The fault state could not be read or written.
165    Unavailable,
166}
167
168impl fmt::Display for FaultStateError {
169    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
170        match self {
171            Self::CapacityExhausted { max } => {
172                write!(
173                    formatter,
174                    "step already retains {max} unresolved retry keys"
175                )
176            }
177            Self::StaleReservation => {
178                formatter.write_str("retry reservation lost to a newer persisted ordinal")
179            }
180            Self::Corrupt(error) => write!(formatter, "durable fault state is unusable: {error}"),
181            Self::Unbound => formatter.write_str("durable fault state has no bound step execution"),
182            Self::Unavailable => formatter.write_str("fault state is unavailable"),
183        }
184    }
185}
186
187impl Error for FaultStateError {
188    fn source(&self) -> Option<&(dyn Error + 'static)> {
189        match self {
190            Self::Corrupt(error) => Some(error),
191            _ => None,
192        }
193    }
194}
195
196impl From<FaultStateFormatError> for FaultStateError {
197    fn from(error: FaultStateFormatError) -> Self {
198        Self::Corrupt(error)
199    }
200}
201
202/// One unresolved retry key retained in durable fault state.
203///
204/// The entry holds only framework-owned classification identity. It never
205/// contains an item value, error text, parameter, or context value.
206#[derive(Clone, Debug, Eq, PartialEq)]
207pub struct FaultStateEntry {
208    key: RetryKey,
209    phase: FaultPhase,
210    category: FailureCategory,
211    ordinal: RetryOrdinal,
212    revision: ClassifierRevision,
213}
214
215impl FaultStateEntry {
216    /// Constructs one retained entry.
217    #[must_use]
218    pub const fn new(
219        key: RetryKey,
220        phase: FaultPhase,
221        category: FailureCategory,
222        ordinal: RetryOrdinal,
223        revision: ClassifierRevision,
224    ) -> Self {
225        Self {
226            key,
227            phase,
228            category,
229            ordinal,
230            revision,
231        }
232    }
233
234    /// Returns the opaque retry key.
235    #[must_use]
236    pub const fn key(&self) -> RetryKey {
237        self.key
238    }
239
240    /// Returns the phase that produced the retained fault.
241    #[must_use]
242    pub const fn phase(&self) -> FaultPhase {
243        self.phase
244    }
245
246    /// Returns the stable category preserved for exhaustion.
247    #[must_use]
248    pub const fn category(&self) -> FailureCategory {
249        self.category
250    }
251
252    /// Returns the reserved retry ordinal.
253    #[must_use]
254    pub const fn ordinal(&self) -> RetryOrdinal {
255        self.ordinal
256    }
257
258    /// Borrows the classifier revision that produced the decision.
259    #[must_use]
260    pub const fn revision(&self) -> &ClassifierRevision {
261        &self.revision
262    }
263
264    fn to_json(&self) -> Value {
265        let mut object = Map::new();
266        object.insert(
267            String::from("category"),
268            Value::String(String::from(self.category.durable_code())),
269        );
270        object.insert(String::from("key"), Value::String(hex(self.key.as_bytes())));
271        object.insert(
272            String::from("ordinal"),
273            Value::Number(Number::from(self.ordinal.get())),
274        );
275        object.insert(
276            String::from("phase"),
277            Value::String(String::from(self.phase.as_str())),
278        );
279        object.insert(
280            String::from("revision"),
281            Value::String(String::from(self.revision.as_str())),
282        );
283        Value::Object(object)
284    }
285
286    fn from_json(value: &Value) -> Result<Self, FaultStateFormatError> {
287        let object = value
288            .as_object()
289            .ok_or(FaultStateFormatError::MalformedEntry)?;
290        let key = object
291            .get("key")
292            .and_then(Value::as_str)
293            .and_then(unhex)
294            .map(RetryKey::from_bytes)
295            .ok_or(FaultStateFormatError::MalformedEntry)?;
296        let phase = object
297            .get("phase")
298            .and_then(Value::as_str)
299            .and_then(FaultPhase::from_durable_name)
300            .ok_or(FaultStateFormatError::UnknownEnumeration)?;
301        let category = object
302            .get("category")
303            .and_then(Value::as_str)
304            .and_then(FailureCategory::from_durable_code)
305            .ok_or(FaultStateFormatError::UnknownEnumeration)?;
306        let ordinal = object
307            .get("ordinal")
308            .and_then(Value::as_u64)
309            .and_then(|value| u32::try_from(value).ok())
310            .and_then(|value| RetryOrdinal::new(value).ok())
311            .ok_or(FaultStateFormatError::MalformedEntry)?;
312        let revision = object
313            .get("revision")
314            .and_then(Value::as_str)
315            .and_then(|value| ClassifierRevision::new(value).ok())
316            .ok_or(FaultStateFormatError::MalformedEntry)?;
317        Ok(Self::new(key, phase, category, ordinal, revision))
318    }
319}
320
321/// The bounded, checksummed fault state of one durable step execution.
322///
323/// Format 1 is canonical JSON containing the prior committed checkpoint digest
324/// and at most [`Self::MAX_ENTRIES`] digest-sorted unresolved retry
325/// entries. The empty envelope carries the zero checkpoint digest because no
326/// entry depends on a checkpoint generation.
327#[derive(Clone, Debug, Eq, PartialEq)]
328pub struct FaultStateEnvelope {
329    checkpoint_digest: [u8; 32],
330    entries: Vec<FaultStateEntry>,
331}
332
333impl FaultStateEnvelope {
334    /// The framework format identifier durable adapters persist.
335    pub const FORMAT: &'static str = "oxide-batch.fault-state";
336    /// The framework format version durable adapters persist.
337    pub const FORMAT_VERSION: u16 = 1;
338    /// The framework schema version durable adapters persist.
339    pub const SCHEMA_VERSION: u32 = 1;
340    /// The canonical byte ceiling accepted by the durable metadata model.
341    pub const MAX_BYTES: usize = 64 * 1024;
342    /// The hard unresolved-key ceiling of format 1.
343    pub const MAX_ENTRIES: usize = 256;
344
345    /// Returns the envelope every step execution starts from.
346    #[must_use]
347    pub const fn empty() -> Self {
348        Self {
349            checkpoint_digest: [0; 32],
350            entries: Vec::new(),
351        }
352    }
353
354    /// Validates a complete envelope, sorting entries by retry-key digest.
355    ///
356    /// # Errors
357    ///
358    /// Returns [`FaultStateFormatError::TooManyEntries`] above the format
359    /// ceiling, [`FaultStateFormatError::DuplicateKey`] for a repeated key, and
360    /// [`FaultStateFormatError::CheckpointMismatch`] when a non-empty envelope
361    /// carries the zero checkpoint digest.
362    pub fn new(
363        checkpoint_digest: [u8; 32],
364        entries: impl IntoIterator<Item = FaultStateEntry>,
365    ) -> Result<Self, FaultStateFormatError> {
366        let mut entries: Vec<FaultStateEntry> = entries.into_iter().collect();
367        if entries.len() > Self::MAX_ENTRIES {
368            return Err(FaultStateFormatError::TooManyEntries {
369                max: Self::MAX_ENTRIES,
370            });
371        }
372        entries.sort_by_key(FaultStateEntry::key);
373        if entries.windows(2).any(|pair| pair[0].key == pair[1].key) {
374            return Err(FaultStateFormatError::DuplicateKey);
375        }
376        if entries.is_empty() {
377            if checkpoint_digest != [0; 32] {
378                return Err(FaultStateFormatError::CheckpointMismatch);
379            }
380        } else if checkpoint_digest == [0; 32] {
381            return Err(FaultStateFormatError::CheckpointMismatch);
382        }
383        Ok(Self {
384            checkpoint_digest,
385            entries,
386        })
387    }
388
389    /// Returns the checkpoint generation the retained entries belong to.
390    #[must_use]
391    pub const fn checkpoint_digest(&self) -> &[u8; 32] {
392        &self.checkpoint_digest
393    }
394
395    /// Borrows the digest-sorted unresolved entries.
396    #[must_use]
397    pub fn entries(&self) -> &[FaultStateEntry] {
398        &self.entries
399    }
400
401    /// Returns whether the envelope retains no unresolved key.
402    #[must_use]
403    pub fn is_empty(&self) -> bool {
404        self.entries.is_empty()
405    }
406
407    /// Returns the number of retained unresolved keys.
408    #[must_use]
409    pub fn len(&self) -> usize {
410        self.entries.len()
411    }
412
413    /// Returns the ordinal already reserved for `key`, when one exists.
414    #[must_use]
415    pub fn reserved_ordinal(&self, key: RetryKey) -> Option<RetryOrdinal> {
416        self.entry(key).map(FaultStateEntry::ordinal)
417    }
418
419    /// Borrows the retained entry for `key`, when one exists.
420    #[must_use]
421    pub fn entry(&self, key: RetryKey) -> Option<&FaultStateEntry> {
422        self.entries
423            .binary_search_by(|entry| entry.key.cmp(&key))
424            .ok()
425            .map(|index| &self.entries[index])
426    }
427
428    /// Returns the envelope after accepting one reservation.
429    ///
430    /// The ordinal must directly follow the persisted one for the same key, and
431    /// a new key must fit within `limit` unresolved keys.
432    ///
433    /// # Errors
434    ///
435    /// Returns [`FaultStateError::StaleReservation`] for a non-consecutive
436    /// ordinal, [`FaultStateError::CapacityExhausted`] at the configured bound,
437    /// and [`FaultStateError::Corrupt`] when the entry belongs to a different
438    /// checkpoint generation.
439    pub fn reserved(
440        &self,
441        entry: FaultStateEntry,
442        checkpoint_digest: [u8; 32],
443        limit: RetryStateLimit,
444    ) -> Result<Self, FaultStateError> {
445        if !self.entries.is_empty() && self.checkpoint_digest != checkpoint_digest {
446            return Err(FaultStateError::Corrupt(
447                FaultStateFormatError::CheckpointMismatch,
448            ));
449        }
450        let expected = self
451            .reserved_ordinal(entry.key())
452            .unwrap_or(RetryOrdinal::INITIAL)
453            .checked_next()
454            .map_err(|_| FaultStateError::StaleReservation)?;
455        if entry.ordinal() != expected {
456            return Err(FaultStateError::StaleReservation);
457        }
458        let mut entries = self.entries.clone();
459        match entries.binary_search_by(|existing| existing.key.cmp(&entry.key())) {
460            Ok(index) => entries[index] = entry,
461            Err(index) => {
462                if entries.len() >= limit.get() as usize {
463                    return Err(FaultStateError::CapacityExhausted { max: limit.get() });
464                }
465                entries.insert(index, entry);
466            }
467        }
468        Ok(Self {
469            checkpoint_digest,
470            entries,
471        })
472    }
473
474    /// Serializes the canonical bytes the durable checksum covers.
475    ///
476    /// # Errors
477    ///
478    /// Returns [`FaultStateFormatError::TooLarge`] above the durable ceiling.
479    pub fn to_canonical_json(&self) -> Result<Vec<u8>, FaultStateFormatError> {
480        let mut object = Map::new();
481        object.insert(
482            String::from("checkpoint"),
483            Value::String(hex(&self.checkpoint_digest)),
484        );
485        object.insert(
486            String::from("entries"),
487            Value::Array(self.entries.iter().map(FaultStateEntry::to_json).collect()),
488        );
489        let bytes = serde_json::to_vec(&Value::Object(object))
490            .map_err(|_| FaultStateFormatError::Malformed)?;
491        if bytes.len() > Self::MAX_BYTES {
492            return Err(FaultStateFormatError::TooLarge {
493                max_bytes: Self::MAX_BYTES,
494            });
495        }
496        Ok(bytes)
497    }
498
499    /// Returns the SHA-256 checksum over the canonical bytes.
500    ///
501    /// # Errors
502    ///
503    /// Returns [`FaultStateFormatError::TooLarge`] above the durable ceiling.
504    pub fn checksum(&self) -> Result<[u8; 32], FaultStateFormatError> {
505        Ok(Sha256::digest(self.to_canonical_json()?).into())
506    }
507
508    /// Validates canonical bytes and their durable checksum.
509    ///
510    /// Unknown format or schema versions, checksum mismatch, invalid
511    /// enumerations, an unsorted or duplicated key, and an over-large payload
512    /// are corruption. No component work may begin after one.
513    ///
514    /// # Errors
515    ///
516    /// Returns the redacted [`FaultStateFormatError`] that rejected the bytes.
517    pub fn from_canonical_json(
518        format_version: u16,
519        schema: &str,
520        schema_version: u32,
521        bytes: &[u8],
522        checksum: &[u8; 32],
523    ) -> Result<Self, FaultStateFormatError> {
524        if format_version != Self::FORMAT_VERSION || schema != Self::FORMAT {
525            return Err(FaultStateFormatError::UnsupportedFormat);
526        }
527        if schema_version != Self::SCHEMA_VERSION {
528            return Err(FaultStateFormatError::UnsupportedSchemaVersion);
529        }
530        if bytes.len() > Self::MAX_BYTES {
531            return Err(FaultStateFormatError::TooLarge {
532                max_bytes: Self::MAX_BYTES,
533            });
534        }
535        let value: Value =
536            serde_json::from_slice(bytes).map_err(|_| FaultStateFormatError::Malformed)?;
537        let object = value.as_object().ok_or(FaultStateFormatError::Malformed)?;
538        let checkpoint_digest = object
539            .get("checkpoint")
540            .and_then(Value::as_str)
541            .and_then(unhex)
542            .ok_or(FaultStateFormatError::Malformed)?;
543        let raw = object
544            .get("entries")
545            .and_then(Value::as_array)
546            .ok_or(FaultStateFormatError::Malformed)?;
547        if raw.len() > Self::MAX_ENTRIES {
548            return Err(FaultStateFormatError::TooManyEntries {
549                max: Self::MAX_ENTRIES,
550            });
551        }
552        let entries = raw
553            .iter()
554            .map(FaultStateEntry::from_json)
555            .collect::<Result<Vec<_>, _>>()?;
556        if entries.windows(2).any(|pair| pair[0].key >= pair[1].key) {
557            return Err(FaultStateFormatError::UnsortedEntries);
558        }
559        let envelope = Self::new(checkpoint_digest, entries)?;
560        if &envelope.checksum()? != checksum {
561            return Err(FaultStateFormatError::ChecksumMismatch);
562        }
563        Ok(envelope)
564    }
565
566    /// Rejects state the current policy and checkpoint cannot own.
567    ///
568    /// # Errors
569    ///
570    /// Returns [`FaultStateFormatError::OrdinalAboveLimit`] for an ordinal the
571    /// configured retry limit cannot reach, [`FaultStateFormatError::
572    /// TooManyEntries`] above the configured capacity, and
573    /// [`FaultStateFormatError::CheckpointMismatch`] when the retained entries
574    /// belong to a superseded checkpoint.
575    pub fn validate_for(
576        &self,
577        retry_limit: RetryLimit,
578        state_limit: RetryStateLimit,
579        checkpoint_digest: &[u8; 32],
580    ) -> Result<(), FaultStateFormatError> {
581        if self.entries.len() > state_limit.get() as usize {
582            return Err(FaultStateFormatError::TooManyEntries {
583                max: state_limit.get() as usize,
584            });
585        }
586        if self
587            .entries
588            .iter()
589            .any(|entry| entry.ordinal().get() > retry_limit.get())
590        {
591            return Err(FaultStateFormatError::OrdinalAboveLimit {
592                max: retry_limit.get(),
593            });
594        }
595        if !self.entries.is_empty() && &self.checkpoint_digest != checkpoint_digest {
596            return Err(FaultStateFormatError::CheckpointMismatch);
597        }
598        Ok(())
599    }
600}
601
602impl Default for FaultStateEnvelope {
603    fn default() -> Self {
604        Self::empty()
605    }
606}
607
608/// A value-redacted durable fault-state format failure.
609///
610/// Every variant is corruption or an unsupported version. The runtime fails
611/// closed before any component work begins.
612#[derive(Clone, Copy, Debug, Eq, PartialEq)]
613#[non_exhaustive]
614pub enum FaultStateFormatError {
615    /// The stored format identifier or version is not format 1.
616    UnsupportedFormat,
617    /// The stored schema version is newer than this runtime understands.
618    UnsupportedSchemaVersion,
619    /// The payload was not canonical fault-state JSON.
620    Malformed,
621    /// One entry was not a valid fault-state entry object.
622    MalformedEntry,
623    /// A stored phase or category name is not a known enumeration value.
624    UnknownEnumeration,
625    /// The stored checksum does not cover the stored payload.
626    ChecksumMismatch,
627    /// The payload exceeded the durable byte ceiling.
628    TooLarge {
629        /// Maximum accepted canonical bytes.
630        max_bytes: usize,
631    },
632    /// The payload retained more keys than the accepted bound.
633    TooManyEntries {
634        /// Maximum accepted unresolved keys.
635        max: usize,
636    },
637    /// The payload retained the same retry key twice.
638    DuplicateKey,
639    /// The payload was not sorted by retry-key digest.
640    UnsortedEntries,
641    /// A retained ordinal is above the configured retry limit.
642    OrdinalAboveLimit {
643        /// Maximum re-invocations the policy allows.
644        max: u32,
645    },
646    /// The retained keys do not belong to the committed checkpoint.
647    CheckpointMismatch,
648}
649
650impl fmt::Display for FaultStateFormatError {
651    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
652        match self {
653            Self::UnsupportedFormat => formatter.write_str("fault-state format is unsupported"),
654            Self::UnsupportedSchemaVersion => {
655                formatter.write_str("fault-state schema version is unsupported")
656            }
657            Self::Malformed => formatter.write_str("fault state is malformed"),
658            Self::MalformedEntry => formatter.write_str("fault-state entry is malformed"),
659            Self::UnknownEnumeration => {
660                formatter.write_str("fault state contains an unknown enumeration value")
661            }
662            Self::ChecksumMismatch => formatter.write_str("fault-state checksum does not match"),
663            Self::TooLarge { max_bytes } => {
664                write!(formatter, "fault state exceeds {max_bytes} bytes")
665            }
666            Self::TooManyEntries { max } => {
667                write!(formatter, "fault state retains more than {max} keys")
668            }
669            Self::DuplicateKey => formatter.write_str("fault state repeats one retry key"),
670            Self::UnsortedEntries => formatter.write_str("fault state is not digest-sorted"),
671            Self::OrdinalAboveLimit { max } => {
672                write!(formatter, "fault state retains an ordinal above {max}")
673            }
674            Self::CheckpointMismatch => {
675                formatter.write_str("fault state belongs to a superseded checkpoint")
676            }
677        }
678    }
679}
680
681impl Error for FaultStateFormatError {}
682
683fn hex(bytes: &[u8; 32]) -> String {
684    let mut text = String::with_capacity(64);
685    for byte in bytes {
686        text.push(char::from_digit(u32::from(byte >> 4), 16).unwrap_or('0'));
687        text.push(char::from_digit(u32::from(byte & 0x0f), 16).unwrap_or('0'));
688    }
689    text
690}
691
692fn unhex(text: &str) -> Option<[u8; 32]> {
693    if text.len() != 64 {
694        return None;
695    }
696    let mut bytes = [0_u8; 32];
697    let raw = text.as_bytes();
698    for (index, slot) in bytes.iter_mut().enumerate() {
699        let high = char::from(raw[index * 2]).to_digit(16)?;
700        let low = char::from(raw[index * 2 + 1]).to_digit(16)?;
701        *slot = u8::try_from(high * 16 + low).ok()?;
702    }
703    Some(bytes)
704}
705
706/// Durable, bounded retry-reservation state for one step execution.
707///
708/// Implementations perform a compare-and-swap: a reservation is accepted only
709/// when its ordinal directly follows the persisted ordinal for the same key.
710/// The reservation must be durable before the runtime waits for backoff.
711pub trait FaultStateStore: Send + Sync {
712    /// Binds durable state to the step execution about to run.
713    ///
714    /// The runtime calls this once before the first chunk attempt when it
715    /// executes through a repository. A process-local store ignores it; a
716    /// durable store cannot read or write state before it is bound.
717    fn bind(
718        &self,
719        _context: ChunkTransactionContext,
720    ) -> BoxFuture<'_, Result<(), FaultStateError>> {
721        Box::pin(std::future::ready(Ok(())))
722    }
723
724    /// Returns the ordinal already reserved for `key`, when one exists.
725    fn reserved_ordinal(
726        &self,
727        key: RetryKey,
728    ) -> BoxFuture<'_, Result<Option<RetryOrdinal>, FaultStateError>>;
729
730    /// Commits one reservation, consuming its ordinal.
731    fn reserve(&self, reservation: RetryReservation) -> BoxFuture<'_, Result<(), FaultStateError>>;
732
733    /// Marks `key` resolved because its unit of work succeeded or was skipped.
734    ///
735    /// The key stays retained until the accepting chunk commits, because
736    /// uncommitted work may still replay.
737    fn resolve(&self, key: RetryKey) -> BoxFuture<'_, Result<(), FaultStateError>>;
738
739    /// Clears every resolved key in the commit that advances the checkpoint.
740    fn clear_resolved(&self) -> BoxFuture<'_, Result<(), FaultStateError>>;
741
742    /// Returns the number of retained unresolved keys.
743    fn unresolved(&self) -> BoxFuture<'_, Result<u32, FaultStateError>>;
744}
745
746#[derive(Clone, Copy, Debug)]
747struct RetryEntry {
748    ordinal: RetryOrdinal,
749    resolved: bool,
750}
751
752/// A bounded, process-local [`FaultStateStore`].
753///
754/// This implementation makes the reservation ordering executable without a
755/// database. It is not durable: a restart starts from an empty state, which the
756/// contract permits because a restart may invoke fewer retries than were
757/// reserved, never more.
758#[derive(Debug)]
759pub struct InMemoryFaultState {
760    limit: RetryStateLimit,
761    entries: Mutex<BTreeMap<RetryKey, RetryEntry>>,
762}
763
764impl InMemoryFaultState {
765    /// Constructs an empty bounded state.
766    #[must_use]
767    pub fn new(limit: RetryStateLimit) -> Self {
768        Self {
769            limit,
770            entries: Mutex::new(BTreeMap::new()),
771        }
772    }
773
774    fn with_entries<T>(&self, body: impl FnOnce(&mut BTreeMap<RetryKey, RetryEntry>) -> T) -> T {
775        let mut entries = self.entries.lock().unwrap_or_else(PoisonError::into_inner);
776        body(&mut entries)
777    }
778}
779
780impl FaultStateStore for InMemoryFaultState {
781    fn reserved_ordinal(
782        &self,
783        key: RetryKey,
784    ) -> BoxFuture<'_, Result<Option<RetryOrdinal>, FaultStateError>> {
785        let result = self.with_entries(|entries| entries.get(&key).map(|entry| entry.ordinal));
786        Box::pin(std::future::ready(Ok(result)))
787    }
788
789    fn reserve(&self, reservation: RetryReservation) -> BoxFuture<'_, Result<(), FaultStateError>> {
790        let limit = self.limit;
791        let result = self.with_entries(|entries| {
792            let expected = entries
793                .get(&reservation.key())
794                .map_or(RetryOrdinal::INITIAL, |entry| entry.ordinal)
795                .checked_next()
796                .map_err(|_| FaultStateError::StaleReservation)?;
797            if reservation.ordinal() != expected {
798                return Err(FaultStateError::StaleReservation);
799            }
800            let unresolved = entries.values().filter(|entry| !entry.resolved).count();
801            let is_new = !entries.contains_key(&reservation.key());
802            if is_new && unresolved >= limit.get() as usize {
803                return Err(FaultStateError::CapacityExhausted { max: limit.get() });
804            }
805            entries.insert(
806                reservation.key(),
807                RetryEntry {
808                    ordinal: reservation.ordinal(),
809                    resolved: false,
810                },
811            );
812            Ok(())
813        });
814        Box::pin(std::future::ready(result))
815    }
816
817    fn resolve(&self, key: RetryKey) -> BoxFuture<'_, Result<(), FaultStateError>> {
818        self.with_entries(|entries| {
819            if let Some(entry) = entries.get_mut(&key) {
820                entry.resolved = true;
821            }
822        });
823        Box::pin(std::future::ready(Ok(())))
824    }
825
826    fn clear_resolved(&self) -> BoxFuture<'_, Result<(), FaultStateError>> {
827        self.with_entries(|entries| entries.retain(|_, entry| !entry.resolved));
828        Box::pin(std::future::ready(Ok(())))
829    }
830
831    fn unresolved(&self) -> BoxFuture<'_, Result<u32, FaultStateError>> {
832        let count = self.with_entries(|entries| entries.values().filter(|e| !e.resolved).count());
833        let result = u32::try_from(count).map_err(|_| FaultStateError::Unavailable);
834        Box::pin(std::future::ready(result))
835    }
836}
837
838/// The validated fault-tolerance capability installed on a chunk step.
839///
840/// The bundle owns the policy, the injected monotonic sleeper, the reservation
841/// store, and the declared delivery mode. Capabilities are validated at
842/// construction so a statically impossible combination cannot reach user work.
843///
844/// ```
845/// use std::sync::Arc;
846/// use std::time::Duration;
847///
848/// use oxide_batch::{
849///     BackoffOutcome, BackoffPolicy, BackoffSleeper, BoxFuture, ChunkDeliveryMode,
850///     ClassifierRevision, FailureCategory, FaultAction, FaultClassifier, FaultPhase, FaultPolicy,
851///     FaultRule, FaultRuntime, InMemoryFaultState, RetryLimit, RetryStateLimit, SkipLimit,
852///     StopToken,
853/// };
854///
855/// struct ImmediateSleeper;
856///
857/// impl BackoffSleeper for ImmediateSleeper {
858///     fn sleep<'a>(
859///         &'a self,
860///         _delay: Duration,
861///         stop: &'a StopToken,
862///     ) -> BoxFuture<'a, BackoffOutcome> {
863///         let stopped = stop.is_stop_requested();
864///         Box::pin(async move {
865///             if stopped { BackoffOutcome::Stopped } else { BackoffOutcome::Elapsed }
866///         })
867///     }
868/// }
869///
870/// let policy = FaultPolicy::new(
871///     FaultClassifier::new(
872///         ClassifierRevision::new("import_v1")?,
873///         [FaultRule::new(
874///             FaultPhase::Write,
875///             FailureCategory::Timeout,
876///             FaultAction::retry(),
877///         )?],
878///     )?,
879///     RetryLimit::new(2)?,
880///     RetryStateLimit::new(16)?,
881///     SkipLimit::NONE,
882///     BackoffPolicy::fixed(Duration::from_millis(10))?,
883/// )?;
884/// let state = Arc::new(InMemoryFaultState::new(policy.retry_state_limit()));
885/// let runtime = FaultRuntime::new(
886///     policy,
887///     Arc::new(ImmediateSleeper),
888///     state,
889///     ChunkDeliveryMode::AtLeastOnce,
890/// )?;
891/// assert_eq!(runtime.delivery_mode(), ChunkDeliveryMode::AtLeastOnce);
892/// # Ok::<(), Box<dyn std::error::Error>>(())
893/// ```
894#[derive(Clone)]
895pub struct FaultRuntime {
896    policy: Arc<FaultPolicy>,
897    sleeper: Arc<dyn BackoffSleeper>,
898    state: Arc<dyn FaultStateStore>,
899    delivery_mode: ChunkDeliveryMode,
900}
901
902impl FaultRuntime {
903    /// Validates and installs the fault-tolerance capability.
904    ///
905    /// # Errors
906    ///
907    /// Returns [`crate::FaultPolicyError::CommitSafeSkipUnsupported`] when the
908    /// policy accepts a commit-safe skip that the declared delivery mode cannot
909    /// commit atomically.
910    pub fn new(
911        policy: FaultPolicy,
912        sleeper: Arc<dyn BackoffSleeper>,
913        state: Arc<dyn FaultStateStore>,
914        delivery_mode: ChunkDeliveryMode,
915    ) -> Result<Self, crate::FaultPolicyError> {
916        policy.validate_capabilities(matches!(
917            delivery_mode,
918            ChunkDeliveryMode::AtomicSameResource
919        ))?;
920        Ok(Self {
921            policy: Arc::new(policy),
922            sleeper,
923            state,
924            delivery_mode,
925        })
926    }
927
928    /// Borrows the validated step policy.
929    #[must_use]
930    pub fn policy(&self) -> &FaultPolicy {
931        &self.policy
932    }
933
934    /// Borrows the injected monotonic sleeper.
935    #[must_use]
936    pub fn sleeper(&self) -> &dyn BackoffSleeper {
937        self.sleeper.as_ref()
938    }
939
940    /// Borrows the reservation store.
941    #[must_use]
942    pub fn state(&self) -> &dyn FaultStateStore {
943        self.state.as_ref()
944    }
945
946    /// Returns the delivery mode declared for this step.
947    #[must_use]
948    pub const fn delivery_mode(&self) -> ChunkDeliveryMode {
949        self.delivery_mode
950    }
951}
952
953impl fmt::Debug for FaultRuntime {
954    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
955        formatter
956            .debug_struct("FaultRuntime")
957            .field("retry_limit", &self.policy.retry_limit())
958            .field("retry_state_limit", &self.policy.retry_state_limit())
959            .field("skip_limit", &self.policy.skip_limit())
960            .field("backoff", &self.policy.backoff().kind())
961            .field("delivery_mode", &self.delivery_mode)
962            .finish_non_exhaustive()
963    }
964}
965
966/// The committed fault-tolerance totals one step attempt inherits.
967///
968/// A restart copies the latest committed totals to the new attempt, so a
969/// bounded limit spans every attempt of one job instance rather than resetting
970/// per process.
971#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)]
972pub struct FaultProgress {
973    retries: RetryCounts,
974    skips: SkipCounts,
975    rollbacks: u64,
976    no_rollbacks: u64,
977}
978
979impl FaultProgress {
980    /// The totals a first attempt of a new job instance inherits.
981    pub const NONE: Self = Self {
982        retries: RetryCounts::ZERO,
983        skips: SkipCounts::ZERO,
984        rollbacks: 0,
985        no_rollbacks: 0,
986    };
987
988    /// Constructs a committed total snapshot.
989    #[must_use]
990    pub const fn new(
991        retries: RetryCounts,
992        skips: SkipCounts,
993        rollbacks: u64,
994        no_rollbacks: u64,
995    ) -> Self {
996        Self {
997            retries,
998            skips,
999            rollbacks,
1000            no_rollbacks,
1001        }
1002    }
1003
1004    /// Returns inherited per-phase reserved retry counts.
1005    #[must_use]
1006    pub const fn retries(self) -> RetryCounts {
1007        self.retries
1008    }
1009
1010    /// Returns inherited per-phase committed skip counts.
1011    #[must_use]
1012    pub const fn skips(self) -> SkipCounts {
1013        self.skips
1014    }
1015
1016    /// Returns inherited acknowledged framework rollback decisions.
1017    #[must_use]
1018    pub const fn rollbacks(self) -> u64 {
1019        self.rollbacks
1020    }
1021
1022    /// Returns inherited commits that accepted a commit-safe skip.
1023    #[must_use]
1024    pub const fn no_rollbacks(self) -> u64 {
1025        self.no_rollbacks
1026    }
1027}
1028
1029/// Durable retry attempts, kept distinct per phase.
1030///
1031/// A count records one reserved retry ordinal, not one component call.
1032#[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)]
1033pub struct RetryCounts {
1034    read: u64,
1035    process: u64,
1036    write: u64,
1037}
1038
1039impl RetryCounts {
1040    /// Counts inherited by a first attempt.
1041    pub const ZERO: Self = Self {
1042        read: 0,
1043        process: 0,
1044        write: 0,
1045    };
1046
1047    /// Constructs per-phase retry counts.
1048    #[must_use]
1049    pub const fn new(read: u64, process: u64, write: u64) -> Self {
1050        Self {
1051            read,
1052            process,
1053            write,
1054        }
1055    }
1056
1057    /// Returns reserved read retries.
1058    #[must_use]
1059    pub const fn read(self) -> u64 {
1060        self.read
1061    }
1062
1063    /// Returns reserved process retries.
1064    #[must_use]
1065    pub const fn process(self) -> u64 {
1066        self.process
1067    }
1068
1069    /// Returns reserved write retries.
1070    #[must_use]
1071    pub const fn write(self) -> u64 {
1072        self.write
1073    }
1074
1075    /// Returns the counts after one reserved retry in `phase`.
1076    ///
1077    /// A phase that cannot reserve a retry leaves the counts unchanged.
1078    #[must_use]
1079    pub const fn increment(mut self, phase: FaultPhase) -> Self {
1080        let counter = match phase {
1081            FaultPhase::Read => &mut self.read,
1082            FaultPhase::Process => &mut self.process,
1083            FaultPhase::Write => &mut self.write,
1084            _ => return self,
1085        };
1086        *counter = counter.saturating_add(1);
1087        self
1088    }
1089}
oxide_batch/fault_state.rs

oxide_batch/
fault_state.rs
