this repo has no description
split out validator
+89
-84
+1
-1
cmd/relayered/handlers.go
+1
-1
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/relay/slurper"
15
14
"github.com/bluesky-social/indigo/cmd/relayered/relay"
15
+
"github.com/bluesky-social/indigo/cmd/relayered/relay/slurper"
16
16
"github.com/bluesky-social/indigo/xrpc"
17
17
18
18
"github.com/labstack/echo/v4"
+1
-1
cmd/relayered/handlers_admin.go
+1
-1
cmd/relayered/handlers_admin.go
···
9
9
"strconv"
10
10
"strings"
11
11
12
-
"github.com/bluesky-social/indigo/cmd/relayered/relay/slurper"
13
12
"github.com/bluesky-social/indigo/cmd/relayered/relay"
13
+
"github.com/bluesky-social/indigo/cmd/relayered/relay/slurper"
14
14
15
15
"github.com/labstack/echo/v4"
16
16
dto "github.com/prometheus/client_model/go"
+3
-4
cmd/relayered/main.go
+3
-4
cmd/relayered/main.go
···
19
19
20
20
"github.com/bluesky-social/indigo/atproto/identity"
21
21
"github.com/bluesky-social/indigo/cmd/relayered/relay"
22
-
"github.com/bluesky-social/indigo/cmd/relayered/relay/slurper"
22
+
"github.com/bluesky-social/indigo/cmd/relayered/relay/validator"
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"
···
188
188
}
189
189
cacheDir := identity.NewCacheDirectory(&baseDir, cctx.Int("did-cache-size"), time.Hour*24, time.Minute*2, time.Minute*5)
190
190
191
-
// TODO: rename repoman
192
-
repoman := slurper.NewValidator(&cacheDir)
191
+
vldtr := validator.NewValidator(&cacheDir)
193
192
194
193
var persister persist.EventPersistence
195
194
···
248
247
logger.Info("generated random admin key", "header", "Authorization: Bearer "+svcConfig.AdminToken)
249
248
}
250
249
251
-
r, err := relay.NewRelay(db, repoman, evtman, &cacheDir, relayConfig)
250
+
r, err := relay.NewRelay(db, vldtr, evtman, &cacheDir, relayConfig)
252
251
if err != nil {
253
252
return err
254
253
}
+1
-2
cmd/relayered/models/models.go
+1
-2
cmd/relayered/models/models.go
+17
-17
cmd/relayered/relay/relay.go
+17
-17
cmd/relayered/relay/relay.go
···
7
7
8
8
"github.com/bluesky-social/indigo/atproto/identity"
9
9
"github.com/bluesky-social/indigo/cmd/relayered/relay/slurper"
10
+
"github.com/bluesky-social/indigo/cmd/relayered/relay/validator"
10
11
"github.com/bluesky-social/indigo/cmd/relayered/stream/eventmgr"
11
12
"github.com/bluesky-social/indigo/xrpc"
12
13
···
19
20
var tracer = otel.Tracer("relay")
20
21
21
22
type Relay struct {
22
-
db *gorm.DB
23
-
dir identity.Directory
23
+
db *gorm.DB
24
+
dir identity.Directory
24
25
25
-
Slurper *slurper.Slurper
26
-
Events *eventmgr.EventManager
27
-
Validator *slurper.Validator
26
+
Slurper *slurper.Slurper
27
+
Events *eventmgr.EventManager
28
+
Validator *validator.Validator
28
29
29
30
Config RelayConfig
30
-
31
31
32
32
// extUserLk serializes a section of syncPDSAccount()
33
33
// TODO: at some point we will want to lock specific DIDs, this lock as is
···
41
41
42
42
// Account cache
43
43
userCache *lru.Cache[string, *slurper.Account]
44
-
Logger *slog.Logger
44
+
Logger *slog.Logger
45
45
}
46
46
47
47
type RelayConfig struct {
48
-
SSL bool
49
-
DefaultRepoLimit int64
50
-
ConcurrencyPerPDS int64
51
-
MaxQueuePerPDS int64
48
+
SSL bool
49
+
DefaultRepoLimit int64
50
+
ConcurrencyPerPDS int64
51
+
MaxQueuePerPDS int64
52
52
ApplyPDSClientSettings func(c *xrpc.Client)
53
53
}
54
54
···
61
61
}
62
62
}
63
63
64
-
func NewRelay(db *gorm.DB, validator *slurper.Validator, evtman *eventmgr.EventManager, dir identity.Directory, config *RelayConfig) (*Relay, error) {
64
+
func NewRelay(db *gorm.DB, vldtr *validator.Validator, evtman *eventmgr.EventManager, dir identity.Directory, config *RelayConfig) (*Relay, error) {
65
65
66
66
if config == nil {
67
67
config = DefaultRelayConfig()
···
70
70
uc, _ := lru.New[string, *slurper.Account](1_000_000)
71
71
72
72
r := &Relay{
73
-
db: db,
74
-
Events: evtman,
75
-
Validator: validator,
73
+
db: db,
74
+
Events: evtman,
75
+
Validator: vldtr,
76
76
dir: dir,
77
-
Config: *config,
77
+
Config: *config,
78
78
79
79
consumersLk: sync.RWMutex{},
80
80
consumers: make(map[uint64]*SocketConsumer),
···
106
106
return r, nil
107
107
}
108
108
109
-
func (r *Relay) MigrateDatabase() error {
109
+
func (r *Relay) MigrateDatabase() error {
110
110
if err := r.db.AutoMigrate(slurper.DomainBan{}); err != nil {
111
111
return err
112
112
}
-28
cmd/relayered/relay/slurper/metrics.go
-28
cmd/relayered/relay/slurper/metrics.go
···
9
9
Name: "relay_connected_inbound",
10
10
Help: "Number of inbound firehoses we are consuming",
11
11
})
12
-
13
-
var commitVerifyStarts = promauto.NewCounter(prometheus.CounterOpts{
14
-
Name: "validator_commit_verify_starts",
15
-
})
16
-
17
-
var commitVerifyWarnings = promauto.NewCounterVec(prometheus.CounterOpts{
18
-
Name: "validator_commit_verify_warnings",
19
-
}, []string{"host", "warn"})
20
-
21
-
// verify error and short code for why
22
-
var commitVerifyErrors = promauto.NewCounterVec(prometheus.CounterOpts{
23
-
Name: "validator_commit_verify_errors",
24
-
}, []string{"host", "err"})
25
-
26
-
// ok and *fully verified*
27
-
var commitVerifyOk = promauto.NewCounterVec(prometheus.CounterOpts{
28
-
Name: "validator_commit_verify_ok",
29
-
}, []string{"host"})
30
-
31
-
// it's ok, but... {old protocol, no previous root cid, ...}
32
-
var commitVerifyOkish = promauto.NewCounterVec(prometheus.CounterOpts{
33
-
Name: "validator_commit_verify_okish",
34
-
}, []string{"host", "but"})
35
-
36
-
// verify error and short code for why
37
-
var syncVerifyErrors = promauto.NewCounterVec(prometheus.CounterOpts{
38
-
Name: "validator_sync_verify_errors",
39
-
}, []string{"host", "err"})
+4
-4
cmd/relayered/relay/slurper/models.go
+4
-4
cmd/relayered/relay/slurper/models.go
···
17
17
}
18
18
19
19
type AccountPreviousState struct {
20
-
Uid models.Uid `gorm:"column:uid;primaryKey"`
21
-
Cid DbCID `gorm:"column:cid"`
22
-
Rev string `gorm:"column:rev"`
23
-
Seq int64 `gorm:"column:seq"`
20
+
Uid models.Uid `gorm:"column:uid;primaryKey"`
21
+
Cid DbCID `gorm:"column:cid"`
22
+
Rev string `gorm:"column:rev"`
23
+
Seq int64 `gorm:"column:seq"`
24
24
}
25
25
26
26
func (ups *AccountPreviousState) GetCid() cid.Cid {
+21
-19
cmd/relayered/relay/slurper/validator.go
cmd/relayered/relay/validator/validator.go
+21
-19
cmd/relayered/relay/slurper/validator.go
cmd/relayered/relay/validator/validator.go
···
1
-
package slurper
1
+
package validator
2
2
3
3
import (
4
4
"bytes"
···
9
9
"sync/atomic"
10
10
"time"
11
11
12
-
atproto "github.com/bluesky-social/indigo/api/atproto"
12
+
comatproto "github.com/bluesky-social/indigo/api/atproto"
13
13
"github.com/bluesky-social/indigo/atproto/identity"
14
-
atrepo "github.com/bluesky-social/indigo/atproto/repo"
14
+
"github.com/bluesky-social/indigo/atproto/repo"
15
15
"github.com/bluesky-social/indigo/atproto/syntax"
16
16
"github.com/bluesky-social/indigo/cmd/relayered/models"
17
+
"github.com/bluesky-social/indigo/cmd/relayered/relay/slurper"
18
+
17
19
"github.com/ipfs/go-cid"
18
20
"go.opentelemetry.io/otel"
19
21
)
···
25
27
ErrRevTooFarFuture := fmt.Errorf("new rev is > %s in the future", maxRevFuture)
26
28
27
29
return &Validator{
28
-
userLocks: make(map[models.Uid]*userLock),
29
-
log: slog.Default().With("system", "validator"),
30
-
directory: directory,
30
+
userLocks: make(map[models.Uid]*userLock),
31
+
log: slog.Default().With("system", "validator"),
32
+
directory: directory,
31
33
32
34
maxRevFuture: maxRevFuture,
33
35
ErrRevTooFarFuture: ErrRevTooFarFuture,
···
40
42
lklk sync.Mutex
41
43
userLocks map[models.Uid]*userLock
42
44
43
-
log *slog.Logger
45
+
log *slog.Logger
44
46
45
47
directory identity.Directory
46
48
···
57
59
}
58
60
59
61
type NextCommitHandler interface {
60
-
HandleCommit(ctx context.Context, host *PDS, uid models.Uid, did string, commit *atproto.SyncSubscribeRepos_Commit) error
62
+
HandleCommit(ctx context.Context, host *slurper.PDS, uid models.Uid, did string, commit *comatproto.SyncSubscribeRepos_Commit) error
61
63
}
62
64
63
65
type userLock struct {
···
98
100
}
99
101
}
100
102
101
-
func (val *Validator) HandleCommit(ctx context.Context, host *PDS, account *Account, commit *atproto.SyncSubscribeRepos_Commit, prevRoot *AccountPreviousState) (newRoot *cid.Cid, err error) {
103
+
func (val *Validator) HandleCommit(ctx context.Context, host *slurper.PDS, account *slurper.Account, commit *comatproto.SyncSubscribeRepos_Commit, prevRoot *slurper.AccountPreviousState) (newRoot *cid.Cid, err error) {
102
104
uid := account.GetUid()
103
105
unlock := val.lockUser(ctx, uid)
104
106
defer unlock()
···
123
125
124
126
var ErrNewRevBeforePrevRev = &revOutOfOrderError{}
125
127
126
-
func (val *Validator) VerifyCommitMessage(ctx context.Context, host *PDS, msg *atproto.SyncSubscribeRepos_Commit, prevRoot *AccountPreviousState) (*atrepo.Repo, error) {
128
+
func (val *Validator) VerifyCommitMessage(ctx context.Context, host *slurper.PDS, msg *comatproto.SyncSubscribeRepos_Commit, prevRoot *slurper.AccountPreviousState) (*repo.Repo, error) {
127
129
hostname := host.Host
128
130
hasWarning := false
129
131
commitVerifyStarts.Inc()
···
262
264
commitVerifyErrors.WithLabelValues(hostname, "pop").Inc()
263
265
return nil, err
264
266
}
265
-
ops, err = atrepo.NormalizeOps(ops)
267
+
ops, err = repo.NormalizeOps(ops)
266
268
if err != nil {
267
269
commitVerifyErrors.WithLabelValues(hostname, "nop").Inc()
268
270
return nil, err
···
270
272
271
273
invTree := repoFragment.MST.Copy()
272
274
for _, op := range ops {
273
-
if err := atrepo.InvertOp(&invTree, &op); err != nil {
275
+
if err := repo.InvertOp(&invTree, &op); err != nil {
274
276
commitVerifyErrors.WithLabelValues(hostname, "inv").Inc()
275
277
return nil, err
276
278
}
···
305
307
}
306
308
307
309
// HandleSync checks signed commit from a #sync message
308
-
func (val *Validator) HandleSync(ctx context.Context, host *PDS, msg *atproto.SyncSubscribeRepos_Sync) (newRoot *cid.Cid, err error) {
310
+
func (val *Validator) HandleSync(ctx context.Context, host *slurper.PDS, msg *comatproto.SyncSubscribeRepos_Sync) (newRoot *cid.Cid, err error) {
309
311
hostname := host.Host
310
312
hasWarning := false
311
313
···
354
356
}
355
357
356
358
// TODO: lift back to indigo/atproto/repo util code?
357
-
func ParseCommitOps(ops []*atproto.SyncSubscribeRepos_RepoOp) ([]atrepo.Operation, error) {
358
-
out := []atrepo.Operation{}
359
+
func ParseCommitOps(ops []*comatproto.SyncSubscribeRepos_RepoOp) ([]repo.Operation, error) {
360
+
out := []repo.Operation{}
359
361
for _, rop := range ops {
360
362
switch rop.Action {
361
363
case "create":
362
364
if rop.Cid == nil || rop.Prev != nil {
363
365
return nil, fmt.Errorf("invalid repoOp: create")
364
366
}
365
-
op := atrepo.Operation{
367
+
op := repo.Operation{
366
368
Path: rop.Path,
367
369
Prev: nil,
368
370
Value: (*cid.Cid)(rop.Cid),
···
372
374
if rop.Cid != nil || rop.Prev == nil {
373
375
return nil, fmt.Errorf("invalid repoOp: delete")
374
376
}
375
-
op := atrepo.Operation{
377
+
op := repo.Operation{
376
378
Path: rop.Path,
377
379
Prev: (*cid.Cid)(rop.Prev),
378
380
Value: nil,
···
382
384
if rop.Cid == nil || rop.Prev == nil {
383
385
return nil, fmt.Errorf("invalid repoOp: update")
384
386
}
385
-
op := atrepo.Operation{
387
+
op := repo.Operation{
386
388
Path: rop.Path,
387
389
Prev: (*cid.Cid)(rop.Prev),
388
390
Value: (*cid.Cid)(rop.Cid),
···
397
399
398
400
// VerifyCommitSignature get's repo's registered public key from Identity Directory, verifies Commit
399
401
// hostname is just for metrics in case of error
400
-
func (val *Validator) VerifyCommitSignature(ctx context.Context, commit *atrepo.Commit, hostname string, hasWarning *bool) error {
402
+
func (val *Validator) VerifyCommitSignature(ctx context.Context, commit *repo.Commit, hostname string, hasWarning *bool) error {
401
403
if val.directory == nil {
402
404
return nil
403
405
}
+34
cmd/relayered/relay/validator/metrics.go
+34
cmd/relayered/relay/validator/metrics.go
···
1
+
package validator
2
+
3
+
import (
4
+
"github.com/prometheus/client_golang/prometheus"
5
+
"github.com/prometheus/client_golang/prometheus/promauto"
6
+
)
7
+
8
+
var commitVerifyStarts = promauto.NewCounter(prometheus.CounterOpts{
9
+
Name: "validator_commit_verify_starts",
10
+
})
11
+
12
+
var commitVerifyWarnings = promauto.NewCounterVec(prometheus.CounterOpts{
13
+
Name: "validator_commit_verify_warnings",
14
+
}, []string{"host", "warn"})
15
+
16
+
// verify error and short code for why
17
+
var commitVerifyErrors = promauto.NewCounterVec(prometheus.CounterOpts{
18
+
Name: "validator_commit_verify_errors",
19
+
}, []string{"host", "err"})
20
+
21
+
// ok and *fully verified*
22
+
var commitVerifyOk = promauto.NewCounterVec(prometheus.CounterOpts{
23
+
Name: "validator_commit_verify_ok",
24
+
}, []string{"host"})
25
+
26
+
// it's ok, but... {old protocol, no previous root cid, ...}
27
+
var commitVerifyOkish = promauto.NewCounterVec(prometheus.CounterOpts{
28
+
Name: "validator_commit_verify_okish",
29
+
}, []string{"host", "but"})
30
+
31
+
// verify error and short code for why
32
+
var syncVerifyErrors = promauto.NewCounterVec(prometheus.CounterOpts{
33
+
Name: "validator_sync_verify_errors",
34
+
}, []string{"host", "err"})
+7
-8
cmd/relayered/service.go
+7
-8
cmd/relayered/service.go
···
27
27
const serverListenerBootTimeout = 5 * time.Second
28
28
29
29
type Service struct {
30
-
db *gorm.DB // XXX
31
-
relay *relay.Relay
32
-
dir identity.Directory
30
+
db *gorm.DB // XXX
31
+
relay *relay.Relay
32
+
dir identity.Directory
33
33
34
34
// TODO: work on doing away with this flag in favor of more pluggable
35
35
// pieces that abstract the need for explicit ssl checks
36
36
ssl bool
37
-
38
37
39
38
// nextCrawlers gets forwarded POST /xrpc/com.atproto.sync.requestCrawl
40
39
nextCrawlers []*url.URL
41
40
httpClient http.Client
42
41
43
-
log *slog.Logger
42
+
log *slog.Logger
44
43
45
44
config ServiceConfig
46
45
}
···
64
63
}
65
64
66
65
svc := &Service{
67
-
db: db,
66
+
db: db,
68
67
relay: r,
69
-
dir: dir,
70
-
ssl: r.Config.SSL,
68
+
dir: dir,
69
+
ssl: r.Config.SSL,
71
70
72
71
log: slog.Default().With("system", "relay"),
73
72