fix(runner): 稳定异步任务重启恢复
ci / verify (pull_request) Successful in 10m32s

避免恢复任务将已到期的 next_run_at 误插入 River scheduled 状态,并使用专用模型隔离重启集成测试。\n\n验证:Go 全量测试、govulncheck、pnpm lint/test/build/audit、Compose、镜像、迁移、流水线及 SemVer 门禁全部通过。
This commit is contained in:
2026-07-21 11:29:56 +08:00
parent 1e55f7df8b
commit 8c38714296
3 changed files with 50 additions and 4 deletions
+5 -3
View File
@@ -156,7 +156,7 @@ func (s *Service) recoverAsyncRiverJobs(ctx context.Context) error {
return err
}
for _, item := range items {
result, err := s.riverClient.Insert(ctx, asyncTaskArgs{TaskID: item.ID}, asyncTaskRecoveryInsertOpts(item))
result, err := s.riverClient.Insert(ctx, asyncTaskArgs{TaskID: item.ID}, asyncTaskRecoveryInsertOpts(item, time.Now()))
if err != nil {
return err
}
@@ -196,9 +196,11 @@ func asyncTaskInsertOpts(task store.GatewayTask) *river.InsertOpts {
}
}
func asyncTaskRecoveryInsertOpts(item store.AsyncTaskQueueItem) *river.InsertOpts {
func asyncTaskRecoveryInsertOpts(item store.AsyncTaskQueueItem, now time.Time) *river.InsertOpts {
opts := asyncTaskInsertOpts(store.GatewayTask{ID: item.ID})
opts.ScheduledAt = item.NextRunAt
if item.NextRunAt.After(now) {
opts.ScheduledAt = item.NextRunAt
}
// A replacement process must not be blocked by a River row that the dead
// process left in running state. PostgreSQL execution leases still ensure
// that only one recovery job can call the upstream provider.
@@ -0,0 +1,33 @@
package runner
import (
"testing"
"time"
"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
)
func TestAsyncTaskRecoveryInsertOptsMakesDueTaskImmediatelyAvailable(t *testing.T) {
now := time.Date(2026, time.July, 21, 3, 30, 0, 0, time.UTC)
opts := asyncTaskRecoveryInsertOpts(store.AsyncTaskQueueItem{
ID: "due-task",
NextRunAt: now.Add(-time.Second),
}, now)
if !opts.ScheduledAt.IsZero() {
t.Fatalf("due recovery task should be immediately available, scheduled at %s", opts.ScheduledAt)
}
}
func TestAsyncTaskRecoveryInsertOptsPreservesFutureDelay(t *testing.T) {
now := time.Date(2026, time.July, 21, 3, 30, 0, 0, time.UTC)
nextRunAt := now.Add(30 * time.Second)
opts := asyncTaskRecoveryInsertOpts(store.AsyncTaskQueueItem{
ID: "delayed-task",
NextRunAt: nextRunAt,
}, now)
if !opts.ScheduledAt.Equal(nextRunAt) {
t.Fatalf("future recovery task should stay delayed until %s, got %s", nextRunAt, opts.ScheduledAt)
}
}