Store events and create deliveries in the delivery engine (closes #544)

The receiver, resubmit and replay handlers now parse the request and
make one call. Engine.StoreEvent stores an event with a pending
delivery for each active target and the totals in one transaction,
then queues them; Engine.CreateReplay does the same for one replay
delivery. The handlers take *delivery.Engine, so the Notifier
interface and its bridge are gone, and the engine's notify is no
longer exported. Archives and CircuitBreakers stay because the handler
tests fake them. The tests that read a queued task moved to the
engine's tests; the handler tests use an engine that is never started.

Model: opus-5-5

internal/delivery/engine.go

@@ -1,5 +1,5 @@
-// Package delivery manages asynchronous event delivery
-// to configured targets.
+// Package delivery stores each event with its deliveries and delivers
+// them, asynchronously, to configured targets.
 package delivery
 
 import (
@@ -23,10 +23,10 @@ import (
 
 const (
 	// deliveryChannelSize is the buffer size for the delivery
-	// channel. New Tasks from the webhook handler are sent
+	// channel. The Tasks of newly stored deliveries are sent
 	// here. Workers drain this channel. Sized large enough
-	// that the webhook handler should never block under
-	// normal load.
+	// that storing an event should never block under normal
+	// load.
 	deliveryChannelSize = 10000
 
 	// retryChannelSize is the buffer size for the retry
@@ -127,17 +127,11 @@ type Task struct {
 	AttemptNum int
 }
 
-// Notifier is the interface for notifying the delivery
-// engine about new deliveries.
-type Notifier interface {
-	Notify(tasks []Task)
-}
-
 // CircuitBreakers is how the handlers read a target's circuit
 // breaker, so the webhook page and the event log can say that
-// deliveries to the target are paused and until when. It keeps the
-// handlers free of the engine's internals and is trivially faked in
-// tests.
+// deliveries to the target are paused and until when. The handlers
+// hold the engine itself as well; they read breakers through this
+// interface so that a test can make one read as open.
 type CircuitBreakers interface {
 	StateAndCooldown(targetID string) (CircuitState, time.Duration)
 }
@@ -232,38 +226,6 @@ func New(
 	return e
 }
 
-// Notify signals the delivery engine that new deliveries
-// are ready.
-func (e *Engine) Notify(tasks []Task) {
-	for i := range tasks {
-		// Owned before it is queued, and until the worker that runs
-		// it returns. A task can sit in a 10000-deep channel for a
-		// long time on a healthy system, and nothing may re-send it
-		// while it waits. See inflight.go.
-		if !e.inflight.retainIdle(tasks[i].DeliveryID) {
-			e.log.Warn(
-				"delivery already in flight, not queued again",
-				"delivery_id", tasks[i].DeliveryID,
-				"event_id", tasks[i].EventID,
-			)
-
-			continue
-		}
-
-		select {
-		case e.deliveryCh <- tasks[i]:
-		default:
-			e.inflight.release(tasks[i].DeliveryID)
-			e.log.Warn(
-				"delivery channel full, "+
-					"task will be recovered by the sweep",
-				"delivery_id", tasks[i].DeliveryID,
-				"event_id", tasks[i].EventID,
-			)
-		}
-	}
-}
-
 // StateAndCooldown implements CircuitBreakers. It returns the state of
 // the target's circuit breaker and, while the breaker is open, what is
 // left of its cooldown; the cooldown is zero once that has passed and
@@ -324,6 +286,38 @@ func (e *Engine) ScheduleRetry(
 	})
 }
 
+// notify queues the tasks of deliveries already stored. StoreEvent and
+// CreateReplay call it once their transaction has committed.
+func (e *Engine) notify(tasks []Task) {
+	for i := range tasks {
+		// Owned before it is queued, and until the worker that runs
+		// it returns. A task can sit in a 10000-deep channel for a
+		// long time on a healthy system, and nothing may re-send it
+		// while it waits. See inflight.go.
+		if !e.inflight.retainIdle(tasks[i].DeliveryID) {
+			e.log.Warn(
+				"delivery already in flight, not queued again",
+				"delivery_id", tasks[i].DeliveryID,
+				"event_id", tasks[i].EventID,
+			)
+
+			continue
+		}
+
+		select {
+		case e.deliveryCh <- tasks[i]:
+		default:
+			e.inflight.release(tasks[i].DeliveryID)
+			e.log.Warn(
+				"delivery channel full, "+
+					"task will be recovered by the sweep",
+				"delivery_id", tasks[i].DeliveryID,
+				"event_id", tasks[i].EventID,
+			)
+		}
+	}
+}
+
 // userAgent is the User-Agent header of every http and slack
 // delivery request: the program name and the running build's
 // version.
@@ -472,9 +466,9 @@ func (e *Engine) processNewTask(
 	}
 
 	// The recovery routine's run at start can send and release this
-	// delivery before the receiver's Notify queues it. Ownership cannot
-	// refuse a delivery nobody holds, so the row decides whether it
-	// still needs sending.
+	// delivery before StoreEvent or CreateReplay queues it. Ownership
+	// cannot refuse a delivery nobody holds, so the row decides whether
+	// it still needs sending.
 	row, err := e.loadDelivery(webhookDB, task.DeliveryID)
 	if err != nil {
 		e.log.Error(
@@ -899,7 +893,7 @@ func (e *Engine) recoverSingleRetry(
 		return
 	}
 
-	e.ScheduleRetry(buildRecoveryTask(
+	e.ScheduleRetry(buildTask(
 		d, webhookID, &event, &target, attemptNum+1,
 	), remaining)
 
@@ -921,7 +915,7 @@ func (e *Engine) recoverSingleRetry(
 // it stopped working on before then: for example one whose bookkeeping
 // write failed, whose attempt a stop cut off, that was still queued
 // when the engine last stopped, whose task was dropped because the
-// delivery channel was full (by Notify or by queueRecovered), or whose
+// delivery channel was full (by notify or by queueRecovered), or whose
 // worker could not open the webhook's database.
 //
 // At start every pending delivery is taken, however recent, so one a
@@ -1822,7 +1816,11 @@ func (e *Engine) loadTarget(
 	return target, nil
 }
 
-func buildRecoveryTask(
+// buildTask builds the task that carries delivery d of event to target
+// through the engine's channels, as attempt attemptNum. The body is
+// carried only when it is under MaxInlineBodySize; a larger one is read
+// from the event row when the attempt runs (see hydrateEvent).
+func buildTask(
 	d *database.Delivery,
 	webhookID string,
 	event *database.Event,
@@ -1998,7 +1996,7 @@ func (e *Engine) sendRecoveredDeliveries(
 			continue
 		}
 
-		task := buildRecoveryTask(
+		task := buildTask(
 			&deliveries[i], webhookID, &event, &target,
 			attempts[deliveries[i].ID]+1,
 		)