[Prism] Honor the resume delay of self-checkpointing SDF residuals

[Eliaaazzz]

Fixes #39848.

The bug

PersistBundle decoded the SDK's requested resume delay into engine.Residual.Delay and then returned every residual to the pending queue immediately, at the TODO "actually reschedule based on the residuals delay". A polling SDF therefore busy-spins: on a master build of Prism, a Python Watch poll loop requesting 3 second pacing via defer_remainder executed 7,034 poll rounds in 9 seconds of wall time.

The change

PersistBundle partitions residuals by delay. A residual with no delay returns to pending as before. A delayed residual is parked on its stage in stageState.delayedResiduals, keyed by the processing time it becomes schedulable, and that time is scheduled through the machinery processing-time timers already use: ptRefreshes → em.processTimeEvents.Schedule → em.wakeUpAt. The watermark evaluation loop releases due parked residuals back to pending before checking bundle readiness.

Parked elements pin the stage's input watermark by being included in minPendingTimestampLocked, the same way pending elements do. An earlier revision used watermarkHolds instead; those clamp only the output watermark, and since the input watermark is monotonic, a bounded pipeline's global-window aggregation could fire early with partial data once the input watermark ran ahead during a park. The TestSeparation ProcessContinuations subtests catch exactly this and pass with the final design.

Under the fast-forward clock (EnableRTC=false), processingTimeNow peeks the event queue, so delayed residuals still fire immediately in synthetic time and test pipelines stay fast. Under the default real-time clock the delay is honored; --experiments=prism_disable_rtc remains the opt-out. The ProcessContinuations subtests of TestSeparation now take real time (~15s instead of ~4s locally) because the 1 second resume delays they request are honored.

Verification

• New TestPersistBundle_ResidualResumeDelay engine tests: a delayed residual is parked rather than pending, pins minPendingTimestampLocked, and schedules a processing time event; under a real-time clock it fires only after the delay; under the fast-forward clock it fires without real waiting. Reverting the parking fails the first two subtests.
• Full runners/prism/... Go tests pass.
• End to end with the Python SDK against a rebuilt prism binary: the 3s-pacing probe went from 7,034 polls in 9 seconds to 4 polls at 3.01 second spacing; a MatchContinuously pipeline that previously missed files added mid-run on Prism now passes; Watch pipelines in both deduplication modes still pass.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

[codecov]

Codecov Report

❌ Patch coverage is 91.22807% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 59.69%. Comparing base (c101795) to head (f85f0cb).
⚠️ Report is 49 commits behind head on master.

Files with missing lines	Patch %	Lines
...am/runners/prism/internal/engine/elementmanager.go	91.22%	3 Missing and 2 partials ⚠️

Additional details and impacted files

@@            Coverage Diff            @@
##             master   #39849   +/-   ##
=========================================
  Coverage     59.69%   59.69%           
+ Complexity    16818    16813    -5     
=========================================
  Files          2863     2863           
  Lines        297877   297972   +95     
  Branches      14679    14679           
=========================================
+ Hits         177805   177875   +70     
- Misses       112468   112486   +18     
- Partials       7604     7611    +7     

Flag	Coverage Δ
go	29.09% <91.22%> (+0.06%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[Eliaaazzz]

Run GoPrism PreCommit

[Eliaaazzz]

Run Java_PVR_Prism_Loopback PreCommit

[github-actions]

Assigning reviewers:

R: @shunping for label go.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).