txhandler: service does not restart after node catching up

data/txHandler.go

@@ -91,13 +91,12 @@ func MakeTxHandler(txPool *pools.TransactionPool, ledger *Ledger, net network.Go
 		postVerificationQueue: make(chan *txBacklogMsg, txBacklogSize),
 		net:                   net,
 	}
-
-	handler.ctx, handler.ctxCancel = context.WithCancel(context.Background())
 	return handler
 }
 
 // Start enables the processing of incoming messages at the transaction handler
 func (handler *TxHandler) Start() {
+	handler.ctx, handler.ctxCancel = context.WithCancel(context.Background())
 	handler.net.RegisterHandlers([]network.TaggedMessageHandler{
 		{Tag: protocol.TxnTag, MessageHandler: network.HandlerFunc(handler.processIncomingTxn)},
 	})

data/txHandler_test.go

@@ -17,6 +17,7 @@
 package data
 
 import (
+	"context"
 	"encoding/binary"
 	"fmt"
 	"io"
@@ -301,6 +302,8 @@ func incomingTxHandlerProcessing(maxGroupSize int, t *testing.T) {
 	tp := pools.MakeTransactionPool(l.Ledger, cfg, logging.Base())
 	backlogPool := execpool.MakeBacklog(nil, 0, execpool.LowPriority, nil)
 	handler := MakeTxHandler(tp, l, &mocks.MockNetwork{}, "", crypto.Digest{}, backlogPool)
+	// since Start is not called, set the context here
+	handler.ctx, handler.ctxCancel = context.WithCancel(context.Background())
 	defer handler.ctxCancel()
 
 	outChan := make(chan *txBacklogMsg, 10)
@@ -522,6 +525,8 @@ func runHandlerBenchmark(maxGroupSize int, b *testing.B) {
 	tp := pools.MakeTransactionPool(l.Ledger, cfg, logging.Base())
 	backlogPool := execpool.MakeBacklog(nil, 0, execpool.LowPriority, nil)
 	handler := MakeTxHandler(tp, l, &mocks.MockNetwork{}, "", crypto.Digest{}, backlogPool)
+	// since Start is not called, set the context here
+	handler.ctx, handler.ctxCancel = context.WithCancel(context.Background())
 	defer handler.ctxCancel()
 
 	// Prepare the transactions

test/e2e-go/features/catchup/catchpointCatchup_test.go

@@ -385,7 +385,6 @@ func TestCatchpointLabelGeneration(t *testing.T) {
 					break
 				}
 				currentRound++
-
 			}
 			log.Infof("done building!\n")
 
@@ -400,3 +399,140 @@ func TestCatchpointLabelGeneration(t *testing.T) {
 		})
 	}
 }
+
+// TestNodeTxHandlerRestart starts a two-node and one relay network
+// Waits until a catchpoint is created
+// Lets the primary node have the majority of the stake
+// Sends a transaction from the second node
+// The transaction will be confirmed only if the txHandler gets the transaction
+func TestNodeTxHandlerRestart(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	defer fixtures.ShutdownSynchronizedTest(t)
+
+	if testing.Short() {
+		t.Skip()
+	}
+	a := require.New(fixtures.SynchronizedTest(t))
+
+	consensus := make(config.ConsensusProtocols)
+	protoVersion := protocol.ConsensusCurrentVersion
+	catchpointCatchupProtocol := config.Consensus[protoVersion]
+	catchpointCatchupProtocol.ApprovedUpgrades = map[protocol.ConsensusVersion]uint64{}
+	// MaxBalLookback  =  2 x SeedRefreshInterval x SeedLookback
+	// ref. https://github.com/algorandfoundation/specs/blob/master/dev/abft.md
+	catchpointCatchupProtocol.SeedLookback = 2
+	catchpointCatchupProtocol.SeedRefreshInterval = 2
+	catchpointCatchupProtocol.MaxBalLookback = 2 * catchpointCatchupProtocol.SeedLookback * catchpointCatchupProtocol.SeedRefreshInterval // 8
+	catchpointCatchupProtocol.CatchpointLookback = catchpointCatchupProtocol.MaxBalLookback
+	catchpointCatchupProtocol.EnableOnlineAccountCatchpoints = true
+	catchpointCatchupProtocol.StateProofInterval = 0
+	if runtime.GOOS == "darwin" || runtime.GOARCH == "amd64" {
+		// amd64/macos platforms are generally quite capable, so accelerate the round times to make the test run faster.
+		catchpointCatchupProtocol.AgreementFilterTimeoutPeriod0 = 1 * time.Second
+		catchpointCatchupProtocol.AgreementFilterTimeout = 1 * time.Second
+	}
+	consensus[protoVersion] = catchpointCatchupProtocol
+
+	var fixture fixtures.RestClientFixture
+	fixture.SetConsensus(consensus)
+	fixture.SetupNoStart(t, filepath.Join("nettemplates", "TwoNodes50EachWithRelay.json"))
+
+	// Get primary node
+	primaryNode, err := fixture.GetNodeController("Node1")
+	a.NoError(err)
+	// Get secondary node
+	secondNode, err := fixture.GetNodeController("Node2")
+	a.NoError(err)
+	// Get the relay
+	relayNode, err := fixture.GetNodeController("Relay")
+	a.NoError(err)
+
+	// prepare it's configuration file to set it to generate a catchpoint every 16 rounds.
+	cfg, err := config.LoadConfigFromDisk(primaryNode.GetDataDir())
+	a.NoError(err)
+	const catchpointInterval = 16
+	cfg.CatchpointInterval = catchpointInterval
+	cfg.CatchpointTracking = 2
+	cfg.MaxAcctLookback = 2
+	cfg.Archival = false
+
+	cfg.TxSyncIntervalSeconds = 200000 // disable txSync
+
+	cfg.SaveToDisk(primaryNode.GetDataDir())
+	cfg.SaveToDisk(secondNode.GetDataDir())
+
+	cfg, err = config.LoadConfigFromDisk(relayNode.GetDataDir())
+	a.NoError(err)
+	cfg.TxSyncIntervalSeconds = 200000 // disable txSync
+	cfg.SaveToDisk(relayNode.GetDataDir())
+
+	fixture.Start()
+	defer fixture.LibGoalFixture.Shutdown()
+
+	client1 := fixture.GetLibGoalClientFromNodeController(primaryNode)
+	client2 := fixture.GetLibGoalClientFromNodeController(secondNode)
+	wallet1, err := client1.GetUnencryptedWalletHandle()
+	a.NoError(err)
+	wallet2, err := client2.GetUnencryptedWalletHandle()
+	a.NoError(err)
+	addrs1, err := client1.ListAddresses(wallet1)
+	a.NoError(err)
+	addrs2, err := client2.ListAddresses(wallet2)
+	a.NoError(err)
+
+	// let the second node have insufficient stake for proposing a block
+	tx, err := client2.SendPaymentFromUnencryptedWallet(addrs2[0], addrs1[0], 1000, 4999999999000000, nil)
+	a.NoError(err)
+	status, err := client1.Status()
+	a.NoError(err)
+	_, err = fixture.WaitForConfirmedTxn(status.LastRound+100, addrs1[0], tx.ID().String())
+	a.NoError(err)
+	targetCatchpointRound := status.LastRound
+
+	// ensure the catchpoint is created for targetCatchpointRound
+	timer := time.NewTimer(100 * time.Second)
+outer:
+	for {
+		status, err = client1.Status()
+		a.NoError(err)
+
+		var round basics.Round
+		if status.LastCatchpoint != nil && len(*status.LastCatchpoint) > 0 {
+			round, _, err = ledgercore.ParseCatchpointLabel(*status.LastCatchpoint)
+			a.NoError(err)
+			if uint64(round) >= targetCatchpointRound {
+				break
+			}
+		}
+		select {
+		case <-timer.C:
+			a.Failf("timeout waiting a catchpoint", "target: %d, got %d", targetCatchpointRound, round)
+			break outer
+		default:
+			time.Sleep(250 * time.Millisecond)
+		}
+	}
+
+	// let the primary node catchup
+	err = client1.Catchup(*status.LastCatchpoint)
+	a.NoError(err)
+
+	status1, err := client1.Status()
+	a.NoError(err)
+	targetRound := status1.LastRound + 5
+
+	// Wait for the network to start making progress again
+	primaryNodeRestClient := fixture.GetAlgodClientForController(primaryNode)
+	err = fixture.ClientWaitForRound(primaryNodeRestClient, targetRound,
+		10*catchpointCatchupProtocol.AgreementFilterTimeout)
+	a.NoError(err)
+
+	// let the 2nd client send a transaction
+	tx, err = client2.SendPaymentFromUnencryptedWallet(addrs2[0], addrs1[0], 1000, 50000, nil)
+	a.NoError(err)
+
+	status, err = client2.Status()
+	a.NoError(err)
+	_, err = fixture.WaitForConfirmedTxn(status.LastRound+50, addrs2[0], tx.ID().String())
+	a.NoError(err)
+}