From d143757598091456d5c343a3307c9fb2273d5c4c Mon Sep 17 00:00:00 2001 From: Julian Sommer <115516918+juliansommer@users.noreply.github.com> Date: Tue, 1 Sep 2026 12:25:10 +1000 Subject: [PATCH] fix: stop the eks tests waiting out fixed fan-out and backoff windows --- spinifex/handlers/eks/k3s_ha_control_plane.go | 24 +++++++++++++++---- .../handlers/eks/k3s_ha_control_plane_test.go | 12 +++++++--- spinifex/handlers/eks/nats_bootstrap.go | 6 ++--- spinifex/handlers/eks/nats_bootstrap_test.go | 8 +++---- 4 files changed, 35 insertions(+), 15 deletions(-) diff --git a/spinifex/handlers/eks/k3s_ha_control_plane.go b/spinifex/handlers/eks/k3s_ha_control_plane.go index b9bcf2e6c..ced8ad950 100644 --- a/spinifex/handlers/eks/k3s_ha_control_plane.go +++ b/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 { diff --git a/spinifex/handlers/eks/k3s_ha_control_plane_test.go b/spinifex/handlers/eks/k3s_ha_control_plane_test.go index 26e58db28..d7bdac40c 100644 --- a/spinifex/handlers/eks/k3s_ha_control_plane_test.go +++ b/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{} diff --git a/spinifex/handlers/eks/nats_bootstrap.go b/spinifex/handlers/eks/nats_bootstrap.go index 877a03a84..4b59b45f8 100644 --- a/spinifex/handlers/eks/nats_bootstrap.go +++ b/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 } diff --git a/spinifex/handlers/eks/nats_bootstrap_test.go b/spinifex/handlers/eks/nats_bootstrap_test.go index 9c7ca411f..5e891cb3b 100644 --- a/spinifex/handlers/eks/nats_bootstrap_test.go +++ b/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) })