diff --git a/.gitignore b/.gitignore index e11aa7e..5d7fa42 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,10 @@ /repl-test /plans messages.db +pando.db +*.db +*.db-journal +*.db-wal +*.db-shm +__pycache__/ +*.pyc diff --git a/Cargo.lock b/Cargo.lock index 3564886..24a0448 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/Cargo.toml b/Cargo.toml index 0816f19..7a72f02 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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"] } diff --git a/README.md b/README.md index 5ac728f..dc26279 100644 --- a/README.md +++ b/README.md @@ -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 < config.json +cat < 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 ( diff --git a/dashboard.py b/dashboard.py index 37838b4..76b6076 100644 --- a/dashboard.py +++ b/dashboard.py @@ -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. """ diff --git a/pando.json b/pando.json new file mode 100644 index 0000000..7a9aefa --- /dev/null +++ b/pando.json @@ -0,0 +1,15 @@ +{ + "mode": "strict", + "fields": { + "version": { + "type": "integer", + "value": 2 + }, + "message": { + "type": "string" + }, + "name": { + "type": "string" + } + } +} diff --git a/repl-test.sh b/repl-test.sh index 15b0520..e6adb23 100755 --- a/repl-test.sh +++ b/repl-test.sh @@ -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,9 +450,9 @@ 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 " Logs Dir: $WORK/" echo "" echo " Press Ctrl+C to stop all instances." echo "======================================================================" diff --git a/src/handlers.rs b/src/handlers.rs index 92e755d..f2da99e 100644 --- a/src/handlers.rs +++ b/src/handlers.rs @@ -140,8 +140,9 @@ pub async fn get_info() -> Json { 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! 🌳" }), }) } diff --git a/src/main.rs b/src/main.rs index bc9260b..888606d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -40,8 +40,12 @@ use crate::{ }, }; -fn env_parse(name: &str, default: T) -> T { - std::env::var(name) +fn get_env(pando_name: &str, app_name: &str) -> Result { + std::env::var(pando_name).or_else(|_| std::env::var(app_name)) +} + +fn env_parse(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::().expect("invalid APP_CORS_ORIGIN")) + .allow_origin(origin.parse::().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 diff --git a/src/replication.rs b/src/replication.rs index 761a309..a9f5c42 100644 --- a/src/replication.rs +++ b/src/replication.rs @@ -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}, diff --git a/src/tls.rs b/src/tls.rs index fd9e35a..e1dc1f8 100644 --- a/src/tls.rs +++ b/src/tls.rs @@ -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