fix lockservice orphan remote txn validation

pkg/lockservice/orphan_txn_test.go

@@ -804,6 +804,20 @@ func TestValidTxnWithInvalidRemoteTxn(t *testing.T) {
 	require.False(t, hold.isValidRemoteTxn(pb.WaitTxn{TxnID: []byte{1}, CreatedOn: "s0"}))
 }
 
+func TestValidTxnWithInactiveRemoteTxnAndNotifyFoundCommitting(t *testing.T) {
+	hold := newMapBasedTxnHandler(
+		"s1",
+		getLogger(""),
+		newFixedSlicePool(16),
+		func(sid string) (bool, error) { return false, nil },
+		func(ot []pb.OrphanTxn) ([][]byte, error) { return [][]byte{{1}}, nil },
+		func(txn pb.WaitTxn) (bool, error) {
+			return false, nil
+		},
+	).(*mapBasedTxnHolder)
+	require.True(t, hold.isValidRemoteTxn(pb.WaitTxn{TxnID: []byte{1}, CreatedOn: "s0"}))
+}
+
 func TestValidTxnWithInvalidRemoteTxnAndNotifyOK(t *testing.T) {
 	hold := newMapBasedTxnHandler(
 		"s1",

pkg/lockservice/service.go

@@ -1001,7 +1001,13 @@ func (h *mapBasedTxnHolder) isValidRemoteTxn(txn pb.WaitTxn) bool {
 
 	valid, err := h.validTxn(txn)
 	if err == nil {
-		return valid
+		if valid {
+			return true
+		}
+		// A remote CN active-txn miss does not prove the txn is safe to unlock.
+		// The commit response may be lost after TN has already committed it, so
+		// require the allocator to confirm the txn cannot still be committing.
+		return !h.canUnlockRemoteTxn(txn)
 	}
 	logValidTxnFailed(h.logger, txn, err)
 	if isRetryError(err) {