457 lines
16 KiB
Markdown
457 lines
16 KiB
Markdown
# Dynamic Schema Records API
|
||
|
||
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
|
||
|
||
- **Dynamic Schema Validation**:
|
||
- Enforce required types: `string`, `integer`, `number`, `boolean`, `object`, `array`, `null`.
|
||
- Force exact values on any field (e.g., pin `"version": 2`).
|
||
- Validation modes:
|
||
- **`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, 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.
|
||
|
||
---
|
||
|
||
## Prerequisites
|
||
|
||
- [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
|
||
|
||
### Schema file
|
||
|
||
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
|
||
{
|
||
"mode": "strict",
|
||
"fields": {
|
||
"version": {
|
||
"type": "integer",
|
||
"value": 2
|
||
},
|
||
"message": {
|
||
"type": "string"
|
||
},
|
||
"enabled": {
|
||
"type": "boolean"
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
#### Schema Specification
|
||
|
||
| Property | Type | Description |
|
||
| :--- | :--- | :--- |
|
||
| `mode` | `string` | `"strict"` or `"loose"`. |
|
||
| `fields` | `object` | Map of field names to validation specifications. Must contain at least one field. |
|
||
| `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` | `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
|
||
|
||
### 1. Build and Run
|
||
|
||
```bash
|
||
# Clone the repository and enter the directory
|
||
cd your-project-name
|
||
|
||
# Create your config file
|
||
cat <<EOF > config.json
|
||
{
|
||
"mode": "strict",
|
||
"fields": {
|
||
"version": {
|
||
"type": "integer",
|
||
"value": 1
|
||
},
|
||
"message": {
|
||
"type": "string"
|
||
}
|
||
}
|
||
}
|
||
EOF
|
||
|
||
# Run the server (default: 127.0.0.1:3000)
|
||
cargo run
|
||
```
|
||
|
||
To specify a custom configuration file path:
|
||
|
||
```bash
|
||
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
|
||
|
||
### Health / Service Info
|
||
- **Endpoint:** `GET /` or `GET /api/now`
|
||
- **Description:** Returns server timestamp and status payload.
|
||
- **Example Response:**
|
||
```json
|
||
{
|
||
"uuid": "4fc736f8-45e0-47da-8da9-6c30fef6a72e",
|
||
"date": "2026-10-25T10:19:00Z",
|
||
"data": {
|
||
"version": 1,
|
||
"message": "Hello from Rust!"
|
||
}
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
### 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
|
||
{
|
||
"data": {
|
||
"version": 1,
|
||
"message": "Hello, world!"
|
||
}
|
||
}
|
||
```
|
||
- **Response (`201 Created`):**
|
||
```json
|
||
{
|
||
"seq": 42,
|
||
"uuid": "f3b392b9-dc81-42cb-b9ef-dc7e90aa7ce5",
|
||
"date": "2026-10-25T10:20:15Z",
|
||
"data": {
|
||
"version": 1,
|
||
"message": "Hello, world!"
|
||
}
|
||
}
|
||
```
|
||
- **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 `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": {
|
||
"version": 1,
|
||
"message": "Hello, world!"
|
||
}
|
||
}
|
||
]
|
||
```
|
||
|
||
---
|
||
|
||
### Get Active Configuration
|
||
- **Endpoint:** `GET /api/admin/config`
|
||
- **Description:** Inspect the currently running schema and mode.
|
||
- **Response (`200 OK`):**
|
||
```json
|
||
{
|
||
"mode": "strict",
|
||
"fields": {
|
||
"version": {
|
||
"type": "integer",
|
||
"value": 1
|
||
},
|
||
"message": {
|
||
"type": "string"
|
||
}
|
||
}
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
### Hot-Reload Configuration
|
||
- **Endpoint:** `POST /api/admin/reload-config`
|
||
- **Description:** Reads and validates the file at `APP_CONFIG`. If valid, atomically updates the active schema in memory.
|
||
- **Response (`200 OK`):**
|
||
```json
|
||
{
|
||
"status": "ok",
|
||
"mode": "strict",
|
||
"fields": 2
|
||
}
|
||
```
|
||
- **Error Response (`422 Unprocessable Entity`):** Returned if the configuration file is unreadable, invalid JSON, or contains no fields. The active schema remains untouched.
|
||
|
||
---
|
||
|
||
### 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 `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
|
||
# 1. Create a valid record
|
||
curl -i -X POST http://127.0.0.1:3000/api/records \
|
||
-H "Content-Type: application/json" \
|
||
-d '{"data": {"version": 1, "message": "First record"}}'
|
||
|
||
# 2. Modify config.json (e.g. bump required version to 2)
|
||
# Update "value": 2 in config.json...
|
||
|
||
# 3. Reload config without downtime
|
||
curl -i -X POST http://127.0.0.1:3000/api/admin/reload-config
|
||
|
||
# 4. Old version requests are now rejected
|
||
curl -i -X POST http://127.0.0.1:3000/api/records \
|
||
-H "Content-Type: application/json" \
|
||
-d '{"data": {"version": 1, "message": "Rejected"}}'
|
||
# -> HTTP 400: Field 'version' must equal 2
|
||
|
||
# 5. New version requests succeed
|
||
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 (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 (
|
||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
uuid TEXT NOT NULL UNIQUE,
|
||
date TEXT NOT NULL,
|
||
data TEXT NOT NULL
|
||
);
|
||
```
|
||
|
||
`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
|
||
|
||
This project is licensed under the MIT License.
|