distributed 4.8.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

use super::*;

/// Stage one compiler-sealed same-transaction projected upsert against cloned
/// row/protocol state. The caller publishes both clones only after every
/// domain, ledger, and projection participant has passed validation.
pub(in crate::in_memory_repo) fn stage_same_transaction_projection(
    protocol: &mut InMemoryProjectionProtocolState,
    staged_rows: &mut HashMap<String, StoredRow>,
    batch: &SameTransactionProjectionBatch,
    retention: ProjectionChangeRetention,
) -> Result<SameTransactionProjectionEvidence, ProjectionProtocolError> {
    batch.validate()?;
    protocol.require_registered_topology(&batch.topology)?;

    let partition_key = PartitionKey::new(&batch.topology, &batch.partition);
    protocol.ensure_partition(&partition_key, &batch.change_epoch)?;
    if let Some(failure_id) = protocol
        .partitions
        .get(&partition_key)
        .and_then(|partition| partition.stopped_failure_id.clone())
    {
        return Err(ProjectionProtocolError::PartitionStopped { failure_id });
    }
    if protocol
        .partitions
        .get(&partition_key)
        .is_some_and(|partition| partition.pending_retry_failure_id.is_some())
    {
        return Err(ProjectionProtocolError::IncomparableInput);
    }
    protocol.register_same_transaction_ownership(&partition_key, batch)?;

    let mutation = &batch.mutations[0];
    let lock_key = mutation.mutation.lock_key();
    let row_exists = staged_rows.contains_key(&lock_key);
    let revision = match protocol.records.get(&mutation.scope) {
        None if !row_exists => RecordRevision::new(mutation.scope.clone(), 1, 1)?,
        None => {
            return Err(ProjectionProtocolError::RecordAlreadyExists {
                model: mutation.scope.model().to_string(),
            });
        }
        Some(metadata) if metadata.tombstone => {
            return Err(ProjectionProtocolError::RecordTombstoned {
                model: mutation.scope.model().to_string(),
            });
        }
        Some(_) if !row_exists => {
            return Err(ProjectionProtocolError::RecordMissing {
                model: mutation.scope.model().to_string(),
            });
        }
        Some(metadata) => RecordRevision::new(
            mutation.scope.clone(),
            metadata.revision.incarnation(),
            checked_next(metadata.revision.revision(), "record revision")?,
        )?,
    };

    let change = protocol.append_change(
        &partition_key,
        PendingChange {
            kind: ProjectionChangeKind::RecordUpsert,
            causation_id: batch.causation_id.clone(),
            observation_kind: None,
            scope: Some(mutation.scope.clone()),
            revision: Some(revision.clone()),
            failure_id: None,
        },
    )?;
    let metadata = ProjectionRecordMetadata {
        revision: revision.clone(),
        tombstone: false,
        change: change.cursor.clone(),
    };
    protocol.ensure_live_record_identity_available(&metadata)?;
    protocol
        .records
        .insert(mutation.scope.clone(), metadata.clone());

    apply_read_model_write_plan(
        TableWritePlan::new(vec![mutation.mutation.clone()]),
        staged_rows,
    )?;
    if !staged_rows.contains_key(&lock_key) {
        return Err(ProjectionProtocolError::InvalidBatch(format!(
            "direct projection upsert left model `{}` without a physical row",
            mutation.scope.model()
        )));
    }

    let observation_key = ObservationKey {
        causation_id: batch.causation_id.clone(),
        scope: mutation.scope.clone(),
        kind: ProjectionObservationKind::Record,
    };
    if protocol.observations.contains_key(&observation_key) {
        return Err(ProjectionProtocolError::InvalidBatch(format!(
            "direct projection causation `{}` already observed this record",
            batch.causation_id
        )));
    }
    let observation = ProjectionObservation {
        causation_id: batch.causation_id.clone(),
        kind: ProjectionObservationKind::Record,
        revision: Some(revision),
        scope: mutation.scope.clone(),
        change: change.cursor.clone(),
    };
    protocol
        .observations
        .insert(observation_key, observation.clone());
    protocol.retain_change_suffix(&partition_key, retention)?;

    Ok(SameTransactionProjectionEvidence {
        records: vec![metadata],
        changes: vec![change],
        observations: vec![observation],
    })
}