refactor: Merge Heartbeat and OT Buckets for Watermark Progression

[yhl25]

Problem Statement

The existing watermark system used two separate KV buckets for tracking watermark progression:

Offset-Timeline (OT) Bucket: Stored watermark-offset mappings to track message progression through the pipeline.
Heartbeat (HB) Bucket: A dedicated bucket solely for tracking processor liveness, processors would periodically write to this bucket to signal they're alive.

This dual-bucket architecture introduced several challenges:

• Synchronization Complexity: Having two sources of truth for processor state created race conditions and synchronization issues. A processor could appear "alive" in the heartbeat bucket while its OT entries were stale, or vice versa. The system had to reconcile state from both buckets, adding complexity to processor lifecycle management.

• No Single Source of Truth: When determining if a processor is alive and what its current watermark is, the system had to consult both buckets. This made reasoning about the system harder and introduced edge cases where the two buckets could be out of sync.

• Infrastructure Overhead: Each watermark-enabled edge required creating, managing, and monitoring an extra KV bucket. This doubled the number of buckets in the system for watermark tracking.

Additionally, for source watermarks, there was no mechanism to know how many processors (partitions) should be reporting before the watermark could be considered valid. During startup or rebalancing, fetching watermarks before all processors were up would result in incorrect watermark values.

Solution

Unified Bucket with KV Creation Timestamps: Instead of maintaining a separate heartbeat bucket, we now use the KV store's native entry creation timestamp (KVEntry.created) as the liveness signal. When a processor publishes a WMB to the OT bucket, the KV store records when that entry was created. The ProcessorManager uses this timestamp to determine if a processor is still alive, if the entry is too old, the processor is marked inactive or deleted.

Single Source of Truth: All processor state (watermark, offset, and liveness) now comes from a single OT bucket. The ProcessorManager watches this bucket for updates and uses the entry timestamps to track processor health. No reconciliation between buckets is needed.

Processor Count in WMB for Source Watermarks: Sources now include partition information in the WMB itself:

• Each source partition publishes its watermark with processor_count indicating the total number of partitions.
• The SourceWatermarkFetcher reads the maximum processor_count from all active processors' head WMBs.
• Watermarks are only computed when the number of active processors matches the expected count.
• This ensures watermark stability during startup, we wait for all processors to be up before fetching a valid watermark.

Periodic Publishing for Liveness : The ISBWatermarkPublisher now publishes WMBs periodically (based on a configurable delay) even when the watermark hasn't changed. This ensures the KV entry timestamps stay fresh, allowing downstream vertices to detect processor liveness without a separate heartbeat mechanism.

Key Changes

• Removed HB bucket creation from the controller, only OT buckets are created now.
• ProcessorManager determines liveness using KVEntry.created timestamp instead of watching a separate HB bucket.
• WMB struct now includes an optional processor_count field for source watermarks.
• SourceWatermarkFetcher waits for all expected processors (based on processor_count in WMBs) before computing a valid watermark.
• Source implementations (Kafka, Pulsar, JetStream, NATS, SQS, etc.) now report active_partitions and total_partitions, enabling proper watermark coordination.
• SourceIdleDetector handles both heartbeat publishing (via periodic WMB updates) and idle detection.

Testing

• Unit tests are updated to progress watermark using single bucket.
• Performance testing is done and we don't see any regression, we see slight improvements in all the parameters.

[codecov]

Codecov Report

❌ Patch coverage is 87.75304% with 121 lines in your changes missing coverage. Please review.
✅ Project coverage is 80.91%. Comparing base (81c82e4) to head (950db9e).
⚠️ Report is 7 commits behind head on main.

Files with missing lines	Patch %	Lines
rust/extns/numaflow-kafka/src/source.rs	0.00%	32 Missing ⚠️
rust/numaflow-core/src/source.rs	48.38%	16 Missing ⚠️
rust/numaflow-core/src/source/kafka.rs	0.00%	14 Missing ⚠️
...t/numaflow-core/src/watermark/processor/manager.rs	90.42%	9 Missing ⚠️
rust/numaflow-core/src/watermark/source.rs	89.23%	7 Missing ⚠️
rust/numaflow-core/src/shared/create_components.rs	25.00%	6 Missing ⚠️
...ust/numaflow-core/src/config/pipeline/watermark.rs	54.54%	5 Missing ⚠️
rust/numaflow-core/src/source/generator.rs	0.00%	5 Missing ⚠️
rust/numaflow-core/src/source/nats.rs	0.00%	5 Missing ⚠️
rust/numaflow-core/src/source/sqs.rs	16.66%	5 Missing ⚠️
... and 6 more

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3287      +/-   ##
==========================================
- Coverage   81.10%   80.91%   -0.19%     
==========================================
  Files         316      316              
  Lines       72304    72770     +466     
==========================================
+ Hits        58639    58884     +245     
- Misses      13107    13331     +224     
+ Partials      558      555       -3     

☔ View full report in Codecov by Sentry.
📢 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.