fix(event): classify personal ticket errors by status before reading body

A 401 whose error body failed mid-read (e.g. unexpected EOF) was wrapped
as retryable by the body-read path, letting the outer reconnect loop
re-enter fetchTicket and refresh again on every iteration, bypassing the
single refresh-retry guard.

Classify non-2xx responses by status first; the body is only drained
best-effort since it is never used for error reporting here. 401 stays
fatal regardless of body state, while 2xx body-read failures remain
retryable transport errors.
This commit is contained in:
shangguanxuan.sgx committed 2026-07-20 17:57:27 +08:00
1 parent 2eca203e74
commit 377ebc5e85
2 files changed
+81 -4

No files matched your search

+10 -4
View File
@@ -245,17 +245,23 @@ func (s *PersonalSource) fetchTicketAttempt(ctx context.Context, accessToken str
return nil, 0, retryPersonal(fmt.Errorf("personal source: fetch ticket: %w", err))
}
defer resp.Body.Close()
data, err := io.ReadAll(io.LimitReader(resp.Body, config.MaxResponseBodySize))
if err != nil {
return nil, resp.StatusCode, retryPersonal(fmt.Errorf("personal source: read ticket response: %w", err))
}
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
// Classify by status before touching the body: a truncated error body
// must not upgrade a fatal status (notably 401) into a retryable
// error, or the outer reconnect loop would bypass the single
// refresh-retry guard. The body is not used here, so drain it only
// best-effort for connection reuse.
_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, config.MaxResponseBodySize))
err := fmt.Errorf("personal source: ticket HTTP %d", resp.StatusCode)
if retryableTicketStatus(resp.StatusCode) {
return nil, resp.StatusCode, retryPersonal(err)
}
return nil, resp.StatusCode, err
}
data, err := io.ReadAll(io.LimitReader(resp.Body, config.MaxResponseBodySize))
if err != nil {
return nil, resp.StatusCode, retryPersonal(fmt.Errorf("personal source: read ticket response: %w", err))
}
ticket, err := decodeTicket(data)
if err != nil {
return nil, resp.StatusCode, err
@@ -314,3 +314,74 @@ func TestPersonalFetchTicket401RefreshFailureStaysFatal(t *testing.T) {
t.Fatalf("failed refresh should stay fatal, got retryable %v", err)
}
}
// brokenBody simulates a response body that fails mid-read, e.g. the server
// closing the connection before the error payload is fully written.
type brokenBody struct{}
func (brokenBody) Read([]byte) (int, error) { return 0, io.ErrUnexpectedEOF }
func (brokenBody) Close() error { return nil }
// TestPersonalFetchTicket401TruncatedBodyStaysFatal guards the single
// refresh-retry protection: a 401 whose body fails with unexpected EOF must
// be classified by status (fatal) and never wrapped as retryable, otherwise
// the outer reconnect loop would refresh again on every iteration.
func TestPersonalFetchTicket401TruncatedBodyStaysFatal(t *testing.T) {
attempts := 0
refreshCalls := 0
src, err := NewPersonal(PersonalConfig{
AccessToken: "stale",
ForceRefreshToken: func(context.Context, string) (string, error) {
refreshCalls++
return "rotated", nil
},
ClientID: "client",
SourceID: "source",
TicketURL: "https://ticket.test",
HTTPClient: &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
attempts++
return &http.Response{StatusCode: http.StatusUnauthorized, Body: brokenBody{}, Header: make(http.Header)}, nil
})},
})
if err != nil {
t.Fatal(err)
}
_, err = src.fetchTicket(context.Background())
if err == nil || !strings.Contains(err.Error(), "HTTP 401") {
t.Fatalf("fatal 401 error expected, got %v", err)
}
if isRetryablePersonalError(err) {
t.Fatalf("401 with truncated body must stay fatal, got retryable %v", err)
}
if refreshCalls != 1 {
t.Fatalf("refresh calls = %d, want 1", refreshCalls)
}
if attempts != 2 {
t.Fatalf("attempts = %d, want 2", attempts)
}
}
// TestPersonalFetchTicket200TruncatedBodyStaysRetryable pins the existing
// behavior for success responses: a body read failure on 2xx is a transient
// transport problem and remains retryable.
func TestPersonalFetchTicket200TruncatedBodyStaysRetryable(t *testing.T) {
src, err := NewPersonal(PersonalConfig{
AccessToken: "token",
ClientID: "client",
SourceID: "source",
TicketURL: "https://ticket.test",
HTTPClient: &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
return &http.Response{StatusCode: 200, Body: brokenBody{}, Header: make(http.Header)}, nil
})},
})
if err != nil {
t.Fatal(err)
}
_, err = src.fetchTicket(context.Background())
if !errors.Is(err, io.ErrUnexpectedEOF) {
t.Fatalf("error should wrap io.ErrUnexpectedEOF, got %v", err)
}
if !isRetryablePersonalError(err) {
t.Fatalf("2xx body read failure should stay retryable, got %v", err)
}
}