Seal factions immediately after reply on start

# Conflicts:
#	fracmanager/fracmanager.go

Author: eguguchkin

fracmanager/fracmanager.go

@@ -63,6 +63,14 @@ type activeRef struct {
 	frac *proxyFrac
 }
 
+func (fm *FracManager) newActiveRef(active *frac.Active) activeRef {
+	f := &proxyFrac{active: active, fp: fm.fracProvider}
+	return activeRef{
+		frac: f,
+		ref:  &fracRef{instance: f},
+	}
+}
+
 func NewFracManager(cfg *Config) *FracManager {
 	FillConfigWithDefault(cfg)
 
@@ -280,11 +288,19 @@ func (fm *FracManager) Start() {
 
 func (fm *FracManager) Load(ctx context.Context) error {
 	var err error
-	var notSealed []activeRef
 
 	l := NewLoader(fm.config, fm.fracProvider, fm.fracCache)
 
-	if fm.fracs, notSealed, err = l.load(ctx); err != nil {
+	actives, sealed, err := l.load()
+	if err != nil {
+		return err
+	}
+
+	for _, s := range sealed {
+		fm.fracs = append(fm.fracs, &fracRef{instance: s})
+	}
+
+	if err := fm.replayAll(ctx, actives); err != nil {
 		return err
 	}
 
@@ -298,16 +314,39 @@ func (fm *FracManager) Load(ctx context.Context) error {
 		}
 	}
 
-	if len(notSealed) == 0 {
-		fm.rotate()
-	} else {
-		if len(notSealed) > 1 {
-			logger.Info("sealing active fractions")
-			for _, active := range notSealed[:len(notSealed)-1] {
-				fm.seal(active)
-			}
+	if fm.active.ref == nil { // no active
+		_ = fm.rotate() // make new empty active
+	}
+
+	return nil
+}
+
+func (fm *FracManager) replayAll(ctx context.Context, actives []*frac.Active) error {
+	wg := sync.WaitGroup{}
+	defer wg.Wait()
+
+	for i, a := range actives {
+		if err := a.Replay(ctx); err != nil {
+			return err
+		}
+		if a.Info().DocsTotal == 0 {
+			a.Suicide() // remove empty
+			continue
+		}
+		r := fm.newActiveRef(a)
+		fm.fracs = append(fm.fracs, r.ref)
+
+		if i == len(actives)-1 { // last and not empty
+			fm.active = r
+		} else {
+			wg.Wait() // wait previous sealing complete
+
+			wg.Add(1)
+			go func() {
+				fm.seal(r)
+				wg.Done()
+			}()
 		}
-		fm.active = notSealed[len(notSealed)-1]
 	}
 
 	return nil
@@ -371,7 +410,7 @@ func (fm *FracManager) rotate() activeRef {
 	baseFilePath := filepath.Join(fm.config.DataDir, filePath)
 	logger.Info("creating new fraction", zap.String("filepath", baseFilePath))
 
-	next := fm.fracProvider.newActiveRef(fm.fracProvider.NewActive(baseFilePath))
+	next := fm.newActiveRef(fm.fracProvider.NewActive(baseFilePath))
 
 	fm.fracMu.Lock()
 	prev := fm.active

fracmanager/fraction_provider.go

@@ -69,11 +69,3 @@ func (fp *fractionProvider) NewSealedPreloaded(name string, preloadedData *frac.
 func (fp *fractionProvider) Stop() {
 	fp.activeIndexer.Stop()
 }
-
-func (fp *fractionProvider) newActiveRef(active *frac.Active) activeRef {
-	f := &proxyFrac{active: active, fp: fp}
-	return activeRef{
-		frac: f,
-		ref:  &fracRef{instance: f},
-	}
-}

fracmanager/loader.go

@@ -1,7 +1,6 @@
 package fracmanager
 
 import (
-	"context"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -44,7 +43,7 @@ func NewLoader(config *Config, fracProvider *fractionProvider, fracCache *sealed
 	}
 }
 
-func (l *loader) load(ctx context.Context) ([]*fracRef, []activeRef, error) {
+func (l *loader) load() ([]*frac.Active, []*frac.Sealed, error) {
 	fracIDs, infos := l.makeInfos(l.getFileList())
 	sort.Strings(fracIDs)
 
@@ -58,7 +57,7 @@ func (l *loader) load(ctx context.Context) ([]*fracRef, []activeRef, error) {
 	infosList := l.filterInfos(fracIDs, infos)
 	cnt := len(infosList)
 
-	fracs := make([]*fracRef, 0, cnt)
+	fracs := make([]*frac.Sealed, 0, cnt)
 	actives := make([]*frac.Active, 0)
 
 	diskFracCache := NewFracCacheFromDisk(filepath.Join(l.config.DataDir, consts.FracCacheFileSuffix))
@@ -73,13 +72,13 @@ func (l *loader) load(ctx context.Context) ([]*fracRef, []activeRef, error) {
 				removeFile(info.base + consts.DocsFileSuffix)
 			}
 			sealed := l.loadSealedFrac(diskFracCache, info)
-			fracs = append(fracs, &fracRef{instance: sealed})
+			fracs = append(fracs, sealed)
 		} else {
 			if info.hasMeta {
 				actives = append(actives, l.fracProvider.NewActive(info.base))
 			} else {
 				sealed := l.loadSealedFrac(diskFracCache, info)
-				fracs = append(fracs, &fracRef{instance: sealed})
+				fracs = append(fracs, sealed)
 			}
 		}
 
@@ -97,22 +96,7 @@ func (l *loader) load(ctx context.Context) ([]*fracRef, []activeRef, error) {
 
 	logger.Info("fractions list created", zap.Int("cached", l.cachedFracs), zap.Int("uncached", l.uncachedFracs))
 
-	logger.Info("replaying active fractions", zap.Int("count", len(actives)))
-	notSealed := make([]activeRef, 0)
-	for _, a := range actives {
-		if err := a.Replay(ctx); err != nil {
-			return nil, nil, fmt.Errorf("while replaying blocks: %w", err)
-		}
-		if a.Info().DocsTotal == 0 { // skip empty
-			removeFractionFiles(a.BaseFileName)
-			continue
-		}
-		activeRef := l.fracProvider.newActiveRef(a)
-		fracs = append(fracs, activeRef.ref)
-		notSealed = append(notSealed, activeRef)
-	}
-
-	return fracs, notSealed, nil
+	return actives, fracs, nil
 }
 
 func (l *loader) loadSealedFrac(diskFracCache *sealedFracCache, info *fracInfo) *frac.Sealed {