memstore 0.2.0

An in-memory implementation of the `async-raft::RaftStorage` trait.
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
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
use std::collections::HashSet;

use super::*;
use async_raft::raft::EntryConfigChange;

const NODE_ID: u64 = 0;

//////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////

#[tokio::test]
async fn test_get_membership_config_default() -> Result<()> {
    let store = MemStore::new(NODE_ID);
    let membership = store.get_membership_config().await?;
    assert_eq!(membership.members.len(), 1, "expected members len of 1");
    assert!(
        membership.members_after_consensus.is_none(),
        "expected None for default members_after_consensus"
    );
    Ok(())
}

#[tokio::test]
async fn test_get_membership_config_with_previous_state() -> Result<()> {
    let mut log = BTreeMap::new();
    let mut members: HashSet<NodeId> = Default::default();
    members.insert(1);
    members.insert(2);
    members.insert(3);
    log.insert(
        1,
        Entry {
            term: 1,
            index: 1,
            payload: EntryPayload::ConfigChange(EntryConfigChange {
                membership: MembershipConfig {
                    members: members.clone(),
                    members_after_consensus: None,
                },
            }),
        },
    );
    let sm = MemStoreStateMachine::default();
    let hs = HardState {
        current_term: 1,
        voted_for: Some(NODE_ID),
    };
    let store = MemStore::new_with_state(NODE_ID, log, sm, Some(hs.clone()), None);

    let initial = store.get_membership_config().await?;

    assert_eq!(&initial.members, &members, "unexpected len for members");
    assert!(initial.members_after_consensus.is_none(), "unexpected value for members_after_consensus");
    Ok(())
}

//////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////

#[tokio::test]
async fn test_get_initial_state_default() -> Result<()> {
    let store = MemStore::new(NODE_ID);
    let expected_hs = HardState {
        current_term: 0,
        voted_for: None,
    };
    let expected_membership = MembershipConfig::new_initial(NODE_ID);

    let initial = store.get_initial_state().await?;

    assert_eq!(initial.last_log_index, 0, "unexpected default value for last log index");
    assert_eq!(initial.last_log_term, 0, "unexpected default value for last log term");
    assert_eq!(initial.last_applied_log, 0, "unexpected value for last applied log");
    assert_eq!(initial.hard_state, expected_hs, "unexpected value for default hard state");
    assert_eq!(initial.membership, expected_membership, "unexpected value for default membership config");
    Ok(())
}

#[tokio::test]
async fn test_get_initial_state_with_previous_state() -> Result<()> {
    let mut log = BTreeMap::new();
    log.insert(
        1,
        Entry {
            term: 1,
            index: 1,
            payload: EntryPayload::Blank,
        },
    );
    let sm = MemStoreStateMachine {
        last_applied_log: 1, // Just stubbed in for testing.
        ..Default::default()
    };
    let hs = HardState {
        current_term: 1,
        voted_for: Some(NODE_ID),
    };
    let store = MemStore::new_with_state(NODE_ID, log, sm, Some(hs.clone()), None);

    let initial = store.get_initial_state().await?;

    assert_eq!(initial.last_log_index, 1, "unexpected default value for last log index");
    assert_eq!(initial.last_log_term, 1, "unexpected default value for last log term");
    assert_eq!(initial.last_applied_log, 1, "unexpected value for last applied log");
    assert_eq!(initial.hard_state, hs, "unexpected value for default hard state");
    Ok(())
}

//////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////

#[tokio::test]
async fn test_save_hard_state() -> Result<()> {
    let store = MemStore::new(NODE_ID);
    let new_hs = HardState {
        current_term: 100,
        voted_for: Some(NODE_ID),
    };

    let initial = store.get_initial_state().await?;
    store.save_hard_state(&new_hs).await?;
    let post = store.get_initial_state().await?;

    assert_ne!(
        initial.hard_state, post.hard_state,
        "hard state was expected to be different after update"
    );
    Ok(())
}

//////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////

#[tokio::test]
async fn test_get_log_entries_returns_emptry_vec_when_start_gt_stop() -> Result<()> {
    let store = default_store_with_logs();

    let logs = store.get_log_entries(10, 1).await?;

    assert_eq!(logs.len(), 0, "expected no logs to be returned");
    Ok(())
}

#[tokio::test]
async fn test_get_log_entries_returns_expected_entries() -> Result<()> {
    let store = default_store_with_logs();

    let logs = store.get_log_entries(5, 7).await?;

    assert_eq!(logs.len(), 2, "expected two logs to be returned");
    assert_eq!(logs[0].index, 5, "unexpected value for log index");
    assert_eq!(logs[0].term, 1, "unexpected value for log term");
    assert_eq!(logs[1].index, 6, "unexpected value for log index");
    assert_eq!(logs[1].term, 1, "unexpected value for log term");
    Ok(())
}

//////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////

#[tokio::test]
async fn test_delete_logs_from_does_nothing_if_start_gt_stop() -> Result<()> {
    let store = default_store_with_logs();

    store.delete_logs_from(10, Some(1)).await?;
    let logs = store.get_log_entries(1, 11).await?;

    assert_eq!(logs.len(), 10, "expected all (10) logs to be preserved");
    Ok(())
}

#[tokio::test]
async fn test_delete_logs_from_deletes_target_logs() -> Result<()> {
    let store = default_store_with_logs();

    store.delete_logs_from(1, Some(11)).await?;
    let logs = store.get_log_entries(0, 100).await?;

    assert_eq!(logs.len(), 0, "expected all logs to be deleted");
    Ok(())
}

#[tokio::test]
async fn test_delete_logs_from_deletes_target_logs_no_stop() -> Result<()> {
    let store = default_store_with_logs();

    store.delete_logs_from(1, None).await?;
    let logs = store.get_log_entries(0, 100).await?;

    assert_eq!(logs.len(), 0, "expected all logs to be deleted");
    Ok(())
}

#[tokio::test]
async fn test_delete_logs_from_deletes_only_target_logs() -> Result<()> {
    let store = default_store_with_logs();

    store.delete_logs_from(1, Some(10)).await?;
    let logs = store.get_log_entries(0, 100).await?;

    assert_eq!(logs.len(), 1, "expected one log to be preserved");
    assert_eq!(logs[0].index, 10, "unexpected log index");
    Ok(())
}

//////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////

#[tokio::test]
async fn test_append_entry_to_log() -> Result<()> {
    let store = default_store_with_logs();

    store
        .append_entry_to_log(&Entry {
            term: 2,
            index: 10,
            payload: EntryPayload::Blank,
        })
        .await?;
    let log = store.get_log().await;

    assert_eq!(log.len(), 10, "expected 10 entries to exist in the log");
    assert_eq!(log[&10].index, 10, "unexpected log index");
    assert_eq!(log[&10].term, 2, "unexpected log term");
    Ok(())
}

//////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////

#[tokio::test]
async fn test_replicate_to_log() -> Result<()> {
    let store = default_store_with_logs();

    store
        .replicate_to_log(&[Entry {
            term: 1,
            index: 11,
            payload: EntryPayload::Blank,
        }])
        .await?;
    let log = store.get_log().await;

    assert_eq!(log.len(), 11, "expected 11 entries to exist in the log");
    assert_eq!(log[&11].index, 11, "unexpected log index");
    assert_eq!(log[&11].term, 1, "unexpected log term");
    Ok(())
}

//////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////

#[tokio::test]
async fn test_apply_entry_to_state_machine() -> Result<()> {
    let store = default_store_with_logs();

    store
        .apply_entry_to_state_machine(
            &1,
            &ClientRequest {
                client: "0".into(),
                serial: 0,
                status: "lit".into(),
            },
        )
        .await?;
    let sm = store.get_state_machine().await;

    assert_eq!(sm.last_applied_log, 1, "expected last_applied_log to be 1, got {}", sm.last_applied_log);
    let client_serial = sm
        .client_serial_responses
        .get("0")
        .expect("expected entry to exist in client_serial_responses");
    assert_eq!(client_serial.0, 0, "unexpected client serial response");
    assert_eq!(client_serial.1, None, "unexpected client serial response");
    let client_status = sm.client_status.get("0").expect("expected entry to exist in client_status");
    assert_eq!(client_status, "lit", "expected client_status to be 'lit', got '{}'", client_status);
    Ok(())
}

//////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////

#[tokio::test]
async fn test_replicate_to_state_machine() -> Result<()> {
    let store = default_store_with_logs();

    let req0 = ClientRequest {
        client: "1".into(),
        serial: 0,
        status: "old".into(),
    };
    let req1 = ClientRequest {
        client: "1".into(),
        serial: 1,
        status: "new".into(),
    };
    let req2 = ClientRequest {
        client: "2".into(),
        serial: 0,
        status: "other".into(),
    };
    let entries = vec![(&1u64, &req0), (&2u64, &req1), (&3u64, &req2)];
    store.replicate_to_state_machine(&entries).await?;
    let sm = store.get_state_machine().await;

    assert_eq!(sm.last_applied_log, 3, "expected last_applied_log to be 3, got {}", sm.last_applied_log);
    let client_serial1 = sm
        .client_serial_responses
        .get("1")
        .expect("expected entry to exist in client_serial_responses for client 1");
    assert_eq!(client_serial1.0, 1, "unexpected client serial response");
    assert_eq!(client_serial1.1, Some(String::from("old")), "unexpected client serial response");
    let client_serial2 = sm
        .client_serial_responses
        .get("2")
        .expect("expected entry to exist in client_serial_responses for client 2");
    assert_eq!(client_serial2.0, 0, "unexpected client serial response");
    assert_eq!(client_serial2.1, None, "unexpected client serial response");
    let client_status1 = sm.client_status.get("1").expect("expected entry to exist in client_status for client 1");
    let client_status2 = sm.client_status.get("2").expect("expected entry to exist in client_status for client 2");
    assert_eq!(client_status1, "new", "expected client_status to be 'new', got '{}'", client_status1);
    assert_eq!(client_status2, "other", "expected client_status to be 'other', got '{}'", client_status2);
    Ok(())
}

//////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////

fn default_store_with_logs() -> MemStore {
    let mut log = BTreeMap::new();
    log.insert(
        1,
        Entry {
            term: 1,
            index: 1,
            payload: EntryPayload::Blank,
        },
    );
    log.insert(
        2,
        Entry {
            term: 1,
            index: 2,
            payload: EntryPayload::Blank,
        },
    );
    log.insert(
        3,
        Entry {
            term: 1,
            index: 3,
            payload: EntryPayload::Blank,
        },
    );
    log.insert(
        4,
        Entry {
            term: 1,
            index: 4,
            payload: EntryPayload::Blank,
        },
    );
    log.insert(
        5,
        Entry {
            term: 1,
            index: 5,
            payload: EntryPayload::Blank,
        },
    );
    log.insert(
        6,
        Entry {
            term: 1,
            index: 6,
            payload: EntryPayload::Blank,
        },
    );
    log.insert(
        7,
        Entry {
            term: 1,
            index: 7,
            payload: EntryPayload::Blank,
        },
    );
    log.insert(
        8,
        Entry {
            term: 1,
            index: 8,
            payload: EntryPayload::Blank,
        },
    );
    log.insert(
        9,
        Entry {
            term: 1,
            index: 9,
            payload: EntryPayload::Blank,
        },
    );
    log.insert(
        10,
        Entry {
            term: 1,
            index: 10,
            payload: EntryPayload::Blank,
        },
    );
    let sm = MemStoreStateMachine::default();
    let hs = HardState {
        current_term: 1,
        voted_for: Some(NODE_ID),
    };
    MemStore::new_with_state(NODE_ID, log, sm, Some(hs), None)
}