fix: add .compaction-plan to handle in-mid compaction crash

compaction/merge.go

@@ -1,6 +1,7 @@
 package compaction
 
 import (
+	"encoding/json"
 	"errors"
 	"os"
 
@@ -14,6 +15,10 @@ func Merge(filename string, params common.SealParams, srcs ...Source) (*sealed.P
 	w := indexwriter.New(params)
 	src := NewMergeSource(filename, srcs)
 
+	if err := createCompactionPlan(filename, srcs...); err != nil {
+		return nil, err
+	}
+
 	if err := createAndWrite(
 		filename+consts.OffsetsTmpFileSuffix,
 		filename+consts.OffsetsFileSuffix,
@@ -83,6 +88,37 @@ func Merge(filename string, params common.SealParams, srcs ...Source) (*sealed.P
 	return preloaded, nil
 }
 
+func createCompactionPlan(filename string, srcs ...Source) error {
+	type plan struct {
+		Participants []string `json:"participants"`
+	}
+
+	// Write which fractions are participating in compaction.
+	//
+	// Since we cannot remove all stale fractions and move
+	// the merged one into final state, we have to keep our
+	// intention on disk.
+	//
+	// Later on, it will help us to make correct decision
+	// whether we should drop participants (delete them from disk).
+	return createAndWrite(
+		filename+consts.CompactionPlan,
+		filename+consts.CompactionPlan,
+		func(f *os.File) error {
+			var p plan
+
+			for i := range srcs {
+				p.Participants = append(
+					p.Participants,
+					srcs[i].Info().Name(),
+				)
+			}
+
+			return json.NewEncoder(f).Encode(p)
+		},
+	)
+}
+
 func mergeDocs(filename string, srcs ...Source) error {
 	return createAndWrite(
 		filename+consts.DocsTmpFileSuffix,

compaction/merge_source.go

@@ -69,18 +69,24 @@ func (s *MergeSource) prepareInfo() *common.Info {
 	info := common.NewInfo(s.filename, 0, 0)
 
 	var (
-		from seq.MID = seq.MaxID.MID
-		to   seq.MID = seq.MinID.MID
+		from     seq.MID = seq.MaxID.MID
+		to       seq.MID = seq.MinID.MID
+		creation uint64  = info.CreationTime
 	)
 
 	for _, src := range s.sources {
+		creation = min(creation, src.Info().CreationTime)
 		from = min(from, src.Info().From)
 		to = max(to, src.Info().To)
 	}
 
 	info.From, info.To = from, to
 	info.SealingTime = info.CreationTime
 
+	// TODO(dkharms): This is workaround which will
+	// force compacted fraction to stay in the same bin.
+	info.CreationTime = creation
+
 	info.InitEmptyDistribution()
 	return info
 }

compaction/merge_source_test.go

@@ -7,6 +7,7 @@ import (
 	"math/rand"
 	"slices"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -29,6 +30,8 @@ func (m *mockSealingSource) Info() *common.Info {
 		DocsTotal:  uint32(len(m.ids)),
 		DocsOnDisk: m.docsOnDisk,
 
+		CreationTime: uint64(time.Now().UnixMilli()),
+
 		From: slices.MinFunc(m.ids, func(x, y seq.ID) int {
 			return cmp.Compare(x.MID, y.MID)
 		}).MID,

compaction/planner.go

@@ -11,6 +11,7 @@ import (
 
 	"go.uber.org/zap"
 
+	"github.com/ozontech/seq-db/consts"
 	"github.com/ozontech/seq-db/frac/common"
 	"github.com/ozontech/seq-db/frac/sealed"
 	"github.com/ozontech/seq-db/fracmanager"
@@ -183,10 +184,13 @@ func (p *planner) pick() (task, bool) {
 				}
 
 				compactionResultTotal.WithLabelValues(bucketSize, "success").Inc()
-				// TODO(dkharms): Is it fine to substitute and delete?
-				// We need somehow substitute and delete atomically.
+
 				p.fm.SubstituteWithSealed(s, csnapshot)
 				csnapshot.Destroy()
+
+				// We have destroyed all sealed fractions which participated
+				// in compaction and now stale. So we can drop compaction plan.
+				util.RemoveFile(s.Info.Path + consts.CompactionPlan)
 			},
 		}, true
 	}

config.example.yaml

@@ -7,7 +7,7 @@ address:
   debug: :9200
 
 storage:
-  frac_size: 16MiB
+  frac_size: 1MiB
   total_size: 1GiB
 
 compaction:

consts/consts.go

@@ -88,6 +88,7 @@ const (
 	RemoteFractionSuffix = ".remote"
 
 	FracCacheFileSuffix = ".frac-cache"
+	CompactionPlan      = ".compaction-plan"
 
 	// tracing
 	JaegerDebugKey = "jaeger-debug-id"

fracmanager/frac_manifest.go

@@ -1,9 +1,13 @@
 package fracmanager
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
+	"os"
 	"path/filepath"
+	"slices"
 	"sort"
 	"strings"
 
@@ -37,6 +41,8 @@ type fracManifest struct {
 	hasDocsDel  bool // documents deletion marker
 	hasSdocsDel bool // sorted documents deletion marker
 	hasIndexDel bool // index deletion marker
+
+	hasCompactionPlan bool
 }
 
 // hasAllIndexFiles reports whether all 5 split index files are present.
@@ -79,6 +85,9 @@ func (m *fracManifest) AddExtension(ext string) error {
 	case consts.IndexDelFileSuffix:
 		m.hasIndexDel = true
 
+	case consts.CompactionPlan:
+		m.hasCompactionPlan = true
+
 	case consts.IndexTmpFileSuffix, consts.InfoTmpFileSuffix,
 		consts.TokenTmpFileSuffix, consts.OffsetsTmpFileSuffix,
 		consts.IDTmpFileSuffix, consts.LIDTmpFileSuffix,
@@ -148,6 +157,11 @@ func removeMeta(m *fracManifest) {
 	}
 }
 
+func removeCompactionPlan(m *fracManifest) {
+	util.RemoveFile(m.basePath + consts.CompactionPlan)
+	m.hasCompactionPlan = false
+}
+
 func removeIndexFiles(m *fracManifest) {
 	for _, suffix := range []string{
 		consts.InfoFileSuffix,
@@ -233,7 +247,14 @@ func analyzeFiles(files []string) ([]*fracManifest, error) {
 // filterValid filters valid fractions and handles invalid ones
 // Removes partially deleted and unknown fractions
 func filterValid(ids []string, manifests map[string]*fracManifest) ([]*fracManifest, error) {
+	// We need to drop stale (compacted) fractions first.
+	ids, err := dropCompacted(ids, manifests)
+	if err != nil {
+		return nil, err
+	}
+
 	validated := make([]*fracManifest, 0, len(manifests))
+
 	for _, id := range ids {
 		manifest := manifests[id]
 		if manifest == nil {
@@ -242,6 +263,16 @@ func filterValid(ids []string, manifests map[string]*fracManifest) ([]*fracManif
 
 		switch manifest.Stage() {
 		case fracStageUnknown:
+			// Processing partially compacted fraction.
+			if manifest.hasCompactionPlan {
+				logger.Warn(
+					"dropping partially compacted fraction",
+					zap.String("base_path", manifest.basePath),
+				)
+				removeAllFiles(manifest.basePath)
+				continue
+			}
+
 			logger.Error("unknown fraction stage", zap.Object("manifest", manifest))
 			fractionLoadErrors.Inc()
 			removeAllFiles(manifest.basePath)
@@ -256,9 +287,80 @@ func filterValid(ids []string, manifests map[string]*fracManifest) ([]*fracManif
 		cleanupFrac(manifest)
 		validated = append(validated, manifest)
 	}
+
 	return validated, nil
 }
 
+func dropCompacted(ids []string, manifests map[string]*fracManifest) ([]string, error) {
+	type plan struct {
+		Participants []string `json:"participants"`
+	}
+
+	filtered := make(map[string]struct{})
+
+	for _, id := range ids {
+		filtered[id] = struct{}{}
+	}
+
+	for _, id := range ids {
+		m := manifests[id]
+		if m == nil {
+			return nil, errors.New("inconsistent fraction file analysis")
+		}
+
+		skip := !m.hasCompactionPlan ||
+			m.Stage() != fracStageSealed
+
+		if skip {
+			continue
+		}
+
+		f, err := os.Open(m.basePath + consts.CompactionPlan)
+		if err != nil {
+			return nil, err
+		}
+		defer f.Close() //nolint
+
+		var p plan
+		if err := json.NewDecoder(f).Decode(&p); err != nil {
+			// Well, we cannot decode compaction plan so let's drop whatever the result is.
+
+			logger.Warn(
+				"dropping possibly correctly compacted fraction: cannot decode compaction plan",
+				zap.Error(err),
+				zap.String("base_path", m.basePath),
+			)
+
+			delete(filtered, id)
+			removeAllFiles(m.basePath)
+			continue
+		}
+
+		for _, pname := range p.Participants {
+			pid := pname[len(fileBasePattern):]
+
+			pm := manifests[pid]
+			if pm == nil {
+				// NOTE(dkharms): It is possible that compaction participants
+				// were dropped but the plan itself was not deleted.
+				continue
+			}
+
+			logger.Warn(
+				"dropping fraction: it was merged into another one",
+				zap.Error(err),
+				zap.String("merged_base_path", m.basePath),
+				zap.String("participant_base_path", pm.basePath),
+			)
+
+			delete(filtered, pid)
+			removeAllFiles(pm.basePath)
+		}
+	}
+
+	return slices.Collect(maps.Keys(filtered)), nil
+}
+
 // cleanupFrac performs cleanup of unnecessary files depending on fraction stage
 // Called after stage determination to optimize storage
 func cleanupFrac(m *fracManifest) {
@@ -284,6 +386,7 @@ func cleanupRemoteFrac(m *fracManifest) {
 // Removes redundant files after finishing work with the fraction
 func cleanupSealedFrac(m *fracManifest) {
 	removeMeta(m)
+	removeCompactionPlan(m)
 	if m.hasSdocs {
 		removeDocs(m) // remove orig docs, but keeping sorted
 	}
@@ -315,6 +418,7 @@ func removeAllFiles(basePath string) {
 
 		consts.MetaFileSuffix,
 		consts.WalFileSuffix,
+		consts.CompactionPlan,
 	} {
 		util.RemoveFile(basePath + suffix)
 	}