Add atxSync flag to disable atx syncing and http bulk sync api now sy…

…ncs in batches
spacemeshos · Apr 17, 2024 · 2108597 · 2108597
1 parent d48c6c3
commit 2108597
Show file tree

Hide file tree

Showing 9 changed files with 98 additions and 17 deletions.
diff --git a/cmd/collector/main.go b/cmd/collector/main.go
@@ -36,6 +36,7 @@ var (
 	apiHostFlag                   string
 	apiPortFlag                   int
 	recalculateEpochStatsBoolFlag bool
+	atxSyncFlag                   bool
 )
 
 var flags = []cli.Flag{
@@ -134,6 +135,14 @@ var flags = []cli.Flag{
 		Destination: &apiPortFlag,
 		EnvVars:     []string{"SPACEMESH_API_PORT"},
 	},
+	&cli.BoolFlag{
+		Name:        "atxSync",
+		Usage:       ``,
+		Required:    false,
+		Value:       true,
+		Destination: &atxSyncFlag,
+		EnvVars:     []string{"SPACEMESH_ATX_SYNC"},
+	},
 }
 
 func main() {
@@ -165,7 +174,7 @@ func main() {
 		dbClient := &sql.Client{}
 
 		c := collector.NewCollector(nodePublicAddressStringFlag, nodePrivateAddressStringFlag,
-			syncMissingLayersBoolFlag, syncFromLayerFlag, recalculateEpochStatsBoolFlag, mongoStorage, db, dbClient)
+			syncMissingLayersBoolFlag, syncFromLayerFlag, recalculateEpochStatsBoolFlag, mongoStorage, db, dbClient, atxSyncFlag)
 		mongoStorage.AccountUpdater = c
 
 		sigs := make(chan os.Signal, 1)

diff --git a/collector/collector.go b/collector/collector.go
@@ -57,6 +57,7 @@ type Collector struct {
 	syncMissingLayersFlag     bool
 	recalculateEpochStatsFlag bool
 	syncFromLayerFlag         uint32
+	atxSyncFlag               bool
 
 	listener Listener
 	db       *sql2.Database
@@ -79,7 +80,9 @@ type Collector struct {
 	notify chan int
 }
 
-func NewCollector(nodePublicAddress string, nodePrivateAddress string, syncMissingLayersFlag bool, syncFromLayerFlag int, recalculateEpochStatsFlag bool, listener Listener, db *sql2.Database, dbClient sql.DatabaseClient) *Collector {
+func NewCollector(nodePublicAddress string, nodePrivateAddress string, syncMissingLayersFlag bool,
+	syncFromLayerFlag int, recalculateEpochStatsFlag bool,
+	listener Listener, db *sql2.Database, dbClient sql.DatabaseClient, atxSyncFlag bool) *Collector {
 	return &Collector{
 		apiPublicUrl:              nodePublicAddress,
 		apiPrivateUrl:             nodePrivateAddress,
@@ -90,6 +93,7 @@ func NewCollector(nodePublicAddress string, nodePrivateAddress string, syncMissi
 		notify:                    make(chan int),
 		db:                        db,
 		dbClient:                  dbClient,
+		atxSyncFlag:               atxSyncFlag,
 	}
 }
 
@@ -132,9 +136,11 @@ func (c *Collector) Run() error {
 		return errors.Join(errors.New("cannot get network info"), err)
 	}
 
-	err = c.syncActivations()
-	if err != nil {
-		return errors.Join(errors.New("cannot sync activations"), err)
+	if c.atxSyncFlag {
+		err = c.syncActivations()
+		if err != nil {
+			return errors.Join(errors.New("cannot sync activations"), err)
+		}
 	}
 
 	if c.syncMissingLayersFlag {

diff --git a/collector/collector_test.go b/collector/collector_test.go
@@ -88,7 +88,7 @@ func TestMain(m *testing.M) {
 
 	collectorApp = collector.NewCollector(fmt.Sprintf("localhost:%d", node.NodePort),
 		fmt.Sprintf("localhost:%d", privateNode.NodePort), false,
-		0, false, storageDB, sqlDb, dbClient)
+		0, false, storageDB, sqlDb, dbClient, true)
 	storageDB.AccountUpdater = collectorApp
 	defer storageDB.Close()
 	go collectorApp.Run()

diff --git a/collector/http.go b/collector/http.go
@@ -94,16 +94,27 @@ func (c *Collector) StartHttpServer(apiHost string, apiPort int) {
 
 		log.Info("http syncing atxs for epoch %s", epoch)
 		go func() {
-			var atxs []*model.Activation
-			err = c.dbClient.GetAtxsByEpoch(c.db, epochId, func(atx *types.VerifiedActivationTx) bool {
-				atxs = append(atxs, model.NewActivation(atx))
-				return true
-			})
+			count, err := c.dbClient.CountAtxsByEpoch(c.db, epochId)
 			if err != nil {
 				log.Warning("syncing atxs for %s failed with error %d", epoch, err)
 				return
 			}
-			c.listener.OnActivations(atxs)
+			batchSize := 100000
+			totalPages := (count + batchSize - 1) / batchSize
+			for page := 0; page < totalPages; page++ {
+				offset := page * batchSize
+				var atxs []*model.Activation
+				err = c.dbClient.GetAtxsByEpochPaginated(c.db, epochId, int64(batchSize), int64(offset), func(atx *types.VerifiedActivationTx) bool {
+					atxs = append(atxs, model.NewActivation(atx))
+					return true
+				})
+				if err != nil {
+					log.Warning("syncing atxs for %s failed with error %d", epoch, err)
+					return
+				}
+				c.listener.OnActivations(atxs)
+				atxs = nil
+			}
 			c.listener.RecalculateEpochStats()
 		}()
 

diff --git a/collector/mesh.go b/collector/mesh.go
@@ -3,6 +3,7 @@ package collector
 import (
 	"context"
 	"fmt"
+	"github.com/spacemeshos/explorer-backend/model"
 	"github.com/spacemeshos/explorer-backend/utils"
 	"github.com/spacemeshos/go-spacemesh/common/types"
 	"go.mongodb.org/mongo-driver/bson"
@@ -238,14 +239,17 @@ func (c *Collector) syncActivations() error {
 	received := c.listener.GetLastActivationReceived()
 	log.Info("Syncing activations from %d", received)
 
+	var atxs []*model.Activation
 	err := c.dbClient.GetAtxsReceivedAfter(c.db, received, func(atx *types.VerifiedActivationTx) bool {
-		c.listener.OnActivation(atx)
+		atxs = append(atxs, model.NewActivation(atx))
 		return true
 	})
 	if err != nil {
 		return err
 	}
 
+	c.listener.OnActivations(atxs)
+
 	return nil
 }
 

diff --git a/collector/node.go b/collector/node.go
@@ -54,9 +54,11 @@ func (c *Collector) syncStatusPump() error {
 					fmt.Errorf("syncNotProcessedTxs error: %v", err)
 				}
 
-				err = c.syncActivations()
-				if err != nil {
-					fmt.Errorf("syncActivations error: %v", err)
+				if c.atxSyncFlag {
+					err = c.syncActivations()
+					if err != nil {
+						fmt.Errorf("syncActivations error: %v", err)
+					}
 				}
 
 				err = c.createFutureEpoch()

diff --git a/collector/sql/atxs.go b/collector/sql/atxs.go
@@ -53,7 +53,7 @@ func decoder(fn decoderCallback) sql.Decoder {
 func (c *Client) GetAtxsReceivedAfter(db *sql.Database, ts int64, fn func(tx *types.VerifiedActivationTx) bool) error {
 	var derr error
 	_, err := db.Exec(
-		fullQuery+` WHERE received > ?1 ORDER BY epoch asc, id asc`,
+		fullQuery+` WHERE received > ?1`,
 		func(stmt *sql.Statement) {
 			stmt.BindInt64(1, ts)
 		},
@@ -91,3 +91,42 @@ func (c *Client) GetAtxsByEpoch(db *sql.Database, epoch int64, fn func(tx *types
 	}
 	return derr
 }
+
+func (c *Client) CountAtxsByEpoch(db *sql.Database, epoch int64) (int, error) {
+	var totalCount int
+	_, err := db.Exec(
+		`SELECT COUNT(*) FROM atxs WHERE epoch = ?1`,
+		func(stmt *sql.Statement) {
+			stmt.BindInt64(1, epoch)
+		}, func(stmt *sql.Statement) bool {
+			totalCount = stmt.ColumnInt(0)
+			return true
+		})
+	if err != nil {
+		return 0, err
+	}
+	return totalCount, nil
+}
+
+func (c *Client) GetAtxsByEpochPaginated(db *sql.Database, epoch, limit, offset int64, fn func(tx *types.VerifiedActivationTx) bool) error {
+	var derr error
+	_, err := db.Exec(
+		fullQuery+` WHERE epoch = ?1 ORDER BY epoch asc, id asc LIMIT ?2 OFFSET ?3`,
+		func(stmt *sql.Statement) {
+			stmt.BindInt64(1, epoch)
+			stmt.BindInt64(2, limit)
+			stmt.BindInt64(3, offset)
+		},
+		decoder(func(atx *types.VerifiedActivationTx, err error) bool {
+			if atx != nil {
+				return fn(atx)
+			}
+			derr = err
+			return derr == nil
+		}),
+	)
+	if err != nil {
+		return err
+	}
+	return derr
+}
diff --git a/collector/sql/sql.go b/collector/sql/sql.go
@@ -14,6 +14,8 @@ type DatabaseClient interface {
 	AccountsSnapshot(db *sql.Database, lid types.LayerID) (rst []*types.Account, err error)
 	GetAtxsReceivedAfter(db *sql.Database, ts int64, fn func(tx *types.VerifiedActivationTx) bool) error
 	GetAtxsByEpoch(db *sql.Database, epoch int64, fn func(tx *types.VerifiedActivationTx) bool) error
+	CountAtxsByEpoch(db *sql.Database, epoch int64) (int, error)
+	GetAtxsByEpochPaginated(db *sql.Database, epoch, limit, offset int64, fn func(tx *types.VerifiedActivationTx) bool) error
 }
 
 type Client struct{}

diff --git a/test/testseed/db.go b/test/testseed/db.go
@@ -206,6 +206,14 @@ func (c *Client) GetAtxsByEpoch(db *sql.Database, epoch int64, fn func(tx *types
 	return nil
 }
 
+func (c *Client) CountAtxsByEpoch(db *sql.Database, epoch int64) (int, error) {
+	return 0, nil
+}
+
+func (c *Client) GetAtxsByEpochPaginated(db *sql.Database, epoch, limit, offset int64, fn func(tx *types.VerifiedActivationTx) bool) error {
+	return nil
+}
+
 func mustParse(str string) []byte {
 	res, err := utils.StringToBytes(str)
 	if err != nil {