feat(lite)!: delete-on-empty

[shikhar]

Closes #152

[greptile-apps]

Greptile Overview

Greptile Summary

This PR introduces delete-on-empty (DOE) behavior to the lite backend by:

• Writing and scanning a new persisted key type (StreamDeleteOnEmptyDeadline) to schedule deletions.
• Extending stream_tail_position to include a persisted write_timestamp_secs used for DOE eligibility checks.
• Adding a new background task (stream-delete-on-empty) and wiring it into the bgtask scheduler.
• Updating stream trimming logic to arm DOE when trims empty a stream and changing trim-point encoding to NonZeroSeqNum.

Key places this integrates with the rest of the codebase are lite/src/backend/streamer.rs (writes DOE and tail-position state during appends), lite/src/backend/streams.rs (initializes state on create/reconfigure), and lite/src/backend/bgtasks/stream_doe.rs (periodic cleanup/deletion based on expired DOE keys).

Blocking issues to address before merge are centered around DOE safety/progress: stale DOE keys can cause deletion even when DOE is disabled in current config; DOE scanning pagination can repeatedly process the same first page without ever making forward progress under load; and the new generic jitter helper in bgtasks should handle edge-case intervals defensively.

Confidence Score: 3/5

• This PR is close to merge-ready but has a few correctness issues in DOE scheduling/deletion logic that should be fixed first.
• Core DOE plumbing and tests look solid, but (1) deletion can proceed even if DOE is currently disabled due to stale deadline keys, (2) expired DOE scans can fail to make forward progress without a resume cursor when the pending set is large, and (3) the new jitter helper is generic and should handle zero/small intervals to avoid panics if reused. These are fixable but should be addressed before shipping a destructive background task.
• lite/src/backend/bgtasks/stream_doe.rs, lite/src/backend/bgtasks/mod.rs, lite/src/backend/streams.rs

Important Files Changed

Filename	Overview
Cargo.lock	Adds indexmap dependency (lockfile update) to support new DOE pending grouping.
Cargo.toml	Adds workspace dependency on indexmap for lite backend DOE feature.
lite/Cargo.toml	Pulls indexmap from workspace for use in lite backend background task implementation.
lite/src/backend/bgtasks/basin_deletion.rs	Narrows visibility of basin deletion tick method to super; no functional change beyond module access.
lite/src/backend/bgtasks/mod.rs	Adds stream delete-on-empty bgtask and jittered sleep scheduling; jitter helper and sleep reset logic need edge-case hardening.
lite/src/backend/bgtasks/stream_doe.rs	Introduces DOE bgtask scanning expired deadlines and deleting eligible empty streams; has config-staleness and pagination/progress concerns.
lite/src/backend/bgtasks/stream_trim.rs	Stream trim now uses NonZeroSeqNum trim points, returns whether records remain, and arms DOE when a trim empties a stream.
lite/src/backend/core.rs	Updates stream state construction for new tail position value format and trim-point semantics; blocks deletion when full-delete trim pending.
lite/src/backend/error.rs	Adds StreamDeleteOnEmptyError enum to surface DOE task failures cleanly.
lite/src/backend/kv/mod.rs	Adds new KeyType for stream DOE deadlines, introduces timestamp module, and reorders persisted KeyType values (requires stability review).
lite/src/backend/kv/stream_doe_deadline.rs	Adds serialization/deserialization for DOE deadline keys and min_age values; provides expired range scan helper and property tests.
lite/src/backend/kv/stream_tail_position.rs	Extends tail position value to include write_timestamp_secs; requires all writers/readers to be updated to avoid decode failures.
lite/src/backend/kv/stream_trim_point.rs	Changes trim point value encoding to NonZeroSeqNum (missing key implies 0) and rejects zero on decode.
lite/src/backend/kv/timestamp.rs	Adds TimestampSecs(u32) helper for persisted timestamps with saturation and SystemTime conversion.
lite/src/backend/read.rs	Moves resolve_timestamp helper from store.rs into read.rs and adds unit test ensuring stream-bounded resolution.
lite/src/backend/store.rs	Removes resolve_timestamp from store layer and keeps db_get / db_txn_get helpers; minimal behavioral impact.
lite/src/backend/streamer.rs	Schedules DOE deadlines on append, adds write_timestamp_secs to tail position, and writes DOE keys; verify interplay with DOE eligibility and config.
lite/src/backend/streams.rs	Writes initial tail position on stream creation and schedules DOE deadlines on config changes; can make never-written streams DOE-eligible based on created_at.

Sequence Diagram

sequenceDiagram
    participant Client
    participant Backend
    participant Streamer
    participant DB as SlateDB
    participant Bgtasks

    Client->>Backend: create_or_reconfigure_stream(basin, stream, config)
    Backend->>DB: txn.put(stream_meta)
    Backend->>DB: txn.put(stream_id_mapping)
    Backend->>DB: txn.put(stream_tail_position(StreamPosition::MIN, created_at_secs))
    alt DOE enabled & changed
        Backend->>DB: txn.put(stream_doe_deadline(after(min_age), stream_id), min_age)
    end
    Backend-->>Client: Ok

    Client->>Backend: append(records)
    Backend->>Streamer: handle append
    Streamer->>Streamer: maybe_doe_deadline(retention_age, min_age)
    Streamer->>DB: write batch
    opt DOE deadline scheduled
        Streamer->>DB: put(stream_doe_deadline(deadline, stream_id), min_age)
    end
    Streamer->>DB: put(stream_tail_position(next_pos, now_secs))

    Bgtasks->>Backend: tick_stream_doe()
    Backend->>DB: scan expired stream_doe_deadline keys
    loop per stream_id (grouped)
        Backend->>DB: scan stream_record_timestamp to check emptiness
        Backend->>DB: get stream_tail_position to check eligibility
        alt empty & eligible & mapping exists
            Backend->>Backend: delete_stream(basin, stream)
            Backend->>DB: mark stream_meta.deleted_at
        end
        Backend->>DB: delete stream_doe_deadline keys (clear deadlines)
    end

Loading

[shikhar]

@greptileai review this

[greptile-apps]

_{18 files reviewed, 4 comments}

_{Edit Code Review Agent Settings | Greptile}

[greptile-apps]

Additional Comments (2)

lite/src/backend/bgtasks/mod.rs
Jitter math can panic

jittered_delay computes max_ms as interval/10 and then calls rand::random_range(-max_ms..=max_ms). For small intervals (<10ms), max_ms becomes 0, so the range is 0..=0 (fine), but random_range expects a non-empty range; and if interval is 0, later interval - Duration::from_millis(...) can underflow/panic. Even if current call sites use 60s, this helper is generic and can be reused. Consider explicitly handling interval.is_zero() and max_ms == 0 by returning interval without calling random_range.

Prompt To Fix With AI

This is a comment left during a code review.
Path: lite/src/backend/bgtasks/mod.rs
Line: 132:140

Comment:
**Jitter math can panic**

`jittered_delay` computes `max_ms` as `interval/10` and then calls `rand::random_range(-max_ms..=max_ms)`. For small intervals (<10ms), `max_ms` becomes 0, so the range is `0..=0` (fine), but `random_range` expects a non-empty range; and if `interval` is 0, later `interval - Duration::from_millis(...)` can underflow/panic. Even if current call sites use 60s, this helper is generic and can be reused. Consider explicitly handling `interval.is_zero()` and `max_ms == 0` by returning `interval` without calling `random_range`.


How can I resolve this? If you propose a fix, please make it concise.

---

lite/src/backend/streams.rs
Tail position init mismatch

In create_or_reconfigure_stream, newly-created streams now write an initial stream_tail_position value (with StreamPosition::MIN and write_timestamp_secs = created_at). But other code paths treat a missing tail position as “never written” (e.g., stream_doe_is_eligible returns false if tail position is missing). With this initialization, a stream that has never had records can become DOE-eligible based purely on created_at, which is a behavioral change. If that’s not intended, consider leaving tail position absent until first append, or storing a sentinel write timestamp distinct from creation time.

Prompt To Fix With AI

This is a comment left during a code review.
Path: lite/src/backend/streams.rs
Line: 2016:2054

Comment:
**Tail position init mismatch**

In `create_or_reconfigure_stream`, newly-created streams now write an initial `stream_tail_position` value (with `StreamPosition::MIN` and `write_timestamp_secs = created_at`). But other code paths treat a missing tail position as “never written” (e.g., `stream_doe_is_eligible` returns `false` if tail position is missing). With this initialization, a stream that has never had records can become DOE-eligible based purely on `created_at`, which is a behavioral change. If that’s not intended, consider leaving tail position absent until first append, or storing a sentinel write timestamp distinct from creation time.


How can I resolve this? If you propose a fix, please make it concise.