mirror of
https://github.com/hanzo-fi/webhooks.git
synced 2026-10-30 20:32:55 +00:00
Maxence MaireauxandGitHub
bc0e5976c2
Redesign the retry worker to support multi-worker scaling with continuous batch processing instead of all-at-once polling. Key changes: - Add "retrying" intermediate status for atomic claim pattern via CTE - Batch processing with configurable LIMIT (--retry-batch-size, default 50) - Reduce tick interval from 1min to 3s (--retry-period) for continuous flow - Parallel batch processing via pond worker pool (bounded concurrency) - Crash recovery for stale "retrying" attempts (runs once per minute) - Partial indexes for retry polling, claimed attempts, and recovery - UpdateAttemptsStatus scoped to "retrying" rows only (preserves history) - Worker resilient to per-webhook errors (log + continue, no crash) - HTTP client timeout (30s) prevents blocking on slow endpoints - Oldest retries prioritized via ORDER BY next_retry_after ASC
32 lines
1.3 KiB
Go
32 lines
1.3 KiB
Go
package storage
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
webhooks "github.com/formancehq/webhooks/pkg"
|
|
"github.com/pkg/errors"
|
|
)
|
|
|
|
var (
|
|
ErrConfigNotFound = errors.New("config not found")
|
|
ErrConfigNotModified = errors.New("config not modified")
|
|
ErrWebhookIDNotFound = errors.New("webhook ID not found")
|
|
ErrAttemptsNotModified = errors.New("attempt not modified")
|
|
)
|
|
|
|
type Store interface {
|
|
FindManyConfigs(ctx context.Context, filter map[string]any) ([]webhooks.Config, error)
|
|
InsertOneConfig(ctx context.Context, cfg webhooks.ConfigUser) (webhooks.Config, error)
|
|
DeleteOneConfig(ctx context.Context, id string) error
|
|
UpdateOneConfigActivation(ctx context.Context, id string, active bool) (webhooks.Config, error)
|
|
UpdateOneConfigSecret(ctx context.Context, id, secret string) (webhooks.Config, error)
|
|
FindAttemptsToRetryByWebhookID(ctx context.Context, webhookID string) ([]webhooks.Attempt, error)
|
|
FindWebhookIDsToRetry(ctx context.Context, limit int) (webhookIDs []string, err error)
|
|
RecoverStaleRetryingAttempts(ctx context.Context, staleDuration time.Duration) error
|
|
UpdateAttemptsStatus(ctx context.Context, webhookID string, status string) ([]webhooks.Attempt, error)
|
|
InsertOneAttempt(ctx context.Context, att webhooks.Attempt) error
|
|
Close(ctx context.Context) error
|
|
UpdateOneConfig(ctx context.Context, id string, cfg webhooks.ConfigUser) error
|
|
}
|