apply suggestions

Author: LexLuthr

deps/config/doc_gen.go

@@ -95,10 +95,9 @@ func DefaultCurioConfig() *CurioConfig {
 					ExpectedSnapSealDuration:  Duration(2 * time.Hour),
 				},
 				IPNI: IPNIConfig{
-					Enable:             true,
-					TopicName:          "",
-					WebHost:            "https://cid.contact",
-					DirectAnnounceURLs: []string{"https://cid.contact/ingest/announce"},
+					EntriesCacheCapacity: 4096,
+					WebHost:              "https://cid.contact",
+					DirectAnnounceURLs:   []string{"https://cid.contact/ingest/announce"},
 				},
 			},
 		},
@@ -668,17 +667,20 @@ type Libp2pConfig struct {
 }
 
 type IPNIConfig struct {
-	// Enable set whether to enable indexing announcement to the network and expose endpoints that
-	// allow indexer nodes to process announcements. Enabled by default.
-	Enable bool
-
-	// TopicName sets the topic name on which the changes to the advertised content are announced.
-	// If not explicitly specified, the topic name is automatically inferred from the network name
-	// in following format: '/indexer/ingest/<network-name>'
-	// Defaults to empty, which implies the topic name is inferred from network name.
-	TopicName string
+	// Disable set whether to disable indexing announcement to the network and expose endpoints that
+	// allow indexer nodes to process announcements. Default: False
+	Disable bool
+
+	// EntriesCacheCapacity sets the maximum capacity to use for caching the indexing advertisement
+	// entries. Defaults to 4096 if not specified. The cache is evicted using LRU policy. The
+	// maximum storage used by the cache is a factor of EntriesCacheCapacity, EntriesChunkSize(16384) and
+	// the length of multihashes being advertised. For example, advertising 128-bit long multihashes
+	// with the default EntriesCacheCapacity, and EntriesChunkSize(16384) means the cache size can grow to
+	// 1GiB when full.
+	EntriesCacheCapacity int
 
 	// The network indexer host that the web UI should link to for published announcements
+	// TODO: should we use this for checking published heas before publishing? Later commit
 	WebHost string
 
 	// The list of URLs of indexing nodes to announce to.

deps/config/types.go

@@ -229,17 +229,13 @@ description: The default curio configuration
   # type: bool
   #EnableBatchSeal = false
 
-<<<<<<< HEAD
   # EnableDealMarket enabled the deal market on the node. This would also enable libp2p on the node, if configured.
   #
   # type: bool
   #EnableDealMarket = false
 
   # EnableCommP enables the commP task on te node. CommP is calculated before sending PublishDealMessage for a Mk12 deal
   # Must have EnableDealMarket = True
-=======
-  # EnableCommP enabled the commP task on te node. CommP is calculated before sending PublishDealMessage for a Mk12 deal
->>>>>>> 563f95c (ipni provider)
   #
   # type: bool
   #EnableCommP = false
@@ -436,21 +432,24 @@ description: The default curio configuration
       #InsertConcurrency = 8
 
     [Market.StorageMarketConfig.IPNI]
-      # Enable set whether to enable indexing announcement to the network and expose endpoints that
-      # allow indexer nodes to process announcements. Enabled by default.
+      # Disable set whether to disable indexing announcement to the network and expose endpoints that
+      # allow indexer nodes to process announcements. Default: False
       #
       # type: bool
-      #Enable = true
-
-      # TopicName sets the topic name on which the changes to the advertised content are announced.
-      # If not explicitly specified, the topic name is automatically inferred from the network name
-      # in following format: '/indexer/ingest/<network-name>'
-      # Defaults to empty, which implies the topic name is inferred from network name.
+      #Disable = false
+
+      # EntriesCacheCapacity sets the maximum capacity to use for caching the indexing advertisement
+      # entries. Defaults to 4096 if not specified. The cache is evicted using LRU policy. The
+      # maximum storage used by the cache is a factor of EntriesCacheCapacity, EntriesChunkSize(16384) and
+      # the length of multihashes being advertised. For example, advertising 128-bit long multihashes
+      # with the default EntriesCacheCapacity, and EntriesChunkSize(16384) means the cache size can grow to
+      # 1GiB when full.
       #
-      # type: string
-      #TopicName = ""
+      # type: int
+      #EntriesCacheCapacity = 4096
 
       # The network indexer host that the web UI should link to for published announcements
+      # TODO: should we use this for checking published heas before publishing? Later commit
       #
       # type: string
       #WebHost = "https://cid.contact"

documentation/en/configuration/default-curio-configuration.md

@@ -20,8 +20,7 @@ import (
 
 var log = logging.Logger("chunker")
 
-const EntriesChunkSize = 16384
-const EntriesCacheCapacity = 4096
+const entriesChunkSize = 16384
 
 // Chunker chunks advertisement entries as a chained series of schema.EntryChunk nodes.
 // See: NewChunker
@@ -36,13 +35,10 @@ type Chunker struct {
 //
 // See: schema.EntryChunk.
 func NewChunker(cache *lru.Cache[ipld.Link, datamodel.Node]) *Chunker {
-	chunker := &Chunker{
-		chunkSize: EntriesChunkSize,
+	return &Chunker{
+		chunkSize: entriesChunkSize,
+		cache:     cache,
 	}
-	if cache != nil {
-		chunker.cache = cache
-	}
-	return chunker
 }
 
 // Chunk chunks all the mulithashes returned by the given iterator into a chain of schema.EntryChunk
@@ -63,7 +59,6 @@ func (ls *Chunker) Chunk(mhi SliceMhIterator) (ipld.Link, error) {
 			return nil, err
 		}
 		mhs = append(mhs, mh)
-		mhCount++
 		if len(mhs) >= ls.chunkSize {
 			cNode, err := newEntriesChunkNode(mhs, next)
 			if err != nil {
@@ -77,6 +72,7 @@ func (ls *Chunker) Chunk(mhi SliceMhIterator) (ipld.Link, error) {
 				ls.cache.Add(next, cNode)
 			}
 			chunkCount++
+			mhCount += len(mhs)
 			// NewLinkedListOfMhs makes it own copy, so safe to reuse mhs
 			mhs = mhs[:0]
 		}
@@ -91,6 +87,7 @@ func (ls *Chunker) Chunk(mhi SliceMhIterator) (ipld.Link, error) {
 			return nil, err
 		}
 		chunkCount++
+		mhCount += len(mhs)
 	}
 
 	log.Infow("Generated linked chunks of multihashes", "totalMhCount", mhCount, "chunkCount", chunkCount)

lib/ipni/chunker/chunker.go

@@ -47,6 +47,7 @@ import (
 const IPNIRoutePath = "/ipni-provider"
 const IPNIPath = "/ipni/v1/ad/"
 const ProviderPath = "/ipni-provider"
+const publishInterval = 10 * time.Minute
 
 var (
 	log         = logging.Logger("ipni-provider")
@@ -68,7 +69,6 @@ type Provider struct {
 	pieceProvider  *pieceprovider.PieceProvider
 	entriesChunker *chunker.Chunker
 	cache          *lru.Cache[ipld.Link, datamodel.Node]
-	topic          string
 	httpPrefix     string
 	keys           map[string]*peerInfo // map[peerID String]Private_Key
 	// announceURLs enables sending direct announcements via HTTP. This is
@@ -78,7 +78,7 @@ type Provider struct {
 }
 
 func NewProvider(api ipniAPI, deps *deps.Deps) (*Provider, error) {
-	c, err := lru.New[ipld.Link, datamodel.Node](chunker.EntriesCacheCapacity)
+	c, err := lru.New[ipld.Link, datamodel.Node](deps.Cfg.Market.StorageMarketConfig.IPNI.EntriesCacheCapacity)
 	if err != nil {
 		return nil, xerrors.Errorf("creating new cache: %w", err)
 	}
@@ -121,30 +121,19 @@ func NewProvider(api ipniAPI, deps *deps.Deps) (*Provider, error) {
 		return nil, err
 	}
 
-	nn, err := api.StateNetworkName(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	topic := "/indexer/ingest/" + string(nn)
-
-	if deps.Cfg.Market.StorageMarketConfig.IPNI.TopicName != "" {
-		topic = deps.Cfg.Market.StorageMarketConfig.IPNI.TopicName
-	}
-
-	var announceURLs []*url.URL
+	announceURLs := make([]*url.URL, 0, len(deps.Cfg.Market.StorageMarketConfig.IPNI.DirectAnnounceURLs))
 
-	for _, us := range deps.Cfg.Market.StorageMarketConfig.IPNI.DirectAnnounceURLs {
+	for i, us := range deps.Cfg.Market.StorageMarketConfig.IPNI.DirectAnnounceURLs {
 		u, err := url.Parse(us)
 		if err != nil {
 			return nil, err
 		}
-		announceURLs = append(announceURLs, u)
+		announceURLs[i] = u
 	}
 
-	var httpServerAddresses []multiaddr.Multiaddr
+	httpServerAddresses := make([]multiaddr.Multiaddr, 0, len(deps.Cfg.Market.HTTP.AnnounceAddresses))
 
-	for _, a := range deps.Cfg.Market.HTTP.AnnounceAddresses {
+	for i, a := range deps.Cfg.Market.HTTP.AnnounceAddresses {
 		addr, err := urltomultiaddr.UrlToMultiaddr(a)
 		if err != nil {
 			return nil, err
@@ -153,7 +142,7 @@ func NewProvider(api ipniAPI, deps *deps.Deps) (*Provider, error) {
 		if err != nil {
 			return nil, err
 		}
-		httpServerAddresses = append(httpServerAddresses, addr)
+		httpServerAddresses[i] = addr
 	}
 
 	return &Provider{
@@ -162,7 +151,6 @@ func NewProvider(api ipniAPI, deps *deps.Deps) (*Provider, error) {
 		pieceProvider:       deps.PieceProvider,
 		cache:               c,
 		keys:                keyMap,
-		topic:               topic,
 		announceURLs:        announceURLs,
 		httpServerAddresses: httpServerAddresses,
 	}, nil
@@ -179,9 +167,17 @@ func (p *Provider) getAd(ctx context.Context, ad cid.Cid, provider string) (sche
 		IsRm       bool
 	}
 
-	err := p.db.Select(ctx, &ads, `SELECT context_id,
-       is_rm, previous, provider, addresses, 
-       signature, entries FROM ipni WHERE ad_cid = $1 AND provider = $2`, ad.String(), provider)
+	err := p.db.Select(ctx, &ads, `SELECT 
+										context_id,
+										is_rm, 
+										previous, 
+										provider, 
+										addresses, 
+										signature, 
+										entries 
+										FROM ipni 
+										WHERE ad_cid = $1 
+										  AND provider = $2`, ad.String(), provider)
 
 	if err != nil {
 		return schema.Advertisement{}, xerrors.Errorf("getting ad from DB: %w", err)
@@ -256,7 +252,7 @@ func (p *Provider) getHead(ctx context.Context, provider string) ([]byte, error)
 		return nil, err
 	}
 
-	signedHead, err := head.NewSignedHead(lnk.(cidlink.Link).Cid, p.topic, p.keys[provider].Key)
+	signedHead, err := head.NewSignedHead(lnk.(cidlink.Link).Cid, "", p.keys[provider].Key)
 	if err != nil {
 		return nil, xerrors.Errorf("failed to generate signed head for peer %s: %w", provider, err)
 	}
@@ -461,7 +457,7 @@ func (p *Provider) handleGet(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 			log.Errorf("failed to get signed head for peer %s: %w", providerID, err)
-			http.Error(w, err.Error(), http.StatusInternalServerError)
+			http.Error(w, "", http.StatusInternalServerError)
 		}
 		w.WriteHeader(http.StatusOK)
 		_, err = w.Write(sh)
@@ -487,7 +483,7 @@ func (p *Provider) handleGet(w http.ResponseWriter, r *http.Request) {
 						return
 					}
 					log.Errorf("failed to get entry %s for peer %s: %w", b.String(), providerID, err)
-					http.Error(w, err.Error(), http.StatusInternalServerError)
+					http.Error(w, "", http.StatusInternalServerError)
 					return
 				}
 				w.WriteHeader(http.StatusOK)
@@ -498,21 +494,21 @@ func (p *Provider) handleGet(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 			log.Errorf("failed to get ad %s for peer %s: %w", b.String(), providerID, err)
-			http.Error(w, err.Error(), http.StatusInternalServerError)
+			http.Error(w, "", http.StatusInternalServerError)
 			return
 		}
 		adn, err := ad.ToNode()
 		if err != nil {
 			log.Errorf("failed to convert ad %s for peer %s to IPLD node: %w", b.String(), providerID, err)
-			http.Error(w, err.Error(), http.StatusInternalServerError)
+			http.Error(w, "", http.StatusInternalServerError)
 			return
 		}
 		// Use local buffer for better error handing
 		resp := new(bytes.Buffer)
 		err = dagjson.Encode(adn, resp)
 		if err != nil {
 			log.Errorf("failed to encode ad %s for peer %s: %w", b.String(), providerID, err)
-			http.Error(w, err.Error(), http.StatusInternalServerError)
+			http.Error(w, "", http.StatusInternalServerError)
 			return
 		}
 		w.WriteHeader(http.StatusOK)
@@ -535,7 +531,7 @@ func Routes(r *mux.Router, p *Provider) {
 func (p *Provider) StartPublishing(ctx context.Context) {
 	// A poller which publishes head for each provider
 	// every 10 minutes
-	ticker := time.NewTicker(10 * time.Minute)
+	ticker := time.NewTicker(publishInterval)
 	go func() {
 		for {
 			select {