[fix][broker] Fix bucket snapshot trim and segment loading

[nodece]

Motivation

Fix delayed-delivery bucket recovery and trim races that could stall progress or break cleanup:

• Segment-boundary advancement could be skipped when the current tail entry is orphaned (ledger below first live ledger), leaving the next snapshot segment unloaded.
• Trim/delete could race with snapshot creation and bucket-id resolution, including failed-create and missing/malformed cursor-property paths.
• Cursor reset needed to wait for delayed-delivery cleanup so in-flight trim/delete does not overlap with reset/replay.

Modifications

• In getScheduledMessages(), process snapshot-segment boundary transition before dropping an orphaned tail entry, while still suppressing delivery of that orphaned entry.
• Kept cutoff gating behavior: next segment is not loaded before cutoff and is loaded once cutoff is reached.
• Hardened bucket-id resolution in ImmutableBucket.asyncDeleteBucketSnapshot():
  • resolve delete bucket id via Optional
  • skip storage delete when create future failed
  • propagate missing/malformed recovered bucket id as failed CompletableFuture (no synchronous throw)
• Kept trim lifecycle checks in tracker delete flow with ownership revalidation before delete start and again before in-memory state removal.
• Made resetCursor() clear delayed delivery state and wait for any in-flight trim to finish before completing the reset.
• Added deterministic regressions in BucketDelayedDeliveryTrackerTest:
  • testDoesNotLoadNextSnapshotSegmentBeforeCutoff
  • testLoadsNextSnapshotSegmentAfterCutoff
  • testTrimDefersDeleteUntilSnapshotCreateCompletes
  • testTrimDeletesAfterCreateCompletesDespiteMarkDeleteChange
  • testResetCursorWaitsForDelayedMessagesClear

[nodece]

@Denovo1998 Fix: #26251 (comment)

[void-ptr974]

Two edge cases remain around resolving the bucket ID.

[void-ptr974]

Nice improvement to make deletion wait for snapshot creation; it closes the missing bucketId race. Two edge cases remain:

1. If snapshot creation fails after trim has started, afterCreateImmutableBucket() resolves the future with INVALID_BUCKET_ID (-1), and this path calls deleteBucketSnapshot(-1). A storage failure would stop the sequential trim chain and skip the subsequent merge. Since no snapshot was persisted, storage deletion should be skipped while orphan-bucket cleanup continues.
2. When no creation future exists, getAndUpdateBucketId() is evaluated before completedFuture() is created. A missing or malformed cursor property therefore throws synchronously. The failure should instead be propagated through the returned CompletableFuture so callers can handle it consistently.

[nodece]

Thanks for the detailed review. Addressed both points in the latest commits:

1. asyncDeleteBucketSnapshot() now resolves bucket id via Optional and skips storage deletion when snapshot creation failed, so trim does not call deleteBucketSnapshot(-1) and the chain is not broken by the old sentinel path.
2. The no-create-future path now catches getAndUpdateBucketId() failures and returns a failed CompletableFuture instead of throwing synchronously, so callers handle failures consistently through the async chain.

[Denovo1998]

Waiting for snapshot creation here resolves the missing bucketId race condition, but the trim decision itself is made earlier in BucketDelayedDeliveryTracker.asyncTrimImmutableBuckets(). Once this future completes, the range is no longer revalidated—neither to confirm it is still orphaned nor to verify that the same bucket remains mapped to the selected range.

#26260 performed both checks after the snapshot‑create future completed, and this PR is intended to subsume that fix. In the current implementation, a cursor reset, backward movement of the mark‑delete position, or a bucket mapping change while snapshot creation is pending can make the original trim decision stale before storage deletion begins.

Could this lifecycle coordination be kept within the tracker—revalidating firstActiveLedgerId() and the exact range‑to‑bucket identity before starting deletion? Ideally, the ownership/eligibility guard should also cover the asynchronous delete completion before tracker state is removed.

[nodece]

Good point on stale trim decisions. The tracker now keeps lifecycle coordination in BucketDelayedDeliveryTracker: it revalidates exact range->bucket ownership before starting deletion and revalidates ownership again before removing tracker state in the async completion. This prevents deleting/removing state for a bucket that has been replaced while delete is in flight.

[void-ptr974]

Thanks for hardening the snapshot-create/delete path. One lifecycle issue remains: asyncTrimImmutableBuckets() decides eligibility before asyncDeleteBucketSnapshot() waits for snapshot creation. The new checks only revalidate range-to-bucket ownership, not whether the range is still eligible after that wait.

For example, bucket [1, 5] is selected when mark-delete is 100. A reset to 3 while creation is pending makes [3, 5] live again, but ownership is unchanged, so the current flow still deletes its snapshot and indexes. Reset replay may see the old index, skip reinsertion, and then lose it when the pending trim completes.

I recommend using mark-delete only for delivery filtering and managedLedger.getLedgersInfo().firstKey() for destructive snapshot/index trim. The retained-ledger boundary is monotonic across cursor resets: [1, 5] becomes removable only after the boundary advances to 6, when those positions are no longer readable. Keeping mark-delete as the destructive boundary instead requires reset/replay to be fenced against the entire async create/delete lifecycle, which the current tracker lifecycle does not provide.

[nodece]

The reset cursor race has been fixed. When the reset cursor, the dispatcher clear the tracker data and then the memory data is consistent.

[void-ptr974]

Thanks, the new fencing closes the stale-trim race. My concern is that it couples reset and consumer recovery to the snapshot lifecycle, since consumers are disconnected before clear() waits for trimFuture and bucket cleanup. In the worst case, a pending snapshot creation could prevent asyncResetCursor() from starting and keep the subscription fenced.

[Denovo1998]

The PR description states that buckets with unfinished snapshot creation are skipped, but the current implementation instead waits for the creation future. Since asyncTrimImmutableBuckets() is part of the global trimFuture chain, a slow or stalled snapshot creation can keep trimFuture.isDone() false, which blocks subsequent trim/merge attempts and causes clear() to wait on the same future.

The latest commit also removed the prior isDone() filtering from asyncTrimImmutableBuckets(). Would it be safer to skip CREATING buckets and retrigger or rescan for trimming once creation finishes, rather than holding the global trim/merge gate? If waiting is intentional, we should at least define and test a bounded-completion guarantee.

[nodece]

Thanks, agreed this is important. The implementation keeps creation gating in the delete path and adds deterministic coverage to ensure we do not start deletion while create is blocked, then cleanup proceeds once create completes (testTrimDefersDeleteUntilSnapshotCreateCompletes). The trim/merge flow remains serialized by trimFuture intentionally so clear/trim observe a single ordered chain.

[Denovo1998]

The PR also addresses the snapshot-create/trim race condition mentioned in the motivation, though the new regression tests only cover snapshot-segment loading.

Could we add a deterministic trim test that:

• Blocks createBucketSnapshot()
• Triggers a trim while the bucket ID is still unavailable
• Verifies that deletion does not begin prematurely
• Then releases the snapshot creation and confirms that cleanup completes

It would also be useful to test revalidation when the bucket ceases to be orphaned while creation is pending. #26260 already includes deterministic coverage for both scenarios that could be adapted for this purpose.

[nodece]

Added deterministic tests for the requested race scenarios in BucketDelayedDeliveryTrackerTest:

• testTrimDefersDeleteUntilSnapshotCreateCompletes
• testTrimDeletesAfterCreateCompletesDespiteMarkDeleteChange

They block createBucketSnapshot(), trigger trim while bucket id is unavailable, verify no premature delete, then release create and assert cleanup behavior.

[Copilot]

Pull request overview

This PR addresses correctness issues in the broker’s delayed-delivery “bucket snapshot” flow, particularly around recovery/segment transitions and trim/delete races that could stall progress or prevent cleanup.

Changes:

• Adjust bucket snapshot segment loading behavior in getScheduledMessages() to handle segment-boundary transitions even when the tail entry is orphaned, while still suppressing delivery of that orphaned entry.
• Harden snapshot delete/trim lifecycles (bucket-id resolution, ownership revalidation during delete, and reset-cursor sequencing with delayed-delivery cleanup).
• Add regression tests covering cutoff-gated segment loading behavior.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.

File	Description
pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java	Adds deterministic tests validating cutoff-gated snapshot segment loading behavior.
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java	Ensures cursor reset waits for delayed-delivery cleanup before proceeding.
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java	Hardens bucket-id resolution for snapshot deletion and avoids sync throws during delete path resolution.
pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java	Fixes segment transition behavior and strengthens trim/delete flow with revalidation checks.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.