SierraDB Distributed Event Store

SierraDB uses a multi-phase protocol to coordinate writes across a cluster. The process involves a Client, a Coordinator (selected per partition), and multiple Replicas.

Write Lifecycle Phases

1. Write Request Routing: A client sends a WriteRequest(transaction, expected_seq?) to any node. If the node is not the coordinator for that partition, it forwards the request to the correct coordinator.
2. Coordinator Write Preparation: The coordinator validates the request, assigns a partition sequence number, and writes the data locally.
3. Replica Write Replication: The coordinator sends ReplicateWrite(partition_seq, transaction) to all replicas in parallel. Replicas validate the coordinator and the sequence. If a sequence gap is detected, the replica buffers the write and returns SEQUENCE_GAP.
4. Quorum Confirmation: Once the coordinator receives enough SUCCESS responses to meet the quorum requirement, it marks the write as confirmed locally.
5. Write Confirmation: The coordinator notifies replicas via ConfirmWrite(partition_seq), and the final WriteResponse(SUCCESS, partition_seq) is returned to the client.

Replica Recovery (Catch-up)

If a replica misses writes, it initiates a catch-up process:

1. Replica sends PartitionSyncRequest(from_seq) to the coordinator.
2. Coordinator responds with PartitionSyncResponse(missing_events).
3. Replica applies the missing events and then processes any writes previously held in its buffer.

@startuml
participant Client as "Client/Node"
participant AnyNode as "Any Replica Node"
participant Coordinator as "Coordinator Node"
participant Replica1 as "Replica Node 1"
participant Replica2 as "Replica Node 2"
participant Replica3 as "Replica Node 3"

== Phase 1: Write Request Routing ==
Client -> AnyNode : WriteRequest(transaction, expected_seq?)
AnyNode -> AnyNode : get_coordinator_for_partition()

alt Current node is coordinator
    AnyNode -> AnyNode : Process as coordinator
else Forward to coordinator
    AnyNode -> Coordinator : WriteRequest(transaction, expected_seq?)
end

== Phase 2: Coordinator Write Preparation ==
Coordinator -> Coordinator : validate_write_request()
Coordinator -> Coordinator : assign_partition_sequence()
Coordinator -> Coordinator : write_locally() ✓

== Phase 3: Replica Write Replication (Parallel) ==
par Replicate to Replica1
    Coordinator -> Replica1 : ReplicateWrite(partition_seq, transaction)
end
par Replicate to Replica2
    Coordinator -> Replica2 : ReplicateWrite(partition_seq, transaction)
end
par Replicate to Replica3
    Coordinator -> Replica3 : ReplicateWrite(partition_seq, transaction)
end

== Each replica processes sequentially ==

Replica1 -> Replica1 : validate_coordinator()
Replica1 -> Replica1 : validate_sequence()
alt Sequence ready (Replica1)
    Replica1 -> Replica1 : write_locally() ✓
    Replica1 -> Coordinator : ReplicateWriteResponse(SUCCESS)
else Sequence gap (Replica1)
    Replica1 -> Replica1 : buffer_write()
    Replica1 -> Coordinator : ReplicateWriteResponse(SEQUENCE_GAP)
end

Replica2 -> Replica2 : validate_coordinator()
Replica2 -> Replica2 : validate_sequence()
Replica2 -> Replica2 : write_locally() ✓
Replica2 -> Coordinator : ReplicateWriteResponse(SUCCESS)

Replica3 -> Replica3 : validate_coordinator()
Replica3 -> Replica3 : validate_sequence()
Replica3 -> Replica3 : write_locally() ✓
Replica3 -> Coordinator : ReplicateWriteResponse(SUCCESS)

== Phase 4: Quorum Confirmation ==
Coordinator -> Coordinator : count_confirmations()
Coordinator -> Coordinator : check_quorum_achieved() ✓

== Phase 5: Write Confirmation ==
Coordinator -> Coordinator : mark_local_confirmed()

par Confirm Replica2
    Coordinator -> Replica2 : ConfirmWrite(partition_seq)
end
par Confirm Replica3
    Coordinator -> Replica3 : ConfirmWrite(partition_seq)
end

Replica2 -> Replica2 : mark_confirmed()
Replica3 -> Replica3 : mark_confirmed()

alt Direct response (if client connected to coordinator)
    Coordinator -> Client : WriteResponse(SUCCESS, partition_seq)
else Via forwarding node
    Coordinator -> AnyNode : WriteResponse(SUCCESS, partition_seq)
    AnyNode -> Client : WriteResponse(SUCCESS, partition_seq)
end

== Background: Failed Replica Recovery ==
note over Replica1 : Replica1 initiates catch-up
Replica1 -> Coordinator : PartitionSyncRequest(from_seq)
Coordinator -> Replica1 : PartitionSyncResponse(missing_events)
Replica1 -> Replica1 : apply_missing_events()
Replica1 -> Replica1 : process_buffered_writes()

@enduml

To maintain high performance while ensuring sequential processing per partition, SierraDB employs several optimization strategies and buffer management rules.

Concurrent Write Optimization

The coordinator can work ahead of the replicas to maximize throughput:

1. Immediate Sequence Assignment: The coordinator assigns sequence numbers without waiting for previous writes to complete.
2. Parallel Replication: Replication requests are sent to all replicas simultaneously.
3. Pipelined Coordination: The coordinator can start replicating write $N+1$ while write $N$ is still undergoing confirmation.
4. Batch Confirmations: (Planned) Grouping multiple confirmations in a single message.

Buffer Management

Replicas use a sequential write buffer to handle out-of-order arrivals (sequence gaps). To prevent memory exhaustion, the following constraints are applied:

• Max Buffer Size: A limit on the number of buffered writes (e.g., MAX_BUFFER_SIZE = 1000).
• Buffer Timeout: Writes that stay in the buffer too long are evicted (e.g., BUFFER_TIMEOUT_MS = 5000).
• Eviction Policy: When the buffer exceeds capacity, the system retains only the writes immediately following the next_expected_seq to ensure continuity.

const MAX_BUFFER_SIZE: usize = 1000;
const BUFFER_TIMEOUT_MS: u64 = 5000;

// Evict old buffered writes to prevent memory issues
fn evict_expired_writes(&mut self, partition_id: PartitionId) {
    let buffer = self.get_buffer_mut(partition_id);
    let now = Instant::now();
    
    buffer.buffered_writes.retain(|_, write| {
        now.duration_since(write.received_at).as_millis() < BUFFER_TIMEOUT_MS as u128
    });
    
    if buffer.buffered_writes.len() > MAX_BUFFER_SIZE {
        // Keep only the next N writes we expect
        let next_seq = buffer.next_expected_seq.load(Ordering::Acquire);
        buffer.buffered_writes = buffer.buffered_writes
            .range(next_seq..next_seq + MAX_BUFFER_SIZE as u64)
            .map(|(k, v)| (*k, v.clone()))
            .collect();
    }
}