oxide_batch/
chunk.rs

1//! Runtime-neutral chunk component and transaction-enlistment contracts.
2
3use std::error::Error;
4use std::fmt;
5use std::future::Future;
6
7use crate::{
8    BoxFuture, Checkpoint, ChunkCounts, ComponentStateEnvelope, ExecutionContext, FailureCategory,
9    FaultProgress, JobExecutionId, SkipCounts, StepExecutionId, StopToken,
10};
11
12/// Borrowed call state for a reader.
13#[derive(Clone, Copy, Debug)]
14pub struct ReadContext<'a> {
15    stop: &'a StopToken,
16}
17
18impl<'a> ReadContext<'a> {
19    /// Constructs a reader call scope.
20    #[must_use]
21    pub const fn new(stop: &'a StopToken) -> Self {
22        Self { stop }
23    }
24
25    /// Borrows the cooperative stop token.
26    #[must_use]
27    pub const fn stop_token(self) -> &'a StopToken {
28        self.stop
29    }
30}
31
32/// One item-reader call outcome.
33#[derive(Clone, Debug, Eq, PartialEq)]
34#[non_exhaustive]
35pub enum ReadOutcome<I> {
36    /// The next input item.
37    Item(I),
38    /// The source is exhausted normally.
39    EndOfInput,
40    /// Cooperative stop was observed before another item was produced.
41    Stopped,
42}
43
44/// A stateful asynchronous item source.
45#[diagnostic::on_unimplemented(
46    message = "`{Self}` is not an OxideBatch item reader for `{I}`",
47    label = "this component cannot read `{I}`",
48    note = "implement `ItemReader<{I}>` with `async fn read(&mut self, context: ReadContext<'_>)`",
49    note = "the returned future must be `Send`: do not hold a non-`Send` value across an await"
50)]
51pub trait ItemReader<I>: Send {
52    /// Reads at most one item while borrowing the reader and call scope.
53    fn read<'a>(
54        &'a mut self,
55        context: ReadContext<'a>,
56    ) -> impl Future<Output = Result<ReadOutcome<I>, ReaderError>> + Send + 'a;
57}
58
59/// Borrowed call state for a processor.
60#[derive(Clone, Copy, Debug)]
61pub struct ProcessContext<'a> {
62    stop: &'a StopToken,
63}
64
65impl<'a> ProcessContext<'a> {
66    /// Constructs a processor call scope.
67    #[must_use]
68    pub const fn new(stop: &'a StopToken) -> Self {
69        Self { stop }
70    }
71
72    /// Borrows the cooperative stop token.
73    #[must_use]
74    pub const fn stop_token(self) -> &'a StopToken {
75        self.stop
76    }
77}
78
79/// One item-processor call outcome.
80#[derive(Clone, Debug, Eq, PartialEq)]
81#[non_exhaustive]
82pub enum ProcessOutcome<O> {
83    /// An output item was produced.
84    Item(O),
85    /// The input was intentionally filtered without producing output.
86    Filtered,
87    /// Cooperative stop was observed before output was produced.
88    Stopped,
89}
90
91/// A shared asynchronous item transformer.
92#[diagnostic::on_unimplemented(
93    message = "`{Self}` is not an OxideBatch item processor from `{I}` to `{O}`",
94    label = "this component cannot process `{I}` into `{O}`",
95    note = "implement `ItemProcessor<{I}, {O}>` with `async fn process(&self, item: &{I}, context: ProcessContext<'_>)`",
96    note = "a processor is shared across the chunk, so it takes `&self` and must be `Sync`"
97)]
98pub trait ItemProcessor<I, O>: Send + Sync {
99    /// Processes one borrowed item.
100    fn process<'a>(
101        &'a self,
102        item: &'a I,
103        context: ProcessContext<'a>,
104    ) -> impl Future<Output = Result<ProcessOutcome<O>, ProcessorError>> + Send + 'a;
105}
106
107/// A stable bound-value type for enlisted business statements.
108#[derive(Clone, Copy, Eq, PartialEq)]
109pub struct BusinessValue<'a>(BusinessValueInner<'a>);
110
111#[derive(Clone, Copy, Eq, PartialEq)]
112enum BusinessValueInner<'a> {
113    Text(&'a str),
114    Bytes(&'a [u8]),
115    I64(i64),
116    Bool(bool),
117    Null,
118}
119
120impl<'a> BusinessValue<'a> {
121    /// Constructs a borrowed UTF-8 value.
122    #[must_use]
123    pub const fn text(value: &'a str) -> Self {
124        Self(BusinessValueInner::Text(value))
125    }
126
127    /// Constructs a borrowed byte value.
128    #[must_use]
129    pub const fn bytes(value: &'a [u8]) -> Self {
130        Self(BusinessValueInner::Bytes(value))
131    }
132
133    /// Constructs a signed integer value.
134    #[must_use]
135    pub const fn i64(value: i64) -> Self {
136        Self(BusinessValueInner::I64(value))
137    }
138
139    /// Constructs a boolean value.
140    #[must_use]
141    pub const fn boolean(value: bool) -> Self {
142        Self(BusinessValueInner::Bool(value))
143    }
144
145    /// Constructs a database null value.
146    #[must_use]
147    pub const fn null() -> Self {
148        Self(BusinessValueInner::Null)
149    }
150
151    /// Returns the stable value kind.
152    #[must_use]
153    pub const fn kind(self) -> BusinessValueKind {
154        match self.0 {
155            BusinessValueInner::Text(_) => BusinessValueKind::Text,
156            BusinessValueInner::Bytes(_) => BusinessValueKind::Bytes,
157            BusinessValueInner::I64(_) => BusinessValueKind::I64,
158            BusinessValueInner::Bool(_) => BusinessValueKind::Bool,
159            BusinessValueInner::Null => BusinessValueKind::Null,
160        }
161    }
162
163    /// Borrows the UTF-8 value when present.
164    #[must_use]
165    pub const fn as_text(self) -> Option<&'a str> {
166        match self.0 {
167            BusinessValueInner::Text(value) => Some(value),
168            _ => None,
169        }
170    }
171
172    /// Borrows the byte value when present.
173    #[must_use]
174    pub const fn as_bytes(self) -> Option<&'a [u8]> {
175        match self.0 {
176            BusinessValueInner::Bytes(value) => Some(value),
177            _ => None,
178        }
179    }
180
181    /// Returns the signed integer when present.
182    #[must_use]
183    pub const fn as_i64(self) -> Option<i64> {
184        match self.0 {
185            BusinessValueInner::I64(value) => Some(value),
186            _ => None,
187        }
188    }
189
190    /// Returns the boolean when present.
191    #[must_use]
192    pub const fn as_bool(self) -> Option<bool> {
193        match self.0 {
194            BusinessValueInner::Bool(value) => Some(value),
195            _ => None,
196        }
197    }
198}
199
200impl fmt::Debug for BusinessValue<'_> {
201    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
202        formatter
203            .debug_struct("BusinessValue")
204            .field("kind", &self.kind())
205            .field("value", &"<redacted>")
206            .finish()
207    }
208}
209
210/// Stable discriminator for a bound business value.
211#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
212#[non_exhaustive]
213pub enum BusinessValueKind {
214    /// UTF-8 text.
215    Text,
216    /// Arbitrary bytes.
217    Bytes,
218    /// Signed 64-bit integer.
219    I64,
220    /// Boolean.
221    Bool,
222    /// Database null.
223    Null,
224}
225
226/// A parameterized business write borrowed for one transaction call.
227pub struct BusinessStatement<'a> {
228    text: &'a str,
229    values: &'a [BusinessValue<'a>],
230}
231
232impl<'a> BusinessStatement<'a> {
233    /// Constructs a statement from SQL text and separately bound values.
234    ///
235    /// The `PostgreSQL` adapter binds every value; it never interpolates these
236    /// values into `text`.
237    #[must_use]
238    pub const fn new(text: &'a str, values: &'a [BusinessValue<'a>]) -> Self {
239        Self { text, values }
240    }
241
242    /// Borrows statement text for the authorized database adapter.
243    #[must_use]
244    pub const fn text(&self) -> &'a str {
245        self.text
246    }
247
248    /// Borrows separately bound values.
249    #[must_use]
250    pub const fn values(&self) -> &'a [BusinessValue<'a>] {
251        self.values
252    }
253}
254
255impl fmt::Debug for BusinessStatement<'_> {
256    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
257        formatter
258            .debug_struct("BusinessStatement")
259            .field("text", &"<redacted>")
260            .field("value_count", &self.values.len())
261            .finish()
262    }
263}
264
265/// Successful effect from one enlisted business statement.
266#[derive(Clone, Copy, Debug, Eq, PartialEq)]
267pub struct BusinessWriteResult {
268    rows_affected: u64,
269}
270
271impl BusinessWriteResult {
272    /// Constructs a result reported by a transaction adapter.
273    #[must_use]
274    pub const fn new(rows_affected: u64) -> Self {
275        Self { rows_affected }
276    }
277
278    /// Returns the database-reported affected-row count.
279    #[must_use]
280    pub const fn rows_affected(self) -> u64 {
281        self.rows_affected
282    }
283}
284
285/// OxideBatch-owned port for the currently enlisted business transaction.
286///
287/// The durable adapter owns the concrete transaction and lends this port to a
288/// writer only for the call. `SQLx` pool, connection, row, error, and transaction
289/// types do not cross this boundary.
290pub trait BusinessTransaction: Send {
291    /// Executes one parameterized business write.
292    fn execute<'a>(
293        &'a mut self,
294        statement: BusinessStatement<'a>,
295    ) -> BoxFuture<'a, Result<BusinessWriteResult, BusinessTransactionError>>;
296}
297
298/// Evidence returned after one chunk transaction is known to have committed.
299///
300/// The receipt owns the durable checkpoint and execution context so
301/// post-commit observers cannot borrow adapter-internal transaction state.
302#[derive(Clone, Debug, Eq, PartialEq)]
303pub struct ChunkCommitReceipt {
304    checkpoint: Checkpoint,
305    execution_context: ExecutionContext,
306}
307
308impl ChunkCommitReceipt {
309    /// Constructs committed durable-state evidence.
310    #[must_use]
311    pub const fn new(checkpoint: Checkpoint, execution_context: ExecutionContext) -> Self {
312        Self {
313            checkpoint,
314            execution_context,
315        }
316    }
317
318    /// Borrows the committed reader checkpoint.
319    #[must_use]
320    pub const fn checkpoint(&self) -> &Checkpoint {
321        &self.checkpoint
322    }
323
324    /// Borrows the committed execution context.
325    #[must_use]
326    pub const fn execution_context(&self) -> &ExecutionContext {
327        &self.execution_context
328    }
329}
330
331/// Stable, payload-redacted chunk-transaction failure.
332#[derive(Clone, Copy, Debug, Eq, PartialEq)]
333#[non_exhaustive]
334pub enum ChunkTransactionError {
335    /// The operation is known not to have committed.
336    NotCommitted,
337    /// The adapter cannot determine whether commit reached durable storage.
338    CommitOutcomeUnknown,
339    /// The adapter does not support committing component state.
340    ///
341    /// Returned only when [`ItemStream`](crate::ItemStream)s are registered
342    /// against a [`ChunkTransaction`] whose
343    /// [`commit_with_component_state`](ChunkTransaction::commit_with_component_state)
344    /// override is missing; an adapter that supports no streams at all never
345    /// observes this variant.
346    ComponentStateUnsupported,
347}
348
349impl fmt::Display for ChunkTransactionError {
350    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
351        formatter.write_str(match self {
352            Self::NotCommitted => "chunk transaction did not commit",
353            Self::CommitOutcomeUnknown => "chunk transaction commit outcome is unknown",
354            Self::ComponentStateUnsupported => {
355                "chunk transaction does not support committing component state"
356            }
357        })
358    }
359}
360
361impl Error for ChunkTransactionError {}
362
363/// The fault-tolerance progress one chunk commit makes authoritative.
364///
365/// The values are deltas contributed by a single chunk attempt. A durable
366/// adapter adds them to the committed totals it read when the transaction
367/// began, so replaying an uncommitted chunk after a crash cannot double-count.
368#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)]
369pub struct ChunkFaultProgress {
370    skips: SkipCounts,
371    no_rollbacks: u64,
372}
373
374impl ChunkFaultProgress {
375    /// The progress of a chunk that accepted no skip.
376    pub const NONE: Self = Self {
377        skips: SkipCounts::ZERO,
378        no_rollbacks: 0,
379    };
380
381    /// Constructs the delta contributed by one chunk attempt.
382    #[must_use]
383    pub const fn new(skips: SkipCounts, no_rollbacks: u64) -> Self {
384        Self {
385            skips,
386            no_rollbacks,
387        }
388    }
389
390    /// Returns the per-phase skips this chunk accepted.
391    #[must_use]
392    pub const fn skips(self) -> SkipCounts {
393        self.skips
394    }
395
396    /// Returns the accepted commit-safe skips this chunk committed.
397    #[must_use]
398    pub const fn no_rollbacks(self) -> u64 {
399        self.no_rollbacks
400    }
401}
402
403/// One adapter-owned transaction for a bounded chunk attempt.
404///
405/// The runtime invokes the writer while this value is open, then commits the
406/// supplied checked counters or rolls the transaction back. Implementations
407/// keep database-driver and serialization types private.
408pub trait ChunkTransaction: Send {
409    /// Reborrows an enlisted business transaction when the selected delivery
410    /// mode supports same-resource atomicity.
411    fn business_transaction(&mut self) -> Option<&mut dyn BusinessTransaction>;
412
413    /// Commits business work and the supplied progress, returning the durable
414    /// checkpoint and context that became authoritative.
415    ///
416    /// `fault` carries the skips this chunk accepted. A durable adapter also
417    /// clears the retained fault state of the superseded checkpoint generation
418    /// in this transaction, so a skip, its counters, and the checkpoint that
419    /// makes it authoritative commit or roll back together.
420    fn commit(
421        &mut self,
422        counts: ChunkCounts,
423        fault: ChunkFaultProgress,
424    ) -> BoxFuture<'_, Result<ChunkCommitReceipt, ChunkTransactionError>>;
425
426    /// Commits business work, progress, and per-namespace
427    /// [`ItemStream`](crate::ItemStream) candidate state as one atomic
428    /// outcome.
429    ///
430    /// The default delegates to [`commit`](Self::commit) and ignores
431    /// `component_state` when it is empty, so every existing implementor is
432    /// unaffected. An adapter that supports [`ItemStream`](crate::ItemStream)
433    /// registration overrides this method to bind each candidate envelope
434    /// into the same transaction/statement as the existing commit, never a
435    /// second connection or transaction. A caller must not register a stream
436    /// against a transaction manager whose commit does not override this: the
437    /// default rejects a nonempty `component_state` with
438    /// [`ChunkTransactionError::ComponentStateUnsupported`] rather than
439    /// silently dropping it.
440    fn commit_with_component_state<'a>(
441        &'a mut self,
442        counts: ChunkCounts,
443        fault: ChunkFaultProgress,
444        component_state: &'a [ComponentStateEnvelope],
445    ) -> BoxFuture<'a, Result<ChunkCommitReceipt, ChunkTransactionError>> {
446        if component_state.is_empty() {
447            self.commit(counts, fault)
448        } else {
449            Box::pin(std::future::ready(Err(
450                ChunkTransactionError::ComponentStateUnsupported,
451            )))
452        }
453    }
454
455    /// Rolls back all provisional work in this chunk attempt.
456    fn rollback(&mut self) -> BoxFuture<'_, Result<(), ChunkTransactionError>>;
457}
458
459/// Repository execution identity for one launched chunk transaction.
460///
461/// Standalone chunk execution has no durable execution graph and therefore
462/// uses [`ChunkTransactionManager::begin`]. The repository-backed launcher
463/// supplies this context through [`ChunkTransactionManager::begin_for`].
464#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
465pub struct ChunkTransactionContext {
466    job_execution_id: JobExecutionId,
467    step_execution_id: StepExecutionId,
468}
469
470impl ChunkTransactionContext {
471    /// Constructs a durable chunk-transaction scope.
472    #[must_use]
473    pub const fn new(job_execution_id: JobExecutionId, step_execution_id: StepExecutionId) -> Self {
474        Self {
475            job_execution_id,
476            step_execution_id,
477        }
478    }
479
480    /// Returns the enclosing job execution.
481    #[must_use]
482    pub const fn job_execution_id(self) -> JobExecutionId {
483        self.job_execution_id
484    }
485
486    /// Returns the step execution whose progress is committed.
487    #[must_use]
488    pub const fn step_execution_id(self) -> StepExecutionId {
489        self.step_execution_id
490    }
491}
492
493/// Committed step progress one chunk-step attempt inherits.
494///
495/// A restart resumes bounded policy limits and stable retry-key identity from
496/// the durable state its attempt inherited, so a retry budget is not refilled
497/// by restarting the process.
498#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)]
499pub struct InheritedStepProgress {
500    read_ordinal: u64,
501    checkpoint_digest: [u8; 32],
502    fault: FaultProgress,
503}
504
505impl InheritedStepProgress {
506    /// The progress a standalone or first-attempt chunk step inherits.
507    pub const NONE: Self = Self {
508        read_ordinal: 0,
509        checkpoint_digest: [0; 32],
510        fault: FaultProgress::NONE,
511    };
512
513    /// Constructs inherited progress from durable step state.
514    #[must_use]
515    pub const fn new(read_ordinal: u64, checkpoint_digest: [u8; 32], fault: FaultProgress) -> Self {
516        Self {
517            read_ordinal,
518            checkpoint_digest,
519            fault,
520        }
521    }
522
523    /// Returns the stable reader ordinal the next chunk continues from.
524    #[must_use]
525    pub const fn read_ordinal(self) -> u64 {
526        self.read_ordinal
527    }
528
529    /// Returns the digest of the last committed checkpoint.
530    ///
531    /// Retry keys are derived from this generation, so an inherited digest
532    /// makes a reserved ordinal resumable after restart.
533    #[must_use]
534    pub const fn checkpoint_digest(self) -> [u8; 32] {
535        self.checkpoint_digest
536    }
537
538    /// Returns the inherited committed fault-tolerance totals.
539    #[must_use]
540    pub const fn fault(self) -> FaultProgress {
541        self.fault
542    }
543}
544
545/// Begins isolated adapter-owned chunk transactions.
546pub trait ChunkTransactionManager: Send + Sync {
547    /// Starts one transaction for a bounded chunk attempt.
548    fn begin(&self)
549    -> BoxFuture<'_, Result<Box<dyn ChunkTransaction + '_>, ChunkTransactionError>>;
550
551    /// Returns the durable progress this step attempt inherits.
552    ///
553    /// The default suits managers without durable state. A durable adapter
554    /// overrides it and fails closed rather than restarting bounded policy
555    /// limits from zero.
556    fn inherited_progress(
557        &self,
558        _context: ChunkTransactionContext,
559    ) -> BoxFuture<'_, Result<InheritedStepProgress, ChunkTransactionError>> {
560        Box::pin(std::future::ready(Ok(InheritedStepProgress::NONE)))
561    }
562
563    /// Returns the last committed [`ItemStream`](crate::ItemStream) envelopes
564    /// this step attempt inherits, one per registered namespace.
565    ///
566    /// The default suits managers without durable component state. A durable
567    /// adapter overrides it and fails closed rather than silently reporting
568    /// no committed state for a namespace that has one.
569    fn inherited_component_state(
570        &self,
571        _context: ChunkTransactionContext,
572    ) -> BoxFuture<'_, Result<Vec<ComponentStateEnvelope>, ChunkTransactionError>> {
573        Box::pin(std::future::ready(Ok(Vec::new())))
574    }
575
576    /// Starts one transaction bound to a durable repository execution.
577    ///
578    /// The default preserves managers that do not need repository identity.
579    /// Durable adapters override this method and reject unbound
580    /// [`Self::begin`] calls rather than guessing an execution target.
581    fn begin_for(
582        &self,
583        _context: ChunkTransactionContext,
584    ) -> BoxFuture<'_, Result<Box<dyn ChunkTransaction + '_>, ChunkTransactionError>> {
585        self.begin()
586    }
587}
588
589/// Stable, value-redacted enlisted-transaction failure.
590#[derive(Clone, Copy, Debug, Eq, PartialEq)]
591#[non_exhaustive]
592pub enum BusinessTransactionError {
593    /// The transaction cannot safely continue after an infrastructure failure.
594    Infrastructure,
595    /// The statement was rejected permanently.
596    Rejected,
597    /// Cooperative cancellation interrupted the operation.
598    Cancelled,
599}
600
601impl fmt::Display for BusinessTransactionError {
602    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
603        formatter.write_str(match self {
604            Self::Infrastructure => "business transaction infrastructure failed",
605            Self::Rejected => "business transaction statement was rejected",
606            Self::Cancelled => "business transaction operation was cancelled",
607        })
608    }
609}
610
611impl Error for BusinessTransactionError {}
612
613/// Borrowed call state for a writer.
614pub struct WriteContext<'a> {
615    stop: &'a StopToken,
616    transaction: Option<&'a mut dyn BusinessTransaction>,
617}
618
619impl<'a> WriteContext<'a> {
620    /// Constructs a non-enlisted writer call scope.
621    #[must_use]
622    pub const fn non_transactional(stop: &'a StopToken) -> Self {
623        Self {
624            stop,
625            transaction: None,
626        }
627    }
628
629    /// Constructs a writer call enlisted in an OxideBatch-owned transaction.
630    #[must_use]
631    pub fn enlisted(stop: &'a StopToken, transaction: &'a mut dyn BusinessTransaction) -> Self {
632        Self {
633            stop,
634            transaction: Some(transaction),
635        }
636    }
637
638    /// Borrows the cooperative stop token.
639    #[must_use]
640    pub const fn stop_token(&self) -> &'a StopToken {
641        self.stop
642    }
643
644    /// Reborrows the enlisted transaction when one is present.
645    #[must_use]
646    pub fn transaction(&mut self) -> Option<&mut (dyn BusinessTransaction + 'a)> {
647        self.transaction.as_deref_mut()
648    }
649
650    /// Returns whether this call participates in the chunk transaction.
651    #[must_use]
652    pub const fn is_enlisted(&self) -> bool {
653        self.transaction.is_some()
654    }
655}
656
657impl fmt::Debug for WriteContext<'_> {
658    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
659        formatter
660            .debug_struct("WriteContext")
661            .field("stop_requested", &self.stop.is_stop_requested())
662            .field("enlisted", &self.transaction.is_some())
663            .finish()
664    }
665}
666
667/// One item-writer call outcome.
668#[derive(Clone, Copy, Debug, Eq, PartialEq)]
669#[non_exhaustive]
670pub enum WriteOutcome {
671    /// Every supplied item was accepted by the writer.
672    Written,
673    /// Cooperative stop was observed before the batch was accepted.
674    Stopped,
675}
676
677/// A shared asynchronous batch writer.
678#[diagnostic::on_unimplemented(
679    message = "`{Self}` is not an OxideBatch item writer for `{I}`",
680    label = "this component cannot write `{I}`",
681    note = "implement `ItemWriter<{I}>` with `async fn write(&self, items: &[{I}], context: WriteContext<'_>)`",
682    note = "a writer that delegates to another component must tie its lifetimes: `async fn write<'a>(&'a self, items: &'a [{I}], context: WriteContext<'a>)`"
683)]
684pub trait ItemWriter<I>: Send + Sync {
685    /// Writes one borrowed, nonempty batch.
686    ///
687    /// A durable step supplies an enlisted transaction in `context`, borrowed
688    /// for exactly this call.
689    fn write<'a>(
690        &'a self,
691        items: &'a [I],
692        context: WriteContext<'a>,
693    ) -> impl Future<Output = Result<WriteOutcome, WriterError>> + Send + 'a;
694}
695
696/// The dyn-compatible mirror of the public component contract.
697///
698/// Nothing here is exported. Its only implementors are the blanket impls
699/// below, so no external crate can observe or depend on this shape, and the
700/// single `Box::pin` per call is the only boxing this item-component erasure
701/// boundary introduces. Other ADR-0002 extension points — tasklets,
702/// transactions, listeners, and the rest — remain boxed by design and are
703/// unaffected by and unrelated to this module.
704mod sealed {
705    use super::{ItemProcessor, ItemReader, ItemWriter};
706    use crate::{
707        BoxFuture, ProcessContext, ProcessOutcome, ProcessorError, ReadContext, ReadOutcome,
708        ReaderError, WriteContext, WriteOutcome, WriterError,
709    };
710
711    pub trait ReaderObject<I>: Send {
712        fn read_boxed<'a>(
713            &'a mut self,
714            context: ReadContext<'a>,
715        ) -> BoxFuture<'a, Result<ReadOutcome<I>, ReaderError>>;
716    }
717
718    impl<I, R: ItemReader<I>> ReaderObject<I> for R {
719        fn read_boxed<'a>(
720            &'a mut self,
721            context: ReadContext<'a>,
722        ) -> BoxFuture<'a, Result<ReadOutcome<I>, ReaderError>> {
723            Box::pin(self.read(context))
724        }
725    }
726
727    pub trait ProcessorObject<I, O>: Send + Sync {
728        fn process_boxed<'a>(
729            &'a self,
730            item: &'a I,
731            context: ProcessContext<'a>,
732        ) -> BoxFuture<'a, Result<ProcessOutcome<O>, ProcessorError>>;
733    }
734
735    impl<I, O, P: ItemProcessor<I, O>> ProcessorObject<I, O> for P {
736        fn process_boxed<'a>(
737            &'a self,
738            item: &'a I,
739            context: ProcessContext<'a>,
740        ) -> BoxFuture<'a, Result<ProcessOutcome<O>, ProcessorError>> {
741            Box::pin(self.process(item, context))
742        }
743    }
744
745    pub trait WriterObject<I>: Send + Sync {
746        fn write_boxed<'a>(
747            &'a self,
748            items: &'a [I],
749            context: WriteContext<'a>,
750        ) -> BoxFuture<'a, Result<WriteOutcome, WriterError>>;
751    }
752
753    impl<I, W: ItemWriter<I>> WriterObject<I> for W {
754        fn write_boxed<'a>(
755            &'a self,
756            items: &'a [I],
757            context: WriteContext<'a>,
758        ) -> BoxFuture<'a, Result<WriteOutcome, WriterError>> {
759            Box::pin(self.write(items, context))
760        }
761    }
762}
763
764/// A reader of any concrete type, behind one dynamic dispatch.
765///
766/// Constructing one is the explicit, greppable point where the pipeline stops
767/// being monomorphized and starts paying a boxed future per call.
768pub struct BoxedReader<I>(Box<dyn sealed::ReaderObject<I>>);
769
770impl<I> BoxedReader<I> {
771    /// Erases a concrete reader.
772    pub fn new<R: ItemReader<I> + 'static>(reader: R) -> Self {
773        Self(Box::new(reader))
774    }
775}
776
777impl<I> ItemReader<I> for BoxedReader<I> {
778    fn read<'a>(
779        &'a mut self,
780        context: ReadContext<'a>,
781    ) -> impl Future<Output = Result<ReadOutcome<I>, ReaderError>> + Send + 'a {
782        self.0.read_boxed(context)
783    }
784}
785
786/// A processor of any concrete type, behind one dynamic dispatch.
787pub struct BoxedProcessor<I, O>(Box<dyn sealed::ProcessorObject<I, O>>);
788
789impl<I, O> BoxedProcessor<I, O> {
790    /// Erases a concrete processor.
791    pub fn new<P: ItemProcessor<I, O> + 'static>(processor: P) -> Self {
792        Self(Box::new(processor))
793    }
794}
795
796impl<I, O> ItemProcessor<I, O> for BoxedProcessor<I, O> {
797    fn process<'a>(
798        &'a self,
799        item: &'a I,
800        context: ProcessContext<'a>,
801    ) -> impl Future<Output = Result<ProcessOutcome<O>, ProcessorError>> + Send + 'a {
802        self.0.process_boxed(item, context)
803    }
804}
805
806/// A writer of any concrete type, behind one dynamic dispatch.
807pub struct BoxedWriter<I>(Box<dyn sealed::WriterObject<I>>);
808
809impl<I> BoxedWriter<I> {
810    /// Erases a concrete writer.
811    pub fn new<W: ItemWriter<I> + 'static>(writer: W) -> Self {
812        Self(Box::new(writer))
813    }
814}
815
816impl<I> ItemWriter<I> for BoxedWriter<I> {
817    fn write<'a>(
818        &'a self,
819        items: &'a [I],
820        context: WriteContext<'a>,
821    ) -> impl Future<Output = Result<WriteOutcome, WriterError>> + Send + 'a {
822        self.0.write_boxed(items, context)
823    }
824}
825
826/// Read-only evidence passed after the chunk transaction commits.
827#[derive(Clone, Copy, Debug)]
828pub struct ChunkCompletionContext<'a> {
829    checkpoint: &'a Checkpoint,
830    execution_context: &'a ExecutionContext,
831    counts: ChunkCounts,
832    stop: &'a StopToken,
833}
834
835impl<'a> ChunkCompletionContext<'a> {
836    /// Constructs committed chunk evidence.
837    #[must_use]
838    pub const fn new(
839        checkpoint: &'a Checkpoint,
840        execution_context: &'a ExecutionContext,
841        counts: ChunkCounts,
842        stop: &'a StopToken,
843    ) -> Self {
844        Self {
845            checkpoint,
846            execution_context,
847            counts,
848            stop,
849        }
850    }
851
852    /// Borrows the committed checkpoint.
853    #[must_use]
854    pub const fn checkpoint(self) -> &'a Checkpoint {
855        self.checkpoint
856    }
857
858    /// Borrows the committed execution context.
859    #[must_use]
860    pub const fn execution_context(self) -> &'a ExecutionContext {
861        self.execution_context
862    }
863
864    /// Returns the committed chunk counts.
865    #[must_use]
866    pub const fn counts(self) -> ChunkCounts {
867        self.counts
868    }
869
870    /// Borrows the cooperative stop token.
871    #[must_use]
872    pub const fn stop_token(self) -> &'a StopToken {
873        self.stop
874    }
875}
876
877/// Post-commit acknowledgement from a chunk-completion component.
878#[derive(Clone, Copy, Debug, Eq, PartialEq)]
879#[non_exhaustive]
880pub enum ChunkCompletionOutcome {
881    /// The component observed and acknowledged the durable commit.
882    Acknowledged,
883    /// Stop was observed after commit; the commit remains authoritative.
884    StoppedAfterCommit,
885}
886
887/// An asynchronous observer called only after a durable chunk commit.
888pub trait ChunkCompletion: Send + Sync {
889    /// Acknowledges committed state without becoming a correctness authority.
890    fn after_commit<'a>(
891        &'a self,
892        context: ChunkCompletionContext<'a>,
893    ) -> BoxFuture<'a, Result<ChunkCompletionOutcome, ChunkCompletionError>>;
894}
895
896macro_rules! component_error {
897    (
898        $name:ident,
899        $message:literal
900        $(, $field:ident : $field_type:ty = $field_default:expr, $field_docs:literal)* $(,)?
901    ) => {
902        #[doc = $message]
903        ///
904        /// The adapter translates its own typed error into a stable
905        /// [`FailureCategory`] at this boundary. The payload, display text, and
906        /// source chain are dropped, so classification never inspects them.
907        #[derive(Clone, Copy, Debug, Eq, PartialEq)]
908        pub struct $name {
909            category: FailureCategory,
910            $(
911                #[doc = $field_docs]
912                $field: $field_type,
913            )*
914        }
915
916        impl $name {
917            /// Constructs a value-redacted [`FailureCategory::UserComponent`]
918            /// failure.
919            #[must_use]
920            pub const fn new() -> Self {
921                Self {
922                    category: FailureCategory::UserComponent,
923                    $($field: $field_default,)*
924                }
925            }
926
927            /// Constructs a failure that declares its own stable category.
928            ///
929            /// A category that is not policy-eligible fails closed: the fault
930            /// is never retried or skipped.
931            #[must_use]
932            pub const fn with_category(category: FailureCategory) -> Self {
933                Self {
934                    category,
935                    $($field: $field_default,)*
936                }
937            }
938
939            /// Classifies an arbitrary user error without retaining its
940            /// payload or display text.
941            #[must_use]
942            pub fn from_error(error: impl Error + Send + Sync + 'static) -> Self {
943                drop(error);
944                Self::new()
945            }
946
947            /// Returns the stable category supplied by the adapter.
948            #[must_use]
949            pub const fn category(self) -> FailureCategory {
950                self.category
951            }
952        }
953
954        impl Default for $name {
955            fn default() -> Self {
956                Self::new()
957            }
958        }
959
960        impl fmt::Display for $name {
961            fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
962                formatter.write_str($message)
963            }
964        }
965
966        impl Error for $name {}
967    };
968}
969
970component_error!(
971    ReaderError,
972    "item reader failed",
973    checkpoint_advanced: bool = false,
974    "Whether the reader proved its checkpoint moved past one failed input.",
975);
976component_error!(ProcessorError, "item processor failed");
977component_error!(
978    WriterError,
979    "item writer failed",
980    rolled_back_output: Option<usize> = None,
981    "The located, known-rolled-back output index, when the writer supplied one.",
982);
983component_error!(ChunkCompletionError, "chunk completion callback failed");
984
985impl ReaderError {
986    /// Records that the reader moved its checkpoint past exactly one failed
987    /// input.
988    ///
989    /// A read skip requires this proof. Without it a repeated failure at the
990    /// same position fails the step instead of skipping forever.
991    #[must_use]
992    pub const fn with_checkpoint_advanced(mut self, advanced: bool) -> Self {
993        self.checkpoint_advanced = advanced;
994        self
995    }
996
997    /// Returns whether the reader proved forward checkpoint progress.
998    #[must_use]
999    pub const fn has_checkpoint_advanced(self) -> bool {
1000        self.checkpoint_advanced
1001    }
1002}
1003
1004impl WriterError {
1005    /// Records that the batch is known to have rolled back and identifies the
1006    /// single failed output by its zero-based index in the supplied batch.
1007    ///
1008    /// A write skip requires this evidence. An unlocated, partially visible, or
1009    /// ambiguous write cannot be skipped.
1010    #[must_use]
1011    pub const fn with_rolled_back_output(mut self, index: usize) -> Self {
1012        self.rolled_back_output = Some(index);
1013        self
1014    }
1015
1016    /// Returns the located failed output index, when the writer supplied one.
1017    #[must_use]
1018    pub const fn rolled_back_output(self) -> Option<usize> {
1019        self.rolled_back_output
1020    }
1021}
oxide_batch/chunk.rs

oxide_batch/
chunk.rs
