feat(dash-exporter): use muttex to prevent concurrent rolling gen while transfering data

Author: invis-bitfly

backend/pkg/exporter/modules/dashboard_data.go

@@ -27,13 +27,14 @@ import (
 
 type dashboardData struct {
 	ModuleContext
-	log               ModuleLog
-	signingDomain     []byte
-	phase0HotfixMutex sync.Mutex
-	latestSafeEpoch   atomic.Int64
-	heavySemaphore    *semaphore.Weighted
-	mediumSemaphore   *semaphore.Weighted
-	lightSemaphore    *semaphore.Weighted
+	log                   ModuleLog
+	signingDomain         []byte
+	phase0HotfixMutex     sync.Mutex
+	sharedRollingGenMutex *sync.Mutex
+	latestSafeEpoch       atomic.Int64
+	heavySemaphore        *semaphore.Weighted
+	mediumSemaphore       *semaphore.Weighted
+	lightSemaphore        *semaphore.Weighted
 }
 
 func NewDashboardDataModule(moduleContext ModuleContext) ModuleInterface {

backend/pkg/exporter/modules/dashboard_data_maintenance.go

@@ -54,6 +54,10 @@ func (d *dashboardData) handleIncompleteTransfers() error {
 	}
 	d.log.Infof("handleIncompleteTransfers, found %d incomplete transfer epochs", len(incomplete))
 
+	d.log.Tracef("acquiring shared rolling gen mutex for incomplete transfers")
+	d.sharedRollingGenMutex.Lock()
+	d.log.Tracef("acquired shared rolling gen mutex for incomplete transfers")
+	defer d.sharedRollingGenMutex.Unlock()
 	err = d.transferEpochs(incomplete)
 	if err != nil {
 		return errors.Wrap(err, "failed to transfer incomplete epochs")
@@ -72,6 +76,11 @@ func (d *dashboardData) handlePendingTransfers() error {
 	}
 	d.log.Infof("handlePendingTransfers, found %d pending transfer epochs", len(pending))
 
+	d.log.Tracef("acquiring shared rolling gen mutex for pending transfers")
+	d.sharedRollingGenMutex.Lock()
+	d.log.Tracef("acquired shared rolling gen mutex for pending transfers")
+	defer d.sharedRollingGenMutex.Unlock()
+
 	// allocate transfer batch ids
 	batchIds := make([]uuid.UUID, 0)
 	for i := range pending {

backend/pkg/exporter/modules/dashboard_data_rollings.go

@@ -36,8 +36,13 @@ func (d *dashboardData) handleRollings() error {
 		edb.Rolling90d,
 		edb.RollingTotal,
 	}
+	d.log.Tracef("acquiring shared rolling gen mutex for rollings")
+	d.sharedRollingGenMutex.Lock()
+	d.log.Tracef("acquired shared rolling gen mutex for rollings")
+	defer d.sharedRollingGenMutex.Unlock()
+
 	eg := errgroup.Group{}
-	//eg.SetLimit(int(utils.Config.DashboardExporter.RollingsInParallel))
+	eg.SetLimit(int(utils.Config.DashboardExporter.RollingsInParallel))
 	for _, rolling := range rollings {
 		rolling := rolling
 		eg.Go(func() error {