draft plan / implementation of peerdb api

Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = ["clickhouse-c-rs", "bench"]
+members = ["clickhouse-c-rs", "bench", "walshadow-peerdb"]
 
 [package]
 name = "walshadow"
@@ -45,7 +45,7 @@ async-trait = "0.1"
 backon = { version = "1", features = ["tokio-sleep"] }
 bytes = "1"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
-clap = { version = "4", features = ["derive"] }
+clap = { version = "4", features = ["derive", "env"] }
 crc32c = "0.6"
 fallible-iterator = "0.2"
 futures = "0.3"
@@ -72,3 +72,4 @@ mimalloc = "0.1.52"
 
 [dev-dependencies]
 tempfile = "3"
+toml = { version = "1", default-features = false, features = ["parse", "display", "serde"] }

docker/Dockerfile

@@ -22,6 +22,7 @@ WORKDIR /src
 COPY Cargo.toml Cargo.lock ./
 COPY clickhouse-c-rs ./clickhouse-c-rs
 COPY bench ./bench
+COPY walshadow-peerdb ./walshadow-peerdb
 COPY src ./src
 # Cache mounts persist cargo's registry/git and the target dir across builds,
 # so a source change recompiles only the changed crates, not all deps. The
@@ -63,7 +64,10 @@ RUN chmod +x /usr/local/bin/walshadow-entrypoint \
         /var/lib/walshadow/shadow-data \
         /var/lib/walshadow/out \
         /var/lib/walshadow/spill \
-        /var/run/postgresql
+        /var/run/postgresql \
+        /var/run/walshadow \
+        /etc/walshadow \
+        /etc/walshadow/ch-config.d
 
 USER postgres
 ENTRYPOINT ["/usr/local/bin/walshadow-entrypoint"]

docker/entrypoint.sh

@@ -15,6 +15,10 @@ mkdir -p "$SHADOW_DATA"
 chmod 700 "$SHADOW_DATA"
 mkdir -p "$OUT_DIR" "$SPILL_DIR" "$SOCKET_DIR"
 
+# conf.d drop-in dir the control API writes fragments into (base is ro).
+CH_CONFIG="${WALSHADOW_CH_CONFIG:-/etc/walshadow/ch-config.toml}"
+mkdir -p "${CH_CONFIG%.toml}.d"
+
 # Pool sizes default here for the local stack, but the EC2 deploy.sh forwards
 # explicit --decoder-pool-size/--inserter-pool-size via "$@"; clap rejects a
 # flag passed twice, so only inject our defaults when the caller didn't.
@@ -49,6 +53,7 @@ exec walshadow-stream \
     --walsender-bind 127.0.0.1:5433 \
     --ch-config "${WALSHADOW_CH_CONFIG:-/etc/walshadow/ch-config.toml}" \
     --metrics-bind 0.0.0.0:9484 \
+    --control-socket "${WALSHADOW_CONTROL_SOCKET:-/var/run/walshadow/control.sock}" \
     --status-interval "${WALSHADOW_STATUS_INTERVAL:-5}" \
     "${POOL_ARGS[@]}" \
     "$@"

plans/INDEX.md

@@ -35,6 +35,11 @@ Cross-doc terminology is collected in [GLOSSARY.md](GLOSSARY.md)
   shared insert tail
 - [ops.md](ops.md) — preflight, metrics, retention, manifest (floor,
   6 LSNs), standby-status triple, kill-restart drill
+- [control.md](control.md) — in-process control plane: `ctl` unix-socket
+  line protocol, base+`conf.d` config merge (API writes only its
+  `50-api.toml` fragment), live reload (mappings/budgets/CH-conn/table
+  selection/pause) with no restart, config-driven table opt-in, pause as
+  `[stream] paused`, `Reloader` (no session lifecycle)
 - [oracle.md](oracle.md) — differential decode oracle, walshadow PG
   extension, `--validate` sampling
 - [clickhouse-c-rs Safety model](../clickhouse-c-rs/README.md#safety-model)

plans/config.md

@@ -255,10 +255,13 @@ point, not the consumer set. Four consumers:
 
 ## SIGHUP
 
-`spawn_sighup_handler` holds the resolver and calls `reload()` on each signal. No
-resolver (metrics-only run, no `--ch-config`) makes it a no-op tap. Install
-failure drops the resolver, so `has_changed` returns `Err` and subscribers freeze
-at the boot snapshot — reload disabled, config still serves.
+`run` installs the handler at boot; install failure is fatal, so a running
+daemon always has the reload path armed. `spawn_sighup_reload` holds the
+process-lifetime `Reloader` and calls `Reloader::reload()` on each signal —
+the same path as the control socket's `config reload`
+([control.md](control.md)). The `Reloader` carries the running session's
+resolver (`set_resolver`); with none registered (metrics-only run, no
+`--ch-config`) reload is a no-op tap.
 
 ## Mapping lifecycle
 

plans/control.md

@@ -0,0 +1,175 @@
+# control — in-process control plane + live reconfigure
+
+The daemon (`walshadow-stream`) embeds a control plane: a request/response Unix
+socket for management, config as a merged TOML with a `conf.d` drop-in, and
+**live reload** so source/dest/table/pause changes apply without dropping
+connections or restarting. No separate binary, no child process. The intended
+external consumer is the `walshadow-peerdb` HTTP shim (branch `origin/peerdbapi`,
+`walshadow-peerdb/`), which translates PeerDB's flow API onto this socket.
+
+## Control socket + `ctl`
+
+`--control-socket <path>` binds a `UnixListener` (`src/ops/control.rs::serve`,
+modeled on `metrics::serve` + the `shadow_stream.rs` bind pattern). Absent →
+disabled. The client is the same binary: `walshadow-stream ctl <verb>` with the
+command body as a TOML fragment on stdin (`ctl apply <<TOML … TOML`); detected in
+`main` before daemon-arg parsing so it needs no daemon flags (`--socket`, env
+`WALSHADOW_CONTROL_SOCKET`, default `/run/walshadow/control.sock`).
+
+The command is a bare **verb**. The TOML body already has section headers, so
+target rides in body, not command: `apply` takes an arbitrary fragment (any mix
+of `[source]` / `[ch]` / `[table.*]` / `[stream]`) and merges it in one atomic
+reload. CLI ergonomics (friendly aliases) are a separate layer above these verbs,
+not the daemon's concern.
+
+Wire protocol (one request per connection, `handle_conn`): a `<verb>` header
+line selects the handler; everything after the first newline is the config, a
+TOML document, read to EOF (the client half-closes its write side). TOML keeps
+full value typing (int/float/±inf/nan/array) and quotes arbitrary strings, so a
+value may contain spaces, `=`, newlines. Response: `OK\n` + an optional payload
+that is *also* TOML (`show`/`status` are tables, `tables`/`columns` are
+`[[tables]]`/`[[columns]]` arrays, `schemas` an array of strings), or
+`ERR <message>\n`. `Request::parse` + `encode_request` are the codec.
+
+### Verbs
+- `apply` — deep-merge the request body (any sections) into the fragment,
+  validate the merged effective config the way boot does
+  (`EmitterConfig::from_table`), reload. A fragment that won't parse is rejected
+  and rolled back so it can't wedge the next reload / restart. Connection config,
+  table opt-in (`[table.<ns>.<rel>] replicate = true`, `initial_load`), and pause
+  (`[stream] paused = true`) are all just sections in the body.
+- `unset` — mask keys out of the fragment, then validate + reload. The body
+  mirrors the config shape (same TOML dialect as `apply`, inverted): a non-table
+  value — the sentinel `""` — removes that key including any subtree, a section
+  recurses. So `[source] password = ""` drops one key, `[table] demo = ""` one
+  namespace, top-level `table = ""` the whole section. The sole delete
+  primitive; only the API's own fragment is touched.
+- `reload` — live reload (re-read merged config + republish; SIGHUP over the
+  socket).
+- `show` — merged effective config (passwords masked).
+- `status` — a TOML table: `state = "running"|"paused"` + `rows_synced`,
+  `backfills_pending`, `lag_bytes`, `lag_seconds`, `uptime_secs` (all from the
+  metrics snapshot).
+- `tables` (`namespace`) — enumerate source `pg_class` as an `[[tables]]` array
+  (`namespace`, `name`, `selected`, `replica_identity`), marking `selected` from
+  the merged config; `schemas` (array of strings), `columns` (`namespace`,
+  `relname` → `[[columns]]` with `name`, `type`, `notnull`) — source-PG
+  introspection.
+
+`SharedCtx { ch_config, source_base, metrics, reloader, frag_lock }` is handed to
+the handlers; `frag_lock` serializes fragment read-modify-write so concurrent
+`apply`/`unset` can't lose an update or race a rollback. `Reloader` holds only
+the running session's `Arc<ConfigResolver>` (`set_resolver`, `reload`) — there is
+no start/stop/restart state machine.
+
+## Config: base + conf.d merge
+
+Config is the daemon's own TOML, `--ch-config` **plus** every `*.toml` in the
+sibling `<ch-config>.d/` directory (e.g. `ch-config.toml` → `ch-config.d/`),
+deep-merged in lexical filename order — Postgres `include_dir` style
+(`ch_emitter::load_merged` / `merge_tables`). `load_effective(path, base)` layers
+the CLI-arg `[source]` defaults *under* the file so source connection resolves
+file-over-CLI (matches `EmitterConfig` boot).
+
+The control API writes **only its own fragment**, `ch-config.d/50-api.toml`
+(`frag_path`) — sparse, only the keys `apply`/`unset` set. The operator's base
+`ch-config.toml` is never rewritten (can be read-only mounted); other channels
+can own other fragments. `show`/introspection/`status` read the merged
+effective config (`get_config` = `load_effective`).
+
+Sections: `[source]` (source conn), `[ch]` (dest conn + emitter knobs),
+`[table.<ns>.<rel>]`, `[namespace.<ns>]`, `[runtime_config] schema`,
+`[stream] paused`. `EmitterConfig::from_table` parses a merged table (the thin
+`from_toml_str` wrapper parses a string then calls it).
+
+## Live reconfigure (no restart)
+
+SIGHUP (`spawn_sighup_reload`, unconditional — independent of the control
+socket) and `ctl reload` both call `Reloader::reload` → `ConfigResolver::reload`
+→ re-read via `load_effective`, rebuild `inner.base`, `republish` on the watch.
+Consumers pick it up live:
+
+- **mappings / namespaces / budgets / compression / drop-strategy / retry** —
+  batcher + inserter + mapping-refresher + DDL applicator already read the watch
+  (`src/config.rs`).
+- **ClickHouse connection** — `ResolvedConfig` carries the CH conn fields; the
+  inserter pool (`pipeline/inserter.rs`) and `DdlApplicator` (`ch_ddl.rs`)
+  compare the conn tuple off the watch and `reconnect()` at a batch/apply
+  boundary (same mechanism as compression).
+- **table selection** — `ResolvedConfig.table_opt_ins` carries the columns-less
+  `[table.*]` opt-in intents; the reorder coordinator
+  (`pipeline/reorder.rs::maybe_apply_reload`, called per commit in the drain
+  driver) diffs desired-vs-applied and runs `apply_table_opt_in` (add,
+  auto-create) / `exclude_table` + `note_opt_out` (remove, CH table retained).
+  Applies at the next commit (deferred while idle).
+- **pause** — `[stream] paused` in `ResolvedConfig`; the pump reads it live from
+  a `config_rx` and gates the `feed.next_chunk` `select!` arm off when paused.
+
+**Source connection is not live**: `reload()` doesn't touch the pump's
+`SourceFeed`. A DNS change is picked up by the pump's existing
+`reconnect_or_fatal`; a real host change needs a process restart.
+
+## Table selection is config-driven
+
+`[table.<ns>.<rel>]` with `columns` → pinned mapping (`EmitterConfig.tables`, as
+before). Without `columns` → an opt-in intent (`runtime_config::TableRow` into
+`EmitterConfig.table_opt_ins`), materialized via `apply_table_opt_in`
+(descriptor-derived auto-create, optional `initial_load` backfill) — the exact
+path the source-PG `config_table` overlay uses. So opting a table in is
+`ctl apply <<'[table.<ns>.<rel>]' replicate = true` (out → `replicate = false`,
+or `unset` the block), applied on reload — **no source-PG writes from control**.
+Because `apply` is a deep-merge, opting one table in leaves every other opt-in
+and every operator-pinned base mapping alone. The `config_table` + WAL overlay
+([config.md], [future/runtime_config_from_pg.md]) still exists independently for
+direct-PG operators.
+
+## Pause
+
+`apply` of `[stream] paused = true` writes the flag to the fragment + reloads;
+the pump stops consuming source WAL (idles at `next_chunk`), freezing its LSN and
+the slot's confirmed position — nothing downstream drops. `paused = false`
+resumes; the pump continues from the same LSN, so every table picks up where it
+left off.
+Retention across a pause requires a replication slot (`[source] slot`); without
+one, `wal_keep_size` bounds pause duration before the source recycles WAL. A
+pause longer than `wal_sender_timeout` may drop the replication connection;
+resume reconnects from the slot (still no data loss with a slot).
+
+## Lifecycle
+
+The daemon runs **one** streaming session (`run_session`), forever, until Ctrl-C
+/ CopyDone / fatal. `run` binds metrics + control socket + SIGHUP, then calls
+`run_session` once; Ctrl-C is a pump-loop `select!` arm that breaks and drains
+the pipeline gracefully. There is no supervisor loop, no restart, no
+running/stopped/exited machine — reconfigure is always a live reload; pause is a
+config flag.
+
+## Deploy
+
+`docker/entrypoint.sh` passes `--control-socket` and `mkdir -p
+"${CH_CONFIG%.toml}.d"` so the API can drop fragments; the image creates
+`/etc/walshadow/ch-config.d` (postgres-owned). Base `ch-config.toml` stays a
+read-only mount.
+
+## Files
+- `src/ops/control.rs` — socket, protocol, handlers, `Reloader`, `SharedCtx`.
+- `src/bin/stream.rs` — `--control-socket`, `ctl` subcommand, `run`/`run_session`,
+  `cli_source_base`, `spawn_sighup_reload`, pump `paused` gate.
+- `src/emit/ch_emitter.rs` — `load_merged`/`load_effective`/`merge_tables`,
+  `from_table` (columns-optional), `EmitterConfig.{table_opt_ins,paused}`.
+- `src/config.rs` — `ResolvedConfig` (CH conn + `table_opt_ins` + `paused`),
+  `reload` via `load_effective`, `ConfigResolver.cli_source_base`.
+- `src/emit/pipeline/inserter.rs`, `src/emit/ch_ddl.rs` — live CH reconnect.
+- `src/emit/pipeline/reorder.rs` — `maybe_apply_reload` opt-in diff at commit.
+
+## Status / open edges
+- e2e-verified live: pause/resume + table add (auto-create) via `ctl apply`
+  (`[stream] paused` / `[table.*] replicate`) + `ctl reload` and via SIGHUP, no
+  restart, base file untouched.
+- Live CH-connection swap needs a second ClickHouse to test fully.
+- `ToastResolver` live CH reconnect (for `[toast] mode = disk`) is a TODO.
+- Control still opens source-PG read connections for introspection
+  (`// TODO` on `pg_connect`) — route through the daemon's catalog later.
+- The `walshadow-peerdb` shim's PAUSED/RUNNING map to `apply [stream] paused`;
+  create-mirror maps to one `apply` carrying `[source]` + `[ch]` + `[table.*]`
+  (source, dest, and tables in a single atomic reload).

plans/future/INDEX.md

@@ -16,6 +16,7 @@ surface; promote into `plans/` once built
 * [coverage100.md](coverage100.md) — drive `cargo llvm-cov` line coverage toward 100%: tiered work list (pure units → fixtures → live e2e → hard tail)
 * [FUZZ.md](FUZZ.md) — continuous coverage-guided fuzzing (cargo-fuzz/libFuzzer) across wal-rus + walshadow + clickhouse-c-rs: tiered targets, round-trip/differential oracles, C-boundary ASan, unattended-VM supervisor
 * [pipeline_backpressure_and_scaling.md](pipeline_backpressure_and_scaling.md) — parallel decode+insert pipeline: WAL-pump backpressure via wire/record split, decode/insert scaling (bootstrap Option B, hot-table sharding, N/M sizing); pipeline substrate in [emitter.md](../emitter.md)
+* [peerdb.md](peerdb.md) — `walshadow-peerdb/` crate: PeerDB flow HTTP API shim translating onto the control daemon's unix-socket protocol; endpoint map, accept-&-ignore surface, control-protocol extensions
 * [dependencies.md](dependencies.md) — crates.io replacement candidates for generic object storage, MPMC, retry, throttling, and metrics code
 * [risks.md](risks.md) — measurement-deferred risks and open questions
 * [parked.md](parked.md) — small operational polish + cross-major fixtures + skipped-test drive