Struct FormatCrdtProcessor 

pub struct FormatCrdtProcessor<P: FormatBatchProcessor> { /* private fields */ }

CRDT-enabled processor for any format implementing FormatBatchProcessor

This wraps a FormatDsonProcessor and adds CRDT semantics:

• Vector clock tracking for causality
• Last-Writer-Wins (LWW) register semantics per field
• Delta-state synchronization
• Operation buffering for causal ordering

Implementations

impl<P: FormatBatchProcessor> FormatCrdtProcessor<P>

pub fn new(batch_processor: P, replica_id: impl Into<String>) -> Self

Create a new CRDT processor with a replica ID

pub fn with_strategy(self, strategy: MergeStrategy) -> Self

Create with a specific merge strategy

pub fn format_kind(&self) -> FormatKind

Get the format kind

pub fn replica_id(&self) -> &str

Get the replica ID

pub const fn vector_clock(&self) -> &VectorClock

Get the current vector clock

pub const fn lamport_timestamp(&self) -> u64

Get the current Lamport timestamp

pub fn process( &mut self, data: &[u8], schema: &CompiledSchema, ) -> Result<FormatBatchResult>

Process data with schema filtering and track as CRDT state

Errors

Returns an error if processing fails

pub fn process_with_operations( &mut self, data: &[u8], schema: &CompiledSchema, operations: &[DsonOperation], ) -> Result<FormatBatchResult>

Process with DSON operations and CRDT tracking

Errors

Returns an error if processing fails

pub fn process_unfiltered(&mut self, data: &[u8]) -> Result<FormatBatchResult>

Process unfiltered data with CRDT tracking

Errors

Returns an error if processing fails

pub fn get_document(&self, doc_id: &str) -> Option<&str>

Get a document by ID

pub fn document_ids(&self) -> Vec<&str>

Get all document IDs

pub fn document_count(&self) -> usize

Get number of tracked documents

pub fn reset(&mut self)

Reset the processor

pub const fn dson_processor(&self) -> &FormatDsonProcessor<P>

Get reference to underlying DSON processor

pub fn dson_processor_mut(&mut self) -> &mut FormatDsonProcessor<P>

Get mutable reference to underlying DSON processor

Trait Implementations

impl<P: FormatBatchProcessor> CrdtMerge for FormatCrdtProcessor<P>

fn merge_operation( &mut self, op: CrdtOperation, ) -> Result<Option<MergeConflict>>

Merge a remote operation into local state

fn merge_field( &mut self, path: &str, value: OperationValue, timestamp: u64, strategy: &MergeStrategy, ) -> Result<Option<MergeConflict>>

Merge field with CRDT semantics

fn vector_clock(&self) -> &VectorClock

Get the current vector clock

fn replica_id(&self) -> &str

Get the replica ID

fn resolve_conflict( &mut self, conflict: &MergeConflict, strategy: &MergeStrategy, ) -> Result<OperationValue>

Resolve a conflict using the specified strategy

impl<P: FormatBatchProcessor> DeltaCrdt for FormatCrdtProcessor<P>

type Delta = FormatDelta

Type representing a delta (difference from previous state)

fn generate_delta(&self, since: &VectorClock) -> Self::Delta

Generate delta since given vector clock

fn apply_delta(&mut self, delta: Self::Delta) -> Result<Vec<MergeConflict>>

Apply delta from remote replica

fn compact(&mut self)

Compact/garbage collect old deltas

impl<P: FormatBatchProcessor> OpBasedCrdt for FormatCrdtProcessor<P>

fn prepare(&self, op: &DsonOperation) -> Result<CrdtOperation>

Prepare an operation for broadcast (downstream precondition)

fn effect(&mut self, op: CrdtOperation) -> Result<Option<MergeConflict>>

Effect an operation (apply after delivery)

fn is_causally_ready(&self, op: &CrdtOperation) -> bool

Check if operation is causally ready to be applied

fn buffer_operation(&mut self, op: CrdtOperation)

Buffer operation until causally ready

fn process_buffered(&mut self) -> Result<Vec<MergeConflict>>

Process buffered operations that are now ready