this repo has no description
quick pass over configs
+64
-58
+4
cmd/rerelay/handlers.go
+4
cmd/rerelay/handlers.go
···
18
18
func (s *Service) handleComAtprotoSyncRequestCrawl(c echo.Context, body *comatproto.SyncRequestCrawl_Input, admin bool) error {
19
19
ctx := c.Request().Context()
20
20
21
+
if s.config.DisableRequestCrawl && !admin {
22
+
return c.JSON(http.StatusForbidden, xrpc.XRPCError{ErrStr: "Forbidden", Message: "public requestCrawl not allowed on this relay"})
23
+
}
24
+
21
25
hostname, noSSL, err := relay.ParseHostname(body.Hostname)
22
26
if err != nil {
23
27
return c.JSON(http.StatusBadRequest, xrpc.XRPCError{ErrStr: "BadRequest", Message: fmt.Sprintf("hostname field empty or invalid: %s", body.Hostname)})
+38
-42
cmd/rerelay/main.go
+38
-42
cmd/rerelay/main.go
···
6
6
"fmt"
7
7
"io"
8
8
"log/slog"
9
-
"net/url"
10
9
"os"
11
10
"os/signal"
12
11
"strings"
···
43
42
Version: versioninfo.Short(),
44
43
}
45
44
app.Flags = []cli.Flag{
46
-
// XXX: actually disabled if empty?
47
45
&cli.StringFlag{
48
46
Name: "admin-password",
49
47
Usage: "secret password/token for accessing admin endpoints (random is used if not set)",
50
48
EnvVars: []string{"RELAY_ADMIN_PASSWORD", "RELAY_ADMIN_KEY"},
51
49
},
52
-
// XXX: not used?
53
50
&cli.StringFlag{
54
51
Name: "plc-host",
55
52
Usage: "method, hostname, and port of PLC registry",
56
53
Value: "https://plc.directory",
57
-
EnvVars: []string{"ATP_PLC_HOST"},
54
+
EnvVars: []string{"RELAY_PLC_HOST", "ATP_PLC_HOST"},
58
55
},
59
56
&cli.StringFlag{
60
57
Name: "log-level",
···
102
99
Name: "host-concurrency",
103
100
Usage: "number of concurrent worker routines per upstream host",
104
101
EnvVars: []string{"RELAY_HOST_CONCURRENCY", "RELAY_CONCURRENCY_PER_PDS"},
105
-
Value: 100,
102
+
Value: 40,
103
+
},
104
+
&cli.IntFlag{
105
+
Name: "host-queue-depth",
106
+
Usage: "size of queue (channel) per-host for unprocessed events",
107
+
EnvVars: []string{"RELAY_HOST_QUEUE_SIZE"},
108
+
Value: 1000,
106
109
},
107
110
&cli.IntFlag{
108
111
Name: "default-account-limit",
···
111
114
EnvVars: []string{"RELAY_DEFAULT_ACCOUUNT_LIMIT", "RELAY_DEFAULT_REPO_LIMIT"},
112
115
},
113
116
&cli.IntFlag{
114
-
Name: "did-cache-size",
117
+
Name: "ident-cache-size",
115
118
Value: 5_000_000,
116
-
Usage: "size of in-process DID (identity) cache",
117
-
EnvVars: []string{"RELAY_DID_CACHE_SIZE"},
119
+
Usage: "size of in-process identity cache (eg, DID docs)",
120
+
EnvVars: []string{"RELAY_IDENT_CACHE_SIZE", "RELAY_DID_CACHE_SIZE"},
121
+
},
122
+
&cli.BoolFlag{
123
+
Name: "disable-request-crawl",
124
+
Usage: "don't process public (un-authenticated) com.atproto.sync.requestCrawl",
125
+
EnvVars: []string{"RELAY_DISABLE_REQUEST_CRAWL"},
126
+
},
127
+
// XXX: should this be handled by rainbow instead of relays?
128
+
&cli.StringSliceFlag{
129
+
Name: "forward-crawl-requests",
130
+
Usage: "servers (eg https://example.com) to forward requestCrawl on to; multiple allowed",
131
+
EnvVars: []string{"RELAY_FORWARD_CRAWL_REQUESTS", "RELAY_NEXT_CRAWLER"},
132
+
},
133
+
&cli.StringSliceFlag{
134
+
Name: "trusted-domains",
135
+
Usage: "domain name suffixes which mark trusted hosts",
136
+
EnvVars: []string{"RELAY_TRUSTED_DOMAINS"},
118
137
},
119
138
&cli.StringFlag{
120
139
Name: "env",
···
142
161
Value: "http://localhost:4328",
143
162
EnvVars: []string{"OTEL_EXPORTER_OTLP_ENDPOINT"},
144
163
},
145
-
// XXX: refactor this flag
146
-
&cli.BoolFlag{
147
-
Name: "crawl-insecure-ws",
148
-
Usage: "when connecting to PDS instances, use ws:// instead of wss://",
149
-
},
150
-
&cli.StringSliceFlag{
151
-
Name: "forward-crawl-requests",
152
-
Usage: "comma-separated list of servers (eg https://example.com) to forward requestCrawl on to",
153
-
EnvVars: []string{"RELAY_FORWARD_CRAWL_REQUESTS", "RELAY_NEXT_CRAWLER"},
154
-
},
155
-
&cli.StringFlag{
156
-
Name: "bsky-social-rate-limit-skip",
157
-
EnvVars: []string{"BSKY_SOCIAL_RATE_LIMIT_SKIP"},
158
-
Usage: "ratelimit bypass secret token for *.bsky.social domains",
159
-
},
160
164
},
161
165
},
162
166
}
···
205
209
SkipHandleVerification: true,
206
210
SkipDNSDomainSuffixes: []string{".bsky.social"},
207
211
TryAuthoritativeDNS: true,
212
+
PLCURL: cctx.String("plc-host"),
208
213
}
209
-
dir := identity.NewCacheDirectory(&baseDir, cctx.Int("did-cache-size"), time.Hour*24, time.Minute*2, time.Minute*5)
214
+
dir := identity.NewCacheDirectory(&baseDir, cctx.Int("ident-cache-size"), time.Hour*24, time.Minute*2, time.Minute*5)
210
215
211
216
persistDir := cctx.String("persist-dir")
212
217
os.MkdirAll(persistDir, os.ModePerm)
···
218
223
return fmt.Errorf("setting up disk persister: %w", err)
219
224
}
220
225
221
-
svcConfig := DefaultServiceConfig()
222
226
relayConfig := relay.DefaultRelayConfig()
223
-
relayConfig.SSL = !cctx.Bool("crawl-insecure-ws")
224
-
relayConfig.ConcurrencyPerHost = cctx.Int64("host-concurrency")
227
+
relayConfig.ConcurrencyPerHost = cctx.Int("host-concurrency")
228
+
relayConfig.QueueDepthPerHost = cctx.Int("host-queue-depth")
225
229
relayConfig.DefaultRepoLimit = cctx.Int64("default-account-limit")
226
-
ratelimitBypass := cctx.String("bsky-social-rate-limit-skip")
227
-
// TODO: actually use ratelimitBypass for host checks?
228
-
_ = ratelimitBypass
229
-
nextCrawlers := cctx.StringSlice("forward-crawl-requests")
230
-
if len(nextCrawlers) > 0 {
231
-
nextCrawlerUrls := make([]*url.URL, len(nextCrawlers))
232
-
for i, tu := range nextCrawlers {
233
-
var err error
234
-
nextCrawlerUrls[i], err = url.Parse(tu)
235
-
if err != nil {
236
-
return fmt.Errorf("invalid crawl request forwarding URL: %w", err)
237
-
}
238
-
}
239
-
svcConfig.NextCrawlers = nextCrawlerUrls
240
-
logger.Info("crawl request forwarding enabled", "servers", svcConfig.NextCrawlers)
230
+
relayConfig.TrustedDomains = cctx.StringSlice("trusted-domains")
231
+
232
+
svcConfig := DefaultServiceConfig()
233
+
svcConfig.DisableRequestCrawl = !cctx.Bool("disable-request-crawl")
234
+
svcConfig.ForwardCrawlRequestHosts = cctx.StringSlice("forward-crawl-requests")
235
+
if len(svcConfig.ForwardCrawlRequestHosts) > 0 {
236
+
logger.Info("crawl request forwarding enabled", "servers", svcConfig.ForwardCrawlRequestHosts)
241
237
}
242
238
if cctx.IsSet("admin-password") {
243
239
svcConfig.AdminPassword = cctx.String("admin-password")
+6
-2
cmd/rerelay/relay/relay.go
+6
-2
cmd/rerelay/relay/relay.go
···
41
41
type RelayConfig struct {
42
42
SSL bool
43
43
DefaultRepoLimit int64
44
-
ConcurrencyPerHost int64
44
+
ConcurrencyPerHost int
45
+
QueueDepthPerHost int
45
46
SkipAccountHostCheck bool // XXX: only used for testing
46
47
LenientSyncValidation bool // XXX: wire through config
47
48
···
51
52
}
52
53
53
54
func DefaultRelayConfig() *RelayConfig {
55
+
// NOTE: many of these defaults are CLI arg defaults
54
56
return &RelayConfig{
55
57
SSL: true,
56
58
DefaultRepoLimit: 100,
57
-
ConcurrencyPerHost: 100,
59
+
ConcurrencyPerHost: 40,
60
+
QueueDepthPerHost: 1000,
58
61
}
59
62
}
60
63
···
91
94
slurpConfig.SSL = config.SSL
92
95
slurpConfig.DefaultRepoLimit = config.DefaultRepoLimit
93
96
slurpConfig.ConcurrencyPerHost = config.ConcurrencyPerHost
97
+
slurpConfig.QueueDepthPerHost = config.QueueDepthPerHost
94
98
// register callbacks to persist cursors and host state in database
95
99
slurpConfig.PersistCursorCallback = r.PersistHostCursors
96
100
slurpConfig.PersistHostStatusCallback = r.UpdateHostStatus
+11
-11
cmd/rerelay/relay/slurper.go
+11
-11
cmd/rerelay/relay/slurper.go
···
7
7
"log/slog"
8
8
"math/rand"
9
9
"sync"
10
+
"sync/atomic"
10
11
"time"
11
12
12
13
comatproto "github.com/bluesky-social/indigo/api/atproto"
···
58
59
DefaultPerHourLimit int64
59
60
DefaultPerDayLimit int64
60
61
DefaultRepoLimit int64
61
-
ConcurrencyPerHost int64
62
+
ConcurrencyPerHost int
63
+
QueueDepthPerHost int
62
64
NewHostPerDayLimit int64
63
65
PersistCursorPeriod time.Duration
64
66
PersistCursorCallback PersistCursorFunc
···
66
68
}
67
69
68
70
func DefaultSlurperConfig() *SlurperConfig {
71
+
// NOTE: many of these defaults are overruled by DefaultRelayConfig, or even process CLI arg defaults
69
72
return &SlurperConfig{
70
73
SSL: false,
71
74
DefaultPerSecondLimit: 50,
···
73
76
DefaultPerDayLimit: 20_000,
74
77
DefaultRepoLimit: 100,
75
78
ConcurrencyPerHost: 40,
79
+
QueueDepthPerHost: 1000,
76
80
PersistCursorPeriod: time.Second * 10,
77
81
}
78
82
}
···
81
85
type Subscription struct {
82
86
Hostname string
83
87
HostID uint64
84
-
LastSeq int64 // XXX: switch to an atomic instead of lock?
88
+
LastSeq atomic.Int64
85
89
Limiters *Limiters // XXX: is this used? or only the separate limiters on Slurper?
86
90
87
91
lk sync.RWMutex
···
90
94
}
91
95
92
96
func (sub *Subscription) UpdateSeq(seq int64) {
93
-
sub.lk.Lock()
94
-
defer sub.lk.Unlock()
95
-
sub.LastSeq = seq
97
+
sub.LastSeq.Store(seq)
96
98
}
97
99
98
100
func (sub *Subscription) HostCursor() HostCursor {
···
100
102
defer sub.lk.Unlock()
101
103
return HostCursor{
102
104
HostID: sub.HostID,
103
-
LastSeq: sub.LastSeq,
105
+
LastSeq: sub.LastSeq.Load(),
104
106
}
105
107
}
106
108
···
448
450
instrumentedRSC := stream.NewInstrumentedRepoStreamCallbacks(limiters, rsc.EventHandler)
449
451
450
452
pool := parallel.NewScheduler(
451
-
100, // XXX: concurrency
452
-
1_000, // XXX: max queue per host
453
+
s.Config.ConcurrencyPerHost,
454
+
s.Config.QueueDepthPerHost,
453
455
conn.RemoteAddr().String(),
454
456
instrumentedRSC.EventHandler,
455
457
)
···
474
476
cursors := make([]HostCursor, len(s.subs))
475
477
i := 0
476
478
for _, sub := range s.subs {
477
-
sub.lk.RLock()
478
479
cursors[i] = HostCursor{
479
480
HostID: sub.HostID,
480
-
LastSeq: sub.LastSeq,
481
+
LastSeq: sub.LastSeq.Load(),
481
482
}
482
-
sub.lk.RUnlock()
483
483
i++
484
484
}
485
485
s.subsLk.Unlock()
+5
-3
cmd/rerelay/service.go
+5
-3
cmd/rerelay/service.go
···
6
6
"log/slog"
7
7
"net"
8
8
"net/http"
9
-
"net/url"
10
9
"strings"
11
10
"time"
12
11
···
29
28
}
30
29
31
30
type ServiceConfig struct {
32
-
// NextCrawlers gets forwarded POST /xrpc/com.atproto.sync.requestCrawl
33
-
NextCrawlers []*url.URL
31
+
// list of hosts which get forwarded com.atproto.sync.requestCrawl (HTTP POST)
32
+
ForwardCrawlRequestHosts []string
34
33
35
34
// verified against Basic admin auth
36
35
AdminPassword string
37
36
38
37
// how long to wait for the requested server socket to become available for use
39
38
ListenerBootTimeout time.Duration
39
+
40
+
// if true, don't process public (unauthenticated) requestCrawl
41
+
DisableRequestCrawl bool
40
42
}
41
43
42
44
func DefaultServiceConfig() *ServiceConfig {