From f531cf9cdf8e33a7cdc91fe243235e80648164c9 Mon Sep 17 00:00:00 2001 From: sneak Date: Fri, 2 Oct 2026 22:04:53 +0000 Subject: [PATCH] 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 --- README.md | 8 ++ cmd/webhooker/main.go | 4 + internal/delivery/engine.go | 37 ++++++- internal/delivery/engine_test.go | 36 +++++++ internal/delivery/export_test.go | 8 ++ internal/delivery/target.go | 1 + internal/handlers/handlers.go | 29 +++--- internal/handlers/handlers_test.go | 48 +++++++++ internal/handlers/source_management.go | 16 ++- internal/handlers/target_list.go | 30 ++++++ internal/handlers/target_paused_test.go | 129 ++++++++++++++++++++++++ internal/resetpw/resetpw_test.go | 10 ++ internal/server/routes_test.go | 12 +++ templates/event_detail.html | 2 +- templates/source_detail.html | 6 ++ templates/source_logs.html | 4 +- 16 files changed, 358 insertions(+), 22 deletions(-) create mode 100644 internal/handlers/target_paused_test.go diff --git a/README.md b/README.md index 8069395..6cc736b 100644 --- a/README.md +++ b/README.md @@ -2331,6 +2331,14 @@ just delayed until the target is healthy again. A delivery already in `retrying` keeps that status without another database write each time the breaker turns it away. +While a target's breaker is turning its deliveries away, the target's row +on the webhook page says its deliveries are paused and until when, in UTC +and as a time from now, and each of its `retrying` deliveries shows as +waiting until then in the event log and on the event's page. That time is +when the cooldown ends; while a half-open breaker's probe delivery is in +flight, it is one whole cooldown from now, the delay the breaker gives the +other deliveries. + ### Metrics `/metrics` serves one Prometheus registry behind basic auth (see diff --git a/cmd/webhooker/main.go b/cmd/webhooker/main.go index 966df32..9269e99 100644 --- a/cmd/webhooker/main.go +++ b/cmd/webhooker/main.go @@ -212,6 +212,10 @@ func newApp() *fx.App { // or renaming a webhook or target reaches its archive // files. func(e *delivery.Engine) delivery.Archives { return e }, + // Wire *delivery.Engine as delivery.CircuitBreakers so + // the pages can show a target whose deliveries are + // paused. + func(e *delivery.Engine) delivery.CircuitBreakers { return e }, server.New, ), fx.Invoke( diff --git a/internal/delivery/engine.go b/internal/delivery/engine.go index 9fe588f..42ddd5d 100644 --- a/internal/delivery/engine.go +++ b/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. diff --git a/internal/delivery/engine_test.go b/internal/delivery/engine_test.go index c06a772..8b08a13 100644 --- a/internal/delivery/engine_test.go +++ b/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() diff --git a/internal/delivery/export_test.go b/internal/delivery/export_test.go index 40585aa..2a6a1e9 100644 --- a/internal/delivery/export_test.go +++ b/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, diff --git a/internal/delivery/target.go b/internal/delivery/target.go index 17c3bf5..5ccb7d4 100644 --- a/internal/delivery/target.go +++ b/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{ diff --git a/internal/handlers/handlers.go b/internal/handlers/handlers.go index 9decb63..88734a3 100644 --- a/internal/handlers/handlers.go +++ b/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 diff --git a/internal/handlers/handlers_test.go b/internal/handlers/handlers_test.go index bbf4f11..d707f92 100644 --- a/internal/handlers/handlers_test.go +++ b/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, diff --git a/internal/handlers/source_management.go b/internal/handlers/source_management.go index 3694f35..7082a61 100644 --- a/internal/handlers/source_management.go +++ b/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 diff --git a/internal/handlers/target_list.go b/internal/handlers/target_list.go index 035a575..bfac39e 100644 --- a/internal/handlers/target_list.go +++ b/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 diff --git a/internal/handlers/target_paused_test.go b/internal/handlers/target_paused_test.go new file mode 100644 index 0000000..46c2a5a --- /dev/null +++ b/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") + 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() +} diff --git a/internal/resetpw/resetpw_test.go b/internal/resetpw/resetpw_test.go index 922a8ac..b48c130 100644 --- a/internal/resetpw/resetpw_test.go +++ b/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, diff --git a/internal/server/routes_test.go b/internal/server/routes_test.go index da41aa2..af98765 100644 --- a/internal/server/routes_test.go +++ b/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, diff --git a/templates/event_detail.html b/templates/event_detail.html index aec0280..03a44eb 100644 --- a/templates/event_detail.html +++ b/templates/event_detail.html @@ -69,7 +69,7 @@
{{.Target.DisplayName}} - {{.Status}} + {{with .Paused}}waiting: target paused after repeated failures, resumes {{.Until}} ({{.Relative}}){{else}}{{.Status}}{{end}} {{.AttemptCount}} attempt{{if ne .AttemptCount 1}}s{{end}}
diff --git a/templates/source_detail.html b/templates/source_detail.html index 2aae5d7..f58697c 100644 --- a/templates/source_detail.html +++ b/templates/source_detail.html @@ -248,6 +248,12 @@ + {{with .Paused}} +
+ Deliveries Paused: + after repeated failures, until {{.Until}} ({{.Relative}}) +
+ {{end}} {{range .Config}}
{{.Label}}: diff --git a/templates/source_logs.html b/templates/source_logs.html index d900661..d6a58a0 100644 --- a/templates/source_logs.html +++ b/templates/source_logs.html @@ -31,7 +31,7 @@ {{range .Deliveries}} - {{.Target.DisplayName}}: {{.Status}} + {{.Target.DisplayName}}: {{if .Paused}}waiting{{else}}{{.Status}}{{end}} {{end}} {{.CreatedAt.Format "2006-01-02 15:04:05"}} @@ -65,7 +65,7 @@