feat(api): add the outbox worker that tears down a deleted team's resources

GitOrigin-RevId: 4675d05f809ca00f11b6dc503fe06dc97d904f5c

Author: pmakanju-e2b

docs/ARCHITECTURE.md

@@ -197,6 +197,25 @@ The control-plane entry point (Gin, OpenAPI-generated from `spec/openapi.yml`, p
   operation.
 - **Extra listeners**: internal gRPC :5009 and edge gRPC :5109 expose `ResumeSandbox` so
   client-proxy can wake paused sandboxes on incoming traffic.
+- **Outbox** (`internal/outbox`): a River client on the shared database's `river` schema
+  works jobs that other services enqueue there for work needing the orchestrator
+  connections. Its job arguments live in `packages/db/pkg/outbox`. The one kind today,
+  `teardown_team_resources`, stops a deleted team's workloads. A guard runs first on every
+  attempt: a team that exists and is not blocked fails the attempt, and the job retries rather
+  than kill a live team's sandboxes; a missing team row does not stop it. Then
+  `kill_sandboxes` kills the team's running sandboxes and succeeds only once the sandbox store
+  holds none of them in any state, so a sandbox still pausing or being killed makes the job
+  retry. A sandbox that is not running cannot be paused, so no new snapshot of the team
+  appears afterwards. Killing through the orchestrator is what removes the sandboxes' store
+  records, so the job deletes no Redis state itself, and every attempt is safe to run again.
+  Builds are not cancelled: they end on their own build timeout. An attempt is bounded at
+  30 minutes; retries back off from 30 seconds to a 15-minute cap, so the 100 attempts last
+  about a day before River discards the job and keeps it. Volumes, templates, snapshots and
+  the team row are left alone. Steps report
+  `api.outbox.steps.finished` and `api.outbox.step.duration`; finished jobs report
+  `outbox.jobs.finished`, and the backlog reports `outbox.jobs` and `outbox.oldest_*_age`.
+  `OUTBOX_MAX_WORKERS` (default 10) and `OUTBOX_BACKLOG_INTERVAL` (default 30s) configure it;
+  shutdown stops it after the sandbox-work drain, before the database and Redis clients close.
 - Reads ClickHouse for sandbox/team metrics endpoints. Sandbox and template-build logs default to
   Loki, with a LaunchDarkly-gated ClickHouse read path (`logs-read-config`) for local-cluster logs
   during the log storage migration. `LOKI_URL` is optional: without it the api has no Loki client
@@ -423,8 +442,9 @@ uncoded or unknown reasons as generic errors.
 `DELETE /v1/management/projects/{teamID}` is declared and answers 501. `envs`, `snapshots` and
 `volumes` reference `teams` with `ON DELETE NO ACTION` and templates are only soft-deleted, so a
 project that ever built one pins its team row — and releasing it needs the API service's
-orchestrator connections, which this service does not have. Projects are not deleted from control
-planes today.
+orchestrator connections, which this service does not have. The API's `teardown_team_resources`
+outbox job (see the API section) is the worker for that release; nothing enqueues it yet, and
+projects are not deleted from control planes today.
 
 ## Data stores
 
@@ -447,7 +467,7 @@ The same database holds River's job tables in the `river` schema, which a goose
 creates. The db-migrator applies the goose migrations and then River's own, under one advisory
 lock; `make migrate` runs it, because the goose CLI cannot apply River's. At startup the API and
 dashboard-api refuse a database whose goose version is older than the one they were built against,
-or whose River migrations are not current.
+or whose River migrations are not current. The API works jobs from that schema (see its Outbox).
 
 A template and a paused-sandbox snapshot have the **same artifact shape** — a snapshot is just a
 new build whose memfile/rootfs are stored as diffs against the template it came from (diff chains

packages/api/go.mod

@@ -48,6 +48,9 @@ require (
 	github.com/oapi-codegen/runtime v1.6.0
 	github.com/posthog/posthog-go v0.0.0-20230801140217-d607812dee69
 	github.com/redis/go-redis/v9 v9.21.0
+	github.com/riverqueue/river v0.47.0
+	github.com/riverqueue/river/riverdriver/riverpgxv5 v0.47.0
+	github.com/riverqueue/river/rivertype v0.47.0
 	github.com/stretchr/testify v1.12.1
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.71.0
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.71.0
@@ -329,12 +332,10 @@ require (
 	github.com/quic-go/quic-go v0.61.0 // indirect
 	github.com/redis/go-redis/extra/rediscmd/v9 v9.17.3 // indirect
 	github.com/redis/go-redis/extra/redisotel/v9 v9.17.3 // indirect
-	github.com/riverqueue/river v0.47.0 // indirect
 	github.com/riverqueue/river/riverdriver v0.47.0 // indirect
 	github.com/riverqueue/river/riverdriver/riverdatabasesql v0.47.0 // indirect
-	github.com/riverqueue/river/riverdriver/riverpgxv5 v0.47.0 // indirect
 	github.com/riverqueue/river/rivershared v0.47.0 // indirect
-	github.com/riverqueue/river/rivertype v0.47.0 // indirect
+	github.com/riverqueue/rivercontrib/otelriver v0.12.0 // indirect
 	github.com/rs/zerolog v1.34.0 // indirect
 	github.com/samber/lo v1.53.0 // indirect
 	github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 // indirect
@@ -357,6 +358,10 @@ require (
 	github.com/tdewolff/test v1.0.12 // indirect
 	github.com/testcontainers/testcontainers-go v0.44.0 // indirect
 	github.com/testcontainers/testcontainers-go/modules/postgres v0.44.0 // indirect
+	github.com/tidwall/gjson v1.19.0 // indirect
+	github.com/tidwall/match v1.2.0 // indirect
+	github.com/tidwall/pretty v1.2.1 // indirect
+	github.com/tidwall/sjson v1.2.5 // indirect
 	github.com/tjhop/slog-gokit v0.1.4 // indirect
 	github.com/tklauser/go-sysconf v0.4.0 // indirect
 	github.com/tklauser/numcpus v0.12.0 // indirect

packages/api/go.sum

@@ -167,6 +167,13 @@ type Config struct {
 	// placement (max of CPU and pool). A node that reports no pool scores
 	// 0.5, so clusters without a pool should set this false and rank on CPU.
 	BestOfKHugepageMemory bool `env:"BEST_OF_K_HUGEPAGE_MEMORY" envDefault:"true"`
+
+	// OutboxMaxWorkers is how many River outbox jobs one API replica works at
+	// once.
+	OutboxMaxWorkers int `env:"OUTBOX_MAX_WORKERS" envDefault:"10"`
+	// OutboxBacklogInterval is how often the outbox backlog gauges are read
+	// from the database.
+	OutboxBacklogInterval time.Duration `env:"OUTBOX_BACKLOG_INTERVAL" envDefault:"30s"`
 }
 
 type FailureCondition string

packages/api/internal/cfg/model.go

@@ -1,132 +1,38 @@
 package handlers
 
 import (
-	"context"
-	"fmt"
 	"net/http"
-	"sync/atomic"
-	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
 	"go.uber.org/zap"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/e2b-dev/infra/packages/api/internal/api"
-	dbtypes "github.com/e2b-dev/infra/packages/db/pkg/types"
-	"github.com/e2b-dev/infra/packages/db/queries"
-	"github.com/e2b-dev/infra/packages/shared/pkg/clusters"
-	templatemanagergrpc "github.com/e2b-dev/infra/packages/shared/pkg/grpc/template-manager"
 	"github.com/e2b-dev/infra/packages/shared/pkg/logger"
 )
 
-// buildDeleteTimeout bounds the node-side delete once the build has been
-// recorded as failed.
-const buildDeleteTimeout = 30 * time.Second
-
-type buildCanceller interface {
-	SetTerminalStatus(ctx context.Context, buildID uuid.UUID, statusGroup dbtypes.BuildStatusGroup, reason *templatemanagergrpc.TemplateBuildStatusReason) (bool, error)
-	DeleteBuild(ctx context.Context, buildID uuid.UUID, templateID string, clusterID uuid.UUID, nodeID string) error
-}
-
-// cancelBuild ends a build and stops it on its node. The node-side delete takes
-// the build's artifacts with it, so it only follows a write that ended the
-// build: one that finished on its own between the listing and here keeps both
-// its outcome and the artifacts that outcome refers to.
-func cancelBuild(ctx context.Context, tm buildCanceller, build queries.GetCancellableTemplateBuildsByTeamRow) error {
-	recorded, err := tm.SetTerminalStatus(ctx, build.BuildID, dbtypes.BuildStatusGroupFailed, &templatemanagergrpc.TemplateBuildStatusReason{
-		Message: "cancelled by admin",
-	})
-	if err != nil {
-		return fmt.Errorf("failed to set build status to failed: %w", err)
-	}
-
-	if !recorded || build.ClusterNodeID == nil {
-		return nil
-	}
-
-	// The write above dropped the build from every listing, so nothing retries
-	// this stop, and the request context dies with the caller or its deadline.
-	deleteCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), buildDeleteTimeout)
-	defer cancel()
-
-	err = tm.DeleteBuild(deleteCtx, build.BuildID, build.TemplateID, clusters.WithClusterFallback(build.ClusterID), *build.ClusterNodeID)
-	if err != nil {
-		return fmt.Errorf("failed to delete build on node: %w", err)
-	}
-
-	return nil
-}
-
 func (a *APIStore) PostAdminTeamsTeamIDBuildsCancel(c *gin.Context, teamID uuid.UUID) {
 	ctx := c.Request.Context()
 	ctx, span := tracer.Start(ctx, "cancel admin-team-builds")
 	defer span.End()
 
 	logger.L().Info(ctx, "Admin cancelling all builds for team", logger.WithTeamID(teamID.String()))
 
-	builds, err := a.sqlcDB.GetCancellableTemplateBuildsByTeam(ctx, teamID)
+	cancelledCount, failedCount, err := a.templateManager.CancelTeamBuilds(ctx, teamID, "cancelled by admin")
 	if err != nil {
 		a.sendAPIStoreError(c, http.StatusInternalServerError, "Failed to get builds")
 
 		return
 	}
 
-	logger.L().Info(ctx, "Found builds to cancel",
-		logger.WithTeamID(teamID.String()),
-		zap.Int("count", len(builds)),
-	)
-
-	cancelledCount := atomic.Int64{}
-	failedCount := atomic.Int64{}
-
-	wg := errgroup.Group{}
-	wg.SetLimit(10)
-
-	for _, b := range builds {
-		wg.Go(func() error {
-			buildID := b.BuildID
-			templateID := b.TemplateID
-
-			err := cancelBuild(ctx, a.templateManager, b)
-			if err != nil {
-				logger.L().Error(ctx, "Failed to cancel build",
-					zap.String("buildID", buildID.String()),
-					zap.String("templateID", templateID),
-					logger.WithTeamID(teamID.String()),
-					zap.Error(err))
-				failedCount.Add(1)
-
-				return nil
-			}
-
-			logger.L().Debug(ctx, "Successfully cancelled build",
-				zap.String("buildID", buildID.String()),
-				zap.String("templateID", templateID),
-				logger.WithTeamID(teamID.String()))
-			cancelledCount.Add(1)
-
-			return nil
-		})
-	}
-
-	err = wg.Wait()
-	if err != nil {
-		a.sendAPIStoreError(c, http.StatusInternalServerError, "Failed to cancel builds")
-
-		return
-	}
-
 	logger.L().Info(ctx, "Completed cancelling team builds",
 		logger.WithTeamID(teamID.String()),
-		zap.Int64("cancelled", cancelledCount.Load()),
-		zap.Int64("failed", failedCount.Load()),
+		zap.Int("cancelled", cancelledCount),
+		zap.Int("failed", failedCount),
 	)
 
-	result := api.AdminBuildCancelResult{
-		CancelledCount: int(cancelledCount.Load()),
-		FailedCount:    int(failedCount.Load()),
-	}
-
-	c.JSON(http.StatusOK, result)
+	c.JSON(http.StatusOK, api.AdminBuildCancelResult{
+		CancelledCount: cancelledCount,
+		FailedCount:    failedCount,
+	})
 }

packages/api/internal/handlers/admin_cancel_team_builds.go

@@ -2,12 +2,10 @@ package handlers
 
 import (
 	"net/http"
-	"sync/atomic"
 
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
 	"go.uber.org/zap"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/e2b-dev/infra/packages/api/internal/api"
 	"github.com/e2b-dev/infra/packages/api/internal/sandbox"
@@ -28,58 +26,13 @@ func (a *APIStore) PostAdminTeamsTeamIDSandboxesKill(c *gin.Context, teamID uuid
 
 	logger.L().Info(ctx, "Admin killing all sandboxes for team", logger.WithTeamID(teamID.String()))
 
-	// Get all running sandboxes for the team
-	sandboxes, err := a.orchestrator.GetSandboxes(ctx, teamID, []sandbox.State{sandbox.StateRunning})
+	killedCount, failedCount, err := a.orchestrator.KillTeamSandboxes(ctx, teamID, sandbox.KillReasonAdmin)
 	if err != nil {
 		a.sendAPIStoreError(c, http.StatusInternalServerError, "Failed to get sandboxes")
 
 		return
 	}
 
-	logger.L().Info(ctx, "Found sandboxes to kill",
-		logger.WithTeamID(teamID.String()),
-		zap.Int("count", len(sandboxes)),
-	)
-
-	killedCount := atomic.Int64{}
-	failedCount := atomic.Int64{}
-
-	wg := errgroup.Group{}
-	wg.SetLimit(10)
-
-	// Kill each sandbox
-	for _, sbx := range sandboxes {
-		wg.Go(func() error {
-			err := a.orchestrator.RemoveSandbox(ctx, sbx.TeamID, sbx.SandboxID, sandbox.RemoveOpts{
-				Action: sandbox.StateActionKill,
-				Reason: sandbox.KillReasonAdmin,
-			})
-			if err != nil {
-				logger.L().Error(ctx, "Failed to kill sandbox",
-					logger.WithSandboxID(sbx.SandboxID),
-					logger.WithTeamID(teamID.String()),
-					zap.String("kill_reason", sandbox.KillReasonAdmin.String()),
-					zap.Error(err))
-				failedCount.Add(1)
-			} else {
-				logger.L().Debug(ctx, "Successfully killed sandbox",
-					logger.WithSandboxID(sbx.SandboxID),
-					logger.WithTeamID(teamID.String()),
-					zap.String("kill_reason", sandbox.KillReasonAdmin.String()))
-				killedCount.Add(1)
-			}
-
-			return nil
-		})
-	}
-
-	err = wg.Wait()
-	if err != nil {
-		a.sendAPIStoreError(c, http.StatusInternalServerError, "Failed to kill sandboxes")
-
-		return
-	}
-
 	// Invalidate auth cache for this team so subsequent requests re-check against DB
 	if err := a.authService.InvalidateTeamCache(ctx, teamID); err != nil {
 		logger.L().Error(ctx, "Failed to invalidate auth cache for team",
@@ -89,15 +42,12 @@ func (a *APIStore) PostAdminTeamsTeamIDSandboxesKill(c *gin.Context, teamID uuid
 
 	logger.L().Info(ctx, "Completed killing team sandboxes",
 		zap.String("teamID", teamID.String()),
-		zap.Int64("killed", killedCount.Load()),
-		zap.Int64("failed", failedCount.Load()),
+		zap.Int("killed", killedCount),
+		zap.Int("failed", failedCount),
 	)
 
-	// Return result
-	result := api.AdminSandboxKillResult{
-		KilledCount: int(killedCount.Load()),
-		FailedCount: int(failedCount.Load()),
-	}
-
-	c.JSON(http.StatusOK, result)
+	c.JSON(http.StatusOK, api.AdminSandboxKillResult{
+		KilledCount: killedCount,
+		FailedCount: failedCount,
+	})
 }

packages/api/internal/handlers/admin_kill_team_sandboxes.go

@@ -26,6 +26,7 @@ import (
 	"github.com/e2b-dev/infra/packages/api/internal/cfg"
 	"github.com/e2b-dev/infra/packages/api/internal/clusters"
 	"github.com/e2b-dev/infra/packages/api/internal/orchestrator"
+	"github.com/e2b-dev/infra/packages/api/internal/outbox"
 	"github.com/e2b-dev/infra/packages/api/internal/sandbox"
 	managementv1 "github.com/e2b-dev/infra/packages/api/internal/secretsstore/management/v1"
 	template_manager "github.com/e2b-dev/infra/packages/api/internal/template-manager"
@@ -473,6 +474,22 @@ func NewAPIStore(ctx context.Context, tel *telemetry.Client, redisClient redis.U
 	return a
 }
 
+// NewOutbox builds the River client that works the API's outbox jobs on the
+// store's clients. Stop it before Close.
+func (a *APIStore) NewOutbox(l logger.Logger) (*outbox.River, error) {
+	return outbox.New(outbox.Dependencies{
+		Pool:      a.sqlcDB.Pool(),
+		Sandboxes: a.orchestrator,
+		Teams:     a.authDB,
+		Logger:    l,
+		Telemetry: a.Telemetry,
+		Config: outbox.Config{
+			MaxWorkers:      a.config.OutboxMaxWorkers,
+			BacklogInterval: a.config.OutboxBacklogInterval,
+		},
+	})
+}
+
 // Drain stops admitting sandbox work that outlives its request and waits for
 // what is in flight. It runs before Close, which tears down the clients that
 // work uses.