aster_forge_runtime/
buffered.rs

1//! Caller-owned, failure-aware buffering. Construction never starts work.
2//!
3//! Records remain in memory until persistence acknowledges them. Sink errors pause the affected
4//! lane; only explicitly retryable failures can be resumed. Dropping the owner cancels/aborts its
5//! worker, while `stop` additionally joins it before dependencies may close.
6
7use std::collections::VecDeque;
8use std::future::Future;
9use std::panic::AssertUnwindSafe;
10use std::pin::Pin;
11use std::sync::{Arc, Mutex, MutexGuard};
12
13use futures::FutureExt;
14use tokio::sync::{Mutex as AsyncMutex, Notify, mpsc, oneshot};
15use tokio::task::JoinHandle;
16use tokio::time::Instant;
17use tokio_util::sync::CancellationToken;
18
19/// Bounded in-memory buffering policy. Zero capacities normalize to one.
20#[derive(Debug, Clone, Copy)]
21pub struct BufferedBatchConfig {
22    /// Maximum unacknowledged batch records, including the batch currently being written.
23    pub queue_capacity: usize,
24    /// Immediate-flush threshold and maximum records passed to a batch sink call.
25    pub batch_size: usize,
26    /// Maximum delay from the first queued record before attempting a partial batch.
27    pub delayed_flush_after: std::time::Duration,
28}
29
30impl BufferedBatchConfig {
31    /// Creates a policy; the writer normalizes zero capacity and batch size to one.
32    #[must_use]
33    pub const fn new(
34        queue_capacity: usize,
35        batch_size: usize,
36        delayed_flush_after: std::time::Duration,
37    ) -> Self {
38        Self {
39            queue_capacity,
40            batch_size,
41            delayed_flush_after,
42        }
43    }
44}
45
46/// Whether an unacknowledged write can safely be replayed.
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub enum WriteFailureClass {
49    /// Persistence guarantees no records committed; an explicit retry is safe.
50    Retryable,
51    /// Persistence rejected the records; retrying unchanged records is inappropriate.
52    Permanent,
53    /// Some or all records may have committed. Reconcile before any replay.
54    UnknownCommit,
55}
56
57/// Original persistence error, or a worker-side interruption with unknown outcome.
58#[derive(Debug)]
59pub enum WriteFailureCause<E> {
60    /// The original error, including its native classification.
61    Persistence(Arc<E>),
62    /// The sink or worker panicked before completing outcome bookkeeping.
63    Panicked,
64    /// The sink future was cancelled before acknowledgment.
65    Cancelled,
66    /// Accepted overflow was cancelled while waiting; persistence was never invoked.
67    NotAttempted,
68}
69
70/// A failure retained alongside the unconfirmed records.
71#[derive(Debug)]
72pub struct WriteFailure<E> {
73    /// Replay eligibility.
74    pub class: WriteFailureClass,
75    /// Original failure cause.
76    pub cause: WriteFailureCause<E>,
77}
78
79impl<E> Clone for WriteFailure<E> {
80    fn clone(&self) -> Self {
81        Self {
82            class: self.class,
83            cause: match &self.cause {
84                WriteFailureCause::Persistence(error) => {
85                    WriteFailureCause::Persistence(error.clone())
86                }
87                WriteFailureCause::Panicked => WriteFailureCause::Panicked,
88                WriteFailureCause::Cancelled => WriteFailureCause::Cancelled,
89                WriteFailureCause::NotAttempted => WriteFailureCause::NotAttempted,
90            },
91        }
92    }
93}
94
95impl<E> WriteFailure<E> {
96    /// Classifies an error without erasing its concrete type.
97    #[must_use]
98    pub fn new(class: WriteFailureClass, error: E) -> Self {
99        Self {
100            class,
101            cause: WriteFailureCause::Persistence(Arc::new(error)),
102        }
103    }
104}
105
106/// Explicit admission policy when the bounded batch queue is full.
107#[derive(Debug, Clone, Copy, PartialEq, Eq)]
108pub enum OverflowPolicy {
109    /// Reject and return ownership of the record.
110    Reject,
111    /// Use a separate single-record persistence lane, bounded to one outstanding record.
112    DirectWrite,
113}
114
115/// Writer lifecycle. A stopped writer cannot be restarted.
116#[derive(Debug, Clone, Copy, PartialEq, Eq)]
117pub enum BufferedWriterState {
118    /// Construction has completed without creating a worker.
119    Created,
120    /// Admission and background persistence are enabled.
121    Running,
122    /// Admission is closed; accepted records can still be persisted.
123    Draining,
124    /// The worker has terminated; unconfirmed records require explicit recovery.
125    Stopped,
126}
127
128/// Successful admission does not imply batch durability.
129#[derive(Debug, Clone, Copy, PartialEq, Eq)]
130pub enum RecordAdmission {
131    /// Accepted into memory only.
132    Queued,
133    /// Only this overflow record was acknowledged by persistence.
134    DirectPersisted,
135}
136
137/// Admission failure. Rejected records are returned; accepted failed direct writes stay in memory.
138#[derive(Debug)]
139pub enum RecordError<T, E> {
140    /// Admission is closed or startup has not completed.
141    NotRunning(T),
142    /// The configured queue/direct-lane bound was reached.
143    Full(T),
144    /// An accepted direct write failed and remains unconfirmed in the writer.
145    DirectFailed(WriteFailure<E>),
146}
147
148/// Flush failure; inspect the snapshot and unconfirmed records before recovery.
149#[derive(Debug)]
150pub enum BufferFlushError<E> {
151    /// Start must be called explicitly before flushing records.
152    NotStarted,
153    /// The worker has stopped with records still awaiting reconciliation.
154    Stopped,
155    /// A retained batch or direct record failed.
156    Write(WriteFailure<E>),
157    /// The caller's deadline elapsed. No durability is implied.
158    Deadline,
159}
160
161/// Low-cardinality events delivered with `try_send`; persistence never calls observer code.
162#[derive(Debug, Clone, Copy, PartialEq, Eq)]
163pub enum BufferedWriterEvent {
164    /// A record was accepted into the batch queue.
165    Queued,
166    /// One overflow record was acknowledged.
167    DirectPersisted,
168    /// One batch was acknowledged.
169    BatchPersisted,
170    /// A sink failed or was interrupted.
171    Failed,
172    /// Admission was rejected.
173    Rejected,
174    /// The worker has exited.
175    Stopped,
176}
177
178/// Observable per-instance state. Counts include in-flight, unacknowledged records.
179#[derive(Debug)]
180pub struct BufferedWriterSnapshot<E> {
181    /// Current admission/worker lifecycle.
182    pub state: BufferedWriterState,
183    /// Unacknowledged batch records, including any in-flight chunk.
184    pub queued: usize,
185    /// Unacknowledged overflow records (zero or one).
186    pub direct_pending: usize,
187    /// Total acknowledged records across both lanes.
188    pub persisted: u64,
189    /// Acknowledged overflow records only; does not imply batch durability.
190    pub direct_persisted: u64,
191    /// Events dropped because the bounded observer channel was full or closed.
192    pub observer_events_dropped: u64,
193    /// Failure blocking the front chunk; originals remain in queue order.
194    pub batch_failure: Option<WriteFailure<E>>,
195    /// Front records affected by `batch_failure`; later queued records were not attempted.
196    pub batch_failure_records: usize,
197    /// Failure blocking the retained overflow record.
198    pub direct_failure: Option<WriteFailure<E>>,
199}
200
201/// Drain result. Once returned, the worker is joined and database dependencies may close.
202#[derive(Debug)]
203pub struct BufferStopReport<E> {
204    /// Whether this stop caller exhausted its drain budget.
205    pub deadline_elapsed: bool,
206    /// Final state after joining the worker; remaining records have no confirmed durability.
207    pub snapshot: BufferedWriterSnapshot<E>,
208}
209
210struct Direct<T, E> {
211    item: T,
212    reply: Option<oneshot::Sender<Result<RecordAdmission, WriteFailure<E>>>>,
213    failure: Option<WriteFailure<E>>,
214}
215
216struct State<T, E> {
217    lifecycle: BufferedWriterState,
218    queue: VecDeque<T>,
219    oldest: Option<Instant>,
220    direct: Option<Direct<T, E>>,
221    failure: Option<WriteFailure<E>>,
222    failure_records: usize,
223    // Highest batch acknowledgment requested by flush/retry. Only the sole worker advances
224    // persisted counters; originals (including in-flight items) remain queued until then.
225    flush_through: u64,
226    persisted: u64,
227    direct_persisted: u64,
228    observer_dropped: u64,
229}
230
231impl<T, E> State<T, E> {
232    fn batch_persisted(&self) -> u64 {
233        self.persisted - self.direct_persisted
234    }
235
236    fn batch_watermark(&self) -> u64 {
237        // Capture admission using batch progress alone: a direct acknowledgment cannot pay for
238        // an unacknowledged queued record. Failed records also remain part of this watermark.
239        self.batch_persisted() + self.queue.len() as u64
240    }
241
242    fn next_batch_size(&self, maximum: usize) -> usize {
243        // Forced partial batches stop at the captured watermark. Later admissions retain their
244        // own threshold/delay instead of being pulled into an earlier flush transaction.
245        let remaining = self.flush_through.saturating_sub(self.batch_persisted());
246        if remaining == 0 {
247            maximum
248        } else {
249            maximum.min(usize::try_from(remaining).unwrap_or(usize::MAX))
250        }
251    }
252}
253
254type Sink<T, E> = dyn Fn(Vec<T>) -> Pin<Box<dyn Future<Output = Result<(), WriteFailure<E>>> + Send>>
255    + Send
256    + Sync;
257
258struct Core<T, E> {
259    config: BufferedBatchConfig,
260    overflow: OverflowPolicy,
261    state: Mutex<State<T, E>>,
262    wake: Notify,
263    changed: Notify,
264    cancel: CancellationToken,
265    sink: Box<Sink<T, E>>,
266    events: Option<mpsc::Sender<BufferedWriterEvent>>,
267}
268
269impl<T, E> Core<T, E> {
270    fn state(&self) -> MutexGuard<'_, State<T, E>> {
271        self.state
272            .lock()
273            .unwrap_or_else(std::sync::PoisonError::into_inner)
274    }
275
276    fn event(&self, state: &mut State<T, E>, event: BufferedWriterEvent) {
277        if let Some(events) = &self.events
278            && events.try_send(event).is_err()
279        {
280            state.observer_dropped = state.observer_dropped.saturating_add(1);
281        }
282    }
283}
284
285/// Explicitly started and stopped buffering with one owned worker and bounded overflow.
286///
287/// The sink must be cancellation-safe: interruption means an unknown commit, not a safe retry.
288/// `T: Clone` permits retaining originals while an owned copy is passed to persistence.
289/// Observers receive a bounded channel created by the caller; callbacks run outside persistence.
290/// No automatic retry occurs, including for `Retryable` failures.
291pub struct ManagedBatchWriter<T, E> {
292    core: Arc<Core<T, E>>,
293    worker: AsyncMutex<Option<JoinHandle<()>>>,
294}
295
296impl<T, E> Drop for ManagedBatchWriter<T, E> {
297    fn drop(&mut self) {
298        self.core.cancel.cancel();
299        if let Some(worker) = self.worker.get_mut().take() {
300            worker.abort();
301        }
302    }
303}
304
305impl<T: Clone + Send + 'static, E: Send + Sync + 'static> ManagedBatchWriter<T, E> {
306    /// Constructs an idle owner. Call `start` from the runtime startup phase.
307    #[must_use]
308    pub fn new<F, Fut>(
309        config: BufferedBatchConfig,
310        overflow: OverflowPolicy,
311        events: Option<mpsc::Sender<BufferedWriterEvent>>,
312        sink: F,
313    ) -> Self
314    where
315        F: Fn(Vec<T>) -> Fut + Send + Sync + 'static,
316        Fut: Future<Output = Result<(), WriteFailure<E>>> + Send + 'static,
317    {
318        let config = BufferedBatchConfig {
319            queue_capacity: config.queue_capacity.max(1),
320            batch_size: config.batch_size.max(1),
321            ..config
322        };
323        Self {
324            core: Arc::new(Core {
325                config,
326                overflow,
327                events,
328                state: Mutex::new(State {
329                    lifecycle: BufferedWriterState::Created,
330                    queue: VecDeque::new(),
331                    oldest: None,
332                    direct: None,
333                    failure: None,
334                    failure_records: 0,
335                    flush_through: 0,
336                    persisted: 0,
337                    direct_persisted: 0,
338                    observer_dropped: 0,
339                }),
340                wake: Notify::new(),
341                changed: Notify::new(),
342                cancel: CancellationToken::new(),
343                sink: Box::new(move |items| Box::pin(sink(items))),
344            }),
345            worker: AsyncMutex::new(None),
346        }
347    }
348
349    /// Starts exactly once. Returns the current state if startup has already been attempted.
350    ///
351    /// # Errors
352    /// Returns the current state unless this is the first start from `Created`.
353    pub async fn start(&self) -> Result<(), BufferedWriterState> {
354        let mut worker = self.worker.lock().await;
355        let mut state = self.core.state();
356        if state.lifecycle != BufferedWriterState::Created {
357            return Err(state.lifecycle);
358        }
359        state.lifecycle = BufferedWriterState::Running;
360        let core = self.core.clone();
361        *worker = Some(tokio::spawn(async move {
362            // Sink panics are handled per write. Also supervise the entire worker: record Clone
363            // implementations and other code outside the sink must not strand admission waiters.
364            if AssertUnwindSafe(run(core.clone()))
365                .catch_unwind()
366                .await
367                .is_err()
368            {
369                core.fail_worker(WriteFailureCause::Panicked);
370            }
371        }));
372        Ok(())
373    }
374
375    /// Admits a record or returns it when rejected. Cancellation of a direct waiter does not
376    /// cancel its accepted record: the owned worker still persists or retains it.
377    ///
378    /// # Errors
379    /// Rejection returns the record. A direct persistence failure retains it in the owner.
380    pub async fn record(&self, item: T) -> Result<RecordAdmission, RecordError<T, E>> {
381        let reply = {
382            let mut state = self.core.state();
383            if state.lifecycle != BufferedWriterState::Running {
384                self.core.event(&mut state, BufferedWriterEvent::Rejected);
385                return Err(RecordError::NotRunning(item));
386            }
387            if state.queue.len() < self.core.config.queue_capacity {
388                if state.queue.is_empty() {
389                    state.oldest = Some(Instant::now());
390                }
391                state.queue.push_back(item);
392                self.core.event(&mut state, BufferedWriterEvent::Queued);
393                self.core.wake.notify_one();
394                return Ok(RecordAdmission::Queued);
395            }
396            if self.core.overflow == OverflowPolicy::Reject || state.direct.is_some() {
397                self.core.event(&mut state, BufferedWriterEvent::Rejected);
398                return Err(RecordError::Full(item));
399            }
400            let (send, receive) = oneshot::channel();
401            state.direct = Some(Direct {
402                item,
403                reply: Some(send),
404                failure: None,
405            });
406            receive
407        };
408        self.core.wake.notify_one();
409        reply
410            .await
411            .unwrap_or_else(|_| {
412                Err(WriteFailure {
413                    class: WriteFailureClass::UnknownCommit,
414                    cause: WriteFailureCause::Cancelled,
415                })
416            })
417            .map_err(RecordError::DirectFailed)
418    }
419
420    /// Atomically stops admission. Accepted records remain owned and can still flush.
421    pub fn begin_drain(&self) {
422        let mut state = self.core.state();
423        if state.lifecycle != BufferedWriterState::Stopped {
424            state.lifecycle = BufferedWriterState::Draining;
425        }
426        self.core.wake.notify_one();
427    }
428
429    /// Flushes records admitted before this future's first poll, using a caller deadline.
430    /// Later admissions cannot extend the captured scope or satisfy it through another lane.
431    /// Dropping this future leaves accepted writes owned by the worker. A failed lane does not
432    /// prevent draining other records in this scope. An expired deadline starts no new flush work.
433    /// An already acknowledged scope succeeds even after the deadline. Normal threshold/timer
434    /// writes and other flush callers proceed independently of this future's scope and deadline.
435    ///
436    /// # Errors
437    /// Returns a lifecycle error, the original in-scope write failure, or deadline exhaustion.
438    pub async fn flush(&self, deadline: Instant) -> Result<(), BufferFlushError<E>> {
439        let (batch_target, direct_target) = {
440            let state = self.core.state();
441            if state.lifecycle == BufferedWriterState::Created {
442                return Err(BufferFlushError::NotStarted);
443            }
444            (
445                state.batch_watermark(),
446                state.direct_persisted + u64::from(state.direct.is_some()),
447            )
448        };
449        loop {
450            let changed = self.core.changed.notified();
451            tokio::pin!(changed);
452            changed.as_mut().enable();
453            {
454                let mut state = self.core.state();
455                let batch_pending = state.batch_persisted() < batch_target;
456                let direct_pending = state.direct_persisted < direct_target;
457                if !batch_pending && !direct_pending {
458                    return Ok(());
459                }
460                let work_pending = (batch_pending && state.failure.is_none())
461                    || (direct_pending
462                        && state.direct.as_ref().is_some_and(|d| d.failure.is_none()));
463                if !work_pending
464                    && let Some(failure) =
465                        state
466                            .failure
467                            .as_ref()
468                            .filter(|_| batch_pending)
469                            .or_else(|| {
470                                state
471                                    .direct
472                                    .as_ref()
473                                    .filter(|_| direct_pending)
474                                    .and_then(|d| d.failure.as_ref())
475                            })
476                {
477                    return Err(BufferFlushError::Write(failure.clone()));
478                }
479                if state.lifecycle == BufferedWriterState::Stopped {
480                    return Err(BufferFlushError::Stopped);
481                }
482                // Check before scheduling too: Tokio timeout polls a ready inner future first,
483                // so repeated progress notifications alone must not bypass the caller deadline.
484                if Instant::now() >= deadline {
485                    return Err(BufferFlushError::Deadline);
486                }
487                if batch_pending {
488                    state.flush_through = state.flush_through.max(batch_target);
489                }
490            }
491            self.core.wake.notify_one();
492            if tokio::time::timeout_at(deadline, changed).await.is_err() {
493                return Err(BufferFlushError::Deadline);
494            }
495        }
496    }
497
498    /// Clears only failures classified as safe to retry. Unknown/permanent records stay blocked.
499    pub fn retry_failed(&self) {
500        let mut state = self.core.state();
501        if state.lifecycle == BufferedWriterState::Stopped {
502            return;
503        }
504        if state
505            .failure
506            .as_ref()
507            .is_some_and(|f| f.class == WriteFailureClass::Retryable)
508        {
509            state.failure = None;
510            state.failure_records = 0;
511            state.flush_through = state.flush_through.max(state.batch_watermark());
512        }
513        if let Some(direct) = &mut state.direct
514            && direct
515                .failure
516                .as_ref()
517                .is_some_and(|f| f.class == WriteFailureClass::Retryable)
518        {
519            direct.failure = None;
520        }
521        self.core.wake.notify_one();
522    }
523
524    /// Stops admission, drains until deadline/failure, then cancels and joins the owned worker.
525    /// Repeated/concurrent calls share the same handle; cancelling a stop caller leaves the handle
526    /// available for another caller. Dependencies must close only after this future returns.
527    pub async fn stop(&self, deadline: Instant) -> BufferStopReport<E> {
528        self.begin_drain();
529        let result = self.flush(deadline).await;
530        self.core.cancel.cancel();
531        let mut worker = self.worker.lock().await;
532        if let Some(task) = worker.as_mut()
533            && let Err(error) = task.await
534        {
535            self.core.fail_worker(if error.is_panic() {
536                WriteFailureCause::Panicked
537            } else {
538                WriteFailureCause::Cancelled
539            });
540        }
541        *worker = None;
542        {
543            let mut state = self.core.state();
544            state.lifecycle = BufferedWriterState::Stopped;
545        }
546        BufferStopReport {
547            deadline_elapsed: matches!(result, Err(BufferFlushError::Deadline)),
548            snapshot: self.snapshot(),
549        }
550    }
551
552    /// Captures independent state and counters, without invoking observer code.
553    #[must_use]
554    pub fn snapshot(&self) -> BufferedWriterSnapshot<E> {
555        let state = self.core.state();
556        BufferedWriterSnapshot {
557            state: state.lifecycle,
558            queued: state.queue.len(),
559            direct_pending: usize::from(state.direct.is_some()),
560            persisted: state.persisted,
561            direct_persisted: state.direct_persisted,
562            observer_events_dropped: state.observer_dropped,
563            batch_failure: state.failure.clone(),
564            batch_failure_records: state.failure_records,
565            direct_failure: state.direct.as_ref().and_then(|d| d.failure.clone()),
566        }
567    }
568
569    /// Copies all unacknowledged records for reconciliation; never implicitly replays them.
570    /// Stop first to obtain a stable recovery snapshot.
571    #[must_use]
572    pub fn unconfirmed(&self) -> Vec<T> {
573        let state = self.core.state();
574        state
575            .queue
576            .iter()
577            .chain(state.direct.iter().map(|d| &d.item))
578            .cloned()
579            .collect()
580    }
581}
582
583async fn run<T: Clone + Send + 'static, E: Send + Sync + 'static>(core: Arc<Core<T, E>>) {
584    // Alternate ready lanes so a flood of overflow records cannot starve a queued batch.
585    let mut prefer_batch = false;
586    loop {
587        let wake = core.wake.notified();
588        tokio::pin!(wake);
589        wake.as_mut().enable();
590        if core.cancel.is_cancelled() {
591            break;
592        }
593        let (work, due) = {
594            let state = core.state();
595            let due = state
596                .oldest
597                .and_then(|at| at.checked_add(core.config.delayed_flush_after));
598            let batch_ready = !state.queue.is_empty()
599                && state.failure.is_none()
600                && (state.batch_persisted() < state.flush_through
601                    || state.lifecycle == BufferedWriterState::Draining
602                    || state.queue.len() >= core.config.batch_size
603                    || due.is_some_and(|at| at <= Instant::now()));
604            if let Some(direct) = &state.direct
605                && direct.failure.is_none()
606                && !(prefer_batch && batch_ready)
607            {
608                (Some((vec![direct.item.clone()], true)), due)
609            } else if batch_ready {
610                (
611                    Some((
612                        state
613                            .queue
614                            .iter()
615                            .take(state.next_batch_size(core.config.batch_size))
616                            .cloned()
617                            .collect::<Vec<_>>(),
618                        false,
619                    )),
620                    due,
621                )
622            } else {
623                (None, if state.failure.is_none() { due } else { None })
624            }
625        };
626        let Some((items, direct)) = work else {
627            tokio::select! {
628                biased;
629                () = core.cancel.cancelled() => break,
630                () = &mut wake => {},
631                () = async { match due { Some(at) => tokio::time::sleep_until(at).await, None => std::future::pending().await } } => {},
632            }
633            continue;
634        };
635        prefer_batch = direct;
636        let count = items.len();
637        let result = tokio::select! {
638            biased;
639            () = core.cancel.cancelled() => Err(WriteFailure { class: WriteFailureClass::UnknownCommit, cause: WriteFailureCause::Cancelled }),
640            result = AssertUnwindSafe(async { (core.sink)(items).await }).catch_unwind() => {
641                result.unwrap_or_else(|_| Err(WriteFailure { class: WriteFailureClass::UnknownCommit, cause: WriteFailureCause::Panicked }))
642            }
643        };
644        core.complete(direct, count, &result);
645        core.changed.notify_waiters();
646        // An immediately-ready sink does not necessarily consume Tokio's cooperative budget.
647        // Yield periodically so the owner's deadline and cancellation tasks can run as well.
648        tokio::task::consume_budget().await;
649    }
650    core.finish();
651}
652
653impl<T, E> Core<T, E> {
654    // An unexpected worker exit has no reliable write boundary. Preserve existing native
655    // failures, conservatively quarantine other records, and wake every accepted direct waiter.
656    fn fail_worker(&self, cause: WriteFailureCause<E>) {
657        let failure = WriteFailure {
658            class: WriteFailureClass::UnknownCommit,
659            cause,
660        };
661        {
662            let mut state = self.state();
663            if !state.queue.is_empty() && state.failure.is_none() {
664                state.failure = Some(failure.clone());
665                state.failure_records = state.queue.len();
666            }
667            if let Some(direct) = &mut state.direct {
668                if direct.failure.is_none() {
669                    direct.failure = Some(failure.clone());
670                }
671                if let Some(reply) = direct.reply.take() {
672                    let _ = reply.send(Err(failure));
673                }
674            }
675            self.event(&mut state, BufferedWriterEvent::Failed);
676        }
677        self.finish();
678    }
679
680    // The worker is the sole consumer. Originals are removed only after acknowledgment, while
681    // producers may append under the same short-held state lock during sink I/O.
682    fn complete(&self, direct: bool, count: usize, result: &Result<(), WriteFailure<E>>) {
683        let mut state = self.state();
684        if direct {
685            if let Some(mut pending) = state.direct.take() {
686                if let Some(reply) = pending.reply.take() {
687                    let _ = reply.send(
688                        result
689                            .as_ref()
690                            .map(|()| RecordAdmission::DirectPersisted)
691                            .map_err(Clone::clone),
692                    );
693                }
694                if let Err(failure) = result {
695                    pending.failure = Some(failure.clone());
696                    state.direct = Some(pending);
697                } else {
698                    state.persisted += 1;
699                    state.direct_persisted += 1;
700                }
701            }
702        } else if let Err(failure) = result {
703            state.failure = Some(failure.clone());
704            state.failure_records = count;
705        } else {
706            state.queue.drain(..count);
707            state.persisted += count as u64;
708            if state.queue.is_empty() {
709                state.oldest = None;
710            }
711        }
712        self.event(
713            &mut state,
714            if result.is_err() {
715                BufferedWriterEvent::Failed
716            } else if direct {
717                BufferedWriterEvent::DirectPersisted
718            } else {
719                BufferedWriterEvent::BatchPersisted
720            },
721        );
722    }
723
724    // Wake every direct/flush waiter on exit. An overflow slot not yet attempted is safe to
725    // recover; an interrupted sink was already marked UnknownCommit by the worker.
726    fn finish(&self) {
727        {
728            let mut state = self.state();
729            // A direct admission may still be waiting behind another in-flight operation.
730            if let Some(pending) = &mut state.direct
731                && pending.failure.is_none()
732            {
733                let failure = WriteFailure {
734                    class: WriteFailureClass::Retryable,
735                    cause: WriteFailureCause::NotAttempted,
736                };
737                if let Some(reply) = pending.reply.take() {
738                    let _ = reply.send(Err(failure.clone()));
739                }
740                pending.failure = Some(failure);
741            }
742            state.lifecycle = BufferedWriterState::Stopped;
743            self.event(&mut state, BufferedWriterEvent::Stopped);
744        }
745        self.changed.notify_waiters();
746    }
747}
748
749impl<E: std::fmt::Display> std::fmt::Display for WriteFailure<E> {
750    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
751        write!(f, "write failed ({:?}): ", self.class)?;
752        match &self.cause {
753            WriteFailureCause::Persistence(error) => error.fmt(f),
754            WriteFailureCause::Panicked => {
755                f.write_str("sink or worker panicked; commit outcome unknown")
756            }
757            WriteFailureCause::Cancelled => f.write_str("sink interrupted; commit outcome unknown"),
758            WriteFailureCause::NotAttempted => {
759                f.write_str("cancelled before persistence was attempted")
760            }
761        }
762    }
763}
764impl<E: std::error::Error + 'static> std::error::Error for WriteFailure<E> {
765    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
766        match &self.cause {
767            WriteFailureCause::Persistence(error) => Some(error.as_ref()),
768            _ => None,
769        }
770    }
771}
772impl<T, E: std::fmt::Display> std::fmt::Display for RecordError<T, E> {
773    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
774        match self {
775            Self::NotRunning(_) => f.write_str("writer admission is closed or not started"),
776            Self::Full(_) => f.write_str("writer admission capacity exhausted"),
777            Self::DirectFailed(error) => {
778                write!(f, "accepted overflow record remains unconfirmed: {error}")
779            }
780        }
781    }
782}
783impl<T: std::fmt::Debug, E: std::error::Error + 'static> std::error::Error for RecordError<T, E> {
784    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
785        match self {
786            Self::DirectFailed(error) => Some(error),
787            _ => None,
788        }
789    }
790}
791impl<E: std::fmt::Display> std::fmt::Display for BufferFlushError<E> {
792    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
793        match self {
794            Self::NotStarted => f.write_str("writer has not started"),
795            Self::Stopped => f.write_str("writer stopped with unconfirmed records"),
796            Self::Write(error) => error.fmt(f),
797            Self::Deadline => f.write_str("flush deadline elapsed; records remain unconfirmed"),
798        }
799    }
800}
801impl<E: std::error::Error + 'static> std::error::Error for BufferFlushError<E> {
802    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
803        match self {
804            Self::Write(error) => Some(error),
805            _ => None,
806        }
807    }
808}
aster_forge_runtime/buffered.rs

aster_forge_runtime/
buffered.rs
