From c0f99e535c86d400d9371909ffc653d797cc69e3 Mon Sep 17 00:00:00 2001 From: GreatRiver Date: Sat, 18 Jul 2026 15:49:14 +0800 Subject: [PATCH 1/4] test(lockservice): bound remote waiter snapshot setup --- pkg/lockservice/service_remote_test.go | 60 +++++++++++++++++++++++--- 1 file changed, 55 insertions(+), 5 deletions(-) diff --git a/pkg/lockservice/service_remote_test.go b/pkg/lockservice/service_remote_test.go index d10f617082e3a..10eef14362e83 100644 --- a/pkg/lockservice/service_remote_test.go +++ b/pkg/lockservice/service_remote_test.go @@ -87,10 +87,18 @@ func TestFetchWhoWaitingMeUsesActiveRemoteWaiterSnapshots(t *testing.T) { require.NoError(t, owner.Unlock(ctx, seedTxn, timestamp.Timestamp{})) mustAddTestLock(t, ctx, holderService, tableID, holderTxn, [][]byte{row}, pb.Granularity_Row) + // The waiter must remain live until the holder is explicitly released. + // Reusing ctx here couples queue admission to all setup RPCs above: once + // that deadline expires, Lock returns before it can enter owner's queue + // and the old unbounded waitWaiters call spins forever. + waiterCtx, cancelWaiter := context.WithCancel(context.Background()) + defer cancelWaiter() waitResult := make(chan error, 1) + holderReleased := false + waiterDone := false go func() { _, err := waiterService.Lock( - ctx, + waiterCtx, tableID, [][]byte{row}, activeWaiterTxn, @@ -99,12 +107,41 @@ func TestFetchWhoWaitingMeUsesActiveRemoteWaiterSnapshots(t *testing.T) { waitResult <- err }() defer func() { - require.NoError(t, holderService.Unlock(ctx, holderTxn, timestamp.Timestamp{})) - require.NoError(t, <-waitResult) - require.NoError(t, waiterService.Unlock(ctx, activeWaiterTxn, timestamp.Timestamp{})) + cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cleanupCancel() + if !holderReleased { + _ = holderService.Unlock(cleanupCtx, holderTxn, timestamp.Timestamp{}) + } + cancelWaiter() + if !waiterDone { + select { + case err := <-waitResult: + if err == nil { + _ = waiterService.Unlock(cleanupCtx, activeWaiterTxn, timestamp.Timestamp{}) + } + case <-cleanupCtx.Done(): + t.Errorf("remote waiter did not exit during cleanup: %v", cleanupCtx.Err()) + } + } }() - waitWaiters(t, owner, tableID, row, 1) + // This is test admission synchronization, not part of the behavior under + // test. Keep it bounded so an unexpected routing failure produces a + // useful failure instead of consuming the package's 40-minute timeout. + require.Eventually(t, func() bool { + lt, err := owner.getLockTable(0, tableID) + if err != nil { + return false + } + local, ok := lt.(*localLockTable) + if !ok { + return false + } + local.mu.Lock() + defer local.mu.Unlock() + lock, ok := local.mu.store.Get(row) + return ok && lock.waiters.size() == 1 + }, 10*time.Second, 10*time.Millisecond, "active remote waiter did not reach owner queue") lt, err := owner.getLockTable(0, tableID) require.NoError(t, err) local := lt.(*localLockTable) @@ -151,6 +188,19 @@ func TestFetchWhoWaitingMeUsesActiveRemoteWaiterSnapshots(t *testing.T) { holderService.getLockTable, )) require.Equal(t, [][]byte{activeWaiterTxn}, waitingTxnIDs) + + releaseCtx, releaseCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer releaseCancel() + require.NoError(t, holderService.Unlock(releaseCtx, holderTxn, timestamp.Timestamp{})) + holderReleased = true + select { + case err := <-waitResult: + waiterDone = true + require.NoError(t, err) + case <-releaseCtx.Done(): + t.Fatalf("remote waiter did not acquire after holder release: %v", releaseCtx.Err()) + } + require.NoError(t, waiterService.Unlock(releaseCtx, activeWaiterTxn, timestamp.Timestamp{})) }, ) } From acb41d7f3febfeec75b729a11fa839b4b544acdd Mon Sep 17 00:00:00 2001 From: GreatRiver Date: Sat, 18 Jul 2026 17:32:35 +0800 Subject: [PATCH 2/4] test(lockservice): give remote waiter a deadline --- pkg/lockservice/service_remote_test.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/pkg/lockservice/service_remote_test.go b/pkg/lockservice/service_remote_test.go index 10eef14362e83..2a82f50ea7bbb 100644 --- a/pkg/lockservice/service_remote_test.go +++ b/pkg/lockservice/service_remote_test.go @@ -91,7 +91,10 @@ func TestFetchWhoWaitingMeUsesActiveRemoteWaiterSnapshots(t *testing.T) { // Reusing ctx here couples queue admission to all setup RPCs above: once // that deadline expires, Lock returns before it can enter owner's queue // and the old unbounded waitWaiters call spins forever. - waiterCtx, cancelWaiter := context.WithCancel(context.Background()) + // Lock may cross MORPC, whose Future contract requires a deadline. + // Give the waiter a budget independent from and larger than setup while + // retaining a hard bound if admission or cleanup regresses. + waiterCtx, cancelWaiter := context.WithTimeout(context.Background(), 30*time.Second) defer cancelWaiter() waitResult := make(chan error, 1) holderReleased := false From 0d406c06dff40a79c99cfcc15a1c53c1fce3f93b Mon Sep 17 00:00:00 2001 From: GreatRiver Date: Sun, 19 Jul 2026 00:20:16 +0800 Subject: [PATCH 3/4] test(lockservice): bound remote unlock cleanup --- pkg/lockservice/service_remote_test.go | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/pkg/lockservice/service_remote_test.go b/pkg/lockservice/service_remote_test.go index 1b4a051bfa1b7..660373ab8a6d4 100644 --- a/pkg/lockservice/service_remote_test.go +++ b/pkg/lockservice/service_remote_test.go @@ -110,18 +110,21 @@ func TestFetchWhoWaitingMeUsesActiveRemoteWaiterSnapshots(t *testing.T) { ) waitResult <- err }() + // Public Unlock deliberately retries remote cleanup with a background + // context. This liveness test uses the internal context-aware path so its + // release and failure cleanup remain bounded by their test deadlines. defer func() { cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), 10*time.Second) defer cleanupCancel() if !holderReleased { - _ = holderService.Unlock(cleanupCtx, holderTxn, timestamp.Timestamp{}) + _ = holderService.unlockWithContext(cleanupCtx, holderTxn, timestamp.Timestamp{}) } cancelWaiter() if !waiterDone { select { case err := <-waitResult: if err == nil { - _ = waiterService.Unlock(cleanupCtx, activeWaiterTxn, timestamp.Timestamp{}) + _ = waiterService.unlockWithContext(cleanupCtx, activeWaiterTxn, timestamp.Timestamp{}) } case <-cleanupCtx.Done(): t.Errorf("remote waiter did not exit during cleanup: %v", cleanupCtx.Err()) @@ -195,7 +198,7 @@ func TestFetchWhoWaitingMeUsesActiveRemoteWaiterSnapshots(t *testing.T) { releaseCtx, releaseCancel := context.WithTimeout(context.Background(), 10*time.Second) defer releaseCancel() - require.NoError(t, holderService.Unlock(releaseCtx, holderTxn, timestamp.Timestamp{})) + require.NoError(t, holderService.unlockWithContext(releaseCtx, holderTxn, timestamp.Timestamp{})) holderReleased = true select { case err := <-waitResult: @@ -204,7 +207,7 @@ func TestFetchWhoWaitingMeUsesActiveRemoteWaiterSnapshots(t *testing.T) { case <-releaseCtx.Done(): t.Fatalf("remote waiter did not acquire after holder release: %v", releaseCtx.Err()) } - require.NoError(t, waiterService.Unlock(releaseCtx, activeWaiterTxn, timestamp.Timestamp{})) + require.NoError(t, waiterService.unlockWithContext(releaseCtx, activeWaiterTxn, timestamp.Timestamp{})) }, func(cfg *Config) { // CheckActiveTxn gets its authoritative transaction liveness from From d6a9b4444a4c40f3e7ad0bef6482e05bc1b2dfbe Mon Sep 17 00:00:00 2001 From: GreatRiver Date: Sun, 19 Jul 2026 11:39:15 +0800 Subject: [PATCH 4/4] test(lockservice): clean failed remote waiters --- pkg/lockservice/service_remote_test.go | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/pkg/lockservice/service_remote_test.go b/pkg/lockservice/service_remote_test.go index 660373ab8a6d4..1c4a315dd444d 100644 --- a/pkg/lockservice/service_remote_test.go +++ b/pkg/lockservice/service_remote_test.go @@ -122,14 +122,15 @@ func TestFetchWhoWaitingMeUsesActiveRemoteWaiterSnapshots(t *testing.T) { cancelWaiter() if !waiterDone { select { - case err := <-waitResult: - if err == nil { - _ = waiterService.unlockWithContext(cleanupCtx, activeWaiterTxn, timestamp.Timestamp{}) - } + case <-waitResult: case <-cleanupCtx.Done(): t.Errorf("remote waiter did not exit during cleanup: %v", cleanupCtx.Err()) } } + // A failed remote Lock can still have reached the owner and is recorded + // locally for exactly this compensating unlock. It is also safe after a + // successful unlock, where the transaction is already absent. + _ = waiterService.unlockWithContext(cleanupCtx, activeWaiterTxn, timestamp.Timestamp{}) }() // This is test admission synchronization, not part of the behavior under