fix(orchestrator): stop nbd read flood for permanently-missing build objects (rebased #3295)

packages/orchestrator/pkg/sandbox/build/cache.go

@@ -4,6 +4,7 @@ package build
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math"
 	"os"
@@ -20,6 +21,7 @@ import (
 	"github.com/e2b-dev/infra/packages/orchestrator/pkg/cfg"
 	"github.com/e2b-dev/infra/packages/shared/pkg/featureflags"
 	"github.com/e2b-dev/infra/packages/shared/pkg/logger"
+	"github.com/e2b-dev/infra/packages/shared/pkg/storage"
 	"github.com/e2b-dev/infra/packages/shared/pkg/units"
 	"github.com/e2b-dev/infra/packages/shared/pkg/utils"
 )
@@ -39,6 +41,12 @@ type deleteDiff struct {
 	closeOnce sync.Once
 }
 
+// negativeCacheTTL is how long a permanently-missing build object is held in
+// the negative cache before being rechecked. It keeps the kernel NBD retry
+// flood from hammering object storage with repeated 404s while still
+// recovering when a build is eventually (re-)uploaded.
+const negativeCacheTTL = 60 * time.Second
+
 type DiffStore struct {
 	cachePath string
 	cache     *ttlcache.Cache[DiffStoreKey, Diff]
@@ -54,6 +62,9 @@ type DiffStore struct {
 	pdDelay time.Duration
 
 	insertionTimes sync.Map // map[DiffStoreKey]time.Time — tracks when each diff was cached
+	// negativeCache holds keys for build objects permanently missing from the
+	// store. ttlcache expires entries automatically so they never accumulate.
+	negativeCache *ttlcache.Cache[DiffStoreKey, struct{}]
 
 	// pinned entries are skipped by disk-pressure eviction (TTL eviction still
 	// applies). Used to protect a diff whose Close would tear down state another
@@ -77,14 +88,19 @@ func NewDiffStore(
 		ttlcache.WithTTL[DiffStoreKey, Diff](ttl),
 	)
 
+	negCache := ttlcache.New(
+		ttlcache.WithTTL[DiffStoreKey, struct{}](negativeCacheTTL),
+	)
+
 	ds := &DiffStore{
-		cachePath: cachePath,
-		cache:     cache,
-		cancel:    func() {},
-		config:    config,
-		flags:     flags,
-		pdSizes:   make(map[DiffStoreKey]*deleteDiff),
-		pdDelay:   delay,
+		cachePath:     cachePath,
+		cache:         cache,
+		negativeCache: negCache,
+		cancel:        func() {},
+		config:        config,
+		flags:         flags,
+		pdSizes:       make(map[DiffStoreKey]*deleteDiff),
+		pdDelay:       delay,
 	}
 
 	cache.OnEviction(func(ctx context.Context, _ ttlcache.EvictionReason, item *ttlcache.Item[DiffStoreKey, Diff]) {
@@ -133,12 +149,14 @@ func (s *DiffStore) Start(ctx context.Context) {
 	s.cancel = cancel
 
 	go s.cache.Start()
+	go s.negativeCache.Start()
 	go s.startDiskSpaceEviction(ctx, s.config, s.flags)
 }
 
 func (s *DiffStore) Close() {
 	s.cancel()
 	s.cache.Stop()
+	s.negativeCache.Stop()
 }
 
 // Get returns the cached Diff for key, refreshing TTL and cancelling any
@@ -160,12 +178,19 @@ func (s *DiffStore) Get(key DiffStoreKey) (Diff, bool) {
 func (s *DiffStore) GetOrCreate(ctx context.Context, key DiffStoreKey, create func(context.Context) (Diff, error)) (Diff, error) {
 	s.resetDelete(key)
 
+	if s.negativeCache.Has(key) {
+		return nil, storage.ErrObjectNotExist
+	}
+
 	if item := s.cache.Get(key); item != nil {
 		return item.Value(), nil
 	}
 
 	v, err, _ := s.initGroup.Do(string(key), func() (any, error) {
-		// Double-check: another goroutine may have cached it while we waited.
+		// Double-check: another goroutine may have populated either cache while we waited.
+		if s.negativeCache.Has(key) {
+			return nil, storage.ErrObjectNotExist
+		}
 		if item := s.cache.Get(key); item != nil {
 			return item.Value(), nil
 		}
@@ -174,6 +199,15 @@ func (s *DiffStore) GetOrCreate(ctx context.Context, key DiffStoreKey, create fu
 
 		diff, err := create(ctx)
 		if err != nil {
+			if errors.Is(err, storage.ErrObjectNotExist) {
+				// Recheck the positive cache before installing the negative entry:
+				// a concurrent Add may have populated it (and cleared any negative
+				// entry) while this storage probe was in flight. Installing one here
+				// then would shadow an available diff for up to negativeCacheTTL.
+				if s.cache.Get(key) == nil {
+					s.negativeCache.Set(key, struct{}{}, ttlcache.DefaultTTL)
+				}
+			}
 			return nil, err
 		}
 
@@ -190,6 +224,8 @@ func (s *DiffStore) GetOrCreate(ctx context.Context, key DiffStoreKey, create fu
 }
 
 func (s *DiffStore) Add(d Diff) {
+	// Clear any negative entry so a re-uploaded build is immediately visible.
+	s.negativeCache.Delete(d.CacheKey())
 	s.resetDelete(d.CacheKey())
 	s.cache.Set(d.CacheKey(), d, ttlcache.DefaultTTL)
 	s.insertionTimes.LoadOrStore(d.CacheKey(), time.Now())

packages/orchestrator/pkg/sandbox/build/cache_test.go

@@ -16,6 +16,7 @@ package build
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"testing"
@@ -32,6 +33,7 @@ import (
 	"github.com/e2b-dev/infra/packages/orchestrator/pkg/cfg"
 	blockmetrics "github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/block/metrics"
 	"github.com/e2b-dev/infra/packages/shared/pkg/featureflags"
+	"github.com/e2b-dev/infra/packages/shared/pkg/storage"
 	"github.com/e2b-dev/infra/packages/shared/pkg/storage/header"
 	"github.com/e2b-dev/infra/packages/shared/pkg/utils"
 )
@@ -773,3 +775,114 @@ func TestDiffStoreEvictionSkipsUnsealedDeferredDiff(t *testing.T) {
 	assert.True(t, ok)
 	assert.True(t, store.isBeingDeleted(deferred.CacheKey()), "sealed deferred diff must be evictable")
 }
+
+func TestGetOrCreate_NegativeCacheShortCircuits(t *testing.T) {
+	store, err := NewDiffStore(mustParseCfg(t), flagsWithMaxBuildCachePercentage(t, 90), t.TempDir(), time.Hour, time.Minute)
+	require.NoError(t, err)
+	store.Start(t.Context())
+	t.Cleanup(store.Close)
+
+	key := DiffStoreKey(uuid.New().String())
+	calls := 0
+	create := func(ctx context.Context) (Diff, error) {
+		calls++
+		return nil, storage.ErrObjectNotExist
+	}
+
+	_, err = store.GetOrCreate(t.Context(), key, create)
+	require.ErrorIs(t, err, storage.ErrObjectNotExist)
+	require.Equal(t, 1, calls, "create should be called once on first miss")
+
+	for i := 0; i < 3; i++ {
+		_, err = store.GetOrCreate(t.Context(), key, create)
+		require.ErrorIs(t, err, storage.ErrObjectNotExist)
+	}
+	require.Equal(t, 1, calls, "create must not be called again while negative cache is hot")
+}
+
+func TestGetOrCreate_NegativeCacheNotSetForOtherErrors(t *testing.T) {
+	store, err := NewDiffStore(mustParseCfg(t), flagsWithMaxBuildCachePercentage(t, 90), t.TempDir(), time.Hour, time.Minute)
+	require.NoError(t, err)
+	store.Start(t.Context())
+	t.Cleanup(store.Close)
+
+	key := DiffStoreKey(uuid.New().String())
+	sentinel := errors.New("transient network error")
+	calls := 0
+	create := func(ctx context.Context) (Diff, error) {
+		calls++
+		return nil, sentinel
+	}
+
+	_, err = store.GetOrCreate(t.Context(), key, create)
+	require.ErrorIs(t, err, sentinel)
+	_, err = store.GetOrCreate(t.Context(), key, create)
+	require.ErrorIs(t, err, sentinel)
+	require.Equal(t, 2, calls, "create should be retried after non-ErrObjectNotExist errors")
+}
+
+func TestGetOrCreate_AddClearsNegativeCache(t *testing.T) {
+	cachePath := t.TempDir()
+	store, err := NewDiffStore(mustParseCfg(t), flagsWithMaxBuildCachePercentage(t, 90), cachePath, time.Hour, time.Minute)
+	require.NoError(t, err)
+	store.Start(t.Context())
+	t.Cleanup(store.Close)
+
+	diff := newRootFSDiff(t, cachePath, uuid.New().String())
+	key := diff.CacheKey()
+
+	calls := 0
+	notFound := func(ctx context.Context) (Diff, error) {
+		calls++
+		return nil, storage.ErrObjectNotExist
+	}
+
+	_, err = store.GetOrCreate(t.Context(), key, notFound)
+	require.ErrorIs(t, err, storage.ErrObjectNotExist)
+	require.Equal(t, 1, calls)
+
+	store.Add(diff)
+
+	returnDiff := func(ctx context.Context) (Diff, error) {
+		calls++
+		return diff, nil
+	}
+	got, err := store.GetOrCreate(t.Context(), key, returnDiff)
+	require.NoError(t, err)
+	require.Equal(t, diff, got)
+	require.Equal(t, 1, calls, "create must not be called after Add populates the positive cache")
+}
+
+func TestGetOrCreate_AddRaceWithStorageProbe(t *testing.T) {
+	cachePath := t.TempDir()
+	store, err := NewDiffStore(mustParseCfg(t), flagsWithMaxBuildCachePercentage(t, 90), cachePath, time.Hour, time.Minute)
+	require.NoError(t, err)
+	store.Start(t.Context())
+	t.Cleanup(store.Close)
+
+	diff := newRootFSDiff(t, cachePath, uuid.New().String())
+	key := diff.CacheKey()
+
+	ready := make(chan struct{})
+	errc := make(chan error, 1)
+	go func() {
+		_, err := store.GetOrCreate(t.Context(), key, func(ctx context.Context) (Diff, error) {
+			<-ready
+			return nil, storage.ErrObjectNotExist
+		})
+		errc <- err
+	}()
+
+	time.Sleep(20 * time.Millisecond)
+	store.Add(diff)
+	close(ready)
+
+	require.ErrorIs(t, <-errc, storage.ErrObjectNotExist)
+
+	got, err := store.GetOrCreate(t.Context(), key, func(ctx context.Context) (Diff, error) {
+		t.Fatal("create must not be called: diff is in the positive cache")
+		return nil, nil
+	})
+	require.NoError(t, err)
+	require.Equal(t, diff, got)
+}

packages/orchestrator/pkg/sandbox/nbd/dispatch.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/block"
 	"github.com/e2b-dev/infra/packages/shared/pkg/logger"
+	"github.com/e2b-dev/infra/packages/shared/pkg/storage"
 	"github.com/e2b-dev/infra/packages/shared/pkg/telemetry"
 	"github.com/e2b-dev/infra/packages/shared/pkg/utils"
 )
@@ -431,7 +432,15 @@ func (d *Dispatch) cmdRead(ctx context.Context, cmd command) error {
 			// Per-request backend failure: signal it to the NBD client via the
 			// response error byte and keep the dispatch loop alive. Only
 			// writeResponse errors (dead NBD socket) escalate through d.fatal.
-			d.logger.Error(ctx, "nbd backend read failed", append(cmd.fields(), zap.Error(readErr))...)
+			// ErrObjectNotExist is permanent (the build object is gone); the
+			// DiffStore negative cache absorbs the kernel's retry storm. Log it at
+			// WARN so dashboards are not flooded with expected 404s, while transient
+			// backend errors stay at ERROR as actionable signal.
+			if errors.Is(readErr, storage.ErrObjectNotExist) {
+				d.logger.Warn(ctx, "nbd backend read failed: build object not found in store", append(cmd.fields(), zap.Error(readErr))...)
+			} else {
+				d.logger.Error(ctx, "nbd backend read failed", append(cmd.fields(), zap.Error(readErr))...)
+			}
 			recordLegFault(ctx)
 
 			return d.writeResponse(1, cmd.handle, []byte{})