fix: cleanup wal on eof in accumulator

[yhl25]

• Track closed windows which are yet to be GC'd for watermark propagation
• Send closed window information in the close request so that the SDK's can send the window information in the eof response.

[codecov]

Codecov Report

❌ Patch coverage is 98.80952% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.88%. Comparing base (02489d3) to head (1072f39).

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.86%   82.88%   +0.01%     
==========================================
  Files         308      308              
  Lines       78820    79024     +204     
==========================================
+ Hits        65318    65497     +179     
- Misses      12932    12960      +28     
+ Partials      570      567       -3     

☔ 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

[tmenjo]

I might have found a solution. My reduce merges a source frame and a inference result and outputs a merged datum. Here, two input datums but only one output. I think I should drop the other. I'm trying the following changes:

        # Output a message with fields (except value) restored from source
        await output.put(
            Message(
                keys=datum_source.keys,
                value=msgspec.msgpack.encode(payload_out),
                watermark=datum_source.watermark,
                event_time=datum_source.event_time,
                headers=datum_source.headers,
                id=datum_source.id,
            )
        )

        # Drop the leftover (inference datum) for watermark and WAL
        await output.put(Message.to_drop(datum_inference))  # <===== here

[vigith]

I might have found a solution. My reduce merges a source frame and a inference result and outputs a merged datum. Here, two input datums but only one output. I think I should drop the other.

Ideally the timeout should reap it via GC, @yhl25 wdyt?

[tmenjo]

I still have the issue. I have tried the changes to my accumulator described above, but the memory usage of the numa container have not reached a ceiling. Here is that chart for the last 24 hours:

[vigith]

I still have the issue. I have tried the changes to my accumulator described above, but the memory usage of the numa container have not reached a ceiling. Here is that chart for the last 24 hours:

Can you share the new logs?

[tmenjo]

Can you share the new logs?

Sorry but I forgot to collect log... I will run the test again today. Please stay tuned.

[vaibhavtiwari33]

These windows have been closed (sent to the SDK)

nit: this should to be reworded.
This makes it sounds like only closed windows are sent to the SDK

User's validation pending

[vaibhavtiwari33]

@tmenjo Let me create a similar fix for the python SDK as well to help you test the with the correct state that the client expects (echo back of closed keyed window).

[tmenjo]

@vigith Thank you for waiting. I've run my pipeline again and collected debug logs.

@tmenjo Let me create a similar fix for the python SDK as well to help you test the with the correct state that the client expects (echo back of closed keyed window).

Thank you so much! Please let me know when you update the Python SDK.

Chart

The yellow line is for the memory usage of the numa container.

Logs

Sorry but there are logs only for the first 4 and a half hours. (How I collect logs may be incorrect. I'll fix it next time.)

• numa-20260615-050000.tar.gz
• numa-20260615-053000.tar.gz
• numa-20260615-060000.tar.gz
• numa-20260615-063000.tar.gz
• numa-20260615-070000.tar.gz
• numa-20260615-073000.tar.gz
• numa-20260615-080000.tar.gz
• numa-20260615-083000.tar.gz
• numa-20260615-090000.tar.gz

[vaibhavtiwari33]

Please let me know when you update the Python SDK.

@tmenjo I've created a PR for Python SDK: numaproj/numaflow-python#360
Please let me know if you're able to utilize this for your testing.

[tmenjo]

@vaibhavtiwari33 Thanks! I'll fetch your pull-request branch and build my own pynumaflow image for my testing, but I have a question: Is the PR numaproj/numaflow-python#358 that @yhl25 told me on this thread also required for fixing the issue?

[vaibhavtiwari33]

Is the PR numaproj/numaflow-python#358 that @yhl25 told me on this thread also required for fixing the issue?

No, you don't need to add explicit drops when running numaproj/numaflow-python#360 SDK on #3461 (this PR)

[vaibhavtiwari33]

Changes verified against both numaproj/numaflow-python#360 and numaproj/numaflow-rs#177

I see consistent memory/disk usage across 2-3 hours when the user code doesn't explicitly returns mapped responses and windows are closed after timeout.

[tmenjo]

@vigith @vaibhavtiwari33 I've run my pipeline for 24 hours on Numaflow nightly-20260618 (51747b2) and numaflow-python numaproj/numaflow-python@e6434ba. The memory usage of the numa container have been drastically improved, but unfortunately it have not reached a ceiling yet.

Screenshot

The yellow line is for the memory usage of the numa container. It looks like a sawtooth wave increasing gradually. I'd say this tendency can be seen on long-running test.

(The Time (X axis) is in UTC.)

Debug logs

Each timestamp is in UTC+9.

• 0.log.20260623-095602.gz
• 0.log.20260623-102634.gz
• 0.log.20260623-105716.gz
• 0.log.20260623-112807.gz
• 0.log.20260623-115838.gz
• 0.log.20260623-122909.gz
• 0.log.20260623-125950.gz
• 0.log.20260623-133031.gz
• 0.log.20260623-140113.gz
• 0.log.20260623-143144.gz
• 0.log.20260623-150225.gz
• 0.log.20260623-153306.gz
• 0.log.20260623-160347.gz
• 0.log.20260623-163418.gz
• 0.log.20260623-170450.gz
• 0.log.20260623-173520.gz
• 0.log.20260623-180602.gz
• 0.log.20260623-183643.gz
• 0.log.20260623-190714.gz
• 0.log.20260623-193745.gz
• 0.log.20260623-200826.gz
• 0.log.20260623-203907.gz
• 0.log.20260623-210939.gz
• 0.log.20260623-214010.gz
• 0.log.20260623-221051.gz
• 0.log.20260623-224132.gz
• 0.log.20260623-231213.gz
• 0.log.20260623-234244.gz
• 0.log.20260624-001315.gz
• 0.log.20260624-004357.gz
• 0.log.20260624-011438.gz
• 0.log.20260624-014509.gz
• 0.log.20260624-021540.gz
• 0.log.20260624-024621.gz
• 0.log.20260624-031703.gz
• 0.log.20260624-034744.gz
• 0.log.20260624-041825.gz
• 0.log.20260624-044856.gz
• 0.log.20260624-051947.gz
• 0.log.20260624-055028.gz
• 0.log.20260624-062059.gz
• 0.log.20260624-065131.gz
• 0.log.20260624-072212.gz
• 0.log.20260624-075253.gz
• 0.log.20260624-082324.gz
• 0.log.20260624-085405.gz
• 0.log.20260624-092436.gz

Environment

• Numaflow: nightly-20260618 (51747b2)
• numaflow-python: numaproj/numaflow-python@e6434ba
• Kubernetes: v1.35.2