Assembler synchronizer integration

config/config.go

@@ -695,6 +695,26 @@ func (config *Configuration) ExtractConsenters() []nodeconfig.ConsenterInfo {
 	return consenters
 }
 
+func (config *Configuration) ExtractAssemblers() []nodeconfig.AssemblerInfo {
+	var assemblers []nodeconfig.AssemblerInfo
+	for _, party := range config.SharedConfig.PartiesConfig {
+		var tlsCACertsCollection []nodeconfig.RawBytes
+		for _, ca := range party.TLSCACerts {
+			tlsCACertsCollection = append(tlsCACertsCollection, ca)
+		}
+
+		assemblerInfo := nodeconfig.AssemblerInfo{
+			PartyID:    types.PartyID(party.PartyID),
+			Endpoint:   net.JoinHostPort(party.AssemblerConfig.Host, strconv.Itoa(int(party.AssemblerConfig.Port))),
+			TLSCACerts: tlsCACertsCollection,
+			TLSCert:    party.AssemblerConfig.TlsCert,
+		}
+		assemblers = append(assemblers, assemblerInfo)
+	}
+
+	return assemblers
+}
+
 func (config *Configuration) ExtractRouterInParty() nodeconfig.RouterInfo {
 	partyID := config.LocalConfig.NodeLocalConfig.PartyID
 	var party *ordererpb.PartyConfig
@@ -1050,3 +1070,30 @@ func ExtractConsenterAddresses(ordererConfig channelconfig.Orderer) (orderers.Pa
 
 	return party2Endpoint, nil
 }
+
+func ExtractAssemblerAddresses(ordererConfig channelconfig.Orderer) (orderers.Party2Endpoint, error) {
+	consensusMeta := ordererConfig.ConsensusMetadata()
+	sharedConfig := &ordererpb.SharedConfig{}
+	if err := proto.Unmarshal(consensusMeta, sharedConfig); err != nil {
+		return nil, errors.Wrap(err, "failed to unmarshal consensus metadata")
+	}
+
+	conf := &Configuration{
+		SharedConfig: sharedConfig,
+	}
+
+	aInfo := conf.ExtractAssemblers()
+
+	party2Endpoint := make(orderers.Party2Endpoint)
+
+	for _, assembler := range aInfo {
+		party2Endpoint[assembler.PartyID] = &orderers.Endpoint{
+			Address: assembler.Endpoint,
+		}
+		for _, cert := range assembler.TLSCACerts {
+			party2Endpoint[assembler.PartyID].RootCerts = append(party2Endpoint[assembler.PartyID].RootCerts, cert)
+		}
+	}
+
+	return party2Endpoint, nil
+}

node/assembler/assembler.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/hyperledger/fabric-lib-go/bccsp/factory"
 	"github.com/hyperledger/fabric-lib-go/common/flogging"
 	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	"github.com/hyperledger/fabric-protos-go-apiv2/orderer"
@@ -19,6 +20,7 @@ import (
 	"github.com/hyperledger/fabric-x-orderer/common/operations"
 	common_utils "github.com/hyperledger/fabric-x-orderer/common/utils"
 	"github.com/hyperledger/fabric-x-orderer/config"
+	"github.com/hyperledger/fabric-x-orderer/node/assembler/synchronizer"
 	node_config "github.com/hyperledger/fabric-x-orderer/node/config"
 	"github.com/hyperledger/fabric-x-orderer/node/consensus/state"
 	"github.com/hyperledger/fabric-x-orderer/node/delivery"
@@ -60,6 +62,8 @@ func (a *Assembler) Deliver(server orderer.AtomicBroadcast_DeliverServer) error
 
 // GetTxCount returns the number of transactions the assembler stored in the ledger. This method is used only in testing.
 func (a *Assembler) GetTxCount() uint64 {
+	a.lock.RLock()
+	defer a.lock.RUnlock()
 	return a.collator.Ledger.(node_ledger.AssemblerLedgerReaderWriter).GetTxCount()
 }
 
@@ -125,6 +129,7 @@ func NewDefaultAssembler(
 	batchBringerFactory BatchBringerFactory,
 	consensusBringerFactory delivery.ConsensusBringerFactory,
 	signer identity.SignerSerializer,
+	synchronizerFactory synchronizer.SynchronizerFactory,
 ) *Assembler {
 	logger.Infof("Creating assembler, party: %d, address: %s", nodeConfig.PartyId, nodeConfig.ListenAddress)
 	if configBlock == nil {
@@ -142,15 +147,19 @@ func NewDefaultAssembler(
 	}
 
 	assembler := &Assembler{
-		logger:       logger,
-		ledger:       ledger,
-		mainExitChan: mainExitChan,
-		status:       utils.NodeStatus{},
-		metrics:      NewMetrics(nodeConfig, ledger.Metrics(), logger),
-		signer:       signer,
+		logger:              logger,
+		ledger:              ledger,
+		mainExitChan:        mainExitChan,
+		status:              utils.NodeStatus{},
+		metrics:             NewMetrics(nodeConfig, ledger.Metrics(), logger),
+		signer:              signer,
+		assemblerNodeConfig: nodeConfig,
+		configuration:       configuration,
 	}
 
-	assembler.initLedger(configBlock)
+	assembler.initLedger(configBlock, nodeConfig, synchronizerFactory)
+
+	// todo - take last config block from ledger and bootstrap the assembler with it
 
 	assembler.initFromConfig(net, nodeConfig, configuration, configBlock, prefetchIndexFactory, prefetcherFactory, batchBringerFactory, consensusBringerFactory)
 
@@ -226,43 +235,69 @@ func (a *Assembler) initFromConfig(
 	a.logger.Infof("Assembler initialized successfully with config sequence number: %d", configSequence)
 }
 
-func (a *Assembler) initLedger(configBlock *common.Block) {
-	var transactionCount, blocksCount uint64
-	height := a.ledger.LedgerReader().Height()
-	if height > 0 {
-		block, err := a.ledger.LedgerReader().RetrieveBlockByNumber(height - 1)
+func (a *Assembler) initLedger(configBlock *common.Block, nodeConfig *node_config.AssemblerNodeConfig, synchronizerFactory synchronizer.SynchronizerFactory) {
+	ledgerHeight := a.ledger.LedgerReader().Height()
+	configBlockNumber := configBlock.GetHeader().GetNumber()
+	a.logger.Infof("Initializing assembler ledger, current height: %d, config block number: %d", ledgerHeight, configBlockNumber)
+
+	if configBlockNumber == 0 && ledgerHeight == 0 {
+		// append config block only if it is the genesis block and the ledger is empty.
+		configBlock.Metadata.Metadata[common.BlockMetadataIndex_ORDERER] = common_utils.GenesisBlockMetadataBytes()
+		ordInfo := &state.OrderingInformation{
+			CommonBlock: configBlock,
+			DecisionNum: 0,
+			BatchIndex:  0,
+			BatchCount:  1,
+		}
+		a.ledger.AppendConfig(ordInfo)
+		a.logger.Infof("Appended genesis block, header digest: %s", hex.EncodeToString(protoutil.BlockHeaderHash(configBlock.GetHeader())))
+	} else if configBlockNumber > ledgerHeight {
+		// genesis block is block number 0, so if config block number is higher than ledger height, it means the assembler needs to sync to the config block before starting.
+		a.logger.Warnf("Config block number %d is higher than current ledger height %d, assembler will need to sync to the config block", configBlockNumber, ledgerHeight)
+		a.logger.Infof("Creating assembler synchronizer to sync to config block number %d, current ledger height is %d", configBlockNumber, ledgerHeight)
+		targetHeight := configBlockNumber + 1
+		synchronizer := synchronizerFactory.CreateSynchronizer(
+			a.logger,
+			uint64(nodeConfig.PartyId),
+			config.Cluster{SendBufferSize: 100, ClientCertificate: nodeConfig.TLSCertificateFile, ClientPrivateKey: nodeConfig.TLSPrivateKeyFile, ReplicationPolicy: "assemblerSync"},
+			&AssemblerSupportAdapter{assembler: a},
+			factory.GetDefault(),
+			targetHeight,
+			configBlock,
+		)
+
+		err := synchronizer.Sync()
+		if err != nil {
+			a.logger.Panicf("error while syncing to config block %v", err)
+		}
+	} else {
+		// todo - should panic?
+		a.logger.Infof("Assembler is starting with config block already in the ledger, config block number: %d, current ledger height: %d", configBlockNumber, ledgerHeight)
+	}
+
+	ledgerHeight = a.ledger.LedgerReader().Height()
+	if ledgerHeight == 0 {
+		a.logger.Panicf("Assembler ledger is empty after initialization, this should not happen")
+	}
+
+	// update metrics with current ledger state only if we didn't just append the genesis block
+	// (if we just appended genesis block, metrics were already updated in AppendConfig)
+	// This happens when the assembler is restarting with an existing ledger or after syncing
+	if ledgerHeight > 1 || configBlockNumber > 0 {
+		var transactionCount uint64
+		block, err := a.ledger.LedgerReader().RetrieveBlockByNumber(ledgerHeight - 1)
 		if err != nil {
 			a.logger.Panicf("error while fetching last block from ledger %v", err)
 		}
 		_, _, transactionCount, err = node_ledger.AssemblerBatchIdOrderingInfoAndTxCountFromBlock(block)
 		if err != nil {
 			a.logger.Panicf("error while fetching last block ordering info %v", err)
 		}
-
-		blocksCount = height
-	}
-	a.ledger.Metrics().TransactionCount.Add(float64(transactionCount))
-	a.ledger.Metrics().BlocksCount.Add(float64(blocksCount))
-
-	a.logger.Infof("Starting with ledger height: %d", a.ledger.LedgerReader().Height())
-
-	if a.ledger.LedgerReader().Height() == 0 {
-		// append config block only if it is the genesis block
-		blockNumber := configBlock.GetHeader().Number
-		if blockNumber == 0 {
-			configBlock.Metadata.Metadata[common.BlockMetadataIndex_ORDERER] = common_utils.GenesisBlockMetadataBytes()
-			ordInfo := &state.OrderingInformation{
-				CommonBlock: configBlock,
-				DecisionNum: 0,
-				BatchIndex:  0,
-				BatchCount:  1,
-			}
-			a.ledger.AppendConfig(ordInfo)
-			a.logger.Infof("Appended genesis block, header digest: %s", hex.EncodeToString(protoutil.BlockHeaderHash(configBlock.GetHeader())))
-		} else {
-			a.logger.Infof("Assembler started with non-genesis config block, block number: %d", blockNumber)
-		}
+		a.ledger.Metrics().TransactionCount.Add(float64(transactionCount))
+		a.ledger.Metrics().BlocksCount.Add(float64(ledgerHeight))
 	}
+
+	a.logger.Infof("Ledger was initialized, current ledger height: %d", ledgerHeight)
 }
 
 func NewAssembler(nodeConfig *node_config.AssemblerNodeConfig, configuration *config.Configuration, configBlock *common.Block, mainExitChan chan struct{}, logger *flogging.FabricLogger, signer identity.SignerSerializer) *Assembler {
@@ -279,6 +314,7 @@ func NewAssembler(nodeConfig *node_config.AssemblerNodeConfig, configuration *co
 		&DefaultBatchBringerFactory{},
 		&delivery.DefaultConsensusBringerFactory{},
 		signer,
+		&synchronizer.SynchronizerCreator{},
 	)
 }
 

node/assembler/assembler_sync_support.go

@@ -12,6 +12,7 @@ import (
 	"github.com/hyperledger/fabric-protos-go-apiv2/common"
 	"github.com/hyperledger/fabric-x-common/common/channelconfig"
 	"github.com/hyperledger/fabric-x-common/protoutil"
+	"github.com/hyperledger/fabric-x-orderer/node/comm"
 )
 
 // AssemblerSupportAdapter adapts an Assembler to satisfy the
@@ -77,6 +78,12 @@ func (a *AssemblerSupportAdapter) Block(number uint64) *common.Block {
 	return block
 }
 
+// ClientConfig returns the TLS client configuration for the assembler, used by
+// the synchronizer to dial other assembler nodes.
+func (a *AssemblerSupportAdapter) ClientConfig() comm.ClientConfig {
+	return extractClientConfigFromAssemblerConfig(a.assembler.assemblerNodeConfig, a.assembler.configuration)
+}
+
 // LastConfigBlock returns the most recent config block at or before the given block,
 // or an error if it cannot be retrieved.
 func (a *AssemblerSupportAdapter) LastConfigBlock(block *common.Block) (*common.Block, error) {

node/assembler/assembler_test.go

@@ -21,6 +21,7 @@ import (
 	orderer_config "github.com/hyperledger/fabric-x-orderer/config"
 	"github.com/hyperledger/fabric-x-orderer/node/assembler"
 	assembler_mocks "github.com/hyperledger/fabric-x-orderer/node/assembler/mocks"
+	synchronizer_mocks "github.com/hyperledger/fabric-x-orderer/node/assembler/synchronizer/mocks"
 	"github.com/hyperledger/fabric-x-orderer/node/config"
 	"github.com/hyperledger/fabric-x-orderer/node/consensus/state"
 	"github.com/hyperledger/fabric-x-orderer/node/delivery"
@@ -52,6 +53,7 @@ type assemblerTest struct {
 	prefetcherMock                 *assembler_mocks.FakePrefetcherController
 	prefetchIndexMock              *assembler_mocks.FakePrefetchIndexer
 	consensusBringerMock           *delivery_mocks.FakeConsensusBringer
+	synchronizerFactoryMock        *synchronizer_mocks.FakeSynchronizerFactory
 }
 
 type dummyAssemblerStopper struct{}
@@ -83,7 +85,9 @@ func setupAssemblerTest(t *testing.T, shards []types.ShardID, parties []types.Pa
 		prefetcherMock:                 &assembler_mocks.FakePrefetcherController{},
 		prefetchIndexMock:              &assembler_mocks.FakePrefetchIndexer{},
 		consensusBringerMock:           &delivery_mocks.FakeConsensusBringer{},
+		synchronizerFactoryMock:        &synchronizer_mocks.FakeSynchronizerFactory{},
 	}
+	test.synchronizerFactoryMock.CreateSynchronizerReturns(&synchronizer_mocks.FakeSynchronizerWithStop{})
 	assemblerEndpoint := ""
 	consenterEndpoint := "consenter"
 
@@ -214,6 +218,7 @@ func (at *assemblerTest) StartAssembler() {
 		batchBringerFactoryMock,
 		consensusBringerFactoryMock,
 		&mocks.SignerSerializer{},
+		at.synchronizerFactoryMock,
 	)
 
 	at.assembler.StartAssemblerService()
@@ -269,7 +274,7 @@ func TestAssembler_StartAndThenStopShouldOnlyWriteGenesisBlockToLedger(t *testin
 	al.Close()
 }
 
-func TestAssembler_SkipNonGenesisConfigBlock(t *testing.T) {
+func TestAssembler_PanicNonGenesisConfigBlock(t *testing.T) {
 	// Arrange
 	shards := []types.ShardID{1, 2}
 	parties := []types.PartyID{1, 2, 3}
@@ -278,21 +283,9 @@ func TestAssembler_SkipNonGenesisConfigBlock(t *testing.T) {
 
 	test := setupAssemblerTest(t, shards, parties, parties[0], configBlock)
 
-	// Act
-	test.StartAssembler()
-	test.WaitAssemblerRunning(t)
-
-	test.StopAssembler()
-	test.WaitAssemblerStopped(t)
-
-	// Assert
-	al, err := node_ledger.NewAssemblerLedger(test.logger, test.ledgerDir)
-	require.NoError(t, err)
-	require.Equal(t, uint64(0), al.Ledger.Height())
-	genesisBlock, err := al.Ledger.RetrieveBlockByNumber(blockNumber)
-	require.ErrorContains(t, err, "no such block number")
-	require.Nil(t, genesisBlock)
-	al.Close()
+	// Act & Assert: starting with a non-genesis config block and an empty ledger should panic
+	// because the assembler cannot operate without first syncing the missing blocks.
+	require.Panics(t, func() { test.StartAssembler() })
 }
 
 func TestAssembler_RestartWithoutAddingBatchesShouldWork(t *testing.T) {

node/assembler/synchronizer/block_verifier.go

@@ -0,0 +1,63 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package synchronizer
+
+import (
+	"github.com/hyperledger/fabric-lib-go/bccsp"
+	"github.com/hyperledger/fabric-lib-go/common/flogging"
+	"github.com/hyperledger/fabric-protos-go-apiv2/common"
+	"github.com/hyperledger/fabric-x-orderer/common/deliverclient"
+)
+
+//go:generate counterfeiter -o mocks/verifier_factory.go --fake-name VerifierFactory . VerifierFactory
+type VerifierFactory interface {
+	CreateBlockVerifier(
+		configBlock *common.Block,
+		lastBlock *common.Block,
+		cryptoProvider bccsp.BCCSP,
+		lg *flogging.FabricLogger,
+	) (deliverclient.CloneableUpdatableBlockVerifier, error)
+}
+
+// noopVerifierCreator creates a block verifier that does not actually verify blocks, which can be used in tests or when block verification is not needed.
+type noopVerifierCreator struct{}
+
+func (*noopVerifierCreator) CreateBlockVerifier(
+	configBlock *common.Block,
+	lastBlock *common.Block,
+	cryptoProvider bccsp.BCCSP,
+	lg *flogging.FabricLogger,
+) (deliverclient.CloneableUpdatableBlockVerifier, error) {
+	return &noopBlockVerifier{}, nil
+}
+
+// noopBlockVerifier is a block verifier that does not actually verify blocks, which can be used in tests or when block verification is not needed.
+type noopBlockVerifier struct{}
+
+// VerifyBlock checks block integrity and its relation to the chain, and verifies the signatures.
+func (*noopBlockVerifier) VerifyBlock(block *common.Block) error {
+	// TODO
+	return nil
+}
+
+func (*noopBlockVerifier) VerifyBlockAttestation(block *common.Block) error {
+	// TODO
+	return nil
+}
+
+func (*noopBlockVerifier) UpdateConfig(configBlock *common.Block) error {
+	// TODO
+	return nil
+}
+
+func (*noopBlockVerifier) UpdateBlockHeader(block *common.Block) {
+	// TODO
+}
+
+func (*noopBlockVerifier) Clone() deliverclient.CloneableUpdatableBlockVerifier {
+	return &noopBlockVerifier{}
+}