diff --git a/internal/events/gate_test.go b/internal/events/gate_test.go new file mode 100644 index 0000000..3ffdc6e --- /dev/null +++ b/internal/events/gate_test.go @@ -0,0 +1,342 @@ +//go:build test + +package events + +import ( + "context" + "database/sql" + "slices" + "strings" + "testing" + "time" + + "windshift/internal/database" + "windshift/internal/testutils" +) + +// countingDatabase counts statement executions so the idle-tick budget can be +// asserted. It wraps the test database, which also proves the gate works on the +// engine the suite is running against. +type countingDatabase struct { + database.Database + statements int + writes int + transactions int + queries []string +} + +func (db *countingDatabase) record(kind, query string) { + db.statements++ + if kind == "write" { + db.writes++ + } + db.queries = append(db.queries, kind+" "+query) +} + +func (db *countingDatabase) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) { + db.record("read", query) + return db.Database.QueryContext(ctx, query, args...) +} + +func (db *countingDatabase) QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row { + db.record("read", query) + return db.Database.QueryRowContext(ctx, query, args...) +} + +func (db *countingDatabase) ExecWriteContext(ctx context.Context, query string, args ...any) (sql.Result, error) { + db.record("write", query) + return db.Database.ExecWriteContext(ctx, query, args...) +} + +func (db *countingDatabase) BeginTx(ctx context.Context, opts *sql.TxOptions) (database.Tx, error) { + db.transactions++ + return db.Database.BeginTx(ctx, opts) +} + +func (db *countingDatabase) reset() { + db.statements = 0 + db.writes = 0 + db.transactions = 0 + db.queries = nil +} + +func gateConfig() Config { + config := DefaultConfig() + config.WorkerCount = 1 + config.RetentionInterval = 0 + return config +} + +// An idle tick must cost one statement and open no transaction. Before the gate +// an idle tick cost one statement per consumer plus one transaction per claim, +// which is what burned CPU on an empty queue in production. +func TestIdleTickCostsOneStatementAndNoTransaction(t *testing.T) { + tdb := testutils.CreateTestDB(t, true) + defer tdb.Close() + counter := &countingDatabase{Database: tdb} + engine := NewEngine(counter, gateConfig()) + store := NewStore(counter) + if err := engine.RegisterHandler("gate.consumer", HandlerFunc(func(_ context.Context, _ Event) error { return nil })); err != nil { + t.Fatal(err) + } + configureConsumer(t, store, "gate.consumer", "item.created") + appendEvent(t, store, testEvent("item", "1", "item.created")) + + // The first tick is the history backstop, so it still runs one scan per + // consumer the gate names. Nothing is scanned for a consumer that does not + // exist or has no events past its cursor. + if _, err := engine.tick(context.Background()); err != nil { + t.Fatal(err) + } + if counter.statements != 2 { + t.Fatalf("first tick executed %d statements, want gate plus one scan: %v", counter.statements, counter.queries) + } + if counter.writes != 1 { + t.Fatalf("first tick executed %d writes, want the reconcile insert", counter.writes) + } + + counter.reset() + if _, err := engine.tick(context.Background()); err != nil { + t.Fatal(err) + } + if counter.statements != 1 { + t.Fatalf("idle tick executed %d statements, want 1 gate statement: %v", counter.statements, counter.queries) + } + if counter.writes != 0 { + t.Fatalf("idle tick executed %d writes, want 0", counter.writes) + } + if counter.transactions != 0 { + t.Fatalf("idle tick opened %d transactions, want 0", counter.transactions) + } +} + +// A worker must not open a transaction when the gate reported no work. +func TestIdleProcessOneClaimsNothing(t *testing.T) { + tdb := testutils.CreateTestDB(t, true) + defer tdb.Close() + counter := &countingDatabase{Database: tdb} + engine := NewEngine(counter, gateConfig()) + if err := engine.RegisterHandler("gate.consumer", HandlerFunc(func(_ context.Context, _ Event) error { return nil })); err != nil { + t.Fatal(err) + } + if _, err := engine.tick(context.Background()); err != nil { + t.Fatal(err) + } + counter.reset() + worked, err := engine.processOne(context.Background()) + if err != nil { + t.Fatal(err) + } + if worked { + t.Fatal("processOne claimed work the gate did not report") + } + if counter.statements != 0 { + t.Fatalf("processOne executed %d statements with no reported work, want 0", counter.statements) + } +} + +// Work found by the gate is reconciled without the per-consumer MAX(id) read, +// and the cursor advances from the gate's highest visible event id. +func TestGateFindsWorkWithoutPerConsumerMaxQuery(t *testing.T) { + tdb := testutils.CreateTestDB(t, true) + defer tdb.Close() + counter := &countingDatabase{Database: tdb} + engine := NewEngine(counter, gateConfig()) + store := NewStore(counter) + handled := 0 + if err := engine.RegisterHandler("gate.consumer", HandlerFunc(func(_ context.Context, _ Event) error { + handled++ + return nil + })); err != nil { + t.Fatal(err) + } + if err := store.ConfigureConsumer(context.Background(), Consumer{ + Key: "gate.consumer", + HandlerVersion: 1, + Active: true, + StartEventID: 1, + EventTypes: []string{"item.created"}, + }); err != nil { + t.Fatal(err) + } + event, err := store.AppendStandalone(context.Background(), testEvent("item", "42", "item.created")) + if err != nil { + t.Fatal(err) + } + + counter.reset() + created, err := engine.tick(context.Background()) + if err != nil { + t.Fatal(err) + } + if created != 1 { + t.Fatalf("tick created %d deliveries, want 1", created) + } + // Gate plus the scan. The per-consumer MAX(id) read the gate replaced must + // not run, so exactly two statements are allowed. + if counter.statements != 2 { + t.Fatalf("tick executed %d statements, want gate plus scan: %v", counter.statements, counter.queries) + } + if engine.reconcileCursors["gate.consumer"] != event.ID { + t.Fatalf("cursor = %d, want event id %d", engine.reconcileCursors["gate.consumer"], event.ID) + } + if !strings.Contains(strings.Join(engine.workKeys, ","), "gate.consumer") { + t.Fatalf("gate did not report the consumer with work: %v", engine.workKeys) + } + + worked, err := engine.processOne(context.Background()) + if err != nil { + t.Fatal(err) + } + if !worked || handled != 1 { + t.Fatalf("processOne handled %d events, want 1", handled) + } + + // With the queue empty the engine returns to the one-statement budget. + counter.reset() + if _, err := engine.tick(context.Background()); err != nil { + t.Fatal(err) + } + if counter.statements != 1 { + t.Fatalf("tick after work executed %d statements, want 1 gate statement: %v", counter.statements, counter.queries) + } +} + +// An inactive consumer must not be scanned on idle ticks. Only the periodic full +// scan touches it, which matches the ActiveConsumerKeys behaviour the gate +// replaced. +func TestInactiveConsumerIsNotScannedOnIdleTicks(t *testing.T) { + tdb := testutils.CreateTestDB(t, true) + defer tdb.Close() + counter := &countingDatabase{Database: tdb} + engine := NewEngine(counter, gateConfig()) + if err := engine.RegisterHandler("inactive.consumer", HandlerFunc(func(_ context.Context, _ Event) error { return nil })); err != nil { + t.Fatal(err) + } + if err := NewStore(counter).ConfigureConsumer(context.Background(), Consumer{ + Key: "inactive.consumer", + HandlerVersion: 1, + Active: false, + StartEventID: 1, + EventTypes: []string{"item.created"}, + }); err != nil { + t.Fatal(err) + } + if _, err := engine.tick(context.Background()); err != nil { + t.Fatal(err) + } + counter.reset() + if _, err := engine.tick(context.Background()); err != nil { + t.Fatal(err) + } + if counter.statements != 1 { + t.Fatalf("idle tick executed %d statements for an inactive consumer, want 1: %v", counter.statements, counter.queries) + } + if counter.writes != 0 { + t.Fatalf("idle tick wrote %d statements for an inactive consumer, want 0", counter.writes) + } +} + +// The scan still covers every active consumer, not only consumers with a handler +// in this process: deliveries must exist for a handler registered later. The gate +// could have been narrowed to registered handlers, which would have broken this, +// so the behavior is pinned here. +func TestActiveConsumerWithoutHandlerIsStillScanned(t *testing.T) { + tdb := testutils.CreateTestDB(t, true) + defer tdb.Close() + store := NewStore(tdb) + configureConsumer(t, store, "unhandled.consumer", "item.created") + appendEvent(t, store, testEvent("item", "7", "item.created")) + + engine := NewEngine(tdb, gateConfig()) + created, err := engine.tick(context.Background()) + if err != nil { + t.Fatal(err) + } + if created != 1 { + t.Fatalf("tick created %d deliveries for an active consumer with no handler, want 1", created) + } + if keys := engine.eligibleHandlerKeys(); len(keys) != 0 { + t.Fatalf("eligibleHandlerKeys() = %v, want none before a handler is registered", keys) + } + + // A handler registered later finds the deliveries the engine already created. + if err := engine.RegisterHandler("unhandled.consumer", HandlerFunc(func(_ context.Context, _ Event) error { + return nil + })); err != nil { + t.Fatal(err) + } + if keys := engine.eligibleHandlerKeys(); len(keys) != 1 || keys[0] != "unhandled.consumer" { + t.Fatalf("eligibleHandlerKeys() = %v, want the consumer the gate named", keys) + } +} + +// Eligibility covers two ranges: pending/retry by next_attempt_at and leased by +// lease_expires_at. The gate runs them as separate indexed branches, so a +// regression that drops one branch leaves expired leases unclaimed until the +// safety reconcile. Pin both branches, and pin that a delivery scheduled for +// the future is not named. +func TestGateNamesDueStatesOnly(t *testing.T) { + tdb := testutils.CreateTestDB(t, true) + defer tdb.Close() + store := NewStore(tdb) + now := time.Now().UTC().Truncate(time.Second) + + for _, key := range []string{"due.lease", "due.pending", "future.pending"} { + configureConsumer(t, store, key, "item.created") + } + event := appendEvent(t, store, testEvent("item", "9", "item.created")) + + insert := func(key, state string, nextAttempt time.Time, leaseExpiry any) { + t.Helper() + if _, err := tdb.ExecWrite( + `INSERT INTO domain_event_deliveries (event_id, consumer_key, state, next_attempt_at, lease_expires_at) + VALUES (?, ?, ?, ?, ?)`, + event.ID, key, state, nextAttempt, leaseExpiry); err != nil { + t.Fatalf("insert %s delivery for %s: %v", state, key, err) + } + } + insert("due.lease", "leased", now, now.Add(-time.Minute)) + insert("due.pending", "pending", now.Add(-time.Minute), nil) + insert("future.pending", "pending", now.Add(time.Hour), nil) + + result, err := store.Gate(context.Background(), event.ID+1, now) + if err != nil { + t.Fatalf("Gate() error = %v", err) + } + slices.Sort(result.ClaimKeys) + if want := []string{"due.lease", "due.pending"}; !slices.Equal(result.ClaimKeys, want) { + t.Fatalf("ClaimKeys = %v, want %v", result.ClaimKeys, want) + } + if len(result.ScanKeys) != 0 { + t.Fatalf("ScanKeys = %v, want none when every cursor is current", result.ScanKeys) + } +} + +// The gate must stay dialect-neutral: no array parameters, no locking clause, +// plain placeholders rewritten by the database layer. SQLite has no locking +// clause and no ON CONFLICT DO NOTHING, so both branches must survive. +func TestGateQueryIsDialectNeutral(t *testing.T) { + if strings.Contains(gateSQL, "ANY(") || strings.Contains(gateSQL, "FOR UPDATE") { + t.Fatalf("gate statement is not dialect-neutral: %s", gateSQL) + } + if !strings.Contains(gateSQL, "?") || strings.Contains(gateSQL, "$1") { + t.Fatalf("gate statement must use plain placeholders: %s", gateSQL) + } +} + +func TestDialectBranchesStaySeparated(t *testing.T) { + if strings.Contains(claimQuery("sqlite"), "FOR UPDATE") { + t.Fatalf("SQLite claim must not carry a locking clause: %s", claimQuery("sqlite")) + } + if !strings.Contains(claimQuery("postgres"), "FOR UPDATE OF delivery SKIP LOCKED") { + t.Fatalf("PostgreSQL claim lost its locking clause: %s", claimQuery("postgres")) + } + if !strings.Contains(reconcileConsumerQuery("sqlite", false), "INSERT OR IGNORE") { + t.Fatalf("SQLite reconcile lost INSERT OR IGNORE: %s", reconcileConsumerQuery("sqlite", false)) + } + if strings.Contains(reconcileConsumerQuery("sqlite", false), "ON CONFLICT") || + !strings.Contains(reconcileConsumerQuery("postgres", true), "ON CONFLICT (event_id, consumer_key) DO NOTHING") { + t.Fatalf("reconcile dialect branch is wrong: sqlite=%s postgres=%s", reconcileConsumerQuery("sqlite", false), reconcileConsumerQuery("postgres", true)) + } +} diff --git a/internal/events/store_test.go b/internal/events/store_test.go index 6ea6ad6..4268f57 100644 --- a/internal/events/store_test.go +++ b/internal/events/store_test.go @@ -251,7 +251,7 @@ func TestEngineRenewsOwnershipWhileHandlerIsStillRunning(t *testing.T) { LeaseDuration: 300 * time.Millisecond, HandlerTimeout: 200 * time.Millisecond, ReconcileBatch: 10, MaxAttempts: 2, BaseRetryDelay: time.Millisecond, MaxRetryDelay: time.Second, - FullReconcileInterval: time.Hour, + SafetyReconcileInterval: time.Hour, ReconcileLookback: time.Hour, } engine := NewEngine(tdb.DB, config) configureConsumer(t, engine.Store(), "actions.items", "item.changed") @@ -275,6 +275,11 @@ func TestEngineRenewsOwnershipWhileHandlerIsStillRunning(t *testing.T) { } done := make(chan error, 1) + // processOne claims only for consumers the gate named, so the gate must run + // before a worker can claim the delivery this test created. + if _, err := engine.tick(ctx); err != nil { + t.Fatalf("tick() error = %v", err) + } go func() { _, err := engine.processOne(ctx) done <- err @@ -470,6 +475,11 @@ func TestEngineProcessesRegisteredConsumerAndReportsStats(t *testing.T) { t.Fatalf("RegisterHandler() error = %v", err) } engine.now = func() time.Time { return time.Now().UTC().Add(time.Second) } + // The gate names the consumers that have work before a worker claims, so a + // tick must run first. This is the order the engine actually runs in. + if _, err := engine.tick(ctx); err != nil { + t.Fatalf("tick() error = %v", err) + } worked, err := engine.processOne(ctx) if err != nil { t.Fatalf("processOne() error = %v", err) @@ -509,6 +519,11 @@ func TestEngineRotatesConsumerClaimPriority(t *testing.T) { } } engine.now = func() time.Time { return time.Now().UTC().Add(time.Second) } + // Workers claim only for consumers the gate named, so the gate must run + // before the rotation is observed. + if _, err := engine.tick(ctx); err != nil { + t.Fatalf("tick() error = %v", err) + } for range 2 { worked, err := engine.processOne(ctx) @@ -543,7 +558,7 @@ func TestEngineShutdownCanBeAwaitedAgainAfterTimeout(t *testing.T) { LeaseDuration: time.Second, HandlerTimeout: 500 * time.Millisecond, ReconcileBatch: 10, MaxAttempts: 2, BaseRetryDelay: time.Millisecond, MaxRetryDelay: time.Second, - FullReconcileInterval: time.Hour, + SafetyReconcileInterval: time.Hour, ReconcileLookback: time.Hour, } engine := NewEngine(tdb.DB, config) configureConsumer(t, engine.Store(), "blocking", "item.changed")