this repo has no description
Fix/events playback (#402)
This is all the refactors and fixes from #388, but without the new
splitter daemon code. should get this merged asap
authored by
Whyrusleeping
and committed by
GitHub
66786961
b761d052
+167
-71
+29
-6
cmd/gosky/debug.go
+29
-6
cmd/gosky/debug.go
···
162
162
}
163
163
164
164
type eventInfo struct {
165
-
LastCid cid.Cid
166
165
LastSeq int64
166
+
LastRev string
167
167
}
168
168
169
169
func cidStr(c *lexutil.LexLink) string {
···
194
194
195
195
h := cctx.String("host")
196
196
197
-
url := fmt.Sprintf("%s/xrpc/com.atproto.sync.subscribeRepos?cursor=%d", h, n-1)
197
+
url := fmt.Sprintf("%s/xrpc/com.atproto.sync.subscribeRepos?cursor=%d", h, n)
198
198
d := websocket.DefaultDialer
199
199
con, _, err := d.Dial(url, http.Header{})
200
200
if err != nil {
···
203
203
204
204
infos := make(map[string]*eventInfo)
205
205
206
+
var lastSeq int64 = -1
206
207
ctx := context.TODO()
207
208
rsc := &events.RepoStreamCallbacks{
208
209
RepoCommit: func(evt *comatproto.SyncSubscribeRepos_Commit) error {
209
210
210
211
fmt.Printf("\rChecking seq: %d ", evt.Seq)
212
+
if lastSeq > 0 && evt.Seq != lastSeq+1 {
213
+
fmt.Println("Gap in sequence numbers: ", lastSeq, evt.Seq)
214
+
}
215
+
lastSeq = evt.Seq
211
216
212
217
if !evt.TooBig {
213
218
r, err := repo.ReadRepoFromCar(ctx, bytes.NewReader(evt.Blocks))
···
237
242
238
243
cur, ok := infos[evt.Repo]
239
244
if ok {
240
-
if cur.LastCid.String() != cidStr(evt.Prev) {
241
-
fmt.Println()
242
-
fmt.Printf("Event at sequence %d, repo=%s had prev=%s head=%s, but last commit we saw was %s (seq=%d)\n", evt.Seq, evt.Repo, cidStr(evt.Prev), evt.Commit.String(), cur.LastCid, cur.LastSeq)
245
+
if evt.Since != nil && cur.LastRev != *evt.Since {
246
+
/*
247
+
fmt.Println()
248
+
fmt.Printf("Event at sequence %d, repo=%s had since=%s, but last rev we saw was %s (seq=%d)\n", evt.Seq, evt.Repo, evt.Since, cur.LastRev, cur.LastSeq)
249
+
*/
243
250
}
244
251
}
245
252
246
253
infos[evt.Repo] = &eventInfo{
247
-
LastCid: cid.Cid(evt.Commit),
248
254
LastSeq: evt.Seq,
255
+
LastRev: evt.Rev,
249
256
}
250
257
258
+
return nil
259
+
},
260
+
RepoHandle: func(evt *comatproto.SyncSubscribeRepos_Handle) error {
261
+
fmt.Printf("\rChecking seq: %d ", evt.Seq)
262
+
if lastSeq > 0 && evt.Seq != lastSeq+1 {
263
+
fmt.Println("Gap in sequence numbers: ", lastSeq, evt.Seq)
264
+
}
265
+
lastSeq = evt.Seq
266
+
return nil
267
+
},
268
+
RepoTombstone: func(evt *comatproto.SyncSubscribeRepos_Tombstone) error {
269
+
fmt.Printf("\rChecking seq: %d ", evt.Seq)
270
+
if lastSeq > 0 && evt.Seq != lastSeq+1 {
271
+
fmt.Println("Gap in sequence numbers: ", lastSeq, evt.Seq)
272
+
}
273
+
lastSeq = evt.Seq
251
274
return nil
252
275
},
253
276
RepoInfo: func(evt *comatproto.SyncSubscribeRepos_Info) error {
-5
events/dbpersist.go
-5
events/dbpersist.go
···
578
578
return nil
579
579
}
580
580
581
-
func (p *DbPersistence) RebaseRepoEvents(ctx context.Context, usr models.Uid) error {
582
-
// a little weird that this is the same action as a takedown
583
-
return p.deleteAllEventsForUser(ctx, usr)
584
-
}
585
-
586
581
func (p *DbPersistence) Shutdown(context.Context) error {
587
582
return nil
588
583
}
+49
-23
events/diskpersist.go
+49
-23
events/diskpersist.go
···
620
620
return err
621
621
}
622
622
623
-
for _, lf := range logs {
624
-
if err := dp.readEventsFrom(ctx, since, filepath.Join(dp.primaryDir, lf.Path), cb); err != nil {
623
+
for i := 0; i < 10; i++ {
624
+
lastSeq, err := dp.PlaybackLogfiles(ctx, since, cb, logs)
625
+
if err != nil {
626
+
return err
627
+
}
628
+
629
+
// No lastSeq implies that we read until the end of known events
630
+
if lastSeq == nil {
631
+
break
632
+
}
633
+
634
+
if err := dp.meta.Debug().Order("seq_start asc").Find(&logs, "seq_start >= ?", *lastSeq).Error; err != nil {
625
635
return err
626
636
}
627
-
since = 0
637
+
since = *lastSeq
628
638
}
629
639
630
640
return nil
631
641
}
632
642
643
+
func (dp *DiskPersistence) PlaybackLogfiles(ctx context.Context, since int64, cb func(*XRPCStreamEvent) error, logFiles []LogFileRef) (*int64, error) {
644
+
for i, lf := range logFiles {
645
+
lastSeq, err := dp.readEventsFrom(ctx, since, filepath.Join(dp.primaryDir, lf.Path), cb)
646
+
if err != nil {
647
+
return nil, err
648
+
}
649
+
since = 0
650
+
if i == len(logFiles)-1 &&
651
+
lastSeq != nil &&
652
+
(*lastSeq-lf.SeqStart) == dp.eventsPerFile-1 {
653
+
// There may be more log files to read since the last one was full
654
+
return lastSeq, nil
655
+
}
656
+
}
657
+
658
+
return nil, nil
659
+
}
660
+
633
661
func postDoNotEmit(flags uint32) bool {
634
662
if flags&(EvtFlagRebased|EvtFlagTakedown) != 0 {
635
663
return true
···
638
666
return false
639
667
}
640
668
641
-
func (dp *DiskPersistence) readEventsFrom(ctx context.Context, since int64, fn string, cb func(*XRPCStreamEvent) error) error {
669
+
func (dp *DiskPersistence) readEventsFrom(ctx context.Context, since int64, fn string, cb func(*XRPCStreamEvent) error) (*int64, error) {
642
670
fi, err := os.OpenFile(fn, os.O_RDONLY, 0)
643
671
if err != nil {
644
-
return err
672
+
return nil, err
645
673
}
646
674
647
675
if since != 0 {
648
676
lastSeq, err := scanForLastSeq(fi, since)
649
677
if err != nil {
650
-
return err
678
+
return nil, err
651
679
}
652
680
if since > lastSeq {
653
681
log.Errorw("playback cursor is greater than last seq of file checked",
···
655
683
"lastSeq", lastSeq,
656
684
"filename", fn,
657
685
)
658
-
return nil
686
+
return nil, nil
659
687
}
660
688
}
661
689
662
690
bufr := bufio.NewReader(fi)
663
691
692
+
lastSeq := int64(0)
693
+
664
694
scratch := make([]byte, headerSize)
665
695
for {
666
696
h, err := readHeader(bufr, scratch)
667
697
if err != nil {
668
698
if errors.Is(err, io.EOF) {
669
-
return nil
699
+
return &lastSeq, nil
670
700
}
671
701
672
-
return err
702
+
return nil, err
673
703
}
704
+
705
+
lastSeq = h.Seq
674
706
675
707
if postDoNotEmit(h.Flags) {
676
708
// event taken down, skip
677
709
_, err := io.CopyN(io.Discard, bufr, h.Len64()) // would be really nice if the buffered reader had a 'skip' method that does a seek under the hood
678
710
if err != nil {
679
-
return fmt.Errorf("failed while skipping event (seq: %d, fn: %q): %w", h.Seq, fn, err)
711
+
return nil, fmt.Errorf("failed while skipping event (seq: %d, fn: %q): %w", h.Seq, fn, err)
680
712
}
681
713
continue
682
714
}
···
685
717
case evtKindCommit:
686
718
var evt atproto.SyncSubscribeRepos_Commit
687
719
if err := evt.UnmarshalCBOR(io.LimitReader(bufr, h.Len64())); err != nil {
688
-
return err
720
+
return nil, err
689
721
}
690
722
evt.Seq = h.Seq
691
723
if err := cb(&XRPCStreamEvent{RepoCommit: &evt}); err != nil {
692
-
return err
724
+
return nil, err
693
725
}
694
726
case evtKindHandle:
695
727
var evt atproto.SyncSubscribeRepos_Handle
696
728
if err := evt.UnmarshalCBOR(io.LimitReader(bufr, h.Len64())); err != nil {
697
-
return err
729
+
return nil, err
698
730
}
699
731
evt.Seq = h.Seq
700
732
if err := cb(&XRPCStreamEvent{RepoHandle: &evt}); err != nil {
701
-
return err
733
+
return nil, err
702
734
}
703
735
case evtKindTombstone:
704
736
var evt atproto.SyncSubscribeRepos_Tombstone
705
737
if err := evt.UnmarshalCBOR(io.LimitReader(bufr, h.Len64())); err != nil {
706
-
return err
738
+
return nil, err
707
739
}
708
740
evt.Seq = h.Seq
709
741
if err := cb(&XRPCStreamEvent{RepoTombstone: &evt}); err != nil {
710
-
return err
742
+
return nil, err
711
743
}
712
744
default:
713
745
log.Warnw("unrecognized event kind coming from log file", "seq", h.Seq, "kind", h.Kind)
714
-
return fmt.Errorf("halting on unrecognized event kind")
746
+
return nil, fmt.Errorf("halting on unrecognized event kind")
715
747
}
716
748
}
717
749
}
···
842
874
return fmt.Errorf("failed to seek: %w", err)
843
875
}
844
876
}
845
-
}
846
-
847
-
func (dp *DiskPersistence) RebaseRepoEvents(ctx context.Context, usr models.Uid) error {
848
-
return dp.forEachShardWithUserEvents(ctx, usr, func(ctx context.Context, fn string) error {
849
-
return dp.mutateUserEventsInLog(ctx, usr, fn, EvtFlagRebased, false)
850
-
})
851
877
}
852
878
853
879
func (dp *DiskPersistence) Flush(ctx context.Context) error {
+89
-28
events/events.go
+89
-28
events/events.go
···
144
144
return nil
145
145
}
146
146
147
-
var ErrPlaybackShutdown = fmt.Errorf("playback shutting down")
147
+
var (
148
+
ErrPlaybackShutdown = fmt.Errorf("playback shutting down")
149
+
ErrCaughtUp = fmt.Errorf("caught up")
150
+
)
148
151
149
152
func (em *EventManager) Subscribe(ctx context.Context, ident string, filter func(*XRPCStreamEvent) bool, since *int64) (<-chan *XRPCStreamEvent, func(), error) {
150
153
if filter == nil {
···
161
164
broadcastCounter: eventsBroadcast.WithLabelValues(ident),
162
165
}
163
166
167
+
cleanup := func() {
168
+
close(done)
169
+
em.rmSubscriber(sub)
170
+
}
171
+
172
+
if since == nil {
173
+
em.addSubscriber(sub)
174
+
return sub.outgoing, cleanup, nil
175
+
}
176
+
177
+
out := make(chan *XRPCStreamEvent, em.bufferSize)
178
+
164
179
go func() {
165
-
if since != nil {
166
-
if err := em.persister.Playback(ctx, *since, func(e *XRPCStreamEvent) error {
167
-
select {
168
-
case <-done:
169
-
return ErrPlaybackShutdown
170
-
case sub.outgoing <- e:
171
-
return nil
180
+
lastSeq := *since
181
+
// run playback to get through *most* of the events, getting our current cursor close to realtime
182
+
if err := em.persister.Playback(ctx, *since, func(e *XRPCStreamEvent) error {
183
+
select {
184
+
case <-done:
185
+
return ErrPlaybackShutdown
186
+
case out <- e:
187
+
seq := sequenceForEvent(e)
188
+
if seq > 0 {
189
+
lastSeq = seq
172
190
}
173
-
}); err != nil {
174
-
if errors.Is(err, ErrPlaybackShutdown) {
175
-
log.Warnf("events playback: %s", err)
176
-
} else {
177
-
log.Errorf("events playback: %s", err)
178
-
}
191
+
return nil
192
+
}
193
+
}); err != nil {
194
+
if errors.Is(err, ErrPlaybackShutdown) {
195
+
log.Warnf("events playback: %s", err)
196
+
} else {
197
+
log.Errorf("events playback: %s", err)
179
198
}
180
-
}
181
199
182
-
// if cancel happens before playback completes
183
-
select {
184
-
case <-done:
200
+
// TODO: send an error frame or something?
201
+
close(out)
185
202
return
186
-
default:
187
203
}
188
204
205
+
// now, start buffering events from the live stream
189
206
em.addSubscriber(sub)
207
+
208
+
first := <-sub.outgoing
209
+
210
+
// run playback again to get us to the events that have started buffering
211
+
if err := em.persister.Playback(ctx, lastSeq, func(e *XRPCStreamEvent) error {
212
+
seq := sequenceForEvent(e)
213
+
if seq > sequenceForEvent(first) {
214
+
return ErrCaughtUp
215
+
}
216
+
217
+
select {
218
+
case <-done:
219
+
return ErrPlaybackShutdown
220
+
case out <- e:
221
+
return nil
222
+
}
223
+
}); err != nil {
224
+
if !errors.Is(err, ErrCaughtUp) {
225
+
log.Errorf("events playback: %s", err)
226
+
227
+
// TODO: send an error frame or something?
228
+
close(out)
229
+
em.rmSubscriber(sub)
230
+
return
231
+
}
232
+
}
233
+
234
+
// now that we are caught up, just copy events from the channel over
235
+
for evt := range sub.outgoing {
236
+
select {
237
+
case out <- evt:
238
+
case <-done:
239
+
em.rmSubscriber(sub)
240
+
return
241
+
}
242
+
}
190
243
}()
191
244
192
-
cleanup := func() {
193
-
close(done)
194
-
em.rmSubscriber(sub)
245
+
return out, cleanup, nil
246
+
}
247
+
248
+
func sequenceForEvent(evt *XRPCStreamEvent) int64 {
249
+
switch {
250
+
case evt.RepoCommit != nil:
251
+
return evt.RepoCommit.Seq
252
+
case evt.RepoHandle != nil:
253
+
return evt.RepoHandle.Seq
254
+
case evt.RepoMigrate != nil:
255
+
return evt.RepoMigrate.Seq
256
+
case evt.RepoTombstone != nil:
257
+
return evt.RepoTombstone.Seq
258
+
case evt.RepoInfo != nil:
259
+
return -1
260
+
default:
261
+
return -1
195
262
}
196
-
197
-
return sub.outgoing, cleanup, nil
198
263
}
199
264
200
265
func (em *EventManager) rmSubscriber(sub *Subscriber) {
···
220
285
func (em *EventManager) TakeDownRepo(ctx context.Context, user models.Uid) error {
221
286
return em.persister.TakeDownRepo(ctx, user)
222
287
}
223
-
224
-
func (em *EventManager) HandleRebase(ctx context.Context, user models.Uid) error {
225
-
return em.persister.RebaseRepoEvents(ctx, user)
226
-
}
-5
events/persist.go
-5
events/persist.go
···
13
13
Persist(ctx context.Context, e *XRPCStreamEvent) error
14
14
Playback(ctx context.Context, since int64, cb func(*XRPCStreamEvent) error) error
15
15
TakeDownRepo(ctx context.Context, usr models.Uid) error
16
-
RebaseRepoEvents(ctx context.Context, usr models.Uid) error
17
16
Flush(context.Context) error
18
17
Shutdown(context.Context) error
19
18
···
81
80
82
81
func (mp *MemPersister) TakeDownRepo(ctx context.Context, uid models.Uid) error {
83
82
return fmt.Errorf("repo takedowns not currently supported by memory persister, test usage only")
84
-
}
85
-
86
-
func (mp *MemPersister) RebaseRepoEvents(ctx context.Context, usr models.Uid) error {
87
-
return fmt.Errorf("repo rebases not currently supported by memory persister, test usage only")
88
83
}
89
84
90
85
func (mp *MemPersister) Flush(ctx context.Context) error {
-4
events/yolopersist.go
-4
events/yolopersist.go
···
52
52
return fmt.Errorf("repo takedowns not currently supported by memory persister, test usage only")
53
53
}
54
54
55
-
func (yp *YoloPersister) RebaseRepoEvents(ctx context.Context, usr models.Uid) error {
56
-
return fmt.Errorf("repo rebases not currently supported by memory persister, test usage only")
57
-
}
58
-
59
55
func (yp *YoloPersister) SetEventBroadcaster(brc func(*XRPCStreamEvent)) {
60
56
yp.broadcast = brc
61
57
}