comparing 02dbec4028c73486e3b97385d8ae929ba1dab1f2 and backfill-next on edavis.dev/monarch

+1 -4

cmd/monarch/cursors.go

···

       3
       3
        
       import (

     

       4
       4
        
       	"context"

     

       5
       5
        
       	"log/slog"

     

       6
       6
       -
       	"sync"

     

       7
       6
        
       	"time"

     

       8
       7
        
       

     

       9
       8
        
       	"gorm.io/gorm"

     

       10
       9
        
       )

     

       11
       10
        
       

     

       12
       11
        
       type CursorService struct {

     

       13
       13
       -
       	store *gorm.DB

     

       14
       14
       -
       

     

       15
       15
       -
       	firehoseLk  sync.Mutex

     

       12
       12
       +
       	store       *gorm.DB

     

       16
       13
        
       	firehoseSeq int64

     

       17
       14
        
       }

     

       18
       15

+6 -9

cmd/monarch/backfill.go

···

       1
       1
        
       package main

     

       2
       2
        
       

     

       3
       3
        
       import (

     

       4
       4
       -
       	"github.com/bluesky-social/indigo/backfill"

     

       4
       4
       +
       	backfill "github.com/bluesky-social/indigo/backfill/next"

     

       5
       5
        
       	"github.com/urfave/cli/v2"

     

       6
       6
        
       )

     

       7
       7
        
       

     

       8
       8
        
       func NewBackfillService(store backfill.Store, h *HandlerService, cctx *cli.Context) *backfill.Backfiller {

     

       9
       9
       -
       	opts := &backfill.BackfillOptions{

     

       10
       10
       -
       		ParallelBackfills:     cctx.Int("backfill-workers"),

     

       11
       11
       -
       		ParallelRecordCreates: cctx.Int("backfill-consumers"),

     

       12
       12
       -
       		NSIDFilter:            "",

     

       13
       13
       -
       		SyncRequestsPerSecond: cctx.Int("sync-requests-limit"),

     

       14
       14
       -
       		RelayHost:             "https://" + cctx.String("relay-host"),

     

       15
       15
       -
       	}

     

       9
       9
       +
       	opts := backfill.DefaultBackfillerOptions()

     

       10
       10
       +
       	opts.PerPDSBackfillConcurrency = cctx.Int("backfill-workers")

     

       11
       11
       +
       	opts.GlobalRecordCreateConcurrency = cctx.Int("backfill-consumers")

     

       12
       12
       +
       	opts.PerPDSSyncsPerSecond = cctx.Float64("sync-requests-limit")

     

       16
       13
        
       

     

       17
       17
       -
       	return backfill.NewBackfiller("backfiller", store, h.HandleCreate, h.HandleUpdate, h.HandleDelete, opts)

     

       14
       14
       +
       	return backfill.NewBackfiller("backfiller", store, h.HandleCreate, opts)

     

       18
       15
        
       }

+5 -5

cmd/monarch/main.go

···

       10
       10
        
       	"time"

     

       11
       11
        
       

     

       12
       12
        
       	comatproto "github.com/bluesky-social/indigo/api/atproto"

     

       13
       13
       -
       	"github.com/bluesky-social/indigo/backfill"

     

       13
       13
       +
       	backfill "github.com/bluesky-social/indigo/backfill/next"

     

       14
       14
        
       	"github.com/bluesky-social/indigo/events"

     

       15
       15
        
       	"github.com/bluesky-social/indigo/events/schedulers/parallel"

     

       16
       16
        
       	"github.com/bluesky-social/indigo/util/cliutil"

     
···

       47
       47
        
       	app.handler = NewHandlerService(app.content)

     

       48
       48
        
       

     

       49
       49
        
       	app.backfill = NewBackfillService(backfill.NewGormstore(app.state), app.handler, cctx)

     

       50
       50
       -
       	go app.backfill.Start()

     

       51
       50
        
       

     

       52
       51
        
       	app.census = NewCensusService(app.cursor, app.backfill)

     

       53
       52
        
       	go app.census.Start(ctx, cctx)

     
···

       61
       60
        
       	rsc := events.RepoStreamCallbacks{

     

       62
       61
        
       		RepoCommit: func(evt *comatproto.SyncSubscribeRepos_Commit) error {

     

       63
       62
        
       			app.cursor.SetFirehoseCursor(evt.Seq)

     

       64
       64
       -
       			return app.backfill.HandleEvent(ctx, evt)

     

       63
       63
       +
       			// return app.backfill.HandleEvent(ctx, evt)

     

       64
       64
       +
       			return nil

     

       65
       65
        
       		},

     

       66
       66
        
       		// TODO account

     

       67
       67
        
       		// TODO identity

     
···

       88
       88
        
       		return nil

     

       89
       89
        
       	}

     

       90
       90
        
       

     

       91
       91
       -
       	if err := app.backfill.Stop(ctx); err != nil {

     

       91
       91
       +
       	if err := app.backfill.Shutdown(ctx); err != nil {

     

       92
       92
        
       		slog.Error("error stopping backfiller", "err", err)

     

       93
       93
        
       	}

     

       94
       94
        
       

     
···

       137
       137
        
       			Name:  "max-repo-crawlers",

     

       138
       138
        
       			Value: 4,

     

       139
       139
        
       		},

     

       140
       140
       -
       		&cli.IntFlag{

     

       140
       140
       +
       		&cli.Float64Flag{

     

       141
       141
        
       			Name:  "sync-requests-limit",

     

       142
       142
        
       			Value: 10, // ratelimit-policy: 3000;w=300

     

       143
       143
        
       		},

Compare changes