pushsync

api/api_test.go

@@ -141,7 +141,7 @@ func TestApiPut(t *testing.T) {
 		resp := testGet(t, api, addr.Hex(), "")
 		checkResponse(t, resp, exp)
 		tag := tags.All()[0]
-		chunktesting.CheckTag(t, tag, 2, 2, 0, 2) //1 chunk data, 1 chunk manifest
+		chunktesting.CheckTag(t, tag, 2, 2, 0, 0, 0, 2) //1 chunk data, 1 chunk manifest
 	})
 }
 
@@ -150,7 +150,7 @@ func TestApiTagLarge(t *testing.T) {
 	const contentLength = 4096 * 4095
 	testAPI(t, func(api *API, tags *chunk.Tags, toEncrypt bool) {
 		randomContentReader := io.LimitReader(crand.Reader, int64(contentLength))
-		tag, err := api.Tags.Create("unnamed-tag", 0)
+		tag, err := api.Tags.Create(context.Background(), "unnamed-tag", 0)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -168,11 +168,11 @@ func TestApiTagLarge(t *testing.T) {
 		if toEncrypt {
 			tag := tags.All()[0]
 			expect := int64(4095 + 64 + 1)
-			chunktesting.CheckTag(t, tag, expect, expect, 0, expect)
+			chunktesting.CheckTag(t, tag, expect, expect, 0, 0, 0, expect)
 		} else {
 			tag := tags.All()[0]
 			expect := int64(4095 + 32 + 1)
-			chunktesting.CheckTag(t, tag, expect, expect, 0, expect)
+			chunktesting.CheckTag(t, tag, expect, expect, 0, 0, 0, expect)
 		}
 	})
 }
@@ -549,18 +549,18 @@ func TestDetectContentType(t *testing.T) {
 // putString provides singleton manifest creation on top of api.API
 func putString(ctx context.Context, a *API, content string, contentType string, toEncrypt bool) (k storage.Address, wait func(context.Context) error, err error) {
 	r := strings.NewReader(content)
-	tag, err := a.Tags.Create("unnamed-tag", 0)
+	tag, err := a.Tags.Create(ctx, "unnamed-tag", 0)
 
-	log.Trace("created new tag", "uid", tag.Uid)
+	log.Trace("created new tag", "id", tag.Uid)
 
-	cCtx := sctx.SetTag(ctx, tag.Uid)
-	key, waitContent, err := a.Store(cCtx, r, int64(len(content)), toEncrypt)
+	ctx = sctx.SetTag(ctx, tag.Uid)
+	key, waitContent, err := a.Store(ctx, r, int64(len(content)), toEncrypt)
 	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(ctx, r, int64(len(manifest)), toEncrypt)
 	if err != nil {
 		return nil, nil, err
 	}

api/client/client_test.go

@@ -51,7 +51,7 @@ func TestClientUploadDownloadRaw(t *testing.T) {
 
 	// check the tag was created successfully
 	tag := srv.Tags.All()[0]
-	chunktesting.CheckTag(t, tag, 1, 1, 0, 1)
+	chunktesting.CheckTag(t, tag, 1, 1, 0, 0, 0, 1)
 }
 
 func TestClientUploadDownloadRawEncrypted(t *testing.T) {
@@ -69,7 +69,7 @@ func TestClientUploadDownloadRawEncrypted(t *testing.T) {
 
 	// check the tag was created successfully
 	tag := srv.Tags.All()[0]
-	chunktesting.CheckTag(t, tag, 1, 1, 0, 1)
+	chunktesting.CheckTag(t, tag, 1, 1, 0, 0, 0, 1)
 }
 
 func testClientUploadDownloadRaw(srv *swarmhttp.TestSwarmServer, toEncrypt bool, t *testing.T, data []byte, toPin bool) string {
@@ -228,7 +228,7 @@ func TestClientUploadDownloadDirectory(t *testing.T) {
 
 	// check the tag was created successfully
 	tag := srv.Tags.All()[0]
-	chunktesting.CheckTag(t, tag, 9, 9, 0, 9)
+	chunktesting.CheckTag(t, tag, 8, 8, 0, 0, 0, 8)
 
 	// check we can download the individual files
 	checkDownloadFile := func(path string, expected []byte) {
@@ -372,7 +372,7 @@ func TestClientMultipartUpload(t *testing.T) {
 
 	// check the tag was created successfully
 	tag := srv.Tags.All()[0]
-	chunktesting.CheckTag(t, tag, 9, 9, 0, 9)
+	chunktesting.CheckTag(t, tag, 8, 8, 0, 0, 0, 8)
 
 	// check we can download the individual files
 	checkDownloadFile := func(path string) {

api/http/middleware.go

@@ -1,6 +1,7 @@
 package http
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"runtime/debug"
@@ -123,7 +124,7 @@ func InitUploadTag(h http.Handler, tags *chunk.Tags) http.Handler {
 
 		log.Trace("creating tag", "tagName", tagName, "estimatedTotal", estimatedTotal)
 
-		t, err := tags.Create(tagName, estimatedTotal)
+		t, err := tags.Create(context.Background(), tagName, estimatedTotal)
 		if err != nil {
 			log.Error("error creating tag", "err", err, "tagName", tagName)
 		}

api/http/server.go

@@ -42,6 +42,7 @@ import (
 	"github.com/ethersphere/swarm/chunk"
 	"github.com/ethersphere/swarm/log"
 	"github.com/ethersphere/swarm/sctx"
+	"github.com/ethersphere/swarm/spancontext"
 	"github.com/ethersphere/swarm/storage"
 	"github.com/ethersphere/swarm/storage/feed"
 	"github.com/ethersphere/swarm/storage/pin"
@@ -276,10 +277,10 @@ func (s *Server) HandlePostRaw(w http.ResponseWriter, r *http.Request) {
 	ruid := GetRUID(r.Context())
 	log.Debug("handle.post.raw", "ruid", ruid)
 
-	tagUid := sctx.GetTag(r.Context())
-	tag, err := s.api.Tags.Get(tagUid)
+	tagUID := sctx.GetTag(r.Context())
+	tag, err := s.api.Tags.Get(tagUID)
 	if err != nil {
-		log.Error("handle post raw got an error retrieving tag for DoneSplit", "tagUid", tagUid, "err", err)
+		log.Error("handle post raw got an error retrieving tag for DoneSplit", "tagUID", tagUID, "err", err)
 	}
 
 	postRawCount.Inc(1)
@@ -334,7 +335,7 @@ func (s *Server) HandlePostRaw(w http.ResponseWriter, r *http.Request) {
 	}
 
 	w.Header().Set("Content-Type", "text/plain")
-	w.Header().Set(TagHeaderName, fmt.Sprint(tagUid))
+	w.Header().Set(TagHeaderName, fmt.Sprint(tagUID))
 	w.WriteHeader(http.StatusOK)
 	fmt.Fprint(w, addr)
 }
@@ -349,6 +350,15 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *http.Request) {
 	log.Debug("handle.post.files", "ruid", ruid)
 	postFilesCount.Inc(1)
 
+	tagUID := sctx.GetTag(r.Context())
+	tag, err := s.api.Tags.Get(tagUID)
+	if err != nil {
+		log.Error("handle post raw got an error retrieving tag", "tagUID", tagUID, "err", err)
+	}
+
+	ctx, sp := spancontext.StartSpan(tag.Context(), "http.post")
+	defer sp.Finish()
+
 	contentType, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
 	if err != nil {
 		postFilesFail.Inc(1)
@@ -375,15 +385,15 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *http.Request) {
 		}
 		log.Debug("resolved key", "ruid", ruid, "key", addr)
 	} else {
-		addr, err = s.api.NewManifest(r.Context(), toEncrypt)
+		addr, err = s.api.NewManifest(ctx, toEncrypt)
 		if err != nil {
 			postFilesFail.Inc(1)
 			respondError(w, r, err.Error(), http.StatusInternalServerError)
 			return
 		}
 		log.Debug("new manifest", "ruid", ruid, "key", addr)
 	}
-	newAddr, err := s.api.UpdateManifest(r.Context(), addr, func(mw *api.ManifestWriter) error {
+	newAddr, err := s.api.UpdateManifest(ctx, addr, func(mw *api.ManifestWriter) error {
 		switch contentType {
 		case tarContentType:
 			_, err := s.handleTarUpload(r, mw)
@@ -405,10 +415,10 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	tagUid := sctx.GetTag(r.Context())
-	tag, err := s.api.Tags.Get(tagUid)
+	tagUID = sctx.GetTag(r.Context())
+	tag, err = s.api.Tags.Get(tagUID)
 	if err != nil {
-		log.Error("got an error retrieving tag for DoneSplit", "tagUid", tagUid, "err", err)
+		log.Error("got an error retrieving tag for DoneSplit", "tagUID", tagUID, "err", err)
 	}
 
 	log.Debug("done splitting, setting tag total", "SPLIT", tag.Get(chunk.StateSplit), "TOTAL", tag.TotalCounter())
@@ -427,7 +437,7 @@ func (s *Server) HandlePostFiles(w http.ResponseWriter, r *http.Request) {
 	log.Debug("stored content", "ruid", ruid, "key", newAddr)
 
 	w.Header().Set("Content-Type", "text/plain")
-	w.Header().Set(TagHeaderName, fmt.Sprint(tagUid))
+	w.Header().Set(TagHeaderName, fmt.Sprint(tagUID))
 	w.WriteHeader(http.StatusOK)
 	fmt.Fprint(w, newAddr)
 }

api/http/server_test.go

@@ -945,7 +945,7 @@ func testBzzTar(encrypted bool, t *testing.T) {
 
 	// check that the tag was written correctly
 	tag := srv.Tags.All()[0]
-	chunktesting.CheckTag(t, tag, 4, 4, 0, 4)
+	chunktesting.CheckTag(t, tag, 3, 3, 0, 0, 0, 3)
 
 	swarmHash, err := ioutil.ReadAll(resp2.Body)
 	resp2.Body.Close()
@@ -1081,7 +1081,7 @@ func TestBzzCorrectTagEstimate(t *testing.T) {
 				<-time.After(10 * time.Millisecond)
 			case 1:
 				tag := srv.Tags.All()[0]
-				chunktesting.CheckTag(t, tag, 0, 0, 0, v.expChunks)
+				chunktesting.CheckTag(t, tag, 0, 0, 0, 0, 0, v.expChunks)
 				srv.Tags.Delete(tag.Uid)
 				done = true
 			}

api/inspector.go

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/metrics"
+	"github.com/ethersphere/swarm/chunk"
 	"github.com/ethersphere/swarm/log"
 	"github.com/ethersphere/swarm/network"
 	stream "github.com/ethersphere/swarm/network/stream/v2"
@@ -53,6 +54,20 @@ func (i *Inspector) KademliaInfo() network.KademliaInfo {
 	return i.hive.KademliaInfo()
 }
 
+func (i *Inspector) IsPushSynced(tagname string) bool {
+	tags := i.api.Tags.All()
+
+	for _, t := range tags {
+		if t.Name == tagname {
+			ds := t.Done(chunk.StateSynced)
+			log.Trace("found tag", "tagname", tagname, "done-syncing", ds)
+			return ds
+		}
+	}
+
+	return false
+}
+
 func (i *Inspector) IsPullSyncing() bool {
 	t := i.stream.LastReceivedChunkTime()
 

api/pullsync/client.go

@@ -0,0 +1,14 @@
+package pullsync
+
+import stream "github.com/ethersphere/swarm/network/stream/v2"
+
+
+
+// the node-wide pullsync.Client
+type Client struct {
+	stream.Syncer // embed stream.Syncer
+	// when pullsync
+	// here  you simply put the update sync logic listening to kademlia depth changes
+	// and call `Request`
+	// remember the request, when no longer relevant just call request.Cancel()
+}

api/pullsync/server.go

@@ -0,0 +1,9 @@
+package pullsync
+
+// pullSync.Server implements stream.Provider
+// uses localstore SubscribePull for the  bins
+// server is node-wide
+type Server struct {
+	// ...
+	*stream.LocalProvider
+}

chunk/chunk.go

@@ -40,12 +40,15 @@ type Chunk interface {
 	Data() []byte
 	PinCounter() uint64
 	WithPinCounter(p uint64) Chunk
+	TagID() uint32
+	WithTagID(t uint32) Chunk
 }
 
 type chunk struct {
 	addr       Address
 	sdata      []byte
 	pinCounter uint64
+	tagID      uint32
 }
 
 func NewChunk(addr Address, data []byte) Chunk {
@@ -60,6 +63,11 @@ func (c *chunk) WithPinCounter(p uint64) Chunk {
 	return c
 }
 
+func (c *chunk) WithTagID(t uint32) Chunk {
+	c.tagID = t
+	return c
+}
+
 func (c *chunk) Address() Address {
 	return c.addr
 }
@@ -72,6 +80,10 @@ func (c *chunk) PinCounter() uint64 {
 	return c.pinCounter
 }
 
+func (c *chunk) TagID() uint32 {
+	return c.tagID
+}
+
 func (self *chunk) String() string {
 	return fmt.Sprintf("Address: %v Chunksize: %v", self.addr.Log(), len(self.sdata))
 }