Add API for changing pin counter (#957)

openapi/Swarm.yaml

@@ -537,6 +537,25 @@ paths:
            $ref: 'SwarmCommon.yaml#/components/responses/500'
         default:
           description: Default response
+    put:
+      summary: Update chunk pin counter
+      tags:
+        - Chunk pinning
+      responses:
+        '200':
+          description: Pinning state of chunk with address
+          content:
+            application/json:
+              schema:
+                $ref: 'SwarmCommon.yaml#/components/schemas/PinningState'
+        '400':
+          $ref: 'SwarmCommon.yaml#/components/responses/400'
+        '403':
+          $ref: 'SwarmCommon.yaml#/components/responses/403'
+        '404':
+          $ref: 'SwarmCommon.yaml#/components/responses/404'
+        default:
+          description: Default response
 
   '/pin/chunks':
     get:

pkg/api/export_test.go

@@ -15,6 +15,7 @@ type (
 	TagRequest               = tagRequest
 	PinnedChunk              = pinnedChunk
 	ListPinnedChunksResponse = listPinnedChunksResponse
+	UpdatePinCounter         = updatePinCounter
 )
 
 var (

pkg/api/pin_chunks.go

@@ -6,7 +6,10 @@ package api
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"io/ioutil"
+	"math"
 	"net/http"
 	"strconv"
 
@@ -184,6 +187,129 @@ func (s *server) getPinnedChunk(w http.ResponseWriter, r *http.Request) {
 	})
 }
 
+type updatePinCounter struct {
+	PinCounter uint64 `json:"pinCounter"`
+}
+
+// updatePinnedChunkPinCounter allows changing the pin counter for the chunk.
+func (s *server) updatePinnedChunkPinCounter(w http.ResponseWriter, r *http.Request) {
+	addr, err := swarm.ParseHexAddress(mux.Vars(r)["address"])
+	if err != nil {
+		s.Logger.Debugf("update pin counter: parse chunk ddress: %v", err)
+		s.Logger.Errorf("update pin counter: parse address")
+		jsonhttp.BadRequest(w, "bad address")
+		return
+	}
+
+	has, err := s.Storer.Has(r.Context(), addr)
+	if err != nil {
+		s.Logger.Debugf("update pin counter: localstore has: %v", err)
+		s.Logger.Errorf("update pin counter: store")
+		jsonhttp.InternalServerError(w, err)
+		return
+	}
+
+	if !has {
+		jsonhttp.NotFound(w, nil)
+		return
+	}
+
+	pinCounter, err := s.Storer.PinCounter(addr)
+	if err != nil {
+		if errors.Is(err, storage.ErrNotFound) {
+			jsonhttp.NotFound(w, nil)
+			return
+		}
+		s.Logger.Debugf("pin counter: get pin counter: %v", err)
+		s.Logger.Errorf("pin counter: get pin counter")
+		jsonhttp.InternalServerError(w, err)
+		return
+	}
+
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		if jsonhttp.HandleBodyReadError(err, w) {
+			return
+		}
+		s.Logger.Debugf("update pin counter: read request body error: %v", err)
+		s.Logger.Error("update pin counter: read request body error")
+		jsonhttp.InternalServerError(w, "cannot read request")
+		return
+	}
+
+	newPinCount := updatePinCounter{}
+	if len(body) > 0 {
+		err = json.Unmarshal(body, &newPinCount)
+		if err != nil {
+			s.Logger.Debugf("update pin counter: unmarshal pin counter error: %v", err)
+			s.Logger.Errorf("update pin counter: unmarshal pin counter error")
+			jsonhttp.InternalServerError(w, "error unmarshaling pin counter")
+			return
+		}
+	}
+
+	if newPinCount.PinCounter > math.MaxInt32 {
+		s.Logger.Errorf("update pin counter: invalid pin counter %d", newPinCount.PinCounter)
+		jsonhttp.BadRequest(w, "invalid pin counter")
+		return
+	}
+
+	diff := newPinCount.PinCounter - pinCounter
+
+	err = s.updatePinCount(r.Context(), addr, int(diff))
+	if err != nil {
+		s.Logger.Debugf("update pin counter: update error: %v, addr %s", err, addr)
+		s.Logger.Error("update pin counter: update")
+		jsonhttp.InternalServerError(w, err)
+		return
+	}
+
+	pinCounter, err = s.Storer.PinCounter(addr)
+	if err != nil {
+		if errors.Is(err, storage.ErrNotFound) {
+			pinCounter = 0
+		} else {
+			s.Logger.Debugf("update pin counter: get pin counter: %v", err)
+			s.Logger.Errorf("update pin counter: get pin counter")
+			jsonhttp.InternalServerError(w, err)
+			return
+		}
+	}
+
+	jsonhttp.OK(w, pinnedChunk{
+		Address:    addr,
+		PinCounter: pinCounter,
+	})
+}
+
+// updatePinCount changes pin counter for a chunk address.
+// This is done with a loop, depending on the delta value supplied.
+// NOTE: If the value is too large, it will result in many database operations.
+func (s *server) updatePinCount(ctx context.Context, reference swarm.Address, delta int) error {
+	diff := delta
+	mode := storage.ModeSetPin
+
+	if diff < 0 {
+		diff = -diff
+		mode = storage.ModeSetUnpin
+	}
+
+	for i := 0; i < diff; i++ {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+
+		err := s.Storer.Set(ctx, mode, reference)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 func (s *server) pinChunkAddressFn(ctx context.Context, reference swarm.Address) func(address swarm.Address) (stop bool) {
 	return func(address swarm.Address) (stop bool) {
 		err := s.Storer.Set(ctx, storage.ModeSetPin, address)

pkg/api/pin_chunks_test.go

@@ -225,4 +225,47 @@ func TestPinChunkHandler(t *testing.T) {
 			}),
 		)
 	})
+
+	t.Run("update-pin-counter-up", func(t *testing.T) {
+		updatePinCounter := api.UpdatePinCounter{
+			PinCounter: 7,
+		}
+
+		jsonhttptest.Request(t, client, http.MethodPut, "/pin/chunks/"+hash.String(), http.StatusOK,
+			jsonhttptest.WithJSONRequestBody(updatePinCounter),
+			jsonhttptest.WithExpectedJSONResponse(api.PinnedChunk{
+				Address:    swarm.MustParseHexAddress("aabbcc"),
+				PinCounter: updatePinCounter.PinCounter,
+			}),
+		)
+
+		jsonhttptest.Request(t, client, http.MethodGet, "/pin/chunks/"+hash.String(), http.StatusOK,
+			jsonhttptest.WithExpectedJSONResponse(api.PinnedChunk{
+				Address:    swarm.MustParseHexAddress("aabbcc"),
+				PinCounter: updatePinCounter.PinCounter,
+			}),
+		)
+	})
+
+	t.Run("update-pin-counter-to-zero", func(t *testing.T) {
+		updatePinCounter := api.UpdatePinCounter{
+			PinCounter: 0,
+		}
+
+		jsonhttptest.Request(t, client, http.MethodPut, "/pin/chunks/"+hash.String(), http.StatusOK,
+			jsonhttptest.WithJSONRequestBody(updatePinCounter),
+			jsonhttptest.WithExpectedJSONResponse(api.PinnedChunk{
+				Address:    swarm.MustParseHexAddress("aabbcc"),
+				PinCounter: updatePinCounter.PinCounter,
+			}),
+		)
+
+		jsonhttptest.Request(t, client, http.MethodGet, "/pin/chunks/"+hash.String(), http.StatusNotFound,
+			jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{
+				Message: http.StatusText(http.StatusNotFound),
+				Code:    http.StatusNotFound,
+			}),
+		)
+	})
+
 }

pkg/api/router.go

@@ -126,6 +126,10 @@ func (s *server) setupRouting() {
 			"GET":    http.HandlerFunc(s.getPinnedChunk),
 			"POST":   http.HandlerFunc(s.pinChunk),
 			"DELETE": http.HandlerFunc(s.unpinChunk),
+			"PUT": web.ChainHandlers(
+				jsonhttp.NewMaxBodyBytesHandler(1024),
+				web.FinalHandlerFunc(s.updatePinnedChunkPinCounter),
+			),
 		})),
 	)
 	handle(router, "/pin/chunks", web.ChainHandlers(