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 _durable = crate::bookkeep::durable_write();
206    let aside = set_aside(dir, unread)?;
207    let path = queue_file(dir);
208    if notes.is_empty() {
209        // Opened, not stat'ed through the path: a dangling link is refused
210        // instead of read as no queue.
211        let present = crate::open_state_dir(&path)
212            .and_then(|(fd, name)| crate::open_regular_at(&fd, &name, &path, libc::O_RDONLY, 0));
213        match present {
214            Ok(_) => crate::rewrite_state_file(&path, b"")
215                .with_context(|| format!("empty {}", path.display()))?,
216            Err(err) if err.raw_os_error() == Some(libc::ENOENT) => {}
217            Err(err) => return Err(err).with_context(|| format!("empty {}", path.display())),
218        }
219        return Ok(aside);
220    }
221    let mut body = String::new();
222    for note in notes {
223        body.push_str(&serde_json::to_string(note)?);
224        body.push('\n');
225    }
226    crate::rewrite_state_file(&path, body.as_bytes())
227        .with_context(|| format!("replace {}", path.display()))?;
228    Ok(aside)
229}
230
231/// The notes waiting in `dir`, oldest first.
232#[must_use]
233pub fn pending_in(dir: &Path) -> Vec<Queued> {
234    let _held = lock(dir);
235    read(dir)
236}
237
238/// Replace the waiting memories in `dir`.
239///
240/// # Errors
241///
242/// The queue file cannot be rewritten.
243pub fn replace_in(dir: &Path, notes: Vec<Queued>) -> Result<()> {
244    let _held = lock(dir);
245    let (_, unread) = read_split(dir);
246    write(dir, &notes, &unread).map(drop)
247}
248
249/// Fit the waiting memories in `dir` with `fit` and set aside the bytes
250/// that are not a memory, rewriting the queue only when either changes
251/// it. Reading and rewriting under one lock keeps a note queued between
252/// them.
253///
254/// # Errors
255///
256/// The queue file cannot be rewritten, or the bytes cannot be set aside.
257pub fn tidy_in(
258    dir: &Path,
259    fit: impl FnOnce(Vec<Queued>) -> Vec<Queued>,
260) -> Result<Option<SetAside>> {
261    let _held = lock(dir);
262    let (notes, unread) = read_split(dir);
263    let fitted = fit(notes.clone());
264    if fitted == notes && unread.is_empty() {
265        return Ok(None);
266    }
267    write(dir, &fitted, &unread)
268}
269
270/// Add `note` to the queue in `dir`. `false` when the same copy already
271/// waits there, including the atoms a lesson closes.
272///
273/// # Errors
274///
275/// The queue file cannot be written.
276pub fn push_in(dir: &Path, note: Queued) -> Result<bool> {
277    let _held = lock(dir);
278    if read(dir).iter().any(|seen| seen.same(&note)) {
279        return Ok(false);
280    }
281    let path = queue_file(dir);
282    let mut line = serde_json::to_string(&note)?;
283    line.push('\n');
284    crate::append_locked_row(&path, line.as_bytes(), crate::Durability::Durable)
285        .with_context(|| format!("append {}", path.display()))?;
286    Ok(true)
287}
288
289/// Send each waiting note in `dir` through `send`, oldest first. A kept
290/// or refused note leaves the queue; the first one that has to wait stops
291/// the drain, and it and every later note stay, in order.
292///
293/// # Errors
294///
295/// The queue file cannot be rewritten.
296pub fn drain_in(dir: &Path, mut send: impl FnMut(&Queued) -> Sent) -> Result<Drained> {
297    let _held = lock(dir);
298    let (notes, unread) = read_split(dir);
299    let mut done = Drained::default();
300    if notes.is_empty() && unread.is_empty() {
301        return Ok(done);
302    }
303    let mut rest = Vec::new();
304    for (at, note) in notes.iter().enumerate() {
305        match send(note) {
306            Sent::Kept(id) => done.kept.push((note.issue.clone(), id)),
307            Sent::Refused(why) => done.refused.push((note.issue.clone(), why)),
308            Sent::Wait(why) => {
309                done.waiting_on = Some(why);
310                rest.extend_from_slice(&notes[at..]);
311                break;
312            }
313        }
314    }
315    done.left = rest.len();
316    done.set_aside.extend(write(dir, &rest, &unread)?);
317    Ok(done)
318}
319
320/// The lines a drain prints: what went, what the pack refused, and what
321/// still waits. Empty when nothing was queued.
322#[must_use]
323pub fn drained_lines(done: &Drained) -> String {
324    let mut out = String::new();
325    if !done.kept.is_empty() {
326        let ids: Vec<String> = done
327            .kept
328            .iter()
329            .map(|(issue, id)| format!("{issue} {}", id.chars().take(8).collect::<String>()))
330            .collect();
331        out.push_str(&format!(
332            "memory: sent {} queued note(s): {}\n",
333            done.kept.len(),
334            ids.join(", ")
335        ));
336    }
337    for (issue, why) in &done.refused {
338        out.push_str(&format!(
339            "memory: the pack refused the queued note on {issue} ({why}); it stays on the logbook\n"
340        ));
341    }
342    for aside in &done.set_aside {
343        out.push_str(&format!(
344            "memory: moved {} byte(s) of the note queue that are not a note to {}; they were not sent\n",
345            aside.bytes,
346            aside.path.display()
347        ));
348    }
349    if done.left > 0 {
350        out.push_str(&format!(
351            "memory: {} note(s) still queued ({})\n",
352            done.left,
353            done.waiting_on
354                .as_deref()
355                .unwrap_or("the pack did not answer")
356        ));
357    }
358    out
359}
360
361#[cfg(test)]
362mod tests {
363    use super::*;
364
365    fn note(issue: &str, text: &str) -> Queued {
366        Queued {
367            issue: issue.into(),
368            text: text.into(),
369            workspace: "ws".into(),
370            at: 1,
371            kind: "note".into(),
372            supersedes: Vec::new(),
373        }
374    }
375
376    fn dir(name: &str) -> PathBuf {
377        let dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
378            .join("../../target/note-queue")
379            .join(name);
380        let _ = std::fs::remove_dir_all(&dir);
381        std::fs::create_dir_all(&dir).unwrap();
382        dir
383    }
384
385    #[test]
386    fn an_old_line_without_kind_is_a_note() {
387        let dir = dir("old");
388        std::fs::write(
389            dir.join("note-queue.jsonl"),
390            "{\"issue\":\"p-1\",\"text\":\"one line\",\"workspace\":\"ws\",\"at\":1}\n",
391        )
392        .unwrap();
393        let pending = pending_in(&dir);
394        assert_eq!(pending.len(), 1);
395        assert_eq!(pending[0].kind, "note");
396        assert_eq!(pending[0].text, "one line");
397        assert!(pending[0].supersedes.is_empty());
398    }
399
400    #[test]
401    fn a_lesson_that_closes_another_atom_is_a_different_copy() {
402        let dir = dir("closes");
403        let mut lesson = note("p-1", "one line");
404        lesson.kind = "lesson".into();
405        lesson.supersedes = vec!["atom-old".into()];
406        assert!(push_in(&dir, lesson.clone()).unwrap());
407        assert!(!push_in(&dir, lesson.clone()).unwrap());
408        let mut other = lesson;
409        other.supersedes = vec!["atom-new".into()];
410        assert!(push_in(&dir, other).unwrap());
411        assert_eq!(pending_in(&dir).len(), 2);
412    }
413
414    /// A copy still being appended is not yet on the queue; the ones
415    /// before it are.
416    #[test]
417    fn a_torn_last_copy_leaves_the_queue() {
418        let dir = dir("torn");
419        let mut bytes =
420            format!("{}\n", serde_json::to_string(&note("p-1", "kept")).unwrap()).into_bytes();
421        bytes.extend_from_slice(b"{\"issue\":\"p-2\",\"text\":\"caf\xc3");
422        std::fs::write(dir.join("note-queue.jsonl"), bytes).unwrap();
423        let pending = pending_in(&dir);
424        assert_eq!(pending.len(), 1, "{pending:?}");
425        assert_eq!(pending[0].text, "kept");
426    }
427
428    /// The queue is one JSON line per copy. writeln! of that line is two
429    /// write calls, and the sidecar lock is skipped when it cannot be
430    /// taken, so two copies can still land on one line.
431    #[test]
432    fn concurrent_queue_rows_each_parse() {
433        use std::os::unix::fs::PermissionsExt;
434        let dir = dir("concurrent");
435        const THREADS: usize = 32;
436        const EACH: usize = 4;
437        let mut handles = Vec::with_capacity(THREADS);
438        for thread in 0..THREADS {
439            let dir = dir.clone();
440            handles.push(std::thread::spawn(move || {
441                for n in 0..EACH {
442                    let mut queued = note(&format!("p-{thread}"), &format!("text {thread}-{n}"));
443                    queued.at = (thread * EACH + n) as u64;
444                    push_in(&dir, queued).unwrap();
445                }
446            }));
447        }
448        for handle in handles {
449            handle.join().expect("queue writer");
450        }
451        let path = dir.join("note-queue.jsonl");
452        let text = std::fs::read_to_string(&path).unwrap();
453        let rows: Vec<Queued> = text
454            .lines()
455            .map(|line| {
456                serde_json::from_str(line).unwrap_or_else(|err| {
457                    panic!("queue row does not parse ({err}): {line}");
458                })
459            })
460            .collect();
461        assert_eq!(rows.len(), THREADS * EACH, "{text}");
462        let mode = std::fs::symlink_metadata(&path)
463            .unwrap()
464            .permissions()
465            .mode()
466            & 0o777;
467        assert_eq!(mode, 0o600, "queue mode {mode:o}");
468    }
469
470    /// A drain rewrites the queue. The file and the lock are ljos's own, so
471    /// the rewrite lands mode 0600, not the 0644 a plain `tmp + rename`
472    /// leaves, and a symlink swapped in for the queue file is refused rather
473    /// than written through to its target.
474    #[test]
475    fn a_rewrite_stays_private_and_refuses_a_symlink() {
476        use std::os::unix::fs::PermissionsExt;
477        let dir = dir("rewrite");
478        push_in(&dir, note("p-1", "a")).unwrap();
479        push_in(&dir, note("p-2", "b")).unwrap();
480        let done = drain_in(&dir, |n| match n.issue.as_str() {
481            "p-1" => Sent::Kept("id".into()),
482            _ => Sent::Wait("down".into()),
483        })
484        .unwrap();
485        assert_eq!(done.left, 1);
486        let path = queue_file(&dir);
487        let mode = std::fs::symlink_metadata(&path)
488            .unwrap()
489            .permissions()
490            .mode()
491            & 0o777;
492        assert_eq!(mode, 0o600, "queue mode after rewrite {mode:o}");
493        let lock_mode = std::fs::symlink_metadata(dir.join("note-queue.lock"))
494            .unwrap()
495            .permissions()
496            .mode()
497            & 0o777;
498        assert_eq!(lock_mode, 0o600, "lock mode {lock_mode:o}");
499
500        let outside = dir.join("outside");
501        std::fs::write(&outside, "kept\n").unwrap();
502        std::fs::remove_file(&path).unwrap();
503        std::os::unix::fs::symlink(&outside, &path).unwrap();
504        assert!(replace_in(&dir, vec![note("p-9", "forged")]).is_err());
505        assert_eq!(std::fs::read_to_string(&outside).unwrap(), "kept\n");
506    }
507
508    /// Emptying a queue that is a dangling link is refused, and the link
509    /// stays for doctor to name.
510    #[test]
511    fn emptying_a_dangling_linked_queue_is_refused() {
512        let dir = dir("empty-dangling");
513        let path = queue_file(&dir);
514        std::os::unix::fs::symlink(dir.join("gone"), &path).unwrap();
515        assert!(replace_in(&dir, Vec::new()).is_err());
516        assert!(std::fs::symlink_metadata(&path).unwrap().is_symlink());
517    }
518
519    /// A drain rewrites the queue with the notes left. A line that does
520    /// not parse and a last line without its newline, even one that
521    /// would, go to the day's damaged file byte for byte, not out with
522    /// the rewrite.
523    #[test]
524    fn a_drain_sets_aside_what_it_cannot_read() {
525        use std::os::unix::fs::PermissionsExt;
526        let dir = dir("aside");
527        let sent = format!("{}\n", serde_json::to_string(&note("p-1", "sent")).unwrap());
528        let waits = format!(
529            "{}\n",
530            serde_json::to_string(&note("p-2", "waits")).unwrap()
531        );
532        let cut: &[u8] = b"{\"issue\":\"p-3\",\"te\n";
533        let tail: &[u8] =
534            br#"{"issue":"p-4","text":"no newline","workspace":"ws","at":1,"kind":"note"}"#;
535        let mut bytes = sent.into_bytes();
536        bytes.extend_from_slice(cut);
537        bytes.extend_from_slice(waits.as_bytes());
538        bytes.extend_from_slice(tail);
539        std::fs::write(queue_file(&dir), &bytes).unwrap();
540        let done = drain_in(&dir, |n| match n.issue.as_str() {
541            "p-1" => Sent::Kept("id".into()),
542            _ => Sent::Wait("down".into()),
543        })
544        .unwrap();
545        assert_eq!(done.left, 1, "{done:?}");
546        assert_eq!(std::fs::read_to_string(queue_file(&dir)).unwrap(), waits);
547        let unread = [cut, tail].concat();
548        let [aside] = done.set_aside.as_slice() else {
549            panic!("{done:?}");
550        };
551        assert_eq!(aside.bytes, unread.len());
552        assert_eq!(std::fs::read(&aside.path).unwrap(), unread);
553        let name = aside.path.file_name().unwrap().to_str().unwrap();
554        assert!(
555            name.starts_with("note-queue.jsonl.") && name.ends_with(".damaged"),
556            "{name}"
557        );
558        let mode = std::fs::metadata(&aside.path).unwrap().permissions().mode() & 0o777;
559        assert_eq!(mode, 0o600, "damaged file mode {mode:o}");
560        let lines = drained_lines(&done);
561        assert!(lines.contains(&aside.path.display().to_string()), "{lines}");
562    }
563
564    /// A queue of nothing but bytes that are not a note is emptied into
565    /// the damaged file, and the next tidy finds nothing to move.
566    #[test]
567    fn a_tidy_sets_aside_a_queue_with_no_note_in_it() {
568        let dir = dir("tidy");
569        std::fs::write(queue_file(&dir), b"{\"issue\":\"p-1\",\"te").unwrap();
570        let aside = tidy_in(&dir, |notes| notes).unwrap().expect("set aside");
571        assert_eq!(
572            std::fs::read(&aside.path).unwrap(),
573            b"{\"issue\":\"p-1\",\"te"
574        );
575        assert_eq!(std::fs::read(queue_file(&dir)).unwrap(), b"");
576        assert_eq!(tidy_in(&dir, |notes| notes).unwrap(), None);
577    }
578
579    /// A damaged name that is a link, to a file outside the state dir or
580    /// to nothing, is not listed, and its target is neither read nor
581    /// changed.
582    #[test]
583    fn a_linked_damaged_name_is_not_followed() {
584        let dir = dir("linked");
585        let real = dir.join("note-queue.jsonl.2026-01-01.damaged");
586        std::fs::write(&real, b"cut").unwrap();
587        let outside = dir.parent().unwrap().join("linked-outside");
588        std::fs::write(&outside, b"not ljos's own").unwrap();
589        std::os::unix::fs::symlink(&outside, dir.join("note-queue.jsonl.2026-01-02.damaged"))
590            .unwrap();
591        std::os::unix::fs::symlink(
592            dir.join("gone"),
593            dir.join("note-queue.jsonl.2026-01-03.damaged"),
594        )
595        .unwrap();
596        assert_eq!(set_aside_in(&dir), [(real, 3)]);
597        assert_eq!(std::fs::read(&outside).unwrap(), b"not ljos's own");
598        assert!(!dir.join("gone").exists());
599    }
600
601    /// A state dir swapped for a link to a dir that holds a damaged file
602    /// lists nothing from that dir.
603    #[test]
604    fn a_linked_state_dir_lists_nothing_from_outside() {
605        let root = dir("linked-dir");
606        let outside = root.join("outside");
607        std::fs::create_dir(&outside).unwrap();
608        std::fs::write(outside.join("note-queue.jsonl.2026-01-01.damaged"), b"x").unwrap();
609        let state = root.join("ljos");
610        std::os::unix::fs::symlink(&outside, &state).unwrap();
611        assert_eq!(set_aside_in(&state), []);
612    }
613
614    #[test]
615    fn the_same_note_waits_once() {
616        let dir = dir("once");
617        assert!(push_in(&dir, note("p-1", "one line")).unwrap());
618        assert!(!push_in(&dir, note("p-1", " one line ")).unwrap());
619        assert!(push_in(&dir, note("p-2", "one line")).unwrap());
620        assert_eq!(pending_in(&dir).len(), 2);
621    }
622
623    #[test]
624    fn a_drain_keeps_order_and_stops_at_the_first_wait() {
625        let dir = dir("order");
626        for (issue, text) in [("p-1", "a"), ("p-2", "b"), ("p-3", "c"), ("p-4", "d")] {
627            push_in(&dir, note(issue, text)).unwrap();
628        }
629        let mut seen = Vec::new();
630        let done = drain_in(&dir, |n| {
631            seen.push(n.issue.clone());
632            match n.issue.as_str() {
633                "p-1" => Sent::Kept("id-1".into()),
634                "p-2" => Sent::Refused("too long".into()),
635                _ => Sent::Wait("connection refused".into()),
636            }
637        })
638        .unwrap();
639        assert_eq!(seen, ["p-1", "p-2", "p-3"]);
640        assert_eq!(done.kept, [("p-1".to_string(), "id-1".to_string())]);
641        assert_eq!(done.refused.len(), 1);
642        assert_eq!(done.left, 2);
643        let left: Vec<String> = pending_in(&dir).into_iter().map(|n| n.issue).collect();
644        assert_eq!(left, ["p-3", "p-4"]);
645        let lines = drained_lines(&done);
646        assert!(lines.contains("sent 1 queued note(s): p-1 id-1"), "{lines}");
647        assert!(
648            lines.contains("2 note(s) still queued (connection refused)"),
649            "{lines}"
650        );
651        let done = drain_in(&dir, |_| Sent::Kept("id".into())).unwrap();
652        assert_eq!(done.kept.len(), 2);
653        assert!(pending_in(&dir).is_empty());
654        assert_eq!(
655            drained_lines(&drain_in(&dir, |_| unreachable!()).unwrap()),
656            ""
657        );
658    }
659}