package jetstream
import (
"context"
"sync"
"sync/atomic"
)
// a crash replays whatever is still pending because the cursor only moves up to the front
type watermark struct {
mu sync.Mutex
front uint64
pending []mark
}
type mark struct {
us int64
done bool
}
// a ticket is a held event, the cursor can't pass it until Done.
// skipping Done pins the cursor here forever
type Ticket struct {
j *JetstreamClient
id uint64
us int64
// shutdown waits for held tickets before the last cursor save
holding atomic.Bool
}
func (j *JetstreamClient) ticket(us int64) *Ticket {
w := &j.marks
w.mu.Lock()
defer w.mu.Unlock()
id := w.front + uint64(len(w.pending))
w.pending = append(w.pending, mark{us: us})
return &Ticket{j: j, id: id, us: us}
}
func (t *Ticket) Done() {
if t == nil {
return
}
defer t.release()
w := &t.j.marks
w.mu.Lock()
defer w.mu.Unlock()
if t.id < w.front {
return
}
w.pending[t.id-w.front].done = true
var newest int64
moved := false
for len(w.pending) > 0 && w.pending[0].done {
newest, moved = max(newest, w.pending[0].us), true
w.pending = w.pending[1:]
w.front++
}
// never move the cursor back, since reconnect replays can be older than it
if moved && newest+1 > t.j.lastSeenUs.Load() {
t.j.lastSeenUs.Store(newest + 1)
}
}
// the cursor stays before this event since dropping doesn't mark it done
func (t *Ticket) Drop() {
if t != nil {
t.release()
}
}
func (t *Ticket) release() {
if t.holding.CompareAndSwap(true, false) {
t.j.holding.Add(-1)
}
}
type holdKey struct{}
type hold struct {
ticket *Ticket
held atomic.Bool
}
// hold before the processor returns, or the cursor moves past it
func Hold(ctx context.Context) *Ticket {
h, ok := ctx.Value(holdKey{}).(*hold)
if !ok {
return nil
}
h.held.Store(true)
if h.ticket.holding.CompareAndSwap(false, true) {
h.ticket.j.holding.Add(1)
}
return h.ticket
}