coalesce resigned wakeups

A follower receiving `resigned` notifications could block the shared
notifier delivery goroutine if its wakeup channel was already full.
That stalls notification delivery beyond leader election, because the
notifier invokes callbacks synchronously on one goroutine.

Treat follower resignation wakeups the same way as forced-resign
wakeups: coalesced, buffered signals instead of preserved blocking
sends. The elector still retries promptly on the pending wakeup, and
its normal election timer remains the backstop if multiple resignations
collapse together.

Regression coverage now asserts the second `resigned` notification is
coalesced instead of blocking, and that a full follower wakeup channel
does not stall notification handling.

Author: bgentry

internal/leadership/elector.go

@@ -305,10 +305,7 @@ func (e *Elector) handleLeadershipNotification(ctx context.Context, topic notifi
 			return
 		}
 
-		select {
-		case <-ctx.Done():
-		case e.leaderResignedChan <- struct{}{}:
-		}
+		trySendWakeup(ctx, e.leaderResignedChan)
 	}
 }
 

internal/leadership/elector_test.go

@@ -395,19 +395,24 @@ func testElector[TElectorBundle any](
 		require.Equal(t, elector2.config.ClientID, leader2.LeaderID)
 	})
 
-	t.Run("RetriesOncePerResignation", func(t *testing.T) {
+	t.Run("CoalescesResignedWakeups", func(t *testing.T) {
 		t.Parallel()
 
 		elector, _ := setup(t, nil)
-		elector.config.ElectInterval = time.Hour
+		elector.config.ElectInterval = 100 * time.Millisecond
+		elector.config.ElectIntervalJitter = 10 * time.Millisecond
 		elector.leaderResignedChan = make(chan struct{}, 1)
 
-		var attempt int
+		var (
+			attempt      int
+			attemptTimes []time.Time
+		)
 
 		elector.exec = &leaderAttemptScriptExecutorMock{
 			Executor: elector.exec,
 			LeaderAttemptElectFunc: func(ctx context.Context, params *riverdriver.LeaderElectParams) (bool, error) {
 				attempt++
+				attemptTimes = append(attemptTimes, time.Now())
 
 				switch attempt {
 				case 1, 2:
@@ -431,22 +436,20 @@ func testElector[TElectorBundle any](
 
 		select {
 		case <-secondNotificationDone:
-			require.Fail(t, "expected second resignation notification to wait for wakeup channel capacity")
 		case <-time.After(50 * time.Millisecond):
+			require.Fail(t, "expected second resignation notification to be coalesced instead of blocking")
 		}
 
-		attemptCtx, cancel := context.WithTimeout(ctx, 500*time.Millisecond)
+		attemptCtx, cancel := context.WithTimeout(ctx, time.Second)
 		defer cancel()
 
+		start := time.Now()
 		require.NoError(t, elector.attemptGainLeadershipLoop(attemptCtx))
-
-		select {
-		case <-secondNotificationDone:
-		case <-time.After(100 * time.Millisecond):
-			require.Fail(t, "expected second resignation notification to complete after wakeup channel was drained")
-		}
+		require.GreaterOrEqual(t, time.Since(start), 75*time.Millisecond)
 
 		require.Equal(t, 3, attempt)
+		require.Len(t, attemptTimes, 3)
+		require.GreaterOrEqual(t, attemptTimes[2].Sub(attemptTimes[1]), 75*time.Millisecond)
 	})
 
 	t.Run("StartStopStress", func(t *testing.T) {
@@ -731,36 +734,25 @@ func TestElectorHandleLeadershipNotification(t *testing.T) {
 		require.Empty(t, elector.leaderResignedChan)
 	})
 
-	t.Run("SignalsLeadershipChangeWaitsForWakeupCapacity", func(t *testing.T) {
+	t.Run("SignalsLeadershipChangeDoesNotBlockOnFullWakeup", func(t *testing.T) {
 		t.Parallel()
 
 		elector, _ := setup(t)
 		elector.leaderResignedChan <- struct{}{}
 
-		ctx, cancel := context.WithCancel(context.Background())
-		defer cancel()
-
 		done := make(chan struct{})
 
 		go func() {
 			defer close(done)
-			elector.handleLeadershipNotification(ctx, notifier.NotificationTopicLeadership, string(mustMarshalJSON(t, validLeadershipChange())))
+			elector.handleLeadershipNotification(context.Background(), notifier.NotificationTopicLeadership, string(mustMarshalJSON(t, validLeadershipChange())))
 		}()
 
-		select {
-		case <-done:
-			require.Fail(t, "expected leadership notification to wait for wakeup channel capacity")
-		case <-time.After(50 * time.Millisecond):
-		}
-
-		<-elector.leaderResignedChan
-
 		select {
 		case <-done:
 		case <-time.After(100 * time.Millisecond):
-			cancel()
-			<-done
-			require.Fail(t, "expected leadership notification to proceed after wakeup channel was drained")
+			require.Fail(t, "expected leadership notification to coalesce the wakeup instead of blocking")
 		}
+
+		require.Len(t, elector.leaderResignedChan, 1)
 	})
 }