fix: fixed connection bug

Author: juicycleff

config.go

@@ -10,6 +10,11 @@ type Config struct {
 	// PollInterval is how often the delivery engine checks for pending deliveries.
 	PollInterval time.Duration
 
+	// MaxPollInterval caps the delivery engine's idle backoff. When polls
+	// come back empty the interval doubles from PollInterval up to this
+	// value, so an idle relay does not hit the store every PollInterval.
+	MaxPollInterval time.Duration
+
 	// BatchSize is the maximum number of deliveries dequeued per poll cycle.
 	BatchSize int
 
@@ -44,6 +49,7 @@ func DefaultConfig() Config {
 	return Config{
 		Concurrency:     10,
 		PollInterval:    1 * time.Second,
+		MaxPollInterval: 30 * time.Second,
 		BatchSize:       50,
 		RequestTimeout:  30 * time.Second,
 		MaxRetries:      5,

delivery/engine.go

@@ -30,13 +30,17 @@ type DLQPusher interface {
 
 // EngineConfig holds engine configuration.
 type EngineConfig struct {
-	Concurrency    int
-	PollInterval   time.Duration
-	BatchSize      int
-	RequestTimeout time.Duration
-	RetrySchedule  []time.Duration
-	Metrics        *observability.Metrics
-	Tracer         *observability.Tracer
+	Concurrency  int
+	PollInterval time.Duration
+	// MaxPollInterval caps the idle backoff. When polls come back empty the
+	// poll interval doubles from PollInterval up to this value, so an idle
+	// engine stops hammering the store every PollInterval. Defaults to 30s.
+	MaxPollInterval time.Duration
+	BatchSize       int
+	RequestTimeout  time.Duration
+	RetrySchedule   []time.Duration
+	Metrics         *observability.Metrics
+	Tracer          *observability.Tracer
 }
 
 // Engine is the delivery worker pool that dequeues and processes deliveries.
@@ -48,6 +52,7 @@ type Engine struct {
 	config  EngineConfig
 	logger  log.Logger
 
+	wakeCh chan struct{}
 	cancel context.CancelFunc
 	wg     sync.WaitGroup
 }
@@ -57,13 +62,20 @@ func NewEngine(store EngineStore, dlq DLQPusher, cfg EngineConfig, logger log.Lo
 	if logger == nil {
 		logger = log.NewNoopLogger()
 	}
+	if cfg.MaxPollInterval <= 0 {
+		cfg.MaxPollInterval = 30 * time.Second
+	}
+	if cfg.MaxPollInterval < cfg.PollInterval {
+		cfg.MaxPollInterval = cfg.PollInterval
+	}
 	return &Engine{
 		store:   store,
 		sender:  NewSender(cfg.RequestTimeout),
 		retrier: NewRetrier(cfg.RetrySchedule),
 		dlq:     dlq,
 		config:  cfg,
 		logger:  logger,
+		wakeCh:  make(chan struct{}, 1),
 	}
 }
 
@@ -86,39 +98,72 @@ func (e *Engine) Stop(_ context.Context) {
 	e.wg.Wait()
 }
 
-// pollLoop periodically dequeues pending deliveries and dispatches them to workers.
+// Wake nudges the poll loop to check for pending deliveries immediately,
+// resetting any idle backoff. It is non-blocking and safe to call from any
+// goroutine, including before Start.
+func (e *Engine) Wake() {
+	select {
+	case e.wakeCh <- struct{}{}:
+	default:
+	}
+}
+
+// pollLoop dequeues pending deliveries and dispatches them to workers. Empty
+// polls double the wait up to MaxPollInterval so an idle engine doesn't issue
+// a dequeue (an UPDATE/findAndModify against the store) every PollInterval;
+// any dequeued work or a Wake call resets the cadence to PollInterval.
 func (e *Engine) pollLoop(ctx context.Context) {
-	ticker := time.NewTicker(e.config.PollInterval)
-	defer ticker.Stop()
+	interval := e.config.PollInterval
+	timer := time.NewTimer(interval)
+	defer timer.Stop()
 
 	sem := make(chan struct{}, e.config.Concurrency)
 
 	for {
 		select {
 		case <-ctx.Done():
 			return
-		case <-ticker.C:
-			batch, err := e.store.Dequeue(ctx, e.config.BatchSize)
-			if err != nil {
-				e.logger.Error("dequeue failed", log.Any("error", err))
-				continue
-			}
-
-			for _, d := range batch {
+		case <-e.wakeCh:
+			interval = e.config.PollInterval
+			if !timer.Stop() {
 				select {
-				case <-ctx.Done():
-					return
-				case sem <- struct{}{}:
+				case <-timer.C:
+				default:
 				}
+			}
+		case <-timer.C:
+		}
+
+		batch, err := e.store.Dequeue(ctx, e.config.BatchSize)
+		if err != nil {
+			e.logger.Error("dequeue failed", log.Any("error", err))
+		}
 
-				e.wg.Add(1)
-				go func(del *Delivery) {
-					defer e.wg.Done()
-					defer func() { <-sem }()
-					e.process(ctx, del)
-				}(d)
+		if len(batch) > 0 {
+			interval = e.config.PollInterval
+		} else {
+			// Back off on empty polls and on errors alike; errors hot-
+			// spinning at PollInterval would only pile onto a struggling
+			// store.
+			interval = min(interval*2, e.config.MaxPollInterval)
+		}
+
+		for _, d := range batch {
+			select {
+			case <-ctx.Done():
+				return
+			case sem <- struct{}{}:
 			}
+
+			e.wg.Add(1)
+			go func(del *Delivery) {
+				defer e.wg.Done()
+				defer func() { <-sem }()
+				e.process(ctx, del)
+			}(d)
 		}
+
+		timer.Reset(interval)
 	}
 }
 

delivery/engine_test.go

@@ -342,3 +342,81 @@ func TestEngineNilDLQ(t *testing.T) {
 
 	engine.Stop(ctx)
 }
+
+// countingStore wraps an EngineStore and counts Dequeue calls.
+type countingStore struct {
+	delivery.EngineStore
+	dequeues atomic.Int32
+}
+
+func (c *countingStore) Dequeue(ctx context.Context, limit int) ([]*delivery.Delivery, error) {
+	c.dequeues.Add(1)
+	return c.EngineStore.Dequeue(ctx, limit)
+}
+
+func TestEngineIdleBackoff(t *testing.T) {
+	cs := &countingStore{EngineStore: memory.New()}
+	cfg := delivery.EngineConfig{
+		Concurrency:     2,
+		PollInterval:    10 * time.Millisecond,
+		MaxPollInterval: 160 * time.Millisecond,
+		BatchSize:       10,
+		RequestTimeout:  time.Second,
+	}
+	engine := delivery.NewEngine(cs, &stubDLQ{}, cfg, nil)
+
+	engine.Start(context.Background())
+	time.Sleep(1 * time.Second)
+	engine.Stop(context.Background())
+
+	// Fixed-rate polling at 10ms would issue ~100 dequeues in 1s. With
+	// doubling backoff capped at 160ms the loop polls at 10, 20, 40, 80,
+	// then every 160ms: roughly 10 calls. Allow generous slack.
+	if n := cs.dequeues.Load(); n > 20 {
+		t.Fatalf("expected backoff to limit idle dequeues to ~10, got %d", n)
+	}
+}
+
+func TestEngineWakeDeliversPromptly(t *testing.T) {
+	var delivered atomic.Int32
+	handler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		delivered.Add(1)
+		w.WriteHeader(http.StatusOK)
+	})
+	srv := httptest.NewServer(handler)
+	defer srv.Close()
+
+	store := memory.New()
+	cfg := delivery.EngineConfig{
+		Concurrency:     2,
+		PollInterval:    10 * time.Millisecond,
+		MaxPollInterval: 5 * time.Second,
+		BatchSize:       10,
+		RequestTimeout:  time.Second,
+		RetrySchedule:   []time.Duration{10 * time.Millisecond},
+	}
+	engine := delivery.NewEngine(store, &stubDLQ{}, cfg, nil)
+
+	ctx := context.Background()
+	engine.Start(ctx)
+	defer engine.Stop(ctx)
+
+	// Let the idle loop back off to a multi-second interval: polls land at
+	// ~10, 30, 70, 150, 310, 630, 1270ms; after that the next poll is
+	// seconds away.
+	time.Sleep(1300 * time.Millisecond)
+
+	// New work arrives; Wake must reset the backoff and trigger an
+	// immediate poll instead of waiting out the inflated interval.
+	createTestData(t, store, srv.URL)
+	engine.Wake()
+
+	deadline := time.After(700 * time.Millisecond)
+	for delivered.Load() == 0 {
+		select {
+		case <-deadline:
+			t.Fatal("delivery not processed promptly after Wake")
+		case <-time.After(10 * time.Millisecond):
+		}
+	}
+}

extension/config.go

@@ -57,6 +57,9 @@ func (c Config) ToRelayOptions() []relay.Option {
 	if c.PollInterval > 0 {
 		opts = append(opts, relay.WithPollInterval(c.PollInterval))
 	}
+	if c.MaxPollInterval > 0 {
+		opts = append(opts, relay.WithMaxPollInterval(c.MaxPollInterval))
+	}
 	if c.BatchSize > 0 {
 		opts = append(opts, relay.WithBatchSize(c.BatchSize))
 	}

extension/extension.go

@@ -321,6 +321,9 @@ func (e *Extension) mergeWithDefaults(cfg Config) Config {
 	if cfg.PollInterval == 0 {
 		cfg.PollInterval = defaults.PollInterval
 	}
+	if cfg.MaxPollInterval == 0 {
+		cfg.MaxPollInterval = defaults.MaxPollInterval
+	}
 	if cfg.BatchSize == 0 {
 		cfg.BatchSize = defaults.BatchSize
 	}
@@ -371,6 +374,9 @@ func (e *Extension) mergeConfigurations(yamlConfig, programmaticConfig Config) C
 	if yamlConfig.PollInterval == 0 && programmaticConfig.PollInterval != 0 {
 		yamlConfig.PollInterval = programmaticConfig.PollInterval
 	}
+	if yamlConfig.MaxPollInterval == 0 && programmaticConfig.MaxPollInterval != 0 {
+		yamlConfig.MaxPollInterval = programmaticConfig.MaxPollInterval
+	}
 	if yamlConfig.BatchSize == 0 && programmaticConfig.BatchSize != 0 {
 		yamlConfig.BatchSize = programmaticConfig.BatchSize
 	}