diff --git a/db/state/aggregator.go b/db/state/aggregator.go index afc8d8570e3..79b829c71ef 100644 --- a/db/state/aggregator.go +++ b/db/state/aggregator.go @@ -1972,7 +1972,7 @@ func (at *AggregatorRoTx) mergeFiles(ctx context.Context, files *FilesForMerge, rng := rng g.Go(func() error { var err error - mf.iis[id], err = at.iis[id].mergeFiles(ctx, files.ii[id], rng.from, rng.to, at.a.ps) + mf.iis[id], err = at.iis[id].mergeFiles(ctx, files.ii[id], rng.from, rng.to, at.a.ps, nil) return err }) } diff --git a/db/state/deduplicate.go b/db/state/deduplicate.go deleted file mode 100644 index eab1bc7c292..00000000000 --- a/db/state/deduplicate.go +++ /dev/null @@ -1,445 +0,0 @@ -// Copyright 2022 The Erigon Authors -// This file is part of Erigon. -// -// Erigon is free software: you can redistribute it and/or modify -// it under the terms of the GNU Lesser General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// Erigon is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Lesser General Public License for more details. -// -// You should have received a copy of the GNU Lesser General Public License -// along with Erigon. If not, see . - -package state - -import ( - "bytes" - "container/heap" - "context" - "fmt" - "path" - - "github.com/erigontech/erigon/common/background" - "github.com/erigontech/erigon/common/log/v3" - "github.com/erigontech/erigon/db/kv" - "github.com/erigontech/erigon/db/recsplit/multiencseq" - "github.com/erigontech/erigon/db/seg" -) - -// this is an internal function which helps to filter content of the history snapshots files -// and deduplicate values in it. -// -// This function is supposed to be used only as part of the snapshot tooling -// to help rebuilding existing snapshots. It should not be used for for -// background merging process because it is not memory-efficient -func (ht *HistoryRoTx) deduplicateFiles(ctx context.Context, indexFiles, historyFiles []*FilesItem, r HistoryRanges, ps *background.ProgressSet) error { - if !r.any() { - return nil - } - - if len(indexFiles) > 1 || len(historyFiles) > 1 { - return fmt.Errorf("wrong deduplication interval from %d to %d", r.history.from, r.history.to) - } - - var decomp *seg.Decompressor - - fromStep, toStep := kv.Step(r.history.from/ht.stepSize), kv.Step(r.history.to/ht.stepSize) - datPath := ht.h.vNewFilePath(fromStep, toStep) - idxPath := ht.h.vAccessorNewFilePath(fromStep, toStep) - - comp, err := seg.NewCompressor(ctx, "dedup hist "+ht.h.FilenameBase, datPath, ht.h.dirs.Tmp, ht.h.CompressorCfg, log.LvlTrace, ht.h.logger) - if err != nil { - return fmt.Errorf("deduo %s history compressor: %w", ht.h.FilenameBase, err) - } - - pagedWr := ht.dataWriter(ctx, comp) - - var cp CursorHeap - heap.Init(&cp) - - dedupKeyEFs := make(map[string]map[uint64]struct{}) // string because slice can not be a key in Golang - - for _, item := range indexFiles { - defer item.closeFilesAndRemove() - - g := ht.iit.dataReader(item.decompressor) - g.Reset(0) - if g.HasNext() { - var g2 *seg.PagedReader - for _, hi := range historyFiles { // full-scan, because it's ok to have different amount files. by unclean-shutdown. - if hi.startTxNum == item.startTxNum && hi.endTxNum == item.endTxNum { - compressedPageValuesCount := hi.decompressor.CompressedPageValuesCount() - - if hi.decompressor.CompressionFormatVersion() == seg.FileCompressionFormatV0 { - compressedPageValuesCount = ht.h.HistoryValuesOnCompressedPage - } - - g2 = seg.NewPagedReader(ht.dataReader(hi.decompressor), compressedPageValuesCount, true) - break - } - } - if g2 == nil { - panic(fmt.Sprintf("for file: %s, not found corresponding file to merge", g.FileName())) - } - key, _ := g.Next(nil) - val, _ := g.Next(nil) - heap.Push(&cp, &CursorItem{ - t: FILE_CURSOR, - kvReader: g, - hist: g2, - key: key, - val: val, - startTxNum: item.startTxNum, - endTxNum: item.endTxNum, - reverse: false, - }) - } - } - // In the loop below, the pair `keyBuf=>valBuf` is always 1 item behind `lastKey=>lastVal`. - // `lastKey` and `lastVal` are taken from the top of the multi-way merge (assisted by the CursorHeap cp), but not processed right away - // instead, the pair from the previous iteration is processed first - `keyBuf=>valBuf`. After that, `keyBuf` and `valBuf` are assigned - // to `lastKey` and `lastVal` correspondingly, and the next step of multi-way merge happens. Therefore, after the multi-way merge loop - // (when CursorHeap cp is empty), there is a need to process the last pair `keyBuf=>valBuf`, because it was one step behind - var lastKey, valBuf []byte - var dedupCount int - for cp.Len() > 0 { - lastKey = append(lastKey[:0], cp[0].key...) - // Advance all the items that have this key (including the top) - for cp.Len() > 0 && bytes.Equal(cp[0].key, lastKey) { - ci1 := heap.Pop(&cp).(*CursorItem) - count := multiencseq.Count(ci1.startTxNum, ci1.val) - - seq := multiencseq.ReadMultiEncSeq(ci1.startTxNum, ci1.val) - ss := seq.Iterator(0) - - var dedupVal *[]byte - var histKeyBuf []byte - var prevTxNum uint64 - - for ss.HasNext() { - txNum, err := ss.Next() - if err != nil { - panic(fmt.Sprintf("failed to extract txNum from ef. File: %s Key: %x", ci1.kvReader.FileName(), ci1.key)) - } - - if !ci1.hist.HasNext() { - panic(fmt.Errorf("assert: no value??? %s, txNum=%d, count=%d, lastKey=%x, ci1.key=%x", ci1.hist.FileName(), txNum, count, lastKey, ci1.key)) - } - - v, _ := ci1.hist.Next(valBuf[:0]) - - // if dedupVal == nil -> we can not insert to the page because next val can be duplicate --> remember the value and decide next iter - if dedupVal == nil { - dd := bytes.Clone(v) // i am not sure if there is a way to avoid extra copy here - dedupVal = &dd - prevTxNum = txNum - - continue - } - - // if dedupVal is the same as current --> can not insert to the page bacause next val can be duplicate as well --> mark prev tx as duplicate - if bytes.Equal(*dedupVal, v) { - if dedupKeyEFs[string(ci1.key)] == nil { - dedupKeyEFs[string(ci1.key)] = make(map[uint64]struct{}) - } - - dedupKeyEFs[string(ci1.key)][prevTxNum] = struct{}{} - prevTxNum = txNum - dedupCount++ - continue - } - - histKeyBuf = historyKey(prevTxNum, ci1.key, histKeyBuf) - - if err = pagedWr.Add(histKeyBuf, *dedupVal); err != nil { - return err - } - - dd := bytes.Clone(v) // i am not sure if there is a way to avoid extra copy here - dedupVal = &dd - prevTxNum = txNum - } - - if dedupVal != nil { - histKeyBuf = historyKey(prevTxNum, ci1.key, histKeyBuf) - - if err = pagedWr.Add(histKeyBuf, *dedupVal); err != nil { - return err - } - } - - // fmt.Printf("fput '%x'->%x\n", lastKey, ci1.val) - if ci1.kvReader.HasNext() { - ci1.key, _ = ci1.kvReader.Next(ci1.key[:0]) - ci1.val, _ = ci1.kvReader.Next(ci1.val[:0]) - heap.Push(&cp, ci1) - } - } - } - if err := pagedWr.Compress(); err != nil { - return err - } - comp.Close() - comp = nil - if decomp, err = seg.NewDecompressor(datPath); err != nil { - return err - } - - fmt.Println("Values to deduplicate:", dedupCount) - - indexIn, err := ht.iit.deduplicateFiles(ctx, indexFiles, r.index.from, r.index.to, ps, dedupKeyEFs) - if err != nil { - return err - } - - if err = ht.h.buildVI(ctx, idxPath, decomp, indexIn.decompressor, indexIn.startTxNum, ps); err != nil { - return err - } - - for _, item := range indexFiles { - if item.StartStep(ht.stepSize) == indexIn.StartStep(ht.stepSize) && item.EndStep(ht.stepSize) == indexIn.EndStep(ht.stepSize) { - continue - } - - item.closeFilesAndRemove() - } - - for _, item := range historyFiles { - if item.StartStep(ht.stepSize) == indexIn.StartStep(ht.stepSize) && item.EndStep(ht.stepSize) == indexIn.EndStep(ht.stepSize) { - continue - } - - item.closeFilesAndRemove() - } - - return nil -} - -func (iit *InvertedIndexRoTx) deduplicateFiles(ctx context.Context, files []*FilesItem, startTxNum, endTxNum uint64, ps *background.ProgressSet, dedupKeyEFs map[string]map[uint64]struct{}) (*FilesItem, error) { - if startTxNum == endTxNum { - panic(fmt.Sprintf("assert: startTxNum(%d) == endTxNum(%d)", startTxNum, endTxNum)) - } - - var outItem *FilesItem - var comp *seg.Compressor - var err error - var closeItem = true - defer func() { - if closeItem { - if comp != nil { - comp.Close() - } - if outItem != nil { - outItem.closeFilesAndRemove() - } - } - }() - if ctx.Err() != nil { - return nil, ctx.Err() - } - fromStep, toStep := kv.Step(startTxNum/iit.stepSize), kv.Step(endTxNum/iit.stepSize) - - datPath := iit.ii.efNewFilePath(fromStep, toStep) - if comp, err = seg.NewCompressor(ctx, iit.ii.FilenameBase+".ii.merge", datPath, iit.ii.dirs.Tmp, iit.ii.CompressorCfg, log.LvlTrace, iit.ii.logger); err != nil { - return nil, fmt.Errorf("merge %s inverted index compressor: %w", iit.ii.FilenameBase, err) - } - if iit.ii.noFsync { - comp.DisableFsync() - } - - write := iit.dataWriter(comp, false) - - cnt := 0 - for _, item := range files { - cnt += item.decompressor.Count() - } - p := ps.AddNew(path.Base(datPath), uint64(cnt/2)) - defer ps.Delete(p) - - var cp CursorHeap - heap.Init(&cp) - - for _, item := range files { - g := iit.dataReader(item.decompressor) - g.Reset(0) - if g.HasNext() { - key, _ := g.Next(nil) - val, _ := g.Next(nil) - //fmt.Printf("heap push %s [%d] %x\n", item.decompressor.FilePath(), item.endTxNum, key) - heap.Push(&cp, &CursorItem{ - t: FILE_CURSOR, - kvReader: g, - key: key, - val: val, - startTxNum: item.startTxNum, - endTxNum: item.endTxNum, - reverse: true, - }) - } - } - - // In the loop below, the pair `keyBuf=>valBuf` is always 1 item behind `lastKey=>lastVal`. - // `lastKey` and `lastVal` are taken from the top of the multi-way merge (assisted by the CursorHeap cp), but not processed right away - // instead, the pair from the previous iteration is processed first - `keyBuf=>valBuf`. After that, `keyBuf` and `valBuf` are assigned - // to `lastKey` and `lastVal` correspondingly, and the next step of multi-way merge happens. Therefore, after the multi-way merge loop - // (when CursorHeap cp is empty), there is a need to process the last pair `keyBuf=>valBuf`, because it was one step behind - var keyBuf, valBuf []byte - var lastKey, lastVal []byte - i := uint64(0) - for cp.Len() > 0 { - lastKey = append(lastKey[:0], cp[0].key...) - lastVal = append(lastVal[:0], cp[0].val...) - - // Pre-rebase the first sequence - preSeq := multiencseq.ReadMultiEncSeq(cp[0].startTxNum, lastVal) - preIt := preSeq.Iterator(0) - - var toInsert []uint64 - - for preIt.HasNext() { - v, err := preIt.Next() - if err != nil { - return nil, err - } - - if dedupKeyEFs != nil && dedupKeyEFs[string(lastKey)] != nil { - if _, ok := dedupKeyEFs[string(lastKey)][v]; ok { - continue - } - } - - toInsert = append(toInsert, v) - } - - newSeq := multiencseq.NewBuilder(startTxNum, uint64(len(toInsert)), preSeq.Max()) - for i := range toInsert { - newSeq.AddOffset(toInsert[i]) - } - newSeq.Build() - lastVal = newSeq.AppendBytes(nil) - var mergedOnce bool - - // Advance all the items that have this key (including the top) - for cp.Len() > 0 && bytes.Equal(cp[0].key, lastKey) { - ci1 := heap.Pop(&cp).(*CursorItem) - if mergedOnce { - var dedupEF map[uint64]struct{} - - if dedupKeyEFs != nil { - dedupEF = dedupKeyEFs[string(lastKey)] - } - - if lastVal, err = dedupNumSeqs(ci1.val, lastVal, ci1.startTxNum, startTxNum, nil, startTxNum, dedupEF); err != nil { - return nil, fmt.Errorf("merge %s inverted index: %w", iit.ii.FilenameBase, err) - } - } else { - mergedOnce = true - } - // fmt.Printf("multi-way %s [%d] %x\n", ii.KeysTable, ci1.endTxNum, ci1.key) - if ci1.kvReader.HasNext() { - ci1.key, _ = ci1.kvReader.Next(ci1.key[:0]) - ci1.val, _ = ci1.kvReader.Next(ci1.val[:0]) - i += 2 - // fmt.Printf("heap next push %s [%d] %x\n", ii.KeysTable, ci1.endTxNum, ci1.key) - heap.Push(&cp, ci1) - } - } - if i%1024 == 0 { - p.Processed.Store(i) - } - if keyBuf != nil { - // fmt.Printf("pput %x->%x\n", keyBuf, valBuf) - if _, err = write.Write(keyBuf); err != nil { - return nil, err - } - if _, err = write.Write(valBuf); err != nil { - return nil, err - } - } - keyBuf = append(keyBuf[:0], lastKey...) - if keyBuf == nil { - keyBuf = []byte{} - } - valBuf = append(valBuf[:0], lastVal...) - } - if keyBuf != nil { - // fmt.Printf("Put %x->%x\n", keyBuf, valBuf) - if _, err = write.Write(keyBuf); err != nil { - return nil, err - } - if _, err = write.Write(valBuf); err != nil { - return nil, err - } - } - if err = write.Compress(); err != nil { - return nil, err - } - comp.Close() - comp = nil - - outItem = newFilesItem(startTxNum, endTxNum) - if outItem.decompressor, err = seg.NewDecompressor(datPath); err != nil { - return nil, fmt.Errorf("merge %s decompressor [%d-%d]: %w", iit.ii.FilenameBase, startTxNum, endTxNum, err) - } - ps.Delete(p) - - if err := iit.ii.buildMapAccessor(ctx, fromStep, toStep, iit.dataReader(outItem.decompressor), ps); err != nil { - return nil, fmt.Errorf("merge %s buildHashMapAccessor [%d-%d]: %w", iit.ii.FilenameBase, startTxNum, endTxNum, err) - } - if outItem.index, err = iit.ii.openHashMapAccessor(iit.ii.efAccessorNewFilePath(fromStep, toStep)); err != nil { - return nil, err - } - - closeItem = false - return outItem, nil -} - -func dedupNumSeqs(preval, val []byte, preBaseNum, baseNum uint64, buf []byte, outBaseNum uint64, dedupEF map[uint64]struct{}) ([]byte, error) { - preSeq := multiencseq.ReadMultiEncSeq(preBaseNum, preval) - seq := multiencseq.ReadMultiEncSeq(baseNum, val) - preIt := preSeq.Iterator(0) - efIt := seq.Iterator(0) - - var toInsert []uint64 - - for preIt.HasNext() { - v, err := preIt.Next() - if err != nil { - return nil, err - } - - if dedupEF != nil { - if _, ok := dedupEF[v]; ok { - continue - } - } - - toInsert = append(toInsert, v) - } - for efIt.HasNext() { - v, err := efIt.Next() - if err != nil { - return nil, err - } - - if dedupEF != nil { - if _, ok := dedupEF[v]; ok { - continue - } - } - - toInsert = append(toInsert, v) - } - - newSeq := multiencseq.NewBuilder(outBaseNum, uint64(len(toInsert)), seq.Max()) - for i := range toInsert { - newSeq.AddOffset(toInsert[i]) - } - - newSeq.Build() - return newSeq.AppendBytes(buf), nil -} diff --git a/db/state/dirty_files.go b/db/state/dirty_files.go index 5db29de8e32..317d27069a0 100644 --- a/db/state/dirty_files.go +++ b/db/state/dirty_files.go @@ -352,6 +352,54 @@ func retireMergeFiles(dirtyFiles *DirtyFiles, outs []*FilesItem, filenameBase st return outs } +// openDirtyDataFile opens item's data file matched by mask in dirPath. +// Returns false when the file is missing or can't be opened; the caller must +// then treat item as invalid and skip opening its accessors. +func openDirtyDataFile(item *FilesItem, mask string, dirEntries []string, dirPath string, ver version.Versions, tag string, logger log.Logger) bool { + fPath, fileVer, found, err := version.MatchVersionedFile(mask, dirEntries, dirPath) + if err != nil { + logger.Debug("[agg] "+tag+": MatchVersionedFile error", "f", filepath.Base(fPath), "err", err) + return false + } + if !found { + logger.Debug("[agg] "+tag+": file does not exists", "f", mask) + return false + } + + fName := filepath.Base(fPath) + ver.MustSupport(fileVer, fName) + + if item.decompressor, err = seg.NewDecompressor(fPath); err != nil { + if errors.Is(err, &seg.ErrCompressedFileCorrupted{}) { + logger.Debug("[agg] "+tag, "err", err, "f", fName) + } else { + logger.Warn("[agg] "+tag, "err", err, "f", fName) + } + return false + } + return true +} + +// openDirtyAccessor opens an accessor file matched by mask in dirPath via open. +// Failures are logged but don't invalidate the item: a missing or broken +// accessor can be rebuilt from the data file. +func openDirtyAccessor(mask string, dirEntries []string, dirPath string, ver version.Versions, open func(fPath string) error, tag string, logger log.Logger) { + fPath, fileVer, found, err := version.MatchVersionedFile(mask, dirEntries, dirPath) + if err != nil { + logger.Warn("[agg] "+tag, "err", err, "f", filepath.Base(fPath)) + } + if !found { + return + } + + fName := filepath.Base(fPath) + ver.MustSupport(fileVer, fName) + + if err := open(fPath); err != nil { + logger.Warn("[agg] "+tag, "err", err, "f", fName) + } +} + func (d *Domain) openDirtyFiles(dirEntries []string) (err error) { var invalidFileItems []*FilesItem iter := d.dirtyFiles.Iter() @@ -359,83 +407,29 @@ func (d *Domain) openDirtyFiles(dirEntries []string) (err error) { item := iter.Item() fromStep, toStep := item.StepRange(d.stepSize) if item.decompressor == nil { - fNameMask := d.kvFileNameMask(fromStep, toStep) - fPath, fileVer, ok, err := version.MatchVersionedFile(fNameMask, dirEntries, d.dirs.SnapDomain) - if err != nil { - fName := filepath.Base(fPath) - d.logger.Debug("[agg] Domain.openDirtyFiles: FileExist err", "f", fName, "err", err) - invalidFileItems = append(invalidFileItems, item) - continue - } - if !ok { - fName := fNameMask - d.logger.Debug("[agg] Domain.openDirtyFiles: file does not exists", "f", fName) - invalidFileItems = append(invalidFileItems, item) - continue - } - - fName := filepath.Base(fPath) - d.FileVersion.DataKV.MustSupport(fileVer, fName) - - if item.decompressor, err = seg.NewDecompressor(fPath); err != nil { - if errors.Is(err, &seg.ErrCompressedFileCorrupted{}) { - d.logger.Debug("[agg] Domain.openDirtyFiles", "err", err, "f", fName) - } else { - d.logger.Warn("[agg] Domain.openDirtyFiles", "err", err, "f", fName) - } + if !openDirtyDataFile(item, d.kvFileNameMask(fromStep, toStep), dirEntries, d.dirs.SnapDomain, d.FileVersion.DataKV, "Domain.openDirtyFiles", d.logger) { invalidFileItems = append(invalidFileItems, item) - // don't interrupt on error. other files may be good. but skip indices open. continue } } if item.index == nil && d.Accessors.Has(statecfg.AccessorHashMap) { - fNameMask := d.kviAccessorFileNameMask(fromStep, toStep) - fPath, fileVer, ok, err := version.MatchVersionedFile(fNameMask, dirEntries, d.dirs.SnapDomain) - if err != nil { - fName := filepath.Base(fPath) - d.logger.Warn("[agg] Domain.openDirtyFiles", "err", err, "f", fName) - } - if ok { - fName := filepath.Base(fPath) - d.FileVersion.AccessorKVI.MustSupport(fileVer, fName) - if item.index, err = d.openHashMapAccessor(fPath); err != nil { - d.logger.Warn("[agg] Domain.openDirtyFiles", "err", err, "f", fName) - // don't interrupt on error. other files may be good - } - } + openDirtyAccessor(d.kviAccessorFileNameMask(fromStep, toStep), dirEntries, d.dirs.SnapDomain, d.FileVersion.AccessorKVI, func(fPath string) (err error) { + item.index, err = d.openHashMapAccessor(fPath) + return err + }, "Domain.openDirtyFiles", d.logger) } if item.bindex == nil && d.Accessors.Has(statecfg.AccessorBTree) { - fNameMask := d.kvBtAccessorFileNameMask(fromStep, toStep) - fPath, fileVer, ok, err := version.MatchVersionedFile(fNameMask, dirEntries, d.dirs.SnapDomain) - if err != nil { - fName := filepath.Base(fPath) - d.logger.Warn("[agg] Domain.openDirtyFiles", "err", err, "f", fName) - } - if ok { - fName := filepath.Base(fPath) - d.FileVersion.AccessorBT.MustSupport(fileVer, fName) - if item.bindex, err = btindex.OpenBtreeIndexWithDecompressor(fPath, btindex.DefaultBtreeM, d.dataReader(item.decompressor)); err != nil { - d.logger.Warn("[agg] Domain.openDirtyFiles", "err", err, "f", fName) - // don't interrupt on error. other files may be good - } - } + openDirtyAccessor(d.kvBtAccessorFileNameMask(fromStep, toStep), dirEntries, d.dirs.SnapDomain, d.FileVersion.AccessorBT, func(fPath string) (err error) { + item.bindex, err = btindex.OpenBtreeIndexWithDecompressor(fPath, btindex.DefaultBtreeM, d.dataReader(item.decompressor)) + return err + }, "Domain.openDirtyFiles", d.logger) } if item.existence == nil && d.Accessors.Has(statecfg.AccessorExistence) { - fNameMask := d.kvExistenceIdxFileNameMask(fromStep, toStep) - fPath, fileVer, ok, err := version.MatchVersionedFile(fNameMask, dirEntries, d.dirs.SnapDomain) - if err != nil { - fName := filepath.Base(fPath) - d.logger.Warn("[agg] Domain.openDirtyFiles", "err", err, "f", fName) - } - if ok { - fName := filepath.Base(fPath) - d.FileVersion.AccessorKVEI.MustSupport(fileVer, fName) - if item.existence, err = d.openExistenceFilter(fPath); err != nil { - d.logger.Warn("[agg] Domain.openDirtyFiles", "err", err, "f", fName) - // don't interrupt on error. other files may be good - } - } + openDirtyAccessor(d.kvExistenceIdxFileNameMask(fromStep, toStep), dirEntries, d.dirs.SnapDomain, d.FileVersion.AccessorKVEI, func(fPath string) (err error) { + item.existence, err = d.openExistenceFilter(fPath) + return err + }, "Domain.openDirtyFiles", d.logger) } } iter.Release() @@ -452,63 +446,17 @@ func (h *History) openDirtyFiles(dataEntries, accessorEntries []string) error { item := iter.Item() fromStep, toStep := item.StepRange(h.stepSize) if item.decompressor == nil { - fNameMask := h.vFileNameMask(fromStep, toStep) - fPath, fileVer, ok, err := version.MatchVersionedFile(fNameMask, dataEntries, h.dirs.SnapHistory) - if err != nil { - fName := filepath.Base(fPath) - h.logger.Debug("[agg] History.openDirtyFiles: FileExist", "f", fName, "err", err) - invalidFileItems = append(invalidFileItems, item) - continue - } - if !ok { - fName := fNameMask - h.logger.Debug("[agg] History.openDirtyFiles: file does not exists", "f", fName) + if !openDirtyDataFile(item, h.vFileNameMask(fromStep, toStep), dataEntries, h.dirs.SnapHistory, h.FileVersion.DataV, "History.openDirtyFiles", h.logger) { invalidFileItems = append(invalidFileItems, item) continue } - fName := filepath.Base(fPath) - h.FileVersion.DataV.MustSupport(fileVer, fName) - - if item.decompressor, err = seg.NewDecompressor(fPath); err != nil { - if errors.Is(err, &seg.ErrCompressedFileCorrupted{}) { - h.logger.Debug("[agg] History.openDirtyFiles", "err", err, "f", fName) - // TODO we do not restore those files so we could just remove them along with indices. Same for domains/indices. - // Those files will keep space on disk and closed automatically as corrupted. So better to remove them, and maybe remove downloading prohibiter to allow downloading them again? - // - // itemPaths := []string{ - // fPath, - // h.vAccessorFilePath(fromStep, toStep), - // } - // for _, fp := range itemPaths { - // err = dir.Remove(fp) - // if err != nil { - // h.logger.Warn("[agg] History.openDirtyFiles cannot remove corrupted file", "err", err, "f", fp) - // } - // } - } else { - h.logger.Warn("[agg] History.openDirtyFiles", "err", err, "f", fName) - } - invalidFileItems = append(invalidFileItems, item) - // don't interrupt on error. other files may be good. but skip indices open. - continue - } } if item.index == nil { - fNameMask := h.vAccessorFileNameMask(fromStep, toStep) - fPath, fileVer, ok, err := version.MatchVersionedFile(fNameMask, accessorEntries, h.dirs.SnapAccessors) - if err != nil { - fName := filepath.Base(fPath) - h.logger.Warn("[agg] History.openDirtyFiles", "err", err, "f", fName) - } - if ok { - fName := filepath.Base(fPath) - h.FileVersion.AccessorVI.MustSupport(fileVer, fName) - if item.index, err = h.openHashMapAccessor(fPath); err != nil { - h.logger.Warn("[agg] History.openDirtyFiles", "err", err, "f", fName) - // don't interrupt on error. other files may be good - } - } + openDirtyAccessor(h.vAccessorFileNameMask(fromStep, toStep), accessorEntries, h.dirs.SnapAccessors, h.FileVersion.AccessorVI, func(fPath string) (err error) { + item.index, err = h.openHashMapAccessor(fPath) + return err + }, "History.openDirtyFiles", h.logger) } } iter.Release() @@ -525,53 +473,17 @@ func (ii *InvertedIndex) openDirtyFiles(dataEntries, accessorEntries []string) e item := iter.Item() fromStep, toStep := item.StepRange(ii.stepSize) if item.decompressor == nil { - fNameMask := ii.efFileNameMask(fromStep, toStep) - fPath, fileVer, ok, err := version.MatchVersionedFile(fNameMask, dataEntries, ii.dirs.SnapIdx) - if err != nil { - fName := filepath.Base(fPath) - ii.logger.Debug("[agg] InvertedIndex.openDirtyFiles: MatchVersionedFile error", "f", fName, "err", err) - invalidFileItems = append(invalidFileItems, item) - continue - } - - if !ok { - fName := fNameMask - ii.logger.Debug("[agg] InvertedIndex.openDirtyFiles: file does not exists", "f", fName) + if !openDirtyDataFile(item, ii.efFileNameMask(fromStep, toStep), dataEntries, ii.dirs.SnapIdx, ii.FileVersion.DataEF, "InvertedIndex.openDirtyFiles", ii.logger) { invalidFileItems = append(invalidFileItems, item) continue } - - fName := filepath.Base(fPath) - ii.FileVersion.DataEF.MustSupport(fileVer, fName) - - if item.decompressor, err = seg.NewDecompressor(fPath); err != nil { - if errors.Is(err, &seg.ErrCompressedFileCorrupted{}) { - ii.logger.Debug("[agg] InvertedIndex.openDirtyFiles", "err", err, "f", fName) - } else { - ii.logger.Warn("[agg] InvertedIndex.openDirtyFiles", "err", err, "f", fName) - } - invalidFileItems = append(invalidFileItems, item) - // don't interrupt on error. other files may be good. but skip indices open. - continue - } } if item.index == nil { - fNameMask := ii.efAccessorFileNameMask(fromStep, toStep) - fPath, fileVer, ok, err := version.MatchVersionedFile(fNameMask, accessorEntries, ii.dirs.SnapAccessors) - if err != nil { - fName := filepath.Base(fPath) - ii.logger.Warn("[agg] InvertedIndex.openDirtyFiles", "err", err, "f", fName) - // don't interrupt on error. other files may be good - } - if ok { - fName := filepath.Base(fPath) - ii.FileVersion.AccessorEFI.MustSupport(fileVer, fName) - if item.index, err = ii.openHashMapAccessor(fPath); err != nil { - ii.logger.Warn("[agg] InvertedIndex.openDirtyFiles", "err", err, "f", fName) - // don't interrupt on error. other files may be good - } - } + openDirtyAccessor(ii.efAccessorFileNameMask(fromStep, toStep), accessorEntries, ii.dirs.SnapAccessors, ii.FileVersion.AccessorEFI, func(fPath string) (err error) { + item.index, err = ii.openHashMapAccessor(fPath) + return err + }, "InvertedIndex.openDirtyFiles", ii.logger) } } iter.Release() diff --git a/db/state/history.go b/db/state/history.go index 096efb063da..ed449fbbeb3 100644 --- a/db/state/history.go +++ b/db/state/history.go @@ -1504,8 +1504,10 @@ func (ht *HistoryRoTx) HistoryDump(fromTxNum, toTxNum int, keyToDump *[]byte, du return nil } -// CompactRange rebuilds the history files within the specified transaction range by performing a forced self-merge. -// If the range contains existing static files, the method collects all files belonging to that span and merges them. +// CompactRange rebuilds the history files within the specified transaction range by performing a forced self-merge +// with value deduplication. If the range contains existing static files, the method collects all files belonging to +// that span and rebuilds each file pair in place: the inputs are removed from disk, so the folder must be reopened +// to use the rebuilt files. func (ht *HistoryRoTx) CompactRange(ctx context.Context, fromTxNum, toTxNum uint64) error { if len(ht.iit.files) == 0 { return nil @@ -1527,9 +1529,19 @@ func (ht *HistoryRoTx) CompactRange(ctx context.Context, fromTxNum, toTxNum uint *NewMergeRange("", true, efFiles[i].startTxNum, efFiles[i].endTxNum), ) - if err := ht.deduplicateFiles(ctx, []*FilesItem{efFiles[i]}, []*FilesItem{vFiles[i]}, mergeRange, background.NewProgressSet()); err != nil { + dedup := newValuesDeduper() + indexIn, historyIn, err := ht.mergeFiles(ctx, []*FilesItem{efFiles[i]}, []*FilesItem{vFiles[i]}, mergeRange, background.NewProgressSet(), dedup) + if err != nil { return err } + fmt.Println("Values to deduplicate:", dedup.dropped) + + efFiles[i].closeFilesAndRemove() + if vFiles[i].StartStep(ht.stepSize) != indexIn.StartStep(ht.stepSize) || vFiles[i].EndStep(ht.stepSize) != indexIn.EndStep(ht.stepSize) { + vFiles[i].closeFilesAndRemove() + } + indexIn.closeFiles() + historyIn.closeFiles() } return nil diff --git a/db/state/history_test.go b/db/state/history_test.go index 34b5faaff58..796150e56e2 100644 --- a/db/state/history_test.go +++ b/db/state/history_test.go @@ -48,6 +48,7 @@ import ( "github.com/erigontech/erigon/db/recsplit/multiencseq" "github.com/erigontech/erigon/db/seg" "github.com/erigontech/erigon/db/state/statecfg" + "github.com/erigontech/erigon/db/version" "github.com/erigontech/erigon/execution/commitment/commitmentdb" ) @@ -1214,7 +1215,7 @@ func collateAndMergeHistory(tb testing.TB, db kv.RwDB, h *History, txs uint64, d } indexOuts, historyOuts, err := hc.staticFilesInRange(r) require.NoError(err) - indexIn, historyIn, err := hc.mergeFiles(ctx, indexOuts, historyOuts, r, background.NewProgressSet()) + indexIn, historyIn, err := hc.mergeFiles(ctx, indexOuts, historyOuts, r, background.NewProgressSet(), nil) require.NoError(err) h.integrateMergedDirtyFiles(indexIn, historyIn) return false @@ -1270,7 +1271,7 @@ func collateAndMergeHistoryWithCollisionRetry(tb testing.TB, db kv.RwDB, h *Hist } indexOuts, historyOuts, err := hc.staticFilesInRange(r) require.NoError(err) - indexIn, historyIn, err := hc.mergeFiles(ctx, indexOuts, historyOuts, r, background.NewProgressSet()) + indexIn, historyIn, err := hc.mergeFiles(ctx, indexOuts, historyOuts, r, background.NewProgressSet(), nil) require.NoError(err) h.integrateMergedDirtyFiles(indexIn, historyIn) return false @@ -2503,6 +2504,97 @@ func BenchmarkHistoryRange_MultiFile(b *testing.B) { } } +// TestHistoryCompactRange pins the snapshot-rebuild dedup semantics: within one +// file, a run of equal values for a key collapses to the run's last txNum (in +// both .v and .ef), values across a file boundary are not deduplicated, and +// HistorySeek answers are unchanged. Inputs are replaced in place: the tool is +// run by a binary whose current file versions are newer than the files on disk. +func TestHistoryCompactRange(t *testing.T) { + t.Parallel() + + logger := log.New() + ctx := t.Context() + require := require.New(t) + + vX, vY, vQ, vZ, vW := []byte("valX"), []byte("valY"), []byte("valQ"), []byte("valZ"), []byte("valW") + keyA, keyB, keyC := []byte("addr-aaaaaaa"), []byte("addr-bbbbbbb"), []byte("addr-ccccccc") + values := map[string][]upd{ + string(keyA): {{1, nil}, {3, vX}, {5, vX}, {7, vX}, {9, vY}, {17, vY}, {19, vY}, {21, vY}, {23, vQ}}, + string(keyB): {{2, nil}, {6, vX}, {10, vY}}, + string(keyC): {{4, nil}, {14, vZ}, {18, vZ}, {25, vW}}, + } + db, h := filledHistoryValues(t, true, values, logger) + + rwTx, err := db.BeginRw(ctx) + require.NoError(err) + defer rwTx.Rollback() + logEvery := time.NewTicker(30 * time.Second) + defer logEvery.Stop() + for step := kv.Step(0); step < 2; step++ { + require.NoError(h.collateBuildIntegrate(ctx, step, rwTx, background.NewProgressSet())) + hc := h.beginForTests() + _, err = hc.Prune(ctx, rwTx, step.ToTxNum(h.stepSize), (step + 1).ToTxNum(h.stepSize), math.MaxUint64, false, logEvery) + hc.Close() + require.NoError(err) + } + require.NoError(rwTx.Commit()) + + keys := [][]byte{keyA, keyB, keyC} + idxTxNums := func(hc *HistoryRoTx, tx kv.Tx, key []byte) []uint64 { + it, err := hc.IdxRange(key, -1, -1, order.Asc, -1, tx) + require.NoError(err) + res, err := stream.ToArrayU64(it) + require.NoError(err) + return res + } + seekAll := func(hc *HistoryRoTx, tx kv.Tx) map[string][]byte { + res := make(map[string][]byte) + for _, key := range keys { + for txNum := uint64(0); txNum <= 33; txNum++ { + v, ok, err := hc.HistorySeek(key, txNum, tx) + require.NoError(err) + if ok { + res[fmt.Sprintf("%s-%d", key, txNum)] = common.Copy(v) + } + } + } + return res + } + + roTx, err := db.BeginRo(ctx) + require.NoError(err) + defer roTx.Rollback() + hc := h.beginForTests() + require.Equal([]uint64{1, 3, 5, 7, 9, 17, 19, 21, 23}, idxTxNums(hc, roTx, keyA)) + seeksBefore := seekAll(hc, roTx) + + bump := func(v *version.Versions) { v.Current.Minor++ } + bump(&h.FileVersion.DataV) + bump(&h.FileVersion.AccessorVI) + bump(&h.InvertedIndex.FileVersion.DataEF) + bump(&h.InvertedIndex.FileVersion.AccessorEFI) + + require.NoError(hc.CompactRange(ctx, 0, 32)) + hc.Close() + roTx.Rollback() + + h.Close() + scanDirsRes, err := scanDirs(h.dirs) + require.NoError(err) + require.NoError(h.openFolder(scanDirsRes)) + + roTx2, err := db.BeginRo(ctx) + require.NoError(err) + defer roTx2.Rollback() + hc2 := h.beginForTests() + defer hc2.Close() + + require.Equal([]uint64{1, 7, 9, 21, 23}, idxTxNums(hc2, roTx2, keyA)) + require.Equal([]uint64{2, 6, 10}, idxTxNums(hc2, roTx2, keyB)) + require.Equal([]uint64{4, 14, 18, 25}, idxTxNums(hc2, roTx2, keyC)) + require.Equal(seeksBefore, seekAll(hc2, roTx2)) +} + // BenchmarkRangeAsOf_MultiFile is like BenchmarkRangeAsOf but keeps all // step-files unmerged so the heap has ~60 elements, actually exercising heap ops. func BenchmarkRangeAsOf_MultiFile(b *testing.B) { diff --git a/db/state/inverted_index_test.go b/db/state/inverted_index_test.go index 0d9384247a0..3d60afa4632 100644 --- a/db/state/inverted_index_test.go +++ b/db/state/inverted_index_test.go @@ -856,7 +856,7 @@ func mergeInverted(tb testing.TB, db kv.RwDB, ii *InvertedIndex, txs uint64) { return true } outs := ic.staticFilesInRange(startTxNum, endTxNum) - in, err := ic.mergeFiles(ctx, outs, startTxNum, endTxNum, background.NewProgressSet()) + in, err := ic.mergeFiles(ctx, outs, startTxNum, endTxNum, background.NewProgressSet(), nil) require.NoError(tb, err) ii.integrateMergedDirtyFiles(in) return false diff --git a/db/state/merge.go b/db/state/merge.go index 746c958512a..df887caec35 100644 --- a/db/state/merge.go +++ b/db/state/merge.go @@ -408,7 +408,7 @@ func (dt *DomainRoTx) mergeFiles(ctx context.Context, domainFiles, indexFiles, h } } }() - if indexIn, historyIn, err = dt.ht.mergeFiles(ctx, indexFiles, historyFiles, r.history, ps); err != nil { + if indexIn, historyIn, err = dt.ht.mergeFiles(ctx, indexFiles, historyFiles, r.history, ps, nil); err != nil { return nil, nil, nil, err } @@ -578,7 +578,110 @@ func (dt *DomainRoTx) mergeFiles(ctx context.Context, domainFiles, indexFiles, h return } -func (iit *InvertedIndexRoTx) mergeFiles(ctx context.Context, files []*FilesItem, startTxNum, endTxNum uint64, ps *background.ProgressSet) (*FilesItem, error) { +// valuesDeduper switches history and inverted-index merges into snapshot-rebuild +// deduplication mode: history entries repeating the previous value of a key are +// dropped, and their txNums are removed from the merged index sequences so both +// files stay consistent. +type valuesDeduper struct { + skipTxNums map[string]map[uint64]struct{} + keptTxNums []uint64 + dropped int +} + +func newValuesDeduper() *valuesDeduper { + return &valuesDeduper{skipTxNums: make(map[string]map[uint64]struct{})} +} + +func (dd *valuesDeduper) skip(key []byte, txNum uint64) { + m, ok := dd.skipTxNums[string(key)] + if !ok { + m = make(map[uint64]struct{}) + dd.skipTxNums[string(key)] = m + } + m[txNum] = struct{}{} + dd.dropped++ +} + +// writeDeduped writes ci's history values for the current key, collapsing each +// run of equal consecutive values to the run's last txNum and recording the +// dropped txNums. Deduplication is scoped to one source file: ci covers a +// single file's sequence for the key. +func (dd *valuesDeduper) writeDeduped(ci *CursorItem, ss *multiencseq.SequenceIterator, wr *seg.PagedWriter, valBuf, histKeyBuf []byte) ([]byte, []byte, error) { + var retainedVal []byte + var retained bool + var prevTxNum uint64 + for ss.HasNext() { + txNum, err := ss.Next() + if err != nil { + panic(fmt.Sprintf("failed to extract txNum from ef. File: %s Key: %x", ci.kvReader.FileName(), ci.key)) + } + + if !ci.hist.HasNext() { + panic(fmt.Errorf("assert: no value??? %s, txNum=%d, key=%x", ci.hist.FileName(), txNum, ci.key)) + } + + var v []byte + _, v, valBuf, _ = ci.hist.Next2(valBuf[:0]) + + if !retained { + retainedVal = append(retainedVal[:0], v...) + retained = true + prevTxNum = txNum + continue + } + if bytes.Equal(retainedVal, v) { + dd.skip(ci.key, prevTxNum) + prevTxNum = txNum + continue + } + + histKeyBuf = historyKey(prevTxNum, ci.key, histKeyBuf) + if err := wr.Add(histKeyBuf, retainedVal); err != nil { + return valBuf, histKeyBuf, err + } + retainedVal = append(retainedVal[:0], v...) + prevTxNum = txNum + } + if retained { + histKeyBuf = historyKey(prevTxNum, ci.key, histKeyBuf) + if err := wr.Add(histKeyBuf, retainedVal); err != nil { + return valBuf, histKeyBuf, err + } + } + return valBuf, histKeyBuf, nil +} + +// mergeSortedSkipping is the dedup-mode counterpart of builder.MergeSorted: +// same single-pass merge, minus the txNums recorded by writeDeduped. +func (dd *valuesDeduper) mergeSortedSkipping(builder *multiencseq.SequenceBuilder, seqReader *multiencseq.SequenceReader, outBaseNum uint64, baseNums []uint64, seqs [][]byte, key []byte) error { + skipTxNums := dd.skipTxNums[string(key)] + dd.keptTxNums = dd.keptTxNums[:0] + var maxTxNum uint64 + var it multiencseq.SequenceIterator + for i, data := range seqs { + seqReader.Reset(baseNums[i], data) + it.Reset(seqReader, 0) + for it.HasNext() { + v, err := it.Next() + if err != nil { + return err + } + maxTxNum = v + if _, drop := skipTxNums[v]; drop { + continue + } + dd.keptTxNums = append(dd.keptTxNums, v) + } + } + builder.Reset(outBaseNum, uint64(len(dd.keptTxNums)), maxTxNum) + for _, v := range dd.keptTxNums { + builder.AddOffset(v) + } + builder.Build() + return nil +} + +func (iit *InvertedIndexRoTx) mergeFiles(ctx context.Context, files []*FilesItem, startTxNum, endTxNum uint64, ps *background.ProgressSet, dedup *valuesDeduper) (*FilesItem, error) { if startTxNum == endTxNum { panic(fmt.Sprintf("assert: startTxNum(%d) == endTxNum(%d)", startTxNum, endTxNum)) } @@ -674,7 +777,11 @@ func (iit *InvertedIndexRoTx) mergeFiles(ctx context.Context, files []*FilesItem } // Merge all sequences for this key in a single pass (O(N·C), one EF allocation). - if err := builder.MergeSorted(&seqReader, startTxNum, mergeBaseNums, mergeSeqs); err != nil { + if dedup == nil { + if err := builder.MergeSorted(&seqReader, startTxNum, mergeBaseNums, mergeSeqs); err != nil { + return nil, err + } + } else if err := dedup.mergeSortedSkipping(&builder, &seqReader, startTxNum, mergeBaseNums, mergeSeqs, lastKey); err != nil { return nil, err } for i := range mergeSeqs { // allow for GC @@ -726,7 +833,7 @@ func (iit *InvertedIndexRoTx) mergeFiles(ctx context.Context, files []*FilesItem return outItem, nil } -func (ht *HistoryRoTx) mergeFiles(ctx context.Context, indexFiles, historyFiles []*FilesItem, r HistoryRanges, ps *background.ProgressSet) (indexIn, historyIn *FilesItem, err error) { +func (ht *HistoryRoTx) mergeFiles(ctx context.Context, indexFiles, historyFiles []*FilesItem, r HistoryRanges, ps *background.ProgressSet, dedup *valuesDeduper) (indexIn, historyIn *FilesItem, err error) { if !r.any() { return nil, nil, nil } @@ -739,8 +846,10 @@ func (ht *HistoryRoTx) mergeFiles(ctx context.Context, indexFiles, historyFiles } }() - if r.index.needMerge { - if indexIn, err = ht.iit.mergeFiles(ctx, indexFiles, r.index.from, r.index.to, ps); err != nil { + // In dedup mode the index merge waits until the history scan below has + // recorded which txNums it dropped. + if r.index.needMerge && dedup == nil { + if indexIn, err = ht.iit.mergeFiles(ctx, indexFiles, r.index.from, r.index.to, ps, nil); err != nil { return nil, nil, err } } @@ -841,23 +950,29 @@ func (ht *HistoryRoTx) mergeFiles(ctx context.Context, indexFiles, historyFiles seq.Reset(ci1.startTxNum, ci1.val) ss.Reset(&seq, 0) - for ss.HasNext() { - txNum, err := ss.Next() - if err != nil { - panic(fmt.Sprintf("failed to extract txNum from ef. File: %s Key: %x", ci1.kvReader.FileName(), ci1.key)) + if dedup != nil { + if valBuf, histKeyBuf, err = dedup.writeDeduped(ci1, &ss, pagedWr, valBuf, histKeyBuf); err != nil { + return nil, nil, err } + } else { + for ss.HasNext() { + txNum, err := ss.Next() + if err != nil { + panic(fmt.Sprintf("failed to extract txNum from ef. File: %s Key: %x", ci1.kvReader.FileName(), ci1.key)) + } - if !ci1.hist.HasNext() { - panic(fmt.Errorf("assert: no value??? %s, txNum=%d, lastKey=%x, ci1.key=%x", ci1.hist.FileName(), txNum, lastKey, ci1.key)) - } + if !ci1.hist.HasNext() { + panic(fmt.Errorf("assert: no value??? %s, txNum=%d, lastKey=%x, ci1.key=%x", ci1.hist.FileName(), txNum, lastKey, ci1.key)) + } - var v []byte - _, v, valBuf, _ = ci1.hist.Next2(valBuf[:0]) // key from .v file can be empty if file is not compressed -> need to be built from txNum + key + var v []byte + _, v, valBuf, _ = ci1.hist.Next2(valBuf[:0]) // key from .v file can be empty if file is not compressed -> need to be built from txNum + key - histKeyBuf = historyKey(txNum, ci1.key, histKeyBuf) + histKeyBuf = historyKey(txNum, ci1.key, histKeyBuf) - if err = pagedWr.Add(histKeyBuf, v); err != nil { - return nil, nil, err + if err = pagedWr.Add(histKeyBuf, v); err != nil { + return nil, nil, err + } } } @@ -880,6 +995,12 @@ func (ht *HistoryRoTx) mergeFiles(ctx context.Context, indexFiles, historyFiles } ps.Delete(p) + if r.index.needMerge && dedup != nil { + if indexIn, err = ht.iit.mergeFiles(ctx, indexFiles, r.index.from, r.index.to, ps, dedup); err != nil { + return nil, nil, err + } + } + if err = ht.h.buildVI(ctx, idxPath, decomp, indexIn.decompressor, indexIn.startTxNum, ps); err != nil { return nil, nil, err } diff --git a/db/state/merge_bench_test.go b/db/state/merge_bench_test.go index bd253c3cb00..b72f862156d 100644 --- a/db/state/merge_bench_test.go +++ b/db/state/merge_bench_test.go @@ -80,7 +80,7 @@ func benchmarkIIMergeFiles(b *testing.B, numFiles int) { b.ResetTimer() for b.Loop() { - out, err := ic.mergeFiles(ctx, inputFiles, mr.from, mr.to, ps) + out, err := ic.mergeFiles(ctx, inputFiles, mr.from, mr.to, ps, nil) b.StopTimer() require.NoError(b, err) out.closeFilesAndRemove() diff --git a/db/state/merge_test.go b/db/state/merge_test.go index 0ab11bdd79b..8c9a2ea5675 100644 --- a/db/state/merge_test.go +++ b/db/state/merge_test.go @@ -1450,7 +1450,7 @@ func TestInvIndexMergeFiles_SharedKey(t *testing.T) { inputFiles := ic.staticFilesInRange(mr.from, mr.to) require.Len(t, inputFiles, numFiles) - out, err := ic.mergeFiles(ctx, inputFiles, mr.from, mr.to, ps) + out, err := ic.mergeFiles(ctx, inputFiles, mr.from, mr.to, ps, nil) require.NoError(t, err) t.Cleanup(out.closeFilesAndRemove)