diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b54e8a..5204e89 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## v1.5.7 - 2026-08-27 +- **可靠队列**: 新增 `task/queue`，提供 `Add`、`Fetch`、`Finish`、`FetchCount` 与泛型 `Decode`；通过全局队列名和 `redis.task` 配置契约管理 pending 及消费者隔离的 processing。 +- **Redis 兼容**: 保持标准 `LMOVE` 原子移动语义，并通过不保留数据的空 List 初始化兼容 SugarDB 对目标 List 预先存在的要求。 +- **延迟任务**: 新增 `AddAfter` 与 `PromoteDelayed`，使用 Redis Sorted Set 按到期时间提升任务，避免消费者反复读取尚未到期的重试项。 +- **运行控制**: 新增包级 `Enable(name)`、`Disable(name)`，便于管理端先操作运行状态、成功后再持久化恢复状态。 +- **日志传导**: 调度器继承生命周期 Logger，为每次任务运行创建独立 Trace，并通过 `WithLogger`、`Logger` 在 Context 中传导；空轮询不额外产生启动日志。 +- **依赖更新**: 升级 `redis` 至 `v1.5.13`、`jsmod` 至 `v1.5.4`。 + ## v1.5.6 - 2026-07-18 - **依赖校验修复**: 更新 `go.sum` 为当前可验证的依赖 checksum。 diff --git a/README.md b/README.md index fc42d34..4576848 100644 --- a/README.md +++ b/README.md @@ -63,15 +63,53 @@ tk.Disable() // 挂起任务 tk.Enable() // 恢复任务 tk.Remove() // 彻底移除 +task.Disable("CleanLog") // 按名称挂起，未知任务返回 false +task.Enable("CleanLog") // 按名称恢复，未知任务返回 false + // 查询 tasks := task.List() ``` +### 4. 任务上下文日志 + +调度器从生命周期管理器接收父 Logger，并为每次实际执行创建独立 Trace。任务函数可直接从 Context 取得本次运行的 Logger： + +```go +task.Add("Report", "@every 1m", func(ctx context.Context) error { + logger := task.Logger(ctx) + logger.Info("report generated", "count", 12) + return nil +}) +``` + +空轮询任务可使用 `SuppressSuccessLog` 避免无效磁盘日志；失败、Panic 和未抑制的成功日志均沿用本次运行的 Trace。 + +### 5. Redis 可靠队列 + +`task/queue` 按全局队列名使用约定连接 `redis.task`，调用方无需持有 Redis Client 或 Context： + +```go +import "apigo.cc/go/task/queue" + +_ = queue.Add("translation", item) +items, _ := queue.Fetch("translation", "default", 50) +// processing 中存在未完成批次时，Fetch 忽略 n 并返回原批次。 +// 业务成功后一次性确认当前消费者的 processing。 +_ = queue.Finish("translation", "default") + +// 持久化延迟任务；到期后由 Fetch 提升到 pending。 +_ = queue.AddAfter("translation", item, 30*time.Second) +``` + +队列使用 `pending`、按消费者隔离的 `processing` 和 Sorted Set `delayed` 三条持久化通道。任务失败时不调用 `Finish`，下一次 `Fetch` 会恢复同一 processing 批次。 + +`Fetch` 使用标准 `LMOVE` 原子地把任务转入 processing。为兼容要求目标 List 预先存在的 SugarDB，首次移动前会用一个立即弹出的内部标记初始化空 List；该标记不会保留在 processing，也不会返回给调用方。标准 Redis 会在弹出最后一个元素后删除 key，随后的 `LMOVE` 会按标准语义创建目标 List。 + ## 🛡️ 健壮性与安全 * **Panic Recovery**: 框架内部自动捕获任务执行中的 Panic，并记录堆栈日志，确保调度引擎持续稳定。 * **Context 传播**: 任务内部应监听 `ctx.Done()` 以响应系统的停止信号。 -* **标准化日志**: 集成 `@go/log`，自动记录每个任务的开始、成功、失败（含耗时）以及 Panic 信息。 +* **标准化日志**: 集成 `@go/log`，为每次实际执行创建独立 Trace，并记录成功、失败（含耗时）以及 Panic 信息。 ## 🧪 验证状态 测试全部通过，性能达标。 diff --git a/TEST.md b/TEST.md index 6095b38..03f8fe3 100644 --- a/TEST.md +++ b/TEST.md @@ -1,5 +1,12 @@ # Test Report +## v1.5.7 验证 +- `go test -v ./...`：通过；在临时标准 Redis 8.8 实例上覆盖任务调度、运行控制和可靠队列行为，测试后已关闭实例。 +- `go test -bench=. ./...`：通过；`BenchmarkTaskRegistration-16` 为 `138.8 ns/op`，Task Get/List 分别为 `7.614 ns/op`、`67.71 ns/op`（Apple M3 Max）。 +- 队列测试验证 processing 恢复、批量 Fetch/Finish、compatibility marker 不残留，以及 Sorted Set 延迟任务只在到期后提升；同一组测试已在未修改的 SugarDB 实例上通过。 +- 现有调度测试覆盖对象级 Enable/Disable；包级 `Enable`/`Disable` 和运行级 Logger 传导已通过编译与 master 集成，尚缺独立单元测试。 +- `git diff --check`：通过。 + ## v1.5.6 验证 - 独立模块解析、全量测试与基准测试均通过。 diff --git a/context.go b/context.go new file mode 100644 index 0000000..b277c7f --- /dev/null +++ b/context.go @@ -0,0 +1,24 @@ +package task + +import ( + "context" + + "apigo.cc/go/log" +) + +type loggerContextKey struct{} + +// WithLogger attaches the logger for the current task run to a context. +func WithLogger(ctx context.Context, logger *log.Logger) context.Context { + return context.WithValue(ctx, loggerContextKey{}, logger) +} + +// Logger returns the logger attached to a task context, or the default logger. +func Logger(ctx context.Context) *log.Logger { + if ctx != nil { + if logger, ok := ctx.Value(loggerContextKey{}).(*log.Logger); ok && logger != nil { + return logger + } + } + return log.DefaultLogger +} diff --git a/go.mod b/go.mod index f7221ac..95e0d45 100644 --- a/go.mod +++ b/go.mod @@ -3,21 +3,24 @@ module apigo.cc/go/task go 1.25.0 require ( - apigo.cc/go/log v1.5.8 + apigo.cc/go/cast v1.5.5 + apigo.cc/go/id v1.5.7 + apigo.cc/go/log v1.5.11 + apigo.cc/go/redis v1.5.13 github.com/robfig/cron/v3 v3.0.1 ) -require apigo.cc/go/jsmod v1.5.3 // indirect +require apigo.cc/go/jsmod v1.5.4 // indirect require ( - apigo.cc/go/cast v1.5.3 // indirect - apigo.cc/go/config v1.5.3 // indirect - apigo.cc/go/encoding v1.5.4 // indirect - apigo.cc/go/file v1.5.5 // indirect - apigo.cc/go/id v1.5.4 // indirect - apigo.cc/go/rand v1.5.3 // indirect - apigo.cc/go/safe v1.5.2 // indirect - apigo.cc/go/shell v1.5.3 // indirect + apigo.cc/go/config v1.5.4 // indirect + apigo.cc/go/crypto v1.5.5 // indirect + apigo.cc/go/encoding v1.5.6 // indirect + apigo.cc/go/file v1.5.6 // indirect + apigo.cc/go/rand v1.5.4 // indirect + apigo.cc/go/safe v1.5.3 // indirect + apigo.cc/go/shell v1.5.6 // indirect + github.com/gomodule/redigo v1.9.3 // indirect golang.org/x/crypto v0.52.0 // indirect golang.org/x/sys v0.45.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index d0a91f5..820c3be 100644 --- a/go.sum +++ b/go.sum @@ -1,31 +1,43 @@ -apigo.cc/go/cast v1.5.3 h1:jk6VX0rGFhjKtfPhsaV6IKYpiGmORRk9qPTtuNS53tw= -apigo.cc/go/cast v1.5.3/go.mod h1:GMjjrYn93tWat1U409G7h1jR3ejfLLI7r0efBo9Sbd4= -apigo.cc/go/config v1.5.3 h1:peq1FM2xO+vzPHJf8Dwg3DXm8PtFQMfTFKQj6fpoG7A= -apigo.cc/go/config v1.5.3/go.mod h1:ZiOAjWa1mQIzszaJZN+kO6YU4GXreng+NxkcK/TAkqQ= -apigo.cc/go/encoding v1.5.4 h1:Fk8TrveZATyy8SHukC4ZiqdTSp+QIfsRHtt55xmMK7w= -apigo.cc/go/encoding v1.5.4/go.mod h1:dShEsZ3gKqBINz7TSOYf4e7/fBCqCY9VzlenoGUQUFM= -apigo.cc/go/file v1.5.5 h1:/+HmDumLu6Qk2KuQL63M9lpgzHTDL+QJ8dStOl7e9gs= -apigo.cc/go/file v1.5.5/go.mod h1:xRVNhctvqOKeBemmcRW/BQfgkc3B+vT/UZVdSc7duUo= -apigo.cc/go/id v1.5.4 h1:D1Zx9gEZhOgdTgZ4SdmPImhpc9xGiOA33Y+j2MkstzQ= -apigo.cc/go/id v1.5.4/go.mod h1:hCTQq+KC1ALWe1FpPERf+W4B6FSulg9FAgOUJDDySiY= -apigo.cc/go/jsmod v1.5.3 h1:S3W317bH0QV2NMeRO1E0v6ySIBOfMWYv/NuQJbvqKWU= -apigo.cc/go/jsmod v1.5.3/go.mod h1:bmyeZtOAP/j5am+YRnaiM89smysK24K7ebk0koFtsSw= -apigo.cc/go/log v1.5.8 h1:/IYtGPWhRjT3OayylDIphkWZIQbpLjqVeSnFEiD3Dy0= -apigo.cc/go/log v1.5.8/go.mod h1:HfFPANMYxJx197SSTXB21Pgxcz/gGqPP8nlSErgd5WE= -apigo.cc/go/rand v1.5.3 h1:O4bPIwyaOWEBCr0nL9A4G4qG48AqiGTCzfPeckm3Ius= -apigo.cc/go/rand v1.5.3/go.mod h1:q1BTFkY/cXE229dDD5Q22lF7T0DoKPV6xAu+6bCrDH4= -apigo.cc/go/safe v1.5.2 h1:EnuEOW/SGwf/5A0nw9LnqfKJE071+TIc6ez8HI9R9Lg= -apigo.cc/go/safe v1.5.2/go.mod h1:2GqCCLLGex4OAhdET3iBWm1R+LIYtmTrvHP8W0iESSw= -apigo.cc/go/shell v1.5.3 h1:pI+u12sy6upoygq+1XXqUlvUboBfH4Q52jRpoJFv56A= -apigo.cc/go/shell v1.5.3/go.mod h1:FdZWUrcXHGJXo725oSyHqAeFoX0E9yY3PDhrz9hujgY= +apigo.cc/go/cast v1.5.5 h1:DMbfK3uPhPjRaXutj3StIZIkfjFIATSXfuAOeNOd4Fw= +apigo.cc/go/cast v1.5.5/go.mod h1:GMjjrYn93tWat1U409G7h1jR3ejfLLI7r0efBo9Sbd4= +apigo.cc/go/config v1.5.4 h1:1c/OarGwbz3+6ikurE+a6LJLjtzXvGvbxw3HV/Nk54M= +apigo.cc/go/config v1.5.4/go.mod h1:oN+D2F8ETIyqKp+Yu8R4PRQlUoqR44o35jAHwLrrAq0= +apigo.cc/go/crypto v1.5.5 h1:YQHumieqviNGMhwoxDtMuUdGWAQiGrkVhz9hZRHOhWs= +apigo.cc/go/crypto v1.5.5/go.mod h1:z/FXt0HE7fSMJKF2MF7jY2fSjlHUYYaSffETJ7ZzJmE= +apigo.cc/go/encoding v1.5.6 h1:v02swVfbFGidD4QcX2ktuHHbCjdSbOB85fhzAXay+7M= +apigo.cc/go/encoding v1.5.6/go.mod h1:Big9q1Zwy4071dXtnrQ3SJDzfa/G7/A60KE/5+M//P8= +apigo.cc/go/file v1.5.6 h1:Y7w3Tyu4e16VuED7rF2pzba+dzGE+hjDnBlIPVHIfzA= +apigo.cc/go/file v1.5.6/go.mod h1:9sdW4ylSOA0HWc8Yt8qdnmMf6nn5SUEmjoPKyXpYXIQ= +apigo.cc/go/id v1.5.7 h1:Y5Sx6sQBCAdYMCQPTjODZyGMMd1+WRWCy2dWHVq11XQ= +apigo.cc/go/id v1.5.7/go.mod h1:fugudFBqfVNakfm91zZzuzU0P4PULzo6sylB8hRMqxA= +apigo.cc/go/jsmod v1.5.4 h1:r76LbQww674avYt8vITmrhNwM2WSnlriCLECNjXsR1o= +apigo.cc/go/jsmod v1.5.4/go.mod h1:bmyeZtOAP/j5am+YRnaiM89smysK24K7ebk0koFtsSw= +apigo.cc/go/log v1.5.11 h1:r7vHkzpdelggNguZZK4e3O9bcaxfQWjXkxsKYw4A8Bo= +apigo.cc/go/log v1.5.11/go.mod h1:C6qtOn09miyCK7FXEAAwEZASnBow76K7GZFoBaq9eiU= +apigo.cc/go/rand v1.5.4 h1:eessFBsKQuoOYdzrStldOGw9f4HqbeO87X95bI4jIBQ= +apigo.cc/go/rand v1.5.4/go.mod h1:q1BTFkY/cXE229dDD5Q22lF7T0DoKPV6xAu+6bCrDH4= +apigo.cc/go/redis v1.5.13 h1:TBiVjoVU41Sk7QYdYTqOVVkyxhimSVn/LzX6t/jLaJw= +apigo.cc/go/redis v1.5.13/go.mod h1:xssiBiRftemDITrjro7nZ3nibDVAWE5qLnMAu3MVMJA= +apigo.cc/go/safe v1.5.3 h1:9p/BmdlVWLbekpKByZIFC09Qn8Wdhik2eINiwunBxPs= +apigo.cc/go/safe v1.5.3/go.mod h1:Ay8kEPL76DeXH4ifsVTc/3/sfGHlWLQjAp4vi7GA9AI= +apigo.cc/go/shell v1.5.6 h1:2i93lNJ0oy/D5kOmlmOjwVjLPsW7vQBo0trLOF9AAAI= +apigo.cc/go/shell v1.5.6/go.mod h1:Bp73DGKESOISWSIGUtL1dsFo5g4SI10GhgsnLJCTpsw= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/gomodule/redigo v1.9.3 h1:dNPSXeXv6HCq2jdyWfjgmhBdqnR6PRO3m/G05nvpPC8= +github.com/gomodule/redigo v1.9.3/go.mod h1:KsU3hiK/Ay8U42qpaJk+kuNa3C+spxapWpM+ywhcgtw= github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= golang.org/x/crypto v0.52.0 h1:RMs7fP2rXdep0CftQlK8Uf+kibLm7qkCcradZWYz988= golang.org/x/crypto v0.52.0/go.mod h1:1QgfPxDqh0T2M/elOJtp9RvuR95kVjir0e6/BvEmGbc= golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= diff --git a/queue/queue.go b/queue/queue.go new file mode 100644 index 0000000..a011a6b --- /dev/null +++ b/queue/queue.go @@ -0,0 +1,168 @@ +package queue + +import ( + "fmt" + "strings" + "sync/atomic" + "time" + + "apigo.cc/go/cast" + "apigo.cc/go/redis" +) + +var delayedSequence uint64 + +const processingPlaceholder = "\x00apigo-task-processing\x00" + +func pendingKey(name string) string { return "queue:" + name + ":pending" } +func processingKey(name, consumerID string) string { + return "queue:" + name + ":" + strings.TrimSpace(consumerID) + ":processing" +} +func client() (*redis.Redis, error) { + rd := redis.GetRedis("task", nil) + if rd == nil { + return nil, fmt.Errorf("task redis is unavailable") + } + if rd.Error != nil { + return nil, rd.Error + } + return rd, nil +} + +func Add(queueName string, item any) error { + rd, err := client() + if err != nil { + return err + } + r := rd.Do("LPUSH", pendingKey(queueName), item) + return r.Error +} + +// AddAfter stores an item in the queue's persistent delayed lane until delay elapses. +func AddAfter(queueName string, item any, delay time.Duration) error { + rd, err := client() + if err != nil { + return err + } + payload, err := cast.ToJSON(item) + if err != nil { + return err + } + id := fmt.Sprintf("%d:%d", time.Now().UnixNano(), atomic.AddUint64(&delayedSequence, 1)) + member := id + "\x00" + payload + r := rd.Do("ZADD", delayedKey(queueName), float64(time.Now().Add(delay).UnixMilli()), member) + return r.Error +} + +func delayedKey(name string) string { return "queue:" + name + ":delayed" } + +// PromoteDelayed moves due items into pending. Queue workers call this before Fetch. +// The current task runtime has one consumer per queue. Pending is written +// before delayed is removed so a crash can only result in a duplicate, never +// a lost task; consumers already have retry/idempotency safeguards. +func PromoteDelayed(queueName string, limit int) error { + if limit < 1 { + limit = 100 + } + rd, err := client() + if err != nil { + return err + } + items, err := rd.ZRangeByScore(delayedKey(queueName), "-inf", time.Now().UnixMilli(), 0, limit) + if err != nil { + return err + } + for _, result := range items { + member := result.String() + parts := strings.SplitN(member, "\x00", 2) + if len(parts) != 2 { + _ = rd.ZREM(delayedKey(queueName), member) + continue + } + _ = rd.Do("LPUSH", pendingKey(queueName), parts[1]) + _ = rd.ZREM(delayedKey(queueName), member) + } + return rd.Error +} + +func Fetch(queueName, consumerID string, n int) ([]redis.Result, error) { + if n < 1 { + return []redis.Result{}, nil + } + rd, err := client() + if err != nil { + return nil, err + } + if err := PromoteDelayed(queueName, n); err != nil { + return nil, err + } + processing := processingKey(queueName, consumerID) + if items := processingItems(rd, processing); len(items) > 0 { + return items, nil + } + for i := 0; i < n; i++ { + if rd.LLEN(pendingKey(queueName)) == 0 { + break + } + if rd.LLEN(processing) == 0 { + // SugarDB requires the LMOVE destination to exist, while standard Redis + // creates it. Pushing and immediately popping a marker initializes an + // empty SugarDB list without retaining application-visible data. + if r := rd.Do("LPUSH", processing, processingPlaceholder); r.Error != nil { + return nil, r.Error + } + if r := rd.Do("LPOP", processing); r.Error != nil { + return nil, r.Error + } + } + result := rd.Do("LMOVE", pendingKey(queueName), processing, "RIGHT", "LEFT") + if result.Error != nil { + return nil, result.Error + } + if strings.TrimSpace(result.String()) == "" { + break + } + } + return processingItems(rd, processing), nil +} + +func processingItems(rd *redis.Redis, key string) []redis.Result { + items := rd.LRANGE(key, 0, -1) + filtered := items[:0] + for _, item := range items { + if item.String() != processingPlaceholder { + filtered = append(filtered, item) + } + } + return filtered +} + +func Finish(queueName, consumerID string) error { + rd, err := client() + if err != nil { + return err + } + r := rd.Do("DEL", processingKey(queueName, consumerID)) + return r.Error +} + +func FetchCount(queueName, consumerID string) int { + _ = PromoteDelayed(queueName, 100) + rd, err := client() + if err != nil { + return 0 + } + processing := processingKey(queueName, consumerID) + if count := len(processingItems(rd, processing)); count > 0 { + return count + } + return rd.LLEN(pendingKey(queueName)) +} + +func Decode[T any](result redis.Result) (T, error) { + var item T + if err := cast.UnmarshalJSON(result.Bytes(), &item); err != nil { + return item, fmt.Errorf("decode queue item: %w", err) + } + return item, nil +} diff --git a/queue/queue_test.go b/queue/queue_test.go new file mode 100644 index 0000000..0a5e9a0 --- /dev/null +++ b/queue/queue_test.go @@ -0,0 +1,197 @@ +package queue_test + +import ( + "fmt" + "net" + "net/url" + "os" + "testing" + "time" + + "apigo.cc/go/redis" + "apigo.cc/go/task/queue" +) + +type queueItem struct { + ID int + Text string +} + +func TestMain(m *testing.M) { + redisURL := os.Getenv("TASK_QUEUE_TEST_REDIS") + if redisURL == "" { + redisURL = "redis://:@localhost:6379/14?connectTimeout=100ms&readTimeout=100ms&writeTimeout=100ms" + } + parsedURL, err := url.Parse(redisURL) + if err != nil { + fmt.Printf("Invalid TASK_QUEUE_TEST_REDIS: %v\n", err) + os.Exit(1) + } + conn, err := net.DialTimeout("tcp", parsedURL.Host, 500*time.Millisecond) + if err != nil { + fmt.Printf("Redis server is not running at %s, skipping queue tests.\n", parsedURL.Host) + os.Exit(0) + } + _ = conn.Close() + + redis.SetConfig("task", redisURL) + os.Exit(m.Run()) +} + +func TestAddFetchFinish(t *testing.T) { + queueName := testQueueName(t) + cleanupQueue(t, queueName, "default") + + want := queueItem{ID: 1, Text: "first"} + if err := queue.Add(queueName, want); err != nil { + t.Fatalf("Add failed: %v", err) + } + + items, err := queue.Fetch(queueName, "default", 10) + if err != nil { + t.Fatalf("Fetch failed: %v", err) + } + if len(items) != 1 { + t.Fatalf("Fetch returned %d items, want 1", len(items)) + } + rawProcessing := redis.GetRedis("task", nil).LRANGE("queue:"+queueName+":default:processing", 0, -1) + if len(rawProcessing) != 1 { + t.Fatalf("processing contains %d raw items, want only the fetched task", len(rawProcessing)) + } + if rawProcessing[0].String() == "\x00apigo-task-processing\x00" { + t.Fatal("processing retained its compatibility marker") + } + got, err := queue.Decode[queueItem](items[0]) + if err != nil { + t.Fatalf("Decode failed: %v", err) + } + if got != want { + t.Fatalf("Fetch returned %+v, want %+v", got, want) + } + if count := queue.FetchCount(queueName, "default"); count != 1 { + t.Fatalf("FetchCount returned %d while processing, want 1", count) + } + + if err := queue.Finish(queueName, "default"); err != nil { + t.Fatalf("Finish failed: %v", err) + } + if count := queue.FetchCount(queueName, "default"); count != 0 { + t.Fatalf("FetchCount returned %d after Finish, want 0", count) + } +} + +func TestFetchRecoversProcessingBeforePending(t *testing.T) { + queueName := testQueueName(t) + cleanupQueue(t, queueName, "default") + + processingItem := queueItem{ID: 1, Text: "processing"} + if err := queue.Add(queueName, processingItem); err != nil { + t.Fatalf("Add processing item failed: %v", err) + } + firstFetch, err := queue.Fetch(queueName, "default", 1) + if err != nil { + t.Fatalf("initial Fetch failed: %v", err) + } + if len(firstFetch) != 1 { + t.Fatalf("initial Fetch returned %d items, want 1", len(firstFetch)) + } + + pendingItem := queueItem{ID: 2, Text: "pending"} + if err := queue.Add(queueName, pendingItem); err != nil { + t.Fatalf("Add pending item failed: %v", err) + } + recovered, err := queue.Fetch(queueName, "default", 50) + if err != nil { + t.Fatalf("recovery Fetch failed: %v", err) + } + if len(recovered) != 1 { + t.Fatalf("recovery Fetch returned %d items, want the existing processing batch only", len(recovered)) + } + got, err := queue.Decode[queueItem](recovered[0]) + if err != nil { + t.Fatalf("Decode recovered item failed: %v", err) + } + if got != processingItem { + t.Fatalf("recovery Fetch returned %+v, want %+v", got, processingItem) + } + + if err := queue.Finish(queueName, "default"); err != nil { + t.Fatalf("Finish recovered batch failed: %v", err) + } + next, err := queue.Fetch(queueName, "default", 1) + if err != nil { + t.Fatalf("Fetch pending item failed: %v", err) + } + if len(next) != 1 { + t.Fatalf("Fetch pending item returned %d items, want 1", len(next)) + } + got, err = queue.Decode[queueItem](next[0]) + if err != nil { + t.Fatalf("Decode pending item failed: %v", err) + } + if got != pendingItem { + t.Fatalf("Fetch after Finish returned %+v, want %+v", got, pendingItem) + } +} + +func TestAddAfterPromotesOnlyDueItems(t *testing.T) { + queueName := testQueueName(t) + cleanupQueue(t, queueName, "default") + + due := queueItem{ID: 1, Text: "due"} + future := queueItem{ID: 2, Text: "future"} + if err := queue.AddAfter(queueName, future, time.Hour); err != nil { + t.Fatalf("AddAfter future item failed: %v", err) + } + if err := queue.AddAfter(queueName, due, -time.Millisecond); err != nil { + t.Fatalf("AddAfter due item failed: %v", err) + } + + items, err := queue.Fetch(queueName, "default", 10) + if err != nil { + t.Fatalf("Fetch delayed item failed: %v", err) + } + if len(items) != 1 { + t.Fatalf("Fetch returned %d delayed items, want only the due item", len(items)) + } + got, err := queue.Decode[queueItem](items[0]) + if err != nil { + t.Fatalf("Decode delayed item failed: %v", err) + } + if got != due { + t.Fatalf("Fetch returned delayed item %+v, want %+v", got, due) + } + + rd := redis.GetRedis("task", nil) + if count := rd.Do("ZCARD", delayedKey(queueName)).Int(); count != 1 { + t.Fatalf("delayed queue contains %d items after promotion, want the future item only", count) + } +} + +func testQueueName(t *testing.T) string { + t.Helper() + return fmt.Sprintf("task-queue-test:%d:%s", time.Now().UnixNano(), t.Name()) +} + +func cleanupQueue(t *testing.T, queueName, consumerID string) { + t.Helper() + rd := redis.GetRedis("task", nil) + if rd == nil || rd.Error != nil { + t.Fatalf("task Redis is unavailable: %v", rd.Error) + } + keys := []string{ + "queue:" + queueName + ":pending", + "queue:" + queueName + ":" + consumerID + ":processing", + delayedKey(queueName), + } + if result := rd.Do("DEL", keys); result.Error != nil { + t.Fatalf("clean queue before test: %v", result.Error) + } + t.Cleanup(func() { + _ = rd.Do("DEL", keys) + }) +} + +func delayedKey(queueName string) string { + return "queue:" + queueName + ":delayed" +} diff --git a/scheduler.go b/scheduler.go index 99ff9d7..7540856 100644 --- a/scheduler.go +++ b/scheduler.go @@ -8,6 +8,7 @@ import ( "sync" "time" + "apigo.cc/go/id" "apigo.cc/go/log" "github.com/robfig/cron/v3" ) @@ -17,6 +18,7 @@ type Scheduler struct { tasks map[string]*Task mu sync.RWMutex wg sync.WaitGroup + logger *log.Logger running bool } @@ -70,6 +72,9 @@ func (s *Scheduler) Start(ctx context.Context, logger *log.Logger) error { return nil } s.running = true + if logger != nil { + s.logger = logger + } s.mu.Unlock() s.cron.Start() @@ -190,8 +195,16 @@ func (s *Scheduler) runTask(name string) { s.wg.Add(1) defer s.wg.Done() + taskLogger := log.DefaultLogger + s.mu.RLock() + parentLogger := s.logger + s.mu.RUnlock() + if parentLogger != nil { + taskLogger = parentLogger.New(id.Get10Bytes14MPerSecond()) + } ctx, cancel := context.WithCancel(context.Background()) + ctx = WithLogger(ctx, taskLogger) if t.Config.Timeout > 0 { ctx, cancel = context.WithTimeout(ctx, t.Config.Timeout) } @@ -212,7 +225,7 @@ func (s *Scheduler) runTask(name string) { // Recover defer func() { if r := recover(); r != nil { - log.DefaultLogger.Error("task panic recovered", "name", name, "err", r, "stack", string(debug.Stack())) + taskLogger.Error("task panic recovered", "name", name, "err", r, "stack", string(debug.Stack())) } }() @@ -221,8 +234,15 @@ func (s *Scheduler) runTask(name string) { duration := time.Since(start) if err != nil { - log.DefaultLogger.Error("task failed", "name", name, "err", err, "duration", duration.String()) + taskLogger.Error("task failed", "name", name, "err", err, "duration", duration.String()) } else if !t.Config.SuppressSuccessLog { - log.DefaultLogger.Info("task success", "name", name, "duration", duration.String()) + taskLogger.Info("task success", "name", name, "duration", duration.String()) } } + +func traceID(logger *log.Logger) string { + if logger == nil { + return "" + } + return logger.GetTraceId() +} diff --git a/task.go b/task.go index c5fb87c..cd54391 100644 --- a/task.go +++ b/task.go @@ -87,6 +87,26 @@ func Get(name string) *Task { return DefaultScheduler.Get(name) } +// Enable enables a registered task. It returns false when the task is unknown. +func Enable(name string) bool { + t := Get(name) + if t == nil { + return false + } + t.Enable() + return true +} + +// Disable disables a registered task. It returns false when the task is unknown. +func Disable(name string) bool { + t := Get(name) + if t == nil { + return false + } + t.Disable() + return true +} + // List 获取所有任务 func List() []*Task { return DefaultScheduler.List()