From 38a331633638e350867f52a9a3beee72921073b9 Mon Sep 17 00:00:00 2001 From: Max Date: Tue, 24 Mar 2026 22:30:37 +0800 Subject: [PATCH] feat(robot): enhance execution management and slot acquisition - Implemented pre-acquisition of execution slots in the Tick method to prevent race conditions, ensuring that robots do not submit duplicate executions. - Updated TryAcquireSlot method to support idempotent behavior, allowing for early slot reservation without consuming additional resources. - Modified worker execution logic to skip pre-checks for robots that have already acquired a slot, streamlining the execution process. - Improved error handling during execution submission to ensure proper tracking and removal of failed executions. --- agent/robot/manager/manager.go | 25 +++++++++++++++++++------ agent/robot/pool/worker.go | 8 ++++---- agent/robot/types/robot.go | 22 +++++++++++++++------- 3 files changed, 38 insertions(+), 17 deletions(-) diff --git a/agent/robot/manager/manager.go b/agent/robot/manager/manager.go index 79147c9b..1725c12b 100644 --- a/agent/robot/manager/manager.go +++ b/agent/robot/manager/manager.go @@ -274,13 +274,28 @@ func (m *Manager) Tick(parentCtx context.Context, now time.Time) error { // continue // } - // Pre-generate execution ID and track for pause/resume/stop - // We need to track BEFORE submit so we can pass the cancellable context to the executor + // Pre-generate execution ID execID := pool.GenerateExecID() + + // Pre-acquire execution slot to prevent daemon-mode race condition: + // Without this, CanRun() stays true between Tick and worker dequeue, + // causing duplicate submissions on every tick interval. + preExec := &types.Execution{ + ID: execID, + MemberID: robot.MemberID, + TeamID: robot.TeamID, + TriggerType: types.TriggerClock, + Status: types.ExecPending, + StartTime: now, + } + if !robot.TryAcquireSlot(preExec) { + continue + } + + // Track for pause/resume/stop — after slot is acquired ctrlExec := m.execController.Track(execID, robot.MemberID, robot.TeamID) // Create context with robot's own identity and cancellable context - // Clock-triggered executions run as the robot itself robotAuth := m.buildRobotAuth(robot) execCtx := types.NewContext(ctrlExec.Context(), robotAuth) @@ -290,10 +305,8 @@ func (m *Manager) Tick(parentCtx context.Context, now time.Time) error { // Submit to pool with the cancellable context and execution control _, err := m.pool.SubmitWithID(execCtx, robot, types.TriggerClock, clockCtx, execID, ctrlExec) if err != nil { - // If submission failed, untrack the execution + robot.RemoveExecution(execID) m.execController.Untrack(execID) - // Log error but continue with other robots - // In production, this would be logged properly continue } diff --git a/agent/robot/pool/worker.go b/agent/robot/pool/worker.go index 69b7246c..def0ab7b 100644 --- a/agent/robot/pool/worker.go +++ b/agent/robot/pool/worker.go @@ -66,10 +66,10 @@ func (w *Worker) run() { // execute processes a single queue item func (w *Worker) execute(item *QueueItem) { - // Pre-check if robot can run (non-atomic, just for early rejection) - // The actual atomic check happens inside Executor.Execute() via TryAcquireSlot() - if !item.Robot.CanRun() { - // Robot likely at quota, re-enqueue for later + // Pre-check if robot can run (non-atomic, just for early rejection). + // Skip for jobs whose slot was pre-acquired by Tick — they already hold + // a reserved slot and will pass TryAcquireSlot idempotently. + if item.Robot.GetExecution(item.ExecID) == nil && !item.Robot.CanRun() { w.requeue(item, "quota pre-check failed") return } diff --git a/agent/robot/types/robot.go b/agent/robot/types/robot.go index c9eb87c7..2ba90f06 100644 --- a/agent/robot/types/robot.go +++ b/agent/robot/types/robot.go @@ -52,25 +52,33 @@ func (r *Robot) CanRun() bool { return len(r.executions) < r.Config.Quota.GetMax() } -// TryAcquireSlot atomically checks if robot can run and reserves a slot -// Returns true if slot was acquired, false if quota is full -// This prevents race conditions between CanRun() check and AddExecution() +// TryAcquireSlot atomically checks if robot can run and reserves a slot. +// Returns true if slot was acquired, false if quota is full. +// Idempotent: if exec.ID already exists in tracking, the entry is updated +// and true is returned without consuming an additional slot. This supports +// the Tick pre-acquisition pattern where the slot is reserved early and +// later confirmed by the executor with a richer Execution object. func (r *Robot) TryAcquireSlot(exec *Execution) bool { r.execMu.Lock() defer r.execMu.Unlock() - // Get max quota + // Idempotent: same ID already tracked — update in place + if r.executions != nil { + if _, exists := r.executions[exec.ID]; exists { + r.executions[exec.ID] = exec + return true + } + } + maxQuota := 2 // default if r.Config != nil { maxQuota = r.Config.Quota.GetMax() } - // Check if we can add if len(r.executions) >= maxQuota { - return false // quota full + return false } - // Reserve slot by adding execution if r.executions == nil { r.executions = make(map[string]*Execution) }