// PullMissingTransactions is a goroutine which continuously pulls missing transactions
// from randomly sampled peers in the network. It does so by sending a list of
// transaction IDs which node is missing.
PullMissingTransactions is a goroutine which continuously pulls missing transactions
from randomly sampled peers in the network. It does so by sending a list of
transaction IDs which node is missing.
// PullMissingTransactions is a goroutine which continuously pulls missing transactions // from randomly sampled peers in the network . It does so by sending a list of // transaction IDs which node is missing .
func (l *Ledger) PullMissingTransactions() {
	defer l.consensus.Done()
	for {
		select {
		case <-time.After(100 * time.Millisecond):
		case <-l.consensusStop:
			return
		}
		snowballK := conf.GetSnowballK()
		peers, err := SelectPeers(l.client.ClosestPeers(), snowballK)
		if err != nil {
			continue
		}
		logger := log.Sync("pull_missing_tx")
		missingIDs := l.transactions.MissingIDs()
		missingTxPullLimit := conf.GetMissingTxPullLimit()
		if uint64(len(missingIDs)) > missingTxPullLimit {
			missingIDs = missingIDs[:missingTxPullLimit]
		}
		req := &TransactionPullRequest{TransactionIds: make([][]byte, 0, len(missingIDs))}
		for _, txID := range missingIDs {
			txID := txID
			req.TransactionIds = append(req.TransactionIds, txID[:])
		}
		type response struct {
			txs []Transaction
		}
		responseChan := make(chan response)
		for _, p := range peers {
			go func(conn *grpc.ClientConn) {
				var response response
				defer func() {
					responseChan <- response
				}()
				client := NewWaveletClient(conn)
				ctx, cancel := context.WithTimeout(context.Background(), conf.GetDownloadTxTimeout())
				defer cancel()
				batch, err := client.PullTransactions(ctx, req)
				if err != nil {
					logger.Error().Err(err).Msg("failed to download missing transactions")
					return
				}
				response.txs = make([]Transaction, 0, len(batch.Transactions))
				for _, buf := range batch.Transactions {
					tx, err := UnmarshalTransaction(bytes.NewReader(buf))
					if err != nil {
						logger.Error().
							Err(err).
							Hex("tx_id", tx.ID[:]).
							Msg("error unmarshaling downloaded tx")
						continue
					}
					response.txs = append(response.txs, tx)
				}
			}(p.Conn())
		}
		count := int64(0)
		pulled := make(map[TransactionID]Transaction)
		for i := 0; i < len(peers); i++ {
			res := <-responseChan
			for i := range res.txs {
				if _, ok := pulled[res.txs[i].ID]; !ok {
					pulled[res.txs[i].ID] = res.txs[i]
					count += int64(res.txs[i].LogicalUnits())
				}
			}
		}
		close(responseChan)
		pulledTXs := make([]Transaction, 0, len(pulled))
		snapshot := l.Snapshot()
		for _, tx := range pulled {
			if err := ValidateTransaction(snapshot, tx); err != nil {
				if err == ErrTxInvalidSignature {
					logger.Error().
						Hex("tx_id", tx.ID[:]).
						Msg("bad signature")
				}
				continue
			}
			pulledTXs = append(pulledTXs, tx)
		}
		l.AddTransaction(pulledTXs...)
		if count > 0 {
			logger.Info().
				Int64("count", count).
				Msg("Pulled missing transaction(s).")
		}
		l.metrics.downloadedTX.Mark(count)
		l.metrics.receivedTX.Mark(count)
	}
}
func ( l * Ledger ) PullMissingTransactions ( ) { defer l . consensus . Done ( ) for { select { case <- time . After ( 100 * time . Millisecond ) : case <- l . consensusStop : return } snowballK := conf . GetSnowballK ( ) peers , err := SelectPeers ( l . client . ClosestPeers ( ) , snowballK ) if err != nil { continue } logger := log . Sync ( " " ) missingIDs := l . transactions . MissingIDs ( ) missingTxPullLimit := conf . GetMissingTxPullLimit ( ) if uint64 ( len ( missingIDs ) ) > missingTxPullLimit { missingIDs = missingIDs [ : missingTxPullLimit ] } req := & TransactionPullRequest { TransactionIds : make ( [ ] [ ] byte , 0 , len ( missingIDs ) ) } for _ , txID := range missingIDs { txID := txID req . TransactionIds = append ( req . TransactionIds , txID [ : ] ) } type response struct { txs [ ] Transaction } responseChan := make ( chan response ) for _ , p := range peers { go func ( conn * grpc . ClientConn ) { var response response defer func ( ) { responseChan <- response } ( ) client := NewWaveletClient ( conn ) ctx , cancel := context . WithTimeout ( context . Background ( ) , conf . GetDownloadTxTimeout ( ) ) defer cancel ( ) batch , err := client . PullTransactions ( ctx , req ) if err != nil { logger . Error ( ) . Err ( err ) . Msg ( " " ) return } response . txs = make ( [ ] Transaction , 0 , len ( batch . Transactions ) ) for _ , buf := range batch . Transactions { tx , err := UnmarshalTransaction ( bytes . NewReader ( buf ) ) if err != nil { logger . Error ( ) . Err ( err ) . Hex ( " " , tx . ID [ : ] ) . Msg ( " " ) continue } response . txs = append ( response . txs , tx ) } } ( p . Conn ( ) ) } count := int64 ( 0 ) pulled := make ( map [ TransactionID ] Transaction ) for i := 0 ; i < len ( peers ) ; i ++ { res := <- responseChan for i := range res . txs { if _ , ok := pulled [ res . txs [ i ] . ID ] ; ! ok { pulled [ res . txs [ i ] . ID ] = res . txs [ i ] count += int64 ( res . txs [ i ] . LogicalUnits ( ) ) } } } close ( responseChan ) pulledTXs := make ( [ ] Transaction , 0 , len ( pulled ) ) snapshot := l . Snapshot ( ) for _ , tx := range pulled { if err := ValidateTransaction ( snapshot , tx ) ; err != nil { if err == ErrTxInvalidSignature { logger . Error ( ) . Hex ( " " , tx . ID [ : ] ) . Msg ( " " ) } continue } pulledTXs = append ( pulledTXs , tx ) } l . AddTransaction ( pulledTXs ... ) if count > 0 { logger . Info ( ) . Int64 ( " " , count ) . Msg ( " " ) } l . metrics . downloadedTX . Mark ( count ) l . metrics . receivedTX . Mark ( count ) } }
PullMissingTransactions is a goroutine which continuously pulls missing transactions
from randomly sampled peers in the network.
PullMissingTransactions is a goroutine which continuously pulls missing transactions from randomly sampled peers in the network .
// Build list of transaction IDs