Show a target paused by its circuit breaker (closes #385)

While an http or slack target's circuit breaker is open, the target's
row on the webhook page says its deliveries are paused until the
cooldown ends, then one is sent to test the target. Each of its
retrying deliveries shows as waiting in the event log and on the
event's page, with the earliest it can be tried next: the later of the
cooldown's end and the end of its own backoff, dated when not today
in UTC. While the breaker is half-open, the row says deliveries are
held while one delivery tests the target, with no time, and deliveries
keep their plain status.

The engine gains one read, StateAndCooldown(targetID), taking a
breaker's state and remaining cooldown under one lock; the handlers
reach it through a one-method interface wired like Archives.

Model: opus-5-5

internal/handlers/delivery_result_view.go

@@ -1,6 +1,8 @@
 package handlers
 
 import (
+	"time"
+
 	"sneak.berlin/go/webhooker/internal/delivery"
 )
 
@@ -24,8 +26,8 @@ const maxRenderedResponseBytes = 4096
 // bytes rather than characters, and they make SQLite do the
 // cut, so an oversized stored response never becomes a Go
 // string at all.
-const deliveryResultColumns = "delivery_id, attempt_num, success, " +
-	"status_code, error, duration, " +
+const deliveryResultColumns = "delivery_id, attempt_num, created_at, " +
+	"success, status_code, error, duration, " +
 	"substr(cast(response_body as blob), 1, ?) AS response_body, " +
 	"length(cast(response_body as blob)) AS response_bytes"
 
@@ -100,6 +102,7 @@ func (v DeliveryResultView) HasStatusCode() bool {
 type deliveryResultRow struct {
 	DeliveryID    string
 	AttemptNum    int
+	CreatedAt     time.Time
 	Success       bool
 	StatusCode    int
 	Error         string

internal/handlers/handlers.go

@@ -57,19 +57,20 @@ var errVerificationBusy = errors.New(
 type HandlersParams struct {
 	fx.In
 
-	Logger       *logger.Logger
-	Globals      *globals.Globals
-	Config       *config.Config
-	Database     *database.Database
-	WebhookDBMgr *database.WebhookDBManager
-	Healthcheck  *healthcheck.Healthcheck
-	Session      *session.Session
-	Middleware   *middleware.Middleware
-	Notifier     delivery.Notifier
-	Archives     delivery.Archives
-	SSRFGuard    *delivery.Guard
-	Metrics      *metrics.Set
-	Registry     *prometheus.Registry
+	Logger          *logger.Logger
+	Globals         *globals.Globals
+	Config          *config.Config
+	Database        *database.Database
+	WebhookDBMgr    *database.WebhookDBManager
+	Healthcheck     *healthcheck.Healthcheck
+	Session         *session.Session
+	Middleware      *middleware.Middleware
+	Notifier        delivery.Notifier
+	Archives        delivery.Archives
+	CircuitBreakers delivery.CircuitBreakers
+	SSRFGuard       *delivery.Guard
+	Metrics         *metrics.Set
+	Registry        *prometheus.Registry
 }
 
 // Handlers provides HTTP handler methods for all application
@@ -84,6 +85,7 @@ type Handlers struct {
 	mw        *middleware.Middleware
 	notifier  delivery.Notifier
 	archives  delivery.Archives
+	breakers  delivery.CircuitBreakers
 	mtr       *metrics.Set
 	templates map[string]*template.Template
 
@@ -148,6 +150,7 @@ func New(
 	s.mw = params.Middleware
 	s.notifier = params.Notifier
 	s.archives = params.Archives
+	s.breakers = params.CircuitBreakers
 	s.mtr = params.Metrics
 	s.ssrf = params.SSRFGuard
 

internal/handlers/handlers_test.go

@@ -9,6 +9,7 @@ import (
 	"net/http/httptest"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -181,6 +182,40 @@ func (r *recordingArchives) Renames() []archiveRename {
 	return out
 }
 
+// testCircuitBreakers is a delivery.CircuitBreakers that reports, for
+// each target, the circuit state and cooldown a test gave it with Set,
+// and a closed breaker for any other target.
+type testCircuitBreakers struct {
+	mu        sync.Mutex
+	states    map[string]delivery.CircuitState
+	cooldowns map[string]time.Duration
+}
+
+// Set makes the target's breaker read as state, with cooldown left.
+func (b *testCircuitBreakers) Set(
+	targetID string, state delivery.CircuitState, cooldown time.Duration,
+) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.states == nil {
+		b.states = map[string]delivery.CircuitState{}
+		b.cooldowns = map[string]time.Duration{}
+	}
+
+	b.states[targetID] = state
+	b.cooldowns[targetID] = cooldown
+}
+
+func (b *testCircuitBreakers) StateAndCooldown(
+	targetID string,
+) (delivery.CircuitState, time.Duration) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	return b.states[targetID], b.cooldowns[targetID]
+}
+
 // newTestApp returns an app whose RequireStart fails the test when
 // starting takes longer than fx's default start timeout of 15s. That
 // limit catches a start that hangs, not a busy host: measured with make
@@ -231,6 +266,12 @@ func newTestAppWithConfig(
 			func(r *recordingArchives) delivery.Archives {
 				return r
 			},
+			func() *testCircuitBreakers {
+				return &testCircuitBreakers{}
+			},
+			func(b *testCircuitBreakers) delivery.CircuitBreakers {
+				return b
+			},
 			metrics.NewRegistry,
 			metrics.New,
 			middleware.New,

internal/handlers/source_management.go

@@ -109,6 +109,10 @@ type DeliveryView struct {
 	// the middle of Results. The page must show it, or the
 	// bound would hide history rather than fold it.
 	AttemptsOmitted int
+
+	// Paused is set while the delivery is retrying and its
+	// target's circuit breaker is open, and nil otherwise.
+	Paused *PausedView
 }
 
 // eventLogTarget is what the event log needs to know about
@@ -1290,7 +1294,7 @@ func (h *Handlers) eventLogViews(
 	}
 
 	for i := range rows {
-		result[i].Deliveries = newDeliveryViews(
+		result[i].Deliveries = h.newDeliveryViews(
 			eventDeliveries[i], targetMap, attempts,
 		)
 		result[i].ResubmitCount = resubmits[rows[i].ID]
@@ -1414,8 +1418,9 @@ func (h *Handlers) loadDeliveryResults(
 
 // newDeliveryViews projects deliveries for rendering,
 // resolving each one's target to its display-safe view and
-// each one's attempts through that target's redactor.
-func newDeliveryViews(
+// each one's attempts through that target's redactor. A
+// retrying delivery also reads its target's circuit breaker.
+func (h *Handlers) newDeliveryViews(
 	deliveries []database.Delivery,
 	targetMap map[string]eventLogTarget,
 	attempts map[string][]deliveryResultRow,
@@ -1438,6 +1443,12 @@ func newDeliveryViews(
 			AttemptCount:    len(rows),
 			AttemptsOmitted: omitted,
 		}
+
+		if deliveries[i].Status == database.DeliveryStatusRetrying {
+			views[i].Paused = h.deliveryPausedView(
+				deliveries[i].TargetID, rows,
+			)
+		}
 	}
 
 	return views

internal/handlers/target_list.go

@@ -24,6 +24,84 @@ type TargetRowView struct {
 	// Archive is a database target's archive file, and nil for a target
 	// of any other type.
 	Archive *ArchiveFileView
+
+	// Paused is set while the target's circuit breaker is turning its
+	// deliveries away, and nil otherwise.
+	Paused *PausedView
+}
+
+// PausedView is a target's circuit breaker turning deliveries away.
+// While the breaker is open, Until is a time in UTC, and Relative how
+// long that is from now: on the target's row, when the cooldown ends;
+// on a delivery, the earliest it can be tried next. While it is
+// half-open both are empty: the cooldown has ended, and the target's
+// deliveries are held while one delivery tests whether the target has
+// recovered.
+type PausedView struct {
+	Until    string
+	Relative string
+}
+
+// pausedView reads the target's circuit breaker for its row, and
+// returns nil when the breaker lets the target's deliveries through.
+func (h *Handlers) pausedView(targetID string) *PausedView {
+	state, cooldown := h.breakers.StateAndCooldown(targetID)
+
+	switch {
+	case state == delivery.CircuitHalfOpen:
+		return &PausedView{}
+	case state == delivery.CircuitOpen && cooldown > 0:
+		return newPausedView(time.Now().Add(cooldown))
+	default:
+		return nil
+	}
+}
+
+// deliveryPausedView reads the circuit breaker of a retrying delivery's
+// target. While it is open, it says the earliest the delivery can be
+// tried next: the later of the cooldown's end and the end of the
+// delivery's own backoff after its last attempt. It is only the
+// earliest: when the cooldown ends, one of the target's waiting
+// deliveries is sent to test it while the others wait at least one more
+// cooldown. Otherwise it returns nil, half-open included, since the
+// delivery may then be the one being sent to test the target.
+func (h *Handlers) deliveryPausedView(
+	targetID string, attempts []deliveryResultRow,
+) *PausedView {
+	state, cooldown := h.breakers.StateAndCooldown(targetID)
+	if state != delivery.CircuitOpen || cooldown <= 0 {
+		return nil
+	}
+
+	next := time.Now().Add(cooldown)
+
+	if len(attempts) > 0 {
+		last := attempts[len(attempts)-1]
+
+		backoffEnd := last.CreatedAt.Add(delivery.Backoff(last.AttemptNum))
+		if backoffEnd.After(next) {
+			next = backoffEnd
+		}
+	}
+
+	return newPausedView(next)
+}
+
+// newPausedView is a PausedView of deliveries paused until the given
+// time. A time not on the current UTC day is written with its date, as
+// the event log writes its times.
+func newPausedView(until time.Time) *PausedView {
+	until = until.UTC()
+
+	layout := time.TimeOnly
+	if until.Format(time.DateOnly) != time.Now().UTC().Format(time.DateOnly) {
+		layout = time.DateTime
+	}
+
+	return &PausedView{
+		Until:    until.Format(layout) + " UTC",
+		Relative: humanize.Time(until),
+	}
 }
 
 // TargetDeliveries is how many of a target's deliveries became
@@ -84,6 +162,8 @@ func (h *Handlers) targetRows(
 		if targets[i].Type == database.TargetTypeDatabase {
 			rows[i].Archive = h.archiveFileView(webhook, &targets[i])
 		}
+
+		rows[i].Paused = h.pausedView(targets[i].ID)
 	}
 
 	return rows

internal/handlers/target_paused_test.go

@@ -0,0 +1,209 @@
+package handlers_test
+
+import (
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/gorm/clause"
+	"sneak.berlin/go/webhooker/internal/database"
+	"sneak.berlin/go/webhooker/internal/delivery"
+	"sneak.berlin/go/webhooker/internal/handlers"
+	"sneak.berlin/go/webhooker/internal/session"
+)
+
+// cooldownEnds is how the pages write the end of a paused target's
+// breaker's cooldown: the time in UTC, with its date when that falls on
+// another UTC day, then how long that is from now.
+const cooldownEnds = `(\d{4}-\d\d-\d\d )?\d\d:\d\d:\d\d UTC ` +
+	`\(\d+ seconds from now\)`
+
+// TestPausedTarget_ShownUntilBreakerCloses takes an http target's
+// circuit breaker from open through half-open to closed.
+//
+// Open, the target's row on the webhook page says its deliveries are
+// paused and until when, and each retrying delivery says it is waiting
+// and why in the event log and on the event's page, with the earliest
+// it can be tried next: the later of the cooldown's end and the end of
+// its own backoff, with the date when that is another UTC day.
+// Half-open, the row says deliveries are held while one delivery tests
+// the target, with no time, and no delivery says it is waiting. Closed,
+// the pages say neither. The delivered delivery and the log target are
+// shown as before throughout.
+func TestPausedTarget_ShownUntilBreakerCloses(t *testing.T) {
+	t.Parallel()
+
+	var (
+		h        *handlers.Handlers
+		sess     *session.Session
+		db       *database.Database
+		dbMgr    *database.WebhookDBManager
+		breakers *testCircuitBreakers
+	)
+
+	app := newTestApp(t, &h, &sess, &db, &dbMgr, &breakers)
+	app.RequireStart()
+
+	t.Cleanup(app.RequireStop)
+
+	wh := seedWebhook(t, db)
+	target := seedTarget(t, db, wh.ID, database.TargetTypeHTTP)
+	seedTarget(t, db, wh.ID, database.TargetTypeLog)
+
+	retrying := seedStoredEvent(t, dbMgr, wh.ID, `{"n":1}`)
+	addDelivery(t, dbMgr, wh.ID, retrying.ID, target.ID,
+		database.DeliveryStatusRetrying)
+
+	delivered := seedStoredEvent(t, dbMgr, wh.ID, `{"n":2}`)
+	addDelivery(t, dbMgr, wh.ID, delivered.ID, target.ID,
+		database.DeliveryStatusDelivered)
+
+	// This delivery's 18th attempt failed a minute ago, so its own
+	// backoff ends over a day from now: long after the cooldown, and on
+	// another UTC day, so the page shows the date.
+	backedOff := seedStoredEvent(t, dbMgr, wh.ID, `{"n":3}`)
+	backedOffID := addDelivery(t, dbMgr, wh.ID, backedOff.ID, target.ID,
+		database.DeliveryStatusRetrying)
+
+	failedAt := time.Now().Add(-time.Minute).Truncate(time.Second)
+	addFailedAttempt(t, dbMgr, wh.ID, backedOffID, 18, failedAt)
+
+	backoffEnds := failedAt.Add(delivery.Backoff(18)).UTC().
+		Format("2006-01-02 15:04:05") + " UTC (1 day from now)"
+
+	const waiting = "waiting: target paused after repeated failures, " +
+		"next try no earlier than "
+
+	breakers.Set(target.ID, delivery.CircuitOpen, 30*time.Second)
+
+	list := targetList(t, renderSourceDetailPage(t, h, sess, wh.ID))
+	assert.Regexp(t, "t-http http Active Edit Deactivate Delete "+
+		"Deliveries Paused: after repeated failures, until "+cooldownEnds+
+		", then one waiting delivery is sent to test the target while "+
+		"the others wait at least one more cooldown", list)
+	assert.Equal(t, 1, strings.Count(list, "Paused"))
+
+	log := renderSourceLogsPage(t, h, sess, wh.ID)
+	assert.Equal(t, 2, strings.Count(log, "t-http: waiting"))
+	assert.Contains(t, log, "t-http: delivered")
+	assert.Regexp(t, waiting+cooldownEnds, log)
+	assert.Contains(t, log, waiting+backoffEnds)
+	assert.NotContains(t, log, "retrying")
+
+	page := eventPage(t, h, sess, wh.ID, retrying.ID)
+	assert.Regexp(t, waiting+cooldownEnds, page)
+	assert.NotContains(t, page, "retrying")
+
+	page = eventPage(t, h, sess, wh.ID, backedOff.ID)
+	assert.Contains(t, page, waiting+backoffEnds)
+	assert.NotContains(t, page, "seconds from now")
+
+	breakers.Set(target.ID, delivery.CircuitHalfOpen, 0)
+
+	list = targetList(t, renderSourceDetailPage(t, h, sess, wh.ID))
+	assert.Contains(t, list, "t-http http Active Edit Deactivate Delete "+
+		"Deliveries Paused: held while one delivery tests whether the "+
+		"target has recovered")
+	assert.NotContains(t, list, "UTC")
+
+	assertRetryingNotWaiting(t, h, sess, wh.ID, retrying, backedOff)
+
+	breakers.Set(target.ID, delivery.CircuitClosed, 0)
+
+	list = targetList(t, renderSourceDetailPage(t, h, sess, wh.ID))
+	assert.NotContains(t, list, "Paused")
+
+	assertRetryingNotWaiting(t, h, sess, wh.ID, retrying, backedOff)
+}
+
+// assertRetryingNotWaiting checks that the event log and each event's
+// page show the http target's delivery of the event as retrying, and
+// none of them as waiting.
+func assertRetryingNotWaiting(
+	t *testing.T,
+	h *handlers.Handlers,
+	sess *session.Session,
+	webhookID string,
+	events ...*database.Event,
+) {
+	t.Helper()
+
+	log := renderSourceLogsPage(t, h, sess, webhookID)
+	assert.Equal(t, len(events), strings.Count(log, "t-http: retrying"))
+	assert.NotContains(t, log, "waiting")
+
+	for _, event := range events {
+		page := eventPage(t, h, sess, webhookID, event.ID)
+		assert.Contains(t, page, ">retrying</span>")
+		assert.NotContains(t, page, "waiting")
+	}
+}
+
+// addDelivery records a delivery of the event to the target, with the
+// given status, in the webhook's own database, and returns its ID.
+func addDelivery(
+	t *testing.T,
+	dbMgr *database.WebhookDBManager,
+	webhookID, eventID, targetID string,
+	status database.DeliveryStatus,
+) string {
+	t.Helper()
+
+	webhookDB, err := dbMgr.GetDB(webhookID)
+	require.NoError(t, err)
+
+	dlv := &database.Delivery{
+		EventID:  eventID,
+		TargetID: targetID,
+		Status:   status,
+	}
+
+	require.NoError(t, webhookDB.Omit(clause.Associations).Create(
+		dlv,
+	).Error)
+
+	return dlv.ID
+}
+
+// addFailedAttempt records the delivery's failed attempt attemptNum,
+// made at the given time.
+func addFailedAttempt(
+	t *testing.T,
+	dbMgr *database.WebhookDBManager,
+	webhookID, deliveryID string,
+	attemptNum int,
+	at time.Time,
+) {
+	t.Helper()
+
+	webhookDB, err := dbMgr.GetDB(webhookID)
+	require.NoError(t, err)
+
+	require.NoError(t, webhookDB.Omit(clause.Associations).Create(
+		&database.DeliveryResult{
+			BaseModel:  database.BaseModel{CreatedAt: at},
+			DeliveryID: deliveryID,
+			AttemptNum: attemptNum,
+			Error:      "connection refused",
+		},
+	).Error)
+}
+
+// eventPage runs the real event page handler and returns the
+// rendered HTML.
+func eventPage(
+	t *testing.T,
+	h *handlers.Handlers,
+	sess *session.Session,
+	webhookID, eventID string,
+) string {
+	t.Helper()
+
+	w := serveEventPage(t, h, sess, webhookID, eventID)
+	require.Equal(t, http.StatusOK, w.Code)
+
+	return w.Body.String()
+}