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
27 lines
595 B
Go
27 lines
595 B
Go
package otlp
|
|
|
|
import (
|
|
"fmt"
|
|
"net/http"
|
|
"time"
|
|
|
|
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
|
|
"go.uber.org/fx"
|
|
)
|
|
|
|
func HttpClientModule() fx.Option {
|
|
return fx.Provide(func() *http.Client {
|
|
return &http.Client{
|
|
Timeout: 30 * time.Second,
|
|
Transport: otelhttp.NewTransport(http.DefaultTransport, otelhttp.WithSpanNameFormatter(func(operation string, r *http.Request) string {
|
|
str := fmt.Sprintf("%s %s", r.Method, r.URL.Path)
|
|
if len(r.URL.Query()) == 0 {
|
|
return str
|
|
}
|
|
|
|
return fmt.Sprintf("%s?%s", str, r.URL.Query().Encode())
|
|
})),
|
|
}
|
|
})
|
|
}
|