Pinning content

api/api.go

@@ -48,7 +48,7 @@ import (
 	"github.com/ethersphere/swarm/storage/feed"
 	"github.com/ethersphere/swarm/storage/feed/lookup"
 
-	opentracing "github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go"
 )
 
 var (
@@ -188,16 +188,18 @@ type API struct {
 	fileStore *storage.FileStore
 	dns       Resolver
 	Tags      *chunk.Tags
+	PinApi    *PinApi
 	Decryptor func(context.Context, string) DecryptFunc
 }
 
 // NewAPI the api constructor initialises a new API instance.
-func NewAPI(fileStore *storage.FileStore, dns Resolver, feedHandler *feed.Handler, pk *ecdsa.PrivateKey, tags *chunk.Tags) (self *API) {
+func NewAPI(fileStore *storage.FileStore, dns Resolver, feedHandler *feed.Handler, pk *ecdsa.PrivateKey, tags *chunk.Tags, pinApi *PinApi) (self *API) {
 	self = &API{
 		fileStore: fileStore,
 		dns:       dns,
 		feed:      feedHandler,
 		Tags:      tags,
+		PinApi:    pinApi,
 		Decryptor: func(ctx context.Context, credentials string) DecryptFunc {
 			return self.doDecrypt(ctx, credentials, pk)
 		},
@@ -211,9 +213,9 @@ func (a *API) Retrieve(ctx context.Context, addr storage.Address) (reader storag
 }
 
 // Store wraps the Store API call of the embedded FileStore
-func (a *API) Store(ctx context.Context, data io.Reader, size int64, toEncrypt bool) (addr storage.Address, wait func(ctx context.Context) error, err error) {
+func (a *API) Store(ctx context.Context, data io.Reader, size int64, toEncrypt bool, pinCounter uint8) (addr storage.Address, wait func(ctx context.Context) error, err error) {
 	log.Debug("api.store", "size", size)
-	return a.fileStore.Store(ctx, data, size, toEncrypt)
+	return a.fileStore.Store(ctx, data, size, toEncrypt, pinCounter)
 }
 
 // Resolve a name into a content-addressed hash
@@ -400,7 +402,7 @@ func (a *API) Get(ctx context.Context, decrypt DecryptFunc, manifestAddr storage
 	return
 }
 
-func (a *API) Delete(ctx context.Context, addr string, path string) (storage.Address, error) {
+func (a *API) Delete(ctx context.Context, addr string, path string, pinCounter uint8) (storage.Address, error) {
 	apiDeleteCount.Inc(1)
 	uri, err := Parse("bzz:/" + addr)
 	if err != nil {
@@ -415,7 +417,7 @@ func (a *API) Delete(ctx context.Context, addr string, path string) (storage.Add
 	newKey, err := a.UpdateManifest(ctx, key, func(mw *ManifestWriter) error {
 		log.Debug(fmt.Sprintf("removing %s from manifest %s", path, key.Log()))
 		return mw.RemoveEntry(path)
-	})
+	}, pinCounter)
 	if err != nil {
 		apiDeleteFail.Inc(1)
 		return nil, err
@@ -561,7 +563,7 @@ func (a *API) GetManifestList(ctx context.Context, decryptor DecryptFunc, addr s
 	return list, nil
 }
 
-func (a *API) UpdateManifest(ctx context.Context, addr storage.Address, update func(mw *ManifestWriter) error) (storage.Address, error) {
+func (a *API) UpdateManifest(ctx context.Context, addr storage.Address, update func(mw *ManifestWriter) error, pinCounter uint8) (storage.Address, error) {
 	apiManifestUpdateCount.Inc(1)
 	mw, err := a.NewManifestWriter(ctx, addr, nil)
 	if err != nil {
@@ -574,7 +576,7 @@ func (a *API) UpdateManifest(ctx context.Context, addr storage.Address, update f
 		return nil, err
 	}
 
-	addr, err = mw.Store()
+	addr, err = mw.Store(pinCounter)
 	if err != nil {
 		apiManifestUpdateFail.Inc(1)
 		return nil, err
@@ -603,7 +605,8 @@ func (a *API) Modify(ctx context.Context, addr storage.Address, path, contentHas
 		trie.deleteEntry(path, quitC)
 	}
 
-	if err := trie.recalcAndStore(); err != nil {
+	// Dont pin manifest as this is used only in test cases
+	if err := trie.recalcAndStore(DONT_PIN); err != nil {
 		apiModifyFail.Inc(1)
 		return nil, err
 	}
@@ -644,13 +647,18 @@ func (a *API) AddFile(ctx context.Context, mhash, path, fname string, content []
 		return nil, "", err
 	}
 
-	fkey, err := mw.AddEntry(ctx, bytes.NewReader(content), entry)
+	// TODO_PIN: support pinning when creating a file in fuse
+	// For now, don't pin it
+	fkey, err := mw.AddEntry(ctx, bytes.NewReader(content), entry, DONT_PIN)
 	if err != nil {
 		apiAddFileFail.Inc(1)
 		return nil, "", err
 	}
 
-	newMkey, err := mw.Store()
+	// TODO_PIN: support pinning manifests when creating a file in fuse
+	// For now, don't pin it
+	// care should be taken to unpin the old manifest
+	newMkey, err := mw.Store(DONT_PIN)
 	if err != nil {
 		apiAddFileFail.Inc(1)
 		return nil, "", err
@@ -660,7 +668,7 @@ func (a *API) AddFile(ctx context.Context, mhash, path, fname string, content []
 	return fkey, newMkey.String(), nil
 }
 
-func (a *API) UploadTar(ctx context.Context, bodyReader io.ReadCloser, manifestPath, defaultPath string, mw *ManifestWriter) (storage.Address, error) {
+func (a *API) UploadTar(ctx context.Context, bodyReader io.ReadCloser, manifestPath, defaultPath string, mw *ManifestWriter, pinCounter uint8) (storage.Address, error) {
 	apiUploadTarCount.Inc(1)
 	var contentKey storage.Address
 	tr := tar.NewReader(bodyReader)
@@ -694,7 +702,7 @@ func (a *API) UploadTar(ctx context.Context, bodyReader io.ReadCloser, manifestP
 			Size:        hdr.Size,
 			ModTime:     hdr.ModTime,
 		}
-		contentKey, err = mw.AddEntry(ctx, tr, entry)
+		contentKey, err = mw.AddEntry(ctx, tr, entry, pinCounter)
 		if err != nil {
 			apiUploadTarFail.Inc(1)
 			return nil, fmt.Errorf("error adding manifest entry from tar stream: %s", err)
@@ -713,7 +721,7 @@ func (a *API) UploadTar(ctx context.Context, bodyReader io.ReadCloser, manifestP
 				Size:        hdr.Size,
 				ModTime:     hdr.ModTime,
 			}
-			contentKey, err = mw.AddEntry(ctx, nil, entry)
+			contentKey, err = mw.AddEntry(ctx, nil, entry, pinCounter)
 			if err != nil {
 				apiUploadTarFail.Inc(1)
 				return nil, fmt.Errorf("error adding default manifest entry from tar stream: %s", err)
@@ -759,7 +767,9 @@ func (a *API) RemoveFile(ctx context.Context, mhash string, path string, fname s
 		return "", err
 	}
 
-	newMkey, err := mw.Store()
+	// TODO_PIN: If a pinned manifest is modified then it needs to pinned too
+	// care should be taken to unpin the old manifest
+	newMkey, err := mw.Store(DONT_PIN)
 	if err != nil {
 		apiRmFileFail.Inc(1)
 		return "", err
@@ -834,13 +844,17 @@ func (a *API) AppendFile(ctx context.Context, mhash, path, fname string, existin
 		ModTime:     time.Now(),
 	}
 
-	fkey, err := mw.AddEntry(ctx, io.Reader(combinedReader), entry)
+	// TODO_PIN: If a pinned file is modified in fuse, pin it here too
+	// For now, this is ignored
+	fkey, err := mw.AddEntry(ctx, io.Reader(combinedReader), entry, DONT_PIN)
 	if err != nil {
 		apiAppendFileFail.Inc(1)
 		return nil, "", err
 	}
 
-	newMkey, err := mw.Store()
+	// TODO_PIN: If a pinned file is modified in fuse, pin the new manifest too
+	// For now, this is ignored
+	newMkey, err := mw.Store(DONT_PIN)
 	if err != nil {
 		apiAppendFileFail.Inc(1)
 		return nil, "", err

api/api_test.go

@@ -58,7 +58,7 @@ func testAPI(t *testing.T, f func(*API, *chunk.Tags, bool)) {
 			return
 		}
 		defer cleanup()
-		api := NewAPI(fileStore, nil, nil, nil, tags)
+		api := NewAPI(fileStore, nil, nil, nil, tags, nil)
 		f(api, tags, v)
 	}
 }
@@ -157,7 +157,7 @@ func TestApiTagLarge(t *testing.T) {
 			t.Fatal(err)
 		}
 		ctx := sctx.SetTag(context.Background(), tag.Uid)
-		key, waitContent, err := api.Store(ctx, randomContentReader, int64(contentLength), toEncrypt)
+		key, waitContent, err := api.Store(ctx, randomContentReader, int64(contentLength), toEncrypt, DONT_PIN)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -438,7 +438,7 @@ func TestDecryptOriginForbidden(t *testing.T) {
 		Access: &AccessEntry{Type: AccessTypePass},
 	}
 
-	api := NewAPI(nil, nil, nil, nil, chunk.NewTags())
+	api := NewAPI(nil, nil, nil, nil, chunk.NewTags(), nil)
 
 	f := api.Decryptor(ctx, "")
 	err := f(me)
@@ -472,7 +472,7 @@ func TestDecryptOrigin(t *testing.T) {
 			Access: &AccessEntry{Type: AccessTypePass},
 		}
 
-		api := NewAPI(nil, nil, nil, nil, chunk.NewTags())
+		api := NewAPI(nil, nil, nil, nil, chunk.NewTags(), nil)
 
 		f := api.Decryptor(ctx, "")
 		err := f(me)
@@ -556,13 +556,13 @@ func putString(ctx context.Context, a *API, content string, contentType string,
 	log.Trace("created new tag", "uid", tag.Uid)
 
 	cCtx := sctx.SetTag(ctx, tag.Uid)
-	key, waitContent, err := a.Store(cCtx, r, int64(len(content)), toEncrypt)
+	key, waitContent, err := a.Store(cCtx, r, int64(len(content)), toEncrypt, DONT_PIN)
 	if err != nil {
 		return nil, nil, err
 	}
 	manifest := fmt.Sprintf(`{"entries":[{"hash":"%v","contentType":"%s"}]}`, key, contentType)
 	r = strings.NewReader(manifest)
-	key, waitManifest, err := a.Store(cCtx, r, int64(len(manifest)), toEncrypt)
+	key, waitManifest, err := a.Store(cCtx, r, int64(len(manifest)), toEncrypt, DONT_PIN)
 	if err != nil {
 		return nil, nil, err
 	}

api/client/client.go

@@ -63,7 +63,7 @@ type Client struct {
 
 // UploadRaw uploads raw data to swarm and returns the resulting hash. If toEncrypt is true it
 // uploads encrypted data
-func (c *Client) UploadRaw(r io.Reader, size int64, toEncrypt bool) (string, error) {
+func (c *Client) UploadRaw(r io.Reader, size int64, toEncrypt bool, toPin bool) (string, error) {
 	if size <= 0 {
 		return "", errors.New("data size must be greater than zero")
 	}
@@ -78,6 +78,11 @@ func (c *Client) UploadRaw(r io.Reader, size int64, toEncrypt bool) (string, err
 	req.ContentLength = size
 	req.Header.Set(swarmhttp.SwarmTagHeaderName, fmt.Sprintf("raw_upload_%d", time.Now().Unix()))
 
+	// Set the pinning header if the file needs to be pinned
+	if toPin {
+		req.Header.Set(swarmhttp.SwarmPinContent, fmt.Sprintf("pin_time_%d", time.Now().Unix()))
+	}
+
 	res, err := http.DefaultClient.Do(req)
 	if err != nil {
 		return "", err
@@ -151,11 +156,11 @@ func Open(path string) (*File, error) {
 // (if the manifest argument is non-empty) or creates a new manifest containing
 // the file, returning the resulting manifest hash (the file will then be
 // available at bzz:/<hash>/<path>)
-func (c *Client) Upload(file *File, manifest string, toEncrypt bool) (string, error) {
+func (c *Client) Upload(file *File, manifest string, toEncrypt bool, toPin bool) (string, error) {
 	if file.Size <= 0 {
 		return "", errors.New("file size must be greater than zero")
 	}
-	return c.TarUpload(manifest, &FileUploader{file}, "", toEncrypt)
+	return c.TarUpload(manifest, &FileUploader{file}, "", toEncrypt, toPin)
 }
 
 // Download downloads a file with the given path from the swarm manifest with
@@ -185,7 +190,7 @@ func (c *Client) Download(hash, path string) (*File, error) {
 // directory will then be available at bzz:/<hash>/path/to/file), with
 // the file specified in defaultPath being uploaded to the root of the manifest
 // (i.e. bzz:/<hash>/)
-func (c *Client) UploadDirectory(dir, defaultPath, manifest string, toEncrypt bool) (string, error) {
+func (c *Client) UploadDirectory(dir, defaultPath, manifest string, toEncrypt bool, toPin bool) (string, error) {
 	stat, err := os.Stat(dir)
 	if err != nil {
 		return "", err
@@ -200,7 +205,7 @@ func (c *Client) UploadDirectory(dir, defaultPath, manifest string, toEncrypt bo
 			return "", fmt.Errorf("default path: %v", err)
 		}
 	}
-	return c.TarUpload(manifest, &DirectoryUploader{dir}, defaultPath, toEncrypt)
+	return c.TarUpload(manifest, &DirectoryUploader{dir}, defaultPath, toEncrypt, toPin)
 }
 
 // DownloadDirectory downloads the files contained in a swarm manifest under
@@ -358,12 +363,13 @@ func (c *Client) DownloadFile(hash, path, dest, credentials string) error {
 }
 
 // UploadManifest uploads the given manifest to swarm
-func (c *Client) UploadManifest(m *api.Manifest, toEncrypt bool) (string, error) {
+func (c *Client) UploadManifest(m *api.Manifest, toEncrypt bool, toPin bool) (string, error) {
 	data, err := json.Marshal(m)
 	if err != nil {
 		return "", err
 	}
-	return c.UploadRaw(bytes.NewReader(data), int64(len(data)), toEncrypt)
+
+	return c.UploadRaw(bytes.NewReader(data), int64(len(data)), toEncrypt, toPin)
 }
 
 // DownloadManifest downloads a swarm manifest
@@ -498,7 +504,7 @@ type UploadFn func(file *File) error
 
 // TarUpload uses the given Uploader to upload files to swarm as a tar stream,
 // returning the resulting manifest hash
-func (c *Client) TarUpload(hash string, uploader Uploader, defaultPath string, toEncrypt bool) (string, error) {
+func (c *Client) TarUpload(hash string, uploader Uploader, defaultPath string, toEncrypt bool, toPin bool) (string, error) {
 	ctx, sp := spancontext.StartSpan(context.Background(), "api.client.tarupload")
 	defer sp.Finish()
 
@@ -540,6 +546,11 @@ func (c *Client) TarUpload(hash string, uploader Uploader, defaultPath string, t
 
 	req.Header.Set(swarmhttp.SwarmTagHeaderName, tag)
 
+	// Set the pinning header if the file is to be pinned
+	if toPin {
+		req.Header.Set(swarmhttp.SwarmPinContent, fmt.Sprintf("pin_time_%d", time.Now().Unix()))
+	}
+
 	// use 'Expect: 100-continue' so we don't send the request body if
 	// the server refuses the request
 	req.Header.Set("Expect", "100-continue")

api/client/client_test.go

@@ -61,7 +61,7 @@ func testClientUploadDownloadRaw(toEncrypt bool, t *testing.T) {
 
 	// upload some raw data
 	data := []byte("foo123")
-	hash, err := client.UploadRaw(bytes.NewReader(data), int64(len(data)), toEncrypt)
+	hash, err := client.UploadRaw(bytes.NewReader(data), int64(len(data)), toEncrypt, false)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -112,7 +112,7 @@ func testClientUploadDownloadFiles(toEncrypt bool, t *testing.T) {
 				Size:        int64(len(data)),
 			},
 		}
-		hash, err := client.Upload(file, manifest, toEncrypt)
+		hash, err := client.Upload(file, manifest, toEncrypt, false)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -206,7 +206,7 @@ func TestClientUploadDownloadDirectory(t *testing.T) {
 	// upload the directory
 	client := NewClient(srv.URL)
 	defaultPath := testDirFiles[0]
-	hash, err := client.UploadDirectory(dir, defaultPath, "", false)
+	hash, err := client.UploadDirectory(dir, defaultPath, "", false, false)
 	if err != nil {
 		t.Fatalf("error uploading directory: %s", err)
 	}
@@ -274,7 +274,7 @@ func testClientFileList(toEncrypt bool, t *testing.T) {
 	defer os.RemoveAll(dir)
 
 	client := NewClient(srv.URL)
-	hash, err := client.UploadDirectory(dir, "", "", toEncrypt)
+	hash, err := client.UploadDirectory(dir, "", "", toEncrypt, false)
 	if err != nil {
 		t.Fatalf("error uploading directory: %s", err)
 	}
@@ -433,7 +433,7 @@ func TestClientBzzWithFeed(t *testing.T) {
 	}
 
 	// upload data to bzz:// and retrieve the content-addressed manifest hash, hex-encoded.
-	manifestAddressHex, err := swarmClient.Upload(f, "", false)
+	manifestAddressHex, err := swarmClient.Upload(f, "", false, false)
 	if err != nil {
 		t.Fatalf("Error creating manifest: %s", err)
 	}

api/filesystem.go

@@ -121,7 +121,9 @@ func (fs *FileSystem) Upload(lpath, index string, toEncrypt bool) (string, error
 			var hash storage.Address
 			var wait func(context.Context) error
 			ctx := context.TODO()
-			hash, wait, err = fs.api.fileStore.Store(ctx, f, stat.Size(), toEncrypt)
+
+			// Fixing the pin counter as DONT_PIN , since this is deprecated
+			hash, wait, err = fs.api.fileStore.Store(ctx, f, stat.Size(), toEncrypt, DONT_PIN)
 			if err != nil {
 				errors[i] = err
 				return
@@ -165,7 +167,8 @@ func (fs *FileSystem) Upload(lpath, index string, toEncrypt bool) (string, error
 		trie.addEntry(entry, quitC)
 	}
 
-	err2 := trie.recalcAndStore()
+	// Fixing the pin counter as DONT_PIN , since this is deprecated
+	err2 := trie.recalcAndStore(DONT_PIN)
 	var hs string
 	if err2 == nil {
 		hs = trie.ref.Hex()

api/filesystem_test.go

@@ -111,7 +111,7 @@ func TestApiDirUploadModify(t *testing.T) {
 			return
 		}
 		ctx := context.TODO()
-		hash, wait, err := api.Store(ctx, bytes.NewReader(index), int64(len(index)), toEncrypt)
+		hash, wait, err := api.Store(ctx, bytes.NewReader(index), int64(len(index)), toEncrypt, DONT_PIN)
 		if err != nil {
 			t.Errorf("unexpected error: %v", err)
 			return