feat: never drop cluster spots + theme-aware spot badges + FR settings
Backend: replace the bounded cluster-event channel (which dropped spots under RBN bursts once enrichment/UI fell behind) with an unbounded clusterQueue (slice + sync.Cond). The socket-read goroutine still never blocks, but a burst now grows the backlog and drains instead of losing spots. Head compaction keeps a sustained burst from growing the backing array without bound. ClusterGrid: drop all hard-coded light-theme hex colours (Time, Country, Spotter, Comment, POTA, band/mode fills) for semantic CSS-var tokens so they read correctly on every theme. Replace the heavy full-cell band/mode fills with small rounded pills (cellChip). Only NEW MODE pills the mode cell — NEW SLOT (band and mode each worked, just not together) no longer highlights the mode cell, which wrongly implied the mode was new. App: stage incoming spots ~50ms and resolve their slot status before inserting, so a row paints with its badge already on instead of flashing plain text then flipping to a pill. Self-spot toast still fires per spot. i18n: translate the General settings that were still English — telemetry and live-status toggles, the Main view pane pickers and their options. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
@@ -435,8 +435,8 @@ type App struct {
|
||||
// to run inline in the read loop, so a single slow step stopped draining the
|
||||
// TCP socket and the whole feed fell behind the node (visible as the grid
|
||||
// lagging telnet). The read loop now just enqueues here; one worker does the work.
|
||||
clusterEventCh chan clusterEvent
|
||||
clusterDropped int64 // spots/lines dropped when the queue was full (atomic)
|
||||
// Unbounded FIFO: a burst grows the backlog instead of dropping spots.
|
||||
clusterEvents *clusterQueue
|
||||
// wcbm is an in-memory "CALL|BAND|MODE" worked-index so the alert engine never
|
||||
// queries the DB per cluster spot (an FT8 firehose would swamp a remote MySQL).
|
||||
// Loaded once, appended to on each log, rebuilt after bulk changes.
|
||||
@@ -906,10 +906,11 @@ func (a *App) startup(ctx context.Context) {
|
||||
// renders the row with all metadata already filled (no flicker of
|
||||
// empty Country / Cont columns while the batch status fetch runs).
|
||||
// Cluster events are processed OFF the socket-read goroutine (see clusterEvent /
|
||||
// clusterEventWorker). Sized large so ordinary traffic and even an SH/DX/100
|
||||
// burst never fill it; a full queue drops-and-counts rather than block the read
|
||||
// loop, which was the actual cause of the feed falling behind telnet.
|
||||
a.clusterEventCh = make(chan clusterEvent, 8192)
|
||||
// clusterEventWorker). The queue is UNBOUNDED so no spot is ever dropped: an
|
||||
// SH/DX or RBN burst grows the backlog and drains once enrichment catches up,
|
||||
// while the read loop still never blocks (that was the cause of the feed falling
|
||||
// behind telnet).
|
||||
a.clusterEvents = newClusterQueue()
|
||||
go a.clusterEventWorker()
|
||||
|
||||
a.cluster = cluster.NewManager(
|
||||
@@ -5902,20 +5903,85 @@ type clusterEvent struct {
|
||||
line *cluster.Line
|
||||
}
|
||||
|
||||
// enqueueClusterEvent hands an event to the worker WITHOUT blocking. It runs on
|
||||
// the cluster session's socket-read goroutine: blocking here would stop draining
|
||||
// the TCP socket, the node's send buffer would fill, and the feed would fall
|
||||
// behind — the exact bug this indirection fixes. If the queue is full (processing
|
||||
// can't keep up), drop and count rather than stall the whole feed.
|
||||
func (a *App) enqueueClusterEvent(ev clusterEvent) {
|
||||
select {
|
||||
case a.clusterEventCh <- ev:
|
||||
default:
|
||||
n := atomic.AddInt64(&a.clusterDropped, 1)
|
||||
if n == 1 || n%100 == 0 {
|
||||
applog.Printf("cluster: processing backlog — dropped %d event(s); the feed is faster than enrichment/UI", n)
|
||||
}
|
||||
// clusterQueue is an unbounded FIFO between the socket-read goroutines (producers,
|
||||
// which must never block or the TCP feed stalls) and the single clusterEventWorker
|
||||
// (consumer). Unlike a bounded channel it NEVER drops an event: a burst just grows
|
||||
// the backlog, which drains once enrichment catches up.
|
||||
type clusterQueue struct {
|
||||
mu sync.Mutex
|
||||
cond *sync.Cond
|
||||
buf []clusterEvent
|
||||
head int // index of the next event to pop (waste reclaimed by compaction)
|
||||
closed bool
|
||||
}
|
||||
|
||||
func newClusterQueue() *clusterQueue {
|
||||
q := &clusterQueue{}
|
||||
q.cond = sync.NewCond(&q.mu)
|
||||
return q
|
||||
}
|
||||
|
||||
// push appends an event and wakes the worker. Amortised O(1); never blocks, never
|
||||
// drops.
|
||||
func (q *clusterQueue) push(ev clusterEvent) {
|
||||
q.mu.Lock()
|
||||
if q.closed {
|
||||
q.mu.Unlock()
|
||||
return
|
||||
}
|
||||
q.buf = append(q.buf, ev)
|
||||
q.mu.Unlock()
|
||||
q.cond.Signal()
|
||||
}
|
||||
|
||||
// pop blocks until an event is available, returning ok=false only once the queue
|
||||
// is closed AND fully drained.
|
||||
func (q *clusterQueue) pop() (clusterEvent, bool) {
|
||||
q.mu.Lock()
|
||||
for q.head == len(q.buf) && !q.closed {
|
||||
q.cond.Wait()
|
||||
}
|
||||
if q.head == len(q.buf) { // closed and drained
|
||||
q.mu.Unlock()
|
||||
return clusterEvent{}, false
|
||||
}
|
||||
ev := q.buf[q.head]
|
||||
q.buf[q.head] = clusterEvent{} // release pointers held by the consumed slot
|
||||
q.head++
|
||||
switch {
|
||||
case q.head == len(q.buf):
|
||||
// Fully drained → reset to reuse the backing array from the front.
|
||||
q.buf = q.buf[:0]
|
||||
q.head = 0
|
||||
case q.head > 1024 && q.head*2 >= len(q.buf):
|
||||
// Head waste is large → compact so a long sustained burst doesn't grow
|
||||
// the backing array without bound.
|
||||
n := copy(q.buf, q.buf[q.head:])
|
||||
for i := n; i < len(q.buf); i++ {
|
||||
q.buf[i] = clusterEvent{}
|
||||
}
|
||||
q.buf = q.buf[:n]
|
||||
q.head = 0
|
||||
}
|
||||
q.mu.Unlock()
|
||||
return ev, true
|
||||
}
|
||||
|
||||
// close wakes any waiting consumer so it can exit once the backlog is drained.
|
||||
func (q *clusterQueue) close() {
|
||||
q.mu.Lock()
|
||||
q.closed = true
|
||||
q.mu.Unlock()
|
||||
q.cond.Broadcast()
|
||||
}
|
||||
|
||||
// enqueueClusterEvent hands an event to the worker WITHOUT blocking and WITHOUT
|
||||
// dropping. It runs on the cluster session's socket-read goroutine: blocking here
|
||||
// would stop draining the TCP socket, the node's send buffer would fill, and the
|
||||
// feed would fall behind — the exact bug this indirection fixes. The queue is
|
||||
// unbounded, so a burst grows the backlog rather than losing a spot.
|
||||
func (a *App) enqueueClusterEvent(ev clusterEvent) {
|
||||
a.clusterEvents.push(ev)
|
||||
}
|
||||
|
||||
// clusterEventWorker drains clusterEventCh and does everything that used to run
|
||||
@@ -5923,7 +5989,11 @@ func (a *App) enqueueClusterEvent(ev clusterEvent) {
|
||||
// to the UI, run alert rules (which may query a remote MySQL) and mirror it to the
|
||||
// Flex panadapter — all serialised on this one goroutine, off the read path.
|
||||
func (a *App) clusterEventWorker() {
|
||||
for ev := range a.clusterEventCh {
|
||||
for {
|
||||
ev, ok := a.clusterEvents.pop()
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if ev.line != nil {
|
||||
if a.ctx != nil {
|
||||
wruntime.EventsEmit(a.ctx, "cluster:line", *ev.line)
|
||||
|
||||
Reference in New Issue
Block a user