From 7eaf4cfcb98b385d95f9d396bbfa8aa8e70707a7 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 15 Oct 2026 14:15:49 +0200 Subject: [PATCH 1/2] test(aws): trace baseline loopback connection closure Capture bounded failure-only lifecycle events around the real SDK fixture to diagnose intermittent response-body connection closure. Refs #414 --- providers/aws/ladder/baseline_sdk_test.go | 25 ++- .../loopback_diagnostic_control_test.go | 184 +++++++++++++++++ .../aws/ladder/loopback_diagnostic_test.go | 191 ++++++++++++++++++ 3 files changed, 395 insertions(+), 5 deletions(-) create mode 100644 providers/aws/ladder/loopback_diagnostic_control_test.go create mode 100644 providers/aws/ladder/loopback_diagnostic_test.go diff --git a/providers/aws/ladder/baseline_sdk_test.go b/providers/aws/ladder/baseline_sdk_test.go index 9b87168..3ac518d 100644 --- a/providers/aws/ladder/baseline_sdk_test.go +++ b/providers/aws/ladder/baseline_sdk_test.go @@ -25,11 +25,18 @@ import ( // Constructed CE responses exercise the public factory without cloud access. func TestScopedBaselineParentCompatible(t *testing.T) { + runLoopbackFixture(t, scopedBaselineFixture) +} + +func scopedBaselineFixture(t *testing.T) { for _, mode := range []string{"account", "reverse", "zero", "lag2", "lag3", "duplicate", "page-duplicate", "hole", "leading", "outside", "future", "bad-start", "nil-start", "nil-period", "nil-end", "bad-end", "short-period", "long-period", "backward-period", "empty", "all-zero", "missing-total", "missing-metric", "missing-amount", "bad-amount", "NaN", "Inf", "negative", "stale", "few", "provider-error"} { - t.Run(mode, func(t *testing.T) { + if !t.Run(mode, func(t *testing.T) { + d := newLoopDiagnostic(t) var calls atomic.Int32 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - calls.Add(1) + page := calls.Add(1) + d.record(loopEvent{Kind: "handler_begin", Page: int(page)}, r.Context(), nil) + defer d.record(loopEvent{Kind: "handler_end", Page: int(page)}, r.Context(), nil) assert.Equal(t, "AWSInsightsIndexService.GetCostAndUsage", r.Header.Get("X-Amz-Target")) var in costexplorer.GetCostAndUsageInput if !assert.NoError(t, json.NewDecoder(r.Body).Decode(&in)) { @@ -82,12 +89,18 @@ func TestScopedBaselineParentCompatible(t *testing.T) { w.Header().Set("Content-Type", "application/x-amz-json-1.1") assert.NoError(t, json.NewEncoder(w).Encode(map[string]any{"ResultsByTime": rows, "NextPageToken": token})) })) - defer srv.Close() + defer func() { + d.record(loopEvent{Kind: "server_close_begin"}, nil, nil) + srv.Close() + d.record(loopEvent{Kind: "server_close_end"}, nil, nil) + }() + transport := d.transport(srv.Client().Transport) + defer transport.http.CloseIdleConnections() endpoint, err := url.Parse(srv.URL) require.NoError(t, err) a, err := awsladder.NewFromConfig(aws.Config{Region: "eu-west-1", RetryMaxAttempts: 1, Credentials: credentials.NewStaticCredentialsProvider("synthetic", "synthetic", ""), - HTTPClient: &http.Client{Transport: ceLoopback{endpoint: endpoint, transport: srv.Client().Transport}}, + HTTPClient: &http.Client{Transport: ceLoopback{endpoint: endpoint, transport: transport}}, }, awsladder.Config{Region: "eu-west-1", AccountID: "fixture-account"}) require.NoError(t, err) scope := ladder.Scope{Provider: common.ProviderAWS, AccountID: "wrong"} @@ -119,7 +132,9 @@ func TestScopedBaselineParentCompatible(t *testing.T) { if mode == "account" || mode == "reverse" { assert.Equal(t, int32(2), calls.Load()) } - }) + }) { + return + } } } diff --git a/providers/aws/ladder/loopback_diagnostic_control_test.go b/providers/aws/ladder/loopback_diagnostic_control_test.go new file mode 100644 index 0000000..126aa86 --- /dev/null +++ b/providers/aws/ladder/loopback_diagnostic_control_test.go @@ -0,0 +1,184 @@ +package ladder_test + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "net/http/httptrace" + "net/url" + "runtime" + "strings" + "sync" + "syscall" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/costexplorer" + "github.com/aws/aws-sdk-go-v2/service/costexplorer/types" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func runLoopbackFixture(t *testing.T, fixture func(*testing.T)) { + iterations := 1 + if runtime.GOOS == "linux" { + iterations = 33 + } + for i := range iterations { + if !t.Run(fmt.Sprint(i), fixture) { + return + } + } +} +func waitLoopControl(t *testing.T, done <-chan struct{}) bool { + t.Helper() + select { + case <-done: + return true + case <-time.After(5 * time.Second): + t.Error("control did not complete") + return false + } +} + +func TestLoopbackDiagnosticLifecycle(t *testing.T) { + for _, cancelRequest := range []bool{true, false} { + t.Run(fmt.Sprint(cancelRequest), func(t *testing.T) { + d := newLoopDiagnostic(t) + deadline, stop := context.WithTimeout(context.Background(), 5*time.Second) + defer stop() + ctx, cancel := context.WithCancelCause(deadline) + release, done, handlerDone, wrote := make(chan struct{}), make(chan struct{}), make(chan struct{}), make(chan struct{}) + var operationErr, writeErr, flushErr error + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + defer close(handlerDone) + d.record(loopEvent{Kind: "handler_begin"}, r.Context(), nil) + defer d.record(loopEvent{Kind: "handler_end"}, r.Context(), nil) + w.Header().Set("Content-Type", "application/x-amz-json-1.1") + _, writeErr = w.Write([]byte("{\"ResultsByTime\":[")) + flushErr = http.NewResponseController(w).Flush() + <-release + })) + tr := d.transport(srv.Client().Transport) + tr.readStarted = make(chan struct{}) + cleanup := sync.OnceFunc(func() { + defer tr.http.CloseIdleConnections() + cancel(nil) + close(release) + waitLoopControl(t, done) + handlerJoined := waitLoopControl(t, handlerDone) + waitLoopControl(t, wrote) + d.record(loopEvent{Kind: "server_close_begin"}, nil, nil) + if !handlerJoined { + srv.CloseClientConnections() + d.record(loopEvent{Kind: "server_close_unjoined"}, nil, srv.Listener.Close()) + return + } + srv.Close() + d.record(loopEvent{Kind: "server_close_end"}, nil, nil) + assert.NoError(t, errors.Join(writeErr, flushErr)) + }) + defer cleanup() + endpoint, err := url.Parse(srv.URL) + require.NoError(t, err) + gotConn := make(chan *loopConn, 1) + ctx = httptrace.WithClientTrace(ctx, &httptrace.ClientTrace{ + GotConn: func(info httptrace.GotConnInfo) { gotConn <- info.Conn.(*loopConn) }, + WroteRequest: func(httptrace.WroteRequestInfo) { close(wrote) }, + }) + client := costexplorer.NewFromConfig(aws.Config{Region: "us-east-1", RetryMaxAttempts: 1, Credentials: credentials.NewStaticCredentialsProvider("synthetic", "synthetic", ""), HTTPClient: &http.Client{Transport: ceLoopback{endpoint: endpoint, transport: tr}}}) + go func() { + defer close(done) + _, operationErr = client.GetCostAndUsage(ctx, &costexplorer.GetCostAndUsageInput{Granularity: types.GranularityDaily, Metrics: []string{"UnblendedCost"}, TimePeriod: &types.DateInterval{Start: aws.String("2026-10-01"), End: aws.String("2026-10-02")}}) + }() + require.True(t, waitLoopControl(t, tr.readStarted)) + var conn *loopConn + select { + case conn = <-gotConn: + case <-deadline.Done(): + t.Fatal("connection not observed") + } + var _ io.ReaderFrom = conn + var _ io.WriterTo = conn + var _ syscall.Conn = conn + var _ interface { + CloseRead() error + CloseWrite() error + } = conn + if cancelRequest { + cancel(errors.New("control cancellation cause")) + } else { + require.NoError(t, conn.Close()) + } + require.True(t, waitLoopControl(t, done)) + require.Error(t, operationErr) + if !cancelRequest { + require.NoError(t, ctx.Err()) + } + cleanup() + kinds := map[string]loopEvent{} + explicitClose := false + for _, e := range d.freeze().Events { + kinds[e.Kind] = e + if e.Kind == "connection_close_begin" { + require.Equal(t, [2]string{}, [2]string{e.Context, e.Cause}) + require.LessOrEqual(t, len(e.Frames), 8) + for _, frame := range e.Frames { + explicitClose = explicitClose || strings.Contains(frame, "TestLoopbackDiagnosticLifecycle") + require.Regexp(t, `^\S+ [^/ ]+:\d+$`, frame) + } + } + } + require.Equal(t, 200, kinds["headers"].N) + require.Contains(t, []string{"canceled", "closed", "other"}, kinds["body_read"].Err) + require.Positive(t, kinds["got_conn"].Conn) + require.Equal(t, kinds["got_conn"].Conn, kinds["body_read"].Conn) + require.Equal(t, "none", kinds["wrote_request"].Err) + require.IsIncreasing(t, []int{kinds["headers"].Seq, kinds["body_read"].Seq, kinds["body_close_end"].Seq}) + require.IsIncreasing(t, []int{0, kinds["handler_end"].Seq, kinds["server_close_end"].Seq}) + require.True(t, kinds["body_read"].Deadline) + require.Equal(t, map[bool]string{true: "canceled", false: "none"}[cancelRequest], kinds["body_read"].Context) + require.Equal(t, map[bool]string{true: "other", false: "none"}[cancelRequest], kinds["body_read"].Cause) + require.True(t, cancelRequest || explicitClose) + }) + } +} + +func TestLoopbackDiagnosticCutoff(t *testing.T) { + d := newLoopDiagnostic(t) + start, recorded, cut := make(chan struct{}), make(chan struct{}), make(chan struct{}) + var raced loopSnapshot + joinWorkers := func() { + require.NotContains(t, []bool{waitLoopControl(t, recorded), waitLoopControl(t, cut)}, false) + } + defer joinWorkers() + go func() { defer close(recorded); <-start; d.record(loopEvent{Kind: "racing"}, nil, nil) }() + go func() { defer close(cut); <-start; raced = d.freeze() }() + close(start) + joinWorkers() + require.True(t, len(raced.Events) <= 1 && strings.Contains(raced.Incomplete, "INCOMPLETE"), "%+v", raced) + bounded := newLoopDiagnostic(t) + for range 130 { + bounded.record(loopEvent{Kind: "error"}, nil, errors.New("sensitive-token/path")) + } + local, peer := net.Pipe() + defer local.Close() + defer peer.Close() + tr := bounded.transport(&http.Transport{DialContext: func(context.Context, string, string) (net.Conn, error) { return local, nil }}) + conn, err := tr.http.DialContext(context.Background(), "tcp", "unused") + require.NoError(t, err) + require.Same(t, local, conn) + snapshot := bounded.freeze() + require.Len(t, snapshot.Events, 128) + require.Equal(t, 2, snapshot.Overflow) + require.True(t, snapshot.Unsupported) + require.NotContains(t, fmt.Sprint(snapshot), "sensitive-token/path") + bounded.record(loopEvent{Kind: "after_overflow_cutoff"}, nil, nil) + require.Equal(t, snapshot, bounded.freeze()) +} diff --git a/providers/aws/ladder/loopback_diagnostic_test.go b/providers/aws/ladder/loopback_diagnostic_test.go new file mode 100644 index 0000000..d6c9b4e --- /dev/null +++ b/providers/aws/ladder/loopback_diagnostic_test.go @@ -0,0 +1,191 @@ +package ladder_test + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httptrace" + "path/filepath" + "runtime" + "sync" + "sync/atomic" + "testing" + "time" +) + +type loopEvent struct { + Seq, Page, Conn, N int + Kind, Err, Context, Cause string + Deadline bool + Remaining time.Duration + Frames []string +} +type loopSnapshot struct { + Incomplete string + Events []loopEvent + Overflow int + Unsupported bool +} +type loopDiagnostic struct { + mu sync.Mutex + events []loopEvent + seq, overflow int + unsupported bool + snapshot *loopSnapshot + pages, conns atomic.Int32 + http *http.Transport + readStarted chan struct{} +} + +func newLoopDiagnostic(t *testing.T) *loopDiagnostic { + d := &loopDiagnostic{} + t.Cleanup(func() { + if t.Failed() { + t.Logf("loopback diagnostic: %+v", d.freeze()) + } + }) + return d +} +func loopError(err error) string { + switch { + case err == nil: + return "none" + case errors.Is(err, io.EOF): + return "eof" + case errors.Is(err, context.Canceled): + return "canceled" + case errors.Is(err, context.DeadlineExceeded): + return "deadline" + case errors.Is(err, net.ErrClosed): + return "closed" + default: + return "other" + } +} +func (d *loopDiagnostic) record(e loopEvent, ctx context.Context, err error) { + e.Err = loopError(err) + if ctx != nil { + e.Context, e.Cause = loopError(ctx.Err()), loopError(context.Cause(ctx)) + if deadline, ok := ctx.Deadline(); ok { + e.Deadline, e.Remaining = true, time.Until(deadline) + } + } + d.mu.Lock() + defer d.mu.Unlock() + if d.snapshot != nil { + return + } + d.seq++ + e.Seq = d.seq + if len(d.events) == 128 { + d.overflow++ + return + } + d.events = append(d.events, e) +} +func (d *loopDiagnostic) freeze() loopSnapshot { + d.mu.Lock() + defer d.mu.Unlock() + if d.snapshot == nil { + d.snapshot = &loopSnapshot{"INCOMPLETE: callbacks not joined; post-cutoff/future events may be omitted", append([]loopEvent(nil), d.events...), d.overflow, d.unsupported} + } + return *d.snapshot +} +func loopFrames() []string { + var pcs [8]uintptr + frames := runtime.CallersFrames(pcs[:runtime.Callers(3, pcs[:])]) + var out []string + for { + frame, more := frames.Next() + out = append(out, fmt.Sprintf("%s %s:%d", frame.Function, filepath.Base(frame.File), frame.Line)) + if !more { + return out + } + } +} + +func (d *loopDiagnostic) transport(base http.RoundTripper) *loopDiagnostic { + tr := base.(*http.Transport).Clone() + dial := tr.DialContext + tr.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) { + conn, err := dial(ctx, network, address) + tcp, ok := conn.(*net.TCPConn) + if err != nil { + return conn, err + } + if !ok { + d.mu.Lock() + d.unsupported = true + d.mu.Unlock() + return conn, err + } + id := int(d.conns.Add(1)) + return &loopConn{TCPConn: tcp, d: d, id: id}, err + } + d.http = tr + return d +} +func (d *loopDiagnostic) RoundTrip(req *http.Request) (*http.Response, error) { + page := int(d.pages.Add(1)) + d.record(loopEvent{Kind: "request", Page: page}, req.Context(), nil) + connID := 0 + trace := &httptrace.ClientTrace{ + GotConn: func(info httptrace.GotConnInfo) { + if conn, ok := info.Conn.(*loopConn); ok { + connID = conn.id + } + d.record(loopEvent{Kind: "got_conn", Page: page, Conn: connID}, req.Context(), nil) + }, + WroteRequest: func(info httptrace.WroteRequestInfo) { + d.record(loopEvent{Kind: "wrote_request", Page: page}, req.Context(), info.Err) + }, + } + resp, err := d.http.RoundTrip(req.Clone(httptrace.WithClientTrace(req.Context(), trace))) + if resp != nil { + d.record(loopEvent{Kind: "headers", Page: page, Conn: connID, N: resp.StatusCode}, req.Context(), err) + resp.Body = &loopBody{ReadCloser: resp.Body, d: d, ctx: req.Context(), page: page, conn: connID, readStarted: d.readStarted} + } + return resp, err +} + +type loopConn struct { + *net.TCPConn + d *loopDiagnostic + id int +} + +func (c *loopConn) Close() error { + c.d.record(loopEvent{Kind: "connection_close_begin", Conn: c.id, Frames: loopFrames()}, nil, nil) + err := c.TCPConn.Close() + c.d.record(loopEvent{Kind: "connection_close_end", Conn: c.id}, nil, err) + return err +} + +type loopBody struct { + io.ReadCloser + d *loopDiagnostic + ctx context.Context + page, conn int + readStarted chan struct{} + readOnce sync.Once +} + +func (b *loopBody) Read(p []byte) (int, error) { + if b.readStarted != nil { + b.readOnce.Do(func() { close(b.readStarted) }) + } + n, err := b.ReadCloser.Read(p) + if err != nil { + b.d.record(loopEvent{Kind: "body_read", Page: b.page, Conn: b.conn, N: n}, b.ctx, err) + } + return n, err +} +func (b *loopBody) Close() error { + b.d.record(loopEvent{Kind: "body_close_begin", Page: b.page, Conn: b.conn}, b.ctx, nil) + err := b.ReadCloser.Close() + b.d.record(loopEvent{Kind: "body_close_end", Page: b.page, Conn: b.conn}, b.ctx, err) + return err +} From 3426bc6f9992a811c6873e41d290fa8f552a135e Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 15 Oct 2026 15:42:56 +0200 Subject: [PATCH 2/2] test(aws): isolate loopback request body ownership Prevent the SDK's post-header body close from invalidating a pending HTTP request write in local fixtures. Replace the temporary tracer with a deterministic public SDK regression. Refs #414 Vendor runtime exposure is tracked separately in #416. --- providers/aws/ladder/baseline_sdk_test.go | 25 +- providers/aws/ladder/ce_loopback_body_test.go | 382 ++++++++++++++++++ .../aws/ladder/commitment_rates_sdk_test.go | 16 +- .../loopback_diagnostic_control_test.go | 184 --------- .../aws/ladder/loopback_diagnostic_test.go | 191 --------- 5 files changed, 402 insertions(+), 396 deletions(-) create mode 100644 providers/aws/ladder/ce_loopback_body_test.go delete mode 100644 providers/aws/ladder/loopback_diagnostic_control_test.go delete mode 100644 providers/aws/ladder/loopback_diagnostic_test.go diff --git a/providers/aws/ladder/baseline_sdk_test.go b/providers/aws/ladder/baseline_sdk_test.go index 3ac518d..9b87168 100644 --- a/providers/aws/ladder/baseline_sdk_test.go +++ b/providers/aws/ladder/baseline_sdk_test.go @@ -25,18 +25,11 @@ import ( // Constructed CE responses exercise the public factory without cloud access. func TestScopedBaselineParentCompatible(t *testing.T) { - runLoopbackFixture(t, scopedBaselineFixture) -} - -func scopedBaselineFixture(t *testing.T) { for _, mode := range []string{"account", "reverse", "zero", "lag2", "lag3", "duplicate", "page-duplicate", "hole", "leading", "outside", "future", "bad-start", "nil-start", "nil-period", "nil-end", "bad-end", "short-period", "long-period", "backward-period", "empty", "all-zero", "missing-total", "missing-metric", "missing-amount", "bad-amount", "NaN", "Inf", "negative", "stale", "few", "provider-error"} { - if !t.Run(mode, func(t *testing.T) { - d := newLoopDiagnostic(t) + t.Run(mode, func(t *testing.T) { var calls atomic.Int32 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - page := calls.Add(1) - d.record(loopEvent{Kind: "handler_begin", Page: int(page)}, r.Context(), nil) - defer d.record(loopEvent{Kind: "handler_end", Page: int(page)}, r.Context(), nil) + calls.Add(1) assert.Equal(t, "AWSInsightsIndexService.GetCostAndUsage", r.Header.Get("X-Amz-Target")) var in costexplorer.GetCostAndUsageInput if !assert.NoError(t, json.NewDecoder(r.Body).Decode(&in)) { @@ -89,18 +82,12 @@ func scopedBaselineFixture(t *testing.T) { w.Header().Set("Content-Type", "application/x-amz-json-1.1") assert.NoError(t, json.NewEncoder(w).Encode(map[string]any{"ResultsByTime": rows, "NextPageToken": token})) })) - defer func() { - d.record(loopEvent{Kind: "server_close_begin"}, nil, nil) - srv.Close() - d.record(loopEvent{Kind: "server_close_end"}, nil, nil) - }() - transport := d.transport(srv.Client().Transport) - defer transport.http.CloseIdleConnections() + defer srv.Close() endpoint, err := url.Parse(srv.URL) require.NoError(t, err) a, err := awsladder.NewFromConfig(aws.Config{Region: "eu-west-1", RetryMaxAttempts: 1, Credentials: credentials.NewStaticCredentialsProvider("synthetic", "synthetic", ""), - HTTPClient: &http.Client{Transport: ceLoopback{endpoint: endpoint, transport: transport}}, + HTTPClient: &http.Client{Transport: ceLoopback{endpoint: endpoint, transport: srv.Client().Transport}}, }, awsladder.Config{Region: "eu-west-1", AccountID: "fixture-account"}) require.NoError(t, err) scope := ladder.Scope{Provider: common.ProviderAWS, AccountID: "wrong"} @@ -132,9 +119,7 @@ func scopedBaselineFixture(t *testing.T) { if mode == "account" || mode == "reverse" { assert.Equal(t, int32(2), calls.Load()) } - }) { - return - } + }) } } diff --git a/providers/aws/ladder/ce_loopback_body_test.go b/providers/aws/ladder/ce_loopback_body_test.go new file mode 100644 index 0000000..e997e45 --- /dev/null +++ b/providers/aws/ladder/ce_loopback_body_test.go @@ -0,0 +1,382 @@ +package ladder_test + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "net" + "net/http" + "net/http/httptest" + "net/http/httptrace" + "net/url" + "runtime" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/costexplorer" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" + awsladder "github.com/LeanerCloud/cloud-commitments-go/providers/aws/ladder" +) + +type ceRoundTripFunc func(*http.Request) (*http.Response, error) + +func (f ceRoundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +type ceSource struct { + io.Reader + readErr, closeErr error + reads, closes int + closed bool +} + +func (b *ceSource) Read(p []byte) (int, error) { + b.reads++ + if b.closed { + return 0, net.ErrClosed + } + if b.readErr != nil { + n, _ := b.Reader.Read(p) + return n, b.readErr + } + return b.Reader.Read(p) +} +func (b *ceSource) Close() error { + b.closes++ + b.closed = true + return b.closeErr +} + +func TestCELoopbackBodyOwnership(t *testing.T) { + readErr, closeErr, downstreamErr := errors.New("read"), errors.New("close"), errors.New("transport") + for _, mode := range []string{"binary", "get-body", "nil", "no-body", "empty", "unknown", "zero", "read", "close", "both", "scheme", "host", "guard-close", "transport"} { + t.Run(mode, func(t *testing.T) { + payload := []byte{'a', 0, 'b'} + source := &ceSource{Reader: bytes.NewReader(payload)} + if mode == "read" || mode == "both" { + source.readErr = readErr + } + if mode == "close" || mode == "both" || mode == "guard-close" { + source.closeErr = closeErr + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://ce.us-east-1.amazonaws.com/path?q=value", source) + require.NoError(t, err) + req.ContentLength = int64(len(payload)) + req.Header.Set("X-Fixture", "preserved") + getterCalls := 0 + if mode == "get-body" { + req.GetBody = func() (io.ReadCloser, error) { + getterCalls++ + return io.NopCloser(bytes.NewReader(payload)), nil + } + } + if mode == "nil" { + req.Body, payload = nil, nil + } + if mode == "no-body" { + req.Body, payload = http.NoBody, nil + } + if mode == "empty" { + source.Reader, payload, req.ContentLength = bytes.NewReader(nil), []byte{}, 0 + } + if mode == "unknown" { + req.ContentLength = -1 + } + if mode == "zero" { + req.ContentLength = 0 + } + if mode == "scheme" { + req.URL.Scheme = "http" + } + if mode == "host" || mode == "guard-close" { + req.URL.Host = "rejected.invalid" + } + originalBody, originalURL, length := req.Body, req.URL.String(), req.ContentLength + endpoint, err := url.Parse("http://fixture.invalid") + require.NoError(t, err) + response, calls := &http.Response{StatusCode: 200, Body: http.NoBody}, 0 + transport := ceRoundTripFunc(func(local *http.Request) (*http.Response, error) { + calls++ + assert.Same(t, ctx, local.Context()) + assert.Equal(t, length, local.ContentLength) + assert.Equal(t, req.Header, local.Header) + assert.Equal(t, req.Method, local.Method) + assert.Equal(t, "/path?q=value", local.URL.RequestURI()) + assert.Equal(t, "http://fixture.invalid/path?q=value", local.URL.String()) + assert.Equal(t, endpoint.Host, local.Host) + if mode == "get-body" { + replay, getErr := local.GetBody() + assert.NoError(t, getErr) + got, read := io.ReadAll(replay) + assert.NoError(t, errors.Join(read, replay.Close())) + assert.Equal(t, payload, got) + assert.Equal(t, 1, getterCalls) + } else { + assert.Nil(t, local.GetBody) + } + if originalBody == nil || originalBody == http.NoBody { + assert.Equal(t, originalBody, local.Body) + } else { + assert.True(t, source.closed) + got, read := io.ReadAll(local.Body) + assert.NoError(t, read) + assert.Equal(t, payload, got) + assert.NoError(t, local.Body.Close()) + } + if mode == "transport" { + return response, downstreamErr + } + return response, nil + }) + got, err := (ceLoopback{endpoint: endpoint, transport: transport}).RoundTrip(req) + if got != nil { + assert.NoError(t, got.Body.Close()) + } + badBody := source.readErr != nil || source.closeErr != nil + guard := mode == "scheme" || mode == "host" || mode == "guard-close" + if badBody || guard { + require.Error(t, err) + assert.Nil(t, got) + assert.Zero(t, calls) + } else { + assert.Same(t, response, got) + assert.Equal(t, 1, calls) + if mode == "transport" { + assert.ErrorIs(t, err, downstreamErr) + } else { + assert.NoError(t, err) + } + } + if source.readErr != nil { + assert.ErrorIs(t, err, readErr) + } + if source.closeErr != nil { + assert.ErrorIs(t, err, closeErr) + } + if guard { + assert.Zero(t, source.reads) + } + if originalBody != nil && originalBody != http.NoBody { + assert.Equal(t, 1, source.closes) + assert.Same(t, originalBody, req.Body) + } else { + assert.Equal(t, originalBody, req.Body) + } + assert.Equal(t, originalURL, req.URL.String()) + assert.Equal(t, length, req.ContentLength) + }) + } +} + +func ceAwait(done, abort <-chan struct{}) error { + select { + case <-done: + return nil + case <-abort: + return errors.New("schedule aborted") + case <-time.After(5 * time.Second): + return errors.New("schedule did not complete") + } +} + +type ceSchedule struct { + abort, entered, sdkClosed, wrote, connClosed chan struct{} + enterOnce, sdkOnce, writeOnce, connOnce sync.Once + returned atomic.Bool + copyErr, writeErr error + status int +} +type ceSDKBody struct { + io.ReadCloser + s *ceSchedule +} + +func (b *ceSDKBody) WriteTo(w io.Writer) (int64, error) { return b.ReadCloser.(io.WriterTo).WriteTo(w) } +func (b *ceSDKBody) Close() error { + err := b.ReadCloser.Close() + if b.s.returned.Load() { + var pcs [16]uintptr + frames := runtime.CallersFrames(pcs[:runtime.Callers(2, pcs[:])]) + for { + frame, more := frames.Next() + if frame.Function == "github.com/aws/smithy-go/transport/http.ClientHandler.Handle" { + b.s.sdkOnce.Do(func() { close(b.s.sdkClosed) }) + } + if !more { + break + } + } + } + return err +} + +type ceTerminalBody struct { + io.ReadCloser + s *ceSchedule + read, length int64 +} + +func (b *ceTerminalBody) Read(p []byte) (int, error) { + n, err := b.ReadCloser.Read(p) + b.read += int64(n) + return n, err +} +func (b *ceTerminalBody) WriteTo(w io.Writer) (int64, error) { + if b.read != b.length { + return 0, errors.New("terminal copy before ContentLength consumed") + } + b.s.enterOnce.Do(func() { close(b.s.entered) }) + if err := ceAwait(b.s.sdkClosed, b.s.abort); err != nil { + return 0, err + } + n, err := b.ReadCloser.(io.WriterTo).WriteTo(w) + b.s.copyErr = err + return n, err +} + +type ceScheduledConn struct { + *net.TCPConn + s *ceSchedule +} + +func (c *ceScheduledConn) Close() error { + err := c.TCPConn.Close() + c.s.connOnce.Do(func() { close(c.s.connClosed) }) + return err +} + +type ceScheduledResponse struct { + io.ReadCloser + s *ceSchedule +} + +func (r *ceScheduledResponse) Read(p []byte) (int, error) { + if err := ceAwait(r.s.wrote, r.s.abort); err != nil { + return 0, err + } + if r.s.writeErr != nil { + if err := ceAwait(r.s.connClosed, r.s.abort); err != nil { + return 0, err + } + } + return r.ReadCloser.Read(p) +} + +type ceScheduledClient struct { + client *http.Client + s *ceSchedule +} + +func (c ceScheduledClient) Do(req *http.Request) (*http.Response, error) { + req.Body = &ceSDKBody{ReadCloser: req.Body, s: c.s} + resp, err := c.client.Do(req) + c.s.returned.Store(true) + return resp, err +} + +func TestCELoopbackPublicBaselineAfterBodyClose(t *testing.T) { + s := &ceSchedule{abort: make(chan struct{}), entered: make(chan struct{}), sdkClosed: make(chan struct{}), wrote: make(chan struct{}), connClosed: make(chan struct{})} + ctx, cancel := context.WithCancel(context.Background()) + done, handlerDone := make(chan struct{}), make(chan struct{}) + var handlerErr, operationErr error + var baseline ladder.UsageBaseline + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + defer close(handlerDone) + var input costexplorer.GetCostAndUsageInput + handlerErr = json.NewDecoder(req.Body).Decode(&input) + if handlerErr != nil { + return + } + if req.Header.Get("X-Amz-Target") != "AWSInsightsIndexService.GetCostAndUsage" || input.TimePeriod == nil { + handlerErr = errors.New("unexpected CE request") + return + } + start, err := time.Parse(time.DateOnly, aws.ToString(input.TimePeriod.Start)) + if err != nil { + handlerErr = err + return + } + w.Header().Set("Content-Type", "application/x-amz-json-1.1") + handlerErr = json.NewEncoder(w).Encode(map[string]any{"ResultsByTime": baselineRows(start, "account"), "padding": strings.Repeat("x", 64<<10)}) + })) + tr := srv.Client().Transport.(*http.Transport).Clone() + dial := tr.DialContext + tr.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) { + conn, err := dial(ctx, network, address) + if err != nil { + return conn, err + } + return &ceScheduledConn{TCPConn: conn.(*net.TCPConn), s: s}, nil + } + cleanup := sync.OnceFunc(func() { + cancel() + close(s.abort) + opJoin, handlerJoin, writeJoin := ceAwait(done, nil), ceAwait(handlerDone, nil), ceAwait(s.wrote, nil) + if handlerJoin == nil { + srv.Close() + } else { + srv.CloseClientConnections() + assert.NoError(t, srv.Listener.Close()) + } + tr.CloseIdleConnections() + assert.NoError(t, errors.Join(opJoin, handlerJoin, writeJoin)) + if handlerJoin == nil { + assert.NoError(t, handlerErr) + } + }) + defer cleanup() + endpoint, err := url.Parse(srv.URL) + require.NoError(t, err) + inner := ceRoundTripFunc(func(req *http.Request) (*http.Response, error) { + if req.ContentLength <= 0 || req.Method != http.MethodPost || req.Header.Get("X-Amz-Target") != "AWSInsightsIndexService.GetCostAndUsage" { + return nil, errors.New("unexpected scheduled CE request") + } + req.Body = &ceTerminalBody{ReadCloser: req.Body, s: s, length: req.ContentLength} + trace := &httptrace.ClientTrace{WroteRequest: func(info httptrace.WroteRequestInfo) { + s.writeOnce.Do(func() { s.writeErr = info.Err; close(s.wrote) }) + }} + resp, tripErr := tr.RoundTrip(req.Clone(httptrace.WithClientTrace(req.Context(), trace))) + if tripErr != nil { + return resp, tripErr + } + if waitErr := ceAwait(s.entered, s.abort); waitErr != nil { + resp.Body.Close() + return nil, waitErr + } + s.status = resp.StatusCode + resp.Body = &ceScheduledResponse{ReadCloser: resp.Body, s: s} + return resp, nil + }) + client := ceScheduledClient{client: &http.Client{Transport: ceLoopback{endpoint: endpoint, transport: inner}}, s: s} + a, err := awsladder.NewFromConfig(aws.Config{Region: "eu-west-1", RetryMaxAttempts: 1, Credentials: credentials.NewStaticCredentialsProvider("synthetic", "synthetic", ""), HTTPClient: client}, awsladder.Config{Region: "eu-west-1", AccountID: "fixture-account"}) + require.NoError(t, err) + go func() { + defer close(done) + baseline, operationErr = a.GetUsageBaseline(ctx, ladder.Scope{Provider: common.ProviderAWS, AccountID: "fixture-account"}, 30, 5) + }() + joined, callerErr := ceAwait(done, nil), ctx.Err() + cleanup() + require.NoError(t, joined) + t.Logf("headers=%d terminalEOF=%t wroteEOF=%t", s.status, errors.Is(s.copyErr, io.EOF), errors.Is(s.writeErr, io.EOF)) + require.NoError(t, operationErr) + require.NoError(t, callerErr) + require.NoError(t, ceAwait(s.sdkClosed, nil)) + require.NoError(t, ceAwait(s.entered, nil)) + require.NoError(t, s.copyErr) + require.NoError(t, s.writeErr) + require.Equal(t, 200, s.status) + require.NotNil(t, baseline.LowWaterUSDPerHour) + require.InDelta(t, 2, *baseline.LowWaterUSDPerHour, 1e-9) +} diff --git a/providers/aws/ladder/commitment_rates_sdk_test.go b/providers/aws/ladder/commitment_rates_sdk_test.go index 405dbd8..893fb8c 100644 --- a/providers/aws/ladder/commitment_rates_sdk_test.go +++ b/providers/aws/ladder/commitment_rates_sdk_test.go @@ -1,9 +1,12 @@ package ladder_test import ( + "bytes" "context" "encoding/json" + "errors" "fmt" + "io" "math/big" "net/http" "net/http/httptest" @@ -29,9 +32,20 @@ type ceLoopback struct { func (c ceLoopback) RoundTrip(req *http.Request) (*http.Response, error) { if req.URL.Scheme != "https" || req.URL.Host != "ce.us-east-1.amazonaws.com" { - return nil, fmt.Errorf("fixture rejects host %s", req.URL.Host) + err := fmt.Errorf("fixture rejects host %s", req.URL.Host) + if req.Body != nil { + err = errors.Join(err, req.Body.Close()) + } + return nil, err } local := req.Clone(req.Context()) + if req.Body != nil && req.Body != http.NoBody { + payload, readErr := io.ReadAll(req.Body) + if err := errors.Join(readErr, req.Body.Close()); err != nil { + return nil, fmt.Errorf("fixture request body: %w", err) + } + local.Body = io.NopCloser(bytes.NewReader(payload)) + } local.URL.Scheme, local.URL.Host = c.endpoint.Scheme, c.endpoint.Host local.Host = c.endpoint.Host return c.transport.RoundTrip(local) diff --git a/providers/aws/ladder/loopback_diagnostic_control_test.go b/providers/aws/ladder/loopback_diagnostic_control_test.go deleted file mode 100644 index 126aa86..0000000 --- a/providers/aws/ladder/loopback_diagnostic_control_test.go +++ /dev/null @@ -1,184 +0,0 @@ -package ladder_test - -import ( - "context" - "errors" - "fmt" - "io" - "net" - "net/http" - "net/http/httptest" - "net/http/httptrace" - "net/url" - "runtime" - "strings" - "sync" - "syscall" - "testing" - "time" - - "github.com/aws/aws-sdk-go-v2/aws" - "github.com/aws/aws-sdk-go-v2/credentials" - "github.com/aws/aws-sdk-go-v2/service/costexplorer" - "github.com/aws/aws-sdk-go-v2/service/costexplorer/types" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" -) - -func runLoopbackFixture(t *testing.T, fixture func(*testing.T)) { - iterations := 1 - if runtime.GOOS == "linux" { - iterations = 33 - } - for i := range iterations { - if !t.Run(fmt.Sprint(i), fixture) { - return - } - } -} -func waitLoopControl(t *testing.T, done <-chan struct{}) bool { - t.Helper() - select { - case <-done: - return true - case <-time.After(5 * time.Second): - t.Error("control did not complete") - return false - } -} - -func TestLoopbackDiagnosticLifecycle(t *testing.T) { - for _, cancelRequest := range []bool{true, false} { - t.Run(fmt.Sprint(cancelRequest), func(t *testing.T) { - d := newLoopDiagnostic(t) - deadline, stop := context.WithTimeout(context.Background(), 5*time.Second) - defer stop() - ctx, cancel := context.WithCancelCause(deadline) - release, done, handlerDone, wrote := make(chan struct{}), make(chan struct{}), make(chan struct{}), make(chan struct{}) - var operationErr, writeErr, flushErr error - srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - defer close(handlerDone) - d.record(loopEvent{Kind: "handler_begin"}, r.Context(), nil) - defer d.record(loopEvent{Kind: "handler_end"}, r.Context(), nil) - w.Header().Set("Content-Type", "application/x-amz-json-1.1") - _, writeErr = w.Write([]byte("{\"ResultsByTime\":[")) - flushErr = http.NewResponseController(w).Flush() - <-release - })) - tr := d.transport(srv.Client().Transport) - tr.readStarted = make(chan struct{}) - cleanup := sync.OnceFunc(func() { - defer tr.http.CloseIdleConnections() - cancel(nil) - close(release) - waitLoopControl(t, done) - handlerJoined := waitLoopControl(t, handlerDone) - waitLoopControl(t, wrote) - d.record(loopEvent{Kind: "server_close_begin"}, nil, nil) - if !handlerJoined { - srv.CloseClientConnections() - d.record(loopEvent{Kind: "server_close_unjoined"}, nil, srv.Listener.Close()) - return - } - srv.Close() - d.record(loopEvent{Kind: "server_close_end"}, nil, nil) - assert.NoError(t, errors.Join(writeErr, flushErr)) - }) - defer cleanup() - endpoint, err := url.Parse(srv.URL) - require.NoError(t, err) - gotConn := make(chan *loopConn, 1) - ctx = httptrace.WithClientTrace(ctx, &httptrace.ClientTrace{ - GotConn: func(info httptrace.GotConnInfo) { gotConn <- info.Conn.(*loopConn) }, - WroteRequest: func(httptrace.WroteRequestInfo) { close(wrote) }, - }) - client := costexplorer.NewFromConfig(aws.Config{Region: "us-east-1", RetryMaxAttempts: 1, Credentials: credentials.NewStaticCredentialsProvider("synthetic", "synthetic", ""), HTTPClient: &http.Client{Transport: ceLoopback{endpoint: endpoint, transport: tr}}}) - go func() { - defer close(done) - _, operationErr = client.GetCostAndUsage(ctx, &costexplorer.GetCostAndUsageInput{Granularity: types.GranularityDaily, Metrics: []string{"UnblendedCost"}, TimePeriod: &types.DateInterval{Start: aws.String("2026-10-01"), End: aws.String("2026-10-02")}}) - }() - require.True(t, waitLoopControl(t, tr.readStarted)) - var conn *loopConn - select { - case conn = <-gotConn: - case <-deadline.Done(): - t.Fatal("connection not observed") - } - var _ io.ReaderFrom = conn - var _ io.WriterTo = conn - var _ syscall.Conn = conn - var _ interface { - CloseRead() error - CloseWrite() error - } = conn - if cancelRequest { - cancel(errors.New("control cancellation cause")) - } else { - require.NoError(t, conn.Close()) - } - require.True(t, waitLoopControl(t, done)) - require.Error(t, operationErr) - if !cancelRequest { - require.NoError(t, ctx.Err()) - } - cleanup() - kinds := map[string]loopEvent{} - explicitClose := false - for _, e := range d.freeze().Events { - kinds[e.Kind] = e - if e.Kind == "connection_close_begin" { - require.Equal(t, [2]string{}, [2]string{e.Context, e.Cause}) - require.LessOrEqual(t, len(e.Frames), 8) - for _, frame := range e.Frames { - explicitClose = explicitClose || strings.Contains(frame, "TestLoopbackDiagnosticLifecycle") - require.Regexp(t, `^\S+ [^/ ]+:\d+$`, frame) - } - } - } - require.Equal(t, 200, kinds["headers"].N) - require.Contains(t, []string{"canceled", "closed", "other"}, kinds["body_read"].Err) - require.Positive(t, kinds["got_conn"].Conn) - require.Equal(t, kinds["got_conn"].Conn, kinds["body_read"].Conn) - require.Equal(t, "none", kinds["wrote_request"].Err) - require.IsIncreasing(t, []int{kinds["headers"].Seq, kinds["body_read"].Seq, kinds["body_close_end"].Seq}) - require.IsIncreasing(t, []int{0, kinds["handler_end"].Seq, kinds["server_close_end"].Seq}) - require.True(t, kinds["body_read"].Deadline) - require.Equal(t, map[bool]string{true: "canceled", false: "none"}[cancelRequest], kinds["body_read"].Context) - require.Equal(t, map[bool]string{true: "other", false: "none"}[cancelRequest], kinds["body_read"].Cause) - require.True(t, cancelRequest || explicitClose) - }) - } -} - -func TestLoopbackDiagnosticCutoff(t *testing.T) { - d := newLoopDiagnostic(t) - start, recorded, cut := make(chan struct{}), make(chan struct{}), make(chan struct{}) - var raced loopSnapshot - joinWorkers := func() { - require.NotContains(t, []bool{waitLoopControl(t, recorded), waitLoopControl(t, cut)}, false) - } - defer joinWorkers() - go func() { defer close(recorded); <-start; d.record(loopEvent{Kind: "racing"}, nil, nil) }() - go func() { defer close(cut); <-start; raced = d.freeze() }() - close(start) - joinWorkers() - require.True(t, len(raced.Events) <= 1 && strings.Contains(raced.Incomplete, "INCOMPLETE"), "%+v", raced) - bounded := newLoopDiagnostic(t) - for range 130 { - bounded.record(loopEvent{Kind: "error"}, nil, errors.New("sensitive-token/path")) - } - local, peer := net.Pipe() - defer local.Close() - defer peer.Close() - tr := bounded.transport(&http.Transport{DialContext: func(context.Context, string, string) (net.Conn, error) { return local, nil }}) - conn, err := tr.http.DialContext(context.Background(), "tcp", "unused") - require.NoError(t, err) - require.Same(t, local, conn) - snapshot := bounded.freeze() - require.Len(t, snapshot.Events, 128) - require.Equal(t, 2, snapshot.Overflow) - require.True(t, snapshot.Unsupported) - require.NotContains(t, fmt.Sprint(snapshot), "sensitive-token/path") - bounded.record(loopEvent{Kind: "after_overflow_cutoff"}, nil, nil) - require.Equal(t, snapshot, bounded.freeze()) -} diff --git a/providers/aws/ladder/loopback_diagnostic_test.go b/providers/aws/ladder/loopback_diagnostic_test.go deleted file mode 100644 index d6c9b4e..0000000 --- a/providers/aws/ladder/loopback_diagnostic_test.go +++ /dev/null @@ -1,191 +0,0 @@ -package ladder_test - -import ( - "context" - "errors" - "fmt" - "io" - "net" - "net/http" - "net/http/httptrace" - "path/filepath" - "runtime" - "sync" - "sync/atomic" - "testing" - "time" -) - -type loopEvent struct { - Seq, Page, Conn, N int - Kind, Err, Context, Cause string - Deadline bool - Remaining time.Duration - Frames []string -} -type loopSnapshot struct { - Incomplete string - Events []loopEvent - Overflow int - Unsupported bool -} -type loopDiagnostic struct { - mu sync.Mutex - events []loopEvent - seq, overflow int - unsupported bool - snapshot *loopSnapshot - pages, conns atomic.Int32 - http *http.Transport - readStarted chan struct{} -} - -func newLoopDiagnostic(t *testing.T) *loopDiagnostic { - d := &loopDiagnostic{} - t.Cleanup(func() { - if t.Failed() { - t.Logf("loopback diagnostic: %+v", d.freeze()) - } - }) - return d -} -func loopError(err error) string { - switch { - case err == nil: - return "none" - case errors.Is(err, io.EOF): - return "eof" - case errors.Is(err, context.Canceled): - return "canceled" - case errors.Is(err, context.DeadlineExceeded): - return "deadline" - case errors.Is(err, net.ErrClosed): - return "closed" - default: - return "other" - } -} -func (d *loopDiagnostic) record(e loopEvent, ctx context.Context, err error) { - e.Err = loopError(err) - if ctx != nil { - e.Context, e.Cause = loopError(ctx.Err()), loopError(context.Cause(ctx)) - if deadline, ok := ctx.Deadline(); ok { - e.Deadline, e.Remaining = true, time.Until(deadline) - } - } - d.mu.Lock() - defer d.mu.Unlock() - if d.snapshot != nil { - return - } - d.seq++ - e.Seq = d.seq - if len(d.events) == 128 { - d.overflow++ - return - } - d.events = append(d.events, e) -} -func (d *loopDiagnostic) freeze() loopSnapshot { - d.mu.Lock() - defer d.mu.Unlock() - if d.snapshot == nil { - d.snapshot = &loopSnapshot{"INCOMPLETE: callbacks not joined; post-cutoff/future events may be omitted", append([]loopEvent(nil), d.events...), d.overflow, d.unsupported} - } - return *d.snapshot -} -func loopFrames() []string { - var pcs [8]uintptr - frames := runtime.CallersFrames(pcs[:runtime.Callers(3, pcs[:])]) - var out []string - for { - frame, more := frames.Next() - out = append(out, fmt.Sprintf("%s %s:%d", frame.Function, filepath.Base(frame.File), frame.Line)) - if !more { - return out - } - } -} - -func (d *loopDiagnostic) transport(base http.RoundTripper) *loopDiagnostic { - tr := base.(*http.Transport).Clone() - dial := tr.DialContext - tr.DialContext = func(ctx context.Context, network, address string) (net.Conn, error) { - conn, err := dial(ctx, network, address) - tcp, ok := conn.(*net.TCPConn) - if err != nil { - return conn, err - } - if !ok { - d.mu.Lock() - d.unsupported = true - d.mu.Unlock() - return conn, err - } - id := int(d.conns.Add(1)) - return &loopConn{TCPConn: tcp, d: d, id: id}, err - } - d.http = tr - return d -} -func (d *loopDiagnostic) RoundTrip(req *http.Request) (*http.Response, error) { - page := int(d.pages.Add(1)) - d.record(loopEvent{Kind: "request", Page: page}, req.Context(), nil) - connID := 0 - trace := &httptrace.ClientTrace{ - GotConn: func(info httptrace.GotConnInfo) { - if conn, ok := info.Conn.(*loopConn); ok { - connID = conn.id - } - d.record(loopEvent{Kind: "got_conn", Page: page, Conn: connID}, req.Context(), nil) - }, - WroteRequest: func(info httptrace.WroteRequestInfo) { - d.record(loopEvent{Kind: "wrote_request", Page: page}, req.Context(), info.Err) - }, - } - resp, err := d.http.RoundTrip(req.Clone(httptrace.WithClientTrace(req.Context(), trace))) - if resp != nil { - d.record(loopEvent{Kind: "headers", Page: page, Conn: connID, N: resp.StatusCode}, req.Context(), err) - resp.Body = &loopBody{ReadCloser: resp.Body, d: d, ctx: req.Context(), page: page, conn: connID, readStarted: d.readStarted} - } - return resp, err -} - -type loopConn struct { - *net.TCPConn - d *loopDiagnostic - id int -} - -func (c *loopConn) Close() error { - c.d.record(loopEvent{Kind: "connection_close_begin", Conn: c.id, Frames: loopFrames()}, nil, nil) - err := c.TCPConn.Close() - c.d.record(loopEvent{Kind: "connection_close_end", Conn: c.id}, nil, err) - return err -} - -type loopBody struct { - io.ReadCloser - d *loopDiagnostic - ctx context.Context - page, conn int - readStarted chan struct{} - readOnce sync.Once -} - -func (b *loopBody) Read(p []byte) (int, error) { - if b.readStarted != nil { - b.readOnce.Do(func() { close(b.readStarted) }) - } - n, err := b.ReadCloser.Read(p) - if err != nil { - b.d.record(loopEvent{Kind: "body_read", Page: b.page, Conn: b.conn, N: n}, b.ctx, err) - } - return n, err -} -func (b *loopBody) Close() error { - b.d.record(loopEvent{Kind: "body_close_begin", Page: b.page, Conn: b.conn}, b.ctx, nil) - err := b.ReadCloser.Close() - b.d.record(loopEvent{Kind: "body_close_end", Page: b.page, Conn: b.conn}, b.ctx, err) - return err -}