distributed 2.3.1

CQRS/ES framework for Rust using Plain Old Rust Structs — append-only events, replay, snapshots, outbox, service bus, and pluggable infrastructure
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

#![cfg(feature = "sqlite")]

#[path = "../persistent_repository_conformance/mod.rs"]
mod conformance;

use distributed::SqliteRepository;

async fn repository() -> SqliteRepository {
    SqliteRepository::connect_and_migrate("sqlite::memory:")
        .await
        .expect("sqlite conformance repository should migrate")
}

#[tokio::test]
async fn aggregate_checkout_flow_persists_reloaded_state() {
    conformance::scenario::aggregate_checkout_flow_persists_reloaded_state(repository().await)
        .await;
}

#[tokio::test]
async fn get_all_and_commit_all_round_trip() {
    conformance::scenario::get_all_and_commit_all_round_trip(repository().await).await;
}

#[tokio::test]
async fn multi_stream_conflict_rolls_back_other_stream_and_snapshot() {
    conformance::scenario::multi_stream_conflict_rolls_back_other_stream_and_snapshot(
        repository().await,
    )
    .await;
}

#[tokio::test]
async fn duplicate_stream_identity_is_rejected_before_write() {
    conformance::scenario::duplicate_stream_identity_is_rejected_before_write(repository().await)
        .await;
}

#[tokio::test]
async fn metadata_round_trips() {
    conformance::scenario::metadata_round_trips(repository().await).await;
}

#[tokio::test]
async fn unsupported_codec_is_rejected_on_write() {
    conformance::scenario::unsupported_codec_is_rejected_on_write(repository().await).await;
}

#[tokio::test]
async fn snapshots_use_full_stream_identity() {
    conformance::scenario::snapshots_use_full_stream_identity(repository().await).await;
}

#[tokio::test]
async fn high_level_outbox_commit_persists_row_without_stream() {
    let repo = repository().await;
    conformance::outbox::high_level_outbox_commit_persists_row_without_stream(
        repo.clone(),
        repo.outbox_store(),
    )
    .await;
}

#[tokio::test]
async fn duplicate_outbox_insert_rolls_back_aggregate() {
    conformance::outbox::duplicate_outbox_insert_rolls_back_aggregate(repository().await).await;
}

#[tokio::test]
async fn aggregate_conflict_rolls_back_outbox() {
    let repo = repository().await;
    conformance::outbox::aggregate_conflict_rolls_back_outbox(repo.clone(), repo.outbox_store())
        .await;
}

#[tokio::test]
async fn worker_claim_complete_and_retry_lifecycle() {
    let repo = repository().await;
    conformance::outbox::worker_claim_complete_and_retry_lifecycle(
        repo.clone(),
        repo.outbox_store(),
    )
    .await;
}

#[tokio::test]
async fn worker_claim_by_ids_claims_only_requested() {
    let repo = repository().await;
    conformance::outbox::worker_claim_by_ids_claims_only_requested(
        repo.clone(),
        repo.outbox_store(),
    )
    .await;
}

#[tokio::test]
async fn worker_completes_claims_in_one_batch() {
    let repo = repository().await;
    conformance::outbox::worker_completes_claims_in_one_batch(repo.clone(), repo.outbox_store())
        .await;
}

#[tokio::test]
async fn consumer_inbox_records_dedupes_and_fences_with_real_effects() {
    let repo = repository().await;
    conformance::inbox::inbox_records_dedupes_and_fences_with_real_effects(
        repo.clone(),
        repo.outbox_store(),
    )
    .await;
}

#[tokio::test]
async fn consumer_inbox_rejects_empty_receipt() {
    conformance::inbox::inbox_rejects_empty_receipt(repository().await).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn racing_commits_one_wins_one_conflicts() {
    conformance::scenario::racing_commits_one_wins_one_conflicts(repository().await, 8).await;
}

#[tokio::test]
async fn expired_outbox_lease_is_reclaimed_by_second_worker() {
    let repo = repository().await;
    conformance::outbox::expired_outbox_lease_is_reclaimed_by_second_worker(
        repo.clone(),
        repo.outbox_store(),
    )
    .await;
}

#[tokio::test]
async fn publish_failure_after_commit_retains_outbox_row_until_delivered() {
    let repo = repository().await;
    conformance::outbox::publish_failure_after_commit_retains_outbox_row_until_delivered(
        repo.clone(),
        repo.outbox_store(),
    )
    .await;
}