Basic Filestore Utilties

core/commands/filestore.go

@@ -0,0 +1,251 @@
+package commands
+
+import (
+	"context"
+	"fmt"
+
+	cmds "github.com/ipfs/go-ipfs/commands"
+	"github.com/ipfs/go-ipfs/core"
+	"github.com/ipfs/go-ipfs/filestore"
+	cid "gx/ipfs/QmV5gPoRsjN1Gid3LMdNZTyfCtP2DsvqEbMAmz82RmmiGk/go-cid"
+	u "gx/ipfs/QmZuY8aV7zbNXVy6DyN9SmnuH3o9nG852F4aTiSBpts8d1/go-ipfs-util"
+)
+
+var FileStoreCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "Interact with filestore objects.",
+	},
+	Subcommands: map[string]*cmds.Command{
+		"ls":     lsFileStore,
+		"verify": verifyFileStore,
+		"dups":   dupsFileStore,
+	},
+}
+
+var lsFileStore = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "List objects in filestore.",
+		LongDescription: `
+List objects in the filestore.
+
+If one or more <obj> is specified only list those specific objects,
+otherwise list all objects.
+
+The output is:
+
+<hash> <size> <path> <offset>
+`,
+	},
+	Arguments: []cmds.Argument{
+		cmds.StringArg("obj", false, true, "Cid of objects to list."),
+	},
+	Run: func(req cmds.Request, res cmds.Response) {
+		_, fs, err := getFilestore(req)
+		if err != nil {
+			res.SetError(err, cmds.ErrNormal)
+			return
+		}
+		args := req.Arguments()
+		if len(args) > 0 {
+			out := perKeyActionToChan(args, func(c *cid.Cid) *filestore.ListRes {
+				return filestore.List(fs, c)
+			}, req.Context())
+			res.SetOutput(out)
+		} else {
+			next, err := filestore.ListAll(fs)
+			if err != nil {
+				res.SetError(err, cmds.ErrNormal)
+				return
+			}
+			out := listResToChan(next, req.Context())
+			res.SetOutput(out)
+		}
+	},
+	PostRun: func(req cmds.Request, res cmds.Response) {
+		if res.Error() != nil {
+			return
+		}
+		outChan, ok := res.Output().(<-chan interface{})
+		if !ok {
+			res.SetError(u.ErrCast(), cmds.ErrNormal)
+			return
+		}
+		res.SetOutput(nil)
+		errors := false
+		for r0 := range outChan {
+			r := r0.(*filestore.ListRes)
+			if r.ErrorMsg != "" {
+				errors = true
+				fmt.Fprintf(res.Stderr(), "%s\n", r.ErrorMsg)
+			} else {
+				fmt.Fprintf(res.Stdout(), "%s\n", r.FormatLong())
+			}
+		}
+		if errors {
+			res.SetError(fmt.Errorf("errors while displaying some entries"), cmds.ErrNormal)
+		}
+	},
+	Type: filestore.ListRes{},
+}
+
+var verifyFileStore = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "Verify objects in filestore.",
+		LongDescription: `
+Verify objects in the filestore.
+
+If one or more <obj> is specified only verify those specific objects,
+otherwise verify all objects.
+
+The output is:
+
+<status> <hash> <size> <path> <offset>
+
+Where <status> is one of:
+ok:       the block can be reconstructed
+changed:  the contents of the backing file have changed
+no-file:  the backing file could not be found
+error:    there was some other problem reading the file
+missing:  <obj> could not be found in the filestore
+ERROR:    internal error, most likely due to a corrupt database
+
+For ERROR entries the error will also be printed to stderr.
+`,
+	},
+	Arguments: []cmds.Argument{
+		cmds.StringArg("obj", false, true, "Cid of objects to verify."),
+	},
+	Run: func(req cmds.Request, res cmds.Response) {
+		_, fs, err := getFilestore(req)
+		if err != nil {
+			res.SetError(err, cmds.ErrNormal)
+			return
+		}
+		args := req.Arguments()
+		if len(args) > 0 {
+			out := perKeyActionToChan(args, func(c *cid.Cid) *filestore.ListRes {
+				return filestore.Verify(fs, c)
+			}, req.Context())
+			res.SetOutput(out)
+		} else {
+			next, err := filestore.VerifyAll(fs)
+			if err != nil {
+				res.SetError(err, cmds.ErrNormal)
+				return
+			}
+			out := listResToChan(next, req.Context())
+			res.SetOutput(out)
+		}
+	},
+	PostRun: func(req cmds.Request, res cmds.Response) {
+		if res.Error() != nil {
+			return
+		}
+		outChan, ok := res.Output().(<-chan interface{})
+		if !ok {
+			res.SetError(u.ErrCast(), cmds.ErrNormal)
+			return
+		}
+		res.SetOutput(nil)
+		for r0 := range outChan {
+			r := r0.(*filestore.ListRes)
+			if r.Status == filestore.StatusOtherError {
+				fmt.Fprintf(res.Stderr(), "%s\n", r.ErrorMsg)
+			}
+			fmt.Fprintf(res.Stdout(), "%s %s\n", r.Status.Format(), r.FormatLong())
+		}
+	},
+	Type: filestore.ListRes{},
+}
+
+var dupsFileStore = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "List blocks that are both in the filestore and standard block storage.",
+	},
+	Run: func(req cmds.Request, res cmds.Response) {
+		_, fs, err := getFilestore(req)
+		if err != nil {
+			res.SetError(err, cmds.ErrNormal)
+			return
+		}
+		ch, err := fs.FileManager().AllKeysChan(req.Context())
+		if err != nil {
+			res.SetError(err, cmds.ErrNormal)
+			return
+		}
+
+		out := make(chan interface{}, 128)
+		res.SetOutput((<-chan interface{})(out))
+
+		go func() {
+			defer close(out)
+			for cid := range ch {
+				have, err := fs.MainBlockstore().Has(cid)
+				if err != nil {
+					out <- &RefWrapper{Err: err.Error()}
+					return
+				}
+				if have {
+					out <- &RefWrapper{Ref: cid.String()}
+				}
+			}
+		}()
+	},
+	Marshalers: refsMarshallerMap,
+	Type:       RefWrapper{},
+}
+
+func getFilestore(req cmds.Request) (*core.IpfsNode, *filestore.Filestore, error) {
+	n, err := req.InvocContext().GetNode()
+	if err != nil {
+		return nil, nil, err
+	}
+	fs := n.Filestore
+	if fs == nil {
+		return n, nil, fmt.Errorf("filestore not enabled")
+	}
+	return n, fs, err
+}
+
+func listResToChan(next func() *filestore.ListRes, ctx context.Context) <-chan interface{} {
+	out := make(chan interface{}, 128)
+	go func() {
+		defer close(out)
+		for {
+			r := next()
+			if r == nil {
+				return
+			}
+			select {
+			case out <- r:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	return out
+}
+
+func perKeyActionToChan(args []string, action func(*cid.Cid) *filestore.ListRes, ctx context.Context) <-chan interface{} {
+	out := make(chan interface{}, 128)
+	go func() {
+		defer close(out)
+		for _, arg := range args {
+			c, err := cid.Decode(arg)
+			if err != nil {
+				out <- &filestore.ListRes{
+					Status:   filestore.StatusOtherError,
+					ErrorMsg: fmt.Sprintf("%s: %v", arg, err),
+				}
+				continue
+			}
+			r := action(c)
+			select {
+			case out <- r:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	return out
+}

core/commands/root.go

@@ -47,6 +47,7 @@ ADVANCED COMMANDS
   pin           Pin objects to local storage
   repo          Manipulate the IPFS repository
   stats         Various operational stats
+  filestore     Manage the filestore (experimental)
 
 NETWORK COMMANDS
   id            Show info about IPFS peers
@@ -124,6 +125,7 @@ var rootSubcommands = map[string]*cmds.Command{
 	"update":    ExternalBinary(),
 	"version":   VersionCmd,
 	"bitswap":   BitswapCmd,
+	"filestore": FileStoreCmd,
 }
 
 // RootRO is the readonly version of Root

filestore/filestore.go

@@ -19,6 +19,14 @@ type Filestore struct {
 	bs blockstore.Blockstore
 }
 
+func (f *Filestore) FileManager() *FileManager {
+	return f.fm
+}
+
+func (f *Filestore) MainBlockstore() blockstore.Blockstore {
+	return f.bs
+}
+
 func NewFilestore(bs blockstore.Blockstore, fm *FileManager) *Filestore {
 	return &Filestore{fm, bs}
 }

filestore/fsrefstore.go

@@ -28,7 +28,8 @@ type FileManager struct {
 }
 
 type CorruptReferenceError struct {
-	Err error
+	Code Status
+	Err  error
 }
 
 func (c CorruptReferenceError) Error() string {
@@ -108,6 +109,10 @@ func (f *FileManager) getDataObj(c *cid.Cid) (*pb.DataObj, error) {
 		//
 	}
 
+	return unmarshalDataObj(o)
+}
+
+func unmarshalDataObj(o interface{}) (*pb.DataObj, error) {
 	data, ok := o.([]byte)
 	if !ok {
 		return nil, fmt.Errorf("stored filestore dataobj was not a []byte")
@@ -127,20 +132,24 @@ func (f *FileManager) readDataObj(c *cid.Cid, d *pb.DataObj) ([]byte, error) {
 	abspath := filepath.Join(f.root, p)
 
 	fi, err := os.Open(abspath)
-	if err != nil {
-		return nil, &CorruptReferenceError{err}
+	if os.IsNotExist(err) {
+		return nil, &CorruptReferenceError{StatusFileNotFound, err}
+	} else if err != nil {
+		return nil, &CorruptReferenceError{StatusFileError, err}
 	}
 	defer fi.Close()
 
 	_, err = fi.Seek(int64(d.GetOffset()), os.SEEK_SET)
 	if err != nil {
-		return nil, &CorruptReferenceError{err}
+		return nil, &CorruptReferenceError{StatusFileError, err}
 	}
 
 	outbuf := make([]byte, d.GetSize_())
 	_, err = io.ReadFull(fi, outbuf)
-	if err != nil {
-		return nil, &CorruptReferenceError{err}
+	if err == io.EOF || err == io.ErrUnexpectedEOF {
+		return nil, &CorruptReferenceError{StatusFileChanged, err}
+	} else if err != nil {
+		return nil, &CorruptReferenceError{StatusFileError, err}
 	}
 
 	outcid, err := c.Prefix().Sum(outbuf)
@@ -149,7 +158,8 @@ func (f *FileManager) readDataObj(c *cid.Cid, d *pb.DataObj) ([]byte, error) {
 	}
 
 	if !c.Equals(outcid) {
-		return nil, &CorruptReferenceError{fmt.Errorf("data in file did not match. %s offset %d", d.GetFilePath(), d.GetOffset())}
+		return nil, &CorruptReferenceError{StatusFileChanged,
+			fmt.Errorf("data in file did not match. %s offset %d", d.GetFilePath(), d.GetOffset())}
 	}
 
 	return outbuf, nil