commit 4aa8b136289bc10ca8cf27db6009c89be05f2dc0 · edavis.dev/monarch

+4 -4

cmd/monarch/backfill.go

···

       2
       2
        
       

     

       3
       3
        
       import "github.com/bluesky-social/indigo/backfill"

     

       4
       4
        
       

     

       5
       5
       -
       func NewBackfillService(store backfill.Store, h *HandlerService) *backfill.Backfiller {

     

       5
       5
       +
       func NewBackfillService(store backfill.Store, h *HandlerService, workers int, consumers int) *backfill.Backfiller {

     

       6
       6
        
       	opts := &backfill.BackfillOptions{

     

       7
       7
       -
       		ParallelBackfills:     1,

     

       8
       8
       -
       		ParallelRecordCreates: 1,

     

       7
       7
       +
       		ParallelBackfills:     workers,

     

       8
       8
       +
       		ParallelRecordCreates: consumers,

     

       9
       9
        
       		NSIDFilter:            "",

     

       10
       10
       -
       		SyncRequestsPerSecond: 2,

     

       10
       10
       +
       		SyncRequestsPerSecond: 10,

     

       11
       11
        
       		RelayHost:             "https://bsky.network",

     

       12
       12
        
       	}

     

       13
       13

+15 -4

cmd/monarch/main.go

···

       6
       6
        
       	"log/slog"

     

       7
       7
        
       	"os"

     

       8
       8
        
       	"os/signal"

     

       9
       9
       +
       	"runtime"

     

       9
       10
        
       	"syscall"

     

       10
       11
        
       	"time"

     

       11
       12
        
       

     
···

       38
       39
        
       	}

     

       39
       40
        
       }

     

       40
       41
        
       

     

       41
       41
       -
       func (app *App) Start(ctx context.Context) error {

     

       42
       42
       +
       func (app *App) Start(ctx context.Context, cctx *cli.Context) error {

     

       42
       43
        
       	slog.Info("starting up")

     

       43
       44
        
       

     

       44
       45
        
       	app.cursor = NewCursorService(app.state)

     
···

       46
       47
        
       

     

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

     

       48
       49
        
       

     

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

     

       50
       50
       +
       	workers := cctx.Int("backfill-workers")

     

       51
       51
       +
       	consumers := cctx.Int("backfill-consumers")

     

       52
       52
       +
       	app.backfill = NewBackfillService(backfill.NewGormstore(app.state), app.handler, workers, consumers)

     

       50
       53
        
       	go app.backfill.Start()

     

       51
       54
        
       

     

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

     
···

       119
       122
        
       		},

     

       120
       123
        
       		&cli.IntFlag{

     

       121
       124
        
       			Name:  "max-db-connections",

     

       122
       122
       -
       			Value: 1,

     

       125
       125
       +
       			Value: runtime.NumCPU(),

     

       126
       126
       +
       		},

     

       127
       127
       +
       		&cli.IntFlag{

     

       128
       128
       +
       			Name:  "backfill-workers",

     

       129
       129
       +
       			Value: 50,

     

       130
       130
       +
       		},

     

       131
       131
       +
       		&cli.IntFlag{

     

       132
       132
       +
       			Name:  "backfill-consumers",

     

       133
       133
       +
       			Value: 100,

     

       123
       134
        
       		},

     

       124
       135
        
       	}

     

       125
       136
        
       

     
···

       142
       153
        
       		}

     

       143
       154
        
       

     

       144
       155
        
       		app := NewApp(statedb, contentdb)

     

       145
       145
       -
       		if err := app.Start(ctx); err != nil {

     

       156
       156
       +
       		if err := app.Start(ctx, cctx); err != nil {

     

       146
       157
        
       			slog.Error("failed to start backfiller", "err", err)

     

       147
       158
        
       		}

     

       148
       159