ffs: enable auto-repair & update to new interopnet & ipfs pinset cache

Makefile

@@ -14,5 +14,5 @@ build: build-cli build-server
 .PHONY: build
 
 test:
-	go test -short -p 1 -race ./... 
+	go test -short -p 4 -race ./... 
 .PHONY: test

deals/deals.go

@@ -101,10 +101,10 @@ func (m *Module) Store(ctx context.Context, waddr string, data io.Reader, dcfgs
 			Data: &storagemarket.DataRef{
 				Root: dataCid,
 			},
-			BlocksDuration: dur,
-			EpochPrice:     types.NewInt(c.EpochPrice),
-			Miner:          maddr,
-			Wallet:         addr,
+			MinBlocksDuration: dur,
+			EpochPrice:        types.NewInt(c.EpochPrice),
+			Miner:             maddr,
+			Wallet:            addr,
 		}
 		p, err := m.api.ClientStartDeal(ctx, params)
 		if err != nil {

docker/docker-compose-embedded.yaml

@@ -32,7 +32,7 @@ services:
       - 5001:5001
 
   lotus:
-    image: textile/lotus-devnet:sha-9aab2c6
+    image: textile/lotus-devnet:sha-c755791
     ports:
       - 7777:7777
     environment:

docker/docker-compose.yaml

@@ -31,7 +31,7 @@ services:
       - textile-fc-lotus:/root/lotus
 
   lotus:
-    image: textile/lotus:custom
+    image: textile/lotus:a99875f
     volumes:
       - textile-fc-lotus:/data
     ports:

fchost/config.go

@@ -9,8 +9,8 @@ import (
 
 var (
 	addrs = []string{
-		"/dns4/t01000.miner.interopnet.kittyhawk.wtf/tcp/1347/p2p/12D3KooWGzxzKZYveHXtpG6AsrUJBcWxHBFS2HsEoGTxrMLvKXtf",
-		"/ip4/52.36.61.156/tcp/1347/p2p/12D3KooWFETiESTf1v4PGUvtnxMAcEFMzLZbJGg4tjWfGEimYior",
+		"/dns4/t01000.miner.interopnet.kittyhawk.wtf/tcp/1347/p2p/12D3KooWNjqGTNZ592wG5DmX7Rbmb76V3NHMdZMqjqPJ1xiax7w6",
+		"/ip4/54.186.82.90/tcp/1347/p2p/12D3KooWGf4ggd3cdZJKZPHkgSe8Lxf4MDsKhVC7npRGXgsPP4fz",
 	}
 )
 

ffs/coreipfs/coreipfs.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"sync"
 
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
@@ -25,6 +26,9 @@ var (
 type CoreIpfs struct {
 	ipfs iface.CoreAPI
 	l    ffs.CidLogger
+
+	lock   sync.Mutex
+	pinset map[cid.Cid]struct{}
 }
 
 var _ ffs.HotStorage = (*CoreIpfs)(nil)
@@ -58,16 +62,13 @@ func (ci *CoreIpfs) Remove(ctx context.Context, c cid.Cid) error {
 
 // IsStored return if a particular Cid is stored.
 func (ci *CoreIpfs) IsStored(ctx context.Context, c cid.Cid) (bool, error) {
-	pins, err := ci.ipfs.Pin().Ls(ctx)
-	if err != nil {
-		return false, fmt.Errorf("getting pins from IPFS: %s", err)
-	}
-	for _, p := range pins {
-		if p.Path().Cid() == c {
-			return true, nil
+	if ci.pinset == nil {
+		if err := ci.ensurePinsetCache(ctx); err != nil {
+			return false, err
 		}
 	}
-	return false, nil
+	_, ok := ci.pinset[c]
+	return ok, nil
 }
 
 // Add adds an io.Reader data as file in the IPFS node.
@@ -106,6 +107,12 @@ func (ci *CoreIpfs) Store(ctx context.Context, c cid.Cid) (int, error) {
 	if err != nil {
 		return 0, fmt.Errorf("getting stats of cid %s: %s", c, err)
 	}
+	if err := ci.ensurePinsetCache(ctx); err != nil {
+		return 0, err
+	}
+	ci.lock.Lock()
+	ci.pinset[c] = struct{}{}
+	ci.lock.Unlock()
 	return s.Size(), nil
 }
 
@@ -123,3 +130,21 @@ func (ci *CoreIpfs) Replace(ctx context.Context, c1 cid.Cid, c2 cid.Cid) (int, e
 	}
 	return stat.Size(), nil
 }
+
+func (ci *CoreIpfs) ensurePinsetCache(ctx context.Context) error {
+	ci.lock.Lock()
+	defer ci.lock.Unlock()
+	if ci.pinset != nil {
+		return nil
+	}
+	pins, err := ci.ipfs.Pin().Ls(ctx)
+	if err != nil {
+		ci.lock.Unlock()
+		return fmt.Errorf("getting pins from IPFS: %s", err)
+	}
+	ci.pinset = make(map[cid.Cid]struct{}, len(pins))
+	for _, p := range pins {
+		ci.pinset[p.Path().Cid()] = struct{}{}
+	}
+	return nil
+}