spate_core/feed/
end_check.rs

1//! The end check: when a bounded job on a feed has read all of its work.
2//!
3//! [`EndCheck`] decides from the reads a source reports, and does no I/O
4//! itself. The source asks it what to read next with
5//! [`next`](EndCheck::next), starts that read, and reports the answer. It
6//! holds no clock: every call takes the instant the source reads from its
7//! own clock.
8
9use super::{Depth, DepthOf, FeedError};
10use crate::error::ErrorClass;
11use std::time::{Duration, Instant};
12
13/// The longest gap between two reads of the feed inside a settling window,
14/// and between a read and its answer.
15pub const END_CHECK_INTERVAL: Duration = Duration::from_secs(20);
16
17/// The step between scheduled reads. Shorter than [`END_CHECK_INTERVAL`], so
18/// a read taken on a poll that comes a little after its time still falls
19/// within the bound.
20const READ_EVERY: Duration = Duration::from_secs(15);
21
22/// What a read of a bounded job's end marker found.
23#[derive(Clone, Debug, PartialEq, Eq)]
24#[non_exhaustive]
25pub enum MarkerRead {
26    /// The marker exists and names this lister run.
27    Present(String),
28    /// No marker exists.
29    Absent,
30    /// The read failed and may succeed later. A source ends its run itself
31    /// on a read that cannot succeed.
32    Failed,
33}
34
35/// The next step an [`EndCheck`] asks of its source.
36#[derive(Clone, Debug, PartialEq, Eq)]
37#[non_exhaustive]
38pub enum EndAction {
39    /// Start nothing. With an instant, call [`next`](EndCheck::next) again
40    /// at or after it; with `None`, a read is outstanding and its answer
41    /// comes first.
42    Wait(Option<Instant>),
43    /// Read the end marker and report it with
44    /// [`marker_read`](EndCheck::marker_read).
45    ReadMarker,
46    /// Read the depth named and report it with
47    /// [`depth_read`](EndCheck::depth_read).
48    ReadDepth(DepthOf),
49    /// Receive from the feed once and report whether it took a unit with
50    /// [`received`](EndCheck::received).
51    Receive,
52    /// The job has ended.
53    Ended(EndOutcome),
54}
55
56/// How a bounded job ended.
57#[derive(Clone, Debug, PartialEq, Eq)]
58#[non_exhaustive]
59pub enum EndOutcome {
60    /// The feed stayed empty over the window and its dead-letter destination
61    /// holds nothing.
62    Completed,
63    /// The feed's dead-letter destination holds units, counted here.
64    Stalled(Depth),
65    /// A read of the feed's depth failed with an error that sending it again
66    /// cannot clear, such as a missing permission.
67    Failed(FeedError),
68}
69
70#[derive(Clone, Copy, Debug, PartialEq, Eq)]
71enum Read {
72    Marker,
73    Depth(DepthOf),
74    Receive,
75}
76
77#[derive(Debug)]
78enum Phase {
79    /// Waiting to open a window: read the marker on a windowed feed, the
80    /// feed itself on an exact one.
81    Opening {
82        due: Instant,
83    },
84    /// A window opened by a read of `run`'s marker. It spans from `start`,
85    /// its first read of the feed, unset until that read. The next read is
86    /// due at `due`, and must be taken within [`END_CHECK_INTERVAL`] of
87    /// `last`, the window's last read.
88    Window {
89        run: String,
90        start: Option<Instant>,
91        last: Instant,
92        due: Instant,
93    },
94    /// The window has passed with every read empty; the marker is read again.
95    Closing {
96        run: String,
97    },
98    /// One receive that must take nothing.
99    Receiving,
100    /// The dead-letter destination's read, due at `due`.
101    DeadLetters {
102        due: Instant,
103    },
104    Ended(EndOutcome),
105}
106
107/// Decides when a bounded job on a feed has ended.
108///
109/// A windowed check opens a window on a read of the end marker. The window
110/// spans from its first read of the feed. Inside it the feed is read every
111/// 15 s, each read taken within [`END_CHECK_INTERVAL`] of the one before and
112/// answered within it of its request, and every read must find the feed
113/// empty. A read requested a window or more after the first that finds it
114/// empty is followed by a second read of the marker: the same run
115/// passes, another run opens a new window, and no marker restarts the
116/// check. An exact check, for a feed whose counts are exact and whose lister
117/// has finished, reads the feed once with no window or marker. Either way
118/// one receive that takes nothing follows, and then the dead-letter
119/// destination decides between [`Completed`](EndOutcome::Completed) and
120/// [`Stalled`](EndOutcome::Stalled).
121///
122/// A non-empty read, a read that fails retryably, a read taken or answered
123/// later than that bound, a marker read answered later than it included, or
124/// a [`restart`](EndCheck::restart) sends a windowed check back to its marker
125/// read, 15 s after its previous read at most. A dead-letter read that fails
126/// retryably is retried 15 s later and ends nothing. A depth read that fails
127/// with a [`Fatal`](ErrorClass::Fatal) class ends the check
128/// [`Failed`](EndOutcome::Failed). The source calls `restart` whenever its reader holds
129/// any work, so every read that counts was answered while it held none.
130///
131/// ```
132/// use spate_core::feed::{Depth, DepthOf, EndAction, EndCheck, EndOutcome, MarkerRead};
133/// use std::time::{Duration, Instant};
134///
135/// let start = Instant::now();
136/// let mut check = EndCheck::windowed(Duration::from_secs(30), start);
137/// assert_eq!(check.next(start), EndAction::ReadMarker);
138/// check.marker_read(start, MarkerRead::Present("run-a".into()));
139///
140/// let empty = || Ok(Depth::new("queue", false));
141/// for at in [0, 15, 30] {
142///     let now = start + Duration::from_secs(at);
143///     assert_eq!(check.next(now), EndAction::ReadDepth(DepthOf::Feed));
144///     check.depth_read(now, empty());
145/// }
146/// let now = start + Duration::from_secs(30);
147/// assert_eq!(check.next(now), EndAction::ReadMarker);
148/// check.marker_read(now, MarkerRead::Present("run-a".into()));
149/// assert_eq!(check.next(now), EndAction::Receive);
150/// check.received(now, false);
151/// assert_eq!(check.next(now), EndAction::ReadDepth(DepthOf::DeadLetters));
152/// check.depth_read(now, empty());
153/// assert_eq!(check.next(now), EndAction::Ended(EndOutcome::Completed));
154/// ```
155#[derive(Debug)]
156pub struct EndCheck {
157    /// `None` on an exact check.
158    window: Option<Duration>,
159    phase: Phase,
160    /// The read outstanding, and when it was requested.
161    pending: Option<(Read, Instant)>,
162    /// When the last read of the feed or the marker was requested, so a
163    /// restart waits 15 s from it.
164    last_request: Option<Instant>,
165    /// The run whose marker last opened a window, so a window reopened on
166    /// the same run is not announced again.
167    announced: Option<String>,
168}
169
170impl EndCheck {
171    /// A check that ends a job once the feed stays empty over `window`
172    /// between two reads of the same lister run's end marker. Its first read
173    /// is due at `now`.
174    #[must_use]
175    pub fn windowed(window: Duration, now: Instant) -> EndCheck {
176        EndCheck {
177            window: Some(window),
178            phase: Phase::Opening { due: now },
179            pending: None,
180            last_request: None,
181            announced: None,
182        }
183    }
184
185    /// A check for a feed whose counts are exact and whose lister has
186    /// finished: one empty read of the feed passes, with no window and no
187    /// marker. Its first read is due at `now`.
188    #[must_use]
189    pub fn exact(now: Instant) -> EndCheck {
190        EndCheck {
191            window: None,
192            phase: Phase::Opening { due: now },
193            pending: None,
194            last_request: None,
195            announced: None,
196        }
197    }
198
199    /// What to do at `now`. A read returned here is outstanding until its
200    /// answer is reported; until then `next` returns `Wait(None)`.
201    pub fn next(&mut self, now: Instant) -> EndAction {
202        if self.pending.is_some() {
203            return EndAction::Wait(None);
204        }
205        if let Phase::Window { last, .. } = self.phase
206            && now > last + END_CHECK_INTERVAL
207        {
208            tracing::info!(
209                gap = ?now - last,
210                "the feed went unread longer than the read interval; restarting the end check"
211            );
212            self.reopen(now);
213        }
214        let read = match &self.phase {
215            Phase::Ended(outcome) => return EndAction::Ended(outcome.clone()),
216            Phase::Opening { due } | Phase::Window { due, .. } | Phase::DeadLetters { due }
217                if now < *due =>
218            {
219                return EndAction::Wait(Some(*due));
220            }
221            Phase::Opening { .. } if self.window.is_none() => Read::Depth(DepthOf::Feed),
222            Phase::Opening { .. } | Phase::Closing { .. } => Read::Marker,
223            Phase::Window { .. } => Read::Depth(DepthOf::Feed),
224            Phase::Receiving => Read::Receive,
225            Phase::DeadLetters { .. } => Read::Depth(DepthOf::DeadLetters),
226        };
227        self.pending = Some((read, now));
228        if matches!(read, Read::Marker | Read::Depth(DepthOf::Feed)) {
229            self.last_request = Some(now);
230        }
231        match read {
232            Read::Marker => EndAction::ReadMarker,
233            Read::Depth(of) => EndAction::ReadDepth(of),
234            Read::Receive => EndAction::Receive,
235        }
236    }
237
238    /// Report the end marker read that [`next`](Self::next) asked for. An
239    /// answer nothing asked for is ignored.
240    pub fn marker_read(&mut self, now: Instant, read: MarkerRead) {
241        // The window and the next read count from the request, so an answer
242        // the source notices late does not shift them.
243        let Some(requested) = self.take_pending(Read::Marker) else {
244            return;
245        };
246        if now > requested + END_CHECK_INTERVAL {
247            tracing::info!(
248                late = ?now - requested,
249                "an end marker read answered after the read interval; restarting the end check"
250            );
251            self.reopen(now);
252            return;
253        }
254        let window = self.window.unwrap_or_default();
255        let closing = match &self.phase {
256            Phase::Closing { run } => Some(run.clone()),
257            _ => None,
258        };
259        match (read, closing) {
260            (MarkerRead::Present(seen), Some(run)) if seen == run => {
261                self.phase = Phase::Receiving;
262            }
263            (MarkerRead::Present(seen), closing) => {
264                if closing.is_some() {
265                    tracing::info!(
266                        run = %seen,
267                        "a newer lister run's end marker replaced the window's; \
268                         opening a new settling window"
269                    );
270                } else if self.announced.as_deref() != Some(seen.as_str()) {
271                    tracing::info!(
272                        run = %seen,
273                        window = ?window,
274                        "end marker found; the job ends once the feed stays empty over \
275                         the settling window"
276                    );
277                }
278                self.announced = Some(seen.clone());
279                self.phase = Phase::Window {
280                    run: seen,
281                    start: None,
282                    last: requested,
283                    due: now,
284                };
285            }
286            (MarkerRead::Absent, closing) => {
287                self.announced = None;
288                if closing.is_some() {
289                    tracing::info!(
290                        "the end marker is gone at the settling window's end; waiting for a \
291                         lister run's marker"
292                    );
293                }
294                self.phase = Phase::Opening {
295                    due: requested + READ_EVERY,
296                };
297            }
298            (MarkerRead::Failed, _) => {
299                tracing::warn!("reading the end marker failed; reading it again");
300                self.phase = Phase::Opening {
301                    due: requested + READ_EVERY,
302                };
303            }
304        }
305    }
306
307    /// Report the depth read that [`next`](Self::next) asked for. An answer
308    /// nothing asked for is ignored.
309    pub fn depth_read(&mut self, now: Instant, read: Result<Depth, FeedError>) {
310        let Some((Read::Depth(of), requested)) = self.pending else {
311            return;
312        };
313        self.pending = None;
314        let due = requested + READ_EVERY;
315        match (of, read) {
316            (_, Err(e)) if e.class() == ErrorClass::Fatal => {
317                self.phase = Phase::Ended(EndOutcome::Failed(e));
318            }
319            (DepthOf::DeadLetters, Ok(depth)) if depth.is_empty() => {
320                self.phase = Phase::Ended(EndOutcome::Completed);
321            }
322            (DepthOf::DeadLetters, Ok(depth)) => {
323                self.phase = Phase::Ended(EndOutcome::Stalled(depth));
324            }
325            (DepthOf::DeadLetters, Err(e)) => {
326                tracing::warn!(error = %e, "reading the dead-letter destination failed; reading it again");
327                self.phase = Phase::DeadLetters { due };
328            }
329            (DepthOf::Feed, Err(e)) => {
330                tracing::warn!(error = %e, "reading the feed's depth failed; restarting the end check");
331                self.reopen(due);
332            }
333            (DepthOf::Feed, Ok(depth)) if !depth.is_empty() => {
334                // Every interval while peers still read, so not at INFO.
335                tracing::debug!(%depth, "the feed is not empty; restarting the end check");
336                self.reopen(due);
337            }
338            (DepthOf::Feed, Ok(_))
339                if self.window.is_some() && now > requested + END_CHECK_INTERVAL =>
340            {
341                tracing::info!(
342                    late = ?now - requested,
343                    "a depth read answered after the read interval; restarting the end check"
344                );
345                self.reopen(now);
346            }
347            (DepthOf::Feed, Ok(_)) => self.advance(requested),
348        }
349    }
350
351    /// Report whether the receive that [`next`](Self::next) asked for took a
352    /// unit. One that did restarts the check.
353    pub fn received(&mut self, now: Instant, took: bool) {
354        if self.take_pending(Read::Receive).is_none() {
355            return;
356        }
357        if took {
358            self.restart(now);
359        } else {
360            self.phase = Phase::DeadLetters { due: now };
361        }
362    }
363
364    /// Send the check back to its opening read, because the reader holds
365    /// work. A windowed check reads again 15 s after its last read; an exact
366    /// one at once. A check that has ended stays ended.
367    ///
368    /// Drop every read outstanding when calling it. The check tells answers
369    /// apart only by kind, so an answer reported once the next read of its
370    /// kind has started counts as that read's.
371    pub fn restart(&mut self, now: Instant) {
372        if matches!(self.phase, Phase::Ended(_)) {
373            return;
374        }
375        if !matches!(self.phase, Phase::Opening { .. }) {
376            tracing::info!("the reader took work; restarting the end check");
377        }
378        self.pending = None;
379        let due = match (self.window, self.last_request) {
380            (Some(_), Some(last)) => (last + READ_EVERY).max(now),
381            _ => now,
382        };
383        self.phase = Phase::Opening { due };
384    }
385
386    /// Whether the check has ended.
387    #[must_use]
388    pub fn outcome(&self) -> Option<&EndOutcome> {
389        match &self.phase {
390            Phase::Ended(outcome) => Some(outcome),
391            _ => None,
392        }
393    }
394
395    /// Take the outstanding read if it is `read`, returning when it was
396    /// requested.
397    fn take_pending(&mut self, read: Read) -> Option<Instant> {
398        match self.pending {
399            Some((pending, requested)) if pending == read => {
400                self.pending = None;
401                Some(requested)
402            }
403            _ => None,
404        }
405    }
406
407    fn reopen(&mut self, due: Instant) {
408        self.phase = Phase::Opening { due };
409    }
410
411    /// An empty, timely read of the feed, requested at `requested`.
412    fn advance(&mut self, requested: Instant) {
413        match (&self.phase, self.window) {
414            (Phase::Opening { .. }, None) => self.phase = Phase::Receiving,
415            (Phase::Window { run, start, .. }, Some(window)) => {
416                // The window spans the feed's reads, so its first read starts
417                // it. A window too long to add to an instant never closes.
418                let start = start.unwrap_or(requested);
419                let closes = start
420                    .checked_add(window)
421                    .is_some_and(|end| requested >= end);
422                self.phase = if closes {
423                    Phase::Closing { run: run.clone() }
424                } else {
425                    Phase::Window {
426                        run: run.clone(),
427                        start: Some(start),
428                        last: requested,
429                        due: requested + READ_EVERY,
430                    }
431                };
432            }
433            _ => {}
434        }
435    }
436}
437
438#[cfg(test)]
439mod tests {
440    use super::*;
441    use crate::feed::FeedErrorKind;
442
443    const WINDOW: Duration = Duration::from_secs(60);
444
445    fn at(start: Instant, secs: u64) -> Instant {
446        start + Duration::from_secs(secs)
447    }
448
449    fn empty() -> Result<Depth, FeedError> {
450        Ok(Depth::new("queue", true))
451    }
452
453    fn visible(n: u64) -> Result<Depth, FeedError> {
454        let mut depth = Depth::new("queue", true);
455        depth.visible = n;
456        Ok(depth)
457    }
458
459    fn failed() -> Result<Depth, FeedError> {
460        Err(FeedError::new(FeedErrorKind::Retryable, "throttled"))
461    }
462
463    fn present(run: &str) -> MarkerRead {
464        MarkerRead::Present(run.into())
465    }
466
467    /// Opens a window on `run` at `start` and returns the check.
468    fn opened(start: Instant, run: &str) -> EndCheck {
469        let mut check = EndCheck::windowed(WINDOW, start);
470        assert_eq!(check.next(start), EndAction::ReadMarker);
471        check.marker_read(start, present(run));
472        check
473    }
474
475    /// Answers every empty read of the feed due from `from` to `to`, in
476    /// 15-second steps, and returns how many it answered.
477    fn empty_reads(check: &mut EndCheck, start: Instant, from: u64, to: u64) -> usize {
478        let mut reads = 0;
479        for secs in (from..=to).step_by(15) {
480            let now = at(start, secs);
481            assert_eq!(
482                check.next(now),
483                EndAction::ReadDepth(DepthOf::Feed),
484                "read at {secs}s"
485            );
486            check.depth_read(now, empty());
487            reads += 1;
488        }
489        reads
490    }
491
492    /// Takes a check whose closing marker passed through the receive and
493    /// the dead-letter read, answering the latter with `dead`.
494    fn finish(check: &mut EndCheck, now: Instant, dead: Result<Depth, FeedError>) -> EndAction {
495        assert_eq!(check.next(now), EndAction::Receive);
496        check.received(now, false);
497        assert_eq!(check.next(now), EndAction::ReadDepth(DepthOf::DeadLetters));
498        check.depth_read(now, dead);
499        check.next(now)
500    }
501
502    /// Pins that a window ends one window after its first read of the feed:
503    /// five empty reads at 0, 15, 30, 45 and 60 s, then the closing marker
504    /// read.
505    #[test]
506    fn a_window_ends_one_window_after_its_opening_marker_read() {
507        let start = Instant::now();
508        let mut check = opened(start, "a");
509        assert_eq!(empty_reads(&mut check, start, 0, 45), 4);
510        assert_eq!(
511            check.next(at(start, 59)),
512            EndAction::Wait(Some(at(start, 60)))
513        );
514        empty_reads(&mut check, start, 60, 60);
515        assert_eq!(check.next(at(start, 60)), EndAction::ReadMarker);
516        check.marker_read(at(start, 60), present("a"));
517        assert_eq!(
518            finish(&mut check, at(start, 60), empty()),
519            EndAction::Ended(EndOutcome::Completed)
520        );
521    }
522
523    /// Pins that one non-empty read inside a window, as an approximate count
524    /// can give between zeros, restarts it from a new marker read.
525    #[test]
526    fn a_nonzero_read_restarts_the_window() {
527        let start = Instant::now();
528        let mut check = opened(start, "a");
529        empty_reads(&mut check, start, 0, 15);
530        assert_eq!(
531            check.next(at(start, 30)),
532            EndAction::ReadDepth(DepthOf::Feed)
533        );
534        check.depth_read(at(start, 30), visible(1));
535        assert_eq!(
536            check.next(at(start, 44)),
537            EndAction::Wait(Some(at(start, 45)))
538        );
539        assert_eq!(check.next(at(start, 45)), EndAction::ReadMarker);
540        check.marker_read(at(start, 45), present("a"));
541        empty_reads(&mut check, start, 45, 90);
542        assert_eq!(
543            check.next(at(start, 105)),
544            EndAction::ReadDepth(DepthOf::Feed),
545            "the window counts from the marker read at 45 s"
546        );
547        check.depth_read(at(start, 105), empty());
548        assert_eq!(check.next(at(start, 105)), EndAction::ReadMarker);
549    }
550
551    /// Pins that a failed read of the feed restarts the window and ends
552    /// nothing.
553    #[test]
554    fn a_failed_read_restarts_the_window() {
555        let start = Instant::now();
556        let mut check = opened(start, "a");
557        empty_reads(&mut check, start, 0, 30);
558        assert_eq!(
559            check.next(at(start, 45)),
560            EndAction::ReadDepth(DepthOf::Feed)
561        );
562        check.depth_read(at(start, 45), failed());
563        assert_eq!(
564            check.next(at(start, 59)),
565            EndAction::Wait(Some(at(start, 60))),
566            "a failed read is followed 15 s after its request"
567        );
568        assert_eq!(check.next(at(start, 60)), EndAction::ReadMarker);
569        assert!(check.outcome().is_none());
570    }
571
572    /// Pins the answer bound: an empty read answered more than 20 s after it
573    /// was requested restarts the window, and one answered at 20 s does not.
574    #[test]
575    fn a_read_answered_late_restarts_the_window() {
576        let start = Instant::now();
577        let mut check = opened(start, "a");
578        empty_reads(&mut check, start, 0, 15);
579        assert_eq!(
580            check.next(at(start, 30)),
581            EndAction::ReadDepth(DepthOf::Feed)
582        );
583        check.depth_read(at(start, 51), empty());
584        assert_eq!(check.next(at(start, 51)), EndAction::ReadMarker);
585
586        let mut check = opened(start, "a");
587        assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
588        check.depth_read(at(start, 20), empty());
589        assert_eq!(
590            check.next(at(start, 20)),
591            EndAction::ReadDepth(DepthOf::Feed)
592        );
593        // The read that would close the window, answered late, opens a new
594        // window from the next marker read and does not pass to the receive.
595        let mut check = opened(start, "a");
596        empty_reads(&mut check, start, 0, 45);
597        assert_eq!(
598            check.next(at(start, 60)),
599            EndAction::ReadDepth(DepthOf::Feed)
600        );
601        check.depth_read(at(start, 81), empty());
602        assert_eq!(check.next(at(start, 81)), EndAction::ReadMarker);
603        check.marker_read(at(start, 81), present("a"));
604        assert_eq!(
605            check.next(at(start, 81)),
606            EndAction::ReadDepth(DepthOf::Feed)
607        );
608    }
609
610    /// Pins the gap bound: a read taken more than 20 s after the window's last
611    /// one, as a clock that jumped or a poll that stalled gives, restarts the
612    /// window from a marker read, and one taken at 20 s does not.
613    #[test]
614    fn a_read_taken_after_a_gap_restarts_the_window() {
615        let start = Instant::now();
616        let mut check = opened(start, "a");
617        empty_reads(&mut check, start, 0, 15);
618        assert_eq!(check.next(at(start, 36)), EndAction::ReadMarker);
619        assert!(check.outcome().is_none());
620
621        let mut check = opened(start, "a");
622        empty_reads(&mut check, start, 0, 15);
623        assert_eq!(
624            check.next(at(start, 35)),
625            EndAction::ReadDepth(DepthOf::Feed),
626            "20 s after the last read is within the bound"
627        );
628
629        // A jump of a whole window cannot close it.
630        let mut check = opened(start, "a");
631        empty_reads(&mut check, start, 0, 0);
632        assert_eq!(check.next(at(start, 60)), EndAction::ReadMarker);
633    }
634
635    /// Pins that a closing marker naming another run opens a new window on
636    /// that run instead of ending.
637    #[test]
638    fn a_closing_marker_of_another_run_opens_a_new_window() {
639        let start = Instant::now();
640        let mut check = opened(start, "a");
641        empty_reads(&mut check, start, 0, 60);
642        assert_eq!(check.next(at(start, 60)), EndAction::ReadMarker);
643        check.marker_read(at(start, 60), present("b"));
644        empty_reads(&mut check, start, 60, 120);
645        assert_eq!(check.next(at(start, 120)), EndAction::ReadMarker);
646        check.marker_read(at(start, 120), present("b"));
647        assert_eq!(check.next(at(start, 120)), EndAction::Receive);
648    }
649
650    /// Pins that a marker gone at the window's end ends nothing, and the
651    /// check reads it again 15 s later.
652    #[test]
653    fn an_absent_closing_marker_does_not_end() {
654        let start = Instant::now();
655        let mut check = opened(start, "a");
656        empty_reads(&mut check, start, 0, 60);
657        assert_eq!(check.next(at(start, 60)), EndAction::ReadMarker);
658        check.marker_read(at(start, 60), MarkerRead::Absent);
659        assert_eq!(
660            check.next(at(start, 60)),
661            EndAction::Wait(Some(at(start, 75))),
662            "an absent marker is read again 15 s later"
663        );
664        assert_eq!(check.next(at(start, 75)), EndAction::ReadMarker);
665        check.marker_read(at(start, 75), MarkerRead::Absent);
666        assert!(check.outcome().is_none());
667
668        // A failed closing read ends nothing either.
669        let mut check = opened(start, "a");
670        empty_reads(&mut check, start, 0, 60);
671        assert_eq!(check.next(at(start, 60)), EndAction::ReadMarker);
672        check.marker_read(at(start, 60), MarkerRead::Failed);
673        assert_eq!(
674            check.next(at(start, 60)),
675            EndAction::Wait(Some(at(start, 75))),
676            "a failed marker read is retried 15 s later"
677        );
678        assert_eq!(check.next(at(start, 75)), EndAction::ReadMarker);
679    }
680
681    /// Pins that a marker read's schedule counts from its request, so an
682    /// answer noticed late does not delay the next read, and that the window
683    /// counts from its first read of the feed.
684    #[test]
685    fn a_marker_answer_noticed_late_keeps_the_schedule() {
686        let start = Instant::now();
687        let mut check = EndCheck::windowed(WINDOW, start);
688        assert_eq!(check.next(start), EndAction::ReadMarker);
689        check.marker_read(at(start, 10), MarkerRead::Absent);
690        assert_eq!(
691            check.next(at(start, 10)),
692            EndAction::Wait(Some(at(start, 15)))
693        );
694        assert_eq!(check.next(at(start, 15)), EndAction::ReadMarker);
695        // Answered at 25 s, and the first read of the feed taken at 30 s:
696        // that read starts the window.
697        check.marker_read(at(start, 25), present("a"));
698        empty_reads(&mut check, start, 30, 75);
699        assert_eq!(
700            check.next(at(start, 75)),
701            EndAction::Wait(Some(at(start, 90))),
702            "the read at 75 s is short of a window after the first read"
703        );
704        empty_reads(&mut check, start, 90, 90);
705        assert_eq!(check.next(at(start, 90)), EndAction::ReadMarker);
706    }
707
708    /// Pins that a window at the 20 s floor spans 20 s of reads even when its
709    /// marker read was answered at the 20 s bound: the first read of the feed
710    /// starts it, and the read 20 s or more later closes it.
711    #[test]
712    fn a_window_spans_its_reads_from_the_first() {
713        let start = Instant::now();
714        let mut check = EndCheck::windowed(END_CHECK_INTERVAL, start);
715        assert_eq!(check.next(start), EndAction::ReadMarker);
716        check.marker_read(at(start, 20), present("a"));
717        empty_reads(&mut check, start, 20, 35);
718        assert_eq!(
719            check.next(at(start, 35)),
720            EndAction::Wait(Some(at(start, 50))),
721            "15 s of reads do not close a 20 s window"
722        );
723        empty_reads(&mut check, start, 50, 50);
724        assert_eq!(check.next(at(start, 50)), EndAction::ReadMarker);
725    }
726
727    /// Pins that an answer of a kind the check did not ask for is ignored and
728    /// leaves the outstanding read outstanding.
729    #[test]
730    fn an_answer_nothing_asked_for_is_ignored() {
731        let start = Instant::now();
732        let mut check = opened(start, "a");
733        assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
734        check.marker_read(start, present("b"));
735        check.received(start, true);
736        assert_eq!(check.next(start), EndAction::Wait(None));
737        check.depth_read(start, empty());
738        assert_eq!(
739            check.next(start),
740            EndAction::Wait(Some(at(start, 15))),
741            "the window on run a carries on"
742        );
743        // With nothing outstanding, a depth answer is ignored too.
744        check.depth_read(start, visible(1));
745        assert_eq!(
746            check.next(at(start, 15)),
747            EndAction::ReadDepth(DepthOf::Feed)
748        );
749    }
750
751    /// Everything the check logs while `f` runs, as text.
752    fn logs(f: impl FnOnce()) -> String {
753        #[derive(Clone, Default)]
754        struct Buf(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
755        impl std::io::Write for Buf {
756            fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
757                self.0.lock().unwrap().extend_from_slice(bytes);
758                Ok(bytes.len())
759            }
760            fn flush(&mut self) -> std::io::Result<()> {
761                Ok(())
762            }
763        }
764        let buf = Buf::default();
765        let writer = buf.clone();
766        let subscriber = tracing_subscriber::fmt()
767            .with_writer(move || writer.clone())
768            .with_ansi(false)
769            .finish();
770        tracing::subscriber::with_default(subscriber, f);
771        let bytes = buf.0.lock().unwrap().clone();
772        String::from_utf8(bytes).unwrap()
773    }
774
775    /// Pins the transitions an operator sees at INFO: a window opening with
776    /// its run, and a restart once a window is open. A restart before any
777    /// window opens, as every poll of a busy reader makes, logs nothing.
778    #[test]
779    fn the_check_logs_a_window_opening_and_its_restart() {
780        let start = Instant::now();
781        let mut check = EndCheck::windowed(WINDOW, start);
782        let quiet = logs(|| check.restart(start));
783        assert!(quiet.is_empty(), "{quiet}");
784        assert_eq!(check.next(start), EndAction::ReadMarker);
785        let opened = logs(|| check.marker_read(start, present("run-a")));
786        assert!(
787            opened.contains("INFO")
788                && opened.contains("end marker found")
789                && opened.contains("run-a"),
790            "{opened}"
791        );
792        let restarted = logs(|| check.restart(start));
793        assert!(
794            restarted.contains("INFO") && restarted.contains("restarting the end check"),
795            "{restarted}"
796        );
797
798        // The same run's marker opening the window again is not announced
799        // again; a busy feed restarts the check at DEBUG only.
800        assert_eq!(check.next(at(start, 15)), EndAction::ReadMarker);
801        let again = logs(|| check.marker_read(at(start, 15), present("run-a")));
802        assert!(again.is_empty(), "{again}");
803        assert_eq!(
804            check.next(at(start, 15)),
805            EndAction::ReadDepth(DepthOf::Feed)
806        );
807        let busy = logs(|| check.depth_read(at(start, 15), visible(1)));
808        assert!(!busy.contains("INFO") && !busy.contains("WARN"), "{busy}");
809
810        // An absent marker is no failure; a failed read warns.
811        assert_eq!(check.next(at(start, 30)), EndAction::ReadMarker);
812        let absent = logs(|| check.marker_read(at(start, 30), MarkerRead::Absent));
813        assert!(!absent.contains("WARN"), "{absent}");
814        assert_eq!(check.next(at(start, 45)), EndAction::ReadMarker);
815        let failed = logs(|| check.marker_read(at(start, 45), MarkerRead::Failed));
816        assert!(
817            failed.contains("WARN") && failed.contains("reading the end marker failed"),
818            "{failed}"
819        );
820    }
821
822    /// Pins that an exact check's wording names no settling window, which
823    /// the in-process feed does not have.
824    #[test]
825    fn an_exact_check_logs_no_settling_window() {
826        let start = Instant::now();
827        let mut check = EndCheck::exact(start);
828        assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
829        let failed = logs(|| check.depth_read(start, failed()));
830        assert!(failed.contains("restarting the end check"), "{failed}");
831        let shown = failed + &logs(|| check.restart(start));
832        assert!(!shown.contains("settling window"), "{shown}");
833    }
834
835    /// Pins that no window opens before a marker is found: the feed is not
836    /// read until a marker is.
837    #[test]
838    fn no_window_opens_before_a_marker_is_found() {
839        let start = Instant::now();
840        let mut check = EndCheck::windowed(WINDOW, start);
841        for secs in [0, 15, 30] {
842            assert_eq!(check.next(at(start, secs)), EndAction::ReadMarker);
843            check.marker_read(at(start, secs), MarkerRead::Absent);
844        }
845        assert_eq!(check.next(at(start, 45)), EndAction::ReadMarker);
846        check.marker_read(at(start, 45), present("a"));
847        assert_eq!(
848            check.next(at(start, 45)),
849            EndAction::ReadDepth(DepthOf::Feed)
850        );
851    }
852
853    /// Pins that the closing marker is read only after the read of the feed
854    /// that closes the window, never at the window's end alone.
855    #[test]
856    fn the_closing_marker_is_read_after_the_last_depth_read() {
857        let start = Instant::now();
858        let mut check = opened(start, "a");
859        empty_reads(&mut check, start, 0, 45);
860        assert_eq!(
861            check.next(at(start, 60)),
862            EndAction::ReadDepth(DepthOf::Feed)
863        );
864        assert_eq!(check.next(at(start, 61)), EndAction::Wait(None));
865        check.depth_read(at(start, 61), empty());
866        assert_eq!(check.next(at(start, 61)), EndAction::ReadMarker);
867    }
868
869    /// Pins that a receive taking a unit after the closing marker restarts
870    /// the check, and that a restart waits 15 s from the last read.
871    #[test]
872    fn a_receive_that_takes_a_unit_restarts_the_window() {
873        let start = Instant::now();
874        let mut check = opened(start, "a");
875        empty_reads(&mut check, start, 0, 60);
876        assert_eq!(check.next(at(start, 60)), EndAction::ReadMarker);
877        check.marker_read(at(start, 60), present("a"));
878        assert_eq!(check.next(at(start, 62)), EndAction::Receive);
879        check.received(at(start, 62), true);
880        assert_eq!(
881            check.next(at(start, 62)),
882            EndAction::Wait(Some(at(start, 75))),
883            "15 s from the closing marker read, the last read of the marker or feed"
884        );
885        assert_eq!(check.next(at(start, 75)), EndAction::ReadMarker);
886    }
887
888    /// Pins that a restart drops the outstanding read: its answer is ignored.
889    #[test]
890    fn a_restart_ignores_the_outstanding_read() {
891        let start = Instant::now();
892        let mut check = opened(start, "a");
893        assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
894        check.restart(at(start, 5));
895        check.depth_read(at(start, 5), empty());
896        assert_eq!(
897            check.next(at(start, 5)),
898            EndAction::Wait(Some(at(start, 15)))
899        );
900        assert_eq!(check.next(at(start, 15)), EndAction::ReadMarker);
901    }
902
903    /// Pins that a failed dead-letter read is retried 15 s later and ends
904    /// the job neither way.
905    #[test]
906    fn a_failed_dead_letter_read_neither_drains_nor_stalls() {
907        let start = Instant::now();
908        let mut check = opened(start, "a");
909        empty_reads(&mut check, start, 0, 60);
910        assert_eq!(check.next(at(start, 60)), EndAction::ReadMarker);
911        check.marker_read(at(start, 60), present("a"));
912        assert_eq!(
913            finish(&mut check, at(start, 60), failed()),
914            EndAction::Wait(Some(at(start, 75)))
915        );
916        assert!(check.outcome().is_none());
917        assert_eq!(
918            check.next(at(start, 75)),
919            EndAction::ReadDepth(DepthOf::DeadLetters)
920        );
921        check.depth_read(at(start, 75), empty());
922        assert_eq!(check.outcome(), Some(&EndOutcome::Completed));
923    }
924
925    /// Pins the dead-letter verdict at its boundary: one unit stalls the job
926    /// with the depth read, none completes it.
927    #[test]
928    fn one_dead_letter_stalls_and_none_completes() {
929        let start = Instant::now();
930        for (field, set) in [
931            ("visible", (|d: &mut Depth| d.visible = 1) as fn(&mut Depth)),
932            ("delayed", |d: &mut Depth| d.delayed = 1),
933            ("in_flight", |d: &mut Depth| d.in_flight = 1),
934        ] {
935            let mut check = EndCheck::exact(start);
936            assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
937            check.depth_read(start, empty());
938            let mut dead = Depth::new("dead letters", true);
939            set(&mut dead);
940            assert_eq!(
941                finish(&mut check, start, Ok(dead.clone())),
942                EndAction::Ended(EndOutcome::Stalled(dead)),
943                "{field}"
944            );
945        }
946        let mut check = EndCheck::exact(start);
947        assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
948        check.depth_read(start, empty());
949        assert_eq!(
950            finish(&mut check, start, empty()),
951            EndAction::Ended(EndOutcome::Completed)
952        );
953    }
954
955    /// Pins that an exact check ends on its first empty read with no window,
956    /// no marker and no clock advance, and that a non-empty read holds it.
957    #[test]
958    fn an_exact_feed_ends_on_its_first_empty_read() {
959        let start = Instant::now();
960        let mut check = EndCheck::exact(start);
961        assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
962        check.depth_read(start, visible(1));
963        assert!(check.outcome().is_none());
964        check.restart(start);
965        assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
966        check.depth_read(start, empty());
967        assert_eq!(
968            finish(&mut check, start, empty()),
969            EndAction::Ended(EndOutcome::Completed)
970        );
971    }
972
973    /// Pins that a depth read failing with a fatal class ends the check
974    /// `Failed` with that error, for the feed and for its dead letters, and
975    /// that a retryable failure does not.
976    #[test]
977    fn a_fatal_depth_read_ends_the_check_failed() {
978        let start = Instant::now();
979        let fatal = FeedError::new(FeedErrorKind::Fatal, "access denied");
980        let mut check = opened(start, "a");
981        assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
982        check.depth_read(start, Err(fatal.clone()));
983        assert_eq!(
984            check.next(start),
985            EndAction::Ended(EndOutcome::Failed(fatal.clone()))
986        );
987
988        let mut check = EndCheck::exact(start);
989        assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
990        check.depth_read(start, empty());
991        assert_eq!(check.next(start), EndAction::Receive);
992        check.received(start, false);
993        assert_eq!(
994            check.next(start),
995            EndAction::ReadDepth(DepthOf::DeadLetters)
996        );
997        check.depth_read(start, Err(fatal.clone()));
998        assert_eq!(check.outcome(), Some(&EndOutcome::Failed(fatal)));
999
1000        let mut check = opened(start, "a");
1001        assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
1002        check.depth_read(start, failed());
1003        assert!(check.outcome().is_none());
1004    }
1005
1006    /// Pins that a marker read answered more than 20 s after its request
1007    /// restarts the check: a closing read that comes back late does not pass
1008    /// on to the receive.
1009    #[test]
1010    fn a_marker_read_answered_late_restarts_the_check() {
1011        let start = Instant::now();
1012        let mut check = opened(start, "a");
1013        empty_reads(&mut check, start, 0, 60);
1014        assert_eq!(check.next(at(start, 60)), EndAction::ReadMarker);
1015        check.marker_read(at(start, 81), present("a"));
1016        assert_eq!(check.next(at(start, 81)), EndAction::ReadMarker);
1017
1018        // At 20 s it is on time.
1019        let mut check = opened(start, "a");
1020        empty_reads(&mut check, start, 0, 60);
1021        assert_eq!(check.next(at(start, 60)), EndAction::ReadMarker);
1022        check.marker_read(at(start, 80), present("a"));
1023        assert_eq!(check.next(at(start, 80)), EndAction::Receive);
1024    }
1025
1026    /// Pins that an exact check takes an empty read answered late, since its
1027    /// counts are exact and it has no window to keep.
1028    #[test]
1029    fn an_exact_check_takes_a_read_answered_late() {
1030        let start = Instant::now();
1031        let mut check = EndCheck::exact(start);
1032        assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
1033        check.depth_read(at(start, 100), empty());
1034        assert_eq!(check.next(at(start, 100)), EndAction::Receive);
1035    }
1036
1037    /// Pins that a restart after the check has ended keeps its outcome.
1038    #[test]
1039    fn a_restart_after_the_end_keeps_the_outcome() {
1040        let start = Instant::now();
1041        let mut check = EndCheck::exact(start);
1042        assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
1043        check.depth_read(start, empty());
1044        assert_eq!(
1045            finish(&mut check, start, empty()),
1046            EndAction::Ended(EndOutcome::Completed)
1047        );
1048        check.restart(at(start, 1));
1049        assert_eq!(
1050            check.next(at(start, 1)),
1051            EndAction::Ended(EndOutcome::Completed)
1052        );
1053    }
1054
1055    /// Pins that a window too long to add to an instant never closes, and
1056    /// does not panic.
1057    #[test]
1058    fn a_window_too_long_for_an_instant_never_closes() {
1059        let start = Instant::now();
1060        let mut check = EndCheck::windowed(Duration::MAX, start);
1061        assert_eq!(check.next(start), EndAction::ReadMarker);
1062        check.marker_read(start, present("a"));
1063        empty_reads(&mut check, start, 0, 45);
1064        assert_eq!(
1065            check.next(at(start, 60)),
1066            EndAction::ReadDepth(DepthOf::Feed)
1067        );
1068    }
1069
1070    /// Pins that every count of the feed's depth, the unsent one included,
1071    /// holds the window open.
1072    #[test]
1073    fn each_count_holds_the_window_open() {
1074        let start = Instant::now();
1075        let setters: [fn(&mut Depth); 4] = [
1076            |d| d.visible = 1,
1077            |d| d.delayed = 1,
1078            |d| d.in_flight = 1,
1079            |d| d.unsent = 1,
1080        ];
1081        for set in setters {
1082            let mut check = opened(start, "a");
1083            assert_eq!(check.next(start), EndAction::ReadDepth(DepthOf::Feed));
1084            let mut depth = Depth::new("queue", true);
1085            set(&mut depth);
1086            check.depth_read(start, Ok(depth));
1087            assert_eq!(check.next(at(start, 15)), EndAction::ReadMarker);
1088        }
1089    }
1090}
spate_core/feed/end_check.rs

spate_core/feed/
end_check.rs
