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
53 lines
1.4 KiB
Go
53 lines
1.4 KiB
Go
package flag
|
|
|
|
import (
|
|
"time"
|
|
|
|
"github.com/sirupsen/logrus"
|
|
"github.com/spf13/pflag"
|
|
)
|
|
|
|
const (
|
|
LogLevel = "log-level"
|
|
Listen = "listen"
|
|
Worker = "worker"
|
|
|
|
RetryPeriod = "retry-period"
|
|
RetryBatchSize = "retry-batch-size"
|
|
AbortAfter = "abort-after"
|
|
MinBackoffDelay = "min-backoff-delay"
|
|
MaxBackoffDelay = "max-backoff-delay"
|
|
|
|
KafkaTopics = "kafka-topics"
|
|
AutoMigrate = "auto-migrate"
|
|
)
|
|
|
|
const (
|
|
DefaultBindAddressServer = ":8080"
|
|
|
|
DefaultPostgresConnString = "postgresql://webhooks:[email protected]/webhooks?sslmode=disable"
|
|
|
|
DefaultKafkaTopic = "default"
|
|
)
|
|
|
|
var (
|
|
DefaultRetryPeriod = 3 * time.Second
|
|
DefaultRetryBatchSize = 50
|
|
)
|
|
|
|
func Init(flagSet *pflag.FlagSet) {
|
|
flagSet.String(LogLevel, logrus.InfoLevel.String(), "Log level")
|
|
|
|
flagSet.String(Listen, DefaultBindAddressServer, "server HTTP bind address")
|
|
flagSet.Duration(RetryPeriod, DefaultRetryPeriod, "worker retry period")
|
|
flagSet.Int(RetryBatchSize, DefaultRetryBatchSize, "number of webhook IDs to claim per retry tick")
|
|
flagSet.Bool(Worker, false, "Enable worker on server")
|
|
|
|
flagSet.StringSlice(KafkaTopics, []string{DefaultKafkaTopic}, "Kafka topics")
|
|
|
|
flagSet.Duration(AbortAfter, 30*24*time.Hour, "consider a webhook as failed after retrying it for this duration.")
|
|
flagSet.Duration(MinBackoffDelay, time.Minute, "minimum backoff delay")
|
|
flagSet.Duration(MaxBackoffDelay, time.Hour, "maximum backoff delay")
|
|
flagSet.Bool(AutoMigrate, false, "auto migrate database")
|
|
}
|