Preserve HTTP status and surface non-HTTP failures in async results (llm-d#300)

* Preserve HTTP status and surface non-HTTP failures in async results

Signed-off-by: Jooyeon Mok <jmok@redhat.com>

* address review: InferenceResponse struct, pool gate re-enqueue on shutdown, test fixes

Signed-off-by: Jooyeon Mok <jmok@redhat.com>

* docs: add release-notes fragment for ResultMessage wire-format changes

Record the breaking wire-format changes (StatusCode/ErrorCode/ErrorMessage,
gate-drop payload, SendRequest signature) as a release-notes.d fragment,
following the llm-d-router convention so the change has a durable home.

Signed-off-by: Shimi Bandiel <shimib@google.com>

---------

Signed-off-by: Jooyeon Mok <jmok@redhat.com>
Signed-off-by: Shimi Bandiel <shimib@google.com>
Co-authored-by: Shimi Bandiel <shimib@google.com>

Author: j-mok-dev

api/api.go

@@ -1,5 +1,7 @@
 package api
 
+import "encoding/json"
+
 // Request is the public interface for submitting requests to the async queue.
 // It exposes only the caller-visible fields. Concrete types like RequestMessage,
 // RedisRequest, and PubSubRequest satisfy this interface.
@@ -58,11 +60,71 @@ var (
 	_ Request = (*PubSubRequest)(nil)
 )
 
-// ResultMessage is the async inference result returned to callers. ID and Payload are
-// JSON fields; Routing and Metadata are infrastructure pass-through (json:"-").
+// ResultMessage is the async inference result returned to callers.
+//
+// Wire-format semantics:
+//   - StatusCode > 0: an HTTP response was received. Payload contains the response body.
+//   - StatusCode == 0: no HTTP response. ErrorCode/ErrorMessage describe the failure.
+//
+// Routing and Metadata are infrastructure pass-through (json:"-").
 type ResultMessage struct {
-	ID       string            `json:"id"`
-	Payload  string            `json:"payload"`
-	Routing  InternalRouting   `json:"-"`
-	Metadata map[string]string `json:"-"`
+	ID           string            `json:"id"`
+	StatusCode   int               `json:"status_code,omitempty"`
+	Payload      string            `json:"payload"`
+	ErrorCode    string            `json:"error_code,omitempty"`
+	ErrorMessage string            `json:"error_message,omitempty"`
+	Routing      InternalRouting   `json:"-"`
+	Metadata     map[string]string `json:"-"`
+}
+
+// Error codes for non-HTTP failures surfaced in ResultMessage.ErrorCode.
+// These are result-level codes describing why a request could not be completed.
+const (
+	ErrCodeDeadlineExceeded = "DEADLINE_EXCEEDED"
+	ErrCodeGateDropped      = "GATE_DROPPED"
+	ErrCodeGateError        = "GATE_ERROR"
+	ErrCodeInferenceError   = "INFERENCE_ERROR"
+	ErrCodeInvalidRequest   = "INVALID_REQUEST"
+)
+
+// NewErrorResult builds a non-HTTP error ResultMessage.
+// errorCode must be one of the ErrCode* constants; errMsg is a human-readable description.
+// Payload is populated with a JSON error object for backward compatibility with
+// consumers that only read Payload.
+func NewErrorResult(req Request, routing InternalRouting, errorCode, errMsg string) ResultMessage {
+	errorPayload := map[string]string{"error": errMsg}
+	payloadBytes, err := json.Marshal(errorPayload)
+	if err != nil {
+		payloadBytes = []byte(`{"error": "internal error"}`)
+	}
+	return ResultMessage{
+		ID:           req.ReqID(),
+		Payload:      string(payloadBytes),
+		ErrorCode:    errorCode,
+		ErrorMessage: errMsg,
+		Routing:      routing,
+		Metadata:     req.ReqMetadata(),
+	}
+}
+
+// NewHTTPResult builds a ResultMessage for any HTTP response (success or error).
+// statusCode is the actual HTTP status code; responseBody is the raw body.
+func NewHTTPResult(req Request, routing InternalRouting, statusCode int, responseBody []byte) ResultMessage {
+	return ResultMessage{
+		ID:         req.ReqID(),
+		StatusCode: statusCode,
+		Payload:    string(responseBody),
+		Routing:    routing,
+		Metadata:   req.ReqMetadata(),
+	}
+}
+
+// NewGateDroppedResult builds a ResultMessage for a gate-dropped request.
+func NewGateDroppedResult(req Request, routing InternalRouting) ResultMessage {
+	return NewErrorResult(req, routing, ErrCodeGateDropped, "Pool gating dropped request")
+}
+
+// NewDeadlineExceededResult builds a ResultMessage for deadline expiry.
+func NewDeadlineExceededResult(req Request, routing InternalRouting) ResultMessage {
+	return NewErrorResult(req, routing, ErrCodeDeadlineExceeded, "deadline exceeded")
 }

api/inference_client.go

@@ -2,11 +2,25 @@ package api
 
 import "context"
 
+// InferenceResponse holds the HTTP response from an upstream inference request.
+// StatusCode is non-zero whenever an HTTP response was received (including 4xx/5xx).
+// A zero StatusCode means no HTTP response was obtained (e.g. transport/network failure).
+// Body contains the response body; it may be partial if a read error occurred.
+type InferenceResponse struct {
+	StatusCode int
+	Body       []byte
+}
+
 // InferenceClient defines the interface for sending inference requests.
 // This interface allows for pluggable implementations beyond the default HTTP client.
 type InferenceClient interface {
 	// SendRequest sends an inference request to the specified URL with the given headers and payload.
-	// Returns the response body and any error that occurred.
+	//
+	// On success (nil error), the returned *InferenceResponse is always non-nil.
+	// On error, *InferenceResponse may still be non-nil if the upstream sent an HTTP
+	// response (e.g. 4xx/5xx); callers should check resp.StatusCode > 0 to determine
+	// whether an HTTP response was received.
+	//
 	// Errors should implement InferenceError to provide an ErrorCategory via Category().
 	// ErrorCategory determines retry and shedding behavior through its Fatal() and Sheddable() methods:
 	//   - ErrCategoryRateLimit:  retryable, sheddable (e.g. 429)
@@ -15,6 +29,5 @@ type InferenceClient interface {
 	//   - ErrCategoryAuth:       not retryable
 	//   - ErrCategoryParse:      not retryable
 	//   - ErrCategoryUnknown:    not retryable
-	// A nil error indicates a successful response.
-	SendRequest(ctx context.Context, url string, headers map[string]string, payload []byte) (responseBody []byte, err error)
+	SendRequest(ctx context.Context, url string, headers map[string]string, payload []byte) (*InferenceResponse, error)
 }

api/inference_error.go

@@ -42,6 +42,7 @@ type ClientError struct {
 	Message       string
 	RawError      error         // original error if available
 	RetryAfter    time.Duration // server-specified retry delay from Retry-After header (0 means not set)
+	StatusCode    int           // HTTP status code; 0 means no HTTP response was received
 }
 
 func (e *ClientError) Error() string {

pkg/asyncworker/http_client.go

@@ -25,7 +25,7 @@ func NewHTTPInferenceClient(client *http.Client) *HTTPInferenceClient {
 }
 
 // SendRequest implements InferenceClient for HTTP-based inference requests.
-func (h *HTTPInferenceClient) SendRequest(ctx context.Context, url string, headers map[string]string, payload []byte) ([]byte, error) {
+func (h *HTTPInferenceClient) SendRequest(ctx context.Context, url string, headers map[string]string, payload []byte) (*asyncapi.InferenceResponse, error) {
 	request, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(payload))
 	if err != nil {
 		return nil, &asyncapi.ClientError{
@@ -51,44 +51,43 @@ func (h *HTTPInferenceClient) SendRequest(ctx context.Context, url string, heade
 
 	body, err := io.ReadAll(result.Body)
 	if err != nil {
-		// Response read errors are retryable as the request may have succeeded
-		return nil, &asyncapi.ClientError{
+		return &asyncapi.InferenceResponse{StatusCode: result.StatusCode, Body: body}, &asyncapi.ClientError{
 			ErrorCategory: asyncapi.ErrCategoryServer,
 			Message:       "failed to read response",
 			RawError:      err,
+			StatusCode:    result.StatusCode,
 		}
 	}
 
-	// Check for rate limiting / load shedding (429)
+	resp := &asyncapi.InferenceResponse{StatusCode: result.StatusCode, Body: body}
+
 	if result.StatusCode == 429 {
 		retryAfter, _ := parseRetryAfter(result.Header.Get("Retry-After"))
-		return body, &asyncapi.ClientError{
+		return resp, &asyncapi.ClientError{
 			ErrorCategory: asyncapi.ErrCategoryRateLimit,
 			Message:       fmt.Sprintf("rate limited: status code %d", result.StatusCode),
-			RawError:      nil,
 			RetryAfter:    retryAfter,
+			StatusCode:    result.StatusCode,
 		}
 	}
 
-	// Check for client errors (4xx, non-429)
 	if result.StatusCode >= 400 && result.StatusCode < 500 {
-		return body, &asyncapi.ClientError{
+		return resp, &asyncapi.ClientError{
 			ErrorCategory: asyncapi.ErrCategoryInvalidReq,
 			Message:       fmt.Sprintf("client error: status code %d", result.StatusCode),
-			RawError:      nil,
+			StatusCode:    result.StatusCode,
 		}
 	}
 
-	// Check for server errors (5xx)
 	if result.StatusCode >= 500 && result.StatusCode < 600 {
-		return body, &asyncapi.ClientError{
+		return resp, &asyncapi.ClientError{
 			ErrorCategory: asyncapi.ErrCategoryServer,
 			Message:       fmt.Sprintf("server error: status code %d", result.StatusCode),
-			RawError:      nil,
+			StatusCode:    result.StatusCode,
 		}
 	}
 
-	return body, nil
+	return resp, nil
 }
 
 // parseRetryAfter parses a Retry-After header value, which can be either

pkg/asyncworker/http_client_test.go

@@ -23,12 +23,15 @@ func TestSendRequest_success(t *testing.T) {
 		}, nil
 	}))
 
-	got, err := client.SendRequest(context.Background(), "http://localhost/v1/completions", nil, []byte(`{}`))
+	resp, err := client.SendRequest(context.Background(), "http://localhost/v1/completions", nil, []byte(`{}`))
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
-	if string(got) != body {
-		t.Errorf("body = %q, want %q", string(got), body)
+	if string(resp.Body) != body {
+		t.Errorf("body = %q, want %q", string(resp.Body), body)
+	}
+	if resp.StatusCode != http.StatusOK {
+		t.Errorf("statusCode = %d, want %d", resp.StatusCode, http.StatusOK)
 	}
 }
 
@@ -68,7 +71,7 @@ func TestSendRequest_rateLimitWithoutRetryAfter(t *testing.T) {
 		}, nil
 	}))
 
-	body, err := client.SendRequest(context.Background(), "http://localhost/v1/completions", nil, []byte(`{}`))
+	resp, err := client.SendRequest(context.Background(), "http://localhost/v1/completions", nil, []byte(`{}`))
 	if err == nil {
 		t.Fatal("expected error for 429 response")
 	}
@@ -79,11 +82,17 @@ func TestSendRequest_rateLimitWithoutRetryAfter(t *testing.T) {
 	if ce.ErrorCategory != asyncapi.ErrCategoryRateLimit {
 		t.Errorf("category = %s, want %s", ce.ErrorCategory, asyncapi.ErrCategoryRateLimit)
 	}
+	if ce.StatusCode != http.StatusTooManyRequests {
+		t.Errorf("StatusCode = %d, want %d", ce.StatusCode, http.StatusTooManyRequests)
+	}
+	if resp.StatusCode != http.StatusTooManyRequests {
+		t.Errorf("returned statusCode = %d, want %d", resp.StatusCode, http.StatusTooManyRequests)
+	}
 	if ce.RetryAfter != 0 {
 		t.Errorf("RetryAfter = %v, want 0 (no header)", ce.RetryAfter)
 	}
-	if string(body) != respBody {
-		t.Errorf("body = %q, want %q", string(body), respBody)
+	if string(resp.Body) != respBody {
+		t.Errorf("body = %q, want %q", string(resp.Body), respBody)
 	}
 }
 
@@ -120,7 +129,7 @@ func TestSendRequest_clientError(t *testing.T) {
 		}, nil
 	}))
 
-	body, err := client.SendRequest(context.Background(), "http://localhost/v1/completions", nil, []byte(`{}`))
+	resp, err := client.SendRequest(context.Background(), "http://localhost/v1/completions", nil, []byte(`{}`))
 	if err == nil {
 		t.Fatal("expected error for 400 response")
 	}
@@ -131,7 +140,13 @@ func TestSendRequest_clientError(t *testing.T) {
 	if ce.ErrorCategory != asyncapi.ErrCategoryInvalidReq {
 		t.Errorf("category = %s, want %s", ce.ErrorCategory, asyncapi.ErrCategoryInvalidReq)
 	}
-	if len(body) == 0 {
+	if ce.StatusCode != http.StatusBadRequest {
+		t.Errorf("StatusCode = %d, want %d", ce.StatusCode, http.StatusBadRequest)
+	}
+	if resp.StatusCode != http.StatusBadRequest {
+		t.Errorf("returned statusCode = %d, want %d", resp.StatusCode, http.StatusBadRequest)
+	}
+	if len(resp.Body) == 0 {
 		t.Error("expected response body to be returned with 4xx error")
 	}
 }
@@ -145,7 +160,7 @@ func TestSendRequest_serverError(t *testing.T) {
 		}, nil
 	}))
 
-	body, err := client.SendRequest(context.Background(), "http://localhost/v1/completions", nil, []byte(`{}`))
+	resp, err := client.SendRequest(context.Background(), "http://localhost/v1/completions", nil, []byte(`{}`))
 	if err == nil {
 		t.Fatal("expected error for 500 response")
 	}
@@ -156,7 +171,13 @@ func TestSendRequest_serverError(t *testing.T) {
 	if ce.ErrorCategory != asyncapi.ErrCategoryServer {
 		t.Errorf("category = %s, want %s", ce.ErrorCategory, asyncapi.ErrCategoryServer)
 	}
-	if len(body) == 0 {
+	if ce.StatusCode != http.StatusInternalServerError {
+		t.Errorf("StatusCode = %d, want %d", ce.StatusCode, http.StatusInternalServerError)
+	}
+	if resp.StatusCode != http.StatusInternalServerError {
+		t.Errorf("returned statusCode = %d, want %d", resp.StatusCode, http.StatusInternalServerError)
+	}
+	if len(resp.Body) == 0 {
 		t.Error("expected response body to be returned with 5xx error")
 	}
 }
@@ -166,7 +187,7 @@ func TestSendRequest_transportError(t *testing.T) {
 		return nil, fmt.Errorf("connection refused")
 	}))
 
-	_, err := client.SendRequest(context.Background(), "http://localhost/v1/completions", nil, []byte(`{}`))
+	resp, err := client.SendRequest(context.Background(), "http://localhost/v1/completions", nil, []byte(`{}`))
 	if err == nil {
 		t.Fatal("expected error for transport failure")
 	}
@@ -177,6 +198,12 @@ func TestSendRequest_transportError(t *testing.T) {
 	if ce.ErrorCategory != asyncapi.ErrCategoryUnknown {
 		t.Errorf("category = %s, want %s", ce.ErrorCategory, asyncapi.ErrCategoryUnknown)
 	}
+	if ce.StatusCode != 0 {
+		t.Errorf("StatusCode = %d, want 0 for transport error", ce.StatusCode)
+	}
+	if resp != nil {
+		t.Errorf("resp = %+v, want nil for transport error", resp)
+	}
 }
 
 func TestSendRequest_invalidURL(t *testing.T) {
@@ -220,6 +247,43 @@ func TestSendRequest_contextCancellation(t *testing.T) {
 	}
 }
 
+func TestSendRequest_bodyReadFailurePreservesStatusCode(t *testing.T) {
+	client := NewHTTPInferenceClient(NewTestClient(func(req *http.Request) (*http.Response, error) {
+		return &http.Response{
+			StatusCode: http.StatusOK,
+			Body:       io.NopCloser(&failReader{}),
+			Header:     make(http.Header),
+		}, nil
+	}))
+
+	resp, err := client.SendRequest(context.Background(), "http://localhost/v1/completions", nil, []byte(`{}`))
+	if err == nil {
+		t.Fatal("expected error for body read failure")
+	}
+	var ce *asyncapi.ClientError
+	if !errors.As(err, &ce) {
+		t.Fatalf("expected *ClientError, got %T", err)
+	}
+	if ce.ErrorCategory != asyncapi.ErrCategoryServer {
+		t.Errorf("category = %s, want %s", ce.ErrorCategory, asyncapi.ErrCategoryServer)
+	}
+	if resp == nil {
+		t.Fatal("expected non-nil response when HTTP response was received")
+	}
+	if resp.StatusCode != http.StatusOK {
+		t.Errorf("returned statusCode = %d, want %d (response was received)", resp.StatusCode, http.StatusOK)
+	}
+	if ce.StatusCode != http.StatusOK {
+		t.Errorf("ClientError.StatusCode = %d, want %d", ce.StatusCode, http.StatusOK)
+	}
+}
+
+type failReader struct{}
+
+func (f *failReader) Read([]byte) (int, error) {
+	return 0, fmt.Errorf("simulated read error")
+}
+
 func TestNewHTTPInferenceClient(t *testing.T) {
 	httpClient := &http.Client{}
 	client := NewHTTPInferenceClient(httpClient)