From 292b7780fc1248444d121e82755e176586e58431 Mon Sep 17 00:00:00 2001 From: Dylan Tinianov Date: Thu, 9 Jul 2026 13:42:23 -0400 Subject: [PATCH 1/3] Async commit for batch deletes --- pkg/durableemitter/store.go | 21 ++++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/pkg/durableemitter/store.go b/pkg/durableemitter/store.go index 5e9d986f91..26b8097df7 100644 --- a/pkg/durableemitter/store.go +++ b/pkg/durableemitter/store.go @@ -79,12 +79,27 @@ DELETE FROM ` + chipDurableEventsTable + ` WHERE id IN ( WHERE id = ANY($1) FOR UPDATE SKIP LOCKED )` - res, err := s.ds.ExecContext(ctx, q, pq.Array(ids)) + // Delete durability is not required under delete-on-delivery: a delete commit + // lost to a crash just leaves the row to be re-delivered (idempotent) or reclaimed + // by the expiry loop. We can relax synchronous_commit for the delete transaction, + // which removes the commit WAL-flush wait (IO:XactSync) which dominates at high + // delete rates, especially on Aurora where commit blocks on a storage-quorum ack. + var deleted int64 + err := sqlutil.TransactDataSource(ctx, s.ds, nil, func(tx sqlutil.DataSource) error { + if _, err := tx.ExecContext(ctx, "SET LOCAL synchronous_commit = off"); err != nil { + return fmt.Errorf("failed to set synchronous_commit=off: %w", err) + } + res, err := tx.ExecContext(ctx, q, pq.Array(ids)) + if err != nil { + return err + } + deleted, _ = res.RowsAffected() + return nil + }) if err != nil { return 0, fmt.Errorf("failed to batch delete delivered chip durable events: %w", err) } - n, _ := res.RowsAffected() - return n, nil + return deleted, nil } func (s *PgDurableEventStore) ListPending(ctx context.Context, createdBefore, afterCreatedAt time.Time, afterID int64, limit int) ([]DurableEvent, error) { From 5c23d17cbde6c838b2cc5634645aaa356269baa5 Mon Sep 17 00:00:00 2001 From: Dylan Tinianov Date: Fri, 10 Jul 2026 11:35:36 -0400 Subject: [PATCH 2/3] Add comment --- pkg/durableemitter/store.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pkg/durableemitter/store.go b/pkg/durableemitter/store.go index 26b8097df7..30920a9fbe 100644 --- a/pkg/durableemitter/store.go +++ b/pkg/durableemitter/store.go @@ -73,6 +73,9 @@ func (s *PgDurableEventStore) BatchDelete(ctx context.Context, ids []int64) (int if len(ids) == 0 { return 0, nil } + // FOR UPDATE SKIP LOCKED avoids deadlocks (40P01) when a backlog re-delivers + // the same id in two concurrent delete batches: each batch skips rows the + // other already locked instead of waiting. Safe since deletes are idempotent. const q = ` DELETE FROM ` + chipDurableEventsTable + ` WHERE id IN ( SELECT id FROM ` + chipDurableEventsTable + ` From c564653427850ff037e24a09135dad6f68e6aa6e Mon Sep 17 00:00:00 2001 From: Dylan Tinianov Date: Fri, 10 Jul 2026 12:27:18 -0400 Subject: [PATCH 3/3] Update store.go --- pkg/durableemitter/store.go | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/pkg/durableemitter/store.go b/pkg/durableemitter/store.go index 30920a9fbe..6f4e63b7d1 100644 --- a/pkg/durableemitter/store.go +++ b/pkg/durableemitter/store.go @@ -73,9 +73,6 @@ func (s *PgDurableEventStore) BatchDelete(ctx context.Context, ids []int64) (int if len(ids) == 0 { return 0, nil } - // FOR UPDATE SKIP LOCKED avoids deadlocks (40P01) when a backlog re-delivers - // the same id in two concurrent delete batches: each batch skips rows the - // other already locked instead of waiting. Safe since deletes are idempotent. const q = ` DELETE FROM ` + chipDurableEventsTable + ` WHERE id IN ( SELECT id FROM ` + chipDurableEventsTable + ` @@ -86,7 +83,7 @@ DELETE FROM ` + chipDurableEventsTable + ` WHERE id IN ( // lost to a crash just leaves the row to be re-delivered (idempotent) or reclaimed // by the expiry loop. We can relax synchronous_commit for the delete transaction, // which removes the commit WAL-flush wait (IO:XactSync) which dominates at high - // delete rates, especially on Aurora where commit blocks on a storage-quorum ack. + // delete rates. var deleted int64 err := sqlutil.TransactDataSource(ctx, s.ds, nil, func(tx sqlutil.DataSource) error { if _, err := tx.ExecContext(ctx, "SET LOCAL synchronous_commit = off"); err != nil {