mirror of
https://github.com/J3vb/OwnCord.git
synced 2026-09-03 03:50:00 +03:00
jevb
f3c9f98b91
- BUG-085: ring buffer EventsSince off-by-one — change < to <= so afterSeq == oldestSeq returns nil (triggers full ready payload) - BUG-087: GracefulStop not idempotent — wrap body in sync.Once to prevent double lkProcess.Stop() on concurrent calls - BUG-090: FTS query truncation at byte boundary — use []rune truncation to preserve valid UTF-8 for CJK/emoji input - BUG-091: updater downloadFile double-closes file on Windows — add closed sentinel to guard defer against explicit Close()
80 lines
2.0 KiB
Go
80 lines
2.0 KiB
Go
package ws
|
|
|
|
import "github.com/owncord/server/syncutil"
|
|
|
|
// eventEntry stores a broadcast event for potential replay.
|
|
type eventEntry struct {
|
|
seq uint64
|
|
data []byte
|
|
}
|
|
|
|
// EventRingBuffer is a bounded, thread-safe ring buffer for recent broadcast events.
|
|
type EventRingBuffer struct {
|
|
mu syncutil.RWMutex
|
|
entries []eventEntry
|
|
size int
|
|
pos int // next write position
|
|
count int // total entries stored (up to size)
|
|
}
|
|
|
|
// NewEventRingBuffer creates a ring buffer with the given capacity.
|
|
func NewEventRingBuffer(size int) *EventRingBuffer {
|
|
return &EventRingBuffer{
|
|
entries: make([]eventEntry, size),
|
|
size: size,
|
|
}
|
|
}
|
|
|
|
// Push adds an event to the ring buffer.
|
|
func (rb *EventRingBuffer) Push(seq uint64, data []byte) {
|
|
rb.mu.Lock()
|
|
defer rb.mu.Unlock()
|
|
rb.entries[rb.pos] = eventEntry{seq: seq, data: data}
|
|
rb.pos = (rb.pos + 1) % rb.size
|
|
if rb.count < rb.size {
|
|
rb.count++
|
|
}
|
|
}
|
|
|
|
// EventsSince returns all events with seq > afterSeq, in order.
|
|
// Returns nil if afterSeq is too old (no longer in the buffer).
|
|
func (rb *EventRingBuffer) EventsSince(afterSeq uint64) [][]byte {
|
|
rb.mu.RLock()
|
|
defer rb.mu.RUnlock()
|
|
|
|
if rb.count == 0 {
|
|
return nil
|
|
}
|
|
|
|
// Find the oldest entry in the buffer.
|
|
oldestIdx := (rb.pos - rb.count + rb.size) % rb.size
|
|
oldestSeq := rb.entries[oldestIdx].seq
|
|
|
|
// If the requested seq is at or older than our oldest, we can't guarantee
|
|
// full coverage — return nil to trigger a full ready payload.
|
|
if afterSeq <= oldestSeq {
|
|
return nil
|
|
}
|
|
|
|
result := make([][]byte, 0)
|
|
for i := 0; i < rb.count; i++ {
|
|
idx := (oldestIdx + i) % rb.size
|
|
e := rb.entries[idx]
|
|
if e.seq > afterSeq {
|
|
result = append(result, e.data)
|
|
}
|
|
}
|
|
return result
|
|
}
|
|
|
|
// OldestSeq returns the oldest sequence number in the buffer, or 0 if empty.
|
|
func (rb *EventRingBuffer) OldestSeq() uint64 {
|
|
rb.mu.RLock()
|
|
defer rb.mu.RUnlock()
|
|
if rb.count == 0 {
|
|
return 0
|
|
}
|
|
oldestIdx := (rb.pos - rb.count + rb.size) % rb.size
|
|
return rb.entries[oldestIdx].seq
|
|
}
|