From d460159ba9d215c83b4d5c344b8388618c60face Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Fri, 2 Oct 2026 16:19:27 +0000 Subject: [PATCH 1/2] test: add routed truncated response regression Signed-off-by: Perfloop Agent --- handlers/forwarding_response_test.go | 383 +++++++++++++++++++++++++++ 1 file changed, 383 insertions(+) create mode 100644 handlers/forwarding_response_test.go diff --git a/handlers/forwarding_response_test.go b/handlers/forwarding_response_test.go new file mode 100644 index 00000000..c45ceec6 --- /dev/null +++ b/handlers/forwarding_response_test.go @@ -0,0 +1,383 @@ +package handlers + +import ( + "bufio" + "bytes" + "context" + "crypto/tls" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/tigrisdata/tag/auth" + "github.com/tigrisdata/tag/config" + "github.com/tigrisdata/tag/proxy" +) + +const ( + forwardingTestAccessKey = "AKIAIOSFODNN7EXAMPLE" + forwardingTestSecretKey = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY" + forwardingTestRegion = "us-east-1" +) + +func newSigningListBucketsRouter(t *testing.T, upstreamURL string) http.Handler { + t.Helper() + + credentials := auth.NewCredentialStore() + credentials.AddCredential(forwardingTestAccessKey, forwardingTestSecretKey) + forwarder := proxy.NewForwarder(credentials, upstreamURL, forwardingTestRegion, 1, nil, nil) + cfg := config.NewDefault() + cfg.Mode = config.ModeSigning + service := proxy.NewService(forwarder, nil, cfg) + return NewServer(service, "127.0.0.1", 0, false, 0).Router() +} + +func newSigningListBucketsGateway(t *testing.T, upstreamURL string) *httptest.Server { + t.Helper() + gateway := httptest.NewServer(newSigningListBucketsRouter(t, upstreamURL)) + t.Cleanup(gateway.Close) + return gateway +} + +func newSigningListBucketsTLSGateway(t *testing.T, upstreamURL string) *httptest.Server { + t.Helper() + gateway := httptest.NewTLSServer(newSigningListBucketsRouter(t, upstreamURL)) + t.Cleanup(gateway.Close) + return gateway +} + +func newSigningListBucketsHTTP2Gateway(t *testing.T, upstreamURL string) (*httptest.Server, <-chan string) { + t.Helper() + protoSeen := make(chan string, 1) + router := newSigningListBucketsRouter(t, upstreamURL) + gateway := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + protoSeen <- r.Proto + router.ServeHTTP(w, r) + })) + gateway.EnableHTTP2 = true + gateway.StartTLS() + t.Cleanup(gateway.Close) + return gateway, protoSeen +} + +func signedListBucketsRequest(t *testing.T, gatewayURL string) *http.Request { + t.Helper() + + request, err := auth.NewRequestSigner(gatewayURL, forwardingTestRegion).SignRequest( + context.Background(), http.MethodGet, "/", nil, "", forwardingTestAccessKey, forwardingTestSecretKey, nil, + ) + if err != nil { + t.Fatalf("sign ListBuckets request: %v", err) + } + return request +} + +type truncatedListBucketsUpstream struct { + server *httptest.Server + calls atomic.Int32 + result chan error + prefix []byte +} + +func newTruncatedListBucketsUpstream(t *testing.T, prefix []byte) *truncatedListBucketsUpstream { + t.Helper() + upstream := &truncatedListBucketsUpstream{ + result: make(chan error, 1), + prefix: prefix, + } + upstream.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + upstream.calls.Add(1) + if r.Method != http.MethodGet || r.URL.Path != "/" || r.Header.Get("Authorization") == "" { + upstream.result <- fmt.Errorf("unexpected forwarded request: %s %s", r.Method, r.URL.Path) + return + } + hijacker, ok := w.(http.Hijacker) + if !ok { + upstream.result <- errors.New("upstream response writer does not support hijacking") + return + } + conn, rw, err := hijacker.Hijack() + if err != nil { + upstream.result <- fmt.Errorf("hijack upstream connection: %w", err) + return + } + _, writeErr := fmt.Fprintf(rw, "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n%x\r\n", len(prefix)) + if writeErr == nil { + _, writeErr = rw.Write(prefix) + } + if writeErr == nil { + _, writeErr = io.WriteString(rw, "\r\n") + } + if writeErr == nil { + writeErr = rw.Flush() + } + _ = conn.Close() + upstream.result <- writeErr + })) + t.Cleanup(upstream.server.Close) + return upstream +} + +func (u *truncatedListBucketsUpstream) check(t *testing.T) { + t.Helper() + if got := u.calls.Load(); got != 1 { + t.Fatalf("upstream calls = %d, want 1", got) + } + select { + case err := <-u.result: + if err != nil { + t.Fatalf("write truncated upstream response: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("upstream fixture did not finish writing the truncated response") + } +} + +func sendSignedHTTP10ListBuckets(t *testing.T, gatewayURL string, request *http.Request) (*http.Response, net.Conn, error) { + t.Helper() + address := strings.TrimPrefix(strings.TrimPrefix(gatewayURL, "http://"), "https://") + conn, err := net.DialTimeout("tcp", address, 5*time.Second) + if err != nil { + return nil, nil, err + } + if strings.HasPrefix(gatewayURL, "https://") { + tlsConn := tls.Client(conn, &tls.Config{ + InsecureSkipVerify: true, // httptest server uses a self-signed certificate. + NextProtos: []string{"http/1.1"}, + }) + if err := tlsConn.Handshake(); err != nil { + _ = conn.Close() + return nil, nil, err + } + conn = tlsConn + } + _ = conn.SetDeadline(time.Now().Add(10 * time.Second)) + + headers := request.Header.Clone() + headers.Set("Host", request.URL.Host) + headers.Set("Connection", "close") + if _, err := fmt.Fprintf(conn, "%s %s HTTP/1.0\r\n", request.Method, request.URL.RequestURI()); err != nil { + _ = conn.Close() + return nil, nil, err + } + if err := headers.Write(conn); err != nil { + _ = conn.Close() + return nil, nil, err + } + if _, err := io.WriteString(conn, "\r\n"); err != nil { + _ = conn.Close() + return nil, nil, err + } + response, err := http.ReadResponse(bufio.NewReader(conn), request) + return response, conn, err +} + +func isClientAbortTimeout(err error) bool { + if errors.Is(err, context.DeadlineExceeded) { + return true + } + var networkErr net.Error + return errors.As(err, &networkErr) && networkErr.Timeout() +} + +func assertTruncatedListBucketsResponse(t *testing.T, response *http.Response, responseErr error, prefix []byte, upstream *truncatedListBucketsUpstream) { + t.Helper() + if responseErr != nil { + upstream.check(t) + if isClientAbortTimeout(responseErr) { + t.Fatalf("response abort timed out: %v", responseErr) + } + return + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + t.Fatalf("ListBuckets status = %d, want upstream status %d", response.StatusCode, http.StatusOK) + } + upstream.check(t) + + body, readErr := io.ReadAll(response.Body) + if isClientAbortTimeout(readErr) { + t.Fatalf("response body abort timed out: %v", readErr) + } + if readErr == nil { + t.Errorf("truncated upstream body completed as a clean HTTP success: body bytes = %d", len(body)) + } + if bytes.Contains(body, []byte("")) { + t.Errorf("S3 error XML was appended after the response was committed: %q", body) + } + if !bytes.HasPrefix(prefix, body) { + t.Errorf("downstream received bytes outside the upstream prefix: %d", len(body)) + } +} + +func TestRoutedListBucketsTruncatedUpstreamResponseAborts(t *testing.T) { + t.Run("HTTP/1.1", func(t *testing.T) { + upstream := newTruncatedListBucketsUpstream(t, []byte("abc")) + gateway := newSigningListBucketsGateway(t, upstream.server.URL) + response, err := gateway.Client().Do(signedListBucketsRequest(t, gateway.URL)) + assertTruncatedListBucketsResponse(t, response, err, upstream.prefix, upstream) + }) + + for _, tc := range []struct { + name string + tls bool + }{ + {name: "HTTP/1.0"}, + {name: "HTTP/1.0 TLS", tls: true}, + } { + t.Run(tc.name, func(t *testing.T) { + prefix := bytes.Repeat([]byte("a"), 2<<20) + upstream := newTruncatedListBucketsUpstream(t, prefix) + var gateway *httptest.Server + if tc.tls { + gateway = newSigningListBucketsTLSGateway(t, upstream.server.URL) + } else { + gateway = newSigningListBucketsGateway(t, upstream.server.URL) + } + request := signedListBucketsRequest(t, gateway.URL) + response, conn, err := sendSignedHTTP10ListBuckets(t, gateway.URL, request) + if conn != nil { + defer conn.Close() + } + assertTruncatedListBucketsResponse(t, response, err, prefix, upstream) + }) + } + + t.Run("HTTP/2", func(t *testing.T) { + upstream := newTruncatedListBucketsUpstream(t, []byte("abc")) + gateway, protoSeen := newSigningListBucketsHTTP2Gateway(t, upstream.server.URL) + response, err := gateway.Client().Do(signedListBucketsRequest(t, gateway.URL)) + select { + case proto := <-protoSeen: + if proto != "HTTP/2.0" { + t.Fatalf("handler saw protocol %q, want HTTP/2.0", proto) + } + case <-time.After(5 * time.Second): + t.Fatal("HTTP/2 request did not reach the handler") + } + assertTruncatedListBucketsResponse(t, response, err, upstream.prefix, upstream) + }) +} + +func TestRoutedListBucketsCompleteResponsesRemainComplete(t *testing.T) { + body := "complete bucket list" + for _, tc := range []struct { + name string + knownLength bool + http10 bool + }{ + {name: "known length HTTP/1.1", knownLength: true}, + {name: "unknown length HTTP/1.1"}, + {name: "unknown length HTTP/1.0", http10: true}, + } { + t.Run(tc.name, func(t *testing.T) { + var upstreamCalls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + upstreamCalls.Add(1) + if tc.knownLength { + w.Header().Set("Content-Type", "application/xml") + w.Header().Set("Content-Length", fmt.Sprint(len(body))) + w.WriteHeader(http.StatusOK) + _, _ = io.WriteString(w, body) + return + } + + hijacker, ok := w.(http.Hijacker) + if !ok { + return + } + conn, rw, err := hijacker.Hijack() + if err != nil { + return + } + _, _ = fmt.Fprintf(rw, "HTTP/1.1 200 OK\r\nContent-Type: application/xml\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n%x\r\n%s\r\n0\r\n\r\n", len(body), body) + _ = rw.Flush() + _ = conn.Close() + })) + t.Cleanup(upstream.Close) + gateway := newSigningListBucketsGateway(t, upstream.URL) + + request := signedListBucketsRequest(t, gateway.URL) + var response *http.Response + var err error + if tc.http10 { + var conn net.Conn + response, conn, err = sendSignedHTTP10ListBuckets(t, gateway.URL, request) + if conn != nil { + defer conn.Close() + } + } else { + response, err = gateway.Client().Do(request) + } + if err != nil { + t.Fatalf("ListBuckets request: %v", err) + } + defer response.Body.Close() + gotBody, readErr := io.ReadAll(response.Body) + if readErr != nil { + t.Fatalf("read complete ListBuckets response: %v", readErr) + } + if response.StatusCode != http.StatusOK { + t.Errorf("ListBuckets status = %d, want %d", response.StatusCode, http.StatusOK) + } + if string(gotBody) != body { + t.Errorf("ListBuckets body = %q, want %q", gotBody, body) + } + if tc.knownLength && response.ContentLength != int64(len(body)) { + t.Errorf("ContentLength = %d, want %d", response.ContentLength, len(body)) + } + if got := response.Header.Get("Content-Type"); got != "application/xml" { + t.Errorf("Content-Type = %q, want application/xml", got) + } + if got := upstreamCalls.Load(); got != 1 { + t.Errorf("upstream calls = %d, want 1", got) + } + }) + } +} + +func TestRoutedListBucketsAuthFailureBeforeCommitmentRemainsS3Error(t *testing.T) { + var upstreamCalls atomic.Int32 + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + upstreamCalls.Add(1) + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(upstream.Close) + gateway := newSigningListBucketsGateway(t, upstream.URL) + + request := signedListBucketsRequest(t, gateway.URL) + authorization := request.Header.Get("Authorization") + last := authorization[len(authorization)-1] + if last == '0' { + authorization = authorization[:len(authorization)-1] + "1" + } else { + authorization = authorization[:len(authorization)-1] + "0" + } + request.Header.Set("Authorization", authorization) + + response, err := gateway.Client().Do(request) + if err != nil { + t.Fatalf("ListBuckets auth error request: %v", err) + } + defer response.Body.Close() + body, readErr := io.ReadAll(response.Body) + if readErr != nil { + t.Fatalf("read auth error response: %v", readErr) + } + if response.StatusCode != http.StatusForbidden { + t.Errorf("auth failure status = %d, want %d", response.StatusCode, http.StatusForbidden) + } + if !bytes.Contains(body, []byte("SignatureDoesNotMatch")) { + t.Errorf("auth failure body = %q, want SignatureDoesNotMatch S3 XML", body) + } + if got := upstreamCalls.Load(); got != 0 { + t.Errorf("upstream calls after local auth rejection = %d, want 0", got) + } +} From 712e2b5b5e869277c208dc7c5f5eba88e579a3d9 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Fri, 2 Oct 2026 18:17:50 +0000 Subject: [PATCH 2/2] fix: abort incomplete upstream response streams Signed-off-by: Perfloop Agent --- README.md | 4 ++ docs/s3-compatibility.md | 8 +++ handlers/server.go | 51 +++++++++++++++++-- proxy/forwarder.go | 14 ++++-- proxy/forwarder_test.go | 52 ++++++++++++++++++++ proxy/reinvalidation_test.go | 84 ++++++++++++++++++++++++++++++++ proxy/service.go | 16 +++--- proxy/tiered.go | 19 +++++--- proxy/tiered_test.go | 86 +++++++++++++++++++++++++++++++++ proxy/write_through_test.go | 94 +++++++++++++++++++++++++++++++++++- 10 files changed, 403 insertions(+), 25 deletions(-) diff --git a/README.md b/README.md index c8d6fdb7..30927e4e 100644 --- a/README.md +++ b/README.md @@ -144,6 +144,10 @@ For production, TAG ships manifests, Compose files, and guides under [`deploy/`] 2. **Cache Hit**: Returns cached object with `X-Cache: HIT` header 3. **Cache Miss**: Forwards request to upstream Tigris, caches response, returns with `X-Cache: MISS` +For directly streamed upstream responses, TAG aborts the downstream response if copying the body fails. It does not complete a truncated prefix as a successful response. Clients may see a transport or body-read error. + +For streamed PUT and DELETE operations, an upstream 2xx still triggers cache invalidation or a tiered marker when caching is enabled. A delivery error alone does not prove the mutation failed. + ### Request Coalescing When multiple concurrent requests arrive for the same uncached object: diff --git a/docs/s3-compatibility.md b/docs/s3-compatibility.md index 4ec6c94d..d30746e8 100644 --- a/docs/s3-compatibility.md +++ b/docs/s3-compatibility.md @@ -57,6 +57,14 @@ All bucket operations support both `/{bucket}` and `/{bucket}/` path forms for c | AbortMultipartUpload | DELETE | `/{bucket}/{key}?uploadId=` | | | ListParts | GET | `/{bucket}/{key}?uploadId=` | | +## Response Delivery + +The direct-streaming path forwards the upstream status and headers before streaming the body. If copying a response body fails after headers are sent, TAG aborts the downstream response instead of completing its framing around the received prefix. Clients may see a transport error or body-read error. TAG does not append S3 error XML after commitment. + +For directly streamed PUT and DELETE operations, an upstream 2xx still triggers cache invalidation or a tiered marker when caching is enabled, even if response delivery fails. A delivery error alone does not establish that the mutation failed. + +This describes the direct-streaming path. Operations that capture an upstream body for XML processing use a separate response path. + ## AWS Chunked Transfer Encoding TAG supports AWS chunked transfer encoding (streaming SigV4), used by tools like [warp](https://github.com/minio/warp) and many AWS SDKs for large uploads. diff --git a/handlers/server.go b/handlers/server.go index 8c2e8436..dbdfd3d9 100644 --- a/handlers/server.go +++ b/handlers/server.go @@ -2,7 +2,9 @@ package handlers import ( "context" + "crypto/tls" "fmt" + "net" "net/http" "net/http/pprof" "regexp" @@ -304,17 +306,60 @@ func (rt *responseTracker) Flush() { } } +func (rt *responseTracker) Unwrap() http.ResponseWriter { + return rt.ResponseWriter +} + +// abortHTTP10Response resets the TCP connection because HTTP/1.0 uses connection +// close as the normal end of an unknown-length body. A reset makes a truncated +// response distinguishable from a complete one after its headers are committed. +// For TLS, close the underlying TCP connection directly so no close-notify turns +// the truncated body into a successful TLS EOF. +func abortHTTP10Response(w http.ResponseWriter) error { + conn, _, err := http.NewResponseController(w).Hijack() + if err != nil { + return fmt.Errorf("hijack HTTP/1.0 response: %w", err) + } + + transportConn := conn + if tlsConn, ok := conn.(*tls.Conn); ok { + transportConn = tlsConn.NetConn() + } + tcpConn, ok := transportConn.(*net.TCPConn) + if !ok { + _ = conn.Close() + return fmt.Errorf("HTTP/1.0 response connection %T is not TCP", transportConn) + } + if err := tcpConn.SetLinger(0); err != nil { + _ = tcpConn.Close() + return fmt.Errorf("enable abortive HTTP/1.0 response close: %w", err) + } + if err := tcpConn.Close(); err != nil { + return fmt.Errorf("reset HTTP/1.0 response connection: %w", err) + } + return nil +} + // handleWithError calls a handler function and handles any returned error. // If headers have already been committed (e.g., the handler started streaming -// a response before encountering an error), the error is logged but no error -// response is written to avoid corrupting the HTTP stream. +// a response before encountering an error), the response is aborted so the +// server does not complete framing for a partial body or append error XML. func handleWithError(w http.ResponseWriter, r *http.Request, handler func(http.ResponseWriter, *http.Request) error) { rt := &responseTracker{ResponseWriter: w} err := handler(rt, r) if err != nil { if rt.committed { log.Warn().Err(err).Str("path", r.URL.Path). - Msg("Error after response headers committed, cannot send error response") + Msg("Error after response headers committed, aborting response") + if r.ProtoMajor == 1 && r.ProtoMinor == 0 { + if abortErr := abortHTTP10Response(rt); abortErr == nil { + return + } else { + log.Warn().Err(abortErr).Str("path", r.URL.Path). + Msg("Failed to reset HTTP/1.0 response connection") + } + } + panic(http.ErrAbortHandler) } else { handleError(rt, r, err) } diff --git a/proxy/forwarder.go b/proxy/forwarder.go index 076f79fa..13ba9cad 100644 --- a/proxy/forwarder.go +++ b/proxy/forwarder.go @@ -195,8 +195,10 @@ func newBaseForwarder(tigrisEndpoint, region string, maxIdleConnsPerHost int) ba } // executeAndStream executes the request and streams the response to the client. -// originalReq is the original client request, passed to the response interceptor -// for parsing auth info. It can be nil if no interceptor is set. +// It returns a body-copy error even though the upstream status has already been +// committed; the HTTP handler must abort that incomplete response. originalReq is +// the original client request, passed to the response interceptor for parsing auth +// info. It can be nil if no interceptor is set. func (b *baseForwarder) executeAndStream(w http.ResponseWriter, fwdReq *http.Request, inContentLength int64, originalReq *http.Request) error { _, _, err := b.executeAndStreamWithMeta(w, fwdReq, inContentLength, originalReq, false) return err @@ -205,13 +207,15 @@ func (b *baseForwarder) executeAndStream(w http.ResponseWriter, fwdReq *http.Req // executeAndStreamReturningMeta is executeAndStream that also returns the upstream // status code and a clone of the upstream response headers. Used by the write-through // tee path, which needs the response ETag to build cache metadata for the just-written -// object without a read-back GET. On a transport error it returns (0, nil, err). +// object without a read-back GET. On a transport error it returns (0, nil, err); a +// body-copy error returns the received status and headers with the error. func (b *baseForwarder) executeAndStreamReturningMeta(w http.ResponseWriter, fwdReq *http.Request, inContentLength int64, originalReq *http.Request) (int, http.Header, error) { return b.executeAndStreamWithMeta(w, fwdReq, inContentLength, originalReq, true) } // executeAndStreamWithMeta streams an upstream response and optionally retains an -// owned copy of its headers for a caller that needs metadata after streaming. +// owned copy of its headers for a caller that needs metadata after streaming. It +// returns any body-copy error even though the upstream status is already committed. func (b *baseForwarder) executeAndStreamWithMeta(w http.ResponseWriter, fwdReq *http.Request, inContentLength int64, originalReq *http.Request, captureMeta bool) (int, http.Header, error) { if inContentLength > 0 { metrics.BytesTransferred.WithLabelValues("in").Add(float64(inContentLength)) @@ -248,7 +252,7 @@ func (b *baseForwarder) executeAndStreamWithMeta(w http.ResponseWriter, fwdReq * metrics.BytesTransferred.WithLabelValues("out").Add(float64(n)) logUpstreamResponse(fwdReq, originalReq, resp.StatusCode, upstreamStart) - return resp.StatusCode, respHeaders, nil + return resp.StatusCode, respHeaders, copyErr } // executeAndCapture executes the request, streams to client, and captures the response. diff --git a/proxy/forwarder_test.go b/proxy/forwarder_test.go index 72344c2c..64d3b8dc 100644 --- a/proxy/forwarder_test.go +++ b/proxy/forwarder_test.go @@ -1,6 +1,7 @@ package proxy import ( + "errors" "io" "net/http" "net/http/httptest" @@ -14,6 +15,57 @@ func (f forwarderTestTransport) RoundTrip(r *http.Request) (*http.Response, erro return f(r) } +type failedResponseBody struct { + reader *strings.Reader + err error + closed bool +} + +func (b *failedResponseBody) Read(p []byte) (int, error) { + if b.reader.Len() > 0 { + return b.reader.Read(p) + } + return 0, b.err +} + +func (b *failedResponseBody) Close() error { + b.closed = true + return nil +} + +func TestExecuteAndStreamReturningMetaReturnsBodyReadError(t *testing.T) { + readErr := errors.New("upstream response body failed") + body := &failedResponseBody{reader: strings.NewReader("prefix"), err: readErr} + headers := make(http.Header) + headers.Set("ETag", `"upstream-etag"`) + forwarder := newBaseForwarder("https://upstream.example.com", "us-east-1", 1) + forwarder.httpClient = &http.Client{Transport: forwarderTestTransport(func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusOK, Header: headers, Body: body}, nil + })} + request, err := http.NewRequest(http.MethodGet, "https://upstream.example.com/bucket", nil) + if err != nil { + t.Fatal(err) + } + writer := httptest.NewRecorder() + + status, gotHeaders, err := forwarder.executeAndStreamReturningMeta(writer, request, 0, nil) + if !errors.Is(err, readErr) { + t.Fatalf("executeAndStreamReturningMeta error = %v, want %v", err, readErr) + } + if status != http.StatusOK { + t.Errorf("status = %d, want %d", status, http.StatusOK) + } + if gotHeaders.Get("ETag") != `"upstream-etag"` { + t.Errorf("captured ETag = %q, want upstream ETag", gotHeaders.Get("ETag")) + } + if writer.Body.String() != "prefix" { + t.Errorf("client body = %q, want received prefix", writer.Body.String()) + } + if !body.closed { + t.Error("upstream response body was not closed") + } +} + func TestExecuteAndStreamReturningMetaOwnsHeaders(t *testing.T) { upstreamHeaders := make(http.Header) upstreamHeaders.Set("ETag", `"upstream-etag"`) diff --git a/proxy/reinvalidation_test.go b/proxy/reinvalidation_test.go index 5829f0cc..146ea324 100644 --- a/proxy/reinvalidation_test.go +++ b/proxy/reinvalidation_test.go @@ -2,6 +2,7 @@ package proxy import ( "context" + "errors" "net/http" "net/http/httptest" "strings" @@ -40,6 +41,89 @@ func TestHandlePutObject_ReinvalidatesAfterForward(t *testing.T) { } } +// A confirmed upstream PUT must re-invalidate a racing refill even when its +// response body fails to reach the client: delivery failure does not undo the write. +func TestHandlePutObject_ReinvalidatesAfterResponseCopyError(t *testing.T) { + copyErr := errors.New("response body delivery failed") + var c *cache.Cache + forward := func(_ context.Context, w http.ResponseWriter, r *http.Request) error { + b, k := ParseBucketKey(r) + stale := &cache.CachedObjectMeta{Bucket: b, Key: k, ETag: `"stale"`, ContentLength: 5, StatusCode: http.StatusOK} + if err := c.PutWithMeta(context.Background(), b, k, stale, []byte("stale"), 60); err != nil { + t.Errorf("seed racing refill: %v", err) + } + w.WriteHeader(http.StatusOK) + return copyErr + } + svc, c := newTestService(&mockForwarder{forwardFunc: forward}, true) + + r := httptest.NewRequest(http.MethodPut, "/test-bucket/test-key", strings.NewReader("new body")) + w := httptest.NewRecorder() + if err := svc.HandlePutObject(w, r); !errors.Is(err, copyErr) { + t.Fatalf("HandlePutObject error = %v, want %v", err, copyErr) + } + if w.Code != http.StatusOK { + t.Fatalf("upstream status = %d, want %d", w.Code, http.StatusOK) + } + b, k := ParseBucketKey(r) + if _, found, _ := c.GetMeta(context.Background(), b, k); found { + t.Error("stale entry survived a 2xx mutation whose response body failed to reach the client") + } +} + +func TestHandlePutObject_Non2xxResponseCopyErrorKeepsRacingRefill(t *testing.T) { + copyErr := errors.New("response body delivery failed") + var c *cache.Cache + forward := func(_ context.Context, w http.ResponseWriter, r *http.Request) error { + b, k := ParseBucketKey(r) + refill := &cache.CachedObjectMeta{Bucket: b, Key: k, ETag: `"refill"`, ContentLength: 6, StatusCode: http.StatusOK} + if err := c.PutWithMeta(context.Background(), b, k, refill, []byte("refill"), 60); err != nil { + t.Errorf("seed racing refill: %v", err) + } + w.WriteHeader(http.StatusForbidden) + return copyErr + } + svc, c := newTestService(&mockForwarder{forwardFunc: forward}, true) + + r := httptest.NewRequest(http.MethodPut, "/test-bucket/test-key", strings.NewReader("new body")) + w := httptest.NewRecorder() + if err := svc.HandlePutObject(w, r); !errors.Is(err, copyErr) { + t.Fatalf("HandlePutObject error = %v, want %v", err, copyErr) + } + b, k := ParseBucketKey(r) + if _, found, _ := c.GetMeta(context.Background(), b, k); !found { + t.Error("valid racing refill was discarded after the upstream rejected the PUT") + } +} + +func TestHandleDeleteObject_ReinvalidatesAfterResponseCopyError(t *testing.T) { + copyErr := errors.New("response body delivery failed") + var c *cache.Cache + forward := func(_ context.Context, w http.ResponseWriter, r *http.Request) error { + b, k := ParseBucketKey(r) + stale := &cache.CachedObjectMeta{Bucket: b, Key: k, ETag: `"stale"`, ContentLength: 5, StatusCode: http.StatusOK} + if err := c.PutWithMeta(context.Background(), b, k, stale, []byte("stale"), 60); err != nil { + t.Errorf("seed racing refill: %v", err) + } + w.WriteHeader(http.StatusNoContent) + return copyErr + } + svc, c := newTestService(&mockForwarder{forwardFunc: forward}, true) + + r := httptest.NewRequest(http.MethodDelete, "/test-bucket/test-key", nil) + w := httptest.NewRecorder() + if err := svc.HandleDeleteObject(w, r); !errors.Is(err, copyErr) { + t.Fatalf("HandleDeleteObject error = %v, want %v", err, copyErr) + } + if w.Code != http.StatusNoContent { + t.Fatalf("upstream status = %d, want %d", w.Code, http.StatusNoContent) + } + b, k := ParseBucketKey(r) + if _, found, _ := c.GetMeta(context.Background(), b, k); found { + t.Error("stale entry survived a 2xx DELETE whose response body failed to reach the client") + } +} + // A rejected CopyObject leaves the destination unchanged, so the post-forward // re-invalidation must NOT fire — otherwise a racing refill of the still-valid // destination is discarded, causing an unnecessary later miss. diff --git a/proxy/service.go b/proxy/service.go index eb9e9958..5cd1574c 100644 --- a/proxy/service.go +++ b/proxy/service.go @@ -659,11 +659,11 @@ func (s *Service) releaseCacheSlot(weight int64) { } // statusRecorder wraps http.ResponseWriter to capture the response status code -// written while forwarding an upstream response. Forward() returns nil even when -// upstream responds 4xx/5xx (the response streamed successfully), so mutating -// handlers use this to gate post-forward cache re-invalidation on an actual 2xx — -// otherwise a rejected PUT/DELETE/COPY would still fence the destination and -// discard a valid racing refill, causing later reads to miss unnecessarily. +// written while forwarding an upstream response. Mutating handlers use the +// upstream status to gate post-forward cache re-invalidation on an actual 2xx, +// even when response delivery later fails; otherwise a rejected PUT/DELETE/COPY +// would still fence the destination and discard a valid racing refill, causing +// later reads to miss unnecessarily. type statusRecorder struct { http.ResponseWriter status int @@ -730,7 +730,8 @@ func (s *Service) HandlePutObject(w http.ResponseWriter, r *http.Request) error // would only discard a valid racing refill and cause an unnecessary later miss. // Routed through invalidateObject (like the pre-forward call) so a failure of this // read-after-write-critical invalidation is recorded and logged, not discarded. - if err == nil && rec.wroteSuccess() && s.cache.IsEnabled() { + // The upstream 2xx confirms the mutation even if delivery of its response body failed. + if rec.wroteSuccess() && s.cache.IsEnabled() { s.convergeInvalidation(context.Background(), bucket, key) teeHandled := requestRejectsCache if teed != nil { @@ -820,7 +821,8 @@ func (s *Service) HandleDeleteObject(w http.ResponseWriter, r *http.Request) err // fence bump blocks that stale repopulation. // Gated on a 2xx: a rejected DELETE leaves the object present, so re-invalidating // would only discard a valid racing refill and cause an unnecessary later miss. - if err == nil && rec.wroteSuccess() && s.cache.IsEnabled() { + // The upstream status remains authoritative when response-body delivery fails. + if rec.wroteSuccess() && s.cache.IsEnabled() { if s.config.IsTiered() { s.convergeTieredDelete(bucket, key, delPriorVer, delPriorKnown) } else { diff --git a/proxy/tiered.go b/proxy/tiered.go index 66d54c16..1a48ab4b 100644 --- a/proxy/tiered.go +++ b/proxy/tiered.go @@ -208,13 +208,15 @@ func (s *Service) handleTieredPut(w http.ResponseWriter, r *http.Request) error // that makes this object exist in TAG's authoritative view. // // No pre-forward invalidation, unlike HandlePutObject: for a local-tier - // prior the cache holds the ONLY copy, and a failed forward must leave it - // intact — S3 semantics say a rejected PUT changes nothing. Reads racing - // the in-flight PUT serve the prior version, which is the atomic-replace - // behavior clients expect. The one read-triggered populate in this mode — - // the re-tier — cannot be ordered against this path's writes here (it - // performs none pre-forward); it defends itself with a claim plus its own - // pre-fetch decision token instead (see maybeRetierOnRead). + // prior the cache holds the ONLY copy, and an upstream-rejected PUT must + // leave it intact — S3 semantics say a rejected mutation changes nothing. + // The status recorder distinguishes that rejection from a response-delivery + // error after upstream has confirmed the write. Reads racing the in-flight + // PUT serve the prior version, which is the atomic-replace behavior clients + // expect. The one read-triggered populate in this mode — the re-tier — cannot + // be ordered against this path's writes here (it performs none pre-forward); it + // defends itself with a claim plus its own pre-fetch decision token instead + // (see maybeRetierOnRead). // No post-success invalidation either: the marker overwrites the prior // metadata directly (a displaced local body ages out by TTL, the engine's // own overwrite semantics), which lets the marker commit under the @@ -243,7 +245,8 @@ func (s *Service) handleTieredPut(w http.ResponseWriter, r *http.Request) error rec := &statusRecorder{ResponseWriter: w} err = s.forwarder.Forward(ctx, rec, r) - if err == nil && rec.wroteSuccess() && markerOwning && s.cache.IsEnabled() { + // A confirmed upstream 2xx still needs its marker if response-body delivery fails. + if rec.wroteSuccess() && markerOwning && s.cache.IsEnabled() { s.putUpstreamMarker(r, w.Header().Get("ETag"), bucket, key, prior, priorVersion, priorKnown) } diff --git a/proxy/tiered_test.go b/proxy/tiered_test.go index c3d5e8ae..0c32c5c7 100644 --- a/proxy/tiered_test.go +++ b/proxy/tiered_test.go @@ -70,6 +70,92 @@ func tieredMock() (*mockForwarder, *atomic.Int64, *atomic.Int64) { return m, forwards, deletes } +func TestTieredUpstreamPutCommitsMarkerAfterResponseCopyError(t *testing.T) { + copyErr := errors.New("response body delivery failed") + mock := &mockForwarder{ + forwardFunc: func(_ context.Context, w http.ResponseWriter, _ *http.Request) error { + w.Header().Set("ETag", `"new-upstream-version"`) + w.WriteHeader(http.StatusOK) + return copyErr + }, + } + service, c := newTieredTestService(mock, 4) + request := httptest.NewRequest(http.MethodPut, "/b/large-object", strings.NewReader("large object body")) + writer := httptest.NewRecorder() + + if err := service.HandlePutObject(writer, request); !errors.Is(err, copyErr) { + t.Fatalf("HandlePutObject error = %v, want %v", err, copyErr) + } + if writer.Code != http.StatusOK { + t.Fatalf("upstream status = %d, want %d", writer.Code, http.StatusOK) + } + meta, found, err := c.GetMeta(context.Background(), "b", "large-object") + if err != nil { + t.Fatalf("GetMeta: %v", err) + } + if !found || meta == nil || !meta.BodyUpstream || meta.ETag != `"new-upstream-version"` { + t.Fatalf("upstream mutation marker missing after body delivery error: %+v", meta) + } +} + +func TestTieredDeleteConvergesAfterResponseCopyError(t *testing.T) { + copyErr := errors.New("response body delivery failed") + mock := &mockForwarder{ + forwardFunc: func(_ context.Context, w http.ResponseWriter, _ *http.Request) error { + w.WriteHeader(http.StatusNoContent) + return copyErr + }, + } + service, c := newTieredTestService(mock, 1024) + marker := &cache.CachedObjectMeta{Bucket: "b", Key: "object", ETag: `"upstream-version"`, BodyUpstream: true, ContentLength: 5000, StatusCode: http.StatusOK} + if err := c.PutWithMeta(context.Background(), "b", "object", marker, []byte("body"), 60); err != nil { + t.Fatalf("seed upstream marker: %v", err) + } + request := httptest.NewRequest(http.MethodDelete, "/b/object", nil) + writer := httptest.NewRecorder() + + if err := service.HandleDeleteObject(writer, request); !errors.Is(err, copyErr) { + t.Fatalf("HandleDeleteObject error = %v, want %v", err, copyErr) + } + if writer.Code != http.StatusNoContent { + t.Fatalf("upstream status = %d, want %d", writer.Code, http.StatusNoContent) + } + if meta, found, err := c.GetMeta(context.Background(), "b", "object"); err != nil { + t.Fatalf("GetMeta: %v", err) + } else if found { + t.Fatalf("deleted upstream marker survived response copy failure: %+v", meta) + } +} + +func TestTieredRejectedDeleteWithResponseCopyErrorKeepsMarker(t *testing.T) { + copyErr := errors.New("response body delivery failed") + mock := &mockForwarder{ + forwardFunc: func(_ context.Context, w http.ResponseWriter, _ *http.Request) error { + w.WriteHeader(http.StatusForbidden) + return copyErr + }, + } + service, c := newTieredTestService(mock, 1024) + marker := &cache.CachedObjectMeta{Bucket: "b", Key: "object", ETag: `"upstream-version"`, BodyUpstream: true, ContentLength: 5000, StatusCode: http.StatusOK} + if err := c.PutWithMeta(context.Background(), "b", "object", marker, []byte("body"), 60); err != nil { + t.Fatalf("seed upstream marker: %v", err) + } + request := httptest.NewRequest(http.MethodDelete, "/b/object", nil) + writer := httptest.NewRecorder() + + if err := service.HandleDeleteObject(writer, request); !errors.Is(err, copyErr) { + t.Fatalf("HandleDeleteObject error = %v, want %v", err, copyErr) + } + if writer.Code != http.StatusForbidden { + t.Fatalf("upstream status = %d, want %d", writer.Code, http.StatusForbidden) + } + if current, found, err := c.GetMeta(context.Background(), "b", "object"); err != nil { + t.Fatalf("GetMeta: %v", err) + } else if !found || current == nil || current.ETag != marker.ETag { + t.Fatalf("upstream-rejected DELETE changed the marker: %+v", current) + } +} + func tieredDo(t *testing.T, svc *Service, method, path, body string, hdr map[string]string) *httptest.ResponseRecorder { t.Helper() var req *http.Request diff --git a/proxy/write_through_test.go b/proxy/write_through_test.go index 36eca8ec..da327834 100644 --- a/proxy/write_through_test.go +++ b/proxy/write_through_test.go @@ -11,6 +11,8 @@ import ( "sync/atomic" "testing" "time" + + "github.com/tigrisdata/tag/cache" ) // teeMockForwarder is a mockForwarder that also implements bodyTeeingForwarder, so @@ -21,8 +23,8 @@ import ( type teeMockForwarder struct { *mockForwarder teeFunc func(ctx context.Context, w http.ResponseWriter, r *http.Request, tee io.Writer) (int, http.Header, string, string, error) - // headHook, if set, runs when the tee issues its authoritative HEAD — used to simulate a - // competing write landing during the HEAD window. + // headHook, if set, runs when the tee issues its authoritative HEAD — used to gate that + // work or simulate a competing write during the HEAD window. headHook func() } @@ -62,6 +64,94 @@ func headResp(etag, contentType string, contentLength int64) *http.Response { return &http.Response{StatusCode: http.StatusOK, Header: h, Body: http.NoBody, ContentLength: contentLength} } +// A response delivery error after the upstream PUT's 2xx still converges the cache, and the +// tee reservation remains owned until its asynchronous cache write finishes and releases it. +func TestHandlePutObject_WriteThroughTee_ResponseCopyErrorInvalidatesAndReleasesReservation(t *testing.T) { + copyErr := errors.New("response body delivery failed") + body := "tee-body-after-successful-upstream-put" + headStarted := make(chan struct{}, 1) + releaseHead := make(chan struct{}) + headReleased := false + defer func() { + if !headReleased { + close(releaseHead) + } + }() + + var c *cache.Cache + forwarder := &teeMockForwarder{ + mockForwarder: &mockForwarder{ + conditionalResp: headResp(`"tee-etag"`, "text/plain", int64(len(body))), + }, + teeFunc: func(_ context.Context, w http.ResponseWriter, r *http.Request, tee io.Writer) (int, http.Header, string, string, error) { + if _, err := io.Copy(tee, r.Body); err != nil { + return 0, nil, "", "", err + } + bucket, key := ParseBucketKey(r) + stale := &cache.CachedObjectMeta{Bucket: bucket, Key: key, ETag: `"stale"`, ContentLength: 5, StatusCode: http.StatusOK} + if err := c.PutWithMeta(context.Background(), bucket, key, stale, []byte("stale"), 60); err != nil { + return 0, nil, "", "", err + } + headers := http.Header{} + headers.Set("ETag", `"tee-etag"`) + w.Header().Set("ETag", `"tee-etag"`) + w.WriteHeader(http.StatusOK) + _, _ = io.WriteString(w, "response prefix") + return http.StatusOK, headers, "access", "secret", copyErr + }, + headHook: func() { + headStarted <- struct{}{} + <-releaseHead + }, + } + service, c := newTestService(forwarder, true) + service.config.Cache.WarmOnWrite = true + service.config.Cache.SizeThreshold = 1 << 20 + service.config.Cache.BlockSize = 1 << 20 + service.cacheSemaphore = make(chan struct{}, 1) + + request := authedPut(wowBucket, wowKey, body) + writer := httptest.NewRecorder() + if err := service.HandlePutObject(writer, request); !errors.Is(err, copyErr) { + t.Fatalf("HandlePutObject error = %v, want %v", err, copyErr) + } + if writer.Code != http.StatusOK { + t.Fatalf("upstream status = %d, want %d", writer.Code, http.StatusOK) + } + if _, found, _ := c.GetMeta(context.Background(), wowBucket, wowKey); found { + t.Fatal("the 2xx PUT's post-forward invalidation did not remove the racing refill") + } + select { + case <-headStarted: + case <-time.After(5 * time.Second): + t.Fatal("write-through worker did not reach its gated HEAD") + } + select { + case service.cacheSemaphore <- struct{}{}: + <-service.cacheSemaphore + t.Fatal("tee reservation was released before its asynchronous owner finished using the body") + default: + } + + slotAcquired := make(chan struct{}) + go func() { + service.cacheSemaphore <- struct{}{} + close(slotAcquired) + }() + close(releaseHead) + headReleased = true + select { + case <-slotAcquired: + case <-time.After(5 * time.Second): + t.Fatal("tee reservation was not released after the write-through owner completed") + } + <-service.cacheSemaphore + if !service.acquireCacheSlot(context.Background(), int64(len(body)), priorityReadMiss) { + t.Fatal("tee completion did not release both the count slot and byte reservation") + } + service.releaseCacheSlot(int64(len(body))) +} + // A single authenticated PutObject within the size threshold is cached by teeing the body, // with metadata sourced from an authoritative HEAD (not the read-back full-object GET). The // cached ETag/Content-Type/Content-Length/Last-Modified come from the HEAD, and the body