Pando, Pando the Aspen colony.

This commit is contained in:
Kalzu Rekku committed 2026-10-27 21:27:30 +02:00
1 parent a6d014601d
commit ac73968de7
11 files changed
+151 -88

No files matched your search

+7
View File
@@ -2,3 +2,10 @@
/repl-test
/plans
messages.db
pando.db
*.db
*.db-journal
*.db-wal
*.db-shm
__pycache__/
*.pyc
Generated
+22 -22
View File
@@ -974,6 +974,28 @@ version = "1.21.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
[[package]]
name = "pando"
version = "0.1.0"
dependencies = [
"axum",
"axum-server",
"chrono",
"rcgen",
"reqwest",
"rusqlite",
"rustls",
"rustls-pemfile",
"rustls-pki-types",
"serde",
"serde_json",
"tokio",
"tower-http 0.7.1",
"tracing",
"tracing-subscriber",
"uuid",
]
[[package]]
name = "pem"
version = "4.0.0"
@@ -1213,28 +1235,6 @@ dependencies = [
"sqlite-wasm-rs",
]
[[package]]
name = "rust_test"
version = "0.1.0"
dependencies = [
"axum",
"axum-server",
"chrono",
"rcgen",
"reqwest",
"rusqlite",
"rustls",
"rustls-pemfile",
"rustls-pki-types",
"serde",
"serde_json",
"tokio",
"tower-http 0.7.1",
"tracing",
"tracing-subscriber",
"uuid",
]
[[package]]
name = "rustc-hash"
version = "2.1.3"
+2 -1
View File
@@ -1,7 +1,8 @@
[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"] }
+14 -13
View File
@@ -1,4 +1,5 @@
# Dynamic Schema Records API
# Pando! 🌳
* Dynamic Schema Records API *
A lightweight, high-performance JSON records service built with **Rust**, **Axum**, and **SQLite**.
@@ -17,7 +18,7 @@ Instances can run as a **primary** (accepts writes) or as a **read-only replica*
- **`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`).
- **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.
@@ -86,7 +87,7 @@ The schema is only enforced on writes to a primary. A replica still loads the fi
| Variable | Default | Description |
| :--- | :--- | :--- |
| `APP_CONFIG` | `config.json` | Path to the schema file. |
| `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.** |
@@ -104,10 +105,10 @@ The schema is only enforced on writes to a primary. A replica still loads the fi
```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": {
@@ -134,23 +135,23 @@ 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.
`pando.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
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/rust_test
../target/debug/pando
# Replica (in another terminal; start with an EMPTY database)
mkdir -p replica && cp config.json replica/ && cd replica
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/rust_test
../target/debug/pando
```
A ready-made script, `repl-test.sh`, starts both on localhost, writes a few records and checks that they arrive:
@@ -407,11 +408,11 @@ Step 4 is atomic, so a crash can never leave the cursor out of step with the dat
| 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.
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 `messages.db`, because the copy shares the primary's instance ID and the replica will refuse to sync.
- **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`.
@@ -456,7 +457,7 @@ 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`:
On startup, the application creates (or opens) a local SQLite database named `pando.db`:
```sql
CREATE TABLE IF NOT EXISTS meta (
+1 -1
View File
@@ -1,6 +1,6 @@
#!/usr/bin/env python3
"""
Lightweight test server and transparent proxy for the Records API.
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.
"""
+15
View File
@@ -0,0 +1,15 @@
{
"mode": "strict",
"fields": {
"version": {
"type": "integer",
"value": 2
},
"message": {
"type": "string"
},
"name": {
"type": "string"
}
}
}
+23 -9
View File
@@ -1,6 +1,6 @@
#!/usr/bin/env bash
# ==============================================================================
# repl-test.sh: Integration harness for Records API & Replication (mTLS)
# 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)
@@ -27,7 +27,13 @@ set -euo pipefail
# Configuration
# ---------------------------------------------------------------------------
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
BIN="${BIN:-$ROOT_DIR/target/debug/rust_test}"
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)
@@ -188,7 +194,7 @@ rm -rf "$WORK"
for idx in $(seq 0 $(( NODES - 1 ))); do
name=$(node_name "$idx")
mkdir -p "$WORK/$name"
cat > "$WORK/$name/config.json" <<'CFGEOF'
cat > "$WORK/$name/pando.json" <<'CFGEOF'
{
"mode": "loose",
"fields": {
@@ -196,6 +202,7 @@ for idx in $(seq 0 $(( NODES - 1 ))); do
}
}
CFGEOF
cp "$WORK/$name/pando.json" "$WORK/$name/config.json"
done
# ---------------------------------------------------------------------------
@@ -205,10 +212,17 @@ start_node() {
local name=$1 port=$2
shift 2
( cd "$WORK/$name" && exec env \
APP_CONFIG=config.json \
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+=("$!")
@@ -292,8 +306,8 @@ for idx in $(seq 1 $(( NODES - 1 ))); do
wait_up "$name" "$port"
color_echo green " ✔ $name is up and enrolled."
if [[ -f "$WORK/$name/messages.db" ]]; then
color_echo green " ✔ mTLS credentials persisted in SQLite."
if [[ -f "$WORK/$name/pando.db" || -f "$WORK/$name/messages.db" ]]; then
color_echo green " ✔ mTLS credentials persisted in SQLite (pando.db)."
fi
done
@@ -421,9 +435,9 @@ fi
# ==============================================================================
echo ""
echo "======================================================================"
echo " Cluster is healthy • Topology: $TOPO • Nodes: $NODES"
echo " Pando Grove is healthy 🌳 • Topology: $TOPO • Nodes: $NODES"
echo ""
echo " Nodes:"
echo " Aspen Nodes (pando-node):"
for idx in $(seq 0 $(( NODES - 1 ))); do
port="${PORTS[$idx]}"
name="${NAMES[$idx]}"
@@ -436,7 +450,7 @@ for idx in $(seq 0 $(( NODES - 1 ))); do
done
echo ""
if [[ "$START_DASHBOARD" -eq 1 ]]; then
echo " Web Dashboard: http://127.0.0.1:$DASHBOARD_PORT"
echo " Pando Dashboard: http://127.0.0.1:$DASHBOARD_PORT"
fi
echo " Logs Dir: $WORK/"
echo ""
+2 -1
View File
@@ -140,8 +140,9 @@ pub async fn get_info() -> Json<ApiResponse> {
uuid: Uuid::new_v4(),
date: Utc::now().to_rfc3339(),
data: serde_json::json!({
"service": "pando",
"version": 1,
"message": "Hello from Rust!"
"message": "Hello from Pando! 🌳"
}),
})
}
+59 -40
View File
@@ -40,8 +40,12 @@ use crate::{
},
};
fn env_parse<T: std::str::FromStr>(name: &str, default: T) -> T {
std::env::var(name)
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))
}
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)
@@ -56,33 +60,53 @@ async fn main() {
.with_target(false)
.init();
let config_path = std::env::var("APP_CONFIG").unwrap_or_else(|_| "config.json".into());
let schema = load_config(&config_path).unwrap_or_else(|e| panic!("Failed to load configuration: {e}"));
info!("Starting Pando 🌳 node...");
info!(path = %config_path, fields = schema.fields.len(), "Configuration loaded");
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}"));
let role = match std::env::var("APP_ROLE").as_deref() {
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!("APP_ROLE must be 'primary' or 'replica', got '{other}'"),
Ok(other) => panic!("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());
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 APP_PRIMARY_URL and APP_REPL_TOKEN");
panic!("Replica requires PANDO_PRIMARY_URL (or APP_PRIMARY_URL) and PANDO_REPL_TOKEN (or APP_REPL_TOKEN)");
}
if repl_token.is_none() {
warn!("APP_REPL_TOKEN not set: /api/replication/changes is disabled");
warn!("Replication 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 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 mut db = Connection::open("messages.db").expect("Failed to open SQLite database");
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");
@@ -105,7 +129,7 @@ async fn main() {
meta_set(&db, "tls_ca_key_pem", &ca_key).unwrap();
ca_cert_pem = Some(ca_cert);
ca_key_pem = Some(ca_key);
info!("Cluster Root CA initialized");
info!("Pando Cluster Root CA initialized");
}
// Initialize Primary Server Certificate
@@ -113,7 +137,7 @@ async fn main() {
let (cert, key) = issue_leaf_cert(
ca_cert_pem.as_ref().unwrap(),
ca_key_pem.as_ref().unwrap(),
"primary",
"pando-primary",
vec!["localhost".into(), "127.0.0.1".into()],
)
.expect("Failed to issue primary server certificate");
@@ -124,8 +148,8 @@ async fn main() {
info!("Primary server certificate generated");
}
// Preload an initial bootstrap join token if APP_JOIN_TOKEN is passed
if let Ok(seed_token) = std::env::var("APP_JOIN_TOKEN") {
// 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 {
@@ -138,16 +162,16 @@ async fn main() {
} else {
// Replica: Check if dynamic bootstrap enrollment is needed
if node_cert_pem.is_none() {
// APP_JOIN_URL lets the replication source (APP_PRIMARY_URL) differ from the
// CA enrollment target (APP_JOIN_URL). This is required in chain topologies
// 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 APP_PRIMARY_URL when not set.
let join_url = std::env::var("APP_JOIN_URL")
// 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("APP_PRIMARY_URL (or APP_JOIN_URL) required for replica enrollment");
let token = std::env::var("APP_JOIN_TOKEN")
.expect("APP_JOIN_TOKEN required for initial replica enrollment");
.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)
@@ -175,19 +199,14 @@ async fn main() {
instance_id = %instance_id,
role = role.as_str(),
last_applied_seq,
"Instance initialised"
db = %db_path,
"pando-node initialised"
);
let sync_cfg = SyncConfig {
poll_interval: Duration::from_secs(env_parse("APP_SYNC_INTERVAL_SECS", 5)),
max_backoff: Duration::from_secs(env_parse("APP_SYNC_MAX_BACKOFF_SECS", 60)),
batch: env_parse("APP_SYNC_BATCH", 200i64).clamp(1, 1000),
};
let state = AppState {
db: Arc::new(Mutex::new(db)),
schema: Arc::new(RwLock::new(schema)),
config_path: Arc::new(config_path),
config_path: Arc::new(config_path.clone()),
instance_id,
role: Arc::new(RwLock::new(role)),
primary_url: Arc::new(RwLock::new(primary_url)),
@@ -232,8 +251,8 @@ async fn main() {
.layer(tower_http::compression::CompressionLayer::new())
.with_state(state);
let bind_str = std::env::var("APP_BIND").unwrap_or_else(|_| "127.0.0.1:3000".into());
let addr: SocketAddr = bind_str.parse().expect("Failed to parse APP_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");
// Build mTLS ServerConfig
let server_config = create_mtls_server_config(
@@ -243,12 +262,12 @@ async fn main() {
)
.expect("Failed to build mTLS server configuration");
let app = match std::env::var("APP_CORS_ORIGIN") {
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 APP_CORS_ORIGIN"))
.allow_origin(origin.parse::<HeaderValue>().expect("invalid CORS origin"))
.allow_methods([Method::GET, Method::POST, Method::OPTIONS])
.allow_headers([CONTENT_TYPE, AUTHORIZATION]),
)
@@ -258,7 +277,7 @@ async fn main() {
let rustls_config = RustlsConfig::from_config(Arc::new(server_config));
info!(address = %addr, "mTLS HTTPS server listening");
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
+5
View File
@@ -1,3 +1,8 @@
//! 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},
+1 -1
View File
@@ -49,7 +49,7 @@ pub fn generate_ca() -> Result<(String, String), String> {
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, "Cluster Root CA");
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