this repo has no description
Avoid using mst diff to calculate ops, trust pds (#205)
Running the diff is pretty expensive, and the primary issue that could
come up with trusting the pds ops is that they could leave out ops
referencing records being added, which seems like a bit of a non-attack
as its just the pds inconveniencing its own users.
authored by
Whyrusleeping
and committed by
GitHub
279ef73a
7ec50856
+41
-70
+1
-1
bgs/bgs.go
+1
-1
bgs/bgs.go
···
557
557
return bgs.Index.Crawler.AddToCatchupQueue(ctx, host, ai, evt)
558
558
}
559
559
560
-
if err := bgs.repoman.HandleExternalUserEvent(ctx, host.ID, u.ID, u.Did, (*cid.Cid)(evt.Prev), evt.Blocks); err != nil {
560
+
if err := bgs.repoman.HandleExternalUserEvent(ctx, host.ID, u.ID, u.Did, (*cid.Cid)(evt.Prev), evt.Blocks, evt.Ops); err != nil {
561
561
log.Warnw("failed handling event", "err", err, "host", host.Host, "seq", evt.Seq, "repo", u.Did, "prev", stringLink(evt.Prev), "commit", evt.Commit.String())
562
562
563
563
if errors.Is(err, carstore.ErrRepoBaseMismatch) {
+10
pds/handlers.go
+10
pds/handlers.go
···
10
10
comatprototypes "github.com/bluesky-social/indigo/api/atproto"
11
11
appbskytypes "github.com/bluesky-social/indigo/api/bsky"
12
12
lexutil "github.com/bluesky-social/indigo/lex/util"
13
+
"github.com/bluesky-social/indigo/models"
13
14
"github.com/ipfs/go-cid"
14
15
"github.com/lestrrat-go/jwx/v2/jwt"
15
16
)
···
322
323
323
324
u.Did = d
324
325
if err := s.db.Save(&u).Error; err != nil {
326
+
return nil, err
327
+
}
328
+
329
+
ai := &models.ActorInfo{
330
+
Uid: u.ID,
331
+
Did: d,
332
+
Handle: body.Handle,
333
+
}
334
+
if err := s.db.Create(ai).Error; err != nil {
325
335
return nil, err
326
336
}
327
337
+1
-1
pds/server.go
+1
-1
pds/server.go
···
143
143
u.ID = subj.Uid
144
144
}
145
145
146
-
return s.repoman.HandleExternalUserEvent(ctx, host.ID, u.ID, u.Did, (*cid.Cid)(evt.Prev), evt.Blocks)
146
+
return s.repoman.HandleExternalUserEvent(ctx, host.ID, u.ID, u.Did, (*cid.Cid)(evt.Prev), evt.Blocks, evt.Ops)
147
147
default:
148
148
return fmt.Errorf("invalid fed event")
149
149
}
+16
-7
repomgr/ingest_test.go
+16
-7
repomgr/ingest_test.go
···
8
8
"path/filepath"
9
9
"testing"
10
10
11
+
atproto "github.com/bluesky-social/indigo/api/atproto"
11
12
bsky "github.com/bluesky-social/indigo/api/bsky"
12
13
"github.com/bluesky-social/indigo/carstore"
13
14
"github.com/bluesky-social/indigo/models"
···
122
123
ctx := context.TODO()
123
124
var prev *cid.Cid
124
125
for i := 0; i < 5; i++ {
125
-
slice, head := doPost(t, cs2, did, prev, i)
126
+
slice, head, tid := doPost(t, cs2, did, prev, i)
127
+
128
+
ops := []*atproto.SyncSubscribeRepos_RepoOp{
129
+
{
130
+
Action: "create",
131
+
Path: "app.bsky.feed.post/" + tid,
132
+
},
133
+
}
126
134
127
-
if err := repoman.HandleExternalUserEvent(ctx, 1, 1, did, prev, slice); err != nil {
135
+
if err := repoman.HandleExternalUserEvent(ctx, 1, 1, did, prev, slice, ops); err != nil {
128
136
t.Fatal(err)
129
137
}
130
138
···
135
143
136
144
// now do a few outside of the standard event stream flow
137
145
for i := 0; i < 5; i++ {
138
-
_, head := doPost(t, cs2, did, prev, i)
146
+
_, head, _ := doPost(t, cs2, did, prev, i)
139
147
prev = &head
140
148
}
141
149
···
149
157
}
150
158
}
151
159
152
-
func doPost(t *testing.T, cs *carstore.CarStore, did string, prev *cid.Cid, postid int) ([]byte, cid.Cid) {
160
+
func doPost(t *testing.T, cs *carstore.CarStore, did string, prev *cid.Cid, postid int) ([]byte, cid.Cid, string) {
153
161
ctx := context.TODO()
154
162
ds, err := cs.NewDeltaSession(ctx, 1, prev)
155
163
if err != nil {
···
158
166
159
167
r := repo.NewRepo(ctx, did, ds)
160
168
161
-
if _, _, err := r.CreateRecord(ctx, "app.bsky.feed.post", &bsky.FeedPost{
169
+
_, tid, err := r.CreateRecord(ctx, "app.bsky.feed.post", &bsky.FeedPost{
162
170
Text: fmt.Sprintf("hello friend %d", postid),
163
-
}); err != nil {
171
+
})
172
+
if err != nil {
164
173
t.Fatal(err)
165
174
}
166
175
···
174
183
t.Fatal(err)
175
184
}
176
185
177
-
return slice, root
186
+
return slice, root, tid
178
187
}
+13
-61
repomgr/repomgr.go
+13
-61
repomgr/repomgr.go
···
13
13
bsky "github.com/bluesky-social/indigo/api/bsky"
14
14
"github.com/bluesky-social/indigo/carstore"
15
15
lexutil "github.com/bluesky-social/indigo/lex/util"
16
-
"github.com/bluesky-social/indigo/models"
17
16
"github.com/bluesky-social/indigo/mst"
18
17
"github.com/bluesky-social/indigo/repo"
19
18
"github.com/bluesky-social/indigo/util"
···
426
425
User: user,
427
426
NewRoot: root,
428
427
Ops: []RepoOp{{
429
-
Kind: EvtKindInitActor,
430
-
ActorInfo: &ActorInfo{
431
-
Did: did,
432
-
Handle: handle,
433
-
DisplayName: displayname,
434
-
Type: actortype,
435
-
},
428
+
Kind: EvtKindCreateRecord,
429
+
Collection: "app.bsky.actor.profile",
430
+
Rkey: "self",
431
+
Record: profile,
436
432
}},
437
433
RepoSlice: rslice,
438
434
})
···
679
675
return nil
680
676
}
681
677
682
-
func (rm *RepoManager) HandleExternalUserEvent(ctx context.Context, pdsid uint, uid util.Uid, did string, prev *cid.Cid, carslice []byte) error {
678
+
func (rm *RepoManager) HandleExternalUserEvent(ctx context.Context, pdsid uint, uid util.Uid, did string, prev *cid.Cid, carslice []byte, ops []*atproto.SyncSubscribeRepos_RepoOp) error {
683
679
ctx, span := otel.Tracer("repoman").Start(ctx, "HandleExternalUserEvent")
684
680
defer span.End()
685
681
···
702
698
return err
703
699
}
704
700
705
-
var pcid cid.Cid
706
-
if prev != nil {
707
-
pcid = *prev
708
-
}
709
-
710
-
ops, err := r.DiffSince(ctx, pcid)
711
-
if err != nil {
712
-
return fmt.Errorf("calculating operations in event: %w", err)
713
-
}
714
701
var evtops []RepoOp
715
702
716
-
if prev == nil {
717
-
// send an implicit init actor event
718
-
var ai models.ActorInfo
719
-
if err := rm.db.First(&ai, "uid = ?", uid).Error; err != nil {
720
-
return fmt.Errorf("expected initialized user: %w", err)
721
-
}
722
-
723
-
evtops = append(evtops, RepoOp{
724
-
Kind: EvtKindInitActor,
725
-
ActorInfo: &ActorInfo{
726
-
Did: ai.Did,
727
-
Handle: ai.Handle,
728
-
DisplayName: ai.DisplayName,
729
-
Type: ai.Type,
730
-
},
731
-
})
732
-
}
733
-
734
703
for _, op := range ops {
735
-
parts := strings.SplitN(op.Rpath, "/", 2)
704
+
parts := strings.SplitN(op.Path, "/", 2)
736
705
if len(parts) != 2 {
737
706
return fmt.Errorf("invalid rpath in mst diff, must have collection and rkey")
738
707
}
739
708
740
-
switch op.Op {
741
-
case "add":
742
-
recid, rec, err := r.GetRecord(ctx, op.Rpath)
709
+
switch EventKind(op.Action) {
710
+
case EvtKindCreateRecord:
711
+
recid, rec, err := r.GetRecord(ctx, op.Path)
743
712
if err != nil {
744
713
return fmt.Errorf("reading changed record from car slice: %w", err)
745
714
}
···
751
720
Record: rec,
752
721
RecCid: &recid,
753
722
})
754
-
/*
755
-
case EvtKindInitActor:
756
-
var ai models.ActorInfo
757
-
if err := rm.db.First(&ai, "id = ?", uid).Error; err != nil {
758
-
return fmt.Errorf("expected initialized user: %w", err)
759
-
}
760
-
761
-
evtops = append(evtops, RepoOp{
762
-
Kind: EvtKindInitActor,
763
-
ActorInfo: &ActorInfo{
764
-
Did: ai.Did,
765
-
Handle: ai.Handle,
766
-
DisplayName: ai.DisplayName,
767
-
Type: ai.Type,
768
-
},
769
-
})
770
-
*/
771
-
case "mut":
772
-
recid, rec, err := r.GetRecord(ctx, op.Rpath)
723
+
case EvtKindUpdateRecord:
724
+
recid, rec, err := r.GetRecord(ctx, op.Path)
773
725
if err != nil {
774
726
return fmt.Errorf("reading changed record from car slice: %w", err)
775
727
}
···
781
733
Record: rec,
782
734
RecCid: &recid,
783
735
})
784
-
case "del":
736
+
case EvtKindDeleteRecord:
785
737
evtops = append(evtops, RepoOp{
786
738
Kind: EvtKindDeleteRecord,
787
739
Collection: parts[0],
788
740
Rkey: parts[1],
789
741
})
790
742
default:
791
-
return fmt.Errorf("unrecognized external user event kind: %q", op.Op)
743
+
return fmt.Errorf("unrecognized external user event kind: %q", op.Action)
792
744
}
793
745
}
794
746