this repo has no description
relay: resolve a batch of TODOs (#1042)
Batch of small fixes resolving "TODOs" in the code. Some are trivial,
some impact behaviors. Tried to describe with commit names/messages.
If any of these individual commits are concerning, I can pull them out
as separate PRs.
+42
-47
+5
-3
cmd/relay/handlers_admin.go
+5
-3
cmd/relay/handlers_admin.go
···
282
282
}
283
283
284
284
func (s *Service) handleAdminKillUpstreamConn(c echo.Context) error {
285
+
ctx := c.Request().Context()
286
+
285
287
queryHost := strings.TrimSpace(c.QueryParam("host"))
286
288
hostname, _, err := relay.ParseHostname(queryHost)
287
289
if err != nil {
···
294
296
banHost := strings.ToLower(c.QueryParam("block")) == "true"
295
297
296
298
// TODO: move this method to relay (for updating the database)
297
-
if err := s.relay.Slurper.KillUpstreamConnection(hostname, banHost); err != nil {
298
-
if errors.Is(err, relay.ErrNoActiveConnection) {
299
+
if err := s.relay.Slurper.KillUpstreamConnection(ctx, hostname, banHost); err != nil {
300
+
if errors.Is(err, relay.ErrHostInactive) {
299
301
return &echo.HTTPError{
300
302
Code: http.StatusBadRequest,
301
303
Message: "no active connection to given host",
···
333
335
}
334
336
335
337
// kill any active connection (there may not be one, so ignore error)
336
-
_ = s.relay.Slurper.KillUpstreamConnection(host.Hostname, false)
338
+
_ = s.relay.Slurper.KillUpstreamConnection(ctx, host.Hostname, false)
337
339
338
340
// TODO: forward to SiblingRelayHosts
339
341
return c.JSON(http.StatusOK, map[string]any{
+12
-1
cmd/relay/main.go
+12
-1
cmd/relay/main.go
···
108
108
EnvVars: []string{"RELAY_DEFAULT_ACCOUUNT_LIMIT", "RELAY_DEFAULT_REPO_LIMIT"},
109
109
},
110
110
&cli.IntFlag{
111
+
Name: "new-hosts-per-day-limit",
112
+
Value: 50,
113
+
Usage: "max number of new upstream hosts subscribed per day via public requestCrawl",
114
+
EnvVars: []string{"RELAY_NEW_HOSTS_PER_DAY_LIMIT"},
115
+
},
116
+
&cli.IntFlag{
111
117
Name: "ident-cache-size",
112
118
Value: 5_000_000,
113
119
Usage: "size of in-process identity cache (eg, DID docs)",
···
239
245
relayConfig.UserAgent = fmt.Sprintf("indigo-relay/%s", versioninfo.Short())
240
246
relayConfig.ConcurrencyPerHost = cctx.Int("host-concurrency")
241
247
relayConfig.DefaultRepoLimit = cctx.Int64("default-account-limit")
248
+
relayConfig.HostPerDayLimit = cctx.Int64("new-hosts-per-day-limit")
242
249
relayConfig.TrustedDomains = cctx.StringSlice("trusted-domains")
243
250
relayConfig.LenientSyncValidation = cctx.Bool("lenient-sync-validation")
244
251
···
289
296
}
290
297
}
291
298
292
-
svcErr := make(chan error, 1)
299
+
// restart any existing subscriptions as worker goroutines
300
+
if err := r.ResubscribeAllHosts(); err != nil {
301
+
return err
302
+
}
293
303
304
+
svcErr := make(chan error, 1)
294
305
go func() {
295
306
err := svc.StartAPI(cctx.String("bind"))
296
307
svcErr <- err
-1
cmd/relay/relay/account.go
-1
cmd/relay/relay/account.go
+1
-1
cmd/relay/relay/crawl.go
+1
-1
cmd/relay/relay/crawl.go
···
26
26
// check if we're over the limit for new hosts today (bypass if admin mode)
27
27
if !adminForce && !r.HostPerDayLimiter.Allow() {
28
28
// TODO: is this the correct error code?
29
-
return ErrNewSubsDisabled
29
+
return ErrNewHostsDisabled
30
30
}
31
31
32
32
accountLimit := r.Config.DefaultRepoLimit
+3
-6
cmd/relay/relay/errors.go
+3
-6
cmd/relay/relay/errors.go
···
6
6
7
7
var (
8
8
ErrHostNotFound = errors.New("unknown host or PDS")
9
+
ErrHostInactive = errors.New("no active connection to host")
10
+
ErrHostNotPDS = errors.New("server is not a PDS")
11
+
ErrNewHostsDisabled = errors.New("new host subscriptions temporarily disabled")
9
12
ErrAccountNotFound = errors.New("unknown account")
10
13
ErrAccountRepoNotFound = errors.New("repository state not available")
11
-
ErrNotPDS = errors.New("server is not a PDS")
12
-
13
-
// TODO: these might need better names
14
-
ErrTimeoutShutdown = errors.New("timed out waiting for new events")
15
-
ErrNewSubsDisabled = errors.New("new subscriptions temporarily disabled")
16
-
ErrNoActiveConnection = errors.New("no active connection to host")
17
14
)
+3
-2
cmd/relay/relay/host.go
+3
-2
cmd/relay/relay/host.go
···
55
55
56
56
func (r *Relay) ListHosts(ctx context.Context, cursor int64, limit int) ([]*models.Host, error) {
57
57
58
-
// TODO: filter based on active status?
58
+
// filters for accounts which have seen at least one event
59
+
// TODO: also filter based on status?
59
60
hosts := []*models.Host{}
60
-
if err := r.db.Model(&models.Host{}).Where("id > ?", cursor).Order("id").Limit(limit).Find(&hosts).Error; err != nil {
61
+
if err := r.db.Model(&models.Host{}).Where("id > ? AND last_seq > 0", cursor).Order("id").Limit(limit).Find(&hosts).Error; err != nil {
61
62
return nil, err
62
63
}
63
64
return hosts, nil
+2
-2
cmd/relay/relay/host_checker.go
+2
-2
cmd/relay/relay/host_checker.go
···
45
45
46
46
_, err := comatproto.ServerDescribeServer(ctx, &xrpcc)
47
47
if err != nil {
48
-
return fmt.Errorf("%w: %w", ErrNotPDS, err)
48
+
return fmt.Errorf("%w: %w", ErrHostNotPDS, err)
49
49
}
50
50
return nil
51
51
}
···
90
90
func (hc *MockHostChecker) CheckHost(ctx context.Context, host string) error {
91
91
_, ok := hc.Hosts[host]
92
92
if !ok {
93
-
return ErrNotPDS
93
+
return ErrHostNotPDS
94
94
}
95
95
return nil
96
96
}
+1
-6
cmd/relay/relay/ingest.go
+1
-6
cmd/relay/relay/ingest.go
···
81
81
}
82
82
}
83
83
84
-
if acc == nil {
85
-
// TODO: this is defensive and could be removed
86
-
return nil, nil, ErrAccountNotFound
87
-
}
88
-
89
84
// verify that the account is on the subscribed host (or update if it should be)
90
85
if err := r.EnsureAccountHost(ctx, acc, hostID, hostname); err != nil {
91
86
return nil, nil, err
···
123
118
124
119
prevRepo, err := r.GetAccountRepo(ctx, acc.UID)
125
120
if err != nil && !errors.Is(err, ErrAccountRepoNotFound) {
126
-
// TODO: should this be a hard error?
127
121
logger.Error("failed to read previous repo state", "err", err)
122
+
return err
128
123
}
129
124
130
125
// fast check for stale revision (will be re-checked in VerifyRepoCommit)
+1
-1
cmd/relay/relay/models/models.go
+1
-1
cmd/relay/relay/models/models.go
···
66
66
AccountStatusSuspended = AccountStatus("suspended")
67
67
AccountStatusTakendown = AccountStatus("takendown")
68
68
AccountStatusThrottled = AccountStatus("throttled")
69
-
AccountStatusHostThrottled = AccountStatus("host-throttled") // TODO: not yet implemented
69
+
AccountStatusHostThrottled = AccountStatus("host-throttled")
70
70
71
71
// generic "not active, but not known" status
72
72
AccountStatusInactive = AccountStatus("inactive")
-4
cmd/relay/relay/relay.go
-4
cmd/relay/relay/relay.go
+14
-20
cmd/relay/relay/slurper.go
+14
-20
cmd/relay/relay/slurper.go
···
2
2
3
3
import (
4
4
"context"
5
-
"errors"
6
5
"fmt"
7
6
"log/slog"
8
7
"math/rand"
···
19
18
"github.com/RussellLuo/slidingwindow"
20
19
"github.com/gorilla/websocket"
21
20
)
22
-
23
-
// TODO: this isn't actually getting setup or used?
24
-
var EventsTimeout = time.Minute
25
21
26
22
type ProcessMessageFunc func(ctx context.Context, evt *stream.XRPCStreamEvent, hostname string, hostID uint64) error
27
23
type PersistCursorFunc func(ctx context.Context, cursors *[]HostCursor) error
···
197
193
198
194
sub, ok := s.subs[hostname]
199
195
if !ok {
200
-
return fmt.Errorf("updating limits for %s: %w", hostname, ErrNoActiveConnection)
196
+
return fmt.Errorf("updating limits for %s: %w", hostname, ErrHostInactive)
201
197
}
202
198
203
199
sub.Limiters.PerSecond.SetLimit(newLims.PerSecond)
···
213
209
214
210
sub, ok := s.subs[hostname]
215
211
if !ok {
216
-
return nil, fmt.Errorf("reading limits for %s: %w", hostname, ErrNoActiveConnection)
212
+
return nil, fmt.Errorf("reading limits for %s: %w", hostname, ErrHostInactive)
217
213
}
218
214
219
215
slc := sub.Limiters.Counts()
···
309
305
}
310
306
311
307
u := host.SubscribeReposURL()
312
-
if !newHost {
308
+
if !newHost && cursor > 0 {
313
309
u = fmt.Sprintf("%s?cursor=%d", u, cursor)
314
310
}
315
311
hdr := make(http.Header)
···
335
331
336
332
curCursor := cursor
337
333
if err := s.handleConnection(ctx, conn, &cursor, sub); err != nil {
338
-
if errors.Is(err, ErrTimeoutShutdown) {
339
-
s.logger.Info("shutting down host subscription after timeout", "host", host.Hostname, "time", EventsTimeout.String())
340
-
return
341
-
}
342
334
s.logger.Warn("connection to failed", "host", host.Hostname, "err", err)
343
335
// TODO: measure the last N connection error times and if they're coming too fast reconnect slower or don't reconnect and wait for requestCrawl
344
336
}
···
408
400
return nil
409
401
},
410
402
Error: func(evt *stream.ErrorFrame) error {
411
-
// TODO: verbose logging
403
+
s.logger.Warn("error event from upstream", "name", evt.Error, "message", evt.Message, "host", sub.Hostname)
412
404
switch evt.Error {
413
405
case "FutureCursor":
414
406
// TODO: need test coverage for this code path (including re-connect)
415
407
// if we get a FutureCursor frame, reset our sequence number for this host
416
408
if s.Config.PersistCursorCallback != nil {
417
-
hc := []HostCursor{sub.HostCursor()}
409
+
hc := []HostCursor{HostCursor{
410
+
HostID: sub.HostID,
411
+
LastSeq: -1, // -1 will result in "current stream" on reconnect
412
+
}}
418
413
if err := s.Config.PersistCursorCallback(context.Background(), &hc); err != nil {
419
-
s.logger.Error("failed to reset cursor for host which sent FutureCursor error message", "hostname", sub.Hostname, "err", err)
414
+
s.logger.Error("failed to reset cursor for host which sent FutureCursor error message", "host", sub.Hostname, "err", err)
420
415
}
421
416
} else {
422
-
s.logger.Warn("skipping FutureCursor fix because PersistCursorCallback registered", "hostname", sub.Hostname)
417
+
s.logger.Warn("skipping FutureCursor fix because PersistCursorCallback registered", "host", sub.Hostname)
423
418
}
424
419
*lastCursor = 0
425
-
// TODO: should this really return an error?
426
-
return fmt.Errorf("got FutureCursor frame, reset cursor tracking for host")
420
+
return fmt.Errorf("got FutureCursor error")
427
421
default:
428
422
return fmt.Errorf("error frame: %s: %s", evt.Error, evt.Message)
429
423
}
···
523
517
return keys
524
518
}
525
519
526
-
func (s *Slurper) KillUpstreamConnection(hostname string, ban bool) error {
520
+
func (s *Slurper) KillUpstreamConnection(ctx context.Context, hostname string, ban bool) error {
527
521
s.subsLk.Lock()
528
522
defer s.subsLk.Unlock()
529
523
530
524
sub, ok := s.subs[hostname]
531
525
if !ok {
532
-
return fmt.Errorf("killing connection %q: %w", hostname, ErrNoActiveConnection)
526
+
return fmt.Errorf("killing connection %q: %w", hostname, ErrHostInactive)
533
527
}
534
528
sub.cancel()
535
529
// cleanup in the run thread subscribeWithRedialer() will delete(s.active, host)
536
530
537
531
if ban && s.Config.PersistHostStatusCallback != nil {
538
-
if err := s.Config.PersistHostStatusCallback(context.TODO(), sub.HostID, models.HostStatusBanned); err != nil {
532
+
if err := s.Config.PersistHostStatusCallback(ctx, sub.HostID, models.HostStatusBanned); err != nil {
539
533
return fmt.Errorf("failed to set host as banned: %w", err)
540
534
}
541
535
}