kvserver: span config queue notification can be lost during lease transfer

## Summary

A span-config-triggered queue notification can be lost when the affected range transfers its lease concurrently. The split (and potentially other queue-driven policy work) is then delayed until the periodic range scan, normally 10 minutes.

Discovered while investigating #173682. This is not caused by range coalescing.

Related umbrella issues: #124426 and #127180. This issue tracks the specific lease-handoff mechanism and its regression test/fix.

## Failure and root cause

In the failing run (`8499e276da6`, system tenant), table 106 required a boundary at `/Table/106`, but remained in `r87: /Table/84-Max` for the test's 45-second deadline.

The relevant events were nearly simultaneous:

```
06:05:29.893808 r87:/Table/84-Max transferring lease n1 -> n3
06:05:29.894329 span-config subscriber begins applying the update on n1/n3
```

`Store.onSpanConfigUpdate` offers each affected replica to `Replica.MaybeQueue` once. `baseQueue.maybeAdd` calls `replicaCanBeProcessed(..., acquireLeaseIfNeeded=false)`, which skips a replica when its valid lease is owned by another store. During a handoff, the future holder can process the notification before acquiring the lease, while the old holder processes it after relinquishing the lease; all replicas can therefore skip it.

Lease acquisition does not re-enqueue the split queue. The periodic scanner is the eventual backstop.

Code paths:

- [`Store.onSpanConfigUpdate`](https://github.com/cockroachdb/cockroach/blob/8499e276da6fc483c164d1c99fc5f0c33a0e995f/pkg/kv/kvserver/store.go#L2903-L2976)
- [`Replica.MaybeQueue`](https://github.com/cockroachdb/cockroach/blob/8499e276da6fc483c164d1c99fc5f0c33a0e995f/pkg/kv/kvserver/replica.go#L1314-L1338)
- [lease-gated `maybeAdd`](https://github.com/cockroachdb/cockroach/blob/8499e276da6fc483c164d1c99fc5f0c33a0e995f/pkg/kv/kvserver/queue.go#L1326-L1357)

## Deterministic verification

At the failure SHA and seed, a temporary hook immediately before `Replica.MaybeQueue` was used to:

1. Hold all three stores' callbacks for the `/Table/106` update.
2. Release the two non-leaseholders while n1 held the lease.
3. Transfer the lease n1 -> n3.
4. Release n1's callback after the transfer.

Result:

```
lost notification reproduced: required=/Table/106 actual-range=/{Table/84-Max}
```

`ComputeSplitKey` still returned `/Table/106`, proving the boundary was required. A forced split-queue scan immediately created the boundary. The instrumented test passed in 23.5s.

## Possible fix

Make span-config-triggered queueing survive lease handoffs, for example by rechecking the relevant queues on lease acquisition or retrying a notification rejected solely because another replica owns the lease. A regression test should orchestrate the callback ordering above.

No data loss is known. The observed impact is delayed split/policy convergence until the scanner backstop.

Deterministic reproduction patch

This temporary patch was applied at `8499e276da6`. It adds a test-only hook immediately before `Replica.MaybeQueue`, orders the three store callbacks around a lease transfer, verifies that the required split was missed, then verifies that a forced scan performs it.

```diff
diff --git a/pkg/kv/kvserver/testing_knobs.go b/pkg/kv/kvserver/testing_knobs.go
@@
SpanConfigUpdateInterceptor func(spanconfig.Update)
+ // BeforeSpanConfigMaybeQueueInterceptor is called immediately before an
+ // affected replica is offered to queues after a span config update.
+ BeforeSpanConfigMaybeQueueInterceptor func(roachpb.StoreID, roachpb.RangeID, roachpb.Span)

diff --git a/pkg/kv/kvserver/store.go b/pkg/kv/kvserver/store.go
@@
if changed {
+ if fn := s.TestingKnobs().BeforeSpanConfigMaybeQueueInterceptor; fn != nil {
+ fn(s.StoreID(), repl.RangeID, updated)
+ }
repl.MaybeQueue(ctx, now)

diff --git a/pkg/sql/tests/table_split_test.go b/pkg/sql/tests/table_split_test.go
@@
import (
"context"
+ "sync"
+ "sync/atomic"
"testing"
+ "time"

"github.com/cockroachdb/cockroach/pkg/base"
+ "github.com/cockroachdb/cockroach/pkg/keys"
+ "github.com/cockroachdb/cockroach/pkg/kv/kvserver"
+ "github.com/cockroachdb/cockroach/pkg/roachpb"
@@
- tc := testcluster.StartTestCluster(t, 3, base.TestClusterArgs{})
+ const expectedTableID = 106
+ target := keys.SystemSQLCodec.TablePrefix(expectedTableID)
+ arrived := make(chan roachpb.StoreID, 3)
+ release := [4]chan struct{}{nil, make(chan struct{}), make(chan struct{}), make(chan struct{})}
+ var releaseOnce [4]sync.Once
+ releaseStore := func(storeID roachpb.StoreID) {
+ releaseOnce[storeID].Do(func() { close(release[storeID]) })
+ }
+ var armed atomic.Bool
+ var storeOnce [4]sync.Once
+ tc := testcluster.StartTestCluster(t, 3, base.TestClusterArgs{
+ ServerArgs: base.TestServerArgs{
+ DefaultTestTenant: base.TestIsSpecificToStorageLayerAndNeedsASystemTenant,
+ Knobs: base.TestingKnobs{Store: &kvserver.StoreTestingKnobs{
+ BeforeSpanConfigMaybeQueueInterceptor: func(
+ storeID roachpb.StoreID, _ roachpb.RangeID, updated roachpb.Span,
+ ) {
+ if !armed.Load() || !updated.ContainsKey(target) {
+ return
+ }
+ storeOnce[storeID].Do(func() {
+ arrived <- storeID
+ <-release[storeID]
+ })
+ },
+ }},
+ },
+ })
defer tc.Stopper().Stop(context.Background())
+ defer func() {
+ for storeID := roachpb.StoreID(1); storeID <= 3; storeID++ {
+ releaseStore(storeID)
+ }
+ }()
@@
runner := sqlutils.MakeSQLRunner(s.SQLConn(t, serverutils.DBName("system")))
+ runner.Exec(t, `SET CLUSTER SETTING kv.allocator.load_based_lease_rebalancing.enabled = false`)
runner.Exec(t, `CREATE DATABASE test`)
+ desc := tc.LookupRangeOrFatal(t, target)
+ tc.TransferRangeLeaseOrFatal(t, desc, tc.Target(0))
+ armed.Store(true)
runner.Exec(t, `CREATE TABLE test.t (k SERIAL PRIMARY KEY, v INT)`)
@@
runner.QueryRow(t, tableIDQuery, "test", "t").Scan(&tableID)
+ if tableID != expectedTableID {
+ t.Fatalf("expected table ID %d, got %d", expectedTableID, tableID)
+ }
tableStartKey := s.Codec().TablePrefix(tableID)
+
+ seen := map[roachpb.StoreID]bool{}
+ for len(seen) < 3 {
+ select {
+ case storeID := <-arrived:
+ seen[storeID] = true
+ case <-time.After(testutils.DefaultSucceedsSoonDuration):
+ t.Fatalf("timed out waiting for callbacks; saw stores %v", seen)
+ }
+ }
+
+ // Make the two non-leaseholders reject the notification first.
+ releaseStore(2)
+ releaseStore(3)
+ time.Sleep(500 * time.Millisecond)
+
+ // Transfer the lease, then make the old leaseholder reject it too.
+ desc = tc.LookupRangeOrFatal(t, tableStartKey)
+ tc.TransferRangeLeaseOrFatal(t, desc, tc.Target(2))
+ releaseStore(1)
+ time.Sleep(time.Second)
+
+ desc = tc.LookupRangeOrFatal(t, tableStartKey)
+ splitKey, err := tc.GetFirstStoreFromServer(t, 2).GetStoreConfig().
+ SpanConfigSubscriber.ComputeSplitKey(context.Background(), desc.StartKey, desc.EndKey)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !splitKey.Equal(roachpb.RKey(tableStartKey)) {
+ t.Fatalf("expected required split %s, got %s", tableStartKey, splitKey)
+ }
+ if desc.StartKey.Equal(roachpb.RKey(tableStartKey)) {
+ t.Fatal("range unexpectedly split after all notification enqueues missed the leaseholder")
+ }
+ t.Logf("lost notification reproduced: required=%s actual-range=%s", splitKey, desc.RSpan())
+
+ for i := 0; i < tc.NumServers(); i++ {
+ if err := tc.GetFirstStoreFromServer(t, i).ForceSplitScanAndProcess(); err != nil {
+ t.Fatal(err)
+ }
+ }
```

Jira issue: CRDB-67047

Read Store.onSpanConfigUpdate in pkg/kv/kvserver/store.go, then follow Replica.MaybeQueue and the lease-gated maybeAdd path in replica.go and queue.go. Run the affected table split test in pkg/sql/tests/table_split_test.go and use the documented callback ordering as the reproduction. Done means a span-config queue notification is not lost across lease transfer and the regression test confirms the required split occurs without waiting for the periodic scan.

Written by the indexing model from the issue text.