Spec #137: per-Series poll failure state, completion hint, and outbound owner notification (#174)

Implements spec #137 (spec 4 of 4 from wayfinder map #114).

Closes #137.

Tickets: #164, #165, #166, #167, #168, #169, #170, #171, #172, #173 — all closed, landed on this branch.

## Summary
- #164/#168: sixth outcome word `not_found`; per-Site completed marker predicate.
- #165/#169: `poll_failures` row is the failure state; the pass remembers a Site-reported completion.
- #166/#171: two new admin filters (`failing`, `unverified`); outbound owner notification + stall condition.
- #167/#170: a failure names itself on the Series page; the completion hint reaches the owner and decides nothing.
- #172: the other three fault conditions (no-browser-route, sidecar-down, adapter-broken) feeding the notifier.
- #173: the landing verdict line shares the same `latest.FaultsFrom` judgement the notifier uses, so the page and the push cannot disagree.

`cd backend && go test ./...` green on the merged branch (8 packages).

Reviewed-on: #174
Co-authored-by: Sulthan Zaki <sultankiki05@gmail.com>
Co-committed-by: Sulthan Zaki <sultankiki05@gmail.com>

backend/internal/latest/poller.go

@@ -46,7 +46,10 @@ type Poller struct {
 	// CoverBytesFetch is optional; it handles plain-TLS sources through the
 	// same failure-isolated prefetch path.
 	CoverBytesFetch CoverBytesFetcher
-	Now             func() time.Time // injected so tests can freeze it
+	// Notify delivers owner notices. Nil disables the whole path (issue #171):
+	// the poller is not the place a missing webhook becomes an error.
+	Notify Notifier
+	Now    func() time.Time // injected so tests can freeze it
 	// eligibleCount reports how many of a Site's Series are eligible for
 	// polling, defaulting to Store.EligibleSeriesCount. Injected so tests can
 	// fail the count alone: the eligible query shares the due query's tables,
@@ -273,7 +276,9 @@ const (
 // (issue #141). The classification the read already makes is counted, never a
 // second taxonomy: refused is the Site holding a challenge, unreachable the
 // browser interrupting, noChapter a 200 with real HTML but no chapter links,
-// unfetchable the host pin or a missing fetcher, and errors everything else.
+// unfetchable the host pin or a missing fetcher, notFound a 4xx other than
+// the 403 refusal — the Site answered with a client status — and errors
+// everything else.
 type readOutcome int
 
 const (
@@ -283,14 +288,33 @@ const (
 	outcomeNoChapter
 	outcomeUnfetchable
 	outcomeError
+	outcomeNotFound
 )
+// word returns the wire spelling this outcome stores in poll_failures — the
+// same strings the pass log's columns use (C1, issue #164). Success, refusal
+// and browser loss return "" so recordFailure's "no statement" case is one
+// return: a challenge or a lost sidecar is no evidence about any particular
+// Series (ADR-0016).
+func (o readOutcome) word() string {
+	switch o {
+	case outcomeNotFound:
+		return "not_found"
+	case outcomeNoChapter:
+		return "no_chapter"
+	case outcomeUnfetchable:
+		return "unfetchable"
+	case outcomeError:
+		return "errors"
+	}
+	return ""
+}
 
-// outcomeCounts are the five named outcome counts of one pass. A success
-// count is derived, never stored: checked minus the four, with unreachable
-// excluded because the sidecar-loss path returns before the checked counter
-// increments (issue #141).
+// outcomeCounts are the six named outcome counts of one pass. A success
+// count is derived, never stored: checked minus the five named failures,
+// with unreachable excluded because the sidecar-loss path returns before the
+// checked counter increments (issue #141).
 type outcomeCounts struct {
-	refused, unreachable, noChapter, unfetchable, errors int
+	refused, unreachable, noChapter, unfetchable, errors, notFound int
 }
 
 func (c *outcomeCounts) add(o readOutcome) {
@@ -303,6 +327,8 @@ func (c *outcomeCounts) add(o readOutcome) {
 		c.noChapter++
 	case outcomeUnfetchable:
 		c.unfetchable++
+	case outcomeNotFound:
+		c.notFound++
 	case outcomeError:
 		c.errors++
 	}
@@ -334,7 +360,7 @@ func (p *Poller) runLanePass(ctx context.Context, name string, paced bool) time.
 	// carries the previous pass's forward inside recordPass.
 	fig := passFigures{}
 	rec := passRecord{site: name, ranAt: now.UnixMilli()}
-	defer func() { p.recordPass(rec, fig) }()
+	defer func() { p.recordPass(ctx, rec, fig) }()
 
 	// One Lane row read at the top of a pass, serving two gates (issue #139).
 	// Both stamps outlive our process, so the gates read the durable row
@@ -451,6 +477,7 @@ func (p *Poller) runLanePass(ctx context.Context, name string, paced bool) time.
 			}
 		}
 		outcome := p.checkOne(ctx, sr)
+		p.recordFailure(sr, outcome)
 		if outcome == outcomeUnreachable {
 			// The mid-loop browser loss writes an empty skip on purpose: the
 			// pass returns before the checked counter increments, so its row
@@ -503,8 +530,9 @@ type passFigures struct {
 // pass's due, gap, clamped and checked forward rather than stating zeroes it
 // did not measure; the skip column says why it declined, so the zeroes that
 // remain (due-query, no-fetcher) read as explanations rather than
-// measurements.
-func (p *Poller) recordPass(rec passRecord, fig passFigures) {
+// measurements. Once the row is durable, the owner-notice judgement runs
+// beside it (issue #171).
+func (p *Poller) recordPass(ctx context.Context, rec passRecord, fig passFigures) {
 	row := store.LanePass{
 		Site:        rec.site,
 		RanAt:       rec.ranAt,
@@ -517,6 +545,7 @@ func (p *Poller) recordPass(rec passRecord, fig passFigures) {
 		Unreachable: rec.counts.unreachable,
 		NoChapter:   rec.counts.noChapter,
 		Unfetchable: rec.counts.unfetchable,
+		NotFound:    rec.counts.notFound,
 		Errors:      rec.counts.errors,
 	}
 	if row.GapMS == 0 {
@@ -532,7 +561,111 @@ func (p *Poller) recordPass(rec passRecord, fig passFigures) {
 	}
 	if err := p.Store.RecordLanePass(row, rec.ranAt-lanePassRetention.Milliseconds()); err != nil {
 		log.Printf("latest poll %s: record lane pass: %v", rec.site, err)
+		return
 	}
+	p.ownerNotices(ctx, row)
+}
+
+// ownerNotices judges the owner-notice conditions for the pass just recorded
+// and fires (issue #171). It sits in recordPass because that deferred call is
+// the one place every return path passes through: three of the four
+// conditions occur on early returns and the success path can never see them.
+// The judgement reads the whole pass log plus three derived reads — the
+// other Lanes' latest passes, each Site's refusing-run start, its last
+// sidecar-reaching pass, and its no-chapter share — so one pass judges every
+// condition (issue #172). Per fault: NoticeSent → send → MarkNoticeSent, so
+// a fault lasting a month sends one message, not one per pass; a condition
+// absent from this pass's fault list forgets its episode, so the next
+// occurrence sends again. Everything here is best-effort: a failed send, a
+// failed store read and a failed notice write are all logged and never
+// change the pass's outcome counts or its return value. The clear runs even
+// when Notify is nil, so a deployment that turns the webhook off does not
+// leave stale rows that suppress the first real notice after it is turned
+// back on.
+func (p *Poller) ownerNotices(ctx context.Context, row store.LanePass) {
+	now := p.Now()
+	in := FaultInput{Passes: []store.LanePass{row}}
+	// Each read fails independently: a failure logs and contributes no fault,
+	// never a false one.
+	if passes, err := p.Store.LatestLanePasses(); err != nil {
+		log.Printf("latest poll %s: latest lane passes: %v", row.Site, err)
+	} else {
+		in.Passes = passes
+	}
+	if since, err := p.Store.RefusingSince(now.UnixMilli()); err != nil {
+		log.Printf("latest poll %s: refusing since: %v", row.Site, err)
+	} else {
+		in.RefusingSince = since
+	}
+	if ok, err := p.Store.SidecarOK(browserBackedSites()); err != nil {
+		log.Printf("latest poll %s: sidecar ok: %v", row.Site, err)
+	} else {
+		in.SidecarOK = ok
+	}
+	if share, err := p.Store.NoChapterShare(now.Add(-OwnerWindow).UnixMilli()); err != nil {
+		log.Printf("latest poll %s: no-chapter share: %v", row.Site, err)
+	} else {
+		in.NoChapterShare = share
+	}
+	faults := FaultsFrom(in, now)
+	bySite := make(map[string]store.LanePass, len(in.Passes))
+	for _, pass := range in.Passes {
+		bySite[pass.Site] = pass
+	}
+	for _, f := range faults {
+		if p.Notify == nil {
+			continue
+		}
+		sent, err := p.Store.NoticeSent(f.Condition, f.Site)
+		if err != nil {
+			log.Printf("latest poll %s: notice sent: %v", row.Site, err)
+			continue
+		}
+		if sent {
+			continue
+		}
+		// The fault's own Site's pass renders its sentence — a stall judged
+		// from another Lane's pass must not quote this pass's figures.
+		pass, ok := bySite[f.Site]
+		if !ok {
+			pass = row
+		}
+		sentence, href := noticeFor(f, pass, now)
+		if err := p.Notify.Notify(ctx, f, sentence, href); err != nil {
+			// The stamp stays unset: no queue, no backoff — the condition is
+			// durable, so the next pass tries again while it holds.
+			log.Printf("latest poll %s: owner notice %s: %v", row.Site, f.Condition, err)
+			continue
+		}
+		if err := p.Store.MarkNoticeSent(f.Condition, f.Site, now.UnixMilli()); err != nil {
+			log.Printf("latest poll %s: mark notice sent: %v", row.Site, err)
+		}
+	}
+	for _, cond := range ownerNoticeConditions {
+		if !hasFault(faults, cond, row.Site) {
+			if err := p.Store.ClearNotice(cond, row.Site); err != nil {
+				log.Printf("latest poll %s: clear owner notice: %v", row.Site, err)
+			}
+		}
+	}
+	// sidecar-down suppresses under the empty Site — one row across all
+	// browser Lanes (AC4) — so clear that row when it is absent from this
+	// pass's fault list, and a lifted sidecar fires again when it returns.
+	if !hasFault(faults, ConditionSidecarDown, "") {
+		if err := p.Store.ClearNotice(ConditionSidecarDown, ""); err != nil {
+			log.Printf("latest poll %s: clear owner notice: %v", row.Site, err)
+		}
+	}
+}
+
+// hasFault reports whether faults hold the given condition for the site.
+func hasFault(faults []Fault, condition, site string) bool {
+	for _, f := range faults {
+		if f.Condition == condition && f.Site == site {
+			return true
+		}
+	}
+	return false
 }
 
 // countEligible routes the eligible count through the test seam when one is
@@ -607,13 +740,35 @@ func maxSeriesWait(due []store.Series, now time.Time, rest time.Duration) time.D
 	}
 	return oldest
 }
+// recordFailure keeps one Series' failure row in step with its read
+// (ADR-0016): a failure word is upserted, a successful read deletes the row,
+// and refused or unreachable issue no statement at all. Called for every
+// outcome from the pass loop, so the four failure words and the success path
+// share one write point, and a forced Poll that reads the page clears through
+// the ordinary success path — no branch of its own. Log a store failure and
+// carry on: this is best-effort, and no single bad Series may stall a Lane.
+func (p *Poller) recordFailure(sr store.Series, outcome readOutcome) {
+	if outcome == outcomeSuccess {
+		if err := p.Store.ClearSeriesFailure(sr.Site, sr.SeriesID); err != nil {
+			log.Printf("latest poll %q: clear failure: %v", sr.Key(), err)
+		}
+		return
+	}
+	word := outcome.word()
+	if word == "" {
+		return
+	}
+	if err := p.Store.RecordSeriesFailure(sr.Site, sr.SeriesID, word, p.Now().UnixMilli()); err != nil {
+		log.Printf("latest poll %q: record failure: %v", sr.Key(), err)
+	}
+}
 
 // checkOne re-checks one series. Every failure path here is "log and move on":
 // the poller is a best-effort enhancement, and no single bad series may stall a
 // Lane or take down the process. The returned outcome classifies the read for
 // the pass row (issue #141), so the Lane can count a refusal, a lost browser,
-// a chapter-less page, an unfetchable address or a transport error without
-// re-deriving the taxonomy.
+// a chapter-less page, an unfetchable address, a missing page or a transport
+// error without re-deriving the taxonomy.
 func (p *Poller) checkOne(ctx context.Context, sr store.Series) (outcome readOutcome) {
 	defer func() {
 		if r := recover(); r != nil {
@@ -657,6 +812,9 @@ func (p *Poller) checkOne(ctx context.Context, sr store.Series) (outcome readOut
 		if errors.Is(err, errBrowserInterrupted) {
 			return outcomeUnreachable
 		}
+		if errors.Is(err, errNotFound) {
+			return outcomeNotFound
+		}
 		return outcomeError
 	}
 	// A legacy cover source is healed independently of the page read.
@@ -669,6 +827,25 @@ func (p *Poller) checkOne(ctx context.Context, sr store.Series) (outcome readOut
 	} else {
 		p.fillBlankCover(ctx, sr, facts.Cover)
 	}
+
+	// Learned from the successful read: the write sits after the error switch
+	// (a refused, unreachable or errored read reaches nothing) and before the
+	// returns below — a completed page whose chapter number did not change
+	// still has to write. The transition is zero-versus-nonzero, not the
+	// stamp's value: a Series still completed keeps its original stamp, so the
+	// age #170 prints is "since the Site first said so"; one that stopped
+	// being completed is zeroed.
+	stamp := int64(0)
+	if facts.SiteCompleted {
+		stamp = p.Now().UnixMilli()
+	}
+	if (sr.SiteCompletedAt == 0) != (stamp == 0) {
+		if err := p.Store.SetSiteCompletedAt(sr.Site, sr.SeriesID, stamp); err != nil {
+			// Best-effort, like every poller write: never change the outcome
+			// word the pass counts.
+			log.Printf("latest poll %q: set site completed: %v", sr.Key(), err)
+		}
+	}
 	if !facts.HasLatest {
 		// Most likely a challenge page or a layout change. Either way the row is
 		// already stamped, so this waits out a rest instead of hot-looping.