From b8fc67c5125fd60a08fd88865d691fc7ec590a1d Mon Sep 17 00:00:00 2001 From: AdaAibaby <8872412+adababys@users.noreply.github.com> Date: Thu, 22 Oct 2026 10:50:28 +0800 Subject: [PATCH] fix(orchestrator): stop nbd read flood for permanently-missing build objects When a build object is permanently absent from the store, the Linux NBD kernel driver retries the failing block read in a tight loop. Each retry reaches DiffStore.GetOrCreate, which does not cache negative results, so every retry triggers a fresh object-store 404 lookup - producing a high-frequency ERROR log flood and burning object-store quota. - Negative cache in GetOrCreate: store a short-lived (60s) negative entry when create returns storage.ErrObjectNotExist. Subsequent calls for the same key short-circuit immediately without hitting storage. The TTL allows recovery if the build is re-uploaded; Add clears the entry so a fresh upload is visible at once. A double-check inside the singleflight closure, plus a positive-cache recheck before installing the negative entry, prevent a race with a concurrent Add shadowing an available diff. - Log-level split in the nbd dispatcher: ErrObjectNotExist (permanent, expected when a build is gone) is logged at WARN with a clearer message; transient backend errors stay at ERROR, keeping dashboards and alerting focused on actionable signals. Rebased onto current main (the diff cache and NBD dispatch were refactored since the original attempt). Fixes #3294. --- .../orchestrator/pkg/sandbox/build/cache.go | 52 ++++++-- .../pkg/sandbox/build/cache_test.go | 113 ++++++++++++++++++ .../orchestrator/pkg/sandbox/nbd/dispatch.go | 11 +- 3 files changed, 167 insertions(+), 9 deletions(-) diff --git a/packages/orchestrator/pkg/sandbox/build/cache.go b/packages/orchestrator/pkg/sandbox/build/cache.go index c8a1152365..16e36d681c 100644 --- a/packages/orchestrator/pkg/sandbox/build/cache.go +++ b/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()) diff --git a/packages/orchestrator/pkg/sandbox/build/cache_test.go b/packages/orchestrator/pkg/sandbox/build/cache_test.go index cfd615e4c4..7b72c82341 100644 --- a/packages/orchestrator/pkg/sandbox/build/cache_test.go +++ b/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) +} diff --git a/packages/orchestrator/pkg/sandbox/nbd/dispatch.go b/packages/orchestrator/pkg/sandbox/nbd/dispatch.go index 5e3911068a..af112e4adc 100644 --- a/packages/orchestrator/pkg/sandbox/nbd/dispatch.go +++ b/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{})