Design proposal: bound segment time span ( segment_ms )

[djanand]

Goal

Put a ceiling on how much time a segment spans, so retention works on partitions that do not fill a segment quickly. This is finer retention granularity, not a bound on message lifetime. [edited]

Related to #3312. A partition that never seals cannot be compressed or offloaded either, so bounding the span can be a prerequisite for the tiering work discussed there rather than a parallel concern.

Issue

Segments seal on size alone. Segment::is_full() compares against the per-topic segment_size, 1 GiB by default, and there is no time trigger. Retention currently looks at sealed segments. Segment::is_expired() and bails early on anything unsealed. So essentially,

• Expiration : message_expiry runs on message timestamps, continuously regardless of segment state.
• Eligibility : sealing decides if a segment is eligible for expiration.

The expiry condition can be true for weeks without sealing the segment.

What it looks like in practice

With segment_size = 1 GiB and message_expiry = 30 days :

1. A topic at 1 MB/hour takes about ~45 days ( six weeks) to fill a segment. For those six weeks nothing seals, so nothing expires. The operator asked for 30 days and gets six weeks.
2. A topic at 1 MB/s fills a segment in roughly 18 minutes, seals constantly, and behaves exactly as documented. Which is probably why this has not come up.
3. A decommissioned topic stops leaves every partition sitting on an unsealed segment, maybe 500 MB each, none reaching 1 GiB and none ever sealing. The same thing happens inside a live topic whenever traffic skews: a partition that stops receiving keys is idle whether or not the topic is.

How late deletion runs

Expiration compates max_timestamp, the newest message in a segment, so nothing is deleted until the newest message is aged out. The oldest message in a segment lives as long as the segment spans. For e.g taking message_expiry = 1h

Current state

Throughput	1 GiB spans	oldest data lives
1 MB/s	~18 minutes	~1h 18m
12 KB/s	~24 hours	~25 hours
1 KB/s	~12 days	~12 days

Proposed State (with segment_ms)

Setting segment_ms = 1h caps the span, and the worst case stops depending on throughput:

Throughput	span becomes	oldest data lives
1 MB/s	~18 minutes, size still wins	~1h 18m
12 KB/s	1 hour	~2 hours
1 KB/s	1 hour	~2 hours

Worst case goes from message_expiry + however long a segment takes to fill to message_expiry + segment_ms, which is the guarantee Kafka gives with segment.ms.

High-level shape

• A per-topic segment_ms alongside segment_size, parsed the same way. Rotation becomes size-or-span.

segment_ms = "none"        size-only, today's behaviour
segment_ms = "default"     take the server default (7days) - [refer Open Questions (Q2)]
segment_ms = "7 days"      bound the span at this

• Keyed on replicated message timestamps, not wall clock - Segment boundaries land at the same offset on every replica today, and that is deliberate. The rollover is keyed on accumulated bytes so flush grouping cannot shift it.
• Compare arriving batch against the segment i.e.

roll if  segment.start_timestamp != 0
     &&  incoming.base_timestamp - segment.start_timestamp >= segment_ms

Both values are primary-assigned and replicated, so every replica rolls at the same message and boundaries stay aligned. Additionally, the roll has to happen before the arriving batch is appended. If it lands in the old segment first, that segment's max_timestamp jumps forward and its contents stop being expirable for another full message_expiry

Preallocated segments need trimming at seal

• preallocate_segments reserves exactly segment_size up front. That is safe today because a segment only seals once it has reached or exceeded that size, so the reservation is always fully used and there is nothing to reclaim.
• A slow topic with preallocation on would reserve 1 GiB, write a few MB into it, and roll. fallocate is called with FALLOC_FL_KEEP_SIZE, so those blocks stay allocated to the file, and nothing in the seal path truncates. The file reports a few MB to ls and holds 1 GiB on disk until retention deletes it, which could be the whole retention window away
• max_topic_size compares against segment.size, which counts written bytes. It would see a few MB while the disk fills
• So the proposal needs a trim at seal, reducing the file to its written length. Kafka does this in onBecomeInactiveSegment, which calls log.trim()

What this does not cover

• A partition that stops receiving writes entirely. Stream time only advances when messages arrive, so the decommissioned case above still never seals.
• Catching this needs actual wall-clock time, and no replica can read a clock without drifting from its peers. So it is either 1) a timer per node, accepting the divergence, or 2) the primary decides and replicates the rollover. The second seems right to me, but it means a new Operation, and router.rs says the version fence that would make that safe during a rolling upgrade is not there yet. Both are bigger than this change, so I would keep them separate.

Open Questions

1. Mutable, not create-only, unless someone sees a reason I have missed. segment_size is fixed at creation because max_topic_size admission uses it to work out the smallest cap it can enforce, which only holds if existing segments really are that size. No calculation uses a time setting that way. Mutable matters here anyway: the topics with this problem already exist, and create-only would mean recreating them to fix it.
2. Server default. Filling 1 GiB in 7 days works out at about 1.8 KB/s, so a 7 day default would only ever affect partitions writing slower than that. Anything faster fills a segment inside the week and still rotates on size, exactly as it does now. Turning it on by default therefore cannot disturb a healthy topic, and the only partitions it touches are the ones that need it already. Inclined to have default as on ( in the near future ), rather than none
3. How the seal-time trim gets done. - Going through compio submits IORING_OP_FTRUNCATE, which only exists from kernel 6.9, and on anything older the driver reaches for a blocking pool the shard does not have and panics. Proposal is copy FileStorage::truncate rather than invent something, but I would rather hear it is the house pattern than assume. It would make a failed trim non-fatal, a warning and carry on, since the segment is perfectly good either way and you have only failed to give some blocks back.

Affected area / component

• Iggy server
• Configuration

Alternatives considered

1. A wall-clock timer per partition. Covers the idle case, which stream time does not, and breaks byte-identical segments across replicas. Rejected because we 1) loose ability to compare cross replicas (i.e checksum ) 2) Tiering gets harder 3) Its a one-way door and everything has to tolerate divergence
2. Lower segment_size at creation. Only helps new topics, since the key is create-only, and expresses the wrong intent by trading segment count against a size budget when the requirement is about time.
3. Idle topic. But it needs a new consensus Operation, and a node that has not been upgraded wedges its whole group silently on meeting one, with no version fence yet to stop that. Bigger than this change, probably its own issue

What I'd like from this discussion

• Confirmation that retention skipping unsealed segments is a gap rather than a deliberate trade
• Whether I have the replica-boundary constraint right. The whole design is shaped around keeping segment boundaries identical, and if that turns out to be softer than I think, a wall-clock timer is simpler and covers the idle case too
• A steer on the server default, since it is the one decision that changes behaviour for deployments that exist today
• Whether FileStorage::truncate is the pattern to copy for the seal-time trim, or there is a better one
• Whether the uncovered idle-partition case wants its own issue now, or can sit documented until someone hits it

Contribution

[x] I'm willing to submit a pull request to implement this feature

Good first issue

[ ] unchecked - touches persistence and the rotation path

[hubcio]

i skimmed thru this proposal and I like the idea. i will review this in detail this week.

[djanand]

thanks for looking into this @hubcio.

[hubcio]

@djanand overall i like this. a few things need to be addressed:

1. expiry works on whole segments, not messages. the cleaner only removes sealed segments, and only once max_timestamp + message_expiry <= now. active-segment protection is deliberate and tested, so i'd keep it and add the missing rotation trigger. with your numbers (45-day fill, 30-day expiry) the oldest messages become eligible at ~75 days, not six weeks. (expiry)

2. message_expiry + segment_ms isn't a hard bound. you already scope out idle partitions, but sparse ones hit the same wall: rotation is only checked on append, so the roll waits for the next batch. cleanup ticks and persistence checkpoints add delay. consumer and consumer-group offsets also block removal, with no timeout (external groups excepted). a stopped consumer pins expired segments past its offset until that offset or its group is deleted. i'd frame this as finer retention granularity, not a max message lifetime, and open the idle-partition issue now. (deletion gates, offsets)

3. the WAL and materialization have to make the same call. SegmentState::reserve picks the physical segment before the batch is written, and materialization checks that placement. if only the logical rotation looks at time, that check rejects the commit, the partition gets fenced and the node shuts down. the decision has to be per batch, independent of flush grouping, and reproducible after restart, rollback and state transfer. SegmentState has no timestamp today. the open segment's first timestamp has to be persisted with it, or recovered from the segment wherever the WAL rebuilds its tail. segments also only seal during materialization (by default every 1024 messages or 1 MiB), so a time roll has to make materialization due too. (WAL placement, placement check, fatal commit, flush trigger)

4. if it's mutable, the design needs to say which batch first uses a new value and how replay gets the same answer. the runtime knobs are create-only because nothing re-pushes them to a live partition and they describe the storage layout. max_topic_size admission depends on that, it isn't the cause. message_expiry and max_topic_size can be mutable because only the local, unreplicated cleaner enforces them, rereading metadata each pass. segment_ms feeds replicated WAL placement instead, and the WAL already refuses a changed segment size. topic updates also commit in the metadata consensus group, not the partition's, so replicas can pick up a new threshold at different batches. re the version fence: since feat(consensus): persist retry receipts and shared sessions #4387 (merged after you posted), the replica handshake rejects peers from a different release, so an old replica never sees a new Operation. (runtime knobs, cleaner, WAL size check, build identity)

5. for the trim, call server_common::fs_utils::truncate_file. it avoids IORING_OP_FTRUNCATE, and since perf(partitions): keep shard traffic moving during file I/O #4311 FileStorage::truncate delegates to it, so there's nothing to copy. the length is harder: the public segment and the retained WAL path are hard links to one inode, and WAL writes can run ahead of materialization. trim only after the WAL tail has left the segment, and only to the length where WAL placement, segment.size and the file length agree. anything shorter deletes bodies the WAL still references. a failed trim can be nonfatal, as long as that doesn't weaken persistence or recovery error handling. (helper, shared inode, WAL ahead)

for the default, i'd go opt-in. low throughput can be normal, and this changes rotation, allocation and retention for those topics. topic defaults are catalog constants, not server settings, and admission persists them on new topics. so please spell out separately what new topics get and what a missing key means for existing ones. (defaults)

it also needs a lower bound. MIN_TOPIC_SEGMENT_SIZE exists because too many segments make a partition unservable for state transfer (65536 manifest entries max), and a time trigger skips that floor. (floor)

tests i'd want to see:

• sparse arrivals, including a time roll below the flush thresholds
• the threshold-crossing batch landing entirely in the new segment
• identical boundaries under different flush grouping
• recovery with WAL writes ahead of materialization
• setting changes mid-write
• preallocation cleanup that checks allocated blocks and message contents

i think PR for this will be huge 🥲

[djanand]

Thanks, these are really useful details and some that I overlooked. Still working through it.

[hubcio]

you better do that after #4561 is merged ;)

[djanand]

Thanks, this is all actionable. Going through 1 to 5, then the three below them:

On 1) expiry. Agreed on both counts. ~75 days, not six weeks. Keeping active-segment protection and adding the missing rotation trigger is what I'll build, with the trigger being the rule under 3.

On 2) deletion gates, offsets. Agreed, I'll reframe the post as finer retention granularity rather than a max message lifetime. The gates as I understand them: sparse writes mean the roll waits for the next append, the cleaner tick
(data_maintenance.messages.interval, 1m default), the checkpoint, since retention_is_ready
refuses while one is pending or while segment.end_offset >= checkpoint.next_offset, and committed consumer offsets with no TTL. Idle partitions are now #4650.

On the consumer offset gate specifically, I hadn't known about it. Does it deserve its own issue? One stopped consumer quietly disables retention for a topic, including max_topic_size, since remove_sealed_segments_up_to is the only removal path, and all you get is a log line. Not something this change should try to fix.

On 3) WAL placement. Agreed, and the rule moves into reserve. The rule: seal when the arriving batch's timestamp is more than segment_ms past the open segment's first batch timestamp. The primary mints both in stamp_and_append_messages and backups validate rather than assign, so the decision is a function of the committed batches and nothing else. No clock read anywhere in it.

Against the three conditions you listed:

• per batch, independent of flush grouping: reserve sees every batch, so grouping can't move a boundary
• restart and rollback: covered by persisting the timestamp, below
• state transfer: persisted is the primary path, with the install path reading the first batch header out of the installed segment as the fallback, since it already opens and reads it

On persist vs recover, I'd persist as the primary path and recover only in the install path.

Two reasons. The first is direction of dependency: today the WAL decides placement and the segment file follows. Recovering the first timestamp from the segment would make WAL placement depend on segment file contents, so the dependency runs both ways, and that is the coupling the placement check exists to police in the first place. The second is practical:
segment_boundary is a synchronous walk over in-memory entries, so a file read there means making it async and changing its callers, whereas reset_with_segment_checkpoint is already async and already opens the segment, so reading the first batch header there is nearly free. Hence recover where there is genuinely no value to carry, persist everywhere else.

So: one new field, the open segment's first batch timestamp, as an eleventh u64 in the frontier block. SEGMENT_STATE_BYTES grows and SEALED_STATE_MAGIC_OFFSET is derived from it, so nothing after it has to move by hand.

On the magic, I'd bump IGGYWAL4 to IGGYWAL5. I looked at carrying it on a flag inside the block instead, following SEGMENT_REFERENCES_FLAG, but that works there because the flag changes the meaning of existing bytes rather than the block's length. Here the field sits before the sealed-magic copy, so the copy moves either way. Without a bump an old build still passes the leading magic check in JournalState::decode, then fails the trailing sealed-magic comparison and reads the block as torn rather than as a newer format. Bumping makes it refuse the open instead, which is the same argument the STATE_MAGIC doc already makes for IGGYWAL4. The release will likely move STORAGE_FORMAT anyway, so this is belt and braces, but it costs one constant and gives the right error.

And yes on materialization. Without it the WAL rolls at the right batch while the seal waits for 1024 messages or 1 MiB that a sparse topic may never produce. Two separate pieces, as far as I can tell:

• making it due: should_persist_messages already escapes the two thresholds when active_segment().is_full(), so a time roll wants a sibling condition there rather than a new trigger.
• doing the rotation: drive_materialization decides it on segment.size >= segment.max_size, which a time roll never satisfies. I would rather the logical side follow the WAL than re-evaluate the rule, since two implementations of one rule is the divergence we are trying to avoid.validate_segment_prefix already reads reference.start_offset, so a reference pointing past the active segment is the roll signal. Today that difference is the fatal mismatch; it would become the instruction to rotate first and validate after.

On 4) runtime knobs, build identity. Going create-only, so to answer "which batch first uses a new value": none, it can't change after create.

On 5) trim. Agreed, nothing to write for the truncate itself.

On the length, seal is the wrong gate. It is logical only and says nothing about whether the WAL is finished with the bytes. The file still carries both names at that point (hard_link(&public, &retained), and the preallocation is even requested through the retained one), and a roll back into it is still possible: the comment at the open says a view change "can truncate a suffix that rolled into this offset".

The gate I would use is the one retention already applies, in retention_is_ready: no checkpoint in flight, and segment.end_offset < checkpoint.next_offset. Simplest argument for reusing it is that
retention UNLINKS the whole file once that holds, and a trim is strictly less destructive than an unlink, so anything that gate already permits covers this too. The length to trim to is then segment.size, which is final once the segment is sealed.

Taking you up on the nonfatal part. A failed trim logs and retries on the next pass, with no change to how persistence or recovery handle their own errors.

[djanand]

On defaults. Opt-in, and the cleanest way I can see is OptionDefault::None with CreateOnly, the same shape EncryptionKey already uses. Then admission persists nothing and the two cases you asked about collapse into one:

• new topics: no key written, feature off
• existing topics: no key either, feature off

So absent is the only off state, and there is nothing to migrate or special-case on read.

On the floor/ lower bound Agreed. Worth saying first that the code already fails safe here: the offer builder checks the entry count before encoding and returns ManifestTooLarge rather than tripping the encoder's assert, with a comment noting a small configured segment size makes such a chain ordinary. So this is about catching misconfiguration early, not about preventing a crash, which is why I would keep it light.

MIN_TOPIC_SEGMENT_MS of 1 minute, as the direct mirror of MIN_TOPIC_SEGMENT_SIZE
One minute is picked by comparison rather than taste, so that the time floor lands about where the size floor already does:

floor	value	what 65536 segments needs
MIN_TOPIC_SEGMENT_SIZE	1 MiB	64 GiB in one partition, ~44 days at 17 kB/s
MIN_TOPIC_SEGMENT_MS	1 minute	65536 minutes, ~45 days

For the running case I would copy the cleaner's disclosure for an unenforceable max_topic_size: log once when a partition's segment count gets close to the cap.

On tests. All six. Adding one: the roll decision should return which trigger fired, otherwise a test can't tell a time roll from a size roll that happened to land in the same place.