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

//! Direct-against-Redis tests for the Stage 1 mutex backend
//! (`crate::lock::mutex_backend`): owner-gated acquire / refresh / release.

use redis::aio::ConnectionManager;

use crate::lock::mutex_backend;
use crate::lock::tests::common::make_options;

const OWNER_A: &str = "owner_a";
const OWNER_B: &str = "owner_b";

/// Returns a cloned connection and a process-unique raw key for `name`.
async fn conn_and_key(name: &str) -> (ConnectionManager, String) {
    let opts = make_options(name).await;
    (opts.connection_manager.clone(), opts.key.to_string())
}

async fn exists(conn: &mut ConnectionManager, key: &str) -> bool {
    let n: i64 = redis::cmd("EXISTS")
        .arg(key)
        .query_async(conn)
        .await
        .expect("EXISTS");
    n == 1
}

async fn pttl(conn: &mut ConnectionManager, key: &str) -> i64 {
    redis::cmd("PTTL")
        .arg(key)
        .query_async(conn)
        .await
        .expect("PTTL")
}

#[tokio::test]
async fn acquire_is_exclusive() {
    let (mut conn, key) = conn_and_key("acquire_is_exclusive").await;

    assert!(
        mutex_backend::acquire(&mut conn, &key, OWNER_A, 30_000)
            .await
            .unwrap(),
        "first acquire should take the lock"
    );
    assert!(
        !mutex_backend::acquire(&mut conn, &key, OWNER_B, 30_000)
            .await
            .unwrap(),
        "second acquire on a held key should fail"
    );
}

#[tokio::test]
async fn refresh_is_owner_gated() {
    let (mut conn, key) = conn_and_key("refresh_is_owner_gated").await;

    // Short initial lease so a successful refresh visibly bumps the PTTL.
    assert!(
        mutex_backend::acquire(&mut conn, &key, OWNER_A, 2_000)
            .await
            .unwrap()
    );

    assert!(
        mutex_backend::refresh(&mut conn, &key, OWNER_A, 30_000)
            .await
            .unwrap(),
        "owner refresh should succeed"
    );
    assert!(
        pttl(&mut conn, &key).await > 2_000,
        "owner refresh should extend the lease"
    );

    assert!(
        !mutex_backend::refresh(&mut conn, &key, OWNER_B, 30_000)
            .await
            .unwrap(),
        "non-owner refresh should fail"
    );
}

#[tokio::test]
async fn release_is_owner_gated() {
    let (mut conn, key) = conn_and_key("release_is_owner_gated").await;

    assert!(
        mutex_backend::acquire(&mut conn, &key, OWNER_A, 30_000)
            .await
            .unwrap()
    );

    assert!(
        !mutex_backend::release(&mut conn, &key, OWNER_B).await.unwrap(),
        "non-owner release should fail"
    );
    assert!(
        exists(&mut conn, &key).await,
        "key should survive a non-owner release"
    );

    assert!(
        mutex_backend::release(&mut conn, &key, OWNER_A).await.unwrap(),
        "owner release should succeed"
    );
    assert!(
        !exists(&mut conn, &key).await,
        "key should be gone after owner release"
    );
}

#[tokio::test]
async fn acquire_succeeds_after_release() {
    let (mut conn, key) = conn_and_key("acquire_succeeds_after_release").await;

    assert!(
        mutex_backend::acquire(&mut conn, &key, OWNER_A, 30_000)
            .await
            .unwrap()
    );
    assert!(mutex_backend::release(&mut conn, &key, OWNER_A).await.unwrap());
    assert!(
        mutex_backend::acquire(&mut conn, &key, OWNER_B, 30_000)
            .await
            .unwrap(),
        "key should be acquirable again after release"
    );
}

/// EXPECTED (ttl validation): `acquire` and `refresh` must reject a non-positive
/// `ttl_ms` (0 or negative) with an error, in the Rust layer, before touching
/// Redis. Mirrors the rwlock_backend validation tests.
#[tokio::test]
async fn acquire_rejects_nonpositive_ttl() {
    let (mut conn, key) = conn_and_key("acquire_rejects_nonpositive_ttl").await;
    for ttl in [0_i64, -1] {
        assert!(
            mutex_backend::acquire(&mut conn, &key, OWNER_A, ttl).await.is_err(),
            "acquire must error for ttl_ms = {ttl}"
        );
    }
}

#[tokio::test]
async fn refresh_rejects_nonpositive_ttl() {
    let (mut conn, key) = conn_and_key("refresh_rejects_nonpositive_ttl").await;
    for ttl in [0_i64, -1] {
        assert!(
            mutex_backend::refresh(&mut conn, &key, OWNER_A, ttl).await.is_err(),
            "refresh must error for ttl_ms = {ttl}"
        );
    }
}

/// EXPECTED (owner validation): an empty owner must be rejected with an error so
/// it can never become a lock holder.
#[tokio::test]
async fn acquire_rejects_empty_owner() {
    let (mut conn, key) = conn_and_key("acquire_rejects_empty_owner").await;
    assert!(
        mutex_backend::acquire(&mut conn, &key, "", 30_000).await.is_err(),
        "acquire must error for an empty owner"
    );
}

#[tokio::test]
async fn lease_expiry_frees_key() {
    let (mut conn, key) = conn_and_key("lease_expiry_frees_key").await;

    assert!(mutex_backend::acquire(&mut conn, &key, OWNER_A, 100).await.unwrap());

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

    assert!(
        mutex_backend::acquire(&mut conn, &key, OWNER_B, 30_000)
            .await
            .unwrap(),
        "key should be acquirable once the lease expires"
    );
}