Add a statistics pane to the webhook page (closes #368)

Each webhook's event database keeps running totals: one row for its
events, with when the newest arrived, and one row per target for its
deliveries, delivered and failed, each with what retention removed.
Every write to them shares the transaction of the rows it counts, and
a delivery already delivered or failed is not settled again.
Deliveries get a finished_at column; it and target_id end the status
index, so each target's deliveries finished in a window come from one
index-range query grouped by target. Retention deletes 1000 expired
events per transaction, pausing 200 ms between them so other writers
get in, and stops between them on shutdown. The pane is its own
template, its figures in tables.

The schema changes in place with nothing back-filled, so an existing
database must be recreated.

Model: opus-5-5

internal/database/event_tier_indexes_test.go

@@ -93,11 +93,11 @@ func TestEventTierQueriesUseTheirIndexes(t *testing.T) {
 		deliveries []database.Delivery
 		results    []database.DeliveryResult
 		depths     []struct{ Depth int }
+		removed    []database.TargetTotals
 	)
 
 	byStatus := "idx_deliveries_status (status=? AND deleted_at=?)"
 	byEvent := "idx_deliveries_event_id (event_id=? AND deleted_at=?)"
-	byAge := "idx_events_deleted_at_created_at (deleted_at=? AND created_at<?)"
 
 	// The delivery engine: recovery and the retry sweep, the sweep for
 	// stranded pending deliveries, and the queue depth count.
@@ -123,25 +123,80 @@ func TestEventTierQueriesUseTheirIndexes(t *testing.T) {
 		Order("attempt_num ASC").Find(&results),
 		"idx_delivery_results_delivery_id (delivery_id=? AND deleted_at=?)")
 
-	// Retention's three deletes (reapExpired), whose subqueries are built
-	// afresh for each statement as it builds them.
-	expiredEventIDs := func() *gorm.DB {
-		return dry.Model(&database.Event{}).Select("id").
-			Where("created_at < ?", cutoff)
-	}
+	// Retention (reapExpired, deleteEvents): one batch of expired
+	// events, then their attempts, deliveries and the events.
+	var expired []string
 
+	assertPlanUses(t, db, dry.Unscoped().Model(&database.Event{}).
+		Where("created_at < ?", cutoff).
+		Limit(database.ExportReapBatchSize).Pluck("id", &expired),
+		"idx_events_created_at (created_at<?)")
 	assertPlanUses(t, db, dry.Unscoped().Where(
-		"delivery_id IN (?)", dry.Model(&database.Delivery{}).
-			Select("id").Where("event_id IN (?)", expiredEventIDs()),
+		"delivery_id IN (?)", dry.Unscoped().Model(&database.Delivery{}).
+			Select("id").Where("event_id IN ?", ids),
 	).Delete(&database.DeliveryResult{}),
-		"idx_delivery_results_delivery_id (delivery_id=?)", byEvent, byAge)
-	assertPlanUses(t, db, dry.Unscoped().Where(
-		"event_id IN (?)", expiredEventIDs(),
-	).Delete(&database.Delivery{}),
-		"idx_deliveries_event_id (event_id=?)", byAge)
-	assertPlanUses(t, db, dry.Unscoped().Where(
-		"created_at < ?", cutoff,
-	).Delete(&database.Event{}), "idx_events_created_at (created_at<?)")
+		"idx_delivery_results_delivery_id (delivery_id=?)",
+		"idx_deliveries_event_id (event_id=?)")
+	assertPlanUses(t, db, dry.Unscoped().Model(&database.Delivery{}).
+		Select("target_id, count(*) AS deliveries_removed, "+
+			"count(CASE WHEN status = ? THEN 1 END) AS failed_removed",
+			database.DeliveryStatusFailed).
+		Where("event_id IN ?", ids).Group("target_id").Find(&removed),
+		"idx_deliveries_event_id (event_id=?)")
+	assertPlanUses(t, db, dry.Unscoped().Where("event_id IN ?", ids).
+		Delete(&database.Delivery{}), "idx_deliveries_event_id (event_id=?)")
+	assertPlanUses(t, db, dry.Unscoped().Where("id IN ?", ids).
+		Delete(&database.Event{}), "sqlite_autoindex_events_1 (id=?)")
+}
+
+// TestStatisticsQueriesUseTheirIndexes does the same for the webhook
+// page's statistics (readEventStats in the handlers): deliveries in
+// progress, each target's deliveries finished since a time, which must
+// come from the index alone, and events received since a time.
+func TestStatisticsQueriesUseTheirIndexes(t *testing.T) {
+	t.Parallel()
+
+	mgr, lc := setupTestWebhookDBManager(t)
+	ctx := context.Background()
+	require.NoError(t, lc.Start(ctx))
+
+	defer func() { require.NoError(t, lc.Stop(ctx)) }()
+
+	db, err := mgr.GetDB(uuid.New().String())
+	require.NoError(t, err)
+
+	dry := db.Session(&gorm.Session{DryRun: true})
+	since := time.Now()
+
+	var (
+		count    int64
+		byTarget []struct{ TargetID string }
+	)
+
+	assertPlanUses(t, db, dry.Model(&database.Delivery{}).
+		Where("status IN ?", []database.DeliveryStatus{
+			database.DeliveryStatusPending,
+			database.DeliveryStatusRetrying,
+		}).Count(&count),
+		"idx_deliveries_status (status=? AND deleted_at=?)")
+	assertPlanUses(t, db, dry.Model(&database.Delivery{}).
+		Select("target_id, "+
+			"count(CASE WHEN status = ? THEN 1 END) AS delivered, "+
+			"count(CASE WHEN status = ? THEN 1 END) AS failed",
+			database.DeliveryStatusDelivered,
+			database.DeliveryStatusFailed).
+		Where("status IN ? AND finished_at >= ?",
+			[]database.DeliveryStatus{
+				database.DeliveryStatusDelivered,
+				database.DeliveryStatusFailed,
+			}, since).
+		Group("target_id").Find(&byTarget),
+		"COVERING INDEX idx_deliveries_status "+
+			"(status=? AND deleted_at=? AND finished_at>?)")
+	assertPlanUses(t, db, dry.Model(&database.Event{}).
+		Where("created_at >= ?", since).Count(&count),
+		"idx_events_deleted_at_created_at "+
+			"(deleted_at=? AND created_at>?)")
 }
 
 // assertPlanUses asserts that SQLite's plan for a statement GORM built

internal/database/export_test.go

@@ -29,6 +29,10 @@ func NewTestRetentionReaper(
 	}
 }
 
+// ExportReapBatchSize exposes how many expired events one retention
+// transaction deletes.
+const ExportReapBatchSize = reapBatchSize
+
 // ExportSweep runs a single retention sweep synchronously for tests.
 func (r *RetentionReaper) ExportSweep(ctx context.Context) {
 	r.sweep(ctx)

internal/database/model_delivery.go

@@ -1,6 +1,10 @@
 package database
 
-import "gorm.io/gorm"
+import (
+	"time"
+
+	"gorm.io/gorm"
+)
 
 // DeliveryStatus represents the status of a delivery
 type DeliveryStatus string
@@ -37,7 +41,7 @@ type Delivery struct {
 	BaseModel
 
 	EventID  string         `gorm:"type:uuid;not null;index:idx_deliveries_event_id,priority:1"       json:"eventId"`
-	TargetID string         `gorm:"type:uuid;not null"                                                json:"targetId"`
+	TargetID string         `gorm:"type:uuid;not null;index:idx_deliveries_status,priority:4"         json:"targetId"`
 	Status   DeliveryStatus `gorm:"not null;default:'pending';index:idx_deliveries_status,priority:1" json:"status"`
 
 	// DeletedAt repeats the BaseModel field only to be the second column
@@ -45,6 +49,13 @@ type Delivery struct {
 	// gives.
 	DeletedAt gorm.DeletedAt `gorm:"index:idx_deliveries_event_id,priority:2;index:idx_deliveries_status,priority:2" json:"deletedAt,omitzero"`
 
+	// FinishedAt is when the delivery became delivered or failed, and
+	// nil while it is pending or retrying. It and then TargetID end the
+	// status index, so the webhook page counts each target's deliveries
+	// that finished in a recent window by reading just that window from
+	// the index.
+	FinishedAt *time.Time `gorm:"index:idx_deliveries_status,priority:3" json:"finishedAt,omitempty"`
+
 	// Relations
 	Event           Event            `json:"event,omitzero"`
 	Target          Target           `json:"target,omitzero"`

internal/database/model_totals.go

@@ -0,0 +1,99 @@
+package database
+
+import (
+	"fmt"
+	"time"
+
+	"gorm.io/gorm"
+)
+
+// The running totals in a webhook's event database keep the webhook
+// page's lifetime figures right after retention has removed the rows
+// they count, and let the page show them without counting every row.
+// Each total changes in the transaction that writes or deletes the
+// rows it counts.
+
+// EventTotals is the single row counting a webhook's events: every
+// event ever stored, how many of them retention has deleted, and when
+// the newest arrived, which retention leaves as it is.
+type EventTotals struct {
+	ID int64 `gorm:"primaryKey"`
+
+	Events        int64 `gorm:"not null"`
+	EventsRemoved int64 `gorm:"not null"`
+
+	// LastEventAt is when the newest event arrived, or nil before the
+	// first.
+	LastEventAt *time.Time
+}
+
+// TableName names the table AddEventTotals updates.
+func (EventTotals) TableName() string {
+	return "event_totals"
+}
+
+// TargetTotals is one row per target counting its deliveries: every
+// delivery ever created, how many became delivered and how many
+// failed, and how many deliveries and failed deliveries retention has
+// deleted. The webhook's delivery figures are these rows summed.
+type TargetTotals struct {
+	TargetID string `gorm:"type:uuid;primaryKey"`
+
+	Deliveries int64 `gorm:"not null"`
+	Delivered  int64 `gorm:"not null"`
+	Failed     int64 `gorm:"not null"`
+
+	DeliveriesRemoved int64 `gorm:"not null"`
+	FailedRemoved     int64 `gorm:"not null"`
+}
+
+// TableName names the table AddTargetTotals updates.
+func (TargetTotals) TableName() string {
+	return "target_totals"
+}
+
+// AddEventTotals adds each count in add to the webhook's event totals,
+// and records add.LastEventAt as when the newest event arrived if it is
+// set. Call it on the transaction that writes or deletes the events it
+// counts.
+func AddEventTotals(tx *gorm.DB, add EventTotals) error {
+	err := tx.Exec(
+		`UPDATE event_totals SET
+			events = events + ?,
+			events_removed = events_removed + ?,
+			last_event_at = coalesce(?, last_event_at)`,
+		add.Events, add.EventsRemoved, add.LastEventAt,
+	).Error
+	if err != nil {
+		return fmt.Errorf("adding to event totals: %w", err)
+	}
+
+	return nil
+}
+
+// AddTargetTotals adds each count in add to the totals of the target
+// add.TargetID names, creating its row the first time. Call it on the
+// transaction that writes or deletes the deliveries it counts.
+func AddTargetTotals(tx *gorm.DB, add TargetTotals) error {
+	err := tx.Exec(
+		`INSERT INTO target_totals (target_id, deliveries, delivered,
+			failed, deliveries_removed, failed_removed)
+		VALUES (?, ?, ?, ?, ?, ?)
+		ON CONFLICT (target_id) DO UPDATE SET
+			deliveries = deliveries + excluded.deliveries,
+			delivered = delivered + excluded.delivered,
+			failed = failed + excluded.failed,
+			deliveries_removed =
+				deliveries_removed + excluded.deliveries_removed,
+			failed_removed = failed_removed + excluded.failed_removed`,
+		add.TargetID, add.Deliveries, add.Delivered,
+		add.Failed, add.DeliveriesRemoved, add.FailedRemoved,
+	).Error
+	if err != nil {
+		return fmt.Errorf(
+			"adding to totals of target %s: %w", add.TargetID, err,
+		)
+	}
+
+	return nil
+}

internal/database/models.go

@@ -2,7 +2,8 @@ package database
 
 // Migrate runs database migrations for the main application database.
 // Only configuration-tier models are stored in the main database.
-// Event-tier models (Event, Delivery, DeliveryResult) live in
+// Event-tier models (Event, Delivery, DeliveryResult, EventTotals,
+// TargetTotals) live in
 // per-webhook dedicated databases managed by WebhookDBManager.
 func (d *Database) Migrate() error {
 	return d.db.AutoMigrate(

internal/database/retention.go

@@ -18,6 +18,19 @@ import (
 // computation.
 const hoursPerDay = 24
 
+// reapBatchSize is how many expired events one retention transaction
+// deletes. A transaction holds the event database's write lock, which
+// the receiver and the delivery workers wait for, so a large prune is
+// split into transactions each short enough to finish well inside the
+// busy timeout.
+const reapBatchSize = 1000
+
+// reapBatchPause is how long retention waits after one batch before
+// starting the next. A writer waiting for the write lock checks for it
+// again after at most 100 ms, so a longer pause lets it in between two
+// batches instead of only after the whole prune.
+const reapBatchPause = 200 * time.Millisecond
+
 // RetentionReaperParams holds the fx dependencies for the
 // RetentionReaper.
 type RetentionReaperParams struct {
@@ -187,13 +200,15 @@ func (r *RetentionReaper) sweep(ctx context.Context) {
 			continue
 		}
 
-		r.reapWebhook(wh.ID, wh.RetentionDays)
+		r.reapWebhook(ctx, wh.ID, wh.RetentionDays)
 	}
 }
 
 // reapWebhook removes every expired event (and its dependents) from a
-// single webhook's database.
+// single webhook's database, or as many as it reaches before ctx is
+// cancelled.
 func (r *RetentionReaper) reapWebhook(
+	ctx context.Context,
 	webhookID string,
 	retentionDays int,
 ) {
@@ -213,7 +228,7 @@ func (r *RetentionReaper) reapWebhook(
 		return
 	}
 
-	deleted, err := reapExpired(db, cutoff)
+	deleted, err := reapExpired(ctx, db, cutoff)
 	if err != nil {
 		r.log.Error(
 			"retention sweep: failed to reap expired events",
@@ -265,57 +280,107 @@ func retentionCutoff(
 	), true
 }
 
-// reapExpired hard-deletes, in foreign-key-safe order, the delivery
-// results, deliveries, and events associated with events older than
-// cutoff. Deletes are unscoped so rows are physically removed rather
-// than soft-deleted, reclaiming disk. It returns the number of events
-// deleted.
-func reapExpired(db *gorm.DB, cutoff time.Time) (int64, error) {
-	// Fresh subqueries are built per statement to avoid reusing a
-	// mutated builder across executions.
-	expiredEventIDs := func() *gorm.DB {
-		return db.Model(&Event{}).
-			Select("id").
-			Where("created_at < ?", cutoff)
-	}
-	expiredDeliveryIDs := func() *gorm.DB {
-		return db.Model(&Delivery{}).
-			Select("id").
-			Where("event_id IN (?)", expiredEventIDs())
-	}
+// reapExpired hard-deletes the events older than cutoff, with their
+// deliveries and delivery results, reapBatchSize events per
+// transaction with reapBatchPause between transactions, until none is
+// left. Once ctx is cancelled it returns after the batch in hand,
+// leaving the rest to the next sweep, so stopping the app does not
+// wait for a long prune. It returns the number of events deleted.
+func reapExpired(
+	ctx context.Context, db *gorm.DB, cutoff time.Time,
+) (int64, error) {
+	var total int64
 
-	// 1. Delivery results whose delivery belongs to an expired event.
-	res := db.Unscoped().
-		Where("delivery_id IN (?)", expiredDeliveryIDs()).
-		Delete(&DeliveryResult{})
-	if res.Error != nil {
-		return 0, fmt.Errorf(
-			"deleting expired delivery results: %w",
-			res.Error,
-		)
-	}
+	for {
+		var eventIDs []string
 
-	// 2. Deliveries belonging to an expired event.
-	del := db.Unscoped().
-		Where("event_id IN (?)", expiredEventIDs()).
-		Delete(&Delivery{})
-	if del.Error != nil {
-		return 0, fmt.Errorf(
-			"deleting expired deliveries: %w",
-			del.Error,
-		)
-	}
+		err := db.Transaction(func(tx *gorm.DB) error {
+			err := tx.Unscoped().Model(&Event{}).
+				Where("created_at < ?", cutoff).
+				Limit(reapBatchSize).
+				Pluck("id", &eventIDs).Error
+			if err != nil {
+				return fmt.Errorf("selecting expired events: %w", err)
+			}
 
-	// 3. The expired events themselves.
-	ev := db.Unscoped().
-		Where("created_at < ?", cutoff).
-		Delete(&Event{})
-	if ev.Error != nil {
-		return 0, fmt.Errorf(
-			"deleting expired events: %w",
-			ev.Error,
-		)
-	}
+			if len(eventIDs) == 0 {
+				return nil
+			}
 
-	return ev.RowsAffected, nil
+			return deleteEvents(tx, eventIDs)
+		})
+		if err != nil {
+			return total, err
+		}
+
+		total += int64(len(eventIDs))
+
+		if len(eventIDs) < reapBatchSize {
+			return total, nil
+		}
+
+		select {
+		case <-ctx.Done():
+			return total, nil
+		case <-time.After(reapBatchPause):
+		}
+	}
+}
+
+// deleteEvents hard-deletes the given events and, in foreign-key-safe
+// order before them, their delivery results and deliveries, then adds
+// what it deleted to the running totals. It runs on reapExpired's
+// transaction, so the totals change exactly when the rows do. Deletes
+// are unscoped so rows are physically removed rather than
+// soft-deleted, reclaiming disk.
+func deleteEvents(tx *gorm.DB, eventIDs []string) error {
+	// 1. The delivery results of the events' deliveries.
+	err := tx.Unscoped().
+		Where("delivery_id IN (?)", tx.Unscoped().Model(&Delivery{}).
+			Select("id").
+			Where("event_id IN ?", eventIDs)).
+		Delete(&DeliveryResult{}).Error
+	if err != nil {
+		return fmt.Errorf("deleting expired delivery results: %w", err)
+	}
+
+	// 2. The events' deliveries, after counting them, and the failed
+	// ones among them, per target. The status is tested in the select
+	// list rather than the WHERE clause: there, SQLite would read every
+	// failed delivery the webhook has through the status index,
+	// instead of only these through the event_id index.
+	var removed []TargetTotals
+
+	err = tx.Unscoped().Model(&Delivery{}).
+		Select("target_id, count(*) AS deliveries_removed, "+
+			"count(CASE WHEN status = ? THEN 1 END) AS failed_removed",
+			DeliveryStatusFailed).
+		Where("event_id IN ?", eventIDs).
+		Group("target_id").
+		Find(&removed).Error
+	if err != nil {
+		return fmt.Errorf("counting expired deliveries: %w", err)
+	}
+
+	err = tx.Unscoped().
+		Where("event_id IN ?", eventIDs).
+		Delete(&Delivery{}).Error
+	if err != nil {
+		return fmt.Errorf("deleting expired deliveries: %w", err)
+	}
+
+	// 3. The events themselves.
+	ev := tx.Unscoped().Where("id IN ?", eventIDs).Delete(&Event{})
+	if ev.Error != nil {
+		return fmt.Errorf("deleting expired events: %w", ev.Error)
+	}
+
+	for i := range removed {
+		err = AddTargetTotals(tx, removed[i])
+		if err != nil {
+			return err
+		}
+	}
+
+	return AddEventTotals(tx, EventTotals{EventsRemoved: ev.RowsAffected})
 }

internal/database/totals_test.go

@@ -0,0 +1,410 @@
+package database_test
+
+import (
+	"context"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/gorm"
+	"sneak.berlin/go/webhooker/internal/database"
+)
+
+// readEventTotals reads a webhook database's row of event totals,
+// asserting that it has exactly one.
+func readEventTotals(t *testing.T, db *gorm.DB) database.EventTotals {
+	t.Helper()
+
+	var rows []database.EventTotals
+
+	require.NoError(t, db.Find(&rows).Error)
+	require.Len(t, rows, 1)
+
+	return rows[0]
+}
+
+// readTargetTotals reads a webhook database's target totals, keyed by
+// target.
+func readTargetTotals(
+	t *testing.T, db *gorm.DB,
+) map[string]database.TargetTotals {
+	t.Helper()
+
+	var rows []database.TargetTotals
+
+	require.NoError(t, db.Find(&rows).Error)
+
+	byTarget := make(map[string]database.TargetTotals, len(rows))
+	for _, row := range rows {
+		byTarget[row.TargetID] = row
+	}
+
+	return byTarget
+}
+
+// TestWebhookDBManager_TotalsSurviveReopen verifies that a new event
+// database starts with one row of zero event totals and no target
+// totals, that adding to a target twice adds to the one row, and that
+// opening the database again keeps everything added.
+func TestWebhookDBManager_TotalsSurviveReopen(t *testing.T) {
+	t.Parallel()
+
+	mgr, lc := setupTestWebhookDBManager(t)
+	ctx := context.Background()
+	require.NoError(t, lc.Start(ctx))
+
+	defer func() { require.NoError(t, lc.Stop(ctx)) }()
+
+	webhookID := uuid.New().String()
+
+	db, err := mgr.GetDB(webhookID)
+	require.NoError(t, err)
+
+	fresh := readEventTotals(t, db)
+	assert.Equal(t, database.EventTotals{ID: fresh.ID}, fresh)
+	assert.Empty(t, readTargetTotals(t, db))
+
+	first, second := uuid.New().String(), uuid.New().String()
+
+	require.NoError(t, database.AddEventTotals(db, database.EventTotals{
+		Events: 2,
+	}))
+	require.NoError(t, database.AddTargetTotals(db, database.TargetTotals{
+		TargetID: first, Deliveries: 2, Delivered: 1,
+	}))
+	require.NoError(t, database.AddTargetTotals(db, database.TargetTotals{
+		TargetID: first, Failed: 1,
+	}))
+	require.NoError(t, database.AddTargetTotals(db, database.TargetTotals{
+		TargetID: second, Deliveries: 1,
+	}))
+
+	// Drop the cached connection so the next open reopens the file,
+	// as a restart would.
+	require.NoError(t, mgr.CloseAll())
+
+	db, err = mgr.GetDB(webhookID)
+	require.NoError(t, err)
+
+	assert.Equal(t, database.EventTotals{ID: fresh.ID, Events: 2},
+		readEventTotals(t, db))
+	assert.Equal(t, map[string]database.TargetTotals{
+		first: {
+			TargetID: first, Deliveries: 2, Delivered: 1, Failed: 1,
+		},
+		second: {TargetID: second, Deliveries: 1},
+	}, readTargetTotals(t, db))
+}
+
+// seedExpiredEvents stores count events created at the given time,
+// each with a delivered delivery to one target and a failed delivery
+// to the other, and one attempt for each delivery.
+func seedExpiredEvents(
+	t *testing.T,
+	db *gorm.DB,
+	webhookID string,
+	count int,
+	createdAt time.Time,
+	delivered, failed string,
+) {
+	t.Helper()
+
+	events := make([]database.Event, count)
+	deliveries := make([]database.Delivery, 0, 2*count)
+
+	for i := range events {
+		events[i] = database.Event{
+			WebhookID:    webhookID,
+			EntrypointID: uuid.New().String(),
+			Method:       http.MethodPost,
+		}
+		events[i].ID = uuid.New().String()
+		events[i].CreatedAt = createdAt
+
+		deliveries = append(deliveries,
+			database.Delivery{
+				EventID:  events[i].ID,
+				TargetID: delivered,
+				Status:   database.DeliveryStatusDelivered,
+			},
+			database.Delivery{
+				EventID:  events[i].ID,
+				TargetID: failed,
+				Status:   database.DeliveryStatusFailed,
+			},
+		)
+	}
+
+	require.NoError(t, db.CreateInBatches(events, 500).Error)
+	require.NoError(t, db.CreateInBatches(deliveries, 500).Error)
+
+	results := make([]database.DeliveryResult, len(deliveries))
+	for i := range deliveries {
+		results[i] = database.DeliveryResult{
+			DeliveryID: deliveries[i].ID, AttemptNum: 1,
+		}
+	}
+
+	require.NoError(t, db.CreateInBatches(results, 500).Error)
+}
+
+// seedBareEvents stores count events created at the given time, with
+// no deliveries.
+func seedBareEvents(
+	t *testing.T,
+	db *gorm.DB,
+	webhookID string,
+	count int,
+	createdAt time.Time,
+) {
+	t.Helper()
+
+	events := make([]database.Event, count)
+	for i := range events {
+		events[i] = database.Event{
+			WebhookID:    webhookID,
+			EntrypointID: uuid.New().String(),
+			Method:       http.MethodPost,
+		}
+		events[i].CreatedAt = createdAt
+	}
+
+	require.NoError(t, db.CreateInBatches(events, 500).Error)
+}
+
+// TestRetentionReaper_PrunesMoreThanOneBatch verifies that a prune
+// larger than one transaction's batch removes every expired event with
+// its deliveries and delivery results, keeps the recent event, and
+// adds what it removed to the event and target totals, so the totals
+// within retention match the rows still stored.
+func TestRetentionReaper_PrunesMoreThanOneBatch(t *testing.T) {
+	t.Parallel()
+
+	env := setupRetentionTest(t)
+
+	webhookID := createWebhook(t, env.mainDB.DB(), 30)
+
+	db, err := env.mgr.GetDB(webhookID)
+	require.NoError(t, err)
+
+	expired := database.ExportReapBatchSize + 1
+	delivered, failed := uuid.New().String(), uuid.New().String()
+	seedExpiredEvents(t, db, webhookID, expired,
+		time.Now().Add(-40*24*time.Hour), delivered, failed)
+
+	// One recent event, delivered to the first target.
+	recent := seedEventChain(t, db, webhookID, time.Now())
+	require.NoError(t, db.Model(&database.Delivery{}).
+		Where("id = ?", recent.deliveryID).
+		Update("target_id", delivered).Error)
+
+	// The totals storing those rows would have left.
+	n := int64(expired)
+	require.NoError(t, database.AddEventTotals(db, database.EventTotals{
+		Events: n + 1,
+	}))
+	require.NoError(t, database.AddTargetTotals(db, database.TargetTotals{
+		TargetID: delivered, Deliveries: n + 1, Delivered: n + 1,
+	}))
+	require.NoError(t, database.AddTargetTotals(db, database.TargetTotals{
+		TargetID: failed, Deliveries: n, Failed: n,
+	}))
+
+	env.reaper.ExportSweep(context.Background())
+
+	// Only the recent event's rows are left.
+	for _, model := range []any{
+		&database.Event{}, &database.Delivery{}, &database.DeliveryResult{},
+	} {
+		var count int64
+
+		require.NoError(t, db.Model(model).Count(&count).Error)
+		assert.Equal(t, int64(1), count, "%T rows left", model)
+	}
+
+	assertChainPresent(t, db, recent)
+
+	eventTotals := readEventTotals(t, db)
+	assert.Equal(t, database.EventTotals{
+		ID: eventTotals.ID, Events: n + 1, EventsRemoved: n,
+	}, eventTotals)
+
+	targetTotals := readTargetTotals(t, db)
+	assert.Equal(t, map[string]database.TargetTotals{
+		delivered: {
+			TargetID: delivered, Deliveries: n + 1, Delivered: n + 1,
+			DeliveriesRemoved: n,
+		},
+		failed: {
+			TargetID: failed, Deliveries: n, Failed: n,
+			DeliveriesRemoved: n, FailedRemoved: n,
+		},
+	}, targetTotals)
+
+	// A sweep with nothing left to remove changes nothing.
+	env.reaper.ExportSweep(context.Background())
+
+	assert.Equal(t, eventTotals, readEventTotals(t, db))
+	assert.Equal(t, targetTotals, readTargetTotals(t, db))
+}
+
+// TestRetentionReaper_WriteDuringPruneSucceeds verifies that a prune
+// of several batches lets other writers in between its batches: an
+// event stored once the first batch is deleted is stored while expired
+// events are still left, not only after the prune has finished.
+func TestRetentionReaper_WriteDuringPruneSucceeds(t *testing.T) {
+	t.Parallel()
+
+	env := setupRetentionTest(t)
+
+	webhookID := createWebhook(t, env.mainDB.DB(), 30)
+
+	db, err := env.mgr.GetDB(webhookID)
+	require.NoError(t, err)
+
+	// Three batches of expired events, with nothing else stored: only
+	// the number of batches matters here.
+	expired := 3 * database.ExportReapBatchSize
+	seedBareEvents(t, db, webhookID, expired,
+		time.Now().Add(-40*24*time.Hour))
+
+	cutoff := time.Now().Add(-30 * 24 * time.Hour)
+	countExpired := func() int64 {
+		var count int64
+
+		require.NoError(t, db.Model(&database.Event{}).
+			Where("created_at < ?", cutoff).
+			Count(&count).Error)
+
+		return count
+	}
+
+	pruned := make(chan struct{})
+
+	go func() {
+		defer close(pruned)
+
+		env.reaper.ExportSweep(context.Background())
+	}()
+
+	t.Cleanup(func() { <-pruned })
+
+	// Every stored event is expired until the write below.
+	require.Eventually(t, func() bool {
+		var count int64
+
+		err := db.Model(&database.Event{}).Count(&count).Error
+
+		return err == nil && count < int64(expired)
+	}, 10*time.Second, 10*time.Millisecond)
+
+	event := &database.Event{
+		WebhookID:    webhookID,
+		EntrypointID: uuid.New().String(),
+		Method:       http.MethodPost,
+	}
+	require.NoError(t, db.Create(event).Error)
+
+	assert.Positive(t, countExpired(),
+		"the event was stored only after the whole prune")
+
+	<-pruned
+
+	assert.Zero(t, countExpired())
+
+	var stored database.Event
+
+	require.NoError(t, db.First(&stored, "id = ?", event.ID).Error)
+}
+
+// TestRetentionReaper_StopDuringPruneLeavesTheRest verifies that
+// stopping the reaper during a prune of several batches returns
+// between two batches, well inside the stop timeout, leaving the
+// remaining expired events for the next sweep, and that the totals
+// match the rows left.
+func TestRetentionReaper_StopDuringPruneLeavesTheRest(t *testing.T) {
+	t.Parallel()
+
+	env := setupRetentionTest(t)
+
+	webhookID := createWebhook(t, env.mainDB.DB(), 30)
+
+	db, err := env.mgr.GetDB(webhookID)
+	require.NoError(t, err)
+
+	// Two batches and one more of expired events, a few of them with a
+	// delivered and a failed delivery for the target totals to count.
+	// Most carry nothing else, to keep the test quick.
+	const withDeliveries = 10
+
+	expiredAt := time.Now().Add(-40 * 24 * time.Hour)
+	delivered, failed := uuid.New().String(), uuid.New().String()
+	seedExpiredEvents(t, db, webhookID, withDeliveries, expiredAt,
+		delivered, failed)
+	seedBareEvents(t, db, webhookID,
+		2*database.ExportReapBatchSize+1-withDeliveries, expiredAt)
+
+	n := int64(2*database.ExportReapBatchSize + 1)
+	require.NoError(t, database.AddEventTotals(db, database.EventTotals{
+		Events: n,
+	}))
+	require.NoError(t, database.AddTargetTotals(db, database.TargetTotals{
+		TargetID: delivered, Deliveries: withDeliveries,
+		Delivered: withDeliveries,
+	}))
+	require.NoError(t, database.AddTargetTotals(db, database.TargetTotals{
+		TargetID: failed, Deliveries: withDeliveries,
+		Failed: withDeliveries,
+	}))
+
+	env.reaper.ExportSetInterval(time.Millisecond)
+	env.reaper.ExportStart()
+
+	// Stop once the first batch is deleted. The stop lands in the pause
+	// after it, or at worst during the second batch, so at least the
+	// last event is left.
+	require.Eventually(t, func() bool {
+		var count int64
+
+		err := db.Model(&database.Event{}).Count(&count).Error
+
+		return err == nil && count < n
+	}, 10*time.Second, 10*time.Millisecond)
+
+	// The app's stop timeout.
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	require.NoError(t, env.reaper.ExportStop(ctx))
+
+	var events int64
+
+	require.NoError(t, db.Model(&database.Event{}).Count(&events).Error)
+	assert.Positive(t, events, "the stop waited for the whole prune")
+
+	eventTotals := readEventTotals(t, db)
+	assert.Equal(t, events, eventTotals.Events-eventTotals.EventsRemoved)
+
+	targetTotals := readTargetTotals(t, db)
+	require.Len(t, targetTotals, 2)
+
+	for target, totals := range targetTotals {
+		var deliveries, failures int64
+
+		require.NoError(t, db.Model(&database.Delivery{}).
+			Where("target_id = ?", target).
+			Count(&deliveries).Error)
+		require.NoError(t, db.Model(&database.Delivery{}).
+			Where("target_id = ? AND status = ?",
+				target, database.DeliveryStatusFailed).
+			Count(&failures).Error)
+
+		assert.Equal(t, deliveries,
+			totals.Deliveries-totals.DeliveriesRemoved, target)
+		assert.Equal(t, failures, totals.Failed-totals.FailedRemoved,
+			target)
+	}
+}

internal/database/webhook_db_manager.go

@@ -35,7 +35,8 @@ var errInvalidCachedDBType = errors.New(
 
 // WebhookDBManager manages per-webhook SQLite database files
 // for event storage. Each webhook gets its own dedicated
-// database containing Events, Deliveries, and DeliveryResults.
+// database containing Events, Deliveries, DeliveryResults and the
+// running totals of them (EventTotals, TargetTotals).
 // Database connections are opened lazily and cached.
 type WebhookDBManager struct {
 	dataDir string
@@ -295,6 +296,7 @@ func (m *WebhookDBManager) openDB(
 	// Run migrations for event-tier models only
 	err = db.AutoMigrate(
 		&Event{}, &Delivery{}, &DeliveryResult{},
+		&EventTotals{}, &TargetTotals{},
 	)
 	if err != nil {
 		_ = sqlDB.Close()
@@ -305,6 +307,18 @@ func (m *WebhookDBManager) openDB(
 		)
 	}
 
+	// A new database gets its row of event totals, all zero. Target
+	// totals rows are created by the first delivery to each target.
+	err = db.FirstOrCreate(&EventTotals{}).Error
+	if err != nil {
+		_ = sqlDB.Close()
+
+		return nil, fmt.Errorf(
+			"creating event totals for webhook database %s: %w",
+			webhookID, err,
+		)
+	}
+
 	m.log.Info(
 		"opened per-webhook database",
 		"webhook_id", webhookID,