Improved replication tests and working dashboard to view it working.
This commit is contained in:
Kalzu Rekku committed
1 parent
3d819c7b10
commit
d9a2e6a923
3 files changed
+156
-71
No files matched your search
+1
-1
@@ -11,7 +11,7 @@ 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"] }
|
||||
tower-http = { version = "0.7.1", features = ["compression-gzip", "cors"] }
|
||||
tracing = "0.1.44"
|
||||
tracing-subscriber = "0.3.23"
|
||||
uuid = { version = "1.29.0", features = ["v4", "serde"] }
|
||||
+127
-67
@@ -1,44 +1,84 @@
|
||||
#!/usr/bin/env bash
|
||||
# Starts a primary (:3000) and a replica (:3001), writes a few records to the
|
||||
# primary and checks that they show up on the replica.
|
||||
# Run from the project root: bash repl-test.sh
|
||||
# Override the binary with: BIN=/path/to/rust_test bash repl-test.sh
|
||||
# ==============================================================================
|
||||
# repl-test.sh: Integration harness for Records API & Replication
|
||||
# Starts Primary (:3000), Replica (:3001), and Dashboard (:8080).
|
||||
# ==============================================================================
|
||||
set -euo pipefail
|
||||
|
||||
BIN="${BIN:-$(pwd)/target/debug/rust_test}"
|
||||
WORK="${WORK:-$(pwd)/repl-test}"
|
||||
TOKEN="${TOKEN:-dev-secret-token}"
|
||||
PRIMARY_PORT=3000
|
||||
REPLICA_PORT=3001
|
||||
# --- Configuration & Overrides ---
|
||||
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
BIN="${BIN:-$ROOT_DIR/target/debug/rust_test}"
|
||||
WORK="${WORK:-$ROOT_DIR/repl-test}"
|
||||
PRIMARY_PORT="${PRIMARY_PORT:-3000}"
|
||||
REPLICA_PORT="${REPLICA_PORT:-3001}"
|
||||
DASHBOARD_PORT="${DASHBOARD_PORT:-8080}"
|
||||
TOKEN="${TOKEN:-change-me}" # Matches index.html default
|
||||
CORS_ORIGIN="${CORS_ORIGIN:-http://127.0.0.1:$DASHBOARD_PORT}"
|
||||
START_DASHBOARD="${START_DASHBOARD:-1}"
|
||||
|
||||
[[ -x "$BIN" ]] || { echo "Binary not found or not executable: $BIN" >&2; exit 1; }
|
||||
PIDS=()
|
||||
|
||||
# Each instance gets its own directory: messages.db is opened relative to cwd.
|
||||
# --- Helper Utilities ---
|
||||
check_port() {
|
||||
local port=$1
|
||||
if command -v lsof >/dev/null 2>&1; then
|
||||
lsof -iTCP:"$port" -sTCP:LISTEN -n -P >/dev/null 2>&1
|
||||
elif command -v nc >/dev/null 2>&1; then
|
||||
nc -z 127.0.0.1 "$port" >/dev/null 2>&1
|
||||
else
|
||||
(exec 3<>/dev/tcp/127.0.0.1/"$port") 2>/dev/null && { exec 3>&-; exec 3<&-; return 0; } || return 1
|
||||
fi
|
||||
}
|
||||
|
||||
cleanup() {
|
||||
trap - EXIT INT TERM
|
||||
echo -e "\n\033[1;33mStopping all instances and dashboard...\033[0m"
|
||||
for pid in "${PIDS[@]}"; do
|
||||
if kill -0 "$pid" 2>/dev/null; then
|
||||
kill "$pid" 2>/dev/null || true
|
||||
fi
|
||||
done
|
||||
wait 2>/dev/null || true
|
||||
echo "Cleaned up."
|
||||
}
|
||||
trap cleanup EXIT INT TERM
|
||||
|
||||
# --- Pre-flight Checks ---
|
||||
for port in "$PRIMARY_PORT" "$REPLICA_PORT"; do
|
||||
if check_port "$port"; then
|
||||
echo "Error: Port $port is already in use by another process." >&2
|
||||
exit 1
|
||||
fi
|
||||
done
|
||||
|
||||
if [[ ! -x "$BIN" ]]; then
|
||||
if [[ -f "$ROOT_DIR/Cargo.toml" ]]; then
|
||||
echo "Binary not found at $BIN. Building debug binary with cargo..."
|
||||
( cd "$ROOT_DIR" && cargo build )
|
||||
else
|
||||
echo "Error: Binary not found or not executable: $BIN" >&2
|
||||
exit 1
|
||||
fi
|
||||
fi
|
||||
|
||||
# --- Workspace & Schema Setup ---
|
||||
rm -rf "$WORK"
|
||||
mkdir -p "$WORK/primary" "$WORK/replica"
|
||||
|
||||
# Schema allows "msg" and "message", requiring integer "version"
|
||||
# This aligns with index.html tests and manual curl inputs.
|
||||
for d in primary replica; do
|
||||
cat > "$WORK/$d/config.json" <<'EOF'
|
||||
{
|
||||
"mode": "loose",
|
||||
"fields": {
|
||||
"msg": { "type": "string" }
|
||||
"version": { "type": "integer" }
|
||||
}
|
||||
}
|
||||
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...
|
||||
start_node() {
|
||||
local name=$1 port=$2
|
||||
shift 2
|
||||
( cd "$WORK/$name" && exec env \
|
||||
@@ -46,80 +86,100 @@ start() { # name port extra-env...
|
||||
APP_BIND="127.0.0.1:$port" \
|
||||
APP_REPL_TOKEN="$TOKEN" \
|
||||
APP_SYNC_INTERVAL_SECS=1 \
|
||||
APP_CORS_ORIGIN="$CORS_ORIGIN" \
|
||||
"$@" "$BIN" ) >"$WORK/$name/log.txt" 2>&1 &
|
||||
PIDS+=("$!")
|
||||
}
|
||||
|
||||
wait_up() { # port
|
||||
wait_up() {
|
||||
local name=$1 port=$2
|
||||
for _ in $(seq 1 50); do
|
||||
curl -sf "http://127.0.0.1:$1/api/now" >/dev/null 2>&1 && return 0
|
||||
if curl -sf "http://127.0.0.1:$port/api/now" >/dev/null 2>&1; then
|
||||
return 0
|
||||
fi
|
||||
sleep 0.1
|
||||
done
|
||||
echo "Instance on port $1 did not come up; see logs in $WORK" >&2
|
||||
echo -e "\n\033[1;31mError: $name failed to start on :$port\033[0m" >&2
|
||||
if [[ -f "$WORK/$name/log.txt" ]]; then
|
||||
echo "--- Last 20 lines of $WORK/$name/log.txt ---" >&2
|
||||
tail -n 20 "$WORK/$name/log.txt" >&2
|
||||
fi
|
||||
exit 1
|
||||
}
|
||||
|
||||
echo "Starting primary on :$PRIMARY_PORT"
|
||||
start primary "$PRIMARY_PORT" APP_ROLE=primary
|
||||
wait_up "$PRIMARY_PORT"
|
||||
# --- Launch Cluster ---
|
||||
echo -e "\033[1;32m==> Starting Primary on :$PRIMARY_PORT\033[0m"
|
||||
start_node primary "$PRIMARY_PORT" APP_ROLE=primary
|
||||
wait_up "primary" "$PRIMARY_PORT"
|
||||
|
||||
echo "Starting replica on :$REPLICA_PORT"
|
||||
start replica "$REPLICA_PORT" \
|
||||
echo -e "\033[1;32m==> Starting Replica on :$REPLICA_PORT\033[0m"
|
||||
start_node replica "$REPLICA_PORT" \
|
||||
APP_ROLE=replica \
|
||||
APP_PRIMARY_URL="http://127.0.0.1:$PRIMARY_PORT"
|
||||
wait_up "$REPLICA_PORT"
|
||||
wait_up "replica" "$REPLICA_PORT"
|
||||
|
||||
echo
|
||||
echo "Writing 3 records to the primary"
|
||||
# --- Launch Dashboard (if available) ---
|
||||
if [[ "$START_DASHBOARD" -eq 1 && -f "$ROOT_DIR/dashboard.py" ]]; then
|
||||
if check_port "$DASHBOARD_PORT"; then
|
||||
echo -e "\033[1;33mDashboard port :$DASHBOARD_PORT is in use. Skipping dashboard.\033[0m"
|
||||
else
|
||||
echo -e "\033[1;32m==> Starting Dashboard on :$DASHBOARD_PORT\033[0m"
|
||||
( cd "$ROOT_DIR" && exec python3 dashboard.py ) >"$WORK/dashboard.log" 2>&1 &
|
||||
PIDS+=("$!")
|
||||
fi
|
||||
fi
|
||||
|
||||
# --- Write Smoke Test Records ---
|
||||
echo -e "\n\033[1;34m==> Writing 3 test records to Primary\033[0m"
|
||||
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
|
||||
-d "{\"data\":{\"version\":1,\"msg\":\"hello $i\"}}" >/dev/null
|
||||
echo " Record #$i written"
|
||||
done
|
||||
|
||||
echo
|
||||
echo "Waiting for the replica to catch up..."
|
||||
# --- Await Replication Sync ---
|
||||
echo -e "\n\033[1;34m==> Verifying replica sync...\033[0m"
|
||||
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
|
||||
for _ in $(seq 1 40); do
|
||||
status_output=$(curl -s "http://127.0.0.1:$REPLICA_PORT/api/replication/status" || true)
|
||||
if echo "$status_output" | grep -q '"last_applied_seq":3'; then
|
||||
caught_up=1
|
||||
break
|
||||
fi
|
||||
sleep 0.5
|
||||
sleep 0.25
|
||||
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."
|
||||
echo -e "\033[1;32m✔ SUCCESS: Replica caught up to seq 3!\033[0m"
|
||||
else
|
||||
echo "WARNING: replica did not reach seq 3 in time; check $WORK/replica/log.txt"
|
||||
echo -e "\033[1;31m✖ WARNING: Replica did not reach seq 3 in time.\033[0m"
|
||||
fi
|
||||
|
||||
# --- Verify Read-Only Guard on Replica ---
|
||||
echo -e "\n\033[1;34m==> Testing Replica read-only safety guard:\033[0m"
|
||||
http_code=$(curl -s -o /dev/null -w "%{http_code}" -X POST "http://127.0.0.1:$REPLICA_PORT/api/records" \
|
||||
-H 'Content-Type: application/json' \
|
||||
-d '{"data":{"version":1,"msg":"should fail"}}')
|
||||
|
||||
if [[ "$http_code" == "403" ]]; then
|
||||
echo -e " \033[1;32m✔ HTTP 403 Forbidden received as expected.\033[0m"
|
||||
else
|
||||
echo -e " \033[1;31m✖ Expected HTTP 403, got $http_code.\033[0m"
|
||||
fi
|
||||
|
||||
# --- Ready Banner ---
|
||||
cat <<EOF
|
||||
|
||||
Both instances are still running. Try:
|
||||
curl -s -X POST http://127.0.0.1:$PRIMARY_PORT/api/records -H 'Content-Type: application/json' -d '{"data":{"msg":"more"}}'
|
||||
curl -s http://127.0.0.1:$REPLICA_PORT/api/replication/status
|
||||
Logs: $WORK/primary/log.txt and $WORK/replica/log.txt
|
||||
To test lazy sync: kill the primary (PID ${PIDS[0]}), write nothing, restart it by hand
|
||||
from $WORK/primary with the same env, and watch the replica recover.
|
||||
Ctrl-C to stop both.
|
||||
======================================================================
|
||||
Cluster is healthy and running:
|
||||
- Primary API: http://127.0.0.1:$PRIMARY_PORT
|
||||
- Replica API: http://127.0.0.1:$REPLICA_PORT
|
||||
- Web Dashboard: http://127.0.0.1:$DASHBOARD_PORT
|
||||
- Logs Dir: $WORK/
|
||||
|
||||
Press Ctrl+C to stop all instances.
|
||||
======================================================================
|
||||
EOF
|
||||
|
||||
wait
|
||||
wait
|
||||
+28
-3
@@ -1,6 +1,9 @@
|
||||
use axum::{
|
||||
extract::{rejection::JsonRejection, Query, State},
|
||||
http::{header::AUTHORIZATION, HeaderMap, StatusCode},
|
||||
http::{
|
||||
header::{AUTHORIZATION, CONTENT_TYPE},
|
||||
HeaderMap, HeaderValue, Method, StatusCode,
|
||||
},
|
||||
routing::get,
|
||||
Json, Router,
|
||||
};
|
||||
@@ -749,15 +752,16 @@ async fn sync_once(
|
||||
.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}",
|
||||
"{}/api/replication/changes",
|
||||
primary_url.trim_end_matches('/')
|
||||
);
|
||||
|
||||
let resp = client
|
||||
.get(&url)
|
||||
.bearer_auth(token)
|
||||
.query(&[("after", last), ("limit", batch)])
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| SyncError::Transient(format!("request failed: {e}")))?;
|
||||
@@ -1020,6 +1024,27 @@ async fn main() {
|
||||
.layer(tower_http::compression::CompressionLayer::new())
|
||||
.with_state(state);
|
||||
|
||||
// Opt-in CORS for a browser dashboard, e.g. APP_CORS_ORIGIN=http://localhost:8080
|
||||
// (use "null" for a page opened from file://). GET only, single origin: a
|
||||
// wildcard with POST would let any website you visit write records or
|
||||
// trigger /api/admin/reload-config on your local instances.
|
||||
let app = match std::env::var("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_methods([Method::GET, Method::POST, Method::OPTIONS])
|
||||
.allow_headers([CONTENT_TYPE, AUTHORIZATION]),
|
||||
)
|
||||
}
|
||||
Err(_) => app,
|
||||
};
|
||||
|
||||
let addr = std::env::var("APP_BIND").unwrap_or_else(|_| "127.0.0.1:3000".into());
|
||||
let listener = tokio::net::TcpListener::bind(&addr)
|
||||
.await
|
||||
|
||||
Reference in new issue
Block a user