refactor(provider): single event-stream loop with one reconnect policy and leak-free sends

pkg/provider/docker/events.go

@@ -2,16 +2,13 @@ package docker
 
 import (
 	"context"
-	"errors"
-	"io"
-	"log/slog"
 	"slices"
 	"strings"
-	"time"
 
 	"github.com/moby/moby/api/types/events"
 	"github.com/moby/moby/client"
 	"github.com/sablierapp/sablier/pkg/provider"
+	"github.com/sablierapp/sablier/pkg/provider/internal/mobyevents"
 	"github.com/sablierapp/sablier/pkg/sablier"
 )
 
@@ -85,76 +82,5 @@ func (p *Provider) InstanceEvents(ctx context.Context, opts provider.InstanceEve
 			return sablier.InstanceEvent{}, false
 		}
 	}
-	return streamEvents(ctx, p.l, dial, build, linearBackoff)
-}
-
-func linearBackoff(attempt int) time.Duration {
-	return min(time.Duration(attempt)*time.Second, 30*time.Second)
-}
-
-// streamEvents runs the reconnect loop. dial is called on each connection attempt;
-// backoff returns how long to wait before the next retry (return 0 in tests for instant retries).
-func streamEvents(
-	ctx context.Context,
-	l *slog.Logger,
-	dial func(ctx context.Context) client.EventsResult,
-	build func(context.Context, events.Message) (sablier.InstanceEvent, bool),
-	backoff func(attempt int) time.Duration,
-) sablier.InstanceEventStream {
-	eventsC := make(chan sablier.InstanceEvent)
-	errC := make(chan error, 1)
-	go func() {
-		defer close(eventsC)
-		defer close(errC)
-		for attempt := 0; ; attempt++ {
-			if attempt > 0 {
-				d := backoff(attempt)
-				l.WarnContext(ctx, "reconnecting event stream", "attempt", attempt, "backoff", d)
-				select {
-				case <-time.After(d):
-				case <-ctx.Done():
-					return
-				}
-			}
-
-			if reconnect := consumeEvents(ctx, l, eventsC, dial(ctx), build); !reconnect {
-				return
-			}
-		}
-	}()
-	return sablier.InstanceEventStream{Events: eventsC, Err: errC}
-}
-
-// consumeEvents drains an EventsResult, forwarding instance events built by the
-// build function. Returns true when the stream ended and the caller should
-// reconnect, or false when the context was cancelled and the caller should stop.
-func consumeEvents(
-	ctx context.Context,
-	l *slog.Logger,
-	instance chan<- sablier.InstanceEvent,
-	result client.EventsResult,
-	build func(context.Context, events.Message) (sablier.InstanceEvent, bool),
-) (reconnect bool) {
-	for {
-		select {
-		case msg, ok := <-result.Messages:
-			if !ok {
-				l.WarnContext(ctx, "event stream closed")
-				return true
-			}
-			l.DebugContext(ctx, "event received", "event", msg)
-			if event, ok := build(ctx, msg); ok {
-				instance <- event
-			}
-		case err, ok := <-result.Err:
-			if !ok || errors.Is(err, io.EOF) {
-				l.WarnContext(ctx, "event stream closed")
-				return true
-			}
-			l.ErrorContext(ctx, "event stream error", slog.Any("error", err))
-			return true
-		case <-ctx.Done():
-			return false
-		}
-	}
+	return mobyevents.Stream(ctx, p.l, dial, build)
 }

pkg/provider/dockerswarm/events.go

@@ -2,21 +2,15 @@ package dockerswarm
 
 import (
 	"context"
-	"errors"
-	"fmt"
-	"io"
-	"log/slog"
 	"slices"
-	"time"
 
 	"github.com/moby/moby/api/types/events"
 	"github.com/moby/moby/client"
 	"github.com/sablierapp/sablier/pkg/provider"
+	"github.com/sablierapp/sablier/pkg/provider/internal/mobyevents"
 	"github.com/sablierapp/sablier/pkg/sablier"
 )
 
-const maxReconnectAttempts = 10
-
 func (p *Provider) InstanceEvents(ctx context.Context, opts provider.InstanceEventsOptions) sablier.InstanceEventStream {
 	wantStopped := len(opts.Types) == 0 || slices.Contains(opts.Types, provider.InstanceEventStopped)
 	wantStarted := len(opts.Types) == 0 || slices.Contains(opts.Types, provider.InstanceEventStarted)
@@ -77,77 +71,5 @@ func (p *Provider) InstanceEvents(ctx context.Context, opts provider.InstanceEve
 			return sablier.InstanceEvent{}, false
 		}
 	}
-	return streamEvents(ctx, p.l, dial, build, linearBackoff)
-}
-
-func linearBackoff(attempt int) time.Duration {
-	return min(time.Duration(attempt)*time.Second, 30*time.Second)
-}
-
-// streamEvents runs the reconnect loop. dial is called on each connection attempt;
-// backoff returns how long to wait before the next retry.
-func streamEvents(
-	ctx context.Context,
-	l *slog.Logger,
-	dial func(ctx context.Context) client.EventsResult,
-	build func(context.Context, events.Message) (sablier.InstanceEvent, bool),
-	backoff func(attempt int) time.Duration,
-) sablier.InstanceEventStream {
-	eventsC := make(chan sablier.InstanceEvent)
-	errC := make(chan error, 1)
-	go func() {
-		defer close(eventsC)
-		defer close(errC)
-		for attempt := range maxReconnectAttempts + 1 {
-			if attempt > 0 {
-				d := backoff(attempt)
-				l.WarnContext(ctx, "reconnecting event stream", "attempt", attempt, "backoff", d)
-				select {
-				case <-time.After(d):
-				case <-ctx.Done():
-					return
-				}
-			}
-
-			if reconnect := consumeEvents(ctx, l, eventsC, dial(ctx), build); !reconnect {
-				return
-			}
-		}
-
-		errC <- fmt.Errorf("event stream reconnect failed after %d attempts", maxReconnectAttempts)
-	}()
-	return sablier.InstanceEventStream{Events: eventsC, Err: errC}
-}
-
-// consumeEvents drains a swarm EventsResult. Returns true if the caller should
-// reconnect, false if the context was cancelled.
-func consumeEvents(
-	ctx context.Context,
-	l *slog.Logger,
-	instance chan<- sablier.InstanceEvent,
-	result client.EventsResult,
-	build func(context.Context, events.Message) (sablier.InstanceEvent, bool),
-) (reconnect bool) {
-	for {
-		select {
-		case msg, ok := <-result.Messages:
-			if !ok {
-				l.WarnContext(ctx, "event stream closed")
-				return true
-			}
-			l.DebugContext(ctx, "event received", "event", msg)
-			if event, ok := build(ctx, msg); ok {
-				instance <- event
-			}
-		case err, ok := <-result.Err:
-			if !ok || errors.Is(err, io.EOF) {
-				l.WarnContext(ctx, "event stream closed")
-				return true
-			}
-			l.ErrorContext(ctx, "event stream error", slog.Any("error", err))
-			return true
-		case <-ctx.Done():
-			return false
-		}
-	}
+	return mobyevents.Stream(ctx, p.l, dial, build)
 }