enh: refactor sync flow to channel-based pipeline

[evacchi]

Main commit is 16a2ca8

Optionally, commit 34450fd also deletes the old executeJob() (for now deprecate and call executeJobAsync()` to keep the diff smaller) and all related dead code.

Test cases should be already migrated to equivalent cases in the new flow.

We can also do the deletion in a separate PR.

---

Current design for the "sync" flow makes it harder to integrate a truly async processing flow.

In PR #581 we introducef a fully-async pipeline for llm-d-async integration, while we mention that the same design can be adapted for the sync case as well (where HTTP requests are handled directly).

The unified design makes it easier to handle both cases and should be overall more composable, this PR migrates to the same design the "sync" flow as well.

Source ──[requestCh]──> Optional Gate ──> DirectDispatcher ──[resultCh]──> Collector
                            │                   │                              │
                       e.g AIMD         spawn goroutine                .Record()
                                        call Generate()                     │
                                        send result                     Tracker

What does this PR do?

This PR introduces AIMDDispatcher and DirectDispatcher and it refactors some cross-cutting concerns into PreDispatcher (publishing metrics, filtering out parse errors)

...──[requestCh]──> PreProcessor ──> Optional Gate ──> DirectDispatcher ──[resultCh]──>...    
                                          │                   │                     
                                     e.g AIMD         spawn goroutine                
                                                         call Generate()             
                                                         send result                 

How was this tested?

• added test coverage, verified existing e2e show the same behavior,

• being verified on a real workload on a real cluster.

• Unit tests added/updated/verified

• Integration/e2e tests added/updated/verified

• Manual testing performed

Checklist

• Commits are signed off (git commit -s) per DCO
• Code follows project contributing guidelines
• CI checks pass (make ci)
• E2E tests pass (make test-e2e)

Related Issues

Related to #581.
Closes #580 once both are merged.

[Copilot]

Pull request overview

Refactors the “sync” job execution flow to use the same channel-based pipeline architecture introduced for async processing, and updates async client plumbing to a shared-client + broadcaster model to support composable dispatch stages.

Changes:

• Introduces a new internal/processor/pipeline package (source → dispatcher chain → collector → tracker) and migrates sync execution to executeJobAsync().
• Reworks async inference client resolution to use a shared async client (asyncSharedClient) with external result broadcasting, and updates tests accordingly.
• Adds plan-file request source (PlanFileSource) and worker-level broadcaster registry for async result propagation; adds small API behavior improvement for cancelling already-cancelled batches.

Reviewed changes

Copilot reviewed 43 out of 45 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
pkg/clients/inference/async_shared_client.go	Adds shared async client that decouples submit from collect and propagates trace metadata.
pkg/clients/inference/async_inference_client_test.go	Updates unit tests to cover shared async client behavior and updated Cancel API.
pkg/clients/inference/async_inference_client_resolver.go	Switches resolver to shared clients; adds model listing and pool uniqueness validation.
pkg/clients/inference/async_inference_client_resolver_test.go	Updates resolver tests to assert shared-client reuse and new API behavior.
pkg/clients/inference/async_inference_client_interface.go	Changes async Cancel API to accept explicit request IDs.
pkg/clients/inference/async_inference_client_integration_test.go	Migrates integration test to shared async client roundtrip.
pkg/clients/inference/async_inference_client_impl.go	Removes per-job async producer client/dispatcher; keeps shared pool struct + deadline constant.
internal/shared/syncutil/mutex_map.go	Adds a type-safe mutex-guarded map utility for concurrency-safe storage.
internal/shared/openai/batch.go	Adds AllSucceeded() helper on BatchRequestCounts.
internal/processor/worker/worker.go	Initializes and runs a per-model async broadcaster registry when async inference is enabled.
internal/processor/worker/test_helpers_test.go	Removes outdated mock async inference client tied to old Cancel signature.
internal/processor/worker/source_planfile.go	Adds a plan-file-based RequestSource implementation producing pipeline.RequestItems.
internal/processor/worker/source_planfile_test.go	Adds comprehensive tests for plan file parsing, headers, and cancellation behavior.
internal/processor/worker/job_runner.go	Routes job execution through executeJobAsync() (pipeline-based).
internal/processor/worker/finalizer_test.go	Removes now-obsolete execution progress unit test.
internal/processor/worker/execute_pipeline.go	Adds pipeline-based job execution and dispatcher-chain composition logic.
internal/processor/worker/broadcasters.go	Adds worker-level registry to manage per-model result broadcasters for async mode.
internal/processor/pipeline/result_router.go	Implements ResultBroadcaster + BroadcasterGroup to fan-out async results to jobs.
internal/processor/pipeline/request_source.go	Defines RequestSource interface for pipeline producers.
internal/processor/pipeline/request_item.go	Introduces RequestItem / ResultItem types + output error struct.
internal/processor/pipeline/progress_tracker.go	Adds throttled progress tracking + status-store updating.
internal/processor/pipeline/pending.go	Adds pending request registry for async result enrichment and cancellation draining.
internal/processor/pipeline/pending_test.go	Adds unit tests for pending registry behavior and concurrency.
internal/processor/pipeline/main_test.go	Initializes metrics for pipeline test package.
internal/processor/pipeline/executor.go	Adds JobExecutor orchestration with errgroup and tracker lifecycle.
internal/processor/pipeline/executor_test.go	Adds end-to-end and cancellation tests for executor behavior.
internal/processor/pipeline/dispatcher.go	Defines RequestDispatcher interface for composable dispatcher chains.
internal/processor/pipeline/dispatcher_pre.go	Adds pre-dispatch stage for parse filtering, cancellation draining, timestamps, and inflight metrics.
internal/processor/pipeline/dispatcher_pre_test.go	Adds tests for PreDispatcher parse filtering, SubmittedAt stamping, and cancellation draining.
internal/processor/pipeline/dispatcher_metricsgate.go	Adds metrics-based gating dispatcher stage.
internal/processor/pipeline/dispatcher_metricsgate_test.go	Adds tests validating gate behavior under budget changes and cancellation.
internal/processor/pipeline/dispatcher_direct.go	Adds direct (HTTP) dispatcher leaf for sync inference.
internal/processor/pipeline/dispatcher_direct_test.go	Adds tests for result building, routing, and error semantics.
internal/processor/pipeline/dispatcher_async.go	Adds async queue submit dispatcher integrating pending registry + broadcaster subscription.
internal/processor/pipeline/dispatcher_aimd.go	Adds AIMD concurrency gating stage wrapping another dispatcher.
internal/processor/pipeline/collector.go	Adds result collector writing JSONL outputs and updating tracker/metrics.
internal/processor/pipeline/collector_test.go	Adds tests for collector routing, cancellation-drain behavior, and tracker throttling.
internal/processor/pipeline/async_test.go	Adds async end-to-end tests for broadcaster + async dispatcher + cancellation.
internal/processor/pipeline/aimd_test.go	Adds AIMD signaling and cancellation/expiry coverage tests.
internal/processor/pipeline/aggregate.go	Adds Pipe() composition helper for declarative dispatcher chains.
internal/apiserver/batch/batch_handler.go	Makes cancel idempotent for already-cancelled batches (returns 200 with batch).
internal/apiserver/batch/batch_handler_test.go	Adds test coverage for cancelling an already-cancelled batch.
.gitignore	Ignores .idea/ directory.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[evacchi]

rebased

[evacchi]

the original version of this PR also included a DSL for piping Dispatchers:

	switch {
	case p.asyncInference != nil:
		broadcasters := p.broadcasters.forModels(modelMap)
		return pipeline.Pipe(
			pipeline.PreDispatcherStage(),
			pipeline.AsyncDispatcherStage(p.asyncInference, broadcasters, pending, logger),
		)
	case p.cfg.Concurrency.AIMD.Enabled:
		models := buildAIMDModels(modelMap, p.inference, p.endpointLimits)
		return pipeline.Pipe(
			pipeline.PreDispatcherStage(),
			pipeline.AIMDDispatcherStage(models, p.cfg.Concurrency.Global, logger),
			pipeline.DirectDispatcherStage(p.inference, logger),
		)
	default:
		return pipeline.Pipe(
			pipeline.PreDispatcherStage(),
			pipeline.DirectDispatcherStage(p.inference, logger),
		)
	}

It does not add much to readability and it can be reintroduced later

[lioraron]

Bug: RecordRequestError is double-counted in the sync flow.

buildResult() calls it here for every non-200 result, and then collector.go:120 calls it again for the same result in Receive(). The result flows through the pipeline chain to the collector, so both fire → error rate metrics are 2x inflated for all sync-mode jobs.

Fix: remove this call from buildResult() — the collector already handles RecordRequestError for both sync and async flows.

[lioraron]

Bug: recordTokenUsage is double-counted in the sync flow — same pattern as RecordRequestError.

handleSuccess() calls it here, and then collector.go:110 calls it again for the same successful result. Both fire for every 200 response → token usage metrics (prompt_tokens, completion_tokens) are 2x inflated for sync-mode jobs.

Fix: remove this call from handleSuccess() — the collector handles it for both flows.

[lioraron]

Fragile implicit contract: buildResult() intentionally does NOT copy SubmittedAt from RequestItem to ResultItem. This is what prevents the collector from double-decrementing inflight metrics (the collector guards Dec calls with !msg.SubmittedAt.IsZero()). But it's undocumented — if someone later adds SubmittedAt: msg.SubmittedAt here, inflight metrics will silently double-decrement.

Worth adding a short comment explaining why SubmittedAt is not copied (DirectDispatcher handles its own Dec/Duration in submitRequest(), so the collector must skip them).

[lioraron]

semaphore.New returns (nil, ErrCap) when capacity <= 0. The error is discarded here, so a misconfigured globalLimit (e.g. concurrency.global: 0) would leave globalSem nil → panic at acquireSlot.

Config validation may ensure this doesn't happen today, but a defensive check would prevent a cryptic nil-pointer panic:

globalSem, err := semaphore.New(globalLimit, nil)
if err != nil {
    // return error or log.Fatal with a clear message about the config field
}

[lioraron]

source_planfile.go:82 — Produce ignores context cancellation

After context cancellation (SLO expiry, user cancel), the Produce loop continues doing ReadAt + json.Unmarshal for every remaining plan entry before sending it into the channel. PreDispatcher will drain everything and convert to cancel errors, so there's no deadlock, but for a large job with many remaining requests this delays shutdown with unnecessary I/O.

Consider adding a ctx check:

select {
case outgoingRequestCh <- *item:
case <-ctx.Done():
    return nil
}

[lioraron]

collector.go:104 — RecordModelRequestExecutionDuration fires for requests that never ran inference

This fires for any result with non-zero SubmittedAt — including cancel errors from AIMDDispatcher.acquireSlot failures. Those requests never ran inference; the recorded duration is just AIMD semaphore queue wait time, which pollutes the execution duration histogram.

Consider guarding:

if msg.Error == nil {
    metrics.RecordModelRequestExecutionDuration(time.Since(msg.SubmittedAt), msg.ModelID)
}

(The Dec calls should remain unconditional since PreDispatcher incremented for all submitted requests.)

[j-mok-dev]

If Receive() has already hit a write error, the collector keeps draining until dispatch finishes and only returns that error at the end. In the unified pipeline that means we can continue issuing/completing backend work after output persistence has already failed. Should this fail fast and cancel the pipeline instead, since at that point we can no longer durably record subsequent results?