@@ -2,40 +2,68 @@ package webhook
22
33import (
44 "context"
5+ "fmt"
56 "log/slog"
67 "sync"
78 "time"
89)
910
11+ // RetryConfig configures exponential backoff retry for sync handlers.
12+ type RetryConfig struct {
13+ // MaxAttempts is the total number of attempts (1 = no retry). Default: 3.
14+ MaxAttempts int
15+ // BaseDelay is the initial backoff duration. Default: 1s.
16+ BaseDelay time.Duration
17+ // MaxDelay caps the backoff duration. Default: 30s.
18+ MaxDelay time.Duration
19+ }
20+
21+ func defaultRetryConfig () RetryConfig {
22+ return RetryConfig {
23+ MaxAttempts : 3 ,
24+ BaseDelay : 1 * time .Second ,
25+ MaxDelay : 30 * time .Second ,
26+ }
27+ }
28+
1029type syncPayload struct {
1130 repoFullName string
1231 cloneURL string
1332}
1433
1534type SyncQueue struct {
16- ctx context.Context
17- handler SyncFunc
18- mu sync.Mutex
19- queue []string
20- dirty map [string ]bool
21- processing map [string ]bool
22- payloads map [string ]syncPayload
23- cond * sync.Cond
24- shutdown bool
25- wg sync.WaitGroup
35+ ctx context.Context
36+ handler SyncHandlerFunc
37+ retryConfig RetryConfig
38+ mu sync.Mutex
39+ queue []string
40+ dirty map [string ]bool
41+ processing map [string ]bool
42+ payloads map [string ]syncPayload
43+ cond * sync.Cond
44+ shutdown bool
45+ wg sync.WaitGroup
2646}
2747
28- func NewSyncQueue (workers int , handler SyncFunc ) * SyncQueue {
48+ func NewSyncQueue (workers int , handler SyncHandlerFunc ) * SyncQueue {
2949 return NewSyncQueueWithContext (context .Background (), workers , handler )
3050}
3151
32- func NewSyncQueueWithContext (ctx context.Context , workers int , handler SyncFunc ) * SyncQueue {
52+ func NewSyncQueueWithContext (ctx context.Context , workers int , handler SyncHandlerFunc ) * SyncQueue {
53+ return NewSyncQueueWithOptions (ctx , workers , handler , defaultRetryConfig ())
54+ }
55+
56+ func NewSyncQueueWithOptions (ctx context.Context , workers int , handler SyncHandlerFunc , retry RetryConfig ) * SyncQueue {
57+ if retry .MaxAttempts <= 0 {
58+ retry .MaxAttempts = 1
59+ }
3360 q := & SyncQueue {
34- ctx : ctx ,
35- handler : handler ,
36- dirty : make (map [string ]bool ),
37- processing : make (map [string ]bool ),
38- payloads : make (map [string ]syncPayload ),
61+ ctx : ctx ,
62+ handler : handler ,
63+ retryConfig : retry ,
64+ dirty : make (map [string ]bool ),
65+ processing : make (map [string ]bool ),
66+ payloads : make (map [string ]syncPayload ),
3967 }
4068 q .cond = sync .NewCond (& q .mu )
4169
@@ -109,12 +137,44 @@ func (q *SyncQueue) worker() {
109137}
110138
111139func (q * SyncQueue ) safeHandle (repo string , payload syncPayload ) {
140+ cfg := q .retryConfig
141+ delay := cfg .BaseDelay
142+
143+ for attempt := 1 ; attempt <= cfg .MaxAttempts ; attempt ++ {
144+ err := q .tryHandle (repo , payload )
145+ if err == nil {
146+ return
147+ }
148+
149+ if attempt == cfg .MaxAttempts {
150+ slog .Error ("sync handler failed, giving up" , "repo" , repo , "attempts" , attempt , "error" , err )
151+ return
152+ }
153+
154+ slog .Warn ("sync handler failed, retrying" , "repo" , repo , "attempt" , attempt , "retryIn" , delay , "error" , err )
155+
156+ select {
157+ case <- q .ctx .Done ():
158+ slog .Warn ("sync retry cancelled" , "repo" , repo , "attempt" , attempt )
159+ return
160+ case <- time .After (delay ):
161+ }
162+
163+ delay *= 2
164+ if delay > cfg .MaxDelay {
165+ delay = cfg .MaxDelay
166+ }
167+ }
168+ }
169+
170+ func (q * SyncQueue ) tryHandle (repo string , payload syncPayload ) (err error ) {
112171 defer func () {
113172 if r := recover (); r != nil {
114173 slog .Error ("sync handler panicked" , "repo" , repo , "panic" , r )
174+ err = fmt .Errorf ("panic: %v" , r )
115175 }
116176 }()
117- q .handler (q .ctx , payload .repoFullName , payload .cloneURL )
177+ return q .handler (q .ctx , payload .repoFullName , payload .cloneURL )
118178}
119179
120180func (q * SyncQueue ) get () (string , syncPayload , bool ) {
0 commit comments