test(load): replication sanity check

Two-node local smoke test: leader on 9005, follower on 9006, each with
the other as a replication peer. Inserts 1000 docs on the leader, polls
the follower every 200 ms until it sees them. Reports catch-up time and
content integrity.

Validates that the replication path still fires alongside v1.7.0's
eviction + admission-control + client-retry changes. Not a full load
test — a smoke test to catch regressions during the v1.7.0 refactor.

First run already caught a latent pre-v1.7.0 bug (commit a96c168,
2026-02-05): the leader's broadcast path serializes only doc.data, but
the follower's applyReplicatedEntry calls Document::fromJson() which
expects the full envelope — so inserts replicate structurally but with
an empty body. README documents the reproducer and two candidate fixes.

tests/load_test/README.md

@@ -265,3 +265,75 @@ The recovery system decided the snapshot+WAL combination was unrecoverable. `run
 
     smartbotic-database --config /tmp/smartbotic-loadtest-crash/config_crash.json \
         --recovery-mode=snapshot_fallback --force-readwrite
+
+---
+
+## Test 6 — Replication sanity check (`run_replication_test.sh`)
+
+Spawns two local server instances (leader on 9005, follower on 9006), each configured with the other as a replication peer, inserts 1000 docs on the leader, then polls the follower every 200 ms until it sees them. Reports catch-up time and content integrity.
+
+This is a smoke test — it does NOT exercise replication at real-deployment scale. It verifies the replication mechanism still fires alongside v1.7.0's eviction / admission-control / client-retry changes. Because v1.7.0 didn't touch replication code, the expected outcome is "no regression" — if the mechanism breaks here, bisect toward something we did change (WAL ordering, store callbacks).
+
+### Build
+
+    cd tests/load_test
+    g++ -std=c++20 -O2 -I/usr/include load_test_replication.cpp \
+        -lsmartbotic-db-client $(pkg-config --libs grpc++) \
+        -lspdlog -lfmt -pthread -o load_test_replication
+    chmod +x run_replication_test.sh
+
+### Run
+
+    ./run_replication_test.sh [num_docs=1000] [max_wait_sec=30]
+
+### Configs
+
+- `config_replica_leader.json` — port 9005, node_id `replica-leader`, replication enabled, peer = `127.0.0.1:9006`, data dir `/tmp/smartbotic-loadtest-replica-leader/data`
+- `config_replica_follower.json` — port 9006, node_id `replica-follower`, replication enabled, peer = `127.0.0.1:9005`, data dir `/tmp/smartbotic-loadtest-replica-follower/data`
+
+Both configs use symmetric multi-master peering (as the replication code supports) — each side lists the other in `peers`. Inserts are performed only on the leader in this test; the follower is a pure consumer, but could also accept writes.
+
+### PASS criteria
+
+- All N docs visible on the follower
+- Zero content mismatches (the body field matches what the leader inserted)
+- Catch-up time < `max_wait_sec` seconds
+
+Exit codes:
+
+- `0` — PASS (full catch-up, zero corruption)
+- `2` — PARTIAL (some docs replicated, no corruption, did not fully catch up in time)
+- `1` — FAIL (either nothing replicated OR content mismatches observed)
+
+### Known quirk found while writing this test (pre-v1.7.0)
+
+A first run of this test against v1.7.0 produced **FAIL — 200 docs replicated but zero content matches**. The follower's copy of each document contained only the system fields (`_id`, `_version`, `_created_at`, `_updated_at`, etc.), with the user-supplied payload fields missing.
+
+Root cause is in the leader-side broadcast code at `service/src/database_service.cpp:515`:
+
+    if (doc) entry.set_data(doc->data.dump());
+
+This serializes only the document's `data` field. The follower-side apply code at `service/src/database_service.cpp:612` expects the full Document envelope:
+
+    auto json = nlohmann::json::parse(entry.data());
+    Document doc = Document::fromJson(json);   // looks for "data", "version", etc.
+
+`Document::fromJson` reads `j.value("data", {})` — i.e. it expects the payload nested under `data`, not at the top level. So the follower effectively discards the body.
+
+`git blame` shows this mismatch predates v1.7.0 (commit `a96c168`, 2026-02-05). It is a latent bug in the push path: inserts/updates are getting replicated structurally (the system fields propagate) but with an empty body on the follower.
+
+The pull path (`GetEntriesSince` → WAL) uses a different code path — it serializes the WAL entry's `data` field directly, so pull-based catch-up may or may not exhibit the same issue depending on WAL format. Not investigated here.
+
+This test catches the bug but is deliberately not wired to fix it — the goal of Test 6 is v1.7.0 sanity, and the bug is unrelated to the eviction refactor. File a separate ticket to either:
+
+1. On the leader, use `doc->toJson().dump()` instead of `doc->data.dump()`, OR
+2. On the follower, parse `entry.data()` as the raw document body and construct the Document from `entry`'s other fields (`collection`, `document_id`, `node_id`).
+
+### If replication silently doesn't happen
+
+Check both server logs for the peer handshake:
+
+    "Connected to peer 127.0.0.1:9006 (node: replica-follower, ...)"
+    "Connected to peer 127.0.0.1:9005 (node: replica-leader, ...)"
+
+Without both lines, peering never established — likely a firewall or port conflict. `ss -tlnp | grep 900[56]` will confirm both are listening.

tests/load_test/config_replica_follower.json

@@ -0,0 +1,58 @@
+{
+  "log_level": "info",
+  "storage": {
+    "bind_address": "127.0.0.1",
+    "rpc_port": 9006,
+    "node_id": "replica-follower",
+    "data_directory": "/tmp/smartbotic-loadtest-replica-follower/data",
+    "memory": {
+      "max_memory_mb": 64,
+      "eviction_threshold_percent": 80,
+      "eviction_target_percent": 50,
+      "eviction_check_interval_ms": 200,
+      "eviction_chunk_size": 500,
+      "eviction_chunk_pause_ms": 30,
+      "max_eviction_passes_per_trigger": 15,
+      "hot_write_floor_ms": 2000,
+      "memory_soft_percent": 60,
+      "memory_hard_percent": 80,
+      "memory_emergency_percent": 92,
+      "eviction_burst_threshold": 2000
+    },
+    "persistence": {
+      "wal_sync_interval_ms": 100,
+      "snapshot_interval_sec": 3600,
+      "compression": "lz4",
+      "snapshots": {
+        "validate_after_write": true,
+        "cleanup_only_if_verified": true
+      },
+      "recovery": {
+        "mode": "normal",
+        "auto_escalate": true,
+        "allow_empty_on_fresh_install": true
+      }
+    },
+    "encryption": {
+      "enabled": false
+    },
+    "migrations": {
+      "enabled": false
+    },
+    "files": {
+      "max_file_size_mb": 100
+    },
+    "replication": {
+      "enabled": true,
+      "conflict_resolution": "last_writer_wins",
+      "peers": ["127.0.0.1:9005"]
+    },
+    "grpc": {
+      "max_receive_message_size_mb": 100,
+      "max_send_message_size_mb": 100,
+      "resource_quota_memory_mb": 64,
+      "max_concurrent_subscribe_streams": 50,
+      "max_concurrent_file_streams": 10
+    }
+  }
+}

tests/load_test/config_replica_leader.json

@@ -0,0 +1,58 @@
+{
+  "log_level": "info",
+  "storage": {
+    "bind_address": "127.0.0.1",
+    "rpc_port": 9005,
+    "node_id": "replica-leader",
+    "data_directory": "/tmp/smartbotic-loadtest-replica-leader/data",
+    "memory": {
+      "max_memory_mb": 64,
+      "eviction_threshold_percent": 80,
+      "eviction_target_percent": 50,
+      "eviction_check_interval_ms": 200,
+      "eviction_chunk_size": 500,
+      "eviction_chunk_pause_ms": 30,
+      "max_eviction_passes_per_trigger": 15,
+      "hot_write_floor_ms": 2000,
+      "memory_soft_percent": 60,
+      "memory_hard_percent": 80,
+      "memory_emergency_percent": 92,
+      "eviction_burst_threshold": 2000
+    },
+    "persistence": {
+      "wal_sync_interval_ms": 100,
+      "snapshot_interval_sec": 3600,
+      "compression": "lz4",
+      "snapshots": {
+        "validate_after_write": true,
+        "cleanup_only_if_verified": true
+      },
+      "recovery": {
+        "mode": "normal",
+        "auto_escalate": true,
+        "allow_empty_on_fresh_install": true
+      }
+    },
+    "encryption": {
+      "enabled": false
+    },
+    "migrations": {
+      "enabled": false
+    },
+    "files": {
+      "max_file_size_mb": 100
+    },
+    "replication": {
+      "enabled": true,
+      "conflict_resolution": "last_writer_wins",
+      "peers": ["127.0.0.1:9006"]
+    },
+    "grpc": {
+      "max_receive_message_size_mb": 100,
+      "max_send_message_size_mb": 100,
+      "resource_quota_memory_mb": 64,
+      "max_concurrent_subscribe_streams": 50,
+      "max_concurrent_file_streams": 10
+    }
+  }
+}

tests/load_test/load_test_replication.cpp

@@ -0,0 +1,192 @@
+// Test 6 — Replication sanity check
+//
+// Spawns two local server instances (leader on 9005, follower on 9006)
+// via the orchestrator script; this binary inserts N deterministic docs
+// into a collection on the leader, then polls the follower every 200ms
+// counting how many of those docs are visible. Reports catch-up time,
+// docs propagated, and any content mismatches.
+//
+// This is a smoke test — validates that v1.7.0's eviction / pressure /
+// admission-control refactor did not break the replication path.
+// It does NOT exercise replication at real-deployment scale.
+//
+// Exit codes:
+//   0 — PASS     (all N docs replicated, zero content mismatches)
+//   1 — FAIL     (nothing replicated or content corrupted)
+//   2 — PARTIAL  (some docs replicated, no corruption, didn't fully catch up)
+
+#include <smartbotic/database/client.hpp>
+
+#include <chrono>
+#include <cstdio>
+#include <iostream>
+#include <string>
+#include <thread>
+
+using namespace smartbotic::database;
+using Clock = std::chrono::steady_clock;
+
+static std::string makeContent(int i) {
+    return "repl-" + std::to_string(i);
+}
+
+int main(int argc, char* argv[]) {
+    int numDocs = 1000;
+    int maxWaitSec = 30;
+    if (argc > 1) numDocs = std::stoi(argv[1]);
+    if (argc > 2) maxWaitSec = std::stoi(argv[2]);
+
+    std::cout << "=== smartbotic-database v1.7.0 replication sanity check ===\n"
+              << "Leader:        localhost:9005\n"
+              << "Follower:      localhost:9006\n"
+              << "Docs:          " << numDocs << "\n"
+              << "Max wait:      " << maxWaitSec << "s\n\n";
+
+    Client::Config lcfg;
+    lcfg.address = "localhost:9005";
+    lcfg.timeoutMs = 3000;
+    Client leader(lcfg);
+    if (!leader.connect()) {
+        std::cerr << "leader connect failed at " << lcfg.address << "\n";
+        return 1;
+    }
+
+    Client::Config fcfg = lcfg;
+    fcfg.address = "localhost:9006";
+    Client follower(fcfg);
+    if (!follower.connect()) {
+        std::cerr << "follower connect failed at " << fcfg.address << "\n";
+        return 1;
+    }
+
+    // Create collection on leader. The follower picks it up implicitly —
+    // applyReplicatedEntry() handles OP_CREATE_COLLECTION, and the push
+    // path of insert() routes through loadDocument() which auto-creates
+    // the collection if missing. We don't call createCollection() on the
+    // follower ourselves — replication is the thing under test.
+    try {
+        leader.createCollection("replica_test");
+    } catch (const std::exception& e) {
+        // Collection may already exist from a previous run — that's fine.
+        std::cout << "leader createCollection note: " << e.what() << "\n";
+    }
+
+    // ===== Phase 1: write to the leader =====
+    std::cout << "Phase 1: inserting " << numDocs << " docs on leader...\n";
+    auto t0 = Clock::now();
+    int inserted = 0;
+    int insertFailed = 0;
+    for (int i = 0; i < numDocs; ++i) {
+        nlohmann::json doc{{"seq", i}, {"content", makeContent(i)}};
+        try {
+            std::string id = "r" + std::to_string(i);
+            leader.insert("replica_test", doc, id);
+            ++inserted;
+        } catch (const std::exception& e) {
+            ++insertFailed;
+            if (insertFailed <= 5) {
+                std::cout << "  insert failed on leader: id=r" << i
+                          << " err=" << e.what() << "\n";
+            }
+        }
+    }
+    auto insertMs = std::chrono::duration_cast<std::chrono::milliseconds>(
+        Clock::now() - t0).count();
+    std::cout << "  inserted " << inserted << "/" << numDocs << " in " << insertMs
+              << "ms (" << insertFailed << " failed)\n\n";
+
+    if (inserted == 0) {
+        std::cerr << "zero successful inserts on leader — cannot test replication\n";
+        return 1;
+    }
+
+    // ===== Phase 2: poll follower until caught up =====
+    std::cout << "Phase 2: polling follower every 200ms for catch-up...\n";
+    auto t1 = Clock::now();
+    int lastSeen = -1;
+    int catchUpMs = 0;
+    bool caughtUp = false;
+
+    for (int attempt = 0; attempt < maxWaitSec * 5; ++attempt) {
+        int found = 0;
+        for (int i = 0; i < numDocs; ++i) {
+            try {
+                auto doc = follower.get("replica_test", "r" + std::to_string(i));
+                if (doc) ++found;
+            } catch (...) {
+                // transient — just means "not visible yet"
+            }
+        }
+        auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
+            Clock::now() - t1).count();
+        if (found != lastSeen) {
+            std::cout << "  [t=" << elapsed << "ms] follower has " << found
+                      << "/" << numDocs << " docs\n";
+            lastSeen = found;
+        }
+        if (found >= inserted) {
+            catchUpMs = static_cast<int>(elapsed);
+            caughtUp = true;
+            break;
+        }
+        std::this_thread::sleep_for(std::chrono::milliseconds(200));
+    }
+
+    if (!caughtUp) {
+        std::cout << "  did NOT fully catch up within " << maxWaitSec << "s\n";
+    }
+
+    // ===== Phase 3: final verification (content integrity) =====
+    std::cout << "\nPhase 3: content integrity verification on follower...\n";
+    int finalFound = 0;
+    int contentMatches = 0;
+    int contentMismatches = 0;
+    int sampleMismatches = 0;
+    for (int i = 0; i < numDocs; ++i) {
+        std::optional<nlohmann::json> doc;
+        try {
+            doc = follower.get("replica_test", "r" + std::to_string(i));
+        } catch (...) {
+            continue;
+        }
+        if (!doc) continue;
+        ++finalFound;
+        if (doc->contains("content") &&
+            (*doc)["content"].is_string() &&
+            (*doc)["content"].get<std::string>() == makeContent(i)) {
+            ++contentMatches;
+        } else {
+            ++contentMismatches;
+            if (sampleMismatches < 5) {
+                ++sampleMismatches;
+                std::cout << "  mismatch: id=r" << i
+                          << " body=" << doc->dump() << "\n";
+            }
+        }
+    }
+
+    std::cout << "\n=== Summary ===\n"
+              << "Inserted on leader:     " << inserted << "/" << numDocs << "\n"
+              << "Insert failures:        " << insertFailed << "\n"
+              << "Seen on follower:       " << finalFound << "\n"
+              << "Content matches:        " << contentMatches << "\n"
+              << "Content mismatches:     " << contentMismatches << "\n"
+              << "Catch-up time:          "
+              << (caughtUp ? (std::to_string(catchUpMs) + "ms") : std::string("NOT REACHED"))
+              << "\n\n";
+
+    if (finalFound == inserted && contentMismatches == 0 && caughtUp) {
+        std::cout << "Verdict: PASS — all " << inserted
+                  << " docs replicated in " << catchUpMs << "ms, zero corruption\n";
+        return 0;
+    } else if (finalFound > 0 && contentMismatches == 0) {
+        std::cout << "Verdict: PARTIAL — " << finalFound << "/" << inserted
+                  << " replicated, no corruption (replication functional but slow)\n";
+        return 2;
+    } else {
+        std::cout << "Verdict: FAIL — "
+                  << "finalFound=" << finalFound
+                  << " mismatches=" << contentMismatches << "\n";
+        return 1;
+    }
+}

tests/load_test/run_replication_test.sh

@@ -0,0 +1,98 @@
+#!/usr/bin/env bash
+# Replication sanity test orchestrator.
+#
+# 1) Wipe both replica data dirs, recreate them.
+# 2) Start leader (9005) and follower (9006) — both with replication enabled,
+#    each configured with the other as a peer.
+# 3) Verify both are listening.
+# 4) Run load_test_replication to insert 1000 docs on the leader and poll
+#    the follower until caught up. Propagate verdict.
+# 5) Always tail both logs for diagnosis.
+#
+# Usage:   ./run_replication_test.sh [num_docs] [max_wait_sec]
+# Exit:    0 PASS, 1 FAIL, 2 PARTIAL, non-0 orchestration error.
+
+set -euo pipefail
+
+cd "$(dirname "$0")"
+
+LDIR=/tmp/smartbotic-loadtest-replica-leader
+FDIR=/tmp/smartbotic-loadtest-replica-follower
+
+NUM_DOCS="${1:-1000}"
+MAX_WAIT_SEC="${2:-30}"
+
+echo "=== Phase 0: reset data dirs ==="
+rm -rf "$LDIR" "$FDIR"
+mkdir -p "$LDIR/data" "$LDIR/files" "$FDIR/data" "$FDIR/files"
+
+cleanup() {
+    if [[ -n "${LPID:-}" ]]; then kill "$LPID" 2>/dev/null || true; fi
+    if [[ -n "${FPID:-}" ]]; then kill "$FPID" 2>/dev/null || true; fi
+}
+trap cleanup EXIT
+
+echo ""
+echo "=== Phase 1: start leader on port 9005 ==="
+smartbotic-database --config ./config_replica_leader.json \
+    > "$LDIR/server.log" 2>&1 &
+LPID=$!
+sleep 2
+
+if ! kill -0 "$LPID" 2>/dev/null; then
+    echo "leader failed to start:"
+    tail -40 "$LDIR/server.log"
+    exit 1
+fi
+
+echo ""
+echo "=== Phase 2: start follower on port 9006 ==="
+smartbotic-database --config ./config_replica_follower.json \
+    > "$FDIR/server.log" 2>&1 &
+FPID=$!
+sleep 2
+
+if ! kill -0 "$FPID" 2>/dev/null; then
+    echo "follower failed to start:"
+    tail -40 "$FDIR/server.log"
+    exit 1
+fi
+
+# Sanity: both listening?
+if ! ss -tlnp 2>/dev/null | grep -q ":9005 "; then
+    echo "leader not listening on :9005"
+    tail -40 "$LDIR/server.log"
+    exit 1
+fi
+if ! ss -tlnp 2>/dev/null | grep -q ":9006 "; then
+    echo "follower not listening on :9006"
+    tail -40 "$FDIR/server.log"
+    exit 1
+fi
+
+# Give the SyncProtocol's connection loop one reconnect tick (default 5s)
+# to establish bidirectional peering before we start pounding the leader.
+echo "waiting 6s for peer handshake..."
+sleep 6
+
+echo ""
+echo "=== Phase 3: run load_test_replication ==="
+set +e
+./load_test_replication "$NUM_DOCS" "$MAX_WAIT_SEC"
+VERDICT=$?
+set -e
+
+echo ""
+echo "--- leader log tail (last 20 lines) ---"
+tail -20 "$LDIR/server.log"
+echo ""
+echo "--- follower log tail (last 20 lines) ---"
+tail -20 "$FDIR/server.log"
+
+echo ""
+case "$VERDICT" in
+    0) echo "=== Orchestrator: PASS ===" ;;
+    2) echo "=== Orchestrator: PARTIAL (exit=$VERDICT) ===" ;;
+    *) echo "=== Orchestrator: FAIL (exit=$VERDICT) ===" ;;
+esac
+exit "$VERDICT"