eventcore-postgres 1.0.1

PostgreSQL event store adapter for EventCore event sourcing library
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

mod common;

use std::sync::Arc;

use eventcore_types::{
    Event, EventStore, EventStoreError, StreamId, StreamVersion, StreamWrites, collect_events,
};
use serde::{Deserialize, Serialize};
use tokio::sync::Barrier;
use uuid::Uuid;

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct TestEvent {
    stream_id: StreamId,
    payload: String,
}

impl Event for TestEvent {
    fn stream_id(&self) -> &StreamId {
        &self.stream_id
    }

    fn event_type_name() -> &'static str {
        "TestEvent"
    }
}

fn unique_stream_id(prefix: &str) -> StreamId {
    StreamId::try_new(format!("{}-{}", prefix, Uuid::now_v7())).expect("valid stream id")
}

fn build_single_stream_writes(
    stream_id: &StreamId,
    expected_version: StreamVersion,
    payload: &str,
) -> StreamWrites {
    StreamWrites::new()
        .register_stream(stream_id.clone(), expected_version)
        .and_then(|writes| {
            writes.append(TestEvent {
                stream_id: stream_id.clone(),
                payload: payload.to_string(),
            })
        })
        .expect("single stream writes should register stream and append event")
}

#[tokio::test]
#[tracing_test::traced_test]
async fn developer_retries_after_postgres_version_conflict() {
    // Given: A migrated Postgres store that enforces optimistic concurrency
    let store = common::create_test_store().await;

    // Unique stream ID per test run for parallel execution
    let stream_id = unique_stream_id("account/concurrency");

    // And: Two separate writes targeting the same stream at version 0
    let first_writes =
        build_single_stream_writes(&stream_id, StreamVersion::new(0), "first deposit");
    let second_writes =
        build_single_stream_writes(&stream_id, StreamVersion::new(0), "second deposit");

    // When: Both writes attempt to append concurrently
    let barrier = Arc::new(Barrier::new(3));
    let first_store = store.clone();
    let first_barrier = barrier.clone();
    let first_handle = tokio::spawn(async move {
        let _ = first_barrier.wait().await;
        first_store.append_events(first_writes).await
    });

    let second_store = store.clone();
    let second_barrier = barrier.clone();
    let second_handle = tokio::spawn(async move {
        let _ = second_barrier.wait().await;
        second_store.append_events(second_writes).await
    });

    let _ = barrier.wait().await;

    let first_result = first_handle
        .await
        .expect("first append task should join without panic");
    let second_result = second_handle
        .await
        .expect("second append task should join without panic");

    let success_count = [&first_result, &second_result]
        .into_iter()
        .filter(|result| result.is_ok())
        .count();
    let conflict_count = [&first_result, &second_result]
        .into_iter()
        .filter(|result| matches!(result, Err(EventStoreError::VersionConflict { .. })))
        .count();

    // And: Developer retries with expected version 1 after reloading state
    let retry_writes = build_single_stream_writes(
        &stream_id,
        StreamVersion::new(1),
        "retry deposit after conflict",
    );
    let retry_result = store.append_events(retry_writes).await;

    // Then: Exactly one write succeeds initially, the other reports conflict, retry succeeds, two events persist, and instrumentation logs conflict
    let read_result = store.read_stream::<TestEvent>(stream_id.clone()).await;
    let read_succeeded = read_result.is_ok();
    let total_events = match read_result {
        Ok(stream) => collect_events(stream)
            .await
            .map(|events: Vec<TestEvent>| events.len())
            .unwrap_or_default(),
        Err(_) => 0,
    };
    let logs_contain_conflict = logs_contain("postgres.version_conflict");
    let retry_succeeded = retry_result.is_ok();
    let retry_state = match &retry_result {
        Ok(_) => "ok".to_string(),
        Err(err) => err.to_string(),
    };

    assert!(
        success_count == 1
            && conflict_count == 1
            && retry_succeeded
            && read_succeeded
            && total_events == 2
            && logs_contain_conflict,
        "postgres store should surface version conflict, allow retry, persist events, and emit instrumentation; success_count={success_count}, conflict_count={conflict_count}, retry_state={retry_state}, total_events={total_events}, logs_contain_conflict={logs_contain_conflict}",
    );
}