// SyncTransactions is an infinite loop which constantly sends transaction ids from its index
// into a Cuckoo Filter to randomly sampled number of peers and adds to it's state all received
// transactions.
SyncTransactions is an infinite loop which constantly sends transaction ids from its index
into a Cuckoo Filter to randomly sampled number of peers and adds to it's state all received
transactions.
// SyncTransactions is an infinite loop which constantly sends transaction ids from its index // into a Cuckoo Filter to randomly sampled number of peers and adds to it ' s state all received // transactions .
func (l *Ledger) SyncTransactions() { 
	defer l.consensus.Done()
	for {
		select {
		case <-time.After(1 * time.Second):
		case <-l.consensusStop:
			return
		}
		snowballK := conf.GetSnowballK()
		peers, err := SelectPeers(l.client.ClosestPeers(), snowballK)
		if err != nil {
			continue
		}
		logger := log.Sync("sync_tx")
		l.transactionFilterLock.RLock()
		cf := l.transactionFilter.MarshalBinary()
		l.transactionFilterLock.RUnlock()
		if err != nil {
			logger.Error().Err(err).Msg("failed to marshal set membership filter data")
			continue
		}
		snapshot := l.Snapshot()
		var wg sync.WaitGroup
		bfReq := &TransactionsSyncRequest{
			Data: &TransactionsSyncRequest_Filter{
				Filter: cf,
			},
		}
		for _, p := range peers {
			wg.Add(1)
			go func(conn *grpc.ClientConn) {
				defer wg.Done()
				ctx, cancel := context.WithTimeout(context.Background(), conf.GetDownloadTxTimeout())
				defer cancel()
				stream, err := NewWaveletClient(conn).SyncTransactions(ctx)
				if err != nil {
					logger.Error().Err(err).Msg("failed to create sync transactions stream")
					return
				}
				defer func() {
					if err := stream.CloseSend(); err != nil {
						logger.Error().Err(err).Msg("failed to close sync transaction stream")
					}
				}()
				if err := stream.Send(bfReq); err != nil {
					logger.Error().Err(err).Msg("failed to send set membership filter data")
					return
				}
				res, err := stream.Recv()
				if err != nil {
					logger.Error().Err(err).Msg("failed to receive sync transactions header")
					return
				}
				count := res.GetTransactionsNum()
				if count == 0 {
					return
				}
				if count > conf.GetTXSyncLimit() {
					logger.Debug().
						Uint64("count", count).
						Str("peer_address", conn.Target()).
						Msg("Bad number of transactions would be received")
					return
				}
				logger.Debug().
					Uint64("count", count).
					Msg("Requesting transaction(s) to sync.")
				for count > 0 {
					req := TransactionsSyncRequest_ChunkSize{
						ChunkSize: conf.GetTXSyncChunkSize(),
					}
					if count < req.ChunkSize {
						req.ChunkSize = count
					}
					if err := stream.Send(&TransactionsSyncRequest{Data: &req}); err != nil {
						logger.Error().Err(err).Msg("failed to send sync transactions request")
						return
					}
					res, err := stream.Recv()
					if err != nil {
						logger.Error().Err(err).Msg("failed to receive sync transactions response")
						return
					}
					txResponse := res.GetTransactions()
					if txResponse == nil {
						return
					}
					transactions := make([]Transaction, 0, len(txResponse.Transactions))
					for _, txBody := range txResponse.Transactions {
						tx, err := UnmarshalTransaction(bytes.NewReader(txBody))
						if err != nil {
							logger.Error().Err(err).Msg("failed to unmarshal synced transaction")
							continue
						}
						if err := ValidateTransaction(snapshot, tx); err != nil && err != ErrContractAlreadyExists {
							logger.Error().
								Err(err).
								Hex("tx_id", tx.ID[:]).
								Msg("transaction validation error")
							continue
						}
						transactions = append(transactions, tx)
					}
					downloadedNum := len(transactions)
					if downloadedNum == 0 {
						logger.Warn().
							Uint64("transaction_to_sync", count).
							Msg("No transactions to add while there are still missing transactions")
						return
					}
					count -= uint64(downloadedNum)
					l.AddTransaction(transactions...)
					l.metrics.downloadedTX.Mark(int64(downloadedNum))
					l.metrics.receivedTX.Mark(int64(downloadedNum))
				}
			}(p.Conn())
		}
		wg.Wait()
	}
}
func ( l * Ledger ) SyncTransactions ( ) { defer l . consensus . Done ( ) for { select { case <- time . After ( 1 * time . Second ) : case <- l . consensusStop : return } snowballK := conf . GetSnowballK ( ) peers , err := SelectPeers ( l . client . ClosestPeers ( ) , snowballK ) if err != nil { continue } logger := log . Sync ( " " ) l . transactionFilterLock . RLock ( ) cf := l . transactionFilter . MarshalBinary ( ) l . transactionFilterLock . RUnlock ( ) if err != nil { logger . Error ( ) . Err ( err ) . Msg ( " " ) continue } snapshot := l . Snapshot ( ) var wg sync . WaitGroup bfReq := & TransactionsSyncRequest { Data : & TransactionsSyncRequest_Filter { Filter : cf , } , } for _ , p := range peers { wg . Add ( 1 ) go func ( conn * grpc . ClientConn ) { defer wg . Done ( ) ctx , cancel := context . WithTimeout ( context . Background ( ) , conf . GetDownloadTxTimeout ( ) ) defer cancel ( ) stream , err := NewWaveletClient ( conn ) . SyncTransactions ( ctx ) if err != nil { logger . Error ( ) . Err ( err ) . Msg ( " " ) return } defer func ( ) { if err := stream . CloseSend ( ) ; err != nil { logger . Error ( ) . Err ( err ) . Msg ( " " ) } } ( ) if err := stream . Send ( bfReq ) ; err != nil { logger . Error ( ) . Err ( err ) . Msg ( " " ) return } res , err := stream . Recv ( ) if err != nil { logger . Error ( ) . Err ( err ) . Msg ( " " ) return } count := res . GetTransactionsNum ( ) if count == 0 { return } if count > conf . GetTXSyncLimit ( ) { logger . Debug ( ) . Uint64 ( " " , count ) . Str ( " " , conn . Target ( ) ) . Msg ( " " ) return } logger . Debug ( ) . Uint64 ( " " , count ) . Msg ( " " ) for count > 0 { req := TransactionsSyncRequest_ChunkSize { ChunkSize : conf . GetTXSyncChunkSize ( ) , } if count < req . ChunkSize { req . ChunkSize = count } if err := stream . Send ( & TransactionsSyncRequest { Data : & req } ) ; err != nil { logger . Error ( ) . Err ( err ) . Msg ( " " ) return } res , err := stream . Recv ( ) if err != nil { logger . Error ( ) . Err ( err ) . Msg ( " " ) return } txResponse := res . GetTransactions ( ) if txResponse == nil { return } transactions := make ( [ ] Transaction , 0 , len ( txResponse . Transactions ) ) for _ , txBody := range txResponse . Transactions { tx , err := UnmarshalTransaction ( bytes . NewReader ( txBody ) ) if err != nil { logger . Error ( ) . Err ( err ) . Msg ( " " ) continue } if err := ValidateTransaction ( snapshot , tx ) ; err != nil && err != ErrContractAlreadyExists { logger . Error ( ) . Err ( err ) . Hex ( " " , tx . ID [ : ] ) . Msg ( " " ) continue } transactions = append ( transactions , tx ) } downloadedNum := len ( transactions ) if downloadedNum == 0 { logger . Warn ( ) . Uint64 ( " " , count ) . Msg ( " " ) return } count -= uint64 ( downloadedNum ) l . AddTransaction ( transactions ... ) l . metrics . downloadedTX . Mark ( int64 ( downloadedNum ) ) l . metrics . receivedTX . Mark ( int64 ( downloadedNum ) ) } } ( p . Conn ( ) ) } wg . Wait ( ) } }
SyncTransactions is an infinite loop which constantly sends transaction ids from its index
into a Cuckoo Filter to randomly sampled number of peers and adds to it's state all received
transactions.
SyncTransactions is an infinite loop which constantly sends transaction ids from its index into a Cuckoo Filter to randomly sampled number of peers and adds to it ' s state all received transactions .
// nolint:gocognit