feat: db compaction cmd

cmd/bee/cmd/db.go

@@ -7,11 +7,13 @@ package cmd
 import (
 	"archive/tar"
 	"bytes"
+	"context"
 	"encoding/binary"
 	"errors"
 	"fmt"
 	"io"
 	"os"
+	"path"
 	"path/filepath"
 	"strings"
 	"time"
@@ -34,6 +36,7 @@ func (c *command) initDBCmd() {
 	dbImportCmd(cmd)
 	dbNukeCmd(cmd)
 	dbInfoCmd(cmd)
+	dbCompactCmd(cmd)
 
 	c.root.AddCommand(cmd)
 }
@@ -75,6 +78,8 @@ func dbInfoCmd(cmd *cobra.Command) {
 			}
 			defer db.Close()
 
+			logger.Info("getting db info", "path", dataDir)
+
 			info, err := db.DebugInfo(cmd.Context())
 			if err != nil {
 				return fmt.Errorf("fetching db info: %w", err)
@@ -95,6 +100,58 @@ func dbInfoCmd(cmd *cobra.Command) {
 	cmd.AddCommand(c)
 }
 
+func dbCompactCmd(cmd *cobra.Command) {
+
+	optionNameValidation := "validate"
+
+	c := &cobra.Command{
+		Use:   "compact",
+		Short: "Compacts the localstore sharky store.",
+		RunE: func(cmd *cobra.Command, args []string) (err error) {
+			v, err := cmd.Flags().GetString(optionNameVerbosity)
+			if err != nil {
+				return fmt.Errorf("get verbosity: %w", err)
+			}
+			v = strings.ToLower(v)
+			logger, err := newLogger(cmd, v)
+			if err != nil {
+				return fmt.Errorf("new logger: %w", err)
+			}
+
+			dataDir, err := cmd.Flags().GetString(optionNameDataDir)
+			if err != nil {
+				return fmt.Errorf("get data-dir: %w", err)
+			}
+			if dataDir == "" {
+				return errors.New("no data-dir provided")
+			}
+
+			validation, err := cmd.Flags().GetBool(optionNameValidation)
+			if err != nil {
+				return fmt.Errorf("get validation: %w", err)
+			}
+
+			localstorePath := path.Join(dataDir, "localstore")
+
+			err = storer.Compact(context.Background(), localstorePath, &storer.Options{
+				Logger:          logger,
+				RadiusSetter:    noopRadiusSetter{},
+				Batchstore:      new(postage.NoOpBatchStore),
+				ReserveCapacity: node.ReserveCapacity,
+			}, validation)
+			if err != nil {
+				return fmt.Errorf("localstore: %w", err)
+			}
+
+			return nil
+		},
+	}
+	c.Flags().String(optionNameDataDir, "", "data directory")
+	c.Flags().String(optionNameVerbosity, "info", "verbosity level")
+	c.Flags().Bool(optionNameValidation, true, "enable chunk validation")
+	cmd.AddCommand(c)
+}
+
 func dbExportCmd(cmd *cobra.Command) {
 	c := &cobra.Command{
 		Use:   "export",

pkg/sharky/recovery.go

@@ -31,7 +31,7 @@ func NewRecovery(dir string, shardCnt int, datasize int) (*Recovery, error) {
 	shardFiles := make([]*os.File, shardCnt)
 
 	for i := 0; i < shardCnt; i++ {
-		file, err := os.OpenFile(path.Join(dir, fmt.Sprintf("shard_%03d", i)), os.O_RDONLY, 0666)
+		file, err := os.OpenFile(path.Join(dir, fmt.Sprintf("shard_%03d", i)), os.O_RDWR, 0666)
 		if errors.Is(err, fs.ErrNotExist) {
 			return nil, fmt.Errorf("index %d: %w", i, ErrShardNotFound)
 		}
@@ -86,6 +86,27 @@ func (r *Recovery) Read(ctx context.Context, loc Location, buf []byte) error {
 	return err
 }
 
+func (r *Recovery) Move(ctx context.Context, from Location, to Location) error {
+	r.mtx.Lock()
+	defer r.mtx.Unlock()
+
+	chData := make([]byte, from.Length)
+	_, err := r.shardFiles[from.Shard].ReadAt(chData, int64(from.Slot)*int64(r.datasize))
+	if err != nil {
+		return err
+	}
+
+	_, err = r.shardFiles[to.Shard].WriteAt(chData, int64(to.Slot)*int64(r.datasize))
+	return err
+}
+
+func (r *Recovery) TruncateAt(ctx context.Context, shard uint8, slot uint32) error {
+	r.mtx.Lock()
+	defer r.mtx.Unlock()
+
+	return r.shardFiles[shard].Truncate(int64(slot) * int64(r.datasize))
+}
+
 // Save saves all free slots files of the recovery (without closing).
 func (r *Recovery) Save() error {
 	r.mtx.Lock()

pkg/storer/compact.go

@@ -0,0 +1,186 @@
+// Copyright 2023 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package storer
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"path"
+	"sort"
+	"time"
+
+	"github.com/ethersphere/bee/pkg/cac"
+	"github.com/ethersphere/bee/pkg/log"
+	"github.com/ethersphere/bee/pkg/sharky"
+	"github.com/ethersphere/bee/pkg/soc"
+	"github.com/ethersphere/bee/pkg/storage"
+	"github.com/ethersphere/bee/pkg/storer/internal/chunkstore"
+	"github.com/ethersphere/bee/pkg/swarm"
+	"golang.org/x/sync/errgroup"
+)
+
+// Compact minimizes sharky disk usage by, using the current sharky locations from the storer,
+// relocating chunks starting from the end of the used slots to the first available slots.
+func Compact(ctx context.Context, basePath string, opts *Options, validate bool) error {
+
+	logger := opts.Logger
+
+	store, err := initStore(basePath, opts)
+	if err != nil {
+		return fmt.Errorf("failed creating levelDB index store: %w", err)
+	}
+	defer func() {
+		if err := store.Close(); err != nil {
+			logger.Error(err, "failed closing store")
+		}
+	}()
+
+	sharkyRecover, err := sharky.NewRecovery(path.Join(basePath, sharkyPath), sharkyNoOfShards, swarm.SocMaxChunkSize)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		if err := sharkyRecover.Close(); err != nil {
+			logger.Error(err, "failed closing sharky recovery")
+		}
+	}()
+
+	if validate {
+		logger.Info("performing chunk validation before compaction")
+		validationWork(ctx, logger, store, sharkyRecover)
+	}
+
+	logger.Info("starting compaction")
+
+	n := time.Now()
+
+	for shard := 0; shard < sharkyNoOfShards; shard++ {
+
+		items := make([]*chunkstore.RetrievalIndexItem, 0, 1_000_000)
+		// we deliberately choose to iterate the whole store again for each shard
+		// so that we do not store all the items in memory (for operators with huge localstores)
+		_ = chunkstore.Iterate(store, func(item *chunkstore.RetrievalIndexItem) error {
+			if item.Location.Shard == uint8(shard) {
+				items = append(items, item)
+			}
+			return nil
+		})
+
+		sort.Slice(items, func(i, j int) bool {
+			return items[i].Location.Slot < items[j].Location.Slot
+		})
+
+		lastUsedSlot := items[len(items)-1].Location.Slot
+		slots := make([]*chunkstore.RetrievalIndexItem, lastUsedSlot+1) // marks free and used slots
+		for _, l := range items {
+			slots[l.Location.Slot] = l
+		}
+
+		// start begins at the zero slot. The loop below will increment the position of start until a free slot is found.
+		// end points to the last slot, and the loop will decrement the position of end until a used slot is found.
+		// Once start and end point to free and used slots, respectively, the swap of the chunk location will occur.
+		start := uint32(0)
+		end := lastUsedSlot
+
+		for start < end {
+			if slots[start] == nil { // free
+				if slots[end] != nil { // used
+					from := slots[end]
+					to := sharky.Location{Slot: start, Length: from.Location.Length, Shard: from.Location.Shard}
+					if err := sharkyRecover.Move(ctx, from.Location, to); err != nil {
+						return fmt.Errorf("sharky move: %w", err)
+					}
+					if err := sharkyRecover.Add(to); err != nil {
+						return fmt.Errorf("sharky add: %w", err)
+					}
+
+					from.Location = to
+					if err := store.Put(from); err != nil {
+						return fmt.Errorf("store put: %w", err)
+					}
+
+					start++
+					end--
+				} else {
+					end-- // keep moving to the left until a used slot is found
+				}
+			} else {
+				start++ // keep moving to the right until a free slot is found
+			}
+		}
+
+		logger.Info("shard truncated", "shard", shard, "slot", end)
+
+		if err := sharkyRecover.TruncateAt(ctx, uint8(shard), end+1); err != nil {
+			return fmt.Errorf("sharky truncate: %w", err)
+		}
+	}
+
+	if err := sharkyRecover.Save(); err != nil {
+		return fmt.Errorf("sharky save: %w", err)
+	}
+
+	logger.Info("compaction finished", "duration", time.Since(n))
+
+	if validate {
+		logger.Info("performing chunk validation after compaction")
+		validationWork(ctx, logger, store, sharkyRecover)
+	}
+
+	return nil
+}
+
+func validationWork(ctx context.Context, logger log.Logger, store storage.Store, sharky *sharky.Recovery) {
+
+	n := time.Now()
+	defer func() {
+		logger.Info("validation finished", "duration", time.Since(n))
+	}()
+
+	iteratateItemsC := make(chan *chunkstore.RetrievalIndexItem)
+
+	validChunk := func(item *chunkstore.RetrievalIndexItem, buf []byte) error {
+		err := sharky.Read(ctx, item.Location, buf)
+		if err != nil {
+			return err
+		}
+
+		ch := swarm.NewChunk(item.Address, buf)
+		if !cac.Valid(ch) && !soc.Valid(ch) {
+			return errors.New("invalid chunk")
+		}
+
+		return nil
+	}
+
+	eg, ctx := errgroup.WithContext(ctx)
+
+	for i := 0; i < 8; i++ {
+		eg.Go(func() error {
+			buf := make([]byte, swarm.SocMaxChunkSize)
+			for item := range iteratateItemsC {
+				if err := validChunk(item, buf[:item.Location.Length]); err != nil {
+					logger.Info("invalid chunk", "address", item.Address, "error", err)
+				}
+			}
+			return nil
+		})
+	}
+
+	count := 0
+	_ = chunkstore.Iterate(store, func(item *chunkstore.RetrievalIndexItem) error {
+		iteratateItemsC <- item
+		count++
+		if count%100_000 == 0 {
+			logger.Info("..still validating chunks", "count", count)
+		}
+		return nil
+	})
+
+	close(iteratateItemsC)
+
+	_ = eg.Wait()
+}