Merge remote-tracking branch 'upstream/main' into rhoai-3.5

Author: m-rafeeq

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.2
-	github.com/llm-d-incubation/llm-d-async/producer v0.7.2
+	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/pashagolub/pgxmock/v4 v4.9.0
 	github.com/prometheus/client_golang v1.23.2
 	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.2 h1:sf6iFDa5LpVoKDYiOOlsbnv+6Ykj5TA22yRqMdIbOfY=
-github.com/llm-d-incubation/llm-d-async/api v0.7.2/go.mod h1:m2zJUwD/AZypJv8RPws3uvCnfQoNW7iv+hH9wPk/Xw4=
-github.com/llm-d-incubation/llm-d-async/producer v0.7.2 h1:hhnLqi+MXq8kBjsQhnUCWYtTCG6uFTZLCH296CZMlpY=
-github.com/llm-d-incubation/llm-d-async/producer v0.7.2/go.mod h1:IrClO4XwMtgLRnlkAqO1LDsuqmwtQjELDabT2f+5d40=
+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/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/executor.go

@@ -679,11 +679,23 @@ func (p *Processor) processModelAsync(
 		delete(pending, resp.RequestID)
 	}
 
+	// Best-effort: tell the dispatcher to drop still-pending requests before
+	// dispatch. Use a detached timeout — requestAbortCtx is often already
+	// cancelled when we reach this path.
+	if len(pending) > 0 {
+		cancelCtx, cancelFn := context.WithTimeout(context.Background(), 5*time.Second)
+		if err := asyncClient.Cancel(cancelCtx); err != nil {
+			logger.Error(err, "Failed to cancel pending async requests", "pendingCount", len(pending))
+		}
+		cancelFn()
+	}
+
 	// Drain submitted-but-uncollected requests as errors so that
-	// output_lines + error_lines == total_requests.
+	// output_lines + error_lines == total_requests. Error code follows the
+	// same abort-reason priority as drainAndFinalize.
+	errCode, errMsg := uncollectedPendingError(mainCtx, sloCtx, userCancelCtx, modelErr)
 	for _, pr := range pending {
-		out := newErrorOutputLine(pr.batchReqID, pr.customID,
-			string(batch_types.ErrCodeBatchExpired), "result not collected before deadline")
+		out := newErrorOutputLine(pr.batchReqID, pr.customID, string(errCode), errMsg)
 		lineBytes, err := json.Marshal(out)
 		if err != nil {
 			return fmt.Errorf("marshal uncollected error line: %w", err)
@@ -699,6 +711,30 @@ func (p *Processor) processModelAsync(
 		inputFile, entries[submitCount:], writers, progress, modelErr, logger, len(entries), 0)
 }
 
+// uncollectedPendingError selects the error code/message for submitted-but-
+// uncollected async requests, matching drainAndFinalize's abort-reason order.
+// Context parameter order mirrors drainAndFinalize (minus requestAbortCtx).
+func uncollectedPendingError(
+	mainCtx, sloCtx, userCancelCtx context.Context,
+	modelErr error,
+) (batch_types.BatchErrorCode, string) {
+	switch {
+	case errors.Is(sloCtx.Err(), context.DeadlineExceeded):
+		return batch_types.ErrCodeBatchExpired, batch_types.ErrCodeBatchExpired.Message()
+	case userCancelCtx.Err() != nil:
+		return batch_types.ErrCodeBatchCancelled, batch_types.ErrCodeBatchCancelled.Message()
+	case modelErr != nil:
+		return batch_types.ErrCodeBatchFailed, batch_types.ErrCodeBatchFailed.Message()
+	case mainCtx.Err() != nil:
+		// SIGTERM: leave terminalization to orphan recovery, but still record
+		// lines so completed+failed == total for this model pass.
+		return batch_types.ErrCodeBatchFailed, batch_types.ErrCodeBatchFailed.Message()
+	default:
+		// Sibling abort or collect interrupted without a classified reason.
+		return batch_types.ErrCodeBatchFailed, batch_types.ErrCodeBatchFailed.Message()
+	}
+}
+
 // drainAndFinalize drains undispatched entries based on termination reason and
 // returns the appropriate sentinel error. Shared by processModel and processModelAsync.
 func (p *Processor) drainAndFinalize(