feat(asap-precompute-rs): otap full plugin (Phase 5 step C)

[zzylol]

Summary

Implements Phase 5 step C of the OTAP-Rust integration per docs/design-asap-otap-rust-integration.md §5, §6, §11. Layers the full asap_sketches plugin lifecycle on top of the Phase B codec (#256): a Tokio-driven input-stream consumer, Wakeup-style flush ticker, control-channel poll task, and graceful drain — with all five sketch types (DDSketch / KLL / HLL / CountSketch / CountMinSketch) dispatched via the sketch_type config knob. Mirrors the Phase 4 step C Telegraf-side processors.allsketches (#237) structurally.

The OTAP submodule wiring (linkme distributed-slice registration, build_sketchotap.sh, otap-patch/all/mod.rs patch) is deliberately deferred to Phase D per the §11 phase plan.

Plugin shape

AsapSketchesPlugin::start() spawns three concurrent Tokio tasks:

1. Input task consumes the host-supplied Stream<OtapMetricRecords>. Per batch: flatten() projects the sibling-batch family down to a flat RecordBatch → decode_batch() produces Vec<Observation> → each observation routes through Precompute::observe.
2. Flush ticker (tokio::time::interval(window_size)) calls Precompute::tick(now_ms) → encode_batch() → lift() raises Strategy-B _asap_* columns onto the per-row attribute child batch (so the resulting batch passes OTAP's strict crates/pdata/src/schema/payloads.rs::check_match validator) → emits the lifted family on an mpsc channel.
3. Control-channel task polls ControlChannel::poll() on a configurable interval, calls Precompute::update_config() on plan change, then acks the version.

Graceful shutdown (PluginHandle::shutdown().await) signals all three tasks via a oneshot cancel, joins them, runs Precompute::drain() for the in-flight residue, and emits one final batch.

File layout

New files:

Path	Purpose
asap-precompute-rs/src/otap/config.rs	PluginConfig + 5-sketch sketch_type dispatch factory (resolve() returns (PrecomputeConfig, SketchDispatch)).
asap-precompute-rs/src/otap/records.rs	Local OtapMetricRecords model + flatten() / lift() bidirectional projection.
asap-precompute-rs/src/otap/lifecycle.rs	AsapSketchesPlugin Tokio runtime + PluginHandle graceful-shutdown affordance.
asap-precompute-rs/tests/otap_lifecycle.rs	End-to-end harness — one test per sketch_type (5), drain test, control-channel test, smoke tests.
otap-patch/plugins/asap_sketches/sample.toml	Canonical TOML config block (mirrors design doc §8).
otap-patch/plugins/asap_sketches/README.md	User-facing config + lifecycle docs.
otap-patch/plugins/asap_sketches/src/mod.rs	Phase D placeholder for the linkme::distributed_slice entry.

Modified files:

Path	Change
asap-precompute-rs/Cargo.toml	Adds tokio (rt, rt-multi-thread, sync, time, macros) and futures as optional deps gated under the existing otap feature.
asap-precompute-rs/Cargo.lock	Auto-update for the new tokio/futures deps.
asap-precompute-rs/src/otap/mod.rs	Re-exports the three new modules (config, records, lifecycle).

Test summary

cargo test --release --features otap runs 127 tests (up from 118 in Phase B); new coverage:

• lifecycle_ddsketch_emits_envelope_with_correct_sketch_type — DDSketch end-to-end: 5 scalar inputs → drain → assert envelope-bearing attribute row carrying _asap_sketch_type=DDSketch.
• lifecycle_kll_emits_envelope_with_correct_sketch_type — KLL end-to-end.
• lifecycle_hll_emits_envelope_with_correct_sketch_type — HLL end-to-end (single series, distinct values via value column).
• lifecycle_countsketch_emits_envelope_with_correct_sketch_type — CountSketch end-to-end.
• lifecycle_countminsketch_emits_envelope_with_correct_sketch_type — CMS lifecycle smoke (CMSObserver requires Bytes-kind input; the test asserts the structural shape rather than the wrapper-level path that's already covered by tests/runtime.rs).
• drain_flushes_in_flight_observations_before_window_boundary — drain test: 60s window, shutdown immediately, assert envelope is emitted.
• control_channel_plan_change_acks_after_apply — feeds a one-shot ControlChannel, waits for ack, asserts post-change metric_name is reflected on emitted batches.
• shutdown_without_inputs_is_clean_no_op — empty stream + shutdown produces no batches and doesn't panic.
• unknown_sketch_type_rejected_at_construction — from_plugin_config surfaces UnknownSketchType for a bogus spelling.

Plus 11 new unit tests across config.rs / records.rs / lifecycle.rs covering dispatch case-insensitivity, zero-window rejection, flatten/lift round-trips, and the structural Strategy-B contract (no _asap_* columns on the metrics-side schema after lift()).

Phase B's existing 4 codec tests + 27 runtime tests + 11 api-surface tests + 7 cross-language-parity tests + 69 unit tests all still pass — no core module touched.

Cargo [features] and dependency changes

• otap feature now also gates tokio (sub-features rt, rt-multi-thread, sync, time, macros) and futures. No new top-level features. No new default deps. The non-otap build (cargo test --release) is byte-identical to Phase B (43 unit + 11 + 7 + 27 = 88 tests, no tokio in the closure).

Verification

• ✅ cargo test --release -p asap-precompute-rs --features otap — 127 tests, 0 failed.
• ✅ cargo test --release -p asap-precompute-rs (no features) — 88 tests, 0 failed.
• ✅ cargo clippy --release -p asap-precompute-rs --features otap --all-targets -- -D warnings — clean.
• ✅ RUSTDOCFLAGS="-D warnings" cargo doc --no-deps -p asap-precompute-rs --features otap — clean.

Design judgements

1. Local OtapMetricRecords model. The upstream OTAP OtapArrowRecords Rust type lives in the otap-dataflow submodule that Phase D wires in. Phase C tests run without the submodule, so records.rs defines a local two-batch family (metrics + per-row attributes joined by parent_id UInt32) that captures the structural shape Phase D will bind to. The flatten() / lift() API surface is the contract; Phase D adds a thin wrapper over the upstream type without changing it.

2. Strategy-B attribute carrier choice. lift() removes ALL Strategy-B _asap_* columns plus all non-reserved Utf8 label columns from the metrics-side batch and re-emits them as (parent_id, key, typed-value) rows on the attributes batch. The non-reserved-label lift is necessary because OTAP's check_match validator rejects ANY extension column on the metrics batch — not just _asap_* ones. Tests pin this contract (assert_no_strategy_b_top_level_columns).

3. HttpPollChannel deferred. The design doc §10 notes the Rust HttpPollChannel impl was promised alongside the Vector adapter (a parallel Phase 5 effort). Phase C wires the existing ControlChannel trait — operators supply any impl. The lifecycle test ships a OnceChannel stub. When the production HttpPollChannel lands (separate PR), it slots in via Arc<dyn ControlChannel> without changes here.

4. Single-instance plugin model. Each AsapSketchesPlugin instance owns one Precompute (one sketch type per the PrecomputeConfig::sketch_type rule). Multi-sketch deployments declare multiple [[pipelines.metrics.processors]] blocks with distinct id values — same shape as Telegraf's allsketches.

5. Tokio sub-feature gating. Pulled in rt-multi-thread so the supervisor can spawn long-running tasks; sync for mpsc / oneshot / Notify; time for interval. Avoided tokio-util (would have given a stock CancellationToken) by inlining a lightweight Notify-based Cancellation — keeps the dep surface minimal.

Open question / signal for Phase D + doc-tightening

• OtapMetricRecords ↔ upstream OtapPdata binding. Phase D needs to map our two-batch family to whatever upstream representation OTAP's local::Processor<OtapPdata> trait expects. The shape is structurally compatible (two RecordBatches with parent-id join), but Phase D should add a thin From/Into adapter and update the doc §6 file layout to call out our records.rs as the local model.
• Stale doc reference. Doc §10 still calls out the HttpPollChannel Rust impl as "shipped alongside the Vector adapter". The Vector adapter PR is in flight (per docs/design-asap-vector-integration.md); Phase C ships the trait surface only. Worth a doc-tightening note that says "the trait alone unblocks Phase C; the prod impl ships with Vector".
• §10 Open questions resolved in code: "OTAP's strict schema validator" is now structurally enforced by lift() + assert_no_strategy_b_top_level_columns — Phase D inherits it for free.

Test plan

• cargo test --release -p asap-precompute-rs --features otap — green
• cargo test --release -p asap-precompute-rs (no features) — green
• cargo clippy --release -p asap-precompute-rs --features otap --all-targets -- -D warnings — clean
• RUSTDOCFLAGS="-D warnings" cargo doc --no-deps -p asap-precompute-rs --features otap — clean
• Phase D: build_sketchotap.sh produces a sketchotap binary with the plugin registered
• Phase E: cross-host envelope parity vs sketchcollector / sketchtelegraf