fix: cleanup wal on eof in accumulator

[yhl25]

No description provided.

[codecov]

Codecov Report

❌ Patch coverage is 98.80952% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.69%. Comparing base (9bd8b83) to head (7aee66d).

Files with missing lines	Patch %	Lines
...aflow-core/src/reduce/reducer/unaligned/reducer.rs	98.42%	3 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3461      +/-   ##
==========================================
+ Coverage   82.68%   82.69%   +0.01%     
==========================================
  Files         307      307              
  Lines       77569    77773     +204     
==========================================
+ Hits        64135    64318     +183     
- Misses      12876    12898      +22     
+ Partials      558      557       -1     

☔ 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.

[vigith]

should we merge to dev-xxx branch and give a test image for @tmenjo to test?

[tmenjo]

@yhl25 @vigith Thank you for create a patch!

should we merge to dev-xxx branch and give a test image for @tmenjo to test?

Thank you for being considerate. That won't be necessary. I've fetched the fix-eof-accum branch and built my own image.

[yhl25]

@yhl25 @vigith Thank you for create a patch!

should we merge to dev-xxx branch and give a test image for @tmenjo to test?

Thank you for being considerate. That won't be necessary. I've fetched the fix-eof-accum branch and built my own image.

Also, in the accumulator can you make the below change to drop the unused messages once the stream has been ended(timeout), so that the watermark progresses and the WAL gets cleaned up.

self.logger.info('out of datums loop')

# The datums stream has ended (timeout or stream close). Any buffered source
# frames or inference results left here will never find a match, so drop them
# while carrying their watermark forward to progress the output watermark and
# release the remaining WAL state.
while self.sorted_source_frames:
    _, stale_datum = self.sorted_source_frames.popitem(index=0)
    await output.put(Message.to_drop(stale_datum))
while self.sorted_inference_results:
    _, stale_datum = self.sorted_inference_results.popitem(index=0)
    await output.put(Message.to_drop(stale_datum))
self.logger.info('dropped all unmatched buffered datums after datums loop')

You can use this numaflow-python branch to get the drop semantics for accumulator.

[tmenjo]

@yhl25 Thank you for your help.

I've run my pipeline fixed as yhl25 told me for 17 hours on Numaflow 685a84b and pynumaflow numaproj/numaflow-python@0b8e3e1 . However, the memory usage of the numa container (yellow line) have not reached a ceiling, as with the issue #3262.

I don't know if it's related, but the memory usage looks to have periodicity. It bumps in every 140-150 minutes as far as I can see. The input video stream of my pipeline also has periodicity, but its cycle is different. It repeats a 46-second video file forever.

Debug logs are as follows. Sorry I don't know but they ended at 08:41:24.

• numa-20260610-043000.tar.gz
• numa-20260610-050000.tar.gz
• numa-20260610-053000.tar.gz
• numa-20260610-060000.tar.gz
• numa-20260610-063000.tar.gz
• numa-20260610-070000.tar.gz
• numa-20260610-073000.tar.gz
• numa-20260610-080000.tar.gz
• numa-20260610-083000.tar.gz