123 lines
4.1 KiB
Go
123 lines
4.1 KiB
Go
package outbox
|
|
|
|
import (
|
|
"errors"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"gorm.io/driver/sqlite"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
func testDB(t *testing.T) *gorm.DB {
|
|
t.Helper()
|
|
db, err := gorm.Open(sqlite.Open("file:"+uuid.NewString()+"?mode=memory&cache=shared"), &gorm.Config{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
sqlDB, _ := db.DB()
|
|
sqlDB.SetMaxOpenConns(1)
|
|
if err = db.AutoMigrate(&Message{}, &DeliveryRecord{}, &Attempt{}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return db
|
|
}
|
|
|
|
func TestEnqueueClaimLeaseRecoveryBackoffAndIdempotentSuccess(t *testing.T) {
|
|
db := testDB(t)
|
|
now := time.Date(2026, 8, 28, 8, 0, 0, 0, time.UTC)
|
|
message, err := Enqueue(db, EnqueueInput{InternalType: "local_event_candidate", BusinessRef: "event-1", IdempotencyKey: "local-event:event-1:v1", PayloadJSON: []byte(`{"event":"event-1"}`), MaxAttempts: 3}, now)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err = Enqueue(db, EnqueueInput{InternalType: "local_event_candidate", BusinessRef: "event-1", IdempotencyKey: message.IdempotencyKey, PayloadJSON: []byte(`{}`)}, now); !errors.Is(err, ErrDuplicateIdempotency) {
|
|
t.Fatalf("expected duplicate error, got %v", err)
|
|
}
|
|
relay := NewRelay(db)
|
|
relay.Now = func() time.Time { return now }
|
|
relay.LeaseDuration = 10 * time.Second
|
|
relay.Backoff = func(int) time.Duration { return 5 * time.Second }
|
|
first, err := relay.Claim("worker-1", 1)
|
|
if err != nil || len(first) != 1 {
|
|
t.Fatalf("first claim=%#v err=%v", first, err)
|
|
}
|
|
second, err := relay.Claim("worker-2", 1)
|
|
if err != nil || len(second) != 0 {
|
|
t.Fatalf("concurrent claim=%#v err=%v", second, err)
|
|
}
|
|
now = now.Add(11 * time.Second)
|
|
recovered, err := relay.Claim("worker-2", 1)
|
|
if err != nil || len(recovered) != 1 || recovered[0].LeaseOwner != "worker-2" {
|
|
t.Fatalf("recovered=%#v err=%v", recovered, err)
|
|
}
|
|
if err = relay.MarkFailure(message.ID, "worker-2", "temporary outage"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
now = now.Add(4 * time.Second)
|
|
waiting, _ := relay.Claim("worker-3", 1)
|
|
if len(waiting) != 0 {
|
|
t.Fatalf("claimed before backoff elapsed: %#v", waiting)
|
|
}
|
|
now = now.Add(2 * time.Second)
|
|
retry, err := relay.Claim("worker-3", 1)
|
|
if err != nil || len(retry) != 1 {
|
|
t.Fatalf("retry=%#v err=%v", retry, err)
|
|
}
|
|
if err = relay.MarkSuccess(message.ID, "worker-3"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = relay.MarkSuccess(message.ID, "worker-3"); err != nil {
|
|
t.Fatalf("idempotent success failed: %v", err)
|
|
}
|
|
var deliveries int64
|
|
db.Model(&DeliveryRecord{}).Where("idempotency_key = ?", message.IdempotencyKey).Count(&deliveries)
|
|
if deliveries != 1 {
|
|
t.Fatalf("deliveries=%d", deliveries)
|
|
}
|
|
}
|
|
|
|
func TestFailureBecomesDeadAndManualRequeuePreservesHistory(t *testing.T) {
|
|
db := testDB(t)
|
|
now := time.Date(2026, 8, 28, 9, 0, 0, 0, time.UTC)
|
|
message, err := Enqueue(db, EnqueueInput{InternalType: "audit_projection", BusinessRef: "audit-1", IdempotencyKey: "audit:audit-1:v1", PayloadJSON: []byte(`{}`), MaxAttempts: 1}, now)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
relay := NewRelay(db)
|
|
relay.Now = func() time.Time { return now }
|
|
claimed, _ := relay.Claim("worker", 1)
|
|
if len(claimed) != 1 {
|
|
t.Fatal("message not claimed")
|
|
}
|
|
if err = relay.MarkFailure(message.ID, "worker", "permanent failure"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
service := NewService(db)
|
|
service.Now = func() time.Time { return now.Add(time.Minute) }
|
|
detail, err := service.Get(message.ID)
|
|
if err != nil || detail.Message.State != StateDead || len(detail.Attempts) != 1 {
|
|
t.Fatalf("detail=%#v err=%v", detail, err)
|
|
}
|
|
requeued, err := service.Requeue(message.ID, RequeueRequest{ExpectedVersion: detail.Message.Version, Reason: "出口故障已排除"}, 7)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if requeued.State != StatePending {
|
|
t.Fatalf("state=%s", requeued.State)
|
|
}
|
|
detail, _ = service.Get(message.ID)
|
|
if len(detail.Attempts) != 2 || detail.Attempts[0].Outcome != "manual_requeue" || detail.Attempts[0].ActorUserID != 7 {
|
|
t.Fatalf("attempt history=%#v", detail.Attempts)
|
|
}
|
|
}
|
|
|
|
func TestProductionCannotCreateTestSink(t *testing.T) {
|
|
if _, err := NewTestSink("prod"); !errors.Is(err, ErrTestSinkForbidden) {
|
|
t.Fatalf("err=%v", err)
|
|
}
|
|
if _, err := NewTestSink("test"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|