Leave a delivery cut off by a stop for the next start (closes #527)
check / check (push) Waiting to run
check / check (push) Waiting to run
Stopping the engine cancels the context every http and slack request is sent with. An attempt cut off that way was recorded as failed, which for a target with no retries, or on a last retry, ended the delivery failed and nothing sent it again. The shared retry core now records nothing for a failed attempt when the engine's context is cancelled: no result, no circuit breaker outcome, no status change, no count in the attempt metrics. The delivery stays pending or retrying and startup recovery sends it again. A target's own timeout is still a failed attempt. The README says what happens to an attempt in flight at a stop and no longer claims every attempt is recorded. Model: opus-5-5
This commit was merged in pull request #532.
This commit is contained in:
@@ -57,7 +57,11 @@ type httpCore struct {
|
||||
// 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.
|
||||
//
|
||||
// ctx is the context attempt sends its request with. Stopping the
|
||||
// engine cancels it; see interruptedByStop.
|
||||
func (c *httpCore) deliver(
|
||||
ctx context.Context,
|
||||
webhookDB *gorm.DB,
|
||||
d *database.Delivery,
|
||||
task *Task,
|
||||
@@ -66,21 +70,55 @@ func (c *httpCore) deliver(
|
||||
attempt func() attemptResult,
|
||||
) {
|
||||
if maxRetries == 0 {
|
||||
c.fireAndForget(webhookDB, d, attempt())
|
||||
c.fireAndForget(ctx, webhookDB, d, attempt())
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
c.withRetry(
|
||||
webhookDB, d, task, sched, maxRetries, attempt,
|
||||
ctx, webhookDB, d, task, sched, maxRetries, attempt,
|
||||
)
|
||||
}
|
||||
|
||||
// interruptedByStop reports whether an attempt failed because the
|
||||
// engine is stopping, which cancels the context its request is made
|
||||
// with. Such an attempt is not recorded: no DeliveryResult, no circuit
|
||||
// breaker outcome, no status change, no count in the attempt metrics.
|
||||
// The delivery stays pending or retrying, and startup recovery sends
|
||||
// it again on the next start.
|
||||
//
|
||||
// A target's own timeout leaves that context alone, so its attempt is
|
||||
// still recorded as the failure it is.
|
||||
func (c *httpCore) interruptedByStop(
|
||||
ctx context.Context,
|
||||
d *database.Delivery,
|
||||
res attemptResult,
|
||||
) bool {
|
||||
if res.success || ctx.Err() == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
c.eng.log.Warn(
|
||||
"delivery attempt interrupted by shutdown; not recorded, "+
|
||||
"it is sent again on the next start",
|
||||
"delivery_id", d.ID,
|
||||
"event_id", d.EventID,
|
||||
"target_id", d.TargetID,
|
||||
)
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func (c *httpCore) fireAndForget(
|
||||
ctx context.Context,
|
||||
webhookDB *gorm.DB,
|
||||
d *database.Delivery,
|
||||
res attemptResult,
|
||||
) {
|
||||
if c.interruptedByStop(ctx, d, res) {
|
||||
return
|
||||
}
|
||||
|
||||
c.eng.observeAttempt(d.Target.Type, res.elapsed())
|
||||
|
||||
err := c.eng.recordResult(
|
||||
@@ -110,6 +148,7 @@ func (c *httpCore) fireAndForget(
|
||||
}
|
||||
|
||||
func (c *httpCore) withRetry(
|
||||
ctx context.Context,
|
||||
webhookDB *gorm.DB,
|
||||
d *database.Delivery,
|
||||
task *Task,
|
||||
@@ -131,6 +170,10 @@ func (c *httpCore) withRetry(
|
||||
|
||||
res := attempt()
|
||||
|
||||
if c.interruptedByStop(ctx, d, res) {
|
||||
return
|
||||
}
|
||||
|
||||
c.eng.observeAttempt(d.Target.Type, res.elapsed())
|
||||
|
||||
err := c.eng.recordResult(
|
||||
@@ -395,7 +438,7 @@ func (t *httpTarget) Deliver(
|
||||
}
|
||||
|
||||
t.deliver(
|
||||
webhookDB, d, task, sched,
|
||||
ctx, webhookDB, d, task, sched,
|
||||
d.Target.MaxRetries, attempt,
|
||||
)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user