Keep Gantry provider records fresh and reject stale DHT hits

cmd/gantry/agent_metrics.go

@@ -555,7 +555,7 @@ func newPhase5Metrics(reg *metrics.Registry, healthScore, connCount func() float
 // - advertise_reconcile_*: the advertiser's reconcile
 // loop instrumentation. Pairs (duration histogram + digest
 // count gauge + reconcile counters) describe one full pass.
-// - withdraw_*: counter pair around DHT.Withdraw, the // equivalent of dht_provide_*. With kad-dht's 24h TTL, the
+// - withdraw_*: counter pair around DHT.Withdraw, the // equivalent of dht_provide_*. With configured provider validity, the
 // primary signal is the rate of attempted withdrawals (it
 // should track the rate of container deletions on the node).
 // - containerd_lease_*: lease lifecycle counters. Active

cmd/gantry/main.go

@@ -48,6 +48,7 @@ import (
 	"github.com/Azure/unbounded/internal/gantry/metrics"
 	"github.com/Azure/unbounded/internal/gantry/mirror"
 	"github.com/Azure/unbounded/internal/gantry/negcache"
+	"github.com/Azure/unbounded/internal/gantry/providerprobe"
 	"github.com/Azure/unbounded/internal/gantry/registryauth"
 	"github.com/Azure/unbounded/internal/gantry/transfer"
 	"github.com/Azure/unbounded/internal/version"
@@ -439,6 +440,7 @@ func runAgent(args []string) error {
 		realResolver := coldstart.NewChairResolver(coldstart.ChairOptions{
 			Chairs:       chairCache,
 			Discovery:    disco,
+			PeerMetadata: peerClient,
 			Coord:        chairCoord,
 			LocalPull:    coordServer,
 			Inflight:     inflightMap,
@@ -507,19 +509,8 @@ func runAgent(args []string) error {
 				// let kubelet back off than thunder the origin.
 				return disco.Health() >= 0.3
 			},
-			Inflight: inflightMap,
-			Recheck: func(ctx context.Context, d digest.Digest) bool {
-				// Final post-jitter probe: did anyone publish a
-				// provider record while we slept? If so, direct-origin-fallback declines
-				// and the client retries through the warm path on its
-				// next attempt.
-				rcCtx, cancel := context.WithTimeout(ctx, 1*time.Second)
-				defer cancel()
-
-				prov, err := disco.FindProviders(rcCtx, d)
-
-				return err == nil && len(prov) > 0
-			},
+			Inflight:   inflightMap,
+			Recheck:    newUsableProviderRecheck(disco, peerClient),
 			OnFallback: func() { p5.originFallbackTotal.Inc() },
 			OnDecline: func(reason string) {
 				p5.originFallbackDeclineTotal.WithLabelValues(reason).Inc()
@@ -921,6 +912,24 @@ func runAgent(args []string) error {
 	return nil
 }
 
+func newUsableProviderRecheck(dht ifaces.DHT, peer ifaces.PeerMetadataDialer) func(context.Context, ifaces.OriginRef) bool {
+	return func(ctx context.Context, ref ifaces.OriginRef) bool {
+		// Keep this final escape-valve check bounded. A provider record alone
+		// cannot suppress origin fallback; a peer must answer HEAD for ref.
+		recheckCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
+		defer cancel()
+
+		providers, err := dht.FindProviders(recheckCtx, ref.Digest)
+		if err != nil || len(providers) == 0 {
+			return false
+		}
+
+		_, usable := providerprobe.First(recheckCtx, peer, providers, ref, providerprobe.Attempted{}, providerprobe.DefaultConcurrency)
+
+		return usable
+	}
+}
+
 // loadAgentConfig merges YAML, env, and flags into a *config.Config. Two-
 // pass parsing: first pass reads --config; second pass overlays flags onto
 // (defaults < YAML < env).

cmd/gantry/provider_recheck_test.go

@@ -0,0 +1,91 @@
+// Copyright (c) Microsoft Corporation.
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/Azure/unbounded/internal/gantry/digest"
+	"github.com/Azure/unbounded/internal/gantry/ifaces"
+	"github.com/Azure/unbounded/internal/gantry/ifaces/fakes"
+	"github.com/Azure/unbounded/internal/gantry/inflight"
+	"github.com/Azure/unbounded/internal/gantry/mirror"
+)
+
+type recheckMetadataDialer struct {
+	errors map[string]error
+}
+
+func (d *recheckMetadataDialer) HeadFromPeer(_ context.Context, addr string, _ ifaces.OriginRef) (int64, string, error) {
+	return 1, "application/octet-stream", d.errors[addr]
+}
+
+func TestNF5UsabilityRecheckAllowsFallbackForStaleProvider(t *testing.T) {
+	ref := providerRecheckRef()
+	dht := fakes.NewDHT()
+
+	dht.Inject(ref.Digest, ifaces.Provider{NodeID: "stale", Addr: "stale:5001"})
+
+	dialer := &recheckMetadataDialer{errors: map[string]error{"stale:5001": errors.New("unreachable")}}
+	controller := mirror.NewDirectOriginFallback(mirror.DirectOriginFallbackOptions{
+		Inflight:      inflight.New(inflight.DefaultStalls(), nil),
+		InBootstrap:   func() bool { return false },
+		HealthyEnough: func() bool { return true },
+		ClusterSize:   func() int { return 1 },
+		Recheck:       newUsableProviderRecheck(dht, dialer),
+	})
+
+	proceed, release, err := controller.Allow(context.Background(), ref, 0)
+	if err != nil {
+		t.Fatalf("Allow: %v", err)
+	}
+
+	if !proceed {
+		t.Fatal("Allow declined fallback for stale provider")
+	}
+
+	if release == nil {
+		t.Fatal("Allow release is nil")
+	}
+
+	release()
+}
+
+func TestNF5UsabilityRecheckDeclinesFallbackForUsableProvider(t *testing.T) {
+	ref := providerRecheckRef()
+	dht := fakes.NewDHT()
+
+	dht.Inject(ref.Digest, ifaces.Provider{NodeID: "fresh", Addr: "fresh:5001"})
+	controller := mirror.NewDirectOriginFallback(mirror.DirectOriginFallbackOptions{
+		Inflight:      inflight.New(inflight.DefaultStalls(), nil),
+		InBootstrap:   func() bool { return false },
+		HealthyEnough: func() bool { return true },
+		ClusterSize:   func() int { return 1 },
+		Recheck:       newUsableProviderRecheck(dht, &recheckMetadataDialer{}),
+	})
+
+	proceed, release, err := controller.Allow(context.Background(), ref, 0)
+	if err != nil {
+		t.Fatalf("Allow: %v", err)
+	}
+
+	if proceed {
+		t.Fatal("Allow proceeded despite usable provider")
+	}
+
+	if release != nil {
+		t.Fatal("Allow returned release after declining fallback")
+	}
+}
+
+func providerRecheckRef() ifaces.OriginRef {
+	return ifaces.OriginRef{
+		Registry:   "registry.example.com",
+		Repository: "repo/image",
+		Digest:     digest.MustParse("sha256:3434343434343434343434343434343434343434343434343434343434343434"),
+		Kind:       ifaces.KindBlob,
+	}
+}

deploy/gantry/configmap.yaml.tmpl

@@ -183,6 +183,12 @@ data:
     advertise_reconcile_interval: "1m"
 
     # ---------- DHT / NF5 (§7.7) ----------
+    # Provider records expire quickly enough to bound stale results. The
+    # sweeping provider spreads refresh work across each interval; its maximum
+    # delay plus interval must remain below provider validity.
+    dht_provider_validity: "6h"
+    dht_reprovide_interval: "3h"
+    dht_max_reprovide_delay: "10m"
     nf5_jitter_base: "3s"
     nf5_per_node_rate_limit: 2
     bootstrap_window: "30s"

designs/gantry-detailed-design.md

@@ -443,7 +443,7 @@ recent_failures[digest] = {
 ### 5.9 Node joins / leaves cluster
 
 - **Join:** the agent starts up, the Kubernetes informer reports the new node to all other agents within a few seconds, HRW rankings update naturally. The new node bootstraps its libp2p host using peers from the informer's existing pod list and joins the DHT.
-- **Leave:** existing provider records held by the departed node expire from the DHT (TTL, default 24h with 12h refresh). HRW rankings update on all surviving agents as the informer removes the departed node. If the departed node was a designated puller for an in-flight pull, the stall-detection path in §5.6 recovers.
+- **Leave:** existing provider records held by DHT peers expire after the configured 6h validity. Live providers refresh through the 3h sweeping-provider schedule. If the departed node was a designated puller for an in-flight pull, the stall-detection path in §5.6 recovers.
 
 ---
 
@@ -491,7 +491,7 @@ The `server = ...` directive causes containerd to attach `?ns=<registry>` to eve
 - **Transport:** TCP + QUIC. Noise for encryption.
 - **DHT mode:** Server mode on every agent (all agents serve queries). With 10k+ nodes, this is fine  -  Kademlia scales to this size comfortably (IPFS runs at much larger scale).
 - **Bootstrap:** the agent's Kubernetes informer provides a list of peer pod IPs. On startup the agent draws a random subset of **8** peer IPs and dials them in parallel. If fewer than **4** respond within 5 s, the agent draws another random subset of 8 from the remaining pool and retries. Total dials per startup are capped at **32**; after that the agent proceeds with whatever routing-table state it has and relies on lazy routing-table growth as DHT queries flow. The 8-peer subset is sized to populate multiple Kademlia buckets (bucket size 20 in `go-libp2p-kad-dht`) on a single round while remaining cheap; the cap prevents pathological retry on a freshly-rolled-out DaemonSet where no peer is yet ready  -  in that case the bootstrap-window suppression (§7.7) and NF5 jitter/rate-limit handle the genuinely-cold case.
-- **Provider record TTL:** 24h with 12h refresh (libp2p default). Dead nodes age out automatically.
+- **Provider record validity:** 6h with a 3h sweeping-provider interval and 10m maximum schedule delay. Dead providers stop refreshing and age out automatically.
 - **Identity persistence:** the libp2p private key is generated on first start and persisted to `hostPath`. Lost identity is not catastrophic  -  the agent rejoins with a new ID; old DHT records expire.
 
 ### 7.3 Cluster membership