Fix TTL write with Kafka NO_TIMESTAMP being stored expired in 1970 (#1141)

[nirajkumarbarot]

Closes #1141

Summary

A state.set(..., ttl=...) write whose record carries Kafka's NO_TIMESTAMP (-1) could be stored with an expiry in 1970, making the record already expired the moment it lands — deleted silently on the very next TTL sweep.

Root cause

In set() / set_bytes() (both RocksDBPartitionTransaction and MemoryPartitionTransaction), an un-timestamped write correctly avoided setting _batch_has_ttl_writes (so it couldn't wrongly trigger the store flip), but its computed stamp - a bogus NO_TIMESTAMP (-1) + ttl expiry was still recorded in _pending_stamps unconditionally:

if timestamp is not None and timestamp >= 0:
    self._batch_has_ttl_writes = True   # flip trigger correctly withheld
self._pending_stamps[(prefix, key_serialized)] = stamp   # <-- recorded anyway

If any other write in the same batch flipped the store, the stale stamp was applied to this record via _restamp_default_cf_cache_for_flip, producing a born-expired record with nothing logged.

Fix

Record the pending stamp under the same condition as the flip trigger:

key_serialized = self._serialize_key(key, prefix=prefix)
if timestamp is not None and timestamp >= 0:
    self._batch_has_ttl_writes = True
    self._pending_stamps[(prefix, key_serialized)] = stamp
else:
    logger.warning(
        "state.set(..., ttl=...) carried timestamp=%s, which cannot "
        "anchor an expiry; storing the record without one.", timestamp,
    )
self._track_batch_ttl_ms(ttl)

This relies on the existing contract that a key present in the update cache but absent from _pending_stamps falls back to SENTINEL_NEVER (never-expiring) at flip time - trading silent deletion for silent non-expiry, which is the safer default for a dedup store, with a warning now logged for visibility.

_track_batch_ttl_ms(ttl) stays outside the guard: it feeds the implicit legacy TTL used when a populated store flips without legacy_records_ttl, and the duration is still valid information even when the write's timestamp can't anchor an expiry.

Applied identically to all four call sites (RocksDB set/set_bytes and Memory set/set_bytes on the unflipped-partition path).

Alternatives considered (and why rejected — per the issue)

• Reject the write with ValueError — breaks
  TestNegativeEventTimeGuard::test_negative_first_timestamp_does_not_crash_high_water, which deliberately requires such writes to be accepted.

• Anchor on the batch's high-water (high_water + ttl) - gives the record a real TTL but invents an event-time anchor for a record that has none, and makes the result depend on write order within the batch.

Test gap closed

test_a_later_good_timestamp_in_the_same_batch_still_flips already constructed the exact batch that reproduces this bug but only asserted the store flipped. Added an assertion that the un-anchorable key's stamp decodes to SENTINEL_NEVER rather than a 1970 expiry - verified red on the original code, green after the fix.

Testing

• pytest tests/test_quixstreams/test_state/test_rocksdb/test_no_timestamp_flip.py -v - all 4 pass
• Full tests/test_quixstreams/test_state/ suite — 717 passed (remaining failures are pre-existing Docker-fixture errors unrelated to this change)
• Confirmed TestNegativeEventTimeGuard::test_negative_first_timestamp_does_not_crash_high_water still passes (the write-acceptance contract is preserved)