diff --git a/spinifex/handlers/rds/backup_test.go b/spinifex/handlers/rds/backup_test.go index f3520242a..eeb401c7a 100644 --- a/spinifex/handlers/rds/backup_test.go +++ b/spinifex/handlers/rds/backup_test.go @@ -815,7 +815,7 @@ func TestReconciler_ClusterLeaseFollowsLeadership(t *testing.T) { require.NotNil(t, release) assert.False(t, ok, "a node that does not hold the lease does not sweep") - h.rec.evaluateLeadership(t.Context()) + require.True(t, h.rec.lease.TryAcquire(t.Context())) release, ok = h.rec.AcquireClusterLease() require.True(t, ok) diff --git a/spinifex/handlers/rds/reconciler.go b/spinifex/handlers/rds/reconciler.go index 3f0669c46..e5a6d57a3 100644 --- a/spinifex/handlers/rds/reconciler.go +++ b/spinifex/handlers/rds/reconciler.go @@ -11,6 +11,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/kvlease" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go/jetstream" ) @@ -76,8 +77,8 @@ type Reconciler struct { svc *Service holder string - mu sync.Mutex - leader bool + lease *kvlease.Lease + leaseErr error // The payloads already reported as stuck pending. The condition persists // until an operator acts, so without this the event would be re-recorded @@ -94,17 +95,31 @@ type Reconciler struct { // holder identifies this daemon in the lease. func NewReconciler(svc *Service, holder string) *Reconciler { - return &Reconciler{svc: svc, holder: holder, reportedPending: make(map[string]string)} + r := &Reconciler{svc: svc, holder: holder, reportedPending: make(map[string]string)} + r.lease, r.leaseErr = kvlease.New(kvlease.Config{ + Name: "rds/reconciler", + Bucket: r.leaderBucket, + Key: reconcilerLeaderKey, + Holder: holder, + TTL: KVBucketRDSLeaderTTL, + Renew: leaseRefresh, + Retry: leaseRefresh, + }) + return r } // Drives the leadership and reconcile loop until ctx is cancelled. Intended as // a daemon-boot goroutine; panics are the caller's recover concern. func (r *Reconciler) Run(ctx context.Context) { + if r.leaseErr != nil { + slog.ErrorContext(ctx, "rds reconciler: lease config invalid", "holder", r.holder, "err", r.leaseErr) + return + } leadershipCtx, cancelLeadership := context.WithCancel(ctx) leadershipDone := make(chan struct{}) go func() { defer close(leadershipDone) - r.maintainLeadership(leadershipCtx, leaseRefresh) + r.lease.Run(leadershipCtx) }() defer func() { cancelLeadership() @@ -117,9 +132,6 @@ func (r *Reconciler) Run(ctx context.Context) { for { select { case <-ctx.Done(): - cancelLeadership() - <-leadershipDone - r.relinquish() return case <-reconcileTicker.C: if !r.isLeader() { @@ -132,23 +144,6 @@ func (r *Reconciler) Run(ctx context.Context) { } } -// Leadership refresh runs independently because a fleet pass can contain -// blocking network I/O and must not outlive its lease. -func (r *Reconciler) maintainLeadership(ctx context.Context, refresh time.Duration) { - leaseTicker := time.NewTicker(refresh) - defer leaseTicker.Stop() - - r.evaluateLeadership(ctx) - for { - select { - case <-ctx.Done(): - return - case <-leaseTicker.C: - r.evaluateLeadership(ctx) - } - } -} - // The shared GC backstop's cluster-wide gate. The reconciler's lease is already // cluster-singular and held continuously rather than claimed per sweep, so // holding it is the whole answer and there is nothing for the caller to release. @@ -157,63 +152,7 @@ func (r *Reconciler) AcquireClusterLease() (func(), bool) { } func (r *Reconciler) isLeader() bool { - r.mu.Lock() - defer r.mu.Unlock() - return r.leader -} - -func (r *Reconciler) evaluateLeadership(ctx context.Context) { - won := r.acquireOrRefresh(ctx) - r.mu.Lock() - was := r.leader - r.leader = won - r.mu.Unlock() - - switch { - case won && !was: - slog.Info("rds reconciler: elected leader", "holder", r.holder) - case !won && was: - slog.Info("rds reconciler: lost leadership", "holder", r.holder) - } -} - -// Claims the lease, or refreshes it (resetting the TTL) when we already hold it. -func (r *Reconciler) acquireOrRefresh(ctx context.Context) bool { - kv, err := r.leaderBucket(ctx) - if err != nil { - return false - } - if _, err := kv.Create(ctx, reconcilerLeaderKey, []byte(r.holder)); err == nil { - return true - } - entry, err := kv.Get(ctx, reconcilerLeaderKey) - if err != nil { - return false - } - if string(entry.Value()) != r.holder { - return false - } - if _, err := kv.Put(ctx, reconcilerLeaderKey, []byte(r.holder)); err != nil { - return false - } - return true -} - -// Releases the lease on shutdown so the next leader is elected immediately -// rather than after the TTL. -func (r *Reconciler) relinquish() { - // Run's ctx is already cancelled by the time this is called, so the release - // runs on its own — a cancelled ctx would fail the delete. - ctx := context.Background() - kv, err := r.leaderBucket(ctx) - if err != nil { - return - } - if entry, gerr := kv.Get(ctx, reconcilerLeaderKey); gerr == nil && string(entry.Value()) == r.holder { - if err := kv.Delete(ctx, reconcilerLeaderKey); err != nil { - slog.Debug("rds reconciler: release lease failed", "holder", r.holder, "err", err) - } - } + return r.lease.Held() } func (r *Reconciler) leaderBucket(ctx context.Context) (jetstream.KeyValue, error) { diff --git a/spinifex/handlers/rds/reconciler_test.go b/spinifex/handlers/rds/reconciler_test.go index 48b0c49e1..c3b3436a3 100644 --- a/spinifex/handlers/rds/reconciler_test.go +++ b/spinifex/handlers/rds/reconciler_test.go @@ -368,47 +368,34 @@ func TestReconciler_ElectsASingleLeader(t *testing.T) { h := newReconcileHarness(t) other := NewReconciler(h.svc, "node-b") - assert.True(t, h.rec.acquireOrRefresh(t.Context())) - assert.False(t, other.acquireOrRefresh(t.Context())) + assert.True(t, h.rec.lease.TryAcquire(t.Context())) + assert.False(t, other.lease.TryAcquire(t.Context())) - // The holder refreshes its own lease rather than losing it to itself. - assert.True(t, h.rec.acquireOrRefresh(t.Context())) + // The holder reports the leasership it already holds, and does not hand it away. + assert.True(t, h.rec.lease.TryAcquire(t.Context())) // Releasing on shutdown hands over immediately instead of after the TTL. - h.rec.relinquish() - assert.True(t, other.acquireOrRefresh(t.Context())) - assert.False(t, h.rec.acquireOrRefresh(t.Context())) + h.rec.lease.Release(t.Context()) + assert.True(t, other.lease.TryAcquire(t.Context())) + assert.False(t, h.rec.lease.TryAcquire(t.Context())) } -func TestReconciler_RefreshesLeadershipOnAnIndependentLoop(t *testing.T) { +func TestReconciler_RunHoldsLeadershipIndependentlyOfTheReconcileLoop(t *testing.T) { t.Parallel() h := newReconcileHarness(t) ctx, cancel := context.WithCancel(t.Context()) done := make(chan struct{}) - go func() { - defer close(done) - h.rec.maintainLeadership(ctx, 5*time.Millisecond) - }() - t.Cleanup(func() { - cancel() - select { - case <-done: - case <-time.After(time.Second): - t.Error("leadership refresh did not stop after cancellation") - } - }) + go func() { defer close(done); h.rec.Run(ctx) }() - require.Eventually(t, h.rec.isLeader, time.Second, 5*time.Millisecond) - kv, err := h.rec.leaderBucket(t.Context()) - require.NoError(t, err) - entry, err := kv.Get(t.Context(), reconcilerLeaderKey) - require.NoError(t, err) - firstRevision := entry.Revision() + require.Eventually(t, h.rec.isLeader, 2*time.Second, 10*time.Millisecond, + "Run must elect without waiting for a reconcile tick") - require.Eventually(t, func() bool { - entry, err := kv.Get(t.Context(), reconcilerLeaderKey) - return err == nil && entry.Revision() > firstRevision - }, time.Second, 5*time.Millisecond, "the lease revision must advance without the reconcile loop driving it") + cancel() + <-done + + other := NewReconciler(h.svc, "node-b") + assert.True(t, other.lease.TryAcquire(t.Context()), + "Run returned with the lease key still present") } // A node that never won the lease must not delete the holder's key on shutdown. @@ -417,8 +404,8 @@ func TestReconciler_RelinquishOnlyReleasesItsOwnLease(t *testing.T) { h := newReconcileHarness(t) other := NewReconciler(h.svc, "node-b") - require.True(t, h.rec.acquireOrRefresh(t.Context())) - other.relinquish() + require.True(t, h.rec.lease.TryAcquire(t.Context())) + other.lease.Release(t.Context()) - assert.False(t, other.acquireOrRefresh(t.Context()), "the original holder still owns the lease") + assert.False(t, other.lease.TryAcquire(t.Context()), "the original holder still owns the lease") }