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
This commit is contained in:
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user