feat(cluster): persist learnt locators behind a "Chase new grids" option
The callsign->grid map died with the process. Every restart began with an empty Locator column that took an hour of listening to refill, and everything learnt the day before was thrown away. internal/gridcache is its own SQLite file in the data directory, not a table in the settings database: that one sits wherever the operator chose to put it, often a synchronised folder, and a store that rewrites itself every minute has no business there. Deleting the file costs a few days of listening and nothing else. Callsign is the primary key and the newest report wins — operators move, go portable, go on expedition, and a stale locator is worse than none for grid chasing because it reads as a square already worked. Entries not seen in two years are pruned at open: that is the one way this cache can be actively wrong rather than merely empty, and it is what bounds a store that would otherwise only grow. Only CHANGES are queued. The feeds repeat themselves, so writing every report would put the whole stream in the batch instead of the news in it. Batches flush on a timer and on shutdown — a restart is exactly when the cache is worth the most. Rotation is switched off while the store is attached: the cap would discard callsigns the database still holds and the lookup would then miss what we know. Age in the store is the bound instead.
This commit is contained in:
@@ -0,0 +1,224 @@
|
||||
// Package gridcache is the long-term callsign→grid store behind grid chasing.
|
||||
//
|
||||
// A DX-cluster line never carries the DX's locator, so a spot can only show one
|
||||
// if OpsLog learnt it elsewhere: a CQ decoded over the WSJT-X UDP link, or a
|
||||
// PSK Reporter report. Learning it is easy; the problem is that the knowledge
|
||||
// died with the process. Every restart began with an empty column that took an
|
||||
// hour of listening to fill, and everything learnt yesterday was thrown away.
|
||||
//
|
||||
// So the map is persisted. Three things follow from what it is:
|
||||
//
|
||||
// - It is a CACHE, not user data. Deleting the file costs a few days of
|
||||
// listening and nothing else, which is why it lives in its own file rather
|
||||
// than in the settings database — that one sits wherever the operator chose
|
||||
// to put it, often a synchronised folder, and a store that rewrites itself
|
||||
// every minute has no business there.
|
||||
//
|
||||
// - A callsign has ONE grid. The key is unique and the newest report wins:
|
||||
// operators move, go portable, go on expedition. A stale locator is worse
|
||||
// than none for grid chasing, because it reads as a square already worked.
|
||||
//
|
||||
// - Writes are batched. The feeds repeat themselves — the same station is
|
||||
// reported by dozens of receivers a minute — so the store accumulates
|
||||
// changes in memory and flushes them on a timer. Nothing on the ingest path
|
||||
// touches the disk.
|
||||
package gridcache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
// Retention bounds a store that would otherwise only ever grow. Two years is
|
||||
// chosen to be far longer than any propagation interest and short enough that a
|
||||
// reassigned callsign eventually stops carrying its previous holder's square —
|
||||
// the one way this cache can be actively wrong rather than merely empty.
|
||||
const Retention = 2 * 365 * 24 * time.Hour
|
||||
|
||||
// FlushEvery is the batch interval. Long enough that a burst of reports costs
|
||||
// one transaction, short enough that a crash loses a minute of learning.
|
||||
const FlushEvery = 60 * time.Second
|
||||
|
||||
type Store struct {
|
||||
db *sql.DB
|
||||
|
||||
mu sync.Mutex
|
||||
dirty map[string]string // call → grid, waiting to be written
|
||||
|
||||
stop chan struct{}
|
||||
stopOnce sync.Once
|
||||
wg sync.WaitGroup
|
||||
|
||||
logf func(string, ...any)
|
||||
}
|
||||
|
||||
// Open creates or opens the store at path and prunes what has aged out.
|
||||
func Open(path string, logf func(string, ...any)) (*Store, error) {
|
||||
if logf == nil {
|
||||
logf = func(string, ...any) {}
|
||||
}
|
||||
// WAL so a flush never blocks a read, and a busy timeout because the flush
|
||||
// goroutine and the startup load can overlap on a slow disk.
|
||||
db, err := sql.Open("sqlite", path+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("gridcache: open %s: %w", path, err)
|
||||
}
|
||||
if _, err := db.Exec(`CREATE TABLE IF NOT EXISTS grids (
|
||||
call TEXT PRIMARY KEY,
|
||||
grid TEXT NOT NULL,
|
||||
updated_at INTEGER NOT NULL
|
||||
)`); err != nil {
|
||||
db.Close()
|
||||
return nil, fmt.Errorf("gridcache: schema: %w", err)
|
||||
}
|
||||
s := &Store{db: db, dirty: map[string]string{}, stop: make(chan struct{}), logf: logf}
|
||||
if n, err := s.prune(); err != nil {
|
||||
s.logf("gridcache: prune failed: %v", err)
|
||||
} else if n > 0 {
|
||||
s.logf("gridcache: pruned %d locators not heard in %d days", n, int(Retention.Hours()/24))
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
func (s *Store) prune() (int64, error) {
|
||||
cut := time.Now().Add(-Retention).Unix()
|
||||
res, err := s.db.Exec(`DELETE FROM grids WHERE updated_at < ?`, cut)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return res.RowsAffected()
|
||||
}
|
||||
|
||||
// LoadAll returns every stored locator, for seeding the in-memory map at
|
||||
// startup. One query and one map build — the point of the whole package is that
|
||||
// nothing afterwards has to ask the database anything.
|
||||
func (s *Store) LoadAll() (map[string]string, error) {
|
||||
rows, err := s.db.Query(`SELECT call, grid FROM grids`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("gridcache: load: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
out := make(map[string]string, 4096)
|
||||
for rows.Next() {
|
||||
var call, grid string
|
||||
if err := rows.Scan(&call, &grid); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[call] = grid
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// Put queues a locator for writing. Callers pass only what CHANGED — an
|
||||
// unchanged report is the common case by a wide margin and must not reach here,
|
||||
// or the batch would carry the whole feed instead of the news in it.
|
||||
func (s *Store) Put(call, grid string) {
|
||||
call = strings.ToUpper(strings.TrimSpace(call))
|
||||
grid = strings.TrimSpace(grid)
|
||||
if call == "" || grid == "" {
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
s.dirty[call] = grid
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// Start runs the flush loop until Close.
|
||||
func (s *Store) Start(ctx context.Context) {
|
||||
s.wg.Add(1)
|
||||
go func() {
|
||||
defer s.wg.Done()
|
||||
t := time.NewTicker(FlushEvery)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-s.stop:
|
||||
return
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
if err := s.Flush(); err != nil {
|
||||
s.logf("gridcache: flush failed: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Flush writes the pending batch in one transaction. Safe to call with nothing
|
||||
// pending, which is most of the time on a quiet band.
|
||||
func (s *Store) Flush() error {
|
||||
s.mu.Lock()
|
||||
if len(s.dirty) == 0 {
|
||||
s.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
batch := s.dirty
|
||||
s.dirty = map[string]string{}
|
||||
s.mu.Unlock()
|
||||
|
||||
tx, err := s.db.Begin()
|
||||
if err != nil {
|
||||
s.requeue(batch)
|
||||
return err
|
||||
}
|
||||
st, err := tx.Prepare(`INSERT INTO grids (call, grid, updated_at) VALUES (?, ?, ?)
|
||||
ON CONFLICT(call) DO UPDATE SET grid = excluded.grid, updated_at = excluded.updated_at`)
|
||||
if err != nil {
|
||||
tx.Rollback()
|
||||
s.requeue(batch)
|
||||
return err
|
||||
}
|
||||
defer st.Close()
|
||||
now := time.Now().Unix()
|
||||
for call, grid := range batch {
|
||||
if _, err := st.Exec(call, grid, now); err != nil {
|
||||
tx.Rollback()
|
||||
s.requeue(batch)
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
s.requeue(batch)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// requeue puts a failed batch back, without overwriting anything learnt while it
|
||||
// was in flight — the newer value is the right one.
|
||||
func (s *Store) requeue(batch map[string]string) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
for call, grid := range batch {
|
||||
if _, newer := s.dirty[call]; !newer {
|
||||
s.dirty[call] = grid
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Pending reports how many locators are waiting to be written (for diagnostics).
|
||||
func (s *Store) Pending() int {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return len(s.dirty)
|
||||
}
|
||||
|
||||
// Close stops the loop and writes whatever is pending. A restart is the moment
|
||||
// the cache is most valuable, so losing the last minute of learning to a clean
|
||||
// shutdown would be a poor trade.
|
||||
func (s *Store) Close() error {
|
||||
s.stopOnce.Do(func() { close(s.stop) })
|
||||
s.wg.Wait()
|
||||
err := s.Flush()
|
||||
if cerr := s.db.Close(); err == nil {
|
||||
err = cerr
|
||||
}
|
||||
return err
|
||||
}
|
||||
Reference in New Issue
Block a user