fix: alternative identity CID handling

Ref: ipfs/go-graphsync#442

Instead of relying on the transports to do what we ask wrt identity CIDs;
handle it all locally and gate their handling right up to the output block
stream.

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/ipfs/go-block-format v0.2.0
 	github.com/ipfs/go-cid v0.4.1
 	github.com/ipfs/go-datastore v0.6.0
-	github.com/ipfs/go-graphsync v0.14.9
+	github.com/ipfs/go-graphsync v0.14.10
 	github.com/ipfs/go-ipfs-blockstore v1.3.0
 	github.com/ipfs/go-ipfs-blocksutil v0.0.1
 	github.com/ipfs/go-ipfs-delay v0.0.1

go.sum

@@ -264,8 +264,8 @@ github.com/ipfs/go-datastore v0.6.0 h1:JKyz+Gvz1QEZw0LsX1IBn+JFCJQH4SJVFtM4uWU0M
 github.com/ipfs/go-datastore v0.6.0/go.mod h1:rt5M3nNbSO/8q1t4LNkLyUwRs8HupMeN/8O4Vn9YAT8=
 github.com/ipfs/go-detect-race v0.0.1 h1:qX/xay2W3E4Q1U7d9lNs1sU9nvguX0a7319XbyQ6cOk=
 github.com/ipfs/go-detect-race v0.0.1/go.mod h1:8BNT7shDZPo99Q74BpGMK+4D8Mn4j46UU0LZ723meps=
-github.com/ipfs/go-graphsync v0.14.9 h1:AZ1TxnU5LNANXesKr4ZjUtj0rLpqp0Qt5PL6Xf7uhc4=
-github.com/ipfs/go-graphsync v0.14.9/go.mod h1:qyHjUvHey6EfKUDMQPwCuVkMOurRG3hcjRm+FaVP6bE=
+github.com/ipfs/go-graphsync v0.14.10 h1:2slsmWcHxXGS/T6HR5hilo2sSAX1VsZVxjoOzA3XCAQ=
+github.com/ipfs/go-graphsync v0.14.10/go.mod h1:qyHjUvHey6EfKUDMQPwCuVkMOurRG3hcjRm+FaVP6bE=
 github.com/ipfs/go-hamt-ipld v0.1.1/go.mod h1:1EZCr2v0jlCnhpa+aZ0JZYp8Tt2w16+JJOAVz17YcDk=
 github.com/ipfs/go-ipfs-blockstore v1.3.0 h1:m2EXaWgwTzAfsmt5UdJ7Is6l4gJcaM/A12XwJyvYvMM=
 github.com/ipfs/go-ipfs-blockstore v1.3.0/go.mod h1:KgtZyc9fq+P2xJUiCAzbRdhhqJHvsw8u2Dlqy2MyRTE=

pkg/internal/itest/http_fetch_test.go

@@ -817,6 +817,35 @@ func TestHttpFetch(t *testing.T) {
 				return []unixfs.DirEntry{trustlesstestutil.MakeDagWithIdentity(t, *remotes[0].LinkSystem)}
 			},
 		},
+		// noDups variants are important because handling of these happens all the
+		// way up to DuplicateAdderCar
+		{
+			name:             "non-unixfs graphsync /w noDups",
+			setHeader:        noDups,
+			expectNoDups:     true,
+			graphsyncRemotes: 1,
+			generate: func(t *testing.T, rndReader io.Reader, remotes []testpeer.TestPeer) []unixfs.DirEntry {
+				return []unixfs.DirEntry{trustlesstestutil.MakeDagWithIdentity(t, *remotes[0].LinkSystem)}
+			},
+		},
+		{
+			name:           "non-unixfs bitswap /w noDups",
+			setHeader:      noDups,
+			expectNoDups:   true,
+			bitswapRemotes: 1,
+			generate: func(t *testing.T, rndReader io.Reader, remotes []testpeer.TestPeer) []unixfs.DirEntry {
+				return []unixfs.DirEntry{trustlesstestutil.MakeDagWithIdentity(t, *remotes[0].LinkSystem)}
+			},
+		},
+		{
+			name:         "non-unixfs http /w noDups",
+			setHeader:    noDups,
+			expectNoDups: true,
+			httpRemotes:  1,
+			generate: func(t *testing.T, rndReader io.Reader, remotes []testpeer.TestPeer) []unixfs.DirEntry {
+				return []unixfs.DirEntry{trustlesstestutil.MakeDagWithIdentity(t, *remotes[0].LinkSystem)}
+			},
+		},
 	}
 
 	for _, testCase := range testCases {

pkg/internal/itest/testpeer/generator.go

@@ -299,13 +299,13 @@ func newTestPeer(
 	dstoreDelayed := delayed.New(dstore, bsdelay)
 
 	if cfg.bstore == nil {
-		var err error
-		cfg.bstore, err = blockstore.CachedBlockstore(ctx,
-			blockstore.NewIdStore(blockstore.NewBlockstore(dstoreDelayed)),
+		bstore, err := blockstore.CachedBlockstore(ctx,
+			blockstore.NewBlockstore(dstoreDelayed),
 			blockstore.DefaultCacheOpts())
 		if err != nil {
 			return TestPeer{}, nil, err
 		}
+		cfg.bstore = blockstore.NewIdStore(bstore)
 	}
 	lsys := storeutil.LinkSystemForBlockstore(cfg.bstore)
 	tp := TestPeer{

pkg/retriever/bitswapretriever_test.go

@@ -17,7 +17,6 @@ import (
 	"github.com/ipfs/boxo/blockservice"
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
-	gstestutil "github.com/ipfs/go-graphsync/testutil"
 	exchange "github.com/ipfs/go-ipfs-exchange-interface"
 	"github.com/ipld/go-ipld-prime"
 	"github.com/ipld/go-ipld-prime/datamodel"
@@ -42,8 +41,8 @@ func TestBitswapRetriever(t *testing.T) {
 	lsys.SetReadStorage(store)
 	lsys.SetWriteStorage(store)
 	lsys.TrustedStorage = true
-	tbc1 := gstestutil.SetupBlockChain(ctx, t, lsys, 1000, 100)
-	tbc2 := gstestutil.SetupBlockChain(ctx, t, lsys, 1000, 100)
+	tbc1 := trustlesstestutil.SetupBlockChain(ctx, t, lsys, 1000, 100)
+	tbc2 := trustlesstestutil.SetupBlockChain(ctx, t, lsys, 1000, 100)
 	var tbc1Cids []cid.Cid
 	for _, blk := range tbc1.AllBlocks() {
 		tbc1Cids = append(tbc1Cids, blk.Cid())

pkg/retriever/httpretriever_test.go

@@ -18,7 +18,6 @@ import (
 	"github.com/google/uuid"
 	blocks "github.com/ipfs/go-block-format"
 	"github.com/ipfs/go-cid"
-	gstestutil "github.com/ipfs/go-graphsync/testutil"
 	dagpb "github.com/ipld/go-codec-dagpb"
 	"github.com/ipld/go-ipld-prime/datamodel"
 	"github.com/ipld/go-ipld-prime/fluent/qp"
@@ -47,8 +46,8 @@ func TestHTTPRetriever(t *testing.T) {
 	lsys.SetReadStorage(store)
 	lsys.SetWriteStorage(store)
 	lsys.TrustedStorage = true
-	tbc1 := gstestutil.SetupBlockChain(ctx, t, lsys, 1000, 100)
-	tbc2 := gstestutil.SetupBlockChain(ctx, t, lsys, 1000, 100)
+	tbc1 := trustlesstestutil.SetupBlockChain(ctx, t, lsys, 1000, 100)
+	tbc2 := trustlesstestutil.SetupBlockChain(ctx, t, lsys, 1000, 100)
 	var tbc1Cids []cid.Cid
 	for _, blk := range tbc1.AllBlocks() {
 		tbc1Cids = append(tbc1Cids, blk.Cid())

pkg/storage/duplicateaddercar.go

@@ -18,6 +18,7 @@ import (
 	ipldstorage "github.com/ipld/go-ipld-prime/storage"
 	trustlessutils "github.com/ipld/go-trustless-utils"
 	"github.com/ipld/go-trustless-utils/traversal"
+	"github.com/multiformats/go-multihash"
 )
 
 type DeferredWriter interface {
@@ -114,6 +115,13 @@ func (da *DuplicateAdderCar) BlockWriteOpener() linking.BlockWriteOpener {
 		var buf bytes.Buffer
 		var written bool
 		return &buf, func(lnk ipld.Link) error {
+			// assume that we will never want to write out identity CIDs, and they
+			// won't be expected by a traversal (i.e. go-trustless-utils won't expect
+			// them in a block stream)
+			if lnk.(cidlink.Link).Cid.Prefix().MhType == multihash.IDENTITY {
+				return nil
+			}
+
 			if written {
 				return fmt.Errorf("WriteCommitter already used")
 			}