fix(lite): serialize delete-on-empty terminal trim

[shikhar]

Fix the delete-on-empty (DOE) TOCTOU race by making DOE-triggered deletion flow through the streamer terminal-trim path instead of deciding eligibility entirely in the background task.

Summary:

• Route regular stream deletion and DOE deletion through a shared streamer TerminalTrimCondition command path.
• Move DOE eligibility into the streamer: reject when appends are pending, compare against the persisted tail write timestamp, scan for remaining records, then revalidate the stable tail snapshot before appending terminal trim.
• Use SlateDB KV create timestamps for persisted tail write age, while keeping legacy stream_tail_position values readable.
• Batch DOE deadline rows as typed entries, use the latest eligible tail-write cutoff per stream, and clear all processed deadlines after a tick.
• Modernize stream record existence checks with a reverse prefix scan that ignores expired SlateDB rows.

Testing:

• just fmt
• cargo test -p s2-lite stream_doe
• cargo test -p s2-lite terminal_trim
• cargo test -p s2-lite append
• cargo clippy -p s2-lite --all-features --all-targets -- -D warnings --allow deprecated

Fixes #354

[greptile-apps]

Greptile Summary

This PR fixes the DOE TOCTOU race by routing all delete-on-empty eligibility checks through the streamer's single-threaded event loop — making the write-timestamp comparison, pending-append guard, record scan, and stable-position revalidation all happen serially within the same event loop context before the terminal trim is appended.

• Streamer serialization: handle_terminal_trim(DeleteOnEmpty{..}) checks last_tail_write_timestamp and next_assignable_pos synchronously, spawns a stream_has_records scan off the event loop, then revalidates both on the returned DeleteOnEmptyCheckResult message before calling append_terminal_trim. Regular delete_stream continues to work via TerminalTrimCondition::Always.
• KV format migration: stream_tail_position values drop the embedded write_timestamp_secs field (20 → 16 bytes); the timestamp is now sourced from SlateDB's create_ts via load_persisted_stream_tail with DurabilityLevel::Remote. Legacy 20-byte values remain readable via LEGACY_VALUE_LEN, but new 16-byte entries are not backward-compatible with old nodes.
• Deadline batching: PendingDoeBatch pre-computes last_write_cutoff = max(deadline − min_age) across all pending entries for a stream and passes the single tightest cutoff to the streamer.

Confidence Score: 5/5

Safe to merge; the two-phase in-streamer eligibility check soundly closes the TOCTOU window with no data-loss risk.

All eligibility decisions are serialized inside the streamer's single-threaded event loop, so no external state can change between the record scan and the terminal-trim append. The backward-compatible deserialization handles rolling upgrades correctly. No correctness bug was found.

lite/src/backend/kv/stream_tail_position.rs: the new 16-byte format cannot be read by old nodes, so a rollback after any new-format writes requires careful coordination.

Important Files Changed

Filename	Overview
lite/src/backend/streamer.rs	Core change: new TerminalTrimCondition/TerminalTrimOutcome types route both regular delete and DOE deletion through handle_terminal_trim; DOE path spawns a record-check task then revalidates stable_pos and last_tail_write_timestamp before calling append_terminal_trim, correctly serializing all eligibility decisions inside the streamer event loop.
lite/src/backend/bgtasks/stream_doe.rs	PendingDoeEntry replaced by PendingDoeBatch that pre-computes the maximum last_write_cutoff; process_stream_doe now delegates eligibility to the streamer via delete_stream_with_condition; tests updated with real KV create_ts-based timestamps instead of hand-crafted values.
lite/src/backend/kv/stream_tail_position.rs	Drops the embedded write_timestamp_secs field (16 bytes down from 20); deser_value now accepts both new and legacy formats, with trailing 4 bytes silently ignored — correct for rolling upgrades but unreadable by old nodes if rolled back.
lite/src/backend/streams.rs	delete_stream now delegates to delete_stream_with_condition(Always); mark_stream_deleted extracted as a standalone method called only after DeletionPending outcome.
lite/src/backend/core.rs	spawn_streamer now reads tail position via load_persisted_stream_tail (DurabilityLevel::Remote) to get the KV create_ts as last_tail_write_timestamp.
lite/src/backend/kv/stream_doe_deadline.rs	Promotes Entry to pub(in crate::backend) and adds last_write_cutoff() — a checked subtraction deadline - min_age that returns None on underflow.
lite/src/backend/kv/timestamp.rs	Adds ZERO/MAX constants, from_millis() and checked_sub_duration(); all edge cases covered and unit-tested.
lite/src/backend/bgtasks/stream_trim.rs	Minimal change: arm_doe_maybe renamed to arm_doe_on_full_trim; ser_value call drops the timestamp argument.

Sequence Diagram

sequenceDiagram
    participant DOE as DOE BgTask
    participant BE as Backend
    participant SC as StreamerClient
    participant ST as Streamer (event loop)
    participant DB as SlateDB

    Note over DOE: tick_stream_doe()
    DOE->>BE: list_pending_stream_doe(now)
    BE->>DB: scan expired deadlines
    DB-->>BE: entries per stream
    BE-->>DOE: Page StreamId PendingDoeBatch

    loop "For each stream CONCURRENCY=4"
        DOE->>BE: process_stream_doe(stream_id, batch)
        Note over BE: batch.last_write_cutoff = max(deadline - min_age)
        alt last_write_cutoff is None
            BE->>BE: skip delete
        else Some(cutoff)
            BE->>SC: delete_stream_with_condition DeleteOnEmpty cutoff
            SC->>ST: Message TerminalTrim DeleteOnEmpty cutoff
            Note over ST: Check last_tail_write_timestamp > cutoff?
            Note over ST: Check next_assignable_pos == stable_pos?
            alt Ineligible
                ST-->>SC: TerminalTrimOutcome Ineligible
            else Eligible spawn record check
                ST->>DB: stream_has_records async task
                DB-->>ST: Message DeleteOnEmptyCheckResult
                Note over ST: Re-check stable_pos, timestamp, trim_point
                alt has_records OR state changed
                    ST-->>SC: TerminalTrimOutcome Ineligible
                else Empty and stable
                    ST->>ST: append_terminal_trim
                    ST-->>SC: TerminalTrimOutcome DeletionPending
                    SC->>BE: mark_stream_deleted txn
                end
            end
        end
        BE->>DB: clear_doe_deadlines batch
    end

Loading

Prompt To Fix All With AI

Fix the following 2 code review issues. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 2
lite/src/backend/streamer.rs:457-479
**No early guard for already-pending terminal trim**

When `trim_point.state.end == SeqNum::MAX` at the time `handle_terminal_trim` is called for a `DeleteOnEmpty` condition, the code still spawns the async `stream_has_records` scan and sends a message back through the queue — only for `handle_doe_check_result` to immediately short-circuit on the `trim_point` check. The early-exit is in `handle_terminal_trim` for the write-cutoff and pending-append cases, so adding the same guard here would skip the unnecessary spawn-and-roundtrip for the already-deleted case.

### Issue 2 of 2
lite/src/backend/kv/stream_tail_position.rs:46-57
**New 16-byte format is unreadable by old nodes**

`deser_value` now accepts both `VALUE_LEN` (16 bytes) and `LEGACY_VALUE_LEN` (20 bytes), so new nodes handle both formats. However, old nodes use `check_exact_size(..., VALUE_LEN=20)` and will fail with `DeserializationError::InvalidSize` if they read a 16-byte entry written by a new node. A rollback while any new node has already written tail-position entries would leave old nodes unable to load those streams. Worth documenting in the deployment runbook that rollbacks are not safe until all 20-byte entries have been rewritten or the cluster is fully migrated.

_{Reviews (2): Last reviewed commit: "test(lite): avoid wall clock in doe proc..." | Re-trigger Greptile}

[shikhar]

@greptileai re