Skip to content

Async commit for batch deletes #2236

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
DylanTinianov merged 3 commits into from
Jul 10, 2026
Merged

Async commit for batch deletes #2236

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
292b778
Async commit for batch deletes
DylanTinianov Jul 9, 2026
5c23d17
Add comment
DylanTinianov Jul 10, 2026
c564653
Update store.go
DylanTinianov Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 18 additions & 3 deletions pkg/durableemitter/store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
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 {
Comment thread
DylanTinianov marked this conversation as resolved.
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) {
Expand Down
Loading