fix: ListOffsets by timestamp should answer offset -1, not 0, when no record matches

[solace-aross]

Summary

When a timestamp-based ListOffsets lookup finds no matching record, nisshi was answering offset=0 / timestamp=None. Kafka's real broker answers offset=-1 / timestamp=-1 (UNKNOWN_OFFSET / UNKNOWN_TIMESTAMP) on this path, via KafkaApis.scala's buildErrorResponse.

This matters because the Java client's offsetsForTimes() keeps any entry where offset != UNKNOWN_OFFSET. Nisshi's offset=0 answer slips past that filter as if it were a real hit, and the client then throws constructing OffsetAndTimestamp from the negative timestamp.

Earliest/Latest lookups on an empty partition are unaffected and still correctly return offset 0; the bug was specific to Timestamp lookups with no match.

Changes

Fixed across all six storage backends, plus the service-layer default:

• nisshi-storage-sql/src/pg.rs
• nisshi-storage-sql/src/lite.rs
• nisshi-storage-sql/src/limbo.rs
• nisshi-storage-dynostore/src/dynostore.rs
• nisshi-storage-slatedb/src/storage.rs
• nisshi-storage-null/src/lib.rs
• nisshi-storage/src/service/list_offsets.rs (default corrected from 0 to -1)

Tests:

• nisshi-broker/tests/it/list_offsets.rs: wire-level checks on every backend, including a lookup past the last record of a non-empty partition. Adds a turso module mirroring the existing ones; it is #[ignore]d like them, so it compiles but does not run in CI.
• nisshi-broker/tests/it/storage_list_offsets.rs: storage-layer checks for the no-match answer.
• nisshi-storage-null: unit test for the Earliest/Latest/Timestamp answers.

Each backend now distinguishes "empty partition" (Earliest/Latest, offset 0) from "no record matched" (Timestamp with no hit, offset/timestamp None so the caller applies Kafka's not-found default of -1).

Test plan

• Targeted tests: 20/20 passing (list_offsets and storage_list_offsets suites)
• just build-all clean
• just fmt clean
• just clippy clean
• Full workspace test suite: 1070/1101 passing; all 31 remaining failures are the known Iceberg/lake (nisshi-schema::lake::berg) gap from no docker-compose services in this environment, unrelated to this change

🤖 Generated with Claude Code

https://claude.ai/code/session_01D2qdgPhMyLGZN5dLR8CVHs

[solace-aross]

Fixed in c5933d8. You were right: after was captured right after produce returned, and the record's stored timestamp comes from the same-millisecond to_timestamp(&SystemTime::now()) taken just before produce, so lite's r.timestamp >= $4 lookup could legitimately match at offset 0 when both landed in the same millisecond. The old Some(0) assertion couldn't tell that apart from a real match, so it never surfaced.

Pushed after a full second forward (SystemTime::now() + Duration::from_secs(1)) so it's deterministically past any produced record's millisecond timestamp, no sleep needed. Ran list_offsets::lite::single_record 25x back to back, including under concurrent background test load (fetch/produce suites running in parallel), all green.

Thanks for reproducing it directly, that made this quick to nail down.

[sgamelin]

Thanks, good catch, and the fix matches Kafka. I checked it against 3.9.1: when fetchOffsetForTimestamp finds nothing, handleListOffsetRequest answers buildErrorResponse(Errors.NONE, partition), which sends offset -1 and timestamp -1 (KafkaApis.scala#L1171-L1177, #L1214-L1226). The Java client drops an entry whose offset is -1 and builds an OffsetAndTimestamp from any other, and that constructor rejects a negative timestamp (OffsetFetcherUtils.java#L135, OffsetAndTimestamp.java#L34-L39). So the old offset 0 really did make offsetsForTimes() throw. Every backend with a no-match fallback is covered, null included, and ListOffsetsService is the only production code that reads ListOffsetResponse.offset, so nothing else in the tree relied on the old 0. Reverting the service default, or any of the four tested backends' None, fails at least one test that runs in CI.

Nothing here blocks. The inline comments cover the details. Three points that don't fit on a line:

Merge order with #839 and #840. This PR sets the convention: a no-match is offset: None, sent as -1. #839 still returns Some(0) on both of its no-match paths, and #840 parametrizes new_topic/single_record over Some(0) or None per backend, so whichever lands second has to reconcile them. There is also a behaviour change to weigh. On slatedb and dynostore, today's timestamp lookup can report no match when one exists: slatedb keys its index by each batch's base_timestamp (storage.rs#L900-L903), and dynostore compares the object's last_modified with a strict > (dynostore.rs#L1307-L1311). Today a false miss answers 0 and the consumer replays the partition. After this PR it answers -1 and the consumer starts at the end, so it skips records. Both answers are wrong, but skipping is worse for an at-least-once consumer. Landing this in the same release as #839 and #840 avoids shipping that window. I'm not asking for a change here, only flagging the ordering.

CHANGELOG. Clients see this change on the wire, so it deserves an entry under [Unreleased] (the advisory changelog job flags it too). For example:

Fixed

• ListOffsets by timestamp now answers offset -1 and timestamp -1 with error NONE when no record has a timestamp at or after the target, as Apache Kafka does. The broker answered offset 0 before. That made the Java consumer's offsetsForTimes() throw IllegalArgumentException: Invalid negative timestamp, and it sent a client that seeks to the returned offset back to the start of the partition. A partition answered with an error code now also carries offset -1 instead of 0.

Tests (optional). The case in the description, a non-empty partition with a target after its last record, is checked only at the storage layer in single_record. At the wire, every -1 assertion is on an empty partition. A fourth lookup in multiple_record at third + 1s, asserting that partition 0 gets offset -1 and timestamp -1, would pin it on all four backends. single_record's after lookup could also assert that timestamp is None. The null backend change has no test, and the new turso module is #[ignore]d like the existing ones, so limbo.rs compiles but nothing runs it. The description doesn't mention the turso module either.

[solace-aross]

@sgamelin following up on the points in your review body:

• CHANGELOG: added under [Unreleased] / Fixed with your wording.
• Tests: took all three. multiple_record now looks up one second past the last record and asserts offset -1 and timestamp -1 on every partition, partition 0 included. single_record asserts the no-match answer has no timestamp. The null backend has a unit test now. The description mentions the turso module and that it is #[ignore]d like the others.
• Merge order with fix(slatedb): redesign ListOffsets(Timestamp) to match Kafka's time-index semantics #839 and fix(dynostore): ListOffsets by timestamp now answers from record time, not object write time #840: agreed. I'd rather these land in the same release so we don't ship the window where a false miss skips records. Whichever of those lands after this one picks up the None convention.

Rebased on main and pushed as 984735d.

[solace-aross]

Small follow-up push (2cbd40a): the null backend test's tokio dev-dependency needed its Cargo.lock line, which cargo-deny caught. No code change beyond that.

[sgamelin]

Thanks for working through all of it. I re-reviewed at aba4286 by range-diff against 4ea4dcb, my last review. The first three commits are unchanged, and the last push is a clean rebase onto main.

Earlier points:

• The no-offset rule: addressed. ListOffsetResponse::offset now states it, Storage::list_offsets points to it, and each backend's comment says what its own code does. I checked every backend against that doc: Earliest and Latest with no error always answer Some, so the rule holds.
• The "empty partition" comment: addressed. It now describes what the code does, and the retention case is tracked in #864.
• Leader epoch: addressed. The difference from Kafka is now noted next to .leader_epoch(Some(0)) and in the doc comment on timestamp_no_match.
• The single_record comment: addressed.
• CHANGELOG: addressed.
• Tests: addressed. multiple_record now checks for offset -1 and timestamp -1 past the last record of a non-empty partition, on pg, lite, in_memory and slatedb. single_record checks that the no-match answer has no timestamp, and the null backend has its own unit test. The description now covers the ignored turso module.
• Merge order with #839 and #840: agreed in the thread, so nothing more is needed here.

I found nothing new in the latest commits. Approving.