this repo has no description
finish crawling gap aware indexer, integration test for this passes
+287
-90
+7
-2
bgs/bgs.go
+7
-2
bgs/bgs.go
···
159
159
}
160
160
161
161
func (bgs *BGS) handleFedEvent(ctx context.Context, host *models.PDS, evt *events.RepoEvent) error {
162
-
log.Infof("bgs got fed event from %q: %s\n", host.Host, evt.Repo)
162
+
log.Infof("bgs got fed event %d from %q: %s\n", evt.Seq, host.Host, evt.Repo)
163
163
switch {
164
164
case evt.RepoAppend != nil:
165
165
u, err := bgs.lookupUserByDid(ctx, evt.Repo)
···
184
184
return err
185
185
}
186
186
187
-
return bgs.Index.Crawler.AddToCatchupQueue(ctx, host, u.ID, evt)
187
+
ai, err := bgs.Index.LookupUser(ctx, u.ID)
188
+
if err != nil {
189
+
return err
190
+
}
191
+
192
+
return bgs.Index.Crawler.AddToCatchupQueue(ctx, host, ai, evt)
188
193
}
189
194
190
195
return nil
+2
-3
bgs/fedmgr.go
+2
-3
bgs/fedmgr.go
···
57
57
npds := models.PDS{
58
58
Host: host,
59
59
}
60
-
fmt.Println("CREATING NEW PDS", host)
61
60
if err := s.db.Create(&npds).Error; err != nil {
62
61
return err
63
62
}
···
87
86
88
87
con, res, err := d.Dial("ws://"+host.Host+"/events", nil)
89
88
if err != nil {
90
-
fmt.Printf("dialing %q failed: %s", host.Host, err)
89
+
log.Warnf("dialing %q failed: %s", host.Host, err)
91
90
time.Sleep(sleepForBackoff(backoff))
92
91
backoff++
93
92
continue
94
93
}
95
94
96
-
fmt.Println("event subscription response code: ", res.StatusCode)
95
+
log.Info("event subscription response code: ", res.StatusCode)
97
96
98
97
if err := s.handleConnection(host, con); err != nil {
99
98
if errors.Is(err, ErrTimeoutShutdown) {
+1
carstore/bs.go
+1
carstore/bs.go
+108
-17
indexer/crawler.go
+108
-17
indexer/crawler.go
···
10
10
type CrawlDispatcher struct {
11
11
ingest chan *models.ActorInfo
12
12
13
-
repoSync chan *models.ActorInfo
13
+
repoSync chan *crawlWork
14
14
15
-
doRepoCrawl func(context.Context, *models.ActorInfo) error
15
+
catchup chan *catchupJob
16
+
17
+
complete chan uint
18
+
19
+
doRepoCrawl func(context.Context, *crawlWork) error
16
20
}
17
21
18
-
func NewCrawlDispatcher(repoFn func(context.Context, *models.ActorInfo) error) *CrawlDispatcher {
22
+
func NewCrawlDispatcher(repoFn func(context.Context, *crawlWork) error) *CrawlDispatcher {
19
23
return &CrawlDispatcher{
20
24
ingest: make(chan *models.ActorInfo),
21
-
repoSync: make(chan *models.ActorInfo),
25
+
repoSync: make(chan *crawlWork),
26
+
complete: make(chan uint),
27
+
catchup: make(chan *catchupJob),
22
28
doRepoCrawl: repoFn,
23
29
}
24
30
}
···
31
37
}
32
38
}
33
39
40
+
type catchupJob struct {
41
+
evt *events.RepoEvent
42
+
host *models.PDS
43
+
user *models.ActorInfo
44
+
}
45
+
46
+
type crawlWork struct {
47
+
act *models.ActorInfo
48
+
initScrape bool
49
+
50
+
catchup []*catchupJob
51
+
52
+
// for events that come in while this actor is being processed
53
+
next []*catchupJob
54
+
}
55
+
34
56
func (c *CrawlDispatcher) mainLoop() {
35
-
var next *models.ActorInfo
36
-
var buffer []*models.ActorInfo
57
+
var next *crawlWork
58
+
var buffer []*crawlWork
37
59
38
-
set := make(map[uint]*models.ActorInfo)
39
-
//progress := make(map[uint]*models.ActorInfo)
60
+
todo := make(map[uint]*crawlWork)
61
+
inProgress := make(map[uint]*crawlWork)
40
62
41
-
var rs chan *models.ActorInfo
63
+
var rs chan *crawlWork
42
64
for {
43
65
select {
44
66
case act := <-c.ingest:
45
67
// TODO: max buffer size
46
68
47
-
_, has := set[act.Uid]
69
+
_, ok := inProgress[act.Uid]
70
+
if ok {
71
+
break
72
+
}
73
+
74
+
_, has := todo[act.Uid]
48
75
if has {
49
76
break
50
77
}
51
-
set[act.Uid] = act
78
+
79
+
cw := &crawlWork{
80
+
act: act,
81
+
initScrape: true,
82
+
}
83
+
todo[act.Uid] = cw
52
84
53
85
if next == nil {
54
-
next = act
86
+
next = cw
55
87
rs = c.repoSync
56
88
} else {
57
-
buffer = append(buffer, act)
89
+
buffer = append(buffer, cw)
58
90
}
59
91
case rs <- next:
60
-
delete(set, next.Uid)
92
+
delete(todo, next.act.Uid)
93
+
inProgress[next.act.Uid] = next
61
94
62
95
if len(buffer) > 0 {
63
96
next = buffer[0]
···
66
99
next = nil
67
100
rs = nil
68
101
}
102
+
case catchup := <-c.catchup:
103
+
job, ok := todo[catchup.user.Uid]
104
+
if ok {
105
+
job.catchup = append(job.catchup, catchup)
106
+
break
107
+
}
108
+
109
+
job, ok = inProgress[catchup.user.Uid]
110
+
if ok {
111
+
job.next = append(job.next, catchup)
112
+
break
113
+
}
114
+
115
+
cw := &crawlWork{
116
+
act: catchup.user,
117
+
catchup: []*catchupJob{catchup},
118
+
}
119
+
todo[catchup.user.Uid] = cw
120
+
121
+
if next == nil {
122
+
next = cw
123
+
rs = c.repoSync
124
+
} else {
125
+
buffer = append(buffer, cw)
126
+
}
127
+
128
+
case uid := <-c.complete:
129
+
job, ok := inProgress[uid]
130
+
if !ok {
131
+
panic("should not be possible to not have a job in progress we receive a completion signal for")
132
+
}
133
+
delete(inProgress, uid)
134
+
135
+
if len(job.next) > 0 {
136
+
todo[uid] = job
137
+
job.initScrape = false
138
+
job.catchup = job.next
139
+
job.next = nil
140
+
if next == nil {
141
+
next = job
142
+
rs = c.repoSync
143
+
} else {
144
+
buffer = append(buffer, job)
145
+
}
146
+
}
147
+
69
148
}
70
149
}
71
150
}
···
75
154
select {
76
155
case job := <-c.repoSync:
77
156
if err := c.doRepoCrawl(context.TODO(), job); err != nil {
78
-
log.Errorf("failed to perform repo crawl of %q: %s", job, err)
157
+
log.Errorf("failed to perform repo crawl of %q: %s", job.act.Did, err)
79
158
}
159
+
160
+
// TODO: do we still just do this if it errors?
161
+
c.complete <- job.act.Uid
80
162
}
81
163
}
82
164
}
···
90
172
}
91
173
}
92
174
93
-
func (c *CrawlDispatcher) AddToCatchupQueue(ctx context.Context, host *models.PDS, u uint, evt *events.RepoEvent) error {
94
-
return nil
175
+
func (c *CrawlDispatcher) AddToCatchupQueue(ctx context.Context, host *models.PDS, u *models.ActorInfo, evt *events.RepoEvent) error {
176
+
select {
177
+
case c.catchup <- &catchupJob{
178
+
evt: evt,
179
+
host: host,
180
+
user: u,
181
+
}:
182
+
return nil
183
+
case <-ctx.Done():
184
+
return ctx.Err()
185
+
}
95
186
}
+30
-9
indexer/indexer.go
+30
-9
indexer/indexer.go
···
121
121
Repo: did,
122
122
123
123
RepoAppend: &events.RepoAppend{
124
-
Car: evt.RepoSlice,
125
-
Ops: repoOps,
124
+
Prev: evt.OldRoot,
125
+
Car: evt.RepoSlice,
126
+
Ops: repoOps,
126
127
},
127
128
128
129
PrivRelevantPds: relpds,
···
149
150
return nil, err
150
151
}
151
152
152
-
author, err := ix.lookupUser(ctx, fp.Author)
153
+
author, err := ix.LookupUser(ctx, fp.Author)
153
154
if err != nil {
154
155
return nil, err
155
156
}
···
191
192
return nil, err
192
193
}
193
194
194
-
act, err := ix.lookupUser(ctx, post.Author)
195
+
act, err := ix.LookupUser(ctx, post.Author)
195
196
if err != nil {
196
197
return nil, err
197
198
}
···
346
347
}
347
348
348
349
if err := ix.db.Clauses(clause.OnConflict{
350
+
Columns: []clause.Column{clause.Column{Name: "rkey"}, clause.Column{Name: "author"}},
349
351
UpdateAll: true,
350
352
}).Create(&fp).Error; err != nil {
351
353
return err
···
424
426
return ai.Did, nil
425
427
}
426
428
427
-
func (ix *Indexer) lookupUser(ctx context.Context, id uint) (*models.ActorInfo, error) {
429
+
func (ix *Indexer) LookupUser(ctx context.Context, id uint) (*models.ActorInfo, error) {
428
430
var ai models.ActorInfo
429
431
if err := ix.db.First(&ai, "id = ?", id).Error; err != nil {
430
432
return nil, err
···
555
557
}
556
558
557
559
// TODO: since this function is the only place we depend on the repomanager, i wonder if this should be wired some other way?
558
-
func (ix *Indexer) FetchAndIndexRepo(ctx context.Context, ai *models.ActorInfo) error {
560
+
func (ix *Indexer) FetchAndIndexRepo(ctx context.Context, job *crawlWork) error {
559
561
ctx, span := otel.Tracer("indexer").Start(ctx, "FetchAndIndexRepo")
560
562
defer span.End()
561
563
564
+
fmt.Println("FETCH AND INDEX REPO")
565
+
566
+
ai := job.act
567
+
562
568
var pds models.PDS
563
569
if err := ix.db.First(&pds, "id = ?", ai.PDS).Error; err != nil {
564
570
return fmt.Errorf("expected to find pds record in db for crawling one of their users: %w", err)
565
571
}
566
572
573
+
curHead, err := ix.repomgr.GetRepoRoot(ctx, ai.Uid)
574
+
if err != nil && !isNotFound(err) {
575
+
return err
576
+
}
577
+
578
+
fmt.Println("CUR HEAD: ", curHead)
579
+
567
580
c := &xrpc.Client{
568
581
Host: pds.Host,
582
+
}
583
+
584
+
var from string
585
+
if curHead.Defined() {
586
+
from = curHead.String()
569
587
}
570
588
571
589
// TODO: max size on these? A malicious PDS could just send us a petabyte sized repo here and kill us
572
-
repo, err := atproto.SyncGetRepo(ctx, c, ai.Did, "")
590
+
repo, err := atproto.SyncGetRepo(ctx, c, ai.Did, from)
573
591
if err != nil {
574
592
return fmt.Errorf("failed to fetch repo: %w", err)
575
593
}
576
594
577
595
// this process will send individual indexing events back to the indexer, doing a 'fast forward' of the users entire history
578
596
// we probably want alternative ways of doing this for 'very large' or 'very old' repos, but this works for now
579
-
if err := ix.repomgr.ImportNewRepo(ctx, ai.Uid, bytes.NewReader(repo)); err != nil {
580
-
return err
597
+
if err := ix.repomgr.ImportNewRepo(ctx, ai.Uid, bytes.NewReader(repo), curHead); err != nil {
598
+
return fmt.Errorf("importing fetched repo: %w", err)
581
599
}
600
+
601
+
// TODO: this is currently doing too much work, allowing us to ignore the catchup events we've gotten
602
+
// need to do 'just enough' work...
582
603
583
604
return nil
584
605
}
+2
-1
pds/server.go
+2
-1
pds/server.go
+16
-12
repo/repo.go
+16
-12
repo/repo.go
···
329
329
ctx, span := otel.Tracer("repo").Start(ctx, "DiffSince")
330
330
defer span.End()
331
331
332
-
otherRepo, err := OpenRepo(ctx, r.bs, oldrepo)
333
-
if err != nil {
334
-
return nil, err
335
-
}
332
+
var oldTree cid.Cid
333
+
if oldrepo.Defined() {
334
+
otherRepo, err := OpenRepo(ctx, r.bs, oldrepo)
335
+
if err != nil {
336
+
return nil, err
337
+
}
336
338
337
-
oldmst, err := otherRepo.getMst(ctx)
338
-
if err != nil {
339
-
return nil, err
340
-
}
339
+
oldmst, err := otherRepo.getMst(ctx)
340
+
if err != nil {
341
+
return nil, err
342
+
}
341
343
342
-
oldptr, err := oldmst.GetPointer(ctx)
343
-
if err != nil {
344
-
return nil, err
344
+
oldptr, err := oldmst.GetPointer(ctx)
345
+
if err != nil {
346
+
return nil, err
347
+
}
348
+
oldTree = oldptr
345
349
}
346
350
347
351
curmst, err := r.getMst(ctx)
···
354
358
return nil, err
355
359
}
356
360
357
-
return mst.DiffTrees(ctx, r.bs, oldptr, curptr)
361
+
return mst.DiffTrees(ctx, r.bs, oldTree, curptr)
358
362
}
+2
-1
repomgr/ingest_test.go
+2
-1
repomgr/ingest_test.go
···
8
8
"testing"
9
9
10
10
"github.com/bluesky-social/indigo/carstore"
11
+
"github.com/ipfs/go-cid"
11
12
"gorm.io/driver/sqlite"
12
13
"gorm.io/gorm"
13
14
)
···
60
61
}
61
62
62
63
ctx := context.TODO()
63
-
if err := repoman.ImportNewRepo(ctx, 2, fi); err != nil {
64
+
if err := repoman.ImportNewRepo(ctx, 2, fi, cid.Undef); err != nil {
64
65
t.Fatal(err)
65
66
}
66
67
}
+55
-37
repomgr/repomgr.go
+55
-37
repomgr/repomgr.go
···
15
15
"github.com/bluesky-social/indigo/events"
16
16
lexutil "github.com/bluesky-social/indigo/lex/util"
17
17
"github.com/bluesky-social/indigo/models"
18
-
"github.com/bluesky-social/indigo/mst"
19
18
"github.com/bluesky-social/indigo/repo"
20
19
"github.com/ipfs/go-cid"
21
20
"github.com/ipfs/go-datastore"
22
21
blockstore "github.com/ipfs/go-ipfs-blockstore"
22
+
ipld "github.com/ipfs/go-ipld-format"
23
23
logging "github.com/ipfs/go-log/v2"
24
24
"github.com/ipld/go-car"
25
25
cbg "github.com/whyrusleeping/cbor-gen"
···
64
64
65
65
type RepoEvent struct {
66
66
User uint
67
-
OldRoot cid.Cid
67
+
OldRoot *cid.Cid
68
68
NewRoot cid.Cid
69
69
RepoSlice []byte
70
70
PDS uint
···
211
211
if rm.events != nil {
212
212
rm.events(ctx, &RepoEvent{
213
213
User: user,
214
-
OldRoot: head,
214
+
OldRoot: &head,
215
215
NewRoot: nroot,
216
216
Ops: []RepoOp{{
217
217
Kind: EvtKindCreateRecord,
···
273
273
if rm.events != nil {
274
274
rm.events(ctx, &RepoEvent{
275
275
User: user,
276
-
OldRoot: head,
276
+
OldRoot: &head,
277
277
NewRoot: nroot,
278
278
Ops: []RepoOp{{
279
279
Kind: EvtKindUpdateRecord,
···
334
334
if rm.events != nil {
335
335
rm.events(ctx, &RepoEvent{
336
336
User: user,
337
-
OldRoot: head,
337
+
OldRoot: &head,
338
338
NewRoot: nroot,
339
339
Ops: []RepoOp{{
340
340
Kind: EvtKindDeleteRecord,
···
500
500
func (rm *RepoManager) HandleExternalUserEvent(ctx context.Context, pdsid uint, uid uint, prev *cid.Cid, ops []*events.RepoOp, carslice []byte) error {
501
501
ctx, span := otel.Tracer("repoman").Start(ctx, "HandleExternalUserEvent")
502
502
defer span.End()
503
+
504
+
log.Infof("HandleExternalUserEvent: %d %d %s", pdsid, uid, prev)
503
505
504
506
unlock := rm.lockUser(ctx, uid)
505
507
defer unlock()
···
578
580
579
581
if rm.events != nil {
580
582
rm.events(ctx, &RepoEvent{
581
-
User: uid,
582
-
//OldRoot: head,
583
+
User: uid,
584
+
OldRoot: prev,
583
585
NewRoot: root,
584
586
Ops: evtops,
585
587
RepoSlice: rslice,
···
691
693
if rm.events != nil {
692
694
rm.events(ctx, &RepoEvent{
693
695
User: user,
694
-
OldRoot: head,
696
+
OldRoot: &head,
695
697
NewRoot: nroot,
696
698
RepoSlice: rslice,
697
699
Ops: ops,
···
701
703
return nil
702
704
}
703
705
704
-
func (rm *RepoManager) ImportNewRepo(ctx context.Context, user uint, r io.Reader) error {
706
+
func (rm *RepoManager) ImportNewRepo(ctx context.Context, user uint, r io.Reader, oldest cid.Cid) error {
705
707
ctx, span := otel.Tracer("repoman").Start(ctx, "ImportNewRepo")
706
708
defer span.End()
707
709
708
710
unlock := rm.lockUser(ctx, user)
709
711
defer unlock()
710
712
711
-
_, err := rm.getUserRepoHead(ctx, user)
712
-
if err == nil {
713
-
// i predict this will get awkward
714
-
return fmt.Errorf("cannot import new repo for user we already know about")
713
+
head, err := rm.getUserRepoHead(ctx, user)
714
+
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
715
+
return err
715
716
}
716
717
717
-
if !errors.Is(err, gorm.ErrRecordNotFound) {
718
-
return err
718
+
if head != oldest {
719
+
// TODO: we could probably just deal with this
720
+
return fmt.Errorf("ImportNewRepo called with incorrect base")
719
721
}
720
722
721
-
err = rm.processNewRepo(ctx, user, r, func(ctx context.Context, old, nu cid.Cid, slice []byte, bs blockstore.Blockstore) error {
723
+
err = rm.processNewRepo(ctx, user, r, head, func(ctx context.Context, old, nu cid.Cid, slice []byte, bs blockstore.Blockstore) error {
724
+
r, err := repo.OpenRepo(ctx, bs, nu)
725
+
if err != nil {
726
+
return fmt.Errorf("opening new repo: %w", err)
727
+
}
722
728
723
-
diffops, err := mst.DiffTrees(ctx, bs, old, nu)
729
+
diffops, err := r.DiffSince(ctx, old)
724
730
if err != nil {
725
731
return fmt.Errorf("diff trees: %w", err)
726
732
}
···
769
775
}
770
776
}
771
777
778
+
if err := rm.updateUserRepoHead(ctx, user, nu); err != nil {
779
+
return err
780
+
}
781
+
772
782
if rm.events != nil {
773
783
rm.events(ctx, &RepoEvent{
774
784
User: user,
775
-
OldRoot: old,
785
+
OldRoot: &old,
776
786
NewRoot: nu,
777
787
RepoSlice: slice,
778
788
Ops: ops,
···
782
792
return nil
783
793
})
784
794
if err != nil {
785
-
return err
795
+
return fmt.Errorf("process new repo: %w:", err)
786
796
}
787
797
788
798
return nil
789
799
}
790
800
791
-
func (rm *RepoManager) processNewRepo(ctx context.Context, user uint, r io.Reader, cb func(ctx context.Context, old, nu cid.Cid, slice []byte, bs blockstore.Blockstore) error) error {
801
+
func (rm *RepoManager) processNewRepo(ctx context.Context, user uint, r io.Reader, until cid.Cid, cb func(ctx context.Context, old, nu cid.Cid, slice []byte, bs blockstore.Blockstore) error) error {
792
802
ctx, span := otel.Tracer("repoman").Start(ctx, "ImportNewRepo")
793
803
defer span.End()
794
804
···
820
830
head := &carr.Header.Roots[0]
821
831
822
832
var commits []cid.Cid
823
-
for head != nil {
833
+
for head != nil && *head != until {
824
834
825
835
commits = append(commits, *head)
826
836
rep, err := repo.OpenRepo(ctx, membs, *head)
827
837
if err != nil {
828
-
return err
838
+
return fmt.Errorf("opening repo for backwalk: %w", err)
829
839
}
830
840
831
841
prev, err := rep.PrevCommit(ctx)
832
842
if err != nil {
833
-
return err
843
+
return fmt.Errorf("prevCommit: %w", err)
834
844
}
835
845
836
846
head = prev
···
840
850
841
851
seen := make(map[cid.Cid]bool)
842
852
843
-
var prev cid.Cid
853
+
if until.Defined() {
854
+
seen[until] = true
855
+
}
856
+
857
+
prev := until
844
858
for i := len(commits) - 1; i >= 0; i-- {
845
859
root := commits[i]
846
-
cids, err := walkTree(ctx, seen, root, membs)
860
+
cids, err := walkTree(ctx, seen, root, membs, true)
847
861
if err != nil {
848
862
return fmt.Errorf("walkTree: %w", err)
849
863
}
850
864
851
865
ds, err := rm.cs.NewDeltaSession(ctx, user, &prev)
852
866
if err != nil {
853
-
return fmt.Errorf("opening delta session: %w", err)
867
+
return fmt.Errorf("opening delta session (%d / %d): %w", i, len(commits)-1, err)
854
868
}
855
869
856
870
for _, c := range cids {
857
871
blk, err := membs.Get(ctx, c)
858
872
if err != nil {
859
-
return err
873
+
return fmt.Errorf("copying walked cids to carstore: %w", err)
860
874
}
861
875
862
876
if err := ds.Put(ctx, blk); err != nil {
···
870
884
}
871
885
872
886
if err := cb(ctx, prev, root, carslice, membs); err != nil {
873
-
return err
887
+
return fmt.Errorf("cb errored: %w", err)
874
888
}
875
889
876
890
prev = root
···
879
893
return nil
880
894
}
881
895
882
-
func walkTree(ctx context.Context, skip map[cid.Cid]bool, root cid.Cid, bs blockstore.Blockstore) ([]cid.Cid, error) {
883
-
896
+
// walkTree returns all cids linked recursively by the root, skipping any cids
897
+
// in the 'skip' map, and not erroring on 'not found' if prevMissing is set
898
+
func walkTree(ctx context.Context, skip map[cid.Cid]bool, root cid.Cid, bs blockstore.Blockstore, prevMissing bool) ([]cid.Cid, error) {
884
899
// TODO: what if someone puts non-cbor links in their repo?
885
900
if root.Prefix().Codec != cid.DagCBOR {
886
901
return nil, fmt.Errorf("can only handle dag-cbor objects in repos (%s is %d)", root, root.Prefix().Codec)
···
891
906
return nil, err
892
907
}
893
908
894
-
out := []cid.Cid{root}
895
-
skip[root] = true
896
-
909
+
var links []cid.Cid
897
910
if err := cbg.ScanForLinks(bytes.NewReader(blk.RawData()), func(c cid.Cid) {
898
911
if skip[c] {
899
912
return
900
913
}
901
914
902
-
out = append(out, c)
915
+
links = append(links, c)
903
916
skip[c] = true
904
917
905
918
return
···
907
920
return nil, err
908
921
}
909
922
923
+
out := []cid.Cid{root}
924
+
skip[root] = true
925
+
910
926
// TODO: should do this non-recursive since i expect these may get deep
911
-
for _, c := range out[1:] {
912
-
sub, err := walkTree(ctx, skip, c, bs)
927
+
for _, c := range links {
928
+
sub, err := walkTree(ctx, skip, c, bs, prevMissing)
913
929
if err != nil {
914
-
return nil, err
930
+
if prevMissing && !ipld.IsNotFound(err) {
931
+
return nil, err
932
+
}
915
933
}
916
934
917
935
out = append(out, sub...)
+64
-8
testing/integ_test.go
+64
-8
testing/integ_test.go
···
80
80
}
81
81
}
82
82
83
-
func socialSim(t *testing.T, users []*testUser) []*atproto.RepoStrongRef {
83
+
func socialSim(t *testing.T, users []*testUser, postiter, likeiter int) []*atproto.RepoStrongRef {
84
84
var posts []*atproto.RepoStrongRef
85
-
for i := 0; i < 10; i++ {
85
+
for i := 0; i < postiter; i++ {
86
86
for _, u := range users {
87
87
posts = append(posts, u.Post(t, makeRandomPost()))
88
88
}
89
89
}
90
90
91
-
for i := 0; i < 5; i++ {
91
+
for i := 0; i < likeiter; i++ {
92
92
for _, u := range users {
93
93
u.Like(t, posts[rand.Intn(len(posts))])
94
94
}
···
98
98
}
99
99
100
100
func TestBGSMultiPDS(t *testing.T) {
101
-
t.Skip()
102
101
assert := assert.New(t)
103
102
_ = assert
104
103
didr := testPLC(t)
···
120
119
}
121
120
122
121
randomFollows(t, users)
123
-
socialSim(t, users)
122
+
socialSim(t, users, 10, 10)
124
123
125
124
var users2 []*testUser
126
125
for i := 0; i < 5; i++ {
···
128
127
}
129
128
130
129
randomFollows(t, users2)
131
-
p2posts := socialSim(t, users2)
130
+
p2posts := socialSim(t, users2, 10, 10)
132
131
133
132
randomFollows(t, append(users, users2...))
134
133
···
136
135
137
136
// now if we make posts on pds 2, the bgs will not hear about those new posts
138
137
139
-
p2posts2 := socialSim(t, users2)
138
+
p2posts2 := socialSim(t, users2, 10, 10)
140
139
141
140
time.Sleep(time.Second)
142
141
···
144
143
time.Sleep(time.Millisecond * 50)
145
144
146
145
// Now, the bgs will discover a gap, and have to catch up somehow
147
-
socialSim(t, users2)
146
+
socialSim(t, users2, 1, 0)
147
+
148
+
time.Sleep(time.Second)
148
149
149
150
// we expect the bgs to learn about posts that it didnt directly see from
150
151
// repos its already partially scraped, as long as its seen *something* after the missing post
···
154
155
if err != nil {
155
156
t.Fatal(err)
156
157
}
158
+
}
159
+
160
+
func TestBGSMultiGap(t *testing.T) {
161
+
assert := assert.New(t)
162
+
_ = assert
163
+
didr := testPLC(t)
164
+
p1 := mustSetupPDS(t, "localhost:5195", ".pdsuno", didr)
165
+
p1.Run(t)
166
+
167
+
p2 := mustSetupPDS(t, "localhost:5196", ".pdsdos", didr)
168
+
p2.Run(t)
169
+
170
+
b1 := mustSetupBGS(t, "localhost:8291", didr)
171
+
b1.Run(t)
172
+
173
+
p1.RequestScraping(t, b1)
174
+
time.Sleep(time.Millisecond * 50)
175
+
176
+
users := []*testUser{p1.MustNewUser(t, usernames[0]+".pdsuno")}
157
177
178
+
socialSim(t, users, 10, 0)
179
+
180
+
users2 := []*testUser{p2.MustNewUser(t, usernames[1]+".pdsdos")}
181
+
182
+
p2posts := socialSim(t, users2, 10, 0)
183
+
184
+
users[0].Reply(t, p2posts[0], p2posts[0], "what a wonderful life")
185
+
time.Sleep(time.Millisecond * 50)
186
+
187
+
ctx := context.Background()
188
+
_, err := b1.bgs.Index.GetPost(ctx, p2posts[3].Uri)
189
+
if err != nil {
190
+
t.Fatal(err)
191
+
}
192
+
193
+
// now if we make posts on pds 2, the bgs will not hear about those new posts
194
+
195
+
p2posts2 := socialSim(t, users2, 10, 0)
196
+
197
+
time.Sleep(time.Second)
198
+
199
+
p2.RequestScraping(t, b1)
200
+
time.Sleep(time.Millisecond * 50)
201
+
202
+
// Now, the bgs will discover a gap, and have to catch up somehow
203
+
socialSim(t, users2, 1, 0)
204
+
205
+
time.Sleep(time.Second)
206
+
207
+
// we expect the bgs to learn about posts that it didnt directly see from
208
+
// repos its already partially scraped, as long as its seen *something* after the missing post
209
+
// this is the 'catchup' process
210
+
_, err = b1.bgs.Index.GetPost(ctx, p2posts2[4].Uri)
211
+
if err != nil {
212
+
t.Fatal(err)
213
+
}
158
214
}