feat(proxy): count block refetches after failed cache writes

docs/metrics.md

@@ -318,6 +318,36 @@ the unary byte-write fast path or the streaming fallback). Flood-safe, alertable
 populate-write failures; the per-block error is logged only at Debug. Healthy value is 0 —
 alert on any sustained rate.
 
+#### tag_cache_block_refetch_total
+
+**Type:** Counter (labeled by `previous`)
+
+Upstream block fetches for a block whose previous populate **on this pod** did not land in
+cache, so the fetch is a direct cost of a warm-caching failure rather than a cold miss.
+Counted at the start of a real upstream block fetch (after the presence recheck finds the
+block absent).
+
+| `previous` | Meaning |
+|------------|---------|
+| `write_skipped` | The earlier fetch was declined a populate slot (budget/concurrency pressure), so its bytes were served but never cached |
+| `write_failed` | The earlier fetch's `PutBlock` failed (also counted by `tag_cache_block_populate_failed_total`) |
+
+Caveats:
+- The memory of earlier failures is a per-pod, fixed-size table (1M slots), cleared on
+  restart. A block whose earlier populate failed on another pod, or whose slot was
+  overwritten by a colliding key, is not counted, so this is a lower bound.
+- A successful `PutBlock` clears the block's entry, so a later eviction of that block is
+  not attributed here.
+
+```promql
+# Re-fetches caused by warm-caching failures, by cause
+sum by (previous) (rate(tag_cache_block_refetch_total[5m]))
+
+# Share of upstream block fetches that are re-fetches of a block that should have been warm
+sum(rate(tag_cache_block_refetch_total[5m])) /
+sum(rate(tag_cache_block_populated_total[5m]) + rate(tag_cache_block_populate_failed_total[5m]))
+```
+
 #### tag_cache_block_prefetch_windows_total
 
 **Type:** Counter (labeled by `outcome`)

metrics/metrics.go

@@ -311,6 +311,19 @@ var (
 		},
 	)
 
+	// CacheBlockRefetch counts upstream block fetches for a block whose previous
+	// populate on this pod was skipped (write_skipped: populate budget declined)
+	// or failed (write_failed: PutBlock error) — re-fetches caused directly by a
+	// warm-caching failure. Attribution uses a bounded per-pod table, so it is a
+	// lower bound and resets on restart.
+	CacheBlockRefetch = promauto.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "tag_cache_block_refetch_total",
+			Help: "Upstream block fetches caused by a previous populate of the same block being skipped or failing (block-aligned caching)",
+		},
+		[]string{"previous"},
+	)
+
 	// CacheBlockPrefetchWindows counts multi-block serves by the prefetch window they were
 	// granted from the shared staging budget: "full" (requested window granted), "shrunk"
 	// (a smaller window than requested), or "sequential" (budget declined every window size —

proxy/block_cache.go

@@ -1273,6 +1273,9 @@ func (s *Service) runBlockFetch(state *blockFetchState, blockKey, bucket, key, a
 		s.finishBlockFetch(blockKey, state, nil)
 		return
 	}
+	// This is a real upstream fetch; count it if the previous populate of this
+	// block on this pod was skipped or failed.
+	s.blockWriteFailures.countBlockRefetch(blockKey)
 	bStart, bEnd := blockBounds(blockIdx, meta.BlockSize, meta.ContentLength)
 	blockLen := bEnd - bStart + 1
 
@@ -1283,6 +1286,7 @@ func (s *Service) runBlockFetch(state *blockFetchState, blockKey, bucket, key, a
 	reserveWeight := s.populateWeight(blockLen)
 	if !s.acquireCacheSlot(fetchCtx, reserveWeight, priorityReadMiss) {
 		metrics.RecordCachePopulateSkipped(priorityReadMiss.metricSource())
+		s.blockWriteFailures.record(blockKey, blockWriteSkipped)
 		fail(errCachePopulateDeclined)
 		return
 	}
@@ -1348,11 +1352,13 @@ func (s *Service) runBlockFetch(state *blockFetchState, blockKey, bucket, key, a
 		if perr != nil {
 			putBlockBuf(bufp)
 			metrics.CacheBlockPopulateFailed.Inc()
+			s.blockWriteFailures.record(blockKey, blockWriteFailed)
 			state.completeServe(nil, nil, perr)
 			s.finishBlockFetch(blockKey, state, perr)
 			return
 		}
 		metrics.CacheBlockPopulated.Inc()
+		s.blockWriteFailures.clear(blockKey)
 		metrics.CacheBlockBytesPopulated.Add(float64(blockLen))
 		state.completeServe(bufp, blockBuf, nil)
 		s.finishBlockFetch(blockKey, state, nil)
@@ -1367,9 +1373,11 @@ func (s *Service) runBlockFetch(state *blockFetchState, blockKey, bucket, key, a
 		perr := s.cache.PutBlock(fetchCtx, bucket, key, meta.ETag, meta.BlockSize, blockIdx, blockBuf, ttl)
 		if perr != nil {
 			metrics.CacheBlockPopulateFailed.Inc()
+			s.blockWriteFailures.record(blockKey, blockWriteFailed)
 			log.Debug().Err(perr).Str("bucket", bucket).Str("key", key).Int64("block", blockIdx).Msg("Detached remote block cache write failed")
 		} else {
 			metrics.CacheBlockPopulated.Inc()
+			s.blockWriteFailures.clear(blockKey)
 			metrics.CacheBlockBytesPopulated.Add(float64(blockLen))
 		}
 		s.releaseCacheSlot(reserveWeight)

proxy/block_write_failures.go

@@ -0,0 +1,84 @@
+package proxy
+
+import (
+	"hash/maphash"
+	"sync/atomic"
+
+	"github.com/tigrisdata/tag/metrics"
+)
+
+// blockWriteFailureEntries is the slot count of a Service's block write-failure
+// table (8 bytes each: 8 MiB).
+const blockWriteFailureEntries = 1 << 20
+
+// blockWriteFailure is why a fetched block was not cached. Codes fit in the low
+// 3 bits of a table word; 0 is reserved for an empty slot.
+type blockWriteFailure uint64
+
+const (
+	blockWriteNone blockWriteFailure = iota
+	blockWriteSkipped
+	blockWriteFailed
+)
+
+const blockWriteFailureMask = 7
+
+var blockWriteFailureSeed = maphash.MakeSeed()
+
+// blockWriteFailureTable remembers, per block key, that its last populate was
+// skipped or failed on this pod. It is a fixed-size direct-mapped array of
+// atomics: slot = hash & (len-1), word = (hash with the low 3 bits cleared) |
+// reason. A colliding key overwrites the slot and a forgotten key reads as none,
+// so refetch counts are a lower bound. Lock-free; a nil table records nothing.
+type blockWriteFailureTable struct {
+	slots []atomic.Uint64
+	mask  uint64
+}
+
+// newBlockWriteFailureTable sizes the table to n, which must be a power of two.
+func newBlockWriteFailureTable(n int) *blockWriteFailureTable {
+	return &blockWriteFailureTable{slots: make([]atomic.Uint64, n), mask: uint64(n - 1)}
+}
+
+func (t *blockWriteFailureTable) record(blockKey string, r blockWriteFailure) {
+	if t == nil {
+		return
+	}
+	h := maphash.String(blockWriteFailureSeed, blockKey)
+	t.slots[h&t.mask].Store(h&^blockWriteFailureMask | uint64(r))
+}
+
+func (t *blockWriteFailureTable) lookup(blockKey string) blockWriteFailure {
+	if t == nil {
+		return blockWriteNone
+	}
+	h := maphash.String(blockWriteFailureSeed, blockKey)
+	w := t.slots[h&t.mask].Load()
+	if w != 0 && w&^blockWriteFailureMask == h&^blockWriteFailureMask {
+		return blockWriteFailure(w & blockWriteFailureMask)
+	}
+	return blockWriteNone
+}
+
+// clear empties blockKey's slot if it still holds blockKey.
+func (t *blockWriteFailureTable) clear(blockKey string) {
+	if t == nil {
+		return
+	}
+	h := maphash.String(blockWriteFailureSeed, blockKey)
+	slot := &t.slots[h&t.mask]
+	if w := slot.Load(); w != 0 && w&^blockWriteFailureMask == h&^blockWriteFailureMask {
+		slot.CompareAndSwap(w, 0)
+	}
+}
+
+// countBlockRefetch increments tag_cache_block_refetch_total when blockKey's
+// previous populate on this pod was skipped or failed.
+func (t *blockWriteFailureTable) countBlockRefetch(blockKey string) {
+	switch t.lookup(blockKey) {
+	case blockWriteSkipped:
+		metrics.CacheBlockRefetch.WithLabelValues("write_skipped").Inc()
+	case blockWriteFailed:
+		metrics.CacheBlockRefetch.WithLabelValues("write_failed").Inc()
+	}
+}

proxy/service.go

@@ -123,6 +123,10 @@ type Service struct {
 	// already examined. Without it every tail read of a fully-warmed object re-probes
 	// its metadata blocks, which in cluster mode are mostly remote.
 	recentFooterWork *expirable.LRU[string, struct{}]
+	// blockWriteFailures remembers block keys whose populate write was skipped
+	// or failed, so a later re-fetch is counted as a warm-caching failure.
+	// Nil (as in bare test Services) records nothing.
+	blockWriteFailures *blockWriteFailureTable
 }
 
 // NewService creates a new proxy service.
@@ -214,6 +218,7 @@ func NewService(forwarder RequestForwarder, cache *cache.Cache, cfg *config.Conf
 		blockFetches:                make(map[string]*blockFetchState),
 		retierClaims:                make(map[string]int),
 		retierInflight:              make(map[string]context.CancelFunc),
+		blockWriteFailures:          newBlockWriteFailureTable(blockWriteFailureEntries),
 	}
 	// Allocated only when the feature that uses it is on.
 	if cfg.Cache.ParquetOptimization {