fix(proxy): detect incomplete whole-object cache hits

README.md

@@ -144,6 +144,8 @@ 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 whole-object cache hits with a positive recorded length, TAG checks the bytes served against that length. In transparent and signing modes, a short buffered body falls back to upstream before headers are sent. If that fallback fails before upstream response headers arrive, TAG returns the error without committing the cached prefix. If a large stream is already committed, TAG counts a request error, attempts guarded metadata invalidation, and does not append an upstream body. Tiered local reads do not fetch a replacement body from upstream: a short buffered body becomes the documented `NoSuchKey` miss for the caller to fall back, while a committed short stream is counted as an error and does not append an upstream body.
+
 ### Request Coalescing
 
 When multiple concurrent requests arrive for the same uncached object:

docs/architecture.md

@@ -206,6 +206,8 @@ Client                 TAG                    Embedded Cache
   │  X-Cache: HIT       │                       │
 ```
 
+For a whole-object cache entry with a positive recorded length, TAG compares the bytes served with the metadata length before counting a local GET as successful. In transparent and signing modes, a short buffered body falls back to upstream before response headers commit. If that request fails before upstream response headers arrive, TAG returns the error without committing the cached prefix. A large stream can commit on its first chunk; if it ends short, TAG records a request error, attempts guarded metadata invalidation, and does not append an upstream body. Tiered local reads have no upstream fallback: a short buffered body returns the documented authoritative `NoSuchKey` miss without the cached prefix, and a committed short stream is recorded as a local error without an upstream append. Zero-length and unknown-length objects retain their existing paths. Range and block-cache reads use separate helpers.
+
 ### GET Object (Cache Miss)
 
 ```text
@@ -229,6 +231,8 @@ Client                 TAG                    Embedded Cache         Tigris
   │  X-Cache: MISS      │                       │                    │
 ```
 
+If an upstream fetch fails before response headers are published, TAG returns the error before committing a success response.
+
 ### GET Object (Cluster Mode - Remote Key)
 
 ```text

docs/tiered-mode.md

@@ -21,6 +21,8 @@ what makes the entry visible. `If-Match`/`If-None-Match` are honored; GET,
 HEAD, and DELETE are served entirely locally. Reads and writes of small
 objects generate zero upstream traffic.
 
+For a positive-length whole-object local hit, TAG checks the served byte count against metadata. A short body found before headers commit is discarded; TAG attempts version-guarded metadata cleanup, records a local request error, and returns the authoritative `NoSuchKey` miss so the caller can fall back to its system of record. If a large local body is found short after streaming commits headers, the response cannot be replaced: TAG records a local request error, attempts version-guarded metadata cleanup, and does not append an upstream body. Cleanup is best effort.
+
 **Large objects are the upstream tier.** The PUT passes through to upstream and
 TAG stores a metadata marker locally. HEAD answers from the marker; GET
 forwards for the body — the mode's only body traffic. DELETE forwards and

proxy/broadcast/broadcaster.go

@@ -62,6 +62,8 @@ const (
 // The client will see a truncated response and S3 SDKs will automatically retry.
 var ErrSlowConsumer = errors.New("slow consumer")
 
+var errCompletedBeforeHeaders = errors.New("broadcast completed before response headers")
+
 // Chunk represents a piece of streamed data.
 // Consumers should call Release() after processing Data to return the
 // underlying buffer to the pool. If Release() is not called, the buffer
@@ -90,24 +92,33 @@ type Listener struct {
 	ch           chan Chunk
 	headers      http.Header
 	status       int
-	headerCh     chan struct{} // Closed when headers are available
-	disconnected bool          // True if disconnected due to slow consumption
+	headerCh     chan struct{} // Closed when headers or a pre-header completion error is available
+	headerErr    error
+	disconnected bool // True if disconnected due to slow consumption
 }
 
-// WaitForHeaders blocks until headers are available. If headers and context
-// cancellation become ready together, headers win. A cache listener can be
-// scheduled after its request handler returns, when both are already ready; it
-// must still consume the completed response and publish the cache entry.
+// WaitForHeaders blocks until headers are available or the broadcast completes
+// before headers arrive. If headers and context cancellation become ready
+// together, headers win. A cache listener can be scheduled after its request
+// handler returns, when both are already ready; it must still consume the
+// completed response and publish the cache entry.
 func (l *Listener) WaitForHeaders(ctx context.Context) (int, http.Header, error) {
 	select {
 	case <-l.headerCh:
+		if l.headerErr != nil {
+			return 0, nil, l.headerErr
+		}
 		return l.status, l.headers, nil
 	case <-ctx.Done():
 		// A select chooses randomly when both channels are ready. Re-check the
-		// headers so a completed response wins over concurrent cancellation, while
-		// still honoring cancellation when headers have not arrived.
+		// notification so a completed response or terminal error wins over
+		// concurrent cancellation, while still honoring cancellation when neither
+		// has arrived.
 		select {
 		case <-l.headerCh:
+			if l.headerErr != nil {
+				return 0, nil, l.headerErr
+			}
 			return l.status, l.headers, nil
 		default:
 			return 0, nil, ctx.Err()
@@ -167,7 +178,8 @@ func (b *Broadcaster) IsStreaming() bool {
 }
 
 // Subscribe adds a new listener to receive chunks.
-// Returns nil if streaming has already started (no late joiners).
+// Returns nil if streaming has already started (no late joiners). A completed
+// broadcast that never streamed a body returns its terminal snapshot.
 func (b *Broadcaster) Subscribe() *Listener {
 	b.mu.Lock()
 	defer b.mu.Unlock()
@@ -182,11 +194,24 @@ func (b *Broadcaster) Subscribe() *Listener {
 		headerCh: make(chan struct{}),
 	}
 
-	// If headers already set, copy them and close headerCh
 	if b.headerSet {
 		l.headers = b.headers.Clone()
 		l.status = b.status
 		close(l.headerCh)
+	} else if b.done {
+		l.headerErr = b.err
+		if l.headerErr == nil {
+			l.headerErr = errCompletedBeforeHeaders
+		}
+		close(l.headerCh)
+	}
+
+	if b.done {
+		if b.headerSet && b.err != nil {
+			l.ch <- Chunk{Err: b.err}
+		}
+		close(l.ch)
+		return l
 	}
 
 	b.listeners = append(b.listeners, l)
@@ -271,16 +296,34 @@ func (b *Broadcaster) Complete(err error) {
 
 	b.done = true
 	b.err = err
+	if !b.headerSet && b.err == nil {
+		b.err = errCompletedBeforeHeaders
+	}
 	close(b.doneCh)
 
-	// Send final chunk and close all listener channels.
+	// A fetch failure before SetHeaders has no chunk for WaitForHeaders to
+	// observe. Publish the terminal error through the same readiness channel so
+	// every pre-header listener can return it without waiting for cancellation.
+	if !b.headerSet {
+		for _, l := range b.listeners {
+			l.headerErr = b.err
+			select {
+			case <-l.headerCh:
+			default:
+				close(l.headerCh)
+			}
+		}
+	}
+
+	// Send final chunk and close all listener channels. After headers, errors
+	// remain in the chunk stream so listeners retain any already-queued prefix.
 	// Non-blocking send: if buffer is full, skip the error chunk.
 	// Correctness is guaranteed by the version preconditions on cache writes,
 	// which prevents stale data from being cached even if error delivery fails.
 	for _, l := range b.listeners {
 		if !l.disconnected {
 			select {
-			case l.ch <- Chunk{Err: err}:
+			case l.ch <- Chunk{Err: b.err}:
 			default:
 			}
 			close(l.ch)

proxy/broadcast/broadcaster_test.go

@@ -2,6 +2,7 @@ package broadcast
 
 import (
 	"context"
+	"errors"
 	"net/http"
 	"sync"
 	"sync/atomic"
@@ -280,6 +281,146 @@ func TestBroadcasterHeadersWinOverConcurrentCancellation(t *testing.T) {
 	}
 }
 
+func TestBroadcasterCompleteBeforeHeadersNotifiesListener(t *testing.T) {
+	b := NewBroadcaster(1)
+	listener := b.Subscribe()
+	wantErr := errors.New("upstream request failed")
+
+	b.Complete(wantErr)
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	assertPreHeaderError(t, listener, wantErr, ctx)
+}
+
+func TestBroadcasterLateSubscriberReceivesPreHeaderError(t *testing.T) {
+	b := NewBroadcaster(1)
+	wantErr := errors.New("upstream request failed")
+	b.Complete(wantErr)
+
+	listener := b.Subscribe()
+	if listener == nil {
+		t.Fatal("Subscribe returned nil for a completed non-streaming broadcast")
+	}
+	assertPreHeaderError(t, listener, wantErr, context.Background())
+}
+
+func TestBroadcasterLateSubscriberReceivesEmptyCompletion(t *testing.T) {
+	b := NewBroadcaster(1)
+	headers := http.Header{"Content-Length": []string{"0"}}
+	b.SetHeaders(http.StatusOK, headers)
+	b.Complete(nil)
+
+	listener := b.Subscribe()
+	if listener == nil {
+		t.Fatal("Subscribe returned nil for a completed empty broadcast")
+	}
+	status, gotHeaders, err := listener.WaitForHeaders(context.Background())
+	if err != nil {
+		t.Fatalf("WaitForHeaders: %v", err)
+	}
+	if status != http.StatusOK {
+		t.Errorf("status = %d, want %d", status, http.StatusOK)
+	}
+	if got := gotHeaders.Get("Content-Length"); got != "0" {
+		t.Errorf("Content-Length = %q, want 0", got)
+	}
+	select {
+	case chunk, ok := <-listener.Chunks():
+		if ok && (chunk.Err != nil || len(chunk.Data) != 0) {
+			t.Fatalf("unexpected chunk after empty completion: %+v", chunk)
+		}
+		if ok {
+			select {
+			case _, ok := <-listener.Chunks():
+				if ok {
+					t.Fatal("listener received more than one terminal chunk")
+				}
+			default:
+				t.Fatal("completed listener channel remains open")
+			}
+		}
+	default:
+		t.Fatal("completed listener channel remains open")
+	}
+}
+
+func TestBroadcasterLateSubscriberReceivesPostHeaderError(t *testing.T) {
+	b := NewBroadcaster(1)
+	headers := http.Header{"Content-Length": []string{"3"}}
+	wantErr := errors.New("upstream body failed")
+	b.SetHeaders(http.StatusOK, headers)
+	b.Complete(wantErr)
+
+	listener := b.Subscribe()
+	if listener == nil {
+		t.Fatal("Subscribe returned nil for a completed non-streaming broadcast")
+	}
+	status, gotHeaders, err := listener.WaitForHeaders(context.Background())
+	if err != nil {
+		t.Fatalf("WaitForHeaders returned %v, want headers before body error", err)
+	}
+	if status != http.StatusOK || gotHeaders.Get("Content-Length") != "3" {
+		t.Fatalf("headers = %d/%v, want 200 and Content-Length 3", status, gotHeaders)
+	}
+	chunk, ok := <-listener.Chunks()
+	if !ok || !errors.Is(chunk.Err, wantErr) {
+		t.Fatalf("terminal chunk = %+v, open=%v, want %v", chunk, ok, wantErr)
+	}
+	if _, ok := <-listener.Chunks(); ok {
+		t.Fatal("listener channel remains open after terminal error")
+	}
+}
+
+func TestBroadcasterCompleteAfterPrefixKeepsHeadersAndErrorInChunkOrder(t *testing.T) {
+	b := NewBroadcaster(2)
+	listener := b.Subscribe()
+	headers := http.Header{"Content-Length": []string{"6"}}
+	wantErr := errors.New("upstream body failed")
+	b.SetHeaders(http.StatusOK, headers)
+	b.Broadcast([]byte("foo"))
+	b.Complete(wantErr)
+
+	status, gotHeaders, err := listener.WaitForHeaders(context.Background())
+	if err != nil {
+		t.Fatalf("WaitForHeaders returned %v after headers were published", err)
+	}
+	if status != http.StatusOK || gotHeaders.Get("Content-Length") != "6" {
+		t.Fatalf("headers = %d/%v, want 200 and Content-Length 6", status, gotHeaders)
+	}
+	prefix, ok := <-listener.Chunks()
+	if !ok || string(prefix.Data) != "foo" || prefix.Err != nil {
+		t.Fatalf("first chunk = %+v, open=%v, want body prefix", prefix, ok)
+	}
+	prefix.Release()
+	terminal, ok := <-listener.Chunks()
+	if !ok || !errors.Is(terminal.Err, wantErr) {
+		t.Fatalf("terminal chunk = %+v, open=%v, want %v", terminal, ok, wantErr)
+	}
+	if _, ok := <-listener.Chunks(); ok {
+		t.Fatal("listener channel remains open after terminal error")
+	}
+}
+
+func assertPreHeaderError(t *testing.T, listener *Listener, wantErr error, ctx context.Context) {
+	t.Helper()
+
+	select {
+	case <-listener.headerCh:
+	default:
+		t.Fatal("pre-header completion did not notify the listener")
+	}
+	if _, _, err := listener.WaitForHeaders(ctx); !errors.Is(err, wantErr) {
+		t.Fatalf("WaitForHeaders error = %v, want %v", err, wantErr)
+	}
+	for chunk := range listener.Chunks() {
+		if chunk.Err != nil && !errors.Is(chunk.Err, wantErr) {
+			t.Errorf("terminal chunk error = %v, want %v", chunk.Err, wantErr)
+		}
+		chunk.Release()
+	}
+}
+
 func TestManagerGetOrCreate(t *testing.T) {
 	m := NewManager(DefaultChannelBuffer)