api, debugapi, internal, sctx, tags: tags cleanup (#557)

api, api_test, debugapi, debugapi_test, internal, sctx, tags: restore tags API from the debugapi package and iterate on it.

pkg/api/api.go

@@ -5,8 +5,11 @@
 package api
 
 import (
+	"fmt"
 	"net/http"
+	"strconv"
 	"strings"
+	"time"
 
 	"github.com/ethersphere/bee/pkg/logging"
 	m "github.com/ethersphere/bee/pkg/metrics"
@@ -15,6 +18,11 @@ import (
 	"github.com/ethersphere/bee/pkg/tracing"
 )
 
+const (
+	SwarmPinHeader    = "Swarm-Pin"
+	SwarmTagUidHeader = "Swarm-Tag-Uid"
+)
+
 type Service interface {
 	http.Handler
 	m.Collector
@@ -50,10 +58,26 @@ func New(tags *tags.Tags, storer storage.Storer, corsAllowedOrigins []string, lo
 	return s
 }
 
-const (
-	SwarmPinHeader = "Swarm-Pin"
-	TagHeaderUid   = "swarm-tag-uid"
-)
+// getOrCreateTag attempts to get the tag if an id is supplied, and returns an error if it does not exist.
+// If no id is supplied, it will attempt to create a new tag with a generated name and return it.
+func (s *server) getOrCreateTag(tagUid string) (*tags.Tag, bool, error) {
+	// if tag ID is not supplied, create a new tag
+	if tagUid == "" {
+		tagName := fmt.Sprintf("unnamed_tag_%d", time.Now().Unix())
+		var err error
+		tag, err := s.Tags.Create(tagName, 0, false)
+		if err != nil {
+			return nil, false, fmt.Errorf("cannot create tag: %w", err)
+		}
+		return tag, true, nil
+	}
+	uid, err := strconv.Atoi(tagUid)
+	if err != nil {
+		return nil, false, fmt.Errorf("cannot parse taguid: %w", err)
+	}
+	t, err := s.Tags.Get(uint32(uid))
+	return t, false, err
+}
 
 // requestModePut returns the desired storage.ModePut for this request based on the request headers.
 func requestModePut(r *http.Request) storage.ModePut {

pkg/api/bytes.go

@@ -5,18 +5,15 @@
 package api
 
 import (
-	"context"
 	"fmt"
 	"net/http"
-	"strconv"
 	"strings"
-	"time"
 
 	"github.com/ethersphere/bee/pkg/file"
 	"github.com/ethersphere/bee/pkg/file/splitter"
 	"github.com/ethersphere/bee/pkg/jsonhttp"
+	"github.com/ethersphere/bee/pkg/sctx"
 	"github.com/ethersphere/bee/pkg/swarm"
-	"github.com/ethersphere/bee/pkg/tags"
 	"github.com/gorilla/mux"
 )
 
@@ -26,28 +23,31 @@ type bytesPostResponse struct {
 
 // bytesUploadHandler handles upload of raw binary data of arbitrary length.
 func (s *server) bytesUploadHandler(w http.ResponseWriter, r *http.Request) {
-	ta := s.createTag(w, r)
-	if ta == nil {
+	tag, created, err := s.getOrCreateTag(r.Header.Get(SwarmTagUidHeader))
+	if err != nil {
+		s.Logger.Debugf("bytes upload: get or create tag: %v", err)
+		s.Logger.Error("bytes upload: get or create tag")
+		jsonhttp.InternalServerError(w, "cannot get or create tag")
 		return
 	}
 
 	// Add the tag to the context
-	r = r.WithContext(context.WithValue(r.Context(), tags.TagsContextKey{}, ta))
-	ctx := r.Context()
+	ctx := sctx.SetTag(r.Context(), tag)
 
 	toEncrypt := strings.ToLower(r.Header.Get(EncryptHeader)) == "true"
 	sp := splitter.NewSimpleSplitter(s.Storer, requestModePut(r))
 	address, err := file.SplitWriteAll(ctx, sp, r.Body, r.ContentLength, toEncrypt)
 	if err != nil {
-		s.Logger.Debugf("bytes upload: %v", err)
+		s.Logger.Debugf("bytes upload: split write all: %v", err)
+		s.Logger.Error("bytes upload: split write all")
 		jsonhttp.InternalServerError(w, nil)
 		return
 	}
-
-	ta.DoneSplit(address)
-
-	w.Header().Set(TagHeaderUid, fmt.Sprint(ta.Uid))
-	w.Header().Set("Access-Control-Expose-Headers", TagHeaderUid)
+	if created {
+		tag.DoneSplit(address)
+	}
+	w.Header().Set(SwarmTagUidHeader, fmt.Sprint(tag.Uid))
+	w.Header().Set("Access-Control-Expose-Headers", SwarmTagUidHeader)
 	jsonhttp.OK(w, bytesPostResponse{
 		Reference: address,
 	})
@@ -71,39 +71,3 @@ func (s *server) bytesGetHandler(w http.ResponseWriter, r *http.Request) {
 
 	s.downloadHandler(w, r, address, additionalHeaders)
 }
-
-func (s *server) createTag(w http.ResponseWriter, r *http.Request) *tags.Tag {
-	// if tag header is not there create a new one
-	var tag *tags.Tag
-	tagUidStr := r.Header.Get(TagHeaderUid)
-	if tagUidStr == "" {
-		tagName := fmt.Sprintf("unnamed_tag_%d", time.Now().Unix())
-		var err error
-		tag, err = s.Tags.Create(tagName, 0, false)
-		if err != nil {
-			s.Logger.Debugf("bytes upload: tag creation error: %v", err)
-			s.Logger.Error("bytes upload: tag creation")
-			jsonhttp.InternalServerError(w, "cannot create tag")
-			return nil
-		}
-	} else {
-		// if the tag uid header is present, then use the tag sent
-		tagUid, err := strconv.ParseUint(tagUidStr, 10, 32)
-		if err != nil {
-			s.Logger.Debugf("bytes upload: parse taguid %s: %v", tagUidStr, err)
-			s.Logger.Error("bytes upload: parse taguid")
-			jsonhttp.BadRequest(w, "invalid taguid")
-			return nil
-		}
-
-		tag, err = s.Tags.Get(uint32(tagUid))
-		if err != nil {
-			s.Logger.Debugf("bytes upload: get tag error: %v", err)
-			s.Logger.Error("bytes upload: get tag")
-			jsonhttp.InternalServerError(w, "cannot create tag")
-			return nil
-		}
-	}
-
-	return tag
-}

pkg/api/chunk.go

@@ -6,14 +6,14 @@ package api
 
 import (
 	"bytes"
-	"context"
 	"errors"
 	"fmt"
-	"github.com/ethersphere/bee/pkg/netstore"
 	"io"
 	"io/ioutil"
 	"net/http"
 
+	"github.com/ethersphere/bee/pkg/netstore"
+
 	"github.com/ethersphere/bee/pkg/jsonhttp"
 	"github.com/ethersphere/bee/pkg/sctx"
 	"github.com/ethersphere/bee/pkg/storage"
@@ -32,14 +32,16 @@ func (s *server) chunkUploadHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	tag := s.createTag(w, r)
-	if tag == nil {
+	tag, _, err := s.getOrCreateTag(r.Header.Get(SwarmTagUidHeader))
+	if err != nil {
+		s.Logger.Debugf("chunk upload: get or create tag: %v", err)
+		s.Logger.Error("chunk upload: get or create tag")
+		jsonhttp.InternalServerError(w, "cannot get or create tag")
 		return
 	}
 
 	// Add the tag to the context
-	r = r.WithContext(context.WithValue(r.Context(), tags.TagsContextKey{}, tag))
-	ctx := r.Context()
+	ctx := sctx.SetTag(r.Context(), tag)
 
 	// Increment the StateSplit here since we dont have a splitter for the file upload
 	tag.Inc(tags.StateSplit)
@@ -68,10 +70,8 @@ func (s *server) chunkUploadHandler(w http.ResponseWriter, r *http.Request) {
 	// Indicate that the chunk is stored
 	tag.Inc(tags.StateStored)
 
-	tag.DoneSplit(address)
-
-	w.Header().Set(TagHeaderUid, fmt.Sprint(tag.Uid))
-	w.Header().Set("Access-Control-Expose-Headers", TagHeaderUid)
+	w.Header().Set(SwarmTagUidHeader, fmt.Sprint(tag.Uid))
+	w.Header().Set("Access-Control-Expose-Headers", SwarmTagUidHeader)
 	jsonhttp.OK(w, nil)
 }
 

pkg/api/dirs.go

@@ -23,6 +23,7 @@ import (
 	"github.com/ethersphere/bee/pkg/jsonhttp"
 	"github.com/ethersphere/bee/pkg/logging"
 	"github.com/ethersphere/bee/pkg/manifest/jsonmanifest"
+	"github.com/ethersphere/bee/pkg/sctx"
 	"github.com/ethersphere/bee/pkg/storage"
 	"github.com/ethersphere/bee/pkg/swarm"
 )
@@ -44,14 +45,28 @@ func (s *server) dirUploadHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	tag, created, err := s.getOrCreateTag(r.Header.Get(SwarmTagUidHeader))
+	if err != nil {
+		s.Logger.Debugf("dir upload: get or create tag: %v", err)
+		s.Logger.Error("dir upload: get or create tag")
+		jsonhttp.InternalServerError(w, "cannot get or create tag")
+		return
+	}
+
+	// Add the tag to the context
+	ctx = sctx.SetTag(ctx, tag)
+
 	reference, err := storeDir(ctx, r.Body, s.Storer, requestModePut(r), s.Logger)
 	if err != nil {
-		s.Logger.Errorf("dir upload, store dir")
 		s.Logger.Debugf("dir upload, store dir err: %v", err)
+		s.Logger.Errorf("dir upload, store dir")
 		jsonhttp.InternalServerError(w, "could not store dir")
 		return
 	}
-
+	if created {
+		tag.DoneSplit(reference)
+	}
+	w.Header().Set(SwarmTagUidHeader, fmt.Sprint(tag.Uid))
 	jsonhttp.OK(w, fileUploadResponse{
 		Reference: reference,
 	})

pkg/api/export_test.go

@@ -7,6 +7,8 @@ package api
 type (
 	BytesPostResponse  = bytesPostResponse
 	FileUploadResponse = fileUploadResponse
+	TagResponse        = tagResponse
+	TagRequest         = tagRequest
 )
 
 var (

pkg/api/file.go

@@ -7,7 +7,6 @@ package api
 import (
 	"bufio"
 	"bytes"
-	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -31,7 +30,6 @@ import (
 	"github.com/ethersphere/bee/pkg/sctx"
 	"github.com/ethersphere/bee/pkg/storage"
 	"github.com/ethersphere/bee/pkg/swarm"
-	"github.com/ethersphere/bee/pkg/tags"
 	"github.com/gorilla/mux"
 )
 
@@ -66,14 +64,16 @@ func (s *server) fileUploadHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	ta := s.createTag(w, r)
-	if ta == nil {
+	tag, created, err := s.getOrCreateTag(r.Header.Get(SwarmTagUidHeader))
+	if err != nil {
+		s.Logger.Debugf("file upload: get or create tag: %v", err)
+		s.Logger.Error("file upload: get or create tag")
+		jsonhttp.InternalServerError(w, "cannot get or create tag")
 		return
 	}
 
 	// Add the tag to the context
-	r = r.WithContext(context.WithValue(r.Context(), tags.TagsContextKey{}, ta))
-	ctx := r.Context()
+	ctx := sctx.SetTag(r.Context(), tag)
 
 	if mediaType == multiPartFormData {
 		mr := multipart.NewReader(r.Body, params["boundary"])
@@ -205,12 +205,12 @@ func (s *server) fileUploadHandler(w http.ResponseWriter, r *http.Request) {
 		jsonhttp.InternalServerError(w, "could not store entry")
 		return
 	}
-
-	ta.DoneSplit(reference)
-
+	if created {
+		tag.DoneSplit(reference)
+	}
 	w.Header().Set("ETag", fmt.Sprintf("%q", reference.String()))
-	w.Header().Set(TagHeaderUid, fmt.Sprint(ta.Uid))
-	w.Header().Set("Access-Control-Expose-Headers", TagHeaderUid)
+	w.Header().Set(SwarmTagUidHeader, fmt.Sprint(tag.Uid))
+	w.Header().Set("Access-Control-Expose-Headers", SwarmTagUidHeader)
 	jsonhttp.OK(w, fileUploadResponse{
 		Reference: reference,
 	})

pkg/api/router.go

@@ -66,6 +66,22 @@ func (s *server) setupRouting() {
 		"GET": http.HandlerFunc(s.bzzDownloadHandler),
 	})
 
+	router.Handle("/tags", jsonhttp.MethodHandler{
+		"POST": web.ChainHandlers(
+			jsonhttp.NewMaxBodyBytesHandler(1024),
+			web.FinalHandlerFunc(s.createTag),
+		),
+	})
+
+	router.Handle("/tags/{id}", jsonhttp.MethodHandler{
+		"GET":    http.HandlerFunc(s.getTag),
+		"DELETE": http.HandlerFunc(s.deleteTag),
+		"PATCH": web.ChainHandlers(
+			jsonhttp.NewMaxBodyBytesHandler(1024),
+			web.FinalHandlerFunc(s.doneSplit),
+		),
+	})
+
 	s.Handler = web.ChainHandlers(
 		logging.NewHTTPAccessLogHandler(s.Logger, logrus.InfoLevel, "api access"),
 		handlers.CompressHandler,

pkg/debugapi/tag.go → pkg/api/tag.go

@@ -2,12 +2,13 @@
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
 
-package debugapi
+package api
 
 import (
-	crand "crypto/rand"
+	"encoding/json"
 	"errors"
 	"fmt"
+	"io/ioutil"
 	"net/http"
 	"strconv"
 	"time"
@@ -19,6 +20,11 @@ import (
 	"github.com/gorilla/mux"
 )
 
+type tagRequest struct {
+	Name    string        `json:"name,omitempty"`
+	Address swarm.Address `json:"address,omitempty"`
+}
+
 type tagResponse struct {
 	Total     int64         `json:"total"`
 	Split     int64         `json:"split"`
@@ -50,56 +56,148 @@ func newTagResponse(tag *tags.Tag) tagResponse {
 }
 
 func (s *server) createTag(w http.ResponseWriter, r *http.Request) {
-	name := r.URL.Query().Get("name")
-	if name == "" {
-		b := make([]byte, 4)
-		_, err := crand.Read(b)
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		if jsonhttp.HandleBodyReadError(err, w) {
+			return
+		}
+		s.Logger.Debugf("create tag: read request body error: %v", err)
+		s.Logger.Error("create tag: read request body error")
+		jsonhttp.InternalServerError(w, "cannot read request")
+		return
+	}
+
+	tagr := tagRequest{}
+	if len(body) > 0 {
+		err = json.Unmarshal(body, &tagr)
 		if err != nil {
-			s.Logger.Debugf("create tag: read random bytes %v", err)
-			s.Logger.Errorf("create tag: read random bytes error")
-			jsonhttp.InternalServerError(w, nil)
+			s.Logger.Debugf("create tag: unmarshal tag name error: %v", err)
+			s.Logger.Errorf("create tag: unmarshal tag name error")
+			jsonhttp.InternalServerError(w, "error unmarshaling metadata")
 			return
 		}
-		name = fmt.Sprintf("tag-%v-%x", time.Now().UnixNano(), b)
 	}
 
-	tag, err := s.Tags.Create(name, 0, false)
+	if tagr.Name == "" {
+		tagr.Name = fmt.Sprintf("unnamed_tag_%d", time.Now().Unix())
+	}
+
+	tag, err := s.Tags.Create(tagr.Name, 0, false)
 	if err != nil {
-		s.Logger.Debugf("create tag: %s %v", name, err)
-		s.Logger.Errorf("create tag: %s error", name)
+		s.Logger.Debugf("create tag: tag create error: %v", err)
+		s.Logger.Error("create tag: tag create error")
 		jsonhttp.InternalServerError(w, "cannot create tag")
 		return
 	}
 	w.Header().Set("Cache-Control", "no-cache, private, max-age=0")
-	jsonhttp.OK(w, newTagResponse(tag))
-
+	jsonhttp.Created(w, newTagResponse(tag))
 }
 
 func (s *server) getTag(w http.ResponseWriter, r *http.Request) {
-	uidStr := mux.Vars(r)["uid"]
+	idStr := mux.Vars(r)["id"]
 
-	uid, err := strconv.ParseUint(uidStr, 10, 32)
+	id, err := strconv.Atoi(idStr)
 	if err != nil {
-		s.Logger.Debugf("get tag: parse uid  %s: %v", uidStr, err)
-		s.Logger.Error("get tag: parse uid")
-		jsonhttp.BadRequest(w, "invalid uid")
+		s.Logger.Debugf("get tag: parse id  %s: %v", idStr, err)
+		s.Logger.Error("get tag: parse id")
+		jsonhttp.BadRequest(w, "invalid id")
 		return
 	}
 
-	tag, err := s.Tags.Get(uint32(uid))
+	tag, err := s.Tags.Get(uint32(id))
 	if err != nil {
 		if errors.Is(err, tags.ErrNotFound) {
-			s.Logger.Debugf("get tag: tag %v not present: %v", uid, err)
-			s.Logger.Warningf("get tag: tag %v not present", uid)
+			s.Logger.Debugf("get tag: tag not present: %v, id %s", err, idStr)
+			s.Logger.Error("get tag: tag not present")
 			jsonhttp.NotFound(w, "tag not present")
 			return
 		}
-		s.Logger.Debugf("get tag: tag %v: %v", uid, err)
-		s.Logger.Errorf("get tag: %v", uid)
-		jsonhttp.InternalServerError(w, nil)
+		s.Logger.Debugf("get tag: tag %v: %v", idStr, err)
+		s.Logger.Errorf("get tag: %v", idStr)
+		jsonhttp.InternalServerError(w, "cannot get tag")
 		return
 	}
 
 	w.Header().Set("Cache-Control", "no-cache, private, max-age=0")
 	jsonhttp.OK(w, newTagResponse(tag))
 }
+
+func (s *server) deleteTag(w http.ResponseWriter, r *http.Request) {
+	idStr := mux.Vars(r)["id"]
+
+	id, err := strconv.Atoi(idStr)
+	if err != nil {
+		s.Logger.Debugf("delete tag: parse id  %s: %v", idStr, err)
+		s.Logger.Error("delete tag: parse id")
+		jsonhttp.BadRequest(w, "invalid id")
+		return
+	}
+
+	tag, err := s.Tags.Get(uint32(id))
+	if err != nil {
+		if errors.Is(err, tags.ErrNotFound) {
+			s.Logger.Debugf("delete tag: tag not present: %v, id %s", err, idStr)
+			s.Logger.Error("delete tag: tag not present")
+			jsonhttp.NotFound(w, "tag not present")
+			return
+		}
+		s.Logger.Debugf("delete tag: tag %v: %v", idStr, err)
+		s.Logger.Errorf("delete tag: %v", idStr)
+		jsonhttp.InternalServerError(w, "cannot get tag")
+		return
+	}
+
+	s.Tags.Delete(tag.Uid)
+	jsonhttp.NoContent(w)
+}
+
+func (s *server) doneSplit(w http.ResponseWriter, r *http.Request) {
+	idStr := mux.Vars(r)["id"]
+
+	id, err := strconv.Atoi(idStr)
+	if err != nil {
+		s.Logger.Debugf("done split tag: parse id  %s: %v", idStr, err)
+		s.Logger.Error("done split tag: parse id")
+		jsonhttp.BadRequest(w, "invalid id")
+		return
+	}
+
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		if jsonhttp.HandleBodyReadError(err, w) {
+			return
+		}
+		s.Logger.Debugf("done split tag: read request body error: %v", err)
+		s.Logger.Error("done split tag: read request body error")
+		jsonhttp.InternalServerError(w, "cannot read request")
+		return
+	}
+
+	tagr := tagRequest{}
+	if len(body) > 0 {
+		err = json.Unmarshal(body, &tagr)
+		if err != nil {
+			s.Logger.Debugf("done split tag: unmarshal tag name error: %v", err)
+			s.Logger.Errorf("done split tag: unmarshal tag name error")
+			jsonhttp.InternalServerError(w, "error unmarshaling metadata")
+			return
+		}
+	}
+
+	tag, err := s.Tags.Get(uint32(id))
+	if err != nil {
+		if errors.Is(err, tags.ErrNotFound) {
+			s.Logger.Debugf("done split: tag not present: %v, id %s", err, idStr)
+			s.Logger.Error("done split: tag not present")
+			jsonhttp.NotFound(w, "tag not present")
+			return
+		}
+		s.Logger.Debugf("done split: tag %v: %v", idStr, err)
+		s.Logger.Errorf("done split: %v", idStr)
+		jsonhttp.InternalServerError(w, "cannot get tag")
+		return
+	}
+
+	tag.DoneSplit(tagr.Address)
+	jsonhttp.OK(w, "ok")
+}

pkg/debugapi/export_test.go

@@ -12,7 +12,6 @@ type (
 	AddressesResponse        = addressesResponse
 	PinnedChunk              = pinnedChunk
 	ListPinnedChunksResponse = listPinnedChunksResponse
-	TagResponse              = tagResponse
 	WelcomeMessageRequest    = welcomeMessageRequest
 	WelcomeMessageResponse   = welcomeMessageResponse
 	BalancesResponse         = balancesResponse

pkg/debugapi/router.go

@@ -78,12 +78,6 @@ func (s *server) setupRouting() {
 	router.Handle("/chunks-pin", jsonhttp.MethodHandler{
 		"GET": http.HandlerFunc(s.listPinnedChunks),
 	})
-	router.Handle("/tags", jsonhttp.MethodHandler{
-		"POST": http.HandlerFunc(s.createTag),
-	})
-	router.Handle("/tags/{uid}", jsonhttp.MethodHandler{
-		"GET": http.HandlerFunc(s.getTag),
-	})
 	router.Handle("/topology", jsonhttp.MethodHandler{
 		"GET": http.HandlerFunc(s.topologyHandler),
 	})

pkg/file/splitter/internal/job.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/ethersphere/bee/pkg/encryption"
 	"github.com/ethersphere/bee/pkg/file"
+	"github.com/ethersphere/bee/pkg/sctx"
 	"github.com/ethersphere/bee/pkg/swarm"
 	"github.com/ethersphere/bee/pkg/tags"
 	"github.com/ethersphere/bmt"
@@ -51,7 +52,7 @@ type SimpleSplitterJob struct {
 	cursors    []int    // section write position, indexed per level
 	hasher     bmt.Hash // underlying hasher used for hashing the tree
 	buffer     []byte   // keeps data and hashes, indexed by cursors
-	tagg       *tags.Tag
+	tag        *tags.Tag
 	toEncrypt  bool // to encryrpt the chunks or not
 	refSize    int64
 }
@@ -67,10 +68,6 @@ func NewSimpleSplitterJob(ctx context.Context, putter Putter, spanLength int64,
 	}
 	p := bmtlegacy.NewTreePool(hashFunc, swarm.Branches, bmtlegacy.PoolSize)
 
-	ta, ok := ctx.Value(tags.TagsContextKey{}).(*tags.Tag)
-	if !ok {
-		ta = nil
-	}
 	return &SimpleSplitterJob{
 		ctx:        ctx,
 		putter:     putter,
@@ -79,7 +76,7 @@ func NewSimpleSplitterJob(ctx context.Context, putter Putter, spanLength int64,
 		cursors:    make([]int, levelBufferLimit),
 		hasher:     bmtlegacy.New(p),
 		buffer:     make([]byte, swarm.ChunkWithSpanSize*levelBufferLimit*2), // double size as temp workaround for weak calculation of needed buffer space
-		tagg:       ta,
+		tag:        sctx.GetTag(ctx),
 		toEncrypt:  toEncrypt,
 		refSize:    refSize,
 	}
@@ -181,8 +178,8 @@ func (s *SimpleSplitterJob) sumLevel(lvl int) ([]byte, error) {
 
 	// Add tag to the chunk if tag is valid
 	var ch swarm.Chunk
-	if s.tagg != nil {
-		ch = swarm.NewChunk(addr, c).WithTagID(s.tagg.Uid)
+	if s.tag != nil {
+		ch = swarm.NewChunk(addr, c).WithTagID(s.tag.Uid)
 	} else {
 		ch = swarm.NewChunk(addr, c)
 	}
@@ -314,7 +311,7 @@ func (s *SimpleSplitterJob) newDataEncryption(key encryption.Key) *encryption.En
 }
 
 func (s *SimpleSplitterJob) incrTag(state tags.State) {
-	if s.tagg != nil {
-		s.tagg.Inc(state)
+	if s.tag != nil {
+		s.tag.Inc(state)
 	}
 }

pkg/jsonhttp/jsonhttp.go

@@ -108,6 +108,13 @@ func NonAuthoritativeInfo(w http.ResponseWriter, response interface{}) {
 	Respond(w, http.StatusNonAuthoritativeInfo, response)
 }
 
+// NoContent writes a response with status code 204. It does not
+// accept a response value since the HTTP server will not write it
+// to the client when returning a NoContent response.
+func NoContent(w http.ResponseWriter) {
+	Respond(w, http.StatusNoContent, nil)
+}
+
 // ResetContent writes a response with status code 205.
 func ResetContent(w http.ResponseWriter, response interface{}) {
 	Respond(w, http.StatusResetContent, response)

pkg/jsonhttp/jsonhttptest/jsonhttptest.go

@@ -28,7 +28,9 @@ func ResponseDirect(t *testing.T, client *http.Client, method, url string, body
 		t.Fatal(err)
 	}
 	got = bytes.TrimSpace(got)
-
+	if response == nil && len(got) == 0 {
+		return
+	}
 	want, err := json.Marshal(response)
 	if err != nil {
 		t.Error(err)

pkg/pushsync/pushsync.go

@@ -209,6 +209,12 @@ func (ps *PushSync) PushChunkToClosest(ctx context.Context, ch swarm.Chunk) (*Re
 	peer, err := ps.peerSuggester.ClosestPeer(ch.Address())
 	if err != nil {
 		if errors.Is(err, topology.ErrWantSelf) {
+			// this is to make sure that the sent number does not diverge from the synced counter
+			t, err := ps.tagg.Get(ch.TagID())
+			if err == nil && t != nil {
+				t.Inc(tags.StateSent)
+			}
+
 			// if you are the closest node return a receipt immediately
 			return &Receipt{
 				Address: ch.Address(),
@@ -228,6 +234,7 @@ func (ps *PushSync) PushChunkToClosest(ctx context.Context, ch swarm.Chunk) (*Re
 		_ = streamer.Reset()
 		return nil, fmt.Errorf("chunk deliver to peer %s: %w", peer.String(), err)
 	}
+
 	//  if you manage to get a tag, just increment the respective counter
 	t, err := ps.tagg.Get(ch.TagID())
 	if err == nil && t != nil {

pkg/sctx/sctx.go

@@ -6,10 +6,13 @@ package sctx
 
 import (
 	"context"
+
 	"encoding/hex"
 	"errors"
 	"strings"
 
+	"github.com/ethersphere/bee/pkg/tags"
+
 	"github.com/ethersphere/bee/pkg/trojan"
 )
 
@@ -39,18 +42,18 @@ func GetHost(ctx context.Context) string {
 	return ""
 }
 
-// SetTag sets the tag unique identifier in the context
-func SetTag(ctx context.Context, tagId uint32) context.Context {
+// SetTag sets the tag instance in the context
+func SetTag(ctx context.Context, tagId *tags.Tag) context.Context {
 	return context.WithValue(ctx, tagKey{}, tagId)
 }
 
-// GetTag gets the tag unique identifier from the context
-func GetTag(ctx context.Context) uint32 {
-	v, ok := ctx.Value(tagKey{}).(uint32)
-	if ok {
-		return v
+// GetTag gets the tag instance from the context
+func GetTag(ctx context.Context) *tags.Tag {
+	v, ok := ctx.Value(tagKey{}).(*tags.Tag)
+	if !ok {
+		return nil
 	}
-	return 0
+	return v
 }
 
 // SetTargets set the target string in the context to be used downstream in netstore

pkg/tags/tag.go

@@ -177,7 +177,11 @@ func (t *Tag) Done(s State) bool {
 func (t *Tag) DoneSplit(address swarm.Address) int64 {
 	total := atomic.LoadInt64(&t.Split)
 	atomic.StoreInt64(&t.Total, total)
-	t.Address = address
+
+	if !address.Equal(swarm.ZeroAddress) {
+		t.Address = address
+	}
+
 	return total
 }
 

pkg/tags/tags.go

@@ -26,7 +26,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/ethersphere/bee/pkg/sctx"
 	"github.com/ethersphere/bee/pkg/swarm"
 )
 
@@ -35,8 +34,6 @@ var (
 	ErrNotFound = errors.New("tag not found")
 )
 
-type TagsContextKey struct{}
-
 // Tags hold tag information indexed by a unique random uint32
 type Tags struct {
 	tags *sync.Map
@@ -101,16 +98,6 @@ func (ts *Tags) GetByAddress(address swarm.Address) (*Tag, error) {
 	return t, nil
 }
 
-// GetFromContext gets a tag from the tag uid stored in the context
-func (ts *Tags) GetFromContext(ctx context.Context) (*Tag, error) {
-	uid := sctx.GetTag(ctx)
-	t, ok := ts.tags.Load(uid)
-	if !ok {
-		return nil, ErrNotFound
-	}
-	return t.(*Tag), nil
-}
-
 // Range exposes sync.Map's iterator
 func (ts *Tags) Range(fn func(k, v interface{}) bool) {
 	ts.tags.Range(fn)