diff --git a/workflow/service/flow/redis_lock.go b/workflow/service/flow/redis_lock.go new file mode 100644 index 0000000..33686fd --- /dev/null +++ b/workflow/service/flow/redis_lock.go @@ -0,0 +1,43 @@ +package flow + +import ( + "context" + + "github.com/gogf/gf/v2/database/gredis" + "github.com/gogf/gf/v2/frame/g" + "github.com/google/uuid" +) + +// RedisLock 基于 SETNX 的分布式锁:token 标识持有者身份,Lua 原子释放防误删他人锁。 +// 用途:恢复例程抢锁防多节点对同一 exec 重复恢复(spec §6) +type RedisLock struct { + key string + token string + ttl int64 // 秒 +} + +func NewRedisLock(key string, ttlSec int64) *RedisLock { + return &RedisLock{key: key, token: uuid.NewString(), ttl: ttlSec} +} + +// Acquire 抢锁;返回 true 表示抢到(SET NX 成功)。 +// 注意:gogf v2.10.2 的 SetNX(ctx,key,value) 不接收 TTL 参数,直接 SETNX 会永不过期(崩溃后死锁); +// 故改用 Set + SetOption{NX,TTLOption{EX}},原子地执行 `SET key value EX NX`。 +func (l *RedisLock) Acquire(ctx context.Context) (bool, error) { + r, err := g.Redis().Set(ctx, l.key, l.token, gredis.SetOption{ + TTLOption: gredis.TTLOption{EX: &l.ttl}, + NX: true, + }) + if err != nil { + return false, err + } + // SET NX 失败时 Redis 返回 nil(空回复),gogf 会转换为值 nil 的 gvar;成功时返回 "OK" + return !r.IsNil(), nil +} + +// Release 释放锁:仅当 key 值仍为本锁 token 时删除(Lua 原子),防止超时后误删他人已续的锁 +func (l *RedisLock) Release(ctx context.Context) error { + const luaRelease = `if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end` + _, err := g.Redis().Eval(ctx, luaRelease, 1, []string{l.key}, []any{l.token}) + return err +} diff --git a/workflow/service/flow/redis_lock_test.go b/workflow/service/flow/redis_lock_test.go new file mode 100644 index 0000000..9fc2534 --- /dev/null +++ b/workflow/service/flow/redis_lock_test.go @@ -0,0 +1,52 @@ +package flow + +import ( + "context" + "os" + "testing" + + "github.com/gogf/gf/v2/frame/g" +) + +// TestRedisLockMutualExclusion 验证互斥与防误删(需 AI_AGENT_TEST_REDIS=1 且本地 Redis 可用) +func TestRedisLockMutualExclusion(t *testing.T) { + if os.Getenv("AI_AGENT_TEST_REDIS") != "1" { + t.Skip("skip: AI_AGENT_TEST_REDIS not set") + } + ctx := context.Background() + key := "workflow:exec:recover:test-lock" + // 清理残留 + g.Redis().Del(ctx, key) + + l1 := NewRedisLock(key, 60) + l2 := NewRedisLock(key, 60) + ok1, err := l1.Acquire(ctx) + if err != nil || !ok1 { + t.Fatalf("l1 acquire should succeed: ok=%v err=%v", ok1, err) + } + ok2, err := l2.Acquire(ctx) + if err != nil { + t.Fatalf("l2 acquire err: %v", err) + } + if ok2 { + t.Fatalf("l2 不应抢到已被 l1 持有的锁") + } + // l2 尝试释放 l1 的锁:token 不匹配,不应删除 + if err := l2.Release(ctx); err != nil { + t.Fatalf("l2 release err: %v", err) + } + ok2, _ = l2.Acquire(ctx) + if ok2 { + t.Fatalf("l2 释放他人锁后不应能抢到(l1 锁应仍在)") + } + // l1 释放后 l2 可抢到 + if err := l1.Release(ctx); err != nil { + t.Fatalf("l1 release err: %v", err) + } + ok2, err = l2.Acquire(ctx) + if err != nil || !ok2 { + t.Fatalf("l1 释放后 l2 应可抢到: ok=%v err=%v", ok2, err) + } + l2.Release(ctx) + g.Redis().Del(ctx, key) +}