mirror of
https://gh-proxy.org/https://github.com/klinsmaya/multica.git
synced 2026-11-09 01:46:46 +08:00
e5f976144b
Source-context cleanup (MUL-6555) ran at the top of the 30s runtime sweep tick, ahead of the stages that flip a dead runtime offline and reclaim its orphaned tasks. Both cleanup stages call the object store, so a stalled or throttled endpoint delayed runtime/task recovery: one round's work was bounded by the number of due intents rather than the batch size, because the loop counted successes, and the abandoned-capture path deleted objects on the sweeper's root context with no per-object deadline. Move both stages onto their own goroutine and ticker, bound each round with a wall-clock budget, bound the intent loop by attempts, and give every object delete a deadline. Failed deletes now record their retry backoff on a context detached from the round budget, so a round cut short cannot leave an intent claimable with no backoff and starve the rows behind it. Cleanup cadence moves from 30s to 60s; nothing here is latency-critical (intents settle for an hour, captures for 30 days). Co-authored-by: J <[email protected]> Co-authored-by: multica-agent <[email protected]>
73 lines
2.9 KiB
Go
73 lines
2.9 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"github.com/multica-ai/multica/server/internal/service"
|
|
)
|
|
|
|
const (
|
|
// sourceContextSweepInterval paces snapshot cleanup (MUL-6555). Nothing
|
|
// here is latency-critical: an orphaned object intent only becomes eligible
|
|
// one hour after its capture failed and carries a five-minute retry
|
|
// backoff, and an abandoned capture waits out a 30-day retention window.
|
|
// A minute keeps the drain rate for a fresh backlog close to what the old
|
|
// placement in the 30s runtime tick gave, at two cheap queries per round.
|
|
sourceContextSweepInterval = time.Minute
|
|
// sourceContextSweepBudget bounds one round below the interval so a slow
|
|
// object store delays cleanup instead of stacking rounds. Per-object
|
|
// deadlines live in the service; this is the whole-round backstop.
|
|
sourceContextSweepBudget = 45 * time.Second
|
|
// sourceContextCleanupBatchSize bounds delete attempts per round.
|
|
sourceContextCleanupBatchSize = 50
|
|
)
|
|
|
|
// sourceContextCleaner is the slice of TaskService this sweeper drives, kept as
|
|
// an interface so the round can be tested against a stalled object store.
|
|
type sourceContextCleaner interface {
|
|
CleanupSourceContextObjectIntents(ctx context.Context, limit int) (int, error)
|
|
CleanupAbandonedSourceContexts(ctx context.Context, limit int32) (int, error)
|
|
}
|
|
|
|
var _ sourceContextCleaner = (*service.TaskService)(nil)
|
|
|
|
// runSourceContextSweeper reclaims snapshot objects whose capture never
|
|
// committed and captures whose retry window has expired.
|
|
//
|
|
// It deliberately runs on its own goroutine rather than inside the runtime
|
|
// sweep tick. Both stages talk to the object store, and a stalled or throttled
|
|
// endpoint there must never delay marking runtimes offline, reclaiming
|
|
// orphaned tasks, or replaying delegated failure recoveries — those are the
|
|
// stages that keep the board truthful within ~180s of a daemon dying.
|
|
func runSourceContextSweeper(ctx context.Context, cleaner sourceContextCleaner) {
|
|
ticker := time.NewTicker(sourceContextSweepInterval)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
sweepSourceContextsWithBudget(ctx, cleaner, sourceContextSweepBudget)
|
|
}
|
|
}
|
|
}
|
|
|
|
func sweepSourceContextsWithBudget(ctx context.Context, cleaner sourceContextCleaner, budget time.Duration) {
|
|
sweepCtx, cancel := context.WithTimeout(ctx, budget)
|
|
defer cancel()
|
|
|
|
if cleaned, err := cleaner.CleanupSourceContextObjectIntents(sweepCtx, sourceContextCleanupBatchSize); err != nil {
|
|
slog.Warn("source context object intent cleanup failed", "error", err)
|
|
} else if cleaned > 0 {
|
|
slog.Info("source context object intent cleanup completed", "count", cleaned)
|
|
}
|
|
if cleaned, err := cleaner.CleanupAbandonedSourceContexts(sweepCtx, sourceContextCleanupBatchSize); err != nil {
|
|
slog.Warn("source context cleanup sweeper failed", "error", err)
|
|
} else if cleaned > 0 {
|
|
slog.Info("source context cleanup sweeper removed abandoned captures", "count", cleaned)
|
|
}
|
|
}
|