feat: add independent platform modules and multi-provider storage
This commit is contained in:
213
queue/redis/consumer.go
Normal file
213
queue/redis/consumer.go
Normal file
@@ -0,0 +1,213 @@
|
||||
package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/queue"
|
||||
goredis "github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
type ConsumerConfig struct {
|
||||
Workers int `json:"workers"`
|
||||
MaxAttempts int64 `json:"maxAttempts"`
|
||||
TaskTimeout time.Duration `json:"taskTimeout"`
|
||||
VisibilityTimeout time.Duration `json:"visibilityTimeout"`
|
||||
ShutdownTimeout time.Duration `json:"shutdownTimeout"`
|
||||
}
|
||||
type Consumer struct {
|
||||
store *Store
|
||||
cfg ConsumerConfig
|
||||
used atomic.Bool
|
||||
}
|
||||
|
||||
func NewConsumer(store *Store, cfg ConsumerConfig) (*Consumer, error) {
|
||||
if store == nil || cfg.Workers < 0 || cfg.MaxAttempts < 0 || cfg.TaskTimeout < 0 || cfg.VisibilityTimeout < 0 || cfg.ShutdownTimeout < 0 {
|
||||
return nil, errors.New("queue/redis: invalid consumer configuration")
|
||||
}
|
||||
if cfg.Workers == 0 {
|
||||
cfg.Workers = 4
|
||||
}
|
||||
if cfg.MaxAttempts == 0 {
|
||||
cfg.MaxAttempts = 3
|
||||
}
|
||||
if cfg.TaskTimeout == 0 {
|
||||
cfg.TaskTimeout = 30 * time.Second
|
||||
}
|
||||
if cfg.VisibilityTimeout == 0 {
|
||||
cfg.VisibilityTimeout = 2 * cfg.TaskTimeout
|
||||
}
|
||||
if cfg.ShutdownTimeout == 0 {
|
||||
cfg.ShutdownTimeout = 5 * time.Second
|
||||
}
|
||||
if cfg.Workers > 128 || cfg.MaxAttempts > 100 || cfg.TaskTimeout < 10*time.Millisecond || cfg.TaskTimeout > time.Hour || cfg.VisibilityTimeout < 2*cfg.TaskTimeout || cfg.VisibilityTimeout > 24*time.Hour || cfg.ShutdownTimeout > time.Minute {
|
||||
return nil, errors.New("queue/redis: consumer limits exceeded")
|
||||
}
|
||||
return &Consumer{store: store, cfg: cfg}, nil
|
||||
}
|
||||
|
||||
// Run is called once. Cancellation stops fetching and cancels handlers; ACK is
|
||||
// deferred to a later delivery if cancellation makes the outcome uncertain.
|
||||
// A handler must honor context cancellation. Go cannot terminate an arbitrary
|
||||
// function; an ignoring handler can outlive ShutdownTimeout (at most Workers).
|
||||
func (c *Consumer) Run(ctx context.Context, handler queue.Handler) error {
|
||||
if ctx == nil || handler == nil {
|
||||
return errors.New("queue/redis: context and handler required")
|
||||
}
|
||||
if !c.used.CompareAndSwap(false, true) {
|
||||
return errors.New("queue/redis: consumer already run")
|
||||
}
|
||||
signature := fmt.Sprintf("%d:%d:%d", c.cfg.MaxAttempts, c.cfg.TaskTimeout.Milliseconds(), c.cfg.VisibilityTimeout.Milliseconds())
|
||||
if err := goredis.NewScript(`local v=redis.call('GET',KEYS[1]);if v and v~=ARGV[1] then return redis.error_reply('consumer configuration mismatch') end;redis.call('SET',KEYS[1],ARGV[1]);return 1`).Run(ctx, c.store.client, []string{c.store.keys[5]}, signature).Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := c.store.client.XGroupCreateMkStream(ctx, c.store.keys[0], "workers", "0").Err(); err != nil && !strings.HasPrefix(err.Error(), "BUSYGROUP ") {
|
||||
return err
|
||||
}
|
||||
var token [16]byte
|
||||
if _, err := rand.Read(token[:]); err != nil {
|
||||
return err
|
||||
}
|
||||
runCtx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
failures := make(chan error, c.cfg.Workers)
|
||||
var wg sync.WaitGroup
|
||||
for n := 0; n < c.cfg.Workers; n++ {
|
||||
wg.Add(1)
|
||||
go func(n int) {
|
||||
defer wg.Done()
|
||||
if err := c.loop(runCtx, hex.EncodeToString(token[:])+fmt.Sprint(n), handler); err != nil {
|
||||
failures <- err
|
||||
cancel()
|
||||
}
|
||||
}(n)
|
||||
}
|
||||
done := make(chan struct{})
|
||||
go func() { wg.Wait(); close(done) }()
|
||||
select {
|
||||
case <-done:
|
||||
case <-runCtx.Done():
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(c.cfg.ShutdownTimeout):
|
||||
return errors.New("queue/redis: handler shutdown timeout")
|
||||
}
|
||||
}
|
||||
select {
|
||||
case err := <-failures:
|
||||
return err
|
||||
default:
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
func (c *Consumer) loop(ctx context.Context, owner string, handler queue.Handler) error {
|
||||
s := c.store
|
||||
defer func() {
|
||||
cleanup, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
_ = goredis.NewScript(`local p=redis.call('XPENDING',KEYS[1],'workers','-','+',1,ARGV[1]);if #p==0 then return redis.call('XGROUP','DELCONSUMER',KEYS[1],'workers',ARGV[1]) end;return 0`).Run(cleanup, s.client, []string{s.keys[0]}, owner).Err()
|
||||
}()
|
||||
cursor := "0-0"
|
||||
for ctx.Err() == nil {
|
||||
messages, next, err := s.client.XAutoClaim(ctx, &goredis.XAutoClaimArgs{Stream: s.keys[0], Group: "workers", Consumer: owner, MinIdle: c.cfg.VisibilityTimeout, Start: cursor, Count: 1}).Result()
|
||||
if err != nil && err != goredis.Nil {
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
cursor = next
|
||||
if cursor == "" {
|
||||
cursor = "0-0"
|
||||
}
|
||||
if len(messages) == 0 {
|
||||
streams, err := s.client.XReadGroup(ctx, &goredis.XReadGroupArgs{Group: "workers", Consumer: owner, Streams: []string{s.keys[0], ">"}, Count: 1, Block: -1}).Result()
|
||||
if err != nil && err != goredis.Nil {
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
if len(streams) > 0 {
|
||||
messages = streams[0].Messages
|
||||
}
|
||||
}
|
||||
if len(messages) == 0 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case <-time.After(25 * time.Millisecond):
|
||||
}
|
||||
continue
|
||||
}
|
||||
message := messages[0]
|
||||
raw, ok := message.Values["message"].(string)
|
||||
var m queue.Message
|
||||
if !ok || len(raw) > s.cfg.MaxPayloadBytes+1024 || json.Unmarshal([]byte(raw), &m) != nil {
|
||||
return errors.New("queue/redis: corrupt message; operator review required")
|
||||
}
|
||||
pending, err := s.client.XPendingExt(ctx, &goredis.XPendingExtArgs{Stream: s.keys[0], Group: "workers", Start: message.ID, End: message.ID, Count: 1}).Result()
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
if len(pending) != 1 || pending[0].Consumer != owner {
|
||||
continue
|
||||
}
|
||||
delivery := queue.Delivery{Message: m, Attempt: pending[0].RetryCount}
|
||||
reason := ""
|
||||
if delivery.Attempt > c.cfg.MaxAttempts {
|
||||
reason = "attemptsExhausted"
|
||||
} else {
|
||||
workCtx, cancel := context.WithTimeout(ctx, c.cfg.TaskTimeout)
|
||||
err = invoke(workCtx, handler, delivery)
|
||||
deadlineErr := workCtx.Err()
|
||||
cancel()
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
var rejection *queue.Rejection
|
||||
if errors.As(err, &rejection) {
|
||||
reason = "rejected"
|
||||
} else if err != nil || deadlineErr != nil {
|
||||
if delivery.Attempt < c.cfg.MaxAttempts {
|
||||
continue
|
||||
}
|
||||
reason = "attemptsExhausted"
|
||||
}
|
||||
}
|
||||
ackCtx, ackCancel := context.WithTimeout(ctx, 3*time.Second)
|
||||
err = s.finish(ackCtx, message.ID, owner, delivery, reason)
|
||||
ackCancel()
|
||||
if err != nil && !errors.Is(err, queue.ErrLeaseLost) {
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func invoke(ctx context.Context, handler queue.Handler, delivery queue.Delivery) (err error) {
|
||||
defer func() {
|
||||
if recover() != nil {
|
||||
err = errors.New("queue: handler panic")
|
||||
}
|
||||
}()
|
||||
if err = ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
return handler(ctx, delivery)
|
||||
}
|
||||
|
||||
var _ queue.Consumer = (*Consumer)(nil)
|
||||
154
queue/redis/consumer_test.go
Normal file
154
queue/redis/consumer_test.go
Normal file
@@ -0,0 +1,154 @@
|
||||
package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"git.yuntuo.hk/k2safe/go-omni/queue"
|
||||
goredis "github.com/redis/go-redis/v9"
|
||||
"os"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func testStore(t *testing.T, max int) *Store {
|
||||
t.Helper()
|
||||
address := os.Getenv("GO_OMNI_REDIS_TEST_ADDR")
|
||||
if address == "" {
|
||||
t.Skip("requires isolated Redis")
|
||||
}
|
||||
client := goredis.NewClient(&goredis.Options{Addr: address, PoolSize: 4, MaxRetries: -1, ContextTimeoutEnabled: true, ReadTimeout: time.Second, WriteTimeout: time.Second})
|
||||
s, err := New(client, Config{Namespace: fmt.Sprint(t.Name(), time.Now().UnixNano()), MaxRecords: max, DedupTTL: time.Second})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { client.Del(context.Background(), s.keys...); client.Close() })
|
||||
return s
|
||||
}
|
||||
func message(id string) queue.Message {
|
||||
return queue.Message{ID: id, Name: "test", Payload: json.RawMessage(`{"value":1}`)}
|
||||
}
|
||||
func eventually(t *testing.T, f func() bool) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
for !f() {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("condition did not recover")
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
func TestQueueRedisIntegration(t *testing.T) {
|
||||
t.Run("publicationAndCapacity", func(t *testing.T) {
|
||||
s := testStore(t, 1)
|
||||
ctx := context.Background()
|
||||
receipt, err := s.Publish(ctx, message("first"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
again, err := s.Publish(ctx, message("first"))
|
||||
if err != nil || again != receipt {
|
||||
t.Fatal(again, err)
|
||||
}
|
||||
changed := message("first")
|
||||
changed.Payload = json.RawMessage(`2`)
|
||||
if _, err = s.Publish(ctx, changed); !errors.Is(err, queue.ErrConflict) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err = s.Publish(ctx, message("second")); !errors.Is(err, queue.ErrFull) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
})
|
||||
t.Run("staleAcknowledgement", func(t *testing.T) {
|
||||
s := testStore(t, 10)
|
||||
ctx := context.Background()
|
||||
receipt, _ := s.Publish(ctx, message("first"))
|
||||
s.client.XGroupCreateMkStream(ctx, s.keys[0], "workers", "0")
|
||||
s.client.XReadGroup(ctx, &goredis.XReadGroupArgs{Group: "workers", Consumer: "old", Streams: []string{s.keys[0], ">"}, Count: 1, Block: -1})
|
||||
if err := s.client.XClaim(ctx, &goredis.XClaimArgs{Stream: s.keys[0], Group: "workers", Consumer: "new", MinIdle: 0, Messages: []string{receipt.BackendID}}).Err(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
delivery := queue.Delivery{Message: message("first"), Attempt: 2}
|
||||
if err := s.finish(ctx, receipt.BackendID, "old", delivery, ""); !errors.Is(err, queue.ErrLeaseLost) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := s.finish(ctx, receipt.BackendID, "new", delivery, ""); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n := s.client.XLen(ctx, s.keys[0]).Val(); n != 0 {
|
||||
t.Fatal(n)
|
||||
}
|
||||
})
|
||||
t.Run("crashReclaimRetryAndDeadLetters", func(t *testing.T) {
|
||||
s := testStore(t, 10)
|
||||
ctx := context.Background()
|
||||
s.Publish(ctx, message("crashed"))
|
||||
s.Publish(ctx, message("retry"))
|
||||
s.Publish(ctx, message("reject"))
|
||||
s.client.XGroupCreateMkStream(ctx, s.keys[0], "workers", "0")
|
||||
s.client.XReadGroup(ctx, &goredis.XReadGroupArgs{Group: "workers", Consumer: "crashedProcess", Streams: []string{s.keys[0], ">"}, Count: 1, Block: -1})
|
||||
cfg := ConsumerConfig{Workers: 1, MaxAttempts: 3, TaskTimeout: 50 * time.Millisecond, VisibilityTimeout: 100 * time.Millisecond, ShutdownTimeout: time.Second}
|
||||
consumer, _ := NewConsumer(s, cfg)
|
||||
second, _ := NewConsumer(s, cfg)
|
||||
runCtx, cancel := context.WithCancel(ctx)
|
||||
var retries, reclaimed, rejected atomic.Int32
|
||||
handler := func(ctx context.Context, d queue.Delivery) error {
|
||||
switch d.ID {
|
||||
case "crashed":
|
||||
if d.Attempt != 2 {
|
||||
t.Error("reclaim count", d.Attempt)
|
||||
}
|
||||
reclaimed.Add(1)
|
||||
return nil
|
||||
case "retry":
|
||||
retries.Add(1)
|
||||
return errors.New("known retryable failure")
|
||||
default:
|
||||
rejected.Add(1)
|
||||
return queue.Reject("invalidInput")
|
||||
}
|
||||
}
|
||||
done := make(chan error, 2)
|
||||
go func() { done <- consumer.Run(runCtx, handler) }()
|
||||
go func() { done <- second.Run(runCtx, handler) }()
|
||||
defer cancel()
|
||||
eventually(t, func() bool {
|
||||
dead, err := s.DeadLetters(ctx, "", 100)
|
||||
return err == nil && len(dead) == 2 && reclaimed.Load() == 1
|
||||
})
|
||||
cancel()
|
||||
for i := 0; i < 2; i++ {
|
||||
if err := <-done; !errors.Is(err, context.Canceled) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if retries.Load() != 3 || rejected.Load() != 1 {
|
||||
t.Fatal(retries.Load(), rejected.Load())
|
||||
}
|
||||
dead, err := s.DeadLetters(ctx, "", 1)
|
||||
if err != nil || len(dead) != 1 {
|
||||
t.Fatal(dead, err)
|
||||
}
|
||||
rest, err := s.DeadLetters(ctx, dead[0].StreamID, 1)
|
||||
if err != nil || len(rest) != 1 || rest[0].StreamID == dead[0].StreamID {
|
||||
t.Fatal(rest, err)
|
||||
}
|
||||
if err = s.DiscardDead(ctx, dead[0].Message.ID, dead[0].StreamID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
remaining, _ := s.DeadLetters(ctx, "", 100)
|
||||
if len(remaining) != 1 {
|
||||
t.Fatal(remaining)
|
||||
}
|
||||
// Retained completed IDs expire and release publication capacity.
|
||||
eventually(t, func() bool {
|
||||
_, err := s.Publish(ctx, message("crashed"))
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
return s.client.XLen(ctx, s.keys[0]).Val() == 2
|
||||
})
|
||||
})
|
||||
}
|
||||
187
queue/redis/store.go
Normal file
187
queue/redis/store.go
Normal file
@@ -0,0 +1,187 @@
|
||||
// Package redis implements one managed Redis Streams consumer group per queue.
|
||||
// Use Redis 6.2+, persistence, noeviction and a private namespace. The application
|
||||
// owns the client. Delivery durability depends on the Redis deployment.
|
||||
package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"git.yuntuo.hk/k2safe/go-omni/queue"
|
||||
goredis "github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
Namespace string `json:"namespace"`
|
||||
MaxRecords int `json:"maxRecords"` // Includes pending, dead and retained completed identities.
|
||||
MaxPayloadBytes int `json:"maxPayloadBytes"`
|
||||
DedupTTL time.Duration `json:"dedupTtl"` // Starts at ACK; pending/dead records never expire automatically.
|
||||
}
|
||||
type Store struct {
|
||||
client goredis.UniversalClient
|
||||
cfg Config
|
||||
keys []string
|
||||
signature string
|
||||
}
|
||||
|
||||
func New(client goredis.UniversalClient, cfg Config) (*Store, error) {
|
||||
if client == nil || cfg.Namespace == "" || len(cfg.Namespace) > 256 || cfg.MaxRecords < 0 || cfg.MaxPayloadBytes < 0 || cfg.DedupTTL < 0 {
|
||||
return nil, errors.New("queue/redis: invalid configuration")
|
||||
}
|
||||
if cfg.MaxRecords == 0 {
|
||||
cfg.MaxRecords = 1000
|
||||
}
|
||||
if cfg.MaxPayloadBytes == 0 {
|
||||
cfg.MaxPayloadBytes = 64 << 10
|
||||
}
|
||||
if cfg.DedupTTL == 0 {
|
||||
cfg.DedupTTL = 24 * time.Hour
|
||||
}
|
||||
if cfg.MaxRecords > 100000 || cfg.MaxPayloadBytes > 1<<20 || cfg.DedupTTL < time.Second || cfg.DedupTTL > 30*24*time.Hour {
|
||||
return nil, errors.New("queue/redis: configuration exceeds supported bounds")
|
||||
}
|
||||
sum := sha256.Sum256([]byte(cfg.Namespace))
|
||||
base := "goOmni:{" + hex.EncodeToString(sum[:]) + "}:queue"
|
||||
return &Store{client: client, cfg: cfg, keys: []string{base + ":stream", base + ":records", base + ":expiry", base + ":dead", base + ":config", base + ":consumerConfig"}, signature: fmt.Sprintf("%d:%d:%d", cfg.MaxRecords, cfg.MaxPayloadBytes, cfg.DedupTTL.Milliseconds())}, nil
|
||||
}
|
||||
|
||||
var publish = goredis.NewScript(`
|
||||
local config=redis.call('GET',KEYS[5]);if config and config~=ARGV[5] then return redis.error_reply('configuration mismatch') end
|
||||
if not config then redis.call('SET',KEYS[5],ARGV[5]) end
|
||||
local clock=redis.call('TIME');local now=clock[1]*1000+math.floor(clock[2]/1000)
|
||||
local expired=redis.call('ZRANGEBYSCORE',KEYS[3],'-inf',now,'LIMIT',0,128)
|
||||
for _,id in ipairs(expired) do redis.call('HDEL',KEYS[2],id);redis.call('ZREM',KEYS[3],id) end
|
||||
local expiry=redis.call('ZSCORE',KEYS[3],ARGV[1]);if expiry and tonumber(expiry)<=now then redis.call('HDEL',KEYS[2],ARGV[1]);redis.call('ZREM',KEYS[3],ARGV[1]) end
|
||||
local old=redis.call('HGET',KEYS[2],ARGV[1]);if old then local r=cjson.decode(old);if r.fingerprint~=ARGV[2] then return {-1,''} end;return {0,r.streamId} end
|
||||
if redis.call('HLEN',KEYS[2])>=tonumber(ARGV[4]) then return {-2,''} end
|
||||
local id=redis.call('XADD',KEYS[1],'*','message',ARGV[3])
|
||||
redis.call('HSET',KEYS[2],ARGV[1],cjson.encode({fingerprint=ARGV[2],streamId=id,state='pending'}));return {1,id}
|
||||
`)
|
||||
|
||||
func (s *Store) Publish(ctx context.Context, m queue.Message) (queue.Receipt, error) {
|
||||
if err := queue.Validate(ctx, m, s.cfg.MaxPayloadBytes); err != nil {
|
||||
return queue.Receipt{}, err
|
||||
}
|
||||
raw, _ := json.Marshal(m)
|
||||
if len(raw) > s.cfg.MaxPayloadBytes+1024 {
|
||||
return queue.Receipt{}, errors.New("queue/redis: encoded message exceeds limit")
|
||||
}
|
||||
sum := sha256.Sum256(raw)
|
||||
r, err := publish.Run(ctx, s.client, s.keys, m.ID, hex.EncodeToString(sum[:]), string(raw), s.cfg.MaxRecords, s.signature).Slice()
|
||||
if err != nil {
|
||||
return queue.Receipt{}, err
|
||||
}
|
||||
if len(r) != 2 {
|
||||
return queue.Receipt{}, errors.New("queue/redis: invalid publish result")
|
||||
}
|
||||
switch r[0] {
|
||||
case int64(-1):
|
||||
return queue.Receipt{}, queue.ErrConflict
|
||||
case int64(-2):
|
||||
return queue.Receipt{}, queue.ErrFull
|
||||
}
|
||||
id, ok := r[1].(string)
|
||||
if !ok {
|
||||
return queue.Receipt{}, errors.New("queue/redis: invalid receipt")
|
||||
}
|
||||
return queue.Receipt{ID: m.ID, BackendID: id}, nil
|
||||
}
|
||||
|
||||
var finish = goredis.NewScript(`
|
||||
local p=redis.call('XPENDING',KEYS[1],'workers',ARGV[1],ARGV[1],1)
|
||||
if #p~=1 or p[1][2]~=ARGV[2] then return 0 end
|
||||
local raw=redis.call('HGET',KEYS[2],ARGV[3]);if not raw then return redis.error_reply('message record missing') end
|
||||
local record=cjson.decode(raw);if record.streamId~=ARGV[1] or record.state~='pending' then return redis.error_reply('message identity mismatch') end
|
||||
if ARGV[4]~='' then
|
||||
record.state='dead';redis.call('ZADD',KEYS[4],0,ARGV[1]);record.dead=cjson.decode(ARGV[4])
|
||||
else
|
||||
record.state='completed';local clock=redis.call('TIME');local now=clock[1]*1000+math.floor(clock[2]/1000);redis.call('ZADD',KEYS[3],now+tonumber(ARGV[5]),ARGV[3])
|
||||
end
|
||||
redis.call('HSET',KEYS[2],ARGV[3],cjson.encode(record));redis.call('XACK',KEYS[1],'workers',ARGV[1])
|
||||
if ARGV[4]=='' then redis.call('XDEL',KEYS[1],ARGV[1]) end
|
||||
return 1
|
||||
`)
|
||||
|
||||
func (s *Store) finish(ctx context.Context, id, owner string, m queue.Delivery, reason string) error {
|
||||
var dead []byte
|
||||
if reason != "" {
|
||||
dead, _ = json.Marshal(map[string]any{"attempt": m.Attempt, "reason": reason})
|
||||
}
|
||||
n, err := finish.Run(ctx, s.client, s.keys, id, owner, m.ID, string(dead), s.cfg.DedupTTL.Milliseconds()).Int()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if n != 1 {
|
||||
return queue.ErrLeaseLost
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type DeadLetter struct {
|
||||
StreamID string `json:"streamId"`
|
||||
Message queue.Message `json:"message"`
|
||||
Attempt int64 `json:"attempt"`
|
||||
Reason string `json:"reason"`
|
||||
}
|
||||
|
||||
// DeadLetters returns a bounded page, ordered by opaque Redis stream ID. Pass
|
||||
// the last StreamID as after. Payloads remain retained until explicit removal.
|
||||
func (s *Store) DeadLetters(ctx context.Context, after string, limit int64) ([]DeadLetter, error) {
|
||||
if ctx == nil || limit < 1 || limit > 100 {
|
||||
return nil, errors.New("queue/redis: context and limit 1..100 required")
|
||||
}
|
||||
min := "-"
|
||||
if after != "" {
|
||||
min = "(" + after
|
||||
}
|
||||
ids, err := s.client.ZRangeByLex(ctx, s.keys[3], &goredis.ZRangeBy{Min: min, Max: "+", Count: limit}).Result()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]DeadLetter, 0, len(ids))
|
||||
for _, id := range ids {
|
||||
messages, err := s.client.XRangeN(ctx, s.keys[0], id, id, 1).Result()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(messages) == 0 {
|
||||
continue
|
||||
}
|
||||
var m queue.Message
|
||||
raw, _ := messages[0].Values["message"].(string)
|
||||
if err = json.Unmarshal([]byte(raw), &m); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
record, err := s.client.HGet(ctx, s.keys[1], m.ID).Result()
|
||||
if err == goredis.Nil {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var saved struct {
|
||||
Dead struct {
|
||||
Attempt int64 `json:"attempt"`
|
||||
Reason string `json:"reason"`
|
||||
} `json:"dead"`
|
||||
}
|
||||
if err = json.Unmarshal([]byte(record), &saved); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, DeadLetter{StreamID: id, Message: m, Attempt: saved.Dead.Attempt, Reason: saved.Dead.Reason})
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// DiscardDead deletes only a matching dead delivery after operator review.
|
||||
// It retains the producer identity for DedupTTL, preventing immediate resubmit.
|
||||
func (s *Store) DiscardDead(ctx context.Context, messageID, streamID string) error {
|
||||
return goredis.NewScript(`local raw=redis.call('HGET',KEYS[2],ARGV[1]);if not raw then return 0 end;local r=cjson.decode(raw);if r.state~='dead' or r.streamId~=ARGV[2] then return redis.error_reply('dead identity mismatch') end;r.state='discarded';r.dead=nil;redis.call('HSET',KEYS[2],ARGV[1],cjson.encode(r));redis.call('ZREM',KEYS[4],ARGV[2]);redis.call('XDEL',KEYS[1],ARGV[2]);local t=redis.call('TIME');redis.call('ZADD',KEYS[3],t[1]*1000+math.floor(t[2]/1000)+tonumber(ARGV[3]),ARGV[1]);return 1`).Run(ctx, s.client, s.keys, messageID, streamID, s.cfg.DedupTTL.Milliseconds()).Err()
|
||||
}
|
||||
|
||||
var _ queue.Publisher = (*Store)(nil)
|
||||
Reference in New Issue
Block a user