Cover the lock paths the keep-alive added

Patch coverage came in at 54.3% against the 70% gate. Two gaps, both
mine:

Lock had NO test at all before this branch, so every line the keep-alive
added to it was uncovered. It is the one form that cannot abort its
caller when the lock is lost, so what it can be held to is narrower:
that it actually excludes, that release actually frees, and that the
keep-alive interferes with neither. The exclusion test holds the lock
across several keep-alive ticks, so a keep-alive that relinquished on a
healthy connection would show up as the second replica getting in.
0.0% to 92.3%.

And processHost's ErrLockLost branch. Reporting 'did not run' is right
even though the callback did run, because nothing is acked until a batch
completes, so the events stay in flight for redelivery. What must not
happen is the worker treating it as a completed batch and draining on.
77.4% to 96.8%.

Author: getvictor

server/coordination/leader/leader_test.go

@@ -463,3 +463,73 @@ func TestLockSurvivesAnIdleConnectionClose(t *testing.T) {
 		require.ErrorIs(t, err, sentinel, "the callback's diagnosis is more specific than ErrLockLost and must win")
 	})
 }
+
+// TestLockGrantsExclusionAndReleases covers the bare Lock form, which had no test at all before issue #721 added a keep-alive to it.
+//
+// It is the one lock form that cannot abort its caller when the lock is lost, because there is no callback context to cancel, so
+// what it can be held to is narrower: it must actually exclude, its release must actually free the lock, and the keep-alive it now
+// runs must not interfere with either.
+func TestLockGrantsExclusionAndReleases(t *testing.T) {
+	t.Parallel()
+	dbA, dbB := replicaDBs(t)
+	// Keep-alive well inside the test's own lifetime, so the ping fires while the lock is held rather than after it is released.
+	coordA := leader.NewMySQL(dbA, slog.Default(), leader.WithKeepAliveInterval(10*time.Millisecond))
+	coordB := leader.NewMySQL(dbB, slog.Default(), leader.WithKeepAliveInterval(10*time.Millisecond))
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	lockName := uniqueLockName()
+
+	release, err := coordA.Lock(ctx, lockName)
+	require.NoError(t, err)
+
+	// Held long enough for several keep-alive ticks, so a keep-alive that broke the lock (by relinquishing on a healthy
+	// connection, say) would show up as the second replica getting in.
+	time.Sleep(100 * time.Millisecond)
+
+	secondAcquired := make(chan struct{})
+	go func() {
+		r, lerr := coordB.Lock(ctx, lockName)
+		if lerr == nil {
+			r()
+			close(secondAcquired)
+		}
+	}()
+	select {
+	case <-secondAcquired:
+		t.Fatal("a second replica acquired a lock the first still holds")
+	case <-time.After(200 * time.Millisecond):
+	}
+
+	release()
+
+	select {
+	case <-secondAcquired:
+	case <-time.After(10 * time.Second):
+		t.Fatal("release did not free the lock; the second replica never acquired it")
+	}
+}
+
+// TestLockReleaseIsIdempotentUnderCancelledContext pins that a lock taken on a context that is later cancelled still frees on
+// release. The boot sequence is the caller here, and a shutdown racing the migration lock must not leave it held for the next
+// replica to block on: release runs on a cancellation-stripped context precisely so this holds.
+func TestLockReleaseIsIdempotentUnderCancelledContext(t *testing.T) {
+	t.Parallel()
+	db, _ := replicaDBs(t)
+	coord := leader.NewMySQL(db, slog.Default(), leader.WithKeepAliveInterval(10*time.Millisecond))
+	lockName := uniqueLockName()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	release, err := coord.Lock(ctx, lockName)
+	require.NoError(t, err)
+
+	cancel() // the shutdown arrives while the lock is held
+	release()
+
+	// The lock must be free: a fresh acquire on an uncancelled context succeeds immediately.
+	freshCtx, freshCancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer freshCancel()
+	again, err := coord.Lock(freshCtx, lockName)
+	require.NoError(t, err, "a lock released after its context was cancelled must still be free")
+	again()
+}

server/detection/internal/pipeline/processor_hostclaim_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"strings"
+	"sync/atomic"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -225,3 +226,44 @@ func TestHostCandidatesWindow(t *testing.T) {
 		})
 	}
 }
+
+// lostLockCoordinator reports that it acquired the host lock, ran the callback, and then discovered the lock had gone: the shape
+// leader.DoOnceIfLeader returns after issue #721 when the connection holding the lock dies mid-callback.
+type lostLockCoordinator struct {
+	stubCoordinator
+	ran atomic.Bool
+}
+
+func (c *lostLockCoordinator) DoOnceIfLeader(ctx context.Context, _ string, fn func(context.Context) error) (bool, error) {
+	c.ran.Store(true)
+	// The callback DOES run, which is the whole difficulty: work may have been partially done before the lock went.
+	if err := fn(ctx); err != nil {
+		return true, err
+	}
+	return true, leader.ErrLockLost
+}
+
+// TestProcessHostTreatsALostLockAsNotRun pins how the processor handles a host whose claim lock went while it was working
+// (issue #721's ErrLockLost, reaching the processor through the per-host claim added in #717).
+//
+// Reporting "did not run" is right even though the callback did run: nothing is acknowledged until a batch completes, so the events
+// stay in flight and are redelivered when the claim lease expires. What must not happen is the worker treating it as a completed
+// batch and draining on, which would leave a host's events folded by two claimers.
+func TestProcessHostTreatsALostLockAsNotRun(t *testing.T) {
+	t.Parallel()
+	coord := &lostLockCoordinator{}
+	proc, err := NewProcessor(&scriptedEventLog{}, nil, nil, ProcessorOptions{
+		Logger:      discardLogger(),
+		Batch:       10,
+		Concurrency: 1,
+		Coordinator: coord,
+		ConnBudget:  25,
+	})
+	require.NoError(t, err)
+
+	claimed, ran := proc.processHost(context.Background(), "host-a")
+
+	assert.True(t, coord.ran.Load(), "the coordinator did hand the callback the lock before losing it")
+	assert.False(t, ran, "a lost lock must not read as a completed batch; the worker has to move to another host")
+	assert.Zero(t, claimed, "and it must not report progress it cannot vouch for")
+}