persistent-queue 0.1.5

A durable, at-least-once MPSC queue backed by in-memory and durable backends (sled, redb).
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

//! End-to-end: the full stack - a codec or the async facade over a real on-disk
//! backend - survives a crash (drop + reopen the same path) and recovers. Each test
//! is gated on the features it exercises, so they run under `--all-features`.

#[cfg(all(feature = "sled", feature = "serde"))]
#[test]
fn typed_bincode_survives_a_crash_on_sled() {
    use persistent_queue::{Bincode, Builder, SledStore};
    use serde::{Deserialize, Serialize};

    #[derive(Serialize, Deserialize, Debug, PartialEq)]
    struct Job {
        id: u64,
        name: String,
    }

    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("queue");

    // Session 1: push 10 jobs, ack the even ids, hold (then drop) the odd ones.
    {
        let store = SledStore::open(&path).unwrap();
        let (tx, rx) = Builder::new(store).open_typed(Bincode).unwrap();
        for id in 0..10u64 {
            tx.push(&Job {
                id,
                name: format!("job-{id}"),
            })
            .unwrap();
        }
        let mut held = Vec::new();
        while let Some(item) = rx.reserve().unwrap() {
            if item.id % 2 == 0 {
                item.ack().unwrap();
            } else {
                held.push(item);
            }
        }
        // `held` (odd ids, unacked) drops here - a crash mid-processing.
    }

    // Session 2: reopen the same store; the odd jobs return, and still decode.
    let store = SledStore::open(&path).unwrap();
    let (_tx, rx) = Builder::new(store).open_typed::<Job, _>(Bincode).unwrap();
    let mut recovered = Vec::new();
    while let Some(item) = rx.reserve().unwrap() {
        assert_eq!(item.name, format!("job-{}", item.id));
        recovered.push(item.id);
        item.ack().unwrap();
    }
    recovered.sort_unstable();
    assert_eq!(recovered, vec![1u64, 3, 5, 7, 9]);
}

#[cfg(all(feature = "redb", feature = "rkyv"))]
#[test]
fn typed_rkyv_survives_a_crash_on_redb() {
    use persistent_queue::{Builder, RedbStore, Rkyv};

    #[derive(rkyv::Archive, rkyv::Serialize, rkyv::Deserialize, Debug, PartialEq)]
    struct Job {
        id: u64,
        name: String,
    }

    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("queue.redb");

    {
        let store = RedbStore::open(&path).unwrap();
        let (tx, rx) = Builder::new(store).open_typed(Rkyv).unwrap();
        for id in 0..10u64 {
            tx.push(&Job {
                id,
                name: format!("job-{id}"),
            })
            .unwrap();
        }
        let mut held = Vec::new();
        while let Some(item) = rx.reserve().unwrap() {
            if item.id % 2 == 0 {
                item.ack().unwrap();
            } else {
                held.push(item);
            }
        }
    }

    let store = RedbStore::open(&path).unwrap();
    let (_tx, rx) = Builder::new(store).open_typed::<Job, _>(Rkyv).unwrap();
    let mut recovered = Vec::new();
    while let Some(item) = rx.reserve().unwrap() {
        assert_eq!(item.name, format!("job-{}", item.id));
        recovered.push(item.id);
        item.ack().unwrap();
    }
    recovered.sort_unstable();
    assert_eq!(recovered, vec![1u64, 3, 5, 7, 9]);
}

#[cfg(all(feature = "tokio", feature = "sled"))]
#[tokio::test]
async fn async_survives_a_crash_on_sled() {
    use persistent_queue::{Builder, SledStore};

    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("queue");

    // Session 1: push 6, ack the first 3, drop the rest (a crash).
    {
        let store = SledStore::open(&path).unwrap();
        let (tx, rx) = Builder::new(store).open_async().await.unwrap();
        for i in 0..6u64 {
            tx.push(i.to_le_bytes().to_vec()).await.unwrap();
        }
        for _ in 0..3 {
            rx.reserve().await.unwrap().unwrap().ack().await.unwrap();
        }
    }

    // Session 2: reopen; the 3 unacked items come back.
    let store = SledStore::open(&path).unwrap();
    let (tx, rx) = Builder::new(store).open_async().await.unwrap();
    tx.close(); // so reserve terminates once the recovered items drain
    let mut recovered = 0;
    while let Some(item) = rx.reserve().await.unwrap() {
        recovered += 1;
        item.ack().await.unwrap();
    }
    assert_eq!(recovered, 3, "the 3 unacked items survived the crash");
}