feat: harden rota sync and throttle manual /sync
Resolve the current-year sheet tab per fetch instead of freezing it at startup, so a long-running process follows the New Year rollover without a restart. Guard against the regression this introduces: a successful fetch that returns no duty weeks (e.g. a freshly created, empty next-year tab) now keeps the last-good cache instead of clobbering it with empty data. Surface a silently-stale cache — which is not a crash, so process-level supervision can't catch it — by escalating to a loud error log once no sync has succeeded for 6h (new Store.LastSynced accessor backs the check). Throttle manual /sync to once per minute across the chat to avoid spamming the Google Sheets API; the check-and-set is mutex-guarded since Telegram dispatches each update in its own goroutine. Update README accordingly.
This commit is contained in:
parent
15b2fc9093
commit
4f9cc6c359
11
README.md
11
README.md
|
|
@ -18,6 +18,13 @@ The service loads the cached rota, syncs from Google Sheets in the background on
|
|||
`SYNC_INTERVAL` (default 30m), and runs the Telegram bot via long polling until
|
||||
interrupted (Ctrl-C / SIGTERM).
|
||||
|
||||
When `SHEET_RANGE` is unset, the bot reads the current year's tab (e.g.
|
||||
`2026!A1:G100`), resolved on every sync — so it follows the New Year rollover
|
||||
automatically without a restart. A sync that fails, or that returns no duty
|
||||
weeks (e.g. a next-year tab that exists but hasn't been filled in yet), keeps
|
||||
the last-good cache instead of clearing it; if no sync succeeds for 6h the bot
|
||||
logs a loud stale-cache error.
|
||||
|
||||
> **Run exactly one instance per bot token.** The bot uses long polling
|
||||
> (`getUpdates`); two concurrent pollers on the same token make Telegram return
|
||||
> HTTP 409 Conflict and updates get dropped. Do not scale this to multiple
|
||||
|
|
@ -38,7 +45,8 @@ across restarts.
|
|||
### Commands
|
||||
|
||||
- `/kto_sprzata` — who is on cleaning duty this week.
|
||||
- `/sync` — refresh the rota from Google Sheets now.
|
||||
- `/sync` — refresh the rota from Google Sheets now (throttled to once per
|
||||
minute across the chat to avoid spamming the Sheets API).
|
||||
|
||||
### Reminders
|
||||
|
||||
|
|
@ -48,4 +56,3 @@ Schedule (all optional, with defaults): `REMINDER_WEEKDAY` (default `Monday`),
|
|||
is interpreted in Europe/Warsaw time. The reminder announces who is on duty
|
||||
`DUTY_AHEAD_DAYS` from the fire time — e.g. every Monday 09:00, who cleans two
|
||||
weeks out. Leave `GROUP_CHAT_ID` unset to disable reminders.
|
||||
|
||||
|
|
|
|||
85
service.go
85
service.go
|
|
@ -2,6 +2,7 @@ package dyzurbot
|
|||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
|
|
@ -30,13 +31,13 @@ const reminderTimezone = "Europe/Warsaw"
|
|||
func LoadConfig() (Config, error) {
|
||||
_ = godotenv.Load()
|
||||
|
||||
// Default range targets the current year's tab (sheets are named per year,
|
||||
// e.g. "2026"), matching the spreadsheet layout.
|
||||
defaultRange := fmt.Sprintf("%d!A1:G100", time.Now().Year())
|
||||
|
||||
// SheetRange is left empty when unset so the fetcher can resolve the
|
||||
// current-year tab at fetch time (see resolveRange). Freezing the default
|
||||
// here would pin a long-running process to its start year and silently miss
|
||||
// the next year's tab after the New Year rollover.
|
||||
cfg := Config{
|
||||
SpreadsheetID: os.Getenv("GOOGLE_SHEETS_ID"),
|
||||
SheetRange: getenvDefault("SHEET_RANGE", defaultRange),
|
||||
SheetRange: os.Getenv("SHEET_RANGE"),
|
||||
CachePath: getenvDefault("ROTA_CACHE_PATH", "rota.json"),
|
||||
}
|
||||
if cfg.SpreadsheetID == "" {
|
||||
|
|
@ -122,30 +123,72 @@ func getenvDefault(key, fallback string) string {
|
|||
// FetchFunc retrieves and parses the current rota from upstream (the Sheet).
|
||||
type FetchFunc func(ctx context.Context) (Rota, error)
|
||||
|
||||
// errEmptyRota signals that a fetch succeeded but returned no duty weeks, so the
|
||||
// cache was deliberately left untouched rather than clobbered with empty data.
|
||||
var errEmptyRota = errors.New("fetched rota has no duty weeks; keeping cached data")
|
||||
|
||||
// defaultStaleAfter is how long the cache may go without a successful sync
|
||||
// before Sync escalates from a per-failure warning to a stale-cache error. At
|
||||
// the default 30m sync interval this is ~12 consecutive failures.
|
||||
const defaultStaleAfter = 6 * time.Hour
|
||||
|
||||
// Syncer refreshes a Store from an upstream fetch.
|
||||
type Syncer struct {
|
||||
store *Store
|
||||
fetch FetchFunc
|
||||
now func() time.Time
|
||||
store *Store
|
||||
fetch FetchFunc
|
||||
now func() time.Time
|
||||
staleAfter time.Duration
|
||||
}
|
||||
|
||||
// NewSyncer wires a Store to a fetch function.
|
||||
func NewSyncer(store *Store, fetch FetchFunc) *Syncer {
|
||||
return &Syncer{store: store, fetch: fetch, now: time.Now}
|
||||
return &Syncer{store: store, fetch: fetch, now: time.Now, staleAfter: defaultStaleAfter}
|
||||
}
|
||||
|
||||
// Sync fetches once. On success it replaces the cache; on failure it logs a
|
||||
// warning, leaves the last good cache in place, and returns the error so a
|
||||
// caller (e.g. a manual /sync command) can report it.
|
||||
// Sync fetches once. On success with a non-empty rota it replaces the cache.
|
||||
// When the fetch fails — or succeeds but returns zero weeks (e.g. a freshly
|
||||
// created, not-yet-filled year tab) — it leaves the last-good cache in place,
|
||||
// warns, and returns an error so a caller (e.g. a manual /sync) can report it.
|
||||
// A guard against the empty case matters because the range targets the current
|
||||
// year's tab, which around New Year may exist but be empty; replacing the cache
|
||||
// with it would silently wipe the schedule.
|
||||
func (sy *Syncer) Sync(ctx context.Context) error {
|
||||
rota, err := sy.fetch(ctx)
|
||||
if err != nil {
|
||||
slog.Warn("sync fetch failed, keeping cached data", "err", err)
|
||||
sy.warnIfStale("fetch error")
|
||||
return err
|
||||
}
|
||||
if len(rota.Weeks) == 0 {
|
||||
slog.Warn("sync returned no duty weeks, keeping cached data")
|
||||
sy.warnIfStale("empty rota")
|
||||
return errEmptyRota
|
||||
}
|
||||
return sy.store.Replace(rota, sy.now())
|
||||
}
|
||||
|
||||
// warnIfStale emits a loud error when the cache has gone too long without a
|
||||
// successful sync, turning a silently-stale cache (which is not a crash, so
|
||||
// process-level supervision won't catch it) into a visible log signal.
|
||||
func (sy *Syncer) warnIfStale(reason string) {
|
||||
last := sy.store.LastSynced()
|
||||
if age, stale := staleAge(last, sy.now(), sy.staleAfter); stale {
|
||||
slog.Error("rota cache is stale; sync has not succeeded recently",
|
||||
"age", age.Round(time.Minute), "last_synced", last.Format(time.RFC3339), "reason", reason)
|
||||
}
|
||||
}
|
||||
|
||||
// staleAge reports the cache's age and whether it exceeds threshold. A
|
||||
// never-synced cache (zero last) is not "stale" — there is no good data to be
|
||||
// behind — so it reports false.
|
||||
func staleAge(last, now time.Time, threshold time.Duration) (time.Duration, bool) {
|
||||
if last.IsZero() {
|
||||
return 0, false
|
||||
}
|
||||
age := now.Sub(last)
|
||||
return age, age >= threshold
|
||||
}
|
||||
|
||||
// Run polls Sync on the given interval until ctx is canceled.
|
||||
func (sy *Syncer) Run(ctx context.Context, interval time.Duration) {
|
||||
ticker := time.NewTicker(interval)
|
||||
|
|
@ -165,12 +208,15 @@ func (sy *Syncer) Run(ctx context.Context, interval time.Duration) {
|
|||
// the parent context is canceled at shutdown.
|
||||
const fetchTimeout = 30 * time.Second
|
||||
|
||||
// SheetFetcher builds a FetchFunc that reads the rota from Google Sheets.
|
||||
// SheetFetcher builds a FetchFunc that reads the rota from Google Sheets. When
|
||||
// readRange is empty the range is resolved per fetch (see resolveRange), so a
|
||||
// long-running process follows the current-year tab across a New Year rollover
|
||||
// instead of pinning the year it started in.
|
||||
func SheetFetcher(spreadsheetID, readRange string) FetchFunc {
|
||||
return func(ctx context.Context) (Rota, error) {
|
||||
ctx, cancel := context.WithTimeout(ctx, fetchTimeout)
|
||||
defer cancel()
|
||||
rows, err := ReadSheet(ctx, spreadsheetID, readRange)
|
||||
rows, err := ReadSheet(ctx, spreadsheetID, resolveRange(readRange, time.Now()))
|
||||
if err != nil {
|
||||
return Rota{}, err
|
||||
}
|
||||
|
|
@ -178,6 +224,17 @@ func SheetFetcher(spreadsheetID, readRange string) FetchFunc {
|
|||
}
|
||||
}
|
||||
|
||||
// resolveRange returns the explicit range when set, otherwise the default that
|
||||
// targets the current year's tab (sheets are named per year, e.g. "2026").
|
||||
// Resolving against now (not a startup-captured value) is what lets the range
|
||||
// advance to the next year's tab automatically.
|
||||
func resolveRange(explicit string, now time.Time) string {
|
||||
if explicit != "" {
|
||||
return explicit
|
||||
}
|
||||
return fmt.Sprintf("%d!A1:G100", now.Year())
|
||||
}
|
||||
|
||||
// Service ties the Store and Syncer together for the bot to consume.
|
||||
type Service struct {
|
||||
Store *Store
|
||||
|
|
|
|||
|
|
@ -3,7 +3,6 @@ package dyzurbot
|
|||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
|
@ -22,9 +21,10 @@ func TestLoadConfigDefaults(t *testing.T) {
|
|||
if cfg.SpreadsheetID != "sheet-123" {
|
||||
t.Errorf("SpreadsheetID: got %q", cfg.SpreadsheetID)
|
||||
}
|
||||
wantRange := fmt.Sprintf("%d!A1:G100", time.Now().Year())
|
||||
if cfg.SheetRange != wantRange {
|
||||
t.Errorf("SheetRange default: got %q, want %q", cfg.SheetRange, wantRange)
|
||||
// An unset SHEET_RANGE stays empty in config; the fetcher resolves the
|
||||
// current-year tab per request so the year isn't frozen at startup.
|
||||
if cfg.SheetRange != "" {
|
||||
t.Errorf("SheetRange default: got %q, want empty (resolved at fetch time)", cfg.SheetRange)
|
||||
}
|
||||
if cfg.CachePath != "rota.json" {
|
||||
t.Errorf("CachePath default: got %q", cfg.CachePath)
|
||||
|
|
@ -152,6 +152,58 @@ func TestSyncSuccessReplacesCache(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestResolveRange(t *testing.T) {
|
||||
now := time.Date(2027, 1, 1, 0, 0, 0, 0, time.UTC)
|
||||
|
||||
// Empty range resolves to the current year's tab, evaluated against now —
|
||||
// this is what advances the range after a New Year rollover.
|
||||
if got, want := resolveRange("", now), "2027!A1:G100"; got != want {
|
||||
t.Errorf("resolveRange(\"\"): got %q, want %q", got, want)
|
||||
}
|
||||
// An explicit range is honored verbatim regardless of the year.
|
||||
if got, want := resolveRange("Dyżury!A1:C30", now), "Dyżury!A1:C30"; got != want {
|
||||
t.Errorf("resolveRange(explicit): got %q, want %q", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStaleAge(t *testing.T) {
|
||||
now := time.Date(2026, 6, 17, 12, 0, 0, 0, time.UTC)
|
||||
threshold := 6 * time.Hour
|
||||
|
||||
// A never-synced cache (zero time) is not stale: there is no good data behind.
|
||||
if _, stale := staleAge(time.Time{}, now, threshold); stale {
|
||||
t.Error("zero last_synced should not be stale")
|
||||
}
|
||||
// Within the threshold: fresh.
|
||||
if _, stale := staleAge(now.Add(-time.Hour), now, threshold); stale {
|
||||
t.Error("1h-old cache should not be stale under a 6h threshold")
|
||||
}
|
||||
// Past the threshold: stale, with the age reported.
|
||||
if age, stale := staleAge(now.Add(-7*time.Hour), now, threshold); !stale || age != 7*time.Hour {
|
||||
t.Errorf("7h-old cache: stale=%v age=%v, want stale age=7h", stale, age)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSyncEmptyRotaPreservesCache guards the New-Year regression: a fetch that
|
||||
// succeeds but returns no weeks (e.g. a freshly created, empty year tab) must
|
||||
// not clobber the last-good cache.
|
||||
func TestSyncEmptyRotaPreservesCache(t *testing.T) {
|
||||
store := NewStore(filepath.Join(t.TempDir(), "rota.json"))
|
||||
good := time.Date(2026, 6, 17, 20, 0, 0, 0, time.UTC)
|
||||
if err := store.Replace(sampleRota(), good); err != nil {
|
||||
t.Fatalf("seed Replace: %v", err)
|
||||
}
|
||||
|
||||
sy := NewSyncer(store, func(context.Context) (Rota, error) { return Rota{}, nil })
|
||||
if err := sy.Sync(context.Background()); !errors.Is(err, errEmptyRota) {
|
||||
t.Fatalf("Sync error: got %v, want errEmptyRota", err)
|
||||
}
|
||||
rota, ts := store.Snapshot()
|
||||
if len(rota.People) != 1 || !ts.Equal(good) {
|
||||
t.Errorf("cache should be untouched: people=%d ts=%v", len(rota.People), ts)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSyncFailurePreservesCache(t *testing.T) {
|
||||
store := NewStore(filepath.Join(t.TempDir(), "rota.json"))
|
||||
good := time.Date(2026, 6, 17, 20, 0, 0, 0, time.UTC)
|
||||
|
|
|
|||
9
store.go
9
store.go
|
|
@ -56,6 +56,15 @@ func (s *Store) Load() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// LastSynced returns the time of the last successful cache replacement, or the
|
||||
// zero time if the cache has never been populated (cold start). It is a cheap
|
||||
// read for staleness checks that don't need the rota payload Snapshot copies.
|
||||
func (s *Store) LastSynced() time.Time {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
return s.lastSynced
|
||||
}
|
||||
|
||||
// Snapshot returns a copy of the cached rota plus the last successful sync
|
||||
// time. The copy lets callers read without holding the lock and shields the
|
||||
// cache from caller-side mutation.
|
||||
|
|
|
|||
33
telegram.go
33
telegram.go
|
|
@ -5,6 +5,7 @@ import (
|
|||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/go-telegram/bot"
|
||||
|
|
@ -60,10 +61,37 @@ func PublishCommands(ctx context.Context, b *bot.Bot) error {
|
|||
return err
|
||||
}
|
||||
|
||||
// syncCooldown is the minimum gap between manual /sync triggers. It throttles
|
||||
// spam — each sync is an outbound Google Sheets call — without getting in the
|
||||
// way of legitimate use (the background syncer already refreshes on its own
|
||||
// interval, so manual syncs are only for impatience).
|
||||
const syncCooldown = time.Minute
|
||||
|
||||
// botHandlers holds the dependencies the command handlers close over.
|
||||
type botHandlers struct {
|
||||
svc *Service
|
||||
now func() time.Time
|
||||
|
||||
// mu guards lastSync: Telegram dispatches each update in its own goroutine,
|
||||
// so concurrent /sync commands race on the cooldown check.
|
||||
mu sync.Mutex
|
||||
lastSync time.Time // zero until the first /sync attempt
|
||||
}
|
||||
|
||||
// allowSync reports whether a /sync may run now, recording the attempt time
|
||||
// when it may. The clock is reset on every allowed attempt (not just successful
|
||||
// syncs) so a failing endpoint can't be hammered. When denied it returns how
|
||||
// long the caller must wait.
|
||||
func (h *botHandlers) allowSync(now time.Time) (bool, time.Duration) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if !h.lastSync.IsZero() {
|
||||
if elapsed := now.Sub(h.lastSync); elapsed < syncCooldown {
|
||||
return false, syncCooldown - elapsed
|
||||
}
|
||||
}
|
||||
h.lastSync = now
|
||||
return true, 0
|
||||
}
|
||||
|
||||
// ktoSprzata answers /kto_sprzata with this week's duty from the cache.
|
||||
|
|
@ -76,6 +104,11 @@ func (h *botHandlers) ktoSprzata(ctx context.Context, b *bot.Bot, update *models
|
|||
// sync answers /sync by triggering a fetch and reporting the outcome. Sync
|
||||
// keeps the last-good cache on failure, so an error here is informational.
|
||||
func (h *botHandlers) sync(ctx context.Context, b *bot.Bot, update *models.Update) {
|
||||
if ok, wait := h.allowSync(h.now()); !ok {
|
||||
h.reply(ctx, b, update, fmt.Sprintf(
|
||||
"Synchronizowano niedawno. Spróbuj ponownie za %ds.", int(wait.Seconds())+1))
|
||||
return
|
||||
}
|
||||
if err := h.svc.Syncer.Sync(ctx); err != nil {
|
||||
// Log the detailed error; show the chat a generic message so internal
|
||||
// details (API endpoints, keys in error strings) don't leak.
|
||||
|
|
|
|||
|
|
@ -24,6 +24,24 @@ func day(t *testing.T, d string) time.Time {
|
|||
return tm
|
||||
}
|
||||
|
||||
func TestAllowSync(t *testing.T) {
|
||||
base := time.Date(2026, 6, 17, 12, 0, 0, 0, time.UTC)
|
||||
h := &botHandlers{}
|
||||
|
||||
// First attempt is always allowed and starts the cooldown.
|
||||
if ok, _ := h.allowSync(base); !ok {
|
||||
t.Fatal("first /sync should be allowed")
|
||||
}
|
||||
// A second attempt inside the window is denied, reporting the remaining wait.
|
||||
if ok, wait := h.allowSync(base.Add(20 * time.Second)); ok || wait != syncCooldown-20*time.Second {
|
||||
t.Errorf("within cooldown: ok=%v wait=%v, want denied with wait=%v", ok, wait, syncCooldown-20*time.Second)
|
||||
}
|
||||
// Once the cooldown elapses, /sync is allowed again.
|
||||
if ok, _ := h.allowSync(base.Add(syncCooldown)); !ok {
|
||||
t.Error("attempt after cooldown should be allowed")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCurrentDuty(t *testing.T) {
|
||||
rota := Rota{Weeks: []DutyWeek{
|
||||
week(t, "2026-06-15", "Ala", "Bartek"),
|
||||
|
|
|
|||
Loading…
Reference in New Issue