diff --git a/cmd/gantry/agent_metrics.go b/cmd/gantry/agent_metrics.go index 91ea0425c..4d8c86122 100644 --- a/cmd/gantry/agent_metrics.go +++ b/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 diff --git a/cmd/gantry/main.go b/cmd/gantry/main.go index d57f5ac79..c24400b0b 100644 --- a/cmd/gantry/main.go +++ b/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). diff --git a/cmd/gantry/provider_recheck_test.go b/cmd/gantry/provider_recheck_test.go new file mode 100644 index 000000000..0d02e894f --- /dev/null +++ b/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, + } +} diff --git a/deploy/gantry/configmap.yaml.tmpl b/deploy/gantry/configmap.yaml.tmpl index 6bef5b12d..9f18b7ae8 100644 --- a/deploy/gantry/configmap.yaml.tmpl +++ b/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" diff --git a/designs/gantry-detailed-design.md b/designs/gantry-detailed-design.md index f54ac5f4a..6fa769ff4 100644 --- a/designs/gantry-detailed-design.md +++ b/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=` 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 diff --git a/designs/gantry-stale-dht-provider-recovery.md b/designs/gantry-stale-dht-provider-recovery.md new file mode 100644 index 000000000..202fcb757 --- /dev/null +++ b/designs/gantry-stale-dht-provider-recovery.md @@ -0,0 +1,599 @@ +# Gantry Stale DHT Provider Recovery + +**Status:** Draft for discussion + +## Evidence Scope + +Current-behavior statements in this document were checked against this +repository and the exact `go-libp2p-kad-dht v0.42.2` source selected by +`go.mod`. Dependency paths such as `routing.go` and `records/providers_manager.go` +below are relative to that module. + +The `6h` validity, `3h` reprovide interval, `10m` maximum reprovide delay, +in-memory sweeping provider, and bounded provider HEAD validation are +implemented. Lookup expansion, explicit chair completion, cold-start deadlines, +and unconditional eventual fallback remain proposed behavior. The timing +values have not been validated by cluster measurements. + +## Implementation Status + +Implemented in the current change: + +- Section 1: configured provider validity and in-memory sweeping reprovide. +- Section 3: request-scoped failed-provider exclusion during chair polling. +- Section 5: bounded HEAD validation for chair polling and final + direct-origin-fallback recheck. + +Sections 2, 4, 6, and 7 remain design proposals. Section 3 is not yet applied +to ordinary warm-path GET retries. + +## Summary + +Gantry uses libp2p Kademlia provider records as hints that a peer may hold an +OCI digest. A provider record does not prove that the peer is alive, still has +the digest, or belongs to the current node generation. Provider lookup also +does not return records in freshness order. + +This becomes a liveness problem during node replacement. If every node that +advertised a digest is replaced inside the provider-record validity window, +the DHT can continue returning only departed provider identities even after +new chairs have pulled and advertised the same digest. + +This design proposes two complementary changes: + +1. Bound the stale-record window with a shorter provider validity and a + correspondingly shorter reprovide interval. +2. Base cold-start and origin-fallback decisions on a usable provider or + authoritative chair state, never on a merely nonempty DHT result. + +The shorter validity limits how long stale records can interfere. It is not +the correctness mechanism. Correctness comes from allowing the state machine +to reach controlled origin fallback when no chair is pulling and no returned +provider can serve the digest. + +## Current Behavior + +### Provider lookup is bounded and unordered + +Gantry calls the synchronous `FindProviders` API +(`internal/gantry/discovery/discovery.go:459`). In +`go-libp2p-kad-dht v0.42.2`, that API passes the DHT bucket size to +`FindProvidersAsync` (`routing.go:492-500`). The default bucket size is 20 +(`amino/defaults.go:23-27`), so the first 20 unique providers collected satisfy +the lookup. Provider sets are shuffled before the count is applied, but they +are not ordered by publication time (`routing.go:536-626`, +`records/providers_manager.go:317-354`). + +The provider response identifies peer IDs, addresses, and a connection hint. +It does not expose a provider timestamp, sequence, or signed expiry that Gantry +can use to prefer a new record (`pb/dht.proto:30-38`). Records from distinct old +and new provider identities coexist for the same digest because the storage +key includes both the digest and provider peer ID +(`records/providers_manager.go:277-284`). + +Gantry shuffles the returned candidates again before fetching, which +distributes attempts within that result set but cannot introduce provider 21 +or later (`internal/gantry/mirror/mirror.go:2382`). + +### Provider records outlive departed nodes + +Each DHT peer stores a provider record with its local receipt time. The +dependency default is 48 hours (`amino/defaults.go:40-43`), but Gantry now +overrides provider validity to six hours through +`dht.ProviderManagerOpts(records.ProvideValidity(...))` +(`internal/gantry/discovery/discovery.go:312-322`). + +The dependency's sweeping provider is separate from `dht.New`. Gantry +constructs it explicitly with a 3-hour interval and 10-minute maximum delay +(`internal/gantry/discovery/discovery.go:383-405`). A successful synchronous +`Provide` also registers the digest with the sweeper, and `Withdraw` removes it +from the reprovide schedule (`internal/gantry/discovery/discovery.go:497-525`). +The sweeper uses in-memory key and schedule state; startup inventory reconcile +rebuilds that state after a process restart. + +Provider records can outlive a departed provider while DHT peers that stored +the records remain alive. Gantry does not configure a DHT datastore, so the +dependency uses an in-memory datastore +(`internal/gantry/discovery/discovery.go:303`, +`internal/config/config.go:100,138-145`). Replacing a DHT process erases the +provider records held only by that process. A complete process replacement +therefore clears old records unless an old provider issued a fresh `Provide` +that wrote its record into replacement DHT peers during the overlap. + +libp2p Kademlia has no protocol-level provider withdrawal. Gantry's `Withdraw` +therefore stops future refreshes, while already-published remote records remain +eligible until their storing peer's configured validity expires. + +### Failure suppression is requester-local + +The mirror records recent failures in process-local maps: + +- A peer that returns not found for a digest is suppressed for three minutes. +- An unreachable address is suppressed for 30 seconds. +- A peer that returns invalid content is suppressed for five minutes. + +The defaults are in `internal/gantry/mirror/mirror.go:643-645`. Suppression keeps +one requester from immediately retrying a failed candidate. It does not remove +the provider record, affect another requester, or cause a subsequent DHT query +to return different candidates. + +### Recovery requires a usable provider + +Chair cold-start polling treats DHT results as candidates. It sends bounded +metadata `HEAD` requests and returns only a provider that currently reports the +digest available (`internal/gantry/coldstart/chair.go`). Failed candidates are +remembered by peer ID and address for that resolution, so repeated DHT results +do not repeat the same probes. An accepted chair becomes eligible again after +an authoritative chair progress recheck because it may have completed at the +same identity and address. + +The final direct-origin-fallback recheck uses the same HEAD validation with a +three-second total budget and at most four concurrent probes +(`cmd/gantry/main.go`). A stale-only result therefore does not suppress origin +fallback. Metadata probes do not forward request-scoped registry credentials. + +## Failure Scenario + +Consider a digest advertised throughout a cluster before a rolling node +replacement: + +1. Old nodes advertise the digest. +2. During rolling overlap, at least one old provider issues a fresh `Provide` + after replacement DHT peers have joined, so those peers store the old + provider identity. +3. The old provider nodes are replaced and receive new libp2p identities when + their node-local identity storage is not retained. +4. A replacement node misses the digest locally. +5. `FindProviders` returns 20 old provider identities. +6. The requester fails to reach them and suppresses their addresses locally. +7. Cold-start asks the current ranked chair cohort to pull the digest. +8. One or more chairs finish and advertise under their new identities. +9. Chair polling receives a nonempty set containing only old providers and + rejects them after their HEAD probes fail. +10. Repeated lookups returning the same old identities produce no new HEAD + traffic during that resolution. +11. Polling succeeds only after a provider answers HEAD for the digest. If no + usable provider appears, normal chair recheck and exhaustion behavior + continues. + +Successful chair pulls still do not guarantee that a requester sees the +resulting new provider record because the lookup remains capped and unordered. +However, stale records no longer count as successful chair completion or as a +reason for the final direct-origin-fallback recheck to decline. + +A complete node replacement by itself does not guarantee this failure. If +every in-memory DHT store containing an old record is replaced and no old +provider writes into a replacement store during overlap, the old records +disappear with those processes regardless of provider validity. The failure +requires at least one stale record to survive on a current DHT peer. Upgraded +peers serve it for at most six hours after receipt; legacy peers in a mixed +rollout may retain it for the dependency's 48-hour default. + +## Required Invariants + +The implementation must preserve these invariants: + +1. A DHT provider record is a candidate, not proof of availability. +2. A failed candidate must not count as cold-start progress for that request. +3. A chair that reports work still in flight may extend the wait within the + request deadline. +4. A chair that has completed must be distinguishable from one that is still + pulling. +5. If no chair is pulling and no candidate can serve the digest, cold-start + must return `ErrExhausted` so controlled origin fallback can run. +6. Direct-origin-fallback recheck may decline fallback only after finding a + usable provider, not merely a provider record. +7. Chair completion must not force every requester to fetch from one chair. + Chairs are bounded initial seeds, not the permanent data plane. +8. Warm discovery must expand past failed provider sets up to a configured + cluster-size ceiling before declaring a cold miss, subject to one hard warm + lookup deadline. +9. Cold-start has one hard end-to-end deadline. DHT results, chair retries, + backup cohorts, and fallback gates must not extend it indefinitely. +10. When that deadline expires without a usable provider, the request must + enter direct origin fallback regardless of the discovery or coordination + failure that prevented peer recovery. +11. Periodic reprovide must use the dependency's sweeping provider and stagger + work both across the interval and across provider peer identities. Gantry + must not reprovide every digest on every node in one periodic burst. + +## Proposed Design + +### 1. Shorten provider validity and reprovide interval + +Add explicit Gantry configuration for both values: + +```yaml +dht_provider_validity: 6h +dht_reprovide_interval: 3h +``` + +These are proposed initial defaults, subject to scale testing. Pass provider +validity to `dht.New` through `dht.ProviderManagerOpts` and +`records.ProvideValidity`. Integrate the dependency's `provider.SweepingProvider` +as the required periodic reprovide scheduler. The interval is not a `dht.New` +option. Gantry's advertiser must register present digests with +`StartProviding` and remove absent digests with `StopProviding`; event-driven +publication remains the immediate path for newly available content. + +The sweeping provider divides the DHT keyspace into regions and schedules those +regions across the complete reprovide interval. Its region schedule is +permuted using the provider peer ID (`provider/provider.go:630-689`), so nodes +with different identities do not all intentionally walk regions in one common +order. This is a distribution mechanism, not a guarantee that every pair of +peer IDs receives a different offset for a given region. Gantry must preserve +the per-peer ordering when wiring the provider. + +Gantry uses the sweeping provider's in-memory keystore and datastore. A restart +rebuilds the key set from containerd inventory and starts a fresh schedule. +Persisting provider scheduler state is not required. + +Cross-node staggering is a requirement, not an assumption derived only from +the library API. Scale validation must measure publication rate by node and +keyspace region. If synchronized node startup or scheduler recovery still +produces aligned bursts, Gantry may select a fresh random startup phase on each +restart before beginning periodic reprovide. Restart continuity is not a +requirement; the maximum resulting refresh gap must remain below provider +validity. + +Validation must require both values to be positive and the reprovide interval +to be shorter than the provider validity. + +With these values: + +- A live node refreshes its provider records every three hours. +- A departed node cannot refresh. +- A DHT peer configured with the six-hour validity stops serving a provider + record six hours after that peer last accepted it. Reads enforce expiration; + the cleanup cadence only controls physical datastore reclamation. + +This change increases DHT publication traffic. A naive per-digest scheduler on +a node advertising `D` digests would schedule approximately `D/3` refreshes per +node per hour, which is why this design requires the sweeping provider. The +sweeper batches work by DHT keyspace region and spreads it across the interval, +so `3D` is not an estimate of network lookups or RPCs. Cluster-scale +measurements must determine whether these defaults are sustainable. + +A six-hour validity does not make stale results impossible. A replacement can +happen inside six hours, and old providers can write records into replacement +DHT peers during rollout. Mixed-version peers can also retain the old validity. +The recovery state machine must remain correct when every returned candidate +is stale. + +### 2. Expand warm discovery until usable or bounded exhaustion + +Replace the synchronous 20-provider lookup with `FindProvidersAsync` and an +explicit target count. The first round requests 20 candidates. If every new +candidate fails, double the target and query again: + +```text +20, 40, 80, 160, ... min(next target, cluster-size ceiling) +``` + +Each target is a cumulative result count for that lookup because libp2p +provider lookup has no exclusion list or pagination cursor. A request-scoped +set removes providers returned and tried in earlier rounds. Therefore a +40-result round may contain the same failed 20, but only its previously unseen +candidates are eligible for probing. + +```text +target = 20 +attempted = {} + +while warm lookup deadline has not expired: + candidates = FindProvidersAsync(digest, target) + candidates = candidates - self - suppressed - attempted + probe candidates with bounded concurrency + + if one candidate can serve the digest: + fetch from that provider + stop + + attempted += candidates + + if target == cluster-size ceiling: + break + + target = min(target * 2, cluster-size ceiling) + +enter cold-start +``` + +Gantry does not maintain exact cluster membership. The ceiling must therefore +come from an explicit configured cluster-size estimate. The current +`chair_cluster_size_estimate` defaults to 100,000 +(`internal/gantry/config/config.go:484`), but the implementation should expose +a lookup-specific name rather than silently coupling two policies. + +The requested count is an output ceiling, not a request to enumerate every +provider or every cluster node. Each remote `GET_PROVIDERS` response stops when +the next provider would exceed libp2p's message-size limit +(`handlers.go:195-233`), and the protocol provides no pagination cursor. +Kademlia lookup also terminates after its normal closest-peer convergence. +Consequently, even a target equal to the cluster-size estimate cannot prove +that every provider record was examined. + +The warm lookup also needs one wall-clock deadline. The cardinality ceiling +alone is not a time bound: probing a large stale provider population could take +longer than the caller can wait. The deadline is authoritative. If it expires +before the target reaches the cluster-size ceiling, the request enters +cold-start rather than continuing an unbounded warm search. + +Candidate probes need bounded concurrency so sequential dial time does not +scale linearly with the number of stale records. The concurrency and lookup +deadline must be measured and configurable. The existing expensive peer-body +transfer limit remains separate from the cheaper availability-probe limit. + +Expansion improves the chance of discovering replacement providers hidden +behind old records. It does not turn DHT lookup into a freshness guarantee, so +bounded cold-start and origin fallback remain required. + +### 3. Carry failed candidates through the request + +Chair resolution maintains a request-scoped set keyed by provider peer ID and +address. A candidate is added when its HEAD probe starts. Repeated DHT results +skip that candidate, while a changed address for the same peer remains eligible. +Candidates that could not start before the probe context expired are not marked +and can be tried in a later poll window. + +After an authoritative chair recheck reports a chair accepted or still +pulling, that chair's provider entries become eligible again. This permits the +same chair identity and address to transition from unavailable to available +without causing unrelated stale providers to be retried. + +Ordinary mirror GET failures continue to use the existing process-local stale, +unavailable, and suspicious-provider caches. Applying one request-scoped set to +the complete warm lookup and exponential expansion remains future work. + +### 4. Make chair completion explicit + +The current chair protocol reports both an already-running pull and an +already-cached digest as `PleasePullAlreadyPulling` +(`cmd/gantry/main.go:1700`). Add an explicit available outcome carrying the +chair holder's existing transfer endpoint, or add a chair status query that +can report these distinct states: + +- `pulling`: origin work is still in flight. +- `available`: the chair can open and serve the digest. +- `failed`: the pull finished unsuccessfully. +- `absent`: no pull is active and the digest is not available. + +Cold-start may wait while at least one accepted chair reports `pulling`. A DHT +result containing only request-scoped failed candidates must not end that +wait. + +When chairs report `available`, their endpoints provide authoritative initial +seeds. They should be merged with usable DHT candidates and distributed across +requesters rather than directing every requester to the first ranked chair. +The existing transfer concurrency limit and `429 Retry-After` behavior remain +the load-shedding boundary while successful requesters become additional DHT +providers. + +This design does not solve the separate coordination fan-in where many +requesters contact the same ranked chair cohort. It must not increase that +fan-in, and chair RPC scaling needs separate measurement and design work. + +### 5. Validate providers at recovery boundaries + +One narrow operation is shared by cold-start polling and the final +direct-origin-fallback recheck: + +```text +find usable provider(digest, excluded candidates) + collect bounded DHT candidates + remove self and suppressed/excluded candidates + probe candidates within a bounded budget + return success only after a peer confirms the digest is available +``` + +A successful peer metadata `HEAD` establishes that the transfer endpoint was +reachable and reported the digest available at probe time without transferring +the layer. It does not guarantee that the subsequent `GET` will succeed. A +failed `GET` continues through the mirror's existing peer failure handling. + +The helper probes at most four candidates concurrently and stops after the first +success. The direct-origin-fallback recheck gives DHT lookup plus all HEAD probes +one three-second budget. It declines fallback only for a provider that answers +HEAD successfully, not because an unprobed stale record exists. + +### 6. Add a hard cold-start deadline + +Cold-start must receive one absolute deadline when it begins. The deadline +covers the complete operation, including: + +- Chair snapshot and refresh calls. +- Initial and backup chair dispatch. +- Waiting while a chair reports active origin work. +- DHT polling for newly available providers. +- Chair status rechecks and patience rounds. + +Per-kind stall windows may decide when to recheck or move to a backup cohort, +but each sub-operation must be capped by the remaining cold-start time. No +sub-operation may create a new deadline beyond the original absolute deadline. + +Until that deadline, cold-start follows authoritative state: + +```text +usable peer found + -> fetch from peer + +chair reports pulling + -> wait within the bounded chair/request deadline + +chair reports available + -> use the available chair set as initial peer seeds + +no chair pulling or available, and no usable DHT provider before deadline + -> coldstart.ErrExhausted + -> mirror.ErrColdStartExhausted + -> direct origin fallback +``` + +An empty DHT result, DHT error, DHT timeout, stale-only result, chair RPC error, +chair turnover, completed-but-undiscoverable chair, or exhausted backup cohort +must all converge on this same deadline. None may create a terminal state that +permanently bypasses direct origin fallback. + +### 7. Make origin fallback eventual + +The current direct-origin-fallback controller can decline because of bootstrap, +DHT health, local in-flight work, rate limiting, or its final DHT recheck. That +is useful before the hard cold-start deadline, but it does not satisfy the +liveness requirement after the deadline. + +After the hard deadline: + +- A stale or unprobed DHT record cannot decline fallback. +- Bootstrap and DHT-health gates cannot permanently veto fallback. +- Rate limiting and jitter may schedule fallback only within a separately + bounded escape window; they cannot return the request to an indefinite + peer-discovery loop. +- Per-node per-digest in-flight dedup remains valid while a local origin pull is + actually running. A waiter must follow that pull to completion or failure + within the request deadline rather than receiving an unbounded series of + declines. +- If no usable provider materializes, one direct origin attempt must begin by + the end of the escape window. + +This requirement prioritizes liveness over the strict origin-protection goal. +In the worst case, many nodes can reach the hard deadline together and pull the +same digest from origin. Jitter, chair coordination, usable-provider rechecks, +and local in-flight dedup reduce that amplification, but they cannot both +guarantee a direct origin attempt for every isolated requester and guarantee a +single cluster-wide origin pull during a total coordination failure. + +"Regardless of failure scenario" here means failures in DHT discovery, peer +availability, chair state, or chair coordination. It does not mean retrying an +origin operation indefinitely after origin itself returns a terminal response +such as an authorization failure or a confirmed not-found response. + +## Why Simpler Changes Are Insufficient + +### Shorter validity alone + +It bounds stale-record lifetime but permits the same failure until expiry. It +also does not fix false success in chair polling or direct-origin-fallback +recheck. + +### Returning more than 20 providers alone + +It improves the chance of finding a new provider but cannot guarantee one. A +large replacement can leave more stale providers than any practical fixed +candidate limit. Exponential expansion therefore ends at both a cluster-size +ceiling and a wall-clock deadline, then moves to cold-start. + +### Requester-local suppression alone + +The DHT can return the same suppressed records repeatedly. Other requesters +must learn the same failures independently. + +### Fetching only from chairs + +It restores liveness but turns the fixed chair cohort into the data plane. The +normal path must still spread through newly completed requesters and other +providers. + +### Treating DHT results as the newest records + +The provider protocol does not expose or order records by publication time. +There is no "newest providers" query to request. + +## Rollout Considerations + +Provider validity is enforced by the DHT peers storing provider records. A +mixed-version rollout can therefore contain peers using both the old and new +validity settings. The implementation must not assume that lowering the local +setting immediately removes all old records. + +Whether old and new Gantry versions must coexist through the entire rollout is +an open requirement. This document does not introduce a new DHT protocol +namespace or a compatibility bridge without that requirement being decided. +Regardless of rollout policy, usability-based recovery remains necessary. + +The incorrect 24-hour/12-hour documentation must be updated in the same +implementation change so operators can reason from configured values rather +than dependency defaults. + +## Metrics + +At minimum, add or retain measurements for: + +- DHT candidates returned, filtered, probed, and usable per lookup. +- Warm lookup target and round count, including whether the target doubled or + reached the cluster-size ceiling. +- Lookups where every returned provider was request-scoped stale or + unavailable. +- Unique provider identities observed across rediscovery rounds. +- Chair states observed after dispatch: pulling, available, failed, absent. +- Cold-start elapsed time and terminal reason, including hard-deadline expiry. +- Direct-origin-fallback rechecks separated into usable hit, stale-only, empty, + and lookup error. +- Origin fallback attempts that became mandatory after the hard deadline. +- Sweeping-provider operations, failures, queue depth, schedule lag, and + duration by node and keyspace region. +- Per-node and cluster-wide reprovide publication rate, including burst size. +- Time from a chair's successful commit to first successful peer fetch. + +## Validation + +Unit and integration coverage must include: + +1. A warm lookup expands through `20, 40, 80, ...` and never probes the same + provider twice in one request. +2. A usable provider outside the first 20 is found before cold-start. +3. Reaching the configured cluster-size ceiling without a usable provider + enters cold-start. +4. Expiring the warm lookup deadline before reaching the cardinality ceiling + enters cold-start. +5. Twenty stale providers do not make chair polling report success. +6. Twenty stale providers do not make direct-origin-fallback recheck decline. +7. A chair reporting `pulling` extends the wait without starting duplicate + origin work. +8. A chair reporting `available` becomes a usable seed without requiring its + provider record to appear in the first DHT result. +9. A chair reporting neither pulling nor available allows + `ErrColdStartExhausted` and the direct-origin-fallback gates to run. +10. DHT errors, stale-only responses, chair RPC failures, chair turnover, and + exhausted backup cohorts all reach the same hard cold-start deadline. +11. Once the hard deadline expires, bootstrap, DHT health, stale records, and + token exhaustion cannot return the request to indefinite discovery. +12. A mandatory origin attempt starts before the bounded escape window ends + when no usable provider appears. +13. Live providers remain discoverable across multiple six-hour validity + windows through three-hour reprovide. +14. Reprovide work is distributed across the interval rather than emitted as a + full-inventory burst. +15. Across a representative population of persistent peer IDs, reprovide work + for the same keyspace region is distributed across the cycle and does not + form one synchronized cluster-wide burst. +16. The maximum measured gap between successful reprovides remains below the + provider validity after accounting for schedule lag and retry delay. +17. Restarting the provider rebuilds its in-memory key set from containerd + inventory and starts a fresh schedule without requiring continuity from the + previous process. +18. A DHT peer stops serving a departed provider after the configured validity, + independently of when physical cleanup deletes the stored entry. +19. A simulated complete provider-node replacement during a mixed-version + 48-hour validity window, where old providers wrote records into replacement + DHT stores during overlap, makes progress through current chairs or + controlled origin fallback even when every initial DHT candidate is stale. +20. Concurrent requesters distribute available chair seeds rather than all + selecting one transfer endpoint. + +## Open Questions + +1. What warm lookup deadline and probe concurrency give acceptable lookup cost + before cold-start? +2. Should the warm lookup ceiling have a dedicated configuration value or reuse + the chair cluster-size estimate? +3. Can a six-hour validity and three-hour reprovide interval sustain the + measured digest inventory and cluster size? +4. What maximum random startup phase prevents synchronized restart bursts while + keeping the worst-case refresh gap below provider validity? +5. Should chair availability be represented by a new `please_pull` outcome or + a separate status RPC? +6. Must mixed-version agents share one DHT throughout rollout? +7. What bounded escape window limits synchronized origin load while still + guaranteeing an origin attempt after cold-start expiry? +8. What bounded policy should distribute requesters among available chair + seeds without increasing chair coordination fan-in? \ No newline at end of file diff --git a/go.mod b/go.mod index d57493a48..c3e77bfdd 100644 --- a/go.mod +++ b/go.mod @@ -153,6 +153,7 @@ require ( github.com/filecoin-project/go-clock v0.1.0 // indirect github.com/flynn/noise v1.1.0 // indirect github.com/fxamacker/cbor/v2 v2.9.1 // indirect + github.com/gammazero/deque v1.2.1 // indirect github.com/go-errors/errors v1.4.2 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/go-logr/zapr v1.3.0 // indirect @@ -181,12 +182,14 @@ require ( github.com/google/gopacket v1.1.19 // indirect github.com/google/uuid v1.6.0 // indirect github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect + github.com/guillaumemichel/reservedpool v0.3.0 // indirect github.com/hashicorp/golang-lru v1.0.2 // indirect github.com/huandu/xstrings v1.5.0 // indirect github.com/huin/goupnp v1.3.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/ipfs/boxo v0.41.0 // indirect github.com/ipfs/go-datastore v0.9.2 // indirect + github.com/ipfs/go-libdht v0.5.0 // indirect github.com/ipfs/go-log/v2 v2.9.2 // indirect github.com/ipld/go-ipld-prime v0.24.0 // indirect github.com/jackpal/go-nat-pmp v1.0.2 // indirect diff --git a/go.sum b/go.sum index 51b96f4ee..17ef454d6 100644 --- a/go.sum +++ b/go.sum @@ -61,6 +61,8 @@ github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03 github.com/BurntSushi/toml v1.5.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU= github.com/DATA-DOG/go-sqlmock v1.5.2/go.mod h1:88MAG/4G7SMwSE3CeA0ZKzrT5CiOU3OJ+JlNzwDqpNU= +github.com/DataDog/zstd v1.5.7 h1:ybO8RBeh29qrxIhCA9E8gKY6xfONU9T6G6aP9DTKfLE= +github.com/DataDog/zstd v1.5.7/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw= github.com/GoogleCloudPlatform/confidential-space/server v0.0.0-20260522213940-e5c6d01a3007 h1:DoeEFwEGBdqcawmpiWtSsSVVZ+wk3zpqvcvssO2JLmY= github.com/GoogleCloudPlatform/confidential-space/server v0.0.0-20260522213940-e5c6d01a3007/go.mod h1:s8F0JYEods/WL03WxZaGsWCnumZeeLD+WKHzspOV9u0= github.com/Masterminds/goutils v1.1.1 h1:5nUrii3FMTL5diU80unEVvNevw1nH4+ZV4DSLVJLSYI= @@ -73,6 +75,10 @@ github.com/Microsoft/go-winio v0.6.3-0.20251027160822-ad3df93bed29 h1:0kQAzHq8vL github.com/Microsoft/go-winio v0.6.3-0.20251027160822-ad3df93bed29/go.mod h1:ZWa7ssZJT30CCDGJ7fk/2SBTq9BIQrrVjrcss0UW2s0= github.com/Microsoft/hcsshim v0.15.0-rc.1 h1:FbbwtQmiD+BVHynGkx5S65JkLyhkEiiTP8nrpmg2SZw= github.com/Microsoft/hcsshim v0.15.0-rc.1/go.mod h1:HWvvUPIy9HF6LotILj1G4VyS065rcLQ6tqj6tMUdOfI= +github.com/RaduBerinde/axisds v0.1.0 h1:YItk/RmU5nvlsv/awo2Fjx97Mfpt4JfgtEVAGPrLdz8= +github.com/RaduBerinde/axisds v0.1.0/go.mod h1:UHGJonU9z4YYGKJxSaC6/TNcLOBptpmM5m2Cksbnw0Y= +github.com/RaduBerinde/btreemap v0.0.0-20250419174037-3d62b7205d54 h1:bsU8Tzxr/PNz75ayvCnxKZWEYdLMPDkUgticP4a4Bvk= +github.com/RaduBerinde/btreemap v0.0.0-20250419174037-3d62b7205d54/go.mod h1:0tr7FllbE9gJkHq7CVeeDDFAFKQVy5RnCSSNBOvdqbc= github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ= github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw= github.com/apex/log v1.9.0 h1:FHtw/xuaM8AgmvDDTI9fiwoAL25Sq2cxojnZICUU8l0= @@ -141,6 +147,20 @@ github.com/cilium/ebpf v0.22.0 h1:v2ktp0roffpMOj2MMf3idtCQZOsAoC4BJbAJN+ke2bY= github.com/cilium/ebpf v0.22.0/go.mod h1:CDzZbe2hC5JjlDC+CY3KFCzlYwN4gbxppYM+Z10bQt4= github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw= github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc= +github.com/cockroachdb/crlib v0.0.0-20241112164430-1264a2edc35b h1:SHlYZ/bMx7frnmeqCu+xm0TCxXLzX3jQIVuFbnFGtFU= +github.com/cockroachdb/crlib v0.0.0-20241112164430-1264a2edc35b/go.mod h1:Gq51ZeKaFCXk6QwuGM0w1dnaOqc/F5zKT2zA9D6Xeac= +github.com/cockroachdb/errors v1.11.3 h1:5bA+k2Y6r+oz/6Z/RFlNeVCesGARKuC6YymtcDrbC/I= +github.com/cockroachdb/errors v1.11.3/go.mod h1:m4UIW4CDjx+R5cybPsNrRbreomiFqt8o1h1wUVazSd8= +github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b h1:r6VH0faHjZeQy818SGhaone5OnYfxFR/+AzdY3sf5aE= +github.com/cockroachdb/logtags v0.0.0-20230118201751-21c54148d20b/go.mod h1:Vz9DsVWQQhf3vs21MhPMZpMGSht7O/2vFW2xusFUVOs= +github.com/cockroachdb/pebble/v2 v2.1.6 h1:GDo7Z2+LgFZ7LJLdLmBXhDeTVIwgSPGxIT15hE7vGqM= +github.com/cockroachdb/pebble/v2 v2.1.6/go.mod h1:Reo1RTniv1UjVTAu/Fv74y5i3kJ5gmVrPhO9UtFiKn8= +github.com/cockroachdb/redact v1.1.5 h1:u1PMllDkdFfPWaNGMyLD1+so+aq3uUItthCFqzwPJ30= +github.com/cockroachdb/redact v1.1.5/go.mod h1:BVNblN9mBWFyMyqK1k3AAiSxhvhfK2oOZZ2lK+dpvRg= +github.com/cockroachdb/swiss v0.0.0-20251224182025-b0f6560f979b h1:VXvSNzmr8hMj8XTuY0PT9Ane9qZGul/p67vGYwl9BFI= +github.com/cockroachdb/swiss v0.0.0-20251224182025-b0f6560f979b/go.mod h1:yBRu/cnL4ks9bgy4vAASdjIW+/xMlFwuHKqtmh3GZQg= +github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 h1:zuQyyAKVxetITBuuhv3BI9cMrmStnpT18zmgmTxunpo= +github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06/go.mod h1:7nc4anLGjupUW/PeY5qiNYsdNXj7zopG+eqsS7To5IQ= github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA= github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= github.com/containerd/cgroups/v3 v3.1.3 h1:eUNflyMddm18+yrDmZPn3jI7C5hJ9ahABE5q6dyLYXQ= @@ -229,6 +249,10 @@ github.com/fsnotify/fsnotify v1.10.1 h1:b0/UzAf9yR5rhf3RPm9gf3ehBPpf0oZKIjtpKrx5 github.com/fsnotify/fsnotify v1.10.1/go.mod h1:TLheqan6HD6GBK6PrDWyDPBaEV8LspOxvPSjC+bVfgo= github.com/fxamacker/cbor/v2 v2.9.1 h1:2rWm8B193Ll4VdjsJY28jxs70IdDsHRWgQYAI80+rMQ= github.com/fxamacker/cbor/v2 v2.9.1/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= +github.com/gammazero/deque v1.2.1 h1:9fnQVFCCZ9/NOc7ccTNqzoKd1tCWOqeI05/lPqFPMGQ= +github.com/gammazero/deque v1.2.1/go.mod h1:5nSFkzVm+afG9+gy0VIowlqVAW4N8zNcMne+CMQVD2g= +github.com/getsentry/sentry-go v0.27.0 h1:Pv98CIbtB3LkMWmXi4Joa5OOcwbmnX88sF5qbK3r3Ps= +github.com/getsentry/sentry-go v0.27.0/go.mod h1:lc76E2QywIyW8WuBnwl8Lc4bkmQH4+w1gwTf25trprY= github.com/go-errors/errors v1.4.2 h1:J6MZopCL4uSllY1OfXM374weqZFFItUbrImctkmUxIA= github.com/go-errors/errors v1.4.2/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og= github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk= @@ -284,6 +308,8 @@ github.com/gobuffalo/flect v1.0.3 h1:xeWBM2nui+qnVvNM4S3foBhCAL2XgPU+a7FdpelbTq4 github.com/gobuffalo/flect v1.0.3/go.mod h1:A5msMlrHtLqh9umBSnvabjsMrCcCpAyzglnDvkbYKHs= github.com/gofrs/flock v0.10.0 h1:SHMXenfaB03KbroETaCMtbBg3Yn29v4w1r+tgy4ff4k= github.com/gofrs/flock v0.10.0/go.mod h1:FirDy1Ing0mI2+kB6wk+vyyAH+e6xiE+EYA0jnzV9jc= +github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY= github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= @@ -304,6 +330,8 @@ github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/golang/snappy v0.0.5-0.20231225225746-43d5d4cd4e0e h1:4bw4WeyTYPp0smaXiJZCNnLrvVBqirQVreixayXezGc= +github.com/golang/snappy v0.0.5-0.20231225225746-43d5d4cd4e0e/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= github.com/google/btree v1.1.3 h1:CVpQJjYgC4VbzxeGVHfvZrv1ctoYCAI8vbl07Fcxlyg= github.com/google/btree v1.1.3/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4= github.com/google/cel-go v0.29.2 h1:ZtDxkeiMmz0mxbKDYiNkE5Lk7V5edMRcaaDf2jX002k= @@ -363,6 +391,8 @@ github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 h1:JeSE6pjso5T github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674/go.mod h1:r4w70xmWCQKmi1ONH4KIaBptdivuRPyosB9RmPlGEwA= github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk= github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs= +github.com/guillaumemichel/reservedpool v0.3.0 h1:eqqO/QvTllLBrit7LVtVJBqw4cD0WdV9ajUe7WNTajw= +github.com/guillaumemichel/reservedpool v0.3.0/go.mod h1:sXSDIaef81TFdAJglsCFCMfgF5E5Z5xK1tFhjDhvbUc= github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c= github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= @@ -386,6 +416,10 @@ github.com/ipfs/go-datastore v0.9.2 h1:HJOgAmvWPRMHiwD8JHBzGZQNTKhuFGYfp8bNPwye2 github.com/ipfs/go-datastore v0.9.2/go.mod h1:VIjDxnINIcCqBMaB8LGggHfYY7PalKWfPtRMFeOU4q4= github.com/ipfs/go-detect-race v0.0.1 h1:qX/xay2W3E4Q1U7d9lNs1sU9nvguX0a7319XbyQ6cOk= github.com/ipfs/go-detect-race v0.0.1/go.mod h1:8BNT7shDZPo99Q74BpGMK+4D8Mn4j46UU0LZ723meps= +github.com/ipfs/go-ds-pebble v0.5.12 h1:idO/w4i3IBA6vZtVWsyG5IlPIgwd62iUaQZBl/Kv+yI= +github.com/ipfs/go-ds-pebble v0.5.12/go.mod h1:H2zy28KMQSiAflUxpKzKHqbpSHRWPZS5/bi4ymAJOjY= +github.com/ipfs/go-libdht v0.5.0 h1:ZN+eCqwahZvUeT0e4DsIxRtm78Mc9UR5tmZUiMsrGjQ= +github.com/ipfs/go-libdht v0.5.0/go.mod h1:L3YiuFXecLeZZFuuVRM0hjg1GgVhARzUdahFsuqSa7w= github.com/ipfs/go-log/v2 v2.9.2 h1:O/5BB0elpkRILvT24rCJ5976wWd7u0nJ436T3rdYdc4= github.com/ipfs/go-log/v2 v2.9.2/go.mod h1:RziRwwXWhndlk8L75RnEe0zeAYaq2heKtEMc3jqUov0= github.com/ipfs/go-test v0.4.1 h1:n6uNSakIgpTQIRorqNg2O02aMIFDLQk2z4rBfrlD3Uw= @@ -495,6 +529,8 @@ github.com/mikioh/tcpinfo v0.0.0-20190314235526-30a79bb1804b/go.mod h1:lxPUiZwKo github.com/mikioh/tcpopt v0.0.0-20190314235656-172688c1accc h1:PTfri+PuQmWDqERdnNMiD9ZejrlswWrCpBEZgWOiTrc= github.com/mikioh/tcpopt v0.0.0-20190314235656-172688c1accc/go.mod h1:cGKTAVKx4SxOuR/czcZ/E2RSJ3sfHs8FpHhQ5CWMf9s= github.com/minio/blake2b-simd v0.0.0-20160723061019-3f5f724cb5b1/go.mod h1:pD8RvIylQ358TN4wwqatJ8rNavkEINozVn9DtGI3dfQ= +github.com/minio/minlz v1.0.1-0.20250507153514-87eb42fe8882 h1:0lgqHvJWHLGW5TuObJrfyEi6+ASTKDBWikGvPqy9Yiw= +github.com/minio/minlz v1.0.1-0.20250507153514-87eb42fe8882/go.mod h1:qT0aEB35q79LLornSzeDH75LBf3aH1MV+jB5w9Wasec= github.com/minio/sha256-simd v0.1.1-0.20190913151208-6de447530771/go.mod h1:B5e1o+1/KgNmWrSQK08Y6Z1Vb5pwIktudl0J58iy0KM= github.com/minio/sha256-simd v1.0.1 h1:6kaan5IFmwTNynnKKpDHe6FWHohJOHhCPchzK49dzMM= github.com/minio/sha256-simd v1.0.1/go.mod h1:Pz6AKMiUdngCLpeTL/RJY1M9rUuPMYujV5xJjtbRSN8= diff --git a/internal/gantry/advertise/advertise.go b/internal/gantry/advertise/advertise.go index 2e48e36c6..7682e6910 100644 --- a/internal/gantry/advertise/advertise.go +++ b/internal/gantry/advertise/advertise.go @@ -9,7 +9,7 @@ // - The local containerd content store is the source of truth for // "what we can serve to peers". // - The DHT is a hint layer: provider records say "this node might -// have this digest", subject to ≤24 h TTL and eventual consistency. +// have this digest", subject to configured validity and eventual consistency. // - This package owns the local "announced set" - the digests we // currently believe are present-and-advertised - and reconciles it // against an inventory source (typically containerdstore.Inventory) @@ -27,10 +27,9 @@ // longer keep it alive. // // The announced set is local rebuildable state - it is NOT persisted -// across process restarts. On startup the first reconcile pass -// re-Provides every present digest, which is the same operation -// libp2p performs internally on its 12 h refresh schedule, so the -// extra cost is bounded by inventory size. +// across process restarts. On startup the first reconcile pass registers every +// present digest with discovery's sweeping provider. The sweeper owns periodic +// refresh for the rest of the process lifetime. package advertise import ( diff --git a/internal/gantry/coldstart/chair.go b/internal/gantry/coldstart/chair.go index 93a135223..1097f9b76 100644 --- a/internal/gantry/coldstart/chair.go +++ b/internal/gantry/coldstart/chair.go @@ -17,6 +17,7 @@ import ( "github.com/Azure/unbounded/internal/gantry/digest" "github.com/Azure/unbounded/internal/gantry/ifaces" "github.com/Azure/unbounded/internal/gantry/inflight" + "github.com/Azure/unbounded/internal/gantry/providerprobe" "github.com/Azure/unbounded/internal/gantry/registryauth" ) @@ -35,8 +36,10 @@ type ChairClaimer interface { } type ChairOptions struct { - Chairs ChairSnapshotCache - Discovery Discovery + Chairs ChairSnapshotCache + Discovery Discovery + // PeerMetadata verifies DHT candidates before they can complete polling. + PeerMetadata ifaces.PeerMetadataDialer Coord ifaces.ChairCoordinator LocalPull ifaces.LocalChairPullStarter Inflight *inflight.Map @@ -81,6 +84,10 @@ func NewChairResolver(opts ChairOptions) *ChairResolver { panic("coldstart.NewChairResolver: Discovery is required") } + if opts.PeerMetadata == nil { + panic("coldstart.NewChairResolver: PeerMetadata is required") + } + if opts.Coord == nil { panic("coldstart.NewChairResolver: Coord is required") } @@ -155,6 +162,7 @@ func (r *ChairResolver) Resolve(ctx context.Context, d digest.Digest, kind iface } accepted := make([]chairs.Chair, 0, r.opts.SeedCount) + attemptedProviders := providerprobe.Attempted{} sawTransientFailure := false // Recruit the seed cohort in one pass. A chair that does not answer inside @@ -196,7 +204,12 @@ func (r *ChairResolver) Resolve(ctx context.Context, d digest.Digest, kind iface patience := 0 for { - providers, err := r.pollDHT(ctx, d, kind, expectedSize) + providers, err := r.pollDHT(ctx, ifaces.OriginRef{ + Registry: registry, + Repository: repository, + Digest: d, + Kind: kind, + }, expectedSize, attemptedProviders) if err == nil { return &Resolution{Providers: providers, Outcome: "chair_cold_start"}, nil } @@ -221,6 +234,7 @@ func (r *ChairResolver) Resolve(ctx context.Context, d digest.Digest, kind iface // that is already running. The caller's context bounds the wait. if len(recheck.stillPulling) > 0 { accepted = recheck.accepted + forgetChairProviders(attemptedProviders, accepted) continue } @@ -231,6 +245,7 @@ func (r *ChairResolver) Resolve(ctx context.Context, d digest.Digest, kind iface if len(recheck.accepted) > 0 && patience < maxStillPullingRounds { patience++ accepted = recheck.accepted + forgetChairProviders(attemptedProviders, accepted) continue } @@ -618,13 +633,13 @@ func chairCallOutcome(err error) string { } } -func (r *ChairResolver) pollDHT(ctx context.Context, d digest.Digest, kind ifaces.OriginRefKind, expectedSize int64) ([]ifaces.Provider, error) { +func (r *ChairResolver) pollDHT(ctx context.Context, ref ifaces.OriginRef, expectedSize int64, attempted providerprobe.Attempted) ([]ifaces.Provider, error) { interval := r.opts.PollLayer - if kind == ifaces.KindManifest { + if ref.Kind == ifaces.KindManifest { interval = r.opts.PollManifest } - deadline := time.Now().Add(r.opts.Inflight.Stalls().ResolveStall(kind, expectedSize)) + deadline := time.Now().Add(r.opts.Inflight.Stalls().ResolveStall(ref.Kind, expectedSize)) pollCtx, cancel := context.WithDeadline(ctx, deadline) defer cancel() @@ -633,9 +648,14 @@ func (r *ChairResolver) pollDHT(ctx context.Context, d digest.Digest, kind iface defer ticker.Stop() for { - providers, err := r.opts.Discovery.FindProviders(pollCtx, d) + providers, err := r.opts.Discovery.FindProviders(pollCtx, ref.Digest) if err == nil && len(providers) > 0 { - return providers, nil + // Provider records are hints. Only a peer that currently answers HEAD + // for this digest can signal that chair seeding has completed. + provider, usable := providerprobe.First(pollCtx, r.opts.PeerMetadata, providers, ref, attempted, providerprobe.DefaultConcurrency) + if usable { + return []ifaces.Provider{provider}, nil + } } select { @@ -646,6 +666,17 @@ func (r *ChairResolver) pollDHT(ctx context.Context, d digest.Digest, kind iface } } +func forgetChairProviders(attempted providerprobe.Attempted, chairs []chairs.Chair) { + for provider := range attempted { + for _, chair := range chairs { + if provider.NodeID == chair.Holder.PeerID { + delete(attempted, provider) + break + } + } + } +} + func containsStaleChair(outcomes []ifaces.PleasePullOutcome) bool { for _, outcome := range outcomes { if outcome.Outcome == ifaces.PleasePullStaleChair { diff --git a/internal/gantry/coldstart/chair_test.go b/internal/gantry/coldstart/chair_test.go index 82581ded2..4fe4175f3 100644 --- a/internal/gantry/coldstart/chair_test.go +++ b/internal/gantry/coldstart/chair_test.go @@ -103,6 +103,39 @@ func (d *backupDiscovery) FindProviders(context.Context, digest.Digest) ([]iface func (*backupDiscovery) Health() float64 { return 1 } +type peerMetadataStub struct { + mu sync.Mutex + failures map[string]error + failureSequences map[string][]error + calls map[string]int +} + +func (s *peerMetadataStub) HeadFromPeer(_ context.Context, addr string, _ ifaces.OriginRef) (int64, string, error) { + s.mu.Lock() + defer s.mu.Unlock() + + if s.calls == nil { + s.calls = map[string]int{} + } + + s.calls[addr]++ + if len(s.failureSequences[addr]) > 0 { + err := s.failureSequences[addr][0] + s.failureSequences[addr] = s.failureSequences[addr][1:] + + return 1, "application/octet-stream", err + } + + return 1, "application/octet-stream", s.failures[addr] +} + +func (s *peerMetadataStub) callCount(addr string) int { + s.mu.Lock() + defer s.mu.Unlock() + + return s.calls[addr] +} + // Chairs that fail to reply are left out of the cohort rather than replaced. // Backfilling to SeedCount used to pull ranks 8..10 into the cohort, which put // three more nodes on the origin for a layer the top of the ranking was @@ -297,6 +330,65 @@ func TestChairResolverWaitsWhileSeedsAreStillPulling(t *testing.T) { } } +func TestChairResolverIgnoresStaleProviderUntilFreshProviderAppears(t *testing.T) { + d := digest.MustParse("sha256:1212121212121212121212121212121212121212121212121212121212121212") + stale := ifaces.Provider{NodeID: "stale", Addr: "stale:5001"} + fresh := ifaces.Provider{NodeID: "fresh", Addr: "fresh:5001"} + metadata := &peerMetadataStub{failures: map[string]error{stale.Addr: errors.New("unreachable")}} + resolver := newTestChairResolverWithMetadata( + &chairSnapshotStub{snapshot: fullChairSnapshot(8)}, + &chairCoordStub{}, + &stubDisco{providers: [][]ifaces.Provider{{stale}, {stale}, {fresh}}}, + metadata, + ) + + resolution, err := resolver.Resolve(context.Background(), d, ifaces.KindBlob, "registry.example.com", "repo/image", 0) + if err != nil { + t.Fatalf("Resolve: %v", err) + } + + if len(resolution.Providers) != 1 || resolution.Providers[0] != fresh { + t.Fatalf("providers = %+v, want fresh provider", resolution.Providers) + } + + if calls := metadata.callCount(stale.Addr); calls != 1 { + t.Fatalf("stale provider HEAD calls = %d, want 1", calls) + } + + if calls := metadata.callCount(fresh.Addr); calls != 1 { + t.Fatalf("fresh provider HEAD calls = %d, want 1", calls) + } +} + +func TestChairResolverReprobesAcceptedChairAfterProgressRecheck(t *testing.T) { + d := digest.MustParse("sha256:1313131313131313131313131313131313131313131313131313131313131313") + snapshot := fullChairSnapshot(8) + chair := chairs.Rank(snapshot, d)[0] + provider := ifaces.Provider{NodeID: chair.Holder.PeerID, Addr: chair.Holder.TransferAddr} + metadata := &peerMetadataStub{failureSequences: map[string][]error{ + provider.Addr: {errors.New("not available yet"), nil}, + }} + resolver := newTestChairResolverWithMetadata( + &chairSnapshotStub{snapshot: snapshot}, + &chairCoordStub{}, + &stubDisco{providers: [][]ifaces.Provider{{provider}}}, + metadata, + ) + + resolution, err := resolver.Resolve(context.Background(), d, ifaces.KindBlob, "registry.example.com", "repo/image", 0) + if err != nil { + t.Fatalf("Resolve: %v", err) + } + + if len(resolution.Providers) != 1 || resolution.Providers[0] != provider { + t.Fatalf("providers = %+v, want accepted chair provider", resolution.Providers) + } + + if calls := metadata.callCount(provider.Addr); calls != 2 { + t.Fatalf("accepted chair HEAD calls = %d, want 2", calls) + } +} + // The round-trip timer separates a deadline from slow transport: a chair that // never replies inside QueryTimeout must be recorded as "deadline", not folded // in with clean replies. @@ -314,9 +406,10 @@ func TestChairResolverTimesChairCallsByOutcome(t *testing.T) { var mu sync.Mutex resolver := coldstart.NewChairResolver(coldstart.ChairOptions{ - Chairs: &chairSnapshotStub{snapshot: snapshot}, - Discovery: &afterCoordCallsDiscovery{coord: coord, min: 0}, - Coord: coord, + Chairs: &chairSnapshotStub{snapshot: snapshot}, + Discovery: &afterCoordCallsDiscovery{coord: coord, min: 0}, + PeerMetadata: &peerMetadataStub{}, + Coord: coord, Inflight: inflight.New(inflight.Stalls{ ManifestConfig: 20 * time.Millisecond, LayerFloor: 20 * time.Millisecond, @@ -379,9 +472,10 @@ func TestChairResolverKeepsPartialSeedCohort(t *testing.T) { dispatch := map[string]int{} resolver := coldstart.NewChairResolver(coldstart.ChairOptions{ - Chairs: &chairSnapshotStub{snapshot: snapshot}, - Discovery: &afterCoordCallsDiscovery{coord: coord, min: 0}, - Coord: coord, + Chairs: &chairSnapshotStub{snapshot: snapshot}, + Discovery: &afterCoordCallsDiscovery{coord: coord, min: 0}, + PeerMetadata: &peerMetadataStub{}, + Coord: coord, Inflight: inflight.New(inflight.Stalls{ ManifestConfig: 20 * time.Millisecond, LayerFloor: 20 * time.Millisecond, @@ -437,9 +531,10 @@ func TestChairResolverReportsSeedRecruitmentDepth(t *testing.T) { var gotSelectable, gotContacted, gotAccepted int resolver := coldstart.NewChairResolver(coldstart.ChairOptions{ - Chairs: &chairSnapshotStub{snapshot: snapshot}, - Discovery: &afterCoordCallsDiscovery{coord: coord, min: 0}, - Coord: coord, + Chairs: &chairSnapshotStub{snapshot: snapshot}, + Discovery: &afterCoordCallsDiscovery{coord: coord, min: 0}, + PeerMetadata: &peerMetadataStub{}, + Coord: coord, Inflight: inflight.New(inflight.Stalls{ ManifestConfig: 20 * time.Millisecond, LayerFloor: 20 * time.Millisecond, @@ -474,10 +569,15 @@ func TestChairResolverReportsSeedRecruitmentDepth(t *testing.T) { } func newTestChairResolver(cache coldstart.ChairSnapshotCache, coord ifaces.ChairCoordinator, discovery coldstart.Discovery) *coldstart.ChairResolver { + return newTestChairResolverWithMetadata(cache, coord, discovery, &peerMetadataStub{}) +} + +func newTestChairResolverWithMetadata(cache coldstart.ChairSnapshotCache, coord ifaces.ChairCoordinator, discovery coldstart.Discovery, metadata ifaces.PeerMetadataDialer) *coldstart.ChairResolver { return coldstart.NewChairResolver(coldstart.ChairOptions{ - Chairs: cache, - Discovery: discovery, - Coord: coord, + Chairs: cache, + Discovery: discovery, + PeerMetadata: metadata, + Coord: coord, Inflight: inflight.New(inflight.Stalls{ ManifestConfig: 20 * time.Millisecond, LayerFloor: 20 * time.Millisecond, diff --git a/internal/gantry/config/config.go b/internal/gantry/config/config.go index 83bcb18de..196958382 100644 --- a/internal/gantry/config/config.go +++ b/internal/gantry/config/config.go @@ -139,6 +139,10 @@ type Config struct { // before it becomes a trim candidate. Libp2pConnManagerGrace time.Duration `yaml:"libp2p_conn_manager_grace"` + DHTProviderValidity time.Duration `yaml:"dht_provider_validity"` + DHTReprovideInterval time.Duration `yaml:"dht_reprovide_interval"` + DHTMaxReprovideDelay time.Duration `yaml:"dht_max_reprovide_delay"` + // ChairListen binds the HTTPS listener that serves cold-start please_pull. // Keeping the RPC off libp2p puts it on a connection pool the libp2p // connection and resource managers do not govern, so a trimmed DHT @@ -467,6 +471,9 @@ func NewDefault() *Config { Libp2pConnManagerHigh: 900, Libp2pConnManagerLow: 600, Libp2pConnManagerGrace: time.Minute, + DHTProviderValidity: 6 * time.Hour, + DHTReprovideInterval: 3 * time.Hour, + DHTMaxReprovideDelay: 10 * time.Minute, ChairListen: "0.0.0.0:5002", NodeName: "", @@ -610,6 +617,9 @@ func (c *Config) LoadEnv(env func(string) string) error { setInt("LIBP2P_CONN_MANAGER_HIGH", &c.Libp2pConnManagerHigh) setInt("LIBP2P_CONN_MANAGER_LOW", &c.Libp2pConnManagerLow) setDur("LIBP2P_CONN_MANAGER_GRACE", &c.Libp2pConnManagerGrace) + setDur("DHT_PROVIDER_VALIDITY", &c.DHTProviderValidity) + setDur("DHT_REPROVIDE_INTERVAL", &c.DHTReprovideInterval) + setDur("DHT_MAX_REPROVIDE_DELAY", &c.DHTMaxReprovideDelay) setStr("CHAIR_LISTEN", &c.ChairListen) setStr("NODE_NAME", &c.NodeName) @@ -690,6 +700,9 @@ func (c *Config) BindFlags(fs *flag.FlagSet) { fs.IntVar(&c.Libp2pConnManagerHigh, "libp2p-conn-manager-high", c.Libp2pConnManagerHigh, "libp2p connection count above which idle connections are trimmed") fs.IntVar(&c.Libp2pConnManagerLow, "libp2p-conn-manager-low", c.Libp2pConnManagerLow, "libp2p connection count that trimming settles at") fs.DurationVar(&c.Libp2pConnManagerGrace, "libp2p-conn-manager-grace", c.Libp2pConnManagerGrace, "minimum connection age before it becomes a trim candidate") + fs.DurationVar(&c.DHTProviderValidity, "dht-provider-validity", c.DHTProviderValidity, "time a DHT peer serves a provider record after its last publication") + fs.DurationVar(&c.DHTReprovideInterval, "dht-reprovide-interval", c.DHTReprovideInterval, "interval over which the sweeping provider refreshes all local content") + fs.DurationVar(&c.DHTMaxReprovideDelay, "dht-max-reprovide-delay", c.DHTMaxReprovideDelay, "maximum delay beyond the scheduled reprovide interval") fs.StringVar(&c.ChairListen, "chair-listen", c.ChairListen, "address for the HTTPS cold-start please_pull endpoint") fs.StringVar(&c.NodeName, "node-name", c.NodeName, "legacy no-op Kubernetes node name") @@ -974,6 +987,22 @@ func (c *Config) Validate() error { errs = append(errs, fmt.Errorf("libp2p_conn_manager_grace: must be >= 0, got %v", c.Libp2pConnManagerGrace)) } + if c.DHTProviderValidity <= 0 { + errs = append(errs, fmt.Errorf("dht_provider_validity: must be > 0, got %v", c.DHTProviderValidity)) + } + + if c.DHTReprovideInterval <= 0 { + errs = append(errs, fmt.Errorf("dht_reprovide_interval: must be > 0, got %v", c.DHTReprovideInterval)) + } + + if c.DHTMaxReprovideDelay <= 0 { + errs = append(errs, fmt.Errorf("dht_max_reprovide_delay: must be > 0, got %v", c.DHTMaxReprovideDelay)) + } + + if c.DHTReprovideInterval+c.DHTMaxReprovideDelay >= c.DHTProviderValidity { + errs = append(errs, fmt.Errorf("dht reprovide interval plus maximum delay must be less than provider validity: %v + %v >= %v", c.DHTReprovideInterval, c.DHTMaxReprovideDelay, c.DHTProviderValidity)) + } + if c.ChairListen == "" { errs = append(errs, errors.New("chair_listen: must be set")) } else if _, _, err := net.SplitHostPort(c.ChairListen); err != nil { diff --git a/internal/gantry/config/config_test.go b/internal/gantry/config/config_test.go index af189f308..3a33f6294 100644 --- a/internal/gantry/config/config_test.go +++ b/internal/gantry/config/config_test.go @@ -62,6 +62,10 @@ func TestDefaultsValidateAfterMinimalUpstream(t *testing.T) { t.Fatalf("ChairAPITimeout = %v, want 5s", c.ChairAPITimeout) } + if c.DHTProviderValidity != 6*time.Hour || c.DHTReprovideInterval != 3*time.Hour || c.DHTMaxReprovideDelay != 10*time.Minute { + t.Fatalf("DHT provider timing = %v/%v/%v, want 6h/3h/10m", c.DHTProviderValidity, c.DHTReprovideInterval, c.DHTMaxReprovideDelay) + } + // Defaults intentionally have no upstream registries - operator must // supply at least one. Seed one and re-validate. c.UpstreamRegistries = []UpstreamRegistry{ @@ -72,6 +76,32 @@ func TestDefaultsValidateAfterMinimalUpstream(t *testing.T) { } } +func TestValidateDHTProviderTiming(t *testing.T) { + for _, tc := range []struct { + name string + validity time.Duration + interval time.Duration + delay time.Duration + }{ + {name: "zero validity", validity: 0, interval: time.Minute, delay: time.Minute}, + {name: "zero interval", validity: time.Hour, interval: 0, delay: time.Minute}, + {name: "zero delay", validity: time.Hour, interval: time.Minute, delay: 0}, + {name: "refresh reaches validity", validity: time.Hour, interval: 50 * time.Minute, delay: 10 * time.Minute}, + } { + t.Run(tc.name, func(t *testing.T) { + c := NewDefault() + c.UpstreamRegistries = []UpstreamRegistry{{Name: "r", Endpoint: "https://r"}} + c.DHTProviderValidity = tc.validity + c.DHTReprovideInterval = tc.interval + c.DHTMaxReprovideDelay = tc.delay + + if err := c.Validate(); err == nil { + t.Fatal("Validate returned nil") + } + }) + } +} + func TestChairSeedCountConfig(t *testing.T) { t.Run("environment", func(t *testing.T) { c := NewDefault() diff --git a/internal/gantry/discovery/discovery.go b/internal/gantry/discovery/discovery.go index 6143da59e..637e74af1 100644 --- a/internal/gantry/discovery/discovery.go +++ b/internal/gantry/discovery/discovery.go @@ -47,6 +47,8 @@ import ( "github.com/ipfs/go-cid" "github.com/libp2p/go-libp2p" dht "github.com/libp2p/go-libp2p-kad-dht" + "github.com/libp2p/go-libp2p-kad-dht/provider" + "github.com/libp2p/go-libp2p-kad-dht/records" "github.com/libp2p/go-libp2p/core/connmgr" "github.com/libp2p/go-libp2p/core/crypto" "github.com/libp2p/go-libp2p/core/host" @@ -118,6 +120,10 @@ type Options struct { // ConnManagerGrace is the minimum age a connection must reach before it // becomes a trim candidate. Zero uses DefaultConnManagerGrace. ConnManagerGrace time.Duration + + ProviderValidity time.Duration + ReprovideInterval time.Duration + MaxReprovideDelay time.Duration } // Connection-manager defaults. go-libp2p's own defaults (160/192) are sized @@ -128,9 +134,12 @@ type Options struct { // plus peers with an in-flight transfer. That grows logarithmically in // cluster size, so these values do not scale with the fleet. const ( - DefaultConnManagerHigh = 900 - DefaultConnManagerLow = 600 - DefaultConnManagerGrace = time.Minute + DefaultConnManagerHigh = 900 + DefaultConnManagerLow = 600 + DefaultConnManagerGrace = time.Minute + DefaultProviderValidity = 6 * time.Hour + DefaultReprovideInterval = 3 * time.Hour + DefaultMaxReprovideDelay = 10 * time.Minute ) // DefaultTransferPort is the conventional peer-transfer port used when @@ -149,23 +158,27 @@ func FromConfig(c *config.Config) Options { } return Options{ - IdentityPath: c.Libp2pIdentityPath, - ListenAddrs: c.Libp2pListen, - BootstrapPeers: c.Libp2pBootstrapPeers, - ProtocolPrefix: "/gantry", - SelfTestPeriod: 60 * time.Second, - TransferPort: port, - ConnManagerHigh: c.Libp2pConnManagerHigh, - ConnManagerLow: c.Libp2pConnManagerLow, - ConnManagerGrace: c.Libp2pConnManagerGrace, + IdentityPath: c.Libp2pIdentityPath, + ListenAddrs: c.Libp2pListen, + BootstrapPeers: c.Libp2pBootstrapPeers, + ProtocolPrefix: "/gantry", + SelfTestPeriod: 60 * time.Second, + TransferPort: port, + ConnManagerHigh: c.Libp2pConnManagerHigh, + ConnManagerLow: c.Libp2pConnManagerLow, + ConnManagerGrace: c.Libp2pConnManagerGrace, + ProviderValidity: c.DHTProviderValidity, + ReprovideInterval: c.DHTReprovideInterval, + MaxReprovideDelay: c.DHTMaxReprovideDelay, } } // Host wraps a libp2p host + kad-dht and implements ifaces.DHT. type Host struct { - logger *slog.Logger - h host.Host - d *dht.IpfsDHT + logger *slog.Logger + h host.Host + d *dht.IpfsDHT + provider *provider.SweepingProvider // transferPort is the conventional peer-transfer port suffixed to // FindProviders results' IP. Captured from Options at New so @@ -295,7 +308,15 @@ func New(ctx context.Context, opts Options) (*Host, error) { return nil, fmt.Errorf("discovery: libp2p new: %w", err) } - dhtOpts := []dht.Option{dht.Mode(dht.ModeServer)} + providerValidity := opts.ProviderValidity + if providerValidity <= 0 { + providerValidity = DefaultProviderValidity + } + + dhtOpts := []dht.Option{ + dht.Mode(dht.ModeServer), + dht.ProviderManagerOpts(records.ProvideValidity(providerValidity)), + } if opts.ProtocolPrefix != "" { dhtOpts = append(dhtOpts, dht.ProtocolPrefix(protocol.ID(opts.ProtocolPrefix))) } @@ -312,7 +333,20 @@ func New(ctx context.Context, opts Options) (*Host, error) { logger.Warn("dht bootstrap kickoff returned err", slog.Any("err", err)) } - host := &Host{logger: logger, h: h, d: d} + reprovider, err := newSweepingProvider(d, opts) + if err != nil { + _ = d.Close() //nolint:errcheck // best-effort constructor rollback + _ = h.Close() //nolint:errcheck // best-effort constructor rollback + + return nil, fmt.Errorf("discovery: sweeping provider: %w", err) + } + + host := &Host{ + logger: logger, + h: h, + d: d, + provider: reprovider, + } host.transferPort = opts.TransferPort if host.transferPort == 0 { @@ -346,6 +380,31 @@ func New(ctx context.Context, opts Options) (*Host, error) { return host, nil } +func newSweepingProvider(d *dht.IpfsDHT, opts Options) (*provider.SweepingProvider, error) { + reprovideInterval := opts.ReprovideInterval + if reprovideInterval <= 0 { + reprovideInterval = DefaultReprovideInterval + } + + maxReprovideDelay := opts.MaxReprovideDelay + if maxReprovideDelay <= 0 { + maxReprovideDelay = DefaultMaxReprovideDelay + } + + return provider.New( + provider.WithHost(d.Host()), + provider.WithReplicationFactor(d.BucketSize()), + provider.WithSelfAddrs(d.FilteredAddrs), + provider.WithRouter(d), + provider.WithAddLocalRecord(func(ctx context.Context, h multihash.Multihash) error { + return d.Provide(ctx, cid.NewCidV1(cid.Raw, h), false) + }), + provider.WithMessageSender(d.MessageSender()), + provider.WithReprovideInterval(reprovideInterval), + provider.WithMaxReprovideDelay(maxReprovideDelay), + ) +} + // Close tears down the DHT and libp2p host. Safe to call multiple times. func (h *Host) Close() error { var err error @@ -361,12 +420,18 @@ func (h *Host) Close() error { <-h.selfTestDone } + if h.provider != nil { + if cerr := h.provider.Close(); cerr != nil { + err = errors.Join(err, cerr) + } + } + if cerr := h.d.Close(); cerr != nil { - err = cerr + err = errors.Join(err, cerr) } - if cerr := h.h.Close(); cerr != nil && err == nil { - err = cerr + if cerr := h.h.Close(); cerr != nil { + err = errors.Join(err, cerr) } }) @@ -434,19 +499,30 @@ func (h *Host) Provide(ctx context.Context, d digest.Digest) error { return err } - return h.d.Provide(ctx, c, true) + if err := h.d.Provide(ctx, c, true); err != nil { + return err + } + + if h.provider == nil { + return nil + } + + return h.provider.StartProviding(false, c.Hash()) } -// Withdraw implements ifaces.DHT. libp2p kad-dht has no protocol-level -// withdraw - provider records expire at the 24 h TTL. The advertiser -// achieves the same effect by simply not re-calling Provide for -// withdrawn digests on its next refresh tick, so this hook exists -// purely as a cooperation point for the interface contract and is a -// no-op today. Future work may emit a libp2p custom protocol message -// to peers in the routing table to evict the stale record sooner, but -// the plan explicitly accepts TTL drainage as adequate. -func (h *Host) Withdraw(_ context.Context, _ digest.Digest) error { - return nil +// Withdraw stops periodic reprovide. Existing remote records drain through +// provider validity because libp2p kad-dht has no protocol-level withdrawal. +func (h *Host) Withdraw(_ context.Context, d digest.Digest) error { + if h.provider == nil { + return nil + } + + c, err := DigestToCID(d) + if err != nil { + return err + } + + return h.provider.StopProviding(c.Hash()) } // FindProviders implements ifaces.DHT. Returns providers whose multiaddrs diff --git a/internal/gantry/discovery/discovery_test.go b/internal/gantry/discovery/discovery_test.go index 6ebd6f7cb..c2c2b5eac 100644 --- a/internal/gantry/discovery/discovery_test.go +++ b/internal/gantry/discovery/discovery_test.go @@ -134,6 +134,80 @@ func TestHostPersistsIdentity(t *testing.T) { } } +func TestHostProvideRegistersSweeperAndWithdrawRemoves(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + + first, err := New(ctx, Options{ + ListenAddrs: []string{"/ip4/127.0.0.1/tcp/0"}, + ProtocolPrefix: "/gantry-provider-lifecycle-test", + }) + if err != nil { + t.Fatalf("first New: %v", err) + } + + t.Cleanup(func() { _ = first.Close() }) + + p2p, err := multiaddr.NewMultiaddr("/p2p/" + first.PeerID().String()) + if err != nil { + t.Fatalf("peer multiaddr: %v", err) + } + + bootstrap := first.Addrs()[0].Encapsulate(p2p).String() + + second, err := New(ctx, Options{ + ListenAddrs: []string{"/ip4/127.0.0.1/tcp/0"}, + BootstrapPeers: []string{bootstrap}, + ProtocolPrefix: "/gantry-provider-lifecycle-test", + }) + if err != nil { + t.Fatalf("second New: %v", err) + } + + t.Cleanup(func() { _ = second.Close() }) + + convergenceDeadline := time.NewTimer(5 * time.Second) + convergenceTick := time.NewTicker(10 * time.Millisecond) + + defer convergenceDeadline.Stop() + defer convergenceTick.Stop() + + for second.RoutingTableSize() == 0 { + select { + case <-convergenceDeadline.C: + t.Fatal("second DHT routing table did not converge") + case <-convergenceTick.C: + } + } + + d := digest.MustParse("sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc") + if err := second.Provide(ctx, d); err != nil { + t.Fatalf("Provide: %v", err) + } + + stats, err := second.provider.Stats(ctx) + if err != nil { + t.Fatalf("provider Stats after Provide: %v", err) + } + + if stats.Schedule.Keys != 1 { + t.Fatalf("scheduled keys after Provide = %d, want 1", stats.Schedule.Keys) + } + + if err := second.Withdraw(ctx, d); err != nil { + t.Fatalf("Withdraw: %v", err) + } + + stats, err = second.provider.Stats(ctx) + if err != nil { + t.Fatalf("provider Stats after Withdraw: %v", err) + } + + if stats.Schedule.Keys != 0 { + t.Fatalf("scheduled keys after Withdraw = %d, want 0", stats.Schedule.Keys) + } +} + func TestTransferAddrWithPortSkipsLoopback(t *testing.T) { tests := []struct { name string diff --git a/internal/gantry/ifaces/ifaces.go b/internal/gantry/ifaces/ifaces.go index 7f26a370f..6460ae9ef 100644 --- a/internal/gantry/ifaces/ifaces.go +++ b/internal/gantry/ifaces/ifaces.go @@ -319,14 +319,14 @@ type DHT interface { FindProviders(ctx context.Context, d digest.Digest) ([]Provider, error) // Provide advertises that this node holds d. Idempotent at the DHT - // level; refreshing is the implementation's responsibility (libp2p - // default 12 h refresh, 24 h TTL - the design doc). + // level; the discovery implementation registers successful publications + // with its periodic sweeping provider. Provide(ctx context.Context, d digest.Digest) error // Withdraw is a soft "stop advertising" hint sent by the advertiser // when the digest is no longer present in the local content store // (e.g. containerd GC'd it). libp2p has no protocol-level withdraw - // - existing provider records expire at the 24 h TTL - so this is + // - existing provider records expire at configured validity - so this is // implementation-defined cooperation: at minimum the local agent // MUST stop re-Providing the digest on the next refresh cycle so // the stale record drains naturally. Returning a non-nil error diff --git a/internal/gantry/mirror/mirror.go b/internal/gantry/mirror/mirror.go index ddf945141..6b6ec0cc7 100644 --- a/internal/gantry/mirror/mirror.go +++ b/internal/gantry/mirror/mirror.go @@ -908,7 +908,12 @@ func (s *Server) serveDigest(w http.ResponseWriter, r *http.Request, upstream, r return } - proceed, release, err := s.nf5.Allow(ctx, d, kind, 0) + proceed, release, err := s.nf5.Allow(ctx, ifaces.OriginRef{ + Registry: upstream, + Repository: repo, + Digest: d, + Kind: kind, + }, 0) if release != nil { defer release() } diff --git a/internal/gantry/mirror/mirror_nf5_integration_test.go b/internal/gantry/mirror/mirror_nf5_integration_test.go index 71a8d354b..12e8916e6 100644 --- a/internal/gantry/mirror/mirror_nf5_integration_test.go +++ b/internal/gantry/mirror/mirror_nf5_integration_test.go @@ -145,7 +145,7 @@ func TestMirror_NF5_AllGatesPassServesFromOrigin(t *testing.T) { InBootstrap: func() bool { return false }, HealthyEnough: func() bool { return true }, ClusterSize: func() int { return 1 }, // no jitter - Recheck: func(context.Context, digest.Digest) bool { return false }, + Recheck: func(context.Context, ifaces.OriginRef) bool { return false }, OnFallback: func() { atomic.AddInt32(&fallbacks, 1) }, }) @@ -264,7 +264,7 @@ func TestMirror_NF5_RecheckHitAbortsAfterJitter(t *testing.T) { InBootstrap: func() bool { return false }, HealthyEnough: func() bool { return true }, ClusterSize: func() int { return 1 }, // no jitter - recheck still runs - Recheck: func(context.Context, digest.Digest) bool { return true }, + Recheck: func(context.Context, ifaces.OriginRef) bool { return true }, OnFallback: func() { t.Fatalf("must not fire when recheck hits") }, }) diff --git a/internal/gantry/mirror/nf5.go b/internal/gantry/mirror/nf5.go index c1366f768..993b35f04 100644 --- a/internal/gantry/mirror/nf5.go +++ b/internal/gantry/mirror/nf5.go @@ -50,7 +50,6 @@ import ( "sync" "time" - "github.com/Azure/unbounded/internal/gantry/digest" "github.com/Azure/unbounded/internal/gantry/ifaces" "github.com/Azure/unbounded/internal/gantry/inflight" ) @@ -105,11 +104,11 @@ type DirectOriginFallbackOptions struct { // concurrent direct-origin-fallback calls for the same digest collapse to one. Inflight *inflight.Map - // Recheck performs a final DHT + cache + peer probe at the end - // of the jitter window. Returns true if a provider materialized - // during jitter (direct-origin-fallback cancels and the caller retries the warm - // path). - Recheck func(context.Context, digest.Digest) bool + // Recheck performs a final DHT + peer metadata probe at the end of the + // jitter window. Returns true only if a provider currently reports ref + // available, in which case direct-origin-fallback cancels and the caller + // retries the warm path. + Recheck func(context.Context, ifaces.OriginRef) bool // OnFallback is invoked once per origin pull that direct-origin-fallback permits. // Maps to the design doc metric `p2p_origin_fallback_total`. @@ -170,7 +169,7 @@ func NewDirectOriginFallback(opts DirectOriginFallbackOptions) *DirectOriginFall } } -// Allow runs the direct-origin-fallback gating sequence for digest d. When it returns +// Allow runs the direct-origin-fallback gating sequence for ref. When it returns // (true, release, nil), the caller MUST invoke `release` once the // origin pull completes (success or failure) - this frees the // in-flight slot. The token has already been consumed; releasing @@ -183,10 +182,10 @@ func NewDirectOriginFallback(opts DirectOriginFallbackOptions) *DirectOriginFall // returns (false, nil, ctx.Err) and any in-flight handle is // released. // -// kind and expectedSize are forwarded to inflight.Map.Start so the +// ref.Kind and expectedSize are forwarded to inflight.Map.Start so the // in-flight entry carries enough context for the design doc stall detection // in case the direct-origin-fallback origin pull itself stalls. -func (n *DirectOriginFallbackController) Allow(ctx context.Context, d digest.Digest, kind ifaces.OriginRefKind, expectedSize int64) (bool, func(), error) { +func (n *DirectOriginFallbackController) Allow(ctx context.Context, ref ifaces.OriginRef, expectedSize int64) (bool, func(), error) { if n.opts.InBootstrap != nil && n.opts.InBootstrap() { n.decline("bootstrap_window") return false, nil, nil @@ -201,7 +200,7 @@ func (n *DirectOriginFallbackController) Allow(ctx context.Context, d digest.Dig // for atomicity: if it reports alreadyPulling, direct-origin-fallback declines so // the caller 5xxs and lets the existing pull complete and // publish. - handle, _, alreadyPulling := n.opts.Inflight.Start(d, kind, expectedSize) + handle, _, alreadyPulling := n.opts.Inflight.Start(ref.Digest, ref.Kind, expectedSize) if alreadyPulling { n.decline("in_flight") return false, nil, nil @@ -237,7 +236,7 @@ func (n *DirectOriginFallbackController) Allow(ctx context.Context, d digest.Dig // Final re-check: the warm path may have materialized during // jitter. Canceling here keeps `p2p_origin_fallback_total` // near zero even under chaos scenarios. - if n.opts.Recheck != nil && n.opts.Recheck(ctx, d) { + if n.opts.Recheck != nil && n.opts.Recheck(ctx, ref) { release() n.decline("recheck_hit") diff --git a/internal/gantry/mirror/nf5_test.go b/internal/gantry/mirror/nf5_test.go index 0733f141c..229052d3c 100644 --- a/internal/gantry/mirror/nf5_test.go +++ b/internal/gantry/mirror/nf5_test.go @@ -45,6 +45,15 @@ func nf5Digest(t *testing.T, b byte) digest.Digest { return d } +func nf5Ref(t *testing.T, b byte, kind ifaces.OriginRefKind) ifaces.OriginRef { + return ifaces.OriginRef{ + Registry: "registry.example.com", + Repository: "repo/image", + Digest: nf5Digest(t, b), + Kind: kind, + } +} + // TestNF5_DeclinesInBootstrapWindow asserts the bootstrap-window // suppression: direct-origin-fallback must not fire while the local DHT is still // converging. @@ -60,7 +69,7 @@ func TestNF5_DeclinesInBootstrapWindow(t *testing.T) { OnDecline: func(r string) { declineReason = r }, }) - proceed, _, err := ctrl.Allow(context.Background(), nf5Digest(t, 'a'), ifaces.KindBlob, 0) + proceed, _, err := ctrl.Allow(context.Background(), nf5Ref(t, 'a', ifaces.KindBlob), 0) if err != nil { t.Fatalf("Allow err = %v", err) } @@ -87,7 +96,7 @@ func TestNF5_DeclinesWhenUnhealthy(t *testing.T) { OnDecline: func(r string) { declineReason = r }, }) - proceed, _, _ := ctrl.Allow(context.Background(), nf5Digest(t, 'b'), ifaces.KindManifest, 0) + proceed, _, _ := ctrl.Allow(context.Background(), nf5Ref(t, 'b', ifaces.KindManifest), 0) if proceed { t.Fatalf("Allow proceed = true; want false (unhealthy)") } @@ -109,10 +118,10 @@ func TestNF5_DeclinesOnInflightCollision(t *testing.T) { OnFallback: func() {}, }) - d := nf5Digest(t, 'c') + ref := nf5Ref(t, 'c', ifaces.KindBlob) // First call grabs the in-flight slot. - proceed1, release1, err := ctrl.Allow(context.Background(), d, ifaces.KindBlob, 0) + proceed1, release1, err := ctrl.Allow(context.Background(), ref, 0) if err != nil || !proceed1 { t.Fatalf("first Allow: proceed=%v err=%v; want true/nil", proceed1, err) } @@ -129,7 +138,7 @@ func TestNF5_DeclinesOnInflightCollision(t *testing.T) { OnDecline: func(r string) { declineReason = r }, }) - proceed2, _, _ := ctrl2.Allow(context.Background(), d, ifaces.KindBlob, 0) + proceed2, _, _ := ctrl2.Allow(context.Background(), ref, 0) if proceed2 { t.Fatalf("second Allow: proceed=true; want false (in_flight)") } @@ -172,7 +181,7 @@ func TestNF5_TokenBucketExhausts(t *testing.T) { // Burn 2 tokens (distinct digests so dedup doesn't intervene). for i, b := range []byte{'d', 'e'} { - _, release, _ := ctrl.Allow(context.Background(), nf5Digest(t, b), ifaces.KindBlob, 0) + _, release, _ := ctrl.Allow(context.Background(), nf5Ref(t, b, ifaces.KindBlob), 0) if release == nil { t.Fatalf("call #%d: expected release fn, got nil", i) } @@ -180,7 +189,7 @@ func TestNF5_TokenBucketExhausts(t *testing.T) { release() } // Third call: empty bucket -> decline. - proceed, _, _ := ctrl.Allow(context.Background(), nf5Digest(t, 'f'), ifaces.KindBlob, 0) + proceed, _, _ := ctrl.Allow(context.Background(), nf5Ref(t, 'f', ifaces.KindBlob), 0) if proceed { t.Fatalf("3rd call: proceed=true; want false (rate_limited)") } @@ -206,7 +215,7 @@ func TestNF5_TokenBucketExhausts(t *testing.T) { // Advance clock 30s -> 30s × 2/60 = 1 token replenished. clock = clock.Add(30 * time.Second) - proceed, release, _ := ctrl.Allow(context.Background(), nf5Digest(t, '0'), ifaces.KindBlob, 0) + proceed, release, _ := ctrl.Allow(context.Background(), nf5Ref(t, '0', ifaces.KindBlob), 0) if !proceed { t.Fatalf("after 30s refill: proceed=false; want true") } @@ -222,7 +231,7 @@ func TestNF5_DeclinesAfterRecheckHit(t *testing.T) { InBootstrap: func() bool { return false }, HealthyEnough: func() bool { return true }, ClusterSize: func() int { return 1 }, // no jitter - recheck still runs - Recheck: func(context.Context, digest.Digest) bool { return true }, + Recheck: func(context.Context, ifaces.OriginRef) bool { return true }, OnFallback: func() { t.Fatalf("NF5 must not fire when recheck hits") }, }) @@ -233,13 +242,13 @@ func TestNF5_DeclinesAfterRecheckHit(t *testing.T) { InBootstrap: func() bool { return false }, HealthyEnough: func() bool { return true }, ClusterSize: func() int { return 1 }, - Recheck: func(context.Context, digest.Digest) bool { return true }, + Recheck: func(context.Context, ifaces.OriginRef) bool { return true }, OnFallback: func() { t.Fatalf("NF5 must not fire when recheck hits") }, OnDecline: func(r string) { reason = r }, }) _ = ctrl // first ctrl uses no decline hook //nolint:errcheck // best-effort - proceed, _, _ := ctrl2.Allow(context.Background(), nf5Digest(t, '1'), ifaces.KindBlob, 0) + proceed, _, _ := ctrl2.Allow(context.Background(), nf5Ref(t, '1', ifaces.KindBlob), 0) if proceed { t.Fatalf("proceed=true; want false (recheck_hit)") } @@ -260,11 +269,11 @@ func TestNF5_ProceedsWhenGatesPass(t *testing.T) { InBootstrap: func() bool { return false }, HealthyEnough: func() bool { return true }, ClusterSize: func() int { return 1 }, // no jitter - Recheck: func(context.Context, digest.Digest) bool { return false }, + Recheck: func(context.Context, ifaces.OriginRef) bool { return false }, OnFallback: func() { fallbacks++ }, }) - proceed, release, err := ctrl.Allow(context.Background(), nf5Digest(t, '2'), ifaces.KindBlob, 0) + proceed, release, err := ctrl.Allow(context.Background(), nf5Ref(t, '2', ifaces.KindBlob), 0) if err != nil { t.Fatalf("Allow err = %v", err) } @@ -300,7 +309,7 @@ func TestNF5_ContextCancelledDuringJitter(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() // cancel immediately - proceed, _, err := ctrl.Allow(ctx, nf5Digest(t, '3'), ifaces.KindBlob, 0) + proceed, _, err := ctrl.Allow(ctx, nf5Ref(t, '3', ifaces.KindBlob), 0) if proceed { t.Fatalf("proceed=true; want false (ctx canceled)") } diff --git a/internal/gantry/mirror/rediscover_test.go b/internal/gantry/mirror/rediscover_test.go index e59386897..999ddf298 100644 --- a/internal/gantry/mirror/rediscover_test.go +++ b/internal/gantry/mirror/rediscover_test.go @@ -188,6 +188,20 @@ func TestMirror_Rediscover_ColdExhaustedFlushesHeadersBeforeLateProvider(t *test t.Fatalf("peer calls before advertise = %d, want 0", got) } + coldStartDeadline := time.NewTimer(time.Second) + coldStartTick := time.NewTicker(time.Millisecond) + + defer coldStartDeadline.Stop() + defer coldStartTick.Stop() + + for atomic.LoadInt32(&coldStart.calls) == 0 { + select { + case <-coldStartDeadline.C: + t.Fatal("cold-start was not called before late provider injection") + case <-coldStartTick.C: + } + } + dht.Inject(d, ifaces.Provider{NodeID: "late-seed", Addr: lateAddr}) got, err := io.ReadAll(resp.Body) diff --git a/internal/gantry/providerprobe/providerprobe.go b/internal/gantry/providerprobe/providerprobe.go new file mode 100644 index 000000000..145ec8b28 --- /dev/null +++ b/internal/gantry/providerprobe/providerprobe.go @@ -0,0 +1,99 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +// Package providerprobe verifies that DHT provider candidates currently serve +// a requested digest. +package providerprobe + +import ( + "context" + + "github.com/Azure/unbounded/internal/gantry/ifaces" + "github.com/Azure/unbounded/internal/gantry/registryauth" +) + +const DefaultConcurrency = 4 + +// Attempted records provider identities and addresses already probed during +// one resolution. A changed address for the same peer remains eligible. +type Attempted map[ifaces.Provider]struct{} + +// First returns the first provider whose transfer endpoint answers HEAD for +// ref. Each previously unseen candidate is added to attempted before probing. +func First(ctx context.Context, dialer ifaces.PeerMetadataDialer, providers []ifaces.Provider, ref ifaces.OriginRef, attempted Attempted, concurrency int) (ifaces.Provider, bool) { + if dialer == nil { + return ifaces.Provider{}, false + } + + if attempted == nil { + attempted = Attempted{} + } + + if concurrency <= 0 { + concurrency = DefaultConcurrency + } + + candidates := make([]ifaces.Provider, 0, len(providers)) + queued := Attempted{} + + for _, provider := range providers { + if _, seen := attempted[provider]; seen { + continue + } + + if _, seen := queued[provider]; seen { + continue + } + + queued[provider] = struct{}{} + candidates = append(candidates, provider) + } + + type result struct { + provider ifaces.Provider + usable bool + } + + for start := 0; start < len(candidates); start += concurrency { + end := min(start+concurrency, len(candidates)) + probeCtx, cancel := context.WithCancel(ctx) + probeCtx = registryauth.WithoutAuthorization(probeCtx) + results := make(chan result, end-start) + + for _, provider := range candidates[start:end] { + attempted[provider] = struct{}{} + + go func() { + _, _, err := dialer.HeadFromPeer(probeCtx, provider.Addr, ref) + results <- result{provider: provider, usable: err == nil} + }() + } + + var usable ifaces.Provider + + found := false + + for range end - start { + probeResult := <-results + + if probeResult.usable && !found { + usable = probeResult.provider + found = true + + cancel() + } + } + + cancel() + + if found { + return usable, true + } + + if ctx.Err() != nil { + return ifaces.Provider{}, false + } + } + + return ifaces.Provider{}, false +} diff --git a/internal/gantry/providerprobe/providerprobe_test.go b/internal/gantry/providerprobe/providerprobe_test.go new file mode 100644 index 000000000..e019a9ac0 --- /dev/null +++ b/internal/gantry/providerprobe/providerprobe_test.go @@ -0,0 +1,145 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package providerprobe_test + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/Azure/unbounded/internal/gantry/digest" + "github.com/Azure/unbounded/internal/gantry/ifaces" + "github.com/Azure/unbounded/internal/gantry/providerprobe" + "github.com/Azure/unbounded/internal/gantry/registryauth" +) + +type metadataDialer struct { + mu sync.Mutex + errors map[string]error + calls map[string]int + auth []string +} + +func (d *metadataDialer) HeadFromPeer(ctx context.Context, addr string, _ ifaces.OriginRef) (int64, string, error) { + d.mu.Lock() + defer d.mu.Unlock() + + d.calls[addr]++ + d.auth = append(d.auth, registryauth.Authorization(ctx)) + + return 1, "application/octet-stream", d.errors[addr] +} + +func TestFirstDoesNotForwardRegistryAuthorization(t *testing.T) { + dialer := &metadataDialer{calls: map[string]int{}} + ctx := registryauth.WithAuthorization(context.Background(), "Bearer secret") + + _, ok := providerprobe.First(ctx, dialer, []ifaces.Provider{{NodeID: "peer", Addr: "peer:5001"}}, ifaces.OriginRef{}, providerprobe.Attempted{}, 1) + if !ok { + t.Fatal("First did not find usable provider") + } + + if len(dialer.auth) != 1 || dialer.auth[0] != "" { + t.Fatalf("probe authorization = %q, want empty", dialer.auth) + } +} + +func TestFirstSkipsAttemptedAndReturnsUsableProvider(t *testing.T) { + stale := ifaces.Provider{NodeID: "stale", Addr: "stale:5001"} + fresh := ifaces.Provider{NodeID: "fresh", Addr: "fresh:5001"} + dialer := &metadataDialer{ + errors: map[string]error{stale.Addr: errors.New("unreachable")}, + calls: map[string]int{}, + } + attempted := providerprobe.Attempted{} + ref := ifaces.OriginRef{ + Registry: "registry.example.com", + Repository: "repo/image", + Digest: digest.MustParse("sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + Kind: ifaces.KindBlob, + } + + provider, ok := providerprobe.First(context.Background(), dialer, []ifaces.Provider{stale, fresh}, ref, attempted, 1) + if !ok || provider != fresh { + t.Fatalf("First = (%+v, %v), want fresh provider", provider, ok) + } + + if len(attempted) != 2 { + t.Fatalf("attempted providers = %d, want 2", len(attempted)) + } + + _, ok = providerprobe.First(context.Background(), dialer, []ifaces.Provider{stale, fresh}, ref, attempted, 1) + if ok { + t.Fatal("second First found an already-attempted provider") + } + + if dialer.calls[stale.Addr] != 1 || dialer.calls[fresh.Addr] != 1 { + t.Fatalf("calls = %+v, want one probe per provider", dialer.calls) + } +} + +func TestFirstReturnsFalseWhenEveryProviderFails(t *testing.T) { + providers := []ifaces.Provider{ + {NodeID: "a", Addr: "a:5001"}, + {NodeID: "b", Addr: "b:5001"}, + } + dialer := &metadataDialer{ + errors: map[string]error{ + providers[0].Addr: errors.New("unreachable"), + providers[1].Addr: errors.New("not found"), + }, + calls: map[string]int{}, + } + + _, ok := providerprobe.First(context.Background(), dialer, providers, ifaces.OriginRef{}, providerprobe.Attempted{}, 2) + if ok { + t.Fatal("First found an unusable provider") + } +} + +func TestFirstProbesDuplicateProviderOnce(t *testing.T) { + provider := ifaces.Provider{NodeID: "peer", Addr: "peer:5001"} + dialer := &metadataDialer{calls: map[string]int{}} + + _, ok := providerprobe.First(context.Background(), dialer, []ifaces.Provider{provider, provider}, ifaces.OriginRef{}, providerprobe.Attempted{}, 2) + if !ok { + t.Fatal("First did not find usable provider") + } + + if calls := dialer.calls[provider.Addr]; calls != 1 { + t.Fatalf("HEAD calls = %d, want 1", calls) + } +} + +type blockingMetadataDialer struct{} + +func (blockingMetadataDialer) HeadFromPeer(ctx context.Context, _ string, _ ifaces.OriginRef) (int64, string, error) { + <-ctx.Done() + + return 0, "", ctx.Err() +} + +func TestFirstDoesNotMarkUnstartedCandidatesAfterTimeout(t *testing.T) { + first := ifaces.Provider{NodeID: "first", Addr: "first:5001"} + second := ifaces.Provider{NodeID: "second", Addr: "second:5001"} + attempted := providerprobe.Attempted{} + ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond) + + defer cancel() + + _, ok := providerprobe.First(ctx, blockingMetadataDialer{}, []ifaces.Provider{first, second}, ifaces.OriginRef{}, attempted, 1) + if ok { + t.Fatal("First found a provider after timeout") + } + + if _, found := attempted[first]; !found { + t.Fatal("started provider missing from attempted set") + } + + if _, found := attempted[second]; found { + t.Fatal("unstarted provider was added to attempted set") + } +} diff --git a/internal/gantry/registryauth/registryauth.go b/internal/gantry/registryauth/registryauth.go index ed15be1d3..8ad646fb4 100644 --- a/internal/gantry/registryauth/registryauth.go +++ b/internal/gantry/registryauth/registryauth.go @@ -77,6 +77,12 @@ func Authorization(ctx context.Context) string { return authorization } +// WithoutAuthorization returns a child context that preserves cancellation, +// deadlines, and unrelated values while masking delegated registry identity. +func WithoutAuthorization(ctx context.Context) context.Context { + return context.WithValue(ctx, contextKey{}, "") +} + // Detach returns a background context containing only the delegated registry // authorization. It is used for bounded work that must outlive the inbound // mirror or coordination request without retaining unrelated request values.