From a6d014601d509a5b79cfd0d6c423c2e37959331b Mon Sep 17 00:00:00 2001 From: Kalzu Rekku Date: Tue, 27 Oct 2026 14:08:42 +0200 Subject: [PATCH] Support for bigger clusters. --- README.md | 47 ++++- index.html | 515 ++++++++++++++++++++++++++++++++++----------- repl-test.sh | 463 +++++++++++++++++++++++++++++----------- src/handlers.rs | 361 ++++++++++++++++++++++++++++++- src/main.rs | 59 +++--- src/replication.rs | 92 +++++--- src/state.rs | 117 +++++++++- src/tls.rs | 5 + 8 files changed, 1336 insertions(+), 323 deletions(-) diff --git a/README.md b/README.md index b54e4ac..5ac728f 100644 --- a/README.md +++ b/README.md @@ -18,12 +18,12 @@ Instances can run as a **primary** (accepts writes) or as a **read-only replica* - **`loose`**: All defined fields are required, but additional unknown fields are accepted. - **Hot-Reloadable Configuration**: Reload `config.json` via an HTTP admin endpoint with atomic swap and rollback on error. - **Persistent Storage**: Validated records are stored with a generated UUID, an ISO-8601 UTC timestamp and a global sequence number (`seq`) in an SQLite database (`messages.db`). -- **Read Replication**: +- **Replication & Dynamic Role Switching**: - Pull-based and cursor-driven: the replica asks for everything after its last `seq`. - Append-only journal with a global tick (`seq`) that is preserved on replicas. - - Permanent per-instance ID, used to detect wrong, cloned or restored sources. - - Atomic, idempotent batch apply: safe across crashes, retries and duplicate deliveries. - - Exponential backoff with jitter, small batches and gzip for flaky or slow links. + - Seamless role switch / failover without data loss or split-brain. + - Automatic CA credential sharing during dynamic join so promoted replicas can continue minting tokens and signing nodes. + - Coordinated 1-step switchover API (`/api/admin/switchover`) and fine-grained controls (`/step-down`, `/promote`, `/repoint`). - Replicas can serve replication to other replicas (relay). - **Structured Logging**: Built with `tracing` outputting to standard output. @@ -89,7 +89,8 @@ The schema is only enforced on writes to a primary. A replica still loads the fi | `APP_CONFIG` | `config.json` | Path to the schema file. | | `APP_BIND` | `127.0.0.1:3000` | Listen address. Change this for a primary that replicas must reach over a network. | | `APP_ROLE` | `primary` | `primary` or `replica`. | -| `APP_PRIMARY_URL` | *none* | Base URL of the source, e.g. `https://primary.example:3000`. **Required for a replica.** | +| `APP_PRIMARY_URL` | *none* | Base URL of the replication source, e.g. `https://primary.example:3000`. **Required for a replica.** | +| `APP_JOIN_URL` | `APP_PRIMARY_URL` | CA enrollment URL used during first-boot mTLS join. Defaults to `APP_PRIMARY_URL`. Set this when the replication source differs from the Cluster CA holder (e.g., chain topologies where an intermediate replica serves data but cannot issue certificates). | | `APP_REPL_TOKEN` | *none* | Shared bearer token for `/api/replication/changes`. **Required for a replica.** On a primary, replication stays disabled if it is unset. | | `APP_SYNC_INTERVAL_SECS` | `5` | Replica idle poll interval (jittered) once caught up. | | `APP_SYNC_MAX_BACKOFF_SECS` | `60` | Upper bound for retry backoff after failed pulls. | @@ -270,6 +271,40 @@ bash repl-test.sh --- +### Coordinated Graceful Switchover +- **Endpoint:** `POST /api/admin/switchover` +- **Description:** Initiated on the **active primary** to execute a coordinated, zero-downtime role switch with a replica. The primary quiesces (rejects new writes), waits for the replica to catch up to its exact `head_seq`, issues a promotion call to the replica, and reconfigures itself as a replica following the new primary. +- **Request Body:** + ```json + { + "replica_url": "https://127.0.0.1:3001", + "timeout_secs": 15 + } + ``` +- **Response (`200 OK`):** + ```json + { + "status": "ok", + "message": "Role switchover completed successfully", + "old_primary_role": "replica", + "new_primary_url": "https://127.0.0.1:3001", + "switched_at_seq": 42 + } + ``` + +--- + +### Manual Role Management Endpoints + +- **`POST /api/admin/step-down`**: + Demotes a primary to a read-only replica. Optionally takes `{"primary_url": "https://..."}` to immediately begin pulling from a new primary. +- **`POST /api/admin/promote`**: + Promotes a replica to a primary. Halts any active replication sync loop, increments cluster epoch, and begins accepting writes at `POST /api/records`. +- **`POST /api/admin/repoint`**: + Updates a replica's source URL (`{"primary_url": "https://...", "reset_source": true}`). Resets the source instance guard and restarts the sync loop. + +--- + ### Replication: Pull Changes - **Endpoint:** `GET /api/replication/changes?after=&limit=` - **Headers:** `Authorization: Bearer ` @@ -379,7 +414,7 @@ On a fatal mismatch the replica halts, because retrying cannot fix it, and the r - **Bootstrapping:** start a replica with an empty database. Don't seed it by copying the primary's `messages.db`, because the copy shares the primary's instance ID and the replica will refuse to sync. - **Transport security:** replication uses a bearer token over whatever transport you give it. Put TLS in front of the primary (reverse proxy) for anything beyond a trusted network. - **Chaining:** a replica serves `/api/replication/changes` too (if it has a token), so replicas can follow other replicas. Downstream replicas see the upstream replica's instance ID as their source. -- **Promotion:** a replica holds the same `seq` values as its source, so it can be restarted with `APP_ROLE=primary` to take over. Note that its instance ID differs from the old primary's, so other replicas will halt on an ID mismatch until they are reset against the new primary. +- **Promotion & Role Switching:** Replicas can be dynamically promoted at runtime without restart via `POST /api/admin/promote` or gracefully swapped via `POST /api/admin/switchover`. Nodes share Cluster CA credentials during initial join, allowing promoted primaries to issue join tokens and admit new nodes. Existing replicas can follow the new primary without resetting their databases via `POST /api/admin/repoint`. - **Updates and deletes:** the design assumes records are only ever inserted. If updates or deletes are added later, the journal needs separate change entries (with tombstones), since replicas only observe inserts. - **Fan-out:** polling is fine for a handful of replicas. For many, consider long-polling or SSE on `/changes`. diff --git a/index.html b/index.html index 95e4e03..36381e9 100644 --- a/index.html +++ b/index.html @@ -80,24 +80,36 @@ border: 1px dashed var(--border); border-radius: 8px; margin-bottom: 20px; + flex-wrap: wrap; + gap: 16px; } .node-box { text-align: center; - padding: 10px 20px; + padding: 12px 20px; border-radius: 6px; border: 1px solid var(--border); background: var(--card-bg); - min-width: 180px; + min-width: 200px; + flex: 1; + max-width: 320px; } .pipe { display: flex; flex-direction: column; align-items: center; - gap: 4px; + gap: 6px; font-size: 0.8rem; color: var(--text-muted); + min-width: 220px; + padding: 0 10px; + } + .arrow { + font-size: 1.4rem; + color: var(--accent); + letter-spacing: 2px; + font-family: monospace; + font-weight: bold; } - .arrow { font-size: 1.5rem; color: var(--accent); letter-spacing: 2px; } /* Inputs & Buttons */ input, textarea, select { @@ -123,11 +135,22 @@ font-weight: 500; transition: all 0.15s; } - button:hover { background: #30363d; border-color: #8b949e; color: #fff; } + button:hover:not(:disabled) { background: #30363d; border-color: #8b949e; color: #fff; } + button:disabled { opacity: 0.5; cursor: not-allowed; } button.btn-primary { background: #238636; border-color: #2ea043; color: #fff; } - button.btn-primary:hover { background: #2ea043; } + button.btn-primary:hover:not(:disabled) { background: #2ea043; } button.btn-test { background: #1f6feb; border-color: #388bfd; color: #fff; } - button.btn-test:hover { background: #388bfd; } + button.btn-test:hover:not(:disabled) { background: #388bfd; } + button.btn-switch { background: #8957e5; border-color: #a371f7; color: #fff; font-weight: 600; } + button.btn-switch:hover:not(:disabled) { background: #9a67ea; border-color: #b388ff; } + + .write-box { + background: rgba(0, 0, 0, 0.25); + border: 1px solid var(--border); + border-radius: 6px; + padding: 10px; + margin-bottom: 14px; + } /* Tables */ .table-container { max-height: 380px; overflow-y: auto; } @@ -160,7 +183,7 @@

⚡ Records API & Replication Monitor

- Interactive validation and real-time visualization of SQLite append-only journals + Interactive validation and real-time visualization of SQLite append-only journals & mTLS failover

@@ -175,12 +198,12 @@
- - + +
- - + +
@@ -196,50 +219,97 @@
-
- Primary Node -
Seq: --
-
ID: --
+ +
+
+ Node 1 (:3000) + PRIMARY +
+
Seq: --
+
ID: --
+
State: --
+
- Connected -
━━━━ ◄─pull── ━━━━
-
Lag: 0 records
+ CONNECTED +
━━━━ ◄─pull── ━━━━
+
Lag: 0 records
+
-
- Replica Node -
Seq: --
-
ID: --
-
State: --
+ +
+
+ Node 2 (:3001) + REPLICA +
+
Seq: --
+
ID: --
+
State: --
- +
- +
- 📖 Primary Records (`/api/records`) - 0 records +
+ 📖 Node 1 Records + PRIMARY +
+ 0 records
+ + +
+
+ ✍️ Insert Record (Node 1) + POST /api/records +
+ + +
+ +
- +
SeqUUIDDate (UTC)Payload
Loading...
- +
- 📖 Replica Records (`/api/records`) - 0 records +
+ 📖 Node 2 Records + REPLICA +
+ 0 records
+ + +
+
+ ✍️ Insert Record (Node 2) + POST /api/records +
+ + +
+ +
- +
SeqUUIDDate (UTC)Payload
Loading...
@@ -249,24 +319,24 @@
- -
-
✍️ Write Record (Primary)
- - - -
-
🧪 Automated Tests
+ - + +
+
+ + +
+
🎛️ Role Controls
+
+ + +
@@ -276,7 +346,7 @@ ⚙️ Schema Management
- 
Click Inspect to view
+ 
Click Inspect to view
@@ -315,8 +385,10 @@ return nativeFetch(input, init); }; - let primaryHeadSeq = 0; - let replicaHeadSeq = 0; + let node1HeadSeq = 0; + let node2HeadSeq = 0; + let currentNode1Role = 'unknown'; + let currentNode2Role = 'unknown'; function log(msg, type = 'info') { const box = document.getElementById('consoleLog'); @@ -328,74 +400,155 @@ } function getUrls() { + const n1 = document.getElementById('node1Url')?.value || 'https://127.0.0.1:3000'; + const n2 = document.getElementById('node2Url')?.value || 'https://127.0.0.1:3001'; return { - primary: document.getElementById('primaryUrl').value.replace(/\/+$/, ''), - replica: document.getElementById('replicaUrl').value.replace(/\/+$/, ''), - token: document.getElementById('replToken').value + node1: n1.replace(/\/+$/, ''), + node2: n2.replace(/\/+$/, ''), + token: document.getElementById('replToken')?.value || 'change-me' }; } async function fetchAll() { - const { primary, replica } = getUrls(); + const { node1, node2 } = getUrls(); - // 1. Fetch Primary Records + let node1Status = null; + let node2Status = null; + let node1Records = []; + let node2Records = []; + + // 1. Query Node 1 Status & Records try { - const res = await fetch(`${primary}/api/records`); - if (res.ok) { - const records = await res.json(); - renderTable('primaryTable', records); - document.getElementById('primaryCount').innerText = `${records.length} records`; - primaryHeadSeq = records.length > 0 ? records[0].seq : 0; - document.getElementById('primaryHead').innerText = `Seq: ${primaryHeadSeq}`; - } - } catch (e) { - document.getElementById('primaryHead').innerText = `Offline`; + const sRes = await fetch(`${node1}/api/replication/status`); + if (sRes.ok) node1Status = await sRes.json(); + } catch (_) {} + + try { + const rRes = await fetch(`${node1}/api/records`); + if (rRes.ok) node1Records = await rRes.json(); + } catch (_) {} + + // 2. Query Node 2 Status & Records + try { + const sRes = await fetch(`${node2}/api/replication/status`); + if (sRes.ok) node2Status = await sRes.json(); + } catch (_) {} + + try { + const rRes = await fetch(`${node2}/api/records`); + if (rRes.ok) node2Records = await rRes.json(); + } catch (_) {} + + // Update Node 1 UI elements + if (node1Status) { + currentNode1Role = node1Status.role || 'replica'; + node1HeadSeq = node1Status.local_head_seq ?? (node1Records.length > 0 ? node1Records[0].seq : 0); + document.getElementById('node1Head').innerText = `Seq: #${node1HeadSeq}`; + document.getElementById('node1Id').innerText = `ID: ${node1Status.instance_id.slice(0, 8)}…`; + document.getElementById('node1SyncState').innerText = `State: ${node1Status.sync?.state || 'idle'}`; + updateRoleBadge('node1RoleBadge', currentNode1Role); + updateRoleBadge('node1TableRole', currentNode1Role); + } else { + currentNode1Role = 'offline'; + document.getElementById('node1Head').innerText = `Offline`; + document.getElementById('node1Id').innerText = `ID: --`; + document.getElementById('node1SyncState').innerText = `State: --`; + updateRoleBadge('node1RoleBadge', 'offline'); + updateRoleBadge('node1TableRole', 'offline'); } + renderTable('node1Table', node1Records, Math.max(node1HeadSeq, node2HeadSeq)); + document.getElementById('node1Count').innerText = `${node1Records.length} records`; - // 2. Fetch Replica Replication Status - try { - const res = await fetch(`${replica}/api/replication/status`); - if (res.ok) { - const status = await res.json(); - replicaHeadSeq = status.local_head_seq ?? 0; - document.getElementById('replicaHead').innerText = `Seq: ${replicaHeadSeq}`; - document.getElementById('replicaId').innerText = `ID: ${status.instance_id.slice(0, 8)}...`; - document.getElementById('primaryId').innerText = `Source: ${(status.sync?.source_instance_id || 'unlinked').slice(0, 8)}...`; - - const state = status.sync?.state || 'unknown'; - const badge = document.getElementById('pipeState'); - badge.innerText = state.toUpperCase(); - badge.className = `badge ${state === 'idle' ? 'badge-ok' : state === 'syncing' ? 'badge-warn' : 'badge-err'}`; - - const lag = status.lag_records ?? Math.max(0, primaryHeadSeq - replicaHeadSeq); - document.getElementById('lagRecords').innerText = lag; - document.getElementById('replicaSyncState').innerText = `State: ${state}`; - } - } catch (e) { - document.getElementById('replicaHead').innerText = `Offline`; - document.getElementById('pipeState').innerText = 'DISCONNECTED'; - document.getElementById('pipeState').className = 'badge badge-err'; + // Update Node 2 UI elements + if (node2Status) { + currentNode2Role = node2Status.role || 'replica'; + node2HeadSeq = node2Status.local_head_seq ?? (node2Records.length > 0 ? node2Records[0].seq : 0); + document.getElementById('node2Head').innerText = `Seq: #${node2HeadSeq}`; + document.getElementById('node2Id').innerText = `ID: ${node2Status.instance_id.slice(0, 8)}…`; + document.getElementById('node2SyncState').innerText = `State: ${node2Status.sync?.state || 'idle'}`; + updateRoleBadge('node2RoleBadge', currentNode2Role); + updateRoleBadge('node2TableRole', currentNode2Role); + } else { + currentNode2Role = 'offline'; + document.getElementById('node2Head').innerText = `Offline`; + document.getElementById('node2Id').innerText = `ID: --`; + document.getElementById('node2SyncState').innerText = `State: --`; + updateRoleBadge('node2RoleBadge', 'offline'); + updateRoleBadge('node2TableRole', 'offline'); } + renderTable('node2Table', node2Records, Math.max(node1HeadSeq, node2HeadSeq)); + document.getElementById('node2Count').innerText = `${node2Records.length} records`; - // 3. Fetch Replica Records - try { - const res = await fetch(`${replica}/api/records`); - if (res.ok) { - const records = await res.json(); - renderTable('replicaTable', records); - document.getElementById('replicaCount').innerText = `${records.length} records`; - } - } catch (e) {} + // 3. Update the Live Replication Pipeline Banner dynamically + updatePipelineBanner(node1Status, node2Status); } - function renderTable(tableId, records) { + function updateRoleBadge(badgeId, role) { + const el = document.getElementById(badgeId); + if (!el) return; + if (role === 'primary') { + el.className = 'badge badge-primary'; + el.innerText = 'PRIMARY'; + } else if (role === 'replica') { + el.className = 'badge badge-replica'; + el.innerText = 'REPLICA'; + } else { + el.className = 'badge badge-err'; + el.innerText = 'OFFLINE'; + } + } + + function updatePipelineBanner(status1, status2) { + const arrowEl = document.getElementById('pipeArrow'); + const detailsEl = document.getElementById('pipeDetails'); + const stateBadge = document.getElementById('pipeState'); + const switchBtn = document.getElementById('switchoverBtn'); + + if (currentNode1Role === 'primary' && currentNode2Role === 'replica') { + // Node 2 pulls from Node 1 + arrowEl.innerText = '━━━━ ◄─pull── ━━━━'; + const lag = status2?.lag_records ?? Math.max(0, node1HeadSeq - node2HeadSeq); + detailsEl.innerHTML = `Node 2 pulling from Node 1 (Lag: ${lag} records)`; + const syncState = status2?.sync?.state || 'idle'; + stateBadge.innerText = syncState.toUpperCase(); + stateBadge.className = `badge ${syncState === 'idle' ? 'badge-ok' : syncState === 'syncing' ? 'badge-warn' : 'badge-err'}`; + switchBtn.innerText = '🔀 Switchover: Node 1 ➔ Node 2'; + switchBtn.disabled = false; + } else if (currentNode2Role === 'primary' && currentNode1Role === 'replica') { + // Node 1 pulls from Node 2 + arrowEl.innerText = '━━━━ ──pull─► ━━━━'; + const lag = status1?.lag_records ?? Math.max(0, node2HeadSeq - node1HeadSeq); + detailsEl.innerHTML = `Node 1 pulling from Node 2 (Lag: ${lag} records)`; + const syncState = status1?.sync?.state || 'idle'; + stateBadge.innerText = syncState.toUpperCase(); + stateBadge.className = `badge ${syncState === 'idle' ? 'badge-ok' : syncState === 'syncing' ? 'badge-warn' : 'badge-err'}`; + switchBtn.innerText = '🔀 Switchover: Node 2 ➔ Node 1'; + switchBtn.disabled = false; + } else if (currentNode1Role === 'primary' && currentNode2Role === 'primary') { + arrowEl.innerText = '⚠️ DUAL PRIMARY ⚠️'; + detailsEl.innerText = 'Warning: Split-brain condition detected!'; + stateBadge.innerText = 'SPLIT BRAIN'; + stateBadge.className = 'badge badge-err'; + switchBtn.innerText = '🔀 Resolve Split-Brain'; + switchBtn.disabled = false; + } else { + arrowEl.innerText = '━━━━ ✖✖✖ ━━━━'; + detailsEl.innerText = 'No active primary/replica replication stream'; + stateBadge.innerText = 'DISCONNECTED'; + stateBadge.className = 'badge badge-err'; + switchBtn.innerText = '🔀 Graceful Switchover'; + switchBtn.disabled = true; + } + } + + function renderTable(tableId, records, globalHead) { const tbody = document.querySelector(`#${tableId} tbody`); if (!records || records.length === 0) { tbody.innerHTML = `Database is empty`; return; } tbody.innerHTML = records.map(r => ` - + #${r.seq} ${r.uuid.slice(0, 8)}… ${r.date.replace('T', ' ').replace('Z', '')} @@ -404,18 +557,23 @@ `).join(''); } - async function submitRecord() { - const { primary } = getUrls(); + // Submit a record directly to a specific node + async function submitRecordToNode(whichNode) { + const { node1, node2 } = getUrls(); + const targetUrl = whichNode === 'node1' ? node1 : node2; + const nodeName = whichNode === 'node1' ? 'Node 1 (:3000)' : 'Node 2 (:3001)'; + const textarea = document.getElementById(`${whichNode}RecordData`); + let payload; try { - payload = JSON.parse(document.getElementById('recordData').value); + payload = JSON.parse(textarea.value); } catch (err) { - log(`Invalid JSON in editor: ${err.message}`, 'fail'); + log(`Invalid JSON in ${nodeName} write box: ${err.message}`, 'fail'); return; } try { - const res = await fetch(`${primary}/api/records`, { + const res = await fetch(`${targetUrl}/api/records`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ data: payload }) @@ -423,43 +581,140 @@ const text = await res.text(); if (res.status === 201) { const data = JSON.parse(text); - log(`Write successful on Primary! Created seq #${data.seq} (UUID: ${data.uuid.slice(0, 8)})`, 'pass'); + log(`Write SUCCESS on ${nodeName}! Created seq #${data.seq} (UUID: ${data.uuid.slice(0, 8)})`, 'pass'); fetchAll(); + } else if (res.status === 403) { + log(`${nodeName} BLOCKED write [HTTP 403 Forbidden] (Read-Only Replica): ${text}`, 'fail'); } else { - log(`Primary rejected write [HTTP ${res.status}]: ${text}`, 'fail'); + log(`${nodeName} returned [HTTP ${res.status}]: ${text}`, 'fail'); } } catch (e) { - log(`Network failure writing to Primary: ${e.message}`, 'fail'); + log(`Network failure writing to ${nodeName}: ${e.message}`, 'fail'); } } - // TEST 1: Replica read-only safety guard - async function runReplicaWriteGuardTest() { - const { replica } = getUrls(); - log(`Starting Replica Write Guard test against ${replica}/api/records...`, 'info'); + // Perform graceful switchover from the active Primary + async function performSwitchover() { + const { node1, node2 } = getUrls(); + let primaryUrl = null; + let replicaUrl = null; + let primaryName = ''; + let replicaName = ''; + + if (currentNode1Role === 'primary') { + primaryUrl = node1; + replicaUrl = node2; + primaryName = 'Node 1'; + replicaName = 'Node 2'; + } else if (currentNode2Role === 'primary') { + primaryUrl = node2; + replicaUrl = node1; + primaryName = 'Node 2'; + replicaName = 'Node 1'; + } else { + log('Switchover aborted: No active primary detected', 'fail'); + return; + } + + log(`Initiating graceful switchover: ${primaryName} ➔ ${replicaName}...`, 'info'); + const btn = document.getElementById('switchoverBtn'); + if (btn) btn.disabled = true; + try { - const res = await fetch(`${replica}/api/records`, { + const res = await fetch(`${primaryUrl}/api/admin/switchover`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ replica_url: replicaUrl, timeout_secs: 15 }) + }); + + const text = await res.text(); + if (res.ok) { + log(`PASS: Graceful switchover completed! ${replicaName} is now Primary; ${primaryName} is Replica.`, 'pass'); + await fetchAll(); + } else { + log(`FAIL: Switchover rejected [HTTP ${res.status}]: ${text}`, 'fail'); + } + } catch (e) { + log(`Switchover execution error: ${e.message}`, 'fail'); + } finally { + if (btn) btn.disabled = false; + } + } + + // Manual promotion helper + async function manualPromote(whichNode) { + const { node1, node2 } = getUrls(); + const targetUrl = whichNode === 'node1' ? node1 : node2; + const nodeName = whichNode === 'node1' ? 'Node 1' : 'Node 2'; + log(`Promoting ${nodeName} to Primary...`, 'info'); + try { + const res = await fetch(`${targetUrl}/api/admin/promote`, { method: 'POST' }); + const text = await res.text(); + log(`Promote result on ${nodeName} [HTTP ${res.status}]: ${text}`, res.ok ? 'pass' : 'fail'); + fetchAll(); + } catch (e) { + log(`Promote failed: ${e.message}`, 'fail'); + } + } + + // Manual step-down helper + async function manualStepDown() { + const { node1, node2 } = getUrls(); + const primaryUrl = (currentNode1Role === 'primary') ? node1 : (currentNode2Role === 'primary' ? node2 : null); + if (!primaryUrl) { + log('No active primary found to step down', 'fail'); + return; + } + log(`Stepping down active primary (${primaryUrl})...`, 'info'); + try { + const res = await fetch(`${primaryUrl}/api/admin/step-down`, { method: 'POST' }); + const text = await res.text(); + log(`Step-down result [HTTP ${res.status}]: ${text}`, res.ok ? 'pass' : 'fail'); + fetchAll(); + } catch (e) { + log(`Step-down failed: ${e.message}`, 'fail'); + } + } + + // TEST 1: Role Switchover automated test + async function runSwitchoverTest() { + log('Starting Automated Role Switchover verification test...', 'info'); + await performSwitchover(); + } + + // TEST 2: Replica read-only safety guard + async function runReplicaWriteGuardTest() { + const { node1, node2 } = getUrls(); + const target = (currentNode2Role === 'replica') ? node2 : (currentNode1Role === 'replica' ? node1 : node2); + const targetName = (target === node2) ? 'Node 2 (:3001)' : 'Node 1 (:3000)'; + + log(`Testing Write Guard against ${targetName} (${target}/api/records)...`, 'info'); + try { + const res = await fetch(`${target}/api/records`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ data: { message: "Direct write forbidden test" } }) }); if (res.status === 403) { const msg = await res.text(); - log(`PASS: Replica blocked write with HTTP 403 Forbidden! Response: "${msg}"`, 'pass'); + log(`PASS: ${targetName} blocked write with HTTP 403 Forbidden! Response: "${msg}"`, 'pass'); } else { - log(`FAIL: Replica answered with HTTP ${res.status} instead of expected 403!`, 'fail'); + log(`FAIL: ${targetName} answered with HTTP ${res.status} instead of expected 403!`, 'fail'); } } catch (e) { log(`Test failed due to network error: ${e.message}`, 'fail'); } } - // TEST 2: Schema validation test + // TEST 3: Schema validation test async function runSchemaValidationTest() { - const { primary } = getUrls(); - log(`Starting Schema Rejection test on Primary (sending wrong types & extra fields)...`, 'info'); + const { node1, node2 } = getUrls(); + const primaryUrl = (currentNode1Role === 'primary') ? node1 : (currentNode2Role === 'primary' ? node2 : node1); + const primaryName = (primaryUrl === node1) ? 'Node 1' : 'Node 2'; + + log(`Starting Schema Rejection test on ${primaryName} (sending invalid payload)...`, 'info'); try { - const res = await fetch(`${primary}/api/records`, { + const res = await fetch(`${primaryUrl}/api/records`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ data: { unknown_xyz_field: 999999, version: "not-an-integer" } }) @@ -475,15 +730,23 @@ } } - // TEST 3: Sync Latency propagation benchmark + // TEST 4: Sync Latency propagation benchmark async function runSyncLatencyTest() { - const { primary, replica } = getUrls(); + const { node1, node2 } = getUrls(); + const primaryUrl = (currentNode1Role === 'primary') ? node1 : (currentNode2Role === 'primary' ? node2 : null); + const replicaUrl = (currentNode1Role === 'primary') ? node2 : (currentNode2Role === 'primary' ? node1 : null); + + if (!primaryUrl || !replicaUrl) { + log('Latency test aborted: Cluster does not have a distinct Primary and Replica', 'fail'); + return; + } + const testMsg = `probe-${Date.now()}`; log(`Triggering probe write on Primary with marker: ${testMsg}...`, 'info'); const startTime = performance.now(); try { - const res = await fetch(`${primary}/api/records`, { + const res = await fetch(`${primaryUrl}/api/records`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ data: { message: testMsg, msg: testMsg, version: 1 } }) @@ -497,7 +760,7 @@ const interval = setInterval(async () => { attempts++; try { - const statusRes = await fetch(`${replica}/api/replication/status`); + const statusRes = await fetch(`${replicaUrl}/api/replication/status`); const status = await statusRes.json(); if (status.local_head_seq >= targetSeq) { clearInterval(interval); @@ -519,9 +782,10 @@ } async function fetchSchema() { - const { primary } = getUrls(); + const { node1, node2 } = getUrls(); + const primaryUrl = (currentNode1Role === 'primary') ? node1 : (currentNode2Role === 'primary' ? node2 : node1); try { - const res = await fetch(`${primary}/api/admin/config`); + const res = await fetch(`${primaryUrl}/api/admin/config`); const data = await res.json(); document.getElementById('schemaView').innerText = JSON.stringify(data, null, 2); log(`Fetched active configuration schema.`, 'info'); @@ -531,9 +795,10 @@ } async function reloadSchema() { - const { primary } = getUrls(); + const { node1, node2 } = getUrls(); + const primaryUrl = (currentNode1Role === 'primary') ? node1 : (currentNode2Role === 'primary' ? node2 : node1); try { - const res = await fetch(`${primary}/api/admin/reload-config`, { method: 'POST' }); + const res = await fetch(`${primaryUrl}/api/admin/reload-config`, { method: 'POST' }); const text = await res.text(); log(`Reload schema result [HTTP ${res.status}]: ${text}`, res.ok ? 'pass' : 'fail'); fetchSchema(); @@ -553,4 +818,4 @@ fetchAll(); - + \ No newline at end of file diff --git a/repl-test.sh b/repl-test.sh index 836cd73..15b0520 100755 --- a/repl-test.sh +++ b/repl-test.sh @@ -1,24 +1,69 @@ #!/usr/bin/env bash # ============================================================================== # repl-test.sh: Integration harness for Records API & Replication (mTLS) -# Starts Primary (:3000), Replica (:3001), and Dashboard (:8080). +# +# Supports N instances in multiple topologies: +# star - one primary, all replicas pull directly from it (default) +# chain - a linear chain: primary -> r1 -> r2 -> ... -> rN +# +# Environment variable overrides: +# NODES=4 number of total nodes (1 primary + N-1 replicas; default: 2) +# BASE_PORT=3000 base port; nodes occupy BASE_PORT .. BASE_PORT+NODES-1 +# TOPO=star|chain replication topology (default: star) +# TOKEN=... shared replication bearer token +# START_DASHBOARD=1 launch the web dashboard (default: 1) +# SKIP_SWITCHOVER=0 skip the switchover smoke test (default: 0) +# NO_WAIT=0 exit immediately after smoke tests instead of waiting +# +# Examples: +# bash repl-test.sh # 2 nodes, star, default ports +# NODES=3 bash repl-test.sh # 1 primary + 2 replicas, star +# NODES=4 TOPO=chain bash repl-test.sh # 4-node replication chain +# NODES=5 SKIP_SWITCHOVER=1 bash repl-test.sh # ============================================================================== set -euo pipefail -# --- Configuration & Overrides --- +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" BIN="${BIN:-$ROOT_DIR/target/debug/rust_test}" WORK="${WORK:-$ROOT_DIR/repl-test}" -PRIMARY_PORT="${PRIMARY_PORT:-3000}" -REPLICA_PORT="${REPLICA_PORT:-3001}" +BASE_PORT="${BASE_PORT:-3000}" +NODES="${NODES:-2}" # total node count (must be >= 2) +TOPO="${TOPO:-star}" # star | chain +TOKEN="${TOKEN:-change-me}" DASHBOARD_PORT="${DASHBOARD_PORT:-8080}" -TOKEN="${TOKEN:-change-me}" # Replication bearer token CORS_ORIGIN="${CORS_ORIGIN:-http://127.0.0.1:$DASHBOARD_PORT}" START_DASHBOARD="${START_DASHBOARD:-1}" +SKIP_SWITCHOVER="${SKIP_SWITCHOVER:-0}" +NO_WAIT="${NO_WAIT:-0}" + +if [[ "$NODES" -lt 2 ]]; then + echo "Error: NODES must be >= 2." >&2 + exit 1 +fi + +# --------------------------------------------------------------------------- +# Node naming and port helpers +# --------------------------------------------------------------------------- +# node_name 0 -> "primary" +# node_name 1 -> "replica-1" +# node_name 2 -> "replica-2" ... +node_name() { + local idx=$1 + if [[ $idx -eq 0 ]]; then echo "primary"; else echo "replica-$idx"; fi +} + +# node_port 0 -> BASE_PORT +# node_port 1 -> BASE_PORT+1 ... +node_port() { echo $(( BASE_PORT + $1 )); } PIDS=() -# --- Helper Utilities --- +# --------------------------------------------------------------------------- +# Helper utilities +# --------------------------------------------------------------------------- check_port() { local port=$1 if command -v lsof >/dev/null 2>&1; then @@ -43,38 +88,119 @@ cleanup() { } trap cleanup EXIT INT TERM -# --- Pre-flight Checks --- -for port in "$PRIMARY_PORT" "$REPLICA_PORT"; do +json_field() { + # Usage: json_field + local json="$1" field="$2" + if command -v jq >/dev/null 2>&1; then + echo "$json" | jq -r ".${field} // empty" + else + echo "$json" | grep -o "\"${field}\":\"[^\"]*" | head -1 | cut -d'"' -f4 + fi +} + +color_echo() { + local color=$1; shift + local code + case "$color" in + red) code='\033[1;31m' ;; + green) code='\033[1;32m' ;; + yellow) code='\033[1;33m' ;; + blue) code='\033[1;34m' ;; + *) code='\033[0m' ;; + esac + echo -e "${code}$*\033[0m" +} + +wait_up() { + local name=$1 port=$2 + for _ in $(seq 1 80); do + if curl -k -sf "https://127.0.0.1:$port/api/now" >/dev/null 2>&1; then + return 0 + fi + sleep 0.1 + done + color_echo red "Error: $name failed to start on :$port" >&2 + if [[ -f "$WORK/$name/log.txt" ]]; then + echo "--- Last 30 lines of $WORK/$name/log.txt ---" >&2 + tail -n 30 "$WORK/$name/log.txt" >&2 + fi + exit 1 +} + +mint_join_token() { + local primary_port=$1 + local resp + resp=$(curl -k -sf -X POST "https://127.0.0.1:$primary_port/api/cluster/tokens" \ + -H 'Content-Type: application/json' \ + -d '{"ttl_secs": 300}') + json_field "$resp" "token" +} + +wait_replica_caught_up() { + local name=$1 port=$2 target_seq=$3 + for _ in $(seq 1 80); do + local out + out=$(curl -k -s "https://127.0.0.1:$port/api/replication/status" 2>/dev/null || true) + local applied + applied=$(json_field "$out" "sync.last_applied_seq" 2>/dev/null || echo "0") + # Fallback grep for nested .sync.last_applied_seq + if [[ -z "$applied" || "$applied" == "null" ]]; then + applied=$(echo "$out" | grep -o '"last_applied_seq":[0-9]*' | head -1 | cut -d: -f2) + fi + if [[ "${applied:-0}" -ge "$target_seq" ]]; then + return 0 + fi + sleep 0.25 + done + color_echo yellow " ⚠ WARNING: $name (:$port) did not reach seq $target_seq in time." >&2 + return 1 +} + +# --------------------------------------------------------------------------- +# Pre-flight +# --------------------------------------------------------------------------- +for idx in $(seq 0 $(( NODES - 1 ))); do + port=$(node_port "$idx") if check_port "$port"; then - echo "Error: Port $port is already in use by another process." >&2 + color_echo red "Error: Port $port is already in use." >&2 exit 1 fi done - -if [[ -f "$ROOT_DIR/Cargo.toml" ]]; then - echo "Ensuring debug binary is up to date with cargo..." - ( cd "$ROOT_DIR" && cargo build ) -elif [[ ! -x "$BIN" ]]; then - echo "Error: Binary not found or not executable: $BIN" >&2 - exit 1 +if [[ "$START_DASHBOARD" -eq 1 ]]; then + if check_port "$DASHBOARD_PORT"; then + color_echo yellow "Dashboard port :$DASHBOARD_PORT is in use; dashboard will be skipped." + START_DASHBOARD=0 + fi fi -# --- Workspace & Schema Setup --- -rm -rf "$WORK" -mkdir -p "$WORK/primary" "$WORK/replica" +# Build +if [[ -f "$ROOT_DIR/Cargo.toml" ]]; then + echo "Ensuring debug binary is up to date..." + ( cd "$ROOT_DIR" && cargo build ) +elif [[ ! -x "$BIN" ]]; then + color_echo red "Error: Binary not found: $BIN" >&2; exit 1 +fi -# Schema allows "msg" and "message", requiring integer "version" -for d in primary replica; do - cat > "$WORK/$d/config.json" <<'EOF' +# --------------------------------------------------------------------------- +# Workspace setup +# --------------------------------------------------------------------------- +rm -rf "$WORK" +for idx in $(seq 0 $(( NODES - 1 ))); do + name=$(node_name "$idx") + mkdir -p "$WORK/$name" + cat > "$WORK/$name/config.json" <<'CFGEOF' { "mode": "loose", "fields": { "version": { "type": "integer" } } } -EOF +CFGEOF done +# --------------------------------------------------------------------------- +# start_node [extra env vars...] +# --------------------------------------------------------------------------- start_node() { local name=$1 port=$2 shift 2 @@ -88,142 +214,235 @@ start_node() { PIDS+=("$!") } -wait_up() { - local name=$1 port=$2 - for _ in $(seq 1 60); do - # Note: -k used since cluster cert is generated on boot - if curl -k -sf "https://127.0.0.1:$port/api/now" >/dev/null 2>&1; then - return 0 - fi - sleep 0.1 - done - echo -e "\n\033[1;31mError: $name failed to start on :$port\033[0m" >&2 - if [[ -f "$WORK/$name/log.txt" ]]; then - echo "--- Last 20 lines of $WORK/$name/log.txt ---" >&2 - tail -n 20 "$WORK/$name/log.txt" >&2 - fi - exit 1 -} - # ============================================================================== -# 1. Launch Primary (Cluster CA & Server Cert generated automatically) +# 1. Start Primary # ============================================================================== -echo -e "\033[1;32m==> Starting Primary on :$PRIMARY_PORT (HTTPS/mTLS)...\033[0m" +PRIMARY_PORT=$(node_port 0) +color_echo green "==> [1/N] Starting Primary on :$PRIMARY_PORT (HTTPS/mTLS)..." start_node primary "$PRIMARY_PORT" APP_ROLE=primary wait_up "primary" "$PRIMARY_PORT" +color_echo green " ✔ Primary is up." # ============================================================================== -# 2. Test Dynamic Discovery: Retrieve Cluster Public CA +# 2. Retrieve CA cert (sanity check) # ============================================================================== -echo -e "\n\033[1;34m==> Testing dynamic discovery: Fetching Cluster Public CA...\033[0m" +color_echo blue "\n==> Verifying Cluster CA is accessible..." ca_resp=$(curl -k -sf "https://127.0.0.1:$PRIMARY_PORT/api/cluster/ca") if echo "$ca_resp" | grep -q 'BEGIN CERTIFICATE'; then - echo -e " \033[1;32m✔ Cluster CA certificate downloaded successfully.\033[0m" + color_echo green " ✔ Cluster CA certificate downloaded." else - echo -e " \033[1;31m✖ Failed to retrieve Cluster CA.\033[0m" - exit 1 + color_echo red " ✖ Failed to retrieve Cluster CA – aborting." >&2; exit 1 fi # ============================================================================== -# 3. Mint Short-Lived Join Token via Primary +# 3. Start replicas according to topology +# +# star topology: every replica pulls directly from the primary +# chain topology: replica-N pulls from replica-(N-1), which pulls from primary # ============================================================================== -echo -e "\n\033[1;34m==> Minting short-lived Join Token from Primary...\033[0m" -token_resp=$(curl -k -sf -X POST "https://127.0.0.1:$PRIMARY_PORT/api/cluster/tokens" \ - -H 'Content-Type: application/json' \ - -d '{"ttl_secs": 120}') -if command -v jq >/dev/null 2>&1; then - JOIN_TOKEN=$(echo "$token_resp" | jq -r '.token') -else - JOIN_TOKEN=$(echo "$token_resp" | grep -o '"token":"[^"]*' | cut -d'"' -f4) -fi +# Arrays for later use +PORTS=() # PORTS[i] = port for node i +NAMES=() # NAMES[i] = name for node i +SOURCES=() # SOURCES[i] = upstream URL for node i (empty for primary) -echo -e " \033[1;32m✔ Obtained one-time join token: $JOIN_TOKEN\033[0m" +PORTS[0]=$PRIMARY_PORT +NAMES[0]="primary" +SOURCES[0]="" -# ============================================================================== -# 4. Launch Replica (Bootstraps mTLS identity using the Join Token) -# ============================================================================== -echo -e "\n\033[1;32m==> Starting Replica on :$REPLICA_PORT (Dynamic Join Enrollment)...\033[0m" -start_node replica "$REPLICA_PORT" \ - APP_ROLE=replica \ - APP_PRIMARY_URL="https://127.0.0.1:$PRIMARY_PORT" \ - APP_JOIN_TOKEN="$JOIN_TOKEN" -wait_up "replica" "$REPLICA_PORT" +for idx in $(seq 1 $(( NODES - 1 ))); do + name=$(node_name "$idx") + port=$(node_port "$idx") + PORTS[$idx]=$port + NAMES[$idx]="$name" -# Verify credentials were saved to replica's SQLite database -if [[ -f "$WORK/replica/messages.db" ]]; then - echo -e " \033[1;32m✔ Replica enrolled and saved mTLS credentials in SQLite.\033[0m" -fi + case "$TOPO" in + chain) + # Replica pulls from the previous node in the chain + upstream_port=$(node_port $(( idx - 1 ))) + upstream_url="https://127.0.0.1:$upstream_port" + upstream_name=$(node_name $(( idx - 1 ))) + ;; + star|*) + # Replica pulls directly from the primary + upstream_port=$PRIMARY_PORT + upstream_url="https://127.0.0.1:$PRIMARY_PORT" + upstream_name="primary" + ;; + esac -# --- Launch Dashboard (if available) --- -if [[ "$START_DASHBOARD" -eq 1 && -f "$ROOT_DIR/dashboard.py" ]]; then - if check_port "$DASHBOARD_PORT"; then - echo -e "\033[1;33mDashboard port :$DASHBOARD_PORT is in use. Skipping dashboard.\033[0m" - else - echo -e "\033[1;32m==> Starting Dashboard on :$DASHBOARD_PORT\033[0m" - ( cd "$ROOT_DIR" && exec python3 dashboard.py ) >"$WORK/dashboard.log" 2>&1 & - PIDS+=("$!") + SOURCES[$idx]="$upstream_url" + + # The join token must come from the Cluster CA holder (always the primary) + color_echo blue "\n==> Minting join token for $name from primary..." + JOIN_TOKEN=$(mint_join_token "$PRIMARY_PORT") + color_echo green " ✔ Token: $JOIN_TOKEN" + + color_echo green "\n==> [$idx/$(( NODES - 1 ))] Starting $name on :$port" + echo " Topology: $TOPO | upstream: $upstream_name ($upstream_url)" + + # In chain topology the replication upstream differs from the CA issuer (primary). + # APP_JOIN_URL overrides enrollment so the new replica can get a cert from the + # actual primary even though it pulls data from an intermediate relay node. + start_node "$name" "$port" \ + APP_ROLE=replica \ + APP_PRIMARY_URL="$upstream_url" \ + APP_JOIN_URL="https://127.0.0.1:$PRIMARY_PORT" \ + APP_JOIN_TOKEN="$JOIN_TOKEN" + wait_up "$name" "$port" + color_echo green " ✔ $name is up and enrolled." + + if [[ -f "$WORK/$name/messages.db" ]]; then + color_echo green " ✔ mTLS credentials persisted in SQLite." fi +done + +# --------------------------------------------------------------------------- +# Dashboard +# --------------------------------------------------------------------------- +if [[ "$START_DASHBOARD" -eq 1 && -f "$ROOT_DIR/dashboard.py" ]]; then + color_echo green "\n==> Starting Dashboard on :$DASHBOARD_PORT..." + ( cd "$ROOT_DIR" && exec python3 dashboard.py ) >"$WORK/dashboard.log" 2>&1 & + PIDS+=("$!") fi # ============================================================================== -# 5. Write Smoke Test Records to Primary +# 4. Write smoke-test records to primary # ============================================================================== -echo -e "\n\033[1;34m==> Writing 3 test records to Primary via HTTPS\033[0m" -for i in 1 2 3; do +RECORD_COUNT=3 +color_echo blue "\n==> Writing $RECORD_COUNT test records to Primary..." +for i in $(seq 1 "$RECORD_COUNT"); do curl -k -s -X POST "https://127.0.0.1:$PRIMARY_PORT/api/records" \ -H 'Content-Type: application/json' \ -d "{\"data\":{\"version\":1,\"msg\":\"hello $i\"}}" >/dev/null - echo " Record #$i written" + echo " Record #$i written" done # ============================================================================== -# 6. Await Replica mTLS Sync +# 5. Verify every replica has converged # ============================================================================== -echo -e "\n\033[1;34m==> Verifying replica mTLS sync stream...\033[0m" -caught_up=0 -for _ in $(seq 1 40); do - status_output=$(curl -k -s "https://127.0.0.1:$REPLICA_PORT/api/replication/status" || true) - if echo "$status_output" | grep -q '"last_applied_seq":3'; then - caught_up=1 - break +color_echo blue "\n==> Waiting for all replicas to converge to seq $RECORD_COUNT..." +all_ok=1 +for idx in $(seq 1 $(( NODES - 1 ))); do + name="${NAMES[$idx]}" + port="${PORTS[$idx]}" + echo -n " Checking $name (:$port)... " + if wait_replica_caught_up "$name" "$port" "$RECORD_COUNT"; then + color_echo green "✔ seq $RECORD_COUNT reached" + else + echo "✖ timeout" + all_ok=0 + fi +done +if [[ $all_ok -eq 1 ]]; then + color_echo green "✔ All replicas are in sync." +else + color_echo yellow "⚠ Some replicas did not converge in time." +fi + +# ============================================================================== +# 6. Read-only guard check on every replica +# ============================================================================== +color_echo blue "\n==> Verifying read-only guard on all replicas..." +for idx in $(seq 1 $(( NODES - 1 ))); do + name="${NAMES[$idx]}" + port="${PORTS[$idx]}" + http_code=$(curl -k -s -o /dev/null -w "%{http_code}" \ + -X POST "https://127.0.0.1:$port/api/records" \ + -H 'Content-Type: application/json' \ + -d '{"data":{"version":1,"msg":"should be rejected"}}') + if [[ "$http_code" == "403" ]]; then + color_echo green " ✔ $name (:$port) → HTTP 403 Forbidden (correct)" + else + color_echo red " ✖ $name (:$port) → HTTP $http_code (expected 403)" fi - sleep 0.25 done -if [[ $caught_up -eq 1 ]]; then - echo -e "\033[1;32m✔ SUCCESS: Replica caught up to seq 3 over mTLS!\033[0m" -else - echo -e "\033[1;31m✖ WARNING: Replica did not reach seq 3 in time.\033[0m" - echo "Replica Status Output: $status_output" +# ============================================================================== +# 7. Graceful switchover (primary -> first replica) +# Skipped if SKIP_SWITCHOVER=1 or only 1 replica exists (would leave no primary) +# ============================================================================== +if [[ "$SKIP_SWITCHOVER" -eq 0 && "$NODES" -ge 2 ]]; then + FIRST_REPLICA_PORT="${PORTS[1]}" + FIRST_REPLICA_NAME="${NAMES[1]}" + + color_echo blue "\n==> Initiating Graceful Switchover: primary (:$PRIMARY_PORT) ─► $FIRST_REPLICA_NAME (:$FIRST_REPLICA_PORT)..." + switch_resp=$(curl -k -sf -X POST "https://127.0.0.1:$PRIMARY_PORT/api/admin/switchover" \ + -H 'Content-Type: application/json' \ + -d "{\"replica_url\":\"https://127.0.0.1:$FIRST_REPLICA_PORT\",\"timeout_secs\":15}" || true) + color_echo green " ✔ Switchover response: $switch_resp" + + # Old primary is now replica – verify 403 + http_code=$(curl -k -s -o /dev/null -w "%{http_code}" \ + -X POST "https://127.0.0.1:$PRIMARY_PORT/api/records" \ + -H 'Content-Type: application/json' \ + -d '{"data":{"version":1,"msg":"should fail on demoted primary"}}') + if [[ "$http_code" == "403" ]]; then + color_echo green " ✔ Demoted primary returned HTTP 403 (read-only)" + else + color_echo red " ✖ Demoted primary returned HTTP $http_code (expected 403)" + fi + + # Write to new primary + NEXT_SEQ=$(( RECORD_COUNT + 1 )) + color_echo blue "\n==> Writing record #$NEXT_SEQ to new primary ($FIRST_REPLICA_NAME :$FIRST_REPLICA_PORT)..." + write_resp=$(curl -k -sf -X POST "https://127.0.0.1:$FIRST_REPLICA_PORT/api/records" \ + -H 'Content-Type: application/json' \ + -d "{\"data\":{\"version\":1,\"msg\":\"post-switchover write\"}}" || true) + color_echo green " ✔ Created: $write_resp" + + # After switchover in star topology, the other replicas still point to the old primary + # which now points to the new primary – they will converge via chaining. + # In chain topology they chain through naturally. + # Verify the demoted primary (now replica) catches up: + color_echo blue "\n==> Verifying all nodes converge to seq $NEXT_SEQ post-switchover..." + all_ok=1 + for idx in $(seq 0 $(( NODES - 1 ))); do + port="${PORTS[$idx]}" + name="${NAMES[$idx]}" + if [[ $idx -eq 1 ]]; then continue; fi # new primary already has it + echo -n " Checking $name (:$port)... " + if wait_replica_caught_up "$name" "$port" "$NEXT_SEQ"; then + color_echo green "✔ seq $NEXT_SEQ reached" + else + echo "✖ timeout" + all_ok=0 + fi + done + if [[ $all_ok -eq 1 ]]; then + color_echo green "✔ All nodes converged after switchover." + else + color_echo yellow "⚠ Some nodes did not converge after switchover." + fi fi # ============================================================================== -# 7. Verify Read-Only Safety Guard on Replica +# Ready Banner # ============================================================================== -echo -e "\n\033[1;34m==> Testing Replica read-only safety guard:\033[0m" -http_code=$(curl -k -s -o /dev/null -w "%{http_code}" -X POST "https://127.0.0.1:$REPLICA_PORT/api/records" \ - -H 'Content-Type: application/json' \ - -d '{"data":{"version":1,"msg":"should fail"}}') - -if [[ "$http_code" == "403" ]]; then - echo -e " \033[1;32m✔ HTTP 403 Forbidden received as expected.\033[0m" -else - echo -e " \033[1;31m✖ Expected HTTP 403, got $http_code.\033[0m" +echo "" +echo "======================================================================" +echo " Cluster is healthy • Topology: $TOPO • Nodes: $NODES" +echo "" +echo " Nodes:" +for idx in $(seq 0 $(( NODES - 1 ))); do + port="${PORTS[$idx]}" + name="${NAMES[$idx]}" + if [[ $idx -eq 0 ]]; then + echo " [$(( idx + 1 ))] Primary (:$port) https://127.0.0.1:$port" + else + src="${SOURCES[$idx]}" + echo " [$(( idx + 1 ))] Replica (:$port) https://127.0.0.1:$port <─ $src" + fi +done +echo "" +if [[ "$START_DASHBOARD" -eq 1 ]]; then + echo " Web Dashboard: http://127.0.0.1:$DASHBOARD_PORT" fi +echo " Logs Dir: $WORK/" +echo "" +echo " Press Ctrl+C to stop all instances." +echo "======================================================================" -# --- Ready Banner --- -cat <, payload: Option>, ) -> Result, ApiError> { - if state.role == Role::Replica { + if state.is_replica() { return Err(api_error(StatusCode::FORBIDDEN, "Only Primary can generate join tokens")); } @@ -61,7 +69,7 @@ pub async fn join_cluster_handler( State(state): State, Json(req): Json, ) -> Result, ApiError> { - if state.role == Role::Replica { + if state.is_replica() { return Err(api_error(StatusCode::FORBIDDEN, "Only Primary can issue certificates")); } @@ -71,7 +79,9 @@ pub async fn join_cluster_handler( .ok_or_else(|| internal_error("CA cert unavailable"))?; let ca_key_pem = state .ca_key_pem - .as_ref() + .read() + .map_err(|_| internal_error("CA key lock poisoned"))? + .clone() .ok_or_else(|| internal_error("CA key unavailable"))?; // 1. Validate & burn the token @@ -106,7 +116,7 @@ pub async fn join_cluster_handler( let (cert_pem, key_pem) = issue_leaf_cert( ca_cert_pem, - ca_key_pem, + &ca_key_pem, &req.node_id.to_string(), sans, ) @@ -118,6 +128,7 @@ pub async fn join_cluster_handler( ca_pem: ca_cert_pem.to_string(), cert_pem, key_pem, + ca_key_pem: Some(ca_key_pem), })) } @@ -141,12 +152,13 @@ pub async fn create_record( ) -> Result<(StatusCode, Json), ApiError> { info!(endpoint = "/api/records", "Create request received"); - if state.role == Role::Replica { + if state.is_replica() { + let p_url = state.get_primary_url(); return Err(api_error( StatusCode::FORBIDDEN, format!( "This instance is a read-only replica; write to {}", - state.primary_url.as_deref().map(String::as_str).unwrap_or("the primary") + p_url.as_deref().unwrap_or("the primary") ), )); } @@ -285,3 +297,338 @@ pub async fn reload_config(State(state): State) -> Result, "fields": field_count }))) } + +#[derive(Debug, Deserialize, Default)] +pub struct StepDownRequest { + pub primary_url: Option, +} + +#[derive(Debug, Deserialize)] +pub struct RepointRequest { + pub primary_url: String, + pub reset_source: Option, +} + +#[derive(Debug, Deserialize)] +pub struct SwitchoverRequest { + pub replica_url: String, + pub timeout_secs: Option, +} + +/// Admin endpoint: Demote this primary to read-only replica. +pub async fn step_down_handler( + State(state): State, + payload: Option>, +) -> Result, ApiError> { + info!("Step-down requested"); + + // 1. Update role in memory + { + let mut r = state + .role + .write() + .map_err(|_| internal_error("role lock poisoned"))?; + *r = Role::Replica; + } + + // 2. Stop any existing replica loop + state.stop_replica_sync(); + + // 3. Update local head seq as baseline for last_applied_seq + let local_head: i64 = { + let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?; + db.query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| { + r.get(0) + }) + .map_err(|e| db_error("get local head", e))? + }; + + { + let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?; + meta_set(&db, "role", "replica").map_err(|e| db_error("save role", e))?; + meta_set(&db, "last_applied_seq", &local_head.to_string()) + .map_err(|e| db_error("save last_applied_seq", e))?; + let _ = db.execute("DELETE FROM meta WHERE key = 'source_instance_id'", []); + } + + if let Ok(mut s) = state.sync.write() { + s.state = "idle".into(); + s.last_applied_seq = local_head; + s.source_instance_id = None; + s.primary_head_seq = Some(local_head); + s.last_error = None; + } + + if let Some(Json(req)) = payload { + if let Some(p_url) = req.primary_url { + let trimmed = p_url.trim_end_matches('/').to_string(); + if !trimmed.is_empty() { + state.set_primary_url(Some(trimmed.clone())); + let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?; + let _ = meta_set(&db, "primary_url", &trimmed); + drop(db); + state.start_replica_sync(); + } + } + } + + info!(role = "replica", last_applied_seq = local_head, "Node stepped down to replica"); + Ok(Json(serde_json::json!({ + "status": "ok", + "role": "replica", + "primary_url": state.get_primary_url(), + "last_applied_seq": local_head, + }))) +} + +/// Admin endpoint: Promote this replica to Primary. +pub async fn promote_handler( + State(state): State, +) -> Result, ApiError> { + info!("Promotion to Primary requested"); + + // 1. Stop pulling changes + state.stop_replica_sync(); + + // 2. Set role to Primary + { + let mut r = state + .role + .write() + .map_err(|_| internal_error("role lock poisoned"))?; + *r = Role::Primary; + } + + let local_head: i64 = { + let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?; + db.query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| { + r.get(0) + }) + .map_err(|e| db_error("get local head", e))? + }; + + let epoch = { + let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?; + meta_set(&db, "role", "primary").map_err(|e| db_error("save role", e))?; + let cur_epoch: i64 = meta_get(&db, "cluster_epoch") + .ok() + .flatten() + .and_then(|v| v.parse().ok()) + .unwrap_or(1); + let next_epoch = cur_epoch + 1; + meta_set(&db, "cluster_epoch", &next_epoch.to_string()) + .map_err(|e| db_error("save epoch", e))?; + next_epoch + }; + + if let Ok(mut s) = state.sync.write() { + s.state = "primary".into(); + s.primary_head_seq = None; + s.last_error = None; + } + + info!(role = "primary", head_seq = local_head, epoch, "Node successfully promoted to Primary"); + Ok(Json(serde_json::json!({ + "status": "ok", + "role": "primary", + "instance_id": state.instance_id, + "head_seq": local_head, + "epoch": epoch, + }))) +} + +/// Admin endpoint: Repoint replica to a new primary URL. +pub async fn repoint_handler( + State(state): State, + Json(req): Json, +) -> Result, ApiError> { + info!(primary_url = %req.primary_url, "Repoint requested"); + + let p_url = req.primary_url.trim_end_matches('/').to_string(); + state.set_primary_url(Some(p_url.clone())); + + { + let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?; + meta_set(&db, "primary_url", &p_url).map_err(|e| db_error("save primary_url", e))?; + if req.reset_source.unwrap_or(true) { + let _ = db.execute("DELETE FROM meta WHERE key = 'source_instance_id'", []); + } + } + + if req.reset_source.unwrap_or(true) { + if let Ok(mut s) = state.sync.write() { + s.source_instance_id = None; + s.last_error = None; + } + } + + if state.is_replica() { + state.start_replica_sync(); + } + + Ok(Json(serde_json::json!({ + "status": "ok", + "role": state.get_role().as_str(), + "primary_url": p_url, + }))) +} + +/// Admin endpoint: Initiates coordinated graceful switchover from Primary to a chosen Replica. +pub async fn switchover_handler( + State(state): State, + Json(req): Json, +) -> Result, ApiError> { + info!(target = %req.replica_url, "Coordinated cluster role switchover initiated"); + + if !state.is_primary() { + return Err(api_error( + StatusCode::BAD_REQUEST, + "Switchover can only be initiated on the primary node", + )); + } + + let replica_url = req.replica_url.trim_end_matches('/').to_string(); + let timeout_secs = req.timeout_secs.unwrap_or(15); + + // Build mTLS client to interact with the replica + let ca_pem = state + .ca_cert_pem + .as_ref() + .ok_or_else(|| internal_error("ca_cert_pem unavailable"))?; + let cert_pem = state + .node_cert_pem + .as_ref() + .ok_or_else(|| internal_error("node_cert_pem unavailable"))?; + let key_pem = state + .node_key_pem + .as_ref() + .ok_or_else(|| internal_error("node_key_pem unavailable"))?; + + let client = create_mtls_client(ca_pem, cert_pem, key_pem) + .map_err(|e| internal_error(&format!("failed to create mtls client: {e}")))?; + + // Step 1: Quiesce this primary by demoting to replica (stops accepting new writes) + { + let mut r = state + .role + .write() + .map_err(|_| internal_error("role lock poisoned"))?; + *r = Role::Replica; + } + + let local_head: i64 = { + let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?; + db.query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| { + r.get(0) + }) + .map_err(|e| db_error("get local head", e))? + }; + + // Step 2: Poll replica status until last_applied_seq >= local_head + info!(local_head, "Waiting for replica to catch up to head_seq..."); + let start = std::time::Instant::now(); + let mut caught_up = false; + + while start.elapsed() < Duration::from_secs(timeout_secs) { + let status_url = format!("{replica_url}/api/replication/status"); + match client.get(&status_url).send().await { + Ok(resp) if resp.status().is_success() => { + if let Ok(json) = resp.json::().await { + let repl_applied = json + .get("sync") + .and_then(|s| s.get("last_applied_seq")) + .and_then(|v| v.as_i64()) + .unwrap_or(0); + + if repl_applied >= local_head { + caught_up = true; + break; + } + } + } + Ok(resp) => { + warn!(status = %resp.status(), "Replica returned non-success during status check"); + } + Err(e) => { + warn!(error = %e, "Failed to query replica status during switchover"); + } + } + tokio::time::sleep(Duration::from_millis(200)).await; + } + + if !caught_up { + // Rollback to primary + error!("Replica did not catch up within timeout; rolling back role to Primary"); + let mut r = state + .role + .write() + .map_err(|_| internal_error("role lock poisoned"))?; + *r = Role::Primary; + return Err(api_error( + StatusCode::GATEWAY_TIMEOUT, + format!( + "Replica at {replica_url} failed to catch up to seq {local_head} within {timeout_secs}s" + ), + )); + } + + // Step 3: Promote replica to Primary + let promote_url = format!("{replica_url}/api/admin/promote"); + let resp = match client.post(&promote_url).send().await { + Ok(r) => r, + Err(e) => { + error!(error = %e, "Failed to call promote on replica; rolling back"); + let mut r = state + .role + .write() + .map_err(|_| internal_error("role lock poisoned"))?; + *r = Role::Primary; + return Err(api_error( + StatusCode::BAD_GATEWAY, + format!("Failed to reach replica for promotion: {e}"), + )); + } + }; + + if !resp.status().is_success() { + error!(status = %resp.status(), "Replica promotion failed; rolling back"); + let mut r = state + .role + .write() + .map_err(|_| internal_error("role lock poisoned"))?; + *r = Role::Primary; + return Err(api_error( + StatusCode::BAD_GATEWAY, + format!("Replica rejected promotion with status {}", resp.status()), + )); + } + + // Step 4: Configure this node as a replica of the new primary + { + let db = state.db.lock().map_err(|_| internal_error("db poisoned"))?; + let _ = meta_set(&db, "role", "replica"); + let _ = meta_set(&db, "primary_url", &replica_url); + let _ = meta_set(&db, "last_applied_seq", &local_head.to_string()); + let _ = db.execute("DELETE FROM meta WHERE key = 'source_instance_id'", []); + } + + if let Ok(mut s) = state.sync.write() { + s.state = "idle".into(); + s.last_applied_seq = local_head; + s.source_instance_id = None; + s.primary_head_seq = Some(local_head); + s.last_error = None; + } + + state.set_primary_url(Some(replica_url.clone())); + state.start_replica_sync(); + + info!(new_primary = %replica_url, "Graceful switchover succeeded"); + Ok(Json(serde_json::json!({ + "status": "ok", + "message": "Role switchover completed successfully", + "old_primary_role": "replica", + "new_primary_url": replica_url, + "switched_at_seq": local_head, + }))) +} diff --git a/src/main.rs b/src/main.rs index 73018a4..bc9260b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -28,11 +28,12 @@ use crate::{ db::{init_db, load_or_create_instance_id, meta_get, meta_set}, handlers::{ create_record, create_token_handler, get_cluster_ca, get_config, get_info, - join_cluster_handler, list_records, reload_config, + join_cluster_handler, list_records, promote_handler, reload_config, repoint_handler, + step_down_handler, switchover_handler, }, - replication::{replica_loop, replication_changes, replication_status, SyncConfig}, + replication::{replication_changes, replication_status}, schema::load_config, - state::{AppState, Role, SyncStatus}, + state::{AppState, Role, SyncConfig, SyncStatus}, tls::{ bootstrap_replica_join, create_mtls_server_config, generate_ca, issue_leaf_cert, JoinToken, @@ -137,18 +138,24 @@ async fn main() { } else { // Replica: Check if dynamic bootstrap enrollment is needed if node_cert_pem.is_none() { - let p_url = primary_url - .as_ref() - .expect("APP_PRIMARY_URL required for replica enrollment"); + // APP_JOIN_URL lets the replication source (APP_PRIMARY_URL) differ from the + // CA enrollment target (APP_JOIN_URL). This is required in chain topologies + // where intermediate replicas serve data but cannot issue certificates. + // Falls back to APP_PRIMARY_URL when not set. + let join_url = std::env::var("APP_JOIN_URL") + .ok() + .or_else(|| primary_url.clone()) + .expect("APP_PRIMARY_URL (or APP_JOIN_URL) required for replica enrollment"); let token = std::env::var("APP_JOIN_TOKEN") .expect("APP_JOIN_TOKEN required for initial replica enrollment"); - info!(primary = %p_url, "Bootstrapping mTLS credentials via join token..."); - bootstrap_replica_join(p_url, &token, instance_id, &db) + info!(join_url = %join_url, "Bootstrapping mTLS credentials via join token..."); + bootstrap_replica_join(&join_url, &token, instance_id, &db) .await .expect("Dynamic join failed"); ca_cert_pem = meta_get(&db, "tls_ca_pem").ok().flatten(); + ca_key_pem = meta_get(&db, "tls_ca_key_pem").ok().flatten(); node_cert_pem = meta_get(&db, "tls_cert_pem").ok().flatten(); node_key_pem = meta_get(&db, "tls_key_pem").ok().flatten(); } @@ -171,13 +178,19 @@ async fn main() { "Instance initialised" ); + let sync_cfg = SyncConfig { + poll_interval: Duration::from_secs(env_parse("APP_SYNC_INTERVAL_SECS", 5)), + max_backoff: Duration::from_secs(env_parse("APP_SYNC_MAX_BACKOFF_SECS", 60)), + batch: env_parse("APP_SYNC_BATCH", 200i64).clamp(1, 1000), + }; + let state = AppState { db: Arc::new(Mutex::new(db)), schema: Arc::new(RwLock::new(schema)), config_path: Arc::new(config_path), instance_id, - role, - primary_url: primary_url.clone().map(Arc::new), + role: Arc::new(RwLock::new(role)), + primary_url: Arc::new(RwLock::new(primary_url)), repl_token: repl_token.clone().map(Arc::new), sync: Arc::new(RwLock::new(SyncStatus { state: if role == Role::Replica { "idle" } else { "primary" }.into(), @@ -187,27 +200,17 @@ async fn main() { last_success: None, last_error: None, })), + sync_cancel: Arc::new(Mutex::new(None)), + sync_cfg, ca_cert_pem: ca_cert_pem.clone().map(Arc::new), - ca_key_pem: ca_key_pem.map(Arc::new), + ca_key_pem: Arc::new(RwLock::new(ca_key_pem)), + node_cert_pem: node_cert_pem.clone().map(Arc::new), + node_key_pem: node_key_pem.clone().map(Arc::new), join_tokens, }; if role == Role::Replica { - let sync_cfg = SyncConfig { - poll_interval: Duration::from_secs(env_parse("APP_SYNC_INTERVAL_SECS", 5)), - max_backoff: Duration::from_secs(env_parse("APP_SYNC_MAX_BACKOFF_SECS", 60)), - batch: env_parse("APP_SYNC_BATCH", 200i64).clamp(1, 1000), - }; - - tokio::spawn(replica_loop( - state.clone(), - primary_url.expect("checked above"), - repl_token.unwrap_or_else(|| "".into()), - sync_cfg, - ca_cert_pem.clone().unwrap(), - node_cert_pem.clone().unwrap(), - node_key_pem.clone().unwrap(), - )); + state.start_replica_sync(); } let app = Router::new() @@ -216,6 +219,10 @@ async fn main() { .route("/api/records", get(list_records).post(create_record)) .route("/api/admin/config", get(get_config)) .route("/api/admin/reload-config", post(reload_config)) + .route("/api/admin/step-down", post(step_down_handler)) + .route("/api/admin/promote", post(promote_handler)) + .route("/api/admin/repoint", post(repoint_handler)) + .route("/api/admin/switchover", post(switchover_handler)) .route("/api/replication/changes", get(replication_changes)) .route("/api/replication/status", get(replication_status)) // Dynamic Peer Discovery Routes: diff --git a/src/replication.rs b/src/replication.rs index ac5bd1e..761a309 100644 --- a/src/replication.rs +++ b/src/replication.rs @@ -18,7 +18,7 @@ use uuid::Uuid; use crate::{ db::meta_set, - state::{api_error, db_error, internal_error, ApiError, AppState, SyncStatus}, + state::{api_error, db_error, internal_error, ApiError, AppState, SyncConfig, SyncStatus}, }; #[derive(Debug, Serialize, Deserialize)] @@ -43,13 +43,6 @@ pub struct ChangesQuery { pub limit: Option, } -#[derive(Debug, Clone)] -pub struct SyncConfig { - pub poll_interval: Duration, - pub max_backoff: Duration, - pub batch: i64, -} - enum SyncError { Transient(String), Fatal(String), @@ -165,10 +158,13 @@ pub async fn replication_status(State(state): State) -> Result, ) { let client = match create_mtls_client(&ca_pem, &cert_pem, &key_pem) { Ok(c) => c, @@ -348,30 +345,61 @@ pub async fn replica_loop( info!(primary = %primary_url, "Replica mTLS sync loop started"); loop { - match sync_once(&client, &state, &primary_url, &token, cfg.batch).await { - Ok(more) => { - backoff = Duration::from_secs(1); - if !more { - tokio::time::sleep(jitter(cfg.poll_interval)).await; + if *cancel_rx.borrow() { + info!("Replica sync loop stopped via cancellation token"); + break; + } + + tokio::select! { + _ = cancel_rx.changed() => { + if *cancel_rx.borrow() { + info!("Replica sync loop received cancellation"); + break; } } - Err(SyncError::Transient(message)) => { - let wait = jitter(backoff); - warn!(reason = %message, retry_in = ?wait, "Replication pull failed"); - update_status(&state, |s| { - s.state = "retrying".into(); - s.last_error = Some(message); - }); - tokio::time::sleep(wait).await; - backoff = (backoff * 2).min(cfg.max_backoff); - } - Err(SyncError::Fatal(message)) => { - error!(reason = %message, "Replication halted; operator action required"); - update_status(&state, |s| { - s.state = "halted".into(); - s.last_error = Some(message); - }); - return; + res = sync_once(&client, &state, &primary_url, &token, cfg.batch) => { + match res { + Ok(more) => { + backoff = Duration::from_secs(1); + if !more { + tokio::select! { + _ = cancel_rx.changed() => { + if *cancel_rx.borrow() { + info!("Replica sync loop cancelled while waiting for next poll"); + break; + } + } + _ = tokio::time::sleep(jitter(cfg.poll_interval)) => {} + } + } + } + Err(SyncError::Transient(message)) => { + let wait = jitter(backoff); + warn!(reason = %message, retry_in = ?wait, "Replication pull failed"); + update_status(&state, |s| { + s.state = "retrying".into(); + s.last_error = Some(message); + }); + tokio::select! { + _ = cancel_rx.changed() => { + if *cancel_rx.borrow() { + info!("Replica sync loop cancelled during backoff sleep"); + break; + } + } + _ = tokio::time::sleep(wait) => {} + } + backoff = (backoff * 2).min(cfg.max_backoff); + } + Err(SyncError::Fatal(message)) => { + error!(reason = %message, "Replication halted; operator action required"); + update_status(&state, |s| { + s.state = "halted".into(); + s.last_error = Some(message); + }); + return; + } + } } } } diff --git a/src/state.rs b/src/state.rs index 19a680d..32cecfb 100644 --- a/src/state.rs +++ b/src/state.rs @@ -5,30 +5,137 @@ use serde_json::Value; use std::{ collections::HashMap, sync::{Arc, Mutex, RwLock}, + time::Duration, }; -use tracing::error; +use tracing::{error, info, warn}; use uuid::Uuid; use crate::{schema::SchemaConfig, tls::JoinToken}; +#[derive(Debug, Clone)] +pub struct SyncConfig { + pub poll_interval: Duration, + pub max_backoff: Duration, + pub batch: i64, +} + #[derive(Clone)] pub struct AppState { pub db: Arc>, pub schema: Arc>, pub config_path: Arc, pub instance_id: Uuid, - pub role: Role, - pub primary_url: Option>, + pub role: Arc>, + pub primary_url: Arc>>, pub repl_token: Option>, pub sync: Arc>, + // Replication task management: + pub sync_cancel: Arc>>>, + pub sync_cfg: SyncConfig, + // mTLS & Discovery additions: pub ca_cert_pem: Option>, - pub ca_key_pem: Option>, + pub ca_key_pem: Arc>>, + pub node_cert_pem: Option>, + pub node_key_pem: Option>, pub join_tokens: Arc>>, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +impl AppState { + pub fn get_role(&self) -> Role { + *self.role.read().unwrap() + } + + pub fn is_primary(&self) -> bool { + self.get_role() == Role::Primary + } + + pub fn is_replica(&self) -> bool { + self.get_role() == Role::Replica + } + + pub fn get_primary_url(&self) -> Option { + self.primary_url.read().unwrap().clone() + } + + pub fn set_primary_url(&self, url: Option) { + let mut p = self.primary_url.write().unwrap(); + *p = url; + } + + pub fn stop_replica_sync(&self) { + let mut cancel = self.sync_cancel.lock().unwrap(); + if let Some(tx) = cancel.take() { + let _ = tx.send(true); + info!("Sent cancellation signal to replica sync loop"); + } + } + + pub fn start_replica_sync(&self) { + self.stop_replica_sync(); + + let primary_url = match self.get_primary_url() { + Some(u) if !u.is_empty() => u, + _ => { + warn!("Cannot start replica sync loop: primary_url is not set"); + return; + } + }; + + let token = self + .repl_token + .as_ref() + .map(|t| t.as_str().to_string()) + .unwrap_or_default(); + let ca_pem = match self.ca_cert_pem.as_ref() { + Some(ca) => ca.as_str().to_string(), + None => { + warn!("Cannot start replica sync loop: ca_cert_pem missing"); + return; + } + }; + let cert_pem = match self.node_cert_pem.as_ref() { + Some(c) => c.as_str().to_string(), + None => { + warn!("Cannot start replica sync loop: node_cert_pem missing"); + return; + } + }; + let key_pem = match self.node_key_pem.as_ref() { + Some(k) => k.as_str().to_string(), + None => { + warn!("Cannot start replica sync loop: node_key_pem missing"); + return; + } + }; + + let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false); + { + let mut cancel = self.sync_cancel.lock().unwrap(); + *cancel = Some(cancel_tx); + } + + let state_clone = self.clone(); + let sync_cfg = self.sync_cfg.clone(); + + tokio::spawn(async move { + crate::replication::replica_loop( + state_clone, + primary_url, + token, + sync_cfg, + ca_pem, + cert_pem, + key_pem, + cancel_rx, + ) + .await; + }); + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum Role { Primary, Replica, diff --git a/src/tls.rs b/src/tls.rs index f48fc9d..fd9e35a 100644 --- a/src/tls.rs +++ b/src/tls.rs @@ -34,6 +34,8 @@ pub struct JoinClusterResponse { pub ca_pem: String, pub cert_pem: String, pub key_pem: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ca_key_pem: Option, } // --------------------------------------------------------------------------- @@ -213,6 +215,9 @@ pub async fn bootstrap_replica_join( meta_set(db, "tls_ca_pem", &payload.ca_pem).map_err(|e| e.to_string())?; meta_set(db, "tls_cert_pem", &payload.cert_pem).map_err(|e| e.to_string())?; meta_set(db, "tls_key_pem", &payload.key_pem).map_err(|e| e.to_string())?; + if let Some(ref ca_key) = payload.ca_key_pem { + meta_set(db, "tls_ca_key_pem", ca_key).map_err(|e| e.to_string())?; + } info!("Replica successfully joined cluster and persisted mTLS certificates"); Ok(())