93 lines
2.5 KiB
Go
93 lines
2.5 KiB
Go
package outbox
|
|
|
|
import (
|
|
"fmt"
|
|
"net/url"
|
|
"os"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"gorm.io/driver/postgres"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
func TestPostgresConcurrentWorkersDoNotClaimSameMessage(t *testing.T) {
|
|
dsn := os.Getenv("SENSE_OUTBOX_TEST_DATABASE_URL")
|
|
if dsn == "" {
|
|
t.Skip("set SENSE_OUTBOX_TEST_DATABASE_URL to run the PostgreSQL multi-worker test")
|
|
}
|
|
base, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
schema := fmt.Sprintf("sense_outbox_78_%d", time.Now().UnixNano())
|
|
if err = base.Exec("CREATE SCHEMA " + schema).Error; err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = base.Exec("DROP SCHEMA IF EXISTS " + schema + " CASCADE").Error })
|
|
scoped, err := gorm.Open(postgres.Open(withSearchPath(dsn, schema)), &gorm.Config{})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err = scoped.AutoMigrate(&Message{}, &DeliveryRecord{}, &Attempt{}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
now := time.Date(2026, 8, 28, 12, 0, 0, 0, time.UTC)
|
|
for index := 0; index < 20; index++ {
|
|
if _, err = Enqueue(scoped, EnqueueInput{InternalType: "local_event_candidate", BusinessRef: fmt.Sprintf("event-%d", index), IdempotencyKey: fmt.Sprintf("event:%d:v1", index), PayloadJSON: []byte(`{}`)}, now); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
workers := []string{"worker-a", "worker-b"}
|
|
results := make(chan []Message, len(workers))
|
|
errorsCh := make(chan error, len(workers))
|
|
var group sync.WaitGroup
|
|
for _, worker := range workers {
|
|
group.Add(1)
|
|
go func(name string) {
|
|
defer group.Done()
|
|
relay := NewRelay(scoped)
|
|
relay.Now = func() time.Time { return now }
|
|
items, claimErr := relay.Claim(name, 20)
|
|
if claimErr != nil {
|
|
errorsCh <- claimErr
|
|
return
|
|
}
|
|
results <- items
|
|
}(worker)
|
|
}
|
|
group.Wait()
|
|
close(results)
|
|
close(errorsCh)
|
|
for claimErr := range errorsCh {
|
|
t.Fatal(claimErr)
|
|
}
|
|
seen := map[string]string{}
|
|
for batch := range results {
|
|
for _, item := range batch {
|
|
if owner, exists := seen[item.ID]; exists {
|
|
t.Fatalf("message %s claimed by %s and %s", item.ID, owner, item.LeaseOwner)
|
|
}
|
|
seen[item.ID] = item.LeaseOwner
|
|
}
|
|
}
|
|
if len(seen) != 20 {
|
|
t.Fatalf("claimed=%d want=20", len(seen))
|
|
}
|
|
}
|
|
|
|
func withSearchPath(dsn, schema string) string {
|
|
if strings.Contains(dsn, "://") {
|
|
parsed, err := url.Parse(dsn)
|
|
if err == nil {
|
|
query := parsed.Query()
|
|
query.Set("search_path", schema)
|
|
parsed.RawQuery = query.Encode()
|
|
return parsed.String()
|
|
}
|
|
}
|
|
return strings.TrimSpace(dsn) + " search_path=" + schema
|
|
}
|