Replication

Replication Architecture

Marmot v2 implements a Percolator-style Two-Phase Commit (2PC) protocol for distributed SQLite replication. This document provides comprehensive technical details on the architecture, data flows, and edge case handling.

Design Philosophy

  • CDC-first: Row-level Change Data Capture, not SQL statement replay
  • Leaderless: Any node can coordinate writes (no master election)
  • Quorum-based: Configurable consistency levels (ONE, QUORUM, ALL)
  • Eventually consistent: Anti-entropy background sync repairs divergence
  • Full replication: All nodes receive all data (no sharding)

Running Example: Users Table

Throughout this document, we'll use a consistent example to trace data through all stages:

-- Schema on all 3 nodes (Node 1, Node 2, Node 3)
CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    email TEXT UNIQUE,
    name TEXT,
    balance INTEGER DEFAULT 0
);
 
-- Initial data (already replicated)
-- id=1: alice@example.com, Alice, balance=100
-- id=2: bob@example.com, Bob, balance=50

Scenario: Node 1 executes UPDATE users SET balance = 75 WHERE id = 1

We'll trace this UPDATE through CDC capture, 2PC protocol, conflict detection, and recovery scenarios.

Two-Phase Commit Protocol

Marmot's 2PC differs from traditional database 2PC by using write intents for conflict detection and CDC replay for deterministic replication.

Phase 1: PREPARE

  1. Coordinator executes locally with preupdate hooks capturing CDC data
  2. Write intents created for each affected row (distributed locks)
  3. Captured rows encoded as canonical msgpack EncodedCapturedRow records
  4. DML bytes stored in the local CDC segment log; Pebble tracks transaction metadata and row-lock indexes
  5. Broadcast to replicas with encoded row bytes (not SQL)
  6. Replicas create write intents and store the same encoded row bytes in their local CDC segment log
  7. Return SUCCESS or CONFLICT to coordinator
Coordinator                      Replica Nodes
==========                       =============

Execute with CDC hooks
  ↓
Create write intents
  ↓
ReplicateTransaction(PREPARE) →  Create write intents
                                 Store encoded rows in CDC segment log
                              ←  Return: Success/Conflict

Phase 2: COMMIT

  1. Wait for quorum of PREPARE responses
  2. Send COMMIT to remote nodes first (not local)
  3. Wait for remote quorum acknowledgment
  4. Commit locally only after remote quorum achieved
  5. Cleanup: Delete write intents and CDC entries
Coordinator                      Replica Nodes
==========                       =============

Remote quorum PREPARE achieved ✓
  ↓
ReplicateTransaction(COMMIT) →   Decode and apply CDC rows to SQLite
                                 Mark transaction COMMITTED
                              ←  Return: CommitTS

Remote quorum COMMIT achieved ✓
  ↓
Commit locally
  ↓
Cleanup intents

Special Case: LOAD DATA LOCAL INFILE

LOAD DATA LOCAL INFILE is replicated as a bulk-load operation with payload transfer, not as row-by-row CDC extraction:

  1. Coordinator receives file bytes from client over MySQL protocol.
  2. PREPARE broadcasts load metadata (load_id, payload size, chunk hint).
  3. Replicas pull payload chunks from coordinator and persist durable LOAD intents.
  4. COMMIT applies the same payload locally on each node.
  5. Operation is recorded in CDC stream for replay/catch-up.

This keeps LOAD DATA LOCAL INFILE deterministic and durable across cluster members while preserving normal 2PC quorum semantics.

Example: 2PC for UPDATE users SET balance = 75 WHERE id = 1

Step 1: Node 1 (Coordinator) executes locally

Time: T1 (TxnID = 0x...0001)
      Physical: 1736000000000ms | NodeID: 1 | Logical: 1

Node 1 executes UPDATE on hookDB connection
  → Preupdate hook fires: SQLITE_UPDATE on "users"
  → Captures row images and primary-key intent key
  → Encodes one msgpack EncodedCapturedRow

Step 2: Coordinator creates write intent

Row lock on Node 1 (in memory):
  Key:   users/users:1
  Value: {
    TxnID: 0x...0001,
    NodeID: 1,
    StartTS: {Wall: 1736000000000000000, Logical: 1},
    IntentType: UPDATE,
    MarkedForCleanup: false
  }

Step 3: Broadcast PREPARE to Node 2 and Node 3

gRPC ReplicateTransaction (parallel to both):
{
  Phase: PREPARE,
  TxnID: 0x...0001,
  NodeID: 1,
  Database: "mydb",
  Statements: [{
    Type: UPDATE,
    TableName: "users",
    RowChange: {
      EncodedRowCodec: 1,
      EncodedRow: msgpack({
        Table: "users",
        Op: UPDATE,
        IntentKey: binary("users:1"),
        OldValues: {"id": msgpack(1), "balance": msgpack(100)},
        NewValues: {"id": msgpack(1), "balance": msgpack(75)}
      })
    }
  }]
}

Step 4: Node 2 and Node 3 process PREPARE

Node 2:
  1. Check Cuckoo filter for "users:1" → MISS (no conflict)
  2. Acquire row lock users/users:1 (in memory)
  3. Append encoded row to local cdcseg segment and publish transaction manifest
  4. Return: {Status: SUCCESS, PrepareTS: ...}

Node 3:
  1. Check Cuckoo filter for "users:1" → MISS (no conflict)
  2. Acquire row lock users/users:1 (in memory)
  3. Append encoded row to local cdcseg segment and publish transaction manifest
  4. Return: {Status: SUCCESS, PrepareTS: ...}

Step 5: Quorum achieved (2/3), send COMMIT

Node 1 received 2 SUCCESS responses (quorum = 2 for 3-node cluster)

gRPC ReplicateTransaction to Node 2, Node 3:
{
  Phase: COMMIT,
  TxnID: 0x...0001,
  CommitTS: {Wall: 1736000000005000000, Logical: 0}
}

Step 6: Replicas apply CDC and commit

Node 2:
  1. Read encoded row from the local cdcseg transaction manifest
  2. Generate SQL: UPDATE users SET id=1, email='alice@example.com',
                   name='Alice', balance=75 WHERE id=1
  3. Execute on SQLite
  4. Update txn record: Status=COMMITTED
  5. Delete write intent (background)
  6. Return: {CommitTS: ...}

Node 3: (same process)

Step 7: Coordinator commits locally

Node 1:
  1. Remote quorum (2/3) committed ✓
  2. Apply own encoded CDC row to SQLite
  3. Update txn record: Status=COMMITTED
  4. Cleanup write intents
  5. Return success to client

Why Coordinator Commits Last

If remote quorum fails after local commit, the coordinator would be in an inconsistent state. By committing remotely first, abort is always safe.

Quorum Calculation

Critical: Quorum uses total cluster membership, not just alive nodes.

Quorum = floor(TotalMembership / 2) + 1
Cluster SizeQuorumSurvives
3 nodes21 failure
5 nodes32 failures
7 nodes43 failures

Split-brain prevention: In a 6-node cluster split 3x3:

  • Using alive nodes: Each partition sees N=3, quorum=2 → both write (dangerous)
  • Using total membership: N=6, quorum=4 → neither writes (safe)

CDC (Change Data Capture) Architecture

Why CDC Instead of SQL Replay

SQL ReplayCDC Replay
Non-deterministic (UUID, RANDOM, NOW)Exact byte-for-byte values
Parsing differences between dialectsBinary msgpack encoding
Auto-increment sequence conflictsCoordinator assigns distributed IDs
Slower (parse + execute)Faster (direct row application)

CDC Capture Flow

Row changes are captured via SQLite's preupdate hook API:

User Query (INSERT/UPDATE/DELETE)
  ↓
Execute on dedicated hookDB connection
  ↓
Preupdate hook fires BEFORE each row modification
  ↓
Extract: table, intentKey, oldValues, newValues
  ↓
Serialize one EncodedCapturedRow to msgpack
  ↓
Keep in bounded hook-session memory or spill to cdcseg for oversized transactions
  ↓
Release hookDB before 2PC broadcast

Encoded Row Format

type EncodedCapturedRow struct {
    Table     string
    Op        uint8             // INSERT=0, REPLACE=1, UPDATE=2, DELETE=3
    IntentKey []byte            // Binary intent key
    OldValues map[string][]byte // Before image, msgpack per column
    NewValues map[string][]byte // After image, msgpack per column
}

The DML wire payload is the encoded msgpack bytes plus a codec id. Decoded maps are reconstructed locally only for SQLite apply, vector-index hooks, and CDC publishing.

Intent Key Format

Single PK: "table:value" or "table:b64:base64_value"

Composite PK: "table:c:b64_col1:b64_col2:..." (columns sorted alphabetically)

NULL handling: "table:\x00NULL\x00" sentinel value

CDC Application on Replicas

OperationSQL Generated
INSERTINSERT OR REPLACE INTO table (...) VALUES (...)
UPDATEUPDATE table SET ... WHERE pk=... using the decoded old primary key image
DELETEDELETE FROM table WHERE pk=... using the decoded old primary key image

Every applied row change is versioned per primary key in __marmot_row_version, in the same SQLite transaction, with its transaction's commit timestamp: the HLC the coordinator decides after PREPARE and sends with COMMIT, tie-broken by the coordinator's node id. It is the same on every node. A change older than the row's stored version is skipped, so replaying peers' logs in any order ends with the newest image. A DELETE leaves a tombstone, so an older INSERT or UPDATE cannot bring the row back, and an UPDATE whose row is absent inserts its after-image. Tombstones are removed once older than gc_max_retention_hours times the member count; with unlimited retention they are kept. DROP TABLE removes a table's versions, RENAME TABLE carries them, and a change of primary-key columns clears them. LOAD DATA rows are versioned the same way, each at its transaction's commit timestamp.

Known limits: a table rebuilt by DDL (create a new table, copy the rows, drop the old one, rename) loses the old table's versions, so changes to its rows replayed afterwards apply as if the rows were unversioned. Read replicas apply their source's stream in order without row versions, so a replica that switches to another source may apply that source's order.

Example: Encoded Row for Our UPDATE

When Node 1 executes UPDATE users SET balance = 75 WHERE id = 1:

DML payload stored in cdcseg and referenced by /cdc/manifest/{txnID}:

Manifest: /cdc/manifest/0x...0001
Segment range: cdcseg-00000042.log offset=8192 length=148

EncodedCapturedRow (msgpack):
{
  Table:     "users",
  Op:        2,                    // UPDATE
  IntentKey: "users:1",            // Single PK format
  OldValues: {
    "id":      0x91 0x01,          // msgpack fixint: 1
    "email":   0xB1 "alice@example.com",  // msgpack fixstr
    "name":    0xA5 "Alice",       // msgpack fixstr
    "balance": 0xCC 0x64           // msgpack uint8: 100
  },
  NewValues: {
    "id":      0x91 0x01,          // msgpack fixint: 1
    "email":   0xB1 "alice@example.com",
    "name":    0xA5 "Alice",
    "balance": 0xCC 0x4B           // msgpack uint8: 75
  }
}

Generated SQL on Replicas:

-- Replicas execute this (generated from CDC, not original SQL):
UPDATE users SET
  id = 1,
  email = 'alice@example.com',
  name = 'Alice',
  balance = 75
WHERE id = 1

Example: CDC for INSERT and DELETE

INSERT INSERT INTO users (id, email, name, balance) VALUES (3, 'carol@example.com', 'Carol', 200):

EncodedCapturedRow:
{
  Op:        0,                    // INSERT
  Table:     "users",
  IntentKey: "users:3",
  OldValues: {},                   // Empty for INSERT
  NewValues: {
    "id": 3, "email": "carol@example.com", "name": "Carol", "balance": 200
  }
}

Generated SQL: INSERT OR REPLACE INTO users (id, email, name, balance)
               VALUES (3, 'carol@example.com', 'Carol', 200)

DELETE DELETE FROM users WHERE id = 2:

EncodedCapturedRow:
{
  Op:        3,                    // DELETE
  Table:     "users",
  IntentKey: "users:2",
  OldValues: {
    "id": 2, "email": "bob@example.com", "name": "Bob", "balance": 50
  },
  NewValues: {}                    // Empty for DELETE
}

Generated SQL: DELETE FROM users WHERE id = 2

Example: Composite Primary Key

CREATE TABLE order_items (
  order_id INTEGER,
  item_id INTEGER,
  quantity INTEGER,
  PRIMARY KEY (order_id, item_id)
);
 
UPDATE order_items SET quantity = 5 WHERE order_id = 100 AND item_id = 42;
IntentKey Format: "order_items:c:MTAw:NDI="
                  ^table       ^composite marker
                                ^base64(order_id=100):base64(item_id=42)
                  (columns sorted alphabetically: item_id, order_id)

Write Intents & Conflict Detection

Write Intent Structure

Write intents are distributed locks stored in MetaStore:

Row lock: (table, intentKey) → txnID, held in memory (RowLockStore); non-DML intents are also persisted under /intent_txn/{txnID}/{table}/{intentKey}
Value: WriteIntentRecord {
    TxnID, NodeID, StartTS, IntentType,
    Statement, DataSnapshot, MarkedForCleanup
}

Cuckoo Filter Fast Path

A Cuckoo filter provides O(1) conflict detection:

ParameterValue
Buckets250,000
Bucket size4 entries
Fingerprint32-bit
Capacity~1 million
False positive rate~2.3×10⁻¹⁰

Conflict check flow:

WriteIntent(table, intentKey)
  ↓
hash = XXH64(table + ":" + intentKey)
  ↓
filter.Check(hash)
  │
  ├─ MISS → Fast path: write directly to Pebble
  │
  └─ HIT → Slow path: lookup actual intent in Pebble
           ├─ No intent found → False positive, proceed
           ├─ Same txnID → Update in-place
           └─ Different txnID → Check if overwritable:
               ├─ MarkedForCleanup=true → Overwrite
               ├─ Status=COMMITTED/ABORTED → Overwrite
               ├─ Heartbeat timeout → Overwrite (stale)
               └─ Active transaction → Return CONFLICT

Sharded Mutexes

256 sharded mutexes prevent TOCTOU races:

func intentLockFor(table, intentKey string) *sync.Mutex {
    return &intentLocks[xxhash.Sum64String(table+":"+intentKey) % 256]
}

Example: Conflict Detection Scenarios

Using our users table, here are concrete conflict scenarios:

Scenario 1: No Conflict (Different Rows)

Time T1: Node 1 starts UPDATE users SET balance = 75 WHERE id = 1
         → Creates intent for "users:1"

Time T2: Node 2 starts UPDATE users SET balance = 30 WHERE id = 2
         → Cuckoo filter check for "users:2" → MISS
         → Creates intent for "users:2"
         → Both transactions succeed (different rows)

Scenario 2: Conflict (Same Row, Different Nodes)

Time T1: Node 1 starts UPDATE users SET balance = 75 WHERE id = 1
         TxnID: 0x...0001
         → Creates intent: row lock users/users:1 → {TxnID: 0x...0001}
         → Adds to Cuckoo filter: hash("users:1")

Time T2: Node 2 starts UPDATE users SET balance = 50 WHERE id = 1
         TxnID: 0x...0002
         → Cuckoo filter check for "users:1" → HIT
         → Look up row lock: users/users:1
         → Found intent with different TxnID (0x...0001)
         → Check if overwritable:
           - MarkedForCleanup = false
           - Status = PENDING (not COMMITTED/ABORTED)
           - LastHeartbeat = recent (not stale)
         → Return CONFLICT to Node 2
         → Node 2 aborts and retries after backoff

Scenario 3: Stale Intent (Coordinator Crashed)

Time T1: Node 1 starts UPDATE users SET balance = 75 WHERE id = 1
         → Creates intent: row lock users/users:1
         → Node 1 crashes before COMMIT

Time T2: (10 seconds later, heartbeat timeout)
         GC on Node 2 scans pending transactions
         → Finds intent with LastHeartbeat > 10s ago
         → Marks intent for cleanup

Time T3: Node 3 starts UPDATE users SET balance = 50 WHERE id = 1
         → Cuckoo filter check → HIT
         → Look up row lock → Found intent
         → Check: MarkedForCleanup = true
         → Overwrite allowed → Creates new intent
         → Transaction succeeds

Scenario 4: Same Transaction Re-entry

Time T1: Node 1 PREPARE phase creates intent for "users:1"
         TxnID: 0x...0001

Time T2: Network hiccup, Node 1 retries PREPARE
         → Cuckoo filter check → HIT
         → Look up row lock → Found intent with SAME TxnID
         → Update in-place (idempotent)
         → No conflict

Scenario 5: Multi-Row Transaction Partial Conflict

-- Node 1 executes:
BEGIN;
UPDATE users SET balance = balance - 25 WHERE id = 1;
UPDATE users SET balance = balance + 25 WHERE id = 2;
COMMIT;
Time T1: Node 1 creates intents for both rows:
         → row lock users/users:1 → TxnID: 0x...0001
         → row lock users/users:2 → TxnID: 0x...0001

Time T2: Node 2 tries UPDATE users SET name = 'Robert' WHERE id = 2
         → Cuckoo filter check for "users:2" → HIT
         → Conflict detected!
         → Node 2 returns CONFLICT
         → Node 2's client retries after Node 1 commits

Storage Layer (MetaStore)

PebbleDB Architecture

Each database has a separate MetaStore (CockroachDB Pebble):

data_dir/
├── mydb.db                    # SQLite database
├── mydb.db-wal               # SQLite WAL
├── mydb_meta.pebble/         # PebbleDB MetaStore
│   ├── MANIFEST-*
│   ├── CURRENT
│   └── *.sst                 # LSM-tree files
├── cdcseg/                   # Rolling CDC segment-log files
│   └── seg-000000000000000001.log
└── __marmot_system.db        # System database
    └── __marmot_system_meta.pebble/

Key Namespace

PrefixPurpose
/txn/{txnID}Immutable transaction record (id, node, start ts, database)
/txn_status/{txnID}Transaction status
/txn_commit/{txnID}Commit record
/txn_idx/txnid/{txnID}Transaction index by id (used for streaming)
/txn_prepared/{txnID}Prepared payload
/txn_idx/pend/{txnID}Pending transaction index
/txn_idx/seq/{seqNum}/{txnID}Local commit log, in position order
/intent_txn/{txnID}/{table}/{intentKey}Non-DML intents by transaction (DML row locks are in memory)
/cdc/manifest/{txnID}Manifest pointing to the transaction's CDC segment-log range
/log_seq/currentLease end of the local commit log's sequence
/log_pull_cursor/{peerID}This node's pull cursor into a peer's log
/log_consumed_pos/{peerID}How far a peer has consumed this node's log (for GC)
/log_truncated_throughHighest log position GC has deleted through
/log_reapply_pendingLocal-log re-apply after a snapshot restore is pending

CDC Segment Store

The CDC segment store is the local append-only payload store for DML row changes. Pebble keeps the transaction metadata, row-lock indexes, and /cdc/manifest/{txnID} pointer; the actual encoded row payloads live in cdcseg/seg-*.log.

Each row record is:

[32-byte header][msgpack EncodedCapturedRow payload]

The header stores a magic value, format version, record type, transaction id, row sequence, payload length, and payload CRC32. Segment files roll at 64 MiB, and a single record larger than one segment is rejected before append so recovery and validation use the same size rule.

Atomicity Model

CDC segment publication is manifest-based:

  1. Encoded row records are appended to the current segment file.
  2. The in-memory transaction manifest tracks row count, first/last row sequence, CRC, and one or more {segmentID, offset, length} chunks.
  3. SealCapturedRows publishes /cdc/manifest/{txnID} in Pebble after the segment ranges are known.
  4. Readers never scan arbitrary segment bytes for committed DML; they read the manifest and then perform bounded ReadAt calls over the referenced segment ranges.

This gives atomic visibility: a transaction's CDC payload is visible only when its manifest is visible. A crash before manifest publication leaves unreferenced segment bytes, which are ignored and later reclaimed. A crash after manifest publication leaves a manifest that points to specific validated byte ranges.

Durability And Prepare Sync

By default Marmot keeps the hot path aligned with Pebble NoSync: CDC segment fsync is grouped asynchronously by bytes, transaction count, or a short timer. This avoids forcing a disk flush per row or per transaction.

For deployments that require a PREPARE acknowledgement to wait for local CDC segment fsync, enable strict prepare sync:

[metastore]
strict_prepare_sync = true

or set MARMOT_CDC_PREPARE_SYNC=strict. Strict mode waits for the segment sync before acknowledging PREPARE, which strengthens local crash durability at the cost of write latency.

Recovery And Cleanup

On startup, Marmot validates each segment record in order. If it finds a torn header, short payload, invalid magic/version, oversized payload, or CRC mismatch, it truncates the active tail to the last valid record boundary.

Prepared manifests embedded in segment records can reconstruct prepared CDC state if Pebble metadata was lost after the segment record became durable. For committed transactions, /cdc/manifest/{txnID} is the retention root used by read-replica streaming, catch-up, and the anti-entropy log pull (FetchTransactions).

GC deletes segment files only when no current manifest, pending prepare, or active writer can reference them. Current and pending segments are retained; old unreferenced segments are pruned with the transaction metadata retention rules.

Transaction Record Format

type TransactionRecord struct {
    TxnID                uint64
    NodeID               uint64
    SeqNum               uint64      // Position in this node's local commit log
    Status               TxnStatus   // PENDING/COMMITTED/ABORTED
    StartTSWall          int64       // HLC wall time
    StartTSLogical       int32       // HLC logical
    CommitTSWall         int64
    CommitTSLogical      int32
    LastHeartbeat        int64       // Stale detection
    DatabaseName         string
}

Group Commit Optimization

SQLite commit batching groups committed DML transactions before applying decoded row changes. Pebble remains responsible for transaction metadata, row-lock indexes, schema state, and CDC manifests; DML row payload bytes live in rolling CDC segment-log files.

Example: MetaStore Keys After Our UPDATE

After Node 1's UPDATE users SET balance = 75 WHERE id = 1 completes:

During PREPARE phase (all 3 nodes):

row lock users/users:1 → TxnID 0x...0001 (in memory, not a Pebble key)
/cdc/manifest/0x...0001 → CDCManifest{SegmentID, Offset, Length, RowCount}
/txn/0x...0001 → TransactionRecord{Status: PENDING, ...}
/txn_idx/pend/0x...0001 → (pending index)

After COMMIT phase:

/txn/0x...0001 → TransactionRecord{
  TxnID: 0x...0001,
  NodeID: 1,
  SeqNum: 42,
  Status: COMMITTED,
  StartTSWall: 1736000000000000000,
  CommitTSWall: 1736000000005000000,
  DatabaseName: "mydb"
}
/txn_idx/seq/42/0x...0001 → (local commit log entry at seq 42)

-- Deleted (cleanup):
row lock users/users:1 → (released)
/txn_idx/pend/0x...0001 → (removed)

-- CDC segment range kept for peers' log pulls (GC'd after retention rules allow it):
/cdc/manifest/0x...0001 → (kept for peers' log pulls)

Log Pull State (per peer, in mydb's MetaStore):

-- On Node 2: its cursor into Node 1's log, and how far Node 1 has consumed Node 2's log
/log_pull_cursor/node1 → (42, 0x...0001)
/log_consumed_pos/node1 → (last position Node 1 reported)

-- Seq is local: Node 2 logged the same transaction at its own position
/txn_idx/seq/{node2 seq}/0x...0001

Hybrid Logical Clock (HLC)

Purpose

  • Causality tracking: Timestamps preserve happens-before relationship
  • Transaction IDs: Unique 64-bit IDs derived from HLC
  • Conflict resolution: Last-Write-Wins using HLC comparison

Clock Structure

type Clock struct {
    nodeID   uint64      // 0-63 (6-bit limit)
    wallTime int64       // Nanoseconds
    logical  int32       // 0-65535 (16-bit limit)
    mu       sync.Mutex
}

Transaction ID Bit Layout

|-------- 42 bits --------|-- 6 bits --|-- 16 bits --|
|     Physical (ms)       |  Node ID   |   Logical   |
  • 42 bits: Milliseconds since epoch (~139 years)
  • 6 bits: Node ID (max 64 nodes)
  • 16 bits: Logical counter (~65k IDs/ms/node)
func (t Timestamp) ToTxnID() uint64 {
    physicalMS := uint64(t.WallTime / 1_000_000)
    return (physicalMS << 22) | (t.NodeID << 16) | uint64(t.Logical)
}

Clock Synchronization

On receiving remote timestamp:

  1. Take max of: local wall, remote wall, physical now
  2. If same wall time: increment logical beyond both local and remote
  3. If the remote wall is the max: logical = remote logical + 1
  4. If physical time is ahead of both and the millisecond changed: reset logical to 0
  5. Overflow protection: Spin-wait if logical exceeds 65535

Example: HLC and TxnID Generation

Our UPDATE transaction on Node 1:

Physical time: 1736000000000 ms since epoch (Jan 4, 2025 14:13:20 UTC)
Node ID: 1
Logical: 1 (first transaction this millisecond)

TxnID calculation:
  Physical bits (42): 1736000000000 << 22 = 0x650C6A7400000000
  NodeID bits (6):    1 << 16              = 0x0000000000010000
  Logical bits (16):  1                    = 0x0000000000000001
  ─────────────────────────────────────────────────────────────
  TxnID:              0x650C6A7400010001

Binary layout:
  |---- 42 bits ----|-- 6 --|-- 16 --|
  |  1736000000000  |   1   |    1   |

Later examples abbreviate this id as 0x...0001.

Conflict resolution using HLC (Last-Write-Wins):

Scenario: Two updates to same row arrive at Node 3

Update A: commit ts (wall=1736000000000000000, logical=1, node=1)
Update B: commit ts (wall=1736000000001000000, logical=0, node=2)
Node 3 stamps __marmot_row_version with each applied change's commit ts;
a change older than the stored (wall, logical, node) is skipped.
B.wall > A.wall → B wins in either arrival order. Equal wall and logical → higher node wins.

Clock advance on message receive:

Node 2 local clock: Wall=1736000000000, Logical=5
Receives PREPARE from Node 1: Wall=1736000000002, Logical=1

Update logic:
  localWall (1736000000000) < remoteWall (1736000000002)
  → Set wall = max(local, remote, physicalNow) = 1736000000002
  → Remote wall is the max: logical = remote logical + 1 = 2

Node 2 new clock: Wall=1736000000002, Logical=2

Anti-Entropy Background Sync

Purpose

Repairs divergence caused by:

  • Network partitions healing
  • Node restarts with stale data
  • Missed transactions during temporary failures

Local Commit Log

Every node keeps a commit log per database. Each committed transaction gets a position (seq, txnID) in it, whether this node coordinated it, took part in its 2PC, or pulled it from a peer. Positions are local: the same transaction has a different seq on each node.

Because a pulled transaction is logged too, a node can receive any transaction from any member that holds it, even when no single peer holds everything it lacks.

Log Pull Flow

Anti-entropy runs every anti_entropy_interval_seconds. Each round:

  1. Reconcile databases: merge every ALIVE peer's database registry (ListDatabaseRegistry), so a missed CREATE or DROP DATABASE is repaired before any log is pulled.
  2. Pull every ALIVE peer's log for each database. Each (peer, database) pair has its own deadline, so one slow peer does not hold up the others.

One pull for one pair:

  1. List: ListCommittedLog returns up to 256 entries after the pair's cursor, the peer's truncation point and its schema version. The request also carries how far this node has consumed the peer's log; the peer stores that position for its GC.
  2. Diff: an entry whose txn id already has a __marmot_applied_txn marker is covered and skipped. Every commit writes its marker in the same SQLite transaction as its rows, whether it came through 2PC or a pull.
  3. Fetch: missing ids are fetched with FetchTransactions. The peer fails the call for a missing, uncommitted or incomplete transaction rather than send part of one.
  4. Apply: each fetched transaction is first written to this node's own log, then applied in one SQLite transaction together with its marker. Rows are stamped with the transaction's commit timestamp in __marmot_row_version, and a row a newer version owns is skipped. DDL in the transaction bumps __marmot_schema_version in that same SQLite transaction. The commit timestamp is merged into this node's clock.
  5. Advance the cursor: the pair's cursor moves over the contiguous prefix of covered entries and is persisted after every page. An entry that cannot be applied yet (a row held by a local write intent, say) keeps the cursor before it while the entries after it still apply.

The marker makes replay idempotent: a crash between logging and applying re-applies the transaction on the next pass, and a transaction reached through two peers applies once. A transaction whose replay fails 5 times in a row is reported STUCK and then retried every 20th pull of its pair.

Unfinished Local Transactions

Every listed entry is committed in the peer's log. When this node holds the same transaction unfinished:

Local stateAction
PreparedCommitted locally with the peer's commit timestamp
Prepared, but its captured rows were lost (crash before sync)Prepare discarded; the peer's copy is fetched and replayed
Begun, and its PREPARE died with an earlier processDiscarded; the peer's copy is fetched and replayed
Begun, PREPARE still runningLeft for a later pull

Caught Up and Promotion

A database is caught up when, in the last round, every member's log was reachable and this node's cursor reached each one's stable point.

A JOINING node turns ALIVE only when, for every database, at least one ALIVE peer was pulled and every ALIVE peer's pull succeeded, reached the end of the log and needed no snapshot, and its schema versions match or exceed every ALIVE peer's. The check covers ALIVE peers only, so a member that stays down does not keep a restarted node JOINING.

Log Truncation

GC deletes a prefix of the local log:

  • An entry older than gc_min_retention_hours is deleted once every current member has consumed past it. A member that has never reported a position counts as position zero and blocks the deletion.
  • An entry older than gc_max_retention_hours is deleted regardless. 0 sets no maximum.

A departed member's consumed position is dropped, so it no longer blocks GC. The truncation point only moves forward.

Snapshot Fallback

When a peer's truncation point is past this node's cursor into its log, the pull applies nothing and reports that a snapshot is needed. The restore:

  1. Durably marks a re-apply of this node's own log as pending.
  2. Restores the database from the ALIVE peer with the highest schema version (ties go to the lowest node id): GetSnapshotInfo, then StreamSnapshot in stream_chunk_size_kb chunks, each checked with MD5 and the file with SHA-256.
  3. Re-applies every entry of this node's own log that the restored file lacks (the markers say which). A failure leaves the mark set, and the next round retries.
  4. On the next pull from each peer, sets that pair's cursor to the peer's truncation point.

Startup Catch-Up

A starting node picks a seed. With no data of its own it restores a full snapshot. With data, it reconciles the database registry and pulls the seed's log for every database, restoring any database whose cursor the seed has truncated past. A node the pull finds behind is marked JOINING until the promotion check passes.

Mixed-Version Clusters

A peer that does not serve ListCommittedLog is skipped for the round and counts as not caught up. DDL and CREATE/DROP DATABASE return error 1213 (retryable) cluster-wide while any member does not report the log-pull protocol, on the coordinator and on each participant's PREPARE.

Read Replicas

StreamChanges serves read replicas only. A replica starting up compares its max txn id per database with its source's: a gap above delta_sync_threshold_transactions takes a snapshot, a smaller one is streamed. Cluster nodes do not use either threshold for catch-up; delta_sync_threshold_seconds is only read by config validation, which, with anti-entropy enabled, requires gc_min_retention_hours to be at least that long and a gc_max_retention_hours above 0 to be at least twice that.

Example: Anti-Entropy Scenarios

Using our 3-node cluster with the users table:

Scenario 1: Node 3 Brief Network Partition (Log Pull)

Timeline:
  T0: All nodes in sync
  T1: Node 3 loses network for 5 minutes
  T2: Node 1, Node 2 continue processing (quorum 2 of 3):
      - TxnID ...0001: UPDATE users SET balance=75 WHERE id=1
      - TxnID ...0002: INSERT INTO users VALUES (3, 'carol@...', 'Carol', 200)
      - TxnID ...0003: DELETE FROM users WHERE id=2
  T3: Node 3 reconnects

Anti-Entropy on Node 3, pair (Node 1, mydb):
  Node 3 → Node 1: ListCommittedLog(after=cursor, consumed=cursor)
  Node 1 returns: [(41, ...0001), (42, ...0002), (43, ...0003)], truncated=(0,0)
  Markers: none of the three applied
  Node 3 → Node 1: FetchTransactions([...0001, ...0002, ...0003])
  Node 3 applies each with its marker and row versions:
    - ...0001 → balance=75 for id=1
    - ...0002 → insert id=3
    - ...0003 → delete id=2, tombstone for users:2
  Cursor for (Node 1, mydb) → (43, ...0003)

Pair (Node 2, mydb):
  Node 2's log lists the same three txns at its own positions
  Markers: all three applied → covered, nothing fetched
  Cursor for (Node 2, mydb) advances

Scenario 2: New Node Joining (Snapshot Transfer)

Initial state:
  Node 1, Node 2: users table has 10K rows
  Node 4: New node, empty database

Node 4 starts:
  1. Startup catch-up: no local data → full snapshot from a seed
  2. Node 4 → Node 1: GetSnapshotInfo(db="mydb")
     Node 4 → Node 1: StreamSnapshot(db="mydb")
       chunks of stream_chunk_size_kb, each with an MD5,
       file verified with SHA-256
  3. Anti-entropy rounds pull Node 1's and Node 2's logs;
     transactions already in the snapshot have markers and are skipped
  4. Every database promotion-ready, schema versions match
  5. Node 4: JOINING → ALIVE

Scenario 3: Cursor Behind the Truncation Point

Timeline:
  T0: Node 3 goes offline
  T1: Entries on Node 1 pass gc_max_retention_hours (24hr)
      GC deletes them even though Node 3 has not consumed them
      Node 1 truncation point → (1000, ...)
  T2: Node 3 reconnects, its cursor into Node 1's log is (500, ...)

Pull (Node 1, mydb):
  ListCommittedLog returns truncated=(1000, ...)
  cursor (500) < truncated (1000) → needs snapshot, nothing applied

Snapshot fallback on Node 3:
  1. Mark re-apply of its own log pending
  2. Restore mydb from the ALIVE peer with the highest schema version
  3. Re-apply entries of Node 3's own log the restored file lacks
  4. Next pull from each peer starts at that peer's truncation point

Gossip Protocol (SWIM)

Node States

StateDescriptionReplication
ALIVEHealthy, respondingReceives writes
SUSPECTMay be dead, awaiting confirmationExcluded
DEADConfirmed failedExcluded
JOININGCatching upGossip only
LEAVINGGraceful shutdown in progressExcluded
REMOVEDAdmin removedExcluded

State Transitions

New node → JOINING → (catch-up complete) → ALIVE
ALIVE → (no response for SuspectTimeout) → SUSPECT
SUSPECT → (no response for DeadTimeout) → DEAD
SUSPECT → (received message) → ALIVE
SUSPECT/DEAD → (higher incarnation, node back) → ALIVE
ALIVE → (graceful stop) → LEAVING
any → (admin remove) → REMOVED
REMOVED → (admin allow) → DEAD, then rejoins with a higher incarnation

SWIM Protocol

  • Gossip interval: 1 second
  • Fanout: 3 random peers per round
  • Suspect timeout: 5 seconds (suspect_timeout_ms)
  • Dead timeout: 10 seconds (dead_timeout_ms)

Incarnation Numbers

Higher incarnation always wins (prevents stale state propagation):

if incoming.Incarnation > local.Incarnation {
    updateLocalState(incoming)
} else if incoming.Incarnation == local.Incarnation {
    // Only allow escalation: ALIVE → SUSPECT → DEAD
}

Example: Gossip and Failure Detection

Scenario: Node 2 crashes during our UPDATE transaction

Initial cluster state (all nodes ALIVE):
  Node 1: {ID: 1, State: ALIVE, Incarnation: 5, Addr: "10.0.0.1:5000"}
  Node 2: {ID: 2, State: ALIVE, Incarnation: 3, Addr: "10.0.0.2:5000"}
  Node 3: {ID: 3, State: ALIVE, Incarnation: 7, Addr: "10.0.0.3:5000"}

T0: Node 2 crashes (process killed)

T1 (+1s): Gossip round on Node 1
  - Node 1 randomly selects Node 2 for ping
  - gRPC Ping("10.0.0.2:5000") → timeout (5s)
  - No response, try indirect probe via Node 3

T2 (+6s): Indirect probe
  - Node 1 → Node 3: "Please ping Node 2 for me"
  - Node 3 → Node 2: Ping → timeout
  - Node 3 → Node 1: "Node 2 didn't respond"

T3 (+7s): Node 1 marks Node 2 as SUSPECT
  Gossip message: {NodeID: 2, State: SUSPECT, Incarnation: 3}

T4 (+8s): Gossip spreads to Node 3
  - Node 3 receives SUSPECT for Node 2
  - Updates local state: Node 2 = SUSPECT

T5 (+22s): Suspect timeout (15s) expires
  - No refutation from Node 2
  - Node 1 marks Node 2 as DEAD
  Gossip message: {NodeID: 2, State: DEAD, Incarnation: 3}

Effect on replication:
  - Quorum calculation now excludes Node 2
  - Total membership still = 3 (for split-brain safety)
  - Required quorum = 2 (floor(3/2)+1)
  - Node 1 + Node 3 can still achieve quorum

Scenario: False positive (Node 2 was just slow)

T0: Node 2 under heavy load (GC pause)

T1 (+5s): Node 1 marks Node 2 as SUSPECT
  Gossip: {NodeID: 2, State: SUSPECT, Incarnation: 3}

T2 (+7s): Node 2 GC pause ends, receives SUSPECT message about itself
  - "I'm being suspected? I'm alive!"
  - Increments incarnation: 3 → 4
  - Broadcasts refutation: {NodeID: 2, State: ALIVE, Incarnation: 4}

T3 (+8s): Node 1 receives refutation
  - Incoming incarnation (4) > local (3)
  - Updates: Node 2 = ALIVE, Incarnation = 4
  - Node 2 stays in cluster

Quorum unaffected: Node 2 was never marked DEAD

DDL Replication

Cluster-Wide Locking

type DDLLock struct {
    Database   string
    NodeID     uint64
    TxnID      uint64
    ExpiresAt  time.Time  // 30-second lease
}
  • One DDL per database at a time
  • Lock expires automatically (handles node crashes)
  • Different databases can have concurrent DDL

Idempotent Rewriting

DDL automatically rewritten for safe replay:

OriginalRewritten
CREATE TABLE fooCREATE TABLE IF NOT EXISTS foo
DROP TABLE fooDROP TABLE IF EXISTS foo
CREATE INDEX idxCREATE INDEX IF NOT EXISTS idx

Schema Version Tracking

Each database keeps its schema version in its own __marmot_schema_version table:

  • Bumped in the same SQLite transaction as the DDL, on the node that runs it and on every node that replays it from a log pull
  • Exchanged via gossip
  • A PREPARE whose required schema version is above the participant's local version is refused
  • Anti-entropy restores a snapshot from the peer with the highest schema version

Example: DDL Replication for ALTER TABLE

Adding a column to our users table:

-- Node 1 executes:
ALTER TABLE users ADD COLUMN created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP;
T0: Node 1 acquires DDL lock
  Key: /ddl_lock/mydb
  Value: {NodeID: 1, TxnID: 0x...0001, ExpiresAt: now()+30s}

T1: Node 1 broadcasts DDL lock acquisition via gossip
  All nodes pause DML on "mydb" database

T2: Node 1 executes locally
  - ALTER TABLE users ADD COLUMN ...
  - Bump __marmot_schema_version → 2 (same SQLite transaction)

T3: Node 1 broadcasts DDL to replicas (not CDC, raw SQL)
  gRPC ReplicateDDL:
  {
    Database: "mydb",
    Statement: "ALTER TABLE IF NOT EXISTS users ADD COLUMN IF NOT EXISTS
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP",
    SchemaVersion: 2,
    TxnID: 0x...0001
  }

  Note: Statement rewritten for idempotency

T4: Node 2, Node 3 execute DDL
  - Parse and execute SQL directly
  - Update local schema version: 2
  - Return success

T5: Node 1 releases DDL lock
  - Delete /ddl_lock/mydb
  - DML resumes on all nodes

T6: Subsequent DML includes new column
  INSERT INTO users (id, email, name, balance)
  VALUES (4, 'dave@...', 'Dave', 100);

  CDC captures:
  {
    NewValues: {
      "id": 4,
      "email": "dave@example.com",
      "name": "Dave",
      "balance": 100,
      "created_at": "2025-01-05 12:30:00"  ← New column included
    }
  }

Concurrent DDL conflict:

T0: Node 1 tries ALTER TABLE users ADD COLUMN age INT
T0: Node 2 tries ALTER TABLE users ADD COLUMN phone TEXT

Node 1:
  - Acquires /ddl_lock/mydb (wins)
  - Proceeds with DDL

Node 2:
  - Tries to acquire /ddl_lock/mydb
  - Lock exists, owned by Node 1
  - Check expiry: 25 seconds remaining
  - Return error: "DDL lock held by Node 1"
  - Client retries after Node 1 completes

Recovery Scenarios

Network Partition

ScenarioBehavior
Minority partitionWrites fail (cannot achieve quorum)
Majority partitionWrites succeed
Partition healsLog pull from every ALIVE peer; per-row versions keep the newest image

Node Failure & Recovery

ScenarioRecovery Method
Restart with dataStartup log pull from a seed, then log pull from every ALIVE peer; ALIVE once every database is caught up
Down while peers truncated past its cursor (gc_max_retention_hours)Snapshot restore, then re-apply of its own log
New node joiningFull snapshot, then log pull

Transaction Failure

Failure PointRecovery
PREPARE timeoutAbort all nodes
Partial PREPARE successAbort if quorum not achieved
COMMIT after remote quorumLocal must succeed (prepared)
Node crash during PREPAREHeartbeat timeout → intents cleaned
Node crash during COMMITLog pull: a prepared txn a peer committed is committed locally; if its rows were lost, it is replayed from the peer

Example: Recovery Scenarios for Our UPDATE

Scenario 1: Network Partition During PREPARE

Initial: UPDATE users SET balance = 75 WHERE id = 1 on Node 1

T0: Node 1 executes locally, creates intent
T1: Network partition isolates Node 1 from Node 2, Node 3

  Node 1 (minority partition):
    - Sends PREPARE to Node 2, Node 3
    - Both timeout after 5s
    - Required quorum: 2, achieved: 1 (self)
    - Decision: ABORT

  Node 1 rollback:
    - Release local row lock users/users:1
    - Update txn record: Status=ABORTED
    - Return error to client: "quorum not achieved"

  Client retries after partition heals.

Scenario 2: Coordinator Crash After PREPARE, Before COMMIT

T0: Node 1 PREPARE succeeds on all nodes
    All nodes have intent for "users:1"

T1: Node 1 crashes before sending COMMIT

T2 (+10s): Heartbeat timeout on Node 2
    - GC scans /txn_idx/pend/
    - Finds TxnID 0x...0001 with no recent heartbeat
    - Checks txn status: PENDING
    - Marks intents for cleanup

T3: Cleanup on Node 2, Node 3
    - Release row lock users/users:1
    - Update txn: Status=ABORTED
    - Remove from Cuckoo filter

Result: Transaction rolled back, row "users:1" unchanged
        Client receives timeout, can retry

Scenario 3: Replica Crash During COMMIT

T0: Node 1 PREPARE succeeds (quorum achieved)
T1: Node 1 sends COMMIT to Node 2, Node 3
T2: Node 3 crashes while applying CDC

State:
  Node 1: COMMITTED (coordinator)
  Node 2: COMMITTED
  Node 3: CRASHED (didn't complete commit)

T3: Node 3 restarts

T4: Anti-entropy on Node 3
  - Node 1's log lists our UPDATE; Node 3 has no marker for it
  - Node 3 still holds it prepared → commit it locally with
    the commit timestamp from Node 1's log entry
    (rows lost in the crash → discard the prepare, fetch and replay)
  - Node 3 now has balance=75 for id=1

Result: All nodes converged, transaction successful

Scenario 4: Partial PREPARE Success

T0: Node 1 sends PREPARE to Node 2, Node 3

T1: Responses:
  Node 2: SUCCESS (intent created)
  Node 3: CONFLICT (another transaction holds intent)

T2: Quorum check
  Required: 2 (floor(3/2)+1)
  SUCCESS responses: 1 (Node 2) + 1 (self) = 2
  But Node 3 returned CONFLICT

T3: Decision depends on consistency level
  If QUORUM: 2 SUCCESS ≥ 2 required → can proceed
  If ALL: 3 required, only 2 → must ABORT

  With QUORUM consistency:
    - Send COMMIT to Node 2 (achieved remote quorum with just Node 2)
    - Commit locally
    - Node 3 will get update via anti-entropy later

  With ALL consistency:
    - Send ABORT to Node 2
    - Rollback locally
    - Return CONFLICT to client

Garbage Collection

Two-Phase GC

Phase 1 (Critical, always runs):

  • Abort transactions without heartbeat for HeartbeatTimeout (10s)
  • Clean orphaned write intents

Phase 2 (Background):

  • Delete a prefix of the local commit log (below)
  • Purge row-version tombstones, in chunks, once older than gc_max_retention_hours times the member count

GC Coordination

  • Each peer reports, on every ListCommittedLog call, how far it has consumed this node's log; the position is stored per peer (/log_consumed_pos/{peerID})
  • The safe position is the minimum across current members; a member that never reported counts as zero
  • An entry older than gc_min_retention_hours is deleted only at or below the safe position
  • An entry older than gc_max_retention_hours (when above 0) is deleted regardless; a peer whose cursor ends up behind the truncation point restores from a snapshot
  • GC stops at the first entry it may not delete, so the log never has holes, and records the truncation point (/log_truncated_through)

Example: Garbage Collection Timeline

Continuing our UPDATE scenario over time:

T0: UPDATE users SET balance = 75 WHERE id = 1 committed
    TxnID: 0x...0001, at seq 42 in Node 1's log
    All nodes: Status=COMMITTED

T0 + 5s: Phase 1 GC runs
    - Scan /txn_idx/pend/ for stale transactions
    - Our transaction is COMMITTED, not pending
    - No action needed

T0 + 1hr: Phase 2 GC considers deletion
    - Transaction age: 1 hour (< gc_min_retention=2hr)
    - Stop: too recent

T0 + 2.5hr: Phase 2 GC runs again on Node 1
    - Transaction age: 2.5 hours (> gc_min_retention=2hr)
    - Consumed positions of Node 1's log:
      /log_consumed_pos/node2 → seq 42 ✓
      /log_consumed_pos/node3 → seq 42 ✓
    - Safe position = min = seq 42
    - Safe to delete

    Cleanup:
      DELETE /txn/0x...0001
      DELETE /txn_idx/seq/42/0x...0001
      DELETE /cdc/manifest/0x...0001
      /log_truncated_through → (42, 0x...0001)
      Reclaim unreferenced cdcseg ranges/files

T0 + 2.5hr: MetaStore state after GC
    - Transaction record: GONE
    - CDC manifest and reclaimed segment ranges: GONE
    - SQLite data: balance=75 for id=1 (permanent)

GC blocked by lagging peer:

T0: Transactions at seq 40-50 committed on Node 1

T1: Node 3 goes offline

T0 + 3hr: Phase 2 GC on Node 1
    - Consumed positions:
      /log_consumed_pos/node2 → seq 50 ✓
      /log_consumed_pos/node3 → seq 39 ✗ (offline, stale)
    - Safe position = min = seq 39
    - Cannot delete seq 40-50 (Node 3 hasn't consumed them)

T0 + 25hr: Max retention kicks in (> gc_max_retention=24hr)
    - Entries at seq 40-50 are > 24 hours old
    - Deleted regardless of consumed positions
    - Node 3's cursor is now behind Node 1's truncation point:
      it restores from a snapshot when it returns

Performance Optimizations

OptimizationMechanism
Cuckoo filterO(1) conflict detection, FP rate ~2.3×10⁻¹⁰
Sharded mutexes256 shards reduce lock contention
Grouped CDC syncBatch CDC segment fsync by bytes, transaction count, or timer
Sequence pre-allocationBatch 1000 sequence numbers
Object poolingReuse allocations in hot paths
Early quorum exitDon't wait for slow nodes
Parallel replicationGoroutines for all replica communication
CDC replayBinary values, no SQL parsing
NoSync cleanupIntent cleanup is idempotent

Limitations

Design Constraints

  1. Single writer per node: SQLite WAL limitation
  2. 64 nodes max: 6-bit NodeID in transaction ID
  3. ~65k txns/ms/node: 16-bit logical counter
  4. Full replication only: No sharding/partitioning
  5. Eventual consistency: No strict serializability across nodes

Not Supported

  • Distributed deadlock detection (client retries)
  • Read repair (anti-entropy only)
  • Cascading rollback (best-effort abort)
  • XA transactions (uses own 2PC)
  • Table-level selective replication

Complete End-to-End Example

This section traces our UPDATE users SET balance = 75 WHERE id = 1 through the entire system lifecycle:

┌─────────────────────────────────────────────────────────────────────────────┐
│  COMPLETE TRANSACTION LIFECYCLE                                              │
│  UPDATE users SET balance = 75 WHERE id = 1                                  │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                              │
│  T0: CLIENT REQUEST                                                          │
│  ══════════════════                                                          │
│  MySQL client → Node 1 (port 3306)                                          │
│  Query: UPDATE users SET balance = 75 WHERE id = 1                          │
│                                                                              │
│  T1: LOCAL EXECUTION + CDC CAPTURE                                           │
│  ═══════════════════════════════                                            │
│  Node 1 hookDB executes UPDATE                                              │
│    ↓                                                                        │
│  Preupdate hook fires (SQLITE_UPDATE)                                       │
│    ↓                                                                        │
│  EncodedCapturedRow created:                                                │
│    {Op: UPDATE, Table: "users", IntentKey: "users:1",                      │
│     OldValues: {balance: 100}, NewValues: {balance: 75}}                    │
│    Stored in bounded hook memory or cdcseg spill                            │
│                                                                              │
│  T2: WRITE INTENT CREATION                                                   │
│  ══════════════════════════                                                 │
│  Cuckoo filter: Add hash("users:1")                                         │
│  Mutex: Lock shard[hash("users:1") % 256]                                   │
│  MetaStore write:                                                           │
│    row lock users/users:1 → TxnID 0x...0001 (in memory)                      │
│    /cdc/manifest/0x...0001 → {SegmentID, Offset, Length, RowCount}         │
│    /txn/0x...0001 → {Status: PENDING, SeqNum: 42}                          │
│                                                                              │
│  T3: 2PC PREPARE PHASE (parallel to Node 2, Node 3)                         │
│  ═══════════════════════════════════════════════════                        │
│                                                                              │
│  Node 1 ──gRPC──→ Node 2: ReplicateTransaction(PREPARE)                     │
│           │                  ↓                                               │
│           │                Cuckoo check "users:1" → MISS                    │
│           │                Create intent, store encoded CDC row              │
│           │                  ↓                                               │
│           └──────────────← SUCCESS                                           │
│                                                                              │
│  Node 1 ──gRPC──→ Node 3: ReplicateTransaction(PREPARE)                     │
│           │                  ↓                                               │
│           │                Cuckoo check "users:1" → MISS                    │
│           │                Create intent, store encoded CDC row              │
│           │                  ↓                                               │
│           └──────────────← SUCCESS                                           │
│                                                                              │
│  Quorum: 3/3 SUCCESS (need 2 for 3-node cluster) ✓                          │
│                                                                              │
│  T4: 2PC COMMIT PHASE                                                        │
│  ═══════════════════                                                        │
│                                                                              │
│  Node 1 ──gRPC──→ Node 2: ReplicateTransaction(COMMIT)                      │
│           │                  ↓                                               │
│           │                Read encoded row via /cdc/manifest/0x...0001     │
│           │                Generate: UPDATE users SET ... WHERE id=1        │
│           │                Execute on SQLite                                 │
│           │                Update /txn/0x...0001 → COMMITTED                │
│           │                  ↓                                               │
│           └──────────────← CommitTS                                          │
│                                                                              │
│  Node 1 ──gRPC──→ Node 3: ReplicateTransaction(COMMIT)                      │
│           │                  ↓                                               │
│           │                (same as Node 2)                                  │
│           │                  ↓                                               │
│           └──────────────← CommitTS                                          │
│                                                                              │
│  Remote quorum achieved (2/2) ✓                                              │
│                                                                              │
│  T5: COORDINATOR LOCAL COMMIT                                                │
│  ═════════════════════════════                                              │
│  Node 1:                                                                     │
│    - Apply CDC to own SQLite                                                │
│    - Update /txn/0x...0001 → COMMITTED                                      │
│    - Schedule intent cleanup (background)                                   │
│                                                                              │
│  T6: RESPONSE TO CLIENT                                                      │
│  ═══════════════════════                                                    │
│  Node 1 → MySQL client: OK, 1 row affected                                  │
│                                                                              │
│  T7: BACKGROUND CLEANUP (async)                                              │
│  ═════════════════════════════                                              │
│  All nodes:                                                                  │
│    - Release row lock users/users:1                                          │
│    - Remove from Cuckoo filter                                              │
│    - Move /txn_idx/pend/0x...0001 → /txn_idx/seq/42/0x...0001              │
│                                                                              │
│  T8: LATER ANTI-ENTROPY ROUNDS                                               │
│  ═════════════════════════════                                               │
│  Each node pulls its peers' logs; our txn already has a                      │
│  __marmot_applied_txn marker everywhere, so it is skipped                    │
│  and each pull cursor moves past it                                          │
│                                                                              │
│  T9: FINAL STATE (all nodes)                                                 │
│  ════════════════════════════                                               │
│  SQLite: users.balance = 75 WHERE id = 1                                    │
│  MetaStore:                                                                 │
│    /txn/0x...0001 → {Status: COMMITTED, SeqNum: 42}                        │
│    /cdc/manifest/0x...0001 → cdcseg range kept for anti-entropy            │
│  Cuckoo filter: "users:1" removed                                           │
│                                                                              │
│  T10: GC (after 2+ hours)                                                    │
│  ═════════════════════════                                                  │
│  Delete transaction metadata, keep only SQLite data                         │
│                                                                              │
└─────────────────────────────────────────────────────────────────────────────┘

Configuration Reference

[replication]
default_write_consistency = "QUORUM"
default_read_consistency = "LOCAL_ONE"
write_timeout_ms = 5000
read_timeout_ms = 2000
 
# Anti-Entropy
enable_anti_entropy = true
anti_entropy_interval_seconds = 30
gc_interval_seconds = 60
delta_sync_threshold_transactions = 10000   # read-replica bootstrap: snapshot above this txn-id gap
delta_sync_threshold_seconds = 3600         # only checked against the gc_*_retention_hours values
 
# Garbage Collection
gc_min_retention_hours = 2
gc_max_retention_hours = 24
 
[transaction]
heartbeat_timeout_seconds = 10
conflict_window_seconds = 10
lock_wait_timeout_seconds = 50
 
[ddl]
lock_lease_seconds = 30
enable_idempotent = true
 
[cluster]
gossip_interval_ms = 1000
gossip_fanout = 3
suspect_timeout_ms = 5000
dead_timeout_ms = 10000

Key Files Reference

FilePurpose
coordinator/write_coordinator.go2PC protocol implementation
coordinator/handler.goMySQL query routing, CDC extraction
db/meta_store_pebble.goPebbleDB storage, write intents
db/preupdate_hook.goCDC capture via SQLite hooks
db/intent_filter.goCuckoo filter for conflict detection
db/transaction.goTransaction lifecycle, GC
grpc/replication_handler.goRemote 2PC phase handling
grpc/anti_entropy.goAnti-entropy rounds, snapshot fallback, promotion readiness
grpc/log_puller.goPulls a peer's commit log (cursor, markers, STUCK tracking)
grpc/log_server.goServes ListCommittedLog and FetchTransactions
grpc/catch_up.goStartup catch-up and snapshot restore
db/replay_apply.goApplies pulled transactions with their markers
db/row_version.goPer-row versions and tombstone purge
grpc/gossip.goSWIM protocol, membership
hlc/clock.goHybrid Logical Clock
publisher/CDC publishing system

Next Steps

  • Integrations - CDC publishing to Kafka, NATS, and external systems
ArchitectureIntegrations