fix(runtime): 运行稳定性 - 修复技能种子、任务续租与回归检查

保留社区技能作者与时间戳修复、Agent 参数约束、资源删除 worker 生命周期、PostgreSQL/Redis 测试门禁及相关回归。

Original-Commit: 0d05ff88bbe6c875d01964041788023a662ff7e8
This commit is contained in:
ddCat
2026-09-21 09:48:43 +08:00
parent 6a3e454141
commit 7f44332914
21 changed files with 644 additions and 354 deletions

View File

@@ -165,7 +165,7 @@ func (s *Service) DeleteAdminResources(actor *model.User, req AdminResourceDelet
return nil, err
}
if len(deletionJobs) > 0 {
go s.drainResourceDeletionJobs(len(deletionJobs))
s.runWorkerTask(func() { s.drainResourceDeletionJobs(len(deletionJobs)) })
}
result := &AdminResourceDeleteResult{Deleted: []string{}, Blocked: []AdminResourceDeleteBlocked{}}

View File

@@ -170,7 +170,9 @@ func newAdminStorageDeleteTestService(t *testing.T) (*Service, *gorm.DB, string,
t.Fatal(err)
}
dataDir := t.TempDir()
return New(repository.New(db), dataDir), db, dataDir, admin
svc := New(repository.New(db), dataDir)
startDeletionTestWorkers(t, svc)
return svc, db, dataDir, admin
}
func assertModelCount(t *testing.T, db *gorm.DB, value any, query string, expected int64, args ...any) {

View File

@@ -2,7 +2,7 @@ package app
import (
"encoding/json"
"strings"
"reflect"
"testing"
)
@@ -24,31 +24,37 @@ func TestCloudAgentToolSchemaStaysCompact(t *testing.T) {
tools, raw := platformToolSchema(t)
t.Logf("platform tool schema: %d tools, %d bytes", len(tools), len(raw))
// 顶层 type 枚举 + 服务端按 action 强校验已经表达了"每种操作要哪些字段",
// 再挂一份 oneOf 只是重复:同一个约束会被下发两遍。
// 条件必填不是 type 枚举能表达的约束,压缩描述不能删掉协议语义。
found := false
for _, tool := range tools {
function, _ := tool["function"].(map[string]any)
if function["name"] != "canvas_apply_ops" {
continue
}
found = true
parameters, _ := function["parameters"].(map[string]any)
ops, _ := parameters["properties"].(map[string]any)
items, _ := ops["ops"].(map[string]any)
opItem, _ := items["items"].(map[string]any)
if _, exists := opItem["oneOf"]; exists {
t.Fatal("canvas_apply_ops 的 ops.items 不该再带重复的 oneOf")
want := []map[string]any{
{"properties": map[string]any{"type": map[string]any{"const": "add_node"}}, "required": []string{"nodeType"}},
{"properties": map[string]any{"type": map[string]any{"const": "update_node"}}, "required": []string{"patch"}},
{"properties": map[string]any{"type": map[string]any{"const": "connect_nodes"}}, "required": []string{"fromNodeId", "toNodeId"}},
}
if !reflect.DeepEqual(opItem["oneOf"], want) {
t.Fatalf("操作类型条件必填约束丢失: %#v", opItem["oneOf"])
}
if props, ok := opItem["properties"].(map[string]any); !ok || len(props) == 0 {
t.Fatal("canvas_apply_ops 的 ops.items 必须保留 properties")
}
}
if !found {
t.Fatal("canvas_apply_ops 未暴露")
}
// 体积预算:当前 20 个工具约 22.4 KB,这里留 ~7% 余量;新增工具或字段时请重新测量并
// 有意识地调整这个数字,而不是让 schema 悄悄膨胀(它每一步都要发、还在前缀最前面)。
if len(raw) > 24000 {
t.Fatalf("平台工具 schema 体积 %d 字节超出预算 24000:请压缩描述或显式调整预算", len(raw))
}
if strings.Count(string(raw), "oneOf") > 2 {
t.Fatalf("工具 schema 里出现了过多 oneOf:%s", string(raw[:200]))
}
}

View File

@@ -251,6 +251,11 @@ func compileCloudAgentTools(req CloudAgentRequest, includeProfileTool bool) []ma
"properties": opProperties,
"required": []string{"type", "id"},
"additionalProperties": false,
"oneOf": []map[string]any{
{"properties": map[string]any{"type": map[string]any{"const": "add_node"}}, "required": []string{"nodeType"}},
{"properties": map[string]any{"type": map[string]any{"const": "update_node"}}, "required": []string{"patch"}},
{"properties": map[string]any{"type": map[string]any{"const": "connect_nodes"}}, "required": []string{"fromNodeId", "toNodeId"}},
},
}
add("canvas_apply_ops", "创建空白节点、修改提示词或建立引用连线,不提交生成任务、不产生生成费用;先读取画布并传 snapshotHash。提交媒体生成使用 generate_media。每次最多20项,禁止删除、任意 metadata 和媒体 URL。每项都需要 type 和 id:add_node 还需要 nodeType,update_node 还需要按节点能力清单填写 patch,connect_nodes 还需要 fromNodeId 与 toNodeId。连线是生成输入关系,不会改变已提交任务的输入;来源须 canSource,目标须 canTarget 且接受来源 inputKind,能力以注册表为准。", map[string]any{"snapshotHash": str("canvas_get_state返回的snapshotHash"), "ops": map[string]any{"type": "array", "maxItems": 20, "items": opItem}}, "snapshotHash", "ops")
}

View File

@@ -146,7 +146,8 @@ func (s *Service) deleteUserAssetWithResources(userID string, assetID string) er
return fmt.Errorf("素材记录删除失败,请重试:%w", err)
}
if len(deletionJobs) > 0 {
go s.drainResourceDeletionJobs(len(deletionJobs))
// 纳入停机等待;排空期间拒绝新 worker 时,已提交的 outbox 留待下次启动处理。
s.runWorkerTask(func() { s.drainResourceDeletionJobs(len(deletionJobs)) })
}
return nil
}

View File

@@ -513,5 +513,7 @@ func newResourceDeletionTestService(t *testing.T) (*Service, *gorm.DB, string) {
t.Fatal(err)
}
dataDir := t.TempDir()
return New(repository.New(db), dataDir), db, dataDir
svc := New(repository.New(db), dataDir)
startDeletionTestWorkers(t, svc)
return svc, db, dataDir
}

View File

@@ -126,7 +126,7 @@ func (s *Service) cleanupDetachedUserResources(userID string, candidates []model
}
log.Printf("detached resource cleanup: removed %d resource rows for user %s", len(detached), userID)
if len(deletionJobs) > 0 {
go s.drainResourceDeletionJobs(len(deletionJobs))
s.runWorkerTask(func() { s.drainResourceDeletionJobs(len(deletionJobs)) })
}
return nil
}

View File

@@ -227,16 +227,11 @@ func (w *taskWorkerCoordinator) processClaimedTask(task *model.Task, globalSlot
if code, _ := ChannelSlotFailureDetails(err); code != "" {
channelSlotFailedBeforeRequest = true
}
// 我方执行时限到点不是"租约丢失":续租请求可能正好被同一个时限打断,
// 此时必须继续走失败收尾,否则任务停在 running、租约过期后被反复重跑。
deadlineExpired := errors.Is(ctx.Err(), context.DeadlineExceeded)
// 续租使用独立 context;即使执行同时超时,真实租约失效仍必须阻止写入。
select {
case leaseErr := <-leaseLost:
if !deadlineExpired {
_ = s.log(task.UserID, task.ID, "warn", "任务租约失效,等待其他 worker 恢复", leaseErr.Error())
return leaseErr
}
_ = s.log(task.UserID, task.ID, "warn", "任务执行时限到点,按失败收尾(不视为租约失效)", leaseErr.Error())
_ = s.log(task.UserID, task.ID, "warn", "任务租约失效,等待其他 worker 恢复", leaseErr.Error())
return leaseErr
default:
}
decryptedInput, decryptErr := s.decryptTaskInputJSON(task.InputJSON)

View File

@@ -2,8 +2,12 @@ package app
import (
"context"
"errors"
"testing"
"time"
"infinite-canvas/backend/internal/model"
"infinite-canvas/backend/internal/repository"
)
// 续租必须比"这条任务的执行时限"活得更久:否则父 context 到点时,续租请求会被自己的
@@ -39,3 +43,66 @@ func TestTaskLeaseRenewContextSurvivesExecutionDeadline(t *testing.T) {
t.Fatalf("父 context 取消不该连带取消续租 context:%v", err)
}
}
func TestTaskDeadlineRenewalAndTerminalFencing(t *testing.T) {
for _, lost := range []bool{false, true} {
name := "deadline_with_valid_lease"
if lost {
name = "deadline_with_reclaimed_lease"
}
t.Run(name, func(t *testing.T) {
db := newSQLiteTestDB(t)
if err := db.AutoMigrate(&model.Task{}); err != nil {
t.Fatal(err)
}
repo := repository.New(db)
if err := db.Create(&model.Task{ID: "deadline-task", UserID: "user", Type: "canvas_text", Status: model.TaskStatusQueued}).Error; err != nil {
t.Fatal(err)
}
task, err := repo.ClaimNextTask("worker-a", time.Minute)
if err != nil || task == nil {
t.Fatalf("claim: %v %v", task, err)
}
ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
defer cancel()
renew, cancelRenew := taskLeaseRenewContext(ctx)
defer cancelRenew()
if lost {
if err := db.Model(&model.Task{}).Where("id = ?", task.ID).Update("lease_expires_at", time.Now().Add(-time.Second)).Error; err != nil {
t.Fatal(err)
}
reclaimed, err := repo.ClaimNextTask("worker-b", time.Minute)
if err != nil || reclaimed == nil {
t.Fatalf("reclaim: %v %v", reclaimed, err)
}
}
err = repo.WithContext(renew).RenewTaskLease(task.ID, task.LeaseOwner, time.Minute)
if (err != nil) != lost {
t.Fatalf("renew lost=%v: %v", lost, err)
}
billing, replay := &taskTerminalBillingStub{}, &taskTerminalReplayStub{}
terminal := newTaskTerminalCoordinatorForTest(repo, billing, replay, &taskTerminalLoggerStub{}, &taskTerminalOutputStub{})
err = terminal.handleExecutionFailure(task, ctx.Err(), false, false)
stored, readErr := repo.Task(task.ID)
if readErr != nil {
t.Fatal(readErr)
}
if lost {
if !errors.Is(err, repository.ErrTaskStateConflict) || stored.Status != model.TaskStatusRunning || stored.LeaseOwner != "worker-b" {
t.Fatalf("stale terminal write: %v %#v", err, stored)
}
if len(billing.refund)+len(billing.uncertain)+len(replay.statuses) != 0 {
t.Fatal("stale worker emitted terminal side effects")
}
} else {
if !errors.Is(err, context.DeadlineExceeded) || stored.Status != model.TaskStatusFailed {
t.Fatalf("deadline not finalized: %v %#v", err, stored)
}
next, err := repo.ClaimNextTask("worker-b", time.Minute)
if err != nil || next != nil {
t.Fatalf("terminal task reclaimed: %v %v", next, err)
}
}
})
}
}

View File

@@ -1,8 +1,8 @@
package app
import (
"context"
"fmt"
"os"
"path/filepath"
"strings"
"sync"
@@ -21,27 +21,43 @@ import (
// 而当前驱动没有启用。实测并发读写 400 次能撞到 70+ 次,CI 上因此偶发失败
// (TestDeleteGeneratedAssetTaskReferences/cancelled_output 就是这样挂的)。
//
// 因此凡是会触发后台删除 outbox(`go s.drainResourceDeletionJobs`)的用例,统一用
// 因此凡是会触发后台删除 outbox 的用例,统一用
// 文件库 + WAL + busy_timeout:WAL 允许"一写多读"并发,写冲突退化成 SQLITE_BUSY 并由
// busy_timeout 重试。
//
// 库文件刻意放在 `os.MkdirTemp` 而不是 `t.TempDir()` 下:outbox goroutine 可能在用例返回后
// 仍在写库,`t.TempDir()` 的清理遇到这种写入会以 "directory not empty" 直接判失败,
// 而它本身与用例断言无关。
// 清理顺序为等待 service worker、关闭 SQL 连接、删除临时目录,不忽略清理错误。
func newSQLiteTestDB(t *testing.T) *gorm.DB {
t.Helper()
directory, err := os.MkdirTemp("", "canvas-test-db-")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = os.RemoveAll(directory) })
directory := t.TempDir()
db, err := gorm.Open(sqlite.Open(filepath.Join(directory, "test.db")+"?_journal_mode=WAL&_busy_timeout=5000"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
sqlDB, err := db.DB()
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
if err := sqlDB.Close(); err != nil {
t.Errorf("close test database: %v", err)
}
})
return db
}
func startDeletionTestWorkers(t *testing.T, svc *Service) {
t.Helper()
// 只启用生命周期跟踪,不启动生成、支付等无关的周期 worker。
svc.backgroundWorkers().Start()
t.Cleanup(func() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
if err := svc.StopWorker(ctx); err != nil {
t.Errorf("stop deletion workers: %v", err)
}
})
}
// TestSQLiteTestDBAllowsConcurrentReadWrite 守的是 newSQLiteTestDB 的选型:并发读写不得报锁。
// 换回 `mode=memory&cache=shared` 会立刻变红(见上面的机制说明)。
func TestSQLiteTestDBAllowsConcurrentReadWrite(t *testing.T) {

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,86 @@
package skills
import (
"encoding/json"
"path/filepath"
"testing"
"time"
"infinite-canvas/backend/internal/model"
"infinite-canvas/backend/internal/repository"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
)
func TestBuiltinSeedSyncPreservesUserState(t *testing.T) {
db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "skills.db")), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
sqlDB, err := db.DB()
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
if err := sqlDB.Close(); err != nil {
t.Error(err)
}
})
if err := db.AutoMigrate(&model.Skill{}, &model.UserSkillState{}); err != nil {
t.Fatal(err)
}
svc := New(repository.New(db), t.TempDir(), nil)
var definitions []builtinSkillDefinition
if err := json.Unmarshal(builtinSkillsJSON, &definitions); err != nil {
t.Fatal(err)
}
community := 0
for _, def := range definitions {
if def.OwnerUID == "" || def.OwnerUID != def.EffectiveUser.UID {
t.Fatalf("invalid author identity: %s", def.SkillID)
}
for _, stamp := range []int64{def.CreateTime, def.UpdateTime} {
if time.UnixMilli(stamp).Year() < 2020 || time.UnixMilli(stamp).After(time.Now().Add(24*time.Hour)) {
t.Fatalf("invalid millisecond timestamp for %s: %d", def.SkillID, stamp)
}
}
if def.OwnerUID == "community-itswyatt-k" {
community++
}
}
if community != 72 {
t.Fatalf("community skills = %d, want 72", community)
}
if err := svc.EnsureBuiltinSkills(); err != nil {
t.Fatal(err)
}
state := model.UserSkillState{ID: "test-state", UserID: "test-user", SkillID: "16000000000001", Added: true, Liked: true}
if err := db.Create(&state).Error; err != nil {
t.Fatal(err)
}
if err := svc.EnsureBuiltinSkills(); err != nil {
t.Fatal(err)
}
var count int64
if err := db.Model(&model.Skill{}).Count(&count).Error; err != nil {
t.Fatal(err)
}
if want := int64(len(definitions) + len(builtinImageEditingSkillDefinitions())); count != want {
t.Fatalf("seed count = %d, want %d", count, want)
}
var saved model.UserSkillState
if err := db.First(&saved, "id = ?", state.ID).Error; err != nil {
t.Fatal(err)
}
if !saved.Added || !saved.Liked {
t.Fatalf("user state lost: %#v", saved)
}
var skill model.Skill
if err := db.First(&skill, "id = ?", state.SkillID).Error; err != nil {
t.Fatal(err)
}
if skill.OwnerID != "community-itswyatt-k" || skill.CreatedAt.Year() != 2026 {
t.Fatalf("invalid persisted seed: %#v", skill)
}
}