diff --git a/README.md b/README.md
index c8d6fdb7..1e06d346 100644
--- a/README.md
+++ b/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:
diff --git a/docs/architecture.md b/docs/architecture.md
index 092de7b4..81b67de3 100644
--- a/docs/architecture.md
+++ b/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
diff --git a/docs/tiered-mode.md b/docs/tiered-mode.md
index b2d2304c..0ee190c5 100644
--- a/docs/tiered-mode.md
+++ b/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
diff --git a/proxy/broadcast/broadcaster.go b/proxy/broadcast/broadcaster.go
index c0374b4c..28dd50b5 100644
--- a/proxy/broadcast/broadcaster.go
+++ b/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)
diff --git a/proxy/broadcast/broadcaster_test.go b/proxy/broadcast/broadcaster_test.go
index 8a63b844..c2e868d8 100644
--- a/proxy/broadcast/broadcaster_test.go
+++ b/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)
 
diff --git a/proxy/cache_hit_stream_test.go b/proxy/cache_hit_stream_test.go
index 891f06b9..f45a88e0 100644
--- a/proxy/cache_hit_stream_test.go
+++ b/proxy/cache_hit_stream_test.go
@@ -7,6 +7,7 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"strconv"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -23,6 +24,39 @@ type testStreamCacheClient struct {
 	stream func(context.Context, string, io.Writer) error
 }
 
+type originBodyReader struct {
+	io.ReadCloser
+	onError func(error)
+}
+
+func (r *originBodyReader) Read(p []byte) (int, error) {
+	n, err := r.ReadCloser.Read(p)
+	if err != nil && !errors.Is(err, io.EOF) && r.onError != nil {
+		r.onError(err)
+	}
+	return n, err
+}
+
+type commitTrackingResponseWriter struct {
+	*httptest.ResponseRecorder
+	committed bool
+}
+
+func (w *commitTrackingResponseWriter) WriteHeader(statusCode int) {
+	w.committed = true
+	w.ResponseRecorder.WriteHeader(statusCode)
+}
+
+func (w *commitTrackingResponseWriter) Write(p []byte) (int, error) {
+	w.committed = true
+	return w.ResponseRecorder.Write(p)
+}
+
+func (w *commitTrackingResponseWriter) Flush() {
+	w.committed = true
+	w.ResponseRecorder.Flush()
+}
+
 func (c *testStreamCacheClient) GetStream(ctx context.Context, key string, w io.Writer) error {
 	return c.stream(ctx, key, w)
 }
@@ -50,6 +84,30 @@ func newLargeStreamCacheService(t *testing.T, body []byte, stream func(context.C
 	return NewService(&mockForwarder{}, store, cfg), store, meta
 }
 
+func newWholeObjectCacheService(t *testing.T, objectBody, cacheBody []byte) (*Service, *cache.CachedObjectMeta) {
+	t.Helper()
+
+	cfg := &config.Config{
+		Cache: config.CacheConfig{SizeThreshold: config.DefaultCacheSizeThreshold},
+	}
+	cfg.Cache.SetBlockCachingEnabled(false)
+	store := cache.NewCacheWithClient(cacheclient.NewMemoryCache(), &cfg.Cache)
+	meta := &cache.CachedObjectMeta{
+		Bucket:        "bucket",
+		Key:           "key",
+		ETag:          `"etag"`,
+		ContentType:   "text/plain",
+		ContentLength: int64(len(objectBody)),
+		BlockSize:     0,
+		StatusCode:    http.StatusOK,
+	}
+	if err := store.PutWithMeta(context.Background(), meta.Bucket, meta.Key, meta, cacheBody, 0); err != nil {
+		t.Fatalf("seed cache: %v", err)
+	}
+
+	return NewService(&mockForwarder{}, store, cfg), meta
+}
+
 func TestServeFromCache_LargeObject(t *testing.T) {
 	body := bytes.Repeat([]byte("cached body"), 8192)
 	stream := func(_ context.Context, _ string, w io.Writer) error {
@@ -66,7 +124,7 @@ func TestServeFromCache_LargeObject(t *testing.T) {
 	svc, _, meta := newLargeStreamCacheService(t, body, stream)
 
 	w := httptest.NewRecorder()
-	if err := svc.serveFromCache(context.Background(), w, meta.Bucket, meta.Key, meta, time.Now()); err != nil {
+	if err := svc.serveFromCache(context.Background(), w, meta.Bucket, meta.Key, meta, time.Now(), func() {}); err != nil {
 		t.Fatalf("serveFromCache() error = %v", err)
 	}
 	if w.Code != http.StatusOK {
@@ -89,7 +147,7 @@ func TestServeFromCache_LargeObjectUncommittedBeforeFirstByte(t *testing.T) {
 	svc, _, meta := newLargeStreamCacheService(t, body, stream)
 
 	w := httptest.NewRecorder()
-	err := svc.serveFromCache(context.Background(), w, meta.Bucket, meta.Key, meta, time.Now())
+	err := svc.serveFromCache(context.Background(), w, meta.Bucket, meta.Key, meta, time.Now(), func() {})
 	if err == nil || !errors.Is(err, wantErr) {
 		t.Fatalf("serveFromCache() error = %v, want wrapped %v", err, wantErr)
 	}
@@ -191,13 +249,551 @@ func TestHandleGetObject_LargeCacheHitPostCommitErrorDoesNotFallBack(t *testing.
 	}
 }
 
+func TestHandleGetObject_CacheBodyLength(t *testing.T) {
+	largeBody := bytes.Repeat([]byte("cached body"), 8192)
+	smallBody := []byte("small cache body")
+	originBody := []byte("complete origin body")
+
+	tests := []struct {
+		name              string
+		objectBody        []byte
+		cacheBody         []byte
+		wantBody          []byte
+		wantCacheStatus   string
+		wantContentLength int
+		wantUpstreamCalls int32
+		wantSource        string
+		wantRequestStatus string
+	}{
+		{
+			name:              "short-small",
+			objectBody:        smallBody,
+			cacheBody:         smallBody[:5],
+			wantBody:          originBody,
+			wantCacheStatus:   XCacheMiss,
+			wantContentLength: len(originBody),
+			wantUpstreamCalls: 1,
+			wantSource:        metrics.SourceUpstream,
+			wantRequestStatus: "success",
+		},
+		{
+			name:              "short-large",
+			objectBody:        largeBody,
+			cacheBody:         largeBody[:32768],
+			wantBody:          largeBody[:32768],
+			wantCacheStatus:   XCacheHit,
+			wantContentLength: len(largeBody),
+			wantUpstreamCalls: 0,
+			wantSource:        metrics.SourceLocal,
+			wantRequestStatus: "error",
+		},
+		{
+			name:              "exact-small",
+			objectBody:        smallBody,
+			cacheBody:         smallBody,
+			wantBody:          smallBody,
+			wantCacheStatus:   XCacheHit,
+			wantContentLength: len(smallBody),
+			wantSource:        metrics.SourceLocal,
+			wantRequestStatus: "success",
+		},
+		{
+			name:              "exact-large",
+			objectBody:        largeBody,
+			cacheBody:         largeBody,
+			wantBody:          largeBody,
+			wantCacheStatus:   XCacheHit,
+			wantContentLength: len(largeBody),
+			wantSource:        metrics.SourceLocal,
+			wantRequestStatus: "success",
+		},
+		{
+			name:              "zero-length",
+			wantCacheStatus:   XCacheHit,
+			wantContentLength: 0,
+			wantSource:        metrics.SourceLocal,
+			wantRequestStatus: "success",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			svc, _ := newWholeObjectCacheService(t, tt.objectBody, tt.cacheBody)
+			var upstreamCalls atomic.Int32
+			originRequestErrors := make(chan error, 1)
+			originBodyErrors := make(chan error, 1)
+			recordOriginError := func(dst chan<- error, err error) {
+				if err != nil {
+					select {
+					case dst <- err:
+					default:
+					}
+				}
+			}
+			origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+				upstreamCalls.Add(1)
+				w.Header().Set("Content-Length", strconv.Itoa(len(originBody)))
+				w.Header().Set("Content-Type", "text/plain")
+				w.Header().Set("ETag", `"origin"`)
+				w.WriteHeader(http.StatusOK)
+				_, _ = w.Write(originBody)
+			}))
+			defer origin.Close()
+			svc.forwarder = &mockForwarder{
+				doRequestFunc: func(ctx context.Context, r *http.Request, _, _ string) (*http.Response, error) {
+					upstreamReq, err := http.NewRequestWithContext(ctx, r.Method, origin.URL+r.URL.RequestURI(), r.Body)
+					if err != nil {
+						recordOriginError(originRequestErrors, err)
+						return nil, err
+					}
+					upstreamReq.Header = r.Header.Clone()
+					resp, err := http.DefaultClient.Do(upstreamReq)
+					if err != nil {
+						recordOriginError(originRequestErrors, err)
+						return nil, err
+					}
+					resp.Body = &originBodyReader{
+						ReadCloser: resp.Body,
+						onError:    func(err error) { recordOriginError(originBodyErrors, err) },
+					}
+					return resp, nil
+				},
+			}
+
+			requestLabels := []struct {
+				status string
+				source string
+			}{
+				{status: "success", source: metrics.SourceLocal},
+				{status: "error", source: metrics.SourceLocal},
+				{status: "success", source: metrics.SourceUpstream},
+				{status: "error", source: metrics.SourceUpstream},
+			}
+			beforeRequests := make([]float64, len(requestLabels))
+			for i, label := range requestLabels {
+				beforeRequests[i] = requestCountFrom(t, label.status, label.source)
+			}
+
+			w := httptest.NewRecorder()
+			r := httptest.NewRequest(http.MethodGet, "/bucket/key", nil)
+			err := svc.HandleGetObject(w, r)
+			if tt.wantUpstreamCalls > 0 {
+				select {
+				case fixtureErr := <-originRequestErrors:
+					t.Skipf("local origin request failed before returning a response: %v", fixtureErr)
+				default:
+				}
+				select {
+				case fixtureErr := <-originBodyErrors:
+					t.Skipf("local origin body failed before a complete response: %v", fixtureErr)
+				default:
+				}
+			}
+			if err != nil {
+				t.Errorf("whole-object GET %s returned error %v; want nil", tt.name, err)
+			}
+			response := w.Result()
+			if response.StatusCode != http.StatusOK {
+				t.Errorf("whole-object GET %s returned status %d; want %d", tt.name, response.StatusCode, http.StatusOK)
+			}
+			if got := response.Header.Get(XCacheHeader); got != tt.wantCacheStatus {
+				t.Errorf("whole-object GET %s returned %s %q; want %q", tt.name, XCacheHeader, got, tt.wantCacheStatus)
+			}
+			if got := response.Header.Get("Content-Length"); got != strconv.Itoa(tt.wantContentLength) {
+				t.Errorf("whole-object GET %s advertised Content-Length %q; want %d", tt.name, got, tt.wantContentLength)
+			}
+			if !bytes.Equal(w.Body.Bytes(), tt.wantBody) {
+				t.Errorf("whole-object GET %s returned %d body bytes; want %d", tt.name, w.Body.Len(), len(tt.wantBody))
+			}
+			if got := upstreamCalls.Load(); got != tt.wantUpstreamCalls {
+				t.Errorf("whole-object GET %s made %d upstream calls; want %d", tt.name, got, tt.wantUpstreamCalls)
+			}
+			for i, label := range requestLabels {
+				wantDelta := float64(0)
+				if label.source == tt.wantSource && label.status == tt.wantRequestStatus {
+					wantDelta = 1
+				}
+				if got := requestCountFrom(t, label.status, label.source); got != beforeRequests[i]+wantDelta {
+					t.Errorf("whole-object GET %s request count for %s/%s = %v; want delta %v", tt.name, label.source, label.status, got-beforeRequests[i], wantDelta)
+				}
+			}
+		})
+	}
+
+	// Tiered mode has no upstream GET for a local-tier object. A short buffered
+	// body is a miss for the caller's authoritative store, while a committed
+	// stream remains incomplete and must be accounted as an error.
+	tieredTests := []struct {
+		name              string
+		objectBody        []byte
+		cacheBody         []byte
+		wantStatus        int
+		wantCacheStatus   string
+		wantContentLength string
+		wantBody          []byte
+		wantNoSuchKey     bool
+	}{
+		{
+			name:              "tiered-short-small",
+			objectBody:        smallBody,
+			cacheBody:         smallBody[:5],
+			wantStatus:        http.StatusNotFound,
+			wantCacheStatus:   XCacheMiss,
+			wantContentLength: "",
+			wantNoSuchKey:     true,
+		},
+		{
+			name:              "tiered-short-large",
+			objectBody:        largeBody,
+			cacheBody:         largeBody[:32768],
+			wantStatus:        http.StatusOK,
+			wantCacheStatus:   XCacheHit,
+			wantContentLength: strconv.Itoa(len(largeBody)),
+			wantBody:          largeBody[:32768],
+		},
+	}
+	for _, tt := range tieredTests {
+		t.Run(tt.name, func(t *testing.T) {
+			mock, forwards, _ := tieredMock()
+			svc, store := newTieredTestService(mock, int64(len(tt.objectBody)+1))
+			if w := tieredDo(t, svc, http.MethodPut, "/bucket/key", string(tt.objectBody), nil); w.Code != http.StatusOK {
+				t.Fatalf("tiered cache setup PUT status = %d", w.Code)
+			}
+			meta, found, err := store.GetMeta(context.Background(), "bucket", "key")
+			if err != nil || !found || meta == nil || meta.BodyRef == "" {
+				t.Fatalf("tiered cache setup metadata = %+v found=%v err=%v", meta, found, err)
+			}
+			if err := store.PutBodyStream(context.Background(), "bucket", "key", meta.BodyRef, bytes.NewReader(tt.cacheBody), 0); err != nil {
+				t.Fatalf("tiered cache setup short body: %v", err)
+			}
+
+			beforeLocalSuccess := requestCountFrom(t, "success", metrics.SourceLocal)
+			beforeLocalErrors := requestCountFrom(t, "error", metrics.SourceLocal)
+			beforeForwards := forwards.Load()
+			w := tieredDo(t, svc, http.MethodGet, "/bucket/key", "", nil)
+			if w.Code != tt.wantStatus {
+				t.Errorf("whole-object GET %s returned status %d; want %d", tt.name, w.Code, tt.wantStatus)
+			}
+			if got := w.Header().Get(XCacheHeader); got != tt.wantCacheStatus {
+				t.Errorf("whole-object GET %s returned %s %q; want %q", tt.name, XCacheHeader, got, tt.wantCacheStatus)
+			}
+			if got := w.Header().Get("Content-Length"); got != tt.wantContentLength {
+				t.Errorf("whole-object GET %s advertised Content-Length %q; want %q", tt.name, got, tt.wantContentLength)
+			}
+			if tt.wantNoSuchKey {
+				if !bytes.Contains(w.Body.Bytes(), []byte("<Code>NoSuchKey</Code>")) {
+					t.Errorf("whole-object GET %s did not return the tiered NoSuchKey miss: %q", tt.name, w.Body.String())
+				}
+				if bytes.Contains(w.Body.Bytes(), tt.cacheBody) {
+					t.Errorf("whole-object GET %s exposed cached prefix %q", tt.name, tt.cacheBody)
+				}
+			} else if !bytes.Equal(w.Body.Bytes(), tt.wantBody) {
+				t.Errorf("whole-object GET %s returned %d body bytes; want %d", tt.name, w.Body.Len(), len(tt.wantBody))
+			}
+			if got := forwards.Load(); got != beforeForwards {
+				t.Errorf("whole-object GET %s made %d upstream calls; want none", tt.name, got-beforeForwards)
+			}
+			if got := requestCountFrom(t, "error", metrics.SourceLocal); got != beforeLocalErrors+1 {
+				t.Errorf("whole-object GET %s local error count delta = %v; want 1", tt.name, got-beforeLocalErrors)
+			}
+			if got := requestCountFrom(t, "success", metrics.SourceLocal); got != beforeLocalSuccess {
+				t.Errorf("whole-object GET %s local success count delta = %v; want 0", tt.name, got-beforeLocalSuccess)
+			}
+		})
+	}
+}
+
+func TestServeFromCache_OverlongBodyIsNotSuccessful(t *testing.T) {
+	smallObject := []byte("small object")
+	largeObject := bytes.Repeat([]byte("cached body"), 8192)
+	tests := []struct {
+		name              string
+		objectBody        []byte
+		cacheBody         []byte
+		wantErr           bool
+		wantBody          []byte
+		wantCacheStatus   string
+		wantContentLength int
+		wantRequestStatus string
+	}{
+		{
+			name:       "small",
+			objectBody: smallObject,
+			cacheBody:  append(append([]byte(nil), smallObject...), []byte(" extra")...),
+			wantErr:    true,
+		},
+		{
+			name:              "large",
+			objectBody:        largeObject,
+			cacheBody:         append(append([]byte(nil), largeObject...), []byte(" extra")...),
+			wantBody:          append(append([]byte(nil), largeObject...), []byte(" extra")...),
+			wantCacheStatus:   XCacheHit,
+			wantContentLength: len(largeObject),
+			wantRequestStatus: "error",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			svc, meta := newWholeObjectCacheService(t, tt.objectBody, tt.cacheBody)
+			beforeLocalSuccess := requestCountFrom(t, "success", metrics.SourceLocal)
+			beforeLocalErrors := requestCountFrom(t, "error", metrics.SourceLocal)
+			w := &commitTrackingResponseWriter{ResponseRecorder: httptest.NewRecorder()}
+			err := svc.serveFromCache(context.Background(), w, meta.Bucket, meta.Key, meta, time.Now(), func() {
+				svc.invalidateStaleMeta(meta.Bucket, meta.Key, meta.ETag)
+			})
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("serveFromCache() error = %v, want error %v", err, tt.wantErr)
+			}
+			if tt.wantErr {
+				if w.committed || w.Body.Len() != 0 || w.Header().Get(XCacheHeader) != "" {
+					t.Errorf("short-path response committed: headers=%v body=%q", w.Header(), w.Body.String())
+				}
+				if got := requestCountFrom(t, "success", metrics.SourceLocal); got != beforeLocalSuccess {
+					t.Errorf("local success count delta = %v, want 0", got-beforeLocalSuccess)
+				}
+				return
+			}
+			if !w.committed {
+				t.Fatal("large-path response was not committed")
+			}
+			if got := w.Header().Get(XCacheHeader); got != tt.wantCacheStatus {
+				t.Errorf("%s = %q, want %q", XCacheHeader, got, tt.wantCacheStatus)
+			}
+			if got := w.Header().Get("Content-Length"); got != strconv.Itoa(tt.wantContentLength) {
+				t.Errorf("Content-Length = %q, want %d", got, tt.wantContentLength)
+			}
+			if !bytes.Equal(w.Body.Bytes(), tt.wantBody) {
+				t.Errorf("body length = %d, want %d", w.Body.Len(), len(tt.wantBody))
+			}
+			if got := requestCountFrom(t, tt.wantRequestStatus, metrics.SourceLocal); got != beforeLocalErrors+1 {
+				t.Errorf("local error count delta = %v, want 1", got-beforeLocalErrors)
+			}
+			if got := requestCountFrom(t, "success", metrics.SourceLocal); got != beforeLocalSuccess {
+				t.Errorf("local success count delta = %v, want 0", got-beforeLocalSuccess)
+			}
+			if !tt.wantErr {
+				if _, found, err := svc.cache.GetMeta(context.Background(), meta.Bucket, meta.Key); err != nil || found {
+					t.Errorf("overlong cache metadata remains: found=%v err=%v", found, err)
+				}
+			}
+		})
+	}
+}
+
+func TestHandleGetObject_LargeShortCacheBodyInvalidatesBeforeNextGet(t *testing.T) {
+	body := bytes.Repeat([]byte("cached body"), 8192)
+	prefix := body[:32768]
+	svc, meta := newWholeObjectCacheService(t, body, prefix)
+	var upstreamCalls atomic.Int32
+	svc.forwarder = &mockForwarder{
+		doRequestFunc: func(context.Context, *http.Request, string, string) (*http.Response, error) {
+			upstreamCalls.Add(1)
+			return &http.Response{
+				StatusCode: http.StatusOK,
+				Header: http.Header{
+					"Content-Length": []string{strconv.Itoa(len(body))},
+					"Content-Type":   []string{"text/plain"},
+					"ETag":           []string{`"refetched"`},
+				},
+				Body:          io.NopCloser(bytes.NewReader(body)),
+				ContentLength: int64(len(body)),
+			}, nil
+		},
+	}
+
+	beforeLocalErrors := requestCountFrom(t, "error", metrics.SourceLocal)
+	beforeLocalSuccesses := requestCountFrom(t, "success", metrics.SourceLocal)
+	beforeUpstreamSuccesses := requestCountFrom(t, "success", metrics.SourceUpstream)
+	first := httptest.NewRecorder()
+	if err := svc.HandleGetObject(first, httptest.NewRequest(http.MethodGet, "/bucket/key", nil)); err != nil {
+		t.Fatalf("first HandleGetObject() error = %v", err)
+	}
+	if first.Code != http.StatusOK || first.Header().Get(XCacheHeader) != XCacheHit {
+		t.Fatalf("first response = %d X-Cache %q, want 200 HIT", first.Code, first.Header().Get(XCacheHeader))
+	}
+	if first.Header().Get("Content-Length") != strconv.Itoa(len(body)) || !bytes.Equal(first.Body.Bytes(), prefix) {
+		t.Fatalf("first committed response length=%q body_bytes=%d, want length %d and prefix %d", first.Header().Get("Content-Length"), first.Body.Len(), len(body), len(prefix))
+	}
+	if got := upstreamCalls.Load(); got != 0 {
+		t.Fatalf("first GET made %d upstream calls, want 0 after commitment", got)
+	}
+	if _, found, err := svc.cache.GetMeta(context.Background(), meta.Bucket, meta.Key); err != nil || found {
+		t.Fatalf("mismatched cache metadata remains after committed response: found=%v err=%v", found, err)
+	}
+	if got := requestCountFrom(t, "error", metrics.SourceLocal); got != beforeLocalErrors+1 {
+		t.Errorf("first local error count delta = %v, want 1", got-beforeLocalErrors)
+	}
+	if got := requestCountFrom(t, "success", metrics.SourceLocal); got != beforeLocalSuccesses {
+		t.Errorf("first local success count delta = %v, want 0", got-beforeLocalSuccesses)
+	}
+
+	second := httptest.NewRecorder()
+	if err := svc.HandleGetObject(second, httptest.NewRequest(http.MethodGet, "/bucket/key", nil)); err != nil {
+		t.Fatalf("second HandleGetObject() error = %v", err)
+	}
+	if second.Code != http.StatusOK || second.Header().Get(XCacheHeader) != XCacheMiss {
+		t.Fatalf("second response = %d X-Cache %q, want 200 MISS", second.Code, second.Header().Get(XCacheHeader))
+	}
+	if second.Header().Get("Content-Length") != strconv.Itoa(len(body)) || !bytes.Equal(second.Body.Bytes(), body) {
+		t.Fatalf("second response length=%q body_bytes=%d, want %d complete bytes", second.Header().Get("Content-Length"), second.Body.Len(), len(body))
+	}
+	if got := upstreamCalls.Load(); got != 1 {
+		t.Errorf("total upstream calls = %d, want 1", got)
+	}
+	if got := requestCountFrom(t, "success", metrics.SourceUpstream); got != beforeUpstreamSuccesses+1 {
+		t.Errorf("upstream success count delta = %v, want 1", got-beforeUpstreamSuccesses)
+	}
+}
+
+func TestHandleGetObject_RevalidationLargeShortBodyInvalidatesMetadata(t *testing.T) {
+	body := bytes.Repeat([]byte("cached body"), 8192)
+	prefix := body[:32768]
+	tests := []struct {
+		name            string
+		conditionalResp *http.Response
+		conditionalErr  error
+	}{
+		{
+			name: "not-modified",
+			conditionalResp: &http.Response{
+				StatusCode: http.StatusNotModified,
+				Header:     make(http.Header),
+				Body:       io.NopCloser(bytes.NewReader(nil)),
+			},
+		},
+		{
+			name:           "stale-after-error",
+			conditionalErr: errors.New("upstream revalidation failed"),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			svc, meta := newWholeObjectCacheService(t, body, prefix)
+			var fallbackCalls atomic.Int32
+			forwarder := &mockForwarder{
+				conditionalResp: tt.conditionalResp,
+				conditionalErr:  tt.conditionalErr,
+				forwardFunc: func(context.Context, http.ResponseWriter, *http.Request) error {
+					fallbackCalls.Add(1)
+					return errors.New("unexpected fallback after committed cache response")
+				},
+			}
+			svc.forwarder = forwarder
+			beforeErrors := requestCountFrom(t, "error", metrics.SourceLocal)
+			beforeSuccesses := requestCountFrom(t, "success", metrics.SourceLocal)
+			w := httptest.NewRecorder()
+			r := httptest.NewRequest(http.MethodGet, "/bucket/key", nil)
+			r.Header.Set("Cache-Control", "no-cache")
+			if err := svc.HandleGetObject(w, r); err != nil {
+				t.Fatalf("HandleGetObject() error = %v", err)
+			}
+			if !forwarder.conditionalCalled {
+				t.Fatal("HandleGetObject did not take the revalidation path")
+			}
+			if got := fallbackCalls.Load(); got != 0 {
+				t.Errorf("post-commit fallback calls = %d, want 0", got)
+			}
+			if w.Code != http.StatusOK || w.Header().Get(XCacheHeader) != XCacheHit || w.Header().Get("Content-Length") != strconv.Itoa(len(body)) || !bytes.Equal(w.Body.Bytes(), prefix) {
+				t.Errorf("committed response = %d headers=%v body_bytes=%d, want HIT with short prefix and recorded length", w.Code, w.Header(), w.Body.Len())
+			}
+			if _, found, err := svc.cache.GetMeta(context.Background(), meta.Bucket, meta.Key); err != nil || found {
+				t.Errorf("mismatched metadata remains after revalidation serve: found=%v err=%v", found, err)
+			}
+			if got := requestCountFrom(t, "error", metrics.SourceLocal); got != beforeErrors+1 {
+				t.Errorf("local error count delta = %v, want 1", got-beforeErrors)
+			}
+			if got := requestCountFrom(t, "success", metrics.SourceLocal); got != beforeSuccesses {
+				t.Errorf("local success count delta = %v, want 0", got-beforeSuccesses)
+			}
+		})
+	}
+}
+
+func TestHandleGetObject_ShortCacheBodyUpstreamFailureBeforeCommit(t *testing.T) {
+	objectBody := []byte("complete object body")
+	cacheBody := objectBody[:5]
+	svc, _ := newWholeObjectCacheService(t, objectBody, cacheBody)
+	wantErr := errors.New("upstream request failed")
+	var upstreamCalls atomic.Int32
+	svc.forwarder = &mockForwarder{
+		doRequestFunc: func(context.Context, *http.Request, string, string) (*http.Response, error) {
+			upstreamCalls.Add(1)
+			return nil, wantErr
+		},
+	}
+
+	beforeLocalSuccess := requestCountFrom(t, "success", metrics.SourceLocal)
+	beforeUpstreamErrors := requestCountFrom(t, "error", metrics.SourceUpstream)
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	w := &commitTrackingResponseWriter{ResponseRecorder: httptest.NewRecorder()}
+	r := httptest.NewRequest(http.MethodGet, "/bucket/key", nil).WithContext(ctx)
+	err := svc.HandleGetObject(w, r)
+
+	if !errors.Is(err, wantErr) {
+		t.Fatalf("HandleGetObject() error = %v, want %v", err, wantErr)
+	}
+	if got := upstreamCalls.Load(); got != 1 {
+		t.Errorf("upstream calls = %d, want 1", got)
+	}
+	if w.committed || w.Body.Len() != 0 || w.Header().Get(XCacheHeader) != "" {
+		t.Errorf("response committed before the upstream error: committed=%v code=%d headers=%v body=%q", w.committed, w.Code, w.Header(), w.Body.String())
+	}
+	if got := requestCountFrom(t, "success", metrics.SourceLocal); got != beforeLocalSuccess {
+		t.Errorf("local success count delta = %v, want 0", got-beforeLocalSuccess)
+	}
+	if got := requestCountFrom(t, "error", metrics.SourceUpstream); got != beforeUpstreamErrors+1 {
+		t.Errorf("upstream error count delta = %v, want 1", got-beforeUpstreamErrors)
+	}
+}
+
+func TestHandleGetObject_CompletedPreHeaderBroadcastReturnsError(t *testing.T) {
+	wantErr := errors.New("upstream request failed")
+	var upstreamCalls atomic.Int32
+	forwarder := &mockForwarder{
+		doRequestFunc: func(context.Context, *http.Request, string, string) (*http.Response, error) {
+			upstreamCalls.Add(1)
+			return nil, wantErr
+		},
+	}
+	svc, _ := newTestService(forwarder, true)
+	broadcastKey := makeBroadcastKey("bucket", "key", "")
+	broadcaster, first := svc.broadcastManager.GetOrCreate(broadcastKey)
+	if !first {
+		t.Fatal("initial GetOrCreate did not create a broadcaster")
+	}
+	broadcaster.Complete(wantErr)
+	defer svc.broadcastManager.Remove(broadcastKey)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	w := &commitTrackingResponseWriter{ResponseRecorder: httptest.NewRecorder()}
+	r := httptest.NewRequest(http.MethodGet, "/bucket/key", nil).WithContext(ctx)
+	err := svc.HandleGetObject(w, r)
+
+	if !errors.Is(err, wantErr) {
+		t.Fatalf("HandleGetObject() error = %v, want %v", err, wantErr)
+	}
+	if got := upstreamCalls.Load(); got != 0 {
+		t.Errorf("late listener started %d upstream fetches, want 0", got)
+	}
+	if w.committed || w.Body.Len() != 0 || w.Header().Get(XCacheHeader) != "" {
+		t.Errorf("response committed before the broadcast error: committed=%v code=%d headers=%v body=%q", w.committed, w.Code, w.Header(), w.Body.String())
+	}
+}
+
 func requestCount(t *testing.T, status string) float64 {
+	return requestCountFrom(t, status, metrics.SourceLocal)
+}
+
+func requestCountFrom(t *testing.T, status, source string) float64 {
 	t.Helper()
 
 	var metric dto.Metric
-	// mode defaults to "unknown" in tests (SetMode is a main.go concern); the
-	// cache-hit paths this test exercises record source=local.
-	if err := metrics.RequestsTotal.WithLabelValues("GetObject", status, "unknown", metrics.SourceLocal).Write(&metric); err != nil {
+	// mode defaults to "unknown" in tests (SetMode is a main.go concern).
+	if err := metrics.RequestsTotal.WithLabelValues("GetObject", status, "unknown", source).Write(&metric); err != nil {
 		t.Fatalf("read %s request count: %v", status, err)
 	}
 	return metric.GetCounter().GetValue()
diff --git a/proxy/get_object.go b/proxy/get_object.go
index ed3c1731..d51c4613 100644
--- a/proxy/get_object.go
+++ b/proxy/get_object.go
@@ -219,7 +219,9 @@ func (s *Service) HandleGetObject(w http.ResponseWriter, r *http.Request) error
 						return assembleErr
 					}
 					// Fall through to cache miss path
-				} else if cacheBodyErr := s.serveFromCache(ctx, w, bucket, key, meta, start); cacheBodyErr != nil {
+				} else if cacheBodyErr := s.serveFromCache(ctx, w, bucket, key, meta, start, func() {
+					s.invalidateStaleMeta(bucket, key, meta.ETag)
+				}); cacheBodyErr != nil {
 					log.Warn().Err(cacheBodyErr).Str("bucket", bucket).Str("key", key).Msg("Cache body unavailable, falling through to upstream")
 					// Invalidate only when the body is genuinely gone: the metadata is then
 					// orphaned, and clearing it stops a meta-hit/body-miss loop that repeats
diff --git a/proxy/localstore.go b/proxy/localstore.go
index 762c8369..a9e95fb7 100644
--- a/proxy/localstore.go
+++ b/proxy/localstore.go
@@ -10,6 +10,7 @@ import (
 	"crypto/md5"
 	"encoding/base64"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -151,8 +152,14 @@ func (s *Service) serveOriginlessObject(w http.ResponseWriter, r *http.Request,
 		return s.finishServeBodyError(ctx, w, r, bucket, key, metaVersion, rangeErr, operation, start)
 	}
 
-	if bodyErr := s.serveFromCache(ctx, w, bucket, key, meta, start); bodyErr != nil {
-		// serveFromCache fails before committing headers, so a miss response is
+	if bodyErr := s.serveFromCache(ctx, w, bucket, key, meta, start, func() {
+		// The stream has committed, so cleanup is version-guarded but must not
+		// turn the partial response into a miss response.
+		if _, derr := s.cache.DeleteMetaIfVersion(ctx, bucket, key, metaVersion); derr != nil {
+			log.Debug().Err(derr).Str("bucket", bucket).Str("key", key).Msg("Orphaned-meta invalidation failed; ages out by TTL")
+		}
+	}); bodyErr != nil {
+		// serveFromCache errors before committing headers, so a miss response is
 		// still writable.
 		return s.finishServeBodyError(ctx, w, r, bucket, key, metaVersion, bodyErr, operation, start)
 	}
@@ -185,6 +192,11 @@ func (s *Service) finishServeBodyError(ctx context.Context, w http.ResponseWrite
 	if _, derr := s.cache.DeleteMetaIfVersion(ctx, bucket, key, metaVersion); derr != nil {
 		log.Debug().Err(derr).Str("bucket", bucket).Str("key", key).Msg("Orphaned-meta invalidation failed; ages out by TTL")
 	}
+	if errors.Is(serveErr, errCacheBodyLengthMismatch) {
+		// A damaged body still produces the mode's authoritative miss, but it
+		// must not be counted as a successful local object read.
+		return s.originlessMissWithRequestStatus(w, r, operation, start, "error")
+	}
 	return s.originlessMiss(w, r, operation, start)
 }
 
@@ -206,9 +218,15 @@ func (s *Service) entryServable(ctx context.Context, bucket, key string, meta *c
 
 // originlessMiss answers the one thing a miss can be in this mode.
 func (s *Service) originlessMiss(w http.ResponseWriter, r *http.Request, operation string, start time.Time) error {
+	return s.originlessMissWithRequestStatus(w, r, operation, start, "success")
+}
+
+// originlessMissWithRequestStatus emits the authoritative NoSuchKey response and
+// records the outcome that led to the miss.
+func (s *Service) originlessMissWithRequestStatus(w http.ResponseWriter, r *http.Request, operation string, start time.Time, requestStatus string) error {
 	writeCacheStatus(w, XCacheMiss)
 	s3err.WriteError(w, r, s3err.ErrNoSuchKey)
-	metrics.RecordRequest(operation, "success", metrics.SourceLocal, time.Since(start).Seconds())
+	metrics.RecordRequest(operation, requestStatus, metrics.SourceLocal, time.Since(start).Seconds())
 	return nil
 }
 
diff --git a/proxy/revalidation.go b/proxy/revalidation.go
index 245baed6..bae9b0fc 100644
--- a/proxy/revalidation.go
+++ b/proxy/revalidation.go
@@ -3,6 +3,7 @@ package proxy
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -14,6 +15,8 @@ import (
 	"github.com/tigrisdata/tag/metrics"
 )
 
+var errCacheBodyLengthMismatch = errors.New("cache body length mismatch")
+
 // revalidateAndServe sends a conditional GET to upstream and serves the result.
 // Supports both full-object and range requests. For range requests, the Range header
 // is included in the conditional GET so upstream returns 304 or 206.
@@ -147,13 +150,15 @@ func (s *Service) handleRevalidation304(
 	metrics.RecordRevalidationNotModified()
 	log.Debug().Str("bucket", bucket).Str("key", key).Msg("Revalidation 304 - object unchanged")
 
-	// Serve range or full body from cache. Both serveRangeFromCache (via its
-	// pre-header probe) and serveFromCache report an unresolvable body without
-	// writing headers, so the caller can safely forward to upstream.
+	// Serve range or full body from cache. The range helper reports served=true
+	// once it commits, and the whole-object helper returns nil after a committed
+	// stream; only a pre-commit failure leaves the caller free to fetch upstream.
 	if rangeHeader != "" {
 		return s.serveRangeFromCache(ctx, w, r, bucket, key, meta, rangeHeader, start)
 	}
-	if bodyErr := s.serveFromCache(ctx, w, bucket, key, meta, start); bodyErr != nil {
+	if bodyErr := s.serveFromCache(ctx, w, bucket, key, meta, start, func() {
+		s.invalidateStaleMeta(bucket, key, meta.ETag)
+	}); bodyErr != nil {
 		return false, bodyErr
 	}
 	return true, nil
@@ -311,14 +316,17 @@ func (s *Service) handleRevalidation200(
 	return copyErr
 }
 
-// serveFromCache serves an object from the cache body.
-// Used as fallback during revalidation (304 or error paths).
+// serveFromCache serves a whole object from the cache body. If a streamed response
+// has committed and the cache returns nil with a byte count different from its
+// positive ContentLength, invalidateOnLengthMismatch applies the caller's guarded
+// metadata invalidation without triggering an upstream fallback.
 func (s *Service) serveFromCache(
 	ctx context.Context,
 	w http.ResponseWriter,
 	bucket, key string,
 	meta *cache.CachedObjectMeta,
 	start time.Time,
+	invalidateOnLengthMismatch func(),
 ) error {
 	// Zero-byte objects: no body to serve
 	if meta.ContentLength == 0 {
@@ -333,6 +341,11 @@ func (s *Service) serveFromCache(
 
 		bodyErr := s.cache.GetBodyStream(ctx, bucket, key, meta.BodyDiscriminator(), bodyBuf)
 		if bodyErr == nil && bodyBuf.Len() > 0 {
+			if meta.ContentLength > 0 && int64(bodyBuf.Len()) != meta.ContentLength {
+				bodyLen := bodyBuf.Len()
+				putBuffer(bodyBuf)
+				return fmt.Errorf("%w for %s/%s: got %d bytes, want %d", errCacheBodyLengthMismatch, bucket, key, bodyLen, meta.ContentLength)
+			}
 			meta.WriteHeaders(w)
 			writeCacheStatus(w, XCacheHit)
 			w.WriteHeader(meta.StatusCode)
@@ -372,6 +385,16 @@ func (s *Service) serveFromCache(
 	if !cw.committed {
 		return fmt.Errorf("cache body empty for %s/%s", bucket, key)
 	}
+	if bodyErr == nil && meta.ContentLength > 0 && cw.written != meta.ContentLength {
+		log.Warn().
+			Str("bucket", bucket).
+			Str("key", key).
+			Int64("expected_bytes", meta.ContentLength).
+			Int64("written_bytes", cw.written).
+			Msg("Cache body length mismatch after headers committed")
+		status = "error"
+		invalidateOnLengthMismatch()
+	}
 
 	metrics.BytesTransferred.WithLabelValues("out").Add(float64(cw.written))
 	metrics.RecordRequest("GetObject", status, metrics.SourceLocal, time.Since(start).Seconds())
@@ -433,9 +456,10 @@ func (s *Service) handleRevalidation206Range(
 }
 
 // serveStaleFromCache serves stale content from cache, handling both full and
-// range requests. It returns served=false, without writing any response bytes,
-// when the cached body cannot be resolved so the caller can fall back to a direct
-// upstream fetch instead of emitting a truncated or empty response.
+// range requests. It returns served=false only when no response was committed,
+// allowing the caller to fetch upstream. A committed stream that later proves
+// incomplete is accounted for by the serve helper and cannot be replaced by an
+// upstream response.
 func (s *Service) serveStaleFromCache(
 	ctx context.Context,
 	w http.ResponseWriter,
@@ -448,7 +472,9 @@ func (s *Service) serveStaleFromCache(
 	if rangeHeader != "" {
 		return s.serveRangeFromCache(ctx, w, r, bucket, key, meta, rangeHeader, start)
 	}
-	if bodyErr := s.serveFromCache(ctx, w, bucket, key, meta, start); bodyErr != nil {
+	if bodyErr := s.serveFromCache(ctx, w, bucket, key, meta, start, func() {
+		s.invalidateStaleMeta(bucket, key, meta.ETag)
+	}); bodyErr != nil {
 		return false, bodyErr
 	}
 	return true, nil
diff --git a/proxy/revalidation_test.go b/proxy/revalidation_test.go
index be50dfeb..7900641f 100644
--- a/proxy/revalidation_test.go
+++ b/proxy/revalidation_test.go
@@ -499,7 +499,7 @@ func TestServeFromCache_ZeroByteObject(t *testing.T) {
 	}
 
 	w := httptest.NewRecorder()
-	err := svc.serveFromCache(context.Background(), w, "b", "k", meta, time.Now())
+	err := svc.serveFromCache(context.Background(), w, "b", "k", meta, time.Now(), func() {})
 	if err != nil {
 		t.Fatalf("serveFromCache() error = %v", err)
 	}
@@ -533,7 +533,7 @@ func TestServeFromCache_SmallObject(t *testing.T) {
 	_ = c.PutWithMeta(ctx, bucket, key, meta, body, 0)
 
 	w := httptest.NewRecorder()
-	err := svc.serveFromCache(ctx, w, bucket, key, meta, time.Now())
+	err := svc.serveFromCache(ctx, w, bucket, key, meta, time.Now(), func() {})
 	if err != nil {
 		t.Fatalf("serveFromCache() error = %v", err)
 	}
diff --git a/proxy/tiered_test.go b/proxy/tiered_test.go
index c3d5e8ae..f09da84d 100644
--- a/proxy/tiered_test.go
+++ b/proxy/tiered_test.go
@@ -10,6 +10,7 @@ import (
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -19,6 +20,7 @@ import (
 	cacheclient "github.com/tigrisdata/ocache/client"
 	"github.com/tigrisdata/tag/cache"
 	"github.com/tigrisdata/tag/config"
+	"github.com/tigrisdata/tag/metrics"
 )
 
 // newTieredTestService builds a Service in tiered mode with an in-memory cache
@@ -1081,6 +1083,141 @@ func TestTieredStreamedPutBodyRefRoundTrip(t *testing.T) {
 	}
 }
 
+func TestTieredLargeShortBodyInvalidatesForNextGet(t *testing.T) {
+	body := strings.Repeat("cached body", 8192)
+	prefix := body[:32768]
+	mock, forwards, _ := tieredMock()
+	svc, store := newTieredTestService(mock, int64(len(body)+1))
+	if w := tieredDo(t, svc, http.MethodPut, "/bucket/key", body, nil); w.Code != http.StatusOK {
+		t.Fatalf("local PUT status = %d", w.Code)
+	}
+	meta, found, err := store.GetMeta(context.Background(), "bucket", "key")
+	if err != nil || !found || meta == nil || meta.BodyRef == "" {
+		t.Fatalf("local metadata = %+v found=%v err=%v", meta, found, err)
+	}
+	if err := store.PutBodyStream(context.Background(), "bucket", "key", meta.BodyRef, strings.NewReader(prefix), 0); err != nil {
+		t.Fatalf("truncate local-tier body: %v", err)
+	}
+
+	beforeErrors := requestCountFrom(t, "error", metrics.SourceLocal)
+	first := tieredDo(t, svc, http.MethodGet, "/bucket/key", "", nil)
+	if first.Code != http.StatusOK || first.Header().Get(XCacheHeader) != XCacheHit || first.Header().Get("Content-Length") != strconv.Itoa(len(body)) || first.Body.String() != string(prefix) {
+		t.Fatalf("first GET = %d headers=%v body_bytes=%d, want HIT with short prefix", first.Code, first.Header(), first.Body.Len())
+	}
+	if _, found, err := store.GetMeta(context.Background(), "bucket", "key"); err != nil || found {
+		t.Fatalf("short local-tier metadata remains: found=%v err=%v", found, err)
+	}
+	if got := requestCountFrom(t, "error", metrics.SourceLocal); got != beforeErrors+1 {
+		t.Errorf("local error count delta = %v, want 1", got-beforeErrors)
+	}
+	if got := forwards.Load(); got != 0 {
+		t.Errorf("tiered upstream forwards = %d, want 0", got)
+	}
+
+	second := tieredDo(t, svc, http.MethodGet, "/bucket/key", "", nil)
+	if second.Code != http.StatusNotFound || second.Header().Get(XCacheHeader) != XCacheMiss {
+		t.Fatalf("second GET = %d X-Cache=%q, want authoritative 404 miss", second.Code, second.Header().Get(XCacheHeader))
+	}
+	if got := forwards.Load(); got != 0 {
+		t.Errorf("tiered upstream forwards after missing local entry = %d, want 0", got)
+	}
+}
+
+func TestTieredLargeShortBodyCleanupKeepsNewerSameETagPut(t *testing.T) {
+	body := bytes.Repeat([]byte("cached body"), 8192)
+	prefix := body[:32768]
+	cfg := config.NewDefault()
+	cfg.Mode = config.ModeTiered
+	cfg.Cache.SetBlockCachingEnabled(false)
+	cfg.Cache.SizeThreshold = int64(len(body) + 1)
+	if err := cfg.Validate(); err != nil {
+		t.Fatal(err)
+	}
+
+	mem := cacheclient.NewMemoryCache()
+	bodyKey := ""
+	started := make(chan struct{})
+	release := make(chan struct{})
+	var releaseOnce sync.Once
+	releaseBody := func() { releaseOnce.Do(func() { close(release) }) }
+	defer releaseBody()
+	var firstRead atomic.Bool
+	client := &testStreamCacheClient{
+		CacheClient: mem,
+		stream: func(ctx context.Context, key string, w io.Writer) error {
+			if key == bodyKey && !firstRead.Swap(true) {
+				_, err := w.Write(prefix)
+				close(started)
+				if err != nil {
+					return err
+				}
+				<-release
+				return nil
+			}
+			return mem.GetStream(ctx, key, w)
+		},
+	}
+	store := cache.NewCacheWithClient(client, &cfg.Cache)
+	mock, forwards, _ := tieredMock()
+	svc := NewService(mock, store, cfg)
+	bucket, key := "bucket", "key"
+
+	if w := tieredDo(t, svc, http.MethodPut, "/"+bucket+"/"+key, string(body), nil); w.Code != http.StatusOK {
+		t.Fatalf("initial local PUT status = %d", w.Code)
+	}
+	oldMeta, oldVersion, found, err := store.GetMetaWithVersion(context.Background(), bucket, key)
+	if err != nil || !found || oldMeta == nil || oldMeta.BodyRef == "" {
+		t.Fatalf("initial local metadata = %+v found=%v err=%v", oldMeta, found, err)
+	}
+	bodyKey = cache.MakeBodyKey(bucket, key, oldMeta.BodyRef)
+	if err := store.PutBodyStream(context.Background(), bucket, key, oldMeta.BodyRef, bytes.NewReader(prefix), 0); err != nil {
+		t.Fatalf("truncate local-tier test body: %v", err)
+	}
+
+	first := httptest.NewRecorder()
+	firstResult := make(chan error, 1)
+	firstRequest := httptest.NewRequest(http.MethodGet, "/"+bucket+"/"+key, nil)
+	go func() { firstResult <- svc.HandleGetObject(first, firstRequest) }()
+	select {
+	case <-started:
+	case <-time.After(2 * time.Second):
+		t.Fatal("local-tier GET did not commit its first cached chunk")
+	}
+
+	if w := tieredDo(t, svc, http.MethodPut, "/"+bucket+"/"+key, string(body), nil); w.Code != http.StatusOK {
+		t.Fatalf("replacement local PUT status = %d", w.Code)
+	}
+	newMeta, newVersion, found, err := store.GetMetaWithVersion(context.Background(), bucket, key)
+	if err != nil || !found || newMeta == nil {
+		t.Fatalf("replacement metadata = %+v found=%v err=%v", newMeta, found, err)
+	}
+	if newVersion == oldVersion || newMeta.ETag != oldMeta.ETag || newMeta.BodyRef == oldMeta.BodyRef {
+		t.Fatalf("replacement version/ref/etag = %d/%q/%q, old = %d/%q/%q", newVersion, newMeta.BodyRef, newMeta.ETag, oldVersion, oldMeta.BodyRef, oldMeta.ETag)
+	}
+
+	releaseBody()
+	if err := <-firstResult; err != nil {
+		t.Fatalf("first GET error = %v", err)
+	}
+	if first.Code != http.StatusOK || first.Header().Get(XCacheHeader) != XCacheHit || first.Header().Get("Content-Length") != strconv.Itoa(len(body)) || !bytes.Equal(first.Body.Bytes(), prefix) {
+		t.Fatalf("first committed GET = %d headers=%v body_bytes=%d, want HIT with the short prefix", first.Code, first.Header(), first.Body.Len())
+	}
+	if got := forwards.Load(); got != 0 {
+		t.Errorf("tiered upstream forwards = %d, want 0", got)
+	}
+	if _, currentVersion, found, err := store.GetMetaWithVersion(context.Background(), bucket, key); err != nil || !found || currentVersion != newVersion {
+		t.Fatalf("version-guarded cleanup lost replacement: found=%v version=%d want=%d err=%v", found, currentVersion, newVersion, err)
+	}
+
+	second := tieredDo(t, svc, http.MethodGet, "/"+bucket+"/"+key, "", nil)
+	if second.Code != http.StatusOK || second.Header().Get(XCacheHeader) != XCacheHit || !bytes.Equal(second.Body.Bytes(), body) {
+		t.Fatalf("second GET = %d X-Cache=%q body_bytes=%d, want full replacement body", second.Code, second.Header().Get(XCacheHeader), second.Body.Len())
+	}
+	if got := forwards.Load(); got != 0 {
+		t.Errorf("tiered upstream forwards after replacement GET = %d, want 0", got)
+	}
+}
+
 // A pre-existing content-addressed entry (proxy-written shape: body keyed by
 // ETag, BodyRef null) serves through the same engine unchanged — the
 // discriminator falls back to the ETag.