diff --git a/internal/feedwatch/feedwatch.go b/internal/feedwatch/feedwatch.go new file mode 100644 index 0000000..0ef3459 --- /dev/null +++ b/internal/feedwatch/feedwatch.go @@ -0,0 +1,237 @@ +// Package feedwatch tells the owner when a feed breaks, when it comes back, and +// when it goes quiet. +// +// The poller has always recorded every poll's outcome in source_health, and +// /status shows it, but only to someone who goes and looks. Feeds died that way +// for months: Siliconera and Digital Foundry had been refused on every poll +// thousands of times over before anyone noticed. +// +// The watcher reads source_health on a timer rather than hooking the pollers, so +// that one pass can batch every change into a single message, and so that what +// has already been said lives in the database (down_alerted_at, +// quiet_alerted_at) instead of in a goroutine that forgets it on restart. Each +// episode is reported once: a feed that goes down is announced, stays silent +// while it stays down, and is announced again when it comes back. +package feedwatch + +import ( + "context" + "fmt" + "log/slog" + "strings" + "time" + + "pete/internal/config" + "pete/internal/storage" +) + +const ( + // DownAfter is how many failed polls in a row make a feed "down". One or + // two failures is weather; five is a feed that has stopped working. + DownAfter = 5 + // QuietAfter is how long a feed can fetch fine and yield nothing new before + // it is "quiet". Every healthy source Pete carries produces something within + // about four days. + QuietAfter = 7 * 24 * time.Hour + // Every is how often the watcher looks. + Every = 15 * time.Minute +) + +// Report is what one pass found that the owner hasn't been told yet. +type Report struct { + Down []storage.FeedAlertRow + Back []storage.FeedAlertRow + Quiet []storage.FeedAlertRow + // quietOver are feeds whose quiet spell has ended. They are cleared without + // a message: a feed that simply started posting again needs no fanfare. + quietOver []string +} + +// Empty reports whether there is nothing to say. +func (r Report) Empty() bool { return len(r.Down)+len(r.Back)+len(r.Quiet) == 0 } + +// Check reads source_health and works out what has changed, without recording +// anything. Only sources that are enabled in the config are considered, so a +// feed that has been switched off or removed never alerts. +func Check(now time.Time, sources []config.SourceConfig) (Report, error) { + enabled := make(map[string]bool, len(sources)) + for _, s := range sources { + if s.Enabled { + enabled[s.Name] = true + } + } + rows, err := storage.ListFeedAlertRows() + if err != nil { + return Report{}, err + } + var r Report + quietCutoff := now.Add(-QuietAfter).Unix() + for _, row := range rows { + if !enabled[row.Source] { + continue + } + down := row.ConsecutiveFailures >= DownAfter + switch { + case down && row.DownAlertedAt == 0: + r.Down = append(r.Down, row) + // Back means "has fetched since we said it was down", not "is fetching + // right now": a feed that recovers and breaks again between two passes + // still gets its all-clear, and then a fresh down alert next pass. + case row.DownAlertedAt != 0 && row.LastSuccessAt > row.DownAlertedAt: + r.Back = append(r.Back, row) + } + // A feed that isn't fetching can't be judged on whether it has news; + // "down" already covers it. + if row.ConsecutiveFailures > 0 { + continue + } + quiet := row.LastNewAt > 0 && row.LastNewAt < quietCutoff + switch { + case quiet && row.QuietAlertedAt == 0: + r.Quiet = append(r.Quiet, row) + case !quiet && row.QuietAlertedAt != 0: + r.quietOver = append(r.quietOver, row.Source) + } + } + return r, nil +} + +// Commit records that the owner has now been told everything in r. +func Commit(now time.Time, r Report) error { + at := now.Unix() + for _, row := range r.Down { + if err := storage.SetFeedDownAlerted(row.Source, at); err != nil { + return err + } + } + for _, row := range r.Back { + if err := storage.SetFeedDownAlerted(row.Source, 0); err != nil { + return err + } + } + for _, row := range r.Quiet { + if err := storage.SetFeedQuietAlerted(row.Source, at); err != nil { + return err + } + } + for _, src := range r.quietOver { + if err := storage.SetFeedQuietAlerted(src, 0); err != nil { + return err + } + } + return nil +} + +// Message renders r as the plain-text note Pete sends the owner. +func Message(now time.Time, r Report) string { + var b strings.Builder + b.WriteString("Feed check") + if len(r.Down) > 0 { + b.WriteString("\n\nNot fetching:") + for _, row := range r.Down { + since := "has never fetched successfully" + if row.LastSuccessAt > 0 { + since = "last worked " + ago(now, row.LastSuccessAt) + } + fmt.Fprintf(&b, "\n• %s: %s failed polls in a row, %s. Last error: %s", + row.Source, count(row.ConsecutiveFailures), since, short(row.LastError)) + } + } + if len(r.Back) > 0 { + b.WriteString("\n\nWorking again:") + for _, row := range r.Back { + fmt.Fprintf(&b, "\n• %s is fetching again.", row.Source) + } + } + if len(r.Quiet) > 0 { + b.WriteString("\n\nGone quiet:") + for _, row := range r.Quiet { + fmt.Fprintf(&b, "\n• %s fetches fine but hasn't had anything new in %s.", + row.Source, span(now, row.LastNewAt)) + } + } + return b.String() +} + +// Send delivers a message to the owner. An error means it didn't arrive, and +// the watcher will try again next pass rather than mark it said. +type Send func(ctx context.Context, msg string) error + +// Run checks once shortly after startup (so pollers have had a first go) and +// then every Every until ctx ends. +func Run(ctx context.Context, sources []config.SourceConfig, send Send) { + first := time.NewTimer(2 * time.Minute) + defer first.Stop() + t := time.NewTicker(Every) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-first.C: + case <-t.C: + } + pass(ctx, time.Now(), sources, send) + } +} + +func pass(ctx context.Context, now time.Time, sources []config.SourceConfig, send Send) { + r, err := Check(now, sources) + if err != nil { + slog.Error("feedwatch: check failed", "err", err) + return + } + if r.Empty() { + // Still commit: a quiet spell may have ended silently. + if err := Commit(now, r); err != nil { + slog.Error("feedwatch: commit failed", "err", err) + } + return + } + msg := Message(now, r) + if err := send(ctx, msg); err != nil { + slog.Error("feedwatch: alert not delivered, will retry", "err", err) + return + } + slog.Info("feedwatch: alert sent", "down", len(r.Down), "back", len(r.Back), "quiet", len(r.Quiet)) + if err := Commit(now, r); err != nil { + slog.Error("feedwatch: commit failed", "err", err) + } +} + +// span renders the time since unix as a rough human length ("9 days"). +func span(now time.Time, unix int64) string { + d := now.Sub(time.Unix(unix, 0)) + switch { + case d < 2*time.Hour: + return fmt.Sprintf("%d minutes", int(d.Minutes())) + case d < 48*time.Hour: + return fmt.Sprintf("%d hours", int(d.Hours())) + default: + return fmt.Sprintf("%d days", int(d.Hours()/24)) + } +} + +// ago is span as a point in the past ("9 days ago"). +func ago(now time.Time, unix int64) string { return span(now, unix) + " ago" } + +// count renders n with thousands separators. +func count(n int) string { + s := fmt.Sprint(n) + for i := len(s) - 3; i > 0; i -= 3 { + s = s[:i] + "," + s[i:] + } + return s +} + +// short trims an error to one readable line. +func short(s string) string { + s = strings.TrimSpace(strings.ReplaceAll(s, "\n", " ")) + if s == "" { + return "none recorded" + } + if len(s) > 140 { + s = s[:140] + "…" + } + return s +} diff --git a/internal/feedwatch/feedwatch_test.go b/internal/feedwatch/feedwatch_test.go new file mode 100644 index 0000000..8cd201f --- /dev/null +++ b/internal/feedwatch/feedwatch_test.go @@ -0,0 +1,179 @@ +package feedwatch + +import ( + "context" + "errors" + "path/filepath" + "strings" + "testing" + "time" + + "pete/internal/config" + "pete/internal/storage" +) + +func setup(t *testing.T) { + t.Helper() + storage.Close() + if err := storage.Init(filepath.Join(t.TempDir(), "feedwatch.db")); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { storage.Close() }) +} + +func src(names ...string) []config.SourceConfig { + var out []config.SourceConfig + for _, n := range names { + out = append(out, config.SourceConfig{Name: n, Enabled: true}) + } + return out +} + +func fail(name string, n int) { + for range n { + storage.RecordPollResult(name, false, 0, errors.New("feed fetch https://x/feed: status 403")) + } +} + +// setLastNew backdates a source's last new story. +func setLastNew(t *testing.T, name string, at time.Time) { + t.Helper() + if _, err := storage.Get().Exec(`UPDATE source_health SET last_new_at = ? WHERE source = ?`, at.Unix(), name); err != nil { + t.Fatal(err) + } +} + +// capture is a Send that records what it was asked to deliver. +type capture struct { + msgs []string + err error +} + +func (c *capture) send(_ context.Context, msg string) error { + if c.err != nil { + return c.err + } + c.msgs = append(c.msgs, msg) + return nil +} + +func TestDownIsReportedOnceAndBackIsReported(t *testing.T) { + setup(t) + sources := src("Broken", "Fine") + now := time.Now() + + storage.RecordPollResult("Fine", true, 10, nil) + fail("Broken", DownAfter-1) + + c := &capture{} + pass(context.Background(), now, sources, c.send) + if len(c.msgs) != 0 { + t.Fatalf("alerted below the threshold: %q", c.msgs) + } + + fail("Broken", 1) + pass(context.Background(), now, sources, c.send) + if len(c.msgs) != 1 || !strings.Contains(c.msgs[0], "Broken: 5 failed polls in a row, has never fetched successfully") { + t.Fatalf("down alert = %q", c.msgs) + } + if strings.Contains(c.msgs[0], "Fine") { + t.Errorf("a healthy feed was mentioned: %q", c.msgs[0]) + } + + // Still down on the next pass: nothing new to say. + fail("Broken", 3) + pass(context.Background(), now.Add(Every), sources, c.send) + if len(c.msgs) != 1 { + t.Fatalf("down alert repeated: %q", c.msgs[1:]) + } + + // It comes back. The success has to be stamped after the alert, so step + // past the second the alert was recorded in. + time.Sleep(1100 * time.Millisecond) + storage.RecordPollResult("Broken", true, 5, nil) + pass(context.Background(), time.Now(), sources, c.send) + if len(c.msgs) != 2 || !strings.Contains(c.msgs[1], "Broken is fetching again") { + t.Fatalf("recovery = %q", c.msgs) + } + pass(context.Background(), time.Now(), sources, c.send) + if len(c.msgs) != 2 { + t.Fatalf("recovery repeated: %q", c.msgs[2:]) + } +} + +func TestUndeliveredAlertIsRetried(t *testing.T) { + setup(t) + sources := src("Broken") + fail("Broken", DownAfter) + + c := &capture{err: errors.New("homeserver unreachable")} + pass(context.Background(), time.Now(), sources, c.send) + + c.err = nil + pass(context.Background(), time.Now(), sources, c.send) + if len(c.msgs) != 1 || !strings.Contains(c.msgs[0], "Broken") { + t.Fatalf("alert after a failed send = %q, want it delivered on the retry", c.msgs) + } +} + +func TestQuietFeedIsReportedOnce(t *testing.T) { + setup(t) + sources := src("Sleepy", "Busy") + now := time.Now() + storage.RecordPollResult("Sleepy", true, 10, nil) + storage.RecordPollResult("Busy", true, 10, nil) + setLastNew(t, "Sleepy", now.Add(-9*24*time.Hour)) + + c := &capture{} + pass(context.Background(), now, sources, c.send) + if len(c.msgs) != 1 || !strings.Contains(c.msgs[0], "Sleepy fetches fine but hasn't had anything new in 9 days") { + t.Fatalf("quiet alert = %q", c.msgs) + } + if strings.Contains(c.msgs[0], "Busy") { + t.Errorf("a busy feed was called quiet: %q", c.msgs[0]) + } + pass(context.Background(), now.Add(Every), sources, c.send) + if len(c.msgs) != 1 { + t.Fatalf("quiet alert repeated: %q", c.msgs[1:]) + } + + // New stories end the quiet spell without a message, and a later quiet + // spell is a new episode that gets reported again. + storage.RecordNewStories("Sleepy", 2) + pass(context.Background(), now, sources, c.send) + if len(c.msgs) != 1 { + t.Fatalf("end of a quiet spell was announced: %q", c.msgs[1:]) + } + setLastNew(t, "Sleepy", now.Add(-8*24*time.Hour)) + pass(context.Background(), now, sources, c.send) + if len(c.msgs) != 2 { + t.Fatalf("second quiet spell not reported: %q", c.msgs) + } +} + +// A feed that has just been added has no history, so it must not be called +// quiet before it has had a full window to produce something. +func TestNewSourceGetsAGracePeriod(t *testing.T) { + setup(t) + storage.RecordPollResult("Fresh", true, 0, nil) + r, err := Check(time.Now().Add(QuietAfter-time.Hour), src("Fresh")) + if err != nil { + t.Fatal(err) + } + if !r.Empty() { + t.Fatalf("new source reported inside its grace period: %+v", r) + } +} + +func TestDisabledSourcesNeverAlert(t *testing.T) { + setup(t) + fail("Retired", DownAfter*2) + sources := []config.SourceConfig{{Name: "Retired", Enabled: false}} + r, err := Check(time.Now(), sources) + if err != nil { + t.Fatal(err) + } + if !r.Empty() { + t.Fatalf("disabled source alerted: %+v", r) + } +} diff --git a/internal/ingestion/poller.go b/internal/ingestion/poller.go index c99348c..7e054a5 100644 --- a/internal/ingestion/poller.go +++ b/internal/ingestion/poller.go @@ -265,6 +265,7 @@ func (p *Poller) pollOnceWithErr(ctx context.Context, src config.SourceConfig) e if newCount > 0 { slog.Info("ingested new stories", "source", src.Name, "count", newCount) + storage.RecordNewStories(src.Name, newCount) } return nil diff --git a/internal/matrix/client.go b/internal/matrix/client.go index ef73eab..fa155c4 100644 --- a/internal/matrix/client.go +++ b/internal/matrix/client.go @@ -508,23 +508,26 @@ func (c *Client) sendImage(ctx context.Context, roomID id.RoomID, imageURL strin return err } -// SendAdminWarning sends a warning message to the admin room, if configured. -func (c *Client) SendAdminWarning(msg string) { +// SendAdminNotice sends a notice to the admin room. It returns an error when no +// admin room is configured or the send fails, so a caller that must not lose +// the message (the feed watcher) can try again later. +func (c *Client) SendAdminNotice(ctx context.Context, msg string) error { if c.adminRoom == nil { - slog.Warn("admin warning (no admin room configured)", "msg", msg) - return + return fmt.Errorf("no admin room configured") } - content := &event.MessageEventContent{ - MsgType: event.MsgText, + MsgType: event.MsgNotice, Body: msg, } - - if _, err := c.mx.SendMessageEvent(context.Background(), *c.adminRoom, event.EventMessage, content); err != nil { - slog.Error("failed to send admin warning", "err", err, "msg", msg) + if _, err := c.mx.SendMessageEvent(ctx, *c.adminRoom, event.EventMessage, content); err != nil { + return fmt.Errorf("send admin notice: %w", err) } + return nil } +// HasAdminRoom reports whether an admin room is configured. +func (c *Client) HasAdminRoom() bool { return c.adminRoom != nil } + // ChannelRoomID returns the room ID for a named channel. // Returns false if the channel is not configured or has an empty room ID. func (c *Client) ChannelRoomID(channel string) (id.RoomID, bool) { diff --git a/internal/storage/db.go b/internal/storage/db.go index 0190679..3cecc30 100644 --- a/internal/storage/db.go +++ b/internal/storage/db.go @@ -96,6 +96,14 @@ func runMigrations(d *sql.DB) error { addColumnIfMissing(d, "post_log", "url_canonical", "TEXT") addColumnIfMissing(d, "post_log", "forced", "INTEGER NOT NULL DEFAULT 0") addColumnIfMissing(d, "round_robin_state", "last_channel", "TEXT") + // Feed alerts. last_new_at is when a source last produced a story Pete kept; + // the two *_alerted_at stamps remember that the owner has already been told a + // feed is down or has gone quiet, so an alert fires once per episode and + // survives a restart. See backfillLastNewAt for what existing rows start with. + addColumnIfMissing(d, "source_health", "last_new_at", "INTEGER") + addColumnIfMissing(d, "source_health", "down_alerted_at", "INTEGER") + addColumnIfMissing(d, "source_health", "quiet_alerted_at", "INTEGER") + backfillLastNewAt(d) // Occupancy of a shared table. Rows written before the casino went multiplayer // are solo games and read as NULL, which is exactly what they are. addColumnIfMissing(d, "game_live_hands", "table_id", "TEXT") diff --git a/internal/storage/sourcehealth.go b/internal/storage/sourcehealth.go index 6d02826..a1fcdae 100644 --- a/internal/storage/sourcehealth.go +++ b/internal/storage/sourcehealth.go @@ -3,6 +3,7 @@ package storage import ( "database/sql" "fmt" + "log/slog" ) // SourceHealth is one source's persisted poll health, as written by the poller. @@ -30,15 +31,17 @@ type SourceContentStat struct { // RecordPollResult upserts a source's health row after a poll attempt. On // success it clears the error and failure counter and records the item count; // on failure it bumps consecutive_failures and stores the message while -// preserving the last successful timestamp and item count. Fire-and-forget: +// preserving the last successful timestamp and item count. A brand-new row +// starts its last_new_at clock at the first poll, so a just-added source gets a +// full quiet window before it can be called quiet. Fire-and-forget: // a failure here must never disrupt polling, so errors are logged and swallowed. func RecordPollResult(source string, ok bool, itemCount int, pollErr error) { now := nowUnix() if ok { exec("record poll success", `INSERT INTO source_health - (source, last_poll_at, last_success_at, last_error, consecutive_failures, last_item_count, updated_at) - VALUES (?, ?, ?, '', 0, ?, ?) + (source, last_poll_at, last_success_at, last_error, consecutive_failures, last_item_count, updated_at, last_new_at) + VALUES (?, ?, ?, '', 0, ?, ?, ?) ON CONFLICT(source) DO UPDATE SET last_poll_at = excluded.last_poll_at, last_success_at = excluded.last_success_at, @@ -46,7 +49,7 @@ func RecordPollResult(source string, ok bool, itemCount int, pollErr error) { consecutive_failures = 0, last_item_count = excluded.last_item_count, updated_at = excluded.updated_at`, - source, now, now, itemCount, now) + source, now, now, itemCount, now, now) return } msg := "" @@ -55,14 +58,14 @@ func RecordPollResult(source string, ok bool, itemCount int, pollErr error) { } exec("record poll failure", `INSERT INTO source_health - (source, last_poll_at, last_success_at, last_error, consecutive_failures, last_item_count, updated_at) - VALUES (?, ?, NULL, ?, 1, 0, ?) + (source, last_poll_at, last_success_at, last_error, consecutive_failures, last_item_count, updated_at, last_new_at) + VALUES (?, ?, NULL, ?, 1, 0, ?, ?) ON CONFLICT(source) DO UPDATE SET last_poll_at = excluded.last_poll_at, last_error = excluded.last_error, consecutive_failures = source_health.consecutive_failures + 1, updated_at = excluded.updated_at`, - source, now, msg, now) + source, now, msg, now, now) } // ListSourceHealth returns the poll-health row for every source that has been @@ -147,3 +150,79 @@ func SourceContentStats() (map[string]SourceContentStat, error) { } return out, prows.Err() } + +// backfillLastNewAt gives rows that predate last_new_at a starting value: the +// newest story still retained for that source, or, when none is, the time of +// the migration. The second case is a grace period rather than a guess: a +// source with no retained stories might be long dead or might have been added +// yesterday, and nothing on disk says which, so it gets a full quiet window +// before it can be called quiet. +func backfillLastNewAt(d *sql.DB) { + if _, err := d.Exec( + `UPDATE source_health + SET last_new_at = COALESCE( + (SELECT MAX(seen_at) FROM stories WHERE stories.source = source_health.source), + ?) + WHERE last_new_at IS NULL`, nowUnix()); err != nil { + slog.Error("backfill source_health.last_new_at failed", "err", err) + } +} + +// RecordNewStories notes that a poll of source kept n genuinely new stories. +// Only a non-zero n moves last_new_at: a poll that fetched fine but found +// nothing new is exactly what "gone quiet" is made of. +func RecordNewStories(source string, n int) { + if n <= 0 { + return + } + exec("record new stories", + `UPDATE source_health SET last_new_at = ? WHERE source = ?`, nowUnix(), source) +} + +// FeedAlertRow is the slice of source_health the feed watcher reads. +type FeedAlertRow struct { + Source string + ConsecutiveFailures int + LastError string + LastSuccessAt int64 // 0 = never fetched successfully + LastNewAt int64 // 0 = unknown + DownAlertedAt int64 // 0 = owner not told it's down + QuietAlertedAt int64 // 0 = owner not told it's quiet +} + +// ListFeedAlertRows returns the alert-relevant health of every polled source. +func ListFeedAlertRows() ([]FeedAlertRow, error) { + rows, err := Get().Query( + `SELECT source, consecutive_failures, COALESCE(last_error, ''), + COALESCE(last_success_at, 0), COALESCE(last_new_at, 0), + COALESCE(down_alerted_at, 0), COALESCE(quiet_alerted_at, 0) + FROM source_health ORDER BY source`) + if err != nil { + return nil, fmt.Errorf("list feed alert rows: %w", err) + } + defer rows.Close() + var out []FeedAlertRow + for rows.Next() { + var r FeedAlertRow + if err := rows.Scan(&r.Source, &r.ConsecutiveFailures, &r.LastError, + &r.LastSuccessAt, &r.LastNewAt, &r.DownAlertedAt, &r.QuietAlertedAt); err != nil { + return nil, err + } + out = append(out, r) + } + return out, rows.Err() +} + +// SetFeedDownAlerted records (at > 0) or clears (at == 0) that the owner has +// been told source is down. +func SetFeedDownAlerted(source string, at int64) error { + _, err := Get().Exec(`UPDATE source_health SET down_alerted_at = NULLIF(?, 0) WHERE source = ?`, at, source) + return err +} + +// SetFeedQuietAlerted records (at > 0) or clears (at == 0) that the owner has +// been told source has gone quiet. +func SetFeedQuietAlerted(source string, at int64) error { + _, err := Get().Exec(`UPDATE source_health SET quiet_alerted_at = NULLIF(?, 0) WHERE source = ?`, at, source) + return err +} diff --git a/main.go b/main.go index b0ddb2c..7962b41 100644 --- a/main.go +++ b/main.go @@ -14,6 +14,7 @@ import ( "time" "pete/internal/config" + "pete/internal/feedwatch" "pete/internal/ingestion" "pete/internal/matrix" "pete/internal/poster" @@ -243,6 +244,15 @@ func main() { // Start pollers poller := ingestion.NewPoller(cfg.Sources, processItem) poller.Start(ctx) + + // Tell the owner when a feed stops fetching, comes back, or goes quiet. + // Without an admin room there is nowhere to say it, and marking alerts as + // sent into the void would mean they never arrive once a room is set. + if mx.HasAdminRoom() { + go feedwatch.Run(ctx, cfg.Sources, mx.SendAdminNotice) + } else { + slog.Warn("feed alerts off: set matrix.admin_room to have Pete report broken or quiet feeds") + } slog.Info("pollers started") if postingEnabled && roundRobinMode {