fix(aggregator): use StartPeriodicSaving instead of custom loop

Author: razmser

internal/aggregator/aggregator.go

@@ -363,13 +363,7 @@ func MakeAggregator(dc pcache.DiskCache, fj *os.File, fjCompact *os.File, mappin
 
 	sh2.Run(a.aggregatorHost, a.shardKey, a.replicaKey)
 
-	go func() {
-		for {
-			time.Sleep(time.Hour) // arbitrary
-			_, _ = mappingsCache.Save()
-			a.SaveJournals()
-		}
-	}()
+	mappingsCache.StartPeriodicSaving()
 
 	return a, nil
 }

internal/metajournal/journal_fast.go

@@ -572,9 +572,8 @@ func (ms *JournalFast) StartPeriodicSaving() {
 
 func (ms *JournalFast) goPeriodicSaving() {
 	for {
-		sleepDuration := ms.periodicSaveInterval + time.Duration(rand.Intn(int(ms.periodicSaveInterval)))*time.Second
+		sleepDuration := ms.periodicSaveInterval + time.Duration(rand.Intn(int(ms.periodicSaveInterval)))
 		time.Sleep(sleepDuration)
-
 		ok, version, err := ms.Save()
 		if err != nil {
 			log.Printf("Periodic journal save failed: %v", err)

internal/pcache/mappings_cache.go

@@ -329,6 +329,7 @@ func (c *MappingsCache) AddValues(nowUnix uint32, pairs []MappingPair) {
 		for _, p := range pairs {
 			c.addItem(p.Str, p.Value, nowUnix)
 		}
+		c.version += 1
 		return
 	}
 	removeSize := c.sumSize + newItemsSizeMem - maxSize
@@ -542,7 +543,7 @@ func (c *MappingsCache) StartPeriodicSaving() {
 
 func (c *MappingsCache) goPeriodicSaving() {
 	for {
-		sleepDuration := c.periodicSaveInterval + time.Duration(rand.Intn(int(c.periodicSaveInterval)))*time.Second
+		sleepDuration := c.periodicSaveInterval + time.Duration(rand.Intn(int(c.periodicSaveInterval)))
 		time.Sleep(sleepDuration)
 
 		ok, err := c.Save()