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

refactor: merge retry target type into http (max_retries=0 = fire-and-forget)
All checks were successful
check / check (push) Successful in 1m46s
Details

Browse Source
This commit is contained in:
clawbot
2026-03-01 23:51:55 -08:00
parent 4dd4dfa5eb
commit 25e27cc57f
7 changed files with 150 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
internal/delivery/engine.go
View File

@@ -133,7 +133,8 @@ type Engine struct {
workers int
// circuitBreakers stores a *CircuitBreaker per target ID. Only used
// for retry targets — HTTP, database, and log targets do not need
// for HTTP targets with MaxRetries > 0 — fire-and-forget HTTP targets
// (MaxRetries == 0), database targets, and log targets do not need
// circuit breakers because they either fire once or are local ops.
circuitBreakers sync.Map
}
@@ -829,9 +830,7 @@ func (e *Engine) sweepWebhookRetries(ctx context.Context, webhookID string) {
func (e *Engine) processDelivery(ctx context.Context, webhookDB *gorm.DB, d *database.Delivery, task *DeliveryTask) {
switch d.Target.Type {
case database.TargetTypeHTTP:
e.deliverHTTP(ctx, webhookDB, d)
case database.TargetTypeRetry:
e.deliverRetry(ctx, webhookDB, d, task)
e.deliverHTTP(ctx, webhookDB, d, task)
case database.TargetTypeDatabase:
e.deliverDatabase(webhookDB, d)
case database.TargetTypeLog:
@@ -845,47 +844,43 @@ func (e *Engine) processDelivery(ctx context.Context, webhookDB *gorm.DB, d *dat
}
}
func (e *Engine) deliverHTTP(_ context.Context, webhookDB *gorm.DB, d *database.Delivery) {
func (e *Engine) deliverHTTP(_ context.Context, webhookDB *gorm.DB, d *database.Delivery, task *DeliveryTask) {
cfg, err := e.parseHTTPConfig(d.Target.Config)
if err != nil {
e.log.Error("invalid HTTP target config",
"target_id", d.TargetID,
"error", err,
)
e.recordResult(webhookDB, d, 1, false, 0, "", err.Error(), 0)
e.updateDeliveryStatus(webhookDB, d, database.DeliveryStatusFailed)
return
}
statusCode, respBody, duration, err := e.doHTTPRequest(cfg, &d.Event)
success := err == nil && statusCode >= 200 && statusCode < 300
errMsg := ""
if err != nil {
errMsg = err.Error()
}
e.recordResult(webhookDB, d, 1, success, statusCode, respBody, errMsg, duration)
if success {
e.updateDeliveryStatus(webhookDB, d, database.DeliveryStatusDelivered)
} else {
e.updateDeliveryStatus(webhookDB, d, database.DeliveryStatusFailed)
}
}
func (e *Engine) deliverRetry(_ context.Context, webhookDB *gorm.DB, d *database.Delivery, task *DeliveryTask) {
cfg, err := e.parseHTTPConfig(d.Target.Config)
if err != nil {
e.log.Error("invalid retry target config",
"target_id", d.TargetID,
"error", err,
)
e.recordResult(webhookDB, d, task.AttemptNum, false, 0, "", err.Error(), 0)
e.updateDeliveryStatus(webhookDB, d, database.DeliveryStatusFailed)
return
}
maxRetries := d.Target.MaxRetries
// Fire-and-forget mode: max_retries == 0 means attempt once with no
// circuit breaker and no retry scheduling.
if maxRetries == 0 {
statusCode, respBody, duration, reqErr := e.doHTTPRequest(cfg, &d.Event)
success := reqErr == nil && statusCode >= 200 && statusCode < 300
errMsg := ""
if reqErr != nil {
errMsg = reqErr.Error()
}
e.recordResult(webhookDB, d, 1, success, statusCode, respBody, errMsg, duration)
if success {
e.updateDeliveryStatus(webhookDB, d, database.DeliveryStatusDelivered)
} else {
e.updateDeliveryStatus(webhookDB, d, database.DeliveryStatusFailed)
}
return
}
// Retry mode: max_retries > 0 — use circuit breaker and exponential backoff.
// Check the circuit breaker for this target before attempting delivery.
cb := e.getCircuitBreaker(task.TargetID)
if !cb.Allow() {
@@ -910,12 +905,12 @@ func (e *Engine) deliverRetry(_ context.Context, webhookDB *gorm.DB, d *database
// Attempt delivery immediately — backoff is handled by the timer
// that triggered this call, not by polling.
statusCode, respBody, duration, err := e.doHTTPRequest(cfg, &d.Event)
statusCode, respBody, duration, reqErr := e.doHTTPRequest(cfg, &d.Event)
success := err == nil && statusCode >= 200 && statusCode < 300
success := reqErr == nil && statusCode >= 200 && statusCode < 300
errMsg := ""
if err != nil {
errMsg = err.Error()
if reqErr != nil {
errMsg = reqErr.Error()
}
e.recordResult(webhookDB, d, attemptNum, success, statusCode, respBody, errMsg, duration)
@@ -929,11 +924,6 @@ func (e *Engine) deliverRetry(_ context.Context, webhookDB *gorm.DB, d *database
// Delivery failed — record failure in circuit breaker
cb.RecordFailure()
maxRetries := d.Target.MaxRetries
if maxRetries <= 0 {
maxRetries = 5 // default
}
if attemptNum >= maxRetries {
e.updateDeliveryStatus(webhookDB, d, database.DeliveryStatusFailed)
} else {

99
internal/delivery/engine_test.go
View File

@@ -141,13 +141,26 @@ func TestDeliverHTTP_Success(t *testing.T) {
defer ts.Close()
e := testEngine(t, 1)
targetID := uuid.New().String()
event := seedEvent(t, db, `{"hello":"world"}`)
delivery := seedDelivery(t, db, event.ID, uuid.New().String(), database.DeliveryStatusPending)
delivery := seedDelivery(t, db, event.ID, targetID, database.DeliveryStatusPending)
task := &DeliveryTask{
DeliveryID: delivery.ID,
EventID: event.ID,
WebhookID: event.WebhookID,
TargetID: targetID,
TargetName: "test-http",
TargetType: database.TargetTypeHTTP,
TargetConfig: newHTTPTargetConfig(ts.URL),
MaxRetries: 0,
AttemptNum: 1,
}
d := &database.Delivery{
EventID: event.ID,
TargetID: delivery.TargetID,
TargetID: targetID,
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
@@ -158,7 +171,7 @@ func TestDeliverHTTP_Success(t *testing.T) {
}
d.ID = delivery.ID
e.deliverHTTP(context.TODO(), db, d)
e.deliverHTTP(context.TODO(), db, d, task)
assert.True(t, received.Load(), "HTTP target should have received request")
@@ -185,13 +198,26 @@ func TestDeliverHTTP_Failure(t *testing.T) {
defer ts.Close()
e := testEngine(t, 1)
targetID := uuid.New().String()
event := seedEvent(t, db, `{"test":true}`)
delivery := seedDelivery(t, db, event.ID, uuid.New().String(), database.DeliveryStatusPending)
delivery := seedDelivery(t, db, event.ID, targetID, database.DeliveryStatusPending)
task := &DeliveryTask{
DeliveryID: delivery.ID,
EventID: event.ID,
WebhookID: event.WebhookID,
TargetID: targetID,
TargetName: "test-http-fail",
TargetType: database.TargetTypeHTTP,
TargetConfig: newHTTPTargetConfig(ts.URL),
MaxRetries: 0,
AttemptNum: 1,
}
d := &database.Delivery{
EventID: event.ID,
TargetID: delivery.TargetID,
TargetID: targetID,
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
@@ -202,7 +228,7 @@ func TestDeliverHTTP_Failure(t *testing.T) {
}
d.ID = delivery.ID
e.deliverHTTP(context.TODO(), db, d)
e.deliverHTTP(context.TODO(), db, d, task)
// HTTP (fire-and-forget) marks as failed on non-2xx
var updated database.Delivery
@@ -280,7 +306,7 @@ func TestDeliverLog_ImmediateSuccess(t *testing.T) {
assert.True(t, result.Success)
}
func TestDeliverRetry_Success(t *testing.T) {
func TestDeliverHTTP_WithRetries_Success(t *testing.T) {
t.Parallel()
db := testWebhookDB(t)
@@ -300,8 +326,8 @@ func TestDeliverRetry_Success(t *testing.T) {
EventID: event.ID,
WebhookID: event.WebhookID,
TargetID: targetID,
TargetName: "test-retry",
TargetType: database.TargetTypeRetry,
TargetName: "test-http-retry",
TargetType: database.TargetTypeHTTP,
TargetConfig: newHTTPTargetConfig(ts.URL),
MaxRetries: 5,
AttemptNum: 1,
@@ -313,8 +339,8 @@ func TestDeliverRetry_Success(t *testing.T) {
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
Name: "test-retry",
Type: database.TargetTypeRetry,
Name: "test-http-retry",
Type: database.TargetTypeHTTP,
Config: newHTTPTargetConfig(ts.URL),
MaxRetries: 5,
},
@@ -322,7 +348,7 @@ func TestDeliverRetry_Success(t *testing.T) {
d.ID = delivery.ID
d.Target.ID = targetID
e.deliverRetry(context.TODO(), db, d, task)
e.deliverHTTP(context.TODO(), db, d, task)
var updated database.Delivery
require.NoError(t, db.First(&updated, "id = ?", delivery.ID).Error)
@@ -333,7 +359,7 @@ func TestDeliverRetry_Success(t *testing.T) {
assert.Equal(t, CircuitClosed, cb.State())
}
func TestDeliverRetry_MaxRetriesExhausted(t *testing.T) {
func TestDeliverHTTP_MaxRetriesExhausted(t *testing.T) {
t.Parallel()
db := testWebhookDB(t)
@@ -354,8 +380,8 @@ func TestDeliverRetry_MaxRetriesExhausted(t *testing.T) {
EventID: event.ID,
WebhookID: event.WebhookID,
TargetID: targetID,
TargetName: "test-retry-exhaust",
TargetType: database.TargetTypeRetry,
TargetName: "test-http-exhaust",
TargetType: database.TargetTypeHTTP,
TargetConfig: newHTTPTargetConfig(ts.URL),
MaxRetries: maxRetries,
AttemptNum: maxRetries, // final attempt
@@ -367,8 +393,8 @@ func TestDeliverRetry_MaxRetriesExhausted(t *testing.T) {
Status: database.DeliveryStatusRetrying,
Event: event,
Target: database.Target{
Name: "test-retry-exhaust",
Type: database.TargetTypeRetry,
Name: "test-http-exhaust",
Type: database.TargetTypeHTTP,
Config: newHTTPTargetConfig(ts.URL),
MaxRetries: maxRetries,
},
@@ -376,7 +402,7 @@ func TestDeliverRetry_MaxRetriesExhausted(t *testing.T) {
d.ID = delivery.ID
d.Target.ID = targetID
e.deliverRetry(context.TODO(), db, d, task)
e.deliverHTTP(context.TODO(), db, d, task)
// After max retries exhausted, delivery should be failed
var updated database.Delivery
@@ -385,7 +411,7 @@ func TestDeliverRetry_MaxRetriesExhausted(t *testing.T) {
"delivery should be failed after max retries exhausted")
}
func TestDeliverRetry_SchedulesRetryOnFailure(t *testing.T) {
func TestDeliverHTTP_SchedulesRetryOnFailure(t *testing.T) {
t.Parallel()
db := testWebhookDB(t)
@@ -405,8 +431,8 @@ func TestDeliverRetry_SchedulesRetryOnFailure(t *testing.T) {
EventID: event.ID,
WebhookID: event.WebhookID,
TargetID: targetID,
TargetName: "test-retry-schedule",
TargetType: database.TargetTypeRetry,
TargetName: "test-http-schedule",
TargetType: database.TargetTypeHTTP,
TargetConfig: newHTTPTargetConfig(ts.URL),
MaxRetries: 5,
AttemptNum: 1,
@@ -418,8 +444,8 @@ func TestDeliverRetry_SchedulesRetryOnFailure(t *testing.T) {
Status: database.DeliveryStatusPending,
Event: event,
Target: database.Target{
Name: "test-retry-schedule",
Type: database.TargetTypeRetry,
Name: "test-http-schedule",
Type: database.TargetTypeHTTP,
Config: newHTTPTargetConfig(ts.URL),
MaxRetries: 5,
},
@@ -427,7 +453,7 @@ func TestDeliverRetry_SchedulesRetryOnFailure(t *testing.T) {
d.ID = delivery.ID
d.Target.ID = targetID
e.deliverRetry(context.TODO(), db, d, task)
e.deliverHTTP(context.TODO(), db, d, task)
// Delivery should be in retrying status (not failed — retries remain)
var updated database.Delivery
@@ -591,6 +617,21 @@ func TestWorkerPool_BoundedConcurrency(t *testing.T) {
tasks[i].ID = delivery.ID
}
// Build DeliveryTask structs for each delivery (needed by deliverHTTP)
deliveryTasks := make([]DeliveryTask, numTasks)
for i := 0; i < numTasks; i++ {
deliveryTasks[i] = DeliveryTask{
DeliveryID: tasks[i].ID,
EventID: tasks[i].EventID,
TargetID: tasks[i].TargetID,
TargetName: tasks[i].Target.Name,
TargetType: tasks[i].Target.Type,
TargetConfig: tasks[i].Target.Config,
MaxRetries: 0,
AttemptNum: 1,
}
}
// Process all tasks through a bounded pool of goroutines to simulate
// the engine's worker pool behavior
var wg sync.WaitGroup
@@ -606,7 +647,7 @@ func TestWorkerPool_BoundedConcurrency(t *testing.T) {
go func() {
defer wg.Done()
for idx := range taskCh {
e.deliverHTTP(context.TODO(), db, &tasks[idx])
e.deliverHTTP(context.TODO(), db, &tasks[idx], &deliveryTasks[idx])
}
}()
}
@@ -629,7 +670,7 @@ func TestWorkerPool_BoundedConcurrency(t *testing.T) {
}
}
func TestDeliverRetry_CircuitBreakerBlocks(t *testing.T) {
func TestDeliverHTTP_CircuitBreakerBlocks(t *testing.T) {
t.Parallel()
db := testWebhookDB(t)
e := testEngine(t, 1)
@@ -651,7 +692,7 @@ func TestDeliverRetry_CircuitBreakerBlocks(t *testing.T) {
WebhookID: event.WebhookID,
TargetID: targetID,
TargetName: "test-cb-block",
TargetType: database.TargetTypeRetry,
TargetType: database.TargetTypeHTTP,
TargetConfig: newHTTPTargetConfig("http://will-not-be-called.invalid"),
MaxRetries: 5,
AttemptNum: 1,
@@ -664,7 +705,7 @@ func TestDeliverRetry_CircuitBreakerBlocks(t *testing.T) {
Event: event,
Target: database.Target{
Name: "test-cb-block",
Type: database.TargetTypeRetry,
Type: database.TargetTypeHTTP,
Config: newHTTPTargetConfig("http://will-not-be-called.invalid"),
MaxRetries: 5,
},
@@ -672,7 +713,7 @@ func TestDeliverRetry_CircuitBreakerBlocks(t *testing.T) {
d.ID = delivery.ID
d.Target.ID = targetID
e.deliverRetry(context.TODO(), db, d, task)
e.deliverHTTP(context.TODO(), db, d, task)
// Delivery should be retrying (circuit open, no attempt made)
var updated database.Delivery