refactor: GC has supervisor

yetanotherco · Oct 15, 2024 · ef8e5d6 · ef8e5d6
1 parent 7391495
commit ef8e5d6
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 5 deletions.
diff --git a/aggregator/cmd/main.go b/aggregator/cmd/main.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"time"
 
 	"github.com/urfave/cli/v2"
 	"github.com/yetanotherco/aligned_layer/aggregator/internal/pkg"
@@ -38,6 +39,9 @@ func main() {
 	}
 }
 
+const garbageCollectorPeriod = time.Second * 150 //TODO change to time.Day * 1
+const garbageCollectorTasksAge = uint64(10)     //TODO change to 2592000, 1 month of blocks
+
 func aggregatorMain(ctx *cli.Context) error {
 
 	configFilePath := ctx.String(config.ConfigFileFlag.Name)
@@ -49,6 +53,13 @@ func aggregatorMain(ctx *cli.Context) error {
 		return err
 	}
 
+	// Supervisor revives garbage collector
+	go func() {
+		for {
+			aggregator.ClearTasksFromMaps(garbageCollectorPeriod, garbageCollectorTasksAge)
+		}
+	}()
+
 	// Listen for new task created in the ServiceManager contract in a separate goroutine, both V1 and V2 subscriptions:
 	go func() {
 		listenErr := aggregator.SubscribeToNewTasks()

diff --git a/aggregator/internal/pkg/aggregator.go b/aggregator/internal/pkg/aggregator.go
@@ -178,7 +178,6 @@ func NewAggregator(aggregatorConfig config.AggregatorConfig) (*Aggregator, error
 		telemetry:             aggregatorTelemetry,
 	}
 
-	go aggregator.clearTasksFromMaps(garbageCollectorPeriod, garbageCollectorTasksAge)
 
 	return &aggregator, nil
 }
@@ -217,8 +216,6 @@ func (agg *Aggregator) Start(ctx context.Context) error {
 
 const MaxSentTxRetries = 5
 
-const garbageCollectorPeriod = time.Second * 150 //TODO change to time.Day * 1
-const garbageCollectorTasksAge = uint64(10)     //TODO change to 2592000, 1 month of blocks
 
 const BLS_AGG_SERVICE_TIMEOUT = 100 * time.Second
 
@@ -392,8 +389,15 @@ func (agg *Aggregator) AddNewTask(batchMerkleRoot [32]byte, senderAddress [20]by
 }
 
 // long-lived gorouting that periodically checks and removes old Tasks from stored Maps
-func (agg *Aggregator) clearTasksFromMaps(period time.Duration, blocksOld uint64) {
-	agg.AggregatorConfig.BaseConfig.Logger.Info("- Removing finalized Task Infos from Maps every %d seconds", period)
+func (agg *Aggregator) ClearTasksFromMaps(period time.Duration, blocksOld uint64) {
+	defer func() {
+		err := recover() //stops panics
+		if err != nil {
+			agg.logger.Error(err.(string))
+		}
+	}()
+
+	agg.AggregatorConfig.BaseConfig.Logger.Info(fmt.Sprintf("- Removing finalized Task Infos from Maps every %v", period))
 	lastIdxDeleted := uint32(0)
 
 	for {