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

README.md

@@ -1754,10 +1754,11 @@ retries) is individually logged for full observability.
 Running counts in each event database, read by the statistics pane at the
 top of the webhook page. `EventTotals` is one row:
 
-| Field            | Type    | Description |
+| Field            | Type      | Description |
-| ---------------- | ------- | ----------- |
+| ---------------- | --------- | ----------- |
-| `events`         | integer | Events ever stored, resubmitted copies included |
+| `events`         | integer   | Events ever stored, resubmitted copies included |
-| `events_removed` | integer | Events retention has deleted |
+| `events_removed` | integer   | Events retention has deleted |
+| `last_event_at`  | timestamp | When the newest event arrived (nullable; empty before the first); retention leaves it as it is |
 
 `TargetTotals` is one row per target, created by the first delivery to it:
 
@@ -1774,7 +1775,9 @@ Each count changes in the transaction that writes or deletes the rows it
 counts. The pane's lifetime events are `events`, and its lifetime
 deliveries and failures are `deliveries` and `failed` summed over the
 targets; each figure within retention is the same less what retention
-removed, so neither needs the rows themselves. Its last-10-minutes and
+removed, so neither needs the rows themselves. Its last event is
+`last_event_at`, written in the transaction that stores the event, so it
+still shows once retention has removed every event. Its last-10-minutes and
 last-24-hours figures are counted from the `events` and `deliveries`
 indexes over just that window, the deliveries in one query grouped by
 target. Its failure percentage for a window is the deliveries that became
@@ -1791,7 +1794,7 @@ tags, so `AutoMigrate` creates them on a fresh database:
 | `deliveries`       | `status`, `deleted_at`, `finished_at`, `target_id` | Startup recovery, the retry and pending sweeps every 60 seconds and the queue-depth sampler every 30 seconds, which select deliveries by status, and the webhook page's statistics, which count each target's deliveries by status and when they finished |
 | `deliveries`       | `event_id`, `deleted_at`    | The event log, which loads each event's deliveries, and retention, which counts and deletes the deliveries of expired events |
 | `delivery_results` | `delivery_id`, `deleted_at` | The event log, which loads the attempts of a page's deliveries, and retention, which deletes the attempts of expired events |
-| `events`           | `deleted_at`, `created_at`  | The webhook page's statistics, which count recent events and find the newest |
+| `events`           | `deleted_at`, `created_at`  | The webhook page's statistics, which count recent events |
 | `events`           | `created_at`                | Retention, which selects expired events by age |
 
 GORM's soft delete adds `deleted_at IS NULL` to these queries; retention

internal/database/event_tier_indexes_test.go

@@ -152,9 +152,7 @@ func TestEventTierQueriesUseTheirIndexes(t *testing.T) {
 // 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, events received since a time, and the
+// come from the index alone, and events received since a time.
-// newest event, which must come straight off an index rather than from
-// sorting every event.
 func TestStatisticsQueriesUseTheirIndexes(t *testing.T) {
 	t.Parallel()
 
@@ -172,7 +170,6 @@ func TestStatisticsQueriesUseTheirIndexes(t *testing.T) {
 
 	var (
 		count    int64
-		newest   []time.Time
 		byTarget []struct{ TargetID string }
 	)
 
@@ -200,12 +197,6 @@ func TestStatisticsQueriesUseTheirIndexes(t *testing.T) {
 		Where("created_at >= ?", since).Count(&count),
 		"idx_events_deleted_at_created_at "+
 			"(deleted_at=? AND created_at>?)")
-
-	newestEvent := dry.Model(&database.Event{}).
-		Order("created_at DESC").Limit(1).Pluck("created_at", &newest)
-	assertPlanUses(t, db, newestEvent,
-		"idx_events_deleted_at_created_at (deleted_at=?)")
-	assert.NotContains(t, queryPlan(t, db, newestEvent), "TEMP B-TREE")
 }
 
 // assertPlanUses asserts that SQLite's plan for a statement GORM built

internal/database/model_totals.go

@@ -2,6 +2,7 @@ package database
 
 import (
 	"fmt"
+	"time"
 
 	"gorm.io/gorm"
 )
@@ -13,12 +14,17 @@ import (
 // rows it counts.
 
 // EventTotals is the single row counting a webhook's events: every
-// event ever stored, and how many of them retention has deleted.
+// 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.
@@ -46,15 +52,17 @@ func (TargetTotals) TableName() string {
 	return "target_totals"
 }
 
-// AddEventTotals adds each count in add to the webhook's event totals.
+// AddEventTotals adds each count in add to the webhook's event totals,
-// Call it on the transaction that writes or deletes the events it
+// 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 + ?`,
+			events_removed = events_removed + ?,
-		add.Events, add.EventsRemoved,
+			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)

internal/database/retention.go

@@ -200,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,
 ) {
@@ -226,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",
@@ -281,8 +283,12 @@ func retentionCutoff(
 // 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. It returns the number of events deleted.
+// left. Once ctx is cancelled it returns after the batch in hand,
-func reapExpired(db *gorm.DB, cutoff time.Time) (int64, error) {
+// 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
 
 	for {
@@ -313,7 +319,11 @@ func reapExpired(db *gorm.DB, cutoff time.Time) (int64, error) {
 			return total, nil
 		}
 
-		time.Sleep(reapBatchPause)
+		select {
+		case <-ctx.Done():
+			return total, nil
+		case <-time.After(reapBatchPause):
+		}
 	}
 }
 

internal/database/totals_test.go

@@ -151,6 +151,30 @@ func seedExpiredEvents(
 	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
@@ -243,17 +267,9 @@ func TestRetentionReaper_WriteDuringPruneSucceeds(t *testing.T) {
 
 	// Three batches of expired events, with nothing else stored: only
 	// the number of batches matters here.
-	expired := make([]database.Event, 3*database.ExportReapBatchSize)
+	expired := 3 * database.ExportReapBatchSize
-	for i := range expired {
+	seedBareEvents(t, db, webhookID, expired,
-		expired[i] = database.Event{
+		time.Now().Add(-40*24*time.Hour))
-			WebhookID:    webhookID,
-			EntrypointID: uuid.New().String(),
-			Method:       http.MethodPost,
-		}
-		expired[i].CreatedAt = time.Now().Add(-40 * 24 * time.Hour)
-	}
-
-	require.NoError(t, db.CreateInBatches(expired, 500).Error)
 
 	cutoff := time.Now().Add(-30 * 24 * time.Hour)
 	countExpired := func() int64 {
@@ -282,7 +298,7 @@ func TestRetentionReaper_WriteDuringPruneSucceeds(t *testing.T) {
 
 		err := db.Model(&database.Event{}).Count(&count).Error
 
-		return err == nil && count < int64(len(expired))
+		return err == nil && count < int64(expired)
 	}, 10*time.Second, 10*time.Millisecond)
 
 	event := &database.Event{
@@ -303,3 +319,92 @@ func TestRetentionReaper_WriteDuringPruneSucceeds(t *testing.T) {
 
 	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/delivery/delivery_totals_test.go

@@ -114,3 +114,64 @@ func TestUpdateDeliveryStatus_DeletedDeliveryIsNotCounted(t *testing.T) {
 	assert.Equal(t, database.TargetTotals{TargetID: targetID},
 		targetTotals(t, db, targetID))
 }
+
+// TestUpdateDeliveryStatus_FinishedDeliveryIsNotSettledAgain covers a
+// delivery settled a second time, as recovery can do when a worker has
+// settled it since recovery read it. Neither status writes over the
+// first, and the totals do not move.
+func TestUpdateDeliveryStatus_FinishedDeliveryIsNotSettledAgain(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	finished := []database.DeliveryStatus{
+		database.DeliveryStatusDelivered,
+		database.DeliveryStatusFailed,
+	}
+
+	for _, first := range finished {
+		t.Run(string(first), func(t *testing.T) {
+			t.Parallel()
+
+			db := testWebhookDB(t)
+			e := testEngine(t, 1)
+			event := seedEvent(t, db, `{}`)
+			targetID := uuid.New().String()
+			d := seedDelivery(
+				t, db, event.ID, targetID,
+				database.DeliveryStatusRetrying,
+			)
+
+			// The delivery as recovery read it, before the worker
+			// settled it.
+			readBefore := d
+
+			require.NoError(t, e.ExportUpdateDeliveryStatus(
+				db, &d, first,
+			))
+
+			var settled database.Delivery
+
+			require.NoError(t, db.First(&settled, "id = ?", d.ID).Error)
+			require.NotNil(t, settled.FinishedAt)
+
+			totals := targetTotals(t, db, targetID)
+
+			for _, again := range finished {
+				stale := readBefore
+
+				require.NoError(t, e.ExportUpdateDeliveryStatus(
+					db, &stale, again,
+				))
+			}
+
+			var stored database.Delivery
+
+			require.NoError(t, db.First(&stored, "id = ?", d.ID).Error)
+			assert.Equal(t, first, stored.Status)
+			require.NotNil(t, stored.FinishedAt)
+			assert.True(t, settled.FinishedAt.Equal(*stored.FinishedAt))
+			assert.Equal(t, totals, targetTotals(t, db, targetID))
+		})
+	}
+}

internal/delivery/engine.go

@@ -1580,9 +1580,11 @@ func (e *Engine) updateDeliveryStatus(
 
 // writeDeliveryStatus writes a delivery's new status. A delivery that
 // becomes delivered or failed also gets the time it finished, and is
-// added to its target's delivered or failed total. It is counted only
+// added to its target's delivered or failed total. That write changes
-// if the row was still there to update: retention may have deleted it
+// only a delivery not yet delivered or failed, and the total moves
-// while the engine was working on it.
+// only when it changed a row: retention may have deleted the delivery
+// while the engine was working on it, and a recovery path may settle
+// a delivery that a worker has already settled.
 func writeDeliveryStatus(
 	tx *gorm.DB,
 	d *database.Delivery,
@@ -1592,10 +1594,15 @@ func writeDeliveryStatus(
 		return tx.Model(d).Update("status", status).Error
 	}
 
-	res := tx.Model(d).Updates(map[string]any{
+	res := tx.Model(d).
-		"status":      status,
+		Where("status NOT IN ?", []database.DeliveryStatus{
-		"finished_at": time.Now(),
+			database.DeliveryStatusDelivered,
-	})
+			database.DeliveryStatusFailed,
+		}).
+		Updates(map[string]any{
+			"status":      status,
+			"finished_at": time.Now(),
+		})
 	if res.Error != nil || res.RowsAffected == 0 {
 		return res.Error
 	}

internal/handlers/webhook.go

@@ -298,7 +298,9 @@ func (h *Handlers) createAndFanOut(
 		return nil, nil, err
 	}
 
-	err = database.AddEventTotals(tx, database.EventTotals{Events: 1})
+	err = database.AddEventTotals(tx, database.EventTotals{
+		Events: 1, LastEventAt: &event.CreatedAt,
+	})
 	if err != nil {
 		tx.Rollback()
 

internal/handlers/webhook_stats.go

@@ -34,8 +34,8 @@ type WebhookStats struct {
 	// InProgress counts the deliveries still pending or retrying.
 	InProgress int64
 
-	// LastEventAt is when the newest stored event arrived, or nil when
+	// LastEventAt is when the newest event arrived, or nil when none
-	// none is stored.
+	// has. Retention does not change it.
 	LastEventAt *time.Time
 
 	Last10Minutes RecentWindow
@@ -154,20 +154,6 @@ func readEventStats(
 		return fmt.Errorf("counting deliveries in progress: %w", err)
 	}
 
-	var newest []time.Time
-
-	err = db.Model(&database.Event{}).
-		Order("created_at DESC").
-		Limit(1).
-		Pluck("created_at", &newest).Error
-	if err != nil {
-		return fmt.Errorf("reading newest event time: %w", err)
-	}
-
-	if len(newest) > 0 {
-		stats.LastEventAt = &newest[0]
-	}
-
 	stats.Last10Minutes, err = readRecentWindow(
 		db, now.Add(-shortWindow),
 	)
@@ -182,8 +168,9 @@ func readEventStats(
 	return err
 }
 
-// readTotals fills in the lifetime and within-retention figures from
+// readTotals fills in the lifetime and within-retention figures, and
-// the running totals: the events' row, and the targets' rows summed.
+// when the last event arrived, from the running totals: the events'
+// row, and the targets' rows summed.
 func readTotals(db *gorm.DB, stats *WebhookStats) error {
 	var events database.EventTotals
 
@@ -201,6 +188,7 @@ func readTotals(db *gorm.DB, stats *WebhookStats) error {
 
 	stats.Lifetime.Events = events.Events
 	stats.WithinRetention.Events = events.Events - events.EventsRemoved
+	stats.LastEventAt = events.LastEventAt
 
 	for _, t := range targets {
 		stats.Lifetime.Deliveries += t.Deliveries

internal/handlers/webhook_stats_test.go

@@ -383,6 +383,56 @@ func TestWebhookStats_EveryFigureAcrossRetentionPrune(t *testing.T) {
 	assertStatsPaneAfterPrune(t, h, sess, hist)
 }
 
+// TestWebhookStats_LastEventSurvivesPruningEveryEvent checks that once
+// retention has removed every event, the pane still shows when the last
+// one arrived rather than "none".
+func TestWebhookStats_LastEventSurvivesPruningEveryEvent(t *testing.T) {
+	t.Parallel()
+
+	var (
+		h     *handlers.Handlers
+		sess  *session.Session
+		db    *database.Database
+		dbMgr *database.WebhookDBManager
+		log   *logger.Logger
+	)
+
+	app := newTestApp(t, &h, &sess, &db, &dbMgr, &log)
+	app.RequireStart()
+
+	t.Cleanup(app.RequireStop)
+
+	wh := &database.Webhook{
+		UserID: deleteTestUserID, Name: "pruned", RetentionDays: 1,
+	}
+	require.NoError(t, db.DB().Omit(clause.Associations).Create(wh).Error)
+
+	path := statsEntrypoint(t, db, wh.ID, true)
+	require.Equal(t, http.StatusOK,
+		postReceiver(t, receiverRouter(h), path))
+
+	webhookDB, err := dbMgr.GetDB(wh.ID)
+	require.NoError(t, err)
+
+	events := listEvents(t, webhookDB)
+	require.Len(t, events, 1)
+
+	arrived := events[0].CreatedAt
+
+	statsAge(t, webhookDB, events[0].ID, time.Now().Add(-50*time.Hour))
+	statsPrune(t, db, dbMgr, log, webhookDB)
+	require.Empty(t, listEvents(t, webhookDB))
+
+	stats := h.WebhookStatsForTest(wh.ID)
+	require.NotNil(t, stats)
+	require.NotNil(t, stats.LastEventAt)
+	assert.True(t, arrived.Equal(*stats.LastEventAt))
+
+	pane := statsPane(t, renderSourceDetailPage(t, h, sess, wh.ID))
+	assert.Contains(t, pane,
+		"Last event "+arrived.Format("2006-01-02 15:04:05 UTC"))
+}
+
 // TestWebhookStats_PaneShowsRetentionPeriod checks that the statistics
 // pane itself, not only the line at the foot of the page, shows the
 // webhook's retention period, for a finite one and for forever.