Connect replicas — main (unreleased)

SafeMesh gives you records and merge rules; your application moves the bytes. After Persist and restart, this in-memory exercise shows that boundary without pretending to test a physical network.

Use it in your own project

SafeMesh is unreleased; use this pinned Git source instead of a crates.io package. This example’s pin is older than the CI-tested revision in the pinned-source install recipe; use that page for the current tested source install. With Git and Rust 1.96.1 installed, run cargo new --bin --vcs none connect-replicas outside the SafeMesh checkout. Replace the generated Cargo.toml dependency section with this block. The example needs no optional crate features.

[dependencies]
safemesh-crdt = { git = "https://github.com/velvetmonkey/safemesh.git", rev = "1321bf576dcc0e71cfe765b0dae0de518a5ae934" }

Paste the complete Rust example below into src/main.rs, then run cargo run --quiet from your new project. It prints the expected output shown below.

Exchange and repair

Inside the SafeMesh checkout, you can instead run the checkout command below; Try a merge explains how to get the source. to_wire_bytes() produces each outgoing record. At the receiving boundary, check its byte length, decode it, then use admit_with to apply only an accepted record. A duplicate must not apply the edit again.

The peer reports positive contiguous prefixes. The exchange format also retains its independent sequence-zero acknowledgement list, although all current writers start at sequence 1 and checked loaders refuse sequence-zero records. Reconstruct its claim with VersionVector::from_peer_prefixes_with_limits and VersionVectorLimits: the opt-in author and zero-replica budgets are checked before prefix validation and before either collection is cloned. None is unbounded. These budgets do not bound the version message’s decoding allocation: your decoder must enforce byte and entry limits before constructing either collection. The built-in version wire codec defaults to a maximum of 4,096 authors and 4,096 zero acknowledgements. These are configurable defaults: callers can choose tighter or larger entry budgets with VersionVector::from_wire_bytes_with_limits and should bound encoded bytes before decoding. This exercise passes the collections directly between objects; it does not use that codec.

use safemesh_crdt::{
    Admission, Crdt, EventLog, GCounter, GCounterDelta, Record, VersionVector, VersionVectorLimits,
    WireDecode, WireEncode,
};

fn main() {
    let mut left = GCounter::new(2);
    let mut right = GCounter::new(2);
    let mut sent = EventLog::for_crdt(&left);
    let mut received = EventLog::for_crdt(&right);
    for tally in [4, 6] {
        sent.append_with(
            &mut left,
            0,
            GCounterDelta { replica: 0, tally },
            |state, delta| state.apply_delta(delta.clone()),
        )
        .unwrap();
    }
    // Bytes leave the library here. An application would send these frames.
    let frames: Vec<_> = sent
        .records()
        .iter()
        .map(|record| record.to_wire_bytes().unwrap())
        .collect();
    let receive = |bytes: &[u8], state: &mut GCounter, log: &mut EventLog<GCounterDelta>| {
        assert!(bytes.len() <= 1024); // Check before decoding each record.
        let record = Record::<GCounterDelta>::from_wire_bytes(bytes).unwrap();
        log.admit_with(state, record, |state, delta| {
            state.apply_delta(delta.clone())
        })
    };
    // The first frame is lost; receiving sequence 2 must not acknowledge the gap.
    assert_eq!(
        receive(&frames[1], &mut right, &mut received),
        Admission::Accepted
    );
    assert_eq!(received.version().get(0), 0);
    // This example passes both version collections directly between objects.
    // Enforce byte/entry budgets in that decoder before allocating these collections.
    let peer = VersionVector::from_peer_prefixes_with_limits(
        received.version().entries(),
        received.version().zero_replicas(),
        VersionVectorLimits {
            max_authors: Some(2),
            max_zero_replicas: Some(2),
        },
    )
    .unwrap();
    let missing = sent.since(&peer);
    assert_eq!(missing.len(), 2); // Contiguous prefix conservatively resends sequence 2.
    let repair: Vec<_> = missing
        .iter()
        .map(|record| record.to_wire_bytes().unwrap())
        .collect();
    assert_eq!(
        receive(&repair[0], &mut right, &mut received),
        Admission::Accepted
    );
    assert_eq!(
        receive(&repair[1], &mut right, &mut received),
        Admission::Duplicate
    );
    assert_eq!(received.version().get(0), 2);
    assert_eq!(received.records().len(), 2);
    assert_eq!(right.value(), 6);
    assert_eq!(left.state(), right.state());
    assert!(sent.since(received.version()).is_empty());
    println!("recovered records=2 total=6 duplicate=unchanged");
}

cargo run --quiet --locked --manifest-path examples/gold-path/rust/Cargo.toml --example connect_replicas

Expected output:

recovered records=2 total=6 duplicate=unchanged

since(peer_version) selects records the peer has not acknowledged. Receiving sequence 2 before sequence 1 leaves the prefix at zero, so repair sends both; sequence 2 is a harmless duplicate. Version claims must come from records actually admitted: fabricated acknowledgements can hide missing records. Retry repair after lost messages; eventual delivery remains an application assumption. since retains all history and is not compaction.

A version vector tracks a contiguous prefix for each writer: if one sequence number never arrives, since() keeps offering every later record from that writer on every exchange. This is the designed behaviour; deliver the missing record to advance the prefix. See the resend cost and batch limit.

For an in-process transport loop, anti_entropy(transport, from, to, local_log, &mut remote_replica) queues repair, admits available deliveries, and returns (sending_peer, record_id, admission) for every delivered record. Inspect every verdict: Invalid(reason) names the refused record and reason on every retry; Collision is a separate verdict. Refusal does not acknowledge a record. There is no refusal cache to clear: after correcting the receiver’s configuration, the next ordinary sync retries it. The lower-level queue_anti_entropy only queues records and cannot observe remote admission. Rust callers migrating from the former send-only anti_entropy must either use the receiving helper or explicitly choose queue_anti_entropy and handle admission themselves. Network, Python and WASM loops must inspect the receiving merge verdict or exception on every sync and associate it with the sending peer and record ID; a successful send is not an admission acknowledgement.

Every WASM replica class with logBytes exposes the same selection: recordIds() lists record IDs as [author, sequence, ...] pairs without decoding payloads, versionVector() returns [author, versionFor(author), ...], and sinceLogBytes(peerVersion) returns the core’s since selection as one batch for the peer’s mergeLogBytes, under the same optional budgets. The pair shape carries no sequence-zero acknowledgement. Current writers cannot create such records, and checked loaders refuse them. The separate acknowledgement list remains in the full exchange format for compatibility. The short example is rust/crates/safemesh-wasm/examples/node-since-sync.mjs.

What the application owns

Responsibility	Application work
Transport	Framing, byte and entry budgets before decoding, authentication, encryption, routing, retry and exchanging both version collections. SafeMesh does not provide a network or discovery service.
Storage	Persist accepted history before durable acknowledgement; choose transaction and recovery behavior. The Linux durable adapter covers the Rust paths, while the Node journey owns its file writes. Retain history: there is no public snapshot/compaction API.
Identity allocation	Assign exclusive writer identities and consistent replica counts; preserve allocation on restart, and never run a cloned writer or old backup as a live writer.

For WASM OR-Sets created with createAllocated, appendAdd rejects caller-supplied tokens, so use appendAllocatedAdd and preserve exportIdentity() / importIdentity() state across restart.

Read the complete integration contracts and input and history limits, then Evaluate guarantees.