fix(dynostore): ListOffsets by timestamp now answers from record time, not object write time

[solace-aross]

Summary

On S3 and in-memory storage, ListOffsets by timestamp (a consumer's offsetsForTimes) answered from the wrong data. It matched an object's last_modified (write time) instead of any record's own timestamp, used > where Kafka's contract is >=, returned a batch's base offset instead of the first matching record inside it, and listed every batch of the partition on every call. It now answers the first record whose own timestamp is at or after the target, with a per-partition time index that follows Kafka's time-index semantics.

What an operator sees differently

• A consumer that seeks by time can land on a different offset after upgrading.
• The first such lookup on each existing partition reads every stored batch of that partition to build its time index, and logs the batch count and duration at start and end. While a partition's index stays incomplete (a batch still in flight at the end of the log, an interior gap for up to ten minutes, or a produce from an older broker during an upgrade), each lookup on it runs that backfill again.
• Upgrade every broker sharing a bucket before relying on these lookups: a produce from an older broker discards the partition's time index, and the next lookup rebuilds it from every batch.
• A lookup that cannot list, read or inflate a batch answers KAFKA_STORAGE_ERROR in that partition's error_code, which a client retries; the other partitions of the request answer normally.
• A backfill that takes a gap in the log as a failed produce logs a warning naming the gap and its age.

Design

• A TimeIndex is kept per partition as a time_index field in the existing watermark.json document, so there is no new storage shape (nisshi-storage-dynostore/src/dynostore/time_index.rs). The old timestamps key is no longer written and is ignored on read. A document without time_index decodes to an incomplete index, and an older binary can read the new document.
• The index is sparse. A batch gets an entry only when it raises the partition's maximum timestamp and at least interval_bytes of batches have been appended since the last entry. The interval starts at 4096 bytes (Kafka's default index.interval.bytes). Timestamps of -1 or below are never indexed.
• The index is capped at 1024 entries. Past the cap it drops every other entry and doubles the interval, so the document stays about 25 KiB.
• A lookup answers no match without a scan when the target is above the partition's maximum timestamp. Otherwise it takes the floor entry (the greatest entry at or below the target, else offset 0) and scans forward from there, listing lazily and decoding records instead of trusting a batch header's max_timestamp. Removing an entry only lengthens a scan, so thinning the index and pruning entries below the log start are both safe.
• Partitions that pre-date the index are backfilled lazily on first lookup, in three steps: a CAS that records a backfill token with the current log start and high watermark; a listing of every batch below that high watermark, replaying the append rule over their headers into a candidate; and a CAS commit that merges the candidate with whatever is live. Lookups on one partition in one process share one backfill. A missing watermark document answers as an empty, complete index without a write. A GET or decode failure answers an error for that partition; only a batch deleted after the listing is skipped.
• produce assigns a batch's offset before it writes the batch object, so a listing can see a later batch and not an earlier one in flight. The backfill checks that the listed batches tile the log up to the high watermark. A gap below a batch written more than ten minutes earlier is a produce that failed in between and holds no records; the backfill logs a warning for it. Past any other gap, batches raise the candidate's maximum timestamp but get no entry, and the index is left incomplete, so the next lookup lists again. Ages are measured between two write times the store reports (the batch object's and the watermark document's), so the broker's clock does not enter into it.
• The commit marks the index complete only while its token is still in the document, so a produce from an older broker during a backfill leaves the index incomplete and the next lookup rebuilds it. The one case neither the token nor the gap check covers is a partition's first produce from an older broker, still in flight during the first lookup's empty listing; the upgrade note above covers it.
• The merge keeps a live entry only when it is above the candidate's maximum, because a sparse candidate's maximum can belong to an unindexed batch.
• Under READ_COMMITTED, a match at or past the last stable offset is no match.

Known limits, tracked separately

• The index, the backfill and the no-match short-circuit read a batch's max_timestamp from its header. fix(storage): rewrite a produced batch's max_timestamp and reject far-future CreateTime records #847 rewrites that header from the records for every CreateTime batch in the shared produce path, which covers every backend for new batches. Batches stored before fix(storage): rewrite a produced batch's max_timestamp and reject far-future CreateTime records #847, and LogAppendTime batches, keep a header that can disagree with the records; a dynostore-side defence is a follow-up.
• While an index is incomplete, each lookup reruns the whole backfill (a LIST, a GET of every batch, two CAS writes). A ranged GET of the header and saved progress across runs are follow-ups.
• The broker does not bound the time between a produce's offset CAS and its batch write; the ten-minute margin covers the object store client's retry timeout with room to spare. A produce-side deadline is a follow-up.

Cross-PR coordination

• fix: implement DeleteRecords on lite, limbo and dynostore; fix Postgres column names #835 (DynoStore delete_records) calls timestamps.retain(...) on the old field. That no longer compiles against this branch and can be dropped: pruning below the log start is safe but optional.
• fix: ListOffsets by timestamp should answer offset -1, not 0, when no record matches #838 is merged: None is the no-match answer on every backend, and the broker sends -1.
• fix(storage): rewrite a produced batch's max_timestamp and reject far-future CreateTime records #847 (produce-path max_timestamp rewrite) is the fix for the header trust point above; this PR does not depend on it to merge.

Testing

• cargo nextest run -p nisshi-broker -p nisshi-storage -p nisshi-storage-dynostore --all-features --no-fail-fast -E '!test(/::pg::/)': 535 passed; the 5 failures are Postgres tests that need a server the local run did not have, which CI runs.
• cargo fmt --all --check, cargo clippy --workspace --all-features --all-targets -- -D warnings and cargo doc --workspace --all-features --no-deps --document-private-items with RUSTDOCFLAGS="-D warnings" are clean.
• Each index test was checked by mutation: a ceiling lookup instead of a floor, no thinning at the cap, no running-maximum guard, a plain-union merge, a merge without the candidate-maximum filter, a listing without the tiling check, a commit that ignores the token, no trailing-gap check, an age taken from the broker's clock, a whole-request failure on a batch read error, and a commit that gives up on a CAS conflict each fail at least one test.

🤖 Generated with Claude Code

https://claude.ai/code/session_01D8pQUJmdicavN4zsumUdaD

[sgamelin]

Thanks for this. Answering from record time instead of object write time is the right fix, and the ceiling-plus-scan logic is sound for a complete index: every batch before the ceiling entry has a max_timestamp below the target, so the first match can't come earlier. The offset-ordered replay in merge_time_index and the serde compatibility with existing watermark documents also hold up, and rolling back to an older binary is safe.

I'm requesting changes for one correctness problem: the backfill turns a transient object-store error into a permanently wrong index (inline). The other main point is where the index lives. It grows with nearly every produce from the first one, inside the document that every produce and every fetch reads (inline).

Also:

• The client-visible part of the no-match change depends on #838. Until that lands, the service layer still maps None to offset 0 (list_offsets.rs#L168).
• CHANGELOG.md on main now has an [Unreleased] section and an advisory check. Consumers that seek by time on S3 or memory storage land on different offsets after upgrading, and the first lookup on an existing partition runs the backfill, so this deserves a ### Fixed entry.
• On #835: agreed that its retain must be deleted, not renamed. With a ceiling lookup, dropping the entries below the cutoff skips the surviving records of a batch that straddles it. If the lookup becomes a floor (see the growth comment), pruning below low becomes safe again.
• Merging with #838 needs one decision in nisshi-broker/tests/it/list_offsets.rs. #838 makes every backend answer None, so once both land the new timestamp_no_match_offset parameter has no Some(0) callers left and can go (#838's new turso module calls the three-argument form). #838's Timestamp(_) => None arm in the listing path also becomes unreachable after this PR's early continue.

[sgamelin]

Thanks for the rework. Every point from the last round is addressed, and each has a test:

• Read errors: a failed GET or decode now fails the request instead of being skipped (1e5767c).
• Backfill: concurrent lookups in one process share a single backfill, it logs at start and end, and a missing watermark document no longer gets created.
• Lookup: it lists only from the scan's start (list_with_offset), applies the READ_COMMITTED bound, and the running-maximum guard has a test.

The sparse index with a floor lookup and a cap settles the growth question. The backfill token closes the mixed-version race. The in-flight check, where the listing has to tile the log up to high, catches a race I missed. #838 has landed, so the None answer now reaches clients as -1.

I'm requesting changes for one point: the index trusts the batch header's max_timestamp, and a complete index is never rebuilt, so one inconsistent batch affects the partition for good (inline). Two cost points are non-blocking (inline): an incomplete index re-runs the whole backfill on every lookup, and the backfill reads whole batch objects when it needs only the header.

Also:

• A failed batch read now fails the whole ListOffsets request through ?. That takes down every partition in the request (list_offsets.rs#L185), where Kafka answers the error on the failing partition only. The LIST could already do this before, but this path now reads every batch. A stored batch that can't be inflated now fails every lookup whose scan crosses it. Could the error go in that partition's error_code instead?
• Every concurrency test uses one DynoStore, so the commit never hits a CAS conflict, and the path that merges again against a refetched document never runs. Two DynoStores over one InMemory would cover it, with the second producing between the first's collect and commit.
• On #835: its timestamps.retain(...) no longer compiles against this branch. Pruning time_index below the log start is safe now but optional, so it can just be dropped.

[sgamelin]

The index takes max_timestamp from the batch header, and so do the backfill (L814) and the no-match short-circuit (L1854). The client writes that header, and nothing checks it against the records. Kafka doesn't trust it: LogValidator recomputes the maximum from the records and sets it on the batch (LogValidator.java#L282, #L415).

The index is persisted, and a complete index is never rebuilt. So a batch whose header disagrees with its records stays in effect for the life of the partition:

• Header below a record's timestamp: a lookup for a target between the two takes the short-circuit and answers no match, though the record exists. Nisshi writes such a batch itself today. For LogAppendTime, the produce service sets base_timestamp and max_timestamp to now and keeps the client's deltas (produce.rs#L326-L337). Kafka also reports every record of a LogAppendTime batch at the batch's max timestamp, where the scan answers base plus delta (L972).
• Header above every record: the maximum never rises again, so no later batch gets an entry. Each lookup then scans from the last entry before that batch, a range that grows with the partition.

Could the maximum come from the records? One way is to recompute it in the produce service and set it on the batch, as Kafka does, which also covers the other backends. The other is for this path and the backfill to derive it from the records. A test with a header below its records' timestamps would pin it.

[solace-aross]

Agreed that the header is the weak point, and the right place to fix it is the produce path, for every backend. #847 does that: for each CreateTime batch, the shared produce service decodes the records and rewrites max_timestamp to their actual maximum (recomputing the CRC) before anything is written, and rejects a record more than an hour ahead of the broker's clock, so a header can neither understate the records nor run far ahead of them. Once it merges, the index, the backfill and the short-circuit here all read a header that equals the records for every new batch. I'm leaving this path as is rather than adding a second derivation here. What #847 does not cover, and I'd track as a follow-up: batches stored before it merges keep the header the client wrote, so a legacy partition whose headers understate the records answers no match for those targets, and a complete index is never rebuilt; and LogAppendTime batches, where the header is the broker's clock and a positive delta puts a record above it, which also means the scan at L972 answers base plus delta where Kafka answers the batch's max timestamp. A follow-up ticket has been raised for a dynostore-side defence: derive the maximum from the records in the backfill (it reads the whole object anyway) and in the scan, and answer LogAppendTime records at the batch's max.

[sgamelin]

While the index is incomplete, every lookup runs the whole backfill again: a LIST from offset 0, a GET of every batch, and two CAS writes to watermark.json. Three states keep it incomplete:

• this trailing gap on a partition that gets no more produces, for example after a broker stopped between the offset CAS and the batch PUT;
• an interior gap, for up to BATCH_SETTLED_AFTER;
• produces from an older broker during a rolling upgrade.

On a large partition from before the upgrade, each offsetsForTimes costs a full read until the state clears. That also makes the CHANGELOG's "reads every stored batch of that partition once" inaccurate. Two changes would bound it:

• Read only the header. The backfill uses batch_length, last_offset_delta and max_timestamp, all in the first 61 bytes, so a ranged GET saves nearly all the bytes. This needs the header's maximum to be trustworthy (see L1514).
• Keep progress. The batches before the first unsettled gap already give valid entries. The commit could save them with the offset they reach, and the next backfill could list from there instead of from 0.

No test leaves a gap after the last listed batch, so deleting this check keeps the suite green.

[solace-aross]

Agreed on the cost, and the "reads every stored batch once" claim was wrong for an index that stays incomplete. The CHANGELOG entry is gone (CONTRIBUTING now generates the changelog from the squash commit), and the PR description says what actually happens: the first lookup reads every batch, and the backfill runs again on each lookup while the index stays incomplete. The missing test is backfill_is_left_incomplete_by_a_gap_after_the_last_listed_batch: an aged listing whose last assigned offset is unwritten answers from the listed batches, leaves the index incomplete, and the lookup after the batch lands completes it; deleting the check fails it. The two bounds you propose (a ranged GET of the 61-byte header, and saving the entries before the first unsettled gap with the offset they reach so the next backfill lists from there) are deferred to a follow-up ticket that has been raised: the header GET depends on the header being trustworthy (the thread above), and saved progress is a format change I'd rather make on its own.

[solace-aross]

• Per-partition error: done. timestamp_lookup answers one partition, and list_offsets turns its error into that partition's error_code with no offset; the other partitions answer normally. A batch that cannot be listed, read or inflated now answers KAFKA_STORAGE_ERROR rather than UNKNOWN_SERVER_ERROR: it is the code Kafka answers for a log it cannot read, and a client retries it, where UNKNOWN_SERVER_ERROR made offsetsForTimes throw on a transient GET failure. failed_lookup_answers_only_its_own_partition_with_an_error covers a two-partition request, and the two read-failure tests assert the partition's code. The Earliest/Latest path keeps its pre-existing whole-request failure on a LIST error; happy to change that here too if you'd rather, but it's older than this PR.
• CAS conflict: backfill_commit_merges_again_after_a_cas_conflict runs two DynoStores over one store; the second produces between the first's listing and commit, the commit's PUT fails once (counted), and the merged index holds the produced batch's entry. Making with_mut give up on a conflict fails it.
• fix: implement DeleteRecords on lite, limbo and dynostore; fix Postgres column names #835: agreed, its timestamps.retain(...) can be dropped.