diff --git a/spinifex/handlers/dns/reconcile.go b/spinifex/handlers/dns/reconcile.go index 9845847b4..837107c94 100644 --- a/spinifex/handlers/dns/reconcile.go +++ b/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 diff --git a/spinifex/handlers/dns/reconcile_test.go b/spinifex/handlers/dns/reconcile_test.go index b457046c1..a4a067c6f 100644 --- a/spinifex/handlers/dns/reconcile_test.go +++ b/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. diff --git a/spinifex/kvutil/kvutil.go b/spinifex/kvutil/kvutil.go index fd5242b08..c4d93107d 100644 --- a/spinifex/kvutil/kvutil.go +++ b/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. diff --git a/spinifex/kvutil/kvutil_test.go b/spinifex/kvutil/kvutil_test.go index 02a0e1fee..a15f9ccab 100644 --- a/spinifex/kvutil/kvutil_test.go +++ b/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)