this repo has no description
disk based event persistence
+761
-13
+2
-2
events/dbpersist.go
+2
-2
events/dbpersist.go
···
124
124
p.lk.Unlock()
125
125
126
126
if needsFlush {
127
-
if err := p.FlushBatch(context.Background()); err != nil {
127
+
if err := p.Flush(context.Background()); err != nil {
128
128
log.Errorf("failed to flush batch: %s", err)
129
129
}
130
130
}
···
135
135
p.broadcast = brc
136
136
}
137
137
138
-
func (p *DbPersistence) FlushBatch(ctx context.Context) error {
138
+
func (p *DbPersistence) Flush(ctx context.Context) error {
139
139
p.lk.Lock()
140
140
defer p.lk.Unlock()
141
141
return p.flushBatchLocked(ctx)
+3
-3
events/dbpersist_test.go
+3
-3
events/dbpersist_test.go
···
129
129
expectedEvtCount := b.N * numRoutines
130
130
131
131
// Flush manually
132
-
err = dbp.FlushBatch(ctx)
132
+
err = dbp.Flush(ctx)
133
133
if err != nil {
134
134
b.Fatal(err)
135
135
}
···
249
249
expectedEvtCount := n * numRoutines
250
250
251
251
// Flush manually
252
-
err = dbp.FlushBatch(ctx)
252
+
err = dbp.Flush(ctx)
253
253
if err != nil {
254
254
b.Fatal(err)
255
255
}
···
306
306
return nil, nil, nil, "", err
307
307
}
308
308
309
-
return maindb, cardb, cs, "", nil
309
+
return maindb, cardb, cs, dir, nil
310
310
}
+405
-8
events/diskpersist.go
+405
-8
events/diskpersist.go
···
1
1
package events
2
2
3
+
import (
4
+
"bufio"
5
+
"bytes"
6
+
"context"
7
+
"encoding/binary"
8
+
"errors"
9
+
"fmt"
10
+
"io"
11
+
"os"
12
+
"path/filepath"
13
+
"sync"
14
+
15
+
"github.com/bluesky-social/indigo/api/atproto"
16
+
"github.com/bluesky-social/indigo/models"
17
+
lru "github.com/hashicorp/golang-lru"
18
+
"gorm.io/gorm"
19
+
)
20
+
3
21
type DiskPersistence struct {
4
-
dir string
22
+
primaryDir string
23
+
archiveDir string
24
+
eventsPerFile int64
25
+
26
+
meta *gorm.DB
27
+
28
+
broadcast func(*XRPCStreamEvent)
29
+
30
+
logfi *os.File
31
+
32
+
curSeq int64
33
+
seqLk sync.Mutex
34
+
35
+
uidCache *lru.ARCCache
36
+
didCache *lru.ARCCache
37
+
38
+
buffer *bytes.Buffer
39
+
scratch []byte
40
+
}
41
+
42
+
var _ (EventPersistence) = (*DiskPersistence)(nil)
43
+
44
+
type DiskPersistOptions struct {
45
+
UIDCacheSize int
46
+
DIDCacheSize int
47
+
EventsPerFile int64
48
+
}
49
+
50
+
func DefaultDiskPersistOptions() *DiskPersistOptions {
51
+
return &DiskPersistOptions{
52
+
EventsPerFile: 10000,
53
+
UIDCacheSize: 100000,
54
+
DIDCacheSize: 100000,
55
+
}
56
+
}
57
+
58
+
func NewDiskPersistence(primaryDir, archiveDir string, db *gorm.DB, opts *DiskPersistOptions) (*DiskPersistence, error) {
59
+
if opts == nil {
60
+
opts = DefaultDiskPersistOptions()
61
+
}
62
+
63
+
uidCache, err := lru.NewARC(opts.UIDCacheSize)
64
+
if err != nil {
65
+
return nil, fmt.Errorf("failed to create uid cache: %w", err)
66
+
}
67
+
68
+
didCache, err := lru.NewARC(opts.DIDCacheSize)
69
+
if err != nil {
70
+
return nil, fmt.Errorf("failed to create did cache: %w", err)
71
+
}
72
+
73
+
db.AutoMigrate(&LogFileRef{})
74
+
75
+
dp := &DiskPersistence{
76
+
meta: db,
77
+
primaryDir: primaryDir,
78
+
archiveDir: archiveDir,
79
+
buffer: new(bytes.Buffer),
80
+
uidCache: uidCache,
81
+
didCache: didCache,
82
+
eventsPerFile: opts.EventsPerFile,
83
+
scratch: make([]byte, headerSize),
84
+
}
85
+
86
+
if err := dp.resumeLog(); err != nil {
87
+
return nil, err
88
+
}
89
+
90
+
return dp, nil
91
+
}
92
+
93
+
type LogFileRef struct {
94
+
gorm.Model
95
+
Path string
96
+
Archived bool
97
+
SeqStart int64
98
+
}
99
+
100
+
func (dp *DiskPersistence) resumeLog() error {
101
+
var lfr LogFileRef
102
+
if err := dp.meta.Order("seq_start desc").Limit(1).Find(&lfr).Error; err != nil {
103
+
return err
104
+
}
105
+
106
+
if lfr.ID == 0 {
107
+
// no files, start anew!
108
+
return dp.initLogFile()
109
+
}
110
+
111
+
fi, err := os.Open(filepath.Join(dp.primaryDir, lfr.Path))
112
+
if err != nil {
113
+
return err
114
+
}
115
+
116
+
seq, err := scanForLastSeq(fi, -1)
117
+
if err != nil {
118
+
return fmt.Errorf("failed to scan log file for last seqno: %w", err)
119
+
}
120
+
121
+
dp.curSeq = seq
122
+
dp.logfi = fi
123
+
124
+
return nil
125
+
}
126
+
127
+
func (dp *DiskPersistence) initLogFile() error {
128
+
if err := os.MkdirAll(dp.primaryDir, 0775); err != nil {
129
+
return err
130
+
}
131
+
132
+
p := filepath.Join(dp.primaryDir, "evts-0")
133
+
fi, err := os.Create(p)
134
+
if err != nil {
135
+
return err
136
+
}
137
+
138
+
if err := dp.meta.Create(&LogFileRef{
139
+
Path: "evts-0",
140
+
SeqStart: 0,
141
+
}).Error; err != nil {
142
+
return err
143
+
}
144
+
145
+
dp.logfi = fi
146
+
dp.curSeq = 1
147
+
return nil
148
+
}
149
+
150
+
func (dp *DiskPersistence) swapLog(ctx context.Context) error {
151
+
if err := dp.logfi.Close(); err != nil {
152
+
return fmt.Errorf("failed to close current log file: %w", err)
153
+
}
154
+
155
+
fname := fmt.Sprintf("evts-%d", dp.curSeq)
156
+
nextp := filepath.Join(dp.primaryDir, fname)
157
+
158
+
fi, err := os.Create(nextp)
159
+
if err != nil {
160
+
return err
161
+
}
162
+
163
+
if err := dp.meta.Create(&LogFileRef{
164
+
Path: fname,
165
+
SeqStart: dp.curSeq,
166
+
}).Error; err != nil {
167
+
return err
168
+
}
169
+
170
+
dp.logfi = fi
171
+
return nil
172
+
}
173
+
174
+
func scanForLastSeq(fi *os.File, end int64) (int64, error) {
175
+
scratch := make([]byte, headerSize)
176
+
177
+
var lastSeq int64 = -1
178
+
var offset int64
179
+
for {
180
+
eh, err := readHeader(fi, scratch)
181
+
if err != nil {
182
+
if err == io.EOF {
183
+
return lastSeq, nil
184
+
}
185
+
return 0, err
186
+
}
187
+
188
+
if end > 0 && eh.Seq >= end {
189
+
// return to beginning of offset
190
+
n, err := fi.Seek(offset, io.SeekStart)
191
+
if err != nil {
192
+
return 0, err
193
+
}
194
+
195
+
if n != offset {
196
+
return 0, fmt.Errorf("rewind seek failed")
197
+
}
198
+
199
+
return eh.Seq, nil
200
+
}
201
+
202
+
lastSeq = eh.Seq
203
+
204
+
noff, err := fi.Seek(eh.Len, io.SeekCurrent)
205
+
if err != nil {
206
+
return 0, err
207
+
}
208
+
209
+
if noff != offset+headerSize+eh.Len {
210
+
// TODO: must recover from this
211
+
return 0, fmt.Errorf("did not seek to next event properly")
212
+
}
213
+
214
+
offset = noff
215
+
}
216
+
}
217
+
218
+
const (
219
+
evtKindCommit = 1
220
+
evtKindHandle = 2
221
+
)
222
+
223
+
func (p *DiskPersistence) Persist(ctx context.Context, e *XRPCStreamEvent) error {
224
+
p.seqLk.Lock()
225
+
defer p.seqLk.Unlock()
226
+
227
+
p.buffer.Truncate(0)
228
+
seq := p.curSeq
229
+
p.curSeq++
230
+
231
+
var evtKind uint32
232
+
switch {
233
+
case e.RepoCommit != nil:
234
+
evtKind = evtKindCommit
235
+
e.RepoCommit.Seq = seq
236
+
if err := e.RepoCommit.MarshalCBOR(p.buffer); err != nil {
237
+
return fmt.Errorf("failed to marshal: %w", err)
238
+
}
239
+
case e.RepoHandle != nil:
240
+
evtKind = evtKindHandle
241
+
e.RepoHandle.Seq = seq
242
+
if err := e.RepoHandle.MarshalCBOR(p.buffer); err != nil {
243
+
return fmt.Errorf("failed to marshal: %w", err)
244
+
}
245
+
default:
246
+
return nil
247
+
// only those two get peristed right now
248
+
}
249
+
250
+
if err := p.writeHeader(ctx, 0, evtKind, uint32(p.buffer.Len()), seq); err != nil {
251
+
return fmt.Errorf("failed to write header: %w", err)
252
+
}
253
+
254
+
if _, err := io.Copy(p.logfi, p.buffer); err != nil {
255
+
return err
256
+
}
257
+
258
+
p.broadcast(e)
259
+
260
+
if seq%p.eventsPerFile == 0 {
261
+
// time to roll the log file
262
+
if err := p.swapLog(ctx); err != nil {
263
+
return err
264
+
}
265
+
}
266
+
267
+
return nil
268
+
}
269
+
270
+
type evtHeader struct {
271
+
Flags uint32
272
+
Kind uint32
273
+
Seq int64
274
+
Len int64
5
275
}
6
276
7
-
func NewDiskPersistence(dir string) (*DiskPersistence, error) {
8
-
return &DiskPersistence{
9
-
dir: dir,
277
+
const headerSize = 4 + 4 + 4 + 8
278
+
279
+
func readHeader(r io.Reader, scratch []byte) (*evtHeader, error) {
280
+
if len(scratch) < headerSize {
281
+
return nil, fmt.Errorf("must pass scratch buffer of at least %d bytes", headerSize)
282
+
}
283
+
284
+
scratch = scratch[:headerSize]
285
+
_, err := io.ReadFull(r, scratch)
286
+
if err != nil {
287
+
return nil, fmt.Errorf("reading header: %w", err)
288
+
}
289
+
290
+
flags := binary.LittleEndian.Uint32(scratch[:4])
291
+
kind := binary.LittleEndian.Uint32(scratch[4:8])
292
+
l := binary.LittleEndian.Uint32(scratch[8:12])
293
+
seq := binary.LittleEndian.Uint64(scratch[12:20])
294
+
295
+
return &evtHeader{
296
+
Flags: flags,
297
+
Kind: kind,
298
+
Len: int64(l),
299
+
Seq: int64(seq),
10
300
}, nil
11
301
}
12
302
13
-
func (p *DiskPersistence) Persist(e *XRPCStreamEvent) error {
14
-
panic("nyi")
303
+
func (p *DiskPersistence) writeHeader(ctx context.Context, flags uint32, kind uint32, l uint32, seq int64) error {
304
+
binary.LittleEndian.PutUint32(p.scratch, flags)
305
+
binary.LittleEndian.PutUint32(p.scratch[4:], kind)
306
+
binary.LittleEndian.PutUint32(p.scratch[8:], l)
307
+
binary.LittleEndian.PutUint64(p.scratch[12:], uint64(seq))
308
+
309
+
nw, err := p.logfi.Write(p.scratch)
310
+
if err != nil {
311
+
return err
312
+
}
313
+
314
+
if nw != headerSize {
315
+
return fmt.Errorf("only wrote %d bytes for header", nw)
316
+
}
317
+
318
+
return nil
15
319
}
16
320
17
-
func (p *DiskPersistence) Playback(since int64, cb func(*XRPCStreamEvent) error) error {
18
-
panic("nyi")
321
+
func (p *DiskPersistence) uidForDid(ctx context.Context, did string) (models.Uid, error) {
322
+
if uid, ok := p.didCache.Get(did); ok {
323
+
return uid.(models.Uid), nil
324
+
}
325
+
326
+
var u models.ActorInfo
327
+
if err := p.meta.First(&u, "did = ?", did).Error; err != nil {
328
+
return 0, err
329
+
}
330
+
331
+
p.didCache.Add(did, u.Uid)
332
+
333
+
return u.Uid, nil
334
+
}
335
+
336
+
func (p *DiskPersistence) Playback(ctx context.Context, since int64, cb func(*XRPCStreamEvent) error) error {
337
+
var logs []LogFileRef
338
+
if err := p.meta.Debug().Order("seq_start asc").Find(&logs, "seq_start >= ?", since%p.eventsPerFile).Error; err != nil {
339
+
return err
340
+
}
341
+
342
+
for _, lf := range logs {
343
+
if err := p.readEventsFrom(ctx, since, filepath.Join(p.primaryDir, lf.Path), cb); err != nil {
344
+
return err
345
+
}
346
+
since = 0
347
+
}
348
+
349
+
return nil
350
+
}
351
+
352
+
func (p *DiskPersistence) readEventsFrom(ctx context.Context, since int64, fn string, cb func(*XRPCStreamEvent) error) error {
353
+
fi, err := os.OpenFile(fn, os.O_RDONLY, 0)
354
+
if err != nil {
355
+
return err
356
+
}
357
+
358
+
if since != 0 {
359
+
_, err := scanForLastSeq(fi, since)
360
+
if err != nil {
361
+
return err
362
+
}
363
+
}
364
+
365
+
bufr := bufio.NewReader(fi)
366
+
367
+
scratch := make([]byte, headerSize)
368
+
for {
369
+
h, err := readHeader(bufr, scratch)
370
+
if err != nil {
371
+
if errors.Is(err, io.EOF) {
372
+
return nil
373
+
}
374
+
375
+
return err
376
+
}
377
+
378
+
switch h.Kind {
379
+
case evtKindCommit:
380
+
var evt atproto.SyncSubscribeRepos_Commit
381
+
if err := evt.UnmarshalCBOR(io.LimitReader(bufr, h.Len)); err != nil {
382
+
return err
383
+
}
384
+
if err := cb(&XRPCStreamEvent{RepoCommit: &evt}); err != nil {
385
+
return err
386
+
}
387
+
case evtKindHandle:
388
+
var evt atproto.SyncSubscribeRepos_Handle
389
+
if err := evt.UnmarshalCBOR(io.LimitReader(bufr, h.Len)); err != nil {
390
+
return err
391
+
}
392
+
if err := cb(&XRPCStreamEvent{RepoHandle: &evt}); err != nil {
393
+
return err
394
+
}
395
+
default:
396
+
log.Warnw("unrecognized event kind coming from log file", "seq", h.Seq, "kind", h.Kind)
397
+
return fmt.Errorf("halting on unrecognized event kind")
398
+
}
399
+
}
400
+
}
401
+
402
+
func (p *DiskPersistence) TakeDownRepo(ctx context.Context, usr models.Uid) error {
403
+
panic("no")
404
+
}
405
+
406
+
func (p *DiskPersistence) RebaseRepoEvents(ctx context.Context, usr models.Uid) error {
407
+
panic("no")
408
+
}
409
+
410
+
func (p *DiskPersistence) Flush(ctx context.Context) error {
411
+
return nil
412
+
}
413
+
414
+
func (p *DiskPersistence) SetEventBroadcaster(f func(*XRPCStreamEvent)) {
415
+
p.broadcast = f
19
416
}
+334
events/diskpersist_test.go
+334
events/diskpersist_test.go
···
1
+
package events_test
2
+
3
+
import (
4
+
"context"
5
+
"os"
6
+
"path/filepath"
7
+
"reflect"
8
+
"sync"
9
+
"testing"
10
+
"time"
11
+
12
+
atproto "github.com/bluesky-social/indigo/api/atproto"
13
+
"github.com/bluesky-social/indigo/api/bsky"
14
+
"github.com/bluesky-social/indigo/carstore"
15
+
"github.com/bluesky-social/indigo/events"
16
+
lexutil "github.com/bluesky-social/indigo/lex/util"
17
+
"github.com/bluesky-social/indigo/models"
18
+
"github.com/bluesky-social/indigo/pds"
19
+
"github.com/bluesky-social/indigo/repomgr"
20
+
"github.com/bluesky-social/indigo/util"
21
+
"gorm.io/gorm"
22
+
)
23
+
24
+
func TestDiskPersist(t *testing.T) {
25
+
ctx := context.Background()
26
+
27
+
db, _, cs, tempPath, err := setupDBs(t)
28
+
if err != nil {
29
+
t.Fatal(err)
30
+
}
31
+
32
+
db.AutoMigrate(&pds.User{})
33
+
db.AutoMigrate(&pds.Peering{})
34
+
db.AutoMigrate(&models.ActorInfo{})
35
+
36
+
db.Create(&models.ActorInfo{
37
+
Uid: 1,
38
+
Did: "did:example:123",
39
+
})
40
+
41
+
mgr := repomgr.NewRepoManager(repomgr.NewDbHeadStore(db), cs, &util.FakeKeyManager{})
42
+
43
+
err = mgr.InitNewActor(ctx, 1, "alice", "did:example:123", "Alice", "", "")
44
+
if err != nil {
45
+
t.Fatal(err)
46
+
}
47
+
48
+
_, cid, err := mgr.CreateRecord(ctx, 1, "app.bsky.feed.post", &bsky.FeedPost{
49
+
Text: "hello world",
50
+
CreatedAt: time.Now().Format(util.ISO8601),
51
+
})
52
+
if err != nil {
53
+
t.Fatal(err)
54
+
}
55
+
56
+
defer os.RemoveAll(tempPath)
57
+
58
+
// Initialize a DBPersister
59
+
60
+
dp, err := events.NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &events.DiskPersistOptions{
61
+
EventsPerFile: 10,
62
+
UIDCacheSize: 100000,
63
+
DIDCacheSize: 100000,
64
+
})
65
+
if err != nil {
66
+
t.Fatal(err)
67
+
}
68
+
69
+
// Create a bunch of events
70
+
evtman := events.NewEventManager(dp)
71
+
72
+
userRepoHead, err := mgr.GetRepoRoot(ctx, 1)
73
+
if err != nil {
74
+
t.Fatal(err)
75
+
}
76
+
77
+
n := 100
78
+
inEvts := make([]*events.XRPCStreamEvent, n)
79
+
for i := 0; i < n; i++ {
80
+
cidLink := lexutil.LexLink(cid)
81
+
headLink := lexutil.LexLink(userRepoHead)
82
+
inEvts[i] = &events.XRPCStreamEvent{
83
+
RepoCommit: &atproto.SyncSubscribeRepos_Commit{
84
+
Repo: "did:example:123",
85
+
Commit: headLink,
86
+
Ops: []*atproto.SyncSubscribeRepos_RepoOp{
87
+
{
88
+
Action: "add",
89
+
Cid: &cidLink,
90
+
Path: "path1",
91
+
},
92
+
},
93
+
Time: time.Now().Format(util.ISO8601),
94
+
},
95
+
}
96
+
}
97
+
98
+
// Add events in parallel
99
+
for i := 0; i < n; i++ {
100
+
err = evtman.AddEvent(ctx, inEvts[i])
101
+
if err != nil {
102
+
t.Fatal(err)
103
+
}
104
+
}
105
+
106
+
outEvtCount := 0
107
+
expectedEvtCount := n
108
+
109
+
dp.Playback(ctx, 0, func(evt *events.XRPCStreamEvent) error {
110
+
outEvtCount++
111
+
return nil
112
+
})
113
+
114
+
if outEvtCount != expectedEvtCount {
115
+
t.Fatalf("expected %d events, got %d", expectedEvtCount, outEvtCount)
116
+
}
117
+
}
118
+
119
+
func BenchmarkDiskPersist(b *testing.B) {
120
+
db, _, cs, tempPath, err := setupDBs(b)
121
+
if err != nil {
122
+
b.Fatal(err)
123
+
}
124
+
125
+
defer os.RemoveAll(tempPath)
126
+
127
+
// Initialize a DBPersister
128
+
129
+
dp, err := events.NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &events.DiskPersistOptions{
130
+
EventsPerFile: 5000,
131
+
UIDCacheSize: 100000,
132
+
DIDCacheSize: 100000,
133
+
})
134
+
if err != nil {
135
+
b.Fatal(err)
136
+
}
137
+
138
+
runPersisterBenchmark(b, cs, db, dp)
139
+
}
140
+
141
+
func runPersisterBenchmark(b *testing.B, cs *carstore.CarStore, db *gorm.DB, p events.EventPersistence) {
142
+
ctx := context.Background()
143
+
144
+
db.AutoMigrate(&pds.User{})
145
+
db.AutoMigrate(&pds.Peering{})
146
+
db.AutoMigrate(&models.ActorInfo{})
147
+
148
+
db.Create(&models.ActorInfo{
149
+
Uid: 1,
150
+
Did: "did:example:123",
151
+
})
152
+
153
+
mgr := repomgr.NewRepoManager(repomgr.NewDbHeadStore(db), cs, &util.FakeKeyManager{})
154
+
155
+
err := mgr.InitNewActor(ctx, 1, "alice", "did:example:123", "Alice", "", "")
156
+
if err != nil {
157
+
b.Fatal(err)
158
+
}
159
+
160
+
_, cid, err := mgr.CreateRecord(ctx, 1, "app.bsky.feed.post", &bsky.FeedPost{
161
+
Text: "hello world",
162
+
CreatedAt: time.Now().Format(util.ISO8601),
163
+
})
164
+
if err != nil {
165
+
b.Fatal(err)
166
+
}
167
+
168
+
// Create a bunch of events
169
+
evtman := events.NewEventManager(p)
170
+
171
+
userRepoHead, err := mgr.GetRepoRoot(ctx, 1)
172
+
if err != nil {
173
+
b.Fatal(err)
174
+
}
175
+
176
+
inEvts := make([]*events.XRPCStreamEvent, b.N)
177
+
for i := 0; i < b.N; i++ {
178
+
cidLink := lexutil.LexLink(cid)
179
+
headLink := lexutil.LexLink(userRepoHead)
180
+
inEvts[i] = &events.XRPCStreamEvent{
181
+
RepoCommit: &atproto.SyncSubscribeRepos_Commit{
182
+
Repo: "did:example:123",
183
+
Commit: headLink,
184
+
Ops: []*atproto.SyncSubscribeRepos_RepoOp{
185
+
{
186
+
Action: "add",
187
+
Cid: &cidLink,
188
+
Path: "path1",
189
+
},
190
+
},
191
+
Time: time.Now().Format(util.ISO8601),
192
+
},
193
+
}
194
+
}
195
+
196
+
numRoutines := 5
197
+
wg := sync.WaitGroup{}
198
+
199
+
b.ResetTimer()
200
+
201
+
errChan := make(chan error, numRoutines)
202
+
203
+
// Add events in parallel
204
+
for i := 0; i < numRoutines; i++ {
205
+
wg.Add(1)
206
+
go func() {
207
+
defer wg.Done()
208
+
for i := 0; i < b.N; i++ {
209
+
err = evtman.AddEvent(ctx, inEvts[i])
210
+
if err != nil {
211
+
errChan <- err
212
+
}
213
+
}
214
+
}()
215
+
}
216
+
217
+
wg.Wait()
218
+
close(errChan)
219
+
220
+
// Check for errors
221
+
for err := range errChan {
222
+
if err != nil {
223
+
b.Fatal(err)
224
+
}
225
+
}
226
+
227
+
// Flush manually
228
+
if err := p.Flush(ctx); err != nil {
229
+
b.Fatal(err)
230
+
}
231
+
232
+
}
233
+
234
+
func TestDiskPersister(t *testing.T) {
235
+
db, _, cs, tempPath, err := setupDBs(t)
236
+
if err != nil {
237
+
t.Fatal(err)
238
+
}
239
+
240
+
defer os.RemoveAll(tempPath)
241
+
242
+
// Initialize a DBPersister
243
+
244
+
dp, err := events.NewDiskPersistence(filepath.Join(tempPath, "diskPrimary"), filepath.Join(tempPath, "diskArchive"), db, &events.DiskPersistOptions{
245
+
EventsPerFile: 20,
246
+
UIDCacheSize: 100000,
247
+
DIDCacheSize: 100000,
248
+
})
249
+
if err != nil {
250
+
t.Fatal(err)
251
+
}
252
+
253
+
runEventManagerTest(t, cs, db, dp)
254
+
}
255
+
256
+
func runEventManagerTest(t *testing.T, cs *carstore.CarStore, db *gorm.DB, p events.EventPersistence) {
257
+
ctx := context.Background()
258
+
259
+
db.AutoMigrate(&pds.User{})
260
+
db.AutoMigrate(&pds.Peering{})
261
+
db.AutoMigrate(&models.ActorInfo{})
262
+
263
+
db.Create(&models.ActorInfo{
264
+
Uid: 1,
265
+
Did: "did:example:123",
266
+
})
267
+
268
+
mgr := repomgr.NewRepoManager(repomgr.NewDbHeadStore(db), cs, &util.FakeKeyManager{})
269
+
270
+
err := mgr.InitNewActor(ctx, 1, "alice", "did:example:123", "Alice", "", "")
271
+
if err != nil {
272
+
t.Fatal(err)
273
+
}
274
+
275
+
_, cid, err := mgr.CreateRecord(ctx, 1, "app.bsky.feed.post", &bsky.FeedPost{
276
+
Text: "hello world",
277
+
CreatedAt: time.Now().Format(util.ISO8601),
278
+
})
279
+
if err != nil {
280
+
t.Fatal(err)
281
+
}
282
+
283
+
evtman := events.NewEventManager(p)
284
+
285
+
userRepoHead, err := mgr.GetRepoRoot(ctx, 1)
286
+
if err != nil {
287
+
t.Fatal(err)
288
+
}
289
+
290
+
testSize := 100 // you can adjust this number as needed
291
+
inEvts := make([]*events.XRPCStreamEvent, testSize)
292
+
for i := 0; i < testSize; i++ {
293
+
cidLink := lexutil.LexLink(cid)
294
+
headLink := lexutil.LexLink(userRepoHead)
295
+
inEvts[i] = &events.XRPCStreamEvent{
296
+
RepoCommit: &atproto.SyncSubscribeRepos_Commit{
297
+
Repo: "did:example:123",
298
+
Commit: headLink,
299
+
Ops: []*atproto.SyncSubscribeRepos_RepoOp{
300
+
{
301
+
Action: "add",
302
+
Cid: &cidLink,
303
+
Path: "path1",
304
+
},
305
+
},
306
+
Time: time.Now().Format(util.ISO8601),
307
+
},
308
+
}
309
+
310
+
err = evtman.AddEvent(ctx, inEvts[i])
311
+
if err != nil {
312
+
t.Fatal(err)
313
+
}
314
+
}
315
+
316
+
// Flush manually
317
+
if err := p.Flush(ctx); err != nil {
318
+
t.Fatal(err)
319
+
}
320
+
321
+
outEvtCount := 0
322
+
p.Playback(ctx, 0, func(evt *events.XRPCStreamEvent) error {
323
+
// Check that the contents of the output events match the input events
324
+
if !reflect.DeepEqual(inEvts[outEvtCount], evt) {
325
+
t.Fatalf("Event content mismatch: expected %+v, got %+v", inEvts[outEvtCount], evt)
326
+
}
327
+
outEvtCount++
328
+
return nil
329
+
})
330
+
331
+
if outEvtCount != testSize {
332
+
t.Fatalf("expected %d events, got %d", testSize, outEvtCount)
333
+
}
334
+
}
+5
events/persist.go
+5
events/persist.go
···
14
14
Playback(ctx context.Context, since int64, cb func(*XRPCStreamEvent) error) error
15
15
TakeDownRepo(ctx context.Context, usr models.Uid) error
16
16
RebaseRepoEvents(ctx context.Context, usr models.Uid) error
17
+
Flush(context.Context) error
17
18
18
19
SetEventBroadcaster(func(*XRPCStreamEvent))
19
20
}
···
83
84
84
85
func (mp *MemPersister) RebaseRepoEvents(ctx context.Context, usr models.Uid) error {
85
86
return fmt.Errorf("repo rebases not currently supported by memory persister, test usage only")
87
+
}
88
+
89
+
func (mp *MemPersister) Flush(ctx context.Context) error {
90
+
return nil
86
91
}
87
92
88
93
func (mp *MemPersister) SetEventBroadcaster(brc func(*XRPCStreamEvent)) {
+12
repomgr/bench_test.go
+12
repomgr/bench_test.go
···
25
25
t.Fatal(err)
26
26
}
27
27
28
+
if err := db.Exec("PRAGMA synchronous=normal;").Error; err != nil {
29
+
t.Fatal(err)
30
+
}
31
+
32
+
if err := db.Exec("PRAGMA temp_store=memory;").Error; err != nil {
33
+
t.Fatal(err)
34
+
}
35
+
36
+
if err := db.Exec("PRAGMA mmap_size=3000000000;").Error; err != nil {
37
+
t.Fatal(err)
38
+
}
39
+
28
40
return db
29
41
}
30
42