fix: don't truncate downloads when the cache write fails

internal/cache/s3.go

@@ -15,6 +15,7 @@ import (
 	"os"
 	"runtime"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/alecthomas/errors"
@@ -87,6 +88,53 @@ 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 {
+			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 +450,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 +466,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 +512,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 +616,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,
 	}

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")

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,49 @@ 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"),
+	)
+}
+
+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) {

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{