distributed 1.7.0

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
142
143
144
145
146

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

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

use distributed::PostgresRepository;

async fn repository() -> Option<PostgresRepository> {
    let schema = postgres::PostgresTestSchema::create_from_env(
        "pg_conformance",
        "skipping Postgres conformance test",
    )
    .await?;
    Some(schema.repository().await)
}

#[tokio::test]
async fn aggregate_checkout_flow_persists_reloaded_state() {
    let Some(repo) = repository().await else {
        return;
    };
    conformance::scenario::aggregate_checkout_flow_persists_reloaded_state(repo).await;
}

#[tokio::test]
async fn get_all_and_commit_all_round_trip() {
    let Some(repo) = repository().await else {
        return;
    };
    conformance::scenario::get_all_and_commit_all_round_trip(repo).await;
}

#[tokio::test]
async fn multi_stream_conflict_rolls_back_other_stream_and_snapshot() {
    let Some(repo) = repository().await else {
        return;
    };
    conformance::scenario::multi_stream_conflict_rolls_back_other_stream_and_snapshot(repo).await;
}

#[tokio::test]
async fn duplicate_stream_identity_is_rejected_before_write() {
    let Some(repo) = repository().await else {
        return;
    };
    conformance::scenario::duplicate_stream_identity_is_rejected_before_write(repo).await;
}

#[tokio::test]
async fn metadata_round_trips() {
    let Some(repo) = repository().await else {
        return;
    };
    conformance::scenario::metadata_round_trips(repo).await;
}

#[tokio::test]
async fn unsupported_codec_is_rejected_on_write() {
    let Some(repo) = repository().await else {
        return;
    };
    conformance::scenario::unsupported_codec_is_rejected_on_write(repo).await;
}

#[tokio::test]
async fn snapshots_use_full_stream_identity() {
    let Some(repo) = repository().await else {
        return;
    };
    conformance::scenario::snapshots_use_full_stream_identity(repo).await;
}

#[tokio::test]
async fn high_level_outbox_commit_persists_row_without_stream() {
    let Some(repo) = repository().await else {
        return;
    };
    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() {
    let Some(repo) = repository().await else {
        return;
    };
    conformance::outbox::duplicate_outbox_insert_rolls_back_aggregate(repo).await;
}

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

#[tokio::test]
async fn worker_claim_complete_and_retry_lifecycle() {
    let Some(repo) = repository().await else {
        return;
    };
    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 Some(repo) = repository().await else {
        return;
    };
    conformance::outbox::worker_claim_by_ids_claims_only_requested(
        repo.clone(),
        repo.outbox_store(),
    )
    .await;
}

#[tokio::test]
async fn consumer_inbox_records_dedupes_and_fences_with_real_effects() {
    let Some(repo) = repository().await else {
        return;
    };
    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() {
    let Some(repo) = repository().await else {
        return;
    };
    conformance::inbox::inbox_rejects_empty_receipt(repo).await;
}