fix: bound the KV bucket listing so a stuck DNS reconcile cannot wedge the cluster (#1071)

kvutil.BucketNames ranged the KeyValueStoreNames lister channel using whatever context the caller supplied, and the EKS desired-set builder supplies one that lives as long as the daemon. A lost stream-names reply therefore left the channel open with nothing ever sent on it and parked the caller for the life of the process. kvutil.Keys, immediately below it, already bounds exactly this case and says why; BucketNames never got the same treatment.

The DNS reconcile made that hang cluster-wide. A pass takes the reconcile lease before computing its batch and releases it on a deferred call, and the lease renews on a background timer for as long as the holder lives, so a pass that never returns holds it forever and the other nodes never take over. Observed on a three-node cluster: the zone object stopped being written after a rolling deploy, and every name published afterwards stayed NXDOMAIN until the holder was restarted.

The listing now carries its own deadline and reports a timeout as an error, never as a short list, because callers prune on absence. The pass bounds the batch computation as well, so any other deadline-free call reached from the desired-set builders costs a cycle rather than the cluster. Only the read side is bounded: it has no side effects, so abandoning it is safe, while the apply stays inline under the lease.

spinifex/handlers/dns/reconcile.go

@@ -21,6 +21,18 @@ import (
 // path by which a normal lifecycle event reaches DNS.
 const DefaultReconcileInterval = reconciler.DefaultResync
 
+// DefaultComputeTimeout bounds how long one pass spends building the desired
+// set before it gives up the cycle. It exists because the pass holds a
+// cluster-wide lease that renews itself for as long as the holder lives, so a
+// single call that never returns takes DNS publishing down on every node until
+// someone restarts the daemon holding it.
+//
+// It is generous relative to the resync interval on purpose. This is here to
+// break a hang, not to police a slow pass: the enumeration it bounds reads
+// every account's KV across the cluster, and a pass abandoned for being slow
+// costs a cycle for nothing.
+const DefaultComputeTimeout = 2 * time.Minute
+
 // DesiredFunc returns the full desired managed record set built from the live
 // resource inventory across all tenants. The daemon supplies it by enumerating
 // instances, load balancers, and EKS clusters.
@@ -91,8 +103,11 @@ type Reconciler struct {
 	writer   *Writer
 	desired  DesiredFunc
 	interval time.Duration
-	holder   string
-	sources  []reconciler.Source
+	// computeTimeout bounds how long one pass spends building the desired set.
+	// A field rather than a constant so a test can shrink it.
+	computeTimeout time.Duration
+	holder         string
+	sources        []reconciler.Source
 }
 
 // NewReconciler builds the drift backstop. It is disabled (a no-op) when
@@ -103,11 +118,12 @@ type Reconciler struct {
 // watch.
 func NewReconciler(cfg *config.Config, cluster *config.ClusterConfig, nc *nats.Conn, writer *Writer, desired DesiredFunc, sources ...reconciler.Source) *Reconciler {
 	r := &Reconciler{
-		nc:       nc,
-		writer:   writer,
-		desired:  desired,
-		interval: DefaultReconcileInterval,
-		sources:  sources,
+		nc:             nc,
+		writer:         writer,
+		desired:        desired,
+		interval:       DefaultReconcileInterval,
+		computeTimeout: DefaultComputeTimeout,
+		sources:        sources,
 	}
 	if cfg != nil {
 		r.holder = cfg.Node
@@ -167,7 +183,7 @@ func (r *Reconciler) reconcileOnce(ctx context.Context) {
 		}
 		defer release()
 	}
-	batch, err := r.computeBatch()
+	batch, err := r.computeBatchBounded()
 	if err != nil {
 		// A corrupt zone is handled in readZone; anything reaching here left the
 		// desired state unapplied, so surface it rather than burying it in WARN.
@@ -186,6 +202,44 @@ func (r *Reconciler) reconcileOnce(ctx context.Context) {
 	slog.Debug("dns reconcile: converged", "changes", res.Applied, "zones", res.Zones)
 }
 
+// computeBatchBounded runs computeBatch with a deadline, so a pass that never
+// returns gives the cycle up instead of holding the cluster-wide lease forever.
+//
+// Only the read side is bounded. computeBatch has no side effects, so a
+// computation abandoned here writes nothing and the caller can release the
+// lease safely; the apply that follows stays inline under the lease, because a
+// batch written by a node that has stopped being leader is the one thing the
+// election exists to prevent.
+func (r *Reconciler) computeBatchBounded() ([]Change, error) {
+	type outcome struct {
+		batch []Change
+		err   error
+	}
+	// Buffered, so the abandoned goroutine can finish and exit rather than
+	// parking on a send nobody will ever receive.
+	done := make(chan outcome, 1)
+	go func() {
+		batch, err := r.computeBatch()
+		done <- outcome{batch, err}
+	}()
+
+	// Zero means the default, not "no time at all": the reconciler is also built
+	// as a struct literal, and a bound that expires before the pass starts would
+	// turn a missing field into a loop that never computes anything.
+	bound := r.computeTimeout
+	if bound <= 0 {
+		bound = DefaultComputeTimeout
+	}
+	timer := time.NewTimer(bound)
+	defer timer.Stop()
+	select {
+	case res := <-done:
+		return res.batch, res.err
+	case <-timer.C:
+		return nil, fmt.Errorf("desired state not computed within %s", bound)
+	}
+}
+
 // computeBatch reads every zone holding prunable records and converges the
 // desired set against them. That is the base domain, plus the private zone once
 // instance records are prunable — a stale private record is as wrong as a stale

spinifex/handlers/dns/reconcile_test.go

@@ -8,6 +8,7 @@ import (
 
 	nsconfig "github.com/mulgadc/northstar/pkg/config"
 	"github.com/mulgadc/spinifex/spinifex/kvstore"
+	reconcilelock "github.com/mulgadc/spinifex/spinifex/network/reconcile"
 	"github.com/mulgadc/spinifex/spinifex/reconciler"
 	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/stretchr/testify/assert"
@@ -260,6 +261,52 @@ func TestReconciler_RunPerformsStartupPassThenStopsOnCancel(t *testing.T) {
 	}
 }
 
+// A pass that never finishes must give the cycle up rather than hold the
+// cluster-wide lease, because the lease renews itself for as long as the holder
+// lives: one stuck pass used to stop DNS publishing on every node until someone
+// restarted the daemon holding it. The assertion is on the lease being
+// acquirable afterwards, not merely on the call returning, since it is the
+// leadership and not the pass that the rest of the cluster waits on.
+func TestReconcileOnce_AStuckPassSurrendersLeadership(t *testing.T) {
+	_, nc, _ := testutil.StartTestJetStream(t)
+	endpoint, _ := fakeS3(t, "northstar")
+
+	blocked := make(chan struct{})
+	t.Cleanup(func() { close(blocked) })
+
+	r := &Reconciler{
+		enabled:    true,
+		baseDomain: testBase,
+		s3cfg: &nsconfig.S3Config{
+			Endpoint: endpoint, Bucket: "northstar", Region: "us-east-1",
+			AccessKey: "SYSTEM", SecretKey: "SYSTEMSECRET",
+		},
+		nc:             nc,
+		holder:         "stuck-node",
+		computeTimeout: 100 * time.Millisecond,
+		desired: func() DesiredSet {
+			<-blocked
+			return DesiredSet{}
+		},
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		r.reconcileOnce(t.Context())
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("reconcileOnce never returned: a stuck pass still holds the lease forever")
+	}
+
+	release, elected := reconcilelock.AcquireLeader(t.Context(), nc, KVBucketDNSReconcile, "next-node")
+	require.True(t, elected, "another node must be able to take the next cycle")
+	release()
+}
+
 // TestNewReconciler_RetainsItsWatchSources covers the wiring the daemon relies
 // on: sources supplied at construction have to reach the loop, and supplying
 // none is legal because that is the interval-only behaviour.

spinifex/kvutil/kvutil.go

@@ -137,6 +137,13 @@ func DeleteBucketIfExists(ctx context.Context, js jetstream.KeyValueManager, buc
 	return fmt.Errorf("delete KV bucket %s: %w", bucket, err)
 }
 
+// bucketListTimeout bounds one bucket listing. It is longer than the per-bucket
+// Keys bound because this listing pages over every stream in the cluster rather
+// than scanning one bucket, and it exists at all because a caller may hold a
+// context that lives as long as the process. A var, not a const, so a test can
+// shrink it.
+var bucketListTimeout = 10 * time.Second
+
 // BucketNames returns the name of every KV bucket, or an error if the listing
 // could not be completed. The lister closes its channel both when the listing
 // is complete and when the underlying stream-names request fails, so the
@@ -147,18 +154,31 @@ func DeleteBucketIfExists(ctx context.Context, js jetstream.KeyValueManager, buc
 //
 // The names are bucket names, already stripped of the KV_ stream prefix.
 func BucketNames(ctx context.Context, js jetstream.KeyValueManager) ([]string, error) {
-	lister := js.KeyValueStoreNames(ctx)
+	listCtx, cancel := context.WithTimeout(ctx, bucketListTimeout)
+	defer cancel()
 
-	// The lister goroutine blocks on an unbuffered send, so the channel is
-	// drained fully — an early return here would leak it until ctx expires.
+	lister := js.KeyValueStoreNames(listCtx)
+
+	// Selecting on the deadline rather than ranging the channel is what makes the
+	// bound real: a lost stream-names reply leaves the channel open with nothing
+	// ever sent on it, and a range parks here for the life of the process.
+	// Cancelling listCtx on the way out unblocks the lister's send, so the early
+	// return no longer leaks the goroutine it used to.
 	var names []string
-	for name := range lister.Name() {
-		names = append(names, name)
+	for {
+		select {
+		case name, open := <-lister.Name():
+			if !open {
+				if err := lister.Error(); err != nil {
+					return nil, fmt.Errorf("enumerate KV buckets: %w", err)
+				}
+				return names, nil
+			}
+			names = append(names, name)
+		case <-listCtx.Done():
+			return nil, fmt.Errorf("enumerate KV buckets: %w", listCtx.Err())
+		}
 	}
-	if err := lister.Error(); err != nil {
-		return nil, fmt.Errorf("enumerate KV buckets: %w", err)
-	}
-	return names, nil
 }
 
 // Keys lists a bucket's keys with the five-second bound used by the legacy API.

spinifex/kvutil/kvutil_test.go

@@ -3,6 +3,7 @@ package kvutil
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/mulgadc/spinifex/spinifex/testutil"
 	"github.com/mulgadc/spinifex/spinifex/utils"
@@ -137,6 +138,46 @@ func TestBucketNames_SurfacesEnumerationFailure(t *testing.T) {
 	assert.Nil(t, names)
 }
 
+// hangingNamesLister never sends and never closes, which is what a lost
+// stream-names reply looks like from the caller's side.
+type hangingNamesLister struct{ names chan string }
+
+func (l *hangingNamesLister) Name() <-chan string { return l.names }
+func (l *hangingNamesLister) Error() error        { return nil }
+
+// hangingKVManager answers only the listing call. The embedded interface is nil
+// on purpose: anything else this test reaches should fail loudly rather than
+// quietly return a zero value.
+type hangingKVManager struct{ jetstream.KeyValueManager }
+
+func (m *hangingKVManager) KeyValueStoreNames(context.Context) jetstream.KeyValueNamesLister {
+	return &hangingNamesLister{names: make(chan string)}
+}
+
+// A listing that never completes must not park the caller for the life of the
+// process. The context passed here carries no deadline, which is what the EKS
+// desired-set builder passes, so the bound inside BucketNames is the only thing
+// that can end this call.
+func TestBucketNames_GivesUpOnAListingThatNeverCompletes(t *testing.T) {
+	restore := bucketListTimeout
+	bucketListTimeout = 50 * time.Millisecond
+	t.Cleanup(func() { bucketListTimeout = restore })
+
+	done := make(chan error, 1)
+	go func() {
+		names, err := BucketNames(context.Background(), &hangingKVManager{})
+		assert.Nil(t, names, "a listing that did not finish must not read as a complete one")
+		done <- err
+	}()
+
+	select {
+	case err := <-done:
+		require.ErrorIs(t, err, context.DeadlineExceeded)
+	case <-time.After(30 * time.Second):
+		t.Fatal("BucketNames never returned: the listing is still unbounded")
+	}
+}
+
 func TestKeys_ListsKeysAndSurfacesCancellation(t *testing.T) {
 	js := startJetStream(t)
 	kv, err := GetOrCreateBucket(t.Context(), js, "bounded-keys", 1)