rouggy
3c8aadf41f
A batch that never reaches the disk looks exactly like one that does: the locators are in memory either way until the next restart, which is the one moment the difference shows. The shutdown flush was the least observable of all — it runs last, and nothing said whether it had written anything.
227 lines
6.5 KiB
Go
227 lines
6.5 KiB
Go
// Package gridcache is the long-term callsign→grid store behind grid chasing.
|
|
//
|
|
// Its own SQLite file, not a table in the settings database: that one sits
|
|
// wherever the operator put it, often a synchronised folder, and this rewrites
|
|
// itself every minute. Deleting the file costs a few days of listening.
|
|
//
|
|
// A callsign has one grid and the newest report wins — a stale locator reads as
|
|
// a square already worked, which is worse than none.
|
|
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
|
|
|
|
// Sources a locator can come from.
|
|
const (
|
|
SourceDecode = "decode" // a CQ this station's own receiver decoded
|
|
SourceMQTT = "mqtt" // a PSK Reporter report
|
|
)
|
|
|
|
// Entry is one locator and where it came from.
|
|
type Entry struct {
|
|
Grid string
|
|
Source string
|
|
}
|
|
|
|
type Store struct {
|
|
db *sql.DB
|
|
|
|
mu sync.Mutex
|
|
dirty map[string]Entry // call → what to write
|
|
|
|
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, " +
|
|
"source TEXT NOT NULL DEFAULT '')"); err != nil {
|
|
db.Close()
|
|
return nil, fmt.Errorf("gridcache: schema: %w", err)
|
|
}
|
|
// A file written before the column existed keeps its rows; this fails
|
|
// harmlessly when the column is already there.
|
|
db.Exec("ALTER TABLE grids ADD COLUMN source TEXT NOT NULL DEFAULT ''")
|
|
s := &Store{db: db, dirty: map[string]Entry{}, 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, source string) {
|
|
call = strings.ToUpper(strings.TrimSpace(call))
|
|
grid = strings.TrimSpace(grid)
|
|
if call == "" || grid == "" {
|
|
return
|
|
}
|
|
s.mu.Lock()
|
|
s.dirty[call] = Entry{Grid: grid, Source: source}
|
|
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]Entry{}
|
|
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, source) VALUES (?, ?, ?, ?)
|
|
ON CONFLICT(call) DO UPDATE SET grid = excluded.grid, updated_at = excluded.updated_at,
|
|
source = excluded.source`)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
s.requeue(batch)
|
|
return err
|
|
}
|
|
defer st.Close()
|
|
now := time.Now().Unix()
|
|
for call, e := range batch {
|
|
if _, err := st.Exec(call, e.Grid, now, e.Source); err != nil {
|
|
tx.Rollback()
|
|
s.requeue(batch)
|
|
return err
|
|
}
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
s.requeue(batch)
|
|
return err
|
|
}
|
|
// Logged because a batch that never reaches the disk looks exactly like one
|
|
// that does: the locators are in memory either way until the next restart,
|
|
// which is the one moment the difference shows.
|
|
s.logf("gridcache: wrote %d locators", len(batch))
|
|
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]Entry) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
for call, e := range batch {
|
|
if _, newer := s.dirty[call]; !newer {
|
|
s.dirty[call] = e
|
|
}
|
|
}
|
|
}
|
|
|
|
// Pending reports how many locators are waiting to be written.
|
|
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
|
|
}
|