188 lines
7.6 KiB
Go
188 lines
7.6 KiB
Go
// 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)
|