enh: refactor async integration to channel-based pipeline

[evacchi]

Why is this PR needed?

We refactor executeJob() to accommodate an async, channel/gouroutine-based processing pipeline that will work better in the async dispatcher case, and can will fit nicely for sync processing as well.

What does this PR do?

Replaces the executeJob() with executeJobAsync() for async processing, using a channel-based pipeline in a new internal/processor/pipeline package. The old execution path (executeJob() / processModel() ) is preserved; processModelAsync() is dropped, in favor of the new flow.

---

Actors

Actor	Responsibility
RequestSource	Reads plan files, parses JSONL, assembles headers (SLO, fairness, inference objective), produces RequestItem values
RequestDispatcher	Routes requests and produces results. Composable as a delegate chain (useful in the sync case). Implementation is currently AsyncDispatcher (queue-backed)
ResultCollector	Reads results, writes JSONL to output/error files, records progress via tracker
ProgressTracker	Atomic counters + background ticker for throttled status-store pushes
JobExecutor	Orchestrator — creates channels, starts actors via errgroup, waits for cascade shutdown

Dispatcher chain (sync)

The DirectDispatcher sends requests to the HTTP clients as they come in without holdback.

It is designed to be the leaf in the dispatch pipeline: usually a gate of some sort will intercept requests and buffer them to rate limit.

The DirectDispatcher sends requests, receives the responses and sends them out on the result channel for collection.

Dispatcher (async)

AsyncDispatcher → queue (fire-and-forget)
                        │
ResultBroadcaster ← SharedClient.GetResult() → resultCh → Collector

The AsyncDispatcher

1. submits to the queue and returns, and flags the ID in a per-job PendingRequests map
2. subscribes all messages for the corresponding result queue from a ResultBroadcaster

The ResultBroadcaster broadcasts responses from a given queue to all its subscribers (it DOES NOT filter per model ID).

The collector reads from the result channel and matches against a per-job PendingRequests map.

e.g., AsyncDispatcher sends out request ID X for Model/Pool A; then it subscribes for all results for model A; the collector filters out all results for model A that do not match request ID X.

Pending requests

A MutexMap-based PendingRequests tracks in-flight requests. The dispatcher stores entries before dispatching; the collector resolves them when results arrive. Results with CustomID already set (cancels, inline errors) bypass the map. Results without CustomID (async inference) are enriched from the map. Unknown results (broadcast from another job) are silently dropped.

Cascade shutdown

Each dispatcher closes resultCh when done → collector drains and returns → tracker is cancelled and pushes final counts. For cancellation, dispatchers drain remaining requestCh items as cancelled results before closing resultCh.

Cancellation

The executor passes a single context (requestAbortCtx). The pipeline doesn't know why it was stopped. After Execute returns, the caller (executeJobV2) inspects sloCtx, userCancelCtx, and mainCtx to choose the terminal status (errExpired, errCancelled, errShutdown).

How was this tested?

• 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

First part of #580, supersedes #529, #528 and related.

[Copilot]

Pull request overview

Refactors the batch job execution path to support an async, channel/goroutine-based pipeline, introducing a new internal/processor/pipeline package and switching the async execution flow to use shared async clients plus per-model result broadcasters.

Changes:

• Introduces a new pipeline abstraction (source → dispatcher → collector → tracker) and wires it into executeJobAsync().
• Replaces the prior per-job async client + internal routing with a shared async client + external ResultBroadcaster.
• Adds planfile-based RequestSource and a broadcasterRegistry to fan out async results to job-specific collectors.

Reviewed changes

Copilot reviewed 31 out of 32 changed files in this pull request and generated 8 comments.

Show a summary per file

File	Description
pkg/clients/inference/async_shared_client.go	Adds a shared async inference client that decouples submit from result collection.
pkg/clients/inference/async_inference_client_test.go	Updates unit tests to validate trace context propagation using the new shared client.
pkg/clients/inference/async_inference_client_resolver.go	Refactors async resolver to return shared clients and exposes configured model IDs.
pkg/clients/inference/async_inference_client_resolver_test.go	Updates resolver tests for shared-client reuse semantics.
pkg/clients/inference/async_inference_client_interface.go	Updates interface docs to remove per-job client assumptions.
pkg/clients/inference/async_inference_client_integration_test.go	Updates integration test to round-trip with asyncSharedClient.
pkg/clients/inference/async_inference_client_impl.go	Removes the old per-job async producer client + dispatcher routing logic.
internal/shared/syncutil/mutex_map.go	Adds a generic mutex-guarded map used by broadcasters and shared client registry.
internal/processor/worker/worker.go	Adds initialization of per-model result broadcaster registry when async inference is enabled.
internal/processor/worker/test_helpers_test.go	Removes mock async inference client helpers (old async path removed).
internal/processor/worker/source_planfile.go	Adds a planfile-based RequestSource that builds RequestItems for the pipeline.
internal/processor/worker/source_planfile_test.go	Adds tests for planfile parsing/production and header merging behavior.
internal/processor/worker/job_runner.go	Switches job execution to executeJobAsync() when async inference is configured.
internal/processor/worker/executor.go	Removes processModelAsync() and routes execution through sync processModel() in legacy path.
internal/processor/worker/executor_test.go	Removes tests specifically covering the removed processModelAsync() implementation.
internal/processor/worker/execute_pipeline.go	Implements executeJobAsync() and wires pipeline actors together for async execution.
internal/processor/worker/broadcasters.go	Adds broadcasterRegistry to manage per-model ResultBroadcaster lifecycles.
internal/processor/pipeline/result_router.go	Introduces async result broadcasting and subscription management.
internal/processor/pipeline/request_source.go	Defines RequestSource interface for producing requests into the pipeline.
internal/processor/pipeline/request_item.go	Defines pipeline request/result message types and helper constructors for errors/cancellation.
internal/processor/pipeline/progress_tracker.go	Adds a progress tracker that periodically pushes batch counts to a status store.
internal/processor/pipeline/pending.go	Adds pending-request registry for matching async results to job requests.
internal/processor/pipeline/pending_test.go	Adds tests validating pending request enrichment and wait semantics.
internal/processor/pipeline/main_test.go	Initializes metrics for pipeline tests and adds shared test helpers.
internal/processor/pipeline/executor.go	Adds JobExecutor orchestration for source/dispatcher/collector/tracker.
internal/processor/pipeline/executor_test.go	Adds end-to-end and error-handling tests for the pipeline executor.
internal/processor/pipeline/dispatcher.go	Defines the dispatcher interface used by pipeline execution.
internal/processor/pipeline/dispatcher_async.go	Adds async dispatcher that submits to queues and relies on broadcasters for results.
internal/processor/pipeline/collector.go	Adds collector that writes JSONL output/error files and updates progress.
internal/processor/pipeline/collector_test.go	Adds tests for collector routing, cancellation behavior, and tracker bookkeeping.
internal/processor/pipeline/async_test.go	Adds end-to-end tests for async dispatch + broadcast + pending-resolution behavior.
.gitignore	Ignores .idea/ directory.

---

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

[j-mok-dev]

This remembers the first collector error but keeps draining while the rest of the pipeline can continue submitting requests. If local result persistence is already broken, should this abort the pipeline rather than allowing more async work whose results may never be recorded?

[evacchi]

should be addressed now

[j-mok-dev]

The collector now skips further writes after the first persistence failure, but the rest of the pipeline can still keep submitting async work until the source/dispatcher finish. That still means we may enqueue remote work even after local result persistence is already broken.

If we want persistence failure to be terminal for the job, I think we still need to abort the pipeline promptly here rather than only returning the error after drain completes.

[evacchi]

true, but the input file has a limited size and eventually we will reach the end, we can bubble up the signal to the top, though; it depends on how critical it is to short-circuit -- I'd argue, not that critical, we shouldn't actually process when we reach the error case, we just drain+skip so it's relatively quick.

[evacchi]

rebased + added support to cancellation, verified e2e tests still pass
following up with PR for "direct" dispatching (i.e. without llm-d-async)

[j-mok-dev]

Produce() does a plain blocking send here and does not observe ctx.Done() while handing items to the dispatcher. So if the downstream side stops reading or is cancelled, this source can still block indefinitely on the channel send instead of exiting promptly with the request context.
I think this send should be wrapped in a select on outgoingRequestCh <- *item and ctx.Done() so the source respects cancellation during shutdown.

[evacchi]

right, but as commented above #581 (comment), the downstream side (the collector) always drains the channel, so this should be fine.

[lioraron]

recover() catches any panic, not just "send on closed channel". An unrelated nil-pointer dereference or index-out-of-range in the broadcast path would be silently swallowed with Info-level logging, making future debugging very difficult.

Consider checking the panic type:

defer func() {
	if r := recover(); r != nil {
		if msg, ok := r.(string); ok && msg == "send on closed channel" {
			b.logger.Info("Broadcast send recovered (subscriber unsubscribed)",
				"requestID", result.RequestID)
		} else {
			panic(r) // re-panic for unexpected failures
		}
	}
}()

[evacchi]

I'll do it for extra safety (recover() is on the slow path anyway), but the only line in this function is:

	ch <- result

so that's all we're protecting here

[lioraron]

Thanks!