this repo has no description
remove direct dependence of indexer on repomgr (#443)
authored by
Whyrusleeping
and committed by
GitHub
5bdc9f19
9857c23c
+192
-159
+2
-2
bgs/admin.go
+2
-2
bgs/admin.go
···
172
172
MaxEventsPerSecond: p.CrawlRateLimit,
173
173
}
174
174
175
-
limiter = bgs.Index.GetLimiter(p.ID)
175
+
limiter = bgs.repoFetcher.GetLimiter(p.ID)
176
176
if limiter != nil {
177
177
crawlRate.TokenCount = limiter.Tokens()
178
178
}
···
410
410
}
411
411
412
412
// Update the crawl limit in the limiter
413
-
limiter := bgs.Index.GetOrCreateLimiter(pds.ID, limit)
413
+
limiter := bgs.repoFetcher.GetOrCreateLimiter(pds.ID, limit)
414
414
limiter.SetLimit(rate.Limit(limit))
415
415
416
416
return e.JSON(200, map[string]any{
+10
-8
bgs/bgs.go
+10
-8
bgs/bgs.go
···
55
55
const serverListenerBootTimeout = 5 * time.Second
56
56
57
57
type BGS struct {
58
-
Index *indexer.Indexer
59
-
db *gorm.DB
60
-
slurper *Slurper
61
-
events *events.EventManager
62
-
didr did.Resolver
58
+
Index *indexer.Indexer
59
+
db *gorm.DB
60
+
slurper *Slurper
61
+
events *events.EventManager
62
+
didr did.Resolver
63
+
repoFetcher *indexer.RepoFetcher
63
64
64
65
blobs blobs.BlobStore
65
66
hr api.HandleResolver
···
106
107
EventsSent promclient.Counter
107
108
}
108
109
109
-
func NewBGS(db *gorm.DB, ix *indexer.Indexer, repoman *repomgr.RepoManager, evtman *events.EventManager, didr did.Resolver, blobs blobs.BlobStore, hr api.HandleResolver, ssl bool) (*BGS, error) {
110
+
func NewBGS(db *gorm.DB, ix *indexer.Indexer, repoman *repomgr.RepoManager, evtman *events.EventManager, didr did.Resolver, blobs blobs.BlobStore, rf *indexer.RepoFetcher, hr api.HandleResolver, ssl bool) (*BGS, error) {
110
111
db.AutoMigrate(User{})
111
112
db.AutoMigrate(AuthToken{})
112
113
db.AutoMigrate(models.PDS{})
113
114
db.AutoMigrate(models.DomainBan{})
114
115
115
116
bgs := &BGS{
116
-
Index: ix,
117
-
db: db,
117
+
Index: ix,
118
+
db: db,
119
+
repoFetcher: rf,
118
120
119
121
hr: hr,
120
122
repoman: repoman,
+4
-2
cmd/bigsky/main.go
+4
-2
cmd/bigsky/main.go
···
283
283
284
284
notifman := ¬ifs.NullNotifs{}
285
285
286
-
ix, err := indexer.NewIndexer(db, notifman, evtman, cachedidr, repoman, true, cctx.Bool("spidering"), cctx.Bool("aggregation"))
286
+
rf := indexer.NewRepoFetcher(db, repoman)
287
+
288
+
ix, err := indexer.NewIndexer(db, notifman, evtman, cachedidr, rf, true, cctx.Bool("spidering"), cctx.Bool("aggregation"))
287
289
if err != nil {
288
290
return err
289
291
}
···
335
337
}
336
338
337
339
log.Infow("constructing bgs")
338
-
bgs, err := libbgs.NewBGS(db, ix, repoman, evtman, cachedidr, blobstore, hr, !cctx.Bool("crawl-insecure-ws"))
340
+
bgs, err := libbgs.NewBGS(db, ix, repoman, evtman, cachedidr, blobstore, rf, hr, !cctx.Bool("crawl-insecure-ws"))
339
341
if err != nil {
340
342
return err
341
343
}
+2
-143
indexer/indexer.go
+2
-143
indexer/indexer.go
···
1
1
package indexer
2
2
3
3
import (
4
-
"bytes"
5
4
"context"
6
5
"database/sql"
7
6
"errors"
8
7
"fmt"
9
-
"sync"
10
8
"time"
11
9
12
10
comatproto "github.com/bluesky-social/indigo/api/atproto"
···
19
17
"github.com/bluesky-social/indigo/repomgr"
20
18
"github.com/bluesky-social/indigo/util"
21
19
"github.com/bluesky-social/indigo/xrpc"
22
-
"golang.org/x/time/rate"
23
20
24
21
"github.com/ipfs/go-cid"
25
-
ipld "github.com/ipfs/go-ipld-format"
26
22
logging "github.com/ipfs/go-log"
27
23
"go.opentelemetry.io/otel"
28
-
"go.opentelemetry.io/otel/attribute"
29
24
"gorm.io/gorm"
30
25
"gorm.io/gorm/clause"
31
26
)
···
42
37
events *events.EventManager
43
38
didr did.Resolver
44
39
45
-
// TODO: i feel like the repomgr doesnt belong here
46
-
repomgr *repomgr.RepoManager
47
-
48
40
Crawler *CrawlDispatcher
49
-
50
-
Limiters map[uint]*rate.Limiter
51
-
LimitMux sync.RWMutex
52
41
53
42
doAggregations bool
54
43
doSpider bool
···
58
47
ApplyPDSClientSettings func(*xrpc.Client)
59
48
}
60
49
61
-
func NewIndexer(db *gorm.DB, notifman notifs.NotificationManager, evtman *events.EventManager, didr did.Resolver, repoman *repomgr.RepoManager, crawl, aggregate, spider bool) (*Indexer, error) {
50
+
func NewIndexer(db *gorm.DB, notifman notifs.NotificationManager, evtman *events.EventManager, didr did.Resolver, fetcher *RepoFetcher, crawl, aggregate, spider bool) (*Indexer, error) {
62
51
db.AutoMigrate(&models.FeedPost{})
63
52
db.AutoMigrate(&models.ActorInfo{})
64
53
db.AutoMigrate(&models.FollowRecord{})
···
69
58
db: db,
70
59
notifman: notifman,
71
60
events: evtman,
72
-
repomgr: repoman,
73
61
didr: didr,
74
-
Limiters: make(map[uint]*rate.Limiter),
75
62
doAggregations: aggregate,
76
63
doSpider: spider,
77
64
SendRemoteFollow: func(context.Context, string, uint) error {
···
81
68
}
82
69
83
70
if crawl {
84
-
c, err := NewCrawlDispatcher(ix.FetchAndIndexRepo, 10)
71
+
c, err := NewCrawlDispatcher(fetcher.FetchAndIndexRepo, 10)
85
72
if err != nil {
86
73
return nil, err
87
74
}
···
91
78
}
92
79
93
80
return ix, nil
94
-
}
95
-
96
-
func (ix *Indexer) GetLimiter(pdsID uint) *rate.Limiter {
97
-
ix.LimitMux.RLock()
98
-
defer ix.LimitMux.RUnlock()
99
-
100
-
return ix.Limiters[pdsID]
101
-
}
102
-
103
-
func (ix *Indexer) GetOrCreateLimiter(pdsID uint, pdsrate float64) *rate.Limiter {
104
-
ix.LimitMux.RLock()
105
-
defer ix.LimitMux.RUnlock()
106
-
107
-
lim, ok := ix.Limiters[pdsID]
108
-
if !ok {
109
-
lim = rate.NewLimiter(rate.Limit(pdsrate), 1)
110
-
ix.Limiters[pdsID] = lim
111
-
}
112
-
113
-
return lim
114
-
}
115
-
116
-
func (ix *Indexer) SetLimiter(pdsID uint, lim *rate.Limiter) {
117
-
ix.LimitMux.Lock()
118
-
defer ix.LimitMux.Unlock()
119
-
120
-
ix.Limiters[pdsID] = lim
121
81
}
122
82
123
83
func (ix *Indexer) HandleRepoEvent(ctx context.Context, evt *repomgr.RepoEvent) error {
···
412
372
}
413
373
414
374
return false
415
-
}
416
-
417
-
func (ix *Indexer) fetchRepo(ctx context.Context, c *xrpc.Client, pds *models.PDS, did string, rev string) ([]byte, error) {
418
-
ctx, span := otel.Tracer("indexer").Start(ctx, "fetchRepo")
419
-
defer span.End()
420
-
421
-
span.SetAttributes(
422
-
attribute.String("pds", pds.Host),
423
-
attribute.String("did", did),
424
-
attribute.String("rev", rev),
425
-
)
426
-
427
-
limiter := ix.GetOrCreateLimiter(pds.ID, pds.CrawlRateLimit)
428
-
429
-
// Wait to prevent DOSing the PDS when connecting to a new stream with lots of active repos
430
-
limiter.Wait(ctx)
431
-
432
-
log.Debugw("SyncGetRepo", "did", did, "since", rev)
433
-
// TODO: max size on these? A malicious PDS could just send us a petabyte sized repo here and kill us
434
-
repo, err := comatproto.SyncGetRepo(ctx, c, did, rev)
435
-
if err != nil {
436
-
reposFetched.WithLabelValues("fail").Inc()
437
-
return nil, fmt.Errorf("failed to fetch repo (did=%s,rev=%s,host=%s): %w", did, rev, pds.Host, err)
438
-
}
439
-
reposFetched.WithLabelValues("success").Inc()
440
-
441
-
return repo, nil
442
-
}
443
-
444
-
// TODO: since this function is the only place we depend on the repomanager, i wonder if this should be wired some other way?
445
-
func (ix *Indexer) FetchAndIndexRepo(ctx context.Context, job *crawlWork) error {
446
-
ctx, span := otel.Tracer("indexer").Start(ctx, "FetchAndIndexRepo")
447
-
defer span.End()
448
-
449
-
span.SetAttributes(attribute.Int("catchup", len(job.catchup)))
450
-
451
-
ai := job.act
452
-
453
-
var pds models.PDS
454
-
if err := ix.db.First(&pds, "id = ?", ai.PDS).Error; err != nil {
455
-
return fmt.Errorf("expected to find pds record (%d) in db for crawling one of their users: %w", ai.PDS, err)
456
-
}
457
-
458
-
rev, err := ix.repomgr.GetRepoRev(ctx, ai.Uid)
459
-
if err != nil && !isNotFound(err) {
460
-
return fmt.Errorf("failed to get repo root: %w", err)
461
-
}
462
-
463
-
// attempt to process buffered events
464
-
if !job.initScrape && len(job.catchup) > 0 {
465
-
first := job.catchup[0]
466
-
var resync bool
467
-
if first.evt.Since == nil || rev == *first.evt.Since {
468
-
for i, j := range job.catchup {
469
-
catchupEventsProcessed.Inc()
470
-
if err := ix.repomgr.HandleExternalUserEvent(ctx, pds.ID, ai.Uid, ai.Did, j.evt.Since, j.evt.Rev, j.evt.Blocks, j.evt.Ops); err != nil {
471
-
log.Errorw("buffered event catchup failed", "error", err, "did", ai.Did, "i", i, "jobCount", len(job.catchup), "seq", j.evt.Seq)
472
-
resync = true // fall back to a repo sync
473
-
break
474
-
}
475
-
}
476
-
477
-
if !resync {
478
-
return nil
479
-
}
480
-
}
481
-
}
482
-
483
-
if rev == "" {
484
-
span.SetAttributes(attribute.Bool("full", true))
485
-
}
486
-
487
-
c := models.ClientForPds(&pds)
488
-
ix.ApplyPDSClientSettings(c)
489
-
490
-
repo, err := ix.fetchRepo(ctx, c, &pds, ai.Did, rev)
491
-
if err != nil {
492
-
return err
493
-
}
494
-
495
-
if err := ix.repomgr.ImportNewRepo(ctx, ai.Uid, ai.Did, bytes.NewReader(repo), &rev); err != nil {
496
-
span.RecordError(err)
497
-
498
-
if ipld.IsNotFound(err) {
499
-
log.Errorw("partial repo fetch was missing data", "did", ai.Did, "pds", pds.Host, "rev", rev)
500
-
repo, err := ix.fetchRepo(ctx, c, &pds, ai.Did, "")
501
-
if err != nil {
502
-
return err
503
-
}
504
-
505
-
if err := ix.repomgr.ImportNewRepo(ctx, ai.Uid, ai.Did, bytes.NewReader(repo), nil); err != nil {
506
-
span.RecordError(err)
507
-
return fmt.Errorf("failed to import backup repo (%s): %w", ai.Did, err)
508
-
}
509
-
510
-
return nil
511
-
}
512
-
return fmt.Errorf("importing fetched repo (curRev: %s): %w", rev, err)
513
-
}
514
-
515
-
return nil
516
375
}
517
376
518
377
func (ix *Indexer) GetPost(ctx context.Context, uri string) (*models.FeedPost, error) {
+3
-1
indexer/posts_test.go
+3
-1
indexer/posts_test.go
···
61
61
62
62
didr := testPLC(t)
63
63
64
-
ix, err := NewIndexer(maindb, notifman, evtman, didr, repoman, false, true, true)
64
+
rf := NewRepoFetcher(maindb, repoman)
65
+
66
+
ix, err := NewIndexer(maindb, notifman, evtman, didr, rf, false, true, true)
65
67
if err != nil {
66
68
t.Fatal(err)
67
69
}
+165
indexer/repofetch.go
+165
indexer/repofetch.go
···
1
+
package indexer
2
+
3
+
import (
4
+
"bytes"
5
+
"context"
6
+
"fmt"
7
+
"sync"
8
+
9
+
"github.com/bluesky-social/indigo/api/atproto"
10
+
"github.com/bluesky-social/indigo/models"
11
+
"github.com/bluesky-social/indigo/repomgr"
12
+
"github.com/bluesky-social/indigo/xrpc"
13
+
ipld "github.com/ipfs/go-ipld-format"
14
+
"go.opentelemetry.io/otel"
15
+
"go.opentelemetry.io/otel/attribute"
16
+
"golang.org/x/time/rate"
17
+
"gorm.io/gorm"
18
+
)
19
+
20
+
func NewRepoFetcher(db *gorm.DB, rm *repomgr.RepoManager) *RepoFetcher {
21
+
return &RepoFetcher{
22
+
repoman: rm,
23
+
db: db,
24
+
Limiters: make(map[uint]*rate.Limiter),
25
+
ApplyPDSClientSettings: func(*xrpc.Client) {},
26
+
}
27
+
}
28
+
29
+
type RepoFetcher struct {
30
+
repoman *repomgr.RepoManager
31
+
db *gorm.DB
32
+
33
+
Limiters map[uint]*rate.Limiter
34
+
LimitMux sync.RWMutex
35
+
36
+
ApplyPDSClientSettings func(*xrpc.Client)
37
+
}
38
+
39
+
func (rf *RepoFetcher) GetLimiter(pdsID uint) *rate.Limiter {
40
+
rf.LimitMux.RLock()
41
+
defer rf.LimitMux.RUnlock()
42
+
43
+
return rf.Limiters[pdsID]
44
+
}
45
+
46
+
func (rf *RepoFetcher) GetOrCreateLimiter(pdsID uint, pdsrate float64) *rate.Limiter {
47
+
rf.LimitMux.RLock()
48
+
defer rf.LimitMux.RUnlock()
49
+
50
+
lim, ok := rf.Limiters[pdsID]
51
+
if !ok {
52
+
lim = rate.NewLimiter(rate.Limit(pdsrate), 1)
53
+
rf.Limiters[pdsID] = lim
54
+
}
55
+
56
+
return lim
57
+
}
58
+
59
+
func (rf *RepoFetcher) SetLimiter(pdsID uint, lim *rate.Limiter) {
60
+
rf.LimitMux.Lock()
61
+
defer rf.LimitMux.Unlock()
62
+
63
+
rf.Limiters[pdsID] = lim
64
+
}
65
+
66
+
func (rf *RepoFetcher) fetchRepo(ctx context.Context, c *xrpc.Client, pds *models.PDS, did string, rev string) ([]byte, error) {
67
+
ctx, span := otel.Tracer("indexer").Start(ctx, "fetchRepo")
68
+
defer span.End()
69
+
70
+
span.SetAttributes(
71
+
attribute.String("pds", pds.Host),
72
+
attribute.String("did", did),
73
+
attribute.String("rev", rev),
74
+
)
75
+
76
+
limiter := rf.GetOrCreateLimiter(pds.ID, pds.CrawlRateLimit)
77
+
78
+
// Wait to prevent DOSing the PDS when connecting to a new stream with lots of active repos
79
+
limiter.Wait(ctx)
80
+
81
+
log.Debugw("SyncGetRepo", "did", did, "since", rev)
82
+
// TODO: max size on these? A malicious PDS could just send us a petabyte sized repo here and kill us
83
+
repo, err := atproto.SyncGetRepo(ctx, c, did, rev)
84
+
if err != nil {
85
+
reposFetched.WithLabelValues("fail").Inc()
86
+
return nil, fmt.Errorf("failed to fetch repo (did=%s,rev=%s,host=%s): %w", did, rev, pds.Host, err)
87
+
}
88
+
reposFetched.WithLabelValues("success").Inc()
89
+
90
+
return repo, nil
91
+
}
92
+
93
+
// TODO: since this function is the only place we depend on the repomanager, i wonder if this should be wired some other way?
94
+
func (rf *RepoFetcher) FetchAndIndexRepo(ctx context.Context, job *crawlWork) error {
95
+
ctx, span := otel.Tracer("indexer").Start(ctx, "FetchAndIndexRepo")
96
+
defer span.End()
97
+
98
+
span.SetAttributes(attribute.Int("catchup", len(job.catchup)))
99
+
100
+
ai := job.act
101
+
102
+
var pds models.PDS
103
+
if err := rf.db.First(&pds, "id = ?", ai.PDS).Error; err != nil {
104
+
return fmt.Errorf("expected to find pds record (%d) in db for crawling one of their users: %w", ai.PDS, err)
105
+
}
106
+
107
+
rev, err := rf.repoman.GetRepoRev(ctx, ai.Uid)
108
+
if err != nil && !isNotFound(err) {
109
+
return fmt.Errorf("failed to get repo root: %w", err)
110
+
}
111
+
112
+
// attempt to process buffered events
113
+
if !job.initScrape && len(job.catchup) > 0 {
114
+
first := job.catchup[0]
115
+
var resync bool
116
+
if first.evt.Since == nil || rev == *first.evt.Since {
117
+
for i, j := range job.catchup {
118
+
catchupEventsProcessed.Inc()
119
+
if err := rf.repoman.HandleExternalUserEvent(ctx, pds.ID, ai.Uid, ai.Did, j.evt.Since, j.evt.Rev, j.evt.Blocks, j.evt.Ops); err != nil {
120
+
log.Errorw("buffered event catchup failed", "error", err, "did", ai.Did, "i", i, "jobCount", len(job.catchup), "seq", j.evt.Seq)
121
+
resync = true // fall back to a repo sync
122
+
break
123
+
}
124
+
}
125
+
126
+
if !resync {
127
+
return nil
128
+
}
129
+
}
130
+
}
131
+
132
+
if rev == "" {
133
+
span.SetAttributes(attribute.Bool("full", true))
134
+
}
135
+
136
+
c := models.ClientForPds(&pds)
137
+
rf.ApplyPDSClientSettings(c)
138
+
139
+
repo, err := rf.fetchRepo(ctx, c, &pds, ai.Did, rev)
140
+
if err != nil {
141
+
return err
142
+
}
143
+
144
+
if err := rf.repoman.ImportNewRepo(ctx, ai.Uid, ai.Did, bytes.NewReader(repo), &rev); err != nil {
145
+
span.RecordError(err)
146
+
147
+
if ipld.IsNotFound(err) {
148
+
log.Errorw("partial repo fetch was missing data", "did", ai.Did, "pds", pds.Host, "rev", rev)
149
+
repo, err := rf.fetchRepo(ctx, c, &pds, ai.Did, "")
150
+
if err != nil {
151
+
return err
152
+
}
153
+
154
+
if err := rf.repoman.ImportNewRepo(ctx, ai.Uid, ai.Did, bytes.NewReader(repo), nil); err != nil {
155
+
span.RecordError(err)
156
+
return fmt.Errorf("failed to import backup repo (%s): %w", ai.Did, err)
157
+
}
158
+
159
+
return nil
160
+
}
161
+
return fmt.Errorf("importing fetched repo (curRev: %s): %w", rev, err)
162
+
}
163
+
164
+
return nil
165
+
}
+3
-1
pds/server.go
+3
-1
pds/server.go
···
80
80
repoman := repomgr.NewRepoManager(cs, kmgr)
81
81
notifman := notifs.NewNotificationManager(db, repoman.GetRecord)
82
82
83
-
ix, err := indexer.NewIndexer(db, notifman, evtman, didr, repoman, false, true, true)
83
+
rf := indexer.NewRepoFetcher(db, repoman)
84
+
85
+
ix, err := indexer.NewIndexer(db, notifman, evtman, didr, rf, false, true, true)
84
86
if err != nil {
85
87
return nil, err
86
88
}
+3
-2
testing/utils.go
+3
-2
testing/utils.go
···
437
437
diskpersist, err := events.NewDiskPersistence(filepath.Join(dir, "dp-primary"), filepath.Join(dir, "dp-archive"), maindb, opts)
438
438
439
439
evtman := events.NewEventManager(diskpersist)
440
+
rf := indexer.NewRepoFetcher(maindb, repoman)
440
441
441
-
ix, err := indexer.NewIndexer(maindb, notifman, evtman, didr, repoman, true, true, true)
442
+
ix, err := indexer.NewIndexer(maindb, notifman, evtman, didr, rf, true, true, true)
442
443
if err != nil {
443
444
return nil, err
444
445
}
···
451
452
452
453
tr := &api.TestHandleResolver{}
453
454
454
-
b, err := bgs.NewBGS(maindb, ix, repoman, evtman, didr, nil, tr, false)
455
+
b, err := bgs.NewBGS(maindb, ix, repoman, evtman, didr, nil, rf, tr, false)
455
456
if err != nil {
456
457
return nil, err
457
458
}