feat: StreamingConfig hot-reload — phase 1 API + endpoint (PR E)

[zzylol]

PR E phase 1 — gives the backend a runtime-swappable StreamingConfig so an external controller (or a test harness) can change the active aggregation plan without restarting the query engine binary.

The problem this solves

Today the backend loads StreamingConfig — the map of agg_id → AggregationConfig that defines every precomputed sketch, its window, grouping labels, and retention — once at startup from a YAML file. Any change to that plan requires restarting the binary, which drops in-flight windows, loses seconds of ingest, and trips health checks.

That restart-only model breaks the broader pipeline architecture:

1. The DataCollector controller's whole purpose is to continuously re-plan which sketches to precompute as the query workload shifts (SLA violations, new query shapes, capability misses reported by PR #11).
2. When the planner decides on a new plan, it needs a push target to deliver that plan to the backend.
3. Without a hot-reload path, every plan change means a backend restart → ingest gap → operator pain. The loop between controller intelligence and backend behavior only closes on a restart, which isn't acceptable in a steady-state production setting.

This PR adds the narrowest shippable slice that makes runtime plan changes possible: the swap machinery and an HTTP endpoint the controller can POST to. Query execution and ingest routing still take a startup snapshot (documented; deferred to phase 2), but the contract between the controller and the backend is now end-to-end testable.

How this PR fits in the stack

This is one half of the feedback loop that PR #11 (PR G) also lands:

query arrives → SimpleEngine miss (no compatible aggregation)
    ↓
PR #11 fires fire-and-forget POST to controller's /api/v1/plan
    ↓
current query returns §5.2 fallback answer
    ↓
controller's replanner generates a new sketch plan
    ↓
controller POSTs the new plan to backend via THIS PR's endpoint   ← PR #10
    ↓
next query for the same shape finds a precomputed match

• PR #11 is the signal — the query plane tells the controller what's missing.
• This PR is the return path — the controller tells the backend "here is the new plan, start computing it."

Without this PR, the controller has nowhere to push; without PR #11, the controller has no signal to push about. They ship as a pair but are independent PRs so they can be reviewed separately.

What's new

data_model::hot_reload_config (new module)

HotReloadStreamingConfig, cloneable, built on Arc<ArcSwap<StreamingConfig>>:

• new(StreamingConfig) / from_arc(Arc<StreamingConfig>) — two constructors so callers don't double-allocate
• snapshot() -> Arc<StreamingConfig> — cheap, lock-free read
• swap(StreamingConfig) -> Arc<StreamingConfig> — atomic replace, returns the old Arc so callers can diff added/removed agg_ids

Four unit tests: initial snapshot, atomic swap, clone-sharing, concurrent race (writer + reader threads never see a torn state).

GET / POST /api/v1/streaming-config

New routes next to /api/v1/precompute, /api/v1/health, /api/v1/store/metrics:

• GET — return the currently active config as JSON for verification by tests and operators
• POST — accept a YAML body matching the existing StreamingConfig::from_yaml_data shape, parse, validate, and atomically swap via HotReloadStreamingConfig::swap. Returns {status, agg_ids_added, agg_ids_removed, new_aggregation_count}. Logs a warning when a swap removes agg_ids that may have in-flight worker state (phase 1 limitation — see below)

Both endpoints return 503 Service Unavailable when the server was constructed without HttpServer::with_hot_reload_config(...), a clear failure mode for misconfigured deployments.

HttpServer::with_hot_reload_config(handle) builder

New opt-in builder. Without it, the endpoints return 503, so existing code paths (unit tests, legacy binaries) are unaffected by this PR. main.rs calls it to attach the production handle.

main.rs wiring

Constructs a single HotReloadStreamingConfig from the startup Arc<StreamingConfig> and passes it to HttpServer::with_hot_reload_config. Other consumers (SimpleEngine, PrecomputeEngine, SimpleMapStore, KafkaConsumer) still receive the startup Arc<StreamingConfig> and will ignore swaps until phase 2.

What's NOT hot-reloaded yet (documented on the module)

• SimpleEngine query execution — holds a startup snapshot. A query landing after a swap still sees the old config. Phase 2 will re-snapshot on query entry (one Arc::clone per query).
• Ingest router / OTLP receiver — routes by startup AggregationConfig clones. New metric/labels mapped to new agg_ids after a swap would not be routed until phase 2.
• In-flight precompute worker GroupState — each GroupState caches Arc<AggregationConfig> at group creation. Existing groups complete with their original config (correct semantics, not a bug). New groups created after a swap use the new config. The only case where phase 1 is visibly incomplete is removing an agg_id mid-window — existing groups for that id continue ingesting until their tumbling window closes. The POST handler logs a warning.

Phase 1 is still useful on its own: it proves the contract end-to-end, lets the controller start pushing plans, and makes the backend observable via GET for operators. Phase 2 extends the swap to take effect immediately on the query and ingest hot paths.

Tests

• Unit (4 in hot_reload_config.rs): snapshot / swap / clone-sharing / concurrent race
• HTTP integration (3 in http.rs::tests):
  • test_streaming_config_hot_reload_round_trip — starts a test server with a hot-reload handle, POSTs a 2-agg YAML config, asserts the POST response, then GETs and verifies. Also asserts the externally-held handle sees the swap, confirming shared ArcSwap.
  • test_streaming_config_hot_reload_missing_handle_503 — no handle attached → both GET and POST return 503
  • test_streaming_config_hot_reload_rejects_bad_yaml — garbage YAML → 400 + error message

Validation (post-rebase onto latest main)

• ✅ cargo check --all-targets: clean
• ✅ cargo clippy --all-targets -- -D warnings: clean
• ✅ cargo fmt --check: clean
• ✅ cargo test -p query_engine_rust --lib: 487 passed
• ✅ cargo test -p query_engine_rust --lib --features sketchlib-tests: green
• ✅ Rebased onto main — the de-flake fix and persistence clippy drive-bys were inherited from PR #9 and auto-dropped by the rebase.

Follow-ups

• Phase 2: re-snapshot per query in SimpleEngine, per message in the ingest router. Likely changes SimpleEngine's field from Arc<StreamingConfig> to Arc<HotReloadStreamingConfig>.
• Controller → backend channel: once this PR merges, DataCollector's controller/src/replan.rs can call this endpoint from its replanner as a new transport alongside the existing OpAMP push to agents.
• Graceful drain on removal: currently warns but does not block reloads that remove in-flight agg_ids. A stronger guarantee would explicitly drain affected groups before applying the swap.

🤖 Generated with Claude Code