feat(dist): Phase B production-readiness — SWIM indirect probes, hint-queue retry, wire compression, and queueHint race fix

Phase B.1 — SWIM-style indirect heartbeat probes:
- Add WithDistIndirectProbes(k, timeout) option; when a direct probe
  fails, up to k random alive relays probe the target on the caller's
  behalf — target is only marked suspect if every relay also fails
- Add /internal/probe HTTP endpoint and IndirectHealth() transport method
- Refuted direct failures now refresh LastSeen rather than escalating
- Expose dist.heartbeat.indirect_probe.{success,failure,refuted} metrics

Phase B.2 — migration failure retry via hint queue:
- migrateIfNeeded queues a hint on ForwardSet failure instead of
  logging-and-dropping silently
- replicateTo hint enqueue broadened from ErrBackendNotFound-only to
  any transport error (timeouts, 5xx, connection resets)
- Fix race in queueHint: snapshot hintBytes under hintsMu before unlock
  to prevent concurrent adjustHintAccounting in the replay loop from
  racing the metric write

Phase B.3 — on-wire gzip compression for the dist HTTP transport:
- Add DistHTTPLimits.CompressionThreshold; ForwardSet gzip-compresses
  Set request bodies exceeding the threshold; server decompresses
  transparently via fiber v3 Content-Encoding auto-decoding

Refactor: extract membershipSnapshot() helper from Metrics() to keep
function length under the lint cap

Add contract tests for all three phases and the queueHint race fix

Author: hyp3rd

CHANGELOG.md

@@ -6,6 +6,17 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
+### Fixed
+
+- **Race in `queueHint` between hint enqueue and hint replay.** Pre-fix,
+  the metric write `dm.metrics.hintedBytes.Store(dm.hintBytes)` happened
+  *after* releasing `hintsMu`, so a concurrent `adjustHintAccounting`
+  call from the replay loop could race the read. Capturing the value
+  under the lock closes the race. Surfaced when migration failures
+  began funneling through `queueHint` (Phase B.2 below) — previously
+  the migration path swallowed errors silently, so the hint enqueue
+  rate from rebalance ticks was much lower.
+
 ### Added
 
 - **Structured logging on the dist backend.** New `WithDistLogger(*slog.Logger)`
@@ -46,6 +57,46 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   against a stopped backend. Library default is a no-op meter, so
   metrics cost nothing unless the caller opts in. Phase A.3 of the
   production-readiness work.
+- **SWIM-style indirect heartbeat probes.** New
+  `WithDistIndirectProbes(k, timeout)` option enables the indirect-
+  probe refutation path: when a direct heartbeat to a peer fails,
+  this node asks `k` random alive peers to probe the target on its
+  behalf, and only marks the target suspect if every relay also
+  fails. Filters caller-side network blips (transient NIC reset,
+  single stuck connection in this node's pool) that would otherwise
+  cause spurious suspect/dead transitions. New transport method
+  `IndirectHealth(ctx, relayNodeID, targetNodeID)` and HTTP endpoint
+  `GET /internal/probe?target=<id>` carry the probe; auth-wrapped
+  identically to the rest of `/internal/*`. New metrics
+  `dist.heartbeat.indirect_probe.success`, `.failure`, `.refuted`
+  expose probe outcomes. `k = 0` (default) preserves the pre-Phase-B
+  behavior. Phase B.1 of the production-readiness work — note that
+  the heartbeat path still carries the `experimental` marker until
+  self-refutation via incarnation-disseminating gossip lands in a
+  later phase.
+- **Migration failures now retry through the hint queue.** When a
+  rebalance forwards a key to its new primary and the transport
+  returns *any* error (not just `ErrBackendNotFound`), the item is
+  enqueued onto the existing hint-replay queue keyed by the new
+  primary, instead of being logged and dropped. The hint-replay
+  loop drains it on its configured schedule until the hint TTL
+  expires. Same broadening applies to the `replicateTo` fan-out on
+  the primary `Set` path — transient HTTP failures (timeout, 5xx,
+  connection reset) no longer silently drop replicas. Phase B.2 of
+  the production-readiness work.
+- **On-wire compression for the dist HTTP transport.** New
+  `DistHTTPLimits.CompressionThreshold` field opts the auto-created
+  HTTP client into gzip-compressing Set request bodies whose
+  serialized payload exceeds the configured byte threshold. The
+  client sets `Content-Encoding: gzip` and the server transparently
+  decompresses (via fiber v3's auto-decoding `Body()`). Threshold
+  `0` (default) preserves the pre-Phase-B wire format byte-for-byte.
+  Operators on bandwidth-constrained links with values above ~1 KiB
+  typically see meaningful reductions; below-threshold values pay
+  no compression cost. Roll out the threshold to all peers before
+  raising it on any peer — a server with compression disabled will
+  reject a gzip body with HTTP 400. Phase B.3 of the
+  production-readiness work.
 
 ## [0.5.0] — 2026-05-05
 

pkg/backend/dist_http_server.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/subtle"
 	"crypto/tls"
+	"errors"
 	"log/slog"
 	"net"
 	"net/http"
@@ -230,6 +231,20 @@ type DistHTTPLimits struct {
 	// ClientAuth=tls.RequireAndVerifyClientCert. The auto-client uses
 	// the same cert as its client cert via Certificates[0].
 	TLSConfig *tls.Config
+
+	// CompressionThreshold opts the dist HTTP transport into gzip
+	// compression of Set request bodies whose serialized payload size
+	// exceeds this many bytes. The client sets `Content-Encoding:
+	// gzip` and the server transparently decompresses before
+	// unmarshaling. 0 disables compression — matches the pre-Phase-B
+	// wire format byte-for-byte. Operators on bandwidth-constrained
+	// links with large values (>1 KiB) typically see meaningful
+	// reductions; values smaller than the threshold pay no cost.
+	//
+	// Server compatibility: a server with compression disabled will
+	// reject a gzip-encoded body with HTTP 400. Roll out the threshold
+	// to all peers before raising it on any peer.
+	CompressionThreshold int
 }
 
 // withDefaults fills any zero-valued field on l with the package default.
@@ -337,6 +352,7 @@ func (s *distHTTPServer) start(bindCtx context.Context, dm *DistMemory) error {
 	s.registerGet(dm)
 	s.registerRemove(dm)
 	s.registerHealth()
+	s.registerProbe(dm)
 	s.registerMerkle(dm)
 
 	return s.listen(bindCtx)
@@ -346,6 +362,11 @@ func (s *distHTTPServer) start(bindCtx context.Context, dm *DistMemory) error {
 // fan-outs to replicas. Uses s.ctx (server-lifecycle) as the backend
 // operation context — see the comment on distHTTPServer.ctx for why we
 // can't use the per-request fiber.Ctx here.
+//
+// Compression note: fiber v3's Body() auto-decompresses based on the
+// inbound `Content-Encoding` header, so this handler does not need
+// explicit gzip handling — it sees the plaintext JSON regardless of
+// whether the client compressed (CompressionThreshold > 0) or not.
 func (s *distHTTPServer) handleSet(fctx fiber.Ctx, dm *DistMemory) error {
 	var req httpSetRequest
 
@@ -434,6 +455,38 @@ func (s *distHTTPServer) registerHealth() {
 	s.app.Get("/health", s.wrapAuth(func(fctx fiber.Ctx) error { return fctx.SendString("ok") }))
 }
 
+// registerProbe wires `/internal/probe?target=<id>` — the indirect-probe
+// relay endpoint used by the SWIM heartbeat path. The relay node calls
+// its own transport's Health(target) and reports the result. 200 = relay
+// reached the target; 502 = relay's probe failed; 404 = target unknown
+// to the relay; 400 = missing/empty target query parameter. Auth-wrapped
+// like the rest of `/internal/*` because indirectly probing arbitrary
+// node IDs through a member is a directory-enumeration vector.
+func (s *distHTTPServer) registerProbe(dm *DistMemory) {
+	s.app.Get("/internal/probe", s.wrapAuth(func(fctx fiber.Ctx) error {
+		target := fctx.Query("target")
+		if target == "" {
+			return fctx.Status(fiber.StatusBadRequest).JSON(fiber.Map{constants.ErrorLabel: "missing target"})
+		}
+
+		transport := dm.loadTransport()
+		if transport == nil {
+			return fctx.SendStatus(fiber.StatusServiceUnavailable)
+		}
+
+		err := transport.Health(s.ctx, target)
+		if err != nil {
+			if errors.Is(err, sentinel.ErrBackendNotFound) {
+				return fctx.SendStatus(fiber.StatusNotFound)
+			}
+
+			return fctx.SendStatus(fiber.StatusBadGateway)
+		}
+
+		return fctx.SendString("ok")
+	}))
+}
+
 func (s *distHTTPServer) registerMerkle(dm *DistMemory) {
 	s.app.Get("/internal/merkle", s.wrapAuth(func(fctx fiber.Ctx) error {
 		tree := dm.BuildMerkleTree()

pkg/backend/dist_http_transport.go

@@ -2,6 +2,7 @@ package backend
 
 import (
 	"bytes"
+	"compress/gzip"
 	"context"
 	"crypto/tls"
 	"io"
@@ -30,6 +31,10 @@ type DistHTTPTransport struct {
 	// lives on distHTTPServer; the two share the same DistHTTPAuth
 	// struct when constructed via NewDistHTTPTransportWithAuth.
 	auth DistHTTPAuth
+	// compressionThreshold is the body-size byte threshold above
+	// which Set request bodies are gzip-compressed. <=0 disables —
+	// matches the pre-Phase-B wire format byte-for-byte.
+	compressionThreshold int
 }
 
 const statusThreshold = 300
@@ -105,10 +110,11 @@ func NewDistHTTPTransportWithAuth(limits DistHTTPLimits, auth DistHTTPAuth, reso
 	}
 
 	return &DistHTTPTransport{
-		client:        client,
-		baseURLFn:     resolver,
-		respBodyLimit: limits.ResponseLimit,
-		auth:          auth,
+		client:               client,
+		baseURLFn:            resolver,
+		respBodyLimit:        limits.ResponseLimit,
+		auth:                 auth,
+		compressionThreshold: limits.CompressionThreshold,
 	}
 }
 
@@ -162,14 +168,23 @@ func (t *DistHTTPTransport) ForwardSet(ctx context.Context, nodeID string, item
 		return ewrap.Wrap(err, "marshal set request")
 	}
 
+	reqBodyReader, gzipped, err := t.maybeGzip(payloadBytes)
+	if err != nil {
+		return ewrap.Wrap(err, "gzip set body")
+	}
+
 	// prefer canonical endpoint; legacy /internal/cache/set still served
-	hreq, err := t.newNodeRequest(ctx, http.MethodPost, nodeID, "/internal/set", nil, bytes.NewReader(payloadBytes))
+	hreq, err := t.newNodeRequest(ctx, http.MethodPost, nodeID, "/internal/set", nil, reqBodyReader)
 	if err != nil {
 		return ewrap.Wrap(err, errMsgNewRequest)
 	}
 
 	hreq.Header.Set("Content-Type", "application/json")
 
+	if gzipped {
+		hreq.Header.Set("Content-Encoding", "gzip")
+	}
+
 	resp, err := t.doTrusted(hreq)
 	if err != nil {
 		return err
@@ -334,6 +349,37 @@ func (t *DistHTTPTransport) Health(ctx context.Context, nodeID string) error {
 	return nil
 }
 
+// IndirectHealth asks the relay node to probe the target on this
+// caller's behalf. The dist HTTP server's `/internal/probe?target=<id>`
+// endpoint runs a Health() call on its own transport and returns 200 if
+// the target is reachable from the relay's vantage point. Used by the
+// SWIM indirect-probe path to filter caller-side network blips before
+// marking a peer suspect.
+func (t *DistHTTPTransport) IndirectHealth(ctx context.Context, relayNodeID, targetNodeID string) error {
+	hreq, err := t.newNodeRequest(ctx, http.MethodGet, relayNodeID, "/internal/probe",
+		url.Values{"target": []string{targetNodeID}}, nil)
+	if err != nil {
+		return ewrap.Wrap(err, errMsgNewRequest)
+	}
+
+	resp, err := t.doTrusted(hreq)
+	if err != nil {
+		return err
+	}
+
+	defer drainBody(t.limitedBody(resp))
+
+	if resp.StatusCode == http.StatusNotFound {
+		return sentinel.ErrBackendNotFound
+	}
+
+	if resp.StatusCode >= statusThreshold {
+		return ewrap.Newf("indirect probe status %d", resp.StatusCode)
+	}
+
+	return nil
+}
+
 // FetchMerkle retrieves a Merkle tree snapshot from a remote node.
 func (t *DistHTTPTransport) FetchMerkle(ctx context.Context, nodeID string) (*MerkleTree, error) {
 	if t == nil {
@@ -416,6 +462,36 @@ func (t *DistHTTPTransport) limitedBody(resp *http.Response) io.ReadCloser {
 	return http.MaxBytesReader(nil, resp.Body, t.respBodyLimit)
 }
 
+// maybeGzip returns a reader for the request body and a boolean
+// indicating whether the body was gzip-compressed. Compression
+// applies when compressionThreshold > 0 and the payload exceeds it.
+// Below the threshold the original bytes round-trip unchanged so
+// peers without compression support remain compatible. Errors come
+// only from the gzip writer (which closes around an in-memory
+// buffer, so they are practically impossible) — propagated for
+// completeness.
+func (t *DistHTTPTransport) maybeGzip(payload []byte) (io.Reader, bool, error) {
+	if t.compressionThreshold <= 0 || len(payload) <= t.compressionThreshold {
+		return bytes.NewReader(payload), false, nil
+	}
+
+	var buf bytes.Buffer
+
+	gz := gzip.NewWriter(&buf)
+
+	_, writeErr := gz.Write(payload)
+	if writeErr != nil {
+		return nil, false, ewrap.Wrap(writeErr, "gzip write")
+	}
+
+	closeErr := gz.Close()
+	if closeErr != nil {
+		return nil, false, ewrap.Wrap(closeErr, "gzip close")
+	}
+
+	return &buf, true, nil
+}
+
 func (t *DistHTTPTransport) resolveBaseURL(nodeID string) (*url.URL, error) {
 	if t == nil || t.baseURLFn == nil {
 		return nil, errNoTransport