distkit 0.5.0

A toolkit of distributed systems primitives for Rust, backed by Redis
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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232

//! Live-Redis tests for [`RwLock`](crate::lock::RwLock) and its guards:
//! shared reads, writer/reader exclusion, writer preference, bounded waiting,
//! release/drop-frees semantics, auto-refresh, and lost-lease state.

use std::time::Duration;

use crate::DistkitError;
use crate::lock::tests::common::{make_options, make_options_with_rw_keys, raw_connection};
use crate::lock::{LockError, RwLock, LockGuardState};

/// Multiple readers share the lock: two `try_read` guards are held at once.
#[tokio::test]
async fn concurrent_reads_share() {
    let rw_a = RwLock::new(make_options("concurrent_reads_share").await);
    let rw_b = RwLock::new(make_options("concurrent_reads_share").await);

    let _read_a = rw_a.try_read().await.expect("first reader should acquire");
    let _read_b = rw_b
        .try_read()
        .await
        .expect("second reader should share the lock");
}

/// A held write excludes readers: `try_read` fails while a writer holds.
#[tokio::test]
async fn writer_excludes_readers() {
    let writer = RwLock::new(make_options("writer_excludes_readers").await);
    let reader = RwLock::new(make_options("writer_excludes_readers").await);

    let _write_guard = writer.try_write().await.expect("writer should acquire");

    match reader.try_read().await {
        Err(DistkitError::LockError(LockError::AcquireFail)) => {}
        other => panic!("expected AcquireFail while write held, got {other:?}"),
    }
}

/// A writer waits while readers hold, then acquires once they release.
#[tokio::test]
async fn writer_waits_for_readers() {
    let reader = RwLock::new(make_options("writer_waits_for_readers").await);
    let writer = RwLock::new(make_options("writer_waits_for_readers").await);

    let read_guard = reader.try_read().await.expect("reader should acquire");

    let waiter = tokio::spawn(async move {
        writer
            .try_write_for(Duration::from_secs(2), Duration::from_millis(20))
            .await
    });

    tokio::time::sleep(Duration::from_millis(200)).await;
    read_guard.release().await.expect("read release should succeed");

    waiter
        .await
        .expect("waiter task should not panic")
        .expect("writer should acquire after readers release");
}

/// A reader waits while a writer holds, then acquires once it releases.
#[tokio::test]
async fn reader_waits_for_writer() {
    let writer = RwLock::new(make_options("reader_waits_for_writer").await);
    let reader = RwLock::new(make_options("reader_waits_for_writer").await);

    let write_guard = writer.try_write().await.expect("writer should acquire");

    let waiter = tokio::spawn(async move {
        reader
            .try_read_for(Duration::from_secs(2), Duration::from_millis(20))
            .await
    });

    tokio::time::sleep(Duration::from_millis(200)).await;
    write_guard
        .release()
        .await
        .expect("write release should succeed");

    waiter
        .await
        .expect("waiter task should not panic")
        .expect("reader should acquire after writer releases");
}

/// Writer preference: while a writer is waiting (enqueued), new readers are
/// turned away so they cannot jump ahead of it.
#[tokio::test]
async fn waiting_writer_blocks_new_readers() {
    let reader = RwLock::new(make_options("waiting_writer_blocks_new_readers").await);
    let writer = RwLock::new(make_options("waiting_writer_blocks_new_readers").await);
    let late_reader = RwLock::new(make_options("waiting_writer_blocks_new_readers").await);

    let _read_guard = reader.try_read().await.expect("first reader should acquire");

    // A waiting (enqueuing) writer blocked by the held reader.
    let waiter = tokio::spawn(async move {
        writer
            .try_write_for(Duration::from_secs(2), Duration::from_millis(20))
            .await
    });

    // Let the writer register itself as pending at least once.
    tokio::time::sleep(Duration::from_millis(150)).await;

    match late_reader.try_read().await {
        Err(DistkitError::LockError(LockError::AcquireFail)) => {}
        other => panic!("expected AcquireFail — waiting writer should block new readers, got {other:?}"),
    }

    waiter.abort();
}

/// Releasing a write frees the resource for the next writer.
#[tokio::test]
async fn write_release_frees() {
    let writer_a = RwLock::new(make_options("write_release_frees").await);
    let writer_b = RwLock::new(make_options("write_release_frees").await);

    let guard = writer_a.try_write().await.expect("writer should acquire");
    guard.release().await.expect("write release should succeed");

    writer_b
        .try_write()
        .await
        .expect("write should be acquirable after release");
}

/// Dropping a read guard frees the slot (best-effort fire-and-forget release), so
/// a writer can then acquire.
#[tokio::test]
async fn read_drop_frees() {
    let reader = RwLock::new(make_options("read_drop_frees").await);
    let writer = RwLock::new(make_options("read_drop_frees").await);

    {
        let _read_guard = reader.try_read().await.expect("reader should acquire");
    }

    // Drop release is spawned/async — give it a moment to land.
    tokio::time::sleep(Duration::from_millis(150)).await;

    writer
        .try_write()
        .await
        .expect("write should be acquirable after the reader drops");
}

/// Background auto-refresh keeps a held write lease alive past its `ttl`.
#[tokio::test]
async fn auto_refresh_keeps_lease_alive() {
    let mut options = make_options("auto_refresh_keeps_lease_alive").await;
    options.ttl = Duration::from_millis(150);
    let writer_a = RwLock::new(options);
    let writer_b = RwLock::new(make_options("auto_refresh_keeps_lease_alive").await);

    let _guard = writer_a.try_write().await.expect("writer should acquire");

    // Well past ttl (150 ms) — without refresh the lease would have expired.
    tokio::time::sleep(Duration::from_millis(400)).await;

    match writer_b.try_write().await {
        Err(DistkitError::LockError(LockError::AcquireFail)) => {}
        other => panic!("expected AcquireFail — refresh should keep lease alive, got {other:?}"),
    }
}

/// A healthy held read reports [`LockGuardState::Acquired`].
#[tokio::test]
async fn get_state_reports_acquired() {
    let reader = RwLock::new(make_options("get_state_reports_acquired").await);

    let guard = reader.try_read().await.expect("reader should acquire");

    assert_eq!(guard.get_state().await, LockGuardState::Acquired);
}

/// If the write lease is lost (writer key deleted out from under us), the refresh
/// task marks it [`LockGuardState::Lost`]: `get_state` reports it and `release`
/// returns `Ok(Lost)` without issuing a delete we no longer own.
#[tokio::test]
async fn lost_lease_reports_lost() {
    let (mut options, keys) = make_options_with_rw_keys("lost_lease_reports_lost").await;
    options.ttl = Duration::from_millis(150);
    let writer = RwLock::new(options);

    let guard = writer.try_write().await.expect("writer should acquire");

    // Delete the writer key out from under the holder; the next refresh tick
    // (~50 ms) sees GET != owner -> Ok(false) -> lost.
    let mut conn = raw_connection().await;
    let _: () = redis::cmd("DEL")
        .arg(&keys.writer)
        .query_async(&mut conn)
        .await
        .expect("DEL should succeed");

    tokio::time::sleep(Duration::from_millis(200)).await;

    assert_eq!(
        guard.get_state().await,
        LockGuardState::Lost,
        "lease should be marked lost after the writer key was deleted"
    );

    match guard.release().await {
        Ok(LockGuardState::Lost) => {}
        other => panic!("expected Ok(LockGuardState::Lost) after the lease was deleted, got {other:?}"),
    }
}

/// A one-shot `try_write` on a read-held lock fails without enqueuing, so a
/// subsequent `try_read` still succeeds (the failed write did not block readers).
#[tokio::test]
async fn try_write_does_not_block_readers() {
    let reader = RwLock::new(make_options("try_write_does_not_block_readers").await);
    let writer = RwLock::new(make_options("try_write_does_not_block_readers").await);
    let late_reader = RwLock::new(make_options("try_write_does_not_block_readers").await);

    let _read_guard = reader.try_read().await.expect("first reader should acquire");

    match writer.try_write().await {
        Err(DistkitError::LockError(LockError::AcquireFail)) => {}
        other => panic!("expected AcquireFail for one-shot try_write, got {other:?}"),
    }

    late_reader
        .try_read()
        .await
        .expect("reader should still acquire — one-shot try_write must not enqueue");
}