90 lines
2.6 KiB
Go
90 lines
2.6 KiB
Go
package events
|
|
|
|
// Replay ring: every published event gets a monotonic id and lands in a
|
|
// fixed-capacity ring buffer so an SSE client that reconnects with
|
|
// Last-Event-ID can be caught up without a full poll. The ring is
|
|
// best-effort — when a client's resume point has already been overwritten,
|
|
// Replay reports the gap and the caller falls back to the state snapshot
|
|
// it sends on every connect anyway.
|
|
|
|
import (
|
|
"sync"
|
|
|
|
panelv1 "github.com/dbledeez/panel/proto/panel/v1"
|
|
)
|
|
|
|
// Stamped pairs an event with its bus-assigned monotonic id. IDs start at 1
|
|
// and never repeat for the lifetime of the process.
|
|
type Stamped struct {
|
|
ID uint64
|
|
Event *panelv1.Event
|
|
}
|
|
|
|
// DefaultRingCap is how many recent events the bus retains for replay.
|
|
// State/app/player events are small and logs dominate volume; at a burst of
|
|
// ~200 events/s this still covers ~20s of reconnect window, and the SSE
|
|
// snapshot covers anything older.
|
|
const DefaultRingCap = 4096
|
|
|
|
type replayRing struct {
|
|
mu sync.Mutex
|
|
buf []Stamped
|
|
next uint64 // id to assign to the NEXT appended event (starts at 1)
|
|
head int // index of oldest valid entry
|
|
n int // number of valid entries (<= cap)
|
|
}
|
|
|
|
func newReplayRing(capacity int) *replayRing {
|
|
if capacity <= 0 {
|
|
capacity = DefaultRingCap
|
|
}
|
|
return &replayRing{buf: make([]Stamped, capacity), next: 1}
|
|
}
|
|
|
|
// append stamps e with the next monotonic id, stores it, and returns the id.
|
|
func (r *replayRing) append(e *panelv1.Event) uint64 {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
id := r.next
|
|
r.next++
|
|
idx := (r.head + r.n) % len(r.buf)
|
|
if r.n < len(r.buf) {
|
|
r.buf[idx] = Stamped{ID: id, Event: e}
|
|
r.n++
|
|
} else {
|
|
r.buf[r.head] = Stamped{ID: id, Event: e}
|
|
r.head = (r.head + 1) % len(r.buf)
|
|
}
|
|
return id
|
|
}
|
|
|
|
// lastID returns the most recently assigned id (0 when nothing published).
|
|
func (r *replayRing) lastID() uint64 {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
return r.next - 1
|
|
}
|
|
|
|
// since returns every retained event with ID > afterID, oldest first.
|
|
// ok=false means the resume point has been overwritten (afterID is older
|
|
// than the oldest retained event), so the returned slice — while still the
|
|
// full retained suffix — has a gap before it; callers should resync from a
|
|
// snapshot. afterID >= lastID returns (nil, true).
|
|
func (r *replayRing) since(afterID uint64) (out []Stamped, ok bool) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
if r.n == 0 {
|
|
// Nothing retained: only gapless if the caller is already current.
|
|
return nil, afterID >= r.next-1
|
|
}
|
|
oldest := r.buf[r.head].ID
|
|
ok = afterID >= oldest-1
|
|
for i := 0; i < r.n; i++ {
|
|
s := r.buf[(r.head+i)%len(r.buf)]
|
|
if s.ID > afterID {
|
|
out = append(out, s)
|
|
}
|
|
}
|
|
return out, ok
|
|
}
|