hotfix-store 0.1.1

Message store traits and implementations for the HotFIX engine
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
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
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
//! Test utilities for message store implementations.
//!
//! This module provides a test harness for verifying that message store
//! implementations conform to the expected behavior of the [`MessageStore`] trait.

use crate::MessageStore;

/// A factory trait for creating message store instances during testing.
///
/// Implement this trait for each message store implementation you want to test.
/// The test functions in this module will use this factory to create fresh
/// store instances for each test.
#[async_trait::async_trait]
pub trait TestStoreFactory: Send + Sync {
    /// Creates a new message store instance.
    ///
    /// For persistent stores, this should return a store that can be
    /// reopened with the same data (by calling `create_store` again).
    async fn create_store(&self) -> Box<dyn MessageStore>;

    /// Returns `true` if the store persists data across restarts.
    ///
    /// This is used to skip persistence-related tests for in-memory stores.
    fn is_persistent(&self) -> bool {
        true
    }
}

/// Tests that a new store starts with sequence numbers at 1.
pub async fn test_new_store_initialization(factory: &dyn TestStoreFactory) {
    let store = factory.create_store().await;

    assert_eq!(store.next_sender_seq_number(), 1);
    assert_eq!(store.next_target_seq_number(), 1);
}

/// Tests adding and retrieving messages.
pub async fn test_add_and_get_messages(factory: &dyn TestStoreFactory) {
    let mut store = factory.create_store().await;

    let message1 = b"test message 1";
    let message2 = b"test message 2";
    let message3 = b"test message 3";

    store
        .add(1, message1)
        .await
        .expect("Failed to add message 1");
    store
        .add(2, message2)
        .await
        .expect("Failed to add message 2");
    store
        .add(3, message3)
        .await
        .expect("Failed to add message 3");

    let messages = store.get_slice(1, 3).await.expect("Failed to get messages");
    assert_eq!(messages.len(), 3);
    assert_eq!(messages[0], message1);
    assert_eq!(messages[1], message2);
    assert_eq!(messages[2], message3);
}

/// Tests getting a partial range of messages.
pub async fn test_get_slice_partial_range(factory: &dyn TestStoreFactory) {
    let mut store = factory.create_store().await;

    let message1 = b"message 1";
    let message2 = b"message 2";
    let message3 = b"message 3";
    let message4 = b"message 4";

    store
        .add(1, message1)
        .await
        .expect("Failed to add message 1");
    store
        .add(2, message2)
        .await
        .expect("Failed to add message 2");
    store
        .add(3, message3)
        .await
        .expect("Failed to add message 3");
    store
        .add(4, message4)
        .await
        .expect("Failed to add message 4");

    let messages = store.get_slice(2, 3).await.expect("Failed to get messages");
    assert_eq!(messages.len(), 2);
    assert_eq!(messages[0], message2);
    assert_eq!(messages[1], message3);
}

/// Tests getting a slice from an empty store.
pub async fn test_get_slice_empty_range(factory: &dyn TestStoreFactory) {
    let store = factory.create_store().await;

    let messages = store.get_slice(1, 3).await.expect("Failed to get messages");
    assert_eq!(messages.len(), 0);
}

/// Tests incrementing the sender sequence number.
pub async fn test_increment_sender_seq_number(factory: &dyn TestStoreFactory) {
    let mut store = factory.create_store().await;

    assert_eq!(store.next_sender_seq_number(), 1);

    store
        .increment_sender_seq_number()
        .await
        .expect("Failed to increment sender seq number");
    assert_eq!(store.next_sender_seq_number(), 2);

    store
        .increment_sender_seq_number()
        .await
        .expect("Failed to increment sender seq number");
    assert_eq!(store.next_sender_seq_number(), 3);
}

/// Tests incrementing the target sequence number.
pub async fn test_increment_target_seq_number(factory: &dyn TestStoreFactory) {
    let mut store = factory.create_store().await;

    assert_eq!(store.next_target_seq_number(), 1);

    store
        .increment_target_seq_number()
        .await
        .expect("Failed to increment target seq number");
    assert_eq!(store.next_target_seq_number(), 2);

    store
        .increment_target_seq_number()
        .await
        .expect("Failed to increment target seq number");
    assert_eq!(store.next_target_seq_number(), 3);
}

/// Tests setting the target sequence number directly.
pub async fn test_set_target_seq_number(factory: &dyn TestStoreFactory) {
    let mut store = factory.create_store().await;

    assert_eq!(store.next_target_seq_number(), 1);

    store
        .set_target_seq_number(10)
        .await
        .expect("Failed to set target seq number");
    assert_eq!(store.next_target_seq_number(), 11);

    store
        .set_target_seq_number(5)
        .await
        .expect("Failed to set target seq number");
    assert_eq!(store.next_target_seq_number(), 6);
}

/// Tests resetting the store.
pub async fn test_reset_store(factory: &dyn TestStoreFactory) {
    let mut store = factory.create_store().await;

    // Add some messages and increment sequence numbers
    store
        .add(1, b"message 1")
        .await
        .expect("Failed to add message");
    store
        .add(2, b"message 2")
        .await
        .expect("Failed to add message");
    store
        .increment_sender_seq_number()
        .await
        .expect("Failed to increment sender seq number");
    store
        .increment_target_seq_number()
        .await
        .expect("Failed to increment target seq number");

    assert_eq!(store.next_sender_seq_number(), 2);
    assert_eq!(store.next_target_seq_number(), 2);

    let messages_before_reset = store.get_slice(1, 2).await.expect("Failed to get messages");
    assert_eq!(messages_before_reset.len(), 2);

    // Reset the store
    store.reset().await.expect("Failed to reset store");

    // Verify sequence numbers are reset
    assert_eq!(store.next_sender_seq_number(), 1);
    assert_eq!(store.next_target_seq_number(), 1);

    // Verify messages are cleared
    let messages_after_reset = store.get_slice(1, 2).await.expect("Failed to get messages");
    assert_eq!(messages_after_reset.len(), 0);
}

/// Tests that data persists across store instances (for persistent stores only).
pub async fn test_persistence_across_store_instances(factory: &dyn TestStoreFactory) {
    if !factory.is_persistent() {
        return;
    }

    // Create first store instance and add data
    {
        let mut store1 = factory.create_store().await;
        store1
            .add(1, b"persistent message")
            .await
            .expect("Failed to add message");
        store1
            .increment_sender_seq_number()
            .await
            .expect("Failed to increment sender seq number");
        store1
            .set_target_seq_number(5)
            .await
            .expect("Failed to set target seq number");
    }

    // Create second store instance and verify data persists
    {
        let store2 = factory.create_store().await;

        assert_eq!(store2.next_sender_seq_number(), 2);
        assert_eq!(store2.next_target_seq_number(), 6);

        let messages = store2
            .get_slice(1, 1)
            .await
            .expect("Failed to get messages");
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0], b"persistent message");
    }
}

/// Tests getting a slice beyond available messages.
pub async fn test_get_slice_beyond_available_messages(factory: &dyn TestStoreFactory) {
    let mut store = factory.create_store().await;

    store
        .add(1, b"only message")
        .await
        .expect("Failed to add message");

    let messages = store
        .get_slice(1, 10)
        .await
        .expect("Failed to get messages");
    assert_eq!(messages.len(), 1);
    assert_eq!(messages[0], b"only message");
}

/// Tests overwriting an existing message.
pub async fn test_overwrite_existing_message(factory: &dyn TestStoreFactory) {
    let mut store = factory.create_store().await;

    store
        .add(1, b"original message")
        .await
        .expect("Failed to add original message");
    store
        .add(1, b"updated message")
        .await
        .expect("Failed to add updated message");

    let messages = store.get_slice(1, 1).await.expect("Failed to get messages");
    assert_eq!(messages.len(), 1);
    assert_eq!(messages[0], b"updated message");
}

/// Tests that creation time is set on new stores.
pub async fn test_creation_time_is_set(factory: &dyn TestStoreFactory) {
    use chrono::Utc;

    let before = Utc::now();
    let store = factory.create_store().await;
    let after = Utc::now();

    assert!(before <= store.creation_time());
    assert!(store.creation_time() <= after);
}

/// Tests that creation time is preserved across store restarts (for persistent stores only).
pub async fn test_creation_time_is_preserved(factory: &dyn TestStoreFactory) {
    if !factory.is_persistent() {
        return;
    }

    let store = factory.create_store().await;
    let creation_time1 = store.creation_time();
    drop(store);

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

    let store = factory.create_store().await;
    let creation_time2 = store.creation_time();

    assert_eq!(creation_time1, creation_time2);
}

/// Tests that creation time is updated on reset.
pub async fn test_creation_time_gets_reset_correctly(factory: &dyn TestStoreFactory) {
    use chrono::Utc;

    let mut store = factory.create_store().await;

    tokio::time::sleep(std::time::Duration::from_millis(2)).await;
    let after_sleep = Utc::now();
    tokio::time::sleep(std::time::Duration::from_millis(2)).await;

    store.reset().await.expect("failed to reset store");
    let reset_creation_time = store.creation_time();
    assert!(reset_creation_time > after_sleep);

    if !factory.is_persistent() {
        return;
    }

    drop(store);
    let store = factory.create_store().await;
    assert_eq!(reset_creation_time, store.creation_time());
}

/// Generates conformance tests for a message store implementation.
///
/// This macro creates a module containing all the standard conformance tests
/// for a [`MessageStore`] implementation. Each test gets its own test function,
/// allowing for parallel execution and clear test reporting.
///
/// # Arguments
///
/// * `$mod_name` - The name of the module to create (e.g., `in_memory`, `file`, `mongodb`)
/// * `$factory` - An expression that creates a [`TestStoreFactory`] instance.
///   This can be a sync expression (e.g., `MyFactory::new()`) or an async
///   expression (e.g., `MyFactory::new().await`).
///
/// # Example
///
/// ```ignore
/// use hotfix_store::conformance_tests;
///
/// struct MyStoreFactory;
///
/// #[async_trait::async_trait]
/// impl TestStoreFactory for MyStoreFactory {
///     async fn create_store(&self) -> Box<dyn MessageStore> {
///         Box::new(MyStore::new())
///     }
/// }
///
/// conformance_tests!(my_store, MyStoreFactory);
/// ```
#[macro_export]
macro_rules! conformance_tests {
    ($mod_name:ident, $factory:expr) => {
        mod $mod_name {
            use super::*;

            #[tokio::test]
            async fn test_new_store_initialization() {
                let factory = $factory;
                $crate::test_utils::test_new_store_initialization(&factory).await;
            }

            #[tokio::test]
            async fn test_add_and_get_messages() {
                let factory = $factory;
                $crate::test_utils::test_add_and_get_messages(&factory).await;
            }

            #[tokio::test]
            async fn test_get_slice_partial_range() {
                let factory = $factory;
                $crate::test_utils::test_get_slice_partial_range(&factory).await;
            }

            #[tokio::test]
            async fn test_get_slice_empty_range() {
                let factory = $factory;
                $crate::test_utils::test_get_slice_empty_range(&factory).await;
            }

            #[tokio::test]
            async fn test_increment_sender_seq_number() {
                let factory = $factory;
                $crate::test_utils::test_increment_sender_seq_number(&factory).await;
            }

            #[tokio::test]
            async fn test_increment_target_seq_number() {
                let factory = $factory;
                $crate::test_utils::test_increment_target_seq_number(&factory).await;
            }

            #[tokio::test]
            async fn test_set_target_seq_number() {
                let factory = $factory;
                $crate::test_utils::test_set_target_seq_number(&factory).await;
            }

            #[tokio::test]
            async fn test_reset_store() {
                let factory = $factory;
                $crate::test_utils::test_reset_store(&factory).await;
            }

            #[tokio::test]
            async fn test_persistence_across_store_instances() {
                let factory = $factory;
                $crate::test_utils::test_persistence_across_store_instances(&factory).await;
            }

            #[tokio::test]
            async fn test_get_slice_beyond_available_messages() {
                let factory = $factory;
                $crate::test_utils::test_get_slice_beyond_available_messages(&factory).await;
            }

            #[tokio::test]
            async fn test_overwrite_existing_message() {
                let factory = $factory;
                $crate::test_utils::test_overwrite_existing_message(&factory).await;
            }

            #[tokio::test]
            async fn test_creation_time_is_set() {
                let factory = $factory;
                $crate::test_utils::test_creation_time_is_set(&factory).await;
            }

            #[tokio::test]
            async fn test_creation_time_is_preserved() {
                let factory = $factory;
                $crate::test_utils::test_creation_time_is_preserved(&factory).await;
            }

            #[tokio::test]
            async fn test_creation_time_gets_reset_correctly() {
                let factory = $factory;
                $crate::test_utils::test_creation_time_gets_reset_correctly(&factory).await;
            }
        }
    };
}