chore: remove compatibility with .meta file

cmd/seq-db/seq-db.go

@@ -284,7 +284,7 @@ func startStore(
 					},
 				},
 				SkipSortDocs: !cfg.DocsSorting.Enabled,
-				KeepMetaFile: false,
+				KeepWalFile:  false,
 			},
 			OffloadingEnabled:    cfg.Offloading.Enabled,
 			OffloadingRetention:  cfg.Offloading.Retention,

consts/consts.go

@@ -52,8 +52,7 @@ const (
 	BrokenDir = ".broken"
 
 	// known extensions
-	MetaFileSuffix = ".meta"
-	WalFileSuffix  = ".wal"
+	WalFileSuffix = ".wal"
 
 	DocsFileSuffix    = ".docs"
 	DocsTmpFileSuffix = "._docs"

docs/en/13-architecture.md

@@ -24,7 +24,7 @@ seq-db store keeps all document data in three file types:
 | File type | Purpose                                        |
 |-----------|------------------------------------------------|
 | `.docs`   | Stores compressed batches of raw log documents |
-| `.meta`   | Tokenized metadata stream (used for recovery)  |
+| `.wal`    | Tokenized metadata stream (used for recovery)  |
 | `.index`  | On-disk inverted index                         | 
 
 
@@ -38,7 +38,7 @@ Read more about file types and their internal structure [here](internal/fraction
 A write operation is acknowledged only after the payload is safely persisted:
 
 ```
-write, fsync   # .meta file
+write, fsync   # .wal file
 write, fsync   # .data file
 ```
 That is, two write system calls followed by two fsync

docs/ru/13-architecture.md

@@ -21,7 +21,7 @@ seq-db store хранит все данные документов в трех 
 | Тип файла | Назначение                                             |
 |-----------|--------------------------------------------------------|
 | `.docs`   | Хранит сжатые батчи сырых документов логов             |
-| `.meta`   | Токенизированный поток метаданных (для восстановления) |
+| `.wal`    | Токенизированный поток метаданных (для восстановления) |
 | `.index`  | Дисковый обратный индекс.                              |
 
 Поскольку набор данных хранится в этих трех типах файлов, перемещение или восстановление шарда выполняется просто: достаточно скопировать (`cp` / `rsync`) директорию на целевой узел и запустить под.
@@ -32,7 +32,7 @@ seq-db store хранит все данные документов в трех 
 Операция записи подтверждается только после того, как полезная нагрузка гарантированно сохранена:
 
 ```
-write, fsync   # файл .meta
+write, fsync   # файл .wal
 write, fsync   # файл .data
 ```
 То есть, два системных вызова `write`, за которыми следуют два вызова `fsync` — это гарантирует, что данные переживут аварию узла или его перезапуск до того, как клиент получит ответ об успехе.

frac/active.go

@@ -3,7 +3,6 @@ package frac
 import (
 	"context"
 	"fmt"
-	"io"
 	"os"
 	"path/filepath"
 	"sync"
@@ -52,9 +51,8 @@ type Active struct {
 	docsCache  *cache.Cache[[]byte]
 	sortCache  *cache.Cache[[]byte]
 
-	metaFile   *os.File
-	metaReader *storage.DocBlocksReader
-	walReader  *storage.WalReader
+	walFile   *os.File
+	walReader *storage.WalReader
 
 	writer  *ActiveWriter
 	indexer *ActiveIndexer
@@ -72,8 +70,7 @@ func NewActive(
 	skipMaskProvider skipMaskProvider,
 ) *Active {
 	docsFile, docsStats := mustOpenFile(baseFileName+consts.DocsFileSuffix, config.SkipFsync)
-
-	metaFile, writer, metaReader, walReader, metaSize := mustOpenMetaWriter(baseFileName, readLimiter, docsFile, docsStats)
+	walFile, writer, walReader, walSize := mustOpenWalWriter(baseFileName, readLimiter, docsFile, docsStats)
 
 	f := &Active{
 		TokenList:     NewActiveTokenList(config.IndexWorkers),
@@ -89,15 +86,14 @@ func NewActive(
 		docsReader: storage.NewDocsReader(readLimiter, docsFile, docsCache),
 		sortReader: storage.NewDocsReader(readLimiter, docsFile, sortCache),
 
-		metaFile:   metaFile,
-		metaReader: metaReader,
-		walReader:  walReader,
+		walFile:   walFile,
+		walReader: walReader,
 
 		indexer: activeIndexer,
 		writer:  writer,
 
 		BaseFileName: baseFileName,
-		info:         common.NewInfo(baseFileName, uint64(docsStats.Size()), metaSize),
+		info:         common.NewInfo(baseFileName, uint64(docsStats.Size()), walSize),
 		Config:       cfg,
 
 		skipMaskProvider: skipMaskProvider,
@@ -112,34 +108,21 @@ func NewActive(
 	return f
 }
 
-func mustOpenMetaWriter(
+func mustOpenWalWriter(
 	baseFileName string,
 	readLimiter *storage.ReadLimiter,
 	docsFile *os.File,
 	docsStats os.FileInfo,
-) (*os.File, *ActiveWriter, *storage.DocBlocksReader, *storage.WalReader, uint64) {
-	legacyMetaFileName := baseFileName + consts.MetaFileSuffix
-
-	if _, err := os.Stat(legacyMetaFileName); err == nil {
-		// .meta file exists
-		metaFile, metaStats := mustOpenFile(legacyMetaFileName, config.SkipFsync)
-		metaSize := uint64(metaStats.Size())
-		metaReader := storage.NewDocBlocksReader(readLimiter, metaFile)
-		writer := NewActiveWriterLegacy(docsFile, metaFile, docsStats.Size(), metaStats.Size(), config.SkipFsync)
-		logger.Info("using legacy meta file format", zap.String("fraction", baseFileName))
-		return metaFile, writer, &metaReader, nil, metaSize
-	}
-
-	logger.Info("using new WAL format", zap.String("fraction", baseFileName))
+) (*os.File, *ActiveWriter, *storage.WalReader, uint64) {
 	walFileName := baseFileName + consts.WalFileSuffix
-	metaFile, metaStats := mustOpenFile(walFileName, config.SkipFsync)
-	metaSize := uint64(metaStats.Size())
-	writer := NewActiveWriter(docsFile, metaFile, docsStats.Size(), metaStats.Size(), config.SkipFsync)
-	walReader, err := storage.NewWalReader(readLimiter, metaFile, baseFileName)
+	walFile, walStats := mustOpenFile(walFileName, config.SkipFsync)
+	walSize := uint64(walStats.Size())
+	writer := NewActiveWriter(docsFile, walFile, docsStats.Size(), walStats.Size(), config.SkipFsync)
+	walReader, err := storage.NewWalReader(readLimiter, walFile, baseFileName)
 	if err != nil {
 		logger.Fatal("failed to initialize WAL reader", zap.String("fraction", baseFileName), zap.Error(err))
 	}
-	return metaFile, writer, nil, walReader, metaSize
+	return walFile, writer, walReader, walSize
 }
 
 func mustOpenFile(name string, skipFsync bool) (*os.File, os.FileInfo) {
@@ -161,13 +144,6 @@ func mustOpenFile(name string, skipFsync bool) (*os.File, os.FileInfo) {
 }
 
 func (f *Active) Replay(ctx context.Context) error {
-	if f.metaReader != nil {
-		return f.replayMetaFileLegacy(ctx)
-	}
-	return f.replayWalFile(ctx)
-}
-
-func (f *Active) replayWalFile(ctx context.Context) error {
 	logger.Info("start replaying WAL file...", zap.String("name", f.info.Name()))
 
 	t := time.Now()
@@ -258,72 +234,6 @@ func (f *Active) backupCorruptedFiles() error {
 	return nil
 }
 
-// replayMetaFileLegacy replays legacy *.meta files. Only basic corruption detection support is implemented
-func (f *Active) replayMetaFileLegacy(ctx context.Context) error {
-	logger.Info("start replaying...", zap.String("name", f.info.Name()))
-
-	t := time.Now()
-
-	offset := uint64(0)
-	step := f.info.MetaOnDisk / 10
-	next := step
-
-	sw := stopwatch.New()
-	wg := sync.WaitGroup{}
-
-out:
-	for {
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		default:
-			meta, metaSize, err := f.metaReader.ReadDocBlock(int64(offset))
-			if err == io.EOF {
-				if metaSize != 0 {
-					logger.Warn("last meta block is partially written, skipping it")
-				}
-				break out
-			}
-			if err != nil && err != io.EOF {
-				return err
-			}
-
-			if offset > next {
-				next += step
-				progress := float64(offset) / float64(f.info.MetaOnDisk) * 100
-				logger.Info("replaying batch, meta",
-					zap.String("name", f.info.Name()),
-					zap.Uint64("from", offset),
-					zap.Uint64("to", offset+metaSize),
-					zap.Uint64("target", f.info.MetaOnDisk),
-					util.ZapFloat64WithPrec("progress_percentage", progress, 2),
-				)
-			}
-			offset += metaSize
-
-			wg.Add(1)
-
-			walBlock := storage.PackDocBlockToWalBlock(meta)
-			f.indexer.Index(f, walBlock, &wg, sw)
-		}
-	}
-
-	wg.Wait()
-
-	tookSeconds := util.DurationToUnit(time.Since(t), "s")
-	throughputRaw := util.SizeToUnit(f.info.DocsRaw, "mb") / tookSeconds
-	throughputMeta := util.SizeToUnit(f.info.MetaOnDisk, "mb") / tookSeconds
-	logger.Info("active fraction replayed",
-		zap.String("name", f.info.Name()),
-		zap.Uint32("docs_total", f.info.DocsTotal),
-		util.ZapUint64AsSizeStr("docs_size", f.info.DocsOnDisk),
-		util.ZapFloat64WithPrec("took_s", tookSeconds, 1),
-		util.ZapFloat64WithPrec("throughput_raw_mb_sec", throughputRaw, 1),
-		util.ZapFloat64WithPrec("throughput_meta_mb_sec", throughputMeta, 1),
-	)
-	return nil
-}
-
 var bulkStagesSeconds = promauto.NewHistogramVec(prometheus.HistogramOpts{
 	Namespace: "seq_db_store",
 	Subsystem: "bulk",
@@ -466,8 +376,8 @@ func (f *Active) IsIntersecting(from, to seq.MID) bool {
 func (f *Active) Release() {
 	f.releaseMem()
 
-	if !f.Config.KeepMetaFile {
-		util.RemoveFile(f.metaFile.Name())
+	if !f.Config.KeepWalFile {
+		util.RemoveFile(f.walFile.Name())
 	}
 
 	if !f.Config.SkipSortDocs {
@@ -479,7 +389,7 @@ func (f *Active) Release() {
 func (f *Active) Suicide() {
 	f.releaseMem()
 
-	util.RemoveFile(f.metaFile.Name())
+	util.RemoveFile(f.walFile.Name())
 	util.RemoveFile(f.docsFile.Name())
 	util.RemoveFile(f.BaseFileName + consts.SdocsFileSuffix)
 }
@@ -491,8 +401,8 @@ func (f *Active) releaseMem() {
 	f.docsCache.Release()
 	f.sortCache.Release()
 
-	if err := f.metaFile.Close(); err != nil {
-		logger.Error("can't close meta file", zap.String("frac", f.BaseFileName), zap.Error(err))
+	if err := f.walFile.Close(); err != nil {
+		logger.Error("can't close wal file", zap.String("frac", f.BaseFileName), zap.Error(err))
 	}
 	if err := f.docsFile.Close(); err != nil {
 		logger.Error("can't close docs file", zap.String("frac", f.BaseFileName), zap.Error(err))

frac/active_writer.go

@@ -9,27 +9,14 @@ import (
 
 type ActiveWriter struct {
 	docs *storage.FileWriter
-	meta MetaWriter
-}
-
-type MetaWriter interface {
-	Write(data storage.WalBlock, sw *stopwatch.Stopwatch) (int64, error)
-	Stop()
+	wal  *storage.WalWriter
 }
 
 // NewActiveWriter creates a writer for *.wal files
 func NewActiveWriter(docsFile, walFile *os.File, docsOffset, walOffset int64, skipFsync bool) *ActiveWriter {
 	return &ActiveWriter{
 		docs: storage.NewFileWriter(docsFile, docsOffset, skipFsync),
-		meta: storage.NewWalWriter(walFile, walOffset, skipFsync),
-	}
-}
-
-// NewActiveWriterLegacy creates a writer for *.meta files
-func NewActiveWriterLegacy(docsFile, metaFile *os.File, docsOffset, metaOffset int64, skipFsync bool) *ActiveWriter {
-	return &ActiveWriter{
-		docs: storage.NewFileWriter(docsFile, docsOffset, skipFsync),
-		meta: NewLegacyMetaWriter(storage.NewFileWriter(metaFile, metaOffset, skipFsync)),
+		wal:  storage.NewWalWriter(walFile, walOffset, skipFsync),
 	}
 }
 
@@ -45,13 +32,13 @@ func (a *ActiveWriter) Write(docs storage.DocBlock, meta storage.WalBlock, sw *s
 	meta.SetDocsOffset(uint64(offset))
 
 	m = sw.Start("write_meta")
-	_, err = a.meta.Write(meta, sw)
+	_, err = a.wal.Write(meta, sw)
 	m.Stop()
 
 	return err
 }
 
 func (a *ActiveWriter) Stop() {
 	a.docs.Stop()
-	a.meta.Stop()
+	a.wal.Stop()
 }

frac/config.go

@@ -4,7 +4,7 @@ type Config struct {
 	Search SearchConfig
 
 	SkipSortDocs bool
-	KeepMetaFile bool
+	KeepWalFile  bool
 }
 
 type SearchConfig struct {

frac/fraction_test.go

@@ -2516,7 +2516,7 @@ func (s *ActiveReplayedFractionTestSuite) SetupTest() {
 	s.SetupTestCommon()
 	// Setting this flags allows to keep meta and docs files on disk after Active.Release() is called
 	s.config.SkipSortDocs = true
-	s.config.KeepMetaFile = true
+	s.config.KeepWalFile = true
 
 	s.insertDocuments = func(bulks ...[]string) {
 		if s.fraction != nil {