Del after read (#139)

* Add option to remove items from store after reading them

Prevents OOM when reading large datasets and using in-memory datastore.

* Set delAfterRead in client store

* increase crawl batch size

Author: gammazero

pkg/adpub/client.go

@@ -18,7 +18,10 @@ import (
 	"github.com/libp2p/go-libp2p/core/peer"
 )
 
-const syncSegmentSize = 2048
+const (
+	crawlBatchSize  = 16
+	syncSegmentSize = 2048
+)
 
 type Client interface {
 	GetAdvertisement(context.Context, cid.Cid) (*Advertisement, error)
@@ -72,7 +75,7 @@ func NewClient(addrInfo peer.AddrInfo, options ...Option) (Client, error) {
 		ownsHost:  ownsHost,
 		topic:     opts.topic,
 
-		store: newClientStore(),
+		store: newClientStore(opts.delAfterRead),
 	}
 
 	c.sub, err = dagsync.NewSubscriber(c.host, c.store.Batching, c.store.LinkSystem, c.topic, dagsync.HttpTimeout(opts.httpTimeout))
@@ -105,8 +108,6 @@ func (c *client) List(ctx context.Context, latestCid cid.Cid, n int, w io.Writer
 }
 
 func (c *client) Crawl(ctx context.Context, latestCid cid.Cid, n int, ads chan<- *Advertisement) error {
-	const batchSize = 10
-
 	var opts []dagsync.SyncOption
 	if n > syncSegmentSize {
 		prevAdCid := func(adCid cid.Cid) (cid.Cid, error) {
@@ -124,10 +125,10 @@ func (c *client) Crawl(ctx context.Context, latestCid cid.Cid, n int, ads chan<-
 	if n == 0 {
 		n = -1
 	}
-	batch := batchSize
+	batch := crawlBatchSize
 	for n != 0 {
 		if n != -1 {
-			if n < batchSize {
+			if n < crawlBatchSize {
 				batch = n
 			}
 			n -= batch

pkg/adpub/client_store.go

@@ -11,7 +11,6 @@ import (
 	"github.com/ipld/go-ipld-prime"
 	"github.com/ipld/go-ipld-prime/linking"
 	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
-	"github.com/ipld/go-ipld-prime/multicodec"
 	"github.com/ipni/go-libipni/ingest/schema"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"github.com/multiformats/go-multihash"
@@ -24,6 +23,8 @@ import (
 type ClientStore struct {
 	datastore.Batching
 	ipld.LinkSystem
+
+	delAfterRead bool
 }
 
 // Advertisement contains information about a schema.Advertisement
@@ -47,7 +48,7 @@ func (a *Advertisement) HasEntries() bool {
 	return a.Entries != nil && a.Entries.IsPresent()
 }
 
-func newClientStore() *ClientStore {
+func newClientStore(delAfterRead bool) *ClientStore {
 	store := dssync.MutexWrap(datastore.NewMapDatastore())
 	lsys := cidlink.DefaultLinkSystem()
 	lsys.StorageReadOpener = func(lctx ipld.LinkContext, lnk ipld.Link) (io.Reader, error) {
@@ -66,8 +67,9 @@ func newClientStore() *ClientStore {
 		}, nil
 	}
 	return &ClientStore{
-		Batching:   store,
-		LinkSystem: lsys,
+		Batching:     store,
+		LinkSystem:   lsys,
+		delAfterRead: delAfterRead,
 	}
 }
 
@@ -92,6 +94,9 @@ func (s *ClientStore) getEntriesChunk(ctx context.Context, target cid.Cid) (cid.
 	if err != nil {
 		return cid.Undef, nil, err
 	}
+	if s.delAfterRead {
+		s.Batching.Delete(ctx, datastore.NewKey(target.String()))
+	}
 
 	chunk, err := schema.UnwrapEntryChunk(n)
 	if err != nil {
@@ -108,10 +113,14 @@ func (s *ClientStore) getEntriesChunk(ctx context.Context, target cid.Cid) (cid.
 }
 
 func (s *ClientStore) loadAd(ctx context.Context, id cid.Cid) (schema.Advertisement, error) {
-	val, err := s.Batching.Get(ctx, datastore.NewKey(id.String()))
+	dsKey := datastore.NewKey(id.String())
+	val, err := s.Batching.Get(ctx, dsKey)
 	if err != nil {
 		return schema.Advertisement{}, err
 	}
+	if s.delAfterRead {
+		s.Batching.Delete(ctx, dsKey)
+	}
 	return schema.BytesToAdvertisement(id, val)
 }
 
@@ -156,28 +165,10 @@ func (s *ClientStore) getAdvertisement(ctx context.Context, id cid.Cid) (*Advert
 
 func (s *ClientStore) list(ctx context.Context, nextCid cid.Cid, n int, w io.Writer) error {
 	for i := 0; i < n; i++ {
-		val, err := s.Batching.Get(ctx, datastore.NewKey(nextCid.String()))
-		if err != nil {
-			return err
-		}
-
-		nb := schema.AdvertisementPrototype.NewBuilder()
-		decoder, err := multicodec.LookupDecoder(nextCid.Prefix().Codec)
-		if err != nil {
-			return err
-		}
-
-		err = decoder(nb, bytes.NewBuffer(val))
-		if err != nil {
-			return err
-		}
-		node := nb.Build()
-
-		ad, err := schema.UnwrapAdvertisement(node)
+		ad, err := s.loadAd(ctx, nextCid)
 		if err != nil {
 			return err
 		}
-
 		if _, err = io.WriteString(w, nextCid.String()); err != nil {
 			return err
 		}
@@ -195,24 +186,7 @@ func (s *ClientStore) list(ctx context.Context, nextCid cid.Cid, n int, w io.Wri
 
 func (s *ClientStore) crawl(ctx context.Context, nextCid cid.Cid, n int, ads chan<- *Advertisement) (cid.Cid, error) {
 	for i := 0; i < n; i++ {
-		val, err := s.Batching.Get(ctx, datastore.NewKey(nextCid.String()))
-		if err != nil {
-			return cid.Undef, err
-		}
-
-		nb := schema.AdvertisementPrototype.NewBuilder()
-		decoder, err := multicodec.LookupDecoder(nextCid.Prefix().Codec)
-		if err != nil {
-			return cid.Undef, err
-		}
-
-		err = decoder(nb, bytes.NewBuffer(val))
-		if err != nil {
-			return cid.Undef, err
-		}
-		node := nb.Build()
-
-		ad, err := schema.UnwrapAdvertisement(node)
+		ad, err := s.loadAd(ctx, nextCid)
 		if err != nil {
 			return cid.Undef, err
 		}

pkg/adpub/options.go

@@ -20,6 +20,7 @@ type config struct {
 	p2pHost           host.Host
 	syncRetryBackoff  time.Duration
 	topic             string
+	delAfterRead      bool
 }
 
 // Option is a function that sets a value in a config.
@@ -98,3 +99,11 @@ func WithHttpTimeout(to time.Duration) Option {
 		return nil
 	}
 }
+
+// WithDeleteAfterRead deleted ifems from the store after reading them.
+func WithDeleteAfterRead(del bool) Option {
+	return func(c *config) error {
+		c.delAfterRead = del
+		return nil
+	}
+}

pkg/ads/crawl.go

@@ -72,6 +72,7 @@ func adsCrawlAction(cctx *cli.Context) error {
 	}
 
 	provClient, err := adpub.NewClient(*addrInfo,
+		adpub.WithDeleteAfterRead(true),
 		adpub.WithEntriesDepthLimit(0),
 		adpub.WithTopicName(cctx.String("topic")),
 		adpub.WithHttpTimeout(cctx.Duration("timeout")))

pkg/ads/list.go

@@ -44,7 +44,9 @@ func adsListAction(cctx *cli.Context) error {
 		return fmt.Errorf("bad pub-addr-info: %w", err)
 	}
 
-	provClient, err := adpub.NewClient(*addrInfo, adpub.WithTopicName(cctx.String("topic")),
+	provClient, err := adpub.NewClient(*addrInfo,
+		adpub.WithDeleteAfterRead(true),
+		adpub.WithTopicName(cctx.String("topic")),
 		adpub.WithHttpTimeout(cctx.Duration("timeout")))
 	if err != nil {
 		return err