Compare commits

...
10 Commits
17 changed files with 4871 additions and 429 deletions

No files matched your search

+11
View File
@@ -0,0 +1,11 @@
/target
/repl-test
/plans
messages.db
pando.db
*.db
*.db-journal
*.db-wal
*.db-shm
__pycache__/
*.pyc
Generated
+1247 -16
View File
File diff suppressed because it is too large. Load diff
+10 -1
View File
@@ -1,15 +1,24 @@
[package]
name = "rust_test"
name = "pando"
version = "0.1.0"
edition = "2024"
description = "Lightweight Rust + SQLite records service where independent nodes replicate data, reconnect over unreliable links, and can change roles when needed."
[dependencies]
axum = { version = "0.8.9", features = ["json"] }
axum-server = { version = "0.7", features = ["tls-rustls"] }
chrono = { version = "0.4.45", features = ["serde"] }
rcgen = { version = "0.14.10", features = ["pem", "x509-parser"] }
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "gzip"] }
rusqlite = { version = "0.41.0", features = ["bundled"] }
rustls = { version = "0.23.46", features = ["ring"] }
rustls-pki-types = "1.0"
rustls-pemfile = "2.1"
serde = { version = "1.0.229", features = ["derive"] }
serde_json = "1.0.154"
tokio = { version = "1.54.1", features = ["macros", "rt-multi-thread"] }
tower-http = { version = "0.7.1", features = ["compression-gzip", "cors"] }
tracing = "0.1.44"
tracing-subscriber = "0.3.23"
uuid = { version = "1.29.0", features = ["v4", "serde"] }
+259 -17
View File
@@ -1,9 +1,12 @@
# Dynamic Schema Records API
# Pando! 🌳
* Dynamic Schema Records API *
A lightweight, high-performance JSON records service built with **Rust**, **Axum**, and **SQLite**.
A lightweight, high-performance JSON records service built with **Rust**, **Axum**, and **SQLite**.
The service validates incoming JSON payloads against a dynamic schema specification defined in a JSON configuration file. Schemas support validation modes (`strict` vs. `loose`), data type constraints, and exact forced values (e.g., pinning API payload versions). Schemas can be reloaded at runtime without restarting the server.
Instances can run as a **primary** (accepts writes) or as a **read-only replica** that lazily pulls records from a primary. Replication is built for high-latency and unreliable links.
---
## Features
@@ -15,7 +18,14 @@ The service validates incoming JSON payloads against a dynamic schema specificat
- **`strict`**: All defined fields are required, and unknown fields are rejected.
- **`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 and ISO-8601 UTC timestamp in an SQLite database (`messages.db`).
- **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 (`pando.db`).
- **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.
- 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.
---
@@ -25,13 +35,24 @@ The service validates incoming JSON payloads against a dynamic schema specificat
- [Rust toolchain](https://rustup.rs/) (1.75+ recommended)
- SQLite development libraries (if linking dynamically)
### Cargo dependencies of note
```toml
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "gzip"] }
tower-http = { version = "0.5", features = ["compression-gzip"] }
```
`reqwest` is used by the replica to pull changes. `tower-http` provides gzip response compression.
---
## Configuration
The service looks for a configuration file at the path specified by the `APP_CONFIG` environment variable, defaulting to `config.json` in the current working directory.
### Schema file
### Example: `config.json`
The service looks for a schema configuration file at the path specified by the `APP_CONFIG` environment variable, defaulting to `config.json` in the current working directory.
#### Example: `config.json`
```json
{
@@ -51,7 +72,7 @@ The service looks for a configuration file at the path specified by the `APP_CON
}
```
### Schema Specification
#### Schema Specification
| Property | Type | Description |
| :--- | :--- | :--- |
@@ -60,6 +81,22 @@ The service looks for a configuration file at the path specified by the `APP_CON
| `fields.<name>.type` | `string` | Expected JSON type: `string`, `integer`, `number`, `boolean`, `object`, `array`, `null`. |
| `fields.<name>.value` | *any* *(optional)* | Expected exact value. Rejects any request where the value does not strictly match. |
The schema is only enforced on writes to a primary. A replica still loads the file at startup, but it does not re-validate replicated records, since the primary already did.
### Environment variables
| Variable | Default | Description |
| :--- | :--- | :--- |
| `APP_CONFIG` | `pando.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 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. |
| `APP_SYNC_BATCH` | `200` | Records per pull (clamped to 1–1000). Lower it for very poor links. |
---
## Getting Started
@@ -68,10 +105,10 @@ The service looks for a configuration file at the path specified by the `APP_CON
```bash
# Clone the repository and enter the directory
cd your-project-name
cd pando
# Create your config file
cat <<EOF > config.json
cat <<EOF > pando.json
{
"mode": "strict",
"fields": {
@@ -96,6 +133,33 @@ To specify a custom configuration file path:
APP_CONFIG=custom_config.json cargo run
```
### 2. Run a primary and a replica
`pando.db` is opened relative to the working directory, so run each instance from its own directory.
```bash
# Primary
mkdir -p primary && cp pando.json primary/ && cd primary
APP_ROLE=primary \
APP_BIND=0.0.0.0:3000 \
APP_REPL_TOKEN=change-me \
../target/debug/pando
# Replica (in another terminal; start with an EMPTY database)
mkdir -p replica && cp pando.json replica/ && cd replica
APP_ROLE=replica \
APP_BIND=127.0.0.1:3001 \
APP_PRIMARY_URL=http://127.0.0.1:3000 \
APP_REPL_TOKEN=change-me \
../target/debug/pando
```
A ready-made script, `repl-test.sh`, starts both on localhost, writes a few records and checks that they arrive:
```bash
bash repl-test.sh
```
---
## API Reference
@@ -120,6 +184,7 @@ APP_CONFIG=custom_config.json cargo run
### Create Record
- **Endpoint:** `POST /api/records`
- **Headers:** `Content-Type: application/json`
- **Availability:** primary only. A replica answers `403 Forbidden` with the primary's URL in the message.
- **Request Body:**
```json
{
@@ -132,6 +197,7 @@ APP_CONFIG=custom_config.json cargo run
- **Response (`201 Created`):**
```json
{
"seq": 42,
"uuid": "f3b392b9-dc81-42cb-b9ef-dc7e90aa7ce5",
"date": "2026-10-25T10:20:15Z",
"data": {
@@ -140,21 +206,24 @@ APP_CONFIG=custom_config.json cargo run
}
}
```
- **Error Responses (`400 Bad Request`):**
- Missing field: `Missing required field: message`
- Wrong type: `Field 'version' must be of type integer`
- Value mismatch: `Field 'version' must equal 1`
- Unknown field (in strict mode): `Unknown field: extra_field`
- **Error Responses:**
- `400 Bad Request`:
- Missing field: `Missing required field: message`
- Wrong type: `Field 'version' must be of type integer`
- Value mismatch: `Field 'version' must equal 1`
- Unknown field (in strict mode): `Unknown field: extra_field`
- `403 Forbidden`: this instance is a read-only replica.
---
### List Records
- **Endpoint:** `GET /api/records`
- **Description:** Returns up to 100 most recent records ordered by `date DESC`.
- **Description:** Returns up to 100 most recent records ordered by `seq DESC`. Available on primaries and replicas.
- **Response (`200 OK`):**
```json
[
{
"seq": 42,
"uuid": "f3b392b9-dc81-42cb-b9ef-dc7e90aa7ce5",
"date": "2026-10-25T10:20:15Z",
"data": {
@@ -203,6 +272,155 @@ APP_CONFIG=custom_config.json cargo run
---
### 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=<seq>&limit=<n>`
- **Headers:** `Authorization: Bearer <APP_REPL_TOKEN>`
- **Description:** Returns records with `seq > after` in ascending order. `after` defaults to `0`; `limit` defaults to `200` and is clamped to 1–1000. The server holds no per-replica state.
- **Response (`200 OK`):**
```json
{
"instance_id": "9d2c1f0e-3a41-4b8e-8f6e-5d1b7a0c2e11",
"head_seq": 42,
"has_more": false,
"changes": [
{
"seq": 41,
"uuid": "a1e5c0de-6c0b-4f43-9d5e-0f2f3c6b9a77",
"date": "2026-10-25T10:19:58Z",
"data": { "version": 1, "message": "earlier" }
},
{
"seq": 42,
"uuid": "f3b392b9-dc81-42cb-b9ef-dc7e90aa7ce5",
"date": "2026-10-25T10:20:15Z",
"data": { "version": 1, "message": "Hello, world!" }
}
]
}
```
- **Error Responses:** `401 Unauthorized` for a wrong or missing token; `403 Forbidden` if no `APP_REPL_TOKEN` is configured on this instance.
Note that `seq` values are increasing but not guaranteed to be contiguous, so consumers must use them as a cursor and must not count on gap-free numbering.
---
### Replication: Status
- **Endpoint:** `GET /api/replication/status`
- **Description:** Instance identity, role, cursor and lag. Unauthenticated; it contains no secrets.
- **Response (`200 OK`), on a replica:**
```json
{
"instance_id": "5b7f4a3c-0f1d-4c6e-9b1a-2e8d6a4f7c90",
"role": "replica",
"primary_url": "http://127.0.0.1:3000",
"local_head_seq": 42,
"lag_records": 0,
"sync": {
"state": "idle",
"last_applied_seq": 42,
"source_instance_id": "9d2c1f0e-3a41-4b8e-8f6e-5d1b7a0c2e11",
"primary_head_seq": 42,
"last_success": "2026-10-25T10:20:20Z",
"last_error": null
}
}
```
- **`sync.state` values:**
| State | Meaning |
| :--- | :--- |
| `primary` | This instance is a primary; no sync is running. |
| `idle` | Replica is caught up and waiting for the next poll. |
| `syncing` | Replica is catching up; more batches are pending. |
| `retrying` | The last pull failed (network, 5xx, truncated body); retrying with backoff. |
| `halted` | Fatal mismatch detected; the replica has stopped. See *Failure handling*. |
---
## Replication
### Model
```
writes reads
┌───────────────┐ ┌───────────────┐
│ primary │ ◄─pull──│ replica │
│ seq 1..N │ after=K │ seq 1..K..N │
└───────────────┘ └───────────────┘
```
- **Journal and global tick:** `records` is insert-only, so the table itself is the append-only journal. Each row has a `seq` (`INTEGER PRIMARY KEY AUTOINCREMENT`), which is never reused, even after a rollback. A replica stores the *primary's* `seq` values unchanged, so the tick is the same everywhere.
- **Instance ID:** each instance generates a permanent UUID on first start and keeps it in the `meta` table. Every `/changes` response carries the source's ID.
- **Pull, not push:** the replica opens all connections, so only the primary needs to be reachable, and it can sit behind NAT or a firewall on the replica side.
- **Lazy sync:** the replica polls on a timer and catches up whenever the link is available. Reads from a replica can be stale, never inconsistent: records appear in `seq` order with no reordering.
### Replica sync loop
1. Read the cursor (`last_applied_seq`) from memory (restored from the DB at startup).
2. `GET /changes?after=<cursor>&limit=<batch>`.
3. Check identity and history (see below).
4. In **one SQLite transaction**: `INSERT OR IGNORE` every row, then store the new cursor and the source's instance ID.
5. If `has_more`, pull again immediately; otherwise sleep for `APP_SYNC_INTERVAL_SECS` (jittered).
Step 4 is atomic, so a crash can never leave the cursor out of step with the data. `INSERT OR IGNORE` makes re-delivery harmless. A truncated or invalid response body fails to parse and is retried, so a partial batch is never applied.
### Failure handling
| Condition | Behaviour |
| :--- | :--- |
| Network error, timeout, non-2xx, bad body | Transient. Retry with exponential backoff (1s doubling up to `APP_SYNC_MAX_BACKOFF_SECS`) with jitter; state `retrying`. |
| Non-monotonic `seq` in a response | Treated as transient; batch discarded. |
| Source `instance_id` differs from the one stored | **Fatal.** Replica halts. |
| Source `instance_id` equals the replica's own | **Fatal.** Typically a cloned database file. |
| Source `head_seq` < replica cursor | **Fatal.** Source was probably restored from an older backup. |
On a fatal mismatch the replica halts, because retrying cannot fix it, and the reason is in `sync.last_error` and the log. To repoint a replica at a different source, stop it, delete its `pando.db` and start it again so it resyncs from `seq` 0.
### Operational notes
- **Bootstrapping:** start a replica with an empty database. Don't seed it by copying the primary's `pando.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 & 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`.
---
## Example Usage Workflow
```bash
@@ -227,23 +445,47 @@ curl -i -X POST http://127.0.0.1:3000/api/records \
curl -i -X POST http://127.0.0.1:3000/api/records \
-H "Content-Type: application/json" \
-d '{"data": {"version": 2, "message": "Accepted"}}'
# 6. Read it back from a replica (once it has synced)
curl -s http://127.0.0.1:3001/api/records
# 7. Check replication health
curl -s http://127.0.0.1:3001/api/replication/status
```
---
## Database Architecture
On startup, the application creates a local SQLite database named `messages.db` with the following schema:
On startup, the application creates (or opens) a local SQLite database named `pando.db`:
```sql
CREATE TABLE IF NOT EXISTS meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS records (
uuid TEXT PRIMARY KEY,
seq INTEGER PRIMARY KEY AUTOINCREMENT,
uuid TEXT NOT NULL UNIQUE,
date TEXT NOT NULL,
data TEXT NOT NULL
);
```
This project is licensed under the [MIT License](LICENSE) (or choose your preferred license).
`meta` holds the following keys:
| Key | Description |
| :--- | :--- |
| `instance_id` | Permanent UUID of this instance. |
| `last_applied_seq` | Replica cursor (replicas only). |
| `source_instance_id` | ID of the source this replica follows (replicas only). |
### Migration
Databases created by earlier versions (`uuid TEXT PRIMARY KEY`, no `seq`) are migrated automatically on startup, in a single transaction. Existing rows get `seq` values in their original insertion order.
---
## License
-14
View File
@@ -1,17 +1,3 @@
{
"mode": "strict",
"fields": {
"version": {
"type": "integer",
"value": 2
},
"message": {
"type": "string"
}
}
}
{
"mode": "strict",
"fields": {
+150
View File
@@ -0,0 +1,150 @@
#!/usr/bin/env python3
"""
Lightweight test server and transparent proxy for the Pando 🌳 cluster.
Serves index.html on http://127.0.0.1:8080 and transparently proxies
requests to cluster instances, bypassing self-signed TLS certificate checks.
"""
from http.server import SimpleHTTPRequestHandler, HTTPServer
import urllib.request
import urllib.error
import urllib.parse
import ssl
import json
import os
PORT = int(os.environ.get("DASHBOARD_PORT", 8080))
# Create an SSL context that ignores self-signed certificate errors
SSL_CTX = ssl._create_unverified_context()
class ProxyHandler(SimpleHTTPRequestHandler):
def end_headers(self):
self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE, OPTIONS, HEAD')
self.send_header('Access-Control-Allow-Headers', '*')
super().end_headers()
def do_OPTIONS(self):
self.send_response(204)
self.end_headers()
def do_GET(self):
if self.path.startswith('/proxy'):
self.handle_proxy('GET')
else:
super().do_GET()
def do_HEAD(self):
if self.path.startswith('/proxy'):
self.handle_proxy('HEAD')
else:
super().do_HEAD()
def do_POST(self):
if self.path.startswith('/proxy'):
self.handle_proxy('POST')
else:
self.send_error(405, "Method Not Allowed")
def do_PUT(self):
if self.path.startswith('/proxy'):
self.handle_proxy('PUT')
else:
self.send_error(405, "Method Not Allowed")
def do_DELETE(self):
if self.path.startswith('/proxy'):
self.handle_proxy('DELETE')
else:
self.send_error(405, "Method Not Allowed")
def extract_target_url(self):
parsed = urllib.parse.urlparse(self.path)
if parsed.path.rstrip('/') == '/proxy':
if '?url=' in self.path:
return urllib.parse.unquote(self.path.split('?url=', 1)[1])
qs = urllib.parse.parse_qs(parsed.query)
return qs.get('url', [None])[0]
elif parsed.path.startswith('/proxy/'):
rest = self.path[len('/proxy/'):]
if rest.startswith('http://') or rest.startswith('https://'):
return rest
parts = rest.split('/', 1)
host_or_port = parts[0]
subpath = '/' + parts[1] if len(parts) > 1 else ''
if parsed.query:
subpath += '?' + parsed.query
if host_or_port.isdigit():
return f"https://127.0.0.1:{host_or_port}{subpath}"
return f"https://{host_or_port}{subpath}"
return None
def handle_proxy(self, method):
target_url = self.extract_target_url()
if not target_url:
self.send_response(400)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps({'error': 'Missing or invalid target URL'}).encode())
return
content_length = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(content_length) if content_length > 0 else None
# Filter hop-by-hop headers and compression to ensure clean raw data transfer
req_headers = {}
for k, v in self.headers.items():
if k.lower() in ('host', 'content-length', 'connection', 'transfer-encoding', 'accept-encoding'):
continue
req_headers[k] = v
req = urllib.request.Request(target_url, data=body, headers=req_headers, method=method)
try:
with urllib.request.urlopen(req, context=SSL_CTX, timeout=15) as resp:
status_code = resp.status
resp_headers = resp.getheaders()
resp_body = resp.read() if method != 'HEAD' else b''
except urllib.error.HTTPError as e:
status_code = e.code
resp_headers = e.headers.items()
resp_body = e.read() if method != 'HEAD' else b''
except urllib.error.URLError as e:
self.send_response(502)
self.send_header('Content-Type', 'application/json')
self.end_headers()
err_msg = json.dumps({'error': f'Failed to connect to backend: {e.reason}'}).encode()
self.wfile.write(err_msg)
return
except Exception as e:
self.send_response(500)
self.send_header('Content-Type', 'application/json')
self.end_headers()
err_msg = json.dumps({'error': f'Proxy exception: {str(e)}'}).encode()
self.wfile.write(err_msg)
return
self.send_response(status_code)
for k, v in resp_headers:
if k.lower() in ('transfer-encoding', 'connection', 'content-length',
'access-control-allow-origin', 'access-control-allow-methods',
'access-control-allow-headers'):
continue
self.send_header(k, v)
if method != 'HEAD':
self.send_header('Content-Length', str(len(resp_body)))
self.end_headers()
if method != 'HEAD' and resp_body:
self.wfile.write(resp_body)
if __name__ == '__main__':
print(f"🚀 Dashboard running at http://127.0.0.1:{PORT}")
print(f"🔒 TLS unverified proxy active at /proxy?url=<target>")
print("Press Ctrl+C to stop.")
server = HTTPServer(('127.0.0.1', PORT), ProxyHandler)
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nStopping...")
+821
View File
@@ -0,0 +1,821 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Records API & Replication Dashboard</title>
<style>
:root {
--bg: #0d1117;
--card-bg: #161b22;
--border: #30363d;
--text: #c9d1d9;
--text-muted: #8b949e;
--accent: #58a6ff;
--success: #3fb950;
--warning: #d29922;
--danger: #f85149;
--primary-color: #79c0ff;
--replica-color: #bc8cff;
}
* { box-sizing: border-box; margin: 0; padding: 0; }
body {
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, monospace;
background: var(--bg);
color: var(--text);
line-height: 1.5;
padding: 20px;
}
header {
display: flex;
justify-content: space-between;
align-items: center;
padding-bottom: 16px;
margin-bottom: 20px;
border-bottom: 1px solid var(--border);
}
h1 { font-size: 1.4rem; color: #fff; display: flex; align-items: center; gap: 10px; }
.badge {
font-size: 0.75rem;
padding: 3px 8px;
border-radius: 12px;
font-weight: bold;
text-transform: uppercase;
}
.badge-primary { background: rgba(121, 192, 255, 0.15); color: var(--primary-color); }
.badge-replica { background: rgba(188, 140, 255, 0.15); color: var(--replica-color); }
.badge-ok { background: rgba(63, 185, 80, 0.2); color: var(--success); }
.badge-warn { background: rgba(210, 153, 34, 0.2); color: var(--warning); }
.badge-err { background: rgba(248, 81, 73, 0.2); color: var(--danger); }
.grid-2 { display: grid; grid-template-columns: 1fr 1fr; gap: 20px; }
.grid-3 { display: grid; grid-template-columns: 1fr 1fr 1fr; gap: 20px; }
@media (max-width: 900px) { .grid-2, .grid-3 { grid-template-columns: 1fr; } }
.card {
background: var(--card-bg);
border: 1px solid var(--border);
border-radius: 8px;
padding: 16px;
margin-bottom: 20px;
}
.card-header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 12px;
padding-bottom: 8px;
border-bottom: 1px solid var(--border);
font-size: 0.95rem;
font-weight: 600;
}
/* Topology Bar */
.topology {
display: flex;
align-items: center;
justify-content: space-around;
padding: 16px;
background: rgba(255,255,255,0.02);
border: 1px dashed var(--border);
border-radius: 8px;
margin-bottom: 20px;
flex-wrap: wrap;
gap: 16px;
}
.node-box {
text-align: center;
padding: 12px 20px;
border-radius: 6px;
border: 1px solid var(--border);
background: var(--card-bg);
min-width: 200px;
flex: 1;
max-width: 320px;
}
.pipe {
display: flex;
flex-direction: column;
align-items: center;
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;
}
/* Inputs & Buttons */
input, textarea, select {
width: 100%;
background: #0d1117;
border: 1px solid var(--border);
color: #fff;
padding: 8px 10px;
border-radius: 6px;
font-family: inherit;
font-size: 0.85rem;
margin-top: 4px;
margin-bottom: 10px;
}
button {
background: #21262d;
color: var(--text);
border: 1px solid var(--border);
padding: 6px 14px;
border-radius: 6px;
cursor: pointer;
font-size: 0.85rem;
font-weight: 500;
transition: all 0.15s;
}
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:not(:disabled) { background: #2ea043; }
button.btn-test { background: #1f6feb; border-color: #388bfd; color: #fff; }
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; }
table { width: 100%; border-collapse: collapse; font-size: 0.82rem; }
th, td { text-align: left; padding: 6px 10px; border-bottom: 1px solid var(--border); }
th { position: sticky; top: 0; background: var(--card-bg); color: var(--text-muted); font-size: 0.75rem; }
tr.synced { background: rgba(63, 185, 80, 0.05); }
tr.lagging { background: rgba(248, 81, 73, 0.1); }
/* Logs */
.log-box {
height: 180px;
overflow-y: auto;
background: #090d13;
border: 1px solid var(--border);
border-radius: 6px;
padding: 10px;
font-size: 0.78rem;
font-family: monospace;
}
.log-line { margin-bottom: 4px; border-bottom: 1px dotted rgba(255,255,255,0.05); padding-bottom: 2px; }
.log-info { color: var(--accent); }
.log-pass { color: var(--success); }
.log-fail { color: var(--danger); }
</style>
</head>
<body>
<header>
<div>
<h1>⚡ Records API & Replication Monitor</h1>
<p style="color: var(--text-muted); font-size: 0.85rem; margin-top: 4px;">
Interactive validation and real-time visualization of SQLite append-only journals & mTLS failover
</p>
</div>
<div style="display:flex; align-items:center; gap: 10px;">
<label style="font-size:0.8rem; display:flex; align-items:center; gap:6px; cursor:pointer;">
<input type="checkbox" id="autoPoll" checked style="width:auto; margin:0;" /> Live Auto-Poll (1.5s)
</label>
<button onclick="fetchAll()">🔄 Refresh Now</button>
</div>
</header>
<!-- Node Addresses & Endpoints Setup -->
<div class="card" style="padding: 10px 16px;">
<div style="display:flex; gap: 20px; align-items:center; flex-wrap:wrap; font-size:0.85rem;">
<div style="flex:1; min-width:240px;">
<label>Node 1 Base URL:</label>
<input type="text" id="node1Url" value="https://127.0.0.1:3000" style="margin-bottom:0;" />
</div>
<div style="flex:1; min-width:240px;">
<label>Node 2 Base URL:</label>
<input type="text" id="node2Url" value="https://127.0.0.1:3001" style="margin-bottom:0;" />
</div>
<div style="flex:1; min-width:180px;">
<label>Replication Bearer Token:</label>
<input type="password" id="replToken" value="change-me" style="margin-bottom:0;" />
</div>
<div style="min-width:220px; display:flex; align-items:flex-end;">
<label style="font-size:0.8rem; display:flex; align-items:center; gap:6px; cursor:pointer; padding-bottom:6px; color:var(--success);">
<input type="checkbox" id="trustAllCerts" checked style="width:auto; margin:0;" /> 🛡️ Trust Self-Signed Certs (Proxy)
</label>
</div>
</div>
</div>
<!-- Live Replication Pipeline Banner -->
<div class="topology">
<!-- Node 1 Box -->
<div class="node-box" id="node1Box">
<div style="display:flex; justify-content:space-between; align-items:center; margin-bottom:4px;">
<span style="font-size:0.75rem; color:var(--text-muted); font-weight:600;">Node 1 (:3000)</span>
<span id="node1RoleBadge" class="badge badge-primary">PRIMARY</span>
</div>
<div style="font-size: 1.3rem; font-weight: bold; margin-top: 5px;" id="node1Head">Seq: --</div>
<div style="font-size: 0.72rem; color: var(--text-muted);" id="node1Id">ID: --</div>
<div style="font-size: 0.72rem; color: var(--text-muted);" id="node1SyncState">State: --</div>
</div>
<!-- Directional Pipeline & Switchover Control -->
<div class="pipe">
<span id="pipeState" class="badge badge-ok">CONNECTED</span>
<div id="pipeArrow" class="arrow">━━━━ ◄─pull── ━━━━</div>
<div id="pipeDetails" style="font-size:0.78rem;">Lag: <span id="lagRecords">0</span> records</div>
<button id="switchoverBtn" class="btn-switch" style="margin-top:4px;" onclick="performSwitchover()">
🔀 Graceful Switchover
</button>
</div>
<!-- Node 2 Box -->
<div class="node-box" id="node2Box">
<div style="display:flex; justify-content:space-between; align-items:center; margin-bottom:4px;">
<span style="font-size:0.75rem; color:var(--text-muted); font-weight:600;">Node 2 (:3001)</span>
<span id="node2RoleBadge" class="badge badge-replica">REPLICA</span>
</div>
<div style="font-size: 1.3rem; font-weight: bold; margin-top: 5px;" id="node2Head">Seq: --</div>
<div style="font-size: 0.72rem; color: var(--text-muted);" id="node2Id">ID: --</div>
<div style="font-size: 0.72rem; color: var(--text-muted);" id="node2SyncState">State: --</div>
</div>
</div>
<!-- Main Side-by-Side Journal Tables with Dedicated Write Box Per Instance -->
<div class="grid-2">
<!-- Node 1 Column -->
<div class="card">
<div class="card-header">
<div style="display:flex; align-items:center; gap:8px;">
<span>📖 Node 1 Records</span>
<span id="node1TableRole" class="badge badge-primary">PRIMARY</span>
</div>
<span id="node1Count" class="badge badge-primary">0 records</span>
</div>
<!-- Write Box for Node 1 -->
<div class="write-box">
<div style="display:flex; justify-content:space-between; align-items:center; margin-bottom:6px;">
<span style="font-size:0.8rem; font-weight:600;">✍️ Insert Record (Node 1)</span>
<span style="font-size:0.72rem; color:var(--text-muted);">POST /api/records</span>
</div>
<textarea id="node1RecordData" rows="2" style="font-family:monospace; margin-bottom:8px;">{ "version": 1, "message": "Written to Node 1" }</textarea>
<button class="btn-primary" style="width:100%;" onclick="submitRecordToNode('node1')">
Send to Node 1 (:3000)
</button>
</div>
<!-- Node 1 Table -->
<div class="table-container">
<table id="node1Table">
<thead><tr><th>Seq</th><th>UUID</th><th>Date (UTC)</th><th>Payload</th></tr></thead>
<tbody><tr><td colspan="4" style="text-align:center; color:var(--text-muted)">Loading...</td></tr></tbody>
</table>
</div>
</div>
<!-- Node 2 Column -->
<div class="card">
<div class="card-header">
<div style="display:flex; align-items:center; gap:8px;">
<span>📖 Node 2 Records</span>
<span id="node2TableRole" class="badge badge-replica">REPLICA</span>
</div>
<span id="node2Count" class="badge badge-replica">0 records</span>
</div>
<!-- Write Box for Node 2 -->
<div class="write-box">
<div style="display:flex; justify-content:space-between; align-items:center; margin-bottom:6px;">
<span style="font-size:0.8rem; font-weight:600;">✍️ Insert Record (Node 2)</span>
<span style="font-size:0.72rem; color:var(--text-muted);">POST /api/records</span>
</div>
<textarea id="node2RecordData" rows="2" style="font-family:monospace; margin-bottom:8px;">{ "version": 1, "message": "Written to Node 2" }</textarea>
<button class="btn-primary" style="width:100%;" onclick="submitRecordToNode('node2')">
Send to Node 2 (:3001)
</button>
</div>
<!-- Node 2 Table -->
<div class="table-container">
<table id="node2Table">
<thead><tr><th>Seq</th><th>UUID</th><th>Date (UTC)</th><th>Payload</th></tr></thead>
<tbody><tr><td colspan="4" style="text-align:center; color:var(--text-muted)">Loading...</td></tr></tbody>
</table>
</div>
</div>
</div>
<!-- Action & Validation Panel -->
<div class="grid-3">
<!-- Automated Validation Tests -->
<div class="card">
<div class="card-header"><span>🧪 Automated Tests</span></div>
<div style="display:flex; flex-direction:column; gap:8px;">
<button class="btn-test" onclick="runSwitchoverTest()">🔀 Test Graceful Role Switchover</button>
<button class="btn-test" onclick="runReplicaWriteGuardTest()">🛡️ Test Replica Read-Only Guard (Expect 403)</button>
<button class="btn-test" onclick="runSchemaValidationTest()">⚠️ Test Invalid Schema Payload (Expect 400)</button>
<button class="btn-test" onclick="runSyncLatencyTest()">⏱️ Test Insert-to-Replica Propagation</button>
</div>
</div>
<!-- Manual Role Management -->
<div class="card">
<div class="card-header"><span>🎛️ Role Controls</span></div>
<div style="display:flex; flex-direction:column; gap:8px;">
<button onclick="manualPromote('node1')">👑 Promote Node 1 to Primary</button>
<button onclick="manualPromote('node2')">👑 Promote Node 2 to Primary</button>
<button onclick="manualStepDown()">⬇️ Step Down Active Primary</button>
</div>
</div>
<!-- Active Schema & Admin -->
<div class="card">
<div class="card-header">
<span>⚙️ Schema Management</span>
<button onclick="fetchSchema()" style="padding:2px 8px; font-size:0.75rem;">Inspect</button>
</div>
<pre id="schemaView" style="font-size:0.75rem; background:#0d1117; padding:8px; border-radius:4px; max-height:85px; overflow:auto;">Click Inspect to view</pre>
<button onclick="reloadSchema()" style="margin-top:8px; width:100%;">Trigger Hot-Reload (`/reload-config`)</button>
</div>
</div>
<!-- Test Console Log -->
<div class="card">
<div class="card-header">
<span>🖥️ Live Verification Console</span>
<button onclick="document.getElementById('consoleLog').innerHTML = '';" style="font-size:0.75rem;">Clear</button>
</div>
<div class="log-box" id="consoleLog"></div>
</div>
<script>
// Transparently proxy external/HTTPS instance calls via dashboard proxy when self-signed cert trust is active
const nativeFetch = window.fetch;
window.fetch = function(input, init) {
const trustAll = document.getElementById('trustAllCerts')?.checked ?? true;
if (trustAll) {
const urlStr = typeof input === 'string' ? input : (input && input.url ? input.url : null);
if (urlStr) {
try {
const parsed = new URL(urlStr, window.location.href);
if (parsed.origin !== window.location.origin) {
const proxyBase = (window.location.protocol === 'file:') ? 'http://127.0.0.1:8080' : '';
const proxiedUrl = `${proxyBase}/proxy?url=${encodeURIComponent(urlStr)}`;
if (typeof input === 'string') {
input = proxiedUrl;
} else {
input = new Request(proxiedUrl, input);
}
}
} catch (_) {}
}
}
return nativeFetch(input, init);
};
let node1HeadSeq = 0;
let node2HeadSeq = 0;
let currentNode1Role = 'unknown';
let currentNode2Role = 'unknown';
function log(msg, type = 'info') {
const box = document.getElementById('consoleLog');
const time = new Date().toISOString().split('T')[1].replace('Z','');
const div = document.createElement('div');
div.className = `log-line log-${type}`;
div.innerText = `[${time}] ${msg}`;
box.prepend(div);
}
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 {
node1: n1.replace(/\/+$/, ''),
node2: n2.replace(/\/+$/, ''),
token: document.getElementById('replToken')?.value || 'change-me'
};
}
async function fetchAll() {
const { node1, node2 } = getUrls();
let node1Status = null;
let node2Status = null;
let node1Records = [];
let node2Records = [];
// 1. Query Node 1 Status & Records
try {
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`;
// 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. Update the Live Replication Pipeline Banner dynamically
updatePipelineBanner(node1Status, node2Status);
}
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: <strong>${lag}</strong> 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: <strong>${lag}</strong> 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 = `<tr><td colspan="4" style="text-align:center; color:var(--text-muted)">Database is empty</td></tr>`;
return;
}
tbody.innerHTML = records.map(r => `
<tr class="${r.seq === globalHead ? 'synced' : ''}">
<td><strong>#${r.seq}</strong></td>
<td style="font-family:monospace; font-size:0.75rem;" title="${r.uuid}">${r.uuid.slice(0, 8)}…</td>
<td style="font-size:0.75rem;">${r.date.replace('T', ' ').replace('Z', '')}</td>
<td><code style="font-size:0.75rem;">${JSON.stringify(r.data)}</code></td>
</tr>
`).join('');
}
// 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(textarea.value);
} catch (err) {
log(`Invalid JSON in ${nodeName} write box: ${err.message}`, 'fail');
return;
}
try {
const res = await fetch(`${targetUrl}/api/records`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ data: payload })
});
const text = await res.text();
if (res.status === 201) {
const data = JSON.parse(text);
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(`${nodeName} returned [HTTP ${res.status}]: ${text}`, 'fail');
}
} catch (e) {
log(`Network failure writing to ${nodeName}: ${e.message}`, 'fail');
}
}
// 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(`${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: ${targetName} blocked write with HTTP 403 Forbidden! Response: "${msg}"`, 'pass');
} else {
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 3: Schema validation test
async function runSchemaValidationTest() {
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(`${primaryUrl}/api/records`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ data: { unknown_xyz_field: 999999, version: "not-an-integer" } })
});
const msg = await res.text();
if (res.status === 400) {
log(`PASS: Schema validation rejected bad payload [HTTP 400]: "${msg}"`, 'pass');
} else {
log(`FAIL: Expected HTTP 400, got ${res.status}: "${msg}"`, 'fail');
}
} catch (e) {
log(`Test failed: ${e.message}`, 'fail');
}
}
// TEST 4: Sync Latency propagation benchmark
async function runSyncLatencyTest() {
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(`${primaryUrl}/api/records`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ data: { message: testMsg, msg: testMsg, version: 1 } })
});
if (!res.ok) throw new Error(`Primary write failed with HTTP ${res.status}`);
const created = await res.json();
const targetSeq = created.seq;
log(`Record created with seq #${targetSeq}. Waiting for Replica sync poll...`, 'info');
let attempts = 0;
const interval = setInterval(async () => {
attempts++;
try {
const statusRes = await fetch(`${replicaUrl}/api/replication/status`);
const status = await statusRes.json();
if (status.local_head_seq >= targetSeq) {
clearInterval(interval);
const elapsed = ((performance.now() - startTime) / 1000).toFixed(2);
log(`PASS: Record #${targetSeq} successfully replicated in ${elapsed}s (${attempts} checks)!`, 'pass');
fetchAll();
} else if (attempts >= 40) {
clearInterval(interval);
log(`FAIL: Replica did not reach seq #${targetSeq} within 20 seconds.`, 'fail');
}
} catch (e) {
clearInterval(interval);
log(`Replication polling error: ${e.message}`, 'fail');
}
}, 500);
} catch (e) {
log(`Probe failed: ${e.message}`, 'fail');
}
}
async function fetchSchema() {
const { node1, node2 } = getUrls();
const primaryUrl = (currentNode1Role === 'primary') ? node1 : (currentNode2Role === 'primary' ? node2 : node1);
try {
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');
} catch (e) {
log(`Failed to fetch schema: ${e.message}`, 'fail');
}
}
async function reloadSchema() {
const { node1, node2 } = getUrls();
const primaryUrl = (currentNode1Role === 'primary') ? node1 : (currentNode2Role === 'primary' ? node2 : node1);
try {
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();
} catch (e) {
log(`Reload failed: ${e.message}`, 'fail');
}
}
// Auto-polling cycle
setInterval(() => {
if (document.getElementById('autoPoll').checked) {
fetchAll();
}
}, 1500);
// Initial load
fetchAll();
</script>
</body>
</html>
BIN
View File
Binary file not shown.
+15
View File
@@ -0,0 +1,15 @@
{
"mode": "strict",
"fields": {
"version": {
"type": "integer",
"value": 2
},
"message": {
"type": "string"
},
"name": {
"type": "string"
}
}
}
Executable
+462
View File
@@ -0,0 +1,462 @@
#!/usr/bin/env bash
# ==============================================================================
# repl-test.sh: Integration harness for Pando 🌳 cluster & pando-sync (mTLS)
#
# 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
# ---------------------------------------------------------------------------
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
if [[ -n "${BIN:-}" ]]; then
: # BIN provided externally
elif [[ -f "$ROOT_DIR/target/debug/pando" ]]; then
BIN="$ROOT_DIR/target/debug/pando"
else
BIN="$ROOT_DIR/target/debug/rust_test"
fi
WORK="${WORK:-$ROOT_DIR/repl-test}"
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}"
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
# ---------------------------------------------------------------------------
check_port() {
local port=$1
if command -v lsof >/dev/null 2>&1; then
lsof -iTCP:"$port" -sTCP:LISTEN -n -P >/dev/null 2>&1
elif command -v nc >/dev/null 2>&1; then
nc -z 127.0.0.1 "$port" >/dev/null 2>&1
else
(exec 3<>/dev/tcp/127.0.0.1/"$port") 2>/dev/null && { exec 3>&-; exec 3<&-; return 0; } || return 1
fi
}
cleanup() {
trap - EXIT INT TERM
echo -e "\n\033[1;33mStopping all instances and dashboard...\033[0m"
for pid in "${PIDS[@]}"; do
if kill -0 "$pid" 2>/dev/null; then
kill "$pid" 2>/dev/null || true
fi
done
wait 2>/dev/null || true
echo "Cleaned up."
}
trap cleanup EXIT INT TERM
json_field() {
# Usage: json_field <json_string> <field_name>
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
color_echo red "Error: Port $port is already in use." >&2
exit 1
fi
done
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
# 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
# ---------------------------------------------------------------------------
# Workspace setup
# ---------------------------------------------------------------------------
rm -rf "$WORK"
for idx in $(seq 0 $(( NODES - 1 ))); do
name=$(node_name "$idx")
mkdir -p "$WORK/$name"
cat > "$WORK/$name/pando.json" <<'CFGEOF'
{
"mode": "loose",
"fields": {
"version": { "type": "integer" }
}
}
CFGEOF
cp "$WORK/$name/pando.json" "$WORK/$name/config.json"
done
# ---------------------------------------------------------------------------
# start_node <name> <port> [extra env vars...]
# ---------------------------------------------------------------------------
start_node() {
local name=$1 port=$2
shift 2
( cd "$WORK/$name" && exec env \
PANDO_CONFIG=pando.json \
APP_CONFIG=pando.json \
PANDO_DB=pando.db \
APP_DB=pando.db \
PANDO_BIND="127.0.0.1:$port" \
APP_BIND="127.0.0.1:$port" \
PANDO_REPL_TOKEN="$TOKEN" \
APP_REPL_TOKEN="$TOKEN" \
PANDO_SYNC_INTERVAL_SECS=1 \
APP_SYNC_INTERVAL_SECS=1 \
PANDO_CORS_ORIGIN="$CORS_ORIGIN" \
APP_CORS_ORIGIN="$CORS_ORIGIN" \
"$@" "$BIN" ) >"$WORK/$name/log.txt" 2>&1 &
PIDS+=("$!")
}
# ==============================================================================
# 1. Start Primary
# ==============================================================================
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. Retrieve CA cert (sanity check)
# ==============================================================================
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
color_echo green " ✔ Cluster CA certificate downloaded."
else
color_echo red " ✖ Failed to retrieve Cluster CA – aborting." >&2; exit 1
fi
# ==============================================================================
# 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
# ==============================================================================
# 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)
PORTS[0]=$PRIMARY_PORT
NAMES[0]="primary"
SOURCES[0]=""
for idx in $(seq 1 $(( NODES - 1 ))); do
name=$(node_name "$idx")
port=$(node_port "$idx")
PORTS[$idx]=$port
NAMES[$idx]="$name"
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
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/pando.db" || -f "$WORK/$name/messages.db" ]]; then
color_echo green " ✔ mTLS credentials persisted in SQLite (pando.db)."
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
# ==============================================================================
# 4. Write smoke-test records to primary
# ==============================================================================
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"
done
# ==============================================================================
# 5. Verify every replica has converged
# ==============================================================================
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
done
# ==============================================================================
# 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
# ==============================================================================
# Ready Banner
# ==============================================================================
echo ""
echo "======================================================================"
echo " Pando Grove is healthy 🌳 • Topology: $TOPO • Nodes: $NODES"
echo ""
echo " Aspen Nodes (pando-node):"
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 " Pando Dashboard: http://127.0.0.1:$DASHBOARD_PORT"
fi
echo " Logs Dir: $WORK/"
echo ""
echo " Press Ctrl+C to stop all instances."
echo "======================================================================"
if [[ "$NO_WAIT" -ne 1 ]]; then
wait
fi
+83
View File
@@ -0,0 +1,83 @@
use rusqlite::{params, Connection, OptionalExtension};
use tracing::{info, warn};
use uuid::Uuid;
pub fn meta_get(db: &Connection, key: &str) -> rusqlite::Result<Option<String>> {
db.query_row("SELECT value FROM meta WHERE key = ?1", [key], |r| {
r.get::<_, String>(0)
})
.optional()
}
pub fn meta_set(db: &Connection, key: &str, value: &str) -> rusqlite::Result<()> {
db.execute(
"INSERT INTO meta (key, value) VALUES (?1, ?2)
ON CONFLICT(key) DO UPDATE SET value = excluded.value",
params![key, value],
)?;
Ok(())
}
pub fn init_db(db: &mut Connection) -> rusqlite::Result<()> {
db.execute_batch(
"CREATE TABLE IF NOT EXISTS meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);",
)?;
let exists: bool = db.query_row(
"SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'records')",
[],
|r| r.get(0),
)?;
if exists {
let has_seq: bool = db.query_row(
"SELECT EXISTS(SELECT 1 FROM pragma_table_info('records') WHERE name = 'seq')",
[],
|r| r.get(0),
)?;
if !has_seq {
info!("Migrating records table to journal layout");
let tx = db.transaction()?;
tx.execute_batch(
"ALTER TABLE records RENAME TO records_old;
CREATE TABLE records (
seq INTEGER PRIMARY KEY AUTOINCREMENT,
uuid TEXT NOT NULL UNIQUE,
date TEXT NOT NULL,
data TEXT NOT NULL
);
INSERT INTO records (uuid, date, data)
SELECT uuid, date, data FROM records_old ORDER BY rowid;
DROP TABLE records_old;",
)?;
tx.commit()?;
}
}
db.execute_batch(
"CREATE TABLE IF NOT EXISTS records (
seq INTEGER PRIMARY KEY AUTOINCREMENT,
uuid TEXT NOT NULL UNIQUE,
date TEXT NOT NULL,
data TEXT NOT NULL
);",
)?;
Ok(())
}
pub fn load_or_create_instance_id(db: &Connection) -> rusqlite::Result<Uuid> {
if let Some(existing) = meta_get(db, "instance_id")? {
if let Ok(id) = Uuid::parse_str(&existing) {
return Ok(id);
}
warn!("Stored instance_id is invalid; generating a new one");
}
let id = Uuid::new_v4();
meta_set(db, "instance_id", &id.to_string())?;
Ok(id)
}
+635
View File
@@ -0,0 +1,635 @@
use axum::{
extract::{rejection::JsonRejection, State},
http::StatusCode,
Json,
};
use chrono::Utc;
use crate::{
db::{meta_get, meta_set},
tls::{
create_join_token, create_mtls_client, issue_leaf_cert, JoinClusterRequest,
JoinClusterResponse,
},
};
use std::time::Duration;
use serde::Deserialize;
use rusqlite::params;
use serde_json::Value;
use tracing::{error, info, warn};
use uuid::Uuid;
use crate::{
schema::{load_config, validate_data, ValidationMode},
state::{api_error, db_error, internal_error, ApiError, ApiResponse, AppState, CreateRequest, Role},
};
/// Public endpoint: fetch the cluster's public CA certificate.
pub async fn get_cluster_ca(
State(state): State<AppState>,
) -> Result<Json<Value>, ApiError> {
let ca_pem = state
.ca_cert_pem
.as_deref()
.ok_or_else(|| api_error(StatusCode::NOT_FOUND, "Cluster CA not available"))?;
Ok(Json(serde_json::json!({
"instance_id": state.instance_id,
"ca_pem": ca_pem
})))
}
/// Admin endpoint: mint a short-lived join token (valid for 10 minutes by default).
pub async fn create_token_handler(
State(state): State<AppState>,
payload: Option<Json<Value>>,
) -> Result<Json<Value>, ApiError> {
if state.is_replica() {
return Err(api_error(StatusCode::FORBIDDEN, "Only Primary can generate join tokens"));
}
let ttl = payload
.and_then(|p| p.get("ttl_secs").and_then(|v| v.as_u64()))
.unwrap_or(600);
let token = create_join_token(ttl);
let mut tokens = state
.join_tokens
.write()
.map_err(|_| internal_error("tokens lock poisoned"))?;
tokens.insert(token.token.clone(), token.clone());
info!(token = %token.token, expires_at = %token.expires_at, "Issued join token");
Ok(Json(serde_json::json!(token)))
}
/// Dynamic join endpoint: exchanges a valid join token for an mTLS certificate.
pub async fn join_cluster_handler(
State(state): State<AppState>,
Json(req): Json<JoinClusterRequest>,
) -> Result<Json<JoinClusterResponse>, ApiError> {
if state.is_replica() {
return Err(api_error(StatusCode::FORBIDDEN, "Only Primary can issue certificates"));
}
let ca_cert_pem = state
.ca_cert_pem
.as_ref()
.ok_or_else(|| internal_error("CA cert unavailable"))?;
let ca_key_pem = state
.ca_key_pem
.read()
.map_err(|_| internal_error("CA key lock poisoned"))?
.clone()
.ok_or_else(|| internal_error("CA key unavailable"))?;
// 1. Validate & burn the token
{
let mut tokens = state
.join_tokens
.write()
.map_err(|_| internal_error("tokens lock poisoned"))?;
let entry = tokens
.get_mut(&req.token)
.ok_or_else(|| api_error(StatusCode::UNAUTHORIZED, "Invalid join token"))?;
if entry.used {
return Err(api_error(StatusCode::UNAUTHORIZED, "Join token already used"));
}
if entry.expires_at < Utc::now() {
return Err(api_error(StatusCode::UNAUTHORIZED, "Join token expired"));
}
entry.used = true;
}
// 2. Issue leaf certificate signed by Cluster CA
let mut sans = req.san_list;
if !sans.iter().any(|s| s == "localhost") {
sans.push("localhost".into());
}
if !sans.iter().any(|s| s == "127.0.0.1") {
sans.push("127.0.0.1".into());
}
let (cert_pem, key_pem) = issue_leaf_cert(
ca_cert_pem,
&ca_key_pem,
&req.node_id.to_string(),
sans,
)
.map_err(|e| internal_error(&format!("cert issue failed: {e}")))?;
info!(node_id = %req.node_id, "Issued client/server mTLS cert to replica");
Ok(Json(JoinClusterResponse {
ca_pem: ca_cert_pem.to_string(),
cert_pem,
key_pem,
ca_key_pem: Some(ca_key_pem),
}))
}
pub async fn get_info() -> Json<ApiResponse> {
info!(endpoint = "/api/now", "Request received");
Json(ApiResponse {
seq: None,
uuid: Uuid::new_v4(),
date: Utc::now().to_rfc3339(),
data: serde_json::json!({
"service": "pando",
"version": 1,
"message": "Hello from Pando! 🌳"
}),
})
}
pub async fn create_record(
State(state): State<AppState>,
payload: Result<Json<CreateRequest>, JsonRejection>,
) -> Result<(StatusCode, Json<ApiResponse>), ApiError> {
info!(endpoint = "/api/records", "Create request received");
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 {}",
p_url.as_deref().unwrap_or("the primary")
),
));
}
let Json(request) = payload.map_err(|e| {
warn!(error = %e, "Invalid JSON request");
api_error(StatusCode::BAD_REQUEST, "Invalid JSON request body")
})?;
let schema = state
.schema
.read()
.map_err(|_| internal_error("schema lock poisoned"))?
.clone();
if let Err(message) = validate_data(&request.data, &schema) {
warn!(reason = %message, "Schema validation failed");
return Err(api_error(StatusCode::BAD_REQUEST, message));
}
let mut record = ApiResponse {
seq: None,
uuid: Uuid::new_v4(),
date: Utc::now().to_rfc3339(),
data: request.data,
};
let data_json = serde_json::to_string(&record.data)
.map_err(|_| internal_error("serialize data"))?;
{
let db = state
.db
.lock()
.map_err(|_| internal_error("database lock poisoned"))?;
db.execute(
"INSERT INTO records (uuid, date, data) VALUES (?1, ?2, ?3)",
params![record.uuid.to_string(), record.date, data_json],
)
.map_err(|e| {
error!(error = %e, "Failed to insert record");
api_error(StatusCode::INTERNAL_SERVER_ERROR, "Failed to store record")
})?;
record.seq = Some(db.last_insert_rowid());
}
info!(uuid = %record.uuid, seq = ?record.seq, "Record stored");
Ok((StatusCode::CREATED, Json(record)))
}
pub async fn list_records(
State(state): State<AppState>,
) -> Result<Json<Vec<ApiResponse>>, ApiError> {
info!(endpoint = "/api/records", "List request received");
let db = state
.db
.lock()
.map_err(|_| internal_error("database lock poisoned"))?;
let mut statement = db
.prepare("SELECT seq, uuid, date, data FROM records ORDER BY seq DESC LIMIT 100")
.map_err(|e| db_error("prepare list", e))?;
let rows = statement
.query_map([], |row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
))
})
.map_err(|e| db_error("query list", e))?;
let mut records = Vec::new();
for row in rows {
let (seq, uuid, date, data_json) = row.map_err(|e| db_error("read row", e))?;
let uuid = Uuid::parse_str(&uuid).map_err(|_| internal_error("invalid stored UUID"))?;
let data: Value =
serde_json::from_str(&data_json).map_err(|_| internal_error("invalid stored JSON"))?;
records.push(ApiResponse {
seq: Some(seq),
uuid,
date,
data,
});
}
info!(count = records.len(), "Records returned");
Ok(Json(records))
}
pub async fn get_config(State(state): State<AppState>) -> Result<Json<crate::schema::SchemaConfig>, ApiError> {
let schema = state
.schema
.read()
.map_err(|_| internal_error("schema lock poisoned"))?
.clone();
Ok(Json(schema))
}
pub async fn reload_config(State(state): State<AppState>) -> Result<Json<Value>, ApiError> {
info!(path = %state.config_path, "Configuration reload requested");
let new_schema = match load_config(&state.config_path) {
Ok(config) => config,
Err(message) => {
error!(reason = %message, "Configuration reload failed");
return Err(api_error(StatusCode::UNPROCESSABLE_ENTITY, message));
}
};
let mode = match new_schema.mode {
ValidationMode::Strict => "strict",
ValidationMode::Loose => "loose",
};
let field_count = new_schema.fields.len();
{
let mut active = state
.schema
.write()
.map_err(|_| internal_error("schema lock poisoned"))?;
*active = new_schema;
}
info!(mode, fields = field_count, "Configuration reloaded successfully");
Ok(Json(serde_json::json!({
"status": "ok",
"mode": mode,
"fields": field_count
})))
}
#[derive(Debug, Deserialize, Default)]
pub struct StepDownRequest {
pub primary_url: Option<String>,
}
#[derive(Debug, Deserialize)]
pub struct RepointRequest {
pub primary_url: String,
pub reset_source: Option<bool>,
}
#[derive(Debug, Deserialize)]
pub struct SwitchoverRequest {
pub replica_url: String,
pub timeout_secs: Option<u64>,
}
/// Admin endpoint: Demote this primary to read-only replica.
pub async fn step_down_handler(
State(state): State<AppState>,
payload: Option<Json<StepDownRequest>>,
) -> Result<Json<Value>, 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<AppState>,
) -> Result<Json<Value>, 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<AppState>,
Json(req): Json<RepointRequest>,
) -> Result<Json<Value>, 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<AppState>,
Json(req): Json<SwitchoverRequest>,
) -> Result<Json<Value>, 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::<Value>().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,
})))
}
+247 -381
View File
@@ -1,419 +1,285 @@
mod db;
mod handlers;
mod replication;
mod schema;
mod state;
mod tls;
use axum::{
extract::{rejection::JsonRejection, State},
http::StatusCode,
routing::get,
Json, Router,
http::{
header::{AUTHORIZATION, CONTENT_TYPE},
HeaderValue, Method,
},
routing::{get, post},
Router,
};
use chrono::Utc;
use rusqlite::{params, Connection};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use axum_server::tls_rustls::RustlsConfig;
use rusqlite::Connection;
use std::{
fs,
collections::HashMap,
net::SocketAddr,
sync::{Arc, Mutex, RwLock},
time::Duration,
};
use tracing::{error, info, warn};
use tracing::{info, warn};
use uuid::Uuid;
#[derive(Clone)]
struct AppState {
db: Arc<Mutex<Connection>>,
schema: Arc<RwLock<SchemaConfig>>,
config_path: Arc<String>,
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, promote_handler, reload_config, repoint_handler,
step_down_handler, switchover_handler,
},
replication::{replication_changes, replication_status},
schema::load_config,
state::{AppState, Role, SyncConfig, SyncStatus},
tls::{
bootstrap_replica_join, create_mtls_server_config, generate_ca, issue_leaf_cert,
JoinToken,
},
};
fn get_env(pando_name: &str, app_name: &str) -> Result<String, std::env::VarError> {
std::env::var(pando_name).or_else(|_| std::env::var(app_name))
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
enum ValidationMode {
Strict,
Loose,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct FieldSpec {
#[serde(rename = "type")]
data_type: DataType,
#[serde(default, skip_serializing_if = "Option::is_none")]
value: Option<Value>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
enum DataType {
String,
Integer,
Number,
Boolean,
Object,
Array,
Null,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct SchemaConfig {
mode: ValidationMode,
fields: std::collections::HashMap<String, FieldSpec>,
}
#[derive(Debug, Serialize, Deserialize)]
struct ApiResponse {
uuid: Uuid,
date: String,
data: Value,
}
#[derive(Debug, Deserialize)]
struct CreateRequest {
data: Value,
}
type ApiError = (StatusCode, String);
fn api_error(status: StatusCode, message: impl Into<String>) -> ApiError {
(status, message.into())
}
fn internal_error(context: &str) -> ApiError {
error!(context, "Internal server error");
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"Internal server error",
)
}
fn matches_type(value: &Value, expected: &DataType) -> bool {
match expected {
DataType::String => value.is_string(),
DataType::Integer => {
value.as_i64().is_some() || value.as_u64().is_some()
}
DataType::Number => value.is_number(),
DataType::Boolean => value.is_boolean(),
DataType::Object => value.is_object(),
DataType::Array => value.is_array(),
DataType::Null => value.is_null(),
}
}
fn type_name(data_type: &DataType) -> &'static str {
match data_type {
DataType::String => "string",
DataType::Integer => "integer",
DataType::Number => "number",
DataType::Boolean => "boolean",
DataType::Object => "object",
DataType::Array => "array",
DataType::Null => "null",
}
}
fn validate_data(data: &Value, schema: &SchemaConfig) -> Result<(), String> {
let object = data
.as_object()
.ok_or("data must be a JSON object")?;
for (key, spec) in &schema.fields {
let value = object
.get(key)
.ok_or_else(|| format!("Missing required field: {key}"))?;
if !matches_type(value, &spec.data_type) {
return Err(format!(
"Field '{key}' must be of type {}",
type_name(&spec.data_type)
));
}
if let Some(expected) = &spec.value {
if value != expected {
return Err(format!(
"Field '{key}' must equal {}",
expected
));
}
}
}
if matches!(schema.mode, ValidationMode::Strict) {
for key in object.keys() {
if !schema.fields.contains_key(key) {
return Err(format!("Unknown field: {key}"));
}
}
}
Ok(())
}
fn load_config(path: &str) -> Result<SchemaConfig, String> {
let contents = fs::read_to_string(path)
.map_err(|e| format!("Cannot read {path}: {e}"))?;
let config: SchemaConfig = serde_json::from_str(&contents)
.map_err(|e| format!("Invalid configuration JSON: {e}"))?;
if config.fields.is_empty() {
return Err("Configuration must define at least one field".into());
}
Ok(config)
}
async fn get_info() -> Json<ApiResponse> {
info!(endpoint = "/api/now", "Request received");
Json(ApiResponse {
uuid: Uuid::new_v4(),
date: Utc::now().to_rfc3339(),
data: serde_json::json!({
"version": 1,
"message": "Hello from Rust!"
}),
})
}
async fn create_record(
State(state): State<AppState>,
payload: Result<Json<CreateRequest>, JsonRejection>,
) -> Result<(StatusCode, Json<ApiResponse>), ApiError> {
info!(endpoint = "/api/records", "Create request received");
let Json(request) = payload.map_err(|e| {
warn!(error = %e, "Invalid JSON request");
api_error(StatusCode::BAD_REQUEST, "Invalid JSON request body")
})?;
// Take a snapshot; reloads won't affect this request mid-validation.
let schema = state
.schema
.read()
.map_err(|_| internal_error("schema lock poisoned"))?
.clone();
if let Err(message) = validate_data(&request.data, &schema) {
warn!(reason = %message, "Schema validation failed");
return Err(api_error(StatusCode::BAD_REQUEST, message));
}
let record = ApiResponse {
uuid: Uuid::new_v4(),
date: Utc::now().to_rfc3339(),
data: request.data,
};
let data_json = serde_json::to_string(&record.data)
.map_err(|_| internal_error("serialize data"))?;
{
let db = state
.db
.lock()
.map_err(|_| internal_error("database lock poisoned"))?;
db.execute(
"INSERT INTO records (uuid, date, data)
VALUES (?1, ?2, ?3)",
params![
record.uuid.to_string(),
record.date,
data_json,
],
)
.map_err(|e| {
error!(error = %e, "Failed to insert record");
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"Failed to store record",
)
})?;
}
info!(uuid = %record.uuid, "Record stored");
Ok((StatusCode::CREATED, Json(record)))
}
async fn list_records(
State(state): State<AppState>,
) -> Result<Json<Vec<ApiResponse>>, ApiError> {
info!(endpoint = "/api/records", "List request received");
let db = state
.db
.lock()
.map_err(|_| internal_error("database lock poisoned"))?;
let mut statement = db
.prepare(
"SELECT uuid, date, data
FROM records
ORDER BY date DESC
LIMIT 100",
)
.map_err(|e| {
error!(error = %e, "Failed to prepare query");
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"Database query failed",
)
})?;
let rows = statement
.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
))
})
.map_err(|e| {
error!(error = %e, "Failed to query records");
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"Database query failed",
)
})?;
let mut records = Vec::new();
for row in rows {
let (uuid, date, data_json) = row.map_err(|e| {
error!(error = %e, "Failed to read database row");
api_error(
StatusCode::INTERNAL_SERVER_ERROR,
"Database read failed",
)
})?;
let uuid = Uuid::parse_str(&uuid)
.map_err(|_| internal_error("invalid stored UUID"))?;
let data: Value = serde_json::from_str(&data_json)
.map_err(|_| internal_error("invalid stored JSON"))?;
records.push(ApiResponse { uuid, date, data });
}
info!(count = records.len(), "Records returned");
Ok(Json(records))
}
async fn get_config(
State(state): State<AppState>,
) -> Result<Json<SchemaConfig>, ApiError> {
let schema = state
.schema
.read()
.map_err(|_| internal_error("schema lock poisoned"))?
.clone();
Ok(Json(schema))
}
async fn reload_config(
State(state): State<AppState>,
) -> Result<Json<Value>, ApiError> {
info!(path = %state.config_path, "Configuration reload requested");
// Parse and validate the new file before touching active configuration.
let new_schema = match load_config(&state.config_path) {
Ok(config) => config,
Err(message) => {
error!(reason = %message, "Configuration reload failed");
return Err(api_error(
StatusCode::UNPROCESSABLE_ENTITY,
message,
));
}
};
let mode = match new_schema.mode {
ValidationMode::Strict => "strict",
ValidationMode::Loose => "loose",
};
let field_count = new_schema.fields.len();
{
let mut active = state
.schema
.write()
.map_err(|_| internal_error("schema lock poisoned"))?;
*active = new_schema;
}
info!(
mode,
fields = field_count,
"Configuration reloaded successfully"
);
Ok(Json(serde_json::json!({
"status": "ok",
"mode": mode,
"fields": field_count
})))
fn env_parse<T: std::str::FromStr>(pando_name: &str, app_name: &str, default: T) -> T {
get_env(pando_name, app_name)
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(default)
}
#[tokio::main]
async fn main() {
let _ = rustls::crypto::ring::default_provider().install_default();
tracing_subscriber::fmt()
.with_writer(std::io::stdout)
.with_target(false)
.init();
let config_path =
std::env::var("APP_CONFIG").unwrap_or_else(|_| "config.json".into());
info!("Starting Pando 🌳 node...");
let schema = load_config(&config_path)
.unwrap_or_else(|e| panic!("Failed to load configuration: {e}"));
let config_path = get_env("PANDO_CONFIG", "APP_CONFIG").unwrap_or_else(|_| {
if std::path::Path::new("pando.json").exists() {
"pando.json".into()
} else if std::path::Path::new("config.json").exists() {
"config.json".into()
} else {
"pando.json".into()
}
});
let schema = load_config(&config_path).unwrap_or_else(|e| panic!("Failed to load configuration from {config_path}: {e}"));
info!(path = %config_path, fields = schema.fields.len(), "Pando configuration loaded");
let role = match get_env("PANDO_ROLE", "APP_ROLE").as_deref() {
Ok("replica") => Role::Replica,
Ok("primary") | Err(_) => Role::Primary,
Ok(other) => panic!("Role must be 'primary' or 'replica', got '{other}'"),
};
let primary_url = get_env("PANDO_PRIMARY_URL", "APP_PRIMARY_URL").ok();
let repl_token = get_env("PANDO_REPL_TOKEN", "APP_REPL_TOKEN").ok().filter(|t| !t.is_empty());
if role == Role::Replica && (primary_url.is_none() || repl_token.is_none()) {
panic!("Replica requires PANDO_PRIMARY_URL (or APP_PRIMARY_URL) and PANDO_REPL_TOKEN (or APP_REPL_TOKEN)");
}
if repl_token.is_none() {
warn!("Replication token not set: /api/replication/changes is disabled");
}
let sync_cfg = SyncConfig {
poll_interval: Duration::from_secs(env_parse("PANDO_SYNC_INTERVAL_SECS", "APP_SYNC_INTERVAL_SECS", 5)),
max_backoff: Duration::from_secs(env_parse("PANDO_SYNC_MAX_BACKOFF_SECS", "APP_SYNC_MAX_BACKOFF_SECS", 60)),
batch: env_parse("PANDO_SYNC_BATCH", "APP_SYNC_BATCH", 200i64).clamp(1, 1000),
};
let db_path = get_env("PANDO_DB", "APP_DB").unwrap_or_else(|_| {
if std::path::Path::new("pando.db").exists() {
"pando.db".into()
} else if std::path::Path::new("messages.db").exists() {
"messages.db".into()
} else {
"pando.db".into()
}
});
let mut db = Connection::open(&db_path).expect("Failed to open SQLite database");
init_db(&mut db).expect("Failed to initialize database");
let instance_id = load_or_create_instance_id(&db).expect("Failed to load instance id");
// -----------------------------------------------------------------------
// mTLS Credential Initialization & Dynamic Join
// -----------------------------------------------------------------------
let mut ca_cert_pem = meta_get(&db, "tls_ca_pem").ok().flatten();
let mut ca_key_pem = meta_get(&db, "tls_ca_key_pem").ok().flatten();
let mut node_cert_pem = meta_get(&db, "tls_cert_pem").ok().flatten();
let mut node_key_pem = meta_get(&db, "tls_key_pem").ok().flatten();
let join_tokens = Arc::new(RwLock::new(HashMap::<String, JoinToken>::new()));
if role == Role::Primary {
// Initialize CA if not present
if ca_cert_pem.is_none() || ca_key_pem.is_none() {
let (ca_cert, ca_key) = generate_ca().expect("Failed to generate cluster Root CA");
meta_set(&db, "tls_ca_pem", &ca_cert).unwrap();
meta_set(&db, "tls_ca_key_pem", &ca_key).unwrap();
ca_cert_pem = Some(ca_cert);
ca_key_pem = Some(ca_key);
info!("Pando Cluster Root CA initialized");
}
// Initialize Primary Server Certificate
if node_cert_pem.is_none() || node_key_pem.is_none() {
let (cert, key) = issue_leaf_cert(
ca_cert_pem.as_ref().unwrap(),
ca_key_pem.as_ref().unwrap(),
"pando-primary",
vec!["localhost".into(), "127.0.0.1".into()],
)
.expect("Failed to issue primary server certificate");
meta_set(&db, "tls_cert_pem", &cert).unwrap();
meta_set(&db, "tls_key_pem", &key).unwrap();
node_cert_pem = Some(cert);
node_key_pem = Some(key);
info!("Primary server certificate generated");
}
// Preload an initial bootstrap join token if PANDO_JOIN_TOKEN / APP_JOIN_TOKEN is passed
if let Ok(seed_token) = get_env("PANDO_JOIN_TOKEN", "APP_JOIN_TOKEN") {
join_tokens.write().unwrap().insert(
seed_token.clone(),
JoinToken {
token: seed_token,
expires_at: chrono::Utc::now() + chrono::Duration::hours(24),
used: false,
},
);
}
} else {
// Replica: Check if dynamic bootstrap enrollment is needed
if node_cert_pem.is_none() {
// PANDO_JOIN_URL lets the replication source (PANDO_PRIMARY_URL) differ from the
// CA enrollment target (PANDO_JOIN_URL). This is required in chain topologies
// where intermediate replicas serve data but cannot issue certificates.
// Falls back to PANDO_PRIMARY_URL when not set.
let join_url = get_env("PANDO_JOIN_URL", "APP_JOIN_URL")
.ok()
.or_else(|| primary_url.clone())
.expect("PANDO_PRIMARY_URL (or PANDO_JOIN_URL) required for replica enrollment");
let token = get_env("PANDO_JOIN_TOKEN", "APP_JOIN_TOKEN")
.expect("PANDO_JOIN_TOKEN required for initial replica enrollment");
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();
}
}
let last_applied_seq = meta_get(&db, "last_applied_seq")
.ok()
.flatten()
.and_then(|v| v.parse::<i64>().ok())
.unwrap_or(0);
let source_instance_id = meta_get(&db, "source_instance_id")
.ok()
.flatten()
.and_then(|v| Uuid::parse_str(&v).ok());
info!(
path = %config_path,
fields = schema.fields.len(),
"Configuration loaded"
instance_id = %instance_id,
role = role.as_str(),
last_applied_seq,
db = %db_path,
"pando-node initialised"
);
let db = Connection::open("messages.db")
.expect("Failed to open SQLite database");
db.execute_batch(
"CREATE TABLE IF NOT EXISTS records (
uuid TEXT PRIMARY KEY,
date TEXT NOT NULL,
data TEXT NOT NULL
);",
)
.expect("Failed to initialize database");
let state = AppState {
db: Arc::new(Mutex::new(db)),
schema: Arc::new(RwLock::new(schema)),
config_path: Arc::new(config_path),
config_path: Arc::new(config_path.clone()),
instance_id,
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(),
last_applied_seq,
source_instance_id,
primary_head_seq: None,
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: 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 {
state.start_replica_sync();
}
let app = Router::new()
.route("/", get(get_info))
.route("/api/now", get(get_info))
.route("/api/records", get(list_records).post(create_record))
.route("/api/admin/config", get(get_config))
.route(
"/api/admin/reload-config",
axum::routing::post(reload_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:
.route("/api/cluster/ca", get(get_cluster_ca))
.route("/api/cluster/tokens", post(create_token_handler))
.route("/api/cluster/join", post(join_cluster_handler))
.layer(tower_http::compression::CompressionLayer::new())
.with_state(state);
let addr = "127.0.0.1:3000";
let listener = tokio::net::TcpListener::bind(addr)
.await
.expect("Failed to bind address");
let bind_str = get_env("PANDO_BIND", "APP_BIND").unwrap_or_else(|_| "127.0.0.1:3000".into());
let addr: SocketAddr = bind_str.parse().expect("Failed to parse bind address");
info!(address = addr, "HTTP server started");
// Build mTLS ServerConfig
let server_config = create_mtls_server_config(
&ca_cert_pem.expect("CA cert missing"),
&node_cert_pem.expect("Server cert missing"),
&node_key_pem.expect("Server key missing"),
)
.expect("Failed to build mTLS server configuration");
axum::serve(listener, app)
let app = match get_env("PANDO_CORS_ORIGIN", "APP_CORS_ORIGIN") {
Ok(origin) => {
info!(origin = %origin, "CORS enabled");
app.layer(
tower_http::cors::CorsLayer::new()
.allow_origin(origin.parse::<HeaderValue>().expect("invalid CORS origin"))
.allow_methods([Method::GET, Method::POST, Method::OPTIONS])
.allow_headers([CONTENT_TYPE, AUTHORIZATION]),
)
}
Err(_) => app,
};
let rustls_config = RustlsConfig::from_config(Arc::new(server_config));
info!(address = %addr, db = %db_path, config = %config_path, "Pando 🌳 node listening (mTLS HTTPS)");
axum_server::bind_rustls(addr, rustls_config)
.serve(app.into_make_service())
.await
.expect("HTTP server failed");
}
.expect("HTTPS server failed");
}
+411
View File
@@ -0,0 +1,411 @@
//! pando-sync: The replication and data synchronization engine of Pando 🌳.
//!
//! Handles pull-based replication across nodes, monotonic sequence tracking,
//! dynamic peer synchronization, and failover support.
use axum::{
extract::{Query, State},
http::{header::AUTHORIZATION, HeaderMap, StatusCode},
Json,
};
use crate::tls::create_mtls_client;
use chrono::Utc;
use rusqlite::{params, Connection};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::{
sync::{Arc, Mutex},
time::Duration,
};
use tracing::{error, info, warn};
use uuid::Uuid;
use crate::{
db::meta_set,
state::{api_error, db_error, internal_error, ApiError, AppState, SyncConfig, SyncStatus},
};
#[derive(Debug, Serialize, Deserialize)]
pub struct Change {
pub seq: i64,
pub uuid: Uuid,
pub date: String,
pub data: Value,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct ChangesResponse {
pub instance_id: Uuid,
pub head_seq: i64,
pub has_more: bool,
pub changes: Vec<Change>,
}
#[derive(Debug, Deserialize)]
pub struct ChangesQuery {
pub after: Option<i64>,
pub limit: Option<i64>,
}
enum SyncError {
Transient(String),
Fatal(String),
}
fn ct_eq(a: &[u8], b: &[u8]) -> bool {
a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
}
fn check_token(headers: &HeaderMap, expected: &Option<Arc<String>>) -> Result<(), ApiError> {
let Some(expected) = expected else {
return Err(api_error(
StatusCode::FORBIDDEN,
"Replication is disabled (no token configured)",
));
};
let provided = headers
.get(AUTHORIZATION)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.strip_prefix("Bearer "))
.unwrap_or("");
if !ct_eq(provided.as_bytes(), expected.as_bytes()) {
return Err(api_error(StatusCode::UNAUTHORIZED, "Invalid token"));
}
Ok(())
}
pub async fn replication_changes(
State(state): State<AppState>,
headers: HeaderMap,
Query(query): Query<ChangesQuery>,
) -> Result<Json<ChangesResponse>, ApiError> {
check_token(&headers, &state.repl_token)?;
let after = query.after.unwrap_or(0).max(0);
let limit = query.limit.unwrap_or(200).clamp(1, 1000);
let db = state
.db
.lock()
.map_err(|_| internal_error("database lock poisoned"))?;
let head_seq: i64 = db
.query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| {
r.get(0)
})
.map_err(|e| db_error("head seq", e))?;
let mut statement = db
.prepare(
"SELECT seq, uuid, date, data
FROM records
WHERE seq > ?1
ORDER BY seq ASC
LIMIT ?2",
)
.map_err(|e| db_error("prepare changes", e))?;
let rows = statement
.query_map(params![after, limit], |row| {
Ok((
row.get::<_, i64>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
))
})
.map_err(|e| db_error("query changes", e))?;
let mut changes = Vec::new();
for row in rows {
let (seq, uuid, date, data_json) = row.map_err(|e| db_error("read change", e))?;
changes.push(Change {
seq,
uuid: Uuid::parse_str(&uuid).map_err(|_| internal_error("invalid stored UUID"))?,
date,
data: serde_json::from_str(&data_json)
.map_err(|_| internal_error("invalid stored JSON"))?,
});
}
let last = changes.last().map(|c| c.seq).unwrap_or(after);
Ok(Json(ChangesResponse {
instance_id: state.instance_id,
head_seq,
has_more: last < head_seq,
changes,
}))
}
pub async fn replication_status(State(state): State<AppState>) -> Result<Json<Value>, ApiError> {
let status = state
.sync
.read()
.map_err(|_| internal_error("sync status lock poisoned"))?
.clone();
let local_head: i64 = {
let db = state
.db
.lock()
.map_err(|_| internal_error("database lock poisoned"))?;
db.query_row("SELECT COALESCE(MAX(seq), 0) FROM records", [], |r| {
r.get(0)
})
.map_err(|e| db_error("head seq", e))?
};
let lag = status
.primary_head_seq
.map(|head| (head - status.last_applied_seq).max(0));
let current_role = state.get_role();
let primary_url = state.get_primary_url();
Ok(Json(serde_json::json!({
"instance_id": state.instance_id,
"role": current_role.as_str(),
"primary_url": primary_url,
"local_head_seq": local_head,
"lag_records": lag,
"sync": status,
})))
}
fn jitter(d: Duration) -> Duration {
let ms = d.as_millis() as u64;
let extra = (Uuid::new_v4().as_u128() as u64) % (ms / 2 + 1);
Duration::from_millis(ms / 2 + extra)
}
fn update_status(state: &AppState, f: impl FnOnce(&mut SyncStatus)) {
if let Ok(mut s) = state.sync.write() {
f(&mut s);
}
}
fn apply_batch(
db: &Arc<Mutex<Connection>>,
source: Uuid,
changes: &[Change],
) -> Result<i64, String> {
let mut conn = db.lock().map_err(|_| "database lock poisoned".to_string())?;
let tx = conn.transaction().map_err(|e| e.to_string())?;
let mut max_seq = 0;
{
let mut insert = tx
.prepare_cached(
"INSERT OR IGNORE INTO records (seq, uuid, date, data) VALUES (?1, ?2, ?3, ?4)",
)
.map_err(|e| e.to_string())?;
for c in changes {
let data = serde_json::to_string(&c.data).map_err(|e| e.to_string())?;
insert
.execute(params![c.seq, c.uuid.to_string(), c.date, data])
.map_err(|e| e.to_string())?;
max_seq = max_seq.max(c.seq);
}
}
meta_set(&tx, "last_applied_seq", &max_seq.to_string()).map_err(|e| e.to_string())?;
meta_set(&tx, "source_instance_id", &source.to_string()).map_err(|e| e.to_string())?;
tx.commit().map_err(|e| e.to_string())?;
Ok(max_seq)
}
async fn sync_once(
client: &reqwest::Client,
state: &AppState,
primary_url: &str,
token: &str,
batch: i64,
) -> Result<bool, SyncError> {
let (last, known_source) = {
let s = state
.sync
.read()
.map_err(|_| SyncError::Fatal("sync status lock poisoned".into()))?;
(s.last_applied_seq, s.source_instance_id)
};
let url = format!("{}/api/replication/changes", primary_url.trim_end_matches('/'));
let resp = client
.get(&url)
.bearer_auth(token)
.query(&[("after", last), ("limit", batch)])
.send()
.await
.map_err(|e| SyncError::Transient(format!("request failed: {e}")))?;
if !resp.status().is_success() {
return Err(SyncError::Transient(format!(
"primary returned {}",
resp.status()
)));
}
let body: ChangesResponse = resp
.json()
.await
.map_err(|e| SyncError::Transient(format!("bad response body: {e}")))?;
if body.instance_id == state.instance_id {
return Err(SyncError::Fatal(
"source has the same instance_id as this replica (cloned database?)".into(),
));
}
if let Some(known) = known_source {
if known != body.instance_id {
return Err(SyncError::Fatal(format!(
"source instance changed: expected {known}, got {}",
body.instance_id
)));
}
}
if body.head_seq < last {
return Err(SyncError::Fatal(format!(
"source head_seq {} is behind our cursor {last} (source restored from older backup?)",
body.head_seq
)));
}
if body.changes.is_empty() {
update_status(state, |s| {
s.state = "idle".into();
s.primary_head_seq = Some(body.head_seq);
s.last_success = Some(Utc::now().to_rfc3339());
s.last_error = None;
s.source_instance_id = Some(body.instance_id);
});
return Ok(false);
}
let mut prev = last;
for c in &body.changes {
if c.seq <= prev {
return Err(SyncError::Transient(format!(
"non-monotonic seq {} after {prev}",
c.seq
)));
}
prev = c.seq;
}
let count = body.changes.len();
let db = state.db.clone();
let source = body.instance_id;
let changes = body.changes;
let max_seq = tokio::task::spawn_blocking(move || apply_batch(&db, source, &changes))
.await
.map_err(|e| SyncError::Transient(format!("apply task failed: {e}")))?
.map_err(|e| SyncError::Transient(format!("apply failed: {e}")))?;
update_status(state, |s| {
s.state = if body.has_more { "syncing" } else { "idle" }.into();
s.last_applied_seq = max_seq;
s.source_instance_id = Some(source);
s.primary_head_seq = Some(body.head_seq);
s.last_success = Some(Utc::now().to_rfc3339());
s.last_error = None;
});
info!(
applied = count,
last_applied_seq = max_seq,
head = body.head_seq,
"Replication batch applied"
);
Ok(body.has_more)
}
pub async fn replica_loop(
state: AppState,
primary_url: String,
token: String,
cfg: SyncConfig,
ca_pem: String,
cert_pem: String,
key_pem: String,
mut cancel_rx: tokio::sync::watch::Receiver<bool>,
) {
let client = match create_mtls_client(&ca_pem, &cert_pem, &key_pem) {
Ok(c) => c,
Err(e) => {
error!(error = %e, "Cannot build mTLS HTTP client; replication halted");
return;
}
};
let mut backoff = Duration::from_secs(1);
info!(primary = %primary_url, "Replica mTLS sync loop started");
loop {
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;
}
}
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;
}
}
}
}
}
}
+105
View File
@@ -0,0 +1,105 @@
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::{collections::HashMap, fs};
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum ValidationMode {
Strict,
Loose,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FieldSpec {
#[serde(rename = "type")]
pub data_type: DataType,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub value: Option<Value>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum DataType {
String,
Integer,
Number,
Boolean,
Object,
Array,
Null,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SchemaConfig {
pub mode: ValidationMode,
pub fields: HashMap<String, FieldSpec>,
}
pub fn matches_type(value: &Value, expected: &DataType) -> bool {
match expected {
DataType::String => value.is_string(),
DataType::Integer => value.as_i64().is_some() || value.as_u64().is_some(),
DataType::Number => value.is_number(),
DataType::Boolean => value.is_boolean(),
DataType::Object => value.is_object(),
DataType::Array => value.is_array(),
DataType::Null => value.is_null(),
}
}
pub fn type_name(data_type: &DataType) -> &'static str {
match data_type {
DataType::String => "string",
DataType::Integer => "integer",
DataType::Number => "number",
DataType::Boolean => "boolean",
DataType::Object => "object",
DataType::Array => "array",
DataType::Null => "null",
}
}
pub fn validate_data(data: &Value, schema: &SchemaConfig) -> Result<(), String> {
let object = data.as_object().ok_or("data must be a JSON object")?;
for (key, spec) in &schema.fields {
let value = object
.get(key)
.ok_or_else(|| format!("Missing required field: {key}"))?;
if !matches_type(value, &spec.data_type) {
return Err(format!(
"Field '{key}' must be of type {}",
type_name(&spec.data_type)
));
}
if let Some(expected) = &spec.value {
if value != expected {
return Err(format!("Field '{key}' must equal {expected}"));
}
}
}
if matches!(schema.mode, ValidationMode::Strict) {
for key in object.keys() {
if !schema.fields.contains_key(key) {
return Err(format!("Unknown field: {key}"));
}
}
}
Ok(())
}
pub fn load_config(path: &str) -> Result<SchemaConfig, String> {
let contents = fs::read_to_string(path).map_err(|e| format!("Cannot read {path}: {e}"))?;
let config: SchemaConfig =
serde_json::from_str(&contents).map_err(|e| format!("Invalid configuration JSON: {e}"))?;
if config.fields.is_empty() {
return Err("Configuration must define at least one field".into());
}
Ok(config)
}
+191
View File
@@ -0,0 +1,191 @@
use axum::http::StatusCode;
use rusqlite::Connection;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::{
collections::HashMap,
sync::{Arc, Mutex, RwLock},
time::Duration,
};
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<Mutex<Connection>>,
pub schema: Arc<RwLock<SchemaConfig>>,
pub config_path: Arc<String>,
pub instance_id: Uuid,
pub role: Arc<RwLock<Role>>,
pub primary_url: Arc<RwLock<Option<String>>>,
pub repl_token: Option<Arc<String>>,
pub sync: Arc<RwLock<SyncStatus>>,
// Replication task management:
pub sync_cancel: Arc<Mutex<Option<tokio::sync::watch::Sender<bool>>>>,
pub sync_cfg: SyncConfig,
// mTLS & Discovery additions:
pub ca_cert_pem: Option<Arc<String>>,
pub ca_key_pem: Arc<RwLock<Option<String>>>,
pub node_cert_pem: Option<Arc<String>>,
pub node_key_pem: Option<Arc<String>>,
pub join_tokens: Arc<RwLock<HashMap<String, JoinToken>>>,
}
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<String> {
self.primary_url.read().unwrap().clone()
}
pub fn set_primary_url(&self, url: Option<String>) {
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,
}
impl Role {
pub fn as_str(self) -> &'static str {
match self {
Role::Primary => "primary",
Role::Replica => "replica",
}
}
}
#[derive(Debug, Clone, Serialize)]
pub struct SyncStatus {
pub state: String,
pub last_applied_seq: i64,
pub source_instance_id: Option<Uuid>,
pub primary_head_seq: Option<i64>,
pub last_success: Option<String>,
pub last_error: Option<String>,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct ApiResponse {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub seq: Option<i64>,
pub uuid: Uuid,
pub date: String,
pub data: Value,
}
#[derive(Debug, Deserialize)]
pub struct CreateRequest {
pub data: Value,
}
pub type ApiError = (StatusCode, String);
pub fn api_error(status: StatusCode, message: impl Into<String>) -> ApiError {
(status, message.into())
}
pub fn internal_error(context: &str) -> ApiError {
error!(context, "Internal server error");
api_error(StatusCode::INTERNAL_SERVER_ERROR, "Internal server error")
}
pub fn db_error(context: &str, e: rusqlite::Error) -> ApiError {
error!(error = %e, context, "Database error");
api_error(StatusCode::INTERNAL_SERVER_ERROR, "Database error")
}
+224
View File
@@ -0,0 +1,224 @@
use std::sync::Arc;
use chrono::{DateTime, Duration as ChronoDuration, Utc};
use rcgen::{
BasicConstraints, CertificateParams, DnType, ExtendedKeyUsagePurpose, IsCa,
Issuer, KeyPair, KeyUsagePurpose,
};
use rusqlite::Connection;
use rustls::server::WebPkiClientVerifier;
use rustls::RootCertStore;
use rustls_pki_types::{CertificateDer, PrivateKeyDer};
use serde::{Deserialize, Serialize};
use tracing::info;
use uuid::Uuid;
use crate::db::{meta_set};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JoinToken {
pub token: String,
pub expires_at: DateTime<Utc>,
pub used: bool,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct JoinClusterRequest {
pub token: String,
pub node_id: Uuid,
#[serde(default)]
pub san_list: Vec<String>,
}
#[derive(Debug, Serialize, Deserialize)]
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<String>,
}
// ---------------------------------------------------------------------------
// Certificate Generation
// ---------------------------------------------------------------------------
pub fn generate_ca() -> Result<(String, String), String> {
let ca_key = KeyPair::generate_for(&rcgen::PKCS_ECDSA_P256_SHA256)
.map_err(|e| format!("CA key error: {e}"))?;
let mut ca_params = CertificateParams::new(vec![])
.map_err(|e| format!("CA params error: {e}"))?;
ca_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
ca_params.distinguished_name.push(DnType::CommonName, "Pando Cluster Root CA");
ca_params.key_usages = vec![KeyUsagePurpose::KeyCertSign, KeyUsagePurpose::CrlSign];
let ca_cert = ca_params
.self_signed(&ca_key)
.map_err(|e| format!("CA self-sign error: {e}"))?;
Ok((ca_cert.pem(), ca_key.serialize_pem()))
}
pub fn issue_leaf_cert(
ca_cert_pem: &str,
ca_key_pem: &str,
common_name: &str,
san_list: Vec<String>,
) -> Result<(String, String), String> {
let ca_key = KeyPair::from_pem(ca_key_pem)
.map_err(|e| format!("Failed to parse CA private key: {e}"))?;
// In rcgen 0.14+, from_ca_cert_pem belongs to Issuer:
let issuer = Issuer::from_ca_cert_pem(ca_cert_pem, &ca_key)
.map_err(|e| format!("Failed to parse CA cert/create issuer: {e}"))?;
let leaf_key = KeyPair::generate_for(&rcgen::PKCS_ECDSA_P256_SHA256)
.map_err(|e| format!("Leaf key error: {e}"))?;
let mut leaf_params = CertificateParams::new(san_list)
.map_err(|e| format!("Leaf params error: {e}"))?;
leaf_params.distinguished_name.push(DnType::CommonName, common_name);
leaf_params.use_authority_key_identifier_extension = true;
leaf_params.key_usages = vec![KeyUsagePurpose::DigitalSignature];
leaf_params.extended_key_usages = vec![
ExtendedKeyUsagePurpose::ServerAuth,
ExtendedKeyUsagePurpose::ClientAuth,
];
let leaf_cert = leaf_params
.signed_by(&leaf_key, &issuer)
.map_err(|e| format!("Leaf sign error: {e}"))?;
Ok((leaf_cert.pem(), leaf_key.serialize_pem()))
}
// ---------------------------------------------------------------------------
// Rustls Server & Reqwest Client Configuration
// ---------------------------------------------------------------------------
fn parse_certs(pem: &str) -> Result<Vec<CertificateDer<'static>>, String> {
let mut reader = pem.as_bytes();
let certs: Result<Vec<_>, _> = rustls_pemfile::certs(&mut reader).collect();
certs.map_err(|e| format!("failed to parse certs: {e}"))
}
fn parse_key(pem: &str) -> Result<PrivateKeyDer<'static>, String> {
let mut reader = pem.as_bytes();
rustls_pemfile::private_key(&mut reader)
.map_err(|e| format!("failed to parse private key: {e}"))?
.ok_or_else(|| "no private key found in pem".to_string())
}
pub fn create_mtls_server_config(
ca_pem: &str,
server_cert_pem: &str,
server_key_pem: &str,
) -> Result<rustls::ServerConfig, String> {
let mut root_store = RootCertStore::empty();
for cert in parse_certs(ca_pem)? {
root_store
.add(cert)
.map_err(|e| format!("failed to add root cert: {e}"))?;
}
// allow_unauthenticated permits anonymous TLS connections (e.g., initial join requests)
// while enforcing valid CA signatures on any presented client certificates.
let client_verifier = WebPkiClientVerifier::builder(Arc::new(root_store))
.allow_unauthenticated()
.build()
.map_err(|e| format!("failed to build client verifier: {e}"))?;
let cert_chain = parse_certs(server_cert_pem)?;
let key = parse_key(server_key_pem)?;
let mut config = rustls::ServerConfig::builder()
.with_client_cert_verifier(client_verifier)
.with_single_cert(cert_chain, key)
.map_err(|e| format!("failed to build ServerConfig: {e}"))?;
config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
Ok(config)
}
pub fn create_mtls_client(
ca_pem: &str,
client_cert_pem: &str,
client_key_pem: &str,
) -> Result<reqwest::Client, String> {
let root_cert = reqwest::Certificate::from_pem(ca_pem.as_bytes())
.map_err(|e| format!("failed to load CA pem for client: {e}"))?;
let identity_pem = format!("{client_cert_pem}\n{client_key_pem}");
let identity = reqwest::Identity::from_pem(identity_pem.as_bytes())
.map_err(|e| format!("failed to create client identity: {e}"))?;
reqwest::Client::builder()
.use_rustls_tls()
.add_root_certificate(root_cert)
.identity(identity)
.connect_timeout(std::time::Duration::from_secs(10))
.timeout(std::time::Duration::from_secs(60))
.build()
.map_err(|e| format!("failed to build reqwest client: {e}"))
}
// ---------------------------------------------------------------------------
// Dynamic Enrollment / Join Helpers
// ---------------------------------------------------------------------------
pub fn create_join_token(ttl_secs: u64) -> JoinToken {
JoinToken {
token: Uuid::new_v4().to_string().replace('-', ""),
expires_at: Utc::now() + ChronoDuration::seconds(ttl_secs as i64),
used: false,
}
}
pub async fn bootstrap_replica_join(
primary_url: &str,
token: &str,
node_id: Uuid,
db: &Connection,
) -> Result<(), String> {
let base = primary_url.trim_end_matches('/');
// 1. Initial bootstrap client without mTLS client cert
let bootstrap_client = reqwest::Client::builder()
.danger_accept_invalid_certs(true)
.build()
.map_err(|e| format!("bootstrap client build failed: {e}"))?;
// 2. Request enrollment
let join_req = JoinClusterRequest {
token: token.to_string(),
node_id,
san_list: vec!["localhost".into(), "127.0.0.1".into()],
};
let resp = bootstrap_client
.post(format!("{base}/api/cluster/join"))
.json(&join_req)
.send()
.await
.map_err(|e| format!("join request failed: {e}"))?;
if !resp.status().is_success() {
return Err(format!("join rejected with HTTP {}", resp.status()));
}
let payload: JoinClusterResponse = resp
.json()
.await
.map_err(|e| format!("failed to decode join response: {e}"))?;
// 3. Persist credentials locally
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(())
}