Struct AdaptiveCompletionPolicy 

pub struct AdaptiveCompletionPolicy { /* private fields */ }

A bounded, restart-safe adaptive completion policy.

The policy adjusts its target chunk size toward target_duration, clamped to bounds, based on the most recently committed chunk’s observed duration. Install it via crate::ChunkStep::with_completion_policy (directly, or nested inside a CompositeCompletionPolicy at any depth): stream_registrations reports its namespaced ItemStream registration under the same identity returned by identity, bound to this exact instance, so the confirmed target survives restart through the same commit-boundary mechanism as any other component state – never a second persistence path, and never two registrations that could drift onto different instances.

Restart safety

A process crash before a chunk commits leaves the previously committed target authoritative: ItemStream::open restores exactly that target from the durable envelope, never a value a still-open, not-yet-committed attempt only speculated.

Same-process rollback safety

ItemStream::update never mutates the confirmed target: it only computes a candidate from the confirmed baseline and this attempt’s freshly observed duration, storing it as a separate, speculative pending value. That candidate is promoted to confirmed – becoming visible to CompletionPolicy::is_complete – only when CompletionPolicy::end_chunk observes ChunkAttemptOutcome::Committed; every other outcome discards it. A rolled-back attempt that gets replayed therefore always recomputes its candidate from the same unmodified confirmed baseline, never a corrupted or partially-applied one, and a rollback can never leave a speculative target looking authoritative. The recomputed candidate itself is not guaranteed identical to the discarded one: it is a function of this replay’s freshly observed duration, which genuinely can differ from the rolled-back attempt’s (only a fully deterministic clock, as a test might inject, makes the two identical).

Implementations

impl AdaptiveCompletionPolicy

pub fn new( identity: ComponentStreamIdentity, bounds: AdaptiveBounds, target_duration: ChunkTimeThreshold, clock: Arc<dyn Clock>, ) -> Arc<Self>

Constructs an adaptive policy starting at bounds’s minimum target.

Returns Arc<Self> rather than Self: the policy holds a weak self-reference so stream_registrations can bind its persisted state to this exact instance regardless of whether it ends up installed directly or nested inside a CompositeCompletionPolicy.

pub const fn identity(&self) -> &ComponentStreamIdentity

Returns the namespace this policy’s persisted decision is registered under.

pub fn current_target(&self) -> ChunkSize

Returns the current confirmed target: the authoritative decision as of the last chunk this process has observed commit (or restored at ItemStream::open). Never a speculative, not-yet-committed candidate.

Trait Implementations

impl CompletionPolicy for AdaptiveCompletionPolicy

fn begin_chunk(&self)

Resets any per-attempt state at the start of a new chunk attempt.

fn is_complete(&self, items_read: ChunkCount) -> bool

Reports whether the chunk should stop accepting more items.

fn end_chunk(&self, outcome: ChunkAttemptOutcome)

Observes the terminal outcome of the chunk attempt this policy’s most recent begin_chunk started.

fn fingerprint(&self) -> String

Returns a canonical, deterministic description of this policy’s restart-relevant configuration, hashed into the owning chunk definition’s fingerprint so a configuration change that alters completion semantics is never mistaken for the same definition across a restart.

fn stream_registrations( &self, ) -> Vec<(ComponentStreamIdentity, Arc<BoxedStream>, StreamStateContract)>

Returns this policy’s own required ItemStream registration(s), bound to the exact same instance this policy makes completion decisions from – never a second, independently constructed instance that could drift out of sync with it.

impl ItemStream for AdaptiveCompletionPolicy

async fn open( &self, context: StreamOpenContext<'_>, ) -> Result<StreamOpenOutcome, StreamOpenError>

Restores last committed state, or begins initial execution.

async fn update( &self, _context: StreamUpdateContext<'_>, ) -> Result<ComponentStateEnvelope, StreamUpdateError>

Produces the candidate component-state envelope for the chunk about to commit.

async fn close( &self, _context: StreamCloseContext<'_>, ) -> Result<StreamCloseOutcome, StreamCloseError>

Runs once for every stream that opened successfully in this step attempt.