diff --git a/packages/api/internal/orchestrator/create_instance.go b/packages/api/internal/orchestrator/create_instance.go index 46102e9aad..9d3ea5c586 100644 --- a/packages/api/internal/orchestrator/create_instance.go +++ b/packages/api/internal/orchestrator/create_instance.go @@ -215,6 +215,12 @@ const ( resumePublishMargin = 5 * time.Second ) +// maxReserveRecontend bounds how many times a joined waiter re-contends for the +// start reservation after the leader left before completing (its own context +// was cancelled). Each re-contend requires a leader to have released, so a small +// bound covers the common hand-off while preventing a retry storm. +const maxReserveRecontend = 2 + // CreateOption adjusts one CreateSandbox call. type CreateOption func(*createOptions) @@ -257,33 +263,51 @@ func (o *Orchestrator) CreateSandbox( // Calculate total concurrent instances including addons totalConcurrentInstances := team.Limits.SandboxConcurrency - // Check if team has reached max instances - reservedAt := time.Now() - finishStart, waitForStart, err := o.sandboxStore.Reserve(ctx, team.Team.ID, sandboxID, int(totalConcurrentInstances)) - if err != nil { - var limitErr *sandbox.LimitExceededError + // Reserve a start slot. A joined waiter whose leader gave up only because + // the leader's own context was cancelled (ErrReservationLeaderLeft) does + // not inherit that cancellation — it re-contends here, becoming the new + // leader or joining whoever won the re-contention, bounded by + // maxReserveRecontend. + var ( + finishStart func(sandbox.Sandbox, error) + waitForStart func(ctx context.Context) (sandbox.Sandbox, error) + reservedAt time.Time + ) + for recontend := 0; ; recontend++ { + // Check if team has reached max instances + reservedAt = time.Now() + + var reserveErr error + finishStart, waitForStart, reserveErr = o.sandboxStore.Reserve(ctx, team.Team.ID, sandboxID, int(totalConcurrentInstances)) + if reserveErr != nil { + var limitErr *sandbox.LimitExceededError + + switch { + case errors.As(reserveErr, &limitErr): + return sandbox.Sandbox{}, &api.APIError{ + Code: http.StatusTooManyRequests, + ClientMsg: fmt.Sprintf( + "you have reached the maximum number of concurrent E2B sandboxes (%d). If you need more, "+ + "please visit 'https://e2b.dev/docs/billing'", totalConcurrentInstances), + Err: fmt.Errorf("team '%s' has reached the maximum number of instances (%d)", team.ID, totalConcurrentInstances), + } + default: + logger.L().Error(ctx, "failed to reserve sandbox for team", logger.WithSandboxID(sandboxID), zap.Error(reserveErr)) - switch { - case errors.As(err, &limitErr): - return sandbox.Sandbox{}, &api.APIError{ - Code: http.StatusTooManyRequests, - ClientMsg: fmt.Sprintf( - "you have reached the maximum number of concurrent E2B sandboxes (%d). If you need more, "+ - "please visit 'https://e2b.dev/docs/billing'", totalConcurrentInstances), - Err: fmt.Errorf("team '%s' has reached the maximum number of instances (%d)", team.ID, totalConcurrentInstances), + return sandbox.Sandbox{}, &api.APIError{ + Code: http.StatusInternalServerError, + ClientMsg: fmt.Sprintf("Failed to create sandbox: %s", reserveErr), + Err: reserveErr, + } } - default: - logger.L().Error(ctx, "failed to reserve sandbox for team", logger.WithSandboxID(sandboxID), zap.Error(err)) + } - return sandbox.Sandbox{}, &api.APIError{ - Code: http.StatusInternalServerError, - ClientMsg: fmt.Sprintf("Failed to create sandbox: %s", err), - Err: err, - } + if waitForStart == nil { + // We won the reservation: we are the leader. Fall through to the + // start below. + break } - } - if waitForStart != nil { // A joined request rides whatever start is already in flight — which may // be a memory restore (e.g. a traffic-triggered auto-resume) that an // explicit memory:false must never be silently answered with. @@ -301,22 +325,41 @@ func (o *Orchestrator) CreateSandbox( logger.L().Info(ctx, "sandbox is already being started, waiting for it to be ready", logger.WithSandboxID(sandboxID)) - sbx, err = waitForStart(ctx) - if err != nil { - logger.L().Warn(ctx, "Error waiting for sandbox to start", zap.Error(err), logger.WithSandboxID(sandboxID)) + var waitErr error + sbx, waitErr = waitForStart(ctx) + if waitErr == nil { + return sbx, nil + } + + // The leader gave up only because its own context was cancelled, not + // because the start failed. If our own context is still alive and we + // have budget, re-contend rather than adopting the leader's cut. + if errors.Is(waitErr, sandbox.ErrReservationLeaderLeft) { + if ctx.Err() == nil && recontend < maxReserveRecontend { + logger.L().Info(ctx, "start reservation leader left before completing, re-contending", + logger.WithSandboxID(sandboxID), zap.Int("attempt", recontend+1)) - if apiErr, ok := errors.AsType[*api.APIError](err); ok { - return sandbox.Sandbox{}, apiErr + continue } return sandbox.Sandbox{}, &api.APIError{ - Code: http.StatusInternalServerError, - ClientMsg: "Error waiting for sandbox to start", - Err: err, + Code: http.StatusServiceUnavailable, + ClientMsg: "Sandbox start was interrupted, please retry", + Err: waitErr, } } - return sbx, nil + logger.L().Warn(ctx, "Error waiting for sandbox to start", zap.Error(waitErr), logger.WithSandboxID(sandboxID)) + + if apiErr, ok := errors.AsType[*api.APIError](waitErr); ok { + return sandbox.Sandbox{}, apiErr + } + + return sandbox.Sandbox{}, &api.APIError{ + Code: http.StatusInternalServerError, + ClientMsg: "Error waiting for sandbox to start", + Err: waitErr, + } } telemetry.ReportEvent(ctx, "Reserved sandbox for team") @@ -341,7 +384,18 @@ func (o *Orchestrator) CreateSandbox( // Don't change this handling // https://go.dev/play/p/4oy02s7BDMc if apiErr != nil { - finishStart(sbx, apiErr) + // A failure caused only by this leader's own context ending (the + // client disconnected or the deadline fired) says nothing about + // whether the sandbox can start. Publish it wrapped as + // ErrReservationLeaderLeft so a joined waiter with a live context + // re-contends instead of inheriting the cut; the caller still gets + // the original apiErr. + publishErr := error(apiErr) + if ctx.Err() != nil { + publishErr = fmt.Errorf("%w: %w", sandbox.ErrReservationLeaderLeft, apiErr) + } + + finishStart(sbx, publishErr) } else { finishStart(sbx, nil) } diff --git a/packages/api/internal/sandbox/aliases.go b/packages/api/internal/sandbox/aliases.go index 76fe0aa65a..e53807e8f9 100644 --- a/packages/api/internal/sandbox/aliases.go +++ b/packages/api/internal/sandbox/aliases.go @@ -53,14 +53,15 @@ const ( // Errors and pre-defined state actions / transition tables. var ( - ErrAlreadyExists = sandboxtypes.ErrAlreadyExists - ErrNotFound = sandboxtypes.ErrNotFound - ErrEvictionInProgress = sandboxtypes.ErrEvictionInProgress - ErrEvictionNotNeeded = sandboxtypes.ErrEvictionNotNeeded - ErrExecutionMismatch = sandboxtypes.ErrExecutionMismatch - ErrRestoreConflict = sandboxtypes.ErrRestoreConflict - ErrTransitionRestored = sandboxtypes.ErrTransitionRestored - ErrDraining = sandboxtypes.ErrDraining + ErrAlreadyExists = sandboxtypes.ErrAlreadyExists + ErrNotFound = sandboxtypes.ErrNotFound + ErrEvictionInProgress = sandboxtypes.ErrEvictionInProgress + ErrEvictionNotNeeded = sandboxtypes.ErrEvictionNotNeeded + ErrExecutionMismatch = sandboxtypes.ErrExecutionMismatch + ErrRestoreConflict = sandboxtypes.ErrRestoreConflict + ErrTransitionRestored = sandboxtypes.ErrTransitionRestored + ErrDraining = sandboxtypes.ErrDraining + ErrReservationLeaderLeft = sandboxtypes.ErrReservationLeaderLeft AllowedTransitions = sandboxtypes.AllowedTransitions diff --git a/packages/api/internal/sandbox/reservations/redis/reservation_test.go b/packages/api/internal/sandbox/reservations/redis/reservation_test.go index d614900b56..f08d9c8678 100644 --- a/packages/api/internal/sandbox/reservations/redis/reservation_test.go +++ b/packages/api/internal/sandbox/reservations/redis/reservation_test.go @@ -193,6 +193,44 @@ func TestReservation_MultipleTeams(t *testing.T) { require.ErrorAs(t, err, new(&sandboxtypes.LimitExceededError{})) } +// TestReservation_LeaderLeftLetsWaiterRecontend proves the #3686 fix at the +// storage layer: when the leader publishes a failure wrapping +// ErrReservationLeaderLeft (its own context was cut), a waiter decodes the +// sentinel, and — because that result is terminal and clears the pending entry +// — a fresh Reserve wins the reservation again (the waiter becomes the new +// leader). A genuine failure, by contrast, does not carry the sentinel. +func TestReservation_LeaderLeftLetsWaiterRecontend(t *testing.T) { + t.Parallel() + storage, _ := setupTestReservationStorage(t) + + teamID := uuid.New() + sbxID := "leader-left-sandbox" + + // Leader reserves, a waiter joins. + finishStart, _, err := storage.Reserve(t.Context(), teamID, sbxID, 10) + require.NoError(t, err) + require.NotNil(t, finishStart) + + _, waitForStart, err := storage.Reserve(t.Context(), teamID, sbxID, 10) + require.NoError(t, err) + require.NotNil(t, waitForStart) + + // Leader gives up only because its own context was cut. + finishStart(sandboxtypes.Sandbox{}, fmt.Errorf("%w: placement timed out", sandboxtypes.ErrReservationLeaderLeft)) + + // The waiter sees the re-contend sentinel, not a terminal failure. + _, waitErr := waitForStart(t.Context()) + require.Error(t, waitErr) + require.ErrorIs(t, waitErr, sandboxtypes.ErrReservationLeaderLeft) + + // Re-contending wins the reservation (the entry was cleared): the waiter + // can become the new leader rather than inheriting the cut. + finishStart2, waitForStart2, err := storage.Reserve(t.Context(), teamID, sbxID, 10) + require.NoError(t, err) + require.NotNil(t, finishStart2, "re-contend should win the reservation") + require.Nil(t, waitForStart2) +} + func TestReservation_FailedStart(t *testing.T) { t.Parallel() storage, _ := setupTestReservationStorage(t) diff --git a/packages/api/internal/sandbox/reservations/redis/result.go b/packages/api/internal/sandbox/reservations/redis/result.go index 46fa78c232..758b319678 100644 --- a/packages/api/internal/sandbox/reservations/redis/result.go +++ b/packages/api/internal/sandbox/reservations/redis/result.go @@ -13,6 +13,11 @@ import ( type reservationResult struct { Sandbox sandboxtypes.Sandbox `json:"sandbox,omitzero"` Error *reservationError `json:"error,omitempty"` + // LeaderLeft marks a result the leader published only because its own + // context was cancelled (client disconnect / deadline), not a genuine + // start failure. A waiter decodes it as ErrReservationLeaderLeft so it can + // re-contend instead of adopting the leader's cancellation. + LeaderLeft bool `json:"leader_left,omitempty"` } // reservationError preserves api.APIError fields for cross-instance error propagation. @@ -36,6 +41,7 @@ func encodeResult(sbx sandboxtypes.Sandbox, err error) ([]byte, error) { re.ErrorCode = apiErr.ErrorCode } result.Error = re + result.LeaderLeft = errors.Is(err, sandboxtypes.ErrReservationLeaderLeft) } return json.Marshal(result) @@ -47,6 +53,18 @@ func decodeResult(data []byte) (sandboxtypes.Sandbox, error) { return sandboxtypes.Sandbox{}, fmt.Errorf("failed to unmarshal reservation result: %w", err) } + if result.LeaderLeft { + // The leader gave up only because its own context was cancelled. + // Surface the typed sentinel so a waiter with a live context can + // re-contend; preserve the leader's message for the logs. + msg := "leader left" + if result.Error != nil { + msg = result.Error.Message + } + + return sandboxtypes.Sandbox{}, fmt.Errorf("%w: %s", sandboxtypes.ErrReservationLeaderLeft, msg) + } + if result.Error != nil { return sandboxtypes.Sandbox{}, reconstructError(result.Error) } diff --git a/packages/api/internal/sandbox/reservations/redis/result_test.go b/packages/api/internal/sandbox/reservations/redis/result_test.go index bbb555024a..88ce4cb872 100644 --- a/packages/api/internal/sandbox/reservations/redis/result_test.go +++ b/packages/api/internal/sandbox/reservations/redis/result_test.go @@ -2,6 +2,7 @@ package redis import ( "errors" + "fmt" "net/http" "testing" @@ -35,3 +36,47 @@ func TestEncodeDecodeResult_APIErrorRoundTripsAllFields(t *testing.T) { assert.Equal(t, in.ErrorCode, out.ErrorCode) assert.Equal(t, in.Err.Error(), out.Err.Error()) } + +// A failure the leader published only because its own context was cancelled is +// carried across instances as ErrReservationLeaderLeft, so a joined waiter can +// detect it (via errors.Is) and re-contend. See issue #3686. +func TestEncodeDecodeResult_LeaderLeftIsCarriedAsSentinel(t *testing.T) { + t.Parallel() + + leaderErr := fmt.Errorf("%w: %w", sandboxtypes.ErrReservationLeaderLeft, &api.APIError{ + Code: http.StatusGatewayTimeout, + ClientMsg: "Failed to place sandbox: placement timed out, please retry", + ErrorCode: "sandbox_placement_timeout", + Err: errors.New("context canceled"), + }) + + data, err := encodeResult(sandboxtypes.Sandbox{}, leaderErr) + require.NoError(t, err) + + _, decodedErr := decodeResult(data) + require.Error(t, decodedErr) + assert.ErrorIs(t, decodedErr, sandboxtypes.ErrReservationLeaderLeft, + "a leader-left result must decode to the re-contend sentinel") +} + +// A genuine start failure (the leader's context was NOT cut) must not be +// mistaken for a leader-left result: the waiter adopts it as before. +func TestEncodeDecodeResult_GenuineFailureIsNotLeaderLeft(t *testing.T) { + t.Parallel() + + data, err := encodeResult(sandboxtypes.Sandbox{}, &api.APIError{ + Code: http.StatusServiceUnavailable, + ClientMsg: "Failed to place sandbox: no compatible node for this template's requirements", + ErrorCode: "sandbox_no_compatible_node", + Err: errors.New("no eligible nodes"), + }) + require.NoError(t, err) + + _, decodedErr := decodeResult(data) + require.Error(t, decodedErr) + require.NotErrorIs(t, decodedErr, sandboxtypes.ErrReservationLeaderLeft) + + var out *api.APIError + require.ErrorAs(t, decodedErr, &out) + assert.Equal(t, http.StatusServiceUnavailable, out.Code) +} diff --git a/packages/api/internal/sandbox/sandboxtypes/errors.go b/packages/api/internal/sandbox/sandboxtypes/errors.go index 1d73f03119..95cac1d83f 100644 --- a/packages/api/internal/sandbox/sandboxtypes/errors.go +++ b/packages/api/internal/sandbox/sandboxtypes/errors.go @@ -37,6 +37,14 @@ func (e *NotRunningError) Error() string { var ErrAlreadyExists = errors.New("sandbox already exists") +// ErrReservationLeaderLeft is the result a joined waiter sees when the leader +// request that held the reservation gave up only because its own context was +// cancelled (a client disconnect or deadline), not because the start genuinely +// failed. It says nothing about whether the sandbox can start, so a waiter +// whose own context is still alive should re-contend for the reservation +// rather than adopt it as a terminal failure. +var ErrReservationLeaderLeft = errors.New("sandbox start reservation leader left before completing") + var ErrEvictionInProgress = errors.New("sandbox eviction already in progress") var ErrEvictionNotNeeded = errors.New("sandbox eviction not needed")