Fail a pending delivery whose target was deleted (closes #293)
check / check (push) Successful in 3m38s

Restart recovery and the pending sweep skipped a pending delivery
whose target was missing from the batch's target map, every minute,
for the life of the database. A miss now asks loadTarget: no row
fails the delivery terminally with a recorded reason; any other error
leaves it pending, since the map is also empty when its query failed;
a target found there is used.

The failure goes through the ownership-gated function the retrying
paths already used, now failMissingTarget. Once it owns the delivery
it re-reads the row and fails it only if the status is unchanged, so
a delivery sent and settled in between is left alone.

Model: opus-5-5
This commit was merged in pull request #329.
This commit is contained in:
2026-09-29 06:48:19 +02:00
parent e0b211f960
commit 51580a2bc6
3 changed files with 354 additions and 32 deletions
+59 -23
View File
@@ -728,9 +728,7 @@ func (e *Engine) recoverSingleRetry(
// webhook on one bad read would be a far larger fault than
// the strand it is meant to clear.
if errors.Is(err, gorm.ErrRecordNotFound) {
e.failMissingTargetRetry(
webhookDB, webhookID, d,
)
e.failMissingTarget(webhookDB, webhookID, d)
return
}
@@ -1133,9 +1131,7 @@ func (e *Engine) sweepSingleRetry(
// Deleted is terminal, unreadable is not; see
// recoverSingleRetry.
if errors.Is(err, gorm.ErrRecordNotFound) {
e.failMissingTargetRetry(
webhookDB, webhookID, d,
)
e.failMissingTarget(webhookDB, webhookID, d)
return
}
@@ -1249,19 +1245,19 @@ func (e *Engine) failUnretryableRetry(
e.failDelivery(webhookDB, d, target.Type, reason)
}
// failMissingTargetRetry terminally fails an orphaned retrying
// delivery whose target row is gone. Both restart recovery and the
// periodic sweep call it, so the transition exists once.
// failMissingTarget terminally fails a recovered delivery, pending or
// retrying, whose target row is gone. Restart recovery and the periodic
// sweep call it for both statuses, so the transition exists once.
//
// Until it existed both paths logged the failed lookup and returned,
// which left the delivery retrying for the life of the database and
// Until it existed those paths logged the failed lookup and moved on,
// which left the delivery where it was for the life of the database and
// the sweep repeating the same error every minute forever. Failing it
// with a recorded reason is the treatment the other orphaned-retry
// cases already get, so all of them read alike in the event log.
//
// Logged at warn rather than error: a deleted target is an operator
// action, not a system fault.
func (e *Engine) failMissingTargetRetry(
func (e *Engine) failMissingTarget(
webhookDB *gorm.DB,
webhookID string,
d *database.Delivery,
@@ -1274,13 +1270,37 @@ func (e *Engine) failMissingTargetRetry(
defer e.inflight.release(d.ID)
// The batch was read before ownership was taken, and a worker may
// have settled the delivery and let it go in between. Only a row
// still in the status the batch read is failed.
row, err := e.loadDelivery(webhookDB, d.ID)
if err != nil {
e.log.Error(
"failed to load delivery",
"delivery_id", d.ID,
"error", err,
)
return
}
if row.Status != d.Status {
e.log.Debug(
"delivery already handled, not failed",
"delivery_id", d.ID,
"status", row.Status,
)
return
}
targetType, reason := e.missingTargetReason(d.TargetID)
e.log.Warn(
"failing orphaned retrying delivery: "+
"its target no longer exists",
"failing recovered delivery: its target no longer exists",
"webhook_id", webhookID,
"delivery_id", d.ID,
"status", d.Status,
"target_id", d.TargetID,
"target_type", targetType,
)
@@ -1314,15 +1334,14 @@ func (e *Engine) missingTargetReason(
if err != nil {
return "", fmt.Sprintf(
"target %s no longer exists; the delivery "+
"cannot be retried and has been failed "+
"terminally",
"has been failed terminally",
targetID,
)
}
return target.Type, fmt.Sprintf(
"target %q (type %s) was deleted; the delivery "+
"cannot be retried and has been failed terminally",
"has been failed terminally",
target.Name, target.Type,
)
}
@@ -2021,13 +2040,30 @@ func (e *Engine) sendRecoveredDeliveries(
target, ok := targetMap[deliveries[i].TargetID]
if !ok {
e.log.Error(
"target not found for delivery",
"delivery_id", deliveries[i].ID,
"target_id", deliveries[i].TargetID,
)
// A missing entry does not mean the target is gone: the
// map is also empty when its query failed. Only a lookup
// that finds no row ends the delivery; any other error
// leaves it pending for the next sweep. See
// recoverSingleRetry.
var err error
continue
target, err = e.loadTarget(deliveries[i].TargetID)
if errors.Is(err, gorm.ErrRecordNotFound) {
e.failMissingTarget(webhookDB, webhookID, &deliveries[i])
continue
}
if err != nil {
e.log.Error(
"failed to load target for recovered delivery",
"delivery_id", deliveries[i].ID,
"target_id", deliveries[i].TargetID,
"error", err,
)
continue
}
}
if !e.takeForRedispatch(