Fix missing TX registration (#3080)

## Summary

When a 2.1 (i.e. no txsync) client connects to a 3.0 relay (i.e. txsync), the relay needs to request the client to keep sending it a TX messages - otherwise, these transactions would not get propagated.

The 2.1 and 3.0 above are network protocol versions, not a algod release version.

## Test Plan

e2e test added.

data/txHandler.go

@@ -502,6 +502,7 @@ func (handler *solicitedAsyncTxHandler) HandleTransactionGroups(networkPeer inte
 
 func (handler *solicitedAsyncTxHandler) Start() {
 	if handler.stopCtxFunc == nil {
+		handler.txHandler.Start()
 		var ctx context.Context
 		ctx, handler.stopCtxFunc = context.WithCancel(context.Background())
 		handler.stopped.Add(1)
@@ -514,6 +515,7 @@ func (handler *solicitedAsyncTxHandler) Stop() {
 		handler.stopCtxFunc()
 		handler.stopped.Wait()
 		handler.stopCtxFunc = nil
+		handler.txHandler.Stop()
 	}
 }
 

network/wsNetwork.go

@@ -1144,7 +1144,21 @@ func (wn *WebsocketNetwork) ServeHTTP(response http.ResponseWriter, request *htt
 
 	// We are careful to encode this prior to starting the server to avoid needing 'messagesOfInterestMu' here.
 	if wn.messagesOfInterestEnc != nil {
-		err = peer.Unicast(wn.ctx, wn.messagesOfInterestEnc, protocol.MsgOfInterestTag, nil)
+		msg := wn.messagesOfInterestEnc
+		// for older peers, we want to include also the "TX" message, for backward compatibility.
+		// this statement could be safely removed once we've fully migrated.
+		if peer.version == "2.1" {
+			wn.messagesOfInterestMu.Lock()
+			txSendMsgTags := make(map[protocol.Tag]bool)
+			for tag := range wn.messagesOfInterest {
+				txSendMsgTags[tag] = true
+			}
+			wn.messagesOfInterestMu.Unlock()
+			txSendMsgTags[protocol.TxnTag] = true
+			msg = MarshallMessageOfInterestMap(txSendMsgTags)
+		}
+		err = peer.Unicast(wn.ctx, msg, protocol.MsgOfInterestTag, nil)
+
 		if err != nil {
 			wn.log.Infof("ws send msgOfInterest: %v", err)
 		}

test/e2e-go/features/transactions/sendReceive_test.go

@@ -17,12 +17,14 @@
 package transactions
 
 import (
+	"fmt"
 	"math/rand"
 	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/algorand/go-algorand/config"
 	v1 "github.com/algorand/go-algorand/daemon/algod/api/spec/v1"
 	"github.com/algorand/go-algorand/test/framework/fixtures"
 	"github.com/algorand/go-algorand/test/partitiontest"
@@ -66,11 +68,16 @@ func TestDevModeAccountsCanSendMoney(t *testing.T) {
 
 func testAccountsCanSendMoney(t *testing.T, templatePath string, numberOfSends int) {
 	t.Parallel()
-	a := require.New(fixtures.SynchronizedTest(t))
 
 	var fixture fixtures.RestClientFixture
 	fixture.Setup(t, templatePath)
 	defer fixture.Shutdown()
+	testAccountsCanSendMoneyFixture(t, &fixture, numberOfSends)
+}
+
+func testAccountsCanSendMoneyFixture(t *testing.T, fixture *fixtures.RestClientFixture, numberOfSends int) {
+	a := require.New(fixtures.SynchronizedTest(t))
+
 	c := fixture.LibGoalClient
 
 	pingClient := fixture.LibGoalClient
@@ -159,3 +166,45 @@ func testAccountsCanSendMoney(t *testing.T, templatePath string, numberOfSends i
 	a.True(expectedPingBalance <= pingBalance, "ping balance is different than expected.")
 	a.True(expectedPongBalance <= pongBalance, "pong balance is different than expected.")
 }
+
+// this test checks that two accounts' balances stay up to date
+// as they send each other money many times
+func TestAccountsCanSendMoneyAcrossTxSync(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	defer fixtures.ShutdownSynchronizedTest(t)
+
+	numberOfSends := 3
+	a := require.New(fixtures.SynchronizedTest(t))
+
+	networkProtocolVersions := []string{"3.0", "2.1"}
+
+	testMoneySending := func(t *testing.T, primaryNodeVersionIdx, nodeVersionIdx int) {
+		t.Run(fmt.Sprintf("%s->%s", networkProtocolVersions[primaryNodeVersionIdx], networkProtocolVersions[nodeVersionIdx]),
+			func(t *testing.T) {
+				t.Parallel()
+				var fixture fixtures.RestClientFixture
+				fixture.SetupNoStart(t, filepath.Join("nettemplates", "TwoNodes50Each.json"))
+				defer fixture.Shutdown()
+
+				cfg, err := config.LoadConfigFromDisk(fixture.PrimaryDataDir())
+				a.NoError(err)
+				cfg.NetworkProtocolVersion = networkProtocolVersions[primaryNodeVersionIdx]
+				cfg.SaveToDisk(fixture.PrimaryDataDir())
+
+				cfg, err = config.LoadConfigFromDisk(fixture.NodeDataDirs()[0])
+				a.NoError(err)
+				cfg.NetworkProtocolVersion = networkProtocolVersions[primaryNodeVersionIdx]
+				cfg.SaveToDisk(fixture.NodeDataDirs()[0])
+
+				fixture.Start()
+				testAccountsCanSendMoneyFixture(t, &fixture, numberOfSends)
+			})
+	}
+
+	// test to see that we can communicate correctly regardless of the network protocol version.
+	for primaryNodeVersionIdx := 0; primaryNodeVersionIdx < 2; primaryNodeVersionIdx++ {
+		for nodeVersionIdx := 0; nodeVersionIdx < 2; nodeVersionIdx++ {
+			testMoneySending(t, primaryNodeVersionIdx, nodeVersionIdx)
+		}
+	}
+}