Sync upstream llm-d/llm-d-batch-gateway 3002a02

Author: zdtsw

README.md

@@ -74,7 +74,7 @@ The system is designed to facilitate efficient processing of batch workloads in
      - Priority queue, event channels, and status updates: `Redis` or `Valkey`.
      - File storage: `S3`, `filesystem`.
 
-4. **Batch Dispatcher** *(optional, future component)* — implemented in [llm-d-incubation/llm-d-async](https://github.com/llm-d-incubation/llm-d-async)
+4. **Batch Dispatcher** *(optional)* — implemented in [llm-d/llm-d-async](https://github.com/llm-d/llm-d-async)
    - An optional component that provides metrics-driven flow control for batch request dispatching.
    - Monitors downstream inference system metrics (e.g. queue depth, latency, utilization) and dynamically adjusts dispatch flow to minimize interference with interactive requests while meeting batch job SLOs.
    - Not required for batch processing — the batch processor includes built-in AIMD-based adaptive concurrency control.
@@ -462,7 +462,7 @@ limitations under the License.
 ## Related Projects
 
 - [llm-d-router](https://github.com/llm-d/llm-d-router) - llm-d inference router.
-- [llm-d-async](https://github.com/llm-d-incubation/llm-d-async) - Metrics-driven batch dispatch flow control (optional).
+- [llm-d-async](https://github.com/llm-d/llm-d-async) - Metrics-driven batch dispatch flow control (optional).
 
 ## Support
 

docs/design/batch-dispatcher-queue-design.md

@@ -6,20 +6,20 @@
 
 Related:
 - [Batch Dispatcher](batch-dispatcher.md)
-- [Dispatch Budget](https://github.com/llm-d-incubation/llm-d-async/blob/main/docs/dispatch-budget.md) (llm-d-async)
+- [Dispatch Budget](https://github.com/llm-d/llm-d-async/blob/main/docs/dispatch-budget.md) (llm-d-async)
 - [Batch Processor Design](batch_processor_architecture.md)
 - [Batch Inference Architecture](batch_inference_architecture.md)
 
 ---
 
 ## Summary
 
-This document describes the design of the request and result queues that connect the **batch-processor** to the **batch dispatcher** ([llm-d-async](https://github.com/llm-d-incubation/llm-d-async)). At this time we always assume that each target inference pool corresponds to a single connector. In other words, we assume that there will never be 2 queues targeting the same inference pool at once; we reserve this for future extensions, if needed.
+This document describes the design of the request and result queues that connect the **batch-processor** to the **batch dispatcher** ([llm-d-async](https://github.com/llm-d/llm-d-async)). At this time we always assume that each target inference pool corresponds to a single connector. In other words, we assume that there will never be 2 queues targeting the same inference pool at once; we reserve this for future extensions, if needed.
 
 The batch-processor supports two mutually exclusive dispatch modes, selected via `dispatch_mode`:
 
 - **`sync`** (default): The executor dispatches inference requests directly to the inference gateway via HTTP, using the existing AIMD + semaphore flow control.
-- **`async`**: The executor enqueues individual requests into **the dispatcher's request queue**; **the dispatcher pulls and forwards** them to the inference gateway based on the [dispatch budget](https://github.com/llm-d-incubation/llm-d-async/blob/main/docs/dispatch-budget.md). A **result consumer** in the batch-processor reads completed responses from **the dispatcher's result queue** and routes them back to the appropriate job's output writer.
+- **`async`**: The executor enqueues individual requests into **the dispatcher's request queue**; **the dispatcher pulls and forwards** them to the inference gateway based on the [dispatch budget](https://github.com/llm-d/llm-d-async/blob/main/docs/dispatch-budget.md). A **result consumer** in the batch-processor reads completed responses from **the dispatcher's result queue** and routes them back to the appropriate job's output writer.
 
 This document describes the **async** dispatch mode and its queue design.
 
@@ -83,7 +83,7 @@ The Redis URL is read from a mounted secret at runtime (not stored in the config
 
 ### Dispatcher Configuration
 
-The dispatcher (llm-d-async) already supports the Redis sorted-set flow with dispatch budget gating. The request queue is configured via the [JSON queues config file](https://github.com/llm-d-incubation/llm-d-async/blob/main/README.md#redis-sorted-set-persisted) (`--redis.ss.queues-config-file`); the result queue is configured via `--redis.ss.result-queue-name`:
+The dispatcher (llm-d-async) already supports the Redis sorted-set flow with dispatch budget gating. The request queue is configured via the [JSON queues config file](https://github.com/llm-d/llm-d-async/blob/main/README.md#redis-sorted-set-persisted) (`--redis.ss.queues-config-file`); the result queue is configured via `--redis.ss.result-queue-name`:
 
 ```json
 [
@@ -107,7 +107,7 @@ The dispatcher (llm-d-async) already supports the Redis sorted-set flow with dis
 
 The queue names must match those derived by the batch-processor's `RequestQueueName()` and `ResultQueueName()` functions.
 
-The dispatcher pulls up to `max_SYS × budget` requests per poll cycle and forwards them to the inference gateway. See the [llm-d-async README](https://github.com/llm-d-incubation/llm-d-async/blob/main/README.md) and [Helm chart values](https://github.com/llm-d-incubation/llm-d-async/tree/main/charts/async-processor) for the full configuration.
+The dispatcher pulls up to `max_SYS × budget` requests per poll cycle and forwards them to the inference gateway. See the [llm-d-async README](https://github.com/llm-d/llm-d-async/blob/main/README.md) and [Helm chart values](https://github.com/llm-d/llm-d-async/tree/main/charts/async-processor) for the full configuration.
 
 ### Future Extension: Queue Registry
 
@@ -130,7 +130,7 @@ Note: The request queue is currently implemented by a **Redis SortedSet** that h
 
 Unlike direct dispatch to the inference gateway — where the EPP's flow control limits how many requests can be in-flight and excess requests are rejected with HTTP 429 — **the request queue is a passive buffer with no backpressure on writes**. The producer can enqueue requests as fast as it can read plan entries, without throttling or semaphore gating. This is safe because:
 
-1. **Flow control is deferred to the dispatcher.** The [dispatch budget](https://github.com/llm-d-incubation/llm-d-async/blob/main/docs/dispatch-budget.md) gates how many requests leave the queue per poll cycle. The gate returns a `budget` value in [0, 1] representing remaining system capacity (generally `budget = D − B`, where `D` is the dispatch budget and `B` is the reserved baseline). The dispatcher pops up to `max_SYS × budget` requests per cycle, where `max_SYS` is a configurable measure of total system capacity. Enqueuing more requests than the dispatcher can immediately process simply means they wait in the queue until capacity opens up — they do not reach the inference gateway or compete with online traffic.
+1. **Flow control is deferred to the dispatcher.** The [dispatch budget](https://github.com/llm-d/llm-d-async/blob/main/docs/dispatch-budget.md) gates how many requests leave the queue per poll cycle. The gate returns a `budget` value in [0, 1] representing remaining system capacity (generally `budget = D − B`, where `D` is the dispatch budget and `B` is the reserved baseline). The dispatcher pops up to `max_SYS × budget` requests per cycle, where `max_SYS` is a configurable measure of total system capacity. Enqueuing more requests than the dispatcher can immediately process simply means they wait in the queue until capacity opens up — they do not reach the inference gateway or compete with online traffic.
 
 2. **The queue is cheap storage.** Redis sorted sets are memory-efficient for this workload. Each request message is a few KB; even a full 50,000-request batch job at ~2 KB per message is ~100 MB — well within Redis capacity and far cheaper than holding in-flight HTTP connections.
 
@@ -140,7 +140,7 @@ Unlike direct dispatch to the inference gateway — where the EPP's flow control
 
 ### Message Format
 
-Request messages follow the wire format defined in the [llm-d-async README — Request Messages and Consumption](https://github.com/llm-d-incubation/llm-d-async/blob/main/README.md#request-messages-and-consumption). The `metadata` field carries batch-processor correlation data (`job_id`, `request_index`) that the dispatcher passes through opaquely and returns in the result. The `headers` field can carry HTTP headers that the dispatcher forwards to the inference gateway (e.g., fairness/SLO headers that the current executor attaches directly).
+Request messages follow the wire format defined in the [llm-d-async README — Request Messages and Consumption](https://github.com/llm-d/llm-d-async/blob/main/README.md#request-messages-and-consumption). The `metadata` field carries batch-processor correlation data (`job_id`, `request_index`) that the dispatcher passes through opaquely and returns in the result. The `headers` field can carry HTTP headers that the dispatcher forwards to the inference gateway (e.g., fairness/SLO headers that the current executor attaches directly).
 
 The sorted-set score is the request's SLO deadline (Unix timestamp), so earliest-deadline requests are dispatched first across all jobs sharing the same pool — providing cross-job deadline-aware scheduling.
 
@@ -175,7 +175,7 @@ As described above, the producer does not need to throttle enqueue operations. I
 
 ### Dispatcher (reads from request queue)
 
-The dispatcher ([llm-d-async](https://github.com/llm-d-incubation/llm-d-async)) periodically polls the sorted set. On each cycle, it computes the budget on its "Dispatch Gate" to determine the current dispatch budget $D$. If $D > B$ (the reserved baseline), it pops up to `max_SYS × budget` requests (lowest score = earliest deadline first) and forwards them to the inference gateway via HTTP. See [Dispatch Budget](https://github.com/llm-d-incubation/llm-d-async/blob/main/docs/dispatch-budget.md) for the full gating logic.
+The dispatcher ([llm-d-async](https://github.com/llm-d/llm-d-async)) periodically polls the sorted set. On each cycle, it computes the budget on its "Dispatch Gate" to determine the current dispatch budget $D$. If $D > B$ (the reserved baseline), it pops up to `max_SYS × budget` requests (lowest score = earliest deadline first) and forwards them to the inference gateway via HTTP. See [Dispatch Budget](https://github.com/llm-d/llm-d-async/blob/main/docs/dispatch-budget.md) for the full gating logic.
 
 The dispatcher requires no changes to support this integration, it already implements the Redis sorted-set flow with dispatch budget gating. Only the queue names need to match the naming convention.
 
@@ -189,7 +189,7 @@ Note: the result queue is currently implemented by a Redis list.
 
 ### Message Format
 
-Result messages follow the format defined in the [llm-d-async README — Results](https://github.com/llm-d-incubation/llm-d-async/blob/main/README.md#results). The `metadata` from the original request (containing `job_id` and `request_index`) is passed through by the dispatcher, allowing the consumer to route results back to the correct job.
+Result messages follow the format defined in the [llm-d-async README — Results](https://github.com/llm-d/llm-d-async/blob/main/README.md#results). The `metadata` from the original request (containing `job_id` and `request_index`) is passed through by the dispatcher, allowing the consumer to route results back to the correct job.
 
 **Example:**
 

docs/design/batch-dispatcher.md

@@ -5,12 +5,12 @@ Related:
 - [\[PUBLIC\] EPP Flow Controller for Priority, Fairness, and Queuing](https://docs.google.com/document/d/1VZL7opFWuwgWquvgiOzLlXAJ633qZ9U-A0ZixGjBgaI/edit?tab=t.0#heading=h.hfyow92z2d0t)
 - [\[PUBLIC\] Improved Flow Control Request Management](https://docs.google.com/document/d/1JxzJc8gNv2wKK5-a8ohb0btn78ymVKw9XMIb4-S-ncA/edit?tab=t.0#heading=h.rutawybt03nl)
 - [https://gateway-api-inference-extension.sigs.k8s.io/api-types/inferencepool/](https://gateway-api-inference-extension.sigs.k8s.io/api-types/inferencepool/)
-- [Async Processor: Dispatch Gates](https://github.com/llm-d-incubation/llm-d-async/blob/main/README.md#dispatch-gates) (llm-d-async)
-- [Dispatch Budget](https://github.com/llm-d-incubation/llm-d-async/blob/main/docs/dispatch-budget.md) (llm-d-async)
+- [Async Processor: Dispatch Gates](https://github.com/llm-d/llm-d-async/blob/main/README.md#dispatch-gates) (llm-d-async)
+- [Dispatch Budget](https://github.com/llm-d/llm-d-async/blob/main/docs/dispatch-budget.md) (llm-d-async)
 
 ## Summary
 
-This document details the design of a **Batch Dispatcher** (sometimes also referred to as the [**Async Processor**](https://github.com/llm-d-incubation/llm-d-async)) to extend the existing "online batch processing agent" architecture (see [\[Public Doc\] Serving Online Batch via Inference Gateway](https://docs.google.com/document/d/1notkq9s0qOmWmUNonZ8CfI-5jtGtHA4PGMI-xz8sGRE/edit?tab=t.0#heading=h.i76kzr3j3swj)). While the **llm-d Router** acts as the primary component for scheduling and flow control, the Batch Dispatcher serves as a system-load aware gatekeeper. It ensures that batch workloads (low-priority and sheddable) are pulled from message queues and forwarded to the llm-d Router only when the inference pool has sufficient capacity. This prevents low-priority traffic from flooding the system and competing with realtime requests.
+This document details the design of a **Batch Dispatcher** (sometimes also referred to as the [**Async Processor**](https://github.com/llm-d/llm-d-async)) to extend the existing "online batch processing agent" architecture (see [\[Public Doc\] Serving Online Batch via Inference Gateway](https://docs.google.com/document/d/1notkq9s0qOmWmUNonZ8CfI-5jtGtHA4PGMI-xz8sGRE/edit?tab=t.0#heading=h.i76kzr3j3swj)). While the **llm-d Router** acts as the primary component for scheduling and flow control, the Batch Dispatcher serves as a system-load aware gatekeeper. It ensures that batch workloads (low-priority and sheddable) are pulled from message queues and forwarded to the llm-d Router only when the inference pool has sufficient capacity. This prevents low-priority traffic from flooding the system and competing with realtime requests.
 
 ## Problem statement
 

go.mod

@@ -14,8 +14,8 @@ require (
 	github.com/go-resty/resty/v2 v2.17.2
 	github.com/google/uuid v1.6.0
 	github.com/jackc/pgx/v5 v5.10.0
-	github.com/llm-d-incubation/llm-d-async/api v0.7.3
-	github.com/llm-d-incubation/llm-d-async/producer v0.7.3
+	github.com/llm-d/llm-d-async/api v0.7.4
+	github.com/llm-d/llm-d-async/producer v0.7.4
 	github.com/pashagolub/pgxmock/v4 v4.9.0
 	github.com/prometheus/client_golang v1.24.0
 	github.com/quasilyte/go-ruleguard/dsl v0.3.23

go.sum

@@ -88,10 +88,10 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
-github.com/llm-d-incubation/llm-d-async/api v0.7.3 h1:OI4zlNdKeXrWa9Q9AhFOSvJZvLC84Nnsv3maF4d8Tns=
-github.com/llm-d-incubation/llm-d-async/api v0.7.3/go.mod h1:m2zJUwD/AZypJv8RPws3uvCnfQoNW7iv+hH9wPk/Xw4=
-github.com/llm-d-incubation/llm-d-async/producer v0.7.3 h1:f06rUZ92g/2gQmcrquYYd5LNGeCYffb1vZdP+7luiA0=
-github.com/llm-d-incubation/llm-d-async/producer v0.7.3/go.mod h1:7FVFV0ggHyaSbD33DP1V68tV7bOVKqjAezd8UgD9Qik=
+github.com/llm-d/llm-d-async/api v0.7.4 h1:R3CBDMxlT+9GOyJkgNdQO7zqSpwVFgfz085vSlXn91Q=
+github.com/llm-d/llm-d-async/api v0.7.4/go.mod h1:hzjFDTFBJEyW9/1vrHAD2dtCVLTzqggU1GBW+7YhxlQ=
+github.com/llm-d/llm-d-async/producer v0.7.4 h1:dmMNtkorVykYX5UXRYY7qC7q8IesGtbHBY+RFwoaE60=
+github.com/llm-d/llm-d-async/producer v0.7.4/go.mod h1:nHxbBM/mVL7bTu2ZVZ+OMz/Up9m2PhpBmeLfoq899+0=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/pashagolub/pgxmock/v4 v4.9.0 h1:itlO8nrVRnzkdMBXLs8pWUyyB2PC3Gku0WGIj/gGl7I=

internal/processor/worker/execute_pipeline.go

@@ -28,12 +28,6 @@ func (p *Processor) executeJobAsync(ctx, sloCtx, userCancelCtx, requestAbortCtx
 		return nil, fmt.Errorf("read model map: %w", err)
 	}
 
-	if sloCtx.Err() == context.DeadlineExceeded {
-		logger.V(logging.INFO).Info("SLO already expired at execution start",
-			"total", modelMap.LineCount)
-		return &openai.BatchRequestCounts{Total: modelMap.LineCount, Failed: modelMap.RejectedCount}, errExpired
-	}
-
 	files, err := p.openDataFiles(params)
 	if err != nil {
 		return nil, err

internal/processor/worker/executor_test.go

@@ -379,8 +379,9 @@ func TestExecuteJob_AbortCtxCancel_AbortsInflightRequests(t *testing.T) {
 }
 
 // TestExecuteJob_SLOExpiredBeforeDispatch verifies that when the SLO deadline has already
-// passed before execution begins, executeJob returns errExpired immediately with the total
-// request count and no output/error files are written (early-exit fast path).
+// passed before execution begins, the pipeline still runs and drains all requests as
+// batch_expired errors. This ensures that handleExpired can upload a non-empty error file
+// and the DB reflects Failed == Total.
 func TestExecuteJob_SLOExpiredBeforeDispatch(t *testing.T) {
 	cfg := config.NewConfig()
 	cfg.WorkDir = t.TempDir()
@@ -393,7 +394,6 @@ func TestExecuteJob_SLOExpiredBeforeDispatch(t *testing.T) {
 	env, jobInfo := setupExecutionJob(t, cfg, &mockInferenceClient{}, requests, map[string]string{"m1": "m1"})
 
 	ctx := testLoggerCtx(t)
-	// SLO deadline already in the past: early check fires before any files are opened.
 	sloCtx, cancel := context.WithDeadline(ctx, time.Now().Add(-1*time.Second))
 	defer cancel()
 
@@ -408,25 +408,33 @@ func TestExecuteJob_SLOExpiredBeforeDispatch(t *testing.T) {
 		t.Fatal("expected non-nil counts")
 		return
 	}
-	// Early exit: total is known from the model map, but no requests were dispatched or drained.
 	if counts.Total != 3 {
 		t.Fatalf("Total = %d, want 3", counts.Total)
 	}
 	if counts.Completed != 0 {
 		t.Fatalf("Completed = %d, want 0", counts.Completed)
 	}
-	if counts.Failed != 0 {
-		t.Fatalf("Failed = %d, want 0 (no drain on early exit)", counts.Failed)
+	if counts.Failed != 3 {
+		t.Fatalf("Failed = %d, want 3 (all requests drained as batch_expired)", counts.Failed)
 	}
 
-	// No output or error files are written on early exit: files are only opened after the SLO check.
-	outputPath, _ := env.p.jobOutputFilePath(jobInfo.JobID, jobInfo.TenantID)
 	errorPath, _ := env.p.jobErrorFilePath(jobInfo.JobID, jobInfo.TenantID)
-	if _, statErr := os.Stat(outputPath); !errors.Is(statErr, os.ErrNotExist) {
-		t.Fatalf("output.jsonl should not exist on early SLO exit, got stat err: %v", statErr)
+	errorData, readErr := os.ReadFile(errorPath)
+	if readErr != nil {
+		t.Fatalf("error.jsonl should exist: %v", readErr)
+	}
+	errorLines := bytes.Count(bytes.TrimSpace(errorData), []byte("\n")) + 1
+	if errorLines != 3 {
+		t.Fatalf("error.jsonl lines = %d, want 3", errorLines)
+	}
+
+	outputPath, _ := env.p.jobOutputFilePath(jobInfo.JobID, jobInfo.TenantID)
+	outputData, readErr := os.ReadFile(outputPath)
+	if readErr != nil {
+		t.Fatalf("output.jsonl should exist: %v", readErr)
 	}
-	if _, statErr := os.Stat(errorPath); !errors.Is(statErr, os.ErrNotExist) {
-		t.Fatalf("error.jsonl should not exist on early SLO exit, got stat err: %v", statErr)
+	if len(outputData) != 0 {
+		t.Fatalf("output.jsonl should be empty, got %d bytes", len(outputData))
 	}
 }