402 lines
13 KiB
Go
402 lines
13 KiB
Go
package submitworker
|
|
|
|
import (
|
|
"context"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"cmpp-platform/gateway/internal/queue"
|
|
"github.com/alicebob/miniredis/v2"
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
type recordingLimiter struct {
|
|
channelID string
|
|
rate int
|
|
called bool
|
|
}
|
|
|
|
type acknowledgingResultOutbox struct {
|
|
client *redis.Client
|
|
}
|
|
|
|
func (o acknowledgingResultOutbox) PublishSubmitResultAndAck(
|
|
ctx context.Context,
|
|
commandStream string,
|
|
commandGroup string,
|
|
commandMessageID string,
|
|
_ queue.SubmitCommand,
|
|
_ queue.SubmitResult,
|
|
) error {
|
|
if o.client == nil {
|
|
return nil
|
|
}
|
|
return o.client.XAck(ctx, commandStream, commandGroup, commandMessageID).Err()
|
|
}
|
|
|
|
func (l *recordingLimiter) Wait(_ context.Context, channelID string, rate int) (time.Duration, error) {
|
|
l.called = true
|
|
l.channelID = channelID
|
|
l.rate = rate
|
|
return 0, nil
|
|
}
|
|
|
|
func TestCommandFromStreamValuesParsesSubmitCommand(t *testing.T) {
|
|
command, err := CommandFromStreamValues(map[string]interface{}{
|
|
"messageType": "SubmitCommand",
|
|
"data": `{
|
|
"schemaVersion": "v1",
|
|
"messageType": "SubmitCommand",
|
|
"traceId": "trace-worker-0001",
|
|
"messageId": "msg-worker-0001",
|
|
"channelId": "channel-1",
|
|
"createdAt": "2026-07-07T10:00:00Z",
|
|
"tenantId": "tenant-1",
|
|
"applicationId": "app-1",
|
|
"submitId": "submit-1",
|
|
"phoneNumber": "13800138000",
|
|
"content": "hello",
|
|
"signature": "测试",
|
|
"templateId": "tpl-1",
|
|
"billingUnits": 1,
|
|
"queuePriority": "normal",
|
|
"route": {
|
|
"channelCode": "CMPP-A",
|
|
"cmppAccountCode": "account-a",
|
|
"priority": 0,
|
|
"rateLimitPerSecond": 100
|
|
},
|
|
"cmpp": {
|
|
"serviceId": "SMS",
|
|
"srcId": "10690000",
|
|
"registeredDelivery": 1,
|
|
"msgFmt": 8
|
|
},
|
|
"upstream": {
|
|
"gatewayHost": "127.0.0.1",
|
|
"gatewayPort": 17890,
|
|
"account": "account-a",
|
|
"passwordCipher": "secret",
|
|
"cmppVersion": "3.0"
|
|
},
|
|
"retry": {
|
|
"attempt": 0,
|
|
"maxAttempts": 1
|
|
}
|
|
}`,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("parse stream command: %v", err)
|
|
}
|
|
if command.MessageID != "msg-worker-0001" || command.Upstream.Account != "account-a" {
|
|
t.Fatalf("unexpected command: %+v", command)
|
|
}
|
|
}
|
|
|
|
func TestCommandFromStreamValuesRejectsMissingData(t *testing.T) {
|
|
_, err := CommandFromStreamValues(map[string]interface{}{"messageType": "SubmitCommand"})
|
|
if err == nil {
|
|
t.Fatal("expected missing data error")
|
|
}
|
|
}
|
|
|
|
func TestHandleMessageUsesInjectedSubmit(t *testing.T) {
|
|
var got queue.SubmitCommand
|
|
limiter := &recordingLimiter{}
|
|
worker := &Worker{
|
|
Limiter: limiter,
|
|
Submit: func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
|
|
got = command
|
|
return queue.SubmitResult{SubmitStatus: "accepted"}, nil
|
|
},
|
|
}
|
|
if err := worker.handleCommand(context.Background(), queue.SubmitCommand{
|
|
Envelope: queue.Envelope{MessageID: "msg-worker-0002", ChannelID: "channel-1"},
|
|
SubmitID: "submit-2",
|
|
PhoneNumber: "13800138000",
|
|
Content: "hello",
|
|
Upstream: queue.UpstreamConfig{GatewayHost: "127.0.0.1", GatewayPort: 17890, Account: "account-a", PasswordCipher: "secret", CMPPVersion: "3.0"},
|
|
CMPP: queue.CMPP{ServiceID: "SMS", SrcID: "10690000", RegisteredDelivery: 1, MsgFmt: 8},
|
|
Route: queue.Route{ChannelCode: "CMPP-A", RateLimitPerSecond: 320},
|
|
Retry: queue.Retry{Attempt: 0, MaxAttempts: 1},
|
|
ApplicationID: "app-1",
|
|
TenantID: "tenant-1",
|
|
}); err != nil {
|
|
t.Fatalf("handleCommand returned error: %v", err)
|
|
}
|
|
if got.MessageID != "msg-worker-0002" || got.SubmitID != "submit-2" {
|
|
t.Fatalf("unexpected command: %+v", got)
|
|
}
|
|
if !limiter.called || limiter.channelID != "channel-1" || limiter.rate != 320 {
|
|
t.Fatalf("unexpected limiter call: %+v", limiter)
|
|
}
|
|
}
|
|
|
|
func TestMinIdleDefaultsToThirtySeconds(t *testing.T) {
|
|
worker := &Worker{}
|
|
if got := worker.minIdle(); got != defaultMinIdle {
|
|
t.Fatalf("minIdle = %v, want %v", got, defaultMinIdle)
|
|
}
|
|
}
|
|
|
|
func TestConcurrencyUsesBoundedDefaultAndMaximum(t *testing.T) {
|
|
if got := (&Worker{}).ConfiguredConcurrency(); got != defaultConcurrency {
|
|
t.Fatalf("default concurrency = %d, want %d", got, defaultConcurrency)
|
|
}
|
|
if got := (&Worker{Concurrency: 2048}).ConfiguredConcurrency(); got != 1024 {
|
|
t.Fatalf("capped concurrency = %d, want 1024", got)
|
|
}
|
|
}
|
|
|
|
func TestMessageWorkPoolContinuouslyRefillsWithoutWaitingForSlowSibling(t *testing.T) {
|
|
mr := miniredis.RunT(t)
|
|
client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
|
|
startedA := make(chan struct{})
|
|
startedB := make(chan struct{})
|
|
startedC := make(chan struct{})
|
|
releaseA := make(chan struct{})
|
|
worker := &Worker{
|
|
Redis: client,
|
|
ResultOutbox: acknowledgingResultOutbox{client: client},
|
|
Submit: func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
|
|
switch command.ChannelID {
|
|
case "channel-a":
|
|
close(startedA)
|
|
<-releaseA
|
|
case "channel-b":
|
|
close(startedB)
|
|
case "channel-c":
|
|
close(startedC)
|
|
}
|
|
return queue.SubmitResult{SubmitStatus: "accepted"}, nil
|
|
},
|
|
}
|
|
pool := newMessageWorkPool(context.Background(), worker, 2)
|
|
if !pool.dispatch(redis.XMessage{ID: "1-0", Values: submitCommandValues("message-a", "channel-a")}) ||
|
|
!pool.dispatch(redis.XMessage{ID: "2-0", Values: submitCommandValues("message-b", "channel-b")}) {
|
|
t.Fatal("initial messages were not dispatched")
|
|
}
|
|
select {
|
|
case <-startedA:
|
|
case <-time.After(time.Second):
|
|
t.Fatal("channel-a did not start")
|
|
}
|
|
select {
|
|
case <-startedB:
|
|
case <-time.After(200 * time.Millisecond):
|
|
t.Fatal("channel-b was blocked by channel-a")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
|
defer cancel()
|
|
if _, err := pool.waitForCapacity(ctx); err != nil {
|
|
t.Fatalf("wait for refill capacity: %v", err)
|
|
}
|
|
if !pool.dispatch(redis.XMessage{ID: "3-0", Values: submitCommandValues("message-c", "channel-c")}) {
|
|
t.Fatal("refill message was not dispatched")
|
|
}
|
|
select {
|
|
case <-startedC:
|
|
case <-time.After(200 * time.Millisecond):
|
|
t.Fatal("pool waited for the slow sibling instead of refilling its free slot")
|
|
}
|
|
close(releaseA)
|
|
pool.wait()
|
|
}
|
|
|
|
func TestMessageWorkPoolAcknowledgesFastMessageBeforeSlowSiblingCompletes(t *testing.T) {
|
|
mr := miniredis.RunT(t)
|
|
client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
|
|
worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway", ResultOutbox: acknowledgingResultOutbox{client: client}}
|
|
ctx := context.Background()
|
|
if err := worker.ensureGroup(ctx); err != nil {
|
|
t.Fatalf("ensureGroup: %v", err)
|
|
}
|
|
for _, entry := range []struct{ id, messageID, channelID string }{
|
|
{"1-0", "message-slow", "channel-slow"},
|
|
{"2-0", "message-fast", "channel-fast"},
|
|
} {
|
|
if err := client.XAdd(ctx, &redis.XAddArgs{Stream: worker.stream(), ID: entry.id, Values: submitCommandValues(entry.messageID, entry.channelID)}).Err(); err != nil {
|
|
t.Fatalf("xadd %s: %v", entry.id, err)
|
|
}
|
|
}
|
|
streams, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{Group: worker.group(), Consumer: worker.consumer(), Streams: []string{worker.stream(), ">"}, Count: 2}).Result()
|
|
if err != nil || len(streams) != 1 || len(streams[0].Messages) != 2 {
|
|
t.Fatalf("xreadgroup: streams=%+v err=%v", streams, err)
|
|
}
|
|
slowStarted := make(chan struct{})
|
|
fastReturned := make(chan struct{})
|
|
releaseSlow := make(chan struct{})
|
|
worker.Submit = func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
|
|
if command.ChannelID == "channel-slow" {
|
|
close(slowStarted)
|
|
<-releaseSlow
|
|
} else {
|
|
close(fastReturned)
|
|
}
|
|
return queue.SubmitResult{SubmitStatus: "accepted"}, nil
|
|
}
|
|
pool := newMessageWorkPool(ctx, worker, 2)
|
|
for _, message := range streams[0].Messages {
|
|
if !pool.dispatch(message) {
|
|
t.Fatalf("message %s was not dispatched", message.ID)
|
|
}
|
|
}
|
|
<-slowStarted
|
|
<-fastReturned
|
|
deadline := time.Now().Add(time.Second)
|
|
for {
|
|
pending, pendingErr := client.XPending(ctx, worker.stream(), worker.group()).Result()
|
|
if pendingErr != nil {
|
|
t.Fatalf("xpending: %v", pendingErr)
|
|
}
|
|
if pending.Count == 1 {
|
|
break
|
|
}
|
|
if time.Now().After(deadline) {
|
|
t.Fatalf("pending count = %d, want 1 while slow sibling is still running", pending.Count)
|
|
}
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
close(releaseSlow)
|
|
pool.wait()
|
|
}
|
|
|
|
func TestPendingRecoveryDoesNotDuplicateAnActiveOrAlreadyAcknowledgedMessage(t *testing.T) {
|
|
mr := miniredis.RunT(t)
|
|
client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
|
|
worker := &Worker{Redis: client, Stream: "gateway.submit.commands", Group: "cmpp-gateway", MinIdle: time.Millisecond, ResultOutbox: acknowledgingResultOutbox{client: client}}
|
|
ctx := context.Background()
|
|
if err := worker.ensureGroup(ctx); err != nil {
|
|
t.Fatalf("ensureGroup: %v", err)
|
|
}
|
|
if err := client.XAdd(ctx, &redis.XAddArgs{Stream: worker.stream(), ID: "3-0", Values: submitCommandValues("message-active", "channel-active")}).Err(); err != nil {
|
|
t.Fatalf("xadd: %v", err)
|
|
}
|
|
streams, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{Group: worker.group(), Consumer: worker.consumer(), Streams: []string{worker.stream(), ">"}, Count: 1}).Result()
|
|
if err != nil || len(streams) != 1 || len(streams[0].Messages) != 1 {
|
|
t.Fatalf("xreadgroup: streams=%+v err=%v", streams, err)
|
|
}
|
|
message := streams[0].Messages[0]
|
|
started := make(chan struct{})
|
|
release := make(chan struct{})
|
|
var submits atomic.Int32
|
|
worker.Submit = func(_ context.Context, _ queue.SubmitCommand) (queue.SubmitResult, error) {
|
|
submits.Add(1)
|
|
close(started)
|
|
<-release
|
|
return queue.SubmitResult{SubmitStatus: "accepted"}, nil
|
|
}
|
|
pool := newMessageWorkPool(ctx, worker, 2)
|
|
if !pool.dispatch(message) {
|
|
t.Fatal("active message was not dispatched")
|
|
}
|
|
<-started
|
|
time.Sleep(5 * time.Millisecond)
|
|
if err := worker.recoverPending(ctx, pool); err != nil {
|
|
t.Fatalf("recoverPending: %v", err)
|
|
}
|
|
if got := submits.Load(); got != 1 {
|
|
t.Fatalf("active message submit count = %d, want 1", got)
|
|
}
|
|
close(release)
|
|
pool.wait()
|
|
if err := pool.dispatchRecovered(ctx, message); err != nil {
|
|
t.Fatalf("dispatch acknowledged recovery: %v", err)
|
|
}
|
|
if got := submits.Load(); got != 1 {
|
|
t.Fatalf("acknowledged message submit count = %d, want 1", got)
|
|
}
|
|
}
|
|
|
|
func TestProcessMessageDeadLettersAfterMaxFailures(t *testing.T) {
|
|
mr := miniredis.RunT(t)
|
|
client := redis.NewClient(&redis.Options{Addr: mr.Addr()})
|
|
reported := []DeadLetterEvent{}
|
|
worker := &Worker{
|
|
Redis: client,
|
|
Stream: "gateway.submit.commands",
|
|
Group: "cmpp-gateway",
|
|
MaxFailures: 2,
|
|
Submit: func(_ context.Context, command queue.SubmitCommand) (queue.SubmitResult, error) {
|
|
return queue.SubmitResult{}, context.DeadlineExceeded
|
|
},
|
|
ReportDeadLetter: func(_ context.Context, event DeadLetterEvent) error {
|
|
reported = append(reported, event)
|
|
return nil
|
|
},
|
|
}
|
|
ctx := context.Background()
|
|
if err := worker.ensureGroup(ctx); err != nil {
|
|
t.Fatalf("ensureGroup: %v", err)
|
|
}
|
|
message := redis.XMessage{
|
|
ID: "1710000000000-0",
|
|
Values: map[string]interface{}{
|
|
"messageType": "SubmitCommand",
|
|
"data": `{
|
|
"schemaVersion": "v1",
|
|
"messageType": "SubmitCommand",
|
|
"traceId": "trace-worker-0003",
|
|
"messageId": "msg-worker-0003",
|
|
"channelId": "channel-1",
|
|
"createdAt": "2026-07-08T10:00:00Z",
|
|
"tenantId": "tenant-1",
|
|
"applicationId": "app-1",
|
|
"submitId": "submit-3",
|
|
"phoneNumber": "13800138000",
|
|
"content": "hello",
|
|
"signature": "测试",
|
|
"templateId": "tpl-1",
|
|
"billingUnits": 1,
|
|
"queuePriority": "normal",
|
|
"route": { "channelCode": "CMPP-A", "cmppAccountCode": "account-a", "priority": 0 },
|
|
"cmpp": { "serviceId": "SMS", "srcId": "10690000", "registeredDelivery": 1, "msgFmt": 8 },
|
|
"upstream": { "gatewayHost": "127.0.0.1", "gatewayPort": 17890, "account": "account-a", "passwordCipher": "secret", "cmppVersion": "3.0" },
|
|
"retry": { "attempt": 0, "maxAttempts": 1 }
|
|
}`,
|
|
},
|
|
}
|
|
if err := client.XAdd(ctx, &redis.XAddArgs{
|
|
Stream: worker.stream(),
|
|
ID: message.ID,
|
|
Values: message.Values,
|
|
}).Err(); err != nil {
|
|
t.Fatalf("xadd: %v", err)
|
|
}
|
|
|
|
if err := worker.processMessage(ctx, message); err == nil {
|
|
t.Fatal("expected first failure")
|
|
}
|
|
if len(reported) != 0 {
|
|
t.Fatalf("unexpected dead letters on first failure: %+v", reported)
|
|
}
|
|
|
|
if err := worker.processMessage(ctx, message); err != nil {
|
|
t.Fatalf("second failure should dead-letter and ack, got %v", err)
|
|
}
|
|
if len(reported) != 1 {
|
|
t.Fatalf("dead letters = %d, want 1", len(reported))
|
|
}
|
|
if reported[0].FailureCode != "SUBMIT_PROCESSING_FAILED" || reported[0].Attempts != 2 {
|
|
t.Fatalf("unexpected dead letter: %+v", reported[0])
|
|
}
|
|
if client.HGet(ctx, worker.failureAttemptsKey(), message.ID).Err() != redis.Nil {
|
|
t.Fatalf("failure attempt key was not cleared")
|
|
}
|
|
}
|
|
|
|
func submitCommandValues(messageID string, channelID string) map[string]interface{} {
|
|
return map[string]interface{}{
|
|
"data": `{
|
|
"schemaVersion":"v1","messageType":"SubmitCommand","traceId":"trace-1",
|
|
"messageId":"` + messageID + `","channelId":"` + channelID + `","submitId":"submit-1",
|
|
"tenantId":"tenant-1","applicationId":"app-1","phoneNumber":"13800138000","content":"hello",
|
|
"route":{"channelCode":"CMPP-A","rateLimitPerSecond":100},
|
|
"cmpp":{"serviceId":"SMS","srcId":"10690000","registeredDelivery":1,"msgFmt":8},
|
|
"upstream":{"gatewayHost":"127.0.0.1","gatewayPort":17890,"account":"sp","passwordCipher":"secret","cmppVersion":"3.0"},
|
|
"retry":{"attempt":0,"maxAttempts":1}
|
|
}`,
|
|
}
|
|
}
|