refactor: convert the RDS reconciler to the shared kvlease package (#870)

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)
 

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) {

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")
 }