this repo has no description
non-archival relay (#827)
real jank right now, but doesnt keep all the repo data, just does basic
event validation
authored by
Whyrusleeping
and committed by
GitHub
86719b02
749cc07b
+522
-44
+3
bgs/bgs.go
+3
bgs/bgs.go
···
1188
1188
}).Error; err != nil {
1189
1189
return err
1190
1190
}
1191
+
u.SetTombstoned(true)
1191
1192
1192
1193
if err := bgs.db.Model(&models.ActorInfo{}).Where("uid = ?", u.ID).UpdateColumns(map[string]any{
1193
1194
"handle": nil,
···
1416
1417
if err := s.db.Create(&u).Error; err != nil {
1417
1418
return nil, fmt.Errorf("failed to create user after handle conflict: %w", err)
1418
1419
}
1420
+
1421
+
s.userCache.Remove(did)
1419
1422
} else {
1420
1423
return nil, fmt.Errorf("failed to create other pds user: %w", err)
1421
1424
}
+49
-5
carstore/bs.go
+49
-5
carstore/bs.go
···
97
97
}
98
98
99
99
type userView struct {
100
-
cs *FileCarStore
100
+
cs CarStore
101
101
user models.Uid
102
102
103
103
cache map[cid.Cid]blockformat.Block
···
111
111
}
112
112
113
113
func (uv *userView) Has(ctx context.Context, k cid.Cid) (bool, error) {
114
-
return uv.cs.meta.HasUidCid(ctx, uv.user, k)
114
+
_, have := uv.cache[k]
115
+
if have {
116
+
return have, nil
117
+
}
118
+
119
+
fcd, ok := uv.cs.(*FileCarStore)
120
+
if !ok {
121
+
return false, nil
122
+
}
123
+
124
+
return fcd.meta.HasUidCid(ctx, uv.user, k)
115
125
}
116
126
117
127
var CacheHits int64
118
128
var CacheMiss int64
119
129
120
130
func (uv *userView) Get(ctx context.Context, k cid.Cid) (blockformat.Block, error) {
131
+
121
132
if !k.Defined() {
122
133
return nil, fmt.Errorf("attempted to 'get' undefined cid")
123
134
}
···
132
143
}
133
144
atomic.AddInt64(&CacheMiss, 1)
134
145
135
-
path, offset, user, err := uv.cs.meta.LookupBlockRef(ctx, k)
146
+
fcd, ok := uv.cs.(*FileCarStore)
147
+
if !ok {
148
+
return nil, ipld.ErrNotFound{Cid: k}
149
+
}
150
+
151
+
path, offset, user, err := fcd.meta.LookupBlockRef(ctx, k)
136
152
if err != nil {
137
153
return nil, err
138
154
}
···
272
288
baseCid cid.Cid
273
289
seq int
274
290
readonly bool
275
-
cs *FileCarStore
291
+
cs CarStore
276
292
lastRev string
277
293
}
278
294
···
587
603
return nil, fmt.Errorf("cannot write to readonly deltaSession")
588
604
}
589
605
590
-
return ds.cs.writeNewShard(ctx, root, rev, ds.user, ds.seq, ds.blks, ds.rmcids)
606
+
switch ocs := ds.cs.(type) {
607
+
case *FileCarStore:
608
+
return ocs.writeNewShard(ctx, root, rev, ds.user, ds.seq, ds.blks, ds.rmcids)
609
+
case *NonArchivalCarstore:
610
+
slice, err := blocksToCar(ctx, root, rev, ds.blks)
611
+
if err != nil {
612
+
return nil, err
613
+
}
614
+
return slice, ocs.updateLastCommit(ctx, ds.user, rev, root)
615
+
default:
616
+
return nil, fmt.Errorf("unsupported carstore type")
617
+
}
591
618
}
592
619
593
620
func WriteCarHeader(w io.Writer, root cid.Cid) (int64, error) {
···
606
633
}
607
634
608
635
return hnw, nil
636
+
}
637
+
638
+
func blocksToCar(ctx context.Context, root cid.Cid, rev string, blks map[cid.Cid]blockformat.Block) ([]byte, error) {
639
+
buf := new(bytes.Buffer)
640
+
_, err := WriteCarHeader(buf, root)
641
+
if err != nil {
642
+
return nil, fmt.Errorf("failed to write car header: %w", err)
643
+
}
644
+
645
+
for k, blk := range blks {
646
+
_, err := LdWrite(buf, k.Bytes(), blk.RawData())
647
+
if err != nil {
648
+
return nil, fmt.Errorf("failed to write block: %w", err)
649
+
}
650
+
}
651
+
652
+
return buf.Bytes(), nil
609
653
}
610
654
611
655
func (cs *FileCarStore) writeNewShard(ctx context.Context, root cid.Cid, rev string, user models.Uid, seq int, blks map[cid.Cid]blockformat.Block, rmcids map[cid.Cid]bool) ([]byte, error) {
+254
carstore/nonarchive.go
+254
carstore/nonarchive.go
···
1
+
package carstore
2
+
3
+
import (
4
+
"bytes"
5
+
"context"
6
+
"fmt"
7
+
"io"
8
+
"log/slog"
9
+
"sync"
10
+
11
+
"github.com/bluesky-social/indigo/models"
12
+
blockformat "github.com/ipfs/go-block-format"
13
+
"github.com/ipfs/go-cid"
14
+
"github.com/ipfs/go-datastore"
15
+
blockstore "github.com/ipfs/go-ipfs-blockstore"
16
+
car "github.com/ipld/go-car"
17
+
"go.opentelemetry.io/otel"
18
+
"gorm.io/gorm"
19
+
"gorm.io/gorm/clause"
20
+
)
21
+
22
+
type NonArchivalCarstore struct {
23
+
db *gorm.DB
24
+
25
+
lk sync.Mutex
26
+
lastCommitCache map[models.Uid]*commitRefInfo
27
+
28
+
log *slog.Logger
29
+
}
30
+
31
+
func NewNonArchivalCarstore(db *gorm.DB) (*NonArchivalCarstore, error) {
32
+
if err := db.AutoMigrate(&commitRefInfo{}); err != nil {
33
+
return nil, err
34
+
}
35
+
36
+
return &NonArchivalCarstore{
37
+
db: db,
38
+
lastCommitCache: make(map[models.Uid]*commitRefInfo),
39
+
log: slog.Default().With("system", "carstorena"),
40
+
}, nil
41
+
}
42
+
43
+
type commitRefInfo struct {
44
+
ID uint `gorm:"primarykey"`
45
+
Uid models.Uid `gorm:"uniqueIndex"`
46
+
Rev string
47
+
Root models.DbCID
48
+
}
49
+
50
+
func (cs *NonArchivalCarstore) checkLastShardCache(user models.Uid) *commitRefInfo {
51
+
cs.lk.Lock()
52
+
defer cs.lk.Unlock()
53
+
54
+
ls, ok := cs.lastCommitCache[user]
55
+
if ok {
56
+
return ls
57
+
}
58
+
59
+
return nil
60
+
}
61
+
62
+
func (cs *NonArchivalCarstore) removeLastShardCache(user models.Uid) {
63
+
cs.lk.Lock()
64
+
defer cs.lk.Unlock()
65
+
66
+
delete(cs.lastCommitCache, user)
67
+
}
68
+
69
+
func (cs *NonArchivalCarstore) putLastShardCache(ls *commitRefInfo) {
70
+
cs.lk.Lock()
71
+
defer cs.lk.Unlock()
72
+
73
+
cs.lastCommitCache[ls.Uid] = ls
74
+
}
75
+
76
+
func (cs *NonArchivalCarstore) loadCommitRefInfo(ctx context.Context, user models.Uid) (*commitRefInfo, error) {
77
+
var out commitRefInfo
78
+
if err := cs.db.Find(&out, "uid = ?", user).Error; err != nil {
79
+
return nil, err
80
+
}
81
+
82
+
return &out, nil
83
+
}
84
+
85
+
func (cs *NonArchivalCarstore) getCommitRefInfo(ctx context.Context, user models.Uid) (*commitRefInfo, error) {
86
+
ctx, span := otel.Tracer("carstore").Start(ctx, "getCommitRefInfo")
87
+
defer span.End()
88
+
89
+
maybeLs := cs.checkLastShardCache(user)
90
+
if maybeLs != nil {
91
+
return maybeLs, nil
92
+
}
93
+
94
+
lastShard, err := cs.loadCommitRefInfo(ctx, user)
95
+
if err != nil {
96
+
return nil, err
97
+
}
98
+
99
+
cs.putLastShardCache(lastShard)
100
+
return lastShard, nil
101
+
}
102
+
103
+
func (cs *NonArchivalCarstore) updateLastCommit(ctx context.Context, uid models.Uid, rev string, cid cid.Cid) error {
104
+
cri := &commitRefInfo{
105
+
Uid: uid,
106
+
Rev: rev,
107
+
Root: models.DbCID{CID: cid},
108
+
}
109
+
110
+
if err := cs.db.Clauses(clause.OnConflict{
111
+
Columns: []clause.Column{{Name: "uid"}},
112
+
UpdateAll: true,
113
+
}).Create(cri).Error; err != nil {
114
+
return fmt.Errorf("update or set last commit info: %w", err)
115
+
}
116
+
117
+
cs.putLastShardCache(cri)
118
+
119
+
return nil
120
+
}
121
+
122
+
func (cs *NonArchivalCarstore) NewDeltaSession(ctx context.Context, user models.Uid, since *string) (*DeltaSession, error) {
123
+
ctx, span := otel.Tracer("carstore").Start(ctx, "NewSession")
124
+
defer span.End()
125
+
126
+
// TODO: ensure that we don't write updates on top of the wrong head
127
+
// this needs to be a compare and swap type operation
128
+
lastShard, err := cs.getCommitRefInfo(ctx, user)
129
+
if err != nil {
130
+
return nil, err
131
+
}
132
+
133
+
if since != nil && *since != lastShard.Rev {
134
+
cs.log.Warn("revision mismatch: %s != %s: %s", *since, lastShard.Rev, ErrRepoBaseMismatch)
135
+
}
136
+
137
+
return &DeltaSession{
138
+
fresh: blockstore.NewBlockstore(datastore.NewMapDatastore()),
139
+
blks: make(map[cid.Cid]blockformat.Block),
140
+
base: &userView{
141
+
user: user,
142
+
cs: cs,
143
+
prefetch: true,
144
+
cache: make(map[cid.Cid]blockformat.Block),
145
+
},
146
+
user: user,
147
+
baseCid: lastShard.Root.CID,
148
+
cs: cs,
149
+
seq: 0,
150
+
lastRev: lastShard.Rev,
151
+
}, nil
152
+
}
153
+
154
+
func (cs *NonArchivalCarstore) ReadOnlySession(user models.Uid) (*DeltaSession, error) {
155
+
return &DeltaSession{
156
+
base: &userView{
157
+
user: user,
158
+
cs: cs,
159
+
prefetch: false,
160
+
cache: make(map[cid.Cid]blockformat.Block),
161
+
},
162
+
readonly: true,
163
+
user: user,
164
+
cs: cs,
165
+
}, nil
166
+
}
167
+
168
+
// TODO: incremental is only ever called true, remove the param
169
+
func (cs *NonArchivalCarstore) ReadUserCar(ctx context.Context, user models.Uid, sinceRev string, incremental bool, w io.Writer) error {
170
+
return fmt.Errorf("not supported in non-archival mode")
171
+
}
172
+
173
+
func (cs *NonArchivalCarstore) ImportSlice(ctx context.Context, uid models.Uid, since *string, carslice []byte) (cid.Cid, *DeltaSession, error) {
174
+
ctx, span := otel.Tracer("carstore").Start(ctx, "ImportSlice")
175
+
defer span.End()
176
+
177
+
carr, err := car.NewCarReader(bytes.NewReader(carslice))
178
+
if err != nil {
179
+
return cid.Undef, nil, err
180
+
}
181
+
182
+
if len(carr.Header.Roots) != 1 {
183
+
return cid.Undef, nil, fmt.Errorf("invalid car file, header must have a single root (has %d)", len(carr.Header.Roots))
184
+
}
185
+
186
+
ds, err := cs.NewDeltaSession(ctx, uid, since)
187
+
if err != nil {
188
+
return cid.Undef, nil, fmt.Errorf("new delta session failed: %w", err)
189
+
}
190
+
191
+
var cids []cid.Cid
192
+
for {
193
+
blk, err := carr.Next()
194
+
if err != nil {
195
+
if err == io.EOF {
196
+
break
197
+
}
198
+
return cid.Undef, nil, err
199
+
}
200
+
201
+
cids = append(cids, blk.Cid())
202
+
203
+
if err := ds.Put(ctx, blk); err != nil {
204
+
return cid.Undef, nil, err
205
+
}
206
+
}
207
+
208
+
return carr.Header.Roots[0], ds, nil
209
+
}
210
+
211
+
func (cs *NonArchivalCarstore) GetUserRepoHead(ctx context.Context, user models.Uid) (cid.Cid, error) {
212
+
lastShard, err := cs.getCommitRefInfo(ctx, user)
213
+
if err != nil {
214
+
return cid.Undef, err
215
+
}
216
+
if lastShard.ID == 0 {
217
+
return cid.Undef, nil
218
+
}
219
+
220
+
return lastShard.Root.CID, nil
221
+
}
222
+
223
+
func (cs *NonArchivalCarstore) GetUserRepoRev(ctx context.Context, user models.Uid) (string, error) {
224
+
lastShard, err := cs.getCommitRefInfo(ctx, user)
225
+
if err != nil {
226
+
return "", err
227
+
}
228
+
if lastShard.ID == 0 {
229
+
return "", nil
230
+
}
231
+
232
+
return lastShard.Rev, nil
233
+
}
234
+
235
+
func (cs *NonArchivalCarstore) Stat(ctx context.Context, usr models.Uid) ([]UserStat, error) {
236
+
return nil, nil
237
+
}
238
+
239
+
func (cs *NonArchivalCarstore) WipeUserData(ctx context.Context, user models.Uid) error {
240
+
if err := cs.db.Raw("DELETE from commit_ref_infos WHERE uid = ?", user).Error; err != nil {
241
+
return err
242
+
}
243
+
244
+
cs.removeLastShardCache(user)
245
+
return nil
246
+
}
247
+
248
+
func (cs *NonArchivalCarstore) GetCompactionTargets(ctx context.Context, shardCount int) ([]CompactionTarget, error) {
249
+
return nil, fmt.Errorf("compaction not supported on non-archival")
250
+
}
251
+
252
+
func (cs *NonArchivalCarstore) CompactUserShards(ctx context.Context, user models.Uid, skipBigShards bool) (*CompactionStats, error) {
253
+
return nil, fmt.Errorf("compaction not supported in non-archival")
254
+
}
+21
-3
cmd/bigsky/main.go
+21
-3
cmd/bigsky/main.go
···
216
216
Usage: "forward POST requestCrawl to this url, should be machine root url and not xrpc/requestCrawl, comma separated list",
217
217
EnvVars: []string{"RELAY_NEXT_CRAWLER"},
218
218
},
219
+
&cli.BoolFlag{
220
+
Name: "non-archival",
221
+
EnvVars: []string{"RELAY_NON_ARCHIVAL"},
222
+
Value: false,
223
+
},
219
224
}
220
225
221
226
app.Action = runBigsky
···
345
350
}
346
351
}
347
352
348
-
cstore, err := carstore.NewCarStore(csdb, csdirs)
349
-
if err != nil {
350
-
return err
353
+
var cstore carstore.CarStore
354
+
355
+
if cctx.Bool("non-archival") {
356
+
cs, err := carstore.NewNonArchivalCarstore(csdb)
357
+
if err != nil {
358
+
return err
359
+
}
360
+
361
+
cstore = cs
362
+
} else {
363
+
cs, err := carstore.NewCarStore(csdb, csdirs)
364
+
if err != nil {
365
+
return err
366
+
}
367
+
368
+
cstore = cs
351
369
}
352
370
353
371
// DID RESOLUTION
+1
indexer/indexer.go
+1
indexer/indexer.go
···
538
538
case *bsky.ActorProfile:
539
539
ix.log.Debug("TODO: got actor profile record creation, need to do something with this")
540
540
default:
541
+
ix.log.Warn("unrecognized record", "record", op.Record, "collection", op.Collection)
541
542
return nil, fmt.Errorf("unrecognized record type (creation): %s", op.Collection)
542
543
}
543
544
+1
-11
pds/server.go
+1
-11
pds/server.go
···
15
15
16
16
"github.com/bluesky-social/indigo/api/atproto"
17
17
comatproto "github.com/bluesky-social/indigo/api/atproto"
18
-
bsky "github.com/bluesky-social/indigo/api/bsky"
19
18
"github.com/bluesky-social/indigo/carstore"
20
19
"github.com/bluesky-social/indigo/events"
21
20
"github.com/bluesky-social/indigo/indexer"
···
202
201
handle = hurl.Host
203
202
}
204
203
205
-
profile, err := bsky.ActorGetProfile(ctx, c, did)
206
-
if err != nil {
207
-
return nil, err
208
-
}
209
-
210
-
if handle != profile.Handle {
211
-
return nil, fmt.Errorf("mismatch in handle between did document and pds profile (%s != %s)", handle, profile.Handle)
212
-
}
213
-
214
204
// TODO: request this users info from their server to fill out our data...
215
205
u := User{
216
206
Handle: handle,
···
227
217
subj := &models.ActorInfo{
228
218
Uid: u.ID,
229
219
Handle: sql.NullString{String: handle, Valid: true},
230
-
DisplayName: *profile.DisplayName,
220
+
DisplayName: "missing display name",
231
221
Did: did,
232
222
Type: "",
233
223
PDS: peering.ID,
+9
-1
repomgr/bench_test.go
+9
-1
repomgr/bench_test.go
···
54
54
b.Fatal(err)
55
55
}
56
56
57
-
cs, err := carstore.NewCarStore(cardb, []string{cspath})
57
+
// TODO: constructor for 'either type'
58
+
/*
59
+
cs, err := carstore.NewCarStore(cardb, []string{cspath})
60
+
if err != nil {
61
+
b.Fatal(err)
62
+
}
63
+
*/
64
+
cs, err := carstore.NewNonArchivalCarstore(cardb)
58
65
if err != nil {
59
66
b.Fatal(err)
60
67
}
61
68
62
69
repoman := NewRepoManager(cs, &util.FakeKeyManager{})
70
+
repoman.noArchive = true
63
71
64
72
ctx := context.TODO()
65
73
if err := repoman.InitNewActor(ctx, 1, "hello.world", "did:foo:bar", "catdog", "", ""); err != nil {
+17
-9
repomgr/ingest_test.go
+17
-9
repomgr/ingest_test.go
···
69
69
}
70
70
}
71
71
72
-
func testCarstore(t *testing.T, dir string) carstore.CarStore {
72
+
func testCarstore(t *testing.T, dir string, archive bool) carstore.CarStore {
73
73
cardb, err := gorm.Open(sqlite.Open(filepath.Join(dir, "car.sqlite")))
74
74
if err != nil {
75
75
t.Fatal(err)
···
80
80
t.Fatal(err)
81
81
}
82
82
83
-
cs, err := carstore.NewCarStore(cardb, []string{cspath})
84
-
if err != nil {
85
-
t.Fatal(err)
86
-
}
83
+
if archive {
84
+
cs, err := carstore.NewCarStore(cardb, []string{cspath})
85
+
if err != nil {
86
+
t.Fatal(err)
87
+
}
88
+
return cs
89
+
} else {
90
+
cs, err := carstore.NewNonArchivalCarstore(cardb)
91
+
if err != nil {
92
+
t.Fatal(err)
93
+
}
87
94
88
-
return cs
95
+
return cs
96
+
}
89
97
}
90
98
91
99
func TestIngestWithGap(t *testing.T) {
···
106
114
Uid: 1,
107
115
})
108
116
109
-
cs := testCarstore(t, dir)
117
+
cs := testCarstore(t, dir, true)
110
118
111
119
repoman := NewRepoManager(cs, &util.FakeKeyManager{})
112
120
···
114
122
if err != nil {
115
123
t.Fatal(err)
116
124
}
117
-
cs2 := testCarstore(t, dir2)
125
+
cs2 := testCarstore(t, dir2, true)
118
126
119
127
var since *string
120
128
ctx := context.TODO()
···
198
206
Uid: 1,
199
207
})
200
208
201
-
cs := testCarstore(t, dir)
209
+
cs := testCarstore(t, dir, true)
202
210
203
211
repoman := NewRepoManager(cs, &util.FakeKeyManager{})
204
212
+113
-1
repomgr/repomgr.go
+113
-1
repomgr/repomgr.go
···
34
34
35
35
func NewRepoManager(cs carstore.CarStore, kmgr KeyManager) *RepoManager {
36
36
37
+
var noArchive bool
38
+
if _, ok := cs.(*carstore.NonArchivalCarstore); ok {
39
+
noArchive = true
40
+
}
41
+
37
42
return &RepoManager{
38
43
cs: cs,
39
44
userLocks: make(map[models.Uid]*userLock),
40
45
kmgr: kmgr,
41
46
log: slog.Default().With("system", "repomgr"),
47
+
noArchive: noArchive,
42
48
}
43
49
}
44
50
···
62
68
events func(context.Context, *RepoEvent)
63
69
hydrateRecords bool
64
70
65
-
log *slog.Logger
71
+
log *slog.Logger
72
+
noArchive bool
66
73
}
67
74
68
75
type ActorInfo struct {
···
530
537
}
531
538
532
539
func (rm *RepoManager) HandleExternalUserEvent(ctx context.Context, pdsid uint, uid models.Uid, did string, since *string, nrev string, carslice []byte, ops []*atproto.SyncSubscribeRepos_RepoOp) error {
540
+
if rm.noArchive {
541
+
return rm.handleExternalUserEventNoArchive(ctx, pdsid, uid, did, since, nrev, carslice, ops)
542
+
} else {
543
+
return rm.handleExternalUserEventArchive(ctx, pdsid, uid, did, since, nrev, carslice, ops)
544
+
}
545
+
}
546
+
547
+
func (rm *RepoManager) handleExternalUserEventNoArchive(ctx context.Context, pdsid uint, uid models.Uid, did string, since *string, nrev string, carslice []byte, ops []*atproto.SyncSubscribeRepos_RepoOp) error {
548
+
ctx, span := otel.Tracer("repoman").Start(ctx, "HandleExternalUserEvent")
549
+
defer span.End()
550
+
551
+
span.SetAttributes(attribute.Int64("uid", int64(uid)))
552
+
553
+
rm.log.Debug("HandleExternalUserEvent", "pds", pdsid, "uid", uid, "since", since, "nrev", nrev)
554
+
555
+
unlock := rm.lockUser(ctx, uid)
556
+
defer unlock()
557
+
558
+
start := time.Now()
559
+
root, ds, err := rm.cs.ImportSlice(ctx, uid, since, carslice)
560
+
if err != nil {
561
+
return fmt.Errorf("importing external carslice: %w", err)
562
+
}
563
+
564
+
r, err := repo.OpenRepo(ctx, ds, root)
565
+
if err != nil {
566
+
return fmt.Errorf("opening external user repo (%d, root=%s): %w", uid, root, err)
567
+
}
568
+
569
+
if err := rm.CheckRepoSig(ctx, r, did); err != nil {
570
+
return fmt.Errorf("check repo sig: %w", err)
571
+
}
572
+
openAndSigCheckDuration.Observe(time.Since(start).Seconds())
573
+
574
+
evtops := make([]RepoOp, 0, len(ops))
575
+
for _, op := range ops {
576
+
parts := strings.SplitN(op.Path, "/", 2)
577
+
if len(parts) != 2 {
578
+
return fmt.Errorf("invalid rpath in mst diff, must have collection and rkey")
579
+
}
580
+
581
+
switch EventKind(op.Action) {
582
+
case EvtKindCreateRecord:
583
+
rop := RepoOp{
584
+
Kind: EvtKindCreateRecord,
585
+
Collection: parts[0],
586
+
Rkey: parts[1],
587
+
RecCid: (*cid.Cid)(op.Cid),
588
+
}
589
+
590
+
if rm.hydrateRecords {
591
+
_, rec, err := r.GetRecord(ctx, op.Path)
592
+
if err != nil {
593
+
return fmt.Errorf("reading changed record from car slice: %w", err)
594
+
}
595
+
rop.Record = rec
596
+
}
597
+
598
+
evtops = append(evtops, rop)
599
+
case EvtKindUpdateRecord:
600
+
rop := RepoOp{
601
+
Kind: EvtKindUpdateRecord,
602
+
Collection: parts[0],
603
+
Rkey: parts[1],
604
+
RecCid: (*cid.Cid)(op.Cid),
605
+
}
606
+
607
+
if rm.hydrateRecords {
608
+
_, rec, err := r.GetRecord(ctx, op.Path)
609
+
if err != nil {
610
+
return fmt.Errorf("reading changed record from car slice: %w", err)
611
+
}
612
+
613
+
rop.Record = rec
614
+
}
615
+
616
+
evtops = append(evtops, rop)
617
+
case EvtKindDeleteRecord:
618
+
evtops = append(evtops, RepoOp{
619
+
Kind: EvtKindDeleteRecord,
620
+
Collection: parts[0],
621
+
Rkey: parts[1],
622
+
})
623
+
default:
624
+
return fmt.Errorf("unrecognized external user event kind: %q", op.Action)
625
+
}
626
+
}
627
+
628
+
if rm.events != nil {
629
+
rm.events(ctx, &RepoEvent{
630
+
User: uid,
631
+
//OldRoot: prev,
632
+
NewRoot: root,
633
+
Rev: nrev,
634
+
Since: since,
635
+
Ops: evtops,
636
+
RepoSlice: carslice,
637
+
PDS: pdsid,
638
+
})
639
+
}
640
+
641
+
return nil
642
+
}
643
+
644
+
func (rm *RepoManager) handleExternalUserEventArchive(ctx context.Context, pdsid uint, uid models.Uid, did string, since *string, nrev string, carslice []byte, ops []*atproto.SyncSubscribeRepos_RepoOp) error {
533
645
ctx, span := otel.Tracer("repoman").Start(ctx, "HandleExternalUserEvent")
534
646
defer span.End()
535
647
+38
-8
testing/integ_test.go
+38
-8
testing/integ_test.go
···
20
20
)
21
21
22
22
func TestRelayBasic(t *testing.T) {
23
+
t.Helper()
24
+
testRelayBasic(t, true)
25
+
}
26
+
27
+
func TestRelayBasicNonArchive(t *testing.T) {
28
+
t.Helper()
29
+
testRelayBasic(t, false)
30
+
}
31
+
32
+
func testRelayBasic(t *testing.T, archive bool) {
23
33
if testing.Short() {
24
34
t.Skip("skipping Relay test in 'short' test mode")
25
35
}
···
28
38
p1 := MustSetupPDS(t, ".tpds", didr)
29
39
p1.Run(t)
30
40
31
-
b1 := MustSetupRelay(t, didr)
41
+
b1 := MustSetupRelay(t, didr, archive)
32
42
b1.Run(t)
33
43
34
44
b1.tr.TrialHosts = []string{p1.RawHost()}
···
111
121
}
112
122
113
123
func TestRelayMultiPDS(t *testing.T) {
124
+
t.Helper()
125
+
testRelayMultiPDS(t, true)
126
+
}
127
+
128
+
func TestRelayMultiPDSNonArchive(t *testing.T) {
129
+
t.Helper()
130
+
testRelayMultiPDS(t, false)
131
+
}
132
+
133
+
func testRelayMultiPDS(t *testing.T, archive bool) {
114
134
if testing.Short() {
115
135
t.Skip("skipping Relay test in 'short' test mode")
116
136
}
···
125
145
p2 := MustSetupPDS(t, ".pdsdos", didr)
126
146
p2.Run(t)
127
147
128
-
b1 := MustSetupRelay(t, didr)
148
+
b1 := MustSetupRelay(t, didr, archive)
129
149
b1.Run(t)
130
150
131
151
b1.tr.TrialHosts = []string{p1.RawHost(), p2.RawHost()}
···
193
213
p2 := MustSetupPDS(t, ".pdsdos", didr)
194
214
p2.Run(t)
195
215
196
-
b1 := MustSetupRelay(t, didr)
216
+
b1 := MustSetupRelay(t, didr, true)
197
217
b1.Run(t)
198
218
199
219
b1.tr.TrialHosts = []string{p1.RawHost(), p2.RawHost()}
···
251
271
p1 := MustSetupPDS(t, ".pdsuno", didr)
252
272
p1.Run(t)
253
273
254
-
b1 := MustSetupRelay(t, didr)
274
+
b1 := MustSetupRelay(t, didr, true)
255
275
b1.Run(t)
256
276
257
277
b1.tr.TrialHosts = []string{p1.RawHost()}
···
288
308
p1 := MustSetupPDS(t, ".pdsuno", didr)
289
309
p1.Run(t)
290
310
291
-
b1 := MustSetupRelay(t, didr)
311
+
b1 := MustSetupRelay(t, didr, true)
292
312
b1.Run(t)
293
313
294
314
b1.tr.TrialHosts = []string{p1.RawHost()}
···
386
406
}
387
407
388
408
func TestRelayTakedown(t *testing.T) {
409
+
testRelayTakedown(t, true)
410
+
}
411
+
412
+
func TestRelayTakedownNonArchive(t *testing.T) {
413
+
testRelayTakedown(t, false)
414
+
}
415
+
416
+
func testRelayTakedown(t *testing.T, archive bool) {
389
417
if testing.Short() {
390
418
t.Skip("skipping Relay test in 'short' test mode")
391
419
}
···
396
424
p1 := MustSetupPDS(t, ".tpds", didr)
397
425
p1.Run(t)
398
426
399
-
b1 := MustSetupRelay(t, didr)
427
+
b1 := MustSetupRelay(t, didr, true)
400
428
b1.Run(t)
401
429
402
430
b1.tr.TrialHosts = []string{p1.RawHost()}
···
475
503
}
476
504
didr := TestPLC(t)
477
505
478
-
b1 := MustSetupRelay(t, didr)
506
+
b1 := MustSetupRelay(t, didr, true)
479
507
b1.Run(t)
480
508
481
509
b1.BanDomain(t, "foo.com")
···
518
546
p1 := MustSetupPDS(t, ".tpds", didr)
519
547
p1.Run(t)
520
548
521
-
b1 := MustSetupRelay(t, didr)
549
+
b1 := MustSetupRelay(t, didr, true)
522
550
b1.Run(t)
523
551
524
552
b1.tr.TrialHosts = []string{p1.RawHost()}
···
536
564
e1 := evts.Next()
537
565
assert.NotNil(e1.RepoCommit)
538
566
assert.Equal(e1.RepoCommit.Repo, bob.DID())
567
+
fmt.Println(e1.RepoCommit.Ops[0])
539
568
540
569
ctx := context.TODO()
541
570
rm := p1.server.Repoman()
···
544
573
}
545
574
546
575
e2 := evts.Next()
576
+
//fmt.Println(e2.RepoCommit.Ops[0])
547
577
assert.Equal(len(e2.RepoCommit.Ops), 0)
548
578
assert.Equal(e2.RepoCommit.Repo, bob.DID())
549
579
}
+16
-6
testing/utils.go
+16
-6
testing/utils.go
···
518
518
return t.listener.Addr().String()
519
519
}
520
520
521
-
func MustSetupRelay(t *testing.T, didr plc.PLCClient) *TestRelay {
521
+
func MustSetupRelay(t *testing.T, didr plc.PLCClient, archive bool) *TestRelay {
522
522
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
523
523
defer cancel()
524
-
tbgs, err := SetupRelay(ctx, didr)
524
+
tbgs, err := SetupRelay(ctx, didr, archive)
525
525
if err != nil {
526
526
t.Fatal(err)
527
527
}
···
529
529
return tbgs
530
530
}
531
531
532
-
func SetupRelay(ctx context.Context, didr plc.PLCClient) (*TestRelay, error) {
532
+
func SetupRelay(ctx context.Context, didr plc.PLCClient, archive bool) (*TestRelay, error) {
533
533
dir, err := os.MkdirTemp("", "integtest")
534
534
if err != nil {
535
535
return nil, err
···
550
550
return nil, err
551
551
}
552
552
553
-
cs, err := carstore.NewCarStore(cardb, []string{cspath})
554
-
if err != nil {
555
-
return nil, err
553
+
var cs carstore.CarStore
554
+
if archive {
555
+
arccs, err := carstore.NewCarStore(cardb, []string{cspath})
556
+
if err != nil {
557
+
return nil, err
558
+
}
559
+
cs = arccs
560
+
} else {
561
+
nacs, err := carstore.NewNonArchivalCarstore(cardb)
562
+
if err != nil {
563
+
return nil, err
564
+
}
565
+
cs = nacs
556
566
}
557
567
558
568
//kmgr := indexer.NewKeyManager(didr, nil)