transaction: Allow multiple votes per transaction

Right now, we're using the pre-receive hook to call into transactions on
the Gitaly nodes. This hook is well-understood to be executed once and
only once per action, and as a result the current implementation of
transactions allowed for a single vote via this pre-receive hook, only.
The pre-receive hook was a stop-gap implementation of the real mechanism
we wanted to eventually use, though, which is the reference-transaction
hook that's going to be released as part of git-core v2.28. While the
input to both hooks is the same and thus no changes to the actual voting
logic should be required, the most important difference is that the
reference-transaction hook may be invoked arbitrarily many times for
each Git command. E.g. a non-atomic push will execute the hook as many
times as there are updated references.

This surfaces a current design limitation of the transaction mechanism
as it is implemented in Gitaly: as a transaction only allows for a
single vote per node, it's inherently incompatible with the semantics
introduce by the reference-transaction hook. This is why we now extend
the transaction mechanism to allow a sequence of votes instead by
introducing subtransactions.

Subtransactions encapsulate all the voting logic that was previously
part of the transaction, which is casting the vote and collecting the
votes to establish whether quorum was reached. Transactions now start to
be a wrapper of a set of transactions, which transparantly creates new
subtransactions as required whenever a node casts a vote. Whenever a
node casts a vote, one of three things may now happen:

    1. A subtransaction exists where the node's status is "aborted". This
       means that any one of the subtransactions failed and thus the
       complete transaction needs to be labelled as "aborted" for the
       node. It will receive an error and is not allowed to cast any
       votes anymore.

    2. A subtransaction exists where the node's state is "undecided". As
       this means that the node simply didn't cast a vote, it will cast
       its vote for the oldest undecided subtransaction and wait for
       quorum to be reached.

    3. All existing subtransactions are in "committed" state for the
       given node. We'll thus create a new subtransaction, cast our vote
       for this new transaction and wait for quorum to be reached.

To evaluate the complete transaction's outcome, we need to establish
whether for a given node, all created subtransactions are in state
"committed". Only if that's the case will the transaction be treated as
successful for the node.

With this logic, we're now able to transparently allow a voter to cast a
sequence of votes.

1 files changed, 107 insertions, 0 deletions

diff --git a/internal/praefect/transaction_test.go b/internal/praefect/transaction_test.go
index 55882f588..12223c272 100644
--- a/internal/praefect/transaction_test.go
+++ b/internal/praefect/transaction_test.go
@@ -10,6 +10,7 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"gitlab.com/gitlab-org/gitaly/internal/praefect/transactions"
 	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
@@ -443,6 +444,112 @@ func TestTransactionReachesQuorum(t *testing.T) {
 	}
 }
 
+func TestTransactionWithMultipleVotes(t *testing.T) {
+	type multiVoter struct {
+		voteCount     uint
+		votes         []string
+		voteSucceeds  []bool
+		shouldSucceed bool
+	}
+
+	tc := []struct {
+		desc      string
+		voters    []multiVoter
+		threshold uint
+	}{
+		{
+			desc: "quorum is reached with multiple votes",
+			voters: []multiVoter{
+				{voteCount: 1, votes: []string{"foo", "bar"}, voteSucceeds: []bool{true, true}, shouldSucceed: true},
+				{voteCount: 1, votes: []string{"foo", "bar"}, voteSucceeds: []bool{true, true}, shouldSucceed: true},
+			},
+			threshold: 2,
+		},
+		{
+			desc: "quorum is not reached with disagreeing votes",
+			voters: []multiVoter{
+				{voteCount: 1, votes: []string{"foo", "bar"}, voteSucceeds: []bool{true, false}, shouldSucceed: false},
+				{voteCount: 1, votes: []string{"foo", "rab"}, voteSucceeds: []bool{true, false}, shouldSucceed: false},
+			},
+			threshold: 2,
+		},
+		{
+			desc: "quorum is reached with unweighted disagreeing voter",
+			voters: []multiVoter{
+				{voteCount: 1, votes: []string{"foo", "bar", "qux"}, voteSucceeds: []bool{true, true, true}, shouldSucceed: true},
+				{voteCount: 0, votes: []string{"foo", "rab"}, voteSucceeds: []bool{true, false}, shouldSucceed: false},
+			},
+			threshold: 1,
+		},
+		{
+			desc: "quorum is reached with outweighed disagreeing voter",
+			voters: []multiVoter{
+				{voteCount: 1, votes: []string{"foo", "bar", "qux"}, voteSucceeds: []bool{true, true, true}, shouldSucceed: true},
+				{voteCount: 1, votes: []string{"foo", "bar", "qux"}, voteSucceeds: []bool{true, true, true}, shouldSucceed: true},
+				{voteCount: 1, votes: []string{"foo", "rab"}, voteSucceeds: []bool{true, false}, shouldSucceed: false},
+			},
+			threshold: 2,
+		},
+	}
+
+	cc, txMgr, cleanup := runPraefectServerAndTxMgr(t)
+	defer cleanup()
+
+	ctx, cleanup := testhelper.Context()
+	defer cleanup()
+
+	client := gitalypb.NewRefTransactionClient(cc)
+
+	for _, tc := range tc {
+		t.Run(tc.desc, func(t *testing.T) {
+			var voters []transactions.Voter
+
+			for i, voter := range tc.voters {
+				voters = append(voters, transactions.Voter{
+					Name:  fmt.Sprintf("node-%d", i),
+					Votes: voter.voteCount,
+				})
+			}
+
+			transactionID, cancel, err := txMgr.RegisterTransaction(ctx, voters, tc.threshold)
+			require.NoError(t, err)
+
+			var wg sync.WaitGroup
+			for i, v := range tc.voters {
+				wg.Add(1)
+				go func(i int, v multiVoter) {
+					defer wg.Done()
+
+					for j, vote := range v.votes {
+						name := fmt.Sprintf("node-%d", i)
+						hash := sha1.Sum([]byte(vote))
+
+						response, err := client.VoteTransaction(ctx, &gitalypb.VoteTransactionRequest{
+							TransactionId:        transactionID,
+							Node:                 name,
+							ReferenceUpdatesHash: hash[:],
+						})
+						assert.NoError(t, err)
+
+						if v.voteSucceeds[j] {
+							assert.Equal(t, gitalypb.VoteTransactionResponse_COMMIT, response.State, "node should have received COMMIT")
+						} else {
+							assert.Equal(t, gitalypb.VoteTransactionResponse_ABORT, response.State, "node should have received ABORT")
+						}
+					}
+				}(i, v)
+			}
+
+			wg.Wait()
+
+			results, _ := cancel()
+			for i, voter := range tc.voters {
+				require.Equal(t, voter.shouldSucceed, results[fmt.Sprintf("node-%d", i)])
+			}
+		})
+	}
+}
+
 func TestTransactionFailures(t *testing.T) {
 	counter, opts := setupMetrics()
 	cc, _, cleanup := runPraefectServerAndTxMgr(t, opts...)