sneak/webhooker
1
0
Fork 0
Code Issues 5 Pull Requests 1 Actions 61 Packages Projects Releases Wiki Activity

Store a target's number of delivery attempts, at least one (closes #540)
check / check (push) Waiting to run
Details

Browse Source 
Target.MaxRetries (max_retries) becomes Target.MaxAttempts
(max_attempts), and the form field, parser and ceiling are renamed to
match. The column is NOT NULL and a check constraint refuses a value
below 1, so every target, whatever its type, is stored with at least
1. 1 is a single attempt with no circuit breaker; above 1 retries
behind the breaker up to that many attempts. Both target forms refuse
0 and less; a blank field on the add target form gives 1, and on the
edit form keeps the stored value. The target page shows the plain
number, and the README's two paragraphs on 0 against 1 become one
sentence.

Model: opus-5-5
This commit is contained in:
sneak
2026-10-15 17:04:48 +00:00
parent 147d58dd96
commit 11fa383503
51 changed files with 1027 additions and 820 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/delivery/archive_sweeper_test.go
+6 -5
View File
@@ -124,11 +124,12 @@ func (env *archiveEnv) addDatabaseTarget(
t.Helper()
tgt := &database.Target{
WebhookID: webhookID,
Name: "Archive",
Type: database.TargetTypeDatabase,
Active: true,
Config: configJSON,
WebhookID: webhookID,
Name: "Archive",
Type: database.TargetTypeDatabase,
Active: true,
Config: configJSON,
MaxAttempts: 1,
}
require.NoError(
t,
internal/delivery/engine.go
+7 -7
View File
@@ -107,7 +107,7 @@ type Task struct {
TargetName string
TargetType database.TargetType
TargetConfig string
MaxRetries int
MaxAttempts int
Method string
RawQuery string
@@ -1769,10 +1769,10 @@ func buildEventFromTask(task *Task) database.Event {
func buildTargetFromTask(task *Task) database.Target {
target := database.Target{
Name: task.TargetName,
Type: task.TargetType,
Config: task.TargetConfig,
MaxRetries: task.MaxRetries,
Name: task.TargetName,
Type: task.TargetType,
Config: task.TargetConfig,
MaxAttempts: task.MaxAttempts,
}
target.ID = task.TargetID
@@ -2028,7 +2028,7 @@ func (e *Engine) countAttemptsBatch(
// One delivery id per recorded attempt, tallied here rather than
// grouped in SQL: internal/gormlog forbids (*gorm.DB).Scan, which
// a GROUP BY into a struct would need, and an attempt row per
// delivery is bounded by the target's MaxRetries.
// delivery is bounded by the target's MaxAttempts.
var attemptIDs []string
err := webhookDB.
@@ -2106,7 +2106,7 @@ func buildRecoveryTask(
TargetName: target.Name,
TargetType: target.Type,
TargetConfig: target.Config,
MaxRetries: target.MaxRetries,
MaxAttempts: target.MaxAttempts,
Method: event.Method,
RawQuery: event.RawQuery,
Headers: event.Headers,
internal/delivery/engine_integration_test.go
+26 -25
View File
@@ -170,7 +170,7 @@ func iTask(
d database.Delivery,
event database.Event,
webhookID, targetID, name, config string,
maxRetries, attemptNum int,
maxAttempts, attemptNum int,
body *string,
) delivery.Task {
return delivery.Task{
@@ -181,7 +181,7 @@ func iTask(
TargetName: name,
TargetType: database.TargetTypeHTTP,
TargetConfig: config,
MaxRetries: maxRetries,
MaxAttempts: maxAttempts,
Method: event.Method,
Headers: event.Headers,
ContentType: event.ContentType,
@@ -242,7 +242,7 @@ func TestProcessNewTask_InlineBody(t *testing.T) {
cfg := iHTTPConfig(ts.URL)
task := iTask(
d, event, s.WebhookID, targetID,
"test-target", cfg, 0, 1, &bodyStr,
"test-target", cfg, 1, 1, &bodyStr,
)
s.Engine.ExportProcessNewTask(
@@ -293,7 +293,7 @@ func TestProcessNewTask_LargeBody_FetchFromDB(
cfg := iHTTPConfig(ts.URL)
task := iTask(
d, event, s.WebhookID, targetID,
"test-large", cfg, 0, 1, nil,
"test-large", cfg, 1, 1, nil,
)
s.Engine.ExportProcessNewTask(
@@ -320,7 +320,7 @@ func TestProcessNewTask_InvalidWebhookID(t *testing.T) {
TargetName: "test",
TargetType: database.TargetTypeHTTP,
TargetConfig: iHTTPConfig("http://localhost:9999"),
MaxRetries: 0,
MaxAttempts: 1,
Body: nil,
AttemptNum: 1,
}
@@ -506,7 +506,7 @@ func TestWorkerLifecycle_StartStop(t *testing.T) {
task := iTask(
d, event, s.WebhookID, targetID,
"lifecycle-test", "",
0, 1, &bodyStr,
1, 1, &bodyStr,
)
task.TargetType = database.TargetTypeLog
@@ -645,7 +645,7 @@ func TestRecoverPendingDeliveries(t *testing.T) {
iCreateTarget(t, s.MainDB, targetID,
s.WebhookID, "recovery-target",
database.TargetTypeLog, "", 0,
database.TargetTypeLog, "", 1,
)
iSeedPendingDeliveries(
@@ -686,17 +686,17 @@ func iCreateTarget(
targetID, webhookID, name string,
targetType database.TargetType,
config string,
maxRetries int,
maxAttempts int,
) {
t.Helper()
target := database.Target{
WebhookID: webhookID,
Name: name,
Type: targetType,
Active: true,
Config: config,
MaxRetries: maxRetries,
WebhookID: webhookID,
Name: name,
Type: targetType,
Active: true,
Config: config,
MaxAttempts: maxAttempts,
}
target.ID = targetID
@@ -1077,7 +1077,7 @@ func TestRecoverInFlight_WithPendingDeliveries(
iCreateTarget(t, s.MainDB, targetID,
s.WebhookID, "recover-target",
database.TargetTypeLog, "", 0,
database.TargetTypeLog, "", 1,
)
event := iSeedEvent(
@@ -1232,7 +1232,7 @@ func TestDeliverHTTP_CustomTargetHeaders(t *testing.T) {
task := iTask(
d, event, s.WebhookID, targetID,
"auth-target", string(cfgJSON),
0, 1, &bodyStr,
1, 1, &bodyStr,
)
s.Engine.ExportProcessNewTask(
@@ -1277,7 +1277,7 @@ func TestDeliverHTTP_TargetTimeout(t *testing.T) {
task, d := iHTTPTaskAndDelivery(
event, del, "timeout-target",
string(cfgJSON), 0, 1,
string(cfgJSON), 1, 1,
)
e.ExportDeliverHTTP(context.TODO(), db, d, task)
@@ -1339,7 +1339,7 @@ func TestDeliverHTTP_CutsStoredResponseAtMaxBodyLog(
bodyStr := event.Body
task := iTask(
del, event, s.WebhookID, targetID,
"cut-target", cfgJSON, 0, 1, &bodyStr,
"cut-target", cfgJSON, 1, 1, &bodyStr,
)
s.Engine.ExportProcessNewTask(context.TODO(), &task)
@@ -1403,7 +1403,7 @@ func iHTTPTaskAndDelivery(
event database.Event,
del database.Delivery,
name, config string,
maxRetries, attemptNum int,
maxAttempts, attemptNum int,
) (*delivery.Task, *database.Delivery) {
task := &delivery.Task{
DeliveryID: del.ID,
@@ -1413,7 +1413,7 @@ func iHTTPTaskAndDelivery(
TargetName: name,
TargetType: database.TargetTypeHTTP,
TargetConfig: config,
MaxRetries: maxRetries,
MaxAttempts: maxAttempts,
AttemptNum: attemptNum,
}
@@ -1423,9 +1423,10 @@ func iHTTPTaskAndDelivery(
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
Name: name,
Type: database.TargetTypeHTTP,
Config: config,
Name: name,
Type: database.TargetTypeHTTP,
Config: config,
MaxAttempts: maxAttempts,
},
}
d.ID = del.ID
@@ -1466,7 +1467,7 @@ func TestDeliverHTTP_InvalidConfig(t *testing.T) {
)
task, d := iHTTPTaskAndDelivery(
event, del, "bad-config", `not-json`, 0, 1,
event, del, "bad-config", `not-json`, 1, 1,
)
e.ExportDeliverHTTP(context.TODO(), db, d, task)
@@ -1490,7 +1491,7 @@ func TestDeliverHTTP_InvalidConfigUnrecordedStaysPending(t *testing.T) {
)
task, d := iHTTPTaskAndDelivery(
event, del, "bad-config", `not-json`, 0, 1,
event, del, "bad-config", `not-json`, 1, 1,
)
require.NoError(t, db.Exec("drop table delivery_results").Error)
internal/delivery/engine_lifecycle_test.go
+12 -11
View File
@@ -149,7 +149,7 @@ func seedLogTask(
bodyStr := event.Body
task := iTask(
d, event, s.WebhookID, targetID,
"hook-context-test", "", 0, 1, &bodyStr,
"hook-context-test", "", 1, 1, &bodyStr,
)
task.TargetType = database.TargetTypeLog
@@ -278,7 +278,7 @@ func TestEngine_StopHookHonoursStopTimeout(t *testing.T) {
}
// seedHeldDelivery gives the setup's webhook an http target with
// maxRetries, pointed at a receiver that keeps every request open until
// maxAttempts, pointed at a receiver that keeps every request open until
// the client gives up on it, and seeds one delivery to that target at
// status. It returns the task that sends attempt attemptNum of the
// delivery, and a channel that is closed when a request reaches the
@@ -286,7 +286,7 @@ func TestEngine_StopHookHonoursStopTimeout(t *testing.T) {
func seedHeldDelivery(
t *testing.T,
s iSetup,
maxRetries int,
maxAttempts int,
status database.DeliveryStatus,
attemptNum int,
) (delivery.Task, <-chan struct{}) {
@@ -315,7 +315,7 @@ func seedHeldDelivery(
iCreateWebhook(t, s.MainDB, s.WebhookID, "hook")
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, "held",
database.TargetTypeHTTP, cfg, maxRetries,
database.TargetTypeHTTP, cfg, maxAttempts,
)
event := iSeedEvent(t, s.WebhookDB, s.WebhookID, `{"held":true}`)
@@ -325,7 +325,7 @@ func seedHeldDelivery(
return iTask(
d, event, s.WebhookID, targetID, "held", cfg,
maxRetries, attemptNum, &bodyStr,
maxAttempts, attemptNum, &bodyStr,
), arrived
}
@@ -380,7 +380,7 @@ func TestEngine_StopLeavesFireAndForgetAttemptPending(t *testing.T) {
lc := startEngineViaHook(t, s.Engine)
task, arrived := seedHeldDelivery(
t, s, 0, database.DeliveryStatusPending, 1,
t, s, 1, database.DeliveryStatusPending, 1,
)
s.Engine.Notify([]delivery.Task{task})
@@ -404,13 +404,13 @@ func TestEngine_StopLeavesFireAndForgetAttemptPending(t *testing.T) {
func TestEngine_StopLeavesLastRetryAttemptRetrying(t *testing.T) {
t.Parallel()
const maxRetries = 2
const maxAttempts = 2
s := newISetup(t)
lc := startEngineViaHook(t, s.Engine)
task, arrived := seedHeldDelivery(
t, s, maxRetries, database.DeliveryStatusRetrying, maxRetries,
t, s, maxAttempts, database.DeliveryStatusRetrying, maxAttempts,
)
iSeedFailedResult(t, s.WebhookDB, task.DeliveryID)
@@ -435,9 +435,10 @@ func deliverToArchive(t *testing.T, s iSetup) (string, string) {
iCreateWebhook(t, s.MainDB, s.WebhookID, "hook")
tgt := &database.Target{
WebhookID: s.WebhookID,
Name: "archive",
Type: database.TargetTypeDatabase,
WebhookID: s.WebhookID,
Name: "archive",
Type: database.TargetTypeDatabase,
MaxAttempts: 1,
}
require.NoError(
t, s.MainDB.Omit(clause.Associations).Create(tgt).Error,
internal/delivery/engine_test.go
+125 -35
View File
@@ -142,7 +142,7 @@ func buildHTTPFixture(
dlv database.Delivery,
event database.Event,
targetID, name, config string,
maxRetries, attemptNum int,
maxAttempts, attemptNum int,
) httpDeliveryFixture {
task := &delivery.Task{
DeliveryID: dlv.ID,
@@ -152,7 +152,7 @@ func buildHTTPFixture(
TargetName: name,
TargetType: database.TargetTypeHTTP,
TargetConfig: config,
MaxRetries: maxRetries,
MaxAttempts: maxAttempts,
AttemptNum: attemptNum,
}
@@ -162,10 +162,10 @@ func buildHTTPFixture(
Status: dlv.Status,
Event: event,
Target: database.Target{
Name: name,
Type: database.TargetTypeHTTP,
Config: config,
MaxRetries: maxRetries,
Name: name,
Type: database.TargetTypeHTTP,
Config: config,
MaxAttempts: maxAttempts,
},
}
d.ID = dlv.ID
@@ -277,7 +277,7 @@ func TestDeliverHTTP_Success(t *testing.T) {
cfg := newHTTPTargetConfig(ts.URL)
fix := buildHTTPFixture(
dlv, event, targetID, "test-http", cfg, 0, 1,
dlv, event, targetID, "test-http", cfg, 1, 1,
)
e.ExportDeliverHTTP(
@@ -324,7 +324,7 @@ func TestDeliverHTTP_Failure(t *testing.T) {
cfg := newHTTPTargetConfig(ts.URL)
fix := buildHTTPFixture(
dlv, event, targetID,
"test-http-fail", cfg, 0, 1,
"test-http-fail", cfg, 1, 1,
)
e.ExportDeliverHTTP(
@@ -473,14 +473,23 @@ func TestDeliverHTTP_WithRetries_Success(t *testing.T) {
)
}
func TestDeliverHTTP_MaxRetriesExhausted(t *testing.T) {
// TestDeliverHTTP_MaxAttemptsExhausted proves a target with three
// delivery attempts makes at most three: the first two failures are
// retried, the third is final, and each attempt goes through the
// target's circuit breaker.
func TestDeliverHTTP_MaxAttemptsExhausted(t *testing.T) {
t.Parallel()
const maxAttempts = 3
db := testWebhookDB(t)
var requests atomic.Int32
ts := httptest.NewServer(
http.HandlerFunc(
func(w http.ResponseWriter, _ *http.Request) {
requests.Add(1)
w.WriteHeader(http.StatusBadGateway)
},
),
@@ -489,26 +498,103 @@ func TestDeliverHTTP_MaxRetriesExhausted(t *testing.T) {
e := testEngine(t, 1)
targetID := uuid.New().String()
// It opens on the last failure only if every attempt reached it.
cb := delivery.NewTestCircuitBreaker(maxAttempts, time.Minute)
e.ExportSetCircuitBreaker(targetID, cb)
event := seedEvent(t, db, `{"retry":"exhaust"}`)
dlv := seedDelivery(
t, db, event.ID, targetID,
database.DeliveryStatusRetrying,
database.DeliveryStatusPending,
)
cfg := newHTTPTargetConfig(ts.URL)
sched := &recordingScheduler{}
for attempt := 1; attempt <= maxAttempts; attempt++ {
fix := buildHTTPFixture(
dlv, event, targetID,
"test-http-exhaust", cfg, maxAttempts, attempt,
)
e.ExportDeliverHTTPWithScheduler(
context.TODO(), db, fix.Delivery, fix.Task, sched,
)
}
assert.Equal(t, int32(maxAttempts), requests.Load())
assert.Len(t, sched.delays, maxAttempts-1,
"every failure but the last should be retried",
)
assertDeliveryStatus(t, db, dlv.ID,
database.DeliveryStatusFailed,
)
assert.Equal(t, delivery.CircuitOpen, cb.State(),
"every attempt should be counted by the circuit breaker",
)
}
// TestDeliverHTTP_SingleAttemptSkipsCircuitBreaker proves a target
// with one delivery attempt makes exactly that attempt, even while
// its circuit breaker is open, and that a failure is final.
func TestDeliverHTTP_SingleAttemptSkipsCircuitBreaker(t *testing.T) {
t.Parallel()
db := testWebhookDB(t)
var requests atomic.Int32
ts := httptest.NewServer(
http.HandlerFunc(
func(w http.ResponseWriter, _ *http.Request) {
requests.Add(1)
w.WriteHeader(http.StatusBadGateway)
},
),
)
defer ts.Close()
e := testEngine(t, 1)
targetID := uuid.New().String()
cb := e.ExportGetCircuitBreaker(targetID)
for range delivery.ExportDefaultFailureThreshold {
cb.RecordFailure()
}
require.Equal(t, delivery.CircuitOpen, cb.State())
event := seedEvent(t, db, `{"retry":"none"}`)
dlv := seedDelivery(
t, db, event.ID, targetID,
database.DeliveryStatusPending,
)
cfg := newHTTPTargetConfig(ts.URL)
fix := buildHTTPFixture(
dlv, event, targetID,
"test-http-exhaust", cfg, 3, 3,
"test-http-single", cfg, 1, 1,
)
sched := &recordingScheduler{}
e.ExportDeliverHTTPWithScheduler(
context.TODO(), db, fix.Delivery, fix.Task, sched,
)
e.ExportDeliverHTTP(
context.TODO(), db, fix.Delivery, fix.Task,
assert.Equal(t, int32(1), requests.Load(),
"the open circuit breaker should not hold the attempt back",
)
assert.Empty(t, sched.delays, "no retry should be scheduled")
assertDeliveryStatus(t, db, dlv.ID,
database.DeliveryStatusFailed,
)
assertDeliveryResult(
t, db, dlv.ID, false, http.StatusBadGateway,
)
}
func TestDeliverHTTP_SchedulesRetryOnFailure(
@@ -727,9 +813,10 @@ func seedConcurrencyTasks(
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
Name: fmt.Sprintf("task-%d", i),
Type: database.TargetTypeHTTP,
Config: targetCfg,
Name: fmt.Sprintf("task-%d", i),
Type: database.TargetTypeHTTP,
Config: targetCfg,
MaxAttempts: 1,
},
}
tasks[i].ID = dlv.ID
@@ -741,7 +828,7 @@ func seedConcurrencyTasks(
TargetName: tasks[i].Target.Name,
TargetType: tasks[i].Target.Type,
TargetConfig: tasks[i].Target.Config,
MaxRetries: 0,
MaxAttempts: 1,
AttemptNum: 1,
}
}
@@ -1688,9 +1775,10 @@ func buildSlackDelivery(
Status: dlv.Status,
Event: event,
Target: database.Target{
Name: name,
Type: database.TargetTypeSlack,
Config: slackCfg,
Name: name,
Type: database.TargetTypeSlack,
Config: slackCfg,
MaxAttempts: 1,
},
}
d.ID = dlv.ID
@@ -1874,9 +1962,10 @@ func TestDeliverSlack_InvalidConfig(t *testing.T) {
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
Name: "test-slack-bad",
Type: database.TargetTypeSlack,
Config: `{"not_webhook_url":"missing"}`,
Name: "test-slack-bad",
Type: database.TargetTypeSlack,
Config: `{"not_webhook_url":"missing"}`,
MaxAttempts: 1,
},
}
d.ID = dlv.ID
@@ -1933,8 +2022,9 @@ func TestProcessDelivery_RoutesToSlack(t *testing.T) {
)
task := &delivery.Task{
DeliveryID: dlv.ID,
TargetType: database.TargetTypeSlack,
DeliveryID: dlv.ID,
TargetType: database.TargetTypeSlack,
MaxAttempts: 1,
}
e.ExportProcessDelivery(
@@ -2051,10 +2141,10 @@ func buildSlackRetryDelivery(
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
Name: "test-slack-retry",
Type: database.TargetTypeSlack,
Config: cfg,
MaxRetries: 5,
Name: "test-slack-retry",
Type: database.TargetTypeSlack,
Config: cfg,
MaxAttempts: 5,
},
}
d.ID = dlv.ID
@@ -2089,11 +2179,11 @@ func TestDeliverSlack_WithRetries_SchedulesRetry(
)
task := &delivery.Task{
DeliveryID: dlv.ID,
TargetID: targetID,
TargetType: database.TargetTypeSlack,
MaxRetries: 5,
AttemptNum: 1,
DeliveryID: dlv.ID,
TargetID: targetID,
TargetType: database.TargetTypeSlack,
MaxAttempts: 5,
AttemptNum: 1,
}
e.ExportProcessDelivery(context.TODO(), db, d, task)
internal/delivery/event_db_isolation_test.go
+1 -1
View File
@@ -152,7 +152,7 @@ func TestEventDBHoldsNoTargetRowsOnFailedDelivery(t *testing.T) {
)
task := iTask(
d, event, s.WebhookID, targetID,
"leaky-target", cfg, 0, 1, &body,
"leaky-target", cfg, 1, 1, &body,
)
s.Engine.ExportProcessNewTask(context.TODO(), &task)
internal/delivery/event_timestamp_test.go
+7 -6
View File
@@ -147,11 +147,12 @@ func tsSeedTarget(
t.Helper()
target := database.Target{
WebhookID: webhookID,
Name: "slack-sink",
Type: database.TargetTypeSlack,
Config: config,
Active: true,
WebhookID: webhookID,
Name: "slack-sink",
Type: database.TargetTypeSlack,
Config: config,
Active: true,
MaxAttempts: 1,
}
require.NoError(t, mainDB.Create(&target).Error)
@@ -176,7 +177,7 @@ func tsTask(
TargetName: target.Name,
TargetType: database.TargetTypeSlack,
TargetConfig: target.Config,
MaxRetries: 0,
MaxAttempts: 1,
Method: event.Method,
Headers: event.Headers,
ContentType: event.ContentType,
internal/delivery/export_test.go
+4 -3
View File
@@ -142,9 +142,10 @@ func (e *Engine) ExportDeliverSlack(
d *database.Delivery,
) {
task := &Task{
DeliveryID: d.ID,
TargetID: d.TargetID,
AttemptNum: 1,
DeliveryID: d.ID,
TargetID: d.TargetID,
MaxAttempts: d.Target.MaxAttempts,
AttemptNum: 1,
}
e.targets[database.TargetTypeSlack].Deliver(
internal/delivery/inflight_test.go
+2 -2
View File
@@ -32,7 +32,7 @@ func fSweepSetup(
iCreateTarget(t, s.MainDB, targetID,
s.WebhookID, name,
database.TargetTypeLog, "", 0,
database.TargetTypeLog, "", 1,
)
require.NoError(t, s.MainDB.Create(&database.Webhook{
@@ -229,7 +229,7 @@ func TestOwnershipIsReleasedAfterDelivery(t *testing.T) {
iCreateTarget(t, s.MainDB, targetID,
s.WebhookID, "released",
database.TargetTypeLog, "", 0,
database.TargetTypeLog, "", 1,
)
event := iSeedEvent(
internal/delivery/metrics_test.go
+10 -10
View File
@@ -291,20 +291,20 @@ func TestDeliveryMetrics_CircuitBreakerGauge(t *testing.T) {
body := event.Body
cfg := iHTTPConfig(ts.URL)
// A retry budget above the failure threshold, so the breaker
// An attempt budget above the failure threshold, so the breaker
// rather than the budget is what stops the delivery.
maxRetries := delivery.ExportDefaultFailureThreshold + 5
maxAttempts := delivery.ExportDefaultFailureThreshold + 5
// The retries below are only run if their target still exists;
// see https://git.eeqj.de/sneak/webhooker/issues/107.
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, "metrics-trip",
database.TargetTypeHTTP, cfg, maxRetries,
database.TargetTypeHTTP, cfg, maxAttempts,
)
first := iTask(
d, event, s.WebhookID, targetID,
"metrics-trip", cfg, maxRetries, 1, &body,
"metrics-trip", cfg, maxAttempts, 1, &body,
)
s.Engine.ExportProcessNewTask(context.TODO(), &first)
@@ -316,7 +316,7 @@ func TestDeliveryMetrics_CircuitBreakerGauge(t *testing.T) {
ExportDefaultFailureThreshold; attempt++ {
task := iTask(
d, event, s.WebhookID, targetID,
"metrics-trip", cfg, maxRetries, attempt, &body,
"metrics-trip", cfg, maxAttempts, attempt, &body,
)
s.Engine.ExportProcessRetryTask(
@@ -364,16 +364,16 @@ func TestDeliveryMetrics_BreakerBlockedIsNotAnAttempt(
body := event.Body
cfg := iHTTPConfig(ts.URL)
maxRetries := delivery.ExportDefaultFailureThreshold + 5
maxAttempts := delivery.ExportDefaultFailureThreshold + 5
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, "metrics-blocked",
database.TargetTypeHTTP, cfg, maxRetries,
database.TargetTypeHTTP, cfg, maxAttempts,
)
first := iTask(
d, event, s.WebhookID, targetID,
"metrics-blocked", cfg, maxRetries, 1, &body,
"metrics-blocked", cfg, maxAttempts, 1, &body,
)
s.Engine.ExportProcessNewTask(context.TODO(), &first)
@@ -382,7 +382,7 @@ func TestDeliveryMetrics_BreakerBlockedIsNotAnAttempt(
ExportDefaultFailureThreshold; attempt++ {
task := iTask(
d, event, s.WebhookID, targetID,
"metrics-blocked", cfg, maxRetries, attempt, &body,
"metrics-blocked", cfg, maxAttempts, attempt, &body,
)
s.Engine.ExportProcessRetryTask(context.TODO(), &task)
@@ -405,7 +405,7 @@ func TestDeliveryMetrics_BreakerBlockedIsNotAnAttempt(
blocked := iTask(
d, event, s.WebhookID, targetID,
"metrics-blocked", cfg, maxRetries,
"metrics-blocked", cfg, maxAttempts,
delivery.ExportDefaultFailureThreshold+1, &body,
)
internal/delivery/recovery_durability_test.go
+13 -13
View File
@@ -66,7 +66,7 @@ func TestRecoverySkipsPendingWithSuccessfulResult(
iCreateTarget(t, s.MainDB, targetID,
s.WebhookID, "already-delivered",
database.TargetTypeLog, "", 0,
database.TargetTypeLog, "", 1,
)
event := iSeedEvent(
@@ -129,7 +129,7 @@ func TestRecoveryContinuesTheAttemptNumbering(t *testing.T) {
iCreateTarget(t, s.MainDB, targetID,
s.WebhookID, "numbering",
database.TargetTypeLog, "", 0,
database.TargetTypeLog, "", 1,
)
event := iSeedEvent(
@@ -354,16 +354,17 @@ func TestFailedResultWriteLeavesDeliveryRecoverable(
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
Name: "unwritable",
Type: database.TargetTypeHTTP,
Config: iHTTPConfig(ts.URL),
Name: "unwritable",
Type: database.TargetTypeHTTP,
Config: iHTTPConfig(ts.URL),
MaxAttempts: 1,
},
}
full.ID = d.ID
s.Engine.ExportDeliverHTTP(
context.Background(), s.WebhookDB, full,
&delivery.Task{DeliveryID: d.ID, AttemptNum: 1},
&delivery.Task{DeliveryID: d.ID, MaxAttempts: 1, AttemptNum: 1},
)
assert.Equal(
@@ -444,10 +445,10 @@ func TestFailedResultWriteWithRetriesLeavesDeliveryRecoverable(
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
Name: "unwritable",
Type: database.TargetTypeHTTP,
Config: iHTTPConfig(ts.URL),
MaxRetries: 3,
Name: "unwritable",
Type: database.TargetTypeHTTP,
Config: iHTTPConfig(ts.URL),
MaxAttempts: 3,
},
}
full.ID = d.ID
@@ -457,9 +458,8 @@ func TestFailedResultWriteWithRetriesLeavesDeliveryRecoverable(
s.Engine.ExportDeliverHTTPWithScheduler(
context.Background(), s.WebhookDB, full,
&delivery.Task{
DeliveryID: d.ID,
TargetID: targetID,
AttemptNum: 1,
DeliveryID: d.ID, TargetID: targetID,
MaxAttempts: 3, AttemptNum: 1,
},
sched,
)
internal/delivery/redirect_test.go
+1 -1
View File
@@ -96,7 +96,7 @@ func deliverWithProbeHeaders(
body := event.Body
task := iTask(
d, event, s.WebhookID, targetID,
"redirect-target", string(cfg), 0, 1, &body,
"redirect-target", string(cfg), 1, 1, &body,
)
s.Engine.ExportProcessNewTask(context.TODO(), &task)
internal/delivery/target_config_view.go
+9 -14
View File
@@ -120,8 +120,8 @@ func unavailableConfigFields() []ConfigField {
}
// slackConfigFields describes a Slack target: its masked
// webhook URL and its retry count. Only the masked URL is
// shown; the full URL is the credential.
// webhook URL and its number of delivery attempts. Only the
// masked URL is shown; the full URL is the credential.
func slackConfigFields(t *database.Target) []ConfigField {
cfg, err := parseSlackConfig(t.Config)
if err != nil {
@@ -131,7 +131,7 @@ func slackConfigFields(t *database.Target) []ConfigField {
return []ConfigField{{
Label: "Webhook URL",
Value: cfg.MaskedWebhookURL(),
}, maxRetriesField(t)}
}, maxAttemptsField(t)}
}
// httpConfigFields describes an HTTP target: its destination
@@ -178,23 +178,18 @@ func httpConfigFields(t *database.Target) []ConfigField {
})
}
fields = append(fields, maxRetriesField(t))
fields = append(fields, maxAttemptsField(t))
return fields
}
// maxRetriesField describes a target's retry count, which lives
// on the target row rather than in its configuration blob. A
// stored 0 makes a single attempt, so it is shown as 1.
func maxRetriesField(t *database.Target) ConfigField {
attempts := strconv.Itoa(t.MaxRetries)
if t.MaxRetries == 0 {
attempts = "1 (fire-and-forget: no retries, no circuit breaker)"
}
// maxAttemptsField describes a target's number of delivery
// attempts, which lives on the target row rather than in its
// configuration blob.
func maxAttemptsField(t *database.Target) ConfigField {
return ConfigField{
Label: "Delivery attempts",
Value: attempts,
Value: strconv.Itoa(t.MaxAttempts),
}
}
internal/delivery/target_config_view_test.go
+34 -24
View File
@@ -32,7 +32,7 @@ const (
viewMaskedOrigin = viewExampleOrigin + "/..."
viewUnavailable = "(unavailable)"
viewExpiryNever = "never"
viewMaxRetries = "Delivery attempts"
viewMaxAttempts = "Delivery attempts"
)
func TestMaskedWebhookURL(t *testing.T) {
@@ -171,6 +171,7 @@ func slackTarget() database.Target {
Active: true,
Config: `{"webhookUrl":"` +
slackWebhookURL + `"}`,
MaxAttempts: 1,
}
}
@@ -189,28 +190,29 @@ func TestNewTargetViews_Slack(t *testing.T) {
assert.Equal(
t,
map[string]string{
"Webhook URL": slackMaskedURL,
viewMaxRetries: "1 (fire-and-forget: no retries, no circuit breaker)",
"Webhook URL": slackMaskedURL,
viewMaxAttempts: "1",
},
fieldMap(view.Config),
)
}
// TestNewTargetViews_SlackRetries proves a Slack target shows
// its retry count the same way an HTTP target does.
func TestNewTargetViews_SlackRetries(t *testing.T) {
// TestNewTargetViews_SlackAttempts proves a Slack target shows
// its number of delivery attempts the same way an HTTP target
// does.
func TestNewTargetViews_SlackAttempts(t *testing.T) {
t.Parallel()
target := slackTarget()
target.MaxRetries = 2
target.MaxAttempts = 2
view := viewFor(t, target)
assert.Equal(
t,
map[string]string{
"Webhook URL": slackMaskedURL,
viewMaxRetries: "2",
"Webhook URL": slackMaskedURL,
viewMaxAttempts: "2",
},
fieldMap(view.Config),
)
@@ -225,7 +227,7 @@ func TestNewTargetViews_HTTP(t *testing.T) {
`"timeout":30,` +
`"headers":{"Authorization":"Bearer sekrit"},` +
`"forwardQuery":true}`,
MaxRetries: 5,
MaxAttempts: 5,
})
fields := fieldMap(view.Config)
@@ -237,7 +239,7 @@ func TestNewTargetViews_HTTP(t *testing.T) {
"Timeout": "30s",
"Headers": "1 configured",
"Query string": "passed on to this target",
viewMaxRetries: "5",
viewMaxAttempts: "5",
},
fields,
)
@@ -248,19 +250,22 @@ func TestNewTargetViews_HTTP(t *testing.T) {
}
}
func TestNewTargetViews_HTTPFireAndForget(t *testing.T) {
// TestNewTargetViews_HTTPSingleAttempt proves a single attempt is
// shown as the plain number, like any other.
func TestNewTargetViews_HTTPSingleAttempt(t *testing.T) {
t.Parallel()
view := viewFor(t, database.Target{
Type: database.TargetTypeHTTP,
Config: `{"url":"` + viewExampleHook + `"}`,
Type: database.TargetTypeHTTP,
Config: `{"url":"` + viewExampleHook + `"}`,
MaxAttempts: 1,
})
assert.Equal(
t,
map[string]string{
"Destination URL": viewMaskedOrigin,
viewMaxRetries: "1 (fire-and-forget: no retries, no circuit breaker)",
viewMaxAttempts: "1",
},
fieldMap(view.Config),
)
@@ -274,8 +279,9 @@ func TestNewTargetViews_HTTPMasksDestinationURL(t *testing.T) {
t.Parallel()
view := viewFor(t, database.Target{
Type: database.TargetTypeHTTP,
Config: `{"url":"` + slackWebhookURL + `"}`,
Type: database.TargetTypeHTTP,
Config: `{"url":"` + slackWebhookURL + `"}`,
MaxAttempts: 1,
})
fields := fieldMap(view.Config)
@@ -398,19 +404,23 @@ func TestNewTargetViews_Unpresentable(t *testing.T) {
Config: blob,
},
"unparseable json": {
Type: database.TargetTypeSlack,
Config: blob,
Type: database.TargetTypeSlack,
Config: blob,
MaxAttempts: 1,
},
"empty slack config": {
Type: database.TargetTypeSlack,
Type: database.TargetTypeSlack,
MaxAttempts: 1,
},
"slack config without url": {
Type: database.TargetTypeSlack,
Config: `{}`,
Type: database.TargetTypeSlack,
Config: `{}`,
MaxAttempts: 1,
},
"unparseable http json": {
Type: database.TargetTypeHTTP,
Config: `{"url":`,
Type: database.TargetTypeHTTP,
Config: `{"url":`,
MaxAttempts: 1,
},
"unparseable archive json": {
Type: database.TargetTypeDatabase,
internal/delivery/target_headers_test.go
+7 -5
View File
@@ -189,6 +189,7 @@ func TestNewTargetConfigForm(t *testing.T) {
Config: `{"url":"https://example.com/h",` +
`"headers":{"Authorization":"Bearer abc"},` +
`"timeout":9}`,
MaxAttempts: 1,
})
require.NoError(t, err)
assert.Equal(t, "https://example.com/h", form.URL)
@@ -196,8 +197,9 @@ func TestNewTargetConfigForm(t *testing.T) {
assert.Equal(t, "9", form.Timeout)
form, err = delivery.NewTargetConfigForm(&database.Target{
Type: database.TargetTypeSlack,
Config: `{"webhookUrl":"https://hooks.example/s"}`,
Type: database.TargetTypeSlack,
Config: `{"webhookUrl":"https://hooks.example/s"}`,
MaxAttempts: 1,
})
require.NoError(t, err)
assert.Equal(t, "https://hooks.example/s", form.URL)
@@ -242,9 +244,9 @@ func TestNewTargetConfigForm_UnreadableConfigErrors(t *testing.T) {
t.Parallel()
cases := []*database.Target{
{Type: database.TargetTypeHTTP, Config: "not json"},
{Type: database.TargetTypeHTTP, Config: `{}`},
{Type: database.TargetTypeSlack, Config: ""},
{Type: database.TargetTypeHTTP, Config: "not json", MaxAttempts: 1},
{Type: database.TargetTypeHTTP, Config: `{}`, MaxAttempts: 1},
{Type: database.TargetTypeSlack, Config: "", MaxAttempts: 1},
{
Type: database.TargetTypeDatabase,
Config: `{"expiry":"soon"}`,
internal/delivery/target_http.go
+12 -12
View File
@@ -53,10 +53,10 @@ type httpCore struct {
}
// deliver runs one delivery attempt through the retry core.
// A maxRetries of 0 is fire-and-forget: a single attempt is
// recorded and no circuit breaker is consulted. A positive
// maxRetries gates the attempt on the circuit breaker and
// schedules a backed-off retry on failure.
// A maxAttempts of 1 is a single attempt, recorded without
// consulting a circuit breaker. A higher maxAttempts gates the
// attempt on the circuit breaker and schedules a backed-off retry
// on failure, until maxAttempts attempts have been made.
//
// ctx is the context attempt sends its request with. Stopping the
// engine cancels it; see interruptedByStop.
@@ -66,17 +66,17 @@ func (c *httpCore) deliver(
d *database.Delivery,
task *Task,
sched Scheduler,
maxRetries int,
maxAttempts int,
attempt func() attemptResult,
) {
if maxRetries == 0 {
if maxAttempts == 1 {
c.fireAndForget(ctx, webhookDB, d, attempt())
return
}
c.withRetry(
ctx, webhookDB, d, task, sched, maxRetries, attempt,
ctx, webhookDB, d, task, sched, maxAttempts, attempt,
)
}
@@ -153,7 +153,7 @@ func (c *httpCore) withRetry(
d *database.Delivery,
task *Task,
sched Scheduler,
maxRetries int,
maxAttempts int,
attempt func() attemptResult,
) {
cb := c.getCircuitBreaker(task.TargetID)
@@ -205,7 +205,7 @@ func (c *httpCore) withRetry(
cb.RecordFailure()
c.handleRetry(
webhookDB, d, task, sched, maxRetries, attemptNum,
webhookDB, d, task, sched, maxAttempts, attemptNum,
)
}
@@ -266,10 +266,10 @@ func (c *httpCore) handleRetry(
d *database.Delivery,
task *Task,
sched Scheduler,
maxRetries int,
maxAttempts int,
attemptNum int,
) {
if attemptNum >= maxRetries {
if attemptNum >= maxAttempts {
c.eng.settleStatus(
webhookDB, d, d.Target.Type,
database.DeliveryStatusFailed,
@@ -439,7 +439,7 @@ func (t *httpTarget) Deliver(
t.deliver(
ctx, webhookDB, d, task, sched,
d.Target.MaxRetries, attempt,
d.Target.MaxAttempts, attempt,
)
}
internal/delivery/target_http_query_test.go
+4 -3
View File
@@ -47,7 +47,7 @@ func deliverWithQuery(t *testing.T, cfg string) database.DeliveryResult {
database.DeliveryStatusPending,
)
task := iTask(
d, event, s.WebhookID, d.TargetID, "query", cfg, 0, 1, &event.Body,
d, event, s.WebhookID, d.TargetID, "query", cfg, 1, 1, &event.Body,
)
task.RawQuery = eventQuery
@@ -144,8 +144,9 @@ func TestDeliverHTTP_ForwardedQueryKeepsTheTargetURLMasked(t *testing.T) {
t.Cleanup(ts.Close)
target := &database.Target{
Type: database.TargetTypeHTTP,
Config: httpTargetConfig(t, ts.URL+"/in?token="+secret, true),
Type: database.TargetTypeHTTP,
Config: httpTargetConfig(t, ts.URL+"/in?token="+secret, true),
MaxAttempts: 1,
}
deliverWithQuery(t, target.Config)
internal/delivery/target_redact_test.go
+25 -16
View File
@@ -22,8 +22,9 @@ func TestRedactor_RemovesSlackWebhookURL(t *testing.T) {
t.Parallel()
r := delivery.NewRedactor(&database.Target{
Type: database.TargetTypeSlack,
Config: `{"webhookUrl":"` + redactWebhookURL + `"}`,
Type: database.TargetTypeSlack,
Config: `{"webhookUrl":"` + redactWebhookURL + `"}`,
MaxAttempts: 1,
})
got := r.Redact("no_service for " + redactWebhookURL)
@@ -52,8 +53,9 @@ func TestRedactor_RemovesSecretSeveredByACut(t *testing.T) {
t.Parallel()
r := delivery.NewRedactor(&database.Target{
Type: database.TargetTypeSlack,
Config: `{"webhookUrl":"` + redactWebhookURL + `"}`,
Type: database.TargetTypeSlack,
Config: `{"webhookUrl":"` + redactWebhookURL + `"}`,
MaxAttempts: 1,
})
// Every cut position inside the credential, not just a
@@ -96,6 +98,7 @@ func TestRedactor_RedactsCredentialShapedHeaderValues(
`"X-Credential":"HHHHHHHHHHHH",` +
`"Accept":"application/json",` +
`"User-Agent":"webhooker/1.0"}}`,
MaxAttempts: 1,
})
for _, secret := range []string{
@@ -136,6 +139,7 @@ func TestRedactor_IgnoresVeryShortHeaderValues(t *testing.T) {
Type: database.TargetTypeHTTP,
Config: `{"url":"https://example.com/in",` +
`"headers":{"X-Api-Key":"ab"}}`,
MaxAttempts: 1,
})
const response = "rabbit"
@@ -150,8 +154,9 @@ func TestRedactor_RemovesBarePath(t *testing.T) {
t.Parallel()
r := delivery.NewRedactor(&database.Target{
Type: database.TargetTypeSlack,
Config: `{"webhookUrl":"` + redactWebhookURL + `"}`,
Type: database.TargetTypeSlack,
Config: `{"webhookUrl":"` + redactWebhookURL + `"}`,
MaxAttempts: 1,
})
got := r.Redact("POST " + redactSecretPath + " 404")
@@ -183,8 +188,9 @@ func TestRedactor_RemovesHTTPURLQueryAndUserinfo(t *testing.T) {
raw := dest.String()
r := delivery.NewRedactor(&database.Target{
Type: database.TargetTypeHTTP,
Config: `{"url":"` + raw + `"}`,
Type: database.TargetTypeHTTP,
Config: `{"url":"` + raw + `"}`,
MaxAttempts: 1,
})
for _, echoed := range []string{
@@ -232,8 +238,9 @@ func TestRedactor_RemovesEchoedQueryOfURLWithoutPath(t *testing.T) {
"https://example.com?token=" + secret,
} {
r := delivery.NewRedactor(&database.Target{
Type: database.TargetTypeHTTP,
Config: `{"url":"` + dest + `"}`,
Type: database.TargetTypeHTTP,
Config: `{"url":"` + dest + `"}`,
MaxAttempts: 1,
})
for echoed, want := range echoes {
@@ -253,8 +260,9 @@ func TestRedactor_LeavesUnrelatedTextAlone(t *testing.T) {
const response = "ok=false error=channel_not_found"
r := delivery.NewRedactor(&database.Target{
Type: database.TargetTypeSlack,
Config: `{"webhookUrl":"` + redactWebhookURL + `"}`,
Type: database.TargetTypeSlack,
Config: `{"webhookUrl":"` + redactWebhookURL + `"}`,
MaxAttempts: 1,
})
assert.Equal(t, response, r.Redact(response))
@@ -279,8 +287,9 @@ func TestRedactor_EmptyUserinfoDoesNotShredTheBody(t *testing.T) {
require.Empty(t, parsed.User.String())
r := delivery.NewRedactor(&database.Target{
Type: database.TargetTypeHTTP,
Config: `{"url":"` + dest + `"}`,
Type: database.TargetTypeHTTP,
Config: `{"url":"` + dest + `"}`,
MaxAttempts: 1,
})
const body = "ok=false error=channel_not_found"
@@ -314,8 +323,8 @@ func TestRedactor_ZeroValueAndConfiglessTargets(t *testing.T) {
for _, tgt := range []database.Target{
{Type: database.TargetTypeLog},
{Type: database.TargetTypeDatabase},
{Type: database.TargetTypeSlack, Config: "not json"},
{Type: database.TargetTypeHTTP, Config: ""},
{Type: database.TargetTypeSlack, Config: "not json", MaxAttempts: 1},
{Type: database.TargetTypeHTTP, Config: "", MaxAttempts: 1},
} {
assert.Equal(
t, text,
internal/delivery/target_slack.go
+3 -4
View File
@@ -30,9 +30,8 @@ type SlackTargetConfig struct {
// slackTarget delivers events to Slack incoming webhooks. It
// formats the event into a Slack message and posts it as
// JSON. It shares the retry core with the HTTP target: a
// MaxRetries of 0 stays single-attempt fire-and-forget
// (preserving existing Slack targets), while a positive
// MaxRetries adds backoff and circuit breaking.
// MaxAttempts of 1 is a single attempt, while a higher one
// adds backoff and circuit breaking.
type slackTarget struct {
*httpCore
@@ -83,7 +82,7 @@ func (t *slackTarget) Deliver(
t.deliver(
ctx, webhookDB, d, task, sched,
d.Target.MaxRetries, attempt,
d.Target.MaxAttempts, attempt,
)
}
internal/delivery/terminal_state_test.go
+2 -2
View File
@@ -743,7 +743,7 @@ func TestSendRecoveredDeliveries_TargetMissingFromMap(
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, "found-on-lookup",
database.TargetTypeLog, "", 0,
database.TargetTypeLog, "", 1,
)
event := iSeedEvent(
@@ -790,7 +790,7 @@ func TestRecoverPending_TargetUnreadable_LeavesDeliveryAlone(
iCreateTarget(
t, s.MainDB, targetID, s.WebhookID, "healthy",
database.TargetTypeLog, "", 0,
database.TargetTypeLog, "", 1,
)
event := iSeedEvent(