327 lines
10 KiB
Go
327 lines
10 KiB
Go
package events_test
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"chat-agent-system/internal/app/uow"
|
|
"chat-agent-system/internal/db"
|
|
"chat-agent-system/internal/events"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
func init() {
|
|
if _, err := os.Stat("deploy/postgres/goose"); err == nil {
|
|
return
|
|
}
|
|
for _, cand := range []string{"../..", "../../.."} {
|
|
if fi, err := os.Stat(cand + "/deploy/postgres/goose"); err == nil && fi.IsDir() {
|
|
_ = os.Chdir(cand)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func e0bPool(t *testing.T) (*pgxpool.Pool, context.Context) {
|
|
t.Helper()
|
|
if testing.Short() {
|
|
t.Skip("E0B dispatcher needs isolated postgres")
|
|
}
|
|
dsn := db.SkipIfNoTestDatabaseURL(t)
|
|
db.RefuseSharedEnvDSN(t, dsn)
|
|
ctx := context.Background()
|
|
pool, err := pgxpool.New(ctx, dsn)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(pool.Close)
|
|
if err := db.RunMigrationsOnceForTest(ctx, pool, dsn); err != nil {
|
|
t.Fatalf("migrate: %v", err)
|
|
}
|
|
return pool, ctx
|
|
}
|
|
|
|
func seedEnvelope(t *testing.T, ctx context.Context, pool *pgxpool.Pool, env events.Envelope, consumers []string) {
|
|
t.Helper()
|
|
unit := uow.NewPG(pool)
|
|
if err := unit.Within(ctx, func(scope uow.TxScope) error {
|
|
return (events.PGSink{}).Append(uow.WithScope(ctx, scope), env, consumers)
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func sampleEnv(id, agg string, ver int64) events.Envelope {
|
|
body := []byte(`{"doc_id":"` + agg + `"}`)
|
|
return events.Envelope{
|
|
EventID: id, EventName: events.EventVersionCommitted, SchemaVersion: events.EnvelopeSchema,
|
|
Environment: "dev", ProjectID: 9, AggregateType: "document", AggregateID: agg,
|
|
AggregateVersion: ver, ActorRef: "user:1", IdempotencyKey: id, PayloadDigest: events.Digest(body),
|
|
Producer: "docflow", OccurredAt: time.Now().UTC(), Payload: body,
|
|
}
|
|
}
|
|
|
|
func TestE0B_ModeOffDoesNotClaim_PG(t *testing.T) {
|
|
pool, ctx := e0bPool(t)
|
|
suffix := fmt.Sprintf("%d", time.Now().UnixNano())
|
|
seedEnvelope(t, ctx, pool, sampleEnv("off-"+suffix, "doc-off-"+suffix, 1), []string{events.ConsumerNotify})
|
|
if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeOff); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
d := events.NewDispatcher(pool, "dev", "w-off")
|
|
n, err := d.RunOnce(ctx, events.ConsumerNotify)
|
|
if err != nil || n != 0 {
|
|
t.Fatalf("claimed=%d err=%v", n, err)
|
|
}
|
|
}
|
|
|
|
func TestE0B_FAVNeverClaimed_PG(t *testing.T) {
|
|
pool, ctx := e0bPool(t)
|
|
tx, err := pool.Begin(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
if _, err := events.ClaimLeases(ctx, tx, events.ConsumerFAV, "w", 8, events.DefaultLease); err == nil {
|
|
t.Fatal("fav claim must fail")
|
|
}
|
|
if err := events.SetMode(ctx, pool, "dev", events.ConsumerFAV, events.ModeLive); err == nil {
|
|
t.Fatal("fav live must fail")
|
|
}
|
|
}
|
|
|
|
func TestE0B_WorkersNoDoubleLeaseAndSerialAggregate_PG(t *testing.T) {
|
|
pool, ctx := e0bPool(t)
|
|
suffix := fmt.Sprintf("%d", time.Now().UnixNano())
|
|
if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeLive); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for i := 0; i < 8; i++ {
|
|
seedEnvelope(t, ctx, pool, sampleEnv(fmt.Sprintf("par-%s-%d", suffix, i), fmt.Sprintf("doc-par-%s-%d", suffix, i), 1), []string{events.ConsumerNotify})
|
|
}
|
|
seedEnvelope(t, ctx, pool, sampleEnv("ser-a-"+suffix, "doc-ser-"+suffix, 1), []string{events.ConsumerNotify})
|
|
seedEnvelope(t, ctx, pool, sampleEnv("ser-b-"+suffix, "doc-ser-"+suffix, 2), []string{events.ConsumerNotify})
|
|
|
|
visible := "e0b_visible_" + suffix
|
|
if _, err := pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS `+visible+` (event_id TEXT PRIMARY KEY, n INT NOT NULL)`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _, _ = pool.Exec(context.Background(), `DROP TABLE IF EXISTS `+visible) })
|
|
|
|
handler := func(ctx context.Context, lease events.Lease, tx pgx.Tx) error {
|
|
_, err := tx.Exec(ctx, `INSERT INTO `+visible+` (event_id, n) VALUES ($1,1) ON CONFLICT (event_id) DO NOTHING`, lease.EventID)
|
|
return err
|
|
}
|
|
runWorkers := func(n int) {
|
|
var wg sync.WaitGroup
|
|
for i := 0; i < n; i++ {
|
|
wg.Add(1)
|
|
go func(i int) {
|
|
defer wg.Done()
|
|
d := events.NewDispatcher(pool, "dev", fmt.Sprintf("w%d", i))
|
|
d.Heartbeat = 50 * time.Millisecond
|
|
if err := d.Register(events.ConsumerNotify, handler); err != nil {
|
|
return
|
|
}
|
|
for k := 0; k < 12; k++ {
|
|
_, _ = d.RunOnce(ctx, events.ConsumerNotify)
|
|
}
|
|
}(i)
|
|
}
|
|
wg.Wait()
|
|
}
|
|
runWorkers(2)
|
|
runWorkers(4)
|
|
|
|
var leased int
|
|
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM event_deliveries WHERE consumer=$1 AND status='leased' AND event_id LIKE $2`, events.ConsumerNotify, "%"+suffix+"%").Scan(&leased); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if leased != 0 {
|
|
t.Fatalf("leftover leases=%d", leased)
|
|
}
|
|
var vis int
|
|
if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM `+visible).Scan(&vis); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if vis < 8 {
|
|
t.Fatalf("visible=%d", vis)
|
|
}
|
|
var owners int
|
|
if err := pool.QueryRow(ctx, `
|
|
SELECT COUNT(*) FROM (
|
|
SELECT event_id, consumer, attempt, COUNT(DISTINCT lease_owner) FROM event_delivery_attempts
|
|
WHERE event_id LIKE $1 GROUP BY 1,2,3 HAVING COUNT(DISTINCT lease_owner) > 1
|
|
) x
|
|
`, "%"+suffix+"%").Scan(&owners); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if owners != 0 {
|
|
t.Fatalf("double lease attempts=%d", owners)
|
|
}
|
|
}
|
|
|
|
func TestE0B_CrashAfterSideEffectBeforeAck_PG(t *testing.T) {
|
|
pool, ctx := e0bPool(t)
|
|
suffix := fmt.Sprintf("%d", time.Now().UnixNano())
|
|
eventID := "crash-" + suffix
|
|
seedEnvelope(t, ctx, pool, sampleEnv(eventID, "doc-crash-"+suffix, 1), []string{events.ConsumerNotify})
|
|
if _, err := pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS e0b_crash_visible (event_id TEXT PRIMARY KEY, n INT NOT NULL)`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() {
|
|
_, _ = pool.Exec(context.Background(), `DELETE FROM e0b_crash_visible WHERE event_id=$1`, eventID)
|
|
})
|
|
|
|
tx, err := pool.Begin(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
leases, err := events.ClaimLeases(ctx, tx, events.ConsumerNotify, "w1", 1, time.Second)
|
|
if err != nil || len(leases) != 1 {
|
|
_ = tx.Rollback(ctx)
|
|
t.Fatalf("claim1: %v %#v", err, leases)
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := pool.Exec(ctx, `INSERT INTO e0b_crash_visible (event_id, n) VALUES ($1,1) ON CONFLICT (event_id) DO NOTHING`, eventID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_, _ = pool.Exec(ctx, `UPDATE event_deliveries SET lease_until=now() - interval '2 seconds' WHERE event_id=$1 AND consumer=$2`, eventID, events.ConsumerNotify)
|
|
|
|
tx2, err := pool.Begin(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
second, err := events.ClaimLeases(ctx, tx2, events.ConsumerNotify, "w2", 1, events.DefaultLease)
|
|
if err != nil || len(second) != 1 {
|
|
_ = tx2.Rollback(ctx)
|
|
t.Fatalf("claim2: %v %#v", err, second)
|
|
}
|
|
if err := tx2.Commit(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ackOld, err := pool.Begin(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := events.AckLease(ctx, ackOld, leases[0]); err != events.ErrStaleFencing {
|
|
_ = ackOld.Rollback(ctx)
|
|
t.Fatalf("old ack: %v", err)
|
|
}
|
|
_ = ackOld.Rollback(ctx)
|
|
|
|
if _, err := pool.Exec(ctx, `INSERT INTO e0b_crash_visible (event_id, n) VALUES ($1,1) ON CONFLICT (event_id) DO NOTHING`, eventID); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ackNew, err := pool.Begin(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := events.AckLease(ctx, ackNew, second[0]); err != nil {
|
|
_ = ackNew.Rollback(ctx)
|
|
t.Fatal(err)
|
|
}
|
|
if err := ackNew.Commit(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var n int
|
|
if err := pool.QueryRow(ctx, `SELECT n FROM e0b_crash_visible WHERE event_id=$1`, eventID).Scan(&n); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if n != 1 {
|
|
t.Fatalf("visible n=%d", n)
|
|
}
|
|
}
|
|
|
|
func TestE0B_QuarantineNotClaimed_PG(t *testing.T) {
|
|
pool, ctx := e0bPool(t)
|
|
suffix := fmt.Sprintf("%d", time.Now().UnixNano())
|
|
env := sampleEnv("q-"+suffix, "doc-q-"+suffix, 1)
|
|
seedEnvelope(t, ctx, pool, env, []string{events.ConsumerNotify})
|
|
unit := uow.NewPG(pool)
|
|
row := events.ClassifyEnvelope(events.Envelope{EventID: env.EventID, EventName: env.EventName}, events.ConsumerNotify)
|
|
row.Classification = events.ClassificationQuarantine
|
|
row.Reason = "fixture"
|
|
if err := unit.Within(ctx, func(scope uow.TxScope) error {
|
|
return events.PersistInventory(uow.WithScope(ctx, scope), row, "ops", "e0b")
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
tx, err := pool.Begin(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer func() { _ = tx.Rollback(ctx) }()
|
|
got, err := events.ClaimLeases(ctx, tx, events.ConsumerNotify, "w", 8, events.DefaultLease)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, l := range got {
|
|
if l.EventID == env.EventID {
|
|
t.Fatalf("quarantined event claimed: %+v", l)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestE0B_ShadowDoesNotAck_PG(t *testing.T) {
|
|
pool, ctx := e0bPool(t)
|
|
suffix := fmt.Sprintf("%d", time.Now().UnixNano())
|
|
id := "sh-" + suffix
|
|
seedEnvelope(t, ctx, pool, sampleEnv(id, "doc-sh-"+suffix, 1), []string{events.ConsumerNotify})
|
|
if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeShadow); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
d := events.NewDispatcher(pool, "dev", "w-shadow")
|
|
n, err := d.RunOnce(ctx, events.ConsumerNotify)
|
|
if err != nil || n < 1 {
|
|
t.Fatalf("shadow n=%d err=%v", n, err)
|
|
}
|
|
var status string
|
|
if err := pool.QueryRow(ctx, `SELECT status FROM event_deliveries WHERE event_id=$1 AND consumer=$2`, id, events.ConsumerNotify).Scan(&status); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if status != "pending" && status != "leased" {
|
|
t.Fatalf("shadow must not ack, status=%s", status)
|
|
}
|
|
if err := events.SetMode(ctx, pool, "dev", events.ConsumerNotify, events.ModeOff); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func TestE2_WriterLiveModes_PG(t *testing.T) {
|
|
pool, ctx := e0bPool(t)
|
|
env := fmt.Sprintf("e2-%d", time.Now().UnixNano())
|
|
if events.WriterLive(ctx, pool, env, events.ConsumerNotify) {
|
|
t.Fatal("missing row must not be live")
|
|
}
|
|
if err := events.SetMode(ctx, pool, env, events.ConsumerNotify, events.ModeOff); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if events.WriterLive(ctx, pool, env, events.ConsumerNotify) {
|
|
t.Fatal("off must keep hook writers")
|
|
}
|
|
if err := events.SetMode(ctx, pool, env, events.ConsumerNotify, events.ModeShadow); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if events.WriterLive(ctx, pool, env, events.ConsumerNotify) {
|
|
t.Fatal("shadow must keep hook writers")
|
|
}
|
|
if err := events.SetMode(ctx, pool, env, events.ConsumerNotify, events.ModeLive); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if !events.WriterLive(ctx, pool, env, events.ConsumerNotify) {
|
|
t.Fatal("live must stop hook writers")
|
|
}
|
|
if err := events.SetMode(ctx, pool, env, events.ConsumerNotify, events.ModeOff); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|