Skip to content

Commit 4f81799

Browse files
authored
fix(orchestrator): pause upload retain retry (#2993)
## Problem A paused sandbox's snapshot could be **partially uploaded and then permanently lost**. The pause upload was fire-and-forget — a single attempt whose failure was only logged — so a transient source-side failure dropped the snapshot for good. The lost artifact is almost always the **memfile** (memory image) plus its header. Because every later build stores only a *diff* against its ancestors, a lost ancestor memfile is unrecoverable and **cascades**: descendant pauses can never finalize, and any resume crashes at runtime when it page-faults on the missing memory. ### Where it was observed Loki (`grafanacloud-logs`): ```logql {service_name="orchestrator"} |= "error uploading snapshot files" ``` → concentrated on a "hot host" per day Concrete case — build `2210682a-30e2-4567-9091-53d57a625a1a` (host `orch-client-n2-2pkh`, ~2026-05-29 18:13 UTC): - Pause at `18:13:37`; at `18:13:47.923`: `error uploading snapshot files` → `... wait for ancestor d7b75d71…: memfile upload: context deadline exceeded` - The build's objects in storage were **only** `metadata.json`, `snapfile`, `rootfs.ext4.zstd` — `memfile.zstd`, `memfile.header`, `rootfs.ext4.header` were **missing**. - Downstream, same sandbox at `18:15:58`: `UFFD serve data fetch error after retries` → `failed to get GCS object ("b52e4ea0-…/memfile") … object does not exist` → **sandbox killed**. ### Why it happened 1. **Transient source-side failure** — a memfile upload hit the 20‑min `uploadTimeout`, or GCS auth blipped (`401`/`503 Metadata Server not available`), or `MemfileDiff.CachePath` wasn't ready. 2. **Fire-and-forget meant one shot** — `uploadSnapshotAsync` ran `Upload.Run` once and only logged on failure; the local diff was then evicted, so the bytes were gone. 3. **Failure propagated and cascaded** — the failed build's future was stored with the error and broadcast (Redis); every descendant pause's `appendAncestorBuilds` → `Uploads.Wait` inherited it instantly, producing the deep `wait for ancestor …` chains. One sandbox auto-pausing every ~5 min re-failed identically 17×. 4. **The loss surfaced at resume, not pause** — pause returned success (upload is async); memory is paged lazily via UFFD from GCS, so the first fault on the missing `…/memfile` object killed the running sandbox. ## What this PR does - **Retry until durable** — replaces the single attempt with a bounded retry: fresh per-attempt timeout under a ~2h budget, exponential backoff, default-retryable error classification (only `NoDiff` / `object-not-exist` / parent-cancel stop). Recovers the transient GCS/auth/timeout roots, and because **every** pause retries, descendants' ancestor-waits resolve once the root finally lands — the cascade self-heals. - **General `retry.Do` helper** (`packages/shared/pkg/retry`) — the loop runs under one budget `context.WithTimeoutCause`, so each attempt derives from it and is auto-capped to the remaining budget; the loop can't overrun `TotalBudget`. Distinguishes budget exhaustion from caller cancellation via `context.Cause`. - **Cross-orchestrator budget alignment** — `refreshHeaderBudget` and `futureTTL` raised to ≥ the upload budget, so a dependent build's `Wait` doesn't give up (non-retryable `object does not exist` on poll-budget expiry) while the parent is still retrying. - **Success-only finalization** — `uploadedBuilds` is marked only on actual success, so a failed build is never advertised as uploaded to peers; new `orchestrator.snapshot.upload.failed` counter surfaces give-ups. - **Graceful shutdown drains uploads** — instead of cancelling on shutdown, `Server.Close(ctx)` waits for in-flight uploads (bounded by the close context, so a forced stop still exits fast), with periodic drain-progress logging.
1 parent de2f391 commit 4f81799

9 files changed

Lines changed: 536 additions & 45 deletions

File tree

‎packages/orchestrator/pkg/factories/run.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -678,8 +678,8 @@ func run(config cfg.Config, opts Options) (success bool) {
678678
if err != nil {
679679
logger.L().Fatal(ctx, "failed to create orchestrator server", zap.Error(err))
680680
}
681-
closers = append(closers, closer{"orchestrator server", func(context.Context) error {
682-
return orchestratorService.Close()
681+
closers = append(closers, closer{"orchestrator server", func(ctx context.Context) error {
682+
return orchestratorService.Close(ctx)
683683
}})
684684

685685
// template manager sandbox logger

‎packages/orchestrator/pkg/sandbox/uploads.go‎

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -32,14 +32,19 @@ var (
3232
)
3333

3434
const (
35-
futureTTL = 1 * time.Hour
35+
// futureTTL must outlive a parent upload's full retry window so a child's
36+
// in-memory Wait still finds the parent's future. Keep >= the upload retry
37+
// budget (server.uploadTotalBudget, 2h).
38+
futureTTL = 3 * time.Hour
3639

3740
// refreshHeaderBudget bounds how long an upload Wait polls remote storage
3841
// for a parent's V4 header. Crosses orchestrators: A may still be uploading
39-
// on a remote orch when B's runV4 calls Wait(A) here. Matches the
40-
// per-upload bound in server.uploadTimeout — anything longer means the
41-
// parent's upload is itself stuck and would have failed on its own.
42-
refreshHeaderBudget = 20 * time.Minute
42+
// on a remote orch when B's runV4 calls Wait(A) here. It must be >= the
43+
// parent's full retry window (server.uploadTotalBudget, 2h); otherwise the
44+
// poll's budget expiry returns a non-retryable "object does not exist" and
45+
// the child gives up while the parent is still retrying. The per-attempt
46+
// context (server.uploadTimeout) bounds the actual poll duration.
47+
refreshHeaderBudget = 2 * time.Hour
4348

4449
// uploadDoneChannelPrefix is the Redis pub/sub channel prefix for per-build
4550
// upload-finished signals. Empty payload = success; non-empty = upload error.

‎packages/orchestrator/pkg/server/main.go‎

Lines changed: 64 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66
"context"
77
"fmt"
88
"sync"
9+
"sync/atomic"
910
"time"
1011

1112
"github.com/jellydator/ttlcache/v3"
@@ -38,6 +39,10 @@ const uploadedBuildsTTL = 1 * time.Hour
3839
// MaxStartingInstancesPerNode feature flag and resize the semaphore.
3940
const startingSandboxesLimitRefreshInterval = 30 * time.Second
4041

42+
// uploadDrainLogInterval is how often Close logs progress while waiting for
43+
// in-flight snapshot uploads to finish during shutdown.
44+
const uploadDrainLogInterval = 10 * time.Second
45+
4146
type Server struct {
4247
orchestrator.UnimplementedSandboxServiceServer
4348
orchestrator.UnimplementedChunkServiceServer
@@ -58,6 +63,13 @@ type Server struct {
5863
uploads *sandbox.Uploads
5964
sandboxCreateDuration metric.Int64Histogram
6065
sandboxKilledCounter metric.Int64Counter
66+
uploadFailedCounter metric.Int64Counter
67+
68+
// uploadsWG tracks in-flight async snapshot uploads so a graceful shutdown
69+
// can wait for them to finish instead of dropping them. uploadsInFlight is
70+
// the live count, used to log drain progress during shutdown.
71+
uploadsWG sync.WaitGroup
72+
uploadsInFlight atomic.Int64
6173

6274
done chan struct{}
6375
closeOnce sync.Once
@@ -123,6 +135,12 @@ func New(ctx context.Context, cfg ServiceConfig) (*Server, error) {
123135
}
124136
server.sandboxKilledCounter = sandboxKilledCounter
125137

138+
uploadFailedCounter, err := telemetry.GetCounter(meter, telemetry.OrchestratorSnapshotUploadFailedCounterName)
139+
if err != nil {
140+
return nil, fmt.Errorf("failed to register snapshot upload failed counter: %w", err)
141+
}
142+
server.uploadFailedCounter = uploadFailedCounter
143+
126144
_, err = telemetry.GetObservableUpDownCounter(meter, telemetry.OrchestratorSandboxCountMeterName, func(_ context.Context, observer metric.Int64Observer) error {
127145
observer.Observe(int64(server.sandboxFactory.Sandboxes.Count()))
128146

@@ -195,16 +213,61 @@ func New(ctx context.Context, cfg ServiceConfig) (*Server, error) {
195213
return server, nil
196214
}
197215

198-
func (s *Server) Close() error {
216+
func (s *Server) Close(ctx context.Context) error {
199217
s.closeOnce.Do(func() {
200218
close(s.done)
201219
})
202220

221+
// Wait for in-flight snapshot uploads to finish so a graceful shutdown
222+
// doesn't drop a snapshot that is still uploading. ctx is cancelled on a
223+
// forced stop, in which case we stop waiting and let the process exit.
224+
uploadsDone := make(chan struct{})
225+
go func() {
226+
s.uploadsWG.Wait()
227+
close(uploadsDone)
228+
}()
229+
230+
s.drainUploads(ctx, uploadsDone)
231+
203232
s.uploadedBuilds.Stop()
204233

205234
return nil
206235
}
207236

237+
// drainUploads waits for in-flight snapshot uploads to finish, logging progress
238+
// periodically, until they complete or ctx is cancelled (forced stop).
239+
func (s *Server) drainUploads(ctx context.Context, uploadsDone <-chan struct{}) {
240+
inFlight := s.uploadsInFlight.Load()
241+
if inFlight == 0 {
242+
return
243+
}
244+
245+
logger.L().Info(ctx, "waiting for in-flight snapshot uploads to finish", zap.Int64("uploads", inFlight))
246+
247+
ticker := time.NewTicker(uploadDrainLogInterval)
248+
defer ticker.Stop()
249+
250+
for {
251+
select {
252+
case <-uploadsDone:
253+
logger.L().Info(ctx, "all in-flight snapshot uploads finished")
254+
255+
return
256+
case <-ctx.Done():
257+
logger.L().Warn(ctx, "shutting down with snapshot uploads still in flight",
258+
zap.Int64("uploads", s.uploadsInFlight.Load()),
259+
zap.Error(context.Cause(ctx)),
260+
)
261+
262+
return
263+
case <-ticker.C:
264+
logger.L().Info(ctx, "still waiting for in-flight snapshot uploads",
265+
zap.Int64("uploads", s.uploadsInFlight.Load()),
266+
)
267+
}
268+
}
269+
}
270+
208271
func (s *Server) refreshStartingSandboxesLimit(ctx context.Context) {
209272
ticker := time.NewTicker(startingSandboxesLimitRefreshInterval)
210273
defer ticker.Stop()

‎packages/orchestrator/pkg/server/sandboxes.go‎

Lines changed: 51 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ import (
3333
"github.com/e2b-dev/infra/packages/shared/pkg/grpc/orchestrator"
3434
"github.com/e2b-dev/infra/packages/shared/pkg/logger"
3535
sbxlogger "github.com/e2b-dev/infra/packages/shared/pkg/logger/sandbox"
36+
"github.com/e2b-dev/infra/packages/shared/pkg/retry"
3637
"github.com/e2b-dev/infra/packages/shared/pkg/storage"
3738
"github.com/e2b-dev/infra/packages/shared/pkg/telemetry"
3839
"github.com/e2b-dev/infra/packages/shared/pkg/utils"
@@ -45,12 +46,25 @@ const (
4546
// acquireTimeout is the max time to wait for a semaphore for resuming sandboxes snapshot.
4647
acquireTimeout = 15 * time.Second
4748

48-
// uploadTimeout is the max time allowed for uploading snapshot files to
49-
// remote storage.
49+
// uploadTimeout is the max time allowed for a single upload attempt to
50+
// remote storage. The overall retry window is uploadTotalBudget.
5051
uploadTimeout = 20 * time.Minute
51-
// redisPeerKeyTTL is slightly longer than uploadTimeout so the key is still
52-
// valid for the entire upload window before being cleaned up.
53-
redisPeerKeyTTL = uploadTimeout + 2*time.Minute
52+
// uploadTotalBudget bounds how long a snapshot upload is retried before it
53+
// is given up. Covers a long GCS outage without retrying forever.
54+
uploadTotalBudget = 2 * time.Hour
55+
// redisPeerKeyTTL keeps the peer routing key valid across the whole retry
56+
// window so a long retry doesn't drop peer routing mid-upload. It is
57+
// unregistered promptly once the upload finishes (success or give-up).
58+
redisPeerKeyTTL = uploadTotalBudget + 2*time.Minute
59+
60+
// uploadRetryInitialBackoff is the wait before the first retry; it grows
61+
// exponentially up to uploadRetryMaxBackoff.
62+
uploadRetryInitialBackoff = 5 * time.Second
63+
// uploadRetryMaxBackoff caps the backoff between attempts.
64+
uploadRetryMaxBackoff = 2 * time.Minute
65+
// uploadRetryBackoffMultiplier is the exponential growth factor between
66+
// retry attempts.
67+
uploadRetryBackoffMultiplier = 2
5468

5569
// executionEventDataKey is the key used in webhook event data for sandbox execution metrics.
5670
executionEventDataKey = "execution"
@@ -899,7 +913,12 @@ func (s *Server) snapshotAndCacheSandbox(
899913
return
900914
}
901915

902-
s.uploadedBuilds.Set(meta.Template.BuildID, struct{}{}, ttlcache.DefaultTTL)
916+
// Only advertise the build as fully uploaded when it actually landed.
917+
// On abandon/failure the bytes are not in storage, so marking it would
918+
// make chunk-serving falsely report "already uploaded".
919+
if uploadErr == nil {
920+
s.uploadedBuilds.Set(meta.Template.BuildID, struct{}{}, ttlcache.DefaultTTL)
921+
}
903922

904923
if err := s.peerRegistry.Unregister(ctx, meta.Template.BuildID); err != nil {
905924
logger.L().Warn(ctx, "failed to unregister peer address from routing", zap.String("build_id", meta.Template.BuildID), zap.Error(err))
@@ -924,23 +943,40 @@ func (s *Server) snapshotAndCacheSandbox(
924943
// background and cleans up the Redis peer key once done. Used by the Pause
925944
// handler where no prefetch data is available.
926945
func (s *Server) uploadSnapshotAsync(ctx context.Context, sbx *sandbox.Sandbox, res *snapshotResult) {
927-
ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), uploadTimeout)
946+
// Detach from the request: the upload retries for up to uploadTotalBudget.
947+
// A graceful shutdown waits for it to finish (see Server.Close via uploadsWG)
948+
// rather than cancelling, so an in-flight snapshot isn't dropped on restart.
949+
uploadCtx := context.WithoutCancel(ctx)
928950

929-
go func() {
930-
defer cancel()
951+
s.uploadsInFlight.Add(1)
952+
s.uploadsWG.Go(func() {
953+
defer s.uploadsInFlight.Add(-1)
931954

932-
ctx, span := tracer.Start(ctx, "upload snapshot")
955+
spanCtx, span := tracer.Start(uploadCtx, "upload snapshot")
933956
defer span.End()
934957

935-
err := res.upload.Run(ctx)
958+
err := retry.Do(
959+
spanCtx,
960+
defaultUploadRetryPolicy(),
961+
isRetryableUploadErr,
962+
res.upload.Run,
963+
func(attempt int, backoff time.Duration, err error) {
964+
sbxlogger.I(sbx).Warn(spanCtx, "snapshot upload attempt failed, retrying",
965+
zap.Int("attempt", attempt),
966+
zap.Duration("backoff", backoff),
967+
zap.Error(err),
968+
)
969+
},
970+
)
936971
if err != nil {
937-
sbxlogger.I(sbx).Error(ctx, "error uploading snapshot files", zap.Error(err))
972+
sbxlogger.I(sbx).Error(spanCtx, "snapshot upload did not durably land", zap.Error(err))
973+
s.uploadFailedCounter.Add(spanCtx, 1)
938974
} else {
939-
sbxlogger.I(sbx).Info(ctx, "snapshot finished uploading successfully")
975+
sbxlogger.I(sbx).Info(spanCtx, "snapshot finished uploading successfully")
940976
}
941977

942-
res.completeUpload(ctx, err)
943-
}()
978+
res.completeUpload(spanCtx, err)
979+
})
944980
}
945981

946982
// setupSandboxLifecycle sets up the cleanup goroutine for a sandbox.
Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
//go:build linux
2+
3+
package server
4+
5+
import (
6+
"context"
7+
"errors"
8+
9+
"github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/build"
10+
"github.com/e2b-dev/infra/packages/shared/pkg/retry"
11+
"github.com/e2b-dev/infra/packages/shared/pkg/storage"
12+
)
13+
14+
// defaultUploadRetryPolicy is the retry policy for pause-snapshot uploads:
15+
// retry with a fresh per-attempt timeout under the total budget, with
16+
// exponential backoff.
17+
func defaultUploadRetryPolicy() retry.Policy {
18+
return retry.Policy{
19+
TotalBudget: uploadTotalBudget,
20+
AttemptTimeout: uploadTimeout,
21+
InitialBackoff: uploadRetryInitialBackoff,
22+
MaxBackoff: uploadRetryMaxBackoff,
23+
Multiplier: uploadRetryBackoffMultiplier,
24+
}
25+
}
26+
27+
// isRetryableUploadErr classifies an upload failure. The default is RETRYABLE:
28+
// a lost snapshot is unrecoverable and cascades to descendants, so a wasted
29+
// retry is far cheaper than dropping a recoverable build. Only genuinely
30+
// terminal conditions stop the loop.
31+
func isRetryableUploadErr(err error) bool {
32+
switch {
33+
case errors.Is(err, build.NoDiffError{}):
34+
return false // nothing to upload
35+
case errors.Is(err, storage.ErrObjectNotExist):
36+
return false // source vanished; retry cannot recover it
37+
case errors.Is(err, context.Canceled):
38+
return false // parent cancelled (shutdown)
39+
default:
40+
// Includes per-attempt context.DeadlineExceeded, GCS 401/503, rate
41+
// limiting, and unknown errors — all worth retrying within the budget.
42+
return true
43+
}
44+
}
Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
//go:build linux
2+
3+
package server
4+
5+
import (
6+
"context"
7+
"errors"
8+
"fmt"
9+
"testing"
10+
11+
"github.com/stretchr/testify/assert"
12+
13+
"github.com/e2b-dev/infra/packages/orchestrator/pkg/sandbox/build"
14+
"github.com/e2b-dev/infra/packages/shared/pkg/storage"
15+
)
16+
17+
func TestIsRetryableUploadErr(t *testing.T) {
18+
t.Parallel()
19+
20+
tests := []struct {
21+
name string
22+
err error
23+
retryable bool
24+
}{
25+
{"no diff", build.NoDiffError{}, false},
26+
{"object not exist", storage.ErrObjectNotExist, false},
27+
{"object not exist wrapped", fmt.Errorf("load: %w", storage.ErrObjectNotExist), false},
28+
{"parent cancelled", context.Canceled, false},
29+
{"per-attempt deadline", context.DeadlineExceeded, true},
30+
{"gcs 503", errors.New("server error (503)"), true},
31+
{"unknown", errors.New("boom"), true},
32+
}
33+
34+
for _, tt := range tests {
35+
t.Run(tt.name, func(t *testing.T) {
36+
t.Parallel()
37+
assert.Equal(t, tt.retryable, isRetryableUploadErr(tt.err))
38+
})
39+
}
40+
}

0 commit comments

Comments
 (0)