From ff6788d61d21f7c40f8edca2d4bf2bcad56f666d Mon Sep 17 00:00:00 2001 From: swadeley Date: Fri, 21 Aug 2026 09:20:10 +0100 Subject: [PATCH 1/2] RHINENG-30154: close HTTP response bodies after retries HTTPCallRetry discarded responses without closing the body on success and retry, leaking TCP sockets from listener Candlepin calls. Co-authored-by: Cursor --- base/utils/http.go | 17 +++++++++--- base/utils/http_test.go | 60 +++++++++++++++++++++++++++++++++++++++++ listener/upload.go | 6 ++--- 3 files changed, 77 insertions(+), 6 deletions(-) diff --git a/base/utils/http.go b/base/utils/http.go index a979f1ef8..70161451b 100644 --- a/base/utils/http.go +++ b/base/utils/http.go @@ -3,6 +3,7 @@ package utils import ( "context" "fmt" + "io" "net/http" "net/http/httputil" "time" @@ -24,29 +25,39 @@ func HTTPCallRetry(httpCallFun func() (outputDataPtr interface{}, resp *http.Res attempt++ outDataPtr, resp, callErr := httpCallFun() if statusCodeFound(resp, codesToRetry) { + closeHTTPResponse(resp) LogWarn("attempt", attempt, "status_code", TryGetStatusCode(resp), "HTTP call ended with wrong status code") continue } if callErr == nil { + closeHTTPResponse(resp) return outDataPtr, nil } if len(codesToRetry) == 0 { // no "retry" codes specified, continue always + closeHTTPResponse(resp) LogWarn("attempt", attempt, "err", callErr, "HTTP call failed, trying again") continue } responseDetails := tryGetResponseDetails(resp) - if resp != nil { - resp.Body.Close() - } + closeHTTPResponse(resp) return nil, errors.Wrap(callErr, "HTTP call failed"+responseDetails) } return nil, errors.Errorf("HTTP retry call failed, attempts: %d", attempt) } +// closeHTTPResponse drains and closes the body so the Transport can release the TCP connection. +func closeHTTPResponse(resp *http.Response) { + if resp == nil || resp.Body == nil { + return + } + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() +} + func CallAPI(client *http.Client, request *http.Request, debugEnabled bool) (*http.Response, error) { if debugEnabled { dump, err := httputil.DumpRequestOut(request, true) diff --git a/base/utils/http_test.go b/base/utils/http_test.go index e0b5de794..a988133c7 100644 --- a/base/utils/http_test.go +++ b/base/utils/http_test.go @@ -2,12 +2,32 @@ package utils import ( "errors" + "io" "net/http" + "strings" + "sync/atomic" "testing" "github.com/stretchr/testify/assert" ) +type closeTrackingBody struct { + io.Reader + closed atomic.Bool +} + +func (b *closeTrackingBody) Close() error { + b.closed.Store(true) + return nil +} + +func trackingResponse(status int, body string) *http.Response { + return &http.Response{ + StatusCode: status, + Body: &closeTrackingBody{Reader: strings.NewReader(body)}, + } +} + func createFirstlyFailingCallFun() func() (interface{}, *http.Response, error) { i := 0 httpCallFun := func() (interface{}, *http.Response, error) { @@ -36,3 +56,43 @@ func TestHTTPCallRetryFail(t *testing.T) { assert.Equal(t, "HTTP retry call failed, attempts: 2", err.Error()) assert.Nil(t, data) } + +func TestHTTPCallRetryClosesBodyOnSuccess(t *testing.T) { + resp := trackingResponse(http.StatusOK, "ok") + data, err := HTTPCallRetry(func() (interface{}, *http.Response, error) { + return "data", resp, nil + }, false, 1) + + assert.NoError(t, err) + assert.Equal(t, "data", data) + assert.True(t, resp.Body.(*closeTrackingBody).closed.Load()) +} + +func TestHTTPCallRetryClosesBodyOnRetryableStatus(t *testing.T) { + first := trackingResponse(http.StatusServiceUnavailable, "retry") + second := trackingResponse(http.StatusOK, "ok") + attempt := 0 + data, err := HTTPCallRetry(func() (interface{}, *http.Response, error) { + attempt++ + if attempt == 1 { + return nil, first, nil + } + return "data", second, nil + }, false, 3, http.StatusServiceUnavailable) + + assert.NoError(t, err) + assert.Equal(t, "data", data) + assert.True(t, first.Body.(*closeTrackingBody).closed.Load()) + assert.True(t, second.Body.(*closeTrackingBody).closed.Load()) +} + +func TestHTTPCallRetryClosesBodyOnNonRetryableError(t *testing.T) { + resp := trackingResponse(http.StatusBadRequest, "nope") + data, err := HTTPCallRetry(func() (interface{}, *http.Response, error) { + return nil, resp, errors.New("bad request") + }, false, 3, http.StatusServiceUnavailable) + + assert.Error(t, err) + assert.Nil(t, data) + assert.True(t, resp.Body.(*closeTrackingBody).closed.Load()) +} diff --git a/listener/upload.go b/listener/upload.go index eee481220..b916ce0aa 100644 --- a/listener/upload.go +++ b/listener/upload.go @@ -838,12 +838,12 @@ func getYumUpdates(event HostEvent, client *api.Client) (*YumUpdates, error) { if yumUpdatesURL != nil && *yumUpdatesURL != "" { resp, err := client.Request(&base.Context, http.MethodGet, *yumUpdatesURL, nil, &parsed) + if resp != nil && resp.Body != nil { + defer resp.Body.Close() + } if err != nil { return nil, errors.Wrap(err, "unable to get yum updates from S3") } - if err := resp.Body.Close(); err != nil { - return nil, errors.Wrap(err, "response error for yum updates from S3") - } } if (parsed == vmaas.UpdatesV3Response{}) { From a9dabb27bae2bdb60eb4061e66777fd2b8af9bce Mon Sep 17 00:00:00 2001 From: swadeley Date: Fri, 21 Aug 2026 09:33:56 +0100 Subject: [PATCH 2/2] RHINENG-30154: limit HTTP body drain and log close errors Unbounded io.Copy could read huge leftover bodies; cap the drain at net/http's reuse limit and log yum-update Body.Close failures. Co-authored-by: Cursor --- base/utils/http.go | 9 ++++-- base/utils/http_test.go | 69 +++++++++++++++++++++++++++++++++-------- listener/upload.go | 6 +++- 3 files changed, 68 insertions(+), 16 deletions(-) diff --git a/base/utils/http.go b/base/utils/http.go index 70161451b..c4eafff11 100644 --- a/base/utils/http.go +++ b/base/utils/http.go @@ -49,12 +49,17 @@ func HTTPCallRetry(httpCallFun func() (outputDataPtr interface{}, resp *http.Res return nil, errors.Errorf("HTTP retry call failed, attempts: %d", attempt) } -// closeHTTPResponse drains and closes the body so the Transport can release the TCP connection. +// maxHTTPResponseDrain matches net/http maxPostHandlerReadBytes: enough leftover +// body to allow connection reuse without reading arbitrarily large payloads. +const maxHTTPResponseDrain = 256 << 10 + +// closeHTTPResponse drains up to maxHTTPResponseDrain and closes the body so the +// Transport can release the TCP connection. func closeHTTPResponse(resp *http.Response) { if resp == nil || resp.Body == nil { return } - _, _ = io.Copy(io.Discard, resp.Body) + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxHTTPResponseDrain)) _ = resp.Body.Close() } diff --git a/base/utils/http_test.go b/base/utils/http_test.go index a988133c7..a659ef0f5 100644 --- a/base/utils/http_test.go +++ b/base/utils/http_test.go @@ -12,20 +12,44 @@ import ( ) type closeTrackingBody struct { - io.Reader - closed atomic.Bool + rc io.ReadCloser + readBytes atomic.Int64 + closed atomic.Bool +} + +func newCloseTrackingBody(content string) *closeTrackingBody { + return &closeTrackingBody{rc: io.NopCloser(strings.NewReader(content))} +} + +func (b *closeTrackingBody) Read(p []byte) (int, error) { + n, err := b.rc.Read(p) + if n > 0 { + b.readBytes.Add(int64(n)) + } + return n, err } func (b *closeTrackingBody) Close() error { b.closed.Store(true) - return nil + return b.rc.Close() } -func trackingResponse(status int, body string) *http.Response { +func trackingResponse(status int, body string) (*http.Response, *closeTrackingBody) { + tb := newCloseTrackingBody(body) return &http.Response{ StatusCode: status, - Body: &closeTrackingBody{Reader: strings.NewReader(body)}, + Body: tb, + }, tb +} + +func assertBodyDrainedAndClosed(t *testing.T, body *closeTrackingBody, content string) { + t.Helper() + assert.True(t, body.closed.Load()) + expected := int64(len(content)) + if expected > maxHTTPResponseDrain { + expected = maxHTTPResponseDrain } + assert.Equal(t, expected, body.readBytes.Load()) } func createFirstlyFailingCallFun() func() (interface{}, *http.Response, error) { @@ -58,19 +82,22 @@ func TestHTTPCallRetryFail(t *testing.T) { } func TestHTTPCallRetryClosesBodyOnSuccess(t *testing.T) { - resp := trackingResponse(http.StatusOK, "ok") + const content = "ok" + resp, body := trackingResponse(http.StatusOK, content) data, err := HTTPCallRetry(func() (interface{}, *http.Response, error) { return "data", resp, nil }, false, 1) assert.NoError(t, err) assert.Equal(t, "data", data) - assert.True(t, resp.Body.(*closeTrackingBody).closed.Load()) + assertBodyDrainedAndClosed(t, body, content) } func TestHTTPCallRetryClosesBodyOnRetryableStatus(t *testing.T) { - first := trackingResponse(http.StatusServiceUnavailable, "retry") - second := trackingResponse(http.StatusOK, "ok") + const firstContent = "retry" + const secondContent = "ok" + first, firstBody := trackingResponse(http.StatusServiceUnavailable, firstContent) + second, secondBody := trackingResponse(http.StatusOK, secondContent) attempt := 0 data, err := HTTPCallRetry(func() (interface{}, *http.Response, error) { attempt++ @@ -82,17 +109,33 @@ func TestHTTPCallRetryClosesBodyOnRetryableStatus(t *testing.T) { assert.NoError(t, err) assert.Equal(t, "data", data) - assert.True(t, first.Body.(*closeTrackingBody).closed.Load()) - assert.True(t, second.Body.(*closeTrackingBody).closed.Load()) + assertBodyDrainedAndClosed(t, firstBody, firstContent) + assertBodyDrainedAndClosed(t, secondBody, secondContent) } func TestHTTPCallRetryClosesBodyOnNonRetryableError(t *testing.T) { - resp := trackingResponse(http.StatusBadRequest, "nope") + const content = "nope" + resp, body := trackingResponse(http.StatusBadRequest, content) data, err := HTTPCallRetry(func() (interface{}, *http.Response, error) { return nil, resp, errors.New("bad request") }, false, 3, http.StatusServiceUnavailable) assert.Error(t, err) assert.Nil(t, data) - assert.True(t, resp.Body.(*closeTrackingBody).closed.Load()) + assertBodyDrainedAndClosed(t, body, content) +} + +func TestCloseHTTPResponseDrainsBody(t *testing.T) { + const content = "some body data" + body := newCloseTrackingBody(content) + closeHTTPResponse(&http.Response{StatusCode: http.StatusOK, Body: body}) + assertBodyDrainedAndClosed(t, body, content) +} + +func TestCloseHTTPResponseLimitsDrain(t *testing.T) { + content := strings.Repeat("x", maxHTTPResponseDrain+1024) + body := newCloseTrackingBody(content) + closeHTTPResponse(&http.Response{StatusCode: http.StatusOK, Body: body}) + assert.True(t, body.closed.Load()) + assert.Equal(t, int64(maxHTTPResponseDrain), body.readBytes.Load()) } diff --git a/listener/upload.go b/listener/upload.go index b916ce0aa..a626b3e6f 100644 --- a/listener/upload.go +++ b/listener/upload.go @@ -839,7 +839,11 @@ func getYumUpdates(event HostEvent, client *api.Client) (*YumUpdates, error) { if yumUpdatesURL != nil && *yumUpdatesURL != "" { resp, err := client.Request(&base.Context, http.MethodGet, *yumUpdatesURL, nil, &parsed) if resp != nil && resp.Body != nil { - defer resp.Body.Close() + defer func() { + if closeErr := resp.Body.Close(); closeErr != nil { + utils.LogWarn("err", closeErr, "closing yum updates response body") + } + }() } if err != nil { return nil, errors.Wrap(err, "unable to get yum updates from S3")