Merge remote-tracking branch 'origin/main' into feat/splitter-rebase

atproto/identity/identity.go

···
 	base := BaseDirectory{
 		PLCURL: DefaultPLCURL,
 		HTTPClient: http.Client{
-			Timeout: time.Second * 15,
+			Timeout: time.Second * 10,
+			Transport: &http.Transport{
+				// would want this around 100ms for services doing lots of handle resolution. Impacts PLC connections as well, but not too bad.
+				IdleConnTimeout: time.Millisecond * 1000,
+				MaxIdleConns:    100,
+			},
 		},
 		Resolver: net.Resolver{
 			Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
-				d := net.Dialer{Timeout: time.Second * 5}
+				d := net.Dialer{Timeout: time.Second * 3}
 				return d.DialContext(ctx, network, address)
 			},
 		},

bgs/bgs.go

···
 	"github.com/bluesky-social/indigo/models"
 	"github.com/bluesky-social/indigo/repomgr"
 	"github.com/bluesky-social/indigo/xrpc"
+	lru "github.com/hashicorp/golang-lru/v2"
 	"golang.org/x/sync/semaphore"
 	"golang.org/x/time/rate"
 
···
 
 	// Management of Compaction
 	compactor *Compactor
+
+	// User cache
+	userCache *lru.Cache[string, *User]
 }
 
 type PDSResync struct {
···
 	db.AutoMigrate(models.PDS{})
 	db.AutoMigrate(models.DomainBan{})
 
+	uc, _ := lru.New[string, *User](1_000_000)
+
 	bgs := &BGS{
 		Index:       ix,
 		db:          db,
···
 		consumers:   make(map[uint64]*SocketConsumer),
 
 		pdsResyncs: make(map[uint]*PDSResync),
+
+		userCache: uc,
 	}
 
 	ix.CreateExternalUser = bgs.createExternalUser
···
 
 	// UpstreamStatus is the state of the user as reported by the upstream PDS
 	UpstreamStatus string `gorm:"index"`
+
+	lk sync.Mutex
+}
+
+func (u *User) SetTakenDown(v bool) {
+	u.lk.Lock()
+	defer u.lk.Unlock()
+	u.TakenDown = v
+}
+
+func (u *User) GetTakenDown() bool {
+	u.lk.Lock()
+	defer u.lk.Unlock()
+	return u.TakenDown
+}
+
+func (u *User) SetTombstoned(v bool) {
+	u.lk.Lock()
+	defer u.lk.Unlock()
+	u.Tombstoned = v
+}
+
+func (u *User) GetTombstoned() bool {
+	u.lk.Lock()
+	defer u.lk.Unlock()
+	return u.Tombstoned
+}
+
+func (u *User) SetUpstreamStatus(v string) {
+	u.lk.Lock()
+	defer u.lk.Unlock()
+	u.UpstreamStatus = v
+}
+
+func (u *User) GetUpstreamStatus() string {
+	u.lk.Lock()
+	defer u.lk.Unlock()
+	return u.UpstreamStatus
 }
 
 type addTargetBody struct {
···
 	ctx, span := tracer.Start(ctx, "lookupUserByDid")
 	defer span.End()
 
+	cu, ok := bgs.userCache.Get(did)
+	if ok {
+		return cu, nil
+	}
+
 	var u User
 	if err := bgs.db.Find(&u, "did = ?", did).Error; err != nil {
 		return nil, err
···
 	if u.ID == 0 {
 		return nil, gorm.ErrRecordNotFound
 	}
+
+	bgs.userCache.Add(did, &u)
 
 	return &u, nil
 }
···
 		repoCommitsReceivedCounter.WithLabelValues(host.Host).Add(1)
 		evt := env.RepoCommit
 		log.Debugw("bgs got repo append event", "seq", evt.Seq, "pdsHost", host.Host, "repo", evt.Repo)
+
+		s := time.Now()
 		u, err := bgs.lookupUserByDid(ctx, evt.Repo)
+		userLookupDuration.Observe(time.Since(s).Seconds())
 		if err != nil {
 			if !errors.Is(err, gorm.ErrRecordNotFound) {
 				return fmt.Errorf("looking up event user: %w", err)
 			}
 
 			newUsersDiscovered.Inc()
+			start := time.Now()
 			subj, err := bgs.createExternalUser(ctx, evt.Repo)
+			newUserDiscoveryDuration.Observe(time.Since(start).Seconds())
 			if err != nil {
 				return fmt.Errorf("fed event create external user: %w", err)
 			}
···
 			u.Did = evt.Repo
 		}
 
-		span.SetAttributes(attribute.String("upstream_status", u.UpstreamStatus))
+		ustatus := u.GetUpstreamStatus()
+		span.SetAttributes(attribute.String("upstream_status", ustatus))
 
-		if u.TakenDown || u.UpstreamStatus == events.AccountStatusTakendown {
-			span.SetAttributes(attribute.Bool("taken_down_by_relay_admin", u.TakenDown))
+		if u.GetTakenDown() || ustatus == events.AccountStatusTakendown {
+			span.SetAttributes(attribute.Bool("taken_down_by_relay_admin", u.GetTakenDown()))
 			log.Debugw("dropping commit event from taken down user", "did", evt.Repo, "seq", evt.Seq, "pdsHost", host.Host)
 			return nil
 		}
 
-		if u.UpstreamStatus == events.AccountStatusSuspended {
+		if ustatus == events.AccountStatusSuspended {
 			log.Debugw("dropping commit event from suspended user", "did", evt.Repo, "seq", evt.Seq, "pdsHost", host.Host)
 			return nil
 		}
 
-		if u.UpstreamStatus == events.AccountStatusDeactivated {
+		if ustatus == events.AccountStatusDeactivated {
 			log.Debugw("dropping commit event from deactivated user", "did", evt.Repo, "seq", evt.Seq, "pdsHost", host.Host)
 			return nil
 		}
···
 			}
 		}
 
-		if u.Tombstoned {
+		if u.GetTombstoned() {
 			span.SetAttributes(attribute.Bool("tombstoned", true))
 			// we've checked the authority of the users PDS, so reinstate the account
 			if err := bgs.db.Model(&User{}).Where("id = ?", u.ID).UpdateColumn("tombstoned", false).Error; err != nil {
 				return fmt.Errorf("failed to un-tombstone a user: %w", err)
 			}
+			u.SetTombstoned(false)
 
 			ai, err := bgs.Index.LookupUser(ctx, u.ID)
 			if err != nil {
···
 			return fmt.Errorf("failed to look up user by did: %w", err)
 		}
 
-		if u.TakenDown {
+		if u.GetTakenDown() {
 			shouldBeActive = false
 			status = &events.AccountStatusTakendown
 		}
···
 		if err := bgs.db.Model(User{}).Where("id = ?", u.ID).Update("upstream_status", events.AccountStatusActive).Error; err != nil {
 			return fmt.Errorf("failed to set user active status: %w", err)
 		}
+		u.SetUpstreamStatus(events.AccountStatusActive)
 	case events.AccountStatusDeactivated:
 		if err := bgs.db.Model(User{}).Where("id = ?", u.ID).Update("upstream_status", events.AccountStatusDeactivated).Error; err != nil {
 			return fmt.Errorf("failed to set user deactivation status: %w", err)
 		}
+		u.SetUpstreamStatus(events.AccountStatusDeactivated)
 	case events.AccountStatusSuspended:
 		if err := bgs.db.Model(User{}).Where("id = ?", u.ID).Update("upstream_status", events.AccountStatusSuspended).Error; err != nil {
 			return fmt.Errorf("failed to set user suspension status: %w", err)
 		}
+		u.SetUpstreamStatus(events.AccountStatusSuspended)
 	case events.AccountStatusTakendown:
 		if err := bgs.db.Model(User{}).Where("id = ?", u.ID).Update("upstream_status", events.AccountStatusTakendown).Error; err != nil {
 			return fmt.Errorf("failed to set user taken down status: %w", err)
 		}
+		u.SetUpstreamStatus(events.AccountStatusTakendown)
 
 		if err := bgs.db.Model(&models.ActorInfo{}).Where("uid = ?", u.ID).UpdateColumns(map[string]any{
 			"handle": nil,
···
 		}).Error; err != nil {
 			return err
 		}
+		u.SetUpstreamStatus(events.AccountStatusDeleted)
 
 		if err := bgs.db.Model(&models.ActorInfo{}).Where("uid = ?", u.ID).UpdateColumns(map[string]any{
 			"handle": nil,
···
 	if err := bgs.db.Model(User{}).Where("id = ?", u.ID).Update("taken_down", true).Error; err != nil {
 		return err
 	}
+	u.SetTakenDown(true)
 
 	if err := bgs.repoman.TakeDownRepo(ctx, u.ID); err != nil {
 		return err
···
 	if err := bgs.db.Model(User{}).Where("id = ?", u.ID).Update("taken_down", false).Error; err != nil {
 		return err
 	}
+	u.SetTakenDown(false)
 
 	return nil
 }

bgs/compactor.go

···
 	return state, nil
 }
 
-func (c *Compactor) EnqueueRepo(ctx context.Context, user User, fast bool) {
+func (c *Compactor) EnqueueRepo(ctx context.Context, user *User, fast bool) {
 	ctx, span := otel.Tracer("compactor").Start(ctx, "EnqueueRepo")
 	defer span.End()
 	log.Infow("enqueueing compaction for repo", "repo", user.Did, "uid", user.ID, "fast", fast)

bgs/handlers.go

···
 		return nil, echo.NewHTTPError(http.StatusInternalServerError, "failed to lookup user")
 	}
 
-	if u.Tombstoned {
+	if u.GetTombstoned() {
 		return nil, fmt.Errorf("account was deleted")
 	}
 
-	if u.TakenDown {
+	if u.GetTakenDown() {
 		return nil, fmt.Errorf("account was taken down by the Relay")
 	}
 
-	if u.UpstreamStatus == events.AccountStatusTakendown {
+	ustatus := u.GetUpstreamStatus()
+	if ustatus == events.AccountStatusTakendown {
 		return nil, fmt.Errorf("account was taken down by its PDS")
 	}
 
-	if u.UpstreamStatus == events.AccountStatusDeactivated {
+	if ustatus == events.AccountStatusDeactivated {
 		return nil, fmt.Errorf("account is temporarily deactivated")
 	}
 
-	if u.UpstreamStatus == events.AccountStatusSuspended {
+	if ustatus == events.AccountStatusSuspended {
 		return nil, fmt.Errorf("account is suspended by its PDS")
 	}
 
···
 		return nil, echo.NewHTTPError(http.StatusInternalServerError, "failed to lookup user")
 	}
 
-	if u.Tombstoned {
+	if u.GetTombstoned() {
 		return nil, fmt.Errorf("account was deleted")
 	}
 
-	if u.TakenDown {
+	if u.GetTakenDown() {
 		return nil, fmt.Errorf("account was taken down by the Relay")
 	}
 
-	if u.UpstreamStatus == events.AccountStatusTakendown {
+	ustatus := u.GetUpstreamStatus()
+	if ustatus == events.AccountStatusTakendown {
 		return nil, fmt.Errorf("account was taken down by its PDS")
 	}
 
-	if u.UpstreamStatus == events.AccountStatusDeactivated {
+	if ustatus == events.AccountStatusDeactivated {
 		return nil, fmt.Errorf("account is temporarily deactivated")
 	}
 
-	if u.UpstreamStatus == events.AccountStatusSuspended {
+	if ustatus == events.AccountStatusSuspended {
 		return nil, fmt.Errorf("account is suspended by its PDS")
 	}
 
···
 		return nil, echo.NewHTTPError(http.StatusInternalServerError, "failed to lookup user")
 	}
 
-	if u.Tombstoned {
+	if u.GetTombstoned() {
 		return nil, fmt.Errorf("account was deleted")
 	}
 
-	if u.TakenDown {
+	if u.GetTakenDown() {
 		return nil, fmt.Errorf("account was taken down by the Relay")
 	}
 
-	if u.UpstreamStatus == events.AccountStatusTakendown {
+	ustatus := u.GetUpstreamStatus()
+	if ustatus == events.AccountStatusTakendown {
 		return nil, fmt.Errorf("account was taken down by its PDS")
 	}
 
-	if u.UpstreamStatus == events.AccountStatusDeactivated {
+	if ustatus == events.AccountStatusDeactivated {
 		return nil, fmt.Errorf("account is temporarily deactivated")
 	}
 
-	if u.UpstreamStatus == events.AccountStatusSuspended {
+	if ustatus == events.AccountStatusSuspended {
 		return nil, fmt.Errorf("account is suspended by its PDS")
 	}
 

bgs/metrics.go

···
 	Buckets: prometheus.ExponentialBuckets(100, 10, 8),
 }, []string{"code", "method", "path"})
 
+var userLookupDuration = promauto.NewHistogram(prometheus.HistogramOpts{
+	Name:    "relay_user_lookup_duration",
+	Help:    "A histogram of user lookup latencies",
+	Buckets: prometheus.ExponentialBuckets(0.001, 2, 15),
+})
+
+var newUserDiscoveryDuration = promauto.NewHistogram(prometheus.HistogramOpts{
+	Name:    "relay_new_user_discovery_duration",
+	Help:    "A histogram of new user discovery latencies",
+	Buckets: prometheus.ExponentialBuckets(0.001, 2, 15),
+})
+
 // MetricsMiddleware defines handler function for metrics middleware
 func MetricsMiddleware(next echo.HandlerFunc) echo.HandlerFunc {
 	return func(c echo.Context) error {

carstore/bs.go

···
 }
 
 type FileCarStore struct {
-	meta    *CarStoreGormMeta
-	rootDir string
+	meta     *CarStoreGormMeta
+	rootDirs []string
 
 	lscLk          sync.Mutex
 	lastShardCache map[models.Uid]*CarShard
 }
 
-func NewCarStore(meta *gorm.DB, root string) (CarStore, error) {
-	if _, err := os.Stat(root); err != nil {
-		if !os.IsNotExist(err) {
-			return nil, err
-		}
+func NewCarStore(meta *gorm.DB, roots []string) (CarStore, error) {
+	for _, root := range roots {
+		if _, err := os.Stat(root); err != nil {
+			if !os.IsNotExist(err) {
+				return nil, err
+			}
 
-		if err := os.Mkdir(root, 0775); err != nil {
-			return nil, err
+			if err := os.Mkdir(root, 0775); err != nil {
+				return nil, err
+			}
 		}
 	}
 	if err := meta.AutoMigrate(&CarShard{}, &blockRef{}); err != nil {
···
 
 	return &FileCarStore{
 		meta:           &CarStoreGormMeta{meta: meta},
-		rootDir:        root,
+		rootDirs:       roots,
 		lastShardCache: make(map[models.Uid]*CarShard),
 	}, nil
 }
···
 func fnameForShard(user models.Uid, seq int) string {
 	return fmt.Sprintf("sh-%d-%d", user, seq)
 }
+
+func (cs *FileCarStore) dirForUser(user models.Uid) string {
+	return cs.rootDirs[int(user)%len(cs.rootDirs)]
+}
+
 func (cs *FileCarStore) openNewShardFile(ctx context.Context, user models.Uid, seq int) (*os.File, string, error) {
 	// TODO: some overwrite protections
-	fname := filepath.Join(cs.rootDir, fnameForShard(user, seq))
+	fname := filepath.Join(cs.dirForUser(user), fnameForShard(user, seq))
 	fi, err := os.Create(fname)
 	if err != nil {
 		return nil, "", err
···
 	defer span.End()
 
 	// TODO: some overwrite protections
-	fname := filepath.Join(cs.rootDir, fnameForShard(user, seq))
+	fname := filepath.Join(cs.dirForUser(user), fnameForShard(user, seq))
 	if err := os.WriteFile(fname, data, 0664); err != nil {
 		return "", err
 	}
···
 		offset += nw
 	}
 
+	start := time.Now()
 	path, err := cs.writeNewShardFile(ctx, user, seq, buf.Bytes())
 	if err != nil {
 		return nil, fmt.Errorf("failed to write shard file: %w", err)
 	}
+	writeShardFileDuration.Observe(time.Since(start).Seconds())
 
 	shard := CarShard{
 		Root:      models.DbCID{CID: root},
···
 		Rev:       rev,
 	}
 
+	start = time.Now()
 	if err := cs.putShard(ctx, &shard, brefs, rmcids, false); err != nil {
 		return nil, err
 	}
+	writeShardMetadataDuration.Observe(time.Since(start).Seconds())
 
 	return buf.Bytes(), nil
 }
···
 	// TODO: some overwrite protections
 	// NOTE CreateTemp is used for creating a non-colliding file, but we keep it and don't delete it so don't think of it as "temporary".
 	// This creates "sh-%d-%d%s" with some random stuff in the last position
-	fi, err := os.CreateTemp(cs.rootDir, fnameForShard(user, seq))
+	fi, err := os.CreateTemp(cs.dirForUser(user), fnameForShard(user, seq))
 	if err != nil {
 		return nil, "", err
 	}

carstore/metrics.go

···
+package carstore
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+)
+
+var writeShardFileDuration = promauto.NewHistogram(prometheus.HistogramOpts{
+	Name:    "carstore_write_shard_file_duration",
+	Help:    "Duration of writing shard file to disk",
+	Buckets: prometheus.ExponentialBuckets(0.001, 2, 15),
+})
+
+var writeShardMetadataDuration = promauto.NewHistogram(prometheus.HistogramOpts{
+	Name:    "carstore_write_shard_metadata_duration",
+	Help:    "Duration of writing shard metadata to DB",
+	Buckets: prometheus.ExponentialBuckets(0.001, 2, 15),
+})

carstore/repo_test.go

···
 		return nil, nil, err
 	}
 
-	sharddir := filepath.Join(tempdir, "shards")
-	if err := os.MkdirAll(sharddir, 0775); err != nil {
+	sharddir1 := filepath.Join(tempdir, "shards1")
+	if err := os.MkdirAll(sharddir1, 0775); err != nil {
+		return nil, nil, err
+	}
+
+	sharddir2 := filepath.Join(tempdir, "shards2")
+	if err := os.MkdirAll(sharddir2, 0775); err != nil {
 		return nil, nil, err
 	}
 
···
 		return nil, nil, err
 	}
 
-	cs, err := NewCarStore(db, sharddir)
+	cs, err := NewCarStore(db, []string{sharddir1, sharddir2})
 	if err != nil {
 		return nil, nil, err
 	}

cmd/bigsky/main.go

···
 			EnvVars: []string{"RELAY_NUM_COMPACTION_WORKERS"},
 			Value:   2,
 		},
+		&cli.StringSliceFlag{
+			Name:    "carstore-shard-dirs",
+			Usage:   "specify list of shard directories for carstore storage, overrides default storage within datadir",
+			EnvVars: []string{"RELAY_CARSTORE_SHARD_DIRS"},
+		},
 	}
 
 	app.Action = runBigsky
···
 		}
 	}
 
-	os.MkdirAll(filepath.Dir(csdir), os.ModePerm)
-	cstore, err := carstore.NewCarStore(csdb, csdir)
+	csdirs := []string{csdir}
+	if paramDirs := cctx.StringSlice("carstore-shard-dirs"); len(paramDirs) > 0 {
+		csdirs = paramDirs
+	}
+
+	for _, csd := range csdirs {
+		if err := os.MkdirAll(filepath.Dir(csd), os.ModePerm); err != nil {
+			return err
+		}
+	}
+
+	cstore, err := carstore.NewCarStore(csdb, csdirs)
 	if err != nil {
 		return err
 	}

cmd/laputa/main.go

···
 			}
 		}
 
-		cstore, err := carstore.NewCarStore(csdb, csdir)
+		cstore, err := carstore.NewCarStore(csdb, []string{csdir})
 		if err != nil {
 			return err
 		}

cmd/supercollider/main.go

···
 		return nil, nil, err
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		return nil, nil, err
 	}

did/metrics.go

···
 	Name: "multiresolver_resolved_dids_total",
 	Help: "Total number of DIDs resolved",
 }, []string{"resolver"})
+
+var mrResolveDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
+	Name:    "indigo_multiresolver_resolve_duration_seconds",
+	Help:    "A histogram of resolve latencies",
+	Buckets: prometheus.ExponentialBuckets(0.001, 2, 15),
+}, []string{"resolver"})

did/multi.go

···
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/whyrusleeping/go-did"
 )
···
 }
 
 func (mr *MultiResolver) GetDocument(ctx context.Context, didstr string) (*did.Document, error) {
+	s := time.Now()
+
 	pdid, err := did.ParseDID(didstr)
 	if err != nil {
 		return nil, err
 	}
 
 	method := pdid.Protocol()
+	defer func() {
+		mrResolveDuration.WithLabelValues(method).Observe(time.Since(s).Seconds())
+	}()
 
 	res, ok := mr.handlers[method]
 	if !ok {

events/dbpersist_test.go

···
 		return nil, nil, nil, "", err
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		return nil, nil, nil, "", err
 	}

indexer/posts_test.go

···
 		t.Fatal(err)
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		t.Fatal(err)
 	}

pds/handlers_test.go

···
 		t.Fatal(err)
 	}
 
-	cs, err := carstore.NewCarStore(db, sharddir)
+	cs, err := carstore.NewCarStore(db, []string{sharddir})
 	if err != nil {
 		t.Fatal(err)
 	}

repomgr/bench_test.go

···
 		b.Fatal(err)
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		b.Fatal(err)
 	}

repomgr/ingest_test.go

···
 		t.Fatal(err)
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		t.Fatal(err)
 	}
···
 		t.Fatal(err)
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		t.Fatal(err)
 	}

repomgr/metrics.go

···
 	Name: "repomgr_repo_ops_imported",
 	Help: "Number of repo ops imported",
 })
+
+var openAndSigCheckDuration = promauto.NewHistogram(prometheus.HistogramOpts{
+	Name:    "repomgr_open_and_sig_check_duration",
+	Help:    "Duration of opening and signature check",
+	Buckets: prometheus.ExponentialBuckets(0.001, 2, 15),
+})
+
+var calcDiffDuration = promauto.NewHistogram(prometheus.HistogramOpts{
+	Name:    "repomgr_calc_diff_duration",
+	Help:    "Duration of calculating diff",
+	Buckets: prometheus.ExponentialBuckets(0.001, 2, 15),
+})
+
+var writeCarSliceDuration = promauto.NewHistogram(prometheus.HistogramOpts{
+	Name:    "repomgr_write_car_slice_duration",
+	Help:    "Duration of writing car slice",
+	Buckets: prometheus.ExponentialBuckets(0.001, 2, 15),
+})

repomgr/repomgr.go

···
 	"io"
 	"strings"
 	"sync"
+	"time"
 
 	atproto "github.com/bluesky-social/indigo/api/atproto"
 	bsky "github.com/bluesky-social/indigo/api/bsky"
···
 	unlock := rm.lockUser(ctx, uid)
 	defer unlock()
 
+	start := time.Now()
 	root, ds, err := rm.cs.ImportSlice(ctx, uid, since, carslice)
 	if err != nil {
 		return fmt.Errorf("importing external carslice: %w", err)
···
 	if err := rm.CheckRepoSig(ctx, r, did); err != nil {
 		return err
 	}
+	openAndSigCheckDuration.Observe(time.Since(start).Seconds())
 
 	var skipcids map[cid.Cid]bool
 	if ds.BaseCid().Defined() {
···
 		}
 	}
 
+	start = time.Now()
 	if err := ds.CalcDiff(ctx, skipcids); err != nil {
 		return fmt.Errorf("failed while calculating mst diff (since=%v): %w", since, err)
-
 	}
+	calcDiffDuration.Observe(time.Since(start).Seconds())
 
 	evtops := make([]RepoOp, 0, len(ops))
 
···
 		}
 	}
 
+	start = time.Now()
 	rslice, err := ds.CloseWithRoot(ctx, root, nrev)
 	if err != nil {
 		return fmt.Errorf("close with root: %w", err)
 	}
+	writeCarSliceDuration.Observe(time.Since(start).Seconds())
 
 	if rm.events != nil {
 		rm.events(ctx, &RepoEvent{

testing/utils.go

···
 		return nil, err
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		return nil, err
 	}
···
 		return nil, err
 	}
 
-	cs, err := carstore.NewCarStore(cardb, cspath)
+	cs, err := carstore.NewCarStore(cardb, []string{cspath})
 	if err != nil {
 		return nil, err
 	}