+1 -1

cmd/monarch/backfill.go

··· 10 10 ParallelBackfills: cctx.Int("backfill-workers"), 11 11 ParallelRecordCreates: cctx.Int("backfill-consumers"), 12 12 NSIDFilter: "", 13 13 - SyncRequestsPerSecond: 10, 13 13 + SyncRequestsPerSecond: cctx.Int("sync-requests-limit"), 14 14 RelayHost: "https://" + cctx.String("relay-host"), 15 15 } 16 16

+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

+3 -1

cmd/monarch/firehose.go

··· 13 13 func NewFirehoseConnection(ctx context.Context, cctx *cli.Context, cursorSvc *CursorService) (*websocket.Conn, error) { 14 14 url := fmt.Sprintf("wss://%s/xrpc/com.atproto.sync.subscribeRepos", cctx.String("relay-host")) 15 15 curs, err := cursorSvc.GetFirehoseCursor() 16 16 - if err == nil { // reversed 16 16 + if err != nil { 17 17 + slog.Error("error getting firehose cursor", "err", err) 18 18 + } else if curs > 0 { 17 19 url += fmt.Sprintf("?cursor=%d", curs) 18 20 } 19 21