feat(cel-shed): datastore: erase samples

cmd/cel-shed/datastore.go

@@ -0,0 +1,134 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/ipfs/boxo/blockstore"
+	ds "github.com/ipfs/go-datastore"
+	"github.com/ipfs/go-datastore/namespace"
+	dsq "github.com/ipfs/go-datastore/query"
+	"github.com/spf13/cobra"
+	"golang.org/x/sync/errgroup"
+
+	"github.com/celestiaorg/celestia-node/nodebuilder"
+)
+
+func init() {
+	datastoreCmd.AddCommand(eraseCmd, eraseSamplesCmd)
+}
+
+var datastoreCmd = &cobra.Command{
+	Use:   "datastore [subcommand]",
+	Short: "Collection of datastore related utilities",
+}
+
+var eraseCmd = &cobra.Command{
+	Use:   "erase <ds_key>",
+	Short: "Erase datastore namespace",
+	RunE: func(cmd *cobra.Command, args []string) error {
+		path, err := nodebuilder.DiscoverStopped()
+		if err != nil {
+			return fmt.Errorf("discovering stopped node: %w", err)
+		}
+		fmt.Printf("Discovered stopped node at %s\n", path)
+
+		nodestore, err := nodebuilder.OpenStore(path, nil)
+		if err != nil {
+			return err
+		}
+		defer func() {
+			err = errors.Join(err, nodestore.Close())
+		}()
+
+		store, err := nodestore.Datastore()
+		if err != nil {
+			return fmt.Errorf("getting datastore: %w", err)
+		}
+
+		key := ds.NewKey(args[0])
+		err = eraseDatastoreNamespace(cmd.Context(), store, key)
+		if err != nil {
+			return fmt.Errorf("erasing datastore namespace: %w", err)
+		}
+
+		fmt.Printf("Erased %s\n", key)
+		return nil
+	},
+	Args: cobra.ExactArgs(1),
+}
+
+var sampleDataKeys = []ds.Key{
+	ds.NewKey("sampling_result"),
+	ds.NewKey("das"),
+	ds.NewKey("pruner"),
+	blockstore.BlockPrefix,
+}
+
+var eraseSamplesCmd = &cobra.Command{
+	Use:   "erase-samples [subcommand]",
+	Short: "Erase samples data and state. Useful to resample, avoiding resyncing headers",
+	RunE: func(cmd *cobra.Command, _ []string) error {
+		path, err := nodebuilder.DiscoverStopped()
+		if err != nil {
+			return fmt.Errorf("discovering stopped node: %w", err)
+		}
+		fmt.Printf("Discovered stopped node at %s\n", path)
+
+		nodestore, err := nodebuilder.OpenStore(path, nil)
+		if err != nil {
+			return err
+		}
+		defer func() {
+			err = errors.Join(err, nodestore.Close())
+		}()
+
+		ds, err := nodestore.Datastore()
+		if err != nil {
+			return fmt.Errorf("getting datastore: %w", err)
+		}
+
+		group, ctx := errgroup.WithContext(cmd.Context())
+		for _, key := range sampleDataKeys {
+			group.Go(func() error {
+				err := eraseDatastoreNamespace(ctx, ds, key)
+				if err != nil {
+					return fmt.Errorf("erasing datastore namespace: %w", err)
+				}
+				fmt.Printf("Erased %s\n", key)
+				return nil
+			})
+		}
+
+		return group.Wait()
+	},
+}
+
+func eraseDatastoreNamespace(ctx context.Context, store ds.Datastore, key ds.Key) error {
+	store = namespace.Wrap(store, key)
+
+	q := dsq.Query{KeysOnly: true}
+	res, err := store.Query(ctx, q)
+	if err != nil {
+		return fmt.Errorf("querying datastore: %w", err)
+	}
+
+	for {
+		e, ok := res.NextSync()
+		if !ok {
+			break
+		}
+		if e.Error != nil {
+			return fmt.Errorf("getting next key: %w", e.Error)
+		}
+
+		key := ds.RawKey(e.Key)
+		err := store.Delete(ctx, key)
+		if err != nil {
+			return fmt.Errorf("deleting key: %w", err)
+		}
+	}
+
+	return nil
+}

cmd/cel-shed/main.go

@@ -10,7 +10,7 @@ import (
 )
 
 func init() {
-	rootCmd.AddCommand(p2pCmd, headerCmd, edsStoreCmd, shwapCmd)
+	rootCmd.AddCommand(p2pCmd, headerCmd, edsStoreCmd, shwapCmd, datastoreCmd)
 }
 
 var rootCmd = &cobra.Command{

nodebuilder/store.go

@@ -157,6 +157,33 @@ type fsStore struct {
 	dirLock *flock.Flock // protects directory
 }
 
+// DiscoverStopped finds a path of an initialized store of a stopped Node and returns its path.
+// If multiple store exists, it only returns the path of the first found.
+// Network is favored over node type.
+//
+// Network preference order: Mainnet, Mocha, Arabica, Private, Custom
+// Type preference order: Bridge, Full, Light
+func DiscoverStopped() (string, error) {
+	defaultNetwork := p2p.GetNetworks()
+	nodeTypes := nodemod.GetTypes()
+
+	for _, n := range defaultNetwork {
+		for _, tp := range nodeTypes {
+			path, err := DefaultNodeStorePath(tp, n)
+			if err != nil {
+				return "", err
+			}
+
+			ok, _ := IsOpened(path)
+			if !ok && IsInit(path) {
+				return path, nil
+			}
+		}
+	}
+
+	return "", ErrNotInited
+}
+
 // DiscoverOpened finds a path of an opened Node Store and returns its path.
 // If multiple nodes are running, it only returns the path of the first found node.
 // Network is favored over node type.