Fix incorrect log messages (#151)

* Fix incorrect log messages
* Cache advertisement while processing entry chunks for advertisement to prevent reloading re-validating the same data.
* Sync datastore after saving ad mapping
* Log completion of entry block indexing at info level
* Index current chunk even if unable to set mapping for next chunk
* Update config defaults
* Populate defaults if config file missing values
* bump version

config/addresses.go

@@ -23,3 +23,21 @@ func NewAddresses() Addresses {
 		P2PAddr: "/ip4/0.0.0.0/tcp/3003",
 	}
 }
+
+// populateUnset replaces zero-values in the config with default values.
+func (c *Addresses) populateUnset() {
+	def := NewAddresses()
+
+	if c.Admin == "" {
+		c.Admin = def.Admin
+	}
+	if c.Finder == "" {
+		c.Finder = def.Finder
+	}
+	if c.Ingest == "" {
+		c.Ingest = def.Ingest
+	}
+	if c.P2PAddr == "" {
+		c.P2PAddr = def.P2PAddr
+	}
+}

config/config.go

@@ -98,6 +98,8 @@ func Load(filePath string) (*Config, error) {
 		return nil, err
 	}
 
+	cfg.populateUnset()
+
 	return &cfg, nil
 }
 
@@ -138,3 +140,11 @@ func (c *Config) String() string {
 	}
 	return string(b)
 }
+
+func (c *Config) populateUnset() {
+	c.Addresses.populateUnset()
+	c.Datastore.populateUnset()
+	c.Discovery.populateUnset()
+	c.Indexer.populateUnset()
+	c.Ingest.populateUnset()
+}

config/datastore.go

@@ -17,3 +17,15 @@ func NewDatastore() Datastore {
 		Dir:  "datastore",
 	}
 }
+
+// populateUnset replaces zero-values in the config with default values.
+func (c *Datastore) populateUnset() {
+	def := NewDatastore()
+
+	if c.Type == "" {
+		c.Type = def.Type
+	}
+	if c.Dir == "" {
+		c.Dir = def.Dir
+	}
+}

config/discovery.go

@@ -24,7 +24,8 @@ type Discovery struct {
 	// Values are a number ending in "s", "m", "h" for seconds. minutes, hours.
 	PollInterval Duration
 	// RediscoverWait is the amount of time that must pass before a provider
-	// can be discovered following a previous discovery attempt
+	// can be discovered following a previous discovery attempt.  A value of 0
+	// means there is no wait time.
 	RediscoverWait Duration
 	// Timeout is the maximum amount of time that the indexer will spend trying
 	// to discover and verify a new provider.
@@ -41,3 +42,15 @@ func NewDiscovery() Discovery {
 		Timeout:        Duration(2 * time.Minute),
 	}
 }
+
+// populateUnset replaces zero-values in the config with default values.
+func (c *Discovery) populateUnset() {
+	def := NewDiscovery()
+
+	if c.PollInterval == 0 {
+		c.PollInterval = def.PollInterval
+	}
+	if c.Timeout == 0 {
+		c.Timeout = def.Timeout
+	}
+}

config/indexer.go

@@ -19,3 +19,18 @@ func NewIndexer() Indexer {
 		ValueStoreType: "sth",
 	}
 }
+
+// populateUnset replaces zero-values in the config with default values.
+func (c *Indexer) populateUnset() {
+	def := NewIndexer()
+
+	if c.CacheSize == 0 {
+		c.CacheSize = def.CacheSize
+	}
+	if c.ValueStoreDir == "" {
+		c.ValueStoreDir = def.ValueStoreDir
+	}
+	if c.ValueStoreType == "" {
+		c.ValueStoreType = def.ValueStoreType
+	}
+}

config/ingest.go

@@ -7,7 +7,9 @@ type Ingest struct {
 	// PubSubTopic used to advertise ingestion announcements.
 	PubSubTopic string
 	// StoreBatchSize is the number of entries in each write to the value
-	// store.  Specifying a value less than 2 disables batching.
+	// store.  Specifying a value less than 2 disables batching.  This should
+	// be smaller than the maximum number of multihashes in an entry block to
+	// write concurrently to the value store.
 	StoreBatchSize int
 	// SyncTimeout is the maximum amount of time allowed for a sync to complete
 	// before it is canceled. This can be a sync of a chain of advertisements
@@ -24,3 +26,18 @@ func NewIngest() Ingest {
 		SyncTimeout:    Duration(2 * time.Hour),
 	}
 }
+
+// populateUnset replaces zero-values in the config with default values.
+func (c *Ingest) populateUnset() {
+	def := NewIngest()
+
+	if c.PubSubTopic == "" {
+		c.PubSubTopic = def.PubSubTopic
+	}
+	if c.StoreBatchSize == 0 {
+		c.StoreBatchSize = def.StoreBatchSize
+	}
+	if c.SyncTimeout == 0 {
+		c.SyncTimeout = def.SyncTimeout
+	}
+}

go.mod

@@ -17,7 +17,7 @@ require (
 	github.com/ipfs/go-datastore v0.5.1
 	github.com/ipfs/go-ds-leveldb v0.5.0
 	github.com/ipfs/go-ipfs v0.11.0
-	github.com/ipfs/go-log/v2 v2.4.0
+	github.com/ipfs/go-log/v2 v2.5.0
 	github.com/ipld/go-ipld-prime v0.14.3
 	github.com/libp2p/go-libp2p v0.17.0
 	github.com/libp2p/go-libp2p-core v0.13.0

go.sum

@@ -565,8 +565,9 @@ github.com/ipfs/go-log/v2 v2.0.5/go.mod h1:eZs4Xt4ZUJQFM3DlanGhy7TkwwawCZcSByscw
 github.com/ipfs/go-log/v2 v2.1.1/go.mod h1:2v2nsGfZsvvAJz13SyFzf9ObaqwHiHxsPLEHntrv9KM=
 github.com/ipfs/go-log/v2 v2.1.3/go.mod h1:/8d0SH3Su5Ooc31QlL1WysJhvyOTDCjcCZ9Axpmri6g=
 github.com/ipfs/go-log/v2 v2.3.0/go.mod h1:QqGoj30OTpnKaG/LKTGTxoP2mmQtjVMEnK72gynbe/g=
-github.com/ipfs/go-log/v2 v2.4.0 h1:iR/2o9PGWanVJrBgIH5Ff8mPGOwpqLaPIAFqSnsdlzk=
 github.com/ipfs/go-log/v2 v2.4.0/go.mod h1:nPZnh7Cj7lwS3LpRU5Mwr2ol1c2gXIEXuF6aywqrtmo=
+github.com/ipfs/go-log/v2 v2.5.0 h1:+MhAooFd9XZNvR0i9FriKW6HB0ql7HNXUuflWtc0dd4=
+github.com/ipfs/go-log/v2 v2.5.0/go.mod h1:prSpmC1Gpllc9UYWxDiZDreBYw7zp4Iqp1kOLU9U5UI=
 github.com/ipfs/go-merkledag v0.0.6/go.mod h1:QYPdnlvkOg7GnQRofu9XZimC5ZW5Wi3bKys/4GQQfto=
 github.com/ipfs/go-merkledag v0.2.3/go.mod h1:SQiXrtSts3KGNmgOzMICy5c0POOpUNQLvB3ClKnBAlk=
 github.com/ipfs/go-merkledag v0.2.4/go.mod h1:SQiXrtSts3KGNmgOzMICy5c0POOpUNQLvB3ClKnBAlk=

internal/ingest/ingest.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	indexer "github.com/filecoin-project/go-indexer-core/engine"
@@ -49,6 +50,9 @@ type Ingester struct {
 	syncTimeout   time.Duration
 	adLocks       *lockChain
 	watchDone     chan struct{}
+
+	adCache      map[cid.Cid]adCacheItem
+	adCacheMutex sync.Mutex
 }
 
 // NewIngester creates a new Ingester that uses a go-legs Subscriber to handle

internal/ingest/ingest_test.go

@@ -319,7 +319,7 @@ func publishRandomAdv(t *testing.T, i *Ingester, pubHost host.Host, pub legs.Pub
 		requireTrueEventually(t, func() bool {
 			has, err := i.ds.Has(context.Background(), datastore.NewKey(c.String()))
 			return err == nil && has
-		}, 2*time.Second, 15*time.Second, "expected advertisement with ID %s was not received", c)
+		}, 3*time.Second, 21*time.Second, "expected advertisement with ID %s was not received", c)
 	}
 
 	// Check if advertisement in datastore.