Do not resend a delivery that restart recovery already sent (closes #299)
check / check (push) Successful in 3m39s
check / check (push) Successful in 3m39s
Restart recovery could find a just-written delivery pending, send it and release it before the receiver's Notify queued the same delivery. Notify's claim then succeeded on the released id, and the worker sent it again because the new-task path never read the delivery's row. Before sending a new task the worker now reads the delivery's status by primary key and skips the task unless the row still says pending, as the retry path already does for retrying. Nothing else can change the row while the worker owns the delivery. A row left pending by a failed bookkeeping write is still sent again. loadRetryDelivery is renamed loadDelivery now that both paths use it. Model: opus-5-5
This commit is contained in:
@@ -438,6 +438,31 @@ func (e *Engine) processNewTask(
|
||||
return
|
||||
}
|
||||
|
||||
// Restart recovery can send and release this delivery before the
|
||||
// receiver's Notify queues it. Ownership cannot refuse a delivery
|
||||
// nobody holds, so the row decides whether it still needs sending.
|
||||
row, err := e.loadDelivery(webhookDB, task.DeliveryID)
|
||||
if err != nil {
|
||||
e.log.Error(
|
||||
"failed to load delivery",
|
||||
"delivery_id", task.DeliveryID,
|
||||
"error", err,
|
||||
)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if row.Status != database.DeliveryStatusPending {
|
||||
e.log.Info(
|
||||
"delivery already handled, not sent again",
|
||||
"delivery_id", task.DeliveryID,
|
||||
"event_id", task.EventID,
|
||||
"status", row.Status,
|
||||
)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
event := buildEventFromTask(task)
|
||||
|
||||
event, err = e.hydrateEvent(
|
||||
@@ -482,7 +507,7 @@ func (e *Engine) processRetryTask(
|
||||
return
|
||||
}
|
||||
|
||||
d, err := e.loadRetryDelivery(
|
||||
d, err := e.loadDelivery(
|
||||
webhookDB, task.DeliveryID,
|
||||
)
|
||||
if err != nil {
|
||||
@@ -1643,7 +1668,7 @@ func (e *Engine) hydrateEvent(
|
||||
return event, nil
|
||||
}
|
||||
|
||||
func (e *Engine) loadRetryDelivery(
|
||||
func (e *Engine) loadDelivery(
|
||||
webhookDB *gorm.DB, deliveryID string,
|
||||
) (*database.Delivery, error) {
|
||||
var d database.Delivery
|
||||
|
||||
Reference in New Issue
Block a user