Skip to main content

ljos_cli/
note_queue.rs

1//! Memory copies the pack did not take because it was down or did not know
2//! this seat.
3//!
4//! `ljos note` writes the logbook line first, then a `note` atom to the
5//! pack. When the pack is off, not answering, or answers 401, the fitted
6//! copy waits here, one JSON line per copy, and the next note that reaches
7//! the pack, `ljos finish`, or the next session start sends it. `ljos
8//! doctor` only counts it. A refusal is not queued: the pack would refuse
9//! it again. A line left
10//! from before that rule is cut to what the pack takes, or dropped, on the
11//! next send. The queue holds one entry per issue, text, workspace, kind
12//! and, for a lesson, the atoms it closes. packsetd answers an identical
13//! issue and text with the claim it already holds, so a resend after a
14//! lost answer stores nothing new.
15
16use std::path::{Path, PathBuf};
17
18use anyhow::{Context, Result};
19use serde::{Deserialize, Serialize};
20
21fn default_kind() -> String {
22    "note".into()
23}
24
25/// One memory waiting for the pack.
26#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
27pub struct Queued {
28    pub issue: String,
29    pub text: String,
30    /// The pack workspace the note was meant for.
31    pub workspace: String,
32    /// Unix seconds when the pack did not take it.
33    pub at: u64,
34    /// `note` or `lesson`. A line written before this field reads as `note`.
35    #[serde(default = "default_kind")]
36    pub kind: String,
37    /// Atoms a queued lesson closes. Empty for a note, and for a lesson
38    /// that closes nothing. A line written before this field reads as empty.
39    #[serde(default, skip_serializing_if = "Vec::is_empty")]
40    pub supersedes: Vec<String>,
41}
42
43impl Queued {
44    pub(crate) fn same(&self, other: &Self) -> bool {
45        self.issue.trim() == other.issue.trim()
46            && self.text.trim() == other.text.trim()
47            && self.workspace == other.workspace
48            && self.kind == other.kind
49            && self.supersedes == other.supersedes
50    }
51}
52
53/// What one send says about a queued note.
54pub enum Sent {
55    /// The pack holds it, under this id.
56    Kept(String),
57    /// The pack refused it; the reason. The note leaves the queue.
58    Refused(String),
59    /// The pack did not answer; the note and every later one wait.
60    Wait(String),
61}
62
63/// What a drain did.
64#[derive(Debug, Default, Clone, PartialEq, Eq)]
65pub struct Drained {
66    /// Issue and pack id of each note the pack now holds.
67    pub kept: Vec<(String, String)>,
68    /// Issue and reason of each note the pack refused.
69    pub refused: Vec<(String, String)>,
70    /// Notes still waiting.
71    pub left: usize,
72    /// Why the first waiting note did not go, when one did not.
73    pub waiting_on: Option<String>,
74    /// Queue bytes that were not a note, moved out before a rewrite.
75    pub set_aside: Vec<SetAside>,
76}
77
78/// Bytes of the queue a rewrite moved to a file beside it rather than
79/// drop: complete lines that did not parse and a last line without its
80/// newline.
81#[derive(Debug, Clone, PartialEq, Eq)]
82pub struct SetAside {
83    pub path: PathBuf,
84    pub bytes: usize,
85}
86
87/// The state dir that holds the queue: `$XDG_STATE_HOME/ljos`, else
88/// `~/.local/state/ljos`.
89#[must_use]
90pub fn state_dir() -> PathBuf {
91    std::env::var_os("XDG_STATE_HOME")
92        .map(PathBuf::from)
93        .or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".local/state")))
94        .unwrap_or_else(|| PathBuf::from(".local/state"))
95        .join("ljos")
96}
97
98fn queue_file(dir: &Path) -> PathBuf {
99    dir.join("note-queue.jsonl")
100}
101
102fn lock_file(dir: &Path) -> PathBuf {
103    dir.join("note-queue.lock")
104}
105
106fn lock(dir: &Path) -> crate::CommitLock {
107    let _ = std::fs::create_dir_all(dir);
108    crate::CommitLock::acquire_private(&lock_file(dir))
109}
110
111/// The queue lock in `dir` when another ljos holds it now, queueing or
112/// sending; tried without waiting for it.
113#[must_use]
114pub fn busy_in(dir: &Path) -> Option<PathBuf> {
115    let lock = lock_file(dir);
116    crate::CommitLock::is_held_private(&lock).then_some(lock)
117}
118
119fn read(dir: &Path) -> Vec<Queued> {
120    read_split(dir).0
121}
122
123/// The queue's notes, and its bytes that are not one: each complete line
124/// that does not parse, and a last line without its newline. Under the
125/// lock no writer is part way through that last line; every append holds
126/// it.
127fn read_split(dir: &Path) -> (Vec<Queued>, Vec<u8>) {
128    let bytes = crate::read_state_file(&queue_file(dir)).unwrap_or_default();
129    let mut notes = Vec::new();
130    let mut unread = Vec::new();
131    for line in bytes.split_inclusive(|&b| b == b'\n') {
132        match line.strip_suffix(b"\n").map(serde_json::from_slice) {
133            Some(Ok(note)) => notes.push(note),
134            _ if line == b"\n" => {}
135            _ => unread.extend_from_slice(line),
136        }
137    }
138    (notes, unread)
139}
140
141/// Where a rewrite on this day puts the queue bytes it could not read.
142fn damaged_file(dir: &Path) -> PathBuf {
143    dir.join(format!(
144        "note-queue.jsonl.{}.damaged",
145        chrono::Utc::now().format("%Y-%m-%d")
146    ))
147}
148
149/// [`set_aside_at`] for `dir`, opened the way every state-log open is: a
150/// link in place of `dir` is refused and lists nothing.
151#[must_use]
152pub fn set_aside_in(dir: &Path) -> Vec<(PathBuf, u64)> {
153    match crate::open_state_dir(&queue_file(dir)) {
154        Ok((fd, _)) => set_aside_at(&fd, dir),
155        Err(_) => Vec::new(),
156    }
157}
158
159/// The damaged files a rewrite left in the state dir open as `fd`, named
160/// below `dir`, with their sizes, oldest day first. Each name is listed
161/// and stated through `fd`, without following a link and without opening
162/// one, so name and size come from the same directory; a link or
163/// anything else that is not a regular file is left out, since a send
164/// never writes through one.
165#[must_use]
166pub fn set_aside_at(fd: &std::fs::File, dir: &Path) -> Vec<(PathBuf, u64)> {
167    let Ok(names) = crate::deedbag::list_names(fd) else {
168        return Vec::new();
169    };
170    let mut found: Vec<(PathBuf, u64)> = names
171        .iter()
172        .filter_map(|name| {
173            let text = name.to_str().ok()?;
174            text.strip_prefix("note-queue.jsonl.")?
175                .strip_suffix(".damaged")?;
176            let st = crate::deedbag::stat_at(fd, name).ok()??;
177            let size = u64::try_from(st.st_size).ok()?;
178            ((st.st_mode & libc::S_IFMT) == libc::S_IFREG).then(|| (dir.join(text), size))
179        })
180        .collect();
181    found.sort();
182    found
183}
184
185/// Add `unread` to the day's damaged file and put it on disk, before the
186/// rewrite that leaves those bytes out of the queue.
187fn set_aside(dir: &Path, unread: &[u8]) -> Result<Option<SetAside>> {
188    if unread.is_empty() {
189        return Ok(None);
190    }
191    let path = damaged_file(dir);
192    crate::append_locked_row(&path, unread, crate::Durability::Durable)
193        .and_then(|()| crate::sync_state_file(&path))
194        .with_context(|| format!("set aside unreadable queue bytes in {}", path.display()))?;
195    Ok(Some(SetAside {
196        path,
197        bytes: unread.len(),
198    }))
199}
200
201/// Rewrite the queue to `notes`. `unread`, what [`read_split`] could not
202/// read, is set aside first, so a rewrite drops no byte it could not
203/// parse.
204fn write(dir: &Path, notes: &[Queued], unread: &[u8]) -> Result<Option<SetAside>> {
205    let aside = set_aside(dir, unread)?;
206    let path = queue_file(dir);
207    if notes.is_empty() {
208        // Opened, not stat'ed through the path: a dangling link is refused
209        // instead of read as no queue.
210        let present = crate::open_state_dir(&path)
211            .and_then(|(fd, name)| crate::open_regular_at(&fd, &name, &path, libc::O_RDONLY, 0));
212        match present {
213            Ok(_) => crate::rewrite_state_file(&path, b"")
214                .with_context(|| format!("empty {}", path.display()))?,
215            Err(err) if err.raw_os_error() == Some(libc::ENOENT) => {}
216            Err(err) => return Err(err).with_context(|| format!("empty {}", path.display())),
217        }
218        return Ok(aside);
219    }
220    let mut body = String::new();
221    for note in notes {
222        body.push_str(&serde_json::to_string(note)?);
223        body.push('\n');
224    }
225    crate::rewrite_state_file(&path, body.as_bytes())
226        .with_context(|| format!("replace {}", path.display()))?;
227    Ok(aside)
228}
229
230/// The notes waiting in `dir`, oldest first.
231#[must_use]
232pub fn pending_in(dir: &Path) -> Vec<Queued> {
233    let _held = lock(dir);
234    read(dir)
235}
236
237/// Replace the waiting memories in `dir`.
238///
239/// # Errors
240///
241/// The queue file cannot be rewritten.
242pub fn replace_in(dir: &Path, notes: Vec<Queued>) -> Result<()> {
243    let _held = lock(dir);
244    let (_, unread) = read_split(dir);
245    write(dir, &notes, &unread).map(drop)
246}
247
248/// Fit the waiting memories in `dir` with `fit` and set aside the bytes
249/// that are not a memory, rewriting the queue only when either changes
250/// it. Reading and rewriting under one lock keeps a note queued between
251/// them.
252///
253/// # Errors
254///
255/// The queue file cannot be rewritten, or the bytes cannot be set aside.
256pub fn tidy_in(
257    dir: &Path,
258    fit: impl FnOnce(Vec<Queued>) -> Vec<Queued>,
259) -> Result<Option<SetAside>> {
260    let _held = lock(dir);
261    let (notes, unread) = read_split(dir);
262    let fitted = fit(notes.clone());
263    if fitted == notes && unread.is_empty() {
264        return Ok(None);
265    }
266    write(dir, &fitted, &unread)
267}
268
269/// Add `note` to the queue in `dir`. `false` when the same copy already
270/// waits there, including the atoms a lesson closes.
271///
272/// # Errors
273///
274/// The queue file cannot be written.
275pub fn push_in(dir: &Path, note: Queued) -> Result<bool> {
276    let _held = lock(dir);
277    if read(dir).iter().any(|seen| seen.same(&note)) {
278        return Ok(false);
279    }
280    let path = queue_file(dir);
281    let mut line = serde_json::to_string(&note)?;
282    line.push('\n');
283    crate::append_locked_row(&path, line.as_bytes(), crate::Durability::Durable)
284        .with_context(|| format!("append {}", path.display()))?;
285    Ok(true)
286}
287
288/// Send each waiting note in `dir` through `send`, oldest first. A kept
289/// or refused note leaves the queue; the first one that has to wait stops
290/// the drain, and it and every later note stay, in order.
291///
292/// # Errors
293///
294/// The queue file cannot be rewritten.
295pub fn drain_in(dir: &Path, mut send: impl FnMut(&Queued) -> Sent) -> Result<Drained> {
296    let _held = lock(dir);
297    let (notes, unread) = read_split(dir);
298    let mut done = Drained::default();
299    if notes.is_empty() && unread.is_empty() {
300        return Ok(done);
301    }
302    let mut rest = Vec::new();
303    for (at, note) in notes.iter().enumerate() {
304        match send(note) {
305            Sent::Kept(id) => done.kept.push((note.issue.clone(), id)),
306            Sent::Refused(why) => done.refused.push((note.issue.clone(), why)),
307            Sent::Wait(why) => {
308                done.waiting_on = Some(why);
309                rest.extend_from_slice(&notes[at..]);
310                break;
311            }
312        }
313    }
314    done.left = rest.len();
315    done.set_aside.extend(write(dir, &rest, &unread)?);
316    Ok(done)
317}
318
319/// The lines a drain prints: what went, what the pack refused, and what
320/// still waits. Empty when nothing was queued.
321#[must_use]
322pub fn drained_lines(done: &Drained) -> String {
323    let mut out = String::new();
324    if !done.kept.is_empty() {
325        let ids: Vec<String> = done
326            .kept
327            .iter()
328            .map(|(issue, id)| format!("{issue} {}", id.chars().take(8).collect::<String>()))
329            .collect();
330        out.push_str(&format!(
331            "memory: sent {} queued note(s): {}\n",
332            done.kept.len(),
333            ids.join(", ")
334        ));
335    }
336    for (issue, why) in &done.refused {
337        out.push_str(&format!(
338            "memory: the pack refused the queued note on {issue} ({why}); it stays on the logbook\n"
339        ));
340    }
341    for aside in &done.set_aside {
342        out.push_str(&format!(
343            "memory: moved {} byte(s) of the note queue that are not a note to {}; they were not sent\n",
344            aside.bytes,
345            aside.path.display()
346        ));
347    }
348    if done.left > 0 {
349        out.push_str(&format!(
350            "memory: {} note(s) still queued ({})\n",
351            done.left,
352            done.waiting_on
353                .as_deref()
354                .unwrap_or("the pack did not answer")
355        ));
356    }
357    out
358}
359
360#[cfg(test)]
361mod tests {
362    use super::*;
363
364    fn note(issue: &str, text: &str) -> Queued {
365        Queued {
366            issue: issue.into(),
367            text: text.into(),
368            workspace: "ws".into(),
369            at: 1,
370            kind: "note".into(),
371            supersedes: Vec::new(),
372        }
373    }
374
375    fn dir(name: &str) -> PathBuf {
376        let dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
377            .join("../../target/note-queue")
378            .join(name);
379        let _ = std::fs::remove_dir_all(&dir);
380        std::fs::create_dir_all(&dir).unwrap();
381        dir
382    }
383
384    #[test]
385    fn an_old_line_without_kind_is_a_note() {
386        let dir = dir("old");
387        std::fs::write(
388            dir.join("note-queue.jsonl"),
389            "{\"issue\":\"p-1\",\"text\":\"one line\",\"workspace\":\"ws\",\"at\":1}\n",
390        )
391        .unwrap();
392        let pending = pending_in(&dir);
393        assert_eq!(pending.len(), 1);
394        assert_eq!(pending[0].kind, "note");
395        assert_eq!(pending[0].text, "one line");
396        assert!(pending[0].supersedes.is_empty());
397    }
398
399    #[test]
400    fn a_lesson_that_closes_another_atom_is_a_different_copy() {
401        let dir = dir("closes");
402        let mut lesson = note("p-1", "one line");
403        lesson.kind = "lesson".into();
404        lesson.supersedes = vec!["atom-old".into()];
405        assert!(push_in(&dir, lesson.clone()).unwrap());
406        assert!(!push_in(&dir, lesson.clone()).unwrap());
407        let mut other = lesson;
408        other.supersedes = vec!["atom-new".into()];
409        assert!(push_in(&dir, other).unwrap());
410        assert_eq!(pending_in(&dir).len(), 2);
411    }
412
413    /// A copy still being appended is not yet on the queue; the ones
414    /// before it are.
415    #[test]
416    fn a_torn_last_copy_leaves_the_queue() {
417        let dir = dir("torn");
418        let mut bytes =
419            format!("{}\n", serde_json::to_string(&note("p-1", "kept")).unwrap()).into_bytes();
420        bytes.extend_from_slice(b"{\"issue\":\"p-2\",\"text\":\"caf\xc3");
421        std::fs::write(dir.join("note-queue.jsonl"), bytes).unwrap();
422        let pending = pending_in(&dir);
423        assert_eq!(pending.len(), 1, "{pending:?}");
424        assert_eq!(pending[0].text, "kept");
425    }
426
427    /// The queue is one JSON line per copy. writeln! of that line is two
428    /// write calls, and the sidecar lock is skipped when it cannot be
429    /// taken, so two copies can still land on one line.
430    #[test]
431    fn concurrent_queue_rows_each_parse() {
432        use std::os::unix::fs::PermissionsExt;
433        let dir = dir("concurrent");
434        const THREADS: usize = 32;
435        const EACH: usize = 4;
436        let mut handles = Vec::with_capacity(THREADS);
437        for thread in 0..THREADS {
438            let dir = dir.clone();
439            handles.push(std::thread::spawn(move || {
440                for n in 0..EACH {
441                    let mut queued = note(&format!("p-{thread}"), &format!("text {thread}-{n}"));
442                    queued.at = (thread * EACH + n) as u64;
443                    push_in(&dir, queued).unwrap();
444                }
445            }));
446        }
447        for handle in handles {
448            handle.join().expect("queue writer");
449        }
450        let path = dir.join("note-queue.jsonl");
451        let text = std::fs::read_to_string(&path).unwrap();
452        let rows: Vec<Queued> = text
453            .lines()
454            .map(|line| {
455                serde_json::from_str(line).unwrap_or_else(|err| {
456                    panic!("queue row does not parse ({err}): {line}");
457                })
458            })
459            .collect();
460        assert_eq!(rows.len(), THREADS * EACH, "{text}");
461        let mode = std::fs::symlink_metadata(&path)
462            .unwrap()
463            .permissions()
464            .mode()
465            & 0o777;
466        assert_eq!(mode, 0o600, "queue mode {mode:o}");
467    }
468
469    /// A drain rewrites the queue. The file and the lock are ljos's own, so
470    /// the rewrite lands mode 0600, not the 0644 a plain `tmp + rename`
471    /// leaves, and a symlink swapped in for the queue file is refused rather
472    /// than written through to its target.
473    #[test]
474    fn a_rewrite_stays_private_and_refuses_a_symlink() {
475        use std::os::unix::fs::PermissionsExt;
476        let dir = dir("rewrite");
477        push_in(&dir, note("p-1", "a")).unwrap();
478        push_in(&dir, note("p-2", "b")).unwrap();
479        let done = drain_in(&dir, |n| match n.issue.as_str() {
480            "p-1" => Sent::Kept("id".into()),
481            _ => Sent::Wait("down".into()),
482        })
483        .unwrap();
484        assert_eq!(done.left, 1);
485        let path = queue_file(&dir);
486        let mode = std::fs::symlink_metadata(&path)
487            .unwrap()
488            .permissions()
489            .mode()
490            & 0o777;
491        assert_eq!(mode, 0o600, "queue mode after rewrite {mode:o}");
492        let lock_mode = std::fs::symlink_metadata(dir.join("note-queue.lock"))
493            .unwrap()
494            .permissions()
495            .mode()
496            & 0o777;
497        assert_eq!(lock_mode, 0o600, "lock mode {lock_mode:o}");
498
499        let outside = dir.join("outside");
500        std::fs::write(&outside, "kept\n").unwrap();
501        std::fs::remove_file(&path).unwrap();
502        std::os::unix::fs::symlink(&outside, &path).unwrap();
503        assert!(replace_in(&dir, vec![note("p-9", "forged")]).is_err());
504        assert_eq!(std::fs::read_to_string(&outside).unwrap(), "kept\n");
505    }
506
507    /// Emptying a queue that is a dangling link is refused, and the link
508    /// stays for doctor to name.
509    #[test]
510    fn emptying_a_dangling_linked_queue_is_refused() {
511        let dir = dir("empty-dangling");
512        let path = queue_file(&dir);
513        std::os::unix::fs::symlink(dir.join("gone"), &path).unwrap();
514        assert!(replace_in(&dir, Vec::new()).is_err());
515        assert!(std::fs::symlink_metadata(&path).unwrap().is_symlink());
516    }
517
518    /// A drain rewrites the queue with the notes left. A line that does
519    /// not parse and a last line without its newline, even one that
520    /// would, go to the day's damaged file byte for byte, not out with
521    /// the rewrite.
522    #[test]
523    fn a_drain_sets_aside_what_it_cannot_read() {
524        use std::os::unix::fs::PermissionsExt;
525        let dir = dir("aside");
526        let sent = format!("{}\n", serde_json::to_string(&note("p-1", "sent")).unwrap());
527        let waits = format!(
528            "{}\n",
529            serde_json::to_string(&note("p-2", "waits")).unwrap()
530        );
531        let cut: &[u8] = b"{\"issue\":\"p-3\",\"te\n";
532        let tail: &[u8] =
533            br#"{"issue":"p-4","text":"no newline","workspace":"ws","at":1,"kind":"note"}"#;
534        let mut bytes = sent.into_bytes();
535        bytes.extend_from_slice(cut);
536        bytes.extend_from_slice(waits.as_bytes());
537        bytes.extend_from_slice(tail);
538        std::fs::write(queue_file(&dir), &bytes).unwrap();
539        let done = drain_in(&dir, |n| match n.issue.as_str() {
540            "p-1" => Sent::Kept("id".into()),
541            _ => Sent::Wait("down".into()),
542        })
543        .unwrap();
544        assert_eq!(done.left, 1, "{done:?}");
545        assert_eq!(std::fs::read_to_string(queue_file(&dir)).unwrap(), waits);
546        let unread = [cut, tail].concat();
547        let [aside] = done.set_aside.as_slice() else {
548            panic!("{done:?}");
549        };
550        assert_eq!(aside.bytes, unread.len());
551        assert_eq!(std::fs::read(&aside.path).unwrap(), unread);
552        let name = aside.path.file_name().unwrap().to_str().unwrap();
553        assert!(
554            name.starts_with("note-queue.jsonl.") && name.ends_with(".damaged"),
555            "{name}"
556        );
557        let mode = std::fs::metadata(&aside.path).unwrap().permissions().mode() & 0o777;
558        assert_eq!(mode, 0o600, "damaged file mode {mode:o}");
559        let lines = drained_lines(&done);
560        assert!(lines.contains(&aside.path.display().to_string()), "{lines}");
561    }
562
563    /// A queue of nothing but bytes that are not a note is emptied into
564    /// the damaged file, and the next tidy finds nothing to move.
565    #[test]
566    fn a_tidy_sets_aside_a_queue_with_no_note_in_it() {
567        let dir = dir("tidy");
568        std::fs::write(queue_file(&dir), b"{\"issue\":\"p-1\",\"te").unwrap();
569        let aside = tidy_in(&dir, |notes| notes).unwrap().expect("set aside");
570        assert_eq!(
571            std::fs::read(&aside.path).unwrap(),
572            b"{\"issue\":\"p-1\",\"te"
573        );
574        assert_eq!(std::fs::read(queue_file(&dir)).unwrap(), b"");
575        assert_eq!(tidy_in(&dir, |notes| notes).unwrap(), None);
576    }
577
578    /// A damaged name that is a link, to a file outside the state dir or
579    /// to nothing, is not listed, and its target is neither read nor
580    /// changed.
581    #[test]
582    fn a_linked_damaged_name_is_not_followed() {
583        let dir = dir("linked");
584        let real = dir.join("note-queue.jsonl.2026-01-01.damaged");
585        std::fs::write(&real, b"cut").unwrap();
586        let outside = dir.parent().unwrap().join("linked-outside");
587        std::fs::write(&outside, b"not ljos's own").unwrap();
588        std::os::unix::fs::symlink(&outside, dir.join("note-queue.jsonl.2026-01-02.damaged"))
589            .unwrap();
590        std::os::unix::fs::symlink(
591            dir.join("gone"),
592            dir.join("note-queue.jsonl.2026-01-03.damaged"),
593        )
594        .unwrap();
595        assert_eq!(set_aside_in(&dir), [(real, 3)]);
596        assert_eq!(std::fs::read(&outside).unwrap(), b"not ljos's own");
597        assert!(!dir.join("gone").exists());
598    }
599
600    /// A state dir swapped for a link to a dir that holds a damaged file
601    /// lists nothing from that dir.
602    #[test]
603    fn a_linked_state_dir_lists_nothing_from_outside() {
604        let root = dir("linked-dir");
605        let outside = root.join("outside");
606        std::fs::create_dir(&outside).unwrap();
607        std::fs::write(outside.join("note-queue.jsonl.2026-01-01.damaged"), b"x").unwrap();
608        let state = root.join("ljos");
609        std::os::unix::fs::symlink(&outside, &state).unwrap();
610        assert_eq!(set_aside_in(&state), []);
611    }
612
613    #[test]
614    fn the_same_note_waits_once() {
615        let dir = dir("once");
616        assert!(push_in(&dir, note("p-1", "one line")).unwrap());
617        assert!(!push_in(&dir, note("p-1", " one line ")).unwrap());
618        assert!(push_in(&dir, note("p-2", "one line")).unwrap());
619        assert_eq!(pending_in(&dir).len(), 2);
620    }
621
622    #[test]
623    fn a_drain_keeps_order_and_stops_at_the_first_wait() {
624        let dir = dir("order");
625        for (issue, text) in [("p-1", "a"), ("p-2", "b"), ("p-3", "c"), ("p-4", "d")] {
626            push_in(&dir, note(issue, text)).unwrap();
627        }
628        let mut seen = Vec::new();
629        let done = drain_in(&dir, |n| {
630            seen.push(n.issue.clone());
631            match n.issue.as_str() {
632                "p-1" => Sent::Kept("id-1".into()),
633                "p-2" => Sent::Refused("too long".into()),
634                _ => Sent::Wait("connection refused".into()),
635            }
636        })
637        .unwrap();
638        assert_eq!(seen, ["p-1", "p-2", "p-3"]);
639        assert_eq!(done.kept, [("p-1".to_string(), "id-1".to_string())]);
640        assert_eq!(done.refused.len(), 1);
641        assert_eq!(done.left, 2);
642        let left: Vec<String> = pending_in(&dir).into_iter().map(|n| n.issue).collect();
643        assert_eq!(left, ["p-3", "p-4"]);
644        let lines = drained_lines(&done);
645        assert!(lines.contains("sent 1 queued note(s): p-1 id-1"), "{lines}");
646        assert!(
647            lines.contains("2 note(s) still queued (connection refused)"),
648            "{lines}"
649        );
650        let done = drain_in(&dir, |_| Sent::Kept("id".into())).unwrap();
651        assert_eq!(done.kept.len(), 2);
652        assert!(pending_in(&dir).is_empty());
653        assert_eq!(
654            drained_lines(&drain_in(&dir, |_| unreachable!()).unwrap()),
655            ""
656        );
657    }
658}