Fail upload if it uses more then available space (#220)

Before upload, check that more then the disk has more than the minimum
free space. Upload up to the size limit that would exceed the available
space.

Fixes #179

Author: gammazero

blob/blob.go

@@ -12,8 +12,9 @@ import (
 )
 
 var (
-	ErrBlobTooLarge = errors.New("blob size exceeds the maximum allowed")
-	ErrBlobNotFound = errors.New("no blob is found with given ID")
+	ErrBlobNotFound   = errors.New("no blob is found with given ID")
+	ErrBlobTooLarge   = errors.New("blob size exceeds the maximum allowed")
+	ErrNotEnoughSpace = errors.New("insufficient local storage space remaining")
 )
 
 var (
@@ -44,7 +45,7 @@ type (
 		Status      string
 	}
 	Store interface {
-		Put(context.Context, io.ReadCloser) (*Descriptor, error)
+		Put(context.Context, io.Reader) (*Descriptor, error)
 		Describe(context.Context, ID) (*Descriptor, error)
 		Get(context.Context, ID) (io.ReadSeekCloser, error)
 	}

blob/local_store.go

@@ -3,9 +3,12 @@ package blob
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"os"
 	"path/filepath"
+
+	"github.com/gammazero/fsutil/disk"
 )
 
 var _ Store = (*LocalStore)(nil)
@@ -14,15 +17,18 @@ var _ Store = (*LocalStore)(nil)
 // Blobs are stored as flat files, named by their ID with .bin extension.
 // This store is used primarily for testing purposes.
 type LocalStore struct {
-	dir string
+	dir          string
+	minFreeSpace uint64
 }
 
 // NewLocalStore instantiates a new LocalStore and uses the given dir as the place to store blobs.
 // Blobs are stored as flat files, named by their ID with .bin extension.
-func NewLocalStore(dir string) *LocalStore {
+func NewLocalStore(dir string, options ...Option) *LocalStore {
+	opts := getOpts(options)
 	logger.Debugw("Instantiated local store", "dir", dir)
 	return &LocalStore{
-		dir: dir,
+		dir:          dir,
+		minFreeSpace: opts.minFreeSpace,
 	}
 }
 
@@ -32,25 +38,53 @@ func (l *LocalStore) Dir() string {
 }
 
 // Put reads the given reader fully and stores its content in the store directory as flat files.
-// The reader content is first stored in a temporary directory and upon successful storage is moved to the store directory.
-// The Descriptor.ModificationTime is set to the modification date of the file that corresponds to the content.
-// The Descriptor.ID is randomly generated; see NewID.
-func (l *LocalStore) Put(_ context.Context, reader io.ReadCloser) (*Descriptor, error) {
-	// TODO: add size limiter here and return ErrBlobTooLarge.
-	id, err := NewID()
-	if err != nil {
-		return nil, err
+//
+// The reader content is first stored in a temporary directory and upon
+// successful storage is moved to the store directory. The
+// Descriptor.ModificationTime is set to the modification date of the file that
+// corresponds to the content. The Descriptor.ID is randomly generated; see
+// NewID.
+//
+// Before a blob is written, the minimum amount of free space must be available
+// on the local disk. If writing the blob consumes more then the available
+// space (free space - minimum free), then this results in an error.
+func (l *LocalStore) Put(_ context.Context, reader io.Reader) (*Descriptor, error) {
+	var limit int64
+	if l.minFreeSpace != 0 {
+		usage, err := disk.Usage(l.dir)
+		if err != nil {
+			return nil, fmt.Errorf("cannot get disk usage: %w", err)
+		}
+		if usage.Free <= l.minFreeSpace {
+			return nil, ErrNotEnoughSpace
+		}
+
+		// Do not write more than the remaining storage - minimum free space.
+		limit = int64(usage.Free - l.minFreeSpace)
+		reader = io.LimitReader(reader, limit)
 	}
+
 	dest, err := os.CreateTemp(l.dir, "motion_local_store_*.bin.temp")
 	if err != nil {
 		return nil, err
 	}
 	defer dest.Close()
+
 	written, err := io.Copy(dest, reader)
 	if err != nil {
 		os.Remove(dest.Name())
 		return nil, err
 	}
+
+	if limit != 0 && written == limit {
+		os.Remove(dest.Name())
+		return nil, ErrBlobTooLarge
+	}
+
+	id, err := NewID()
+	if err != nil {
+		return nil, err
+	}
 	if err = os.Rename(dest.Name(), filepath.Join(l.dir, id.String()+".bin")); err != nil {
 		return nil, err
 	}

blob/local_store_test.go

@@ -0,0 +1,51 @@
+package blob_test
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"testing"
+
+	"github.com/filecoin-project/motion/blob"
+	"github.com/gammazero/fsutil/disk"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWriteOK(t *testing.T) {
+	tmpDir := t.TempDir()
+
+	store := blob.NewLocalStore(tmpDir)
+	buf := []byte("This is a test")
+	readCloser := io.NopCloser(bytes.NewReader(buf))
+
+	desc, err := store.Put(context.Background(), readCloser)
+	require.NoError(t, err)
+	require.NotNil(t, desc)
+	require.Equal(t, uint64(len(buf)), desc.Size)
+}
+
+func TestInsufficientSpace(t *testing.T) {
+	tmpDir := t.TempDir()
+	usage, err := disk.Usage(tmpDir)
+	require.NoError(t, err)
+
+	store := blob.NewLocalStore(tmpDir, blob.WithMinFreeSpace(int64(usage.Free+blob.Gib)))
+	readCloser := io.NopCloser(bytes.NewReader([]byte("This is a test")))
+
+	_, err = store.Put(context.Background(), readCloser)
+	require.ErrorIs(t, err, blob.ErrNotEnoughSpace)
+}
+
+func TestWriteTooLarge(t *testing.T) {
+	tmpDir := t.TempDir()
+	usage, err := disk.Usage(tmpDir)
+	require.NoError(t, err)
+
+	store := blob.NewLocalStore(tmpDir, blob.WithMinFreeSpace(int64(usage.Free-5*blob.Kib)))
+
+	buf := make([]byte, 32*blob.Kib)
+	readCloser := io.NopCloser(bytes.NewReader(buf))
+
+	_, err = store.Put(context.Background(), readCloser)
+	require.ErrorIs(t, err, blob.ErrBlobTooLarge)
+}

blob/option.go

@@ -0,0 +1,42 @@
+package blob
+
+const (
+	Kib = 1 << (10 * (iota + 1))
+	Mib
+	Gib
+)
+
+const defaultMinFreeSpace = 64 * Mib
+
+// config contains all options for LocalStore.
+type config struct {
+	minFreeSpace uint64
+}
+
+// Option is a function that sets a value in a config.
+type Option func(*config)
+
+// getOpts creates a config and applies Options to it.
+func getOpts(options []Option) config {
+	cfg := config{
+		minFreeSpace: defaultMinFreeSpace,
+	}
+	for _, opt := range options {
+		opt(&cfg)
+	}
+	return cfg
+}
+
+// WithMinFreeSpace seta the minimum amount of free dist space that must remain
+// after writing a blob. If unset or 0 then defaultMinFreeSpace is used. If -1, then
+// no free space checks are performed.
+func WithMinFreeSpace(space int64) Option {
+	return func(c *config) {
+		if space == 0 {
+			space = defaultMinFreeSpace
+		} else if space < 0 {
+			space = 0
+		}
+		c.minFreeSpace = uint64(space)
+	}
+}

cmd/motion/main.go

@@ -37,6 +37,12 @@ func main() {
 				Value:       os.TempDir(),
 				EnvVars:     []string{"MOTION_STORE_DIR"},
 			},
+			&cli.Int64Flag{
+				Name:        "minFreeDiskSpace",
+				Usage:       "Minumum amount of free space that must remaio on disk after writing blob. Set -1 to disable checks.",
+				DefaultText: "64 Mib",
+				EnvVars:     []string{"MIN_FREE_DISK_SPACE"},
+			},
 			&cli.StringFlag{
 				Name:    "walletKey",
 				Usage:   "Hex encoded private key for the wallet to use with motion",
@@ -205,6 +211,7 @@ func main() {
 					singularity.WithScheduleDealNumber(cctx.Int("experimentalSingularityScheduleDealNumber")),
 					singularity.WithVerifiedDeal(cctx.Bool("verifiedDeal")),
 					singularity.WithCleanupInterval(cctx.Duration("experimentalSingularityCleanupInterval")),
+					singularity.WithMinFreeSpace(cctx.Int64("minFreeDiskSpace")),
 				)
 				if err != nil {
 					logger.Errorw("Failed to instantiate singularity store", "err", err)
@@ -222,7 +229,7 @@ func main() {
 				}()
 				store = singularityStore
 			} else {
-				store = blob.NewLocalStore(storeDir)
+				store = blob.NewLocalStore(storeDir, blob.WithMinFreeSpace(cctx.Int64("minFreeDiskSpace")))
 				logger.Infow("Using local blob store", "storeDir", storeDir)
 			}
 

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/data-preservation-programs/singularity v0.5.8
 	github.com/filecoin-project/go-address v1.1.0
 	github.com/filecoin-project/go-state-types v0.12.0
+	github.com/gammazero/fsutil v0.0.1
 	github.com/google/uuid v1.3.1
 	github.com/gotidy/ptr v1.4.0
 	github.com/ipfs/go-log/v2 v2.5.1
@@ -160,7 +161,7 @@ require (
 	golang.org/x/mod v0.12.0 // indirect
 	golang.org/x/net v0.17.0 // indirect
 	golang.org/x/sync v0.3.0 // indirect
-	golang.org/x/sys v0.13.0 // indirect
+	golang.org/x/sys v0.14.0 // indirect
 	golang.org/x/text v0.13.0 // indirect
 	golang.org/x/tools v0.12.1-0.20230815132531-74c255bcf846 // indirect
 	golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect

go.sum

@@ -103,6 +103,8 @@ github.com/francoispqt/gojay v1.2.13/go.mod h1:ehT5mTG4ua4581f1++1WLG0vPdaA9HaiD
 github.com/frankban/quicktest v1.14.3/go.mod h1:mgiwOwqx65TmIk1wJ6Q7wvnVMocbUorkibMOrVTHZps=
 github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/gammazero/fsutil v0.0.1 h1:ELBzTchuwz+f6Xh5a3PS0J1ssuvMRP+bU4/h+eTBGaQ=
+github.com/gammazero/fsutil v0.0.1/go.mod h1:RUdM7Ubfblvprq9CJmUtDFa1HA2Qp1kxsWIKwTVvFvw=
 github.com/getsentry/sentry-go v0.18.0 h1:MtBW5H9QgdcJabtZcuJG80BMOwaBpkRDZkxRkNC1sN0=
 github.com/getsentry/sentry-go v0.18.0/go.mod h1:Kgon4Mby+FJ7ZWHFUAZgVaIa8sxHtnRJRLTXZr51aKQ=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
@@ -791,8 +793,8 @@ golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220704084225-05e143d24a9e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.13.0 h1:Af8nKPmuFypiUBjVoU9V20FiaFXOcuZI21p0ycVYYGE=
-golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.14.0 h1:Vz7Qs629MkJkGyHxUlRHizWJRG2j8fbQKjELVSNhy7Q=
+golang.org/x/sys v0.14.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

integration/ribs/store.go

@@ -86,7 +86,7 @@ func (s *Store) Start(_ context.Context) error {
 	// TODO: change RIBS to take context.
 	return s.ribs.Start()
 }
-func (s *Store) Put(ctx context.Context, in io.ReadCloser) (*blob.Descriptor, error) {
+func (s *Store) Put(ctx context.Context, in io.Reader) (*blob.Descriptor, error) {
 
 	// Generate ID early to fail early if generation fails.
 	id, err := uuid.NewRandom()

integration/singularity/options.go

@@ -43,6 +43,7 @@ type (
 		maxPendingDealSize    string
 		maxPendingDealNumber  int
 		cleanupInterval       time.Duration
+		minFreeSpace          int64
 	}
 )
 
@@ -331,3 +332,13 @@ func WithCleanupInterval(v time.Duration) Option {
 		return nil
 	}
 }
+
+// WithMinFreeSpce configures the minimul free disk space that must remain
+// after storing a blob. A value of zero uses the default value and -1 disabled
+// checks.
+func WithMinFreeSpace(space int64) Option {
+	return func(o *options) error {
+		o.minFreeSpace = space
+		return nil
+	}
+}

integration/singularity/store.go

@@ -52,7 +52,7 @@ func NewStore(o ...Option) (*Store, error) {
 
 	return &Store{
 		options:    opts,
-		local:      blob.NewLocalStore(opts.storeDir),
+		local:      blob.NewLocalStore(opts.storeDir, blob.WithMinFreeSpace(opts.minFreeSpace)),
 		sourceName: "source",
 		toPack:     make(chan uint64, 1),
 		closing:    make(chan struct{}),
@@ -374,7 +374,7 @@ func (s *Store) Shutdown(ctx context.Context) error {
 	return nil
 }
 
-func (s *Store) Put(ctx context.Context, reader io.ReadCloser) (*blob.Descriptor, error) {
+func (s *Store) Put(ctx context.Context, reader io.Reader) (*blob.Descriptor, error) {
 	desc, err := s.local.Put(ctx, reader)
 	if err != nil {
 		return nil, fmt.Errorf("failed to put file locally: %w", err)