feat(monitor/collectors/coinbase): reconnect-with-backoff helper (1s init, 60s max, ctx-aware)

PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/reconnect.go

@@ -0,0 +1,70 @@
+// ©AngelaMos | 2026
+// reconnect.go
+
+package coinbase
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
+)
+
+const (
+	defaultReconnectInitial = time.Second
+	defaultReconnectMax     = 60 * time.Second
+)
+
+type ReconnectConfig struct {
+	InitialInterval time.Duration
+	MaxInterval     time.Duration
+}
+
+type ConnHandler func(ctx context.Context, conn *Conn) error
+
+func PermanentReconnectError(err error) error {
+	return backoff.Permanent(err)
+}
+
+func Reconnect(ctx context.Context, d Dialer, cfg ReconnectConfig, handler ConnHandler) error {
+	if cfg.InitialInterval <= 0 {
+		cfg.InitialInterval = defaultReconnectInitial
+	}
+	if cfg.MaxInterval <= 0 {
+		cfg.MaxInterval = defaultReconnectMax
+	}
+
+	bo := backoff.NewExponentialBackOff()
+	bo.InitialInterval = cfg.InitialInterval
+	bo.MaxInterval = cfg.MaxInterval
+	bo.MaxElapsedTime = 0
+
+	op := func() error {
+		if ctx.Err() != nil {
+			return backoff.Permanent(ctx.Err())
+		}
+		conn, err := d.Dial(ctx)
+		if err != nil {
+			if ctx.Err() != nil {
+				return backoff.Permanent(ctx.Err())
+			}
+			return err
+		}
+		hErr := handler(ctx, conn)
+		_ = conn.Close()
+		if hErr == nil {
+			return nil
+		}
+		var perm *backoff.PermanentError
+		if errors.As(hErr, &perm) {
+			return hErr
+		}
+		if ctx.Err() != nil {
+			return backoff.Permanent(ctx.Err())
+		}
+		return hErr
+	}
+
+	return backoff.Retry(op, backoff.WithContext(bo, ctx))
+}

PROJECTS/advanced/monitor-the-situation-dashboard/backend/internal/collectors/coinbase/reconnect_test.go

@@ -0,0 +1,81 @@
+// ©AngelaMos | 2026
+// reconnect_test.go
+
+package coinbase_test
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/carterperez-dev/monitor-the-situation/backend/internal/collectors/coinbase"
+)
+
+type stubDialer struct {
+	failsBeforeOK int32
+	called        atomic.Int32
+}
+
+func (d *stubDialer) Dial(_ context.Context) (*coinbase.Conn, error) {
+	n := d.called.Add(1)
+	if n <= d.failsBeforeOK {
+		return nil, errors.New("transient dial failure")
+	}
+	return &coinbase.Conn{}, nil
+}
+
+func TestReconnect_StopsOnContextCancel(t *testing.T) {
+	d := &stubDialer{failsBeforeOK: 0}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 80*time.Millisecond)
+	defer cancel()
+
+	calls := atomic.Int32{}
+	err := coinbase.Reconnect(ctx, d, coinbase.ReconnectConfig{
+		InitialInterval: 5 * time.Millisecond,
+		MaxInterval:     20 * time.Millisecond,
+	}, func(_ context.Context, _ *coinbase.Conn) error {
+		calls.Add(1)
+		return errors.New("upstream closed")
+	})
+	require.True(t, errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled))
+	require.Greater(t, calls.Load(), int32(1), "callback should fire multiple times before ctx expires")
+}
+
+func TestReconnect_TransientDialFailureBacksOffThenSucceeds(t *testing.T) {
+	d := &stubDialer{failsBeforeOK: 3}
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+
+	cbInvocations := atomic.Int32{}
+	err := coinbase.Reconnect(ctx, d, coinbase.ReconnectConfig{
+		InitialInterval: 10 * time.Millisecond,
+		MaxInterval:     50 * time.Millisecond,
+	}, func(_ context.Context, _ *coinbase.Conn) error {
+		cbInvocations.Add(1)
+		return errors.New("force loop")
+	})
+	require.True(t, errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled))
+	require.GreaterOrEqual(t, d.called.Load(), int32(4), "must dial at least 4 times (3 failures + 1 success)")
+	require.GreaterOrEqual(t, cbInvocations.Load(), int32(1))
+}
+
+func TestReconnect_PermanentErrorFromCallbackPropagates(t *testing.T) {
+	d := &stubDialer{failsBeforeOK: 0}
+	sentinel := errors.New("hard stop")
+
+	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+	defer cancel()
+
+	err := coinbase.Reconnect(ctx, d, coinbase.ReconnectConfig{
+		InitialInterval: 5 * time.Millisecond,
+		MaxInterval:     10 * time.Millisecond,
+	}, func(_ context.Context, _ *coinbase.Conn) error {
+		return coinbase.PermanentReconnectError(sentinel)
+	})
+	require.ErrorIs(t, err, sentinel)
+}