fix: stop the eks tests waiting out fixed fan-out and backoff windows

spinifex/handlers/eks/k3s_ha_control_plane.go

@@ -459,12 +459,25 @@ var _ HostScheduler = (*natsHostScheduler)(nil)
 const hostFanoutTimeout = time.Second
 
 type natsHostScheduler struct {
-	nc *nats.Conn
+	nc            *nats.Conn
+	fanoutTimeout time.Duration
+}
+
+// HostSchedulerOption customises a NATS fan-out HostScheduler.
+type HostSchedulerOption func(*natsHostScheduler)
+
+// WithFanoutTimeout overrides the window a fan-out waits for host replies.
+func WithFanoutTimeout(d time.Duration) HostSchedulerOption {
+	return func(h *natsHostScheduler) { h.fanoutTimeout = d }
 }
 
 // NewNATSHostScheduler returns a NATS fan-out HostScheduler for EKSServiceDeps.
-func NewNATSHostScheduler(nc *nats.Conn) HostScheduler {
-	return &natsHostScheduler{nc: nc}
+func NewNATSHostScheduler(nc *nats.Conn, opts ...HostSchedulerOption) HostScheduler {
+	h := &natsHostScheduler{nc: nc, fanoutTimeout: hostFanoutTimeout}
+	for _, opt := range opts {
+		opt(h)
+	}
+	return h
 }
 
 func (h *natsHostScheduler) SchedulableHosts(ctx context.Context, instanceType string) []string {
@@ -577,7 +590,8 @@ func (h *natsHostScheduler) InstanceHosts(ctx context.Context, instanceIDs []str
 	return out
 }
 
-// fanout publishes on subject and calls handle for each reply within hostFanoutTimeout.
+// fanout publishes on subject and calls handle for each reply within the
+// scheduler's fan-out window.
 func (h *natsHostScheduler) fanout(ctx context.Context, subject string, handle func([]byte)) {
 	if h.nc == nil {
 		return
@@ -599,7 +613,7 @@ func (h *natsHostScheduler) fanout(ctx context.Context, subject string, handle f
 		return
 	}
 
-	deadline := time.Now().Add(hostFanoutTimeout)
+	deadline := time.Now().Add(h.fanoutTimeout)
 	for {
 		remaining := time.Until(deadline)
 		if remaining <= 0 {

spinifex/handlers/eks/k3s_ha_control_plane_test.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/ec2"
@@ -23,6 +24,11 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// testFanoutTimeout shortens the fan-out window, which has no early exit and so
+// is otherwise the whole runtime of every scheduler test. The replies here are
+// local and land in tens of milliseconds, well inside this.
+const testFanoutTimeout = 250 * time.Millisecond
+
 // serveNodeStatus replies to spinifex.node.status fan-outs with the given node
 // snapshots so the NATS host scheduler can be exercised end-to-end.
 func serveNodeStatus(t *testing.T, nc *nats.Conn, nodes []types.NodeStatusResponse) {
@@ -51,7 +57,7 @@ func TestNATSHostScheduler_SchedulableHosts(t *testing.T) {
 			InstanceTypes: []types.InstanceTypeCap{{Name: "t3.medium", Available: 0}},
 		},
 	})
-	sched := NewNATSHostScheduler(nc)
+	sched := NewNATSHostScheduler(nc, WithFanoutTimeout(testFanoutTimeout))
 
 	// Customer type: only the node advertising free capacity is schedulable.
 	require.Equal(t, []string{"nodeA"}, sched.SchedulableHosts(context.Background(), "t3.medium"))
@@ -129,7 +135,7 @@ func TestNATSHostScheduler_SchedulableHosts_SpreadsByAZ(t *testing.T) {
 		{Node: "node-3", AZ: "az-b", InstanceTypes: []types.InstanceTypeCap{{Name: "t3.medium", Available: 1}}},
 		{Node: "node-4", AZ: "az-c", InstanceTypes: []types.InstanceTypeCap{{Name: "t3.medium", Available: 1}}},
 	})
-	sched := NewNATSHostScheduler(nc)
+	sched := NewNATSHostScheduler(nc, WithFanoutTimeout(testFanoutTimeout))
 
 	hosts := sched.SchedulableHosts(context.Background(), "t3.medium")
 	require.Len(t, hosts, 4)
@@ -404,7 +410,7 @@ func TestPlaceControlPlane_SpreadPrefersDistinctAZs(t *testing.T) {
 		{Node: "node-c", AZ: "az-2", TotalVCPU: 32, TotalMemGB: 128},
 		{Node: "node-d", AZ: "az-3", TotalVCPU: 32, TotalMemGB: 128},
 	})
-	sched := NewNATSHostScheduler(nc)
+	sched := NewNATSHostScheduler(nc, WithFanoutTimeout(testFanoutTimeout))
 	placer := &fakePlacer{} // reserved unset: defaults to the first MaxCount EligibleNodes, like the real service.
 	vpc := &seqK3sVPC{}
 	inst := &seqK3sInst{}

spinifex/handlers/eks/nats_bootstrap.go

@@ -29,7 +29,7 @@ const (
 // persistWithRetry runs op, retrying on transient errors up to
 // bootstrapPersistAttempts. A nil result or an errBootstrapPermanent result
 // returns immediately (no retry).
-func persistWithRetry(op func() error) error {
+func persistWithRetry(backoff time.Duration, op func() error) error {
 	var err error
 	for attempt := 1; attempt <= bootstrapPersistAttempts; attempt++ {
 		err = op()
@@ -37,7 +37,7 @@ func persistWithRetry(op func() error) error {
 			return err
 		}
 		if attempt < bootstrapPersistAttempts {
-			time.Sleep(bootstrapPersistBackoff)
+			time.Sleep(backoff)
 		}
 	}
 	return err
@@ -201,7 +201,7 @@ func (b *NATSBootstrap) RunForKinds(ctx context.Context, kinds []string) error {
 				return
 			}
 			mu.Unlock()
-			if err := persistWithRetry(func() error { return s.handler(ctx, m.Data) }); err != nil {
+			if err := persistWithRetry(bootstrapPersistBackoff, func() error { return s.handler(ctx, m.Data) }); err != nil {
 				errCh <- fmt.Errorf("persist %s: %w", s.kind, err)
 				return
 			}

spinifex/handlers/eks/nats_bootstrap_test.go

@@ -299,14 +299,14 @@ func TestNATSBootstrap_CANotBase64Rejected(t *testing.T) {
 func TestPersistWithRetry(t *testing.T) {
 	t.Run("succeeds first try", func(t *testing.T) {
 		calls := 0
-		err := persistWithRetry(func() error { calls++; return nil })
+		err := persistWithRetry(0, func() error { calls++; return nil })
 		require.NoError(t, err)
 		assert.Equal(t, 1, calls)
 	})
 
 	t.Run("retries transient then succeeds", func(t *testing.T) {
 		calls := 0
-		err := persistWithRetry(func() error {
+		err := persistWithRetry(0, func() error {
 			calls++
 			if calls < 2 {
 				return errors.New("transient kv blip")
@@ -319,7 +319,7 @@ func TestPersistWithRetry(t *testing.T) {
 
 	t.Run("permanent error returns immediately without retry", func(t *testing.T) {
 		calls := 0
-		err := persistWithRetry(func() error {
+		err := persistWithRetry(0, func() error {
 			calls++
 			return fmt.Errorf("%w: malformed envelope", errBootstrapPermanent)
 		})
@@ -329,7 +329,7 @@ func TestPersistWithRetry(t *testing.T) {
 
 	t.Run("transient error exhausts attempts", func(t *testing.T) {
 		calls := 0
-		err := persistWithRetry(func() error { calls++; return errors.New("always transient") })
+		err := persistWithRetry(0, func() error { calls++; return errors.New("always transient") })
 		require.Error(t, err)
 		assert.Equal(t, bootstrapPersistAttempts, calls)
 	})