Batcher restart recover by gengur · Pull Request #146 · hyperledger/fabric-x-orderer

gengur · 2025-07-28T12:39:59Z

Integration tests - Batcher restart and recover
issue #84

test/batcher_test.go

@@ -0,0 +1,366 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+SPDX-License-Identifier: Apache-2.0


test/batcher_test.go

+Copyright IBM Corp. All Rights Reserved.
+SPDX-License-Identifier: Apache-2.0
+*/
+package test


test/batcher_test.go

+
+const (
+	// Number of shards in the test
+	numOfShards = 4


test/batcher_test.go

+	// Number of parties in the test
+	numOfParties = 4
+	// Placeholder for a value that doesn't matter in the test
+	doesntMatter = -1


test/utils_test.go

 			require.ErrorContains(t, err, errString)
-			require.Equal(t, uint64(transactions), totalTxs)
-			require.Equal(t, uint64(blocks), totalBlocks)
+			require.LessOrEqual(t, uint64(transactions), totalTxs)


testutil/utils.go

 	}
 }

+func WaitForAssemblersGotAtLeast(t *testing.T, armaNetwork *ArmaNetwork, parties []types.PartyID, waitFor int, duration time.Duration) {


test/batcher_test.go

+	doesntMatter = -1
+)
+
+func TestPrimaryBatcherRestartRecover(t *testing.T) {


test/batcher_test.go

+	PullFromAssemblers(t, uc, parties, 0, math.MaxUint64, totalTxSent, doesntMatter, "cancelled pull from assembler: %d")
+}
+
+func TestSecondaryBatcherRestartRecover(t *testing.T) {


testutil/client/tx_broadcast_client.go

 	"github.com/hyperledger/fabric-x-orderer/node/comm"
 )

+type StreamInfo struct {


test/batcher_test.go

+	// Pull from Assemblers
+	PullFromAssemblers(t, uc, parties, 0, math.MaxUint64, totalTxSent, doesntMatter, "cancelled pull from assembler: %d")
+
+	partyToRestart := types.PartyID(3)


test/batcher_test.go

+	// make sure clients of correct parties continue to get transactions (expect 2000 TXs).
+	PullFromAssemblers(t, uc, correctParties, 0, math.MaxUint64, totalTxSent, doesntMatter, "cancelled pull from assembler: %d")
+
+	// for shardId := 2; shardId <= 2; shardId++ {


test/utils_test.go

+func PullFromAssembler(t *testing.T, userConfig *armageddon.UserConfig, partyID types.PartyID, startBlock uint64, endBlock uint64, transactions int, blocks int) (uint64, uint64, []types.BatchID, error) {
 	require.NotNil(t, userConfig)
 	dc := client.NewDeliverClient(userConfig)
-	toCtx, toCancel := context.WithTimeout(context.Background(), 30*time.Second)


test/utils_test.go

-			require.Equal(t, uint64(transactions), totalTxs)
-			require.Equal(t, uint64(blocks), totalBlocks)
+			// TODO:  check that we get all of the submitted transactions
+			require.LessOrEqual(t, uint64(transactions), totalTxs)


test/utils_test.go

+			// TODO:  check that we get all of the submitted transactions
+			require.LessOrEqual(t, uint64(transactions), totalTxs)
+			if blocks > 0 {
+				require.LessOrEqual(t, uint64(blocks), totalBlocks)


test/batcher_test.go

+
+const (
+	// Number of shards in the test
+	numOfShards = 1


test/batcher_test.go

+	partyBlockInfos := blockInfos[types.PartyID(1)]
+	primaryBatcherId := partyBlockInfos[len(partyBlockInfos)-1].Primary()
+	primaryBatcher := armaNetwork.GeBatcher(t, primaryBatcherId, types.ShardID(1))
+	correctParties := []types.PartyID{}


test/batcher_test.go

+		}
+	}
+
+	// test that the router of party get stalled in the some point


test/batcher_test.go

+	totalTxSent += totalTxNumber
+
+	// 5.
+	// make sure clients of correct parties continue to get transactions (expect 2000 TXs).


HagarMeir · 2025-08-06T20:26:20Z

test/batcher_test.go

+		txContent := prepareTx(i, 64, []byte("sessionNumber"))
+		err = broadcastClient.SendTx(txContent)
+		if err != nil {
+			require.ErrorContains(t, err, fmt.Sprintf("received error response from %s: INTERNAL_SERVER_ERROR", routerToStall.Listener.Addr().String())) // only such errors are permitted


should this still happen? maybe check there is no error

Yes, it may happen since it takes time to renew faulty streams.

test/batcher_test.go

+	totalTxSent += totalTxNumber
+
+	// Pull from Assemblers
+	// make sure clients of all the parties get transactions (expect 3000 TXs).


test/utils_test.go

-			require.Equal(t, uint64(blocks), totalBlocks)
+			// TODO:  check that we get all of the submitted transactions
+			require.GreaterOrEqual(t, pullInfo.TotalTxs, uint64(transactions))
+			if blocks > 0 {


HagarMeir · 2025-08-10T06:07:51Z

test/utils_test.go

+
+		if shardID != types.ShardIDConsensus {
+			pID := PrimaryID{ShardID: shardID}
+			primaryIDBytes := block.GetMetadata().GetMetadata()[common.BlockMetadataIndex_ORDERER][0:2]


why are you using this and not the AssemblerBatchIdOrderingInfoAndTxCountFromBlock func?

we don't need all the stuff that returns it

HagarMeir · 2025-08-10T06:15:07Z

test/utils_test.go

 			return errors.New("nil block header")
 		}
+
+		atomic.AddUint64(&totalBlocks, uint64(1))


how come this needs to be atomic but there is no lock over the map?
I guess you can remove this atomic add?

no need it's legacy

HagarMeir · 2025-08-10T06:17:48Z

test/utils_test.go

+	ShardID types.ShardID
+}
+
+func (p *PrimaryID) Digest() types.ShardID {


what is this for?

No need, shardID is a comparable

HagarMeir · 2025-08-10T06:18:06Z

test/utils_test.go

 }

-func PullFromAssemblers(t *testing.T, userConfig *armageddon.UserConfig, parties []types.PartyID, startBlock uint64, endBlock uint64, transactions int, blocks int, errString string) {
+type PrimaryID struct {


why create a type for this? just use shardID

You are right, shardID is comparable Fixed

Signed-off-by: Genady Gurevich <genadyg@il.ibm.com>

HagarMeir · 2025-08-10T12:32:00Z

@gengur you still have a TODO for next PR to check that all transactions actually got in

gengur requested review from HagarMeir and tock-ibm July 28, 2025 12:39

gengur marked this pull request as draft July 28, 2025 12:40

gengur force-pushed the batcher_restart_recover branch 10 times, most recently from 59dada3 to 80448d9 Compare August 4, 2025 14:37

gengur marked this pull request as ready for review August 4, 2025 15:59