agent-substrate / agent-substrate/substrate
[P2] Stale-worker release in AssignWorkerStep retry path is fire-and-forget — failure silently leaks worker slot
- Lingua principale
- Go
- Stelle
- 1.8k
- Fork
- 316
- Merge medio
- 2g 43m
- PR unite (30g)
- 287
Descrizione
**Severity:** P2 (worker slot consumed; pool capacity reduced without alert)
**Component:** Control Plane — `cmd/ateapi/internal/controlapi/workflow_resume.go`
**Confirmed:** Yes — code-level analysis
---
## Summary
When `AssignWorkerStep` detects a stale worker assignment from a previous failed resume
attempt, it spawns a background goroutine to release the claim. This goroutine has a
10-second timeout and no retry. If `UpdateWorker` fails (Valkey blip, version conflict),
the goroutine logs an error and exits. The stale worker assignment persists until the
pod is deleted. The pool appears to have one fewer free worker than it actually does,
and no metric or alert is emitted.
---
## Impact
- Repeated resume failures with Valkey transient errors can drain the worker pool
over time without any visible signal.
- The metric `ate.workerpool.workers` (if instrumented) may show a discrepancy between
free workers and actual pod count.
- Operators have no way to know that stale claims exist without inspecting Valkey directly.
---
## Root Cause
**File:** `cmd/ateapi/internal/controlapi/workflow_resume.go` lines 184–192
```go
go func(release *ateapipb.Worker) {
bgCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := s.store.UpdateWorker(bgCtx, release, release.Version); err != nil {
slog.ErrorContext(bgCtx, "Failed to release stale worker assignment",
slog.String("worker", release.GetWorkerNamespace()+"/"+release.GetWorkerPod()),
slog.Any("err", err))
// NO RETRY — stale assignment persists
}
}(releaseWorker)
```
This fires when `AssignWorkerStep` finds a worker already assigned to this actor
(from a previous failed resume attempt where UpdateWorker succeeded but UpdateActor
failed). The release is a best-effort background goroutine with no retry and no metric.
If the goroutine fails, `releaseWorker` still holds `Assignment = `. The
scheduler filters out workers with `Assignment != nil`, so this worker is permanently
removed from the available pool until the pod dies or ateapi restarts and
`reconcileOrphanedWorkers` picks it up.
---
## Steps to Reproduce
1. Set up a Valkey proxy (e.g., toxiproxy) between ateapi and Valkey.
2. Resume an actor, let it fail partway — specifically after `UpdateWorker` succeeds
(worker assigned) but before `UpdateActor` completes (simulate via ateapi crash
or version conflict injection). The actor stays SUSPENDED, worker has stale claim.
3. Retry `ResumeActor` — `AssignWorkerStep` detects the stale claim (line 162–174).
4. During the retry, inject a Valkey write error for 15 seconds (longer than the 10s goroutine timeout).
5. The stale-release goroutine fires at line 184, times out, and exits.
6. Inspect Valkey:
```bash
kubectl exec -n ate-system valkey-cluster-0 -- \
valkey-cli SCAN 0 MATCH "worker:*" COUNT 1000 | xargs -I{} valkey-cli HGET {} assignment
# Worker still has assignment pointing to the SUSPENDED actor
```
7. The actor successfully resumes (a different worker was chosen), but the first
worker remains claimed.
---
## Expected Behavior
The stale-worker release should be retried with exponential backoff, or the stale
worker UID should be enqueued in a work queue for reliable background cleanup.
A metric should track the number of stale releases attempted and failed.
---
## Suggested Fix
Replace the fire-and-forget goroutine with a retry loop:
```go
go func(release *ateapipb.Worker) {
backoff := wait.Backoff{
Steps: 5,
Duration: 1 * time.Second,
Factor: 2.0,
Cap: 30 * time.Second,
}
err := wait.ExponentialBackoff(backoff, func() (bool, error) {
bgCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
err := s.store.UpdateWorker(bgCtx, release, release.Version)
if err == nil {
return true, nil
}
if errors.Is(err, store.ErrVersionConflict) {
return true, nil // another writer cleared it — success
}
slog.WarnContext(bgCtx, "Retrying stale worker release", "err", err)
return false, nil
})
if err != nil {
slog.ErrorContext(context.Background(), "Permanently failed to release stale worker — manual recovery needed",
"worker", release.GetWorkerNamespace()+"/"+release.GetWorkerPod())
// Emit a metric: stale_worker_release_failures_total
}
}(releaseWorker)
```
Also add a counter metric `ate.workerpool.stale_worker_releases` to make this
visible to operators.
Guida per i contributori
Apri la guida per i contributori
Valutazione
Questa issue non è ancora stata valutata.