this repo has no description
back to relay pkg
+356
-314
+51
-48
cmd/relayered/account.go
cmd/relayered/relay/account.go
+51
-48
cmd/relayered/account.go
cmd/relayered/relay/account.go
···
1
-
package main
1
+
package relay
2
2
3
3
import (
4
4
"context"
···
11
11
comatproto "github.com/bluesky-social/indigo/api/atproto"
12
12
"github.com/bluesky-social/indigo/atproto/syntax"
13
13
"github.com/bluesky-social/indigo/cmd/relayered/models"
14
-
"github.com/bluesky-social/indigo/cmd/relayered/slurper"
14
+
"github.com/bluesky-social/indigo/cmd/relayered/relay/slurper"
15
15
"github.com/bluesky-social/indigo/xrpc"
16
16
17
17
"github.com/ipfs/go-cid"
18
18
"gorm.io/gorm"
19
19
)
20
20
21
-
func (svc *Service) DidToUid(ctx context.Context, did string) (models.Uid, error) {
22
-
xu, err := svc.lookupUserByDid(ctx, did)
21
+
var ErrNotFound = errors.New("not found")
22
+
var ErrUserStatusUnavailable = errors.New("user status unavailable")
23
+
24
+
func (r *Relay) DidToUid(ctx context.Context, did string) (models.Uid, error) {
25
+
xu, err := r.LookupUserByDid(ctx, did)
23
26
if err != nil {
24
27
return 0, err
25
28
}
···
29
32
return xu.ID, nil
30
33
}
31
34
32
-
func (svc *Service) lookupUserByDid(ctx context.Context, did string) (*slurper.Account, error) {
35
+
func (r *Relay) LookupUserByDid(ctx context.Context, did string) (*slurper.Account, error) {
33
36
ctx, span := tracer.Start(ctx, "lookupUserByDid")
34
37
defer span.End()
35
38
36
-
cu, ok := svc.userCache.Get(did)
39
+
cu, ok := r.userCache.Get(did)
37
40
if ok {
38
41
return cu, nil
39
42
}
40
43
41
44
var u slurper.Account
42
-
if err := svc.db.Find(&u, "did = ?", did).Error; err != nil {
45
+
if err := r.db.Find(&u, "did = ?", did).Error; err != nil {
43
46
return nil, err
44
47
}
45
48
···
47
50
return nil, gorm.ErrRecordNotFound
48
51
}
49
52
50
-
svc.userCache.Add(did, &u)
53
+
r.userCache.Add(did, &u)
51
54
52
55
return &u, nil
53
56
}
54
57
55
-
func (svc *Service) lookupUserByUID(ctx context.Context, uid models.Uid) (*slurper.Account, error) {
58
+
func (r *Relay) LookupUserByUID(ctx context.Context, uid models.Uid) (*slurper.Account, error) {
56
59
ctx, span := tracer.Start(ctx, "lookupUserByUID")
57
60
defer span.End()
58
61
59
62
var u slurper.Account
60
-
if err := svc.db.Find(&u, "id = ?", uid).Error; err != nil {
63
+
if err := r.db.Find(&u, "id = ?", uid).Error; err != nil {
61
64
return nil, err
62
65
}
63
66
···
68
71
return &u, nil
69
72
}
70
73
71
-
func (svc *Service) newUser(ctx context.Context, host *slurper.PDS, did string) (*slurper.Account, error) {
74
+
func (r *Relay) newUser(ctx context.Context, host *slurper.PDS, did string) (*slurper.Account, error) {
72
75
newUsersDiscovered.Inc()
73
76
start := time.Now()
74
-
account, err := svc.syncPDSAccount(ctx, did, host, nil)
77
+
account, err := r.syncPDSAccount(ctx, did, host, nil)
75
78
newUserDiscoveryDuration.Observe(time.Since(start).Seconds())
76
79
if err != nil {
77
80
repoCommitsResultCounter.WithLabelValues(host.Host, "uerr").Inc()
···
87
90
// did is the user
88
91
// host is the PDS we received this from, not necessarily the canonical PDS in the DID document
89
92
// cachedAccount is (optionally) the account that we have already looked up from cache or database
90
-
func (svc *Service) syncPDSAccount(ctx context.Context, did string, host *slurper.PDS, cachedAccount *slurper.Account) (*slurper.Account, error) {
93
+
func (r *Relay) syncPDSAccount(ctx context.Context, did string, host *slurper.PDS, cachedAccount *slurper.Account) (*slurper.Account, error) {
91
94
ctx, span := tracer.Start(ctx, "syncPDSAccount")
92
95
defer span.End()
93
96
94
97
externalUserCreationAttempts.Inc()
95
98
96
-
svc.log.Debug("create external user", "did", did)
99
+
r.Logger.Debug("create external user", "did", did)
97
100
98
101
// lookup identity so that we know a DID's canonical source PDS
99
102
pdid, err := syntax.ParseDID(did)
100
103
if err != nil {
101
104
return nil, fmt.Errorf("bad did %#v, %w", did, err)
102
105
}
103
-
ident, err := svc.dir.LookupDID(ctx, pdid)
106
+
ident, err := r.dir.LookupDID(ctx, pdid)
104
107
if err != nil {
105
108
return nil, fmt.Errorf("no ident for did %s, %w", did, err)
106
109
}
107
110
if len(ident.Services) == 0 {
108
111
return nil, fmt.Errorf("no services for did %s", did)
109
112
}
110
-
pdsService, ok := ident.Services["atproto_pds"]
113
+
pdsRelay, ok := ident.Services["atproto_pds"]
111
114
if !ok {
112
115
return nil, fmt.Errorf("no atproto_pds service for did %s", did)
113
116
}
114
-
durl, err := url.Parse(pdsService.URL)
117
+
durl, err := url.Parse(pdsRelay.URL)
115
118
if err != nil {
116
-
return nil, fmt.Errorf("pds bad url %#v, %w", pdsService.URL, err)
119
+
return nil, fmt.Errorf("pds bad url %#v, %w", pdsRelay.URL, err)
117
120
}
118
121
119
122
// is the canonical PDS banned?
120
-
ban, err := svc.domainIsBanned(ctx, durl.Host)
123
+
ban, err := r.DomainIsBanned(ctx, durl.Host)
121
124
if err != nil {
122
125
return nil, fmt.Errorf("failed to check pds ban status: %w", err)
123
126
}
···
137
140
// we got the message from an intermediate relay
138
141
// check our db for info on canonical PDS
139
142
var peering slurper.PDS
140
-
if err := svc.db.Find(&peering, "host = ?", durl.Host).Error; err != nil {
141
-
svc.log.Error("failed to find pds", "host", durl.Host)
143
+
if err := r.db.Find(&peering, "host = ?", durl.Host).Error; err != nil {
144
+
r.Logger.Error("failed to find pds", "host", durl.Host)
142
145
return nil, err
143
146
}
144
147
canonicalHost = &peering
···
152
155
// we got an event from a non-canonical PDS (an intermediate relay)
153
156
// a non-canonical PDS we haven't seen before; ping it to make sure it's real
154
157
// TODO: what do we actually want to track about the source we immediately got this message from vs the canonical PDS?
155
-
svc.log.Warn("pds discovered in new user flow", "pds", durl.String(), "did", did)
158
+
r.Logger.Warn("pds discovered in new user flow", "pds", durl.String(), "did", did)
156
159
157
160
// Do a trivial API request against the PDS to verify that it exists
158
161
pclient := &xrpc.Client{Host: durl.String()}
159
-
svc.config.ApplyPDSClientSettings(pclient)
162
+
r.Config.ApplyPDSClientSettings(pclient)
160
163
cfg, err := comatproto.ServerDescribeServer(ctx, pclient)
161
164
if err != nil {
162
165
// TODO: failing this shouldn't halt our indexing
···
169
172
// could check other things, a valid response is good enough for now
170
173
canonicalHost.Host = durl.Host
171
174
canonicalHost.SSL = (durl.Scheme == "https")
172
-
canonicalHost.RateLimit = float64(svc.slurper.DefaultPerSecondLimit)
173
-
canonicalHost.HourlyEventLimit = svc.slurper.DefaultPerHourLimit
174
-
canonicalHost.DailyEventLimit = svc.slurper.DefaultPerDayLimit
175
-
canonicalHost.RepoLimit = svc.slurper.DefaultRepoLimit
175
+
canonicalHost.RateLimit = float64(r.Slurper.DefaultPerSecondLimit)
176
+
canonicalHost.HourlyEventLimit = r.Slurper.DefaultPerHourLimit
177
+
canonicalHost.DailyEventLimit = r.Slurper.DefaultPerDayLimit
178
+
canonicalHost.RepoLimit = r.Slurper.DefaultRepoLimit
176
179
177
-
if svc.ssl && !canonicalHost.SSL {
178
-
return nil, fmt.Errorf("did references non-ssl PDS, this is disallowed in prod: %q %q", did, pdsService.URL)
180
+
if r.Config.SSL && !canonicalHost.SSL {
181
+
return nil, fmt.Errorf("did references non-ssl PDS, this is disallowed in prod: %q %q", did, pdsRelay.URL)
179
182
}
180
183
181
-
if err := svc.db.Create(&canonicalHost).Error; err != nil {
184
+
if err := r.db.Create(&canonicalHost).Error; err != nil {
182
185
return nil, err
183
186
}
184
187
}
···
193
196
}
194
197
195
198
// this lock just governs the lower half of this function
196
-
svc.extUserLk.Lock()
197
-
defer svc.extUserLk.Unlock()
199
+
r.extUserLk.Lock()
200
+
defer r.extUserLk.Unlock()
198
201
199
202
if cachedAccount == nil {
200
-
cachedAccount, err = svc.lookupUserByDid(ctx, did)
203
+
cachedAccount, err = r.LookupUserByDid(ctx, did)
201
204
}
202
205
if errors.Is(err, ErrNotFound) || errors.Is(err, gorm.ErrRecordNotFound) {
203
206
err = nil
···
209
212
caPDS := cachedAccount.GetPDS()
210
213
if caPDS != canonicalHost.ID {
211
214
// Account is now on a different PDS, update
212
-
err = svc.db.Transaction(func(tx *gorm.DB) error {
215
+
err = r.db.Transaction(func(tx *gorm.DB) error {
213
216
if caPDS != 0 {
214
217
// decrement prior PDS's account count
215
218
tx.Model(&slurper.PDS{}).Where("id = ?", caPDS).Update("repo_count", gorm.Expr("repo_count - 1"))
···
234
237
PDS: canonicalHost.ID,
235
238
}
236
239
237
-
err = svc.db.Transaction(func(tx *gorm.DB) error {
240
+
err = r.db.Transaction(func(tx *gorm.DB) error {
238
241
res := tx.Model(&slurper.PDS{}).Where("id = ? AND repo_count < repo_limit", canonicalHost.ID).Update("repo_count", gorm.Expr("repo_count + 1"))
239
242
if res.Error != nil {
240
243
return fmt.Errorf("failed to increment repo count for pds %q: %w", canonicalHost.Host, res.Error)
241
244
}
242
245
if terr := tx.Create(&newAccount).Error; terr != nil {
243
-
svc.log.Error("failed to create user", "did", newAccount.Did, "err", terr)
246
+
r.Logger.Error("failed to create user", "did", newAccount.Did, "err", terr)
244
247
return fmt.Errorf("failed to create other pds user: %w", terr)
245
248
}
246
249
return nil
247
250
})
248
251
if err != nil {
249
-
svc.log.Error("user create and pds inc err", "err", err)
252
+
r.Logger.Error("user create and pds inc err", "err", err)
250
253
return nil, err
251
254
}
252
255
253
-
svc.userCache.Add(did, &newAccount)
256
+
r.userCache.Add(did, &newAccount)
254
257
255
258
return &newAccount, nil
256
259
}
257
260
258
-
func (svc *Service) TakeDownRepo(ctx context.Context, did string) error {
259
-
u, err := svc.lookupUserByDid(ctx, did)
261
+
func (r *Relay) TakeDownRepo(ctx context.Context, did string) error {
262
+
u, err := r.LookupUserByDid(ctx, did)
260
263
if err != nil {
261
264
return err
262
265
}
263
266
264
-
if err := svc.db.Model(slurper.Account{}).Where("id = ?", u.ID).Update("taken_down", true).Error; err != nil {
267
+
if err := r.db.Model(slurper.Account{}).Where("id = ?", u.ID).Update("taken_down", true).Error; err != nil {
265
268
return err
266
269
}
267
270
u.SetTakenDown(true)
268
271
269
-
if err := svc.events.TakeDownRepo(ctx, u.ID); err != nil {
272
+
if err := r.Events.TakeDownRepo(ctx, u.ID); err != nil {
270
273
return err
271
274
}
272
275
273
276
return nil
274
277
}
275
278
276
-
func (svc *Service) ReverseTakedown(ctx context.Context, did string) error {
277
-
u, err := svc.lookupUserByDid(ctx, did)
279
+
func (r *Relay) ReverseTakedown(ctx context.Context, did string) error {
280
+
u, err := r.LookupUserByDid(ctx, did)
278
281
if err != nil {
279
282
return err
280
283
}
281
284
282
-
if err := svc.db.Model(slurper.Account{}).Where("id = ?", u.ID).Update("taken_down", false).Error; err != nil {
285
+
if err := r.db.Model(slurper.Account{}).Where("id = ?", u.ID).Update("taken_down", false).Error; err != nil {
283
286
return err
284
287
}
285
288
u.SetTakenDown(false)
···
287
290
return nil
288
291
}
289
292
290
-
func (svc *Service) GetRepoRoot(ctx context.Context, user models.Uid) (cid.Cid, error) {
293
+
func (r *Relay) GetRepoRoot(ctx context.Context, user models.Uid) (cid.Cid, error) {
291
294
var prevState slurper.AccountPreviousState
292
-
err := svc.db.First(&prevState, user).Error
295
+
err := r.db.First(&prevState, user).Error
293
296
if err == nil {
294
297
return prevState.Cid.CID, nil
295
298
} else if errors.Is(err, gorm.ErrRecordNotFound) {
296
299
return cid.Cid{}, ErrUserStatusUnavailable
297
300
} else {
298
-
svc.log.Error("user db err", "err", err)
301
+
r.Logger.Error("user db err", "err", err)
299
302
return cid.Cid{}, fmt.Errorf("user prev db err, %w", err)
300
303
}
301
304
}
+7
-7
cmd/relayered/domain_ban.go
cmd/relayered/relay/domain_ban.go
+7
-7
cmd/relayered/domain_ban.go
cmd/relayered/relay/domain_ban.go
···
1
-
package main
1
+
package relay
2
2
3
3
import (
4
4
"context"
5
5
"strings"
6
6
7
-
"github.com/bluesky-social/indigo/cmd/relayered/slurper"
7
+
"github.com/bluesky-social/indigo/cmd/relayered/relay/slurper"
8
8
)
9
9
10
-
// domainIsBanned checks if the given host is banned, starting with the host
10
+
// DomainIsBanned checks if the given host is banned, starting with the host
11
11
// itself, then checking every parent domain up to the tld
12
-
func (s *Service) domainIsBanned(ctx context.Context, host string) (bool, error) {
12
+
func (r *Relay) DomainIsBanned(ctx context.Context, host string) (bool, error) {
13
13
// ignore ports when checking for ban status
14
14
hostport := strings.Split(host, ":")
15
15
···
29
29
30
30
for i := 0; i < len(segments)-1; i++ {
31
31
dchk := strings.Join(segments[i:], ".")
32
-
found, err := s.findDomainBan(ctx, dchk)
32
+
found, err := r.findDomainBan(ctx, dchk)
33
33
if err != nil {
34
34
return false, err
35
35
}
···
41
41
return false, nil
42
42
}
43
43
44
-
func (s *Service) findDomainBan(ctx context.Context, host string) (bool, error) {
44
+
func (r *Relay) findDomainBan(ctx context.Context, host string) (bool, error) {
45
45
var db slurper.DomainBan
46
-
if err := s.db.Find(&db, "domain = ?", host).Error; err != nil {
46
+
if err := r.db.Find(&db, "domain = ?", host).Error; err != nil {
47
47
return false, err
48
48
}
49
49
+48
-47
cmd/relayered/firehose.go
cmd/relayered/relay/firehose.go
+48
-47
cmd/relayered/firehose.go
cmd/relayered/relay/firehose.go
···
1
-
package main
1
+
package relay
2
2
3
3
import (
4
4
"context"
···
10
10
comatproto "github.com/bluesky-social/indigo/api/atproto"
11
11
"github.com/bluesky-social/indigo/atproto/syntax"
12
12
"github.com/bluesky-social/indigo/cmd/relayered/models"
13
-
"github.com/bluesky-social/indigo/cmd/relayered/slurper"
13
+
"github.com/bluesky-social/indigo/cmd/relayered/relay/slurper"
14
14
"github.com/bluesky-social/indigo/cmd/relayered/stream"
15
15
16
16
"github.com/ipfs/go-cid"
···
19
19
)
20
20
21
21
// handleFedEvent() is the callback passed to Slurper called from Slurper.handleConnection()
22
-
func (svc *Service) handleFedEvent(ctx context.Context, host *slurper.PDS, env *stream.XRPCStreamEvent) error {
22
+
func (r *Relay) handleFedEvent(ctx context.Context, host *slurper.PDS, env *stream.XRPCStreamEvent) error {
23
23
ctx, span := tracer.Start(ctx, "handleFedEvent")
24
24
defer span.End()
25
25
···
28
28
eventsHandleDuration.WithLabelValues(host.Host).Observe(time.Since(start).Seconds())
29
29
}()
30
30
31
-
eventsReceivedCounter.WithLabelValues(host.Host).Add(1)
31
+
EventsReceivedCounter.WithLabelValues(host.Host).Add(1)
32
32
33
33
switch {
34
34
case env.RepoCommit != nil:
35
35
repoCommitsReceivedCounter.WithLabelValues(host.Host).Add(1)
36
-
return svc.handleCommit(ctx, host, env.RepoCommit)
36
+
return r.handleCommit(ctx, host, env.RepoCommit)
37
37
case env.RepoSync != nil:
38
38
repoSyncReceivedCounter.WithLabelValues(host.Host).Add(1)
39
-
return svc.handleSync(ctx, host, env.RepoSync)
39
+
return r.handleSync(ctx, host, env.RepoSync)
40
40
case env.RepoHandle != nil:
41
41
eventsWarningsCounter.WithLabelValues(host.Host, "handle").Add(1)
42
42
// TODO: rate limit warnings per PDS before we (temporarily?) block them
43
43
return nil
44
44
case env.RepoIdentity != nil:
45
-
svc.log.Info("relay got identity event", "did", env.RepoIdentity.Did)
45
+
r.Logger.Info("relay got identity event", "did", env.RepoIdentity.Did)
46
46
// Flush any cached DID documents for this user
47
-
svc.purgeDidCache(ctx, env.RepoIdentity.Did)
47
+
r.purgeDidCache(ctx, env.RepoIdentity.Did)
48
48
49
49
// Refetch the DID doc and update our cached keys and handle etc.
50
-
account, err := svc.syncPDSAccount(ctx, env.RepoIdentity.Did, host, nil)
50
+
account, err := r.syncPDSAccount(ctx, env.RepoIdentity.Did, host, nil)
51
51
if err != nil {
52
52
return err
53
53
}
54
54
55
55
// Broadcast the identity event to all consumers
56
-
err = svc.events.AddEvent(ctx, &stream.XRPCStreamEvent{
56
+
err = r.Events.AddEvent(ctx, &stream.XRPCStreamEvent{
57
57
RepoIdentity: &comatproto.SyncSubscribeRepos_Identity{
58
58
Did: env.RepoIdentity.Did,
59
59
Seq: env.RepoIdentity.Seq,
···
63
63
PrivUid: account.ID,
64
64
})
65
65
if err != nil {
66
-
svc.log.Error("failed to broadcast Identity event", "error", err, "did", env.RepoIdentity.Did)
66
+
r.Logger.Error("failed to broadcast Identity event", "error", err, "did", env.RepoIdentity.Did)
67
67
return fmt.Errorf("failed to broadcast Identity event: %w", err)
68
68
}
69
69
···
78
78
if env.RepoAccount.Status != nil {
79
79
span.SetAttributes(attribute.String("repo_status", *env.RepoAccount.Status))
80
80
}
81
-
svc.log.Info("relay got account event", "did", env.RepoAccount.Did)
81
+
r.Logger.Info("relay got account event", "did", env.RepoAccount.Did)
82
82
83
83
if !env.RepoAccount.Active && env.RepoAccount.Status == nil {
84
84
accountVerifyWarnings.WithLabelValues(host.Host, "nostat").Inc()
···
86
86
}
87
87
88
88
// Flush any cached DID documents for this user
89
-
svc.purgeDidCache(ctx, env.RepoAccount.Did)
89
+
r.purgeDidCache(ctx, env.RepoAccount.Did)
90
90
91
91
// Refetch the DID doc to make sure the PDS is still authoritative
92
-
account, err := svc.syncPDSAccount(ctx, env.RepoAccount.Did, host, nil)
92
+
account, err := r.syncPDSAccount(ctx, env.RepoAccount.Did, host, nil)
93
93
if err != nil {
94
94
span.RecordError(err)
95
95
return err
···
98
98
// Check if the PDS is still authoritative
99
99
// if not we don't want to be propagating this account event
100
100
if account.GetPDS() != host.ID {
101
-
svc.log.Error("account event from non-authoritative pds",
101
+
r.Logger.Error("account event from non-authoritative pds",
102
102
"seq", env.RepoAccount.Seq,
103
103
"did", env.RepoAccount.Did,
104
104
"event_from", host.Host,
···
115
115
}
116
116
117
117
account.SetUpstreamStatus(repoStatus)
118
-
err = svc.db.Save(account).Error
118
+
err = r.db.Save(account).Error
119
119
if err != nil {
120
120
span.RecordError(err)
121
121
return fmt.Errorf("failed to update account status: %w", err)
···
131
131
}
132
132
133
133
// Broadcast the account event to all consumers
134
-
err = svc.events.AddEvent(ctx, &stream.XRPCStreamEvent{
134
+
err = r.Events.AddEvent(ctx, &stream.XRPCStreamEvent{
135
135
RepoAccount: &comatproto.SyncSubscribeRepos_Account{
136
136
Active: shouldBeActive,
137
137
Did: env.RepoAccount.Did,
···
142
142
PrivUid: account.ID,
143
143
})
144
144
if err != nil {
145
-
svc.log.Error("failed to broadcast Account event", "error", err, "did", env.RepoAccount.Did)
145
+
r.Logger.Error("failed to broadcast Account event", "error", err, "did", env.RepoAccount.Did)
146
146
return fmt.Errorf("failed to broadcast Account event: %w", err)
147
147
}
148
148
···
160
160
}
161
161
}
162
162
163
-
func (svc *Service) handleCommit(ctx context.Context, host *slurper.PDS, evt *comatproto.SyncSubscribeRepos_Commit) error {
164
-
svc.log.Debug("relay got repo append event", "seq", evt.Seq, "pdsHost", host.Host, "repo", evt.Repo)
163
+
func (r *Relay) handleCommit(ctx context.Context, host *slurper.PDS, evt *comatproto.SyncSubscribeRepos_Commit) error {
164
+
r.Logger.Debug("relay got repo append event", "seq", evt.Seq, "pdsHost", host.Host, "repo", evt.Repo)
165
165
166
-
account, err := svc.lookupUserByDid(ctx, evt.Repo)
166
+
account, err := r.LookupUserByDid(ctx, evt.Repo)
167
167
if err != nil {
168
168
if !errors.Is(err, gorm.ErrRecordNotFound) {
169
169
repoCommitsResultCounter.WithLabelValues(host.Host, "nou").Inc()
170
170
return fmt.Errorf("looking up event user: %w", err)
171
171
}
172
172
173
-
account, err = svc.newUser(ctx, host, evt.Repo)
173
+
account, err = r.newUser(ctx, host, evt.Repo)
174
174
if err != nil {
175
175
repoCommitsResultCounter.WithLabelValues(host.Host, "nuerr").Inc()
176
176
return err
···
184
184
ustatus := account.GetUpstreamStatus()
185
185
186
186
if account.GetTakenDown() || ustatus == slurper.AccountStatusTakendown {
187
-
svc.log.Debug("dropping commit event from taken down user", "did", evt.Repo, "seq", evt.Seq, "pdsHost", host.Host)
187
+
r.Logger.Debug("dropping commit event from taken down user", "did", evt.Repo, "seq", evt.Seq, "pdsHost", host.Host)
188
188
repoCommitsResultCounter.WithLabelValues(host.Host, "tdu").Inc()
189
189
return nil
190
190
}
191
191
192
192
if ustatus == slurper.AccountStatusSuspended {
193
-
svc.log.Debug("dropping commit event from suspended user", "did", evt.Repo, "seq", evt.Seq, "pdsHost", host.Host)
193
+
r.Logger.Debug("dropping commit event from suspended user", "did", evt.Repo, "seq", evt.Seq, "pdsHost", host.Host)
194
194
repoCommitsResultCounter.WithLabelValues(host.Host, "susu").Inc()
195
195
return nil
196
196
}
197
197
198
198
if ustatus == slurper.AccountStatusDeactivated {
199
-
svc.log.Debug("dropping commit event from deactivated user", "did", evt.Repo, "seq", evt.Seq, "pdsHost", host.Host)
199
+
r.Logger.Debug("dropping commit event from deactivated user", "did", evt.Repo, "seq", evt.Seq, "pdsHost", host.Host)
200
200
repoCommitsResultCounter.WithLabelValues(host.Host, "du").Inc()
201
201
return nil
202
202
}
···
208
208
209
209
accountPDSId := account.GetPDS()
210
210
if host.ID != accountPDSId && accountPDSId != 0 {
211
-
svc.log.Warn("received event for repo from different pds than expected", "repo", evt.Repo, "expPds", accountPDSId, "gotPds", host.Host)
211
+
r.Logger.Warn("received event for repo from different pds than expected", "repo", evt.Repo, "expPds", accountPDSId, "gotPds", host.Host)
212
212
// Flush any cached DID documents for this user
213
-
svc.purgeDidCache(ctx, evt.Repo)
213
+
r.purgeDidCache(ctx, evt.Repo)
214
214
215
-
account, err = svc.syncPDSAccount(ctx, evt.Repo, host, account)
215
+
account, err = r.syncPDSAccount(ctx, evt.Repo, host, account)
216
216
if err != nil {
217
217
repoCommitsResultCounter.WithLabelValues(host.Host, "uerr2").Inc()
218
218
return err
···
225
225
}
226
226
227
227
var prevState slurper.AccountPreviousState
228
-
err = svc.db.First(&prevState, account.ID).Error
228
+
err = r.db.First(&prevState, account.ID).Error
229
229
prevP := &prevState
230
230
if errors.Is(err, gorm.ErrRecordNotFound) {
231
231
prevP = nil
232
232
} else if err != nil {
233
-
svc.log.Error("failed to get previous root", "err", err)
233
+
r.Logger.Error("failed to get previous root", "err", err)
234
234
prevP = nil
235
235
}
236
236
dbPrevRootStr := ""
···
248
248
if evt.PrevData != nil {
249
249
evtPrevDataStr = ((*cid.Cid)(evt.PrevData)).String()
250
250
}
251
-
newRootCid, err := svc.validator.HandleCommit(ctx, host, account, evt, prevP)
251
+
newRootCid, err := r.Validator.HandleCommit(ctx, host, account, evt, prevP)
252
252
if err != nil {
253
-
svc.inductionTraceLog.Error("commit bad", "seq", evt.Seq, "pseq", dbPrevSeqStr, "pdsHost", host.Host, "repo", evt.Repo, "prev", evtPrevDataStr, "dbprev", dbPrevRootStr, "err", err)
254
-
svc.log.Warn("failed handling event", "err", err, "pdsHost", host.Host, "seq", evt.Seq, "repo", account.Did, "commit", evt.Commit.String())
253
+
// XXX: induction trace log
254
+
r.Logger.Error("commit bad", "seq", evt.Seq, "pseq", dbPrevSeqStr, "pdsHost", host.Host, "repo", evt.Repo, "prev", evtPrevDataStr, "dbprev", dbPrevRootStr, "err", err)
255
+
r.Logger.Warn("failed handling event", "err", err, "pdsHost", host.Host, "seq", evt.Seq, "repo", account.Did, "commit", evt.Commit.String())
255
256
repoCommitsResultCounter.WithLabelValues(host.Host, "err").Inc()
256
257
return fmt.Errorf("handle user event failed: %w", err)
257
258
} else {
258
259
// store now verified new repo state
259
-
err = svc.upsertPrevState(account.ID, newRootCid, evt.Rev, evt.Seq)
260
+
err = r.upsertPrevState(account.ID, newRootCid, evt.Rev, evt.Seq)
260
261
if err != nil {
261
262
return fmt.Errorf("failed to set previous root uid=%d: %w", account.ID, err)
262
263
}
···
266
267
267
268
// Broadcast the identity event to all consumers
268
269
commitCopy := *evt
269
-
err = svc.events.AddEvent(ctx, &stream.XRPCStreamEvent{
270
+
err = r.Events.AddEvent(ctx, &stream.XRPCStreamEvent{
270
271
RepoCommit: &commitCopy,
271
272
PrivUid: account.GetUid(),
272
273
})
273
274
if err != nil {
274
-
svc.log.Error("failed to broadcast commit event", "error", err, "did", evt.Repo)
275
+
r.Logger.Error("failed to broadcast commit event", "error", err, "did", evt.Repo)
275
276
return fmt.Errorf("failed to broadcast commit event: %w", err)
276
277
}
277
278
···
279
280
}
280
281
281
282
// handleSync processes #sync messages
282
-
func (svc *Service) handleSync(ctx context.Context, host *slurper.PDS, evt *comatproto.SyncSubscribeRepos_Sync) error {
283
-
account, err := svc.lookupUserByDid(ctx, evt.Did)
283
+
func (r *Relay) handleSync(ctx context.Context, host *slurper.PDS, evt *comatproto.SyncSubscribeRepos_Sync) error {
284
+
account, err := r.LookupUserByDid(ctx, evt.Did)
284
285
if err != nil {
285
286
if !errors.Is(err, gorm.ErrRecordNotFound) {
286
287
repoCommitsResultCounter.WithLabelValues(host.Host, "nou").Inc()
287
288
return fmt.Errorf("looking up event user: %w", err)
288
289
}
289
290
290
-
account, err = svc.newUser(ctx, host, evt.Did)
291
+
account, err = r.newUser(ctx, host, evt.Did)
291
292
}
292
293
if err != nil {
293
294
return fmt.Errorf("could not get user for did %#v: %w", evt.Did, err)
294
295
}
295
296
296
-
newRootCid, err := svc.validator.HandleSync(ctx, host, evt)
297
+
newRootCid, err := r.Validator.HandleSync(ctx, host, evt)
297
298
if err != nil {
298
299
return err
299
300
}
300
-
err = svc.upsertPrevState(account.ID, newRootCid, evt.Rev, evt.Seq)
301
+
err = r.upsertPrevState(account.ID, newRootCid, evt.Rev, evt.Seq)
301
302
if err != nil {
302
303
return fmt.Errorf("could not sync set previous state uid=%d: %w", account.ID, err)
303
304
}
304
305
305
306
// Broadcast the sync event to all consumers
306
307
evtCopy := *evt
307
-
err = svc.events.AddEvent(ctx, &stream.XRPCStreamEvent{
308
+
err = r.Events.AddEvent(ctx, &stream.XRPCStreamEvent{
308
309
RepoSync: &evtCopy,
309
310
})
310
311
if err != nil {
311
-
svc.log.Error("failed to broadcast sync event", "error", err, "did", evt.Did)
312
+
r.Logger.Error("failed to broadcast sync event", "error", err, "did", evt.Did)
312
313
return fmt.Errorf("failed to broadcast sync event: %w", err)
313
314
}
314
315
315
316
return nil
316
317
}
317
318
318
-
func (svc *Service) upsertPrevState(accountID models.Uid, newRootCid *cid.Cid, rev string, seq int64) error {
319
+
func (r *Relay) upsertPrevState(accountID models.Uid, newRootCid *cid.Cid, rev string, seq int64) error {
319
320
cidBytes := newRootCid.Bytes()
320
-
return svc.db.Exec(
321
+
return r.db.Exec(
321
322
"INSERT INTO account_previous_states (uid, cid, rev, seq) VALUES (?, ?, ?, ?) ON CONFLICT (uid) DO UPDATE SET cid = EXCLUDED.cid, rev = EXCLUDED.rev, seq = EXCLUDED.seq",
322
323
accountID, cidBytes, rev, seq,
323
324
).Error
324
325
}
325
326
326
-
func (svc *Service) purgeDidCache(ctx context.Context, did string) {
327
+
func (r *Relay) purgeDidCache(ctx context.Context, did string) {
327
328
ati, err := syntax.ParseAtIdentifier(did)
328
329
if err != nil {
329
330
return
330
331
}
331
-
_ = svc.dir.Purge(ctx, *ati)
332
+
_ = r.dir.Purge(ctx, *ati)
332
333
}
+7
-8
cmd/relayered/handlers.go
+7
-8
cmd/relayered/handlers.go
···
11
11
"strings"
12
12
13
13
comatproto "github.com/bluesky-social/indigo/api/atproto"
14
-
"github.com/bluesky-social/indigo/cmd/relayered/slurper"
14
+
"github.com/bluesky-social/indigo/cmd/relayered/relay/slurper"
15
+
"github.com/bluesky-social/indigo/cmd/relayered/relay"
15
16
"github.com/bluesky-social/indigo/xrpc"
16
17
17
18
"github.com/labstack/echo/v4"
···
55
56
56
57
host = u.Host // potentially hostname:port
57
58
58
-
banned, err := s.domainIsBanned(ctx, host)
59
+
banned, err := s.relay.DomainIsBanned(ctx, host)
59
60
if banned {
60
61
return echo.NewHTTPError(http.StatusUnauthorized, "domain is banned")
61
62
}
···
102
103
}
103
104
}
104
105
105
-
return s.slurper.SubscribeToPds(ctx, host, true, false, nil)
106
+
return s.relay.Slurper.SubscribeToPds(ctx, host, true, false, nil)
106
107
}
107
108
108
109
func (s *Service) handleComAtprotoSyncListRepos(ctx context.Context, cursor int64, limit int) (*comatproto.SyncListRepos_Output, error) {
···
131
132
for i := range accounts {
132
133
user := accounts[i]
133
134
134
-
root, err := s.GetRepoRoot(ctx, user.ID)
135
+
root, err := s.relay.GetRepoRoot(ctx, user.ID)
135
136
if err != nil {
136
137
s.log.Error("failed to get repo root", "err", err, "did", user.Did)
137
138
return nil, echo.NewHTTPError(http.StatusInternalServerError, fmt.Sprintf("failed to get repo root for (%s): %v", user.Did, err.Error()))
···
152
153
return resp, nil
153
154
}
154
155
155
-
var ErrUserStatusUnavailable = errors.New("user status unavailable")
156
-
157
156
func (s *Service) handleComAtprotoSyncGetLatestCommit(ctx context.Context, did string) (*comatproto.SyncGetLatestCommit_Output, error) {
158
-
u, err := s.lookupUserByDid(ctx, did)
157
+
u, err := s.relay.LookupUserByDid(ctx, did)
159
158
if err != nil {
160
159
if errors.Is(err, gorm.ErrRecordNotFound) {
161
160
return nil, echo.NewHTTPError(http.StatusNotFound, "user not found")
···
185
184
if err == nil {
186
185
// okay!
187
186
} else if errors.Is(err, gorm.ErrRecordNotFound) {
188
-
return nil, ErrUserStatusUnavailable
187
+
return nil, relay.ErrUserStatusUnavailable
189
188
} else {
190
189
s.log.Error("user db err", "err", err)
191
190
return nil, fmt.Errorf("user prev db err, %w", err)
+21
-44
cmd/relayered/handlers_admin.go
+21
-44
cmd/relayered/handlers_admin.go
···
8
8
"slices"
9
9
"strconv"
10
10
"strings"
11
-
"time"
11
+
12
+
"github.com/bluesky-social/indigo/cmd/relayered/relay/slurper"
13
+
"github.com/bluesky-social/indigo/cmd/relayered/relay"
12
14
13
-
"github.com/bluesky-social/indigo/cmd/relayered/slurper"
14
15
"github.com/labstack/echo/v4"
15
16
dto "github.com/prometheus/client_model/go"
16
17
"gorm.io/gorm"
···
25
26
}
26
27
}
27
28
28
-
return svc.slurper.SetNewSubsDisabled(!enabled)
29
+
return svc.relay.Slurper.SetNewSubsDisabled(!enabled)
29
30
}
30
31
31
32
func (svc *Service) handleAdminGetSubsEnabled(e echo.Context) error {
32
33
return e.JSON(200, map[string]bool{
33
-
"enabled": !svc.slurper.GetNewSubsDisabledState(),
34
+
"enabled": !svc.relay.Slurper.GetNewSubsDisabledState(),
34
35
})
35
36
}
36
37
37
38
func (svc *Service) handleAdminGetNewPDSPerDayRateLimit(e echo.Context) error {
38
-
limit := svc.slurper.GetNewPDSPerDayLimit()
39
+
limit := svc.relay.Slurper.GetNewPDSPerDayLimit()
39
40
return e.JSON(200, map[string]int64{
40
41
"limit": limit,
41
42
})
···
50
51
}
51
52
}
52
53
53
-
err = svc.slurper.SetNewPDSPerDayLimit(limit)
54
+
err = svc.relay.Slurper.SetNewPDSPerDayLimit(limit)
54
55
if err != nil {
55
56
return &echo.HTTPError{
56
57
Code: 500,
···
76
77
}
77
78
}
78
79
79
-
err := svc.TakeDownRepo(ctx, did)
80
+
err := svc.relay.TakeDownRepo(ctx, did)
80
81
if err != nil {
81
82
if errors.Is(err, gorm.ErrRecordNotFound) {
82
83
return &echo.HTTPError{
···
95
96
func (svc *Service) handleAdminReverseTakedown(e echo.Context) error {
96
97
did := e.QueryParam("did")
97
98
ctx := e.Request().Context()
98
-
err := svc.ReverseTakedown(ctx, did)
99
+
err := svc.relay.ReverseTakedown(ctx, did)
99
100
100
101
if err != nil {
101
102
if errors.Is(err, gorm.ErrRecordNotFound) {
···
159
160
}
160
161
161
162
func (svc *Service) handleAdminGetUpstreamConns(e echo.Context) error {
162
-
return e.JSON(200, svc.slurper.GetActiveList())
163
+
return e.JSON(200, svc.relay.Slurper.GetActiveList())
163
164
}
164
165
165
166
type rateLimit struct {
···
190
191
191
192
enrichedPDSs := make([]enrichedPDS, len(pds))
192
193
193
-
activePDSHosts := svc.slurper.GetActiveList()
194
+
activePDSHosts := svc.relay.Slurper.GetActiveList()
194
195
195
196
for i, p := range pds {
196
197
enrichedPDSs[i].PDS = p
···
202
203
}
203
204
}
204
205
var m = &dto.Metric{}
205
-
if err := eventsReceivedCounter.WithLabelValues(p.Host).Write(m); err != nil {
206
+
if err := relay.EventsReceivedCounter.WithLabelValues(p.Host).Write(m); err != nil {
206
207
enrichedPDSs[i].EventsSeenSinceStartup = 0
207
208
continue
208
209
}
···
227
228
return e.JSON(200, enrichedPDSs)
228
229
}
229
230
230
-
type consumer struct {
231
-
ID uint64 `json:"id"`
232
-
RemoteAddr string `json:"remote_addr"`
233
-
UserAgent string `json:"user_agent"`
234
-
EventsConsumed uint64 `json:"events_consumed"`
235
-
ConnectedAt time.Time `json:"connected_at"`
236
-
}
237
-
238
231
func (svc *Service) handleAdminListConsumers(e echo.Context) error {
239
-
svc.consumersLk.RLock()
240
-
defer svc.consumersLk.RUnlock()
241
232
242
-
consumers := make([]consumer, 0, len(svc.consumers))
243
-
for id, c := range svc.consumers {
244
-
var m = &dto.Metric{}
245
-
if err := c.EventsSent.Write(m); err != nil {
246
-
continue
247
-
}
248
-
consumers = append(consumers, consumer{
249
-
ID: id,
250
-
RemoteAddr: c.RemoteAddr,
251
-
UserAgent: c.UserAgent,
252
-
EventsConsumed: uint64(m.Counter.GetValue()),
253
-
ConnectedAt: c.ConnectedAt,
254
-
})
255
-
}
256
-
233
+
consumers := svc.relay.ListConsumers()
257
234
return e.JSON(200, consumers)
258
235
}
259
236
···
268
245
269
246
block := strings.ToLower(e.QueryParam("block")) == "true"
270
247
271
-
if err := svc.slurper.KillUpstreamConnection(host, block); err != nil {
248
+
if err := svc.relay.Slurper.KillUpstreamConnection(host, block); err != nil {
272
249
if errors.Is(err, slurper.ErrNoActiveConnection) {
273
250
return &echo.HTTPError{
274
251
Code: 400,
···
298
275
}
299
276
300
277
// don't care if this errors, but we should try to disconnect something we just blocked
301
-
_ = svc.slurper.KillUpstreamConnection(host, false)
278
+
_ = svc.relay.Slurper.KillUpstreamConnection(host, false)
302
279
303
280
return e.JSON(200, map[string]any{
304
281
"success": "true",
···
417
394
}
418
395
419
396
// Update the rate limit in the limiter
420
-
limits := svc.slurper.GetOrCreateLimiters(pds.ID, body.PerSecond, body.PerHour, body.PerDay)
397
+
limits := svc.relay.Slurper.GetOrCreateLimiters(pds.ID, body.PerSecond, body.PerHour, body.PerDay)
421
398
limits.PerSecond.SetLimit(body.PerSecond)
422
399
limits.PerHour.SetLimit(body.PerHour)
423
400
limits.PerDay.SetLimit(body.PerDay)
···
434
411
}
435
412
436
413
// Check if the domain is already trusted
437
-
trustedDomains := svc.slurper.GetTrustedDomains()
414
+
trustedDomains := svc.relay.Slurper.GetTrustedDomains()
438
415
if slices.Contains(trustedDomains, domain) {
439
416
return &echo.HTTPError{
440
417
Code: 400,
···
442
419
}
443
420
}
444
421
445
-
if err := svc.slurper.AddTrustedDomain(domain); err != nil {
422
+
if err := svc.relay.Slurper.AddTrustedDomain(domain); err != nil {
446
423
return err
447
424
}
448
425
···
502
479
503
480
host = u.Host // potentially hostname:port
504
481
505
-
banned, err := svc.domainIsBanned(ctx, host)
482
+
banned, err := svc.relay.DomainIsBanned(ctx, host)
506
483
if banned {
507
484
return echo.NewHTTPError(http.StatusUnauthorized, "domain is banned")
508
485
}
509
486
510
487
// Skip checking if the server is online for now
511
488
rateOverrides := body.PDSRates
512
-
rateOverrides.FromSlurper(svc.slurper)
489
+
rateOverrides.FromSlurper(svc.relay.Slurper)
513
490
514
-
return svc.slurper.SubscribeToPds(ctx, host, true, true, &rateOverrides) // Override Trusted Domain Check
491
+
return svc.relay.Slurper.SubscribeToPds(ctx, host, true, true, &rateOverrides) // Override Trusted Domain Check
515
492
}
+17
-32
cmd/relayered/main.go
+17
-32
cmd/relayered/main.go
···
4
4
"crypto/rand"
5
5
"encoding/base64"
6
6
"fmt"
7
-
"io"
8
7
"log/slog"
9
8
"net/url"
10
9
"os"
···
19
18
_ "net/http/pprof"
20
19
21
20
"github.com/bluesky-social/indigo/atproto/identity"
22
-
"github.com/bluesky-social/indigo/cmd/relayered/slurper"
21
+
"github.com/bluesky-social/indigo/cmd/relayered/relay"
22
+
"github.com/bluesky-social/indigo/cmd/relayered/relay/slurper"
23
23
"github.com/bluesky-social/indigo/cmd/relayered/stream/eventmgr"
24
24
"github.com/bluesky-social/indigo/cmd/relayered/stream/persist"
25
25
"github.com/bluesky-social/indigo/cmd/relayered/stream/persist/diskpersist"
···
141
141
Usage: "forward POST requestCrawl to this url, should be machine root url and not xrpc/requestCrawl, comma separated list",
142
142
EnvVars: []string{"RELAY_NEXT_CRAWLER"},
143
143
},
144
-
&cli.StringFlag{
145
-
Name: "trace-induction",
146
-
Usage: "file path to log debug trace stuff about induction firehose",
147
-
EnvVars: []string{"RELAY_TRACE_INDUCTION"},
148
-
},
149
144
&cli.BoolFlag{
150
145
Name: "time-seq",
151
146
EnvVars: []string{"RELAY_TIME_SEQUENCE"},
···
166
161
logger, logWriter, err := cliutil.SetupSlog(cliutil.LogOptions{})
167
162
if err != nil {
168
163
return err
169
-
}
170
-
171
-
var inductionTraceLog *slog.Logger
172
-
173
-
if cctx.IsSet("trace-induction") {
174
-
traceFname := cctx.String("trace-induction")
175
-
traceFout, err := os.OpenFile(traceFname, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
176
-
if err != nil {
177
-
return fmt.Errorf("%s: could not open trace file: %w", traceFname, err)
178
-
}
179
-
defer traceFout.Close()
180
-
if traceFname != "" {
181
-
inductionTraceLog = slog.New(slog.NewJSONHandler(traceFout, &slog.HandlerOptions{Level: slog.LevelDebug}))
182
-
}
183
-
} else {
184
-
inductionTraceLog = slog.New(slog.NewTextHandler(io.Discard, &slog.HandlerOptions{Level: slog.Level(999)}))
185
164
}
186
165
187
166
// start observability/tracing (OTEL and jaeger)
···
210
189
cacheDir := identity.NewCacheDirectory(&baseDir, cctx.Int("did-cache-size"), time.Hour*24, time.Minute*2, time.Minute*5)
211
190
212
191
// TODO: rename repoman
213
-
repoman := slurper.NewValidator(&cacheDir, inductionTraceLog)
192
+
repoman := slurper.NewValidator(&cacheDir)
214
193
215
194
var persister persist.EventPersistence
216
195
···
241
220
242
221
logger.Info("constructing relay service")
243
222
svcConfig := DefaultServiceConfig()
244
-
svcConfig.SSL = !cctx.Bool("crawl-insecure-ws")
245
-
svcConfig.ConcurrencyPerPDS = cctx.Int64("concurrency-per-pds")
246
-
svcConfig.MaxQueuePerPDS = cctx.Int64("max-queue-per-pds")
247
-
svcConfig.DefaultRepoLimit = cctx.Int64("default-repo-limit")
248
-
svcConfig.ApplyPDSClientSettings = makePdsClientSetup(ratelimitBypass)
249
-
svcConfig.InductionTraceLog = inductionTraceLog
223
+
relayConfig := relay.DefaultRelayConfig()
224
+
relayConfig.SSL = !cctx.Bool("crawl-insecure-ws")
225
+
relayConfig.ConcurrencyPerPDS = cctx.Int64("concurrency-per-pds")
226
+
relayConfig.MaxQueuePerPDS = cctx.Int64("max-queue-per-pds")
227
+
relayConfig.DefaultRepoLimit = cctx.Int64("default-repo-limit")
228
+
relayConfig.ApplyPDSClientSettings = makePdsClientSetup(ratelimitBypass)
250
229
nextCrawlers := cctx.StringSlice("next-crawler")
251
230
if len(nextCrawlers) != 0 {
252
231
nextCrawlerUrls := make([]*url.URL, len(nextCrawlers))
···
268
247
svcConfig.AdminToken = base64.URLEncoding.EncodeToString(rblob[:])
269
248
logger.Info("generated random admin key", "header", "Authorization: Bearer "+svcConfig.AdminToken)
270
249
}
271
-
svc, err := NewService(db, repoman, evtman, &cacheDir, svcConfig)
250
+
251
+
r, err := relay.NewRelay(db, repoman, evtman, &cacheDir, relayConfig)
272
252
if err != nil {
273
253
return err
274
254
}
275
-
dp.SetUidSource(svc)
255
+
256
+
svc, err := NewService(db, r, &cacheDir, svcConfig)
257
+
if err != nil {
258
+
return err
259
+
}
260
+
dp.SetUidSource(r)
276
261
277
262
// set up metrics endpoint
278
263
go func() {
+3
-2
cmd/relayered/metrics.go
cmd/relayered/relay/metrics.go
+3
-2
cmd/relayered/metrics.go
cmd/relayered/relay/metrics.go
···
1
-
package main
1
+
package relay
2
2
3
3
import (
4
4
"github.com/prometheus/client_golang/prometheus"
5
5
"github.com/prometheus/client_golang/prometheus/promauto"
6
6
)
7
7
8
-
var eventsReceivedCounter = promauto.NewCounterVec(prometheus.CounterOpts{
8
+
// TODO: expose an accessor instead of exporting
9
+
var EventsReceivedCounter = promauto.NewCounterVec(prometheus.CounterOpts{
9
10
Name: "events_received_counter",
10
11
Help: "The total number of events received",
11
12
}, []string{"pds"})
+130
cmd/relayered/relay/relay.go
+130
cmd/relayered/relay/relay.go
···
1
+
package relay
2
+
3
+
import (
4
+
"log/slog"
5
+
"sync"
6
+
"time"
7
+
8
+
"github.com/bluesky-social/indigo/atproto/identity"
9
+
"github.com/bluesky-social/indigo/cmd/relayered/relay/slurper"
10
+
"github.com/bluesky-social/indigo/cmd/relayered/stream/eventmgr"
11
+
"github.com/bluesky-social/indigo/xrpc"
12
+
13
+
lru "github.com/hashicorp/golang-lru/v2"
14
+
promclient "github.com/prometheus/client_golang/prometheus"
15
+
"go.opentelemetry.io/otel"
16
+
"gorm.io/gorm"
17
+
)
18
+
19
+
var tracer = otel.Tracer("relay")
20
+
21
+
type Relay struct {
22
+
db *gorm.DB
23
+
dir identity.Directory
24
+
25
+
Slurper *slurper.Slurper
26
+
Events *eventmgr.EventManager
27
+
Validator *slurper.Validator
28
+
29
+
Config RelayConfig
30
+
31
+
32
+
// extUserLk serializes a section of syncPDSAccount()
33
+
// TODO: at some point we will want to lock specific DIDs, this lock as is
34
+
// is overly broad, but i dont expect it to be a bottleneck for now
35
+
extUserLk sync.Mutex
36
+
37
+
// Management of Socket Consumers
38
+
consumersLk sync.RWMutex
39
+
nextConsumerID uint64
40
+
consumers map[uint64]*SocketConsumer
41
+
42
+
// Account cache
43
+
userCache *lru.Cache[string, *slurper.Account]
44
+
Logger *slog.Logger
45
+
}
46
+
47
+
type RelayConfig struct {
48
+
SSL bool
49
+
DefaultRepoLimit int64
50
+
ConcurrencyPerPDS int64
51
+
MaxQueuePerPDS int64
52
+
ApplyPDSClientSettings func(c *xrpc.Client)
53
+
}
54
+
55
+
func DefaultRelayConfig() *RelayConfig {
56
+
return &RelayConfig{
57
+
SSL: true,
58
+
DefaultRepoLimit: 100,
59
+
ConcurrencyPerPDS: 100,
60
+
MaxQueuePerPDS: 1_000,
61
+
}
62
+
}
63
+
64
+
func NewRelay(db *gorm.DB, validator *slurper.Validator, evtman *eventmgr.EventManager, dir identity.Directory, config *RelayConfig) (*Relay, error) {
65
+
66
+
if config == nil {
67
+
config = DefaultRelayConfig()
68
+
}
69
+
70
+
uc, _ := lru.New[string, *slurper.Account](1_000_000)
71
+
72
+
r := &Relay{
73
+
db: db,
74
+
Events: evtman,
75
+
Validator: validator,
76
+
dir: dir,
77
+
Config: *config,
78
+
79
+
consumersLk: sync.RWMutex{},
80
+
consumers: make(map[uint64]*SocketConsumer),
81
+
82
+
userCache: uc,
83
+
84
+
Logger: slog.Default().With("system", "relay"),
85
+
}
86
+
87
+
slOpts := slurper.DefaultSlurperOptions()
88
+
slOpts.SSL = config.SSL
89
+
slOpts.DefaultRepoLimit = config.DefaultRepoLimit
90
+
slOpts.ConcurrencyPerPDS = config.ConcurrencyPerPDS
91
+
slOpts.MaxQueuePerPDS = config.MaxQueuePerPDS
92
+
slOpts.Logger = r.Logger
93
+
s, err := slurper.NewSlurper(db, r.handleFedEvent, slOpts)
94
+
if err != nil {
95
+
return nil, err
96
+
}
97
+
r.Slurper = s
98
+
99
+
if err := r.Slurper.RestartAll(); err != nil {
100
+
return nil, err
101
+
}
102
+
103
+
if err := r.MigrateDatabase(); err != nil {
104
+
return nil, err
105
+
}
106
+
return r, nil
107
+
}
108
+
109
+
func (r *Relay) MigrateDatabase() error {
110
+
if err := r.db.AutoMigrate(slurper.DomainBan{}); err != nil {
111
+
return err
112
+
}
113
+
if err := r.db.AutoMigrate(slurper.PDS{}); err != nil {
114
+
return err
115
+
}
116
+
if err := r.db.AutoMigrate(slurper.Account{}); err != nil {
117
+
return err
118
+
}
119
+
if err := r.db.AutoMigrate(slurper.AccountPreviousState{}); err != nil {
120
+
return err
121
+
}
122
+
return nil
123
+
}
124
+
125
+
type SocketConsumer struct {
126
+
UserAgent string
127
+
RemoteAddr string
128
+
ConnectedAt time.Time
129
+
EventsSent promclient.Counter
130
+
}
+11
-97
cmd/relayered/service.go
+11
-97
cmd/relayered/service.go
···
2
2
3
3
import (
4
4
"context"
5
-
"errors"
6
5
"io"
7
6
"log/slog"
8
7
"net"
9
8
"net/http"
10
9
"net/url"
11
10
"strings"
12
-
"sync"
13
11
"time"
14
12
15
13
"github.com/bluesky-social/indigo/atproto/identity"
16
-
"github.com/bluesky-social/indigo/cmd/relayered/slurper"
17
-
"github.com/bluesky-social/indigo/cmd/relayered/stream/eventmgr"
18
-
"github.com/bluesky-social/indigo/xrpc"
14
+
"github.com/bluesky-social/indigo/cmd/relayered/relay"
19
15
20
-
lru "github.com/hashicorp/golang-lru/v2"
21
16
"github.com/labstack/echo/v4"
22
17
"github.com/labstack/echo/v4/middleware"
23
-
promclient "github.com/prometheus/client_golang/prometheus"
24
18
"github.com/prometheus/client_golang/prometheus/promhttp"
25
-
"go.opentelemetry.io/otel"
26
19
"gorm.io/gorm"
27
20
)
28
-
29
-
var tracer = otel.Tracer("relay")
30
21
31
22
// serverListenerBootTimeout is how long to wait for the requested server socket
32
23
// to become available for use. This is an arbitrary timeout that should be safe
···
36
27
const serverListenerBootTimeout = 5 * time.Second
37
28
38
29
type Service struct {
39
-
db *gorm.DB
40
-
slurper *slurper.Slurper
41
-
events *eventmgr.EventManager
30
+
db *gorm.DB // XXX
31
+
relay *relay.Relay
42
32
dir identity.Directory
43
33
44
34
// TODO: work on doing away with this flag in favor of more pluggable
45
35
// pieces that abstract the need for explicit ssl checks
46
36
ssl bool
47
37
48
-
// extUserLk serializes a section of syncPDSAccount()
49
-
// TODO: at some point we will want to lock specific DIDs, this lock as is
50
-
// is overly broad, but i dont expect it to be a bottleneck for now
51
-
extUserLk sync.Mutex
52
-
53
-
validator *slurper.Validator
54
-
55
-
// Management of Socket Consumers
56
-
consumersLk sync.RWMutex
57
-
nextConsumerID uint64
58
-
consumers map[uint64]*SocketConsumer
59
-
60
-
// Account cache
61
-
userCache *lru.Cache[string, *slurper.Account]
62
38
63
39
// nextCrawlers gets forwarded POST /xrpc/com.atproto.sync.requestCrawl
64
40
nextCrawlers []*url.URL
65
41
httpClient http.Client
66
42
67
43
log *slog.Logger
68
-
inductionTraceLog *slog.Logger
69
44
70
45
config ServiceConfig
71
46
}
72
47
73
-
type SocketConsumer struct {
74
-
UserAgent string
75
-
RemoteAddr string
76
-
ConnectedAt time.Time
77
-
EventsSent promclient.Counter
78
-
}
79
-
80
48
type ServiceConfig struct {
81
-
SSL bool
82
-
DefaultRepoLimit int64
83
-
ConcurrencyPerPDS int64
84
-
MaxQueuePerPDS int64
85
-
86
49
// NextCrawlers gets forwarded POST /xrpc/com.atproto.sync.requestCrawl
87
50
NextCrawlers []*url.URL
88
51
89
-
ApplyPDSClientSettings func(c *xrpc.Client)
90
-
InductionTraceLog *slog.Logger
91
-
92
52
// AdminToken checked against "Authorization: Bearer {}" header
93
53
AdminToken string
94
54
}
95
55
96
56
func DefaultServiceConfig() *ServiceConfig {
97
-
return &ServiceConfig{
98
-
SSL: true,
99
-
DefaultRepoLimit: 100,
100
-
ConcurrencyPerPDS: 100,
101
-
MaxQueuePerPDS: 1_000,
102
-
}
57
+
return &ServiceConfig{}
103
58
}
104
59
105
-
func NewService(db *gorm.DB, validator *slurper.Validator, evtman *eventmgr.EventManager, dir identity.Directory, config *ServiceConfig) (*Service, error) {
60
+
func NewService(db *gorm.DB, r *relay.Relay, dir identity.Directory, config *ServiceConfig) (*Service, error) {
106
61
107
62
if config == nil {
108
63
config = DefaultServiceConfig()
109
64
}
110
-
if err := db.AutoMigrate(slurper.DomainBan{}); err != nil {
111
-
panic(err)
112
-
}
113
-
if err := db.AutoMigrate(slurper.PDS{}); err != nil {
114
-
panic(err)
115
-
}
116
-
if err := db.AutoMigrate(slurper.Account{}); err != nil {
117
-
panic(err)
118
-
}
119
-
if err := db.AutoMigrate(slurper.AccountPreviousState{}); err != nil {
120
-
panic(err)
121
-
}
122
-
123
-
uc, _ := lru.New[string, *slurper.Account](1_000_000)
124
65
125
66
svc := &Service{
126
67
db: db,
127
-
128
-
validator: validator,
129
-
events: evtman,
68
+
relay: r,
130
69
dir: dir,
131
-
ssl: config.SSL,
132
-
133
-
consumersLk: sync.RWMutex{},
134
-
consumers: make(map[uint64]*SocketConsumer),
135
-
136
-
userCache: uc,
70
+
ssl: r.Config.SSL,
137
71
138
72
log: slog.Default().With("system", "relay"),
139
73
140
74
config: *config,
141
-
142
-
inductionTraceLog: config.InductionTraceLog,
143
-
}
144
-
145
-
slOpts := slurper.DefaultSlurperOptions()
146
-
slOpts.SSL = config.SSL
147
-
slOpts.DefaultRepoLimit = config.DefaultRepoLimit
148
-
slOpts.ConcurrencyPerPDS = config.ConcurrencyPerPDS
149
-
slOpts.MaxQueuePerPDS = config.MaxQueuePerPDS
150
-
slOpts.Logger = svc.log
151
-
s, err := slurper.NewSlurper(db, svc.handleFedEvent, slOpts)
152
-
if err != nil {
153
-
return nil, err
154
-
}
155
-
156
-
svc.slurper = s
157
-
158
-
if err := svc.slurper.RestartAll(); err != nil {
159
-
return nil, err
160
75
}
161
76
162
77
svc.nextCrawlers = config.NextCrawlers
···
239
154
240
155
// TODO: this API is temporary until we formalize what we want here
241
156
242
-
e.GET("/xrpc/com.atproto.sync.subscribeRepos", svc.EventsHandler)
157
+
e.GET("/xrpc/com.atproto.sync.subscribeRepos", svc.relay.EventsHandler)
158
+
243
159
e.POST("/xrpc/com.atproto.sync.requestCrawl", svc.HandleComAtprotoSyncRequestCrawl)
244
160
e.GET("/xrpc/com.atproto.sync.listRepos", svc.HandleComAtprotoSyncListRepos)
245
161
e.GET("/xrpc/com.atproto.sync.getRepo", svc.HandleComAtprotoSyncGetRepo) // just returns 3xx redirect to source PDS
···
288
204
}
289
205
290
206
func (svc *Service) Shutdown() []error {
291
-
errs := svc.slurper.Shutdown()
207
+
errs := svc.relay.Slurper.Shutdown()
292
208
293
-
if err := svc.events.Shutdown(context.TODO()); err != nil {
209
+
if err := svc.relay.Events.Shutdown(context.TODO()); err != nil {
294
210
errs = append(errs, err)
295
211
}
296
212
···
347
263
return next(e)
348
264
}
349
265
}
350
-
351
-
var ErrNotFound = errors.New("not found")
cmd/relayered/slurper/account.go
cmd/relayered/relay/slurper/account.go
cmd/relayered/slurper/account.go
cmd/relayered/relay/slurper/account.go
cmd/relayered/slurper/metrics.go
cmd/relayered/relay/slurper/metrics.go
cmd/relayered/slurper/metrics.go
cmd/relayered/relay/slurper/metrics.go
cmd/relayered/slurper/models.go
cmd/relayered/relay/slurper/models.go
cmd/relayered/slurper/models.go
cmd/relayered/relay/slurper/models.go
cmd/relayered/slurper/models_dbcid.go
cmd/relayered/relay/slurper/models_dbcid.go
cmd/relayered/slurper/models_dbcid.go
cmd/relayered/relay/slurper/models_dbcid.go
cmd/relayered/slurper/rate_limits.go
cmd/relayered/relay/slurper/rate_limits.go
cmd/relayered/slurper/rate_limits.go
cmd/relayered/relay/slurper/rate_limits.go
cmd/relayered/slurper/slurper.go
cmd/relayered/relay/slurper/slurper.go
cmd/relayered/slurper/slurper.go
cmd/relayered/relay/slurper/slurper.go
+11
-8
cmd/relayered/slurper/validator.go
cmd/relayered/relay/slurper/validator.go
+11
-8
cmd/relayered/slurper/validator.go
cmd/relayered/relay/slurper/validator.go
···
20
20
21
21
const defaultMaxRevFuture = time.Hour
22
22
23
-
func NewValidator(directory identity.Directory, inductionTraceLog *slog.Logger) *Validator {
23
+
func NewValidator(directory identity.Directory) *Validator {
24
24
maxRevFuture := defaultMaxRevFuture // TODO: configurable
25
25
ErrRevTooFarFuture := fmt.Errorf("new rev is > %s in the future", maxRevFuture)
26
26
27
27
return &Validator{
28
28
userLocks: make(map[models.Uid]*userLock),
29
29
log: slog.Default().With("system", "validator"),
30
-
inductionTraceLog: inductionTraceLog,
31
30
directory: directory,
32
31
33
32
maxRevFuture: maxRevFuture,
···
42
41
userLocks map[models.Uid]*userLock
43
42
44
43
log *slog.Logger
45
-
inductionTraceLog *slog.Logger
46
44
47
45
directory identity.Directory
48
46
···
164
162
if msg.TooBig {
165
163
//logger.Warn("event with tooBig flag set")
166
164
commitVerifyWarnings.WithLabelValues(hostname, "big").Inc()
167
-
val.inductionTraceLog.Warn("commit tooBig", "seq", msg.Seq, "pdsHost", host.Host, "repo", msg.Repo)
165
+
// XXX: induction trace log
166
+
val.log.Warn("commit tooBig", "seq", msg.Seq, "pdsHost", host.Host, "repo", msg.Repo)
168
167
hasWarning = true
169
168
}
170
169
if msg.Rebase {
171
170
//logger.Warn("event with rebase flag set")
172
171
commitVerifyWarnings.WithLabelValues(hostname, "reb").Inc()
173
-
val.inductionTraceLog.Warn("commit rebase", "seq", msg.Seq, "pdsHost", host.Host, "repo", msg.Repo)
172
+
// XXX: induction trace log
173
+
val.log.Warn("commit rebase", "seq", msg.Seq, "pdsHost", host.Host, "repo", msg.Repo)
174
174
hasWarning = true
175
175
}
176
176
···
227
227
case "delete":
228
228
if o.Prev == nil {
229
229
logger.Debug("can't invert legacy op", "action", o.Action)
230
-
val.inductionTraceLog.Warn("commit delete op", "seq", msg.Seq, "pdsHost", host.Host, "repo", msg.Repo)
230
+
// XXX: induction trace log
231
+
val.log.Warn("commit delete op", "seq", msg.Seq, "pdsHost", host.Host, "repo", msg.Repo)
231
232
commitVerifyOkish.WithLabelValues(hostname, "del").Inc()
232
233
return repoFragment, nil
233
234
}
234
235
case "update":
235
236
if o.Prev == nil {
236
237
logger.Debug("can't invert legacy op", "action", o.Action)
237
-
val.inductionTraceLog.Warn("commit update op", "seq", msg.Seq, "pdsHost", host.Host, "repo", msg.Repo)
238
+
// XXX: induction trace log
239
+
val.log.Warn("commit update op", "seq", msg.Seq, "pdsHost", host.Host, "repo", msg.Repo)
238
240
commitVerifyOkish.WithLabelValues(hostname, "up").Inc()
239
241
return repoFragment, nil
240
242
}
···
246
248
if prevRoot != nil {
247
249
if *c != prevRoot.GetCid() {
248
250
commitVerifyWarnings.WithLabelValues(hostname, "pr").Inc()
249
-
val.inductionTraceLog.Warn("commit prevData mismatch", "seq", msg.Seq, "pdsHost", host.Host, "repo", msg.Repo)
251
+
// XXX: induction trace log
252
+
val.log.Warn("commit prevData mismatch", "seq", msg.Seq, "pdsHost", host.Host, "repo", msg.Repo)
250
253
hasWarning = true
251
254
}
252
255
} else {
+50
-21
cmd/relayered/subscribe_repos.go
cmd/relayered/relay/subscribe_repos.go
+50
-21
cmd/relayered/subscribe_repos.go
cmd/relayered/relay/subscribe_repos.go
···
1
-
package main
1
+
package relay
2
2
3
3
import (
4
4
"context"
···
15
15
dto "github.com/prometheus/client_model/go"
16
16
)
17
17
18
-
func (svc *Service) registerConsumer(c *SocketConsumer) uint64 {
19
-
svc.consumersLk.Lock()
20
-
defer svc.consumersLk.Unlock()
18
+
func (r *Relay) registerConsumer(c *SocketConsumer) uint64 {
19
+
r.consumersLk.Lock()
20
+
defer r.consumersLk.Unlock()
21
21
22
-
id := svc.nextConsumerID
23
-
svc.nextConsumerID++
22
+
id := r.nextConsumerID
23
+
r.nextConsumerID++
24
24
25
-
svc.consumers[id] = c
25
+
r.consumers[id] = c
26
26
27
27
return id
28
28
}
29
29
30
-
func (svc *Service) cleanupConsumer(id uint64) {
31
-
svc.consumersLk.Lock()
32
-
defer svc.consumersLk.Unlock()
30
+
func (r *Relay) cleanupConsumer(id uint64) {
31
+
r.consumersLk.Lock()
32
+
defer r.consumersLk.Unlock()
33
33
34
-
c := svc.consumers[id]
34
+
c := r.consumers[id]
35
35
36
36
var m = &dto.Metric{}
37
37
if err := c.EventsSent.Write(m); err != nil {
38
-
svc.log.Error("failed to get sent counter", "err", err)
38
+
r.Logger.Error("failed to get sent counter", "err", err)
39
39
}
40
40
41
-
svc.log.Info("consumer disconnected",
41
+
r.Logger.Info("consumer disconnected",
42
42
"consumer_id", id,
43
43
"remote_addr", c.RemoteAddr,
44
44
"user_agent", c.UserAgent,
45
45
"events_sent", m.Counter.GetValue())
46
46
47
-
delete(svc.consumers, id)
47
+
delete(r.consumers, id)
48
48
}
49
49
50
50
// GET+websocket /xrpc/com.atproto.sync.subscribeRepos
51
-
func (svc *Service) EventsHandler(c echo.Context) error {
51
+
func (r *Relay) EventsHandler(c echo.Context) error {
52
52
var since *int64
53
53
if sinceVal := c.QueryParam("cursor"); sinceVal != "" {
54
54
sval, err := strconv.ParseInt(sinceVal, 10, 64)
···
90
90
}
91
91
92
92
if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(5*time.Second)); err != nil {
93
-
svc.log.Warn("failed to ping client", "err", err)
93
+
r.Logger.Warn("failed to ping client", "err", err)
94
94
cancel()
95
95
return
96
96
}
···
115
115
for {
116
116
_, _, err := conn.ReadMessage()
117
117
if err != nil {
118
-
svc.log.Warn("failed to read message from client", "err", err)
118
+
r.Logger.Warn("failed to read message from client", "err", err)
119
119
cancel()
120
120
return
121
121
}
···
124
124
125
125
ident := c.RealIP() + "-" + c.Request().UserAgent()
126
126
127
-
evts, cleanup, err := svc.events.Subscribe(ctx, ident, func(evt *stream.XRPCStreamEvent) bool { return true }, since)
127
+
evts, cleanup, err := r.Events.Subscribe(ctx, ident, func(evt *stream.XRPCStreamEvent) bool { return true }, since)
128
128
if err != nil {
129
129
return err
130
130
}
···
139
139
sentCounter := eventsSentCounter.WithLabelValues(consumer.RemoteAddr, consumer.UserAgent)
140
140
consumer.EventsSent = sentCounter
141
141
142
-
consumerID := svc.registerConsumer(&consumer)
143
-
defer svc.cleanupConsumer(consumerID)
142
+
consumerID := r.registerConsumer(&consumer)
143
+
defer r.cleanupConsumer(consumerID)
144
144
145
-
logger := svc.log.With(
145
+
logger := r.Logger.With(
146
146
"consumer_id", consumerID,
147
147
"remote_addr", consumer.RemoteAddr,
148
148
"user_agent", consumer.UserAgent,
···
187
187
}
188
188
}
189
189
}
190
+
191
+
type ConsumerInfo struct {
192
+
ID uint64 `json:"id"`
193
+
RemoteAddr string `json:"remote_addr"`
194
+
UserAgent string `json:"user_agent"`
195
+
EventsConsumed uint64 `json:"events_consumed"`
196
+
ConnectedAt time.Time `json:"connected_at"`
197
+
}
198
+
199
+
func (r *Relay) ListConsumers() []ConsumerInfo {
200
+
r.consumersLk.RLock()
201
+
defer r.consumersLk.RUnlock()
202
+
203
+
info := make([]ConsumerInfo, 0, len(r.consumers))
204
+
for id, c := range r.consumers {
205
+
var m = &dto.Metric{}
206
+
if err := c.EventsSent.Write(m); err != nil {
207
+
continue
208
+
}
209
+
info = append(info, ConsumerInfo{
210
+
ID: id,
211
+
RemoteAddr: c.RemoteAddr,
212
+
UserAgent: c.UserAgent,
213
+
EventsConsumed: uint64(m.Counter.GetValue()),
214
+
ConnectedAt: c.ConnectedAt,
215
+
})
216
+
}
217
+
return info
218
+
}