package jetstream
import (
"context"
"errors"
"math/rand/v2"
"sync"
"testing"
"time"
)
func TestTheCursorNeverPassesAnUnfinishedEvent(t *testing.T) {
const n = 64
for range 200 {
j := &JetstreamClient{}
tickets := make([]*Ticket, n)
for i := range tickets {
tickets[i] = j.ticket(int64(100 + i*10))
}
rand.Shuffle(n, func(a, b int) { tickets[a], tickets[b] = tickets[b], tickets[a] })
done := map[int64]bool{}
for _, tk := range tickets {
done[tk.us] = true
tk.Done()
for i := range n {
us := int64(100 + i*10)
if !done[us] {
if got := j.lastSeenUs.Load(); got > us {
t.Fatalf("cursor %d passed the unfinished event at %d", got, us)
}
break
}
}
}
if got, want := j.lastSeenUs.Load(), int64(100+(n-1)*10+1); got != want {
t.Fatalf("cursor = %d once everything is done, want %d", got, want)
}
}
}
func TestAHeldEventKeepsTheCursorBehindIt(t *testing.T) {
j := &JetstreamClient{wantedDids: make(Set[string])}
var held *Ticket
wrapped := j.withDidFilter(func(ctx context.Context, e *Event) error {
if e.TimeUS == 100 {
held = Hold(ctx)
}
return nil
})
_ = wrapped(context.Background(), &Event{Did: "did:plc:a", TimeUS: 100})
_ = wrapped(context.Background(), &Event{Did: "did:plc:a", TimeUS: 200})
if got := j.lastSeenUs.Load(); got != 0 {
t.Fatalf("cursor = %d with an earlier event still held, want it untouched", got)
}
held.Done()
if got := j.lastSeenUs.Load(); got != 201 {
t.Fatalf("cursor = %d after the held event finished, want 201", got)
}
}
func TestHoldOutsideTheClientIsANoOp(t *testing.T) {
if tk := Hold(context.Background()); tk != nil {
t.Fatalf("Hold = %+v with no event in the context", tk)
}
var none *Ticket
none.Done()
}
func TestTicketsFinishingAtOnceStillLandOnTheNewestEvent(t *testing.T) {
const n = 500
j := &JetstreamClient{}
tickets := make([]*Ticket, n)
for i := range tickets {
tickets[i] = j.ticket(int64(i + 1))
}
var wg sync.WaitGroup
for _, tk := range tickets {
wg.Go(tk.Done)
}
wg.Wait()
if got := j.lastSeenUs.Load(); got != n+1 {
t.Fatalf("cursor = %d, want %d", got, n+1)
}
if len(j.marks.pending) != 0 {
t.Fatalf("%d events still pending", len(j.marks.pending))
}
}
func TestShutdownWaitsForHeldEventsBeforeTheLastSave(t *testing.T) {
db := &fakeCursorDB{savedErr: errors.New("no row")}
j, cancel := startClient(t, refusedEndpoint, db, false)
ticket := j.ticket(7)
ticket.holding.Store(true)
j.holding.Add(1)
cancel()
select {
case <-j.Done():
t.Fatal("the client finished while a processor still held an event")
case <-time.After(200 * time.Millisecond):
}
ticket.Drop()
requireStopsWithSavedCursor(t, j, db, "the client never finished after the held event was dropped")
}
// replays come out of time order, and the cursor tracks the max, not the last one
func TestTheCursorLandsOnTheNewestEventReleasedTogether(t *testing.T) {
j := &JetstreamClient{}
a, b, replayed := j.ticket(100), j.ticket(200), j.ticket(100)
b.Done()
replayed.Done()
a.Done()
if got := j.lastSeenUs.Load(); got != 201 {
t.Fatalf("cursor = %d, want 201 once the event at 200 is done too", got)
}
}