recordFailure records a fan-out failure and schedules an aggressive retry. Per design §4.4 there is NO terminal 'failed' state on the outbox — at-least-once requires that we retry indefinitely. After many attempts we'd page ops; today we log and let the row stay pending until human intervention or a
(ctx context.Context, eventID string, errMsg string)
| 420 | // many attempts we'd page ops; today we log and let the row stay |
| 421 | // pending until human intervention or a successful retry. |
| 422 | func (w *OutboxWorker) recordFailure(ctx context.Context, eventID string, errMsg string) { |
| 423 | // Truncate to the CHECK constraint cap to avoid an INSERT failure |
| 424 | // inside the failure-recording path. |
| 425 | if len(errMsg) > 4000 { |
| 426 | errMsg = errMsg[:4000] |
| 427 | } |
| 428 | cap := time.Duration(60) * time.Second |
| 429 | if w.leaseDuration < cap { |
| 430 | cap = w.leaseDuration |
| 431 | } |
| 432 | _, err := w.pool.Exec(ctx, |
| 433 | `UPDATE webhook_events |
| 434 | SET attempts = attempts + 1, |
| 435 | last_error = $2, |
| 436 | next_poll_at = now() + ($3 * interval '1 second') |
| 437 | WHERE id = $1`, |
| 438 | eventID, errMsg, int(cap.Seconds()), |
| 439 | ) |
| 440 | if err != nil { |
| 441 | log.Printf("[outbox-worker] recordFailure err: id=%s err=%v", eventID, err) |
| 442 | } |
| 443 | } |
| 444 | |
| 445 | func derefString(p *string) string { |
| 446 | if p == nil { |