crawl faster, error out less

events/repostream.go

@@ -36,20 +36,23 @@ func ConsumeRepoStreamLite(ctx context.Context, con *websocket.Conn, cb LiteStre
 					if err != nil {
 						e := fmt.Errorf("getting record %s (%s) within seq %d for %s: %w", op.Path, *op.Cid, evt.Seq, evt.Repo, err)
 						log.Error(e)
-						return nil
+						continue
 					}
 
 					if lexutil.LexLink(rc) != *op.Cid {
+						// TODO: do we even error here?
 						return fmt.Errorf("mismatch in record and op cid: %s != %s", rc, *op.Cid)
 					}
 
 					if err := cb(ek, evt.Seq, op.Path, evt.Repo, &rc, rec); err != nil {
-						return err
+						log.Errorf("event consumer callback (%s): %s", ek, err)
+						continue
 					}
 
 				case repomgr.EvtKindDeleteRecord:
 					if err := cb(ek, evt.Seq, op.Path, evt.Repo, nil, nil); err != nil {
-						return err
+						log.Errorf("event consumer callback (%s): %s", ek, err)
+						continue
 					}
 				}
 			}

indexer/crawler.go

@@ -2,6 +2,7 @@ package indexer
 
 import (
 	"context"
+	"fmt"
 
 	comatproto "github.com/bluesky-social/indigo/api/atproto"
 	"github.com/bluesky-social/indigo/models"
@@ -20,22 +21,29 @@ type CrawlDispatcher struct {
 	complete chan util.Uid
 
 	doRepoCrawl func(context.Context, *crawlWork) error
+
+	concurrency int
 }
 
-func NewCrawlDispatcher(repoFn func(context.Context, *crawlWork) error) *CrawlDispatcher {
+func NewCrawlDispatcher(repoFn func(context.Context, *crawlWork) error, concurrency int) (*CrawlDispatcher, error) {
+	if concurrency < 1 {
+		return nil, fmt.Errorf("must specify a non-zero positive integer for crawl dispatcher concurrency")
+	}
+
 	return &CrawlDispatcher{
 		ingest:      make(chan *models.ActorInfo),
 		repoSync:    make(chan *crawlWork),
 		complete:    make(chan util.Uid),
 		catchup:     make(chan *catchupJob),
 		doRepoCrawl: repoFn,
-	}
+		concurrency: concurrency,
+	}, nil
 }
 
 func (c *CrawlDispatcher) Run() {
 	go c.mainLoop()
 
-	for i := 0; i < 3; i++ {
+	for i := 0; i < c.concurrency; i++ {
 		go c.fetchWorker()
 	}
 }

indexer/indexer.go

@@ -68,8 +68,12 @@ func NewIndexer(db *gorm.DB, notifman notifs.NotificationManager, evtman *events
 	}
 
 	if crawl {
-		ix.Crawler = NewCrawlDispatcher(ix.FetchAndIndexRepo)
+		c, err := NewCrawlDispatcher(ix.FetchAndIndexRepo, 10)
+		if err != nil {
+			return nil, err
+		}
 
+		ix.Crawler = c
 		ix.Crawler.Run()
 	}