Use one recovery routine at start and on every sweep tick (closes #542)

Start-up recovery and the 60-second sweep were two parallel sets of
functions doing the same job. recoverDeliveries now does it, run once at
start and then on every tick from one goroutine. A retrying delivery is
scheduled after what is left of its backoff, now if nothing is left, so
backoffElapsed is gone. The run at start still takes every pending
delivery and opens a webhook database whose file is missing; a tick
still takes only pending deliveries untouched for 15 minutes, at most
500, and skips a missing file. A retrying delivery is taken before its
event is read. The README describes one routine.

Model: opus-5-5

internal/delivery/engine.go

@@ -39,8 +39,9 @@ const (
 	// explosions regardless of queue depth.
 	defaultWorkers = 10
 
-	// retrySweepInterval is how often the periodic retry
-	// sweep runs.
+	// retrySweepInterval is how often the recovery routine runs
+	// once it has run at start. Each of those runs is a tick of
+	// the sweep.
 	retrySweepInterval = 60 * time.Second
 
 	// pendingSweepMinAge is how long a delivery must have sat
@@ -169,7 +170,7 @@ type EngineParams struct {
 // Engine processes queued deliveries in the background
 // using a bounded worker pool architecture. It owns only
 // the cross-target machinery: the worker pool, the queue
-// and retry channels, restart recovery, and the persistence
+// and retry channels, the recovery routine, and the persistence
 // helpers and Scheduler that individual targets rely on.
 // Each target type owns its own delivery, including retries,
 // backoff, and circuit breaking.
@@ -207,7 +208,7 @@ type Engine struct {
 	dbTarget *databaseTarget
 
 	// inflight is the set of deliveries this engine currently owns.
-	// Recovery and the sweeps re-dispatch only what it does not
+	// The recovery routine re-dispatches only what it does not
 	// hold. Held by value: its zero value works, so no constructor
 	// can leave it out. See inflight.go.
 	inflight inflightSet
@@ -399,8 +400,8 @@ func (e *Engine) registerHooks(lc fx.Lifecycle) {
 	})
 }
 
-// start launches the worker pool, restart recovery, and the
-// periodic retry sweep.
+// start launches the worker pool and the recovery routine, which
+// runs now and then on every tick of the sweep (see runRecovery).
 //
 // Their context is derived from context.Background(), NOT from
 // the fx OnStart hook context. The hook context carries fx's
@@ -423,11 +424,7 @@ func (e *Engine) start() {
 
 	e.wg.Add(1)
 
-	go e.recoverPending(ctx)
-
-	e.wg.Add(1)
-
-	go e.retrySweep(ctx)
+	go e.runRecovery(ctx)
 
 	e.wg.Add(1)
 
@@ -510,12 +507,6 @@ func (e *Engine) runTask(
 	run(ctx, task)
 }
 
-func (e *Engine) recoverPending(ctx context.Context) {
-	defer e.wg.Done()
-
-	e.recoverInFlight(ctx)
-}
-
 func (e *Engine) processNewTask(
 	ctx context.Context, task *Task,
 ) {
@@ -662,9 +653,9 @@ func (e *Engine) processRetryTask(
 // else on this path reads the target row. Without this check a
 // deletion stops nothing: the timer keeps firing and keeps sending to
 // the destination the operator removed, for the whole remaining
-// backoff chain. Terminalising in the recovery and sweep paths alone
-// is not enough, because those only see the delivery once nothing
-// holds it in memory — which is to say after a restart.
+// backoff chain. Terminalising in the recovery routine alone is not
+// enough, because it only sees the delivery once nothing holds it in
+// memory — which is to say after a restart.
 //
 // The worker already owns this delivery, so the terminal write happens
 // here directly, exactly as a target's own Deliver fails one. Claiming
@@ -712,7 +703,47 @@ func (e *Engine) abandonRetryForMissingTarget(
 	return true
 }
 
-func (e *Engine) recoverInFlight(ctx context.Context) {
+// runRecovery runs the recovery routine, recoverDeliveries, once when
+// the engine starts and then on every tick of the sweep, every
+// retrySweepInterval, until the engine stops.
+func (e *Engine) runRecovery(ctx context.Context) {
+	defer e.wg.Done()
+
+	ticker := time.NewTicker(retrySweepInterval)
+	defer ticker.Stop()
+
+	e.recoverDeliveries(ctx, true)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			e.recoverDeliveries(ctx, false)
+		}
+	}
+}
+
+// recoverDeliveries is the engine's one recovery routine: atStart is
+// true for its run when the engine starts and false on a tick of the
+// sweep.
+//
+// For every webhook it finds the pending and retrying deliveries the
+// engine is not working on: those a stop or a crash cut off, a full
+// channel dropped, or a failed bookkeeping write left behind. One that
+// already has a successful attempt is marked delivered. A retrying one
+// is scheduled after what is left of its backoff, or now when nothing
+// is left. A pending one goes to a worker now.
+//
+// Ownership keeps it off a delivery the engine is working on, has
+// queued, or holds a retry timer for, the same at start as on a tick:
+// every delivery it acts on is taken through the inflight set first.
+// See inflight.go.
+//
+// A run at start differs from a tick in which webhooks it opens (see
+// recoverWebhookDeliveries) and which pending deliveries it takes (see
+// recoverPendingDeliveries).
+func (e *Engine) recoverDeliveries(ctx context.Context, atStart bool) {
 	var webhookIDs []string
 
 	err := e.database.DB().
@@ -734,18 +765,27 @@ func (e *Engine) recoverInFlight(ctx context.Context) {
 		default:
 		}
 
-		// Opened even when its file is missing, so that a lost
-		// database is reported at start, not when the webhook next
-		// receives an event, which for a quiet webhook may be never.
-		e.recoverWebhookDeliveries(ctx, webhookID)
+		e.recoverWebhookDeliveries(ctx, webhookID, atStart)
 	}
 }
 
+// recoverWebhookDeliveries runs the recovery routine over one webhook.
+//
+// At start the webhook's database is opened even when its file is
+// missing, so that a lost database is reported at start, not when the
+// webhook next receives an event, which for a quiet webhook may be
+// never. A tick leaves a missing file alone rather than create one: a
+// webhook being created has its row saved a moment before its file,
+// and a tick in between would report that file as lost.
 func (e *Engine) recoverWebhookDeliveries(
-	ctx context.Context, webhookID string,
+	ctx context.Context, webhookID string, atStart bool,
 ) {
-	// The web interface is already serving, so the webhook may have
-	// been deleted since the list was read. Opening its database then
+	if !atStart && !e.dbManager.DBExists(webhookID) {
+		return
+	}
+
+	// The web interface is serving, so the webhook may have been
+	// deleted since the list was read. Opening its database then
 	// would create the file again after the delete removed it.
 	stillExists := func() (bool, error) {
 		var count int64
@@ -777,16 +817,18 @@ func (e *Engine) recoverWebhookDeliveries(
 	}
 
 	e.recoverPendingDeliveries(
-		ctx, webhookDB, webhookID,
+		ctx, webhookDB, webhookID, atStart,
 	)
 
 	e.recoverRetryingDeliveries(
-		webhookDB, webhookID,
+		ctx, webhookDB, webhookID,
 	)
 }
 
+// recoverRetryingDeliveries recovers every retrying delivery of one
+// webhook, at start and on a tick alike.
 func (e *Engine) recoverRetryingDeliveries(
-	webhookDB *gorm.DB, webhookID string,
+	ctx context.Context, webhookDB *gorm.DB, webhookID string,
 ) {
 	var retrying []database.Delivery
 
@@ -813,6 +855,12 @@ func (e *Engine) recoverRetryingDeliveries(
 	)
 
 	for i := range retrying {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
 		if _, ok := settled[retrying[i].ID]; ok {
 			continue
 		}
@@ -824,8 +872,9 @@ func (e *Engine) recoverRetryingDeliveries(
 }
 
 // recoverSingleRetry hands an orphaned retrying delivery back
-// to its target to recompute the remaining backoff, then
-// reschedules it. Targets that do not own durable retries
+// to its target to work out what is left of its backoff, then
+// schedules its next attempt after that, or now when nothing is
+// left. Targets that do not own durable retries
 // (fire-and-forget) never produce retrying deliveries, so a
 // delivery found in that state has had its target's type
 // changed underneath it and is terminally failed.
@@ -873,6 +922,19 @@ func (e *Engine) recoverSingleRetry(
 		webhookDB, d.ID, attemptNum,
 	)
 
+	// Taken before the event is read, so that a delivery the engine
+	// already holds, most often one waiting out its backoff on a
+	// timer, costs a tick no read of its body.
+	if !e.takeForRedispatch(
+		webhookDB, d.ID, database.DeliveryStatusRetrying,
+	) {
+		return
+	}
+
+	// ScheduleRetry takes a reference of its own before it returns,
+	// and that one keeps the delivery owned through the backoff.
+	defer e.inflight.release(d.ID)
+
 	event, err := e.loadEvent(webhookDB, d.EventID)
 	if err != nil {
 		e.log.Error(
@@ -886,13 +948,9 @@ func (e *Engine) recoverSingleRetry(
 		return
 	}
 
-	task := buildRecoveryTask(
+	e.ScheduleRetry(buildRecoveryTask(
 		d, webhookID, &event, &target, attemptNum+1,
-	)
-
-	if !e.rescheduleRecovered(webhookDB, task, remaining) {
-		return
-	}
+	), remaining)
 
 	e.log.Info(
 		"recovering retrying delivery",
@@ -903,28 +961,51 @@ func (e *Engine) recoverSingleRetry(
 	)
 }
 
+// recoverPendingDeliveries recovers the pending deliveries of one
+// webhook: it marks delivered those that already have a successful
+// attempt and sends the rest.
+//
+// A delivery is created pending and leaves that state only when its
+// outcome is written, so a pending row the engine does not own is one
+// whose bookkeeping write failed, whose attempt a stop cut off, or
+// that was still queued when the engine last stopped.
+//
+// At start every pending delivery is taken, however recent, so one a
+// stop cut off moments ago is sent again now. A tick takes only those
+// untouched for pendingSweepMinAge, at most pendingSweepBatch of them.
+// The age bound is not what keeps a tick off live work: a delivery
+// waiting in deliveryCh is pending and arbitrarily old. Ownership is,
+// checked per delivery in takeForRedispatch. See inflight.go.
 func (e *Engine) recoverPendingDeliveries(
 	ctx context.Context,
 	webhookDB *gorm.DB,
 	webhookID string,
+	atStart bool,
 ) {
+	query := webhookDB.Where(
+		"status = ?", database.DeliveryStatusPending,
+	)
+
+	if !atStart {
+		query = query.
+			Where(
+				"updated_at < ?",
+				time.Now().Add(-pendingSweepMinAge),
+			).
+			Limit(pendingSweepBatch)
+	}
+
 	var deliveries []database.Delivery
 
 	// No Preload: event bodies are read one at a time in
 	// sendRecoveredDeliveries, and only for the deliveries actually
 	// being sent.
-	result := webhookDB.
-		Where(
-			"status = ?",
-			database.DeliveryStatusPending,
-		).
-		Find(&deliveries)
-
-	if result.Error != nil {
+	err := query.Find(&deliveries).Error
+	if err != nil {
 		e.log.Error(
-			"failed to query pending deliveries",
+			"failed to query pending deliveries for recovery",
 			"webhook_id", webhookID,
-			"error", result.Error,
+			"error", err,
 		)
 
 		return
@@ -940,21 +1021,6 @@ func (e *Engine) recoverPendingDeliveries(
 		"count", len(deliveries),
 	)
 
-	e.recoverPendingBatch(
-		ctx, webhookDB, webhookID, deliveries,
-	)
-}
-
-// recoverPendingBatch settles every delivery in the batch that was
-// already delivered, and re-dispatches only the rest. Both the
-// restart-time recovery and the periodic sweep go through it, so a
-// pending delivery is treated the same however it was found.
-func (e *Engine) recoverPendingBatch(
-	ctx context.Context,
-	webhookDB *gorm.DB,
-	webhookID string,
-	deliveries []database.Delivery,
-) {
 	targetMap := e.loadTargetMap(deliveries)
 
 	settled := e.reconcileDelivered(
@@ -982,11 +1048,11 @@ func (e *Engine) recoverPendingBatch(
 // wire I/O happened and was recorded, and all that is missing is the
 // status.
 //
-// Every recovery path runs this, not only the pending one. A delivery
-// abandoned at retrying can hold a successful result just as a pending
-// one can — a second attempt that reached the receiver and whose status
-// write then failed sits at retrying with success recorded — and
-// re-sending it is the same duplicate.
+// The recovery routine runs this for retrying deliveries as well as
+// pending ones. A delivery abandoned at retrying can hold a successful
+// result just as a pending one can — a second attempt that reached the
+// receiver and whose status write then failed sits at retrying with
+// success recorded — and re-sending it is the same duplicate.
 //
 // Deliveries whose result row itself never landed are not in the
 // returned set and are re-sent, recorded as the further attempt they
@@ -1073,247 +1139,8 @@ func (e *Engine) reconcileDelivered(
 	return settled
 }
 
-func (e *Engine) retrySweep(ctx context.Context) {
-	defer e.wg.Done()
-
-	ticker := time.NewTicker(retrySweepInterval)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-ticker.C:
-			e.sweepOrphanedRetries(ctx)
-		}
-	}
-}
-
-func (e *Engine) sweepOrphanedRetries(
-	ctx context.Context,
-) {
-	var webhookIDs []string
-
-	err := e.database.DB().
-		Model(&database.Webhook{}).
-		Pluck("id", &webhookIDs).Error
-	if err != nil {
-		e.log.Error(
-			"retry sweep: failed to query webhook IDs",
-			"error", err,
-		)
-
-		return
-	}
-
-	for _, webhookID := range webhookIDs {
-		select {
-		case <-ctx.Done():
-			return
-		default:
-		}
-
-		if !e.dbManager.DBExists(webhookID) {
-			continue
-		}
-
-		e.sweepWebhookRetries(ctx, webhookID)
-	}
-}
-
-func (e *Engine) sweepWebhookRetries(
-	ctx context.Context, webhookID string,
-) {
-	webhookDB, err := e.dbManager.GetDB(webhookID)
-	if err != nil {
-		e.log.Error(
-			"retry sweep: failed to get webhook database",
-			"webhook_id", webhookID,
-			"error", err,
-		)
-
-		return
-	}
-
-	var retrying []database.Delivery
-
-	err = webhookDB.
-		Where(
-			"status = ?",
-			database.DeliveryStatusRetrying,
-		).
-		Find(&retrying).Error
-	if err != nil {
-		e.log.Error(
-			"retry sweep: "+
-				"failed to query retrying deliveries",
-			"webhook_id", webhookID,
-			"error", err,
-		)
-
-		return
-	}
-
-	settled := e.reconcileDelivered(
-		webhookDB, webhookID, retrying,
-		e.loadTargetMap(retrying),
-	)
-
-	for i := range retrying {
-		select {
-		case <-ctx.Done():
-			return
-		default:
-		}
-
-		if _, ok := settled[retrying[i].ID]; ok {
-			continue
-		}
-
-		e.sweepSingleRetry(
-			webhookDB, webhookID, &retrying[i],
-		)
-	}
-
-	e.sweepWebhookPending(ctx, webhookDB, webhookID)
-}
-
-// sweepWebhookPending recovers deliveries stranded at pending.
-//
-// A delivery is created pending and leaves that state only when its
-// outcome is written, so a pending row the engine does not own is one
-// whose bookkeeping write failed — the state that used to sit there
-// until a restart, and then produce a duplicate at the receiver. The
-// sweep gives it the same reconcile-then-dispatch treatment restart
-// recovery gets, so it costs a minute rather than an operator
-// noticing.
-//
-// What keeps the sweep off live work is ownership, checked per
-// delivery in takeForRedispatch, not the age bound in this query.
-// A delivery waiting in deliveryCh is pending and arbitrarily old —
-// the channel holds 10000 tasks and 10 workers drain it — so
-// reasoning from the row's age alone re-sends it. See inflight.go.
-func (e *Engine) sweepWebhookPending(
-	ctx context.Context,
-	webhookDB *gorm.DB,
-	webhookID string,
-) {
-	var pending []database.Delivery
-
-	err := webhookDB.
-		Where(
-			"status = ? AND updated_at < ?",
-			database.DeliveryStatusPending,
-			time.Now().Add(-pendingSweepMinAge),
-		).
-		Limit(pendingSweepBatch).
-		Find(&pending).Error
-	if err != nil {
-		e.log.Error(
-			"retry sweep: "+
-				"failed to query pending deliveries",
-			"webhook_id", webhookID,
-			"error", err,
-		)
-
-		return
-	}
-
-	if len(pending) == 0 {
-		return
-	}
-
-	e.log.Info(
-		"retry sweep: recovering stranded pending deliveries",
-		"webhook_id", webhookID,
-		"count", len(pending),
-	)
-
-	e.recoverPendingBatch(ctx, webhookDB, webhookID, pending)
-}
-
-// sweepSingleRetry re-enqueues an orphaned retrying delivery
-// whose backoff window has elapsed, delegating the backoff
-// decision to the delivery's target. A delivery whose target
-// no longer owns durable retries is terminally failed.
-func (e *Engine) sweepSingleRetry(
-	webhookDB *gorm.DB,
-	webhookID string,
-	d *database.Delivery,
-) {
-	target, err := e.loadTarget(d.TargetID)
-	if err != nil {
-		// Deleted is terminal, unreadable is not; see
-		// recoverSingleRetry.
-		if errors.Is(err, gorm.ErrRecordNotFound) {
-			e.failMissingTarget(webhookDB, webhookID, d)
-
-			return
-		}
-
-		e.log.Error(
-			"retry sweep: failed to load target",
-			"delivery_id", d.ID,
-			"target_id", d.TargetID,
-			"error", err,
-		)
-
-		return
-	}
-
-	rs, ok := e.targets[target.Type].(rescheduler)
-	if !ok {
-		e.failUnretryableRetry(
-			webhookDB, webhookID, d, &target,
-		)
-
-		return
-	}
-
-	attemptNum := e.countAttempts(webhookDB, d.ID)
-
-	if !rs.backoffElapsed(
-		webhookDB, d.ID, attemptNum,
-	) {
-		return
-	}
-
-	event, err := e.loadEvent(webhookDB, d.EventID)
-	if err != nil {
-		e.log.Error(
-			"retry sweep: failed to load event",
-			"delivery_id", d.ID,
-			"event_id", d.EventID,
-			"error", err,
-		)
-
-		return
-	}
-
-	task := buildRecoveryTask(
-		d, webhookID, &event, &target, attemptNum+1,
-	)
-
-	if !e.redispatch(
-		e.retryCh, webhookDB, task,
-		database.DeliveryStatusRetrying,
-	) {
-		return
-	}
-
-	e.log.Info(
-		"retry sweep: "+
-			"recovered orphaned retrying delivery",
-		"delivery_id", d.ID,
-		"webhook_id", webhookID,
-		"attempt", attemptNum+1,
-	)
-}
-
 // failUnretryableRetry terminally fails an orphaned retrying
-// delivery whose target type no longer supports retries. Both
-// restart recovery and the periodic sweep call it, so the
-// terminal transition exists once.
+// delivery whose target type no longer supports retries.
 //
 // This is only reachable when a target's type has been changed
 // out from under an in-flight retrying delivery (or the type is
@@ -1361,8 +1188,8 @@ func (e *Engine) failUnretryableRetry(
 }
 
 // failMissingTarget terminally fails a recovered delivery, pending or
-// retrying, whose target row is gone. Restart recovery and the periodic
-// sweep call it for both statuses, so the transition exists once.
+// retrying, whose target row is gone. The recovery routine calls it for
+// both statuses.
 //
 // Until it existed those paths logged the failed lookup and moved on,
 // which left the delivery where it was for the life of the database and
@@ -1939,14 +1766,11 @@ func (e *Engine) takeForRedispatch(
 	return true
 }
 
-// queueRecovered puts an owned delivery's task on a worker channel,
-// dropping the ownership the gate took if it does not fit.
-func (e *Engine) queueRecovered(
-	ch chan<- Task, task Task,
-) bool {
+// queueRecovered puts an owned pending delivery's task on the delivery
+// channel, dropping the ownership the gate took if it does not fit.
+func (e *Engine) queueRecovered(task Task) {
 	select {
-	case ch <- task:
-		return true
+	case e.deliveryCh <- task:
 	default:
 		e.inflight.release(task.DeliveryID)
 		e.log.Warn(
@@ -1955,53 +1779,9 @@ func (e *Engine) queueRecovered(
 			"delivery_id", task.DeliveryID,
 			"webhook_id", task.WebhookID,
 		)
-
-		return false
 	}
 }
 
-// redispatch hands a recovered delivery to a worker channel through
-// takeForRedispatch, and reports whether the task was queued.
-func (e *Engine) redispatch(
-	ch chan<- Task,
-	webhookDB *gorm.DB,
-	task Task,
-	eligible database.DeliveryStatus,
-) bool {
-	if !e.takeForRedispatch(
-		webhookDB, task.DeliveryID, eligible,
-	) {
-		return false
-	}
-
-	return e.queueRecovered(ch, task)
-}
-
-// rescheduleRecovered hands an orphaned retrying delivery back to the
-// retry timer, through the same gate. It reports whether the delivery
-// was rescheduled.
-//
-// The reference taken by the gate is dropped as soon as ScheduleRetry
-// has taken its own, which it does before returning: what keeps the
-// delivery owned through the backoff window is ScheduleRetry's
-// reference, not this one.
-func (e *Engine) rescheduleRecovered(
-	webhookDB *gorm.DB, task Task, delay time.Duration,
-) bool {
-	if !e.takeForRedispatch(
-		webhookDB, task.DeliveryID,
-		database.DeliveryStatusRetrying,
-	) {
-		return false
-	}
-
-	defer e.inflight.release(task.DeliveryID)
-
-	e.ScheduleRetry(task, delay)
-
-	return true
-}
-
 // countAttemptsBatch counts the recorded attempts of every delivery
 // in a batch with one grouped query, keyed by delivery id. Deliveries
 // with no attempts are simply absent from the result, which reads back
@@ -2259,6 +2039,6 @@ func (e *Engine) sendRecoveredDeliveries(
 			attempts[deliveries[i].ID]+1,
 		)
 
-		e.queueRecovered(e.deliveryCh, task)
+		e.queueRecovered(task)
 	}
 }