config: add and wire datastore.WriteThrough option

config/datastore.go

@@ -11,6 +11,13 @@ const (
 	// DefaultBlockKeyCacheSize is the size for the blockstore two-queue
 	// cache which caches block keys and sizes.
 	DefaultBlockKeyCacheSize = 64 << 10
+
+	// DefaultWriteThrough specifies whether to use a "write-through"
+	// Blockstore and Blockservice. This means that they will write
+	// without performing any reads to check if the incoming blocks are
+	// already present in the datastore. Enable for datastores with fast
+	// writes and slower reads.
+	DefaultWriteThrough = true
 )
 
 // Datastore tracks the configuration of the datastore.
@@ -30,6 +37,7 @@ type Datastore struct {
 	HashOnRead        bool
 	BloomFilterSize   int
 	BlockKeyCacheSize OptionalInteger
+	WriteThrough      bool
 }
 
 // DataStorePath returns the default data store path given a configuration root

config/import.go

@@ -5,8 +5,15 @@ const (
 	DefaultUnixFSRawLeaves = false
 	DefaultUnixFSChunker   = "size-262144"
 	DefaultHashFunction    = "sha2-256"
-	DefaultBatchMaxNodes   = 128
-	DefaultBatchMaxSize    = 100 << 20 // 20MiB
+
+	// DefaultBatchMaxNodes controls the maximum number of nodes in a
+	// write-batch. The total size of the batch is limited by
+	// BatchMaxnodes and BatchMaxSize.
+	DefaultBatchMaxNodes = 128
+	// DefaultBatchMaxSize controls the maximum size of a single
+	// write-batch. The total size of the batch is limited by
+	// BatchMaxnodes and BatchMaxSize.
+	DefaultBatchMaxSize = 100 << 20 // 20MiB
 )
 
 // Import configures the default options for ingesting data. This affects commands

config/init.go

@@ -135,6 +135,7 @@ func DefaultDatastoreConfig() Datastore {
 		GCPeriod:           "1h",
 		BloomFilterSize:    0,
 		Spec:               flatfsSpec(),
+		WriteThrough:       DefaultWriteThrough,
 	}
 }
 

core/coreapi/coreapi.go

@@ -245,8 +245,7 @@ func (api *CoreAPI) WithOptions(opts ...options.ApiOption) (coreiface.CoreAPI, e
 	if settings.Offline || !settings.FetchBlocks {
 		subAPI.exchange = offlinexch.Exchange(subAPI.blockstore)
 		var bsopts []bserv.Option
-		// If bloom filter disable, do not do Has() when writing.
-		if cfg.Datastore.BloomFilterSize == 0 {
+		if cfg.Datastore.WriteThrough {
 			bsopts = append(bsopts, bserv.WriteThrough())
 		}
 		subAPI.blocks = bserv.New(subAPI.blockstore, subAPI.exchange, bsopts...)

core/coreapi/unixfs.go

@@ -92,8 +92,7 @@ func (api *UnixfsAPI) Add(ctx context.Context, files files.Node, opts ...options
 	}
 
 	var bsopts []blockservice.Option
-	// If bloom filter disabled, do not do Has() when writing.
-	if cfg.Datastore.BloomFilterSize == 0 {
+	if cfg.Datastore.WriteThrough {
 		bsopts = append(bsopts, blockservice.WriteThrough())
 	}
 	bserv := blockservice.New(addblockstore, exch, bsopts...) // hash security 001

core/node/core.go

@@ -33,11 +33,7 @@ import (
 func BlockService(cfg *config.Config) func(lc fx.Lifecycle, bs blockstore.Blockstore, rem exchange.Interface) blockservice.BlockService {
 	return func(lc fx.Lifecycle, bs blockstore.Blockstore, rem exchange.Interface) blockservice.BlockService {
 		var opts []blockservice.Option
-		// If bloom filter is disabled, do not do Has() when writing.
-		// We defer to the datastore how to handle this efficiently,
-		// but we cannot assume that triggering Reads for every white
-		// is fine.
-		if cfg.Datastore.BloomFilterSize == 0 {
+		if cfg.Datastore.WriteThrough {
 			opts = append(opts, blockservice.WriteThrough())
 		}
 		bsvc := blockservice.New(bs, rem, opts...)

core/node/groups.go

@@ -202,7 +202,7 @@ func Storage(bcfg *BuildCfg, cfg *config.Config) fx.Option {
 	return fx.Options(
 		fx.Provide(RepoConfig),
 		fx.Provide(Datastore),
-		fx.Provide(BaseBlockstoreCtor(cacheOpts, cfg.Datastore.HashOnRead, cfg.Datastore.BloomFilterSize == 0)),
+		fx.Provide(BaseBlockstoreCtor(cacheOpts, cfg.Datastore.HashOnRead, cfg.Datastore.WriteThrough)),
 		finalBstore,
 	)
 }