fix(staging): preserve ownership across flush and recovery

[VibeSWE]

Draft: fresh compilation, regression tests and integration checks for this main-based port are pending CI. CPU/RAM non-regression is not yet established; this is a correctness fix, not a qualified performance gain.

Description

A flush detaches disk writers before finishing them. A late batch for the same schema/minute could reopen and truncate the detached writer's .arrows.part, destroying acknowledged data. Startup recovery could also publish a live part, convert another conversion's sources, or delete its temporary Parquet output.

Use exclusive part-file creation with a ULID fallback, coordinate Stream-managed flush/recovery through a lifecycle lock, and track processing directories and temporary outputs with per-Stream RAII claims. Startup plans retain their claims while pending; conversions retain them through publication/cleanup and schema staging. Failed or dropped work releases only its own claims. Recovery publishes legacy parts under the writer lock to prevent an in-process final-name clobber. Conversion parallelism and the two-group periodic bound are preserved.

Keep pub writer: Mutex<Writer> and the exact public arrow_files_grouped_exclude_time signature, including ordinary non-test builds, as restored in R5. Document that direct writer operations do not acquire Stream lifecycle coordination and the grouping method mutates staging but returns an unowned snapshot. Add 12 deterministic lifecycle regressions and an external-consumer API/auto-trait test.

Scope and provenance

Based on origin/main at 9747e2ba7a6dc18e2965b24fc84dd0632dadd87e; only streams.rs, staging/writer.rs, and tests/staging_api_compat.rs change. This is the lifecycle correctness fix, with final R5 API restoration, isolated from the historical 941-file experimental overlay.

Main's streams.rs is byte-identical to the frozen experimental baseline. The resulting file and external API test exactly match R5; the disk-writer implementation also matches R5. No additional private overlay dependency needs porting. The experimental writer baseline's separate memory staging/chunking changes and their tests are excluded: main's entire memory writer implementation, including public MutableBuffer.inner, remains unchanged. Existing per-batch IPC flushing is unchanged. Consequently the full branch is not the source tree used by the historical runs below.

Validation

Fresh checks on this branch:

• cargo +1.96.0 fmt --all --check: passed.
• rustfmt +1.96.0 --check --edition 2024 src/parseable/streams.rs src/parseable/staging/writer.rs tests/staging_api_compat.rs: passed.
• git diff --check: passed.
• cargo +1.96.0 metadata --offline --locked --no-deps --format-version 1: passed; discovers the external API test target.
• Static public-declaration comparison against main: no removals or signature changes in either changed production file. Snapshot isolation checks confirm exact R5 lifecycle code/API test and unchanged main memory code.

Fresh compilation and Rust test execution are deferred to CI. No Cargo build, server, native workload, benchmark, or historical-target mutation was performed for this port. Keep this PR in draft until fresh CI compilation and tests pass. Suggested targeted execution: RAYON_NUM_THREADS=4 cargo test --locked --features kafka --lib parseable::streams::tests:: -- --test-threads=4, then cargo test --locked --features kafka --test staging_api_compat and the full library suite. These commands have not been run on this branch.

Historical evidence from the experimental overlay, not fresh current-branch results (2026-10-17, retained under staging-lifecycle/):

• r3/REPORT-r3.md: five shared ownership/publication regressions fail on R2 control and pass on R3; 807 default-feature library tests pass, one ignored.
• r5/REPORT-r5.md: restored external APIs compile; Kafka library suite has 834 passed, one ignored; Kafka Clippy passes with warnings.
• r6-native/REPORT-r6.md: four prospective repaired server lifetimes, each 18 million Kafka records, pass exact readback and graceful shutdown. A baseline arm committed 18 million offsets but had 10.42 million physical rows (7.58 million deficit). That deficit invalidated A/A comparison; no complete A/B block or relative performance conclusion exists. It is not proof that every missing row had one uniquely established cause.

Limitations

Claims are in-process and per Stream object. Cross-process/shared-path mutation, deletion/recreation, and direct lifecycle manipulation through the historical public writer are not coordinated. Crash/SIGKILL after final Parquet rename but before source cleanup, or failed source removal, can still duplicate rows on retry. There is no new fsync, power-loss, durability, exactly-once, or universal loss-free guarantee. Invalid-source deletion and logged-only Drop errors remain. Metric workers can continue read-only work after an early error. Regression schedules use Unix inode checks and need sufficient Rayon workers; the historical runs used four. No throughput, CPU/RAM nonregression, latency, or 20% improvement claim is made; performance qualification remains incomplete.

This PR has:

• been tested to ensure log ingestion and log query works on this branch (fresh CI pending; historical qualification described above).
• added comments explaining the ownership and locking invariants.
• documented the modified staging and public grouping behavior in source docs.

---

^{Need help on this PR? Tag @codesmith-bot with what you need.}