From d48ee804729aaddb4dbd2587e137a0570aff35cc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Tue, 12 May 2026 22:41:21 +0100 Subject: [PATCH] fix(e2e): release agent ports during E2E reset to prevent shard exhaustion (#888) --- apps/backend/cmd/kandev/e2e_reset.go | 36 ++++++++++++++++--- .../internal/task/repository/sqlite/task.go | 11 ------ .../internal/task/service/service_tasks.go | 7 ++-- apps/web/e2e/fixtures/backend.ts | 12 +++++-- 4 files changed, 45 insertions(+), 21 deletions(-) diff --git a/apps/backend/cmd/kandev/e2e_reset.go b/apps/backend/cmd/kandev/e2e_reset.go index 3ac0c4a8e5..ec0eaa8d30 100644 --- a/apps/backend/cmd/kandev/e2e_reset.go +++ b/apps/backend/cmd/kandev/e2e_reset.go @@ -22,7 +22,7 @@ func registerE2EResetRoutes(router *gin.Engine, repo *sqliterepo.Repository, tas } api := router.Group("/api/v1/e2e") - api.DELETE("/reset/:workspaceId", handleE2EReset(repo, log)) + api.DELETE("/reset/:workspaceId", handleE2EReset(repo, taskSvc, log)) // Hidden-workflow factory: lets E2E tests cover the system-only // workflow path (e.g. improve-kandev) without depending on the real // bootstrap endpoint, which clones from GitHub and shells out to gh. @@ -31,7 +31,7 @@ func registerE2EResetRoutes(router *gin.Engine, repo *sqliterepo.Repository, tas log.Info("registered E2E endpoints (test-only)") } -func handleE2EReset(repo *sqliterepo.Repository, log *logger.Logger) gin.HandlerFunc { +func handleE2EReset(repo *sqliterepo.Repository, taskSvc *taskservice.Service, log *logger.Logger) gin.HandlerFunc { return func(c *gin.Context) { workspaceID := c.Param("workspaceId") @@ -43,12 +43,40 @@ func handleE2EReset(repo *sqliterepo.Repository, log *logger.Logger) gin.Handler ctx := c.Request.Context() - deletedTasks, err := repo.DeleteTasksByWorkspace(ctx, workspaceID) + // Route through the task service (rather than a raw SQL DELETE) so + // each delete spawns the async cleanup goroutine that stops the + // agentctl instance and releases its port. Without this, instances + // accumulate across tests in the same Playwright worker and + // eventually exhaust the per-worker port range. + const resetPageSize = 10000 + tasks, total, err := repo.ListTasksByWorkspace(ctx, workspaceID, "", "", "", 1, resetPageSize, true, true, false, false) if err != nil { - log.Error("e2e reset: failed to delete tasks", zap.Error(err)) + log.Error("e2e reset: failed to list tasks", zap.Error(err)) c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) return } + if total > resetPageSize { + // Fail loudly rather than silently leaving tasks behind, which + // would leak agentctl instances and exhaust ports. + log.Error("e2e reset: task count exceeds page size", + zap.Int("total", total), zap.Int("page_size", resetPageSize)) + c.JSON(http.StatusInternalServerError, gin.H{ + "error": "task count exceeds reset page size", + }) + return + } + var deletedTasks int64 + for _, t := range tasks { + if err := taskSvc.DeleteTask(ctx, t.ID); err != nil { + // Abort: leaving an undeleted task with its workflow gone + // would create orphan rows visible to subsequent tests. + log.Error("e2e reset: failed to delete task", + zap.String("task_id", t.ID), zap.Error(err)) + c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + deletedTasks++ + } deletedWorkflows, err := repo.DeleteWorkflowsByWorkspace(ctx, workspaceID, keepWorkflowIDs) if err != nil { diff --git a/apps/backend/internal/task/repository/sqlite/task.go b/apps/backend/internal/task/repository/sqlite/task.go index 90002e6a50..5b2be575dc 100644 --- a/apps/backend/internal/task/repository/sqlite/task.go +++ b/apps/backend/internal/task/repository/sqlite/task.go @@ -96,17 +96,6 @@ func (r *Repository) UpdateTask(ctx context.Context, task *models.Task) error { return nil } -// DeleteTasksByWorkspace deletes all tasks for a workspace (E2E cleanup). -// Relies on CASCADE foreign keys to remove sessions, messages, turns, etc. -func (r *Repository) DeleteTasksByWorkspace(ctx context.Context, workspaceID string) (int64, error) { - result, err := r.db.ExecContext(ctx, r.db.Rebind(`DELETE FROM tasks WHERE workspace_id = ?`), workspaceID) - if err != nil { - return 0, err - } - rows, _ := result.RowsAffected() - return rows, nil -} - // DeleteTask deletes a task by ID func (r *Repository) DeleteTask(ctx context.Context, id string) error { result, err := r.db.ExecContext(ctx, r.db.Rebind(`DELETE FROM tasks WHERE id = ?`), id) diff --git a/apps/backend/internal/task/service/service_tasks.go b/apps/backend/internal/task/service/service_tasks.go index 398cf0ce48..2c1ac3d191 100644 --- a/apps/backend/internal/task/service/service_tasks.go +++ b/apps/backend/internal/task/service/service_tasks.go @@ -531,10 +531,9 @@ func (s *Service) DeleteTask(ctx context.Context, id string) error { zap.String("task_id", id), zap.Duration("duration", time.Since(start))) - // 6. Return immediately - all remaining cleanup is async - - // 7. Background: Stop agents and cleanup worktrees - if len(stopTargets) > 0 || s.worktreeCleanup != nil || len(sessions) > 0 || task.IsEphemeral { + // 6. Stop agents and cleanup worktrees in the background. + hasCleanup := len(stopTargets) > 0 || s.worktreeCleanup != nil || len(sessions) > 0 || task.IsEphemeral + if hasCleanup { s.runAsyncTaskCleanup(id, sessions, worktrees, stopTargets, task.IsEphemeral, "task deleted", "failed to stop session on task delete", "task cleanup completed") } diff --git a/apps/web/e2e/fixtures/backend.ts b/apps/web/e2e/fixtures/backend.ts index 7b98a7c7da..f61e8234e7 100644 --- a/apps/web/e2e/fixtures/backend.ts +++ b/apps/web/e2e/fixtures/backend.ts @@ -191,8 +191,16 @@ export const backendFixture = base.extend({ // Give each worker its own agentctl port range, offset from the default // range (41001-41100) to avoid conflicts with a running dev instance. - const agentctlPortBase = 30001 + E2E_PORT_OFFSET * 1000 + workerInfo.workerIndex * 50; - const agentctlPortMax = agentctlPortBase + 49; + // The async cleanup of agent instances runs after each test deletes its + // tasks, so during a 60+ test shard the in-flight cleanup queue can hold + // several dozen ports at any given moment. 200 ports per worker keeps + // headroom for that without overflowing the 65535 port space. With the + // current playwright config (workers: 1, so workerIndex == 0) and shard + // offsets capped at 29, the highest port used is 30001 + 29*1000 + 199 + // = 59200. The `workerIndex * 200` term is defensive for the case where + // a future config sets workers > 1. + const agentctlPortBase = 30001 + E2E_PORT_OFFSET * 1000 + workerInfo.workerIndex * 200; + const agentctlPortMax = agentctlPortBase + 199; // Install a `git` shim that can sleep on `fetch`/`pull` before execing // the real git binary. Tests that need to simulate slow network git