this repo has no description
implement repo tombstone handler (#340)
authored by
Whyrusleeping
and committed by
GitHub
cf60dd25
9b35e98e
+103
-9
+70
-1
bgs/bgs.go
+70
-1
bgs/bgs.go
···
453
453
// TakenDown is set to true if the user in question has been taken down.
454
454
// A user in this state will have all future events related to it dropped
455
455
// and no data about this user will be served.
456
-
TakenDown bool
456
+
TakenDown bool
457
+
Tombstoned bool
457
458
}
458
459
459
460
type addTargetBody struct {
···
776
777
return fmt.Errorf("rebase was true in event seq:%d,host:%s", evt.Seq, host.Host)
777
778
}
778
779
780
+
if host.ID != u.PDS {
781
+
log.Infow("received event for repo from different pds than expected", "repo", evt.Repo, "expPds", u.PDS, "gotPds", host.Host)
782
+
subj, err := bgs.createExternalUser(ctx, evt.Repo)
783
+
if err != nil {
784
+
return err
785
+
}
786
+
787
+
if subj.PDS != host.ID {
788
+
return fmt.Errorf("event from non-authoritative pds")
789
+
}
790
+
}
791
+
792
+
if u.Tombstoned {
793
+
// we've checked the authority of the users PDS, so reinstate the account
794
+
if err := bgs.db.Model(&User{}).Where("id = ?", u.ID).UpdateColumn("tombstoned", false).Error; err != nil {
795
+
return fmt.Errorf("failed to un-tombstone a user: %w", err)
796
+
}
797
+
798
+
ai, err := bgs.Index.LookupUser(ctx, u.ID)
799
+
if err != nil {
800
+
return fmt.Errorf("failed to look up user (tombstone recover): %w", err)
801
+
}
802
+
803
+
// Now a simple re-crawl should suffice to bring the user back online
804
+
return bgs.Index.Crawler.AddToCatchupQueue(ctx, host, ai, evt)
805
+
}
806
+
779
807
// skip the fast path for rebases or if the user is already in the slow path
780
808
if bgs.Index.Crawler.RepoInSlowPath(ctx, host, u.ID) {
781
809
rebasesCounter.WithLabelValues(host.Host).Add(1)
···
861
889
}
862
890
863
891
return nil
892
+
case env.RepoTombstone != nil:
893
+
if err := bgs.handleRepoTombstone(ctx, host, env.RepoTombstone); err != nil {
894
+
return err
895
+
}
896
+
897
+
return nil
864
898
default:
865
899
return fmt.Errorf("invalid fed event")
866
900
}
867
901
}
868
902
903
+
func (bgs *BGS) handleRepoTombstone(ctx context.Context, pds *models.PDS, evt *atproto.SyncSubscribeRepos_Tombstone) error {
904
+
u, err := bgs.lookupUserByDid(ctx, evt.Did)
905
+
if err != nil {
906
+
return err
907
+
}
908
+
909
+
if u.PDS != pds.ID {
910
+
return fmt.Errorf("unauthoritative tombstone event from %s for %s", pds.Host, evt.Did)
911
+
}
912
+
913
+
if err := bgs.db.Model(&User{}).Where("id = ?", u.ID).UpdateColumns(map[string]any{
914
+
"tombstoned": true,
915
+
"handle": nil,
916
+
}).Error; err != nil {
917
+
return err
918
+
}
919
+
920
+
if err := bgs.db.Model(&models.ActorInfo{}).Where("uid = ?", u.ID).UpdateColumns(map[string]any{
921
+
"handle": nil,
922
+
}).Error; err != nil {
923
+
return err
924
+
}
925
+
926
+
// delete data from carstore
927
+
if err := bgs.repoman.TakeDownRepo(ctx, u.ID); err != nil {
928
+
// don't let a failure here prevent us from propagating this event
929
+
log.Errorf("failed to delete user data from carstore: %s", err)
930
+
}
931
+
932
+
return bgs.events.AddEvent(ctx, &events.XRPCStreamEvent{
933
+
RepoTombstone: evt,
934
+
})
935
+
}
936
+
869
937
func (s *BGS) syncUserBlobs(ctx context.Context, pds *models.PDS, user models.Uid, blobs []string) error {
870
938
if s.blobs == nil {
871
939
log.Debugf("blob syncing disabled")
···
1009
1077
return nil, fmt.Errorf("failed to update users pds: %w", err)
1010
1078
}
1011
1079
1080
+
exu.PDS = peering.ID
1012
1081
}
1013
1082
1014
1083
if exu.Handle.String != handle {
+33
-8
bgs/handlers.go
+33
-8
bgs/handlers.go
···
21
21
)
22
22
23
23
func (s *BGS) handleComAtprotoSyncGetRecord(ctx context.Context, collection string, commit string, did string, rkey string) (io.Reader, error) {
24
-
u, err := s.Index.LookupUserByDid(ctx, did)
24
+
u, err := s.lookupUserByDid(ctx, did)
25
25
if err != nil {
26
26
if errors.Is(err, gorm.ErrRecordNotFound) {
27
27
return nil, echo.NewHTTPError(http.StatusNotFound, "user not found")
···
29
29
return nil, echo.NewHTTPError(http.StatusInternalServerError, "failed to lookup user")
30
30
}
31
31
32
+
if u.Tombstoned {
33
+
return nil, fmt.Errorf("account was deleted")
34
+
}
35
+
36
+
if u.TakenDown {
37
+
return nil, fmt.Errorf("account was taken down")
38
+
}
39
+
32
40
reqCid := cid.Undef
33
41
if commit != "" {
34
42
reqCid, err = cid.Decode(commit)
···
37
45
}
38
46
}
39
47
40
-
_, record, err := s.repoman.GetRecord(ctx, u.Uid, collection, rkey, reqCid)
48
+
_, record, err := s.repoman.GetRecord(ctx, u.ID, collection, rkey, reqCid)
41
49
if err != nil {
42
50
return nil, fmt.Errorf("failed to get record: %w", err)
43
51
}
···
52
60
}
53
61
54
62
func (s *BGS) handleComAtprotoSyncGetRepo(ctx context.Context, did string, since string) (io.Reader, error) {
55
-
u, err := s.Index.LookupUserByDid(ctx, did)
63
+
u, err := s.lookupUserByDid(ctx, did)
56
64
if err != nil {
57
65
if errors.Is(err, gorm.ErrRecordNotFound) {
58
66
return nil, echo.NewHTTPError(http.StatusNotFound, "user not found")
···
60
68
return nil, echo.NewHTTPError(http.StatusInternalServerError, "failed to lookup user")
61
69
}
62
70
71
+
if u.Tombstoned {
72
+
return nil, fmt.Errorf("account was deleted")
73
+
}
74
+
75
+
if u.TakenDown {
76
+
return nil, fmt.Errorf("account was taken down")
77
+
}
78
+
63
79
// TODO: stream the response
64
80
buf := new(bytes.Buffer)
65
-
if err := s.repoman.ReadRepo(ctx, u.Uid, since, buf); err != nil {
81
+
if err := s.repoman.ReadRepo(ctx, u.ID, since, buf); err != nil {
66
82
return nil, fmt.Errorf("failed to read repo: %w", err)
67
83
}
68
84
···
158
174
}
159
175
160
176
users := []User{}
161
-
if err := s.db.Model(&User{}).Where("id > ?", c).Order("id").Limit(limit).Find(&users).Error; err != nil {
177
+
if err := s.db.Model(&User{}).Where("id > ? AND NOT tombstoned AND NOT taken_down", c).Order("id").Limit(limit).Find(&users).Error; err != nil {
162
178
if err == gorm.ErrRecordNotFound {
163
179
return &comatprototypes.SyncListRepos_Output{}, nil
164
180
}
···
175
191
176
192
for i := range users {
177
193
user := users[i]
194
+
178
195
root, err := s.repoman.GetRepoRoot(ctx, user.ID)
179
196
if err != nil {
180
197
return nil, fmt.Errorf("failed to get repo root for (%s): %w", user.Did, err)
···
194
211
}
195
212
196
213
func (s *BGS) handleComAtprotoSyncGetLatestCommit(ctx context.Context, did string) (*comatprototypes.SyncGetLatestCommit_Output, error) {
197
-
u, err := s.Index.LookupUserByDid(ctx, did)
214
+
u, err := s.lookupUserByDid(ctx, did)
198
215
if err != nil {
199
216
if errors.Is(err, gorm.ErrRecordNotFound) {
200
217
return nil, echo.NewHTTPError(http.StatusNotFound, "user not found")
···
202
219
return nil, echo.NewHTTPError(http.StatusInternalServerError, "failed to lookup user")
203
220
}
204
221
205
-
root, err := s.repoman.GetRepoRoot(ctx, u.Uid)
222
+
if u.Tombstoned {
223
+
return nil, fmt.Errorf("account was deleted")
224
+
}
225
+
226
+
if u.TakenDown {
227
+
return nil, fmt.Errorf("account was taken down")
228
+
}
229
+
230
+
root, err := s.repoman.GetRepoRoot(ctx, u.ID)
206
231
if err != nil {
207
232
return nil, fmt.Errorf("failed to get repo root: %w", err)
208
233
}
209
234
210
-
rev, err := s.repoman.GetRepoRev(ctx, u.Uid)
235
+
rev, err := s.repoman.GetRepoRev(ctx, u.ID)
211
236
if err != nil {
212
237
return nil, fmt.Errorf("failed to get repo rev: %w", err)
213
238
}