fix: implement DeleteRecords on lite, limbo and dynostore; fix Postgres column names

[solace-aross]

What does this change do, and why?

DeleteRecords panicked with todo!() on libSQL (lite) and DynoStore, and failed at SQL-prepare time on Postgres because the query referenced record.id/record.partition, columns that don't exist (the real columns are record.offset_id/topition, with partition living on topition). The Postgres query also deleted the wrong range (offset_id >= cutoff instead of < cutoff) and never persisted the new low watermark, so a "successful" delete had no lasting effect on Fetch, ListOffsets or offset_stage.

A fifth backend, limbo (turso://, a default feature), had the identical todo!() and wasn't mentioned by the ticket; it gets the same treatment as lite since it shares the same schema and watermark helpers.

Shared validation

Adds nisshi_storage::delete_records_cutoff, a single pure function encoding Kafka's DeleteRecords validation (confirmed against real apache/kafka trunk source): offset -1 means "delete up to the high watermark" (the common case, not an edge case); any other negative offset, or one above the high watermark, is OffsetOutOfRange; at or below the current log start is a no-op that reports the existing log start unchanged. All five backends — Postgres, lite, limbo, DynoStore and SlateDB — call this before any mutation, so none can drift from the others.

Corrected deletion semantics

Every backend now physically deletes only up to high_watermark - 1: the record/batch holding the last committed offset is Kafka's active segment and must never be removed, even for the common offset == -1 ("delete everything") request — otherwise ListOffsets(Latest) would regress to 0 until the next produce. SlateDB already had a deletion path but was missing the validation entirely and deleted by batch start offset rather than end offset (over-deleting a batch that straddles the cutoff, and deleting the active segment on the common -1 call). DynoStore's physical reclaim gets the same "never delete the active segment" guarantee for free: an object is only deleted when its successor's base offset is also below the cutoff, and the last object in a partition's listing has no successor.

No backend returns a hard Err for a per-partition problem (unknown topic, out-of-range partition or offset) — each is reported as a per-partition error code, so a sibling partition in the same request still succeeds.

Earliest after a delete

ListOffsets(Earliest) answers the new log start on every engine. list_earliest_offset.sql (pg, lite, limbo) returns the first record at or above watermark.low, or watermark.low with no timestamp when none remains, and DynoStore floors its answer at the low watermark. SlateDB already read watermark.low.

Postgres transactions

Each partition commits its new log start in a short transaction of its own, so a request holds at most one watermark row lock at a time: produce waits only while its own partition advances, and the request cannot deadlock with EndTxn. Removing the records below the log start follows in a separate transaction and is best effort. If it fails or times out, maintain removes the rows on its next run (record_delete_below_log_start.sql, statement timeout lifted like the other sweeps), always keeping the record at high_watermark - 1. A storage failure on one partition answers KAFKA_STORAGE_ERROR for that partition instead of failing the request.

Until #826 lands, Fetch does not reject an offset below the log start on any engine. On Postgres that means a fetch below the new log start can return deleted records between a timed-out removal and the next maintain run (10 minutes by default). The window opens only on a delete large enough to time out.

Follow-ups

• ListOffsets(Latest) should answer from the high watermark on every storage engine #864: answer ListOffsets(Latest) from the high watermark on every engine, which lets the high_watermark - 1 clamps go.
• DeleteRecords should answer POLICY_VIOLATION for a compact-only topic #865: POLICY_VIOLATION for compact-only topics.
• Run the broker integration tests against the Turso engine #866: broker integration tests for the Turso engine.
• DeleteRecords on libSQL and Turso holds the writer lock for the whole request #867: DeleteRecords on libSQL and Turso holds the writer lock for the whole request.
• fix: answer OFFSET_OUT_OF_RANGE for a fetch offset outside the partition #826: Fetch rejects reads below the log start with OFFSET_OUT_OF_RANGE.

How was this tested?

• 8 scenarios × 4 backends (pg/in_memory/lite/slatedb) in nisshi-broker/tests/it/delete_records.rs: cutoff resolution end-to-end (response, offset_stage, ListOffsets, Fetch all agree), the -1/keep-latest-watermark case, out-of-range rejection (both directions), the log-start no-op case, an unknown topic/partition never failing a sibling partition in the same request, and a batch that straddles the cutoff keeping its earlier-offset records reachable (directly exercises the end-vs-start deletion fix).
• limbo gets equivalent #[ignore]d unit tests in its own test module, matching the pre-existing pattern for every other limbo engine-level test in this repo.
• A SlateDB-internal test scans the physical batch-key prefix directly (something no public API can observe) to confirm the active-segment protection actually keeps the active batch's object in the underlying store, not just its logical watermark.
• Every new test mutation-tested: reverting the corrected deletion rule, the active-segment guard, or the physical-delete clamp on any backend makes the relevant test fail in the specific way it's designed to catch; reverting the mutation makes it pass again.
• maintain_removes_records_left_below_log_start (pg unit test, runs in CI) covers the Postgres sweep: rows below the log start stay until maintain, then go, and the record at high_watermark - 1 stays.
• just clippy, cargo fmt --all --check, just build-all clean.

Note: the limbo (turso) DeleteRecords tests are #[ignore]d like the other limbo engine tests, so they don't run in CI.

🤖 Generated with Claude Code

https://claude.ai/code/session_011P97dHLhTRdJ35fMJFpYPg

[sgamelin]

Thanks for taking this on. Removing the todo!() panics and the Postgres query that deleted the wrong range are real fixes, and delete_records_cutoff matches Kafka 3.9.1: -1 resolves to the high watermark, any other negative offset or one above the high watermark gets OFFSET_OUT_OF_RANGE, and an offset at or below the log start is a no-op that reports the current log start. I checked it against Partition.deleteRecordsOnLeader and UnifiedLog.maybeIncrementLogStartOffset at the 3.9.1 tag.

I'm requesting changes for one thing (1). The rest are strong suggestions or optional.

1. ListOffsets(Earliest) stays below the new log start on four of the five engines (blocking)

Kafka answers EARLIEST with logStartOffset and NO_TIMESTAMP (UnifiedLog.scala#L1326-L1336). Here:

• pg, lite and limbo answer from list_earliest_offset.sql, which takes the lowest offset_id still stored. After a -1 delete that is the retained row at high_watermark - 1, as the doc comment on delete_to_high_watermark_keeps_latest already notes.
• DynoStore answers with the base offset of the oldest object (dynostore.rs#L1454-L1460). So it is below the log start after -1 (the last batch's base offset, which can be far below high_watermark - 1) and also after any cutoff that falls inside a batch. In delete_offset_inside_a_batch_keeps_the_whole_batch, Earliest would answer 0 for a log start of 2.
• SlateDB reads watermark.low and is correct.

A consumer that resets to earliest, or a new group, therefore reads records that the DeleteRecords response reported as deleted. Once #826 lands (Fetch answers OFFSET_OUT_OF_RANGE below the log start), this becomes a loop: the fetch at high_watermark - 1 is out of range, the consumer resets to Earliest, gets high_watermark - 1 again, and repeats.

The fix is small enough for this PR: floor Earliest at the watermark's low. For SQL, join watermark in list_earliest_offset.sql, take the first record with offset_id >= w.low, and answer w.low with no timestamp when none remains. For DynoStore, floor the answer at w.low the same way. Please also assert Earliest in delete_to_high_watermark_keeps_latest and in the straddling-batch test.

Related, optional: once Earliest reads the watermark, the record kept at high_watermark - 1 exists only because the SQL Latest queries join the record at w.high - 1 (list_latest_offset_uncommitted.sql), and DynoStore's Latest reads the newest object (#798). SlateDB's Latest reads watermark.high, so protect_active_batch protects nothing there, and SlateDB's own policy_delete already removes the last batch. Answering Latest from the watermark would let the five hand-copied clamps go; if they stay, the comments should name the query that needs the retained row.

2. pg, lite and limbo delete the rows inside one request-wide transaction that holds the watermark locks (strongly suggested)

On pg, delete_records takes for no key update on each partition's watermark row in request order and holds every lock until the single commit, while an unbounded record_delete_by_offset.sql (plus the header cascade) runs for each partition. Three effects:

• Produce to every partition already processed waits on the row lock.
• The delete runs under DEFAULT_STATEMENT_TIMEOUT_MS (30 s). policy_delete opts out with SET LOCAL statement_timeout = 0 (pg.rs#L1695-L1698) for this reason. On a large partition the delete times out, the whole request rolls back, and each retry does the same, so -1 cannot succeed.
• EndTxn locks watermarks in order by t.name, tp.partition (txn_select_produced_topitions.sql), and this locks them in request order, so overlapping requests can deadlock and fail the whole request.

On lite and limbo the IMMEDIATE transaction holds the database's writer lock for the duration.

Kafka only advances logStartOffset in the request path; segment removal happens later in the retention task. Could the request commit only the watermark advance, one short transaction per partition, and leave the row removal to a bounded or background step (for example policy_delete also removing offset_id < w.low)? If it stays in the request for now, sorting the topitions before locking at least removes the deadlock.

3. Smaller points (inline)

• DynoStore: the reclaim is described as best-effort, but a listing error returns Err after the watermark is committed.
• DynoStore and #840: the timestamp pruning is a no-op on main but conflicts with #840's design. Details inline; this decides whether merge order matters.
• SlateDB: every candidate batch is decoded inside the transaction.
• Tests: the SlateDB physical test and the limbo -1 test pass with the guard broken.
• On error, Kafka answers low_watermark = -1 (ReplicaManager.scala#L1127-L1142); this PR answers 0 for an unknown topic or partition and the current log start for OFFSET_OUT_OF_RANGE. Clients act on the error code, so this is minor, but please either return -1 or add a comment at the code site explaining the difference.

4. Follow-ups

The description says the out-of-scope items are filed as follow-ups. I couldn't find GitHub issues for them (Earliest from the watermark, POLICY_VIOLATION for compact topics, broker tests for turso). Could you open them and link them from the description? The Fetch item is #826.

Merge note for #839: it replaces SlateDB's timestamps.retain with prune_time_index_below(new_low_watermark) and adds clear_latest_if_partition_empty. Whichever PR merges second should call both inside this PR's Ok(cutoff) arm with the validated cutoff (not the raw request offset), and re-run #839's delete_records tests against the end-offset deletion rule.

[solace-aross]

Summary points, in bd52b8d after a rebase onto main:

• Earliest after DeleteRecords: list_earliest_offset.sql returns the first record at or above watermark.low, or w.low with a null timestamp when none remains; the pg, lite and limbo parsers accept the null, and DynoStore floors Earliest at the low watermark. Broker tests assert Earliest; reverting the SQL or the floor fails three of them. Answering Latest from the watermark is not done; the SQL clamp comments now name the queries that need the retained row.
• pg now locks the watermark rows in (topic, partition) order, which removes the deadlock with EndTxn. Committing only the watermark advance per partition and removing rows later is left as a design follow-up, as is the statement timeout on a large -1 delete.
• Every engine now answers low_watermark = -1 on an error, and the broker error tests assert it.
• The CHANGELOG entry mentions Earliest and the -1.
• Follow-up issues still to open: Earliest/Latest from the watermark, POLICY_VIOLATION for compact topics, broker tests for turso.

[solace-aross]

@sgamelin closing out the rest of your review body, in ec7a750 (rebased on main):

• Point 2, Postgres: agreed this belonged in this PR. Each partition now commits its log start in its own short transaction, so the request holds one watermark lock at a time and can't deadlock with EndTxn. Removing the rows follows as a separate best-effort step; if it fails or times out, maintain sweeps rows below least(low, high - 1) with the statement timeout lifted, like policy_delete. A storage failure on one partition answers KAFKA_STORAGE_ERROR for that partition. New pg test: maintain_removes_records_left_below_log_start.
• One trade-off to flag: until fix: answer OFFSET_OUT_OF_RANGE for a fetch offset outside the partition #826 lands, a fetch below the new log start on pg can return deleted rows between a timed-out removal and the next maintain run. It only opens on a delete large enough to time out. The description says so.
• Point 2, lite and limbo: left as is and filed as DeleteRecords on libSQL and Turso holds the writer lock for the whole request #867. SQLite has one writer either way, so only chunking really helps, and the limbo tests don't run in CI.
• Point 4: filed ListOffsets(Latest) should answer from the high watermark on every storage engine #864 (Latest from the watermark, which lets the clamps go), DeleteRecords should answer POLICY_VIOLATION for a compact-only topic #865 (POLICY_VIOLATION), Run the broker integration tests against the Turso engine #866 (Turso broker tests), and linked them plus fix: answer OFFSET_OUT_OF_RANGE for a fetch offset outside the partition #826 from the description.
• fix(slatedb): redesign ListOffsets(Timestamp) to match Kafka's time-index semantics #839 merge note: understood. Whichever lands second wires prune_time_index_below(cutoff) and clear_latest_if_partition_empty into the Ok(cutoff) arm and reruns the delete_records tests.

Could you take another look when you have a moment?