Always use basicnode.Prototype.Any as node prototype (#228)

Choosing a specific node prototype based on the cid schema hint adds significant code complexity without significant gain.

dagsync/announce_test.go

@@ -33,8 +33,7 @@ func TestAnnounceReplace(t *testing.T) {
 		blocksSeenByHook[c] = struct{}{}
 	}
 
-	sub, err := dagsync.NewSubscriber(dstHost, dstLnkS, dagsync.RecvAnnounce(testTopic),
-		dagsync.BlockHook(blockHook), dagsync.WithCidSchemaHint(false))
+	sub, err := dagsync.NewSubscriber(dstHost, dstLnkS, dagsync.RecvAnnounce(testTopic), dagsync.BlockHook(blockHook))
 	require.NoError(t, err)
 	defer sub.Close()
 
@@ -377,7 +376,7 @@ func initPubSub(t *testing.T, srcStore, dstStore datastore.Batching, allowPeer f
 	dstHost.Peerstore().AddAddrs(srcHost.ID(), srcHost.Addrs(), time.Hour)
 	dstLnkS := test.MkLinkSystem(dstStore)
 
-	sub, err := dagsync.NewSubscriber(dstHost, dstLnkS, dagsync.WithCidSchemaHint(false),
+	sub, err := dagsync.NewSubscriber(dstHost, dstLnkS,
 		dagsync.RecvAnnounce(testTopic, announce.WithTopic(topics[1]), announce.WithAllowPeer(allowPeer)))
 	require.NoError(t, err)
 

dagsync/ipnisync/publisher.go

@@ -13,11 +13,9 @@ import (
 	"github.com/ipfs/go-cid"
 	"github.com/ipld/go-ipld-prime"
 	"github.com/ipld/go-ipld-prime/codec/dagjson"
-	"github.com/ipld/go-ipld-prime/datamodel"
 	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
 	basicnode "github.com/ipld/go-ipld-prime/node/basic"
 	headschema "github.com/ipni/go-libipni/dagsync/ipnisync/head"
-	"github.com/ipni/go-libipni/ingest/schema"
 	"github.com/ipni/go-libipni/maurl"
 	ic "github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/libp2p/go-libp2p/core/peer"
@@ -229,7 +227,6 @@ func (p *Publisher) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	ipldCtx := ipld.LinkContext{}
 	reqType := r.Header.Get(CidSchemaHeader)
 	if reqType != "" {
-		log.Debug("sync request has cid schema type hint", "hint", reqType)
 		ipldCtx.Ctx, err = CtxWithCidSchema(context.Background(), reqType)
 		if err != nil {
 			// Log warning about unknown cid schema type, but continue on since
@@ -238,17 +235,7 @@ func (p *Publisher) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	var ipldProto datamodel.NodePrototype
-	switch reqType {
-	case CidSchemaAdvertisement:
-		ipldProto = schema.AdvertisementPrototype
-	case CidSchemaEntryChunk:
-		ipldProto = schema.EntryChunkPrototype
-	default:
-		ipldProto = basicnode.Prototype.Any
-	}
-
-	item, err := p.lsys.Load(ipldCtx, cidlink.Link{Cid: c}, ipldProto)
+	item, err := p.lsys.Load(ipldCtx, cidlink.Link{Cid: c}, basicnode.Prototype.Any)
 	if err != nil {
 		if errors.Is(err, ipld.ErrNotExists{}) {
 			http.Error(w, "cid not found", http.StatusNotFound)

dagsync/ipnisync/sync.go

@@ -22,7 +22,6 @@ import (
 	"github.com/ipld/go-ipld-prime/traversal"
 	"github.com/ipld/go-ipld-prime/traversal/selector"
 	headschema "github.com/ipni/go-libipni/dagsync/ipnisync/head"
-	"github.com/ipni/go-libipni/ingest/schema"
 	"github.com/ipni/go-libipni/maurl"
 	"github.com/ipni/go-libipni/mautil"
 	"github.com/libp2p/go-libp2p/core/network"
@@ -228,22 +227,12 @@ func (s *Syncer) Sync(ctx context.Context, nextCid cid.Cid, sel ipld.Node) error
 	}
 
 	// Check for valid cid schema type if set.
-	reqType, err := CidSchemaFromCtx(ctx)
+	_, err = CidSchemaFromCtx(ctx)
 	if err != nil {
 		return err
 	}
 
-	var ipldProto datamodel.NodePrototype
-	switch reqType {
-	case CidSchemaAdvertisement:
-		ipldProto = schema.AdvertisementPrototype
-	case CidSchemaEntryChunk:
-		ipldProto = schema.EntryChunkPrototype
-	default:
-		ipldProto = basicnode.Prototype.Any
-	}
-
-	cids, err := s.walkFetch(ctx, nextCid, xsel, ipldProto)
+	cids, err := s.walkFetch(ctx, nextCid, xsel)
 	if err != nil {
 		return fmt.Errorf("failed to traverse requested dag: %w", err)
 	}
@@ -269,7 +258,7 @@ func (s *Syncer) Sync(ctx context.Context, nextCid cid.Cid, sel ipld.Node) error
 // walkFetch is run by a traversal of the selector. For each block that the
 // selector walks over, walkFetch will look to see if it can find it in the
 // local data store. If it cannot, it will then go and get it over HTTP.
-func (s *Syncer) walkFetch(ctx context.Context, rootCid cid.Cid, sel selector.Selector, ipldProto datamodel.NodePrototype) ([]cid.Cid, error) {
+func (s *Syncer) walkFetch(ctx context.Context, rootCid cid.Cid, sel selector.Selector) ([]cid.Cid, error) {
 	// Track the order of cids seen during traversal so that the block hook
 	// function gets called in the same order.
 	var traversalOrder []cid.Cid
@@ -280,7 +269,7 @@ func (s *Syncer) walkFetch(ctx context.Context, rootCid cid.Cid, sel selector.Se
 	getMissingLs.StorageReadOpener = func(lc ipld.LinkContext, l ipld.Link) (io.Reader, error) {
 		c := l.(cidlink.Link).Cid
 		// fetchBlock checks if the node is already present in storage.
-		err := s.fetchBlock(ctx, c, ipldProto)
+		err := s.fetchBlock(ctx, c)
 		if err != nil {
 			return nil, fmt.Errorf("failed to fetch block for cid %s: %w", c, err)
 		}
@@ -302,7 +291,7 @@ func (s *Syncer) walkFetch(ctx context.Context, rootCid cid.Cid, sel selector.Se
 	}
 
 	// get the direct node.
-	rootNode, err := getMissingLs.Load(ipld.LinkContext{Ctx: ctx}, cidlink.Link{Cid: rootCid}, ipldProto)
+	rootNode, err := getMissingLs.Load(ipld.LinkContext{Ctx: ctx}, cidlink.Link{Cid: rootCid}, basicnode.Prototype.Any)
 	if err != nil {
 		return nil, fmt.Errorf("failed to load node for root cid %s: %w", rootCid, err)
 	}
@@ -385,8 +374,8 @@ retry:
 }
 
 // fetchBlock fetches an item into the datastore at c if not locally available.
-func (s *Syncer) fetchBlock(ctx context.Context, c cid.Cid, ipldProto datamodel.NodePrototype) error {
-	n, err := s.sync.lsys.Load(ipld.LinkContext{Ctx: ctx}, cidlink.Link{Cid: c}, ipldProto)
+func (s *Syncer) fetchBlock(ctx context.Context, c cid.Cid) error {
+	n, err := s.sync.lsys.Load(ipld.LinkContext{Ctx: ctx}, cidlink.Link{Cid: c}, basicnode.Prototype.Any)
 	// node is already present.
 	if n != nil && err == nil {
 		return nil

dagsync/option.go

@@ -48,7 +48,6 @@ type config struct {
 	firstSyncDepth    int64
 	segDepthLimit     int64
 
-	cidSchemaHint   bool
 	strictAdsSelSeq bool
 
 	httpTimeout      time.Duration
@@ -67,7 +66,6 @@ func getOpts(opts []Option) (config, error) {
 		httpTimeout:     defaultHttpTimeout,
 		idleHandlerTTL:  defaultIdleHandlerTTL,
 		segDepthLimit:   defaultSegDepthLimit,
-		cidSchemaHint:   true,
 		strictAdsSelSeq: true,
 	}
 
@@ -341,10 +339,3 @@ func MakeGeneralBlockHook(prevAdCid func(adCid cid.Cid) (cid.Cid, error)) BlockH
 		}
 	}
 }
-
-func WithCidSchemaHint(enable bool) Option {
-	return func(c *config) error {
-		c.cidSchemaHint = enable
-		return nil
-	}
-}

dagsync/subscriber.go

@@ -105,10 +105,6 @@ type Subscriber struct {
 
 	receiver *announce.Receiver
 
-	// cidSchemaHint enables sending the cid schema type hint as
-	// an HTTP header in sync requests.
-	cidSchemaHint bool
-
 	// Track explicit Sync calls in progress and allow them to complete before
 	// closing subscriber.
 	expSyncClosed bool
@@ -240,8 +236,6 @@ func NewSubscriber(host host.Host, lsys ipld.LinkSystem, options ...Option) (*Su
 			ssb.ExploreFields(func(efsb builder.ExploreFieldsSpecBuilder) {
 				efsb.Insert("Next", ssb.ExploreRecursiveEdge()) // Next field in EntryChunk
 			})).Node(),
-
-		cidSchemaHint: opts.cidSchemaHint,
 	}
 
 	if opts.strictAdsSelSeq {
@@ -250,7 +244,6 @@ func NewSubscriber(host host.Host, lsys ipld.LinkSystem, options ...Option) (*Su
 		}).Node()
 	} else {
 		s.adsSelectorSeq = ssb.ExploreAll(ssb.ExploreRecursiveEdge()).Node()
-		s.cidSchemaHint = false
 	}
 
 	if opts.hasRcvr {
@@ -495,11 +488,9 @@ func (s *Subscriber) SyncAdChain(ctx context.Context, peerInfo peer.AddrInfo, op
 
 	sel := ExploreRecursiveWithStopNode(depthLimit, s.adsSelectorSeq, stopLnk)
 
-	if s.cidSchemaHint {
-		ctx, err = ipnisync.CtxWithCidSchema(ctx, ipnisync.CidSchemaAdvertisement)
-		if err != nil {
-			panic(err.Error())
-		}
+	ctx, err = ipnisync.CtxWithCidSchema(ctx, ipnisync.CidSchemaAdvertisement)
+	if err != nil {
+		panic(err.Error())
 	}
 	syncCount, err := hnd.handle(ctx, nextCid, sel, syncer, opts.blockHook, segdl, stopAtCid)
 	if err != nil {
@@ -584,11 +575,9 @@ func (s *Subscriber) syncEntries(ctx context.Context, peerInfo peer.AddrInfo, en
 
 	log.Debugw("Start entries sync", "peer", peerInfo.ID, "cid", entCid)
 
-	if s.cidSchemaHint {
-		ctx, err = ipnisync.CtxWithCidSchema(ctx, ipnisync.CidSchemaEntryChunk)
-		if err != nil {
-			panic(err.Error())
-		}
+	ctx, err = ipnisync.CtxWithCidSchema(ctx, ipnisync.CidSchemaEntryChunk)
+	if err != nil {
+		panic(err.Error())
 	}
 	_, err = hnd.handle(ctx, entCid, sel, syncer, bh, segdl, cid.Undef)
 	if err != nil {
@@ -891,11 +880,9 @@ func (h *handler) asyncSyncAdChain(ctx context.Context) {
 		return
 	}
 
-	if h.subscriber.cidSchemaHint {
-		ctx, err = ipnisync.CtxWithCidSchema(ctx, ipnisync.CidSchemaAdvertisement)
-		if err != nil {
-			panic(err.Error())
-		}
+	ctx, err = ipnisync.CtxWithCidSchema(ctx, ipnisync.CidSchemaAdvertisement)
+	if err != nil {
+		panic(err.Error())
 	}
 	sel := ExploreRecursiveWithStopNode(adsDepthLimit, h.subscriber.adsSelectorSeq, latestSyncLink)
 	syncCount, err := h.handle(ctx, nextCid, sel, syncer, h.subscriber.generalBlockHook, h.subscriber.segDepthLimit, stopAtCid)