From 3d819c7b106138a9d1de3b0dc2dfda9150c2d7d3 Mon Sep 17 00:00:00 2001 From: Kalzu Rekku Date: Sun, 25 Oct 2026 18:54:13 +0200 Subject: [PATCH] Replication! --- Cargo.toml | 2 + README.md | 234 +++++++++++++++-- config.json | 14 -- messages.db | Bin 12288 -> 0 bytes repl-test.sh | 125 +++++++++ src/main.rs | 700 +++++++++++++++++++++++++++++++++++++++++++++++---- 6 files changed, 1004 insertions(+), 71 deletions(-) delete mode 100644 messages.db create mode 100755 repl-test.sh diff --git a/Cargo.toml b/Cargo.toml index a2250eb..39a3893 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,10 +6,12 @@ edition = "2024" [dependencies] axum = { version = "0.8.9", features = ["json"] } chrono = { version = "0.4.45", features = ["serde"] } +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "gzip"] } rusqlite = { version = "0.41.0", features = ["bundled"] } 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"] } tracing = "0.1.44" tracing-subscriber = "0.3.23" uuid = { version = "1.29.0", features = ["v4", "serde"] } diff --git a/README.md b/README.md index e5a89fa..b54e4ac 100644 --- a/README.md +++ b/README.md @@ -1,9 +1,11 @@ # 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 +17,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 (`messages.db`). +- **Read Replication**: + - Pull-based and cursor-driven: the replica asks for everything after its last `seq`. + - Append-only journal with a global tick (`seq`) that is preserved on replicas. + - Permanent per-instance ID, used to detect wrong, cloned or restored sources. + - Atomic, idempotent batch apply: safe across crashes, retries and duplicate deliveries. + - Exponential backoff with jitter, small batches and gzip for flaky or slow links. + - Replicas can serve replication to other replicas (relay). - **Structured Logging**: Built with `tracing` outputting to standard output. --- @@ -25,13 +34,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 +71,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 +80,21 @@ The service looks for a configuration file at the path specified by the `APP_CON | `fields..type` | `string` | Expected JSON type: `string`, `integer`, `number`, `boolean`, `object`, `array`, `null`. | | `fields..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` | `config.json` | Path to the schema file. | +| `APP_BIND` | `127.0.0.1:3000` | Listen address. Change this for a primary that replicas must reach over a network. | +| `APP_ROLE` | `primary` | `primary` or `replica`. | +| `APP_PRIMARY_URL` | *none* | Base URL of the source, e.g. `https://primary.example:3000`. **Required for a replica.** | +| `APP_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 @@ -96,6 +131,33 @@ To specify a custom configuration file path: APP_CONFIG=custom_config.json cargo run ``` +### 2. Run a primary and a replica + +`messages.db` is opened relative to the working directory, so run each instance from its own directory. + +```bash +# Primary +mkdir -p primary && cp config.json primary/ && cd primary +APP_ROLE=primary \ +APP_BIND=0.0.0.0:3000 \ +APP_REPL_TOKEN=change-me \ + ../target/debug/rust_test + +# Replica (in another terminal; start with an EMPTY database) +mkdir -p replica && cp config.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/rust_test +``` + +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 +182,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 +195,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 +204,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 +270,121 @@ APP_CONFIG=custom_config.json cargo run --- +### Replication: Pull Changes +- **Endpoint:** `GET /api/replication/changes?after=&limit=` +- **Headers:** `Authorization: Bearer ` +- **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=&limit=`. +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 `messages.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 `messages.db`, because the copy shares the primary's instance ID and the replica will refuse to sync. +- **Transport security:** replication uses a bearer token over whatever transport you give it. Put TLS in front of the primary (reverse proxy) for anything beyond a trusted network. +- **Chaining:** a replica serves `/api/replication/changes` too (if it has a token), so replicas can follow other replicas. Downstream replicas see the upstream replica's instance ID as their source. +- **Promotion:** a replica holds the same `seq` values as its source, so it can be restarted with `APP_ROLE=primary` to take over. Note that its instance ID differs from the old primary's, so other replicas will halt on an ID mismatch until they are reset against the new primary. +- **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 +409,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 `messages.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 diff --git a/config.json b/config.json index ae3893e..7a9aefa 100644 --- a/config.json +++ b/config.json @@ -1,17 +1,3 @@ - -{ - "mode": "strict", - "fields": { - "version": { - "type": "integer", - "value": 2 - }, - "message": { - "type": "string" - } - } -} - { "mode": "strict", "fields": { diff --git a/messages.db b/messages.db deleted file mode 100644 index 589f3af1acedcbc6d82e53a3e232bfa2b6462c86..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 12288 zcmWFz^vNtqRY=P(%1ta$FlG>7U}R))P*7lCU|?ZjU|?oI07j@7Oa#P-v2oEX40hQMeDjE2By2#kinXb6mkz-S1JhQMeDjE2By2n>o4 zXq07S7nhc1Y!WO&p$-LFVx2eLp~8z9QJRqv9u%~%#VjUDc%su0QKuRK;g;@>g4maGVr(a^YOKUL`Lyw2#kinXb6mk zz-S1JhQMeDjE2By2#kinXb6mkz-S22CTEb3t-N`S~SEI!a}!Ma7x< zc}iA>wdG7Kp~0ypmWCE-#)(O~W`>5Ax+WGTM!FUzCWg9}76vJ)2F9iaX(@?l&M>et zFwrwLG&D80G&Zy((izAWGlDEmF-S@@Ha0WRH8L_t(KRtLG}E<6F*emTGf6fwwMa>^ zOfxh_wb&2; exit 1; } + +# Each instance gets its own directory: messages.db is opened relative to cwd. +rm -rf "$WORK" +mkdir -p "$WORK/primary" "$WORK/replica" + +for d in primary replica; do + cat > "$WORK/$d/config.json" <<'EOF' +{ + "mode": "loose", + "fields": { + "msg": { "type": "string" } + } +} +EOF +done + +PIDS=() +cleanup() { + trap - EXIT INT TERM + echo + echo "Stopping instances..." + for pid in "${PIDS[@]}"; do kill "$pid" 2>/dev/null || true; done + wait 2>/dev/null || true +} +trap cleanup EXIT INT TERM + +start() { # name port extra-env... + local name=$1 port=$2 + shift 2 + ( cd "$WORK/$name" && exec env \ + APP_CONFIG=config.json \ + APP_BIND="127.0.0.1:$port" \ + APP_REPL_TOKEN="$TOKEN" \ + APP_SYNC_INTERVAL_SECS=1 \ + "$@" "$BIN" ) >"$WORK/$name/log.txt" 2>&1 & + PIDS+=("$!") +} + +wait_up() { # port + for _ in $(seq 1 50); do + curl -sf "http://127.0.0.1:$1/api/now" >/dev/null 2>&1 && return 0 + sleep 0.1 + done + echo "Instance on port $1 did not come up; see logs in $WORK" >&2 + exit 1 +} + +echo "Starting primary on :$PRIMARY_PORT" +start primary "$PRIMARY_PORT" APP_ROLE=primary +wait_up "$PRIMARY_PORT" + +echo "Starting replica on :$REPLICA_PORT" +start replica "$REPLICA_PORT" \ + APP_ROLE=replica \ + APP_PRIMARY_URL="http://127.0.0.1:$PRIMARY_PORT" +wait_up "$REPLICA_PORT" + +echo +echo "Writing 3 records to the primary" +for i in 1 2 3; do + curl -s -X POST "http://127.0.0.1:$PRIMARY_PORT/api/records" \ + -H 'Content-Type: application/json' \ + -d "{\"data\":{\"msg\":\"hello $i\"}}" + echo +done + +echo +echo "Waiting for the replica to catch up..." +caught_up=0 +for _ in $(seq 1 30); do + if curl -s "http://127.0.0.1:$REPLICA_PORT/api/replication/status" \ + | grep -q '"last_applied_seq":3'; then + caught_up=1 + break + fi + sleep 0.5 +done + +echo +echo "Replica status:" +curl -s "http://127.0.0.1:$REPLICA_PORT/api/replication/status" +echo +echo +echo "Replica records:" +curl -s "http://127.0.0.1:$REPLICA_PORT/api/records" +echo +echo +echo "Write to the replica (should be rejected):" +curl -s -w ' [HTTP %{http_code}]\n' -X POST "http://127.0.0.1:$REPLICA_PORT/api/records" \ + -H 'Content-Type: application/json' \ + -d '{"data":{"msg":"nope"}}' + +echo +if [[ $caught_up -eq 1 ]]; then + echo "OK: replica caught up." +else + echo "WARNING: replica did not reach seq 3 in time; check $WORK/replica/log.txt" +fi + +cat <>, schema: Arc>, config_path: Arc, + + /// Permanent per-instance identity (stored in the `meta` table). + instance_id: Uuid, + role: Role, + primary_url: Option>, + /// Shared secret for /api/replication/*. None = replication endpoint disabled. + repl_token: Option>, + sync: Arc>, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Role { + Primary, + Replica, +} + +impl Role { + fn as_str(self) -> &'static str { + match self { + Role::Primary => "primary", + Role::Replica => "replica", + } + } +} + +#[derive(Debug, Clone, Serialize)] +struct SyncStatus { + /// "primary" | "idle" | "syncing" | "retrying" | "halted" + state: String, + /// Last global tick (seq) applied locally. Replica's cursor. + last_applied_seq: i64, + source_instance_id: Option, + /// Primary's head seq as of the last successful pull. + primary_head_seq: Option, + last_success: Option, + last_error: Option, +} + +// --------------------------------------------------------------------------- +// Schema validation (unchanged) +// --------------------------------------------------------------------------- + #[derive(Debug, Clone, Copy, Serialize, Deserialize)] #[serde(rename_all = "lowercase")] enum ValidationMode { @@ -58,6 +103,9 @@ struct SchemaConfig { #[derive(Debug, Serialize, Deserialize)] struct ApiResponse { + /// Global tick. Absent on endpoints that aren't records. + #[serde(default, skip_serializing_if = "Option::is_none")] + seq: Option, uuid: Uuid, date: String, data: Value, @@ -82,6 +130,11 @@ fn internal_error(context: &str) -> ApiError { ) } +fn db_error(context: &str, e: rusqlite::Error) -> ApiError { + error!(error = %e, context, "Database error"); + api_error(StatusCode::INTERNAL_SERVER_ERROR, "Database error") +} + fn matches_type(value: &Value, expected: &DataType) -> bool { match expected { DataType::String => value.is_string(), @@ -160,10 +213,107 @@ fn load_config(path: &str) -> Result { Ok(config) } +// --------------------------------------------------------------------------- +// Database init / meta +// --------------------------------------------------------------------------- + +fn meta_get(db: &Connection, key: &str) -> rusqlite::Result> { + db.query_row("SELECT value FROM meta WHERE key = ?1", [key], |r| { + r.get::<_, String>(0) + }) + .optional() +} + +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(()) +} + +/// Creates tables and migrates a pre-replication `records` table +/// (uuid PK, no seq) to the journal layout (seq AUTOINCREMENT + uuid UNIQUE). +/// +/// `records` is insert-only, so it *is* the append-only journal; `seq` is the +/// global tick. AUTOINCREMENT guarantees seq never gets reused, even after +/// a rollback. +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(()) +} + +fn load_or_create_instance_id(db: &Connection) -> rusqlite::Result { + 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) +} + +// --------------------------------------------------------------------------- +// Regular handlers +// --------------------------------------------------------------------------- + async fn get_info() -> Json { info!(endpoint = "/api/now", "Request received"); Json(ApiResponse { + seq: None, uuid: Uuid::new_v4(), date: Utc::now().to_rfc3339(), data: serde_json::json!({ @@ -179,6 +329,16 @@ async fn create_record( ) -> Result<(StatusCode, Json), ApiError> { info!(endpoint = "/api/records", "Create request received"); + if state.role == Role::Replica { + return Err(api_error( + StatusCode::FORBIDDEN, + format!( + "This instance is a read-only replica; write to {}", + state.primary_url.as_deref().map(String::as_str).unwrap_or("the primary") + ), + )); + } + let Json(request) = payload.map_err(|e| { warn!(error = %e, "Invalid JSON request"); api_error(StatusCode::BAD_REQUEST, "Invalid JSON request body") @@ -196,7 +356,8 @@ async fn create_record( return Err(api_error(StatusCode::BAD_REQUEST, message)); } - let record = ApiResponse { + let mut record = ApiResponse { + seq: None, uuid: Uuid::new_v4(), date: Utc::now().to_rfc3339(), data: request.data, @@ -227,9 +388,11 @@ async fn create_record( "Failed to store record", ) })?; + + record.seq = Some(db.last_insert_rowid()); } - info!(uuid = %record.uuid, "Record stored"); + info!(uuid = %record.uuid, seq = ?record.seq, "Record stored"); Ok((StatusCode::CREATED, Json(record))) } @@ -246,45 +409,29 @@ async fn list_records( let mut statement = db .prepare( - "SELECT uuid, date, data + "SELECT seq, uuid, date, data FROM records - ORDER BY date DESC + ORDER BY seq DESC LIMIT 100", ) - .map_err(|e| { - error!(error = %e, "Failed to prepare query"); - api_error( - StatusCode::INTERNAL_SERVER_ERROR, - "Database query failed", - ) - })?; + .map_err(|e| db_error("prepare list", e))?; let rows = statement .query_map([], |row| { Ok(( - row.get::<_, String>(0)?, + row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, + row.get::<_, String>(3)?, )) }) - .map_err(|e| { - error!(error = %e, "Failed to query records"); - api_error( - StatusCode::INTERNAL_SERVER_ERROR, - "Database query failed", - ) - })?; + .map_err(|e| db_error("query list", e))?; 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 (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"))?; @@ -292,7 +439,12 @@ async fn list_records( let data: Value = serde_json::from_str(&data_json) .map_err(|_| internal_error("invalid stored JSON"))?; - records.push(ApiResponse { uuid, date, data }); + records.push(ApiResponse { + seq: Some(seq), + uuid, + date, + data, + }); } info!(count = records.len(), "Records returned"); @@ -358,6 +510,408 @@ async fn reload_config( }))) } +// --------------------------------------------------------------------------- +// Replication: serving side (primary, or a replica acting as a relay) +// --------------------------------------------------------------------------- + +#[derive(Debug, Serialize, Deserialize)] +struct Change { + seq: i64, + uuid: Uuid, + date: String, + data: Value, +} + +#[derive(Debug, Serialize, Deserialize)] +struct ChangesResponse { + instance_id: Uuid, + head_seq: i64, + has_more: bool, + changes: Vec, +} + +#[derive(Debug, Deserialize)] +struct ChangesQuery { + after: Option, + limit: Option, +} + +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>, +) -> 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(()) +} + +/// Cursor-based pull. Stateless on the server: the replica says "give me +/// everything after seq N", so dropped connections, duplicate requests and +/// restarts on either side are all harmless. +async fn replication_changes( + State(state): State, + headers: HeaderMap, + Query(query): Query, +) -> Result, 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, + })) +} + +async fn replication_status( + State(state): State, +) -> Result, 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)); + + Ok(Json(serde_json::json!({ + "instance_id": state.instance_id, + "role": state.role.as_str(), + "primary_url": state.primary_url.as_deref(), + "local_head_seq": local_head, + "lag_records": lag, + "sync": status, + }))) +} + +// --------------------------------------------------------------------------- +// Replication: replica side +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone)] +struct SyncConfig { + poll_interval: Duration, + max_backoff: Duration, + batch: i64, +} + +enum SyncError { + /// Network trouble, 5xx, truncated body... retry with backoff. + Transient(String), + /// Identity/history mismatch. Retrying won't help; needs an operator. + Fatal(String), +} + +fn jitter(d: Duration) -> Duration { + // Uniform in [d/2, d]; avoids a fleet of replicas retrying in lockstep. + 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); + } +} + +/// Applies a batch atomically: rows + cursor + source identity commit together, +/// so a crash can never leave the cursor ahead of (or behind) the data. +/// INSERT OR IGNORE makes re-delivery idempotent. +fn apply_batch( + db: &Arc>, + source: Uuid, + changes: &[Change], +) -> Result { + 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 { + 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?after={last}&limit={batch}", + primary_url.trim_end_matches('/') + ); + + let resp = client + .get(&url) + .bearer_auth(token) + .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() + ))); + } + + // A truncated or garbled body on a flaky link fails to decode here and is + // retried; a partial batch is never applied. + let body: ChangesResponse = resp + .json() + .await + .map_err(|e| SyncError::Transient(format!("bad response body: {e}")))?; + + // --- identity / history checks --- + 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); + } + + // Sequence must be strictly increasing and past our cursor. + 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) +} + +async fn replica_loop( + state: AppState, + primary_url: String, + token: String, + cfg: SyncConfig, +) { + let client = match reqwest::Client::builder() + .connect_timeout(Duration::from_secs(10)) + .timeout(Duration::from_secs(60)) + .build() + { + Ok(c) => c, + Err(e) => { + error!(error = %e, "Cannot build HTTP client; replication halted"); + return; + } + }; + + let mut backoff = Duration::from_secs(1); + + info!(primary = %primary_url, "Replica sync loop started"); + + loop { + match sync_once(&client, &state, &primary_url, &token, cfg.batch).await { + Ok(more) => { + backoff = Duration::from_secs(1); + if !more { + tokio::time::sleep(jitter(cfg.poll_interval)).await; + } + } + Err(SyncError::Transient(message)) => { + let wait = jitter(backoff); + warn!(reason = %message, retry_in = ?wait, "Replication pull failed"); + update_status(&state, |s| { + s.state = "retrying".into(); + s.last_error = Some(message); + }); + tokio::time::sleep(wait).await; + backoff = (backoff * 2).min(cfg.max_backoff); + } + Err(SyncError::Fatal(message)) => { + error!(reason = %message, "Replication halted; operator action required"); + update_status(&state, |s| { + s.state = "halted".into(); + s.last_error = Some(message); + }); + return; + } + } + } +} + +// --------------------------------------------------------------------------- +// main +// --------------------------------------------------------------------------- + +fn env_parse(name: &str, default: T) -> T { + std::env::var(name) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + #[tokio::main] async fn main() { tracing_subscriber::fmt() @@ -377,24 +931,80 @@ async fn main() { "Configuration loaded" ); - let db = Connection::open("messages.db") + // Replication settings + let role = match std::env::var("APP_ROLE").as_deref() { + Ok("replica") => Role::Replica, + Ok("primary") | Err(_) => Role::Primary, + Ok(other) => panic!("APP_ROLE must be 'primary' or 'replica', got '{other}'"), + }; + let primary_url = std::env::var("APP_PRIMARY_URL").ok(); + let repl_token = std::env::var("APP_REPL_TOKEN").ok().filter(|t| !t.is_empty()); + + if role == Role::Replica && (primary_url.is_none() || repl_token.is_none()) { + panic!("Replica requires APP_PRIMARY_URL and APP_REPL_TOKEN"); + } + if repl_token.is_none() { + warn!("APP_REPL_TOKEN not set: /api/replication/changes is disabled"); + } + + let sync_cfg = SyncConfig { + poll_interval: Duration::from_secs(env_parse("APP_SYNC_INTERVAL_SECS", 5)), + max_backoff: Duration::from_secs(env_parse("APP_SYNC_MAX_BACKOFF_SECS", 60)), + batch: env_parse("APP_SYNC_BATCH", 200i64).clamp(1, 1000), + }; + + let mut 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"); + init_db(&mut db).expect("Failed to initialize database"); + + let instance_id = + load_or_create_instance_id(&db).expect("Failed to load instance id"); + + let last_applied_seq = meta_get(&db, "last_applied_seq") + .ok() + .flatten() + .and_then(|v| v.parse::().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!( + instance_id = %instance_id, + role = role.as_str(), + last_applied_seq, + "Instance initialised" + ); let state = AppState { db: Arc::new(Mutex::new(db)), schema: Arc::new(RwLock::new(schema)), config_path: Arc::new(config_path), + instance_id, + role, + primary_url: primary_url.clone().map(Arc::new), + 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, + })), }; + if role == Role::Replica { + tokio::spawn(replica_loop( + state.clone(), + primary_url.expect("checked above"), + repl_token.expect("checked above"), + sync_cfg, + )); + } + let app = Router::new() .route("/", get(get_info)) .route("/api/now", get(get_info)) @@ -404,16 +1014,20 @@ async fn main() { "/api/admin/reload-config", axum::routing::post(reload_config), ) + .route("/api/replication/changes", get(replication_changes)) + .route("/api/replication/status", get(replication_status)) + // gzip matters on slow links; reqwest's "gzip" feature negotiates it. + .layer(tower_http::compression::CompressionLayer::new()) .with_state(state); - let addr = "127.0.0.1:3000"; - let listener = tokio::net::TcpListener::bind(addr) + let addr = std::env::var("APP_BIND").unwrap_or_else(|_| "127.0.0.1:3000".into()); + let listener = tokio::net::TcpListener::bind(&addr) .await .expect("Failed to bind address"); - info!(address = addr, "HTTP server started"); + info!(address = %addr, "HTTP server started"); axum::serve(listener, app) .await .expect("HTTP server failed"); -} \ No newline at end of file +}