From 99b95991f7495e83ec8028a2fe27f86a8a1b3298 Mon Sep 17 00:00:00 2001 From: Faks <1909645+Faks@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:33:07 +0300 Subject: [PATCH 1/3] perf(db): gate idle event-engine ticks with one statement per tick Measured on the deployed instance: 16753 statements in a clean 120 s idle window (140/s average, 218/s sampled), 17 core-seconds of CPU, zero rows returned. The cost was statement and transaction count, not query work: one poll and one MAX(id) read per consumer per tick, plus a transaction per claim. An idle tick now pays one gate statement that names only the consumers that can have work; a claim still runs inside a transaction when a claim happens. The scan still covers every active consumer, so deliveries exist for a handler registered later, and cursors of consumers that go inactive are still pruned. The dialect branches were extracted as claimQuery and reconcileConsumerQuery so they can be tested; the gate statement uses plain placeholders and no locking clause, so SQLite is unaffected. --- internal/events/engine.go | 148 +++++++++++++++++++++++++++++++------ internal/events/store.go | 149 +++++++++++++++++++++++++++++--------- 2 files changed, 239 insertions(+), 58 deletions(-) diff --git a/internal/events/engine.go b/internal/events/engine.go index 7f4793dd0..2e356f68b 100644 --- a/internal/events/engine.go +++ b/internal/events/engine.go @@ -115,6 +115,12 @@ type Engine struct { reconcileCursors map[string]int64 lastSafetyReconcile time.Time + // workKeys is the consumer set the last gate reported as having eligible + // deliveries, plus any consumer the same tick created deliveries for. It is + // guarded by mu and read by workers, so an idle engine claims nothing at all + // instead of probing every consumer once per tick. + workKeys []string + now func() time.Time jitter func(time.Duration) time.Duration // leaseRenewed is a test observation hook set before engine startup. @@ -281,7 +287,7 @@ func (e *Engine) reconcileLoop(ctx context.Context) { ticker := time.NewTicker(e.config.PollInterval) defer ticker.Stop() for { - if _, err := e.reconcile(ctx); err != nil && ctx.Err() == nil { + if _, err := e.tick(ctx); err != nil && ctx.Err() == nil { slog.Error("domain event reconciliation failed", "error", err) } select { @@ -293,39 +299,58 @@ func (e *Engine) reconcileLoop(ctx context.Context) { } } -// reconcile creates missing deliveries. Per consumer it normally scans only -// events past the in-memory cursor. Periodically it also re-checks events -// recorded within ReconcileLookback so an event whose inserting transaction -// committed after a cursor advanced past its (lower) id is still delivered, -// without rescanning the full retained history. -func (e *Engine) reconcile(ctx context.Context) (int64, error) { - keys, err := e.store.ActiveConsumerKeys(ctx) +// tick pays one gate statement per tick, then only does the work the gate proved +// exists: it creates deliveries for consumers whose cursor does not yet cover the +// highest visible event id, and records the consumers with eligible deliveries so +// workers claim for those only. An idle engine costs one statement per tick +// instead of one statement per consumer plus a transaction per claim. +// +// Cursors advance only after an exhaustive scan, as before: a batch-sized scan has +// not covered every event, so its cursor must not jump to the gate's highest id. +func (e *Engine) tick(ctx context.Context) (int64, error) { + since, safety := e.safetyReconcileWindow() + gate, err := e.store.Gate(ctx, e.lowestCursor(), e.now()) if err != nil { return 0, err } - for key := range e.reconcileCursors { - if !slices.Contains(keys, key) { - delete(e.reconcileCursors, key) - } + + // Scanning still covers every active consumer, as before, not only consumers + // with a handler in this process: deliveries for a handler registered later + // must still exist. The gate names the active set whenever any event is past + // the lowest cursor, so the only case the gate cannot name is a late-committed + // event whose id is below every cursor — which is what the safety scan is for. + // On a safety tick scan the consumers the engine already knows as well. + targets := gate.ScanKeys + if safety { + targets = unionKeys(targets, e.knownKeys()) } - since, safety := e.safetyReconcileWindow() + if len(gate.ScanKeys) > 0 { + // A named consumer means EXISTS was true, so ScanKeys is the full active + // set and cursors of consumers no longer active can be dropped. + e.pruneCursors(gate.ScanKeys) + } + e.setWorkKeys(gate.ScanKeys, gate.ClaimKeys) + var total int64 - for _, key := range keys { + for _, key := range targets { afterID := e.reconcileCursors[key] if safety { afterID = 0 + } else if afterID >= gate.MaxEventID { + continue } for { - created, maxSeen, err := e.store.reconcileConsumer(ctx, key, afterID, since, int64(e.config.ReconcileBatch)) + created, err := e.store.reconcileConsumer(ctx, key, afterID, since, int64(e.config.ReconcileBatch)) if err != nil { return total, err } total += created if created > 0 { + e.addWorkKey(key) e.signalWorkers() } if created < int64(e.config.ReconcileBatch) { - e.reconcileCursors[key] = maxSeen + e.reconcileCursors[key] = gate.MaxEventID break } } @@ -333,6 +358,53 @@ func (e *Engine) reconcile(ctx context.Context) (int64, error) { return total, nil } +// lowestCursor is the oldest uncovered point across the consumers the engine +// knows. If no event is past it, no consumer can have new work, and nothing is +// scanned. +func (e *Engine) lowestCursor() int64 { + e.mu.Lock() + defer e.mu.Unlock() + lowest := int64(0) + for _, cursor := range e.reconcileCursors { + if cursor < lowest { + lowest = cursor + } + } + return lowest +} + +func (e *Engine) knownKeys() []string { + e.mu.Lock() + defer e.mu.Unlock() + keys := make([]string, 0, len(e.reconcileCursors)) + for key := range e.reconcileCursors { + keys = append(keys, key) + } + slices.Sort(keys) + return keys +} + +func (e *Engine) pruneCursors(active []string) { + e.mu.Lock() + defer e.mu.Unlock() + for key := range e.reconcileCursors { + if !slices.Contains(active, key) { + delete(e.reconcileCursors, key) + } + } +} + +func unionKeys(a, b []string) []string { + out := slices.Clone(a) + for _, key := range b { + if !slices.Contains(out, key) { + out = append(out, key) + } + } + slices.Sort(out) + return out +} + // safetyReconcileWindow reports the lower recorded_at bound for the periodic // safety rescan. The first pass after startup scans from the (empty) cursor, // which is a full history backstop; the safety clock starts then. @@ -370,7 +442,7 @@ func (e *Engine) workerLoop(ctx context.Context) { } func (e *Engine) processOne(ctx context.Context) (bool, error) { - keys := e.handlerKeys() + keys := e.eligibleHandlerKeys() for _, consumerKey := range keys { delivery, err := e.store.Claim(ctx, consumerKey, e.owner, e.now(), e.config.LeaseDuration) if err != nil { @@ -464,19 +536,49 @@ func (e *Engine) retryDelay(attempt int) time.Duration { return delay + jitter } -func (e *Engine) handlerKeys() []string { +// eligibleHandlerKeys returns the consumers the last gate reported as having an +// eligible delivery, limited to consumers with a registered handler, in rotated +// order so a busy consumer cannot starve the others. Rotation is unchanged. +func (e *Engine) eligibleHandlerKeys() []string { e.mu.Lock() defer e.mu.Unlock() - keys := make([]string, 0, len(e.handlers)) - for key := range e.handlers { - keys = append(keys, key) + keys := make([]string, 0, len(e.workKeys)) + for _, key := range e.workKeys { + if _, exists := e.handlers[key]; exists { + keys = append(keys, key) + } } slices.Sort(keys) + return rotateKeys(keys, &e.nextHandler) +} + +func (e *Engine) setWorkKeys(scan, claim []string) { + e.mu.Lock() + defer e.mu.Unlock() + keys := slices.Clone(scan) + for _, key := range claim { + if !slices.Contains(keys, key) { + keys = append(keys, key) + } + } + slices.Sort(keys) + e.workKeys = keys +} + +func (e *Engine) addWorkKey(key string) { + e.mu.Lock() + defer e.mu.Unlock() + if !slices.Contains(e.workKeys, key) { + e.workKeys = append(e.workKeys, key) + } +} + +func rotateKeys(keys []string, counter *int) []string { if len(keys) == 0 { return keys } - start := e.nextHandler % len(keys) - e.nextHandler = (start + 1) % len(keys) + start := *counter % len(keys) + *counter = (*counter + 1) % len(keys) if start == 0 { return keys } diff --git a/internal/events/store.go b/internal/events/store.go index e91333570..193f6ea10 100644 --- a/internal/events/store.go +++ b/internal/events/store.go @@ -373,7 +373,7 @@ func (s *Store) Reconcile(ctx context.Context, limit int) (int64, error) { if remaining <= 0 { break } - created, _, err := s.reconcileConsumer(ctx, key, 0, time.Time{}, remaining) + created, err := s.reconcileConsumer(ctx, key, 0, time.Time{}, remaining) if err != nil { return total, err } @@ -415,15 +415,20 @@ func (s *Store) ActiveConsumerKeys(ctx context.Context) ([]string, error) { // created deliveries and the highest event id visible once the scan finished, // which callers may store as the next afterID only after a short (exhaustive) // scan. +// reconcileConsumer creates pending deliveries for one active consumer for +// events with an id greater than afterID, bounded by the lookback window when +// one is given. The caller advances the cursor from the gate's highest visible +// event id, so this statement no longer needs a second MAX(id) read per consumer +// per tick. func (s *Store) reconcileConsumer( ctx context.Context, consumerKey string, afterID int64, since time.Time, limit int64, -) (created, maxSeen int64, err error) { +) (int64, error) { if limit <= 0 { - return 0, 0, errors.New("reconcile limit must be positive") + return 0, errors.New("reconcile limit must be positive") } args := []any{afterID, consumerKey} if !since.IsZero() { @@ -433,24 +438,90 @@ func (s *Store) reconcileConsumer( args = append(args, limit) result, err := s.db.ExecWriteContext(ctx, query, args...) if err != nil { - return 0, 0, fmt.Errorf("reconcile domain event deliveries for %q: %w", consumerKey, err) + return 0, fmt.Errorf("reconcile domain event deliveries for %q: %w", consumerKey, err) } - created, err = result.RowsAffected() + created, err := result.RowsAffected() if err != nil { - return 0, 0, fmt.Errorf("read reconciled delivery count for %q: %w", consumerKey, err) + return 0, fmt.Errorf("read reconciled delivery count for %q: %w", consumerKey, err) } - if err := s.db.QueryRowContext(ctx, ` - SELECT COALESCE(MAX(id), 0) FROM domain_events - `).Scan(&maxSeen); err != nil { - return 0, 0, fmt.Errorf("read max domain event id for %q reconcile: %w", consumerKey, err) - } - return created, maxSeen, nil + return created, nil } // reconcileConsumerQuery builds the delivery-insertion statement. withSince // adds the recorded_at lower bound the bounded safety rescan relies on so the // planner can use idx_domain_events_recorded_at instead of scanning the full // retained history (WI-1770). +// GateResult is the answer of the once-per-tick gate statement. +type GateResult struct { + // ScanKeys are active consumers that may have events past the lowest cursor + // the engine holds. Consumers the gate does not name cannot have new work. + ScanKeys []string + // ClaimKeys are consumers with a delivery already eligible for claim. + ClaimKeys []string + // MaxEventID is the highest event id the reconciler can see. A consumer + // whose cursor already equals it is covered and is not scanned again. + MaxEventID int64 +} + +// gateSQL is the once-per-tick statement. It is dialect-neutral: no array +// parameters, no locking clause, plain ? placeholders. +const gateSQL = ` + SELECT 1 AS gate_kind, CAST(NULL AS TEXT) AS consumer_key, CAST(COALESCE(MAX(id), 0) AS INTEGER) AS event_id + FROM domain_events + UNION ALL + SELECT 2, consumer.consumer_key, CAST(NULL AS INTEGER) + FROM domain_event_consumers consumer + WHERE consumer.is_active = true + AND EXISTS (SELECT 1 FROM domain_events event WHERE event.id > ?) + UNION ALL + SELECT 3, delivery.consumer_key, CAST(NULL AS INTEGER) + FROM domain_event_deliveries delivery + JOIN domain_event_consumers consumer ON consumer.consumer_key = delivery.consumer_key + WHERE consumer.is_active = true + AND ( + (delivery.state IN ('pending', 'retry') AND delivery.next_attempt_at <= ?) + OR (delivery.state = 'leased' AND delivery.lease_expires_at <= ?) + ) + GROUP BY delivery.consumer_key +` + +// Gate answers which consumers have work in one statement. Before this, an +// idle engine probed every consumer and opened a transaction per probe, so an +// empty queue still cost a statement per consumer per tick, plus one read per +// consumer for MAX(id). The gate names only consumers that can have work: an +// idle or inactive consumer is named by nothing, so the loop skips it. Claim +// stays inside a transaction whenever a claim happens. +func (s *Store) Gate(ctx context.Context, minCursor int64, now time.Time) (GateResult, error) { + var result GateResult + rows, err := s.db.QueryContext(ctx, gateSQL, minCursor, now.UTC(), now.UTC()) + if err != nil { + return result, fmt.Errorf("gate domain event work: %w", err) + } + defer func() { _ = rows.Close() }() + for rows.Next() { + var kind int + var consumerKey sql.NullString + var eventID sql.NullInt64 + if err := rows.Scan(&kind, &consumerKey, &eventID); err != nil { + return result, fmt.Errorf("scan domain event gate row: %w", err) + } + switch kind { + case 1: + result.MaxEventID = eventID.Int64 + case 2: + result.ScanKeys = append(result.ScanKeys, consumerKey.String) + default: + result.ClaimKeys = append(result.ClaimKeys, consumerKey.String) + } + } + if err := rows.Err(); err != nil { + return result, fmt.Errorf("iterate domain event gate rows: %w", err) + } + slices.Sort(result.ScanKeys) + slices.Sort(result.ClaimKeys) + return result, nil +} + func reconcileConsumerQuery(driver string, withSince bool) string { recordedAtFilter := "" orderBy := "ORDER BY e.id" @@ -497,27 +568,10 @@ func reconcileConsumerQuery(driver string, withSince bool) string { return query + " ON CONFLICT (event_id, consumer_key) DO NOTHING" } -// Claim leases the next causally eligible delivery for one consumer. -func (s *Store) Claim( - ctx context.Context, - consumerKey string, - owner string, - now time.Time, - leaseDuration time.Duration, -) (*Delivery, error) { - if strings.TrimSpace(consumerKey) == "" || strings.TrimSpace(owner) == "" { - return nil, errors.New("consumer key and lease owner are required") - } - if leaseDuration <= 0 { - return nil, errors.New("lease duration must be positive") - } - - tx, err := s.db.BeginTx(ctx, nil) - if err != nil { - return nil, fmt.Errorf("begin domain event claim: %w", err) - } - defer func() { _ = tx.Rollback() }() - +// claimQuery builds the claim SELECT for one dialect. FOR UPDATE OF delivery +// SKIP LOCKED is PostgreSQL-only; SQLite has no locking clause, so this branch +// must stay. The claim still runs inside a transaction in both dialects. +func claimQuery(driver string) string { query := ` SELECT delivery.event_id, delivery.attempt_count FROM domain_event_deliveries delivery @@ -543,10 +597,35 @@ func (s *Store) Claim( ORDER BY delivery.next_attempt_at, event.id LIMIT 1 ` - if s.db.GetDriverName() == "postgres" { - query += " FOR UPDATE OF delivery SKIP LOCKED" + if driver == "postgres" { + return query + " FOR UPDATE OF delivery SKIP LOCKED" + } + return query +} + +// Claim leases the next causally eligible delivery for one consumer. +func (s *Store) Claim( + ctx context.Context, + consumerKey string, + owner string, + now time.Time, + leaseDuration time.Duration, +) (*Delivery, error) { + if strings.TrimSpace(consumerKey) == "" || strings.TrimSpace(owner) == "" { + return nil, errors.New("consumer key and lease owner are required") + } + if leaseDuration <= 0 { + return nil, errors.New("lease duration must be positive") } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return nil, fmt.Errorf("begin domain event claim: %w", err) + } + defer func() { _ = tx.Rollback() }() + + query := claimQuery(s.db.GetDriverName()) + var eventID int64 var previousAttempts int if err := tx.QueryRowContext(ctx, query, consumerKey, now.UTC(), now.UTC()).Scan(&eventID, &previousAttempts); err != nil { From 15f724b3aaaed71c77d6b48f44607c2260a1244f Mon Sep 17 00:00:00 2001 From: Faks <1909645+Faks@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:52:40 +0300 Subject: [PATCH 2/3] fix(db): put the GateResult doc comment on its type The reconcileConsumerQuery doc comment was left above the type, so staticcheck read the whole block as the type comment: ST1021 at internal/events/store.go:450. GitHub CI Lint job failed on it. --- internal/events/store.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/internal/events/store.go b/internal/events/store.go index 193f6ea10..0604b8c2b 100644 --- a/internal/events/store.go +++ b/internal/events/store.go @@ -447,10 +447,6 @@ func (s *Store) reconcileConsumer( return created, nil } -// reconcileConsumerQuery builds the delivery-insertion statement. withSince -// adds the recorded_at lower bound the bounded safety rescan relies on so the -// planner can use idx_domain_events_recorded_at instead of scanning the full -// retained history (WI-1770). // GateResult is the answer of the once-per-tick gate statement. type GateResult struct { // ScanKeys are active consumers that may have events past the lowest cursor @@ -522,6 +518,10 @@ func (s *Store) Gate(ctx context.Context, minCursor int64, now time.Time) (GateR return result, nil } +// reconcileConsumerQuery builds the delivery-insertion statement. withSince +// adds the recorded_at lower bound the bounded safety rescan relies on so the +// planner can use idx_domain_events_recorded_at instead of scanning the full +// retained history (WI-1770). func reconcileConsumerQuery(driver string, withSince bool) string { recordedAtFilter := "" orderBy := "ORDER BY e.id" From e70b56e0988671ca4a4f68624b88d98ddea95858 Mon Sep 17 00:00:00 2001 From: Faks <1909645+Faks@users.noreply.github.com> Date: Tue, 20 Oct 2026 19:05:18 +0300 Subject: [PATCH 3/3] perf(events): probe the gate per consumer instead of scanning the queue --- internal/events/store.go | 32 ++++++++++++++++++++++++++------ 1 file changed, 26 insertions(+), 6 deletions(-) diff --git a/internal/events/store.go b/internal/events/store.go index 0604b8c2b..2c64c228c 100644 --- a/internal/events/store.go +++ b/internal/events/store.go @@ -461,6 +461,18 @@ type GateResult struct { // gateSQL is the once-per-tick statement. It is dialect-neutral: no array // parameters, no locking clause, plain ? placeholders. +// +// The eligibility branch probes deliveries per active consumer instead of +// scanning the queue. Each EXISTS stops at the first matching row, so the cost +// is bounded by the number of consumers, not by how many deliveries are due. +// Asking the same question as one predicate over the queue degrades with the +// due backlog instead: at 10M due rows SQLite took 6.7 s and split two ways +// still 3.0 s, while the probes stayed at 0.4 ms. +// +// The pending/retry probe rides idx_domain_event_deliveries_runnable +// (consumer_key, state, next_attempt_at); the lease probe rides +// idx_domain_event_deliveries_lease. Both indexes already exist, so the gate +// needs no schema change. const gateSQL = ` SELECT 1 AS gate_kind, CAST(NULL AS TEXT) AS consumer_key, CAST(COALESCE(MAX(id), 0) AS INTEGER) AS event_id FROM domain_events @@ -470,15 +482,23 @@ const gateSQL = ` WHERE consumer.is_active = true AND EXISTS (SELECT 1 FROM domain_events event WHERE event.id > ?) UNION ALL - SELECT 3, delivery.consumer_key, CAST(NULL AS INTEGER) - FROM domain_event_deliveries delivery - JOIN domain_event_consumers consumer ON consumer.consumer_key = delivery.consumer_key + SELECT 3, consumer.consumer_key, CAST(NULL AS INTEGER) + FROM domain_event_consumers consumer WHERE consumer.is_active = true AND ( - (delivery.state IN ('pending', 'retry') AND delivery.next_attempt_at <= ?) - OR (delivery.state = 'leased' AND delivery.lease_expires_at <= ?) + EXISTS ( + SELECT 1 FROM domain_event_deliveries delivery + WHERE delivery.consumer_key = consumer.consumer_key + AND delivery.state IN ('pending', 'retry') + AND delivery.next_attempt_at <= ? + ) + OR EXISTS ( + SELECT 1 FROM domain_event_deliveries delivery + WHERE delivery.consumer_key = consumer.consumer_key + AND delivery.state = 'leased' + AND delivery.lease_expires_at <= ? + ) ) - GROUP BY delivery.consumer_key ` // Gate answers which consumers have work in one statement. Before this, an