From 7ffc870520de60a696d17f378c367bd81d3ecd0f Mon Sep 17 00:00:00 2001 From: Daniil Porokhnin Date: Tue, 30 Jun 2026 17:44:35 +0300 Subject: [PATCH 1/4] fix: add .compaction-queue to handle in-mid compaction crash --- compaction/merge.go | 36 ++++++++++++++++++ compaction/planner.go | 8 +++- config.example.yaml | 2 +- consts/consts.go | 1 + fracmanager/frac_manifest.go | 74 ++++++++++++++++++++++++++++++++++++ 5 files changed, 118 insertions(+), 3 deletions(-) diff --git a/compaction/merge.go b/compaction/merge.go index ee6f6d88..fe5209a0 100644 --- a/compaction/merge.go +++ b/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.CompactionQueue, + filename+consts.CompactionQueue, + 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, diff --git a/compaction/planner.go b/compaction/planner.go index c44b80ed..7b8afbe4 100644 --- a/compaction/planner.go +++ b/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 queue. + util.RemoveFile(s.Info.Path + consts.CompactionQueue) }, }, true } diff --git a/config.example.yaml b/config.example.yaml index e4cf0a9a..1491645b 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -7,7 +7,7 @@ address: debug: :9200 storage: - frac_size: 16MiB + frac_size: 1MiB total_size: 1GiB compaction: diff --git a/consts/consts.go b/consts/consts.go index 3341aecd..4a2ccf71 100644 --- a/consts/consts.go +++ b/consts/consts.go @@ -88,6 +88,7 @@ const ( RemoteFractionSuffix = ".remote" FracCacheFileSuffix = ".frac-cache" + CompactionQueue = ".compaction-queue" // tracing JaegerDebugKey = "jaeger-debug-id" diff --git a/fracmanager/frac_manifest.go b/fracmanager/frac_manifest.go index ae4adc4a..fd408493 100644 --- a/fracmanager/frac_manifest.go +++ b/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 + + hasCompactionQueue 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.CompactionQueue: + m.hasCompactionQueue = true + case consts.IndexTmpFileSuffix, consts.InfoTmpFileSuffix, consts.TokenTmpFileSuffix, consts.OffsetsTmpFileSuffix, consts.IDTmpFileSuffix, consts.LIDTmpFileSuffix, @@ -148,6 +157,11 @@ func removeMeta(m *fracManifest) { } } +func removeCompactionQueue(m *fracManifest) { + util.RemoveFile(m.basePath + consts.CompactionQueue) + m.hasCompactionQueue = 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 { @@ -256,9 +277,60 @@ 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.hasCompactionQueue || + m.Stage() != fracStageSealed + + if skip { + continue + } + + f, err := os.Open(m.basePath + consts.CompactionQueue) + if err != nil { + return nil, err + } + + var p plan + if err := json.NewDecoder(f).Decode(&p); err != nil { + return nil, err + } + + for _, pname := range p.Participants { + pid := pname[len(fileBasePattern):] + + f := manifests[pid] + if f == nil { + return nil, errors.New("inconsistent fraction file analysis") + } + + delete(filtered, pid) + removeAllFiles(f.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 +356,7 @@ func cleanupRemoteFrac(m *fracManifest) { // Removes redundant files after finishing work with the fraction func cleanupSealedFrac(m *fracManifest) { removeMeta(m) + removeCompactionQueue(m) if m.hasSdocs { removeDocs(m) // remove orig docs, but keeping sorted } @@ -315,6 +388,7 @@ func removeAllFiles(basePath string) { consts.MetaFileSuffix, consts.WalFileSuffix, + consts.CompactionQueue, } { util.RemoveFile(basePath + suffix) } From 83e6bb39687519193877dbe1a0d19517c0ee8f6b Mon Sep 17 00:00:00 2001 From: Daniil Porokhnin Date: Fri, 3 Jul 2026 15:40:15 +0300 Subject: [PATCH 2/4] fix: close fd, skip deleted fractions --- fracmanager/frac_manifest.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/fracmanager/frac_manifest.go b/fracmanager/frac_manifest.go index fd408493..9ab47044 100644 --- a/fracmanager/frac_manifest.go +++ b/fracmanager/frac_manifest.go @@ -309,6 +309,7 @@ func dropCompacted(ids []string, manifests map[string]*fracManifest) ([]string, if err != nil { return nil, err } + defer f.Close() //nolint var p plan if err := json.NewDecoder(f).Decode(&p); err != nil { @@ -320,7 +321,9 @@ func dropCompacted(ids []string, manifests map[string]*fracManifest) ([]string, f := manifests[pid] if f == nil { - return nil, errors.New("inconsistent fraction file analysis") + // NOTE(dkharms): It is possible that compaction participants + // were dropped but the queue itself was not deleted. + continue } delete(filtered, pid) From a2f3666c7d56c1e59fb50a1d247276921d970593 Mon Sep 17 00:00:00 2001 From: Daniil Porokhnin Date: Fri, 3 Jul 2026 17:29:08 +0300 Subject: [PATCH 3/4] fix: set creation time --- compaction/merge_source.go | 10 ++++++++-- compaction/merge_source_test.go | 3 +++ 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/compaction/merge_source.go b/compaction/merge_source.go index 00e6e708..20d0bd7b 100644 --- a/compaction/merge_source.go +++ b/compaction/merge_source.go @@ -69,11 +69,13 @@ 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) } @@ -81,6 +83,10 @@ func (s *MergeSource) prepareInfo() *common.Info { 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 } diff --git a/compaction/merge_source_test.go b/compaction/merge_source_test.go index c4cb0fcc..c9ea83be 100644 --- a/compaction/merge_source_test.go +++ b/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, From 4d59887a0ede2b755bd231cbcad868e057bb975a Mon Sep 17 00:00:00 2001 From: Daniil Porokhnin Date: Fri, 10 Jul 2026 15:49:39 +0300 Subject: [PATCH 4/4] chore: review fixes --- compaction/merge.go | 4 +-- compaction/planner.go | 4 +-- consts/consts.go | 2 +- fracmanager/frac_manifest.go | 57 ++++++++++++++++++++++++++---------- 4 files changed, 47 insertions(+), 20 deletions(-) diff --git a/compaction/merge.go b/compaction/merge.go index fe5209a0..635a1bf2 100644 --- a/compaction/merge.go +++ b/compaction/merge.go @@ -102,8 +102,8 @@ func createCompactionPlan(filename string, srcs ...Source) error { // Later on, it will help us to make correct decision // whether we should drop participants (delete them from disk). return createAndWrite( - filename+consts.CompactionQueue, - filename+consts.CompactionQueue, + filename+consts.CompactionPlan, + filename+consts.CompactionPlan, func(f *os.File) error { var p plan diff --git a/compaction/planner.go b/compaction/planner.go index 7b8afbe4..a76bca63 100644 --- a/compaction/planner.go +++ b/compaction/planner.go @@ -189,8 +189,8 @@ func (p *planner) pick() (task, bool) { csnapshot.Destroy() // We have destroyed all sealed fractions which participated - // in compaction and now stale. So we can drop compaction queue. - util.RemoveFile(s.Info.Path + consts.CompactionQueue) + // in compaction and now stale. So we can drop compaction plan. + util.RemoveFile(s.Info.Path + consts.CompactionPlan) }, }, true } diff --git a/consts/consts.go b/consts/consts.go index 4a2ccf71..422f229e 100644 --- a/consts/consts.go +++ b/consts/consts.go @@ -88,7 +88,7 @@ const ( RemoteFractionSuffix = ".remote" FracCacheFileSuffix = ".frac-cache" - CompactionQueue = ".compaction-queue" + CompactionPlan = ".compaction-plan" // tracing JaegerDebugKey = "jaeger-debug-id" diff --git a/fracmanager/frac_manifest.go b/fracmanager/frac_manifest.go index 9ab47044..f899bacc 100644 --- a/fracmanager/frac_manifest.go +++ b/fracmanager/frac_manifest.go @@ -42,7 +42,7 @@ type fracManifest struct { hasSdocsDel bool // sorted documents deletion marker hasIndexDel bool // index deletion marker - hasCompactionQueue bool + hasCompactionPlan bool } // hasAllIndexFiles reports whether all 5 split index files are present. @@ -85,8 +85,8 @@ func (m *fracManifest) AddExtension(ext string) error { case consts.IndexDelFileSuffix: m.hasIndexDel = true - case consts.CompactionQueue: - m.hasCompactionQueue = true + case consts.CompactionPlan: + m.hasCompactionPlan = true case consts.IndexTmpFileSuffix, consts.InfoTmpFileSuffix, consts.TokenTmpFileSuffix, consts.OffsetsTmpFileSuffix, @@ -157,9 +157,9 @@ func removeMeta(m *fracManifest) { } } -func removeCompactionQueue(m *fracManifest) { - util.RemoveFile(m.basePath + consts.CompactionQueue) - m.hasCompactionQueue = false +func removeCompactionPlan(m *fracManifest) { + util.RemoveFile(m.basePath + consts.CompactionPlan) + m.hasCompactionPlan = false } func removeIndexFiles(m *fracManifest) { @@ -263,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) @@ -298,14 +308,14 @@ func dropCompacted(ids []string, manifests map[string]*fracManifest) ([]string, return nil, errors.New("inconsistent fraction file analysis") } - skip := !m.hasCompactionQueue || + skip := !m.hasCompactionPlan || m.Stage() != fracStageSealed if skip { continue } - f, err := os.Open(m.basePath + consts.CompactionQueue) + f, err := os.Open(m.basePath + consts.CompactionPlan) if err != nil { return nil, err } @@ -313,21 +323,38 @@ func dropCompacted(ids []string, manifests map[string]*fracManifest) ([]string, var p plan if err := json.NewDecoder(f).Decode(&p); err != nil { - return nil, err + // 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):] - f := manifests[pid] - if f == nil { + pm := manifests[pid] + if pm == nil { // NOTE(dkharms): It is possible that compaction participants - // were dropped but the queue itself was not deleted. + // 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(f.basePath) + removeAllFiles(pm.basePath) } } @@ -359,7 +386,7 @@ func cleanupRemoteFrac(m *fracManifest) { // Removes redundant files after finishing work with the fraction func cleanupSealedFrac(m *fracManifest) { removeMeta(m) - removeCompactionQueue(m) + removeCompactionPlan(m) if m.hasSdocs { removeDocs(m) // remove orig docs, but keeping sorted } @@ -391,7 +418,7 @@ func removeAllFiles(basePath string) { consts.MetaFileSuffix, consts.WalFileSuffix, - consts.CompactionQueue, + consts.CompactionPlan, } { util.RemoveFile(basePath + suffix) }