rpc: retrieve unresolved transactions (#16356)

Signed-off-by: Harshit Gangal <[email protected]>

go/vt/vtcombo/tablet_map.go

@@ -553,6 +553,12 @@ func (itc *internalTabletConn) ReadTransaction(ctx context.Context, target *quer
 	return metadata, tabletconn.ErrorFromGRPC(vterrors.ToGRPC(err))
 }
 
+// UnresolvedTransactions is part of queryservice.QueryService
+func (itc *internalTabletConn) UnresolvedTransactions(ctx context.Context, target *querypb.Target) (transactions []*querypb.TransactionMetadata, err error) {
+	transactions, err = itc.tablet.qsc.QueryService().UnresolvedTransactions(ctx, target)
+	return transactions, tabletconn.ErrorFromGRPC(vterrors.ToGRPC(err))
+}
+
 // BeginExecute is part of queryservice.QueryService
 func (itc *internalTabletConn) BeginExecute(
 	ctx context.Context,

go/vt/vttablet/endtoend/framework/client.go

@@ -178,6 +178,11 @@ func (client *QueryClient) ReadTransaction(dtid string) (*querypb.TransactionMet
 	return client.server.ReadTransaction(client.ctx, client.target, dtid)
 }
 
+// UnresolvedTransactions invokes the UnresolvedTransactions API of TabletServer.
+func (client *QueryClient) UnresolvedTransactions() ([]*querypb.TransactionMetadata, error) {
+	return client.server.UnresolvedTransactions(client.ctx, client.target)
+}
+
 // SetServingType is for testing transitions.
 // It currently supports only primary->replica and back.
 func (client *QueryClient) SetServingType(tabletType topodatapb.TabletType) error {

go/vt/vttablet/endtoend/transaction_test.go

@@ -737,3 +737,32 @@ func TestManualTwopcz(t *testing.T) {
 	fmt.Print("Sleeping for 30 seconds\n")
 	time.Sleep(30 * time.Second)
 }
+
+// TestUnresolvedTransactions tests the UnresolvedTransactions API.
+func TestUnresolvedTransactions(t *testing.T) {
+	client := framework.NewClient()
+
+	participants := []*querypb.Target{
+		{Keyspace: "ks1", Shard: "-80"},
+		{Keyspace: "ks1", Shard: "80-"},
+	}
+	err := client.CreateTransaction("dtid01", participants)
+	require.NoError(t, err)
+
+	// expected no transaction to show here, as 1 second not passed.
+	transactions, err := client.UnresolvedTransactions()
+	require.NoError(t, err)
+	require.Empty(t, transactions)
+
+	// abandon age is 1 second.
+	time.Sleep(2 * time.Second)
+
+	transactions, err = client.UnresolvedTransactions()
+	require.NoError(t, err)
+	want := []*querypb.TransactionMetadata{{
+		Dtid:         "dtid01",
+		State:        querypb.TransactionState_PREPARE,
+		Participants: participants,
+	}}
+	utils.MustMatch(t, want, transactions)
+}

go/vt/vttablet/grpcqueryservice/server.go

@@ -233,6 +233,21 @@ func (q *query) ReadTransaction(ctx context.Context, request *querypb.ReadTransa
 	return &querypb.ReadTransactionResponse{Metadata: result}, nil
 }
 
+// UnresolvedTransactions is part of the queryservice.QueryServer interface
+func (q *query) UnresolvedTransactions(ctx context.Context, request *querypb.UnresolvedTransactionsRequest) (response *querypb.UnresolvedTransactionsResponse, err error) {
+	defer q.server.HandlePanic(&err)
+	ctx = callerid.NewContext(callinfo.GRPCCallInfo(ctx),
+		request.EffectiveCallerId,
+		request.ImmediateCallerId,
+	)
+	transactions, err := q.server.UnresolvedTransactions(ctx, request.Target)
+	if err != nil {
+		return nil, vterrors.ToGRPC(err)
+	}
+
+	return &querypb.UnresolvedTransactionsResponse{Transactions: transactions}, nil
+}
+
 // BeginExecute is part of the queryservice.QueryServer interface
 func (q *query) BeginExecute(ctx context.Context, request *querypb.BeginExecuteRequest) (response *querypb.BeginExecuteResponse, err error) {
 	defer q.server.HandlePanic(&err)

go/vt/vttablet/grpctabletconn/conn.go

@@ -438,6 +438,26 @@ func (conn *gRPCQueryClient) ReadTransaction(ctx context.Context, target *queryp
 	return response.Metadata, nil
 }
 
+// UnresolvedTransactions returns all unresolved distributed transactions.
+func (conn *gRPCQueryClient) UnresolvedTransactions(ctx context.Context, target *querypb.Target) ([]*querypb.TransactionMetadata, error) {
+	conn.mu.RLock()
+	defer conn.mu.RUnlock()
+	if conn.cc == nil {
+		return nil, tabletconn.ConnClosed
+	}
+
+	req := &querypb.UnresolvedTransactionsRequest{
+		Target:            target,
+		EffectiveCallerId: callerid.EffectiveCallerIDFromContext(ctx),
+		ImmediateCallerId: callerid.ImmediateCallerIDFromContext(ctx),
+	}
+	response, err := conn.c.UnresolvedTransactions(ctx, req)
+	if err != nil {
+		return nil, tabletconn.ErrorFromGRPC(err)
+	}
+	return response.Transactions, nil
+}
+
 // BeginExecute starts a transaction and runs an Execute.
 func (conn *gRPCQueryClient) BeginExecute(ctx context.Context, target *querypb.Target, preQueries []string, query string, bindVars map[string]*querypb.BindVariable, reservedID int64, options *querypb.ExecuteOptions) (state queryservice.TransactionState, result *sqltypes.Result, err error) {
 	conn.mu.RLock()

go/vt/vttablet/queryservice/queryservice.go

@@ -76,6 +76,9 @@ type QueryService interface {
 	// ReadTransaction returns the metadata for the specified dtid.
 	ReadTransaction(ctx context.Context, target *querypb.Target, dtid string) (metadata *querypb.TransactionMetadata, err error)
 
+	// UnresolvedTransactions returns the list of unresolved distributed transactions.
+	UnresolvedTransactions(ctx context.Context, target *querypb.Target) ([]*querypb.TransactionMetadata, error)
+
 	// Execute for query execution
 	Execute(ctx context.Context, target *querypb.Target, sql string, bindVariables map[string]*querypb.BindVariable, transactionID, reservedID int64, options *querypb.ExecuteOptions) (*sqltypes.Result, error)
 	// StreamExecute for query execution with streaming

go/vt/vttablet/queryservice/wrapped.go

@@ -176,6 +176,15 @@ func (ws *wrappedService) ReadTransaction(ctx context.Context, target *querypb.T
 	return metadata, err
 }
 
+func (ws *wrappedService) UnresolvedTransactions(ctx context.Context, target *querypb.Target) (transactions []*querypb.TransactionMetadata, err error) {
+	err = ws.wrapper(ctx, target, ws.impl, "UnresolvedTransactions", false, func(ctx context.Context, target *querypb.Target, conn QueryService) (bool, error) {
+		var innerErr error
+		transactions, innerErr = conn.UnresolvedTransactions(ctx, target)
+		return canRetry(ctx, innerErr), innerErr
+	})
+	return transactions, err
+}
+
 func (ws *wrappedService) Execute(ctx context.Context, target *querypb.Target, query string, bindVars map[string]*querypb.BindVariable, transactionID, reservedID int64, options *querypb.ExecuteOptions) (qr *sqltypes.Result, err error) {
 	inDedicatedConn := transactionID != 0 || reservedID != 0
 	err = ws.wrapper(ctx, target, ws.impl, "Execute", inDedicatedConn, func(ctx context.Context, target *querypb.Target, conn QueryService) (bool, error) {

go/vt/vttablet/sandboxconn/sandboxconn.go

@@ -427,6 +427,12 @@ func (sbc *SandboxConn) ReadTransaction(ctx context.Context, target *querypb.Tar
 	return nil, nil
 }
 
+// UnresolvedTransactions is part of the QueryService interface.
+func (sbc *SandboxConn) UnresolvedTransactions(context.Context, *querypb.Target) ([]*querypb.TransactionMetadata, error) {
+	// TODO implement me
+	panic("implement me")
+}
+
 // BeginExecute is part of the QueryService interface.
 func (sbc *SandboxConn) BeginExecute(ctx context.Context, target *querypb.Target, preQueries []string, query string, bindVars map[string]*querypb.BindVariable, reservedID int64, options *querypb.ExecuteOptions) (queryservice.TransactionState, *sqltypes.Result, error) {
 	sbc.panicIfNeeded()

go/vt/vttablet/tabletconntest/fakequeryservice.go

@@ -359,6 +359,18 @@ func (f *FakeQueryService) ReadTransaction(ctx context.Context, target *querypb.
 	return Metadata, nil
 }
 
+// UnresolvedTransactions is part of the queryservice.QueryService interface
+func (f *FakeQueryService) UnresolvedTransactions(ctx context.Context, target *querypb.Target) ([]*querypb.TransactionMetadata, error) {
+	if f.HasError {
+		return nil, f.TabletError
+	}
+	if f.Panics {
+		panic(fmt.Errorf("test-triggered panic"))
+	}
+	f.checkTargetCallerID(ctx, "UnresolvedTransactions", target)
+	return []*querypb.TransactionMetadata{Metadata}, nil
+}
+
 // ExecuteQuery is a fake test query.
 const ExecuteQuery = "executeQuery"
 

go/vt/vttablet/tabletconntest/tabletconntest.go

@@ -399,6 +399,33 @@ func testReadTransactionPanics(t *testing.T, conn queryservice.QueryService, f *
 	})
 }
 
+func testUnresolvedTransactions(t *testing.T, conn queryservice.QueryService, f *FakeQueryService) {
+	t.Log("testUnresolvedTransactions")
+	ctx := context.Background()
+	ctx = callerid.NewContext(ctx, TestCallerID, TestVTGateCallerID)
+	transactions, err := conn.UnresolvedTransactions(ctx, TestTarget)
+	require.NoError(t, err)
+	require.True(t, proto.Equal(transactions[0], Metadata))
+}
+
+func testUnresolvedTransactionsError(t *testing.T, conn queryservice.QueryService, f *FakeQueryService) {
+	t.Log("testUnresolvedTransactionsError")
+	f.HasError = true
+	testErrorHelper(t, f, "UnresolvedTransactions", func(ctx context.Context) error {
+		_, err := conn.UnresolvedTransactions(ctx, TestTarget)
+		return err
+	})
+	f.HasError = false
+}
+
+func testUnresolvedTransactionsPanics(t *testing.T, conn queryservice.QueryService, f *FakeQueryService) {
+	t.Log("testUnresolvedTransactionsPanics")
+	testPanicHelper(t, f, "UnresolvedTransactions", func(ctx context.Context) error {
+		_, err := conn.UnresolvedTransactions(ctx, TestTarget)
+		return err
+	})
+}
+
 func testExecute(t *testing.T, conn queryservice.QueryService, f *FakeQueryService) {
 	t.Log("testExecute")
 	f.ExpectedTransactionID = ExecuteTransactionID
@@ -936,6 +963,7 @@ func TestSuite(ctx context.Context, t *testing.T, protocol string, tablet *topod
 		testSetRollback,
 		testConcludeTransaction,
 		testReadTransaction,
+		testUnresolvedTransactions,
 		testExecute,
 		testBeginExecute,
 		testStreamExecute,
@@ -956,6 +984,7 @@ func TestSuite(ctx context.Context, t *testing.T, protocol string, tablet *topod
 		testSetRollbackError,
 		testConcludeTransactionError,
 		testReadTransactionError,
+		testUnresolvedTransactionsError,
 		testExecuteError,
 		testBeginExecuteErrorInBegin,
 		testBeginExecuteErrorInExecute,
@@ -979,6 +1008,7 @@ func TestSuite(ctx context.Context, t *testing.T, protocol string, tablet *topod
 		testSetRollbackPanics,
 		testConcludeTransactionPanics,
 		testReadTransactionPanics,
+		testUnresolvedTransactionsPanics,
 		testExecutePanics,
 		testBeginExecutePanics,
 		testStreamExecutePanics,

go/vt/vttablet/tabletserver/dt_executor.go

@@ -20,15 +20,13 @@ import (
 	"context"
 	"time"
 
-	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
-
 	"vitess.io/vitess/go/trace"
 	"vitess.io/vitess/go/vt/log"
-
 	querypb "vitess.io/vitess/go/vt/proto/query"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
+	"vitess.io/vitess/go/vt/vttablet/tabletserver/tx"
 )
 
 // DTExecutor is used for executing a distributed transactional request.
@@ -290,3 +288,8 @@ func (dte *DTExecutor) inTransaction(f func(*StatefulConnection) error) error {
 	}
 	return nil
 }
+
+// UnresolvedTransactions returns the list of unresolved distributed transactions.
+func (dte *DTExecutor) UnresolvedTransactions() ([]*querypb.TransactionMetadata, error) {
+	return dte.te.twoPC.UnresolvedTransactions(dte.ctx, time.Now().Add(-dte.te.abandonAge))
+}

go/vt/vttablet/tabletserver/tabletserver.go

@@ -44,6 +44,10 @@ import (
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/logutil"
 	"vitess.io/vitess/go/vt/mysqlctl"
+	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
+	querypb "vitess.io/vitess/go/vt/proto/query"
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/servenv"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/srvtopo"
@@ -67,11 +71,6 @@ import (
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/txserializer"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/txthrottler"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/vstreamer"
-
-	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
-	querypb "vitess.io/vitess/go/vt/proto/query"
-	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
-	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 )
 
 // logPoolFull is for throttling transaction / query pool full messages in the log.
@@ -748,6 +747,21 @@ func (tsv *TabletServer) ReadTransaction(ctx context.Context, target *querypb.Ta
 	return metadata, err
 }
 
+// UnresolvedTransactions returns the unresolved distributed transaction record.
+func (tsv *TabletServer) UnresolvedTransactions(ctx context.Context, target *querypb.Target) (transactions []*querypb.TransactionMetadata, err error) {
+	err = tsv.execRequest(
+		ctx, tsv.loadQueryTimeout(),
+		"UnresolvedTransactions", "unresolved_transaction", nil,
+		target, nil, false, /* allowOnShutdown */
+		func(ctx context.Context, logStats *tabletenv.LogStats) error {
+			txe := NewDTExecutor(ctx, tsv.te, logStats)
+			transactions, err = txe.UnresolvedTransactions()
+			return err
+		},
+	)
+	return
+}
+
 // Execute executes the query and returns the result as response.
 func (tsv *TabletServer) Execute(ctx context.Context, target *querypb.Target, sql string, bindVariables map[string]*querypb.BindVariable, transactionID, reservedID int64, options *querypb.ExecuteOptions) (result *sqltypes.Result, err error) {
 	span, ctx := trace.NewSpan(ctx, "TabletServer.Execute")