Numaflow 1.8.1 breaks accumulator with `Timestamp is not valid 8210266876799`

[th0ger]

Describe the bug
We are testing upgrade from numaflow 1.7.2to 1.8.1, but our accumulator UDFs break with errors:

CRITICAL Accumulator Error: ValueError('Timestamp is not valid: Seconds 8210266876799 must be in range [-62135596800, 253402300799].')

We have multiple independent accumulator pipelines throwing this exact same error, including the same numeric values. It is NOT invalid user data.
We have tried to upgrade to the latest pynumaflow 0.13.0
Non-accumulator pipelines are not affected.
We run a rust-based streamsorter accumulator that is not affected.

The range [-62135596800, 253402300799] is related to the valid range of protobuf timestamps:

Returns true if the given Timestamp is valid. The seconds value must be in the range [-62,135,596,800, +253,402,300,799] (i.e., between 0001-01-01T00:00:00Z and 9999-12-31T23:59:59Z). The nanos value must be in the range [0, +999,999,999].

The errorneous value 8210266876799 is "Friday, March 5, 2230 at 5:41:16.799 AM", i.e. ~1970+260yr.
This looks suspiciously close to the MAX value of Rust NaiveDateTime:

Panics if the number of seconds would be out of range for a NaiveDateTime (more than ca. 262,000 years away from common era)

To Reproduce

This minimal example reproduces the issue.

1. myaccumulator python image:

from collections.abc import AsyncIterable
from pynumaflow.accumulator import AccumulatorAsyncServer, Accumulator, Datum, Message
from pynumaflow.shared.asynciter import NonBlockingIterator

class DemoAccumulator(Accumulator):
    async def handler(
        self, datums: AsyncIterable[Datum], output: NonBlockingIterator
    ) -> None:
        async for datum in datums:
            print(f"Received datum value={datum.value[:200]}")
            await output.put(Message.to_drop())
            continue


def main() -> None:
    server = AccumulatorAsyncServer(DemoAccumulator)
    server.start()

if __name__ == "__main__":
    main()

2. pipeline spec

apiVersion: numaflow.numaproj.io/v1alpha1
kind: Pipeline
metadata:
  name: mypipeline
spec:
  edges:
  - from: myinput
    to: myaccumulator
  - from: myaccumulator
    to: myoutput
  vertices:
  - name: myinput
    source:
      generator:
        duration: '1s'
        jitter: '0s'
        msgSize: 8
        rpu: 5
  - name: myaccumulator
    udf:
      container:
        image: myimage
      groupBy:
        storage:
          emptyDir: {}
        window:
          accumulator:
            timeout: '60s'
  - name: myoutput
    sink:
      log: {}

3. Full log numa+udf:

stream closed: EOF for default/mypipeline-myaccumulator-0-cbflc (init)
init {"level":"info","ts":"2026-06-29T13:53:56.400749164Z","logger":"numaflow.isbsvc-validate","caller":"commands/isbsvc_validate.go:75","msg":"Buffers, buckets and side inputs store might have not been created yet, will retry if the limit is not reached","pipeline":"mypipeline","error":"failed to query information of stream \"default-mypipeline-myaccumulator-0\", nats: stream not found"}
init {"level":"info","ts":"2026-06-29T13:54:01.716493898Z","logger":"numaflow.isbsvc-validate","caller":"commands/isbsvc_validate.go:84","msg":"Validate buffers, buckets and side inputs store successfully","pipeline":"mypipeline"}
udf    Building myimage @ file:///app
udf       Built myimage @ file:///app
udf Uninstalled 1 package in 1ms
udf Installed 5 packages in 545ms
udf 2026-06-29 13:54:08 CRITICAL Accumulator Error: ValueError('Timestamp is not valid: Seconds 8210266876799 must be in range [-62135596800, 253402300799].')
udf Traceback (most recent call last):
udf   File "/app/.venv/lib/python3.12/site-packages/pynumaflow/accumulator/servicer/task_manager.py", line 231, in process_input_stream
udf     async for request in request_iterator:
udf   File "/app/.venv/lib/python3.12/site-packages/pynumaflow/accumulator/servicer/async_servicer.py", line 28, in datum_generator
udf     end=d.operation.keyedWindow.end.ToDatetime(),
udf         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
udf   File "/app/.venv/lib/python3.12/site-packages/google/protobuf/internal/well_known_types.py", line 258, in ToDatetime
udf     _CheckTimestampValid(self.seconds, self.nanos)
udf   File "/app/.venv/lib/python3.12/site-packages/google/protobuf/internal/well_known_types.py", line 317, in _CheckTimestampValid
udf     raise ValueError(
udf ValueError: Timestamp is not valid: Seconds 8210266876799 must be in range [-62135596800, 253402300799].
udf 2026-06-29 13:54:08 CRITICAL UDF_EXECUTION_ERROR(udf): ValueError('Timestamp is not valid: Seconds 8210266876799 must be in range [-62135596800, 253402300799].')
udf NoneType: None
udf 2026-06-29 13:54:08 CRITICAL Traceback (most recent call last):
udf   File "/app/.venv/lib/python3.12/site-packages/pynumaflow/accumulator/servicer/task_manager.py", line 231, in process_input_stream
udf     async for request in request_iterator:
udf   File "/app/.venv/lib/python3.12/site-packages/pynumaflow/accumulator/servicer/async_servicer.py", line 28, in datum_generator
udf     end=d.operation.keyedWindow.end.ToDatetime(),
udf         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
udf   File "/app/.venv/lib/python3.12/site-packages/google/protobuf/internal/well_known_types.py", line 258, in ToDatetime
udf     _CheckTimestampValid(self.seconds, self.nanos)
udf   File "/app/.venv/lib/python3.12/site-packages/google/protobuf/internal/well_known_types.py", line 317, in _CheckTimestampValid
udf     raise ValueError(
udf ValueError: Timestamp is not valid: Seconds 8210266876799 must be in range [-62135596800, 253402300799].
udf 2026-06-29 13:54:08 CRITICAL UDF_EXECUTION_ERROR(udf): ValueError('Timestamp is not valid: Seconds 8210266876799 must be in range [-62135596800, 253402300799].')
udf 2026-06-29 13:54:08 CRITICAL Server exiting due to UDF error: Timestamp is not valid: Seconds 8210266876799 must be in range [-62135596800, 253402300799].
monitor {"timestamp":"2026-06-29T13:54:02.051891Z","level":"INFO","message":"Numaflow runtime details","VERSION_INFO":"Version: v1.8.1, BuildDate: 2026-06-18T02:26:33Z, GitCommit: ff0daa90289fbc3a7bf274c534827e3e0265ad06, GitTag: v1.8.1, GitTreeState: clean, RustVersion: rustc 1.95.0 (59807616e 2026-04-14), Platform: linux/x86_64","tokio_worker_threads":1,"target":"numaflow"}
monitor {"timestamp":"2026-06-29T13:54:02.052082Z","level":"INFO","message":"Starting monitor","target":"numaflow"}
monitor {"timestamp":"2026-06-29T13:54:02.052713Z","level":"INFO","message":"Starting monitor server with config","server_config":"MonitorServerConfig { server_listen_port: 2470, graceful_shutdown_duration: 30 }","target":"numaflow_monitor"}
monitor {"timestamp":"2026-06-29T13:54:02.052819Z","level":"INFO","message":"Starting monitor app server..","app_addr":"0.0.0.0:2470","target":"numaflow_monitor::app"}
numa {"timestamp":"2026-06-29T13:54:07.087337Z","level":"INFO","message":"Numaflow runtime details","VERSION_INFO":"Version: v1.8.1, BuildDate: 2026-06-18T02:26:33Z, GitCommit: ff0daa90289fbc3a7bf274c534827e3e0265ad06, GitTag: v1.8.1, GitTreeState: clean, RustVersion: rustc 1.95.0 (59807616e 2026-04-14), Platform: linux/x86_64","tokio_worker_threads":1,"target":"numaflow"}
numa {"timestamp":"2026-06-29T13:54:07.087522Z","level":"INFO","message":"Starting processing pipeline","target":"numaflow"}
numa {"timestamp":"2026-06-29T13:54:07.087631Z","level":"INFO","message":"Loaded pipeline spec: Vertex [...]

Expected behavior
Upgrade from 1.7 to 1.8 without issue.

Environment (please complete the following information):

• Numaflow: v1.8.1
• Numaflow-python: v0.13.0

---

Message from the maintainers:

Impacted by this bug? Give it a 👍. We often sort issues this way to know what to prioritize.

[vigith]

We are working on cutting a new release which includes #360 /cc @vaibhavtiwari33