From 019e757601b2ad92226598e9c7d607361d600406 Mon Sep 17 00:00:00 2001 From: Stuart Douglas Date: Mon, 2 Nov 2026 18:09:01 +1100 Subject: [PATCH 1/2] fix: don't truncate downloads when the cache write fails A cache write failure aborted the shared copy feeding the client, so clients got 200 OK with a truncated body. The S3 backend triggered this whenever upstream response headers exceeded S3's 2 KB user metadata limit. - handler: keep streaming to the client when the cache writer fails or can't be created, and discard the cache entry. - s3: when headers don't fit in user metadata, store only the headers needed to serve the object (content headers, ETag, Last-Modified, X-Cachew-*), and fail Create up front if even those don't fit. - handler: log the cache key with fetch and serve errors. Amp-Thread-ID: https://ampcode.com/threads/T-01a19b0c-ec15-75f2-b427-a1f18a35f91c Co-authored-by: Amp --- internal/cache/s3.go | 70 +++++++++++++++++++---- internal/cache/s3_test.go | 39 +++++++++++++ internal/strategy/handler/handler.go | 52 +++++++++++------ internal/strategy/handler/handler_test.go | 66 +++++++++++++++++++++ 4 files changed, 199 insertions(+), 28 deletions(-) diff --git a/internal/cache/s3.go b/internal/cache/s3.go index a9b43a2c..6ab04552 100644 --- a/internal/cache/s3.go +++ b/internal/cache/s3.go @@ -15,6 +15,7 @@ import ( "os" "runtime" "strconv" + "strings" "time" "github.com/alecthomas/errors" @@ -87,6 +88,55 @@ type s3Meta struct { // keeps them readable. const s3TagMetadataKey = "Tag" +// s3MaxUserMetadataSize is S3's limit on the combined size of an object's +// user-defined metadata keys and values. PutObject fails if it is exceeded. +const s3MaxUserMetadataSize = 2048 + +// s3EssentialHeaders are the headers kept when the full header set doesn't fit +// in S3 user metadata, along with any X-Cachew-* headers. +var s3EssentialHeaders = []string{ //nolint:gochecknoglobals + "Content-Type", "Content-Encoding", "Content-Disposition", "Content-Language", + "Cache-Control", ETagKey, "Last-Modified", +} + +// s3UserMetadata encodes the tag and headers as data object user metadata. +// Upstreams can send arbitrarily large response headers, so when they don't +// fit within S3's limit only the headers needed to serve the object are kept. +func s3UserMetadata(tag string, headers http.Header) (map[string]string, error) { + metadata := map[string]string{s3TagMetadataKey: tag} + if len(headers) == 0 { + return metadata, nil + } + essential := make(http.Header) + for _, name := range s3EssentialHeaders { + if values := headers.Values(name); len(values) > 0 { + essential[http.CanonicalHeaderKey(name)] = values + } + } + for name, values := range headers { + if strings.HasPrefix(http.CanonicalHeaderKey(name), "X-Cachew-") { + essential[name] = values + } + } + for _, candidate := range []http.Header{headers, essential} { + headersJSON, err := json.Marshal(candidate) + if err != nil { + return nil, errors.Errorf("failed to marshal headers: %w", err) + } + metadata["Headers"] = string(headersJSON) + size := 0 + for k, v := range metadata { + // Count the x-amz-meta- prefix too, to stay under the limit + // regardless of how the store measures keys. + size += len("x-amz-meta-") + len(k) + len(v) + } + if size <= s3MaxUserMetadataSize { + return metadata, nil + } + } + return nil, errors.Errorf("headers exceed S3 user metadata limit of %d bytes", s3MaxUserMetadataSize) +} + // newS3Tag returns a random hex tag used to correlate a data object with its // companion metadata object. func newS3Tag() (string, error) { @@ -402,6 +452,11 @@ func (s *S3) Create(ctx context.Context, key Key, headers http.Header, ttl time. return nil, err } + userMetadata, err := s3UserMetadata(tag, clonedHeaders) + if err != nil { + return nil, err + } + ctx, cancel := context.WithCancelCause(ctx) pr, pw := io.Pipe() @@ -413,6 +468,7 @@ func (s *S3) Create(ctx context.Context, key Key, headers http.Header, ttl time. pipe: pw, expiresAt: expiresAt, headers: clonedHeaders, + metadata: userMetadata, tag: tag, ctx: ctx, cancel: cancel, @@ -458,6 +514,7 @@ type s3Writer struct { pipe *io.PipeWriter expiresAt time.Time headers http.Header + metadata map[string]string tag string ctx context.Context cancel context.CancelCauseFunc @@ -561,21 +618,10 @@ func (w *s3Writer) upload(pr *io.PipeReader, r io.Reader) { objectName := w.s3.keyToPath(w.namespace, w.key) - userMetadata := map[string]string{s3TagMetadataKey: w.tag} - if len(w.headers) > 0 { - headersJSON, err := json.Marshal(w.headers) - if err != nil { - uploadErr = errors.Errorf("failed to marshal headers: %w", err) - w.errCh <- uploadErr - return - } - userMetadata["Headers"] = string(headersJSON) - } - // Configure upload options. CRC64-NVME is computed as data streams through and sent as a // trailing header so S3 validates integrity server-side, preventing corrupt or truncated uploads. opts := minio.PutObjectOptions{ - UserMetadata: userMetadata, + UserMetadata: w.metadata, AutoChecksum: minio.ChecksumCRC64NVME, Expires: w.expiresAt, } diff --git a/internal/cache/s3_test.go b/internal/cache/s3_test.go index b596887c..ab2d71a2 100644 --- a/internal/cache/s3_test.go +++ b/internal/cache/s3_test.go @@ -2,8 +2,11 @@ package cache_test import ( "context" + "io" "log/slog" + "net/http" "os" + "strings" "testing" "time" @@ -71,6 +74,42 @@ func TestS3ContextCancellationAbortsUpload(t *testing.T) { assert.IsError(t, err, os.ErrNotExist) } +func TestS3OversizedHeadersKeepEssentials(t *testing.T) { + bucket := s3clienttest.Start(t) + c := newS3Cache(t, bucket) + defer c.Close() + + ctx := t.Context() + key := cache.NewKey("oversized-headers") + headers := http.Header{ + "Content-Type": {"application/zip"}, + "Last-Modified": {"Mon, 02 Nov 2026 05:42:00 GMT"}, + "X-Cachew-Snapshot-Commit": {"abc123"}, + "Content-Security-Policy": {strings.Repeat("default-src 'self'; ", 150)}, + } + + w, err := c.Create(ctx, key, headers, time.Hour) + assert.NoError(t, err) + _, err = w.Write([]byte("payload")) + assert.NoError(t, err) + assert.NoError(t, w.Close()) + + r, got, err := c.Open(ctx, key) + assert.NoError(t, err) + data, err := io.ReadAll(r) + assert.NoError(t, err) + assert.NoError(t, r.Close()) + assert.Equal(t, "payload", string(data)) + assert.Equal(t, "application/zip", got.Get("Content-Type")) + assert.Equal(t, "Mon, 02 Nov 2026 05:42:00 GMT", got.Get("Last-Modified")) + assert.Equal(t, "abc123", got.Get("X-Cachew-Snapshot-Commit")) + assert.NotEqual(t, "", got.Get(cache.ETagKey)) + assert.Equal(t, "", got.Get("Content-Security-Policy")) + + _, err = c.Create(ctx, cache.NewKey("too-large"), http.Header{"Content-Disposition": {strings.Repeat("x", 3000)}}, time.Hour) + assert.Error(t, err) +} + func TestS3CacheSoak(t *testing.T) { if os.Getenv("SOAK_TEST") == "" { t.Skip("Skipping soak test; set SOAK_TEST=1 to run") diff --git a/internal/strategy/handler/handler.go b/internal/strategy/handler/handler.go index 46ecf0a4..79376174 100644 --- a/internal/strategy/handler/handler.go +++ b/internal/strategy/handler/handler.go @@ -1,6 +1,7 @@ package handler import ( + "cmp" "io" "maps" "net/http" @@ -149,14 +150,14 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { served, err := h.serveCached(w, r, key) if err != nil { - logger.ErrorContext(ctx, "Failed to serve from cache", "error", err) + logger.ErrorContext(ctx, "Failed to serve from cache", "cache_key", parts.Path, "error", err) } if served { return } if err := h.fetchAndCache(w, r, key); err != nil { - logger.ErrorContext(ctx, "Failed to fetch and cache", "error", err) + logger.ErrorContext(ctx, "Failed to fetch and cache", "cache_key", parts.Path, "error", err) } } @@ -212,32 +213,51 @@ func (h *Handler) streamNonOKResponse(w http.ResponseWriter, resp *http.Response return nil } +// streamAndCache streams resp to the client while writing it to the cache. +// Cache failures never interrupt the client stream: the cache entry is +// discarded and the response is streamed in full. func (h *Handler) streamAndCache(w http.ResponseWriter, r *http.Request, key cache.Key, resp *http.Response) error { - ttl := h.ttlFunc(r) - responseHeaders := maps.Clone(resp.Header) - cw, err := h.cache.Create(r.Context(), key, responseHeaders, ttl) + maps.Copy(w.Header(), resp.Header) + + cw, err := h.cache.Create(r.Context(), key, maps.Clone(resp.Header), h.ttlFunc(r)) if err != nil { - h.errorHandler(httputil.Errorf(http.StatusInternalServerError, "failed to create cache entry: %w", err), w, r) - return nil + _, copyErr := io.Copy(w, resp.Body) + return errors.Join(errors.Wrap(err, "create cache entry"), errors.Wrap(copyErr, "stream response")) } + tee := &cacheTee{w: cw} + cacheErrCh := make(chan error, 1) pr, pw := io.Pipe() go func() { - mw := io.MultiWriter(pw, cw) - _, copyErr := io.Copy(mw, resp.Body) - var closeErr error - if copyErr != nil { - closeErr = cw.Abort(copyErr) + _, copyErr := io.Copy(io.MultiWriter(pw, tee), resp.Body) + if abortErr := cmp.Or(copyErr, tee.err); abortErr != nil { + cacheErrCh <- errors.Join(tee.err, cw.Abort(abortErr)) } else { - closeErr = cw.Close() + cacheErrCh <- cw.Close() } - pw.CloseWithError(errors.Join(copyErr, closeErr)) + pw.CloseWithError(copyErr) }() - maps.Copy(w.Header(), resp.Header) _, copyErr := io.Copy(w, pr) closeErr := pr.Close() - return errors.Wrap(errors.Join(copyErr, closeErr), "stream and cache response") + return errors.Join( + errors.Wrap(errors.Join(copyErr, closeErr), "stream response"), + errors.Wrap(<-cacheErrCh, "cache response"), + ) +} + +// cacheTee forwards writes to a cache writer until it fails, then discards +// them, so a cache failure can't abort the copy that feeds the client. +type cacheTee struct { + w io.Writer + err error +} + +func (t *cacheTee) Write(p []byte) (int, error) { + if t.err == nil { + _, t.err = t.w.Write(p) + } + return len(p), nil } func defaultErrorHandler(err error, w http.ResponseWriter, r *http.Request) { diff --git a/internal/strategy/handler/handler_test.go b/internal/strategy/handler/handler_test.go index 83d6faa7..a45aebe3 100644 --- a/internal/strategy/handler/handler_test.go +++ b/internal/strategy/handler/handler_test.go @@ -7,6 +7,7 @@ import ( "net/http" "net/http/httptest" "os" + "strconv" "strings" "testing" "time" @@ -438,6 +439,71 @@ func TestStreamAndCacheAbortsOnUpstreamError(t *testing.T) { assert.IsError(t, err, os.ErrNotExist) } +type failingWriteCache struct { + cache.Cache + createErr error +} + +func (c failingWriteCache) Create(ctx context.Context, key cache.Key, headers http.Header, ttl time.Duration, opts ...cache.Option) (cache.Writer, error) { + if c.createErr != nil { + return nil, c.createErr + } + w, err := c.Cache.Create(ctx, key, headers, ttl, opts...) + return &failingWriter{Writer: w}, errors.WithStack(err) +} + +type failingWriter struct { + cache.Writer + writes int +} + +func (w *failingWriter) Write(p []byte) (int, error) { + w.writes++ + if w.writes > 1 { + return 0, errors.New("failed to put object: metadata too large") + } + return errors.WithStack2(w.Writer.Write(p)) +} + +func TestCacheWriteFailureDoesNotTruncateResponse(t *testing.T) { + payload := strings.Repeat("0123456789abcdef", 64<<10) + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Length", strconv.Itoa(len(payload))) + _, _ = w.Write([]byte(payload)) + })) + defer backend.Close() + + for _, tt := range []struct { + name string + createErr error + }{ + {name: "WriteFails"}, + {name: "CreateFails", createErr: errors.New("headers exceed S3 metadata limit")}, + } { + t.Run(tt.name, func(t *testing.T) { + mem := mustNewMemoryCache() + defer mem.Close() + ctx := logging.ContextWithLogger(context.Background(), slog.Default()) + + h := handler.New(http.DefaultClient, failingWriteCache{Cache: mem, createErr: tt.createErr}). + Transform(func(r *http.Request) (*http.Request, error) { + return http.NewRequestWithContext(r.Context(), http.MethodGet, backend.URL, nil) + }) + + r := httptest.NewRequest(http.MethodGet, "http://example.com/big.zip", nil).WithContext(ctx) + w := httptest.NewRecorder() + h.ServeHTTP(w, r) + + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, len(payload), w.Body.Len()) + assert.True(t, w.Body.String() == payload, "response body differs from upstream") + + _, _, err := mem.Open(ctx, cache.NewKey("http://example.com/big.zip")) + assert.IsError(t, err, os.ErrNotExist) + }) + } +} + func mustNewMemoryCache() cache.Cache { _, ctx := logging.Configure(context.Background(), logging.Config{Level: slog.LevelError}) c, err := cache.NewMemory(ctx, cache.MemoryConfig{ From 7558c3e2de9037d6f871335fbab0e95043260414 Mon Sep 17 00:00:00 2001 From: Stuart Douglas Date: Mon, 2 Nov 2026 18:33:57 +1100 Subject: [PATCH 2/2] chore: drop comments on small helpers per AGENTS.md Amp-Thread-ID: https://ampcode.com/threads/T-01a19b0c-ec15-75f2-b427-a1f18a35f91c Co-authored-by: Amp --- internal/cache/s3.go | 2 -- internal/strategy/handler/handler.go | 2 -- 2 files changed, 4 deletions(-) diff --git a/internal/cache/s3.go b/internal/cache/s3.go index 6ab04552..854ba134 100644 --- a/internal/cache/s3.go +++ b/internal/cache/s3.go @@ -126,8 +126,6 @@ func s3UserMetadata(tag string, headers http.Header) (map[string]string, error) metadata["Headers"] = string(headersJSON) size := 0 for k, v := range metadata { - // Count the x-amz-meta- prefix too, to stay under the limit - // regardless of how the store measures keys. size += len("x-amz-meta-") + len(k) + len(v) } if size <= s3MaxUserMetadataSize { diff --git a/internal/strategy/handler/handler.go b/internal/strategy/handler/handler.go index 79376174..cac4cad5 100644 --- a/internal/strategy/handler/handler.go +++ b/internal/strategy/handler/handler.go @@ -246,8 +246,6 @@ func (h *Handler) streamAndCache(w http.ResponseWriter, r *http.Request, key cac ) } -// cacheTee forwards writes to a cache writer until it fails, then discards -// them, so a cache failure can't abort the copy that feeds the client. type cacheTee struct { w io.Writer err error