feat: drop a NodeInventory whose publishedAt trails the fleet's newest publish by more than --inventory-stale-after

api/v1beta1/nodeinventory_types.go

@@ -87,6 +87,9 @@ type NodeInventory struct {
 	// already holds a warm microVM for a requested (template, net, size).
 	// +optional
 	Pools []PoolCapacity `json:"pools,omitempty"`
+	// publishedAt is the publisher's last apply; the aggregated apiserver skips an inventory older than its staleness window.
+	// +optional
+	PublishedAt metav1.Time `json:"publishedAt,omitempty"`
 }
 
 // +kubebuilder:object:root=true

cmd/sandbox-apiserver/main.go

@@ -53,6 +53,8 @@ type options struct {
 
 	E2BAPI bool
 	E2B    *e2bcompat.Flags
+
+	Inventory kubeinventory.Options
 }
 
 func newOptions() *options {
@@ -88,6 +90,7 @@ func (o *options) addFlags(fs *pflag.FlagSet) {
 	fs.BoolVar(&o.E2BAPI, "enable-e2b-api", o.E2BAPI,
 		"Serve the e2b-compatible REST surface, so an unmodified e2b SDK can claim from the same warm pools (point E2B_API_URL at it).")
 	o.E2B.AddFlags(fs)
+	o.Inventory.AddFlags(fs)
 }
 
 func (o *options) serverConfig() (*genericapiserver.Config, error) {
@@ -139,7 +142,7 @@ func run() error {
 	if err != nil {
 		return err
 	}
-	invSource := kubeinventory.New(reader)
+	invSource := kubeinventory.New(reader, o.Inventory)
 	store := scale.NewScatterGatherStore(
 		invSource,
 		scale.WithClaimRouting(token, scale.NewSandboxdClientFactory()),

cmd/sandbox-envd-proxy/main.go

@@ -32,10 +32,12 @@ type options struct {
 	Domain    string
 	Namespace string
 	Proxy     envdproxy.Flags
+	Inventory kubeinventory.Options
 }
 
 func (o *options) addFlags(fs *pflag.FlagSet) {
 	o.Proxy.AddFlags(fs, "")
+	o.Inventory.AddFlags(fs)
 	fs.StringVar(&o.Domain, "domain", o.Domain,
 		"Base domain sandbox hosts are derived from, as {port}-{sandboxID}.{domain}. Must match the apiserver's --e2b-domain.")
 	fs.StringVar(&o.Namespace, "namespace", o.Namespace,
@@ -70,7 +72,7 @@ func run(ctx context.Context, o *options) error {
 	if err != nil {
 		return err
 	}
-	inv := kubeinventory.New(reader)
+	inv := kubeinventory.New(reader, o.Inventory)
 	resolver, err := envdproxy.NewResolver(scale.NewScatterGatherStore(inv), inv, o.Namespace)
 	if err != nil {
 		return err

docs/api.md

@@ -67,6 +67,7 @@ _Appears in:_
 | `entries` _[InventoryEntry](#inventoryentry) array_ | entries summarizes the node's live sandboxes. |  | Optional: \{\} <br /> |
 | `address` _string_ | address is the node's sandboxd advertise address ("host:port"); the<br />aggregated apiserver routes a claim to this node's sandboxd through it. |  | Optional: \{\} <br /> |
 | `pools` _[PoolCapacity](#poolcapacity) array_ | pools is the node's per-pool warm capacity, used to pick a node that<br />already holds a warm microVM for a requested (template, net, size). |  | Optional: \{\} <br /> |
+| `publishedAt` _[Time](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.37/#time-v1-meta)_ | publishedAt is the publisher's last apply; the aggregated apiserver skips an inventory older than its staleness window. |  | Optional: \{\} <br /> |
 
 
 #### NodeInventoryList

docs/configuration.md

@@ -134,6 +134,31 @@ The driver watches `SandboxWarmPool` and `NodeInventory`, resolves each pool's
 under leader election (`cocoon-warmpool-driver`), so one replica drives the
 pools. An empty sandboxd token leaves it fail-closed: it logs and sets no pools.
 
+### Node inventory freshness
+
+| Flag | Default | Help |
+|---|---|---|
+| `--inventory-stale-after` | `90s` | Drop a node from this process's inventory reads once its NodeInventory `publishedAt` trails the newest publish in the fleet by more than this; set the same value on sandbox-apiserver and sandbox-envd-proxy. An inventory without `publishedAt`, from a vk-sandbox that predates the field, always stays. |
+
+A node that stops publishing, because it died or its vk-sandbox stopped,
+leaves every read and claim path once its `publishedAt` trails the newest
+publish by more than this. At vk-sandbox's default 30 s publish cadence that is
+three missed publishes, and since the newest publish only advances when a live
+node publishes, a dead node leaves between this age and one publish interval
+later. Keep it above vk-sandbox's `--publish-interval`.
+
+The age is measured against the newest publish the process has seen, capped by
+its own clock. A control-plane write outage freezes every stamp together, so it
+never empties the fleet, and a node's clock matters only relative to the other
+nodes: one running ahead cannot evict the rest. A fleet whose nodes all stop
+publishing at once, or a one-node fleet, keeps its last inventory until the
+nodes publish again.
+
+The window works on a node once the `nodeinventories` CRD from `helm/crds` is
+applied and that node runs a vk-sandbox that stamps `publishedAt`. An older CRD
+prunes the field and an older vk-sandbox never sets it, and either way that
+node never goes stale, so an upgrade never empties the fleet.
+
 ### e2b-compatible surface
 
 | Flag | Default | Help |
@@ -176,6 +201,7 @@ informer and relays the request into the owning node's guest-port endpoint.
 | `--tls-cert-file` | — | Wildcard certificate for `*.{domain}`. Omit to serve cleartext h2c behind an edge that terminates TLS. |
 | `--tls-private-key-file` | — | Private key for `--tls-cert-file`. |
 | `--guest-http2` | `false` | Forward to the guest over cleartext HTTP/2. Off by default: envd 0.8.0 installs no h2c handler and refuses it. Clients still reach this proxy over HTTP/2. |
+| `--inventory-stale-after` | `90s` | Drop a node from this process's inventory reads once its NodeInventory `publishedAt` trails the newest publish in the fleet by more than this; set the same value on sandbox-apiserver and sandbox-envd-proxy. An inventory without `publishedAt`, from a vk-sandbox that predates the field, always stays. |
 
 The two TLS flags must be set together. Routing, authorization and failure
 mapping are in [envd-proxy](envd-proxy.md).

docs/e2b-compat.md

@@ -143,6 +143,13 @@ const sandbox = await Sandbox.create('registry.example.com/rt:24.04')
   once by `files`/`commands` works. The proxy's node lookups share one budget
   per replica (200 new sandboxes/s, see [envd-proxy](envd-proxy.md)); past it a
   sandbox its node has not published answers `502` until the node publishes.
+- **A node that stops publishing leaves after `--inventory-stale-after`**
+  (90 s by default). Until then a dead node's sandboxes stay listed and a
+  claim can still sample it; past it they leave `GET /sandboxes`, and
+  `GET`, `DELETE` and the verbs on them answer `404`. The window is measured
+  against the fleet's newest publish, so a control-plane outage never drops
+  a node, and a vk-sandbox restart inside the window is invisible. A
+  vk-sandbox that predates `publishedAt` never goes stale.
 - **`envdVersion`** is reported as `0.4.0` unless `--e2b-envd-version` says
   otherwise. The SDK version-compares it and *kills the sandbox* if it cannot
   parse it, so it is always sent. Set it to the version actually installed in

docs/scaling-design.md

@@ -161,6 +161,8 @@ v1beta1 the `APIService` hands to the aggregated server, which serves only
 | Node partitioned from aggregated server | Its sandboxes briefly absent from `List` (eventual consistency, same as an informer lag) | No |
 | A node `inventory` object lost | Rebuilt from the node's own live state on next publish | No |
 | A node's `Node` object deleted | Its `NodeInventory` is garbage-collected with it: the node leaves the claim path and its sandboxes leave the read view, though sandboxd keeps serving them and their leases still expire there; restarting vk-sandbox registers the node again | No |
+| A node dies, its `Node` object kept | Its `NodeInventory` is no longer republished. Once its `publishedAt` trails the newest publish in the fleet by more than `--inventory-stale-after` (90 s), the node leaves the claim path and its sandboxes leave the read view, the watch emits Deleted for each, Get and the verbs answer 404, and the warm-pool driver and envd-proxy stop calling it. Its next publish brings it back. A vk-sandbox restart shorter than the window changes nothing | No |
+| The control plane stops accepting writes | Every publish fails and every stamp freezes together, so no node goes stale. envd-proxy and every apiserver replica but one keep claims, lists and lookups running from the informer cache, which shows the inventory as of the outage. The replica holding the warm-pool driver's lease exits once it cannot renew it (about 10 s) and starts again when the control plane returns | No |
 | Aggregated server restart | Stateless; rebuilds from node fan-out | No |
 | Client reads before the owning node republishes inventory | Lookups by name and by claim id (e2b, envd proxy), and a list or watch pinned to one name in a namespace, ask the nodes; other lists and watches show it at the next publish | No — fleet `list` and `watch`, and a deleted sandbox until its node publishes, stay eventually consistent |
 

helm/README.md

@@ -51,9 +51,10 @@ kubectl apply -f helm/crds/
 helm upgrade sandbox-operator ./helm --namespace sandbox-system --reuse-values
 ```
 
-An older `nodeinventories` CRD silently prunes entry fields it does not know
-(`claimedAt` is one), so until the CRD is applied every node publishes like an
-older node and the read view falls back accordingly.
+An older `nodeinventories` CRD silently prunes fields it does not know
+(`claimedAt` and `publishedAt` are two), so until the CRD is applied every node
+publishes like an older node: the read view falls back accordingly, and no node
+ever goes stale under `--inventory-stale-after`.
 
 `NodeInventory` moved from `extensions.agents.x-k8s.io` to
 `sandbox.cocoonstack.io`. A fleet coming from the old group runs a vk-sandbox

helm/crds/sandbox.cocoonstack.io_nodeinventories.yaml

@@ -80,6 +80,9 @@ spec:
               - warm
               type: object
             type: array
+          publishedAt:
+            format: date-time
+            type: string
         required:
         - node
         type: object

pkg/scale/inventorycache_bench_test.go

@@ -96,6 +96,7 @@ func benchCachedStore(b *testing.B, nodes, perNode int, noCopy bool) (*scale.Sca
 	invs, pool := scale.BenchInventories(nodes, perNode)
 	list := &unstructured.UnstructuredList{}
 	for _, inv := range invs {
+		inv.PublishedAt = metav1.Now()
 		raw, err := runtime.DefaultUnstructuredConverter.ToUnstructured(inv)
 		if err != nil {
 			b.Fatalf("encode node inventory: %v", err)
@@ -137,5 +138,5 @@ func benchCachedStore(b *testing.B, nodes, perNode int, noCopy bool) (*scale.Sca
 	if !invCache.WaitForCacheSync(syncCtx) {
 		b.Fatal("node inventory cache did not sync")
 	}
-	return scale.NewScatterGatherStore(kubeinventory.New(invCache)).(*scale.ScatterGatherStore), pool
+	return scale.NewScatterGatherStore(kubeinventory.New(invCache, kubeinventory.Options{})).(*scale.ScatterGatherStore), pool
 }

pkg/scale/kubeinventory/source.go

@@ -2,29 +2,53 @@
 package kubeinventory
 
 import (
+	"cmp"
 	"context"
 	"fmt"
 	"slices"
+	"sync/atomic"
+	"time"
 
+	"github.com/spf13/pflag"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"github.com/cocoonstack/sandbox-operator/pkg/scale"
 )
 
+const defaultStaleAfter = 90 * time.Second
+
+var nodeInventories = schema.GroupResource{Group: scale.NodeInventoryGVK.Group, Resource: "nodeinventories"}
+
+// Options tunes a Source.
+type Options struct {
+	// StaleAfter is how old publishedAt may get before its node leaves the fleet; zero means 90 s.
+	StaleAfter time.Duration
+}
+
+// AddFlags registers the source flags on fs.
+func (o *Options) AddFlags(fs *pflag.FlagSet) {
+	fs.DurationVar(&o.StaleAfter, "inventory-stale-after", cmp.Or(o.StaleAfter, defaultStaleAfter),
+		"Drop a node from this process's inventory reads once its NodeInventory publishedAt trails the newest publish in the fleet by more than this; set the same value on sandbox-apiserver and sandbox-envd-proxy. An inventory without publishedAt, from a vk-sandbox that predates the field, always stays.")
+}
+
 var _ scale.InventorySource = (*Source)(nil)
 
 // Source is the production InventorySource over a cache-fed NodeInventory reader.
 // It never mutates what it reads, since NewCache hands out its cached objects themselves.
 type Source struct {
-	reader client.Reader
+	reader     client.Reader
+	staleAfter time.Duration
+	reference  atomic.Int64
 }
 
 // New builds a Source over reader.
-func New(reader client.Reader) *Source {
-	return &Source{reader: reader}
+func New(reader client.Reader, opts Options) *Source {
+	return &Source{reader: reader, staleAfter: cmp.Or(opts.StaleAfter, defaultStaleAfter)}
 }
 
 func (s *Source) ListNodes(ctx context.Context) ([]string, error) {
@@ -33,19 +57,29 @@ func (s *Source) ListNodes(ctx context.Context) ([]string, error) {
 	if err := s.reader.List(ctx, ul); err != nil {
 		return nil, fmt.Errorf("kubeinventory: list node inventories: %w", err)
 	}
-	nodes := make([]string, 0, len(ul.Items))
+	nodes := make([]string, len(ul.Items))
+	stamps := make([]int64, len(ul.Items))
+	var newest int64
 	for i := range ul.Items {
-		nodes = append(nodes, ul.Items[i].GetName())
+		nodes[i], stamps[i] = ul.Items[i].GetName(), publishedAt(ul.Items[i].Object)
+		newest = max(newest, stamps[i])
+	}
+	ref := min(newest, time.Now().UnixNano())
+	s.reference.Store(ref)
+	kept := nodes[:0]
+	for i, node := range nodes {
+		if !s.stale(stamps[i], ref) {
+			kept = append(kept, node)
+		}
 	}
-	slices.Sort(nodes)
-	return nodes, nil
+	slices.Sort(kept)
+	return kept, nil
 }
 
 func (s *Source) NodeInventory(ctx context.Context, node string) (*scale.NodeInventory, error) {
-	u := &unstructured.Unstructured{}
-	u.SetGroupVersionKind(scale.NodeInventoryGVK)
-	if err := s.reader.Get(ctx, types.NamespacedName{Name: node}, u); err != nil {
-		return nil, fmt.Errorf("kubeinventory: get node %q inventory: %w", node, err)
+	u, err := s.get(ctx, node)
+	if err != nil {
+		return nil, err
 	}
 	inv := &scale.NodeInventory{}
 	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(u.Object, inv); err != nil {
@@ -55,10 +89,9 @@ func (s *Source) NodeInventory(ctx context.Context, node string) (*scale.NodeInv
 }
 
 func (s *Source) NodeCapacity(ctx context.Context, node string) (string, []scale.PoolCapacity, error) {
-	u := &unstructured.Unstructured{}
-	u.SetGroupVersionKind(scale.NodeInventoryGVK)
-	if err := s.reader.Get(ctx, types.NamespacedName{Name: node}, u); err != nil {
-		return "", nil, fmt.Errorf("kubeinventory: get node %q inventory: %w", node, err)
+	u, err := s.get(ctx, node)
+	if err != nil {
+		return "", nil, err
 	}
 	addr, _, err := unstructured.NestedString(u.Object, "address")
 	if err != nil {
@@ -82,3 +115,31 @@ func (s *Source) NodeCapacity(ctx context.Context, node string) (string, []scale
 	}
 	return addr, pools, nil
 }
+
+func (s *Source) get(ctx context.Context, node string) (*unstructured.Unstructured, error) {
+	u := &unstructured.Unstructured{}
+	u.SetGroupVersionKind(scale.NodeInventoryGVK)
+	if err := s.reader.Get(ctx, types.NamespacedName{Name: node}, u); err != nil {
+		return nil, fmt.Errorf("kubeinventory: get node %q inventory: %w", node, err)
+	}
+	if s.stale(publishedAt(u.Object), s.reference.Load()) {
+		return nil, fmt.Errorf("kubeinventory: node %q inventory is stale: %w", node, k8serrors.NewNotFound(nodeInventories, node))
+	}
+	return u, nil
+}
+
+func (s *Source) stale(stamp, ref int64) bool {
+	return stamp != 0 && ref-stamp > int64(s.staleAfter)
+}
+
+func publishedAt(obj map[string]any) int64 {
+	raw, _ := obj["publishedAt"].(string)
+	if raw == "" {
+		return 0
+	}
+	at, err := time.Parse(time.RFC3339, raw)
+	if err != nil {
+		return 0
+	}
+	return at.UnixNano()
+}