From 597b18b2a08172a672feab2c202f8de95b04d726 Mon Sep 17 00:00:00 2001
From: Josh Siviour <mrk@mulgadc.com>
Date: Sat, 6 Jun 2026 12:03:51 +1000
Subject: [PATCH] fix(eks): map transient JetStream errors in ListClusters to
 ServiceUnavailable (#302)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

aws eks list-clusters returned InternalErrorException for a few seconds after a
daemon restart, while describe-cluster (by name) succeeded. During the
post-restart window the per-account KV stream's Raft group has no leader yet, so
kv.Keys() (an ordered consumer over the warming stream) fails with a transient
NATS error. ListClusters wrapped it as a fmt.Errorf string the daemon's
ValidErrorCode cannot map, so it sanitized to ServerInternal — a misleading 500
that tells AWS clients the request is non-retryable.

Classify transient NATS conditions (no-responders, timeout, no-stream-response,
connection-closed) and return a retryable ServiceUnavailable (503) instead;
genuine faults still pass through wrapped. describe-cluster survived only
because kv.Get on the same bucket is cheaper than the consumer kv.Keys spins up.
---
 spinifex/handlers/eks/list_clusters_test.go | 27 +++++++++++++++++++
 spinifex/handlers/eks/service_impl.go       | 30 ++++++++++++++++++---
 2 files changed, 54 insertions(+), 3 deletions(-)

diff --git a/spinifex/handlers/eks/list_clusters_test.go b/spinifex/handlers/eks/list_clusters_test.go
index 501345415..a2ff78ea1 100644
--- a/spinifex/handlers/eks/list_clusters_test.go
+++ b/spinifex/handlers/eks/list_clusters_test.go
@@ -1,10 +1,14 @@
 package handlers_eks
 
 import (
+	"errors"
+	"fmt"
 	"testing"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/eks"
+	"github.com/mulgadc/spinifex/spinifex/awserrors"
+	"github.com/nats-io/nats.go"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -53,3 +57,26 @@ func TestEKSServiceImpl_ListClustersMaxResultsClamped(t *testing.T) {
 		assert.Nil(t, out.NextToken, "MaxResults=%d single page must not advertise a NextToken", mr)
 	}
 }
+
+// TestEKSReadUnavailableClassification pins the post-restart behaviour: a
+// transient JetStream error (no leader yet) surfaces as a retryable
+// ServiceUnavailable, while a genuine fault stays wrapped (the daemon maps it
+// to InternalError). Regression guard for list-clusters returning a misleading
+// 500 during the KV warmup window.
+func TestEKSReadUnavailableClassification(t *testing.T) {
+	transient := []error{
+		nats.ErrNoResponders,
+		nats.ErrTimeout,
+		nats.ErrNoStreamResponse,
+		nats.ErrConnectionClosed,
+		fmt.Errorf("kv keys: %w", nats.ErrNoResponders), // wrapped, as ListClusters wraps it
+	}
+	for _, e := range transient {
+		assert.True(t, natsTransient(e), "%v must be classified transient", e)
+		assert.EqualError(t, eksReadUnavailableOr(e, "op"), awserrors.ErrorServiceUnavailable)
+	}
+
+	genuine := errors.New("corrupt cluster meta")
+	assert.False(t, natsTransient(genuine))
+	assert.ErrorContains(t, eksReadUnavailableOr(genuine, "list cluster names"), "list cluster names: corrupt cluster meta")
+}
diff --git a/spinifex/handlers/eks/service_impl.go b/spinifex/handlers/eks/service_impl.go
index 5b0cad159..10d740dbc 100644
--- a/spinifex/handlers/eks/service_impl.go
+++ b/spinifex/handlers/eks/service_impl.go
@@ -533,15 +533,15 @@ func (s *EKSServiceImpl) DescribeCluster(input *eks.DescribeClusterInput, accoun
 func (s *EKSServiceImpl) ListClusters(input *eks.ListClustersInput, accountID string) (*eks.ListClustersOutput, error) {
 	js, err := s.deps.NATSConn.JetStream()
 	if err != nil {
-		return nil, fmt.Errorf("jetstream: %w", err)
+		return nil, eksReadUnavailableOr(err, "jetstream")
 	}
 	acctKV, err := GetOrCreateAccountBucket(js, accountID)
 	if err != nil {
-		return nil, fmt.Errorf("get account bucket: %w", err)
+		return nil, eksReadUnavailableOr(err, "get account bucket")
 	}
 	names, err := listClusterNames(acctKV)
 	if err != nil {
-		return nil, err
+		return nil, eksReadUnavailableOr(err, "list cluster names")
 	}
 	sort.Strings(names)
 
@@ -1234,6 +1234,30 @@ func clusterMetaToAWS(meta *ClusterMeta) *eks.Cluster {
 	return out
 }
 
+// natsTransient reports whether err is a transient NATS/JetStream condition
+// rather than a genuine fault. These occur in the post-restart window before
+// the KV stream's Raft group has elected a leader: JetStream requests get no
+// responder, time out, or the stream is briefly unreachable.
+func natsTransient(err error) bool {
+	return err != nil && (errors.Is(err, nats.ErrNoResponders) ||
+		errors.Is(err, nats.ErrTimeout) ||
+		errors.Is(err, nats.ErrNoStreamResponse) ||
+		errors.Is(err, nats.ErrConnectionClosed))
+}
+
+// eksReadUnavailableOr maps a transient NATS/JetStream failure to a retryable
+// ServiceUnavailable (503) so AWS clients back off and retry, instead of the
+// daemon sanitizing an unrecognised wrapped error to a misleading 500
+// InternalError. Non-transient errors pass through wrapped so real faults stay
+// visible.
+func eksReadUnavailableOr(err error, op string) error {
+	if natsTransient(err) {
+		slog.Warn("EKS read temporarily unavailable", "op", op, "err", err)
+		return errors.New(awserrors.ErrorServiceUnavailable)
+	}
+	return fmt.Errorf("%s: %w", op, err)
+}
+
 func listClusterNames(kv nats.KeyValue) ([]string, error) {
 	keys, err := kv.Keys()
 	if err != nil {