Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3c7b7ebf59 |
@@ -40,10 +40,10 @@ const (
|
|||||||
// explosions regardless of queue depth.
|
// explosions regardless of queue depth.
|
||||||
defaultWorkers = 10
|
defaultWorkers = 10
|
||||||
|
|
||||||
// sweepInterval is how often the recovery routine runs
|
// retrySweepInterval is how often the recovery routine runs
|
||||||
// once it has run at start. Each of those runs is a tick of
|
// once it has run at start. Each of those runs is a tick of
|
||||||
// the sweep.
|
// the sweep.
|
||||||
sweepInterval = 60 * time.Second
|
retrySweepInterval = 60 * time.Second
|
||||||
|
|
||||||
// pendingSweepMinAge is how long a delivery must have sat
|
// pendingSweepMinAge is how long a delivery must have sat
|
||||||
// untouched at pending before the sweep will look at it.
|
// untouched at pending before the sweep will look at it.
|
||||||
@@ -65,7 +65,7 @@ const (
|
|||||||
|
|
||||||
// pendingSweepBatch bounds how many stranded pending deliveries
|
// pendingSweepBatch bounds how many stranded pending deliveries
|
||||||
// one tick of the sweep re-dispatches for one webhook. The sweep
|
// one tick of the sweep re-dispatches for one webhook. The sweep
|
||||||
// ticks every sweepInterval, so a larger backlog drains
|
// ticks every retrySweepInterval, so a larger backlog drains
|
||||||
// across successive ticks instead of arriving as one burst against
|
// across successive ticks instead of arriving as one burst against
|
||||||
// a database that was already struggling to accept writes.
|
// a database that was already struggling to accept writes.
|
||||||
pendingSweepBatch = 500
|
pendingSweepBatch = 500
|
||||||
@@ -649,11 +649,11 @@ func (e *Engine) abandonRetryForMissingTarget(
|
|||||||
|
|
||||||
// runRecovery runs the recovery routine, recoverDeliveries, once when
|
// runRecovery runs the recovery routine, recoverDeliveries, once when
|
||||||
// the engine starts and then on every tick of the sweep, every
|
// the engine starts and then on every tick of the sweep, every
|
||||||
// sweepInterval, until the engine stops.
|
// retrySweepInterval, until the engine stops.
|
||||||
func (e *Engine) runRecovery(ctx context.Context) {
|
func (e *Engine) runRecovery(ctx context.Context) {
|
||||||
defer e.wg.Done()
|
defer e.wg.Done()
|
||||||
|
|
||||||
ticker := time.NewTicker(sweepInterval)
|
ticker := time.NewTicker(retrySweepInterval)
|
||||||
defer ticker.Stop()
|
defer ticker.Stop()
|
||||||
|
|
||||||
e.recoverDeliveries(ctx, true)
|
e.recoverDeliveries(ctx, true)
|
||||||
|
|||||||
Reference in New Issue
Block a user