feedwatch: tell the owner when a feed breaks, comes back, or goes quiet

Every poll's outcome has always been written to source_health and shown on
/status, but only to someone who went and looked. Nobody did: Siliconera has
been refused on 3,769 polls in a row and Digital Foundry on 1,900, and
neither has ever fetched successfully. SendAdminWarning existed and nothing
called it.

A watcher now reads source_health every 15 minutes and sends one notice to
matrix.admin_room covering everything that changed since it last spoke:
feeds that have failed 5 polls in a row, feeds that have fetched since they
were reported down, and feeds that fetch fine but have kept nothing new for 7
days (every healthy source produces something inside about 4). What has been
said is stored on the row (down_alerted_at, quiet_alerted_at), so each
episode is reported once and a restart doesn't repeat it. A notice that
fails to send is not marked, and goes out on the next pass.

last_new_at is new, moved only when a poll keeps a story. Existing rows start
from their newest retained story, or from now if they have none, and a new
source starts at its first poll: nothing on disk says whether a source with
no stories is long dead or was added yesterday, so it gets a full window
before it can be called quiet.

With no admin room configured the watcher doesn't run at all, rather than
marking alerts as delivered to nowhere.

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
+}

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)
+	}
+}

internal/ingestion/poller.go

@@ -118,6 +118,16 @@ func (p *Poller) pollOnceWithErr(ctx context.Context, src config.SourceConfig) e
 			continue
 		}
 
+		// An item older than the retention window is one Pete has either
+		// already shown and since pruned, or never needs to: maintenance would
+		// delete it on the next pass. Without this, a slow feed that still
+		// lists month-old items re-ingests them as new the day after their row
+		// is pruned, since that row was the only memory of them. Items with no
+		// date can't be judged and pass through.
+		if items[i].PublishedAt > 0 && time.Since(time.Unix(items[i].PublishedAt, 0)) > storage.StoryRetention {
+			continue
+		}
+
 		// Drop items in languages this source isn't configured to surface.
 		// Items without a language tag pass through — only filter when the
 		// feed explicitly marks an item with a non-matching language.
@@ -255,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

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) {

internal/storage/db.go

@@ -1,6 +1,7 @@
 package storage
 
 import (
+	"context"
 	"database/sql"
 	"fmt"
 	"log/slog"
@@ -8,6 +9,7 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"time"
 
 	_ "modernc.org/sqlite"
 )
@@ -94,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")
@@ -154,12 +164,41 @@ func runMigrations(d *sql.DB) error {
 	return nil
 }
 
-// RunMaintenance prunes stale data. Called periodically.
+// StoryRetention is how long a story is kept after Pete first saw it. It is also
+// how long Pete remembers having seen it: the stories row is the only record of
+// its GUID, canonical URL and headline, so the poller ignores feed items older
+// than this rather than meet them again as strangers once their row is gone.
+const StoryRetention = 30 * 24 * time.Hour
+
+// StartMaintenance runs RunMaintenance now and then once a day until ctx ends.
+// It used to run only at startup, so on a service that went weeks between
+// deploys nothing was pruned, and then everything was, all at once.
+func StartMaintenance(ctx context.Context) {
+	RunMaintenance()
+	go func() {
+		t := time.NewTicker(24 * time.Hour)
+		defer t.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-t.C:
+				RunMaintenance()
+			}
+		}
+	}()
+}
+
+// RunMaintenance prunes stale data.
 func RunMaintenance() {
-	// Prune old stories (30 days) and their post logs / reactions
-	storyCutoff := nowUnix() - int64(30*86400)
+	// Prune old stories and their post logs / reactions. A story someone has
+	// bookmarked is kept however old it is: the orphan sweep below would
+	// otherwise take the bookmark with it, and a saved story that quietly
+	// vanishes after a month is exactly what saving it was meant to prevent.
+	storyCutoff := nowUnix() - int64(StoryRetention/time.Second)
 	exec("prune old stories",
-		`DELETE FROM stories WHERE seen_at < ? AND classified = 1`, storyCutoff)
+		`DELETE FROM stories WHERE seen_at < ? AND classified = 1
+		   AND id NOT IN (SELECT story_id FROM user_story_state WHERE bookmarked_at IS NOT NULL)`, storyCutoff)
 	exec("prune old post_log",
 		`DELETE FROM post_log WHERE posted_at < ?`, storyCutoff)
 	exec("prune old reactions",

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
+}

internal/storage/userstate_test.go

@@ -124,3 +124,46 @@ func TestUserStoryState_ScopedBySub(t *testing.T) {
 		t.Fatalf("user-b count = %d, want 0", n)
 	}
 }
+
+// Maintenance prunes stories older than the retention window, and then sweeps
+// away user rows pointing at them. A bookmark is someone asking for a story to
+// be kept, so a bookmarked story must survive the prune, and its bookmark with
+// it. Read markers on pruned stories are fair game.
+func TestMaintenanceKeepsBookmarkedStories(t *testing.T) {
+	setupTestDB(t)
+	saved := seedStory(t, "saved")   // seedStory stamps 2023, far past retention
+	readOnly := seedStory(t, "read") // read, not saved
+	seedStory(t, "untouched")
+	const sub = "ak-sub-keep"
+
+	if err := SetBookmark(sub, saved, true); err != nil {
+		t.Fatal(err)
+	}
+	if err := SetRead(sub, readOnly, true); err != nil {
+		t.Fatal(err)
+	}
+
+	RunMaintenance()
+
+	var n int
+	if err := Get().QueryRow(`SELECT COUNT(*) FROM stories`).Scan(&n); err != nil {
+		t.Fatal(err)
+	}
+	if n != 1 {
+		t.Errorf("stories left after maintenance = %d, want 1 (the bookmarked one)", n)
+	}
+	marks, err := ListBookmarks(sub, 10, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(marks) != 1 || marks[0].GUID != "saved" {
+		t.Fatalf("bookmarks after maintenance = %+v, want just the saved story", marks)
+	}
+	read, _, err := UserStoryState(sub, []int64{readOnly})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if read[readOnly] {
+		t.Error("read marker on a pruned story survived the orphan sweep")
+	}
+}

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 {
@@ -305,8 +315,9 @@ func main() {
 		}
 	}
 
-	// Run maintenance on startup
-	storage.RunMaintenance()
+	// Prune now and then daily. Startup-only meant a service that ran for weeks
+	// between deploys never pruned at all.
+	storage.StartMaintenance(ctx)
 
 	// Wait for shutdown signal
 	sig := <-sigCh
@@ -423,7 +434,7 @@ func runLocal(cfg *config.Config) {
 	ws.StartAdventureAlerts(ctx)
 	slog.Info("local: web UI listening", "addr", cfg.Web.ListenAddr)
 
-	storage.RunMaintenance()
+	storage.StartMaintenance(ctx)
 
 	sig := <-sigCh
 	slog.Info("local: shutdown signal received", "signal", sig)