feat: support micro batching alternative at source

[vaibhavtiwari33]

• Currently source reads 1 batch of messages at a time and waits for ack/nack of the whole batch of messages it read and sent downstream.
• This is fine for pipeline which can read multiple batches (read ahead)
• In case of monovertex, it essentially blocks reading of additional messages until the current batch gets fully acked/nacked (which can be delayed in cases of large tail latency).

What this PR does / why we need it

This PR aims to add an opt-in feature for monovertex to enable per message ack loop (instead of per batch acking)

• Added a field in monovertex spec to enable streaming (off by default)
• When streaming is enabled, acks for messages are tracked individually instead of in a batch
• Max in-flight messages are still capped using concurrency
• Semaphores (==concurrency) are distributed per message instead of per batch
• Most of the new logic lives in streaming_source in source.rs where we're doing the same operations as before, but on a per message level.

Implemented spec:

kind: MonoVertex
metadata:
  name: my-mvtx
spec:
  streaming: true
...

Related issues

#3452

Testing

• Unit tests and an e2e test was added
• Tested with running a normal ud source with and without streaming enabled

Tracing Test:

Streaming disabled; transformer returns results

Streaming enabled: transformer returns results

Streaming disabled; transformer omits results

Streaming enabled; transformer omits results

Special notes for reviewers

This is a monovertex only feature currently since pipeline already supports reading multiple batches.

[codecov]

Codecov Report

❌ Patch coverage is 87.37728% with 90 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.67%. Comparing base (38f3306) to head (639f079).

Files with missing lines	Patch %	Lines
rust/numaflow-core/src/source.rs	86.77%	82 Missing ⚠️
rust/numaflow-core/src/shared/create_components.rs	33.33%	6 Missing ⚠️
pkg/apis/numaflow/v1alpha1/mono_vertex_types.go	0.00%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3450      +/-   ##
==========================================
+ Coverage   82.55%   82.67%   +0.12%     
==========================================
  Files         307      307              
  Lines       77618    78169     +551     
==========================================
+ Hits        64077    64629     +552     
+ Misses      12984    12983       -1     
  Partials      557      557              

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

did you test tracing with this?

[vaibhavtiwari33]

I haven't tested end to end tracing with this change but all similar tracing changes have been made as the normal path, so should be good. I'll test this branch as well.

One more change that needs to be made in a separate PR is message level metrics (replacing batch level metrics for source)

[vaibhavtiwari33]

You're right, I missed creating spans for transformer in this branch. Fixing this.

[vaibhavtiwari33]

Updated, ty for catching this!

[vigith]

added a WARN if streaming and readAhead are set simultaneously

[vaibhavtiwari33]

Since currently streaming is a mvtx only feature, I'll move this inside the following block:

if self.streaming {
    self.streaming_source(
        pipeline_labels,
        mvtx_labels,
        bypass_router,
        messages_tx,
        cln_token,
    )
    .await
}