Skip to content

Commit 17ab15d

Browse files
j-mok-devj-mok-dev
authored
fix(processor): propagate async ResultMessage HTTP status to batch output (llm-d#590)
Async dispatch discarded ResultMessage.StatusCode and hardcoded 200, so auth failures (e.g. 403 with empty body) became parse_error instead of preserving the real status. Propagate status/error fields through GetResult and asyncResult, and cover with unit + dispatcher e2e. Fixes llm-d#589 Signed-off-by: Jooyeon Mok <jmok@redhat.com>
1 parent 811674e commit 17ab15d

7 files changed

Lines changed: 379 additions & 60 deletions

File tree

‎internal/processor/pipeline/async_test.go‎

Lines changed: 118 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -55,8 +55,9 @@ func (c *fakeAsyncClient) Close() error { return nil }
5555
func (c *fakeAsyncClient) deliver(requestID string, body map[string]any) {
5656
resp, _ := json.Marshal(body)
5757
c.results <- &inference.GenerateResponse{
58-
RequestID: requestID,
59-
Response: resp,
58+
RequestID: requestID,
59+
Response: resp,
60+
StatusCode: 200,
6061
}
6162
}
6263

@@ -127,34 +128,123 @@ func TestAsyncEndToEnd(t *testing.T) {
127128
}
128129
}
129130

130-
func TestAsyncResult_NilResponseBody(t *testing.T) {
131-
resp := &inference.GenerateResponse{
132-
RequestID: "req-nil-body",
133-
Response: nil,
134-
}
135-
result := asyncResult(resp, logr.Discard())
136-
if result.Error == nil {
137-
t.Fatal("expected error for nil response body")
138-
}
139-
if result.Error.Code != "server_error" {
140-
t.Fatalf("error code = %q, want %q", result.Error.Code, "server_error")
141-
}
142-
if result.Response != nil {
143-
t.Fatalf("expected nil response, got %+v", result.Response)
131+
func TestAsyncResult(t *testing.T) {
132+
tests := []struct {
133+
name string
134+
resp *inference.GenerateResponse
135+
wantStatusCode int
136+
wantErrCode string
137+
wantResponse bool
138+
}{
139+
{
140+
name: "nil body treated as empty 2xx",
141+
resp: &inference.GenerateResponse{
142+
RequestID: "req-nil-body",
143+
Response: nil,
144+
},
145+
wantErrCode: "server_error",
146+
},
147+
{
148+
name: "bad JSON on 2xx is parse_error",
149+
resp: &inference.GenerateResponse{
150+
RequestID: "req-bad-json",
151+
Response: []byte(`{not valid json`),
152+
StatusCode: 200,
153+
},
154+
wantErrCode: "parse_error",
155+
},
156+
{
157+
name: "legacy success without StatusCode defaults to 200",
158+
resp: &inference.GenerateResponse{
159+
RequestID: "req-ok",
160+
Response: []byte(`{"choices":[]}`),
161+
},
162+
wantStatusCode: 200,
163+
wantResponse: true,
164+
},
165+
{
166+
name: "success with StatusCode 200",
167+
resp: &inference.GenerateResponse{
168+
RequestID: "req-ok-200",
169+
Response: []byte(`{"choices":[]}`),
170+
StatusCode: 200,
171+
},
172+
wantStatusCode: 200,
173+
wantResponse: true,
174+
},
175+
{
176+
name: "HTTP 403 with empty body preserves status",
177+
resp: &inference.GenerateResponse{
178+
RequestID: "req-403",
179+
Response: []byte{},
180+
StatusCode: 403,
181+
},
182+
wantStatusCode: 403,
183+
wantResponse: true,
184+
},
185+
{
186+
name: "HTTP 403 with JSON error body preserves status",
187+
resp: &inference.GenerateResponse{
188+
RequestID: "req-403-json",
189+
Response: []byte(`{"error":{"message":"unauthorized"}}`),
190+
StatusCode: 403,
191+
},
192+
wantStatusCode: 403,
193+
wantResponse: true,
194+
},
195+
{
196+
name: "HTTP 422 with unparseable body preserves status",
197+
resp: &inference.GenerateResponse{
198+
RequestID: "req-422",
199+
Response: []byte(`not-json`),
200+
StatusCode: 422,
201+
},
202+
wantStatusCode: 422,
203+
wantResponse: true,
204+
},
205+
{
206+
name: "non-HTTP failure uses ErrorCode",
207+
resp: &inference.GenerateResponse{
208+
RequestID: "req-deadline",
209+
StatusCode: 0,
210+
ErrorCode: "DEADLINE_EXCEEDED",
211+
ErrorMessage: "deadline exceeded",
212+
},
213+
wantErrCode: "DEADLINE_EXCEEDED",
214+
},
144215
}
145-
}
146216

147-
func TestAsyncResult_BadJSONBody(t *testing.T) {
148-
resp := &inference.GenerateResponse{
149-
RequestID: "req-bad-json",
150-
Response: []byte(`{not valid json`),
151-
}
152-
result := asyncResult(resp, logr.Discard())
153-
if result.Error == nil {
154-
t.Fatal("expected error for bad JSON body")
155-
}
156-
if result.Error.Code != "parse_error" {
157-
t.Fatalf("error code = %q, want %q", result.Error.Code, "parse_error")
217+
for _, tt := range tests {
218+
t.Run(tt.name, func(t *testing.T) {
219+
result := asyncResult(tt.resp, logr.Discard())
220+
if tt.wantErrCode != "" {
221+
if result.Error == nil {
222+
t.Fatalf("expected error code %q, got nil error", tt.wantErrCode)
223+
}
224+
if result.Error.Code != tt.wantErrCode {
225+
t.Fatalf("error code = %q, want %q", result.Error.Code, tt.wantErrCode)
226+
}
227+
if result.Response != nil {
228+
t.Fatalf("expected nil response, got %+v", result.Response)
229+
}
230+
return
231+
}
232+
if result.Error != nil {
233+
t.Fatalf("unexpected error: %+v", result.Error)
234+
}
235+
if !tt.wantResponse {
236+
return
237+
}
238+
if result.Response == nil {
239+
t.Fatal("expected response")
240+
}
241+
if result.Response.StatusCode != tt.wantStatusCode {
242+
t.Fatalf("StatusCode = %d, want %d", result.Response.StatusCode, tt.wantStatusCode)
243+
}
244+
if result.Response.RequestID != tt.resp.RequestID {
245+
t.Fatalf("RequestID = %q, want %q", result.Response.RequestID, tt.resp.RequestID)
246+
}
247+
})
158248
}
159249
}
160250

@@ -178,26 +268,6 @@ func TestSafeChannelSend(t *testing.T) {
178268
})
179269
}
180270

181-
func TestAsyncResult_Success(t *testing.T) {
182-
resp := &inference.GenerateResponse{
183-
RequestID: "req-ok",
184-
Response: []byte(`{"choices":[]}`),
185-
}
186-
result := asyncResult(resp, logr.Discard())
187-
if result.Error != nil {
188-
t.Fatalf("unexpected error: %+v", result.Error)
189-
}
190-
if result.Response == nil {
191-
t.Fatal("expected response")
192-
}
193-
if result.Response.StatusCode != 200 {
194-
t.Fatalf("StatusCode = %d, want 200", result.Response.StatusCode)
195-
}
196-
if result.Response.RequestID != "req-ok" {
197-
t.Fatalf("RequestID = %q, want %q", result.Response.RequestID, "req-ok")
198-
}
199-
}
200-
201271
func TestBroadcaster_RetriesTransientError(t *testing.T) {
202272
synctest.Test(t, func(t *testing.T) {
203273
var callCount atomic.Int32

‎internal/processor/pipeline/result_router.go‎

Lines changed: 46 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -136,21 +136,59 @@ func asyncResult(resp *inference.GenerateResponse, logger logr.Logger) ResultIte
136136
result := ResultItem{
137137
RequestID: resp.RequestID,
138138
}
139-
if resp.Response == nil {
140-
result.Error = &OutputError{Code: "server_error", Message: "async response has no body"}
139+
140+
if resp.IsNonHTTPFailure() {
141+
code := resp.ErrorCode
142+
if code == "" {
143+
code = "server_error"
144+
}
145+
msg := resp.ErrorMessage
146+
if msg == "" {
147+
msg = "async request failed"
148+
}
149+
result.Error = &OutputError{Code: code, Message: msg}
150+
return result
151+
}
152+
153+
statusCode := resp.StatusCode
154+
if statusCode == 0 {
155+
// Legacy ResultMessage with only Payload — treat as HTTP 200.
156+
statusCode = 200
157+
}
158+
159+
if len(resp.Response) == 0 {
160+
if statusCode >= 200 && statusCode < 300 {
161+
result.Error = &OutputError{Code: "server_error", Message: "async response has no body"}
162+
return result
163+
}
164+
// Non-2xx with empty body (e.g. 403 from auth): preserve status code.
165+
result.Response = &batch_types.ResponseData{
166+
StatusCode: statusCode,
167+
RequestID: resp.RequestID,
168+
Body: map[string]any{},
169+
}
141170
return result
142171
}
172+
143173
var body map[string]any
144174
if err := json.Unmarshal(resp.Response, &body); err != nil {
145-
logger.Error(err, "Failed to unmarshal async response", "requestID", resp.RequestID)
146-
result.Error = &OutputError{
147-
Code: "parse_error",
148-
Message: fmt.Sprintf("response body could not be parsed: %v", err),
175+
if statusCode >= 200 && statusCode < 300 {
176+
logger.Error(err, "Failed to unmarshal async response", "requestID", resp.RequestID)
177+
result.Error = &OutputError{
178+
Code: "parse_error",
179+
Message: fmt.Sprintf("response body could not be parsed: %v", err),
180+
}
181+
return result
182+
}
183+
// Non-2xx with unparseable body: still preserve the HTTP status.
184+
body = map[string]any{
185+
"error": map[string]any{
186+
"message": string(resp.Response),
187+
},
149188
}
150-
return result
151189
}
152190
result.Response = &batch_types.ResponseData{
153-
StatusCode: 200,
191+
StatusCode: statusCode,
154192
RequestID: resp.RequestID,
155193
Body: body,
156194
}

‎pkg/clients/inference/async_inference_client_integration_test.go‎

Lines changed: 57 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,11 @@ func TestAsyncSharedClient_Submit_roundtrip(t *testing.T) {
5454

5555
go func() {
5656
time.Sleep(50 * time.Millisecond)
57-
data, _ := json.Marshal(api.ResultMessage{ID: "req-1", Payload: `{"choices":[{"text":"hello"}]}`})
57+
data, _ := json.Marshal(api.ResultMessage{
58+
ID: "req-1",
59+
StatusCode: 200,
60+
Payload: `{"choices":[{"text":"hello"}]}`,
61+
})
5862
if _, lpushErr := mr.Lpush(resultQueue, string(data)); lpushErr != nil {
5963
t.Errorf("Lpush: %v", lpushErr)
6064
}
@@ -79,6 +83,9 @@ func TestAsyncSharedClient_Submit_roundtrip(t *testing.T) {
7983
if resp.RequestID != "req-1" {
8084
t.Errorf("RequestID = %q, want %q", resp.RequestID, "req-1")
8185
}
86+
if resp.StatusCode != 200 {
87+
t.Errorf("StatusCode = %d, want 200", resp.StatusCode)
88+
}
8289
if resp.Response == nil {
8390
t.Fatal("expected non-nil Response")
8491
}
@@ -103,4 +110,53 @@ func TestAsyncSharedClient_Submit_roundtrip(t *testing.T) {
103110
t.Errorf("enqueued request ID = %q, want %q", data.ID, "req-1")
104111
}
105112
})
113+
114+
t.Run("propagates HTTP error status code from ResultMessage", func(t *testing.T) {
115+
mr := miniredis.RunT(t)
116+
poolName := "test-pool-403"
117+
resultQueue := asyncQueuePrefix + "results:" + poolName
118+
119+
rdb := redis.NewClient(&redis.Options{Addr: mr.Addr()})
120+
defer func() { _ = rdb.Close() }()
121+
122+
p, err := producer.NewRedisSortedSetProducer(
123+
producer.RedisSortedSetConfig{
124+
RequestQueueName: asyncQueuePrefix + "requests:" + poolName,
125+
ResultQueueName: resultQueue,
126+
},
127+
producer.WithRedisClient(rdb),
128+
)
129+
if err != nil {
130+
t.Fatalf("NewRedisSortedSetProducer: %v", err)
131+
}
132+
defer func() { _ = p.Close() }()
133+
134+
client := newAsyncSharedClient(p, time.Second, testLogger(t))
135+
136+
data, _ := json.Marshal(api.ResultMessage{
137+
ID: "req-403",
138+
StatusCode: 403,
139+
Payload: "",
140+
})
141+
if _, lpushErr := mr.Lpush(resultQueue, string(data)); lpushErr != nil {
142+
t.Fatalf("Lpush: %v", lpushErr)
143+
}
144+
145+
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
146+
defer cancel()
147+
148+
resp, getErr := client.GetResult(ctx)
149+
if getErr != nil {
150+
t.Fatalf("GetResult error: %v", getErr)
151+
}
152+
if resp.RequestID != "req-403" {
153+
t.Errorf("RequestID = %q, want %q", resp.RequestID, "req-403")
154+
}
155+
if resp.StatusCode != 403 {
156+
t.Errorf("StatusCode = %d, want 403", resp.StatusCode)
157+
}
158+
if len(resp.Response) != 0 {
159+
t.Errorf("Response = %q, want empty", resp.Response)
160+
}
161+
})
106162
}

‎pkg/clients/inference/async_shared_client.go‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -78,8 +78,11 @@ func (c *asyncSharedClient) GetResult(ctx context.Context) (*GenerateResponse, e
7878
}
7979

8080
return &GenerateResponse{
81-
RequestID: result.ID,
82-
Response: []byte(result.Payload),
81+
RequestID: result.ID,
82+
Response: []byte(result.Payload),
83+
StatusCode: result.StatusCode,
84+
ErrorCode: result.ErrorCode,
85+
ErrorMessage: result.ErrorMessage,
8386
}, nil
8487
}
8588

‎pkg/clients/inference/inference_client_interface.go‎

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,14 +35,29 @@ type GenerateRequest struct {
3535
Headers map[string]string // extra headers to forward to the endpoint
3636
}
3737

38-
// GenerateResponse represents an inference generation response
38+
// GenerateResponse represents an inference generation response.
39+
//
40+
// For async results (from llm-d-async ResultMessage):
41+
// - StatusCode > 0: an HTTP response was received; Response holds the body.
42+
// - StatusCode == 0 with ErrorCode/ErrorMessage: no HTTP response (deadline,
43+
// cancel, gate drop, etc.).
44+
// - StatusCode == 0 with empty ErrorCode: legacy success payload (treat as 200).
3945
type GenerateResponse struct {
4046
RequestID string
4147
Response []byte
48+
StatusCode int // HTTP status from async ResultMessage; 0 = unset/non-HTTP
49+
ErrorCode string // non-HTTP failure code from async ResultMessage
50+
ErrorMessage string // non-HTTP failure message from async ResultMessage
4251
RawData interface{}
4352
HadCapacityRetry bool // true if any retry was caused by 429/5xx (not network error)
4453
}
4554

55+
// IsNonHTTPFailure reports whether the response represents a failure that did
56+
// not produce an HTTP status (e.g. deadline exceeded, cancel, gate drop).
57+
func (r *GenerateResponse) IsNonHTTPFailure() bool {
58+
return r.StatusCode == 0 && (r.ErrorCode != "" || r.ErrorMessage != "")
59+
}
60+
4661
// ClientError represents an inference client error
4762
type ClientError = httpclient.ClientError
4863

‎test/e2e/dispatcher/processor-async-values.yaml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,3 +6,5 @@ processor:
66
modelGateways:
77
sim-model:
88
inferencePoolName: "sim-pool"
9+
sim-model-inject:
10+
inferencePoolName: "sim-pool-inject"

0 commit comments

Comments
 (0)