+4 -4

events/events.go

reviewed

··· 290 290 case <-done: 291 291 return ErrPlaybackShutdown 292 292 case out <- e: 293 293 - seq := sequenceForEvent(e) 293 293 + seq := SequenceForEvent(e) 294 294 if seq > 0 { 295 295 lastSeq = seq 296 296 } ··· 315 315 316 316 // run playback again to get us to the events that have started buffering 317 317 if err := em.persister.Playback(ctx, lastSeq, func(e *XRPCStreamEvent) error { 318 318 - seq := sequenceForEvent(e) 319 319 - if seq > sequenceForEvent(first) { 318 318 + seq := SequenceForEvent(e) 319 319 + if seq > SequenceForEvent(first) { 320 320 return ErrCaughtUp 321 321 } 322 322 ··· 351 351 return out, sub.cleanup, nil 352 352 } 353 353 354 354 - func sequenceForEvent(evt *XRPCStreamEvent) int64 { 354 354 + func SequenceForEvent(evt *XRPCStreamEvent) int64 { 355 355 switch { 356 356 case evt == nil: 357 357 return -1

+2 -2

splitter/ringbuf.go

reviewed

··· 96 96 for ; i >= 0; i-- { 97 97 c := chunks[i] 98 98 evts := c.events() 99 99 - if since > sequenceForEvent(evts[len(evts)-1]) { 99 99 + if since > events.SequenceForEvent(evts[len(evts)-1]) { 100 100 i++ 101 101 break 102 102 } ··· 112 112 for nread < len(evts) { 113 113 for _, e := range evts[nread:] { 114 114 nread++ 115 115 - seq := sequenceForEvent(e) 115 115 + seq := events.SequenceForEvent(e) 116 116 if seq <= since { 117 117 continue 118 118 }

+2 -18

splitter/splitter.go

reviewed

··· 383 383 defer cancel() 384 384 385 385 sched := sequential.NewScheduler("splitter", func(ctx context.Context, evt *events.XRPCStreamEvent) error { 386 386 - seq := sequenceForEvent(evt) 386 386 + seq := events.SequenceForEvent(evt) 387 387 if seq < 0 { 388 388 // ignore info events and other unsupported types 389 389 return nil ··· 402 402 *lastCursor = seq 403 403 return nil 404 404 }) 405 405 - return events.HandleRepoStream(ctx, con, sched) 406 406 - } 407 405 408 408 - func sequenceForEvent(evt *events.XRPCStreamEvent) int64 { 409 409 - switch { 410 410 - case evt.RepoCommit != nil: 411 411 - return evt.RepoCommit.Seq 412 412 - case evt.RepoHandle != nil: 413 413 - return evt.RepoHandle.Seq 414 414 - case evt.RepoMigrate != nil: 415 415 - return evt.RepoMigrate.Seq 416 416 - case evt.RepoTombstone != nil: 417 417 - return evt.RepoTombstone.Seq 418 418 - case evt.RepoInfo != nil: 419 419 - return -1 420 420 - default: 421 421 - return -1 422 422 - } 406 406 + return events.HandleRepoStream(ctx, con, sched) 423 407 } 424 408 425 409 func (s *Splitter) getLastCursor() (int64, error) {