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

While an http or slack target's circuit breaker turns its deliveries
away, the target's row on the webhook page says its deliveries are
paused and until when, in UTC and from now. Each of its retrying
deliveries shows as waiting, with the reason and the same time, in the
event log and on the event's page, instead of a bare "retrying".

The engine gains one read, CooldownRemaining(targetID), over the
breakers it already keeps; the handlers reach it through a one-method
interface wired like Archives.

Model: opus-5-5

internal/delivery/engine.go

@@ -143,6 +143,15 @@ type Archives interface {
 	Rename(targetID, webhookName, targetName string) error
 }
 
+// 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. Like Archives,
+// it keeps the handlers free of the engine's internals and is
+// trivially faked in tests.
+type CircuitBreakers interface {
+	CooldownRemaining(targetID string) time.Duration
+}
+
 // EngineParams are the fx dependencies for the delivery
 // engine.
 type EngineParams struct {
@@ -186,9 +195,11 @@ type Engine struct {
 	// targets maps each target type to its implementation.
 	targets map[database.TargetType]Target
 
-	// httpTarget is retained so tests can reach the HTTP
-	// target's shared client and circuit breakers.
-	httpTarget *httpTarget
+	// httpTarget and slackTarget are retained so CooldownRemaining
+	// can read their circuit breakers, and so tests can reach the
+	// HTTP target's shared client.
+	httpTarget  *httpTarget
+	slackTarget *slackTarget
 
 	// dbTarget is retained so the engine can reach the archive
 	// writer registry for eviction, renames and the idle sweep.
@@ -300,6 +311,26 @@ func (e *Engine) Rename(
 	return e.dbTarget.rename(targetID, webhookName, targetName)
 }
 
+// CooldownRemaining implements CircuitBreakers. It is
+// CircuitBreaker.CooldownRemaining for the target's breaker: how long
+// the breaker will keep turning the target's deliveries away, and
+// zero when it lets them through. A target with no breaker gets zero,
+// and reading never creates one.
+func (e *Engine) CooldownRemaining(targetID string) time.Duration {
+	for _, core := range []*httpCore{
+		e.httpTarget.httpCore, e.slackTarget.httpCore,
+	} {
+		val, ok := core.circuitBreakers.Load(targetID)
+		if ok {
+			cb, _ := val.(*CircuitBreaker)
+
+			return cb.CooldownRemaining()
+		}
+	}
+
+	return 0
+}
+
 // ScheduleRetry schedules a task to be re-enqueued onto the
 // retry channel after delay. It implements the Scheduler
 // interface the targets use to own their durable retries.

internal/delivery/engine_test.go

@@ -1018,6 +1018,42 @@ func TestGetCircuitBreaker_CreatesOnDemand(t *testing.T) {
 	)
 }
 
+// TestCooldownRemaining_ReadsHTTPAndSlackBreakers proves the engine
+// reads the cooldown of an http or a slack target's circuit breaker
+// while it is open, and zero for a target with no breaker and once the
+// breaker closes.
+func TestCooldownRemaining_ReadsHTTPAndSlackBreakers(t *testing.T) {
+	t.Parallel()
+
+	e := testEngine(t, 1)
+
+	httpID := uuid.New().String()
+	slackID := uuid.New().String()
+
+	assert.Zero(t, e.CooldownRemaining(httpID), "no breaker")
+
+	httpCB := delivery.NewTestCircuitBreaker(1, time.Hour)
+	e.ExportSetCircuitBreaker(httpID, httpCB)
+
+	slackCB := delivery.NewTestCircuitBreaker(1, time.Hour)
+	e.ExportSetSlackCircuitBreaker(slackID, slackCB)
+
+	httpCB.RecordFailure()
+	slackCB.RecordFailure()
+
+	for _, id := range []string{httpID, slackID} {
+		remaining := e.CooldownRemaining(id)
+		assert.Greater(t, remaining, 59*time.Minute)
+		assert.LessOrEqual(t, remaining, time.Hour)
+	}
+
+	httpCB.RecordSuccess()
+	slackCB.RecordSuccess()
+
+	assert.Zero(t, e.CooldownRemaining(httpID), "closed http breaker")
+	assert.Zero(t, e.CooldownRemaining(slackID), "closed slack breaker")
+}
+
 func TestParseHTTPConfig_Valid(t *testing.T) {
 	t.Parallel()
 

internal/delivery/export_test.go

@@ -212,6 +212,14 @@ func (e *Engine) ExportSetCircuitBreaker(
 	e.httpTarget.circuitBreakers.Store(targetID, cb)
 }
 
+// ExportSetSlackCircuitBreaker is ExportSetCircuitBreaker for the
+// slack target.
+func (e *Engine) ExportSetSlackCircuitBreaker(
+	targetID string, cb *CircuitBreaker,
+) {
+	e.slackTarget.circuitBreakers.Store(targetID, cb)
+}
+
 // ExportParseHTTPConfig exposes parseHTTPConfig.
 func (e *Engine) ExportParseHTTPConfig(
 	configJSON string,

internal/delivery/target.go

@@ -105,6 +105,7 @@ func (e *Engine) initTargets(client *http.Client) {
 	dbT := &databaseTarget{eng: e}
 
 	e.httpTarget = httpT
+	e.slackTarget = slackT
 	e.dbTarget = dbT
 
 	e.targets = map[database.TargetType]Target{

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
 
@@ -145,6 +147,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,47 @@ func (r *recordingArchives) Renames() []archiveRename {
 	return out
 }
 
+// testCircuitBreakers is a delivery.CircuitBreakers over real circuit
+// breakers, one per target a test asks for, so a test can trip a
+// target's breaker with RecordFailure and close it with RecordSuccess.
+type testCircuitBreakers struct {
+	mu       sync.Mutex
+	breakers map[string]*delivery.CircuitBreaker
+}
+
+// Breaker returns the target's circuit breaker, making it on first
+// use.
+func (b *testCircuitBreakers) Breaker(
+	targetID string,
+) *delivery.CircuitBreaker {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.breakers == nil {
+		b.breakers = map[string]*delivery.CircuitBreaker{}
+	}
+
+	if b.breakers[targetID] == nil {
+		b.breakers[targetID] = delivery.NewCircuitBreaker()
+	}
+
+	return b.breakers[targetID]
+}
+
+func (b *testCircuitBreakers) CooldownRemaining(
+	targetID string,
+) time.Duration {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	cb := b.breakers[targetID]
+	if cb == nil {
+		return 0
+	}
+
+	return cb.CooldownRemaining()
+}
+
 // 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 +273,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,11 @@ 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 turning deliveries away, and
+	// nil otherwise.
+	Paused *PausedView
 }
 
 // eventLogTarget is what the event log needs to know about
@@ -1181,7 +1186,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]
@@ -1305,8 +1310,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,
@@ -1329,6 +1335,10 @@ func newDeliveryViews(
 			AttemptCount:    len(rows),
 			AttemptsOmitted: omitted,
 		}
+
+		if deliveries[i].Status == database.DeliveryStatusRetrying {
+			views[i].Paused = h.pausedView(deliveries[i].TargetID)
+		}
 	}
 
 	return views

internal/handlers/target_list.go

@@ -24,6 +24,34 @@ 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 when a target's circuit breaker will let deliveries to
+// it through again: Until is the time in UTC, and Relative how long
+// that is from now.
+type PausedView struct {
+	Until    string
+	Relative string
+}
+
+// pausedView reads the target's circuit breaker, and returns nil when
+// the breaker is not turning the target's deliveries away.
+func (h *Handlers) pausedView(targetID string) *PausedView {
+	remaining := h.breakers.CooldownRemaining(targetID)
+	if remaining <= 0 {
+		return nil
+	}
+
+	until := time.Now().Add(remaining)
+
+	return &PausedView{
+		Until:    until.UTC().Format(time.TimeOnly) + " UTC",
+		Relative: humanize.Time(until),
+	}
 }
 
 // TargetDeliveries is how many of a target's deliveries became
@@ -84,6 +112,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,129 @@
+package handlers_test
+
+import (
+	"net/http"
+	"strings"
+	"testing"
+
+	"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"
+)
+
+// resumesAt is how the pages write when a paused target's deliveries
+// resume: the time in UTC, then how long that is from now.
+const resumesAt = `\d\d:\d\d:\d\d UTC \(\d+ seconds from now\)`
+
+// TestPausedTarget_ShownUntilBreakerCloses trips the circuit breaker of
+// an http target, then checks that its row on the webhook page says its
+// deliveries are paused and until when, and that its retrying delivery
+// says it is waiting and why in the event log and on the event's page,
+// while its delivered delivery and the log target are shown as before.
+// Once the breaker closes, none of the pages says so any more.
+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)
+
+	cb := breakers.Breaker(target.ID)
+	for cb.State() != delivery.CircuitOpen {
+		cb.RecordFailure()
+	}
+
+	const waiting = "waiting: target paused after repeated failures, " +
+		"resumes " + resumesAt
+
+	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 "+resumesAt,
+		list)
+	assert.Equal(t, 1, strings.Count(list, "Paused"))
+
+	log := renderSourceLogsPage(t, h, sess, wh.ID)
+	assert.Contains(t, log, "t-http: waiting")
+	assert.Contains(t, log, "t-http: delivered")
+	assert.Regexp(t, waiting, log)
+	assert.NotContains(t, log, "retrying")
+
+	page := eventPage(t, h, sess, wh.ID, retrying.ID)
+	assert.Regexp(t, waiting, page)
+	assert.NotContains(t, page, "retrying")
+
+	cb.RecordSuccess()
+
+	list = targetList(t, renderSourceDetailPage(t, h, sess, wh.ID))
+	assert.NotContains(t, list, "Paused")
+
+	log = renderSourceLogsPage(t, h, sess, wh.ID)
+	assert.Contains(t, log, "t-http: retrying")
+	assert.NotContains(t, log, "waiting")
+
+	page = eventPage(t, h, sess, wh.ID, retrying.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.
+func addDelivery(
+	t *testing.T,
+	dbMgr *database.WebhookDBManager,
+	webhookID, eventID, targetID string,
+	status database.DeliveryStatus,
+) {
+	t.Helper()
+
+	webhookDB, err := dbMgr.GetDB(webhookID)
+	require.NoError(t, err)
+
+	require.NoError(t, webhookDB.Omit(clause.Associations).Create(
+		&database.Delivery{
+			EventID:  eventID,
+			TargetID: targetID,
+			Status:   status,
+		},
+	).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()
+}

internal/resetpw/resetpw_test.go

@@ -11,6 +11,7 @@ import (
 	"path/filepath"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -142,6 +143,12 @@ func (n *noopArchives) Rename(_, _, _ string) error {
 	return nil
 }
 
+type noopCircuitBreakers struct{}
+
+func (n *noopCircuitBreakers) CooldownRemaining(string) time.Duration {
+	return 0
+}
+
 // newServerApp starts the real login path against dir: the handlers,
 // the middleware that bounds password verification, the session store
 // and the database, exactly as internal/handlers builds them.
@@ -174,6 +181,9 @@ func newServerApp(
 			session.New,
 			func() delivery.Notifier { return &noopNotifier{} },
 			func() delivery.Archives { return &noopArchives{} },
+			func() delivery.CircuitBreakers {
+				return &noopCircuitBreakers{}
+			},
 			metrics.NewRegistry,
 			metrics.New,
 			middleware.New,

internal/server/routes_test.go

@@ -62,6 +62,15 @@ func (e *noopArchives) Rename(_, _, _ string) error {
 	return nil
 }
 
+// noopCircuitBreakers satisfies handlers.New's
+// delivery.CircuitBreakers dependency with no target's deliveries
+// paused.
+type noopCircuitBreakers struct{}
+
+func (b *noopCircuitBreakers) CooldownRemaining(string) time.Duration {
+	return 0
+}
+
 // testEnv is the real router from routes.go plus the collaborators
 // tests need to seed users and forge sessions.
 type testEnv struct {
@@ -126,6 +135,9 @@ func newTestEnvWithConfig(
 			session.New,
 			func() delivery.Notifier { return &noopNotifier{} },
 			func() delivery.Archives { return &noopArchives{} },
+			func() delivery.CircuitBreakers {
+				return &noopCircuitBreakers{}
+			},
 			metrics.NewRegistry,
 			metrics.New,
 			middleware.New,