Repository: trustwallet/blockatlas
Pull request #1011 (2020-04-05 06:12:07)

Additional fixes for observer, add batch notifications for mq


Patch:
diff --git a/cmd/observer_notifier/main.go b/cmd/observer_notifier/main.go
index 293357c..623676c 100644
--- a/cmd/observer_notifier/main.go
+++ b/cmd/observer_notifier/main.go
@@ -28,6 +28,7 @@ func init() {
 
 	mqHost := viper.GetString("observer.rabbitmq.uri")
 	prefetchCount := viper.GetInt("observer.rabbitmq.consumer.prefetch_count")
+	maxPushNotificationsBatchLimit := viper.GetUint("observer.push_notifications_batch_limit")
 	pgUri := viper.GetString("postgres.uri")
 
 	internal.InitRabbitMQ(mqHost, prefetchCount)
@@ -36,7 +37,7 @@ func init() {
 		logger.Fatal(err)
 	}
 
-	if err := mq.Transactions.Declare(); err != nil {
+	if err := mq.TxNotifications.Declare(); err != nil {
 		logger.Fatal(err)
 	}
 
@@ -46,6 +47,14 @@ func init() {
 		logger.Fatal(err)
 	}
 
+	if maxPushNotificationsBatchLimit == 0 {
+		notifier.MaxPushNotificationsBatchLimit = notifier.DefaultPushNotificationsBatchLimit
+	} else {
+		notifier.MaxPushNotificationsBatchLimit = maxPushNotificationsBatchLimit
+	}
+
+	logger.Info("maxPushNotificationsBatchLimit ", logger.Params{"limit": maxPushNotificationsBatchLimit})
+
 	go mq.RestoreConnectionWorker(mqHost, mq.RawTransactions, time.Second*10)
 	go db.RestoreConnectionWorker(database, time.Second*10, pgUri)
 	time.Sleep(time.Millisecond)
diff --git a/cmd/observer_parser/main.go b/cmd/observer_parser/main.go
index 41190a9..83e59f3 100644
--- a/cmd/observer_parser/main.go
+++ b/cmd/observer_parser/main.go
@@ -23,11 +23,11 @@ const (
 )
 
 var (
-	confPath                              string
-	backlogTime, minInterval, maxInterval time.Duration
-	maxBackLogBlocks                      int64
-	txsBatchLimit                         uint
-	database                              *db.Instance
+	confPath                                                   string
+	backlogTime, minInterval, maxInterval, fetchBlocksInterval time.Duration
+	maxBackLogBlocks                                           int64
+	txsBatchLimit                                              uint
+	database                                                   *db.Instance
 )
 
 func init() {
@@ -57,6 +57,7 @@ func init() {
 	backlogTime = viper.GetDuration("observer.backlog")
 	minInterval = viper.GetDuration("observer.block_poll.min")
 	maxInterval = viper.GetDuration("observer.block_poll.max")
+	fetchBlocksInterval = viper.GetDuration("observer.fetch_blocks_interval")
 	maxBackLogBlocks = viper.GetInt64("observer.backlog_max_blocks")
 	if minInterval >= maxInterval {
 		logger.Fatal("minimum block polling interval cannot be greater or equal than maximum")
@@ -107,6 +108,7 @@ func main() {
 			Api:                   api,
 			Queue:                 mq.RawTransactions,
 			ParsingBlocksInterval: pollInterval,
+			FetchBlocksTimeout:    fetchBlocksInterval,
 			BacklogCount:          backlogCount,
 			MaxBacklogBlocks:      maxBackLogBlocks,
 			StopChannel:           stopChannel,
@@ -117,10 +119,11 @@ func main() {
 		go parser.RunParser(params)
 
 		logger.Info("Parser params", logger.Params{
-			"interval":        pollInterval,
-			"backlog":         backlogCount,
-			"Max backlog":     maxBackLogBlocks,
-			"Txs Batch limit": txsBatchLimit,
+			"interval":                 pollInterval,
+			"backlog":                  backlogCount,
+			"Max backlog":              maxBackLogBlocks,
+			"Txs Batch limit":          txsBatchLimit,
+			"Fetching blocks interval": fetchBlocksInterval,
 		})
 
 		wg.Done()
diff --git a/config.yml b/config.yml
index 9ae5439..426a9be 100644
--- a/config.yml
+++ b/config.yml
@@ -19,10 +19,14 @@ platform: binance
 observer:
   # Don't request blocks older than this
   backlog: 3h
+  # Amount of time between fetching blocks concurrently
+  fetch_blocks_interval: 1ms
   # Don't request more than N blocks at once
   backlog_max_blocks: 200
   # Limit amount of transactions in batch
   txs_batch_limit: 3000
+  # Limit of push notifications in batch
+  push_notifications_batch_limit: 50
   # Block polling interval
   block_poll:
     min: 3s
diff --git a/db/models/tracker.go b/db/models/tracker.go
index 10e62a2..c5a0355 100644
--- a/db/models/tracker.go
+++ b/db/models/tracker.go
@@ -1,6 +1,6 @@
 package models
 
 type Tracker struct {
-	Coin   uint `gorm:"primary_key:true"`
+	Coin   uint `gorm:"primary_key:true; auto_increment:false"`
 	Height int64
 }
diff --git a/mq/mq.go b/mq/mq.go
index 812fd90..e2c2a60 100644
--- a/mq/mq.go
+++ b/mq/mq.go
@@ -22,11 +22,9 @@ type (
 )
 
 const (
-	Transactions         Queue = "transactions"
-	Subscriptions        Queue = "subscriptions"
-	RawTransactions      Queue = "rawTransactions"
-	defaultPrefetchCount       = 5
-	minPrefetchCount           = 1
+	TxNotifications Queue = "txNotifications"
+	Subscriptions   Queue = "subscriptions"
+	RawTransactions Queue = "rawTransactions"
 )
 
 func Init(uri string) (err error) {
@@ -39,8 +37,15 @@ func Init(uri string) (err error) {
 }
 
 func Close() {
-	amqpChan.Close()
-	conn.Close()
+	err := amqpChan.Close()
+	if err != nil {
+		logger.Error(err)
+	}
+
+	err = conn.Close()
+	if err != nil {
+		logger.Error(err)
+	}
 }
 
 func (mc MessageChannel) GetMessage() amqp.Delivery {
diff --git a/services/observer/notifier/notifier.go b/services/observer/notifier/notifier.go
index 16e97ff..35e1b09 100644
--- a/services/observer/notifier/notifier.go
+++ b/services/observer/notifier/notifier.go
@@ -13,7 +13,11 @@ import (
 	"time"
 )
 
-type DispatchEvent struct {
+const DefaultPushNotificationsBatchLimit = 50
+
+var MaxPushNotificationsBatchLimit uint = DefaultPushNotificationsBatchLimit
+
+type TransactionNotification struct {
 	Action blockatlas.TransactionType `json:"action"`
 	Result *blockatlas.Tx             `json:"result"`
 	Id     uint                       `json:"id"`
@@ -60,40 +64,47 @@ func RunNotifier(database *db.Instance, delivery amqp.Delivery) {
 
 func buildAndPostMessage(blockTransactions blockatlas.TxSetMap, sub blockatlas.Subscription, wg *sync.WaitGroup) {
 	defer wg.Done()
+
 	tx, ok := blockTransactions.Map[sub.Address]
 	if !ok {
 		return
 	}
+	notifications := make([]TransactionNotification, 0, len(tx.Txs()))
 	for _, tx := range tx.Txs() {
 		tx.Direction = tx.GetTransactionDirection(sub.Address)
 		tx.InferUtxoValue(sub.Address, tx.Coin)
-		action := DispatchEvent{
+		notification := TransactionNotification{
 			Action: tx.Type,
 			Result: &tx,
 			Id:     sub.Id,
 		}
-		txJson, err := json.Marshal(action)
-		if err != nil {
-			logger.Panic(err, logger.Params{"coin": tx.Coin})
-		}
 
-		logParams := logger.Params{
-			"Id":   sub.Id,
-			"coin": sub.Coin,
-			"txID": tx.ID,
-		}
+		logger.Info("Notification ready", logger.Params{"Id": sub.Id, "coin": sub.Coin, "txID": tx.ID})
+
+		notifications = append(notifications, notification)
+	}
+
+	batches := getNotificationBatches(notifications, MaxPushNotificationsBatchLimit)
 
-		publishTransaction(sub.Id, txJson, logParams)
+	for _, batch := range batches {
+		publishNotificationBatch(batch)
 	}
 }
 
-func publishTransaction(id uint, rawMessage []byte, logParams logger.Params) {
-	err := mq.Transactions.Publish(rawMessage)
+func publishNotificationBatch(batch []TransactionNotification) {
+	raw, err := json.Marshal(batch)
 	if err != nil {
-		err = errors.E(err, "Failed to dispatch event", errors.Params{"id": id}, logParams)
-		logger.Fatal(err, logger.Params{"id": id}, logParams)
+		err = errors.E(err, " failed to dispatch event")
+		logger.Fatal(err)
 	}
-	logger.Info("Message dispatched", logger.Params{"id": id}, logParams)
+
+	err = mq.TxNotifications.Publish(raw)
+	if err != nil {
+		err = errors.E(err, " failed to dispatch event")
+		logger.Fatal(err)
+	}
+
+	logger.Info("Txs batch dispatched", logger.Params{"txs": len(batch)})
 }
 
 func GetInterval(value int, minInterval, maxInterval time.Duration) time.Duration {
@@ -102,3 +113,18 @@ func GetInterval(value int, minInterval, maxInterval time.Duration) time.Duratio
 	pMax := numbers.Min(int(maxInterval.Nanoseconds()), int(pMin))
 	return time.Duration(pMax)
 }
+
+func getNotificationBatches(notifications []TransactionNotification, sizeUint uint) [][]TransactionNotification {
+	size := int(sizeUint)
+	resultLength := (len(notifications) + size - 1) / size
+	result := make([][]TransactionNotification, resultLength)
+	lo, hi := 0, size
+	for i := range result {
+		if hi > len(notifications) {
+			hi = len(notifications)
+		}
+		result[i] = notifications[lo:hi:hi]
+		lo, hi = hi, hi+size
+	}
+	return result
+}
diff --git a/services/observer/parser/parser.go b/services/observer/parser/parser.go
index 10d71a2..86e6f67 100644
--- a/services/observer/parser/parser.go
+++ b/services/observer/parser/parser.go
@@ -19,15 +19,15 @@ import (
 
 type (
 	Params struct {
-		Ctx                   context.Context
-		Api                   blockatlas.BlockAPI
-		Queue                 mq.Queue
-		ParsingBlocksInterval time.Duration
-		BacklogCount          int
-		MaxBacklogBlocks      int64
-		StopChannel           chan<- struct{}
-		TxBatchLimit          uint
-		Database              *db.Instance
+		Ctx                                       context.Context
+		Api                                       blockatlas.BlockAPI
+		Queue                                     mq.Queue
+		ParsingBlocksInterval, FetchBlocksTimeout time.Duration
+		BacklogCount                              int
+		MaxBacklogBlocks                          int64
+		StopChannel                               chan<- struct{}
+		TxBatchLimit                              uint
+		Database                                  *db.Instance
 	}
 
 	GetBlockByNumber func(num int64) (*blockatlas.Block, error)
@@ -54,13 +54,13 @@ func RunParser(params Params) {
 			return
 		default:
 			lastParsedBlock, currentBlock, err := GetBlocksIntervalToFetch(params)
-			if err != nil {
+			if err != nil || lastParsedBlock > currentBlock {
 				logger.Error(err, logger.Params{"coin": params.Api.Coin().Handle})
 				time.Sleep(params.ParsingBlocksInterval)
 				continue
 			}
 
-			blocks := FetchBlocks(params.Api, lastParsedBlock, currentBlock)
+			blocks := FetchBlocks(params, lastParsedBlock, currentBlock)
 
 			err = SaveLastParsedBlock(params, blocks)
 			if err != nil {
@@ -100,15 +100,15 @@ func GetBlocksIntervalToFetch(params Params) (int64, int64, error) {
 	return lastParsedBlock, currentBlock, nil
 }
 
-func FetchBlocks(api blockatlas.BlockAPI, lastParsedBlock, currentBlock int64) []blockatlas.Block {
+func FetchBlocks(params Params, lastParsedBlock, currentBlock int64) []blockatlas.Block {
 	if lastParsedBlock == currentBlock {
-		logger.Info("No new blocks", logger.Params{"last": lastParsedBlock, "coin": api.Coin().ID, "time": time.Now().Unix()})
+		logger.Info("No new blocks", logger.Params{"last": lastParsedBlock, "coin": params.Api.Coin().ID, "time": time.Now().Unix()})
 		return nil
 	}
 
 	blocksCount := currentBlock - lastParsedBlock
 	if blocksCount < 0 {
-		logger.Error("Current block is 0", logger.Params{"coin": api.Coin().Handle})
+		logger.Error("Current block is 0", logger.Params{"coin": params.Api.Coin().Handle})
 		return nil
 	}
 
@@ -121,9 +121,10 @@ func FetchBlocks(api blockatlas.BlockAPI, lastParsedBlock, currentBlock int64) [
 
 	for i := lastParsedBlock + 1; i <= currentBlock; i++ {
 		wg.Add(1)
+		time.Sleep(params.FetchBlocksTimeout)
 		go func(i int64, wg *sync.WaitGroup) {
 			defer wg.Done()
-			err := fetchBlock(api, i, blocksChan)
+			err := fetchBlock(params.Api, i, blocksChan)
 			if err != nil {
 				errorsChan <- err
 				return
@@ -174,6 +175,9 @@ func SaveLastParsedBlock(params Params, blocks []blockatlas.Block) error {
 	})
 
 	lastBlockNumber := blocks[len(blocks)-1].Number
+	if lastBlockNumber <= 0 {
+		return errors.E(fmt.Sprintf("Parser of %s failed to save last block, lastBlockNumber <= 0", params.Api.Coin().Handle))
+	}
 	err := params.Database.SetLastParsedBlockNumber(params.Api.Coin().ID, lastBlockNumber)
 	if err != nil {
 		return err


Test patch:
diff --git a/services/observer/parser/parser_test.go b/services/observer/parser/parser_test.go
index 6b6d369..d5b88ae 100644
--- a/services/observer/parser/parser_test.go
+++ b/services/observer/parser/parser_test.go
@@ -63,7 +63,19 @@ var (
 )
 
 func TestFetchBlocks(t *testing.T) {
-	blocks := FetchBlocks(getMockedBlockAPI(), 0, 100)
+	params := Params{
+		Ctx:                   nil,
+		Api:                   getMockedBlockAPI(),
+		Queue:                 "",
+		ParsingBlocksInterval: 0,
+		FetchBlocksTimeout:    0,
+		BacklogCount:          0,
+		MaxBacklogBlocks:      0,
+		StopChannel:           nil,
+		TxBatchLimit:          0,
+		Database:              nil,
+	}
+	blocks := FetchBlocks(params, 0, 100)
 	assert.Equal(t, len(blocks), 100)
 }
 
diff --git a/tests/integration/observer_test/full_flow_test.go b/tests/integration/observer_test/full_flow_test.go
index 3b4df80..aeaf88f 100644
--- a/tests/integration/observer_test/full_flow_test.go
+++ b/tests/integration/observer_test/full_flow_test.go
@@ -101,8 +101,8 @@ func (p *PlatformFullFlow) GetBlockByNumber(num int64) (*blockatlas.Block, error
 }
 
 func ConsumerToTestTransactionsFull(delivery amqp.Delivery, t *testing.T, cancel context.CancelFunc, counter int) {
-	var event notifier.DispatchEvent
-	if err := json.Unmarshal(delivery.Body, &event); err != nil {
+	var notifications []notifier.TransactionNotification
+	if err := json.Unmarshal(delivery.Body, &notifications); err != nil {
 		assert.Nil(t, err)
 		return
 	}
@@ -119,7 +119,7 @@ func ConsumerToTestTransactionsFull(delivery amqp.Delivery, t *testing.T, cancel
 		To:       "testAddress",
 	}
 
-	assert.Equal(t, notifier.DispatchEvent{
+	assert.Equal(t, notifier.TransactionNotification{
 		Action: blockatlas.TxNativeTokenTransfer,
 		Result: &blockatlas.Tx{
 			Type:      blockatlas.TxNativeTokenTransfer,
@@ -136,7 +136,7 @@ func ConsumerToTestTransactionsFull(delivery amqp.Delivery, t *testing.T, cancel
 			Meta:      &memo,
 		},
 		Id: 1,
-	}, event)
+	}, notifications[0])
 
 	if counter == 10 {
 		cancel()
diff --git a/tests/integration/observer_test/notifier_test.go b/tests/integration/observer_test/notifier_test.go
index 8bb5b4d..4737b8b 100644
--- a/tests/integration/observer_test/notifier_test.go
+++ b/tests/integration/observer_test/notifier_test.go
@@ -60,8 +60,8 @@ func TestNotifier(t *testing.T) {
 }
 
 func ConsumerToTestTransactions(delivery amqp.Delivery, t *testing.T) {
-	var event notifier.DispatchEvent
-	if err := json.Unmarshal(delivery.Body, &event); err != nil {
+	var notifications []notifier.TransactionNotification
+	if err := json.Unmarshal(delivery.Body, &notifications); err != nil {
 		assert.Nil(t, err)
 		return
 	}
@@ -80,7 +80,7 @@ func ConsumerToTestTransactions(delivery amqp.Delivery, t *testing.T) {
 		To:       "tbnb12hlquylu78cjylk5zshxpdj6hf3t0tahwjt3ex",
 	}
 
-	assert.Equal(t, notifier.DispatchEvent{
+	assert.Equal(t, notifier.TransactionNotification{
 		Action: blockatlas.TxNativeTokenTransfer,
 		Result: &blockatlas.Tx{
 			Type:      blockatlas.TxNativeTokenTransfer,
@@ -97,7 +97,7 @@ func ConsumerToTestTransactions(delivery amqp.Delivery, t *testing.T) {
 			Meta:      &memo,
 		},
 		Id: 1,
-	}, event)
+	}, notifications[0])
 
 	return
 }
diff --git a/tests/integration/observer_test/observer_test.go b/tests/integration/observer_test/observer_test.go
index dc75b91..cdaba9d 100644
--- a/tests/integration/observer_test/observer_test.go
+++ b/tests/integration/observer_test/observer_test.go
@@ -22,7 +22,7 @@ func TestMain(m *testing.M) {
 	if err := mq.RawTransactions.Declare(); err != nil {
 		log.Fatal(err)
 	}
-	if err := mq.Transactions.Declare(); err != nil {
+	if err := mq.TxNotifications.Declare(); err != nil {
 		log.Fatal(err)
 	}
 	if err := mq.Subscriptions.Declare(); err != nil {
@@ -30,7 +30,7 @@ func TestMain(m *testing.M) {
 	}
 	rawTransactionsChannel = mq.RawTransactions.GetMessageChannel()
 	subscriptionChannel = mq.Subscriptions.GetMessageChannel()
-	transactionsChannel = mq.Transactions.GetMessageChannel()
+	transactionsChannel = mq.TxNotifications.GetMessageChannel()
 
 	code := m.Run()