Maptracker: extract optracker and make improvements

License: MIT
Signed-off-by: Hector Sanjuan <code@hector.link>

allocate.go

@@ -119,8 +119,10 @@ func allocationError(hash *cid.Cid, needed, wanted int, candidatesValid []peer.I
 func (c *Cluster) obtainAllocations(
 	hash *cid.Cid,
 	rplMin, rplMax int,
-	currentValidMetrics, candidatesMetrics map[peer.ID]api.Metric,
-	priorityMetrics map[peer.ID]api.Metric) ([]peer.ID, error) {
+	currentValidMetrics map[peer.ID]api.Metric,
+	candidatesMetrics map[peer.ID]api.Metric,
+	priorityMetrics map[peer.ID]api.Metric,
+) ([]peer.ID, error) {
 
 	// The list of peers in current
 	validAllocations := make([]peer.ID, 0, len(currentValidMetrics))

api/types.go

@@ -125,6 +125,21 @@ func (ips IPFSPinStatus) IsPinned() bool {
 	return ips == IPFSPinStatusDirect || ips == IPFSPinStatusRecursive
 }
 
+// ToTrackerStatus converts the IPFSPinStatus value to the
+// appropriate TrackerStatus value.
+func (ips IPFSPinStatus) ToTrackerStatus() TrackerStatus {
+	return ipfsPinStatus2TrackerStatusMap[ips]
+}
+
+var ipfsPinStatus2TrackerStatusMap = map[IPFSPinStatus]TrackerStatus{
+	IPFSPinStatusDirect:    TrackerStatusPinned,
+	IPFSPinStatusRecursive: TrackerStatusPinned,
+	IPFSPinStatusIndirect:  TrackerStatusUnpinned,
+	IPFSPinStatusUnpinned:  TrackerStatusUnpinned,
+	IPFSPinStatusBug:       TrackerStatusBug,
+	IPFSPinStatusError:     TrackerStatusClusterError, //TODO(ajl): check suitability
+}
+
 // GlobalPinInfo contains cluster-wide status information about a tracked Cid,
 // indexed by cluster peer.
 type GlobalPinInfo struct {

cluster.go

@@ -74,7 +74,8 @@ func NewCluster(
 	tracker PinTracker,
 	monitor PeerMonitor,
 	allocator PinAllocator,
-	informer Informer) (*Cluster, error) {
+	informer Informer,
+) (*Cluster, error) {
 
 	err := cfg.Validate()
 	if err != nil {

optracker/operationtracker.go

@@ -0,0 +1,284 @@
+package optracker
+
+import (
+	"context"
+	"sync"
+
+	"github.com/ipfs/ipfs-cluster/api"
+
+	cid "github.com/ipfs/go-cid"
+	logging "github.com/ipfs/go-log"
+	peer "github.com/libp2p/go-libp2p-peer"
+)
+
+var logger = logging.Logger("optracker")
+
+//go:generate stringer -type=OperationType
+
+// OperationType represents the kinds of operations that the PinTracker
+// performs and the operationTracker tracks the status of.
+type OperationType int
+
+const (
+	// OperationUnknown represents an unknown operation.
+	OperationUnknown OperationType = iota
+	// OperationPin represents a pin operation.
+	OperationPin
+	// OperationUnpin represents an unpin operation.
+	OperationUnpin
+)
+
+//go:generate stringer -type=Phase
+
+// Phase represents the multiple phase that an operation can be in.
+type Phase int
+
+const (
+	// PhaseError represents an error state.
+	PhaseError Phase = iota
+	// PhaseQueued represents the queued phase of an operation.
+	PhaseQueued
+	// PhaseInProgress represents the operation as in progress.
+	PhaseInProgress
+)
+
+// Operation represents an ongoing operation involving a
+// particular Cid. It provides the type and phase of operation
+// and a way to mark the operation finished (also used to cancel).
+type Operation struct {
+	Cid    *cid.Cid
+	Op     OperationType
+	Phase  Phase
+	Ctx    context.Context
+	cancel func()
+}
+
+// NewOperation creates a new Operation.
+func NewOperation(ctx context.Context, c *cid.Cid, op OperationType) Operation {
+	ctx, cancel := context.WithCancel(ctx)
+	return Operation{
+		Cid:    c,
+		Op:     op,
+		Phase:  PhaseQueued,
+		Ctx:    ctx,
+		cancel: cancel, // use *OperationTracker.Finish() instead
+	}
+}
+
+// OperationTracker tracks and manages all inflight Operations.
+type OperationTracker struct {
+	ctx context.Context
+
+	mu         sync.RWMutex
+	operations map[string]Operation
+}
+
+// NewOperationTracker creates a new OperationTracker.
+func NewOperationTracker(ctx context.Context) *OperationTracker {
+	return &OperationTracker{
+		ctx:        ctx,
+		operations: make(map[string]Operation),
+	}
+}
+
+// TrackNewOperation tracks a new operation, adding it to the OperationTracker's
+// map of inflight operations.
+func (opt *OperationTracker) TrackNewOperation(ctx context.Context, c *cid.Cid, op OperationType) {
+	op2 := NewOperation(ctx, c, op)
+	logger.Debugf(
+		"'%s' on cid '%s' has been created with phase '%s'",
+		op.String(),
+		c.String(),
+		op2.Phase.String(),
+	)
+	opt.Set(op2)
+}
+
+// UpdateOperationPhase updates the phase of the operation associated with
+// the provided Cid.
+func (opt *OperationTracker) UpdateOperationPhase(c *cid.Cid, p Phase) {
+	opc, ok := opt.Get(c)
+	if !ok {
+		logger.Debugf(
+			"attempted to update non-existent operation with cid: %s",
+			c.String(),
+		)
+		return
+	}
+	opc.Phase = p
+	opt.Set(opc)
+	logger.Debugf(
+		"'%s' on cid '%s' has been updated to phase '%s'",
+		opc.Op.String(),
+		c.String(),
+		p.String(),
+	)
+}
+
+// SetError sets the phase of the operation to PhaseError and cancels
+// the operation context. Error is similar to Finish but doesn't
+// delete the operation from the map.
+func (opt *OperationTracker) SetError(c *cid.Cid) Operation {
+	logger.Debugf("optracker: setting operation in error state: %s", c)
+	opt.mu.Lock()
+	defer opt.mu.Unlock()
+
+	opc, ok := opt.operations[c.String()]
+	if !ok {
+		logger.Debugf(
+			"attempted to remove non-existent operation with cid: %s",
+			c.String(),
+		)
+		return Operation{}
+	}
+
+	opc.Phase = PhaseError
+	logger.Debugf(
+		"'%s' on cid '%s' has been updated to phase '%s'",
+		opc.Op.String(),
+		c.String(),
+		PhaseError,
+	)
+	opc.cancel()
+	return opc
+}
+
+// RemoveErroredOperations removes operations that have errored from
+// the map.
+func (opt *OperationTracker) RemoveErroredOperations(c *cid.Cid) {
+	opt.mu.Lock()
+	defer opt.mu.Unlock()
+
+	if opc, ok := opt.operations[c.String()]; ok && opc.Phase == PhaseError {
+		delete(opt.operations, c.String())
+		logger.Debugf(
+			"'%s' on cid '%s' has been removed",
+			opc.Op.String(),
+			c.String(),
+		)
+	}
+
+	logger.Debugf(
+		"attempted to remove non-errored operation with cid: %s",
+		c.String(),
+	)
+	return
+}
+
+// Finish cancels the operation context and removes it from the map.
+// If the Operations Phase has been set to PhaseError, the operation
+// context will be cancelled but the operation won't be removed from
+// the map.
+func (opt *OperationTracker) Finish(c *cid.Cid) {
+	opt.mu.Lock()
+	defer opt.mu.Unlock()
+
+	opc, ok := opt.operations[c.String()]
+	if !ok {
+		logger.Debugf(
+			"attempted to remove non-existent operation with cid: %s",
+			c.String(),
+		)
+		return
+	}
+
+	if opc.Phase != PhaseError {
+		opc.cancel()
+		delete(opt.operations, c.String())
+		logger.Debugf(
+			"'%s' on cid '%s' has been removed",
+			opc.Op.String(),
+			c.String(),
+		)
+	}
+}
+
+// Set sets the operation in the OperationTrackers map.
+func (opt *OperationTracker) Set(oc Operation) {
+	opt.mu.Lock()
+	opt.operations[oc.Cid.String()] = oc
+	opt.mu.Unlock()
+}
+
+// Get gets the operation associated with the Cid. If the
+// there is no associated operation, Get will return Operation{}, false.
+func (opt *OperationTracker) Get(c *cid.Cid) (Operation, bool) {
+	opt.mu.RLock()
+	opc, ok := opt.operations[c.String()]
+	opt.mu.RUnlock()
+	return opc, ok
+}
+
+// GetAll gets all inflight operations.
+func (opt *OperationTracker) GetAll() []Operation {
+	var ops []Operation
+	opt.mu.RLock()
+	defer opt.mu.RUnlock()
+	for _, op := range opt.operations {
+		ops = append(ops, op)
+	}
+	return ops
+}
+
+// Filter returns a slice that only contains operations
+// with the matching filter. Note, only supports
+// filters of type OperationType or Phase, any other type
+// will result in a nil slice being returned.
+func (opt *OperationTracker) Filter(filter interface{}) []Operation {
+	var ops []Operation
+	opt.mu.RLock()
+	defer opt.mu.RUnlock()
+	for _, op := range opt.operations {
+		switch filter.(type) {
+		case OperationType:
+			if op.Op == filter {
+				ops = append(ops, op)
+			}
+		case Phase:
+			if op.Phase == filter {
+				ops = append(ops, op)
+			}
+		default:
+		}
+	}
+	return ops
+}
+
+// ToPinInfo converts an operation to an api.PinInfo.
+func (op Operation) ToPinInfo(pid peer.ID) api.PinInfo {
+	return api.PinInfo{Cid: op.Cid, Peer: pid, Status: op.ToTrackerStatus()}
+}
+
+// ToTrackerStatus converts an operation and it's phase to
+// the most approriate api.TrackerStatus.
+func (op Operation) ToTrackerStatus() api.TrackerStatus {
+	switch op.Op {
+	case OperationPin:
+		switch op.Phase {
+		case PhaseError:
+			return api.TrackerStatusPinError
+		case PhaseQueued:
+			return api.TrackerStatusPinQueued
+		case PhaseInProgress:
+			return api.TrackerStatusPinning
+		default:
+			logger.Debugf("couldn't match operation to tracker status")
+			return api.TrackerStatusBug
+		}
+	case OperationUnpin:
+		switch op.Phase {
+		case PhaseError:
+			return api.TrackerStatusUnpinError
+		case PhaseQueued:
+			return api.TrackerStatusUnpinQueued
+		case PhaseInProgress:
+			return api.TrackerStatusUnpinning
+		default:
+			logger.Debugf("couldn't match operation to tracker status")
+			return api.TrackerStatusBug
+		}
+	default:
+		logger.Debugf("couldn't match operation to tracker status")
+		return api.TrackerStatusBug
+	}
+}

optracker/operationtracker_test.go

@@ -0,0 +1,115 @@
+package optracker
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/ipfs/ipfs-cluster/test"
+)
+
+func testOperationTracker(ctx context.Context, t *testing.T) *OperationTracker {
+	return NewOperationTracker(ctx)
+}
+
+func TestOperationTracker_TrackNewOperationWithCtx(t *testing.T) {
+	ctx := context.Background()
+	opt := testOperationTracker(ctx, t)
+
+	h := test.MustDecodeCid(test.TestCid1)
+	opt.TrackNewOperation(ctx, h, OperationPin)
+
+	opc, ok := opt.Get(h)
+	if !ok {
+		t.Errorf("operation wasn't set in operationTracker")
+	}
+
+	testopc1 := Operation{
+		Cid:   h,
+		Op:    OperationPin,
+		Phase: PhaseQueued,
+	}
+
+	if opc.Cid != testopc1.Cid {
+		t.Fail()
+	}
+	if opc.Op != testopc1.Op {
+		t.Fail()
+	}
+	if opc.Phase != testopc1.Phase {
+		t.Fail()
+	}
+	if t.Failed() {
+		fmt.Printf("got %#v\nwant %#v", opc, testopc1)
+	}
+}
+
+func TestOperationTracker_Finish(t *testing.T) {
+	ctx := context.Background()
+	opt := testOperationTracker(ctx, t)
+
+	h := test.MustDecodeCid(test.TestCid1)
+	opt.TrackNewOperation(ctx, h, OperationPin)
+
+	opt.Finish(h)
+	_, ok := opt.Get(h)
+	if ok {
+		t.Error("cancelling operation failed to remove it from the map of ongoing operation")
+	}
+}
+
+func TestOperationTracker_UpdateOperationPhase(t *testing.T) {
+	ctx := context.Background()
+	opt := testOperationTracker(ctx, t)
+
+	h := test.MustDecodeCid(test.TestCid1)
+	opt.TrackNewOperation(ctx, h, OperationPin)
+
+	opt.UpdateOperationPhase(h, PhaseInProgress)
+	opc, ok := opt.Get(h)
+	if !ok {
+		t.Error("error getting operation context after updating phase")
+	}
+
+	if opc.Phase != PhaseInProgress {
+		t.Errorf("operation phase failed to be updated to %s, got %s", PhaseInProgress.String(), opc.Phase.String())
+	}
+}
+
+func TestOperationTracker_Filter(t *testing.T) {
+	ctx := context.Background()
+	testOpsMap := map[string]Operation{
+		test.TestCid1: Operation{Cid: test.MustDecodeCid(test.TestCid1), Op: OperationPin, Phase: PhaseQueued},
+		test.TestCid2: Operation{Cid: test.MustDecodeCid(test.TestCid2), Op: OperationPin, Phase: PhaseInProgress},
+		test.TestCid3: Operation{Cid: test.MustDecodeCid(test.TestCid3), Op: OperationUnpin, Phase: PhaseInProgress},
+	}
+	opt := &OperationTracker{ctx: ctx, operations: testOpsMap}
+
+	t.Run("filter to pin operations", func(t *testing.T) {
+		wantLen := 2
+		wantOp := OperationPin
+		got := opt.Filter(wantOp)
+		if len(got) != wantLen {
+			t.Errorf("want: %d %s operations; got: %d", wantLen, wantOp.String(), len(got))
+		}
+		for i := range got {
+			if got[i].Op != wantOp {
+				t.Errorf("want: %v; got: %v", wantOp.String(), got[i])
+			}
+		}
+	})
+
+	t.Run("filter to in progress phase", func(t *testing.T) {
+		wantLen := 2
+		wantPhase := PhaseInProgress
+		got := opt.Filter(PhaseInProgress)
+		if len(got) != wantLen {
+			t.Errorf("want: %d %s operations; got: %d", wantLen, wantPhase.String(), len(got))
+		}
+		for i := range got {
+			if got[i].Phase != wantPhase {
+				t.Errorf("want: %s; got: %v", wantPhase.String(), got[i])
+			}
+		}
+	})
+}

optracker/operationtype_string.go

@@ -0,0 +1,16 @@
+// Code generated by "stringer -type=OperationType"; DO NOT EDIT.
+
+package optracker
+
+import "strconv"
+
+const _OperationType_name = "OperationUnknownOperationPinOperationUnpin"
+
+var _OperationType_index = [...]uint8{0, 16, 28, 42}
+
+func (i OperationType) String() string {
+	if i < 0 || i >= OperationType(len(_OperationType_index)-1) {
+		return "OperationType(" + strconv.FormatInt(int64(i), 10) + ")"
+	}
+	return _OperationType_name[_OperationType_index[i]:_OperationType_index[i+1]]
+}

optracker/phase_string.go

@@ -0,0 +1,16 @@
+// Code generated by "stringer -type=Phase"; DO NOT EDIT.
+
+package optracker
+
+import "strconv"
+
+const _Phase_name = "PhaseErrorPhaseQueuedPhaseInProgress"
+
+var _Phase_index = [...]uint8{0, 10, 21, 36}
+
+func (i Phase) String() string {
+	if i < 0 || i >= Phase(len(_Phase_index)-1) {
+		return "Phase(" + strconv.FormatInt(int64(i), 10) + ")"
+	}
+	return _Phase_name[_Phase_index[i]:_Phase_index[i+1]]
+}

pintracker/maptracker/maptracker.go

@@ -9,6 +9,8 @@ import (
 	"time"
 
 	"github.com/ipfs/ipfs-cluster/api"
+	"github.com/ipfs/ipfs-cluster/optracker"
+	"github.com/ipfs/ipfs-cluster/pintracker/util"
 
 	rpc "github.com/hsanjuan/go-libp2p-gorpc"
 	cid "github.com/ipfs/go-cid"
@@ -32,7 +34,7 @@ type MapPinTracker struct {
 	status map[string]api.PinInfo
 	config *Config
 
-	optracker *operationTracker
+	optracker *optracker.OperationTracker
 
 	ctx    context.Context
 	cancel func()
@@ -59,7 +61,7 @@ func NewMapPinTracker(cfg *Config, pid peer.ID) *MapPinTracker {
 		cancel:    cancel,
 		status:    make(map[string]api.PinInfo),
 		config:    cfg,
-		optracker: newOperationTracker(ctx),
+		optracker: optracker.NewOperationTracker(ctx),
 		rpcReady:  make(chan struct{}, 1),
 		peerID:    pid,
 		pinCh:     make(chan api.Pin, cfg.MaxPinQueueSize),
@@ -77,10 +79,10 @@ func (mpt *MapPinTracker) pinWorker() {
 	for {
 		select {
 		case p := <-mpt.pinCh:
-			if opc, ok := mpt.optracker.get(p.Cid); ok && opc.op == operationPin {
-				mpt.optracker.updateOperationPhase(
+			if opc, ok := mpt.optracker.Get(p.Cid); ok && opc.Op == optracker.OperationPin {
+				mpt.optracker.UpdateOperationPhase(
 					p.Cid,
-					phaseInProgress,
+					optracker.PhaseInProgress,
 				)
 				mpt.pin(p)
 			}
@@ -95,10 +97,10 @@ func (mpt *MapPinTracker) unpinWorker() {
 	for {
 		select {
 		case p := <-mpt.unpinCh:
-			if opc, ok := mpt.optracker.get(p.Cid); ok && opc.op == operationUnpin {
-				mpt.optracker.updateOperationPhase(
+			if opc, ok := mpt.optracker.Get(p.Cid); ok && opc.Op == optracker.OperationUnpin {
+				mpt.optracker.UpdateOperationPhase(
 					p.Cid,
-					phaseInProgress,
+					optracker.PhaseInProgress,
 				)
 				mpt.unpin(p)
 			}
@@ -218,12 +220,12 @@ func (mpt *MapPinTracker) pin(c api.Pin) error {
 	mpt.set(c.Cid, api.TrackerStatusPinning)
 
 	var ctx context.Context
-	opc, ok := mpt.optracker.get(c.Cid)
+	opc, ok := mpt.optracker.Get(c.Cid)
 	if !ok {
 		logger.Debug("pin operation wasn't being tracked")
 		ctx = mpt.ctx
 	} else {
-		ctx = opc.ctx
+		ctx = opc.Ctx
 	}
 
 	err := mpt.rpcClient.CallContext(
@@ -236,11 +238,12 @@ func (mpt *MapPinTracker) pin(c api.Pin) error {
 	)
 	if err != nil {
 		mpt.setError(c.Cid, err)
+		mpt.optracker.SetError(c.Cid)
 		return err
 	}
 
 	mpt.set(c.Cid, api.TrackerStatusPinned)
-	mpt.optracker.finish(c.Cid)
+	mpt.optracker.Finish(c.Cid)
 	return nil
 }
 
@@ -249,12 +252,12 @@ func (mpt *MapPinTracker) unpin(c api.Pin) error {
 	mpt.set(c.Cid, api.TrackerStatusUnpinning)
 
 	var ctx context.Context
-	opc, ok := mpt.optracker.get(c.Cid)
+	opc, ok := mpt.optracker.Get(c.Cid)
 	if !ok {
 		logger.Debug("pin operation wasn't being tracked")
 		ctx = mpt.ctx
 	} else {
-		ctx = opc.ctx
+		ctx = opc.Ctx
 	}
 
 	err := mpt.rpcClient.CallContext(
@@ -267,11 +270,12 @@ func (mpt *MapPinTracker) unpin(c api.Pin) error {
 	)
 	if err != nil {
 		mpt.setError(c.Cid, err)
+		mpt.optracker.SetError(c.Cid)
 		return err
 	}
 
 	mpt.set(c.Cid, api.TrackerStatusUnpinned)
-	mpt.optracker.finish(c.Cid)
+	mpt.optracker.Finish(c.Cid)
 	return nil
 }
 
@@ -279,26 +283,26 @@ func (mpt *MapPinTracker) unpin(c api.Pin) error {
 // possibly triggering Pin operations on the IPFS daemon.
 func (mpt *MapPinTracker) Track(c api.Pin) error {
 	logger.Debugf("tracking %s", c.Cid)
-
-	if opc, ok := mpt.optracker.get(c.Cid); ok {
-		switch {
-		case opc.op == operationPin:
-			return nil // already ongoing
-		case opc.op == operationUnpin && opc.phase == phaseQueued:
-			mpt.optracker.finish(c.Cid)
-			return nil // cancelled while in queue, all done
-		case opc.op == operationUnpin && opc.phase == phaseInProgress:
-			mpt.optracker.finish(c.Cid)
-			// cancelled while unpinning: continue and trigger pin
+	if opc, ok := mpt.optracker.Get(c.Cid); ok {
+		if opc.Op == optracker.OperationUnpin {
+			switch opc.Phase {
+			case optracker.PhaseQueued:
+				mpt.optracker.Finish(c.Cid)
+				return nil
+			case optracker.PhaseInProgress:
+				mpt.optracker.Finish(c.Cid)
+				// NOTE: this may leave the api.PinInfo in an error state
+				// so a pin operation needs to be run on it (same as Recover)
+			}
 		}
 	}
 
-	if mpt.isRemote(c) {
+	if util.IsRemotePin(c, mpt.peerID) {
 		if mpt.get(c.Cid).Status == api.TrackerStatusPinned {
-			mpt.optracker.trackNewOperation(
+			mpt.optracker.TrackNewOperation(
 				mpt.ctx,
 				c.Cid,
-				operationUnpin,
+				optracker.OperationUnpin,
 			)
 			mpt.unpin(c)
 		}
@@ -306,7 +310,7 @@ func (mpt *MapPinTracker) Track(c api.Pin) error {
 		return nil
 	}
 
-	mpt.optracker.trackNewOperation(mpt.ctx, c.Cid, operationPin)
+	mpt.optracker.TrackNewOperation(mpt.ctx, c.Cid, optracker.OperationPin)
 	mpt.set(c.Cid, api.TrackerStatusPinQueued)
 
 	select {
@@ -314,7 +318,7 @@ func (mpt *MapPinTracker) Track(c api.Pin) error {
 	default:
 		err := errors.New("pin queue is full")
 		mpt.setError(c.Cid, err)
-		mpt.optracker.finish(c.Cid)
+		mpt.optracker.SetError(c.Cid)
 		logger.Error(err.Error())
 		return err
 	}
@@ -325,16 +329,16 @@ func (mpt *MapPinTracker) Track(c api.Pin) error {
 // If the Cid is pinned locally, it will be unpinned.
 func (mpt *MapPinTracker) Untrack(c *cid.Cid) error {
 	logger.Debugf("untracking %s", c)
-	if opc, ok := mpt.optracker.get(c); ok {
-		switch {
-		case opc.op == operationUnpin:
-			return nil // already ongoing
-		case opc.op == operationPin && opc.phase == phaseQueued:
-			mpt.optracker.finish(c)
-			return nil // cancelled while in queue, all done
-		case opc.op == operationPin && opc.phase == phaseInProgress:
-			mpt.optracker.finish(c)
-			// cancelled while pinning: continue and trigger unpin
+	if opc, ok := mpt.optracker.Get(c); ok {
+		if opc.Op == optracker.OperationPin {
+			mpt.optracker.Finish(c) // cancel it
+
+			switch opc.Phase {
+			case optracker.PhaseQueued:
+				return nil
+			case optracker.PhaseInProgress:
+				// continues below to run a full unpin
+			}
 		}
 	}
 
@@ -342,7 +346,7 @@ func (mpt *MapPinTracker) Untrack(c *cid.Cid) error {
 		return nil
 	}
 
-	mpt.optracker.trackNewOperation(mpt.ctx, c, operationUnpin)
+	mpt.optracker.TrackNewOperation(mpt.ctx, c, optracker.OperationUnpin)
 	mpt.set(c, api.TrackerStatusUnpinQueued)
 
 	select {
@@ -350,7 +354,7 @@ func (mpt *MapPinTracker) Untrack(c *cid.Cid) error {
 	default:
 		err := errors.New("unpin queue is full")
 		mpt.setError(c, err)
-		mpt.optracker.finish(c)
+		mpt.optracker.SetError(c)
 		logger.Error(err.Error())
 		return err
 	}
@@ -459,7 +463,7 @@ func (mpt *MapPinTracker) syncStatus(c *cid.Cid, ips api.IPFSPinStatus) api.PinI
 		case api.TrackerStatusUnpinned:
 			mpt.setError(c, errPinned)
 		case api.TrackerStatusUnpinError: // nothing, keep error as it was
-		default: //remote
+		default: // remote
 		}
 	} else {
 		switch p.Status {

pintracker/maptracker/operationtracker.go

@@ -1,131 +0,0 @@
-package maptracker
-
-import (
-	"context"
-	"sync"
-
-	"github.com/ipfs/go-cid"
-)
-
-//go:generate stringer -type=operationType
-
-// operationType represents the kinds of operations that the PinTracker
-// performs and the operationTracker tracks the status of.
-type operationType int
-
-const (
-	operationUnknown operationType = iota
-	operationPin
-	operationUnpin
-)
-
-//go:generate stringer -type=phase
-
-// phase represents the multiple phase that an operation can be in.
-type phase int
-
-const (
-	phaseError phase = iota
-	phaseQueued
-	phaseInProgress
-)
-
-type operation struct {
-	cid    *cid.Cid
-	op     operationType
-	phase  phase
-	ctx    context.Context
-	cancel func()
-}
-
-func newOperationWithContext(ctx context.Context, c *cid.Cid, op operationType) operation {
-	ctx, cancel := context.WithCancel(ctx)
-	return operation{
-		cid:    c,
-		op:     op,
-		phase:  phaseQueued,
-		ctx:    ctx,
-		cancel: cancel, // use *operationTracker.cancelOperation() instead
-	}
-}
-
-type operationTracker struct {
-	ctx context.Context
-
-	mu         sync.RWMutex
-	operations map[string]operation
-}
-
-func newOperationTracker(ctx context.Context) *operationTracker {
-	return &operationTracker{
-		ctx:        ctx,
-		operations: make(map[string]operation),
-	}
-}
-
-//TODO(ajl): return error or bool if there is already an ongoing operation
-func (opt *operationTracker) trackNewOperation(ctx context.Context, c *cid.Cid, op operationType) {
-	op2 := newOperationWithContext(ctx, c, op)
-	logger.Debugf(
-		"'%s' on cid '%s' has been created with phase '%s'",
-		op.String(),
-		c.String(),
-		op2.phase.String(),
-	)
-	opt.set(op2)
-}
-
-func (opt *operationTracker) updateOperationPhase(c *cid.Cid, p phase) {
-	opc, ok := opt.get(c)
-	if !ok {
-		logger.Debugf(
-			"attempted to update non-existent operation with cid: %s",
-			c.String(),
-		)
-		return
-	}
-	opc.phase = p
-	opt.set(opc)
-	logger.Debugf(
-		"'%s' on cid '%s' has been updated to phase '%s'",
-		opc.op.String(),
-		c.String(),
-		p.String(),
-	)
-}
-
-func (opt *operationTracker) set(oc operation) {
-	opt.mu.Lock()
-	opt.operations[oc.cid.String()] = oc
-	opt.mu.Unlock()
-}
-
-func (opt *operationTracker) get(c *cid.Cid) (operation, bool) {
-	opt.mu.RLock()
-	opc, ok := opt.operations[c.String()]
-	opt.mu.RUnlock()
-	return opc, ok
-}
-
-// finish cancels the operation context and removes it from the map
-func (opt *operationTracker) finish(c *cid.Cid) {
-	opt.mu.Lock()
-	defer opt.mu.Unlock()
-
-	opc, ok := opt.operations[c.String()]
-	if !ok {
-		logger.Debugf(
-			"attempted to remove non-existent operation with cid: %s",
-			c.String(),
-		)
-		return
-	}
-
-	opc.cancel()
-	delete(opt.operations, c.String())
-	logger.Debugf(
-		"'%s' on cid '%s' has been removed",
-		opc.op.String(),
-		c.String(),
-	)
-}

pintracker/maptracker/operationtracker_test.go

@@ -1,79 +0,0 @@
-package maptracker
-
-import (
-	"context"
-	"fmt"
-	"testing"
-
-	"github.com/ipfs/go-cid"
-
-	"github.com/ipfs/ipfs-cluster/test"
-)
-
-func testOperationTracker(ctx context.Context, t *testing.T) *operationTracker {
-	return newOperationTracker(ctx)
-}
-
-func TestOperationTracker_trackNewOperationWithCtx(t *testing.T) {
-	ctx := context.Background()
-	opt := testOperationTracker(ctx, t)
-
-	h, _ := cid.Decode(test.TestCid1)
-	opt.trackNewOperation(ctx, h, operationPin)
-
-	opc, ok := opt.get(h)
-	if !ok {
-		t.Errorf("operation wasn't set in operationTracker")
-	}
-
-	testopc1 := operation{
-		cid:   h,
-		op:    operationPin,
-		phase: phaseQueued,
-	}
-
-	if opc.cid != testopc1.cid {
-		t.Fail()
-	}
-	if opc.op != testopc1.op {
-		t.Fail()
-	}
-	if opc.phase != testopc1.phase {
-		t.Fail()
-	}
-	if t.Failed() {
-		fmt.Printf("got %#v\nwant %#v", opc, testopc1)
-	}
-}
-
-func TestOperationTracker_finish(t *testing.T) {
-	ctx := context.Background()
-	opt := testOperationTracker(ctx, t)
-
-	h, _ := cid.Decode(test.TestCid1)
-	opt.trackNewOperation(ctx, h, operationPin)
-
-	opt.finish(h)
-	_, ok := opt.get(h)
-	if ok {
-		t.Error("cancelling operation failed to remove it from the map of ongoing operation")
-	}
-}
-
-func TestOperationTracker_updateOperationPhase(t *testing.T) {
-	ctx := context.Background()
-	opt := testOperationTracker(ctx, t)
-
-	h, _ := cid.Decode(test.TestCid1)
-	opt.trackNewOperation(ctx, h, operationPin)
-
-	opt.updateOperationPhase(h, phaseInProgress)
-	opc, ok := opt.get(h)
-	if !ok {
-		t.Error("error getting operation context after updating phase")
-	}
-
-	if opc.phase != phaseInProgress {
-		t.Errorf("operation phase failed to be updated to %s, got %s", phaseInProgress.String(), opc.phase.String())
-	}
-}

pintracker/maptracker/operationtype_string.go

@@ -1,16 +0,0 @@
-// Code generated by "stringer -type=operationType"; DO NOT EDIT.
-
-package maptracker
-
-import "strconv"
-
-const _operationType_name = "operationUnknownoperationPinoperationUnpin"
-
-var _operationType_index = [...]uint8{0, 16, 28, 42}
-
-func (i operationType) String() string {
-	if i < 0 || i >= operationType(len(_operationType_index)-1) {
-		return "operationType(" + strconv.FormatInt(int64(i), 10) + ")"
-	}
-	return _operationType_name[_operationType_index[i]:_operationType_index[i+1]]
-}

pintracker/maptracker/phase_string.go

@@ -1,16 +0,0 @@
-// Code generated by "stringer -type=phase"; DO NOT EDIT.
-
-package maptracker
-
-import "strconv"
-
-const _phase_name = "phaseErrorphaseQueuedphaseInProgress"
-
-var _phase_index = [...]uint8{0, 10, 21, 36}
-
-func (i phase) String() string {
-	if i < 0 || i >= phase(len(_phase_index)-1) {
-		return "phase(" + strconv.FormatInt(int64(i), 10) + ")"
-	}
-	return _phase_name[_phase_index[i]:_phase_index[i+1]]
-}

pintracker/util/pin.go

@@ -0,0 +1,22 @@
+package util
+
+import (
+	"github.com/ipfs/ipfs-cluster/api"
+
+	peer "github.com/libp2p/go-libp2p-peer"
+)
+
+// IsRemotePin determines whether a Pin's ReplicationFactor has
+// been met, so as to either pin or unpin it from the peer.
+func IsRemotePin(c api.Pin, pid peer.ID) bool {
+	if c.ReplicationFactorMax < 0 {
+		return false
+	}
+
+	for _, p := range c.Allocations {
+		if p == pid {
+			return false
+		}
+	}
+	return true
+}

rpc_api.go

@@ -112,7 +112,7 @@ func (rpcapi *RPCAPI) Join(ctx context.Context, in api.MultiaddrSerial, out *str
 // StatusAll runs Cluster.StatusAll().
 func (rpcapi *RPCAPI) StatusAll(ctx context.Context, in struct{}, out *[]api.GlobalPinInfoSerial) error {
 	pinfos, err := rpcapi.c.StatusAll()
-	*out = globalPinInfoSliceToSerial(pinfos)
+	*out = GlobalPinInfoSliceToSerial(pinfos)
 	return err
 }
 
@@ -142,7 +142,7 @@ func (rpcapi *RPCAPI) StatusLocal(ctx context.Context, in api.PinSerial, out *ap
 // SyncAll runs Cluster.SyncAll().
 func (rpcapi *RPCAPI) SyncAll(ctx context.Context, in struct{}, out *[]api.GlobalPinInfoSerial) error {
 	pinfos, err := rpcapi.c.SyncAll()
-	*out = globalPinInfoSliceToSerial(pinfos)
+	*out = GlobalPinInfoSliceToSerial(pinfos)
 	return err
 }
 

util.go

@@ -69,7 +69,9 @@ func pinInfoSliceToSerial(pi []api.PinInfo) []api.PinInfoSerial {
 	return pis
 }
 
-func globalPinInfoSliceToSerial(gpi []api.GlobalPinInfo) []api.GlobalPinInfoSerial {
+// GlobalPinInfoSliceToSerial is a helper function for serializing a slice of
+// api.GlobalPinInfos.
+func GlobalPinInfoSliceToSerial(gpi []api.GlobalPinInfo) []api.GlobalPinInfoSerial {
 	gpis := make([]api.GlobalPinInfoSerial, len(gpi), len(gpi))
 	for i, v := range gpi {
 		gpis[i] = v.ToSerial()