fix(lock): preserve bounded ranges for large DML batches

pkg/lockservice/cfg.go

@@ -71,10 +71,13 @@ type Config struct {
 	// execution path forgets to propagate a session or task deadline. Callers
 	// that retry across Lock calls still need to own and propagate a deadline.
 	MaxLockWaitDuration toml.Duration `toml:"max-lock-wait-duration"`
-	// MaxLockRowCount each time a lock is added, some LockRow is stored in the lockservice, if
-	// too many LockRows are put in each time, it will cause too much memory overhead, this value
-	// limits the maximum count of LocRow put into the LockService each time, beyond this value it
-	// will be converted into a Range of locks
+	// MaxLockRowCount bounds lock keys retained for one transaction and physical lock table
+	// only while its complete ownership consists of non-sharded Exclusive locks, which can be
+	// conservatively coarsened to their observed range. Once the table records a Shared or
+	// row-sharded lock, it stays exact for the rest of the transaction: an overlapping range
+	// cannot preserve independent compatible ownership, and sharded endpoints can belong to
+	// different physical tables. The planner upgrades cardinality-known Shared targets before
+	// acquisition instead.
 	MaxLockRowCount toml.ByteSize `toml:"max-row-lock-count"`
 	// KeepBindTimeout when a locktable is assigned to a lockservice, the lockservice will
 	// continuously hold the bind, and if no hold request is received after the configured time,
@@ -102,6 +105,9 @@ func (c *Config) Validate() {
 	if c.MaxFixedSliceSize == 0 {
 		c.MaxFixedSliceSize = toml.ByteSize(defaultMaxFixedSliceSize)
 	}
+	// Preserve the compatibility contract of existing deployments. Remote
+	// cleanup is routed by table and transaction ID, so cumulative coarsening
+	// must not require extra endpoint capacity in the origin-side key snapshot.
 	if c.MaxLockRowCount > c.MaxFixedSliceSize {
 		panic("This parameter configuration may trigger scenarios that violate MaxFixedSliceSize")
 	}

pkg/lockservice/cfg_test.go

@@ -51,3 +51,20 @@ func TestAdjustConfigRejectsNegativeMaxLockWaitDuration(t *testing.T) {
 	c.MaxLockWaitDuration.Duration = -1
 	assert.Panics(t, c.Validate)
 }
+
+func TestAdjustConfigPreservesFixedSliceCompatibility(t *testing.T) {
+	// These tight settings were accepted before cumulative coarsening existed
+	// and must remain bootable across an upgrade.
+	for _, c := range []Config{
+		{ServiceID: "s1", MaxLockRowCount: 1, MaxFixedSliceSize: 1},
+		{ServiceID: "s1", MaxLockRowCount: 2, MaxFixedSliceSize: 2},
+		{ServiceID: "s1", MaxLockRowCount: 3, MaxFixedSliceSize: 3},
+		{ServiceID: "s1", MaxLockRowCount: 3, MaxFixedSliceSize: 4},
+		{ServiceID: "s1", MaxLockRowCount: 4, MaxFixedSliceSize: 4},
+	} {
+		assert.NotPanics(t, c.Validate)
+	}
+
+	c := Config{ServiceID: "s1", MaxLockRowCount: 5, MaxFixedSliceSize: 4}
+	assert.Panics(t, c.Validate)
+}

pkg/lockservice/deadlock.go

@@ -158,8 +158,16 @@ func (d *detector) doCheck(ctx context.Context) {
 				if err == nil {
 					err = ErrDeadLockDetected
 				}
-				d.ignoreTxns.Store(string(deadlockTxn.TxnID), struct{}{})
-				d.waitTxnAbortFunc(deadlockTxn, err)
+				// Different detector workers can discover the same cycle from
+				// different roots. All traversals choose the same deterministic
+				// victim; LoadOrStore is the linearization point that gives exactly
+				// one worker ownership of the abort notification.
+				if _, loaded := d.ignoreTxns.LoadOrStore(
+					string(deadlockTxn.TxnID),
+					struct{}{},
+				); !loaded {
+					d.waitTxnAbortFunc(deadlockTxn, err)
+				}
 			}
 			d.mu.Lock()
 			delete(d.mu.activeCheckTxn, util.UnsafeBytesToString(txn.waitTxn.TxnID))
@@ -200,8 +208,9 @@ func (d *detector) checkDeadlock(ctx context.Context, w *waiters) (bool, pb.Wait
 
 func (d *detector) deadlockFound(w *waiters) (bool, pb.WaitTxn, error) {
 	node := w.deadlockNode()
-	logDeadLockFound(d.logger, node.txn, printPathFromRoot(node))
-	return true, node.txn, nil
+	victim := w.deadlockVictim()
+	logDeadLockFound(d.logger, victim, printPathFromRoot(node))
+	return true, victim, nil
 }
 
 type txnVisitState uint8
@@ -277,6 +286,21 @@ func (w *waiters) deadlockNode() *lockNode {
 	return w.deadlock
 }
 
+// deadlockVictim returns a root-independent victim for the detected cycle.
+// Transaction IDs are opaque, so lexical order is used only as a stable total
+// order. Walking from the closing node upward also preserves the populated
+// WaiterAddress on the duplicate closing occurrence when the cycle includes
+// the detector root.
+func (w *waiters) deadlockVictim() pb.WaitTxn {
+	var victim pb.WaitTxn
+	for node := w.deadlock; node != nil; node = node.parent {
+		if len(victim.TxnID) == 0 || bytes.Compare(node.txn.TxnID, victim.TxnID) > 0 {
+			victim = node.txn
+		}
+	}
+	return victim
+}
+
 func (w *waiters) setDeadlock(closing *lockNode) {
 	for i, node := range w.stack {
 		if !bytes.Equal(node.txn.TxnID, closing.txn.TxnID) {

pkg/lockservice/deadlock_test.go

@@ -59,13 +59,13 @@ func TestCheckWithDeadlock(t *testing.T) {
 
 		// txn1 - txn2 - txn3 - txn1
 		assert.NoError(t, d.check(txn4, pb.WaitTxn{TxnID: txn1}))
-		assert.Equal(t, txn1, <-abortC)
-		d.txnClosed(txn1)
+		assert.Equal(t, txn3, <-abortC)
+		d.txnClosed(txn3)
 
 		// txn2 - txn3 - txn1 - txn2
 		assert.NoError(t, d.check(nil, pb.WaitTxn{TxnID: txn2}))
-		assert.Equal(t, txn2, <-abortC)
-		d.txnClosed(txn2)
+		assert.Equal(t, txn3, <-abortC)
+		d.txnClosed(txn3)
 
 		// txn3 - txn1 - txn2 - txn3
 		assert.NoError(t, d.check(nil, pb.WaitTxn{TxnID: txn3}))
@@ -129,16 +129,18 @@ func TestCheckWithAcyclicBranchReconvergence(t *testing.T) {
 
 func TestCheckWithCrossBranchDeadlock(t *testing.T) {
 	reuse.RunReuseTests(func() {
-		root := []byte("root")
-		seed := []byte("seed")
+		// Prefix nodes deliberately sort after every cycle member. Victim
+		// selection must consider the cycle only, not an acyclic path into it.
+		root := []byte("zz-root")
+		seed := []byte("zz-seed")
 		a := []byte("a")
 		b := []byte("b")
 		x := []byte("x")
 		y := []byte("y")
 		depends := map[string][]pb.WaitTxn{
 			string(seed): {{TxnID: a}, {TxnID: b}},
 			string(a):    {{TxnID: x}},
-			string(b):    {{TxnID: y}},
+			string(b):    {{TxnID: y, WaiterAddress: "y-service"}},
 			string(x):    {{TxnID: b}},
 			string(y):    {{TxnID: x, WaiterAddress: "closing-service"}},
 		}
@@ -168,8 +170,8 @@ func TestCheckWithCrossBranchDeadlock(t *testing.T) {
 		hasDeadlock, deadlockTxn, err := d.checkDeadlock(context.Background(), w)
 		require.NoError(t, err)
 		require.True(t, hasDeadlock)
-		require.Equal(t, x, deadlockTxn.TxnID)
-		require.Equal(t, "closing-service", deadlockTxn.WaiterAddress)
+		require.Equal(t, y, deadlockTxn.TxnID)
+		require.Equal(t, "y-service", deadlockTxn.WaiterAddress)
 		require.Equal(t, "78 <= 62 <= 79 <= 78", printPathFromRoot(w.deadlockNode()))
 		require.Equal(t, 1, fetchCount[string(seed)])
 		require.Equal(t, 1, fetchCount[string(a)])
@@ -422,25 +424,23 @@ func TestCheckWithComplexDeadlock(t *testing.T) {
 			})
 		defer d.close()
 
-		// Test case 1: Start with txn1, should detect deadlock and abort txn1
+		// Every traversal of the same cycle selects the same victim.
 		assert.NoError(t, d.check([]byte("txn0"), pb.WaitTxn{TxnID: txn1}))
-		assert.Equal(t, txn1, <-abortC)
-		d.txnClosed(txn1)
+		assert.Equal(t, txn9, <-abortC)
+		d.txnClosed(txn9)
 
-		// Test case 2: Start with txn5, should detect deadlock and abort txn5
 		assert.NoError(t, d.check([]byte("txn0"), pb.WaitTxn{TxnID: txn5}))
-		assert.Equal(t, txn5, <-abortC)
-		d.txnClosed(txn5)
+		assert.Equal(t, txn9, <-abortC)
+		d.txnClosed(txn9)
 
-		// Test case 3: Start with txn10, should detect deadlock and abort txn10
 		assert.NoError(t, d.check([]byte("txn0"), pb.WaitTxn{TxnID: txn10}))
-		assert.Equal(t, txn10, <-abortC)
-		d.txnClosed(txn10)
+		assert.Equal(t, txn9, <-abortC)
+		d.txnClosed(txn9)
 
-		// Test case 3: Start with txn11, should detect deadlock and abort txn11
+		// txn11 reaches the same cycle but is not itself a cycle member.
 		assert.NoError(t, d.check([]byte("txn0"), pb.WaitTxn{TxnID: txn11}))
-		assert.Equal(t, txn1, <-abortC)
-		d.txnClosed(txn1)
+		assert.Equal(t, txn9, <-abortC)
+		d.txnClosed(txn9)
 
 		// Test case 5: Break the cycle by removing txn10's dependency on txn1
 		depends[string(txn10)] = []pb.WaitTxn{} // Remove the dependency that creates the cycle
@@ -503,9 +503,9 @@ func TestCheckDeadlock(t *testing.T) {
 			})
 		defer d.close()
 
-		// Test case 1: Start with txn1, should detect deadlock and abort txn1
+		// The cycle is t2..t10; lexical ordering selects t9 deterministically.
 		assert.NoError(t, d.check([]byte("txn0"), pb.WaitTxn{TxnID: txn1}))
-		assert.Equal(t, txn2, <-abortC)
-		d.txnClosed(txn2)
+		assert.Equal(t, txn9, <-abortC)
+		d.txnClosed(txn9)
 	})
 }

pkg/lockservice/lock.go

@@ -220,6 +220,12 @@ func (l Lock) closeTxn(
 
 	// has another holders
 	if l.holders.size() > 0 {
+		if l.isShared() {
+			// A range-merge waiter can itself be one of the remaining
+			// compatible Shared holders. Wake only those waiters so they can
+			// retry when the ownership shape becomes collapsible.
+			l.waiters.notifySharedHolderChange(notify)
+		}
 		return false
 	}
 

pkg/lockservice/lock_table_keeper_test.go

@@ -1130,7 +1130,7 @@ func TestKeepRemoteLockBindChangedFencesActiveTxn(t *testing.T) {
 			txnID := []byte("txn1")
 			txn := svc.activeTxnHolder.getActiveTxn(txnID, true, "")
 			txn.Lock()
-			require.NoError(t, txn.lockAdded(oldBind.Group, oldBind, [][]byte{{1}}, logger))
+			require.NoError(t, txn.lockAdded(oldBind.Group, oldBind, [][]byte{{1}}, pb.LockOptions{}, logger))
 			txn.Unlock()
 
 			keeper := &lockTableKeeper{
@@ -1254,7 +1254,7 @@ func TestKeepRemoteLockBindChangedRefreshFailureInvalidatesOldBind(t *testing.T)
 					addTxn := func(txnID []byte, bind pb.LockTable) *activeTxn {
 						txn := svc.activeTxnHolder.getActiveTxn(txnID, true, "")
 						txn.Lock()
-						require.NoError(t, txn.lockAdded(bind.Group, bind, [][]byte{{1}}, logger))
+						require.NoError(t, txn.lockAdded(bind.Group, bind, [][]byte{{1}}, pb.LockOptions{}, logger))
 						txn.Unlock()
 						return txn
 					}
@@ -1361,7 +1361,7 @@ func TestKeepRemoteLockFailureFetchesNewBindAndFencesActiveTxn(t *testing.T) {
 			txnID := []byte("txn1")
 			txn := svc.activeTxnHolder.getActiveTxn(txnID, true, "")
 			txn.Lock()
-			require.NoError(t, txn.lockAdded(oldBind.Group, oldBind, [][]byte{{1}}, logger))
+			require.NoError(t, txn.lockAdded(oldBind.Group, oldBind, [][]byte{{1}}, pb.LockOptions{}, logger))
 			txn.Unlock()
 
 			keeper := &lockTableKeeper{
@@ -1619,7 +1619,7 @@ func TestKeepRemoteLockIgnoresNonBindResponseErrors(t *testing.T) {
 			txnID := []byte("txn1")
 			txn := svc.activeTxnHolder.getActiveTxn(txnID, true, "")
 			txn.Lock()
-			require.NoError(t, txn.lockAdded(oldBind.Group, oldBind, [][]byte{{1}}, logger))
+			require.NoError(t, txn.lockAdded(oldBind.Group, oldBind, [][]byte{{1}}, pb.LockOptions{}, logger))
 			txn.Unlock()
 
 			keeper := &lockTableKeeper{