feat: SimpleEngine re-snapshots StreamingConfig per query (PR E phase 2)

[zzylol]

PR E phase 2 — makes PR #10's POST /api/v1/streaming-config endpoint actually useful for query-time behavior.

The problem this solves

Phase 1 (merged in PR #10) landed the swap machinery and the HTTP endpoint, but SimpleEngine still took a one-time snapshot of Arc<StreamingConfig> at construction. A controller push via the endpoint would update the shared ArcSwap, but SimpleEngine held its own cached Arc and never re-read — so the swap was observable only through the GET debug endpoint, never through an actual query.

This PR threads the shared HotReloadStreamingConfig handle all the way into SimpleEngine and re-snapshots per query:

main.rs builds one HotReloadStreamingConfig
  ├─> HttpServer::with_hot_reload_config(handle.clone())
  │       ↑ POST /api/v1/streaming-config swaps via this handle
  │
  └─> SimpleEngine::new_with_hot_reload(handle.clone())
          ↑ each query calls streaming_config_snapshot() which
            load_full()s from the same underlying ArcSwap

HotReloadStreamingConfig clones share the same inner Arc<ArcSwap<StreamingConfig>>, so a swap on one clone is immediately visible through the other. That's the entire mechanism.

What this PR closes

Paired with PR #10 (endpoint) and PR #11 (capability-miss notify), phase 2 closes the query-side of the control loop end-to-end:

query arrives → SimpleEngine miss
    ↓
PR #11 fires fire-and-forget POST to controller /api/v1/plan
    ↓
§5.2 fallback serves the current query
    ↓
controller generates a new sketch plan
    ↓
controller POSTs to backend /api/v1/streaming-config (PR #10)
    ↓
swap lands in the shared HotReloadStreamingConfig
    ↓
next query re-snapshots via PR E phase 2 and finds a match   ← THIS PR

Before phase 2 the last step was broken — the swap landed but queries never saw it.

Changes

SimpleEngine

• Field renamed: streaming_config: Arc<StreamingConfig> → streaming_config_source: HotReloadStreamingConfig
• New constructor: SimpleEngine::new_with_hot_reload(...) takes the shared handle. main.rs uses this path.
• Existing constructor preserved: SimpleEngine::new(...) signature unchanged for tests, binaries, and legacy callers. Internally wraps the provided Arc<StreamingConfig> in a fresh HotReloadStreamingConfig so external unrelated handles don't leak in.
• New accessor: streaming_config_snapshot(&self) -> Arc<StreamingConfig>. Each call observes whatever was most recently pushed via PR feat: StreamingConfig hot-reload — phase 1 API + endpoint (PR E) #10.

13 internal call sites in simple_engine.rs

Each site that reads from streaming_config is converted to use one of two patterns:

• Inlined snapshot when the chain's result copies out of the borrowed AggregationConfig (e.g. window_size * 1000, .grouping_labels.clone()):

  self.streaming_config_snapshot()
      .get_aggregation_config(id)
      .map(|c| c.window_size * 1000)?

• Per-helper binding when a borrowed &AggregationConfig outlives the initial expression (e.g. create_store_query_plan reads multiple fields):

  let streaming_config = self.streaming_config_snapshot();
  let cfg = streaming_config.get_aggregation_config(id)?;
  // ... use cfg ...

main.rs

Switches from SimpleEngine::new(streaming_config.clone(), ...) to SimpleEngine::new_with_hot_reload(hot_reload_config.clone(), ...).

What's NOT hot-reloaded yet (phase 3)

• Per-query top-level snapshot binding — today each helper re-snapshots independently, so two helpers in the same query could theoretically see different configs under a concurrent swap. Both views remain internally consistent and the resulting answer is correct under EITHER old or new config; swaps are rare and the torn window is microseconds. Binding a single per-query snapshot and plumbing it through helper signatures is a larger refactor deferred to phase 3.
• Ingest router / OTLP receiver — still routes by startup AggregationConfig clones. Adding a new agg_id via POST does not yet cause incoming metrics to be routed into that agg_id on the ingest side.
• In-flight precompute worker GroupState — unchanged from phase 1. Existing groups complete with their construction-time config; new groups created after the swap pick up the new config. Intended semantics.

Tests

Three new unit tests in a new hot_reload_phase2_tests module:

1. streaming_config_snapshot_starts_at_initial_config — baseline: initial config observable through the new accessor.
2. streaming_config_snapshot_observes_post_construction_swap — the core guarantee: build engine with a shared handle → swap from outside → next accessor call sees the new config. Also asserts the previous snapshot remains internally consistent (per-query stability under ArcSwap).
3. legacy_new_constructor_is_independent_of_external_handle — verifies the backwards-compat SimpleEngine::new path pins the initial config in a fresh internal wrapper; a swap on an unrelated external handle does NOT leak into test/bin-only constructions.

Validation

• ✅ cargo check --all-targets: clean
• ✅ cargo clippy --all-targets -- -D warnings: clean
• ✅ cargo fmt --check: clean
• ✅ cargo test -p query_engine_rust --lib: 495 passed (up 3 from main)

Stack

• PR #9 (merged) — MessagePack encoding parity
• PR #10 (merged) — StreamingConfig hot-reload phase 1
• PR #11 (merged) — capability-miss → ControllerClient notification
• This PR — phase 2, makes PR feat: StreamingConfig hot-reload — phase 1 API + endpoint (PR E) #10's swap observable at query time

🤖 Generated with Claude Code