fix(e2e): release agent ports during E2E reset to prevent shard exhaustion (#888)

This commit is contained in:
Carlos Florêncio
2026-05-12 22:41:21 +01:00
committed by GitHub
parent c72080e133
commit d48ee80472
4 changed files with 45 additions and 21 deletions
+32 -4
View File
@@ -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 {
@@ -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)
@@ -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")
}
+10 -2
View File
@@ -191,8 +191,16 @@ export const backendFixture = base.extend<object, { backend: BackendContext }>({
// 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