Process ad chain in segments during GC.

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/cockroachdb/pebble v0.0.0-20220726144858-a78491c0086f
 	github.com/filecoin-project/go-dagaggregator-unixfs v0.3.0
 	github.com/gammazero/channelqueue v0.2.1
+	github.com/gammazero/deque v0.2.1
 	github.com/gammazero/targz v0.0.3
 	github.com/ipfs/boxo v0.15.0
 	github.com/ipfs/go-cid v0.4.1
@@ -89,7 +90,6 @@ require (
 	github.com/flynn/noise v1.0.0 // indirect
 	github.com/francoispqt/gojay v1.2.13 // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
-	github.com/gammazero/deque v0.2.1 // indirect
 	github.com/gammazero/radixtree v0.3.1 // indirect
 	github.com/go-kit/log v0.2.1 // indirect
 	github.com/go-logfmt/logfmt v0.5.1 // indirect

ipni-gc/cmd/ipnigc/daemon.go

@@ -119,7 +119,6 @@ func daemonAction(cctx *cli.Context) error {
 		reaper.WithCarCompress(cfgMirror.Compress),
 		reaper.WithCarDelete(cfgMirror.Write),
 		reaper.WithCarRead(true),
-		reaper.WithCommit(true),
 		reaper.WithDatastoreDir(dsDir),
 		reaper.WithDatastoreTempDir(dsTmpDir),
 		reaper.WithPCache(pc),

ipni-gc/cmd/ipnigc/provider.go

@@ -34,11 +34,6 @@ var providerFlags = []cli.Flag{
 		Aliases: []string{"b"},
 		Value:   1024,
 	},
-	&cli.BoolFlag{
-		Name:    "commit",
-		Usage:   "Commit changes to storage if set. Otherwise, only report what GC would have deleted.",
-		Aliases: []string{"w"},
-	},
 	&cli.BoolFlag{
 		Name:    "delete-not-found",
 		Usage:   "Delete all provider indexes if provider is not found",
@@ -72,6 +67,12 @@ var providerFlags = []cli.Flag{
 		Usage:   "Set log level for other loggers that are not ipni-gc",
 		Value:   "error",
 	},
+	&cli.IntFlag{
+		Name:    "segment-size",
+		Usage:   "Set advertisement chain segment size. This specifies how many ads to process at a time.",
+		Aliases: []string{"ss"},
+		Value:   16384,
+	},
 }
 
 func providerAction(cctx *cli.Context) error {
@@ -150,12 +151,12 @@ func providerAction(cctx *cli.Context) error {
 		reaper.WithCarCompress(cfgMirror.Compress),
 		reaper.WithCarDelete(cfgMirror.Write),
 		reaper.WithCarRead(true),
-		reaper.WithCommit(cctx.Bool("commit")),
 		reaper.WithDatastoreDir(dsDir),
 		reaper.WithDatastoreTempDir(dsTmpDir),
 		reaper.WithDeleteNotFound(cctx.Bool("delete-not-found")),
 		reaper.WithEntriesFromPublisher(cctx.Bool("ents-from-pub")),
 		reaper.WithPCache(pc),
+		reaper.WithSegmentSize(cctx.Int("segment-size")),
 		reaper.WithTopicName(cfg.Ingest.PubSubTopic),
 		reaper.WithHttpTimeout(time.Duration(cfg.Ingest.HttpSyncTimeout)),
 	)
@@ -164,11 +165,7 @@ func providerAction(cctx *cli.Context) error {
 	}
 	defer grim.Close()
 
-	if cctx.Bool("commit") {
-		fmt.Println("Starting IPNI GC, committing changes")
-	} else {
-		fmt.Println("Starting IPNI GC, dry-run - GC progress and changes will not be saved")
-	}
+	fmt.Println("Starting IPNI GC")
 
 	var gcCount int
 	for _, pid := range peerIDs {

ipni-gc/reaper/option.go

@@ -12,14 +12,14 @@ import (
 
 const (
 	defaultHttpTimeout = 10 * time.Second
+	defaultSegmentSize = 16384
 	defaultTopic       = "/indexer/ingest/mainnet"
 )
 
 type config struct {
 	carCompAlg        string
 	carDelete         bool
 	carRead           bool
-	commit            bool
 	deleteNotFound    bool
 	dstoreDir         string
 	dstoreTmpDir      string
@@ -28,6 +28,7 @@ type config struct {
 	httpTimeout       time.Duration
 	p2pHost           host.Host
 	pcache            *pcache.ProviderCache
+	segmentSize       int
 	topic             string
 }
 
@@ -41,6 +42,7 @@ func getOpts(opts []Option) (config, error) {
 		carRead:     true,
 		entsFromPub: true,
 		httpTimeout: defaultHttpTimeout,
+		segmentSize: defaultSegmentSize,
 		topic:       defaultTopic,
 	}
 
@@ -80,22 +82,17 @@ func WithCarDelete(del bool) Option {
 	}
 }
 
-// WithCommit tells GC to commit changes to storage. Otherwise, GC only reports
-// information about what would have been collected.
-func WithCommit(commit bool) Option {
-	return func(c *config) error {
-		c.commit = commit
-		return nil
-	}
-}
-
+// WithDatastoreDir tells GC the directory to use as the parent for all
+// provider-specific datastores.
 func WithDatastoreDir(dir string) Option {
 	return func(c *config) error {
 		c.dstoreDir = dir
 		return nil
 	}
 }
 
+// WithDatastoreTempDir tells GC the directory to use as the parent for all
+// provider-specific temproary datastores.
 func WithDatastoreTempDir(dir string) Option {
 	return func(c *config) error {
 		c.dstoreTmpDir = dir
@@ -137,6 +134,17 @@ func WithPCache(pc *pcache.ProviderCache) Option {
 	}
 }
 
+// WithSegmentSize sets the size of the segments that the ad chain is broken
+// into for processing.
+func WithSegmentSize(size int) Option {
+	return func(c *config) error {
+		if size > 0 {
+			c.segmentSize = size
+		}
+		return nil
+	}
+}
+
 // WithTopicName sets the topic name on which the provider announces advertised
 // content. Defaults to '/indexer/ingest/mainnet'.
 func WithTopicName(topic string) Option {