Use containerd's content store as Racer's disk copy for Gantry

api/racer/v1alpha1/clustercache_envtest_test.go

@@ -539,3 +539,90 @@ func TestEnvtestClusterCachePassthroughValidation(t *testing.T) {
 		require.NoError(t, update(t, name, nil))
 	})
 }
+
+func TestEnvtestClusterCachePersistenceValidation(t *testing.T) {
+	assets := os.Getenv("KUBEBUILDER_ASSETS")
+	if assets == "" {
+		t.Skip("set KUBEBUILDER_ASSETS for real API validation tests")
+	}
+
+	environment := &envtest.Environment{BinaryAssetsDirectory: assets, CRDDirectoryPaths: []string{"crd"}, ErrorIfCRDPathMissing: true}
+	rc, err := environment.Start()
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, environment.Stop()) })
+
+	scheme := runtime.NewScheme()
+	require.NoError(t, AddToScheme(scheme))
+	c, err := client.New(rc, client.Options{Scheme: scheme})
+	require.NoError(t, err)
+
+	n := 0
+	create := func(spec map[string]any) (string, error) {
+		n++
+		name := fmt.Sprintf("persistence-%d", n)
+		obj := &unstructured.Unstructured{Object: map[string]any{
+			"apiVersion": GroupVersion.String(),
+			"kind":       "ClusterCache",
+			"metadata":   map[string]any{"name": name},
+			"spec":       spec,
+		}}
+
+		return name, c.Create(t.Context(), obj)
+	}
+
+	for name, spec := range map[string]map[string]any{
+		"unset":                  {},
+		"racer":                  {"persistence": "Racer"},
+		"adapter":                {"persistence": "Adapter"},
+		"racer with passthrough": {"persistence": "Racer", "passthrough": map[string]any{}},
+		"racer with origin":      {"persistence": "Racer", "origin": azureCredential(map[string]any{"default": map[string]any{}})},
+	} {
+		t.Run("accepts "+name, func(t *testing.T) {
+			_, err := create(spec)
+			require.NoError(t, err)
+		})
+	}
+
+	for name, tc := range map[string]struct {
+		spec    map[string]any
+		message string
+	}{
+		"unknown":          {map[string]any{"persistence": "Disk"}, "spec.persistence"},
+		"lowercase":        {map[string]any{"persistence": "adapter"}, "spec.persistence"},
+		"with passthrough": {map[string]any{"persistence": "Adapter", "passthrough": map[string]any{"route": "Local"}}, "cannot be combined with passthrough"},
+		"with origin":      {map[string]any{"persistence": "Adapter", "origin": azureCredential(map[string]any{"default": map[string]any{}})}, "cannot be combined with origin"},
+	} {
+		t.Run("rejects "+name, func(t *testing.T) {
+			_, err := create(tc.spec)
+			require.True(t, apierrors.IsInvalid(err), "%v", err)
+			require.ErrorContains(t, err, tc.message)
+		})
+	}
+
+	t.Run("allows switching persistence but not adding passthrough to Adapter", func(t *testing.T) {
+		name, err := create(map[string]any{})
+		require.NoError(t, err)
+
+		set := func(path string, value any) error {
+			obj := &unstructured.Unstructured{}
+			obj.SetGroupVersionKind(GroupVersion.WithKind("ClusterCache"))
+			require.NoError(t, c.Get(t.Context(), client.ObjectKey{Name: name}, obj))
+
+			if value == nil {
+				unstructured.RemoveNestedField(obj.Object, "spec", path)
+			} else {
+				require.NoError(t, unstructured.SetNestedField(obj.Object, value, "spec", path))
+			}
+
+			return c.Update(t.Context(), obj)
+		}
+
+		require.NoError(t, set("persistence", "Adapter"))
+
+		err = set("passthrough", map[string]any{})
+		require.True(t, apierrors.IsInvalid(err), "%v", err)
+		require.NoError(t, set("persistence", "Racer"))
+		require.NoError(t, set("persistence", nil))
+		require.NoError(t, set("persistence", "Adapter"))
+	})
+}

api/racer/v1alpha1/clustercache_test.go

@@ -136,3 +136,18 @@ func TestClusterCachePassthroughJSONDeepCopyAndDefault(t *testing.T) {
 	require.NoError(t, err)
 	require.NotContains(t, string(data), "passthrough")
 }
+
+func TestClusterCachePersistenceJSONAndDefault(t *testing.T) {
+	require.Equal(t, PersistenceRacer, ClusterCacheSpec{}.EffectivePersistence())
+	require.Equal(t, PersistenceRacer, ClusterCacheSpec{Persistence: PersistenceRacer}.EffectivePersistence())
+	require.Equal(t, PersistenceAdapter, ClusterCacheSpec{Persistence: PersistenceAdapter}.EffectivePersistence())
+
+	data, err := json.Marshal(&ClusterCache{ObjectMeta: metav1.ObjectMeta{Name: "cache"}, Spec: ClusterCacheSpec{Persistence: PersistenceAdapter}})
+	require.NoError(t, err)
+	require.JSONEq(t, `{"metadata":{"name":"cache"},"spec":{"persistence":"Adapter"}}`, string(data))
+
+	// An unset persistence is omitted, so existing caches keep their encoding.
+	data, err = json.Marshal(&ClusterCache{ObjectMeta: metav1.ObjectMeta{Name: "cache"}})
+	require.NoError(t, err)
+	require.NotContains(t, string(data), "persistence")
+}

api/racer/v1alpha1/clustercache_types.go

@@ -30,6 +30,8 @@ type ClusterCache struct {
 }
 
 // ClusterCacheSpec configures optional cache features.
+// +kubebuilder:validation:XValidation:rule="!has(self.persistence) || self.persistence != 'Adapter' || !has(self.passthrough)",message="persistence Adapter cannot be combined with passthrough"
+// +kubebuilder:validation:XValidation:rule="!has(self.persistence) || self.persistence != 'Adapter' || !has(self.origin)",message="persistence Adapter requires an origin adapter and cannot be combined with origin"
 type ClusterCacheSpec struct {
 	// Origin makes the Racer dataplane read this cache's objects directly from
 	// object storage, with no separate origin process. Origin can be set only
@@ -66,6 +68,20 @@ type ClusterCacheSpec struct {
 	// +optional
 	Passthrough *Passthrough `json:"passthrough,omitempty"`
 
+	// Persistence selects where this cache's pages are kept on disk. Racer,
+	// the default when unset, writes pages to Racer's own local disk. With
+	// Adapter, Racer never writes this cache's pages to its own disk: pages
+	// live in memory only, and on a memory miss Racer first asks the node's
+	// local origin adapter for a local copy with a local-only origin request,
+	// one the adapter must answer without contacting its remote source,
+	// before it asks peers and the remote origin. Use Adapter when the origin
+	// adapter already keeps a durable local copy of the objects it serves.
+	// Adapter cannot be combined with passthrough or origin. Persistence may
+	// be changed at any time; when a cache switches to Adapter, pages
+	// previously written to Racer's disk for it are dropped.
+	// +optional
+	Persistence Persistence `json:"persistence,omitempty"`
+
 	// NodeSelector selects, by Node labels, the Racer nodes that store, serve,
 	// and read this cache. When unset or empty, every Racer node does.
 	// Changing it moves the cache to a new pool of nodes with no migration:
@@ -75,6 +91,27 @@ type ClusterCacheSpec struct {
 	NodeSelector *metav1.LabelSelector `json:"nodeSelector,omitempty"`
 }
 
+// Persistence selects where a cache's pages are kept on disk.
+// +kubebuilder:validation:Enum=Racer;Adapter
+type Persistence string
+
+const (
+	// PersistenceRacer writes pages to Racer's own local disk.
+	PersistenceRacer Persistence = "Racer"
+	// PersistenceAdapter keeps pages in Racer memory only and relies on the
+	// origin adapter's local copies for durability on each node.
+	PersistenceAdapter Persistence = "Adapter"
+)
+
+// EffectivePersistence returns spec.persistence, or PersistenceRacer when unset.
+func (s ClusterCacheSpec) EffectivePersistence() Persistence {
+	if s.Persistence == "" {
+		return PersistenceRacer
+	}
+
+	return s.Persistence
+}
+
 // PassthroughRoute selects which node's origin serves a passthrough read.
 // +kubebuilder:validation:Enum=Peers;Local
 type PassthroughRoute string

api/racer/v1alpha1/crd/racer.unbounded-cloud.io_clustercaches.yaml

@@ -255,6 +255,23 @@ spec:
                     - Local
                     type: string
                 type: object
+              persistence:
+                description: |-
+                  Persistence selects where this cache's pages are kept on disk. Racer,
+                  the default when unset, writes pages to Racer's own local disk. With
+                  Adapter, Racer never writes this cache's pages to its own disk: pages
+                  live in memory only, and on a memory miss Racer first asks the node's
+                  local origin adapter for a local copy with a local-only origin request,
+                  one the adapter must answer without contacting its remote source,
+                  before it asks peers and the remote origin. Use Adapter when the origin
+                  adapter already keeps a durable local copy of the objects it serves.
+                  Adapter cannot be combined with passthrough or origin. Persistence may
+                  be changed at any time; when a cache switches to Adapter, pages
+                  previously written to Racer's disk for it are dropped.
+                enum:
+                - Racer
+                - Adapter
+                type: string
               priority:
                 description: |-
                   Priority is the relative weight used to share node cache capacity among
@@ -273,6 +290,14 @@ spec:
                 minimum: 0
                 type: integer
             type: object
+            x-kubernetes-validations:
+            - message: persistence Adapter cannot be combined with passthrough
+              rule: '!has(self.persistence) || self.persistence != ''Adapter'' ||
+                !has(self.passthrough)'
+            - message: persistence Adapter requires an origin adapter and cannot be
+                combined with origin
+              rule: '!has(self.persistence) || self.persistence != ''Adapter'' ||
+                !has(self.origin)'
         type: object
         x-kubernetes-validations:
         - message: name must fit the canonical Unix socket path

cmd/gantry/agent_racer.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/Azure/unbounded/internal/gantry/config"
+	"github.com/Azure/unbounded/internal/gantry/contentstore"
 	"github.com/Azure/unbounded/internal/gantry/metrics"
 	"github.com/Azure/unbounded/internal/gantry/mirror"
 	"github.com/Azure/unbounded/internal/gantry/origin"
@@ -27,6 +28,7 @@ const racerShutdownBudget = 10 * time.Second
 
 type racerAgentClient interface {
 	racer.Client
+	racer.Invalidator
 	Close() error
 }
 
@@ -70,6 +72,15 @@ func serveRacerAgent(ctx context.Context, c *config.Config, logger *slog.Logger,
 		return fmt.Errorf("racer origin client: %w", err)
 	}
 
+	replicas := racer.NewReplicaVerifier(logger, telemetry.LocalReplicaVerification)
+
+	originOptions, err := racerOriginOptions(c)
+	if err != nil {
+		return err
+	}
+
+	originOptions = append(originOptions, racer.WithReplicaVerifier(replicas))
+
 	// Signal cancellation starts draining; cancelOrigin runs after HTTP drain.
 	originCtx, cancelOrigin := context.WithCancel(context.WithoutCancel(ctx))
 	originDone := make(chan struct{})
@@ -79,13 +90,14 @@ func serveRacerAgent(ctx context.Context, c *config.Config, logger *slog.Logger,
 	go func() {
 		defer close(originDone)
 
-		originErr = deps.serveOrigin(originCtx, racer.OriginConfig(c, cache), racer.Origin(c, originClient))
+		originErr = deps.serveOrigin(originCtx, racer.OriginConfig(c, cache), racer.Origin(c, originClient, originOptions...))
 	}()
 
 	// Always install the handler, even without a client, so unavailable requests
 	// retain Racer correlation headers and rate-limited failure diagnostics.
 	mirrorSrv := mirror.New(c, nil, originClient,
-		mirror.WithContentBackend(racer.NewHandler(deps.client, originClient, logger, racer.WithRetryBudget(c.RacerRetryBudget), racer.WithRetryObserver(telemetry.MirrorRetry))), mirror.WithLogger(logger), mirror.WithStartupReadinessGate())
+		mirror.WithContentBackend(racer.NewHandler(deps.client, originClient, logger, racer.WithRetryBudget(c.RacerRetryBudget), racer.WithRetryObserver(telemetry.MirrorRetry), racer.WithAccessCheck(originClient),
+			racer.WithInvalidator(deps.client), racer.WithDigestMismatchObserver(telemetry.DigestMismatch), racer.WithLocalReplicaReset(replicas))), mirror.WithLogger(logger), mirror.WithStartupReadinessGate())
 
 	var (
 		ready   atomic.Bool
@@ -260,3 +272,18 @@ func shutdownRacerAgent(servers []*http.Server, cancelOrigin context.CancelFunc,
 
 	return retErr
 }
+
+// racerOriginOptions serves Racer local-only page requests from the node's
+// containerd content store unless racer_content_root is empty.
+func racerOriginOptions(c *config.Config) ([]racer.OriginOption, error) {
+	if c.RacerContentRoot == "" {
+		return nil, nil
+	}
+
+	store, err := contentstore.New(c.RacerContentRoot)
+	if err != nil {
+		return nil, fmt.Errorf("racer content store: %w", err)
+	}
+
+	return []racer.OriginOption{racer.WithLocalStore(racer.ContentStore(store))}, nil
+}

cmd/gantry/agent_racer_test.go

@@ -32,6 +32,10 @@ func (*startupRacerClient) Stat(context.Context, racersdk.Request) (racersdk.Met
 	return racersdk.Metadata{}, errors.New("unexpected metadata request")
 }
 
+func (*startupRacerClient) Invalidate(context.Context, racersdk.Key, string) error {
+	return errors.New("unexpected invalidate request")
+}
+
 func (c *startupRacerClient) Close() error { c.closed.Store(true); return nil }
 
 func TestRacerStartupSelection(t *testing.T) {
@@ -386,3 +390,22 @@ func TestRacerIgnoresSharedCredentialFile(t *testing.T) {
 		t.Fatalf("Racer loaded p2p credentials or leaked client: %v", err)
 	}
 }
+
+func TestRacerOriginOptions(t *testing.T) {
+	c := config.NewDefault()
+
+	opts, err := racerOriginOptions(c)
+	if err != nil || len(opts) != 1 {
+		t.Fatalf("default root: opts=%d err=%v", len(opts), err)
+	}
+
+	c.RacerContentRoot = ""
+	if opts, err := racerOriginOptions(c); err != nil || len(opts) != 0 {
+		t.Fatalf("empty root must disable local replicas: opts=%d err=%v", len(opts), err)
+	}
+
+	c.RacerContentRoot = "relative"
+	if _, err := racerOriginOptions(c); err == nil {
+		t.Fatal("relative root accepted")
+	}
+}

cmd/racer-loadgen/gantry_test.go

@@ -240,14 +240,18 @@ func TestGantryIntegration(t *testing.T) {
 		assertRanges(t, 0, 0, 0)
 	})
 
-	t.Run("consumer rejects transported corruption", func(t *testing.T) {
+	t.Run("mirror aborts transported corruption", func(t *testing.T) {
 		p, metrics := newPull(t)
 		require.True(t, p.opts.Verify, "consumer verification must remain enabled")
 		corrupt.Store(true)
 
 		err := p.pull(t.Context())
-		require.ErrorContains(t, err, "digest mismatch")
-		require.Equal(t, float64(img.Manifest.Size), testutil.ToFloat64(metrics.receivedBytes), "mirror must deliver the complete corrupt object")
+		// The manifest is read first. Manifests fit in one page, so the mirror
+		// verifies the buffered body before committing headers and answers
+		// 502 instead of aborting a started response.
+		require.ErrorContains(t, err, "status 502")
+		require.Less(t, testutil.ToFloat64(metrics.receivedBytes), float64(img.Manifest.Size),
+			"mirror must not complete an object that fails digest verification")
 		require.Zero(t, testutil.ToFloat64(metrics.pulls.WithLabelValues("success")))
 		require.Equal(t, float64(1), testutil.ToFloat64(metrics.pulls.WithLabelValues("error")))
 	})

cmd/racer/README.md

@@ -60,6 +60,59 @@ dashboards and alerts that use the old names:
 | `racer_peer_local_pressure_total` | Removed; `racer_peer_admission_limit` is now a fixed cap |
 
 
+## Adapter persistence
+
+A ClusterCache with `spec.persistence: Adapter` publishes `CacheMode::Adapter`
+(wire spelling `adapter`). `CacheMode::disk()` is false for it, as for
+passthrough, so the store refuses its writes (`StoreWriter::allowed`), fills
+never persist it, reads skip Racer's disk and never spill, and every snapshot
+purges its disk records. Unlike passthrough, its pages and metadata stay in
+memory.
+
+On a miss, `Fill::acquire_once` asks the sources that
+`racer_core::read::local_sources` lists after memory and pending copies, and
+before the ranked owner walk and the remote origin. For an adapter cache that
+is `LocalSource::AdapterReplica`: `Origin::local_page_reserved` sends the
+node's own origin socket a pinned page `GET` with `Racer-Source: local`. A
+`404` or `412` is a local miss. A miss or any failure other than cancellation
+falls through without origin failure notes or link-health accounting.
+
+Next is `LocalSource::Holders`. With no disk copy, once an owner evicts an
+adapter page the durable copies live in the adapters of the nodes that read
+it. An owner that serves an adapter page to a peer `Acquire` records the
+signed requester (`Coordinator::note_holder`) in its worker's holder table
+(`racer_core::read::holders::Holders`, on `CandidatePolicy::holders`). Keys
+are page identities, whose version carries the etag, so a new version never
+routes to holders of an old one. Each page keeps at most `HOLDERS_PER_PAGE`
+(4) nodes, newest first and deduplicated, each for `HOLDER_TTL` (24h); the
+table keeps at most `HOLDER_CAPACITY` (65536) pages and evicts the oldest
+inserted page first. Time is injected, so the table is deterministic.
+
+`Fill::acquire_from_holders` asks up to `HOLDER_ATTEMPTS` (2) unexpired
+holders, other than this node, with a holder probe (`FetchMode::Holder`, wire
+mode `holder`). Like an offer, a probe is a credential-free page request
+with no remote attempts, sent on its own `HOLDER_TIMEOUT` (2s) deadline and a
+fresh one-exchange grant, so a dead holder neither stalls the fill nor spends
+the reader's budget. The holder (`Coordinator::serve_holder`) answers from
+memory or its own adapter (`LocalSource::AdapterReplica`), never acquires,
+forwards, or contacts the remote origin, and answers `Miss` for any other
+cache mode. A returned copy is decrypted, which authenticates it, and
+published to memory. A miss, a failure, or an unauthenticated copy forgets
+that holder and asks the next; after the last the fill continues with the
+ranked walk. Receivers that predate the mode reject it, which counts as a
+failure. `racer_holder_records_total`, `racer_holder_probes_served_total`,
+`racer_holder_hits_total`, `racer_holder_misses_total`, and
+`racer_holder_failures_total` count these steps.
+
+## Invalidation
+
+`DELETE /v1/objects/<key>` with a strong `If-Match` ETag on the client socket
+drops that version (`src/client/invalidate.rs`, `src/read/invalidate.rs`). The
+node drops its metadata, pages, and holder records, then sends a `prefetch` peer operation with
+action `invalidate` to each page's ranked owners. Receivers drop local state
+only and never forward it. See "Invalidation" in
+`docs/content/reference/racer.md` for limits.
+
 ## Origin HTTPS
 
 Origins reached over HTTPS (for example Azure Blob Storage) use rustls for the