From 044aecbd3c7698c83e3ba9915f096221a8ed2c3d Mon Sep 17 00:00:00 2001 From: "Enot (ded) Skelly" Date: Fri, 12 Jun 2026 10:28:32 -0700 Subject: [PATCH] feat: background task scheduler move refresh views and delete tasks to scheduler add new config for refresh and cleanup intervals new deployments can set these in minutes to see data faster --- cmd/beacon/main.go | 52 +++++++++++-------------------- config.yaml.example | 6 ++++ internal/background/background.go | 50 +++++++++++++++++++++++++++++ internal/background/tasks.go | 49 +++++++++++++++++++++++++++++ internal/config/config.go | 12 +++++++ 5 files changed, 135 insertions(+), 34 deletions(-) create mode 100644 internal/background/background.go create mode 100644 internal/background/tasks.go diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index 55e303e..cdfdfa2 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -20,6 +20,7 @@ import ( _ "github.com/MeshCore-Beacon/beacon-server/docs" "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/MeshCore-Beacon/beacon-server/internal/api/router" + "github.com/MeshCore-Beacon/beacon-server/internal/background" "github.com/MeshCore-Beacon/beacon-server/internal/cache" "github.com/MeshCore-Beacon/beacon-server/internal/config" "github.com/MeshCore-Beacon/beacon-server/internal/hub" @@ -100,6 +101,18 @@ func main() { maxConnsPerIP = 5 } + // resolve background intervals with defaults + viewRefreshInterval := cfg.Background.ViewRefresh.Duration + if viewRefreshInterval == 0 { + viewRefreshInterval = time.Hour + } + cleanupInterval := cfg.Background.Cleanup.Duration + if cleanupInterval == 0 { + cleanupInterval = time.Hour + } + log.Printf("config: loaded — telemetryResolution=%s telemetryRetention=%s packetRetention=%s maxConnsPerIP=%d viewRefresh=%s cleanup=%s", + telemetryResolution, telemetryRetention, packetRetention, maxConnsPerIP, viewRefreshInterval, cleanupInterval) + // ── Hub ────────────────────────────────────────────────────────────────── h := hub.New() go h.Run() @@ -147,9 +160,6 @@ func main() { } } - // refresh materialized views on boot or restart to stay fresh - refreshMaterializedViews(ctx, store) - // ── Seed config data ───────────────────────────────────────────────────── if err := config.Seed(ctx, cfg, store); err != nil { log.Fatalf("failed to seed config: %v", err) @@ -250,25 +260,11 @@ func main() { go broker1.Start(ctx) go broker2.Start(ctx) - // ── cleanup and materialized view refresh goroutine ───────────────────────────────────────── - go func() { - ticker := time.NewTicker(time.Hour) - defer ticker.Stop() - for { - select { - case <-ticker.C: - if err := store.DeleteOldTelemetry(ctx, time.Now().Add(-telemetryRetention)); err != nil { - log.Printf("cleanup: delete old telemetry failed: %v", err) - } - if err := store.DeleteOldPackets(ctx, time.Now().Add(-packetRetention)); err != nil { - log.Printf("cleanup: delete old packets failed: %v", err) - } - refreshMaterializedViews(ctx, store) - case <-ctx.Done(): - return - } - } - }() + scheduler := background.New([]background.Task{ + background.ViewRefreshTask(store, viewRefreshInterval), + background.CleanupTask(store, telemetryRetention, packetRetention, cleanupInterval), + }) + go scheduler.Start(ctx) // ── HTTP server ────────────────────────────────────────────────────────── r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, maxConnsPerIP, cfg.CORS) @@ -320,15 +316,3 @@ func getEnv(key string) string { } return v } - -func refreshMaterializedViews(ctx context.Context, store *db.Store) { - if err := store.RefreshHourlyStats(ctx); err != nil { - log.Printf("refresh: materialized view for hourly stats failed: %v", err) - } - if err := store.RefreshTopNodes(ctx); err != nil { - log.Printf("refresh: materialized view for top nodes failed: %v", err) - } - if err := store.RefreshRadioPresets(ctx); err != nil { - log.Printf("refresh: materialized view for radio presets failed: %v", err) - } -} diff --git a/config.yaml.example b/config.yaml.example index 657a8d9..9897bf9 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -117,3 +117,9 @@ cache: # allow_countries: [CA, US] # allow_continents: [NA] +# Background task intervals. +# Shorter intervals are useful during initial deployment to confirm data is +# flowing. Back off to 1h or more once stable. +#background: +# view_refresh: 1h # default: 1h +# cleanup: 1h # default: 1h diff --git a/internal/background/background.go b/internal/background/background.go new file mode 100644 index 0000000..5e8c189 --- /dev/null +++ b/internal/background/background.go @@ -0,0 +1,50 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +// Package background runs periodic maintenance tasks on independent schedules. +package background + +import ( + "context" + "log" + "time" +) + +// Task is a named unit of work that runs on a fixed interval. +type Task struct { + Name string + Interval time.Duration + Run func(ctx context.Context) error +} + +// Scheduler runs a set of tasks on independent tickers. +type Scheduler struct { + tasks []Task +} + +// New creates a Scheduler with the given tasks. +func New(tasks []Task) *Scheduler { + return &Scheduler{tasks: tasks} +} + +// Start launches each task in its own goroutine. Blocks until ctx is cancelled. +func (s *Scheduler) Start(ctx context.Context) { + for _, t := range s.tasks { + go func() { + ticker := time.NewTicker(t.Interval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + log.Printf("background[%s]: running", t.Name) + if err := t.Run(ctx); err != nil { + log.Printf("background[%s]: %v", t.Name, err) + } + log.Printf("background[%s]: complete", t.Name) + case <-ctx.Done(): + return + } + } + }() + } +} diff --git a/internal/background/tasks.go b/internal/background/tasks.go new file mode 100644 index 0000000..67df0f0 --- /dev/null +++ b/internal/background/tasks.go @@ -0,0 +1,49 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package background + +import ( + "context" + "log" + "time" + + "github.com/MeshCore-Beacon/beacon-server/db" +) + +// ViewRefreshTask returns a Task that refreshes all materialized views. +func ViewRefreshTask(store *db.Store, interval time.Duration) Task { + return Task{ + Name: "view_refresh", + Interval: interval, + Run: func(ctx context.Context) error { + if err := store.RefreshHourlyStats(ctx); err != nil { + log.Printf("background[view_refresh]: hourly stats: %v", err) + } + if err := store.RefreshTopNodes(ctx); err != nil { + log.Printf("background[view_refresh]: top nodes: %v", err) + } + if err := store.RefreshRadioPresets(ctx); err != nil { + log.Printf("background[view_refresh]: radio presets: %v", err) + } + return nil + }, + } +} + +// CleanupTask returns a Task that prunes old telemetry and packet rows. +func CleanupTask(store *db.Store, telemetryRetention, packetRetention, interval time.Duration) Task { + return Task{ + Name: "cleanup", + Interval: interval, + Run: func(ctx context.Context) error { + if err := store.DeleteOldTelemetry(ctx, time.Now().Add(-telemetryRetention)); err != nil { + return err + } + if err := store.DeleteOldPackets(ctx, time.Now().Add(-packetRetention)); err != nil { + return err + } + return nil + }, + } +} diff --git a/internal/config/config.go b/internal/config/config.go index 52b4104..658069c 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -24,6 +24,18 @@ type Config struct { Scopes []ScopeConfig `yaml:"scopes"` Cache CacheConfig `yaml:"cache"` CORS CORSConfig `yaml:"cors"` + Background BackgroundConfig `yaml:"background"` +} + +// BackgroundConfig controls the intervals for background maintenance tasks. +type BackgroundConfig struct { + // ViewRefresh is how often materialized views are refreshed. + // Defaults to 1h if not set. + ViewRefresh duration `yaml:"view_refresh"` + + // Cleanup is how often old telemetry and packet rows are pruned. + // Defaults to 1h if not set. + Cleanup duration `yaml:"cleanup"` } // CORSConfig controls Cross-Origin Resource Sharing behaviour.