Spec #134, all ten tickets. Closes #134. ## What ships The admin surface becomes four bookmarkable addresses behind one nav row, and Lane observability stops dying with the process. - **#138** `/admin` splits into Overview, Lanes, Readers, Series, each a real route with the active tab underlined. - **#139** `poll_passes` and `poll_lanes` land as durable tables with their store surface. - **#140** cross-Series admin read model, with the privacy boundary in the projection: the Reader id that raised a Latest Chapter never leaves the store package. - **#141** the poller records exactly one pass row per exit, with a skip reason and outcome counts. - **#142** Series list: eight hygiene filters, Site and Library narrowing, paging — all of it in the query string, so a filtered list is a bookmark. - **#143** Overview: a three-state verdict line and a stats block where every non-zero figure links to the list that counts it. - **#144** per-Series detail page, keyed by the `site:series_id` composite the rest of the system already uses. - **#145** the Lanes page reads the database; the in-memory Lane state, `web.LaneReporter` and `latest.Status` are deleted. - **#146** Forced Poll: *Check now* stamps `series.force_poll_at` and never commands the poller. - **#147** pause and resume one Site's Lane, with a mandatory 1h/6h/24h expiry. ## Shape of the design Two decisions carry the rest. **Commands go through the database, never at the poller**: both *Check now* and a Lane pause write a row the next pass reads, so they survive a restart and the whole surface stays testable with no poller running. And **pending is derived, never stored** — the request stamp being newer than the check stamp — which self-clears on the check stamp with no second write and no sweeper, because the check stamp is written before the fetch. ADRs: `docs/adr/0012-persisted-lane-state.md`, `docs/adr/0013-commands-through-the-database.md`. ## Verification `go test ./...` green on the merged base (`264839e`), all packages, Docker-backed. `gofmt -l` and `go vet` clean. Every ticket was reviewed on both axes (`cr-spec` + `cr-standards`) before merge. ## Known, non-blocking - **#143** the verdict ignores never-reported Lanes when other Lanes have reported, and the per-Site table lists Sites that have Series rather than the whole registry. The ticket prose asks for eight hygiene figures per Site; the design mock and the landed `.tbl.sites` grid both say six columns, and the mock won. - **#146** two `SeriesPage` scans per press instead of a keyed read — `ponytail:`-commented in-tree with the upgrade path. - **#147** a paused Site with no pass row yet renders no row and so no control, since the Lanes page lists Sites that have passed. - **#141** a sibling browser Lane declining at the top of a pass records as `sidecar-down`. Specified deliberately; the later spec in this series settles it. Reviewed-on: #148 Co-authored-by: Sulthan Zaki <sultankiki05@gmail.com> Co-committed-by: Sulthan Zaki <sultankiki05@gmail.com>
This commit was merged in pull request #148.
This commit is contained in:
+273
-12
@@ -88,6 +88,39 @@ type Series struct {
|
||||
// readerCount is the number of bookmarks referencing this series, filled
|
||||
// only by the due-queue query that orders on it.
|
||||
readerCount int
|
||||
// Forced is whether the owner asked for a check now (issue #146): the
|
||||
// request stamp is newer than the check stamp. Derived in the due query,
|
||||
// never stored, and the flag that jumps the queue and opens the browser
|
||||
// wake gate.
|
||||
Forced bool
|
||||
}
|
||||
|
||||
// LanePass is one Poll Lane's durable pass snapshot. Pause and refusal stamps
|
||||
// are joined from poll_lanes on read; they are not pass facts.
|
||||
type LanePass struct {
|
||||
Site string
|
||||
RanAt int64
|
||||
Skip string
|
||||
Due, Checked int
|
||||
GapMS int64
|
||||
Clamped bool
|
||||
Refused, Unreachable, NoChapter int
|
||||
Unfetchable, Errors int
|
||||
PausedUntil, RefuseUntil int64
|
||||
}
|
||||
|
||||
// SiteOutcomes is one Site's summed Lane Pass outcomes over a caller-supplied
|
||||
// window.
|
||||
type SiteOutcomes struct {
|
||||
Site string
|
||||
Refused, Unreachable, NoChapter int
|
||||
Unfetchable, Errors int
|
||||
}
|
||||
|
||||
// LanePause is one persisted Lane pause stamp.
|
||||
type LanePause struct {
|
||||
Site string
|
||||
PausedUntil int64
|
||||
}
|
||||
|
||||
// Key returns the canonical identity in bookmark-key form ("<site>:<series_id>"),
|
||||
@@ -188,9 +221,9 @@ const (
|
||||
//go:embed migrations/*.sql
|
||||
var migrations embed.FS
|
||||
|
||||
// bookmarkColumns is the only value ever concatenated into query text. It is a
|
||||
// compile-time constant; every request value is bound as a parameter. The
|
||||
// series-owned fields are joined in from the series table, in scanBookmark
|
||||
// These column lists are the only values ever concatenated into query text.
|
||||
// They are compile-time constants; every request value is bound as a parameter.
|
||||
// The series-owned fields are joined in from the series table, in scanBookmark
|
||||
// order, so the flat Bookmark reads back whole despite the split (ADR-0004).
|
||||
const bookmarkColumns = `b.site, b.series_id, s.title, s.series_url, s.cover_address,
|
||||
b.last_chapter, b.last_chapter_num, b.last_chapter_url,
|
||||
@@ -202,6 +235,10 @@ const bookmarkColumns = `b.site, b.series_id, s.title, s.series_url, s.cover_add
|
||||
const seriesColumns = `s.site, s.series_id, s.title, s.series_url, s.cover, s.cover_address,
|
||||
s.kind, s.latest_chapter, s.latest_chapter_num, s.latest_checked_at, s.latest_raised_by`
|
||||
|
||||
const lanePassColumns = `p.site, p.ran_at, p.skip, p.due, p.checked, p.gap_ms, p.clamped,
|
||||
p.refused, p.unreachable, p.no_chapter, p.unfetchable, p.errors,
|
||||
COALESCE(l.paused_until, 0), COALESCE(l.refuse_until, 0)`
|
||||
|
||||
// Owner is the person running the service: the first Reader, seeded at startup
|
||||
// so a fresh deployment has a library before anyone logs in. The seed makes
|
||||
// sure exactly one readers row matches their Discord ID, carrying the SHA-256
|
||||
@@ -589,8 +626,9 @@ func (s *Store) scanBookmark(scan func(...any) error) (Bookmark, error) {
|
||||
}
|
||||
|
||||
// scanSeries reads one row in seriesColumns order, plus the due query's
|
||||
// reader_count column. latest_chapter_num and latest_raised_by are both
|
||||
// nullable, same as latest_chapter_num on the bookmark read path.
|
||||
// forced flag and reader_count columns. latest_chapter_num and
|
||||
// latest_raised_by are both nullable, same as latest_chapter_num on the
|
||||
// bookmark read path.
|
||||
func scanSeries(scan func(...any) error) (Series, error) {
|
||||
var (
|
||||
sr Series
|
||||
@@ -600,7 +638,7 @@ func scanSeries(scan func(...any) error) (Series, error) {
|
||||
if err := scan(
|
||||
&sr.Site, &sr.SeriesID, &sr.Title, &sr.SeriesURL, &sr.Cover, &sr.CoverAddress,
|
||||
&sr.Kind, &sr.LatestChapter, &latestChapterNum, &sr.LatestCheckedAt, &latestRaisedBy,
|
||||
&sr.readerCount,
|
||||
&sr.Forced, &sr.readerCount,
|
||||
); err != nil {
|
||||
return Series{}, err
|
||||
}
|
||||
@@ -613,6 +651,18 @@ func scanSeries(scan func(...any) error) (Series, error) {
|
||||
return sr, nil
|
||||
}
|
||||
|
||||
func scanLanePass(scan func(...any) error) (LanePass, error) {
|
||||
var p LanePass
|
||||
if err := scan(
|
||||
&p.Site, &p.RanAt, &p.Skip, &p.Due, &p.Checked, &p.GapMS, &p.Clamped,
|
||||
&p.Refused, &p.Unreachable, &p.NoChapter, &p.Unfetchable, &p.Errors,
|
||||
&p.PausedUntil, &p.RefuseUntil,
|
||||
); err != nil {
|
||||
return LanePass{}, err
|
||||
}
|
||||
return p, nil
|
||||
}
|
||||
|
||||
// Close releases the underlying database handle.
|
||||
func (s *Store) Close() error { return s.db.Close() }
|
||||
|
||||
@@ -934,6 +984,184 @@ func (s *Store) Delete(readerID int64, key string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// RecordLanePass appends one pass and prunes every older row in the same
|
||||
// transaction. retainBefore is supplied by the poller's clock.
|
||||
func (s *Store) RecordLanePass(p LanePass, retainBefore int64) error {
|
||||
tx, err := s.db.Begin()
|
||||
if err != nil {
|
||||
return fmt.Errorf("begin lane pass %s: %w", p.Site, err)
|
||||
}
|
||||
defer tx.Rollback()
|
||||
if _, err := tx.Exec(`
|
||||
INSERT INTO poll_passes
|
||||
(site, ran_at, skip, due, checked, gap_ms, clamped,
|
||||
refused, unreachable, no_chapter, unfetchable, errors)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)`,
|
||||
p.Site, p.RanAt, p.Skip, p.Due, p.Checked, p.GapMS, p.Clamped,
|
||||
p.Refused, p.Unreachable, p.NoChapter, p.Unfetchable, p.Errors); err != nil {
|
||||
return fmt.Errorf("insert lane pass %s at %d: %w", p.Site, p.RanAt, err)
|
||||
}
|
||||
if _, err := tx.Exec(`DELETE FROM poll_passes WHERE ran_at < $1`, retainBefore); err != nil {
|
||||
return fmt.Errorf("prune lane passes before %d: %w", retainBefore, err)
|
||||
}
|
||||
if err := tx.Commit(); err != nil {
|
||||
return fmt.Errorf("commit lane pass %s at %d: %w", p.Site, p.RanAt, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// LatestLanePass returns the newest pass for one Site, with its current Lane
|
||||
// state joined on. A Site without a pass has no durable snapshot yet.
|
||||
func (s *Store) LatestLanePass(site string) (LanePass, bool, error) {
|
||||
p, err := scanLanePass(s.db.QueryRow(`SELECT `+lanePassColumns+`
|
||||
FROM poll_passes p
|
||||
LEFT JOIN poll_lanes l ON l.site = p.site
|
||||
WHERE p.site = $1
|
||||
ORDER BY p.ran_at DESC
|
||||
LIMIT 1`, site).Scan)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return LanePass{}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return LanePass{}, false, fmt.Errorf("latest lane pass %s: %w", site, err)
|
||||
}
|
||||
return p, true, nil
|
||||
}
|
||||
|
||||
// LatestLanePasses returns the newest pass for each Site, with current Lane
|
||||
// state joined on. Sites without a pass have no row yet.
|
||||
func (s *Store) LatestLanePasses() ([]LanePass, error) {
|
||||
rows, err := s.db.Query(`SELECT ` + lanePassColumns + `
|
||||
FROM (
|
||||
SELECT DISTINCT ON (site)
|
||||
site, ran_at, skip, due, checked, gap_ms, clamped,
|
||||
refused, unreachable, no_chapter, unfetchable, errors
|
||||
FROM poll_passes
|
||||
ORDER BY site, ran_at DESC
|
||||
) p
|
||||
LEFT JOIN poll_lanes l ON l.site = p.site
|
||||
ORDER BY p.site`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query latest lane passes: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := []LanePass{}
|
||||
for rows.Next() {
|
||||
p, err := scanLanePass(rows.Scan)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("scan latest lane pass: %w", err)
|
||||
}
|
||||
out = append(out, p)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// LanePassOutcomes sums the named outcomes for each Site at or after since.
|
||||
// The window boundary is supplied by the caller; the store has no clock.
|
||||
func (s *Store) LanePassOutcomes(since int64) ([]SiteOutcomes, error) {
|
||||
rows, err := s.db.Query(`
|
||||
SELECT site, SUM(refused), SUM(unreachable), SUM(no_chapter),
|
||||
SUM(unfetchable), SUM(errors)
|
||||
FROM poll_passes
|
||||
WHERE ran_at >= $1
|
||||
GROUP BY site
|
||||
ORDER BY site`, since)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query lane pass outcomes: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := []SiteOutcomes{}
|
||||
for rows.Next() {
|
||||
var outcomes SiteOutcomes
|
||||
if err := rows.Scan(
|
||||
&outcomes.Site, &outcomes.Refused, &outcomes.Unreachable,
|
||||
&outcomes.NoChapter, &outcomes.Unfetchable, &outcomes.Errors,
|
||||
); err != nil {
|
||||
return nil, fmt.Errorf("scan lane pass outcomes: %w", err)
|
||||
}
|
||||
out = append(out, outcomes)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// SetLaneRefusal persists a Site's refusal backoff stamp without touching its
|
||||
// pause. until is supplied by the caller's clock.
|
||||
func (s *Store) SetLaneRefusal(site string, until int64) error {
|
||||
if _, err := s.db.Exec(`
|
||||
INSERT INTO poll_lanes (site, refuse_until) VALUES ($1, $2)
|
||||
ON CONFLICT (site) DO UPDATE SET refuse_until = EXCLUDED.refuse_until`, site, until); err != nil {
|
||||
return fmt.Errorf("set lane refusal %s: %w", site, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// PauseLane persists a bounded pause. The caller must ensure until is after
|
||||
// its current timestamp; the store has no clock and rejects only the invalid
|
||||
// zero and negative sentinels.
|
||||
func (s *Store) PauseLane(site string, until int64) error {
|
||||
if until <= 0 {
|
||||
return fmt.Errorf("pause lane %s: expiry must be positive", site)
|
||||
}
|
||||
if _, err := s.db.Exec(`
|
||||
INSERT INTO poll_lanes (site, paused_until) VALUES ($1, $2)
|
||||
ON CONFLICT (site) DO UPDATE SET paused_until = EXCLUDED.paused_until`, site, until); err != nil {
|
||||
return fmt.Errorf("pause lane %s: %w", site, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ResumeLane clears only the pause stamp and keeps the Lane state row, along
|
||||
// with any refusal stamp already persisted on it.
|
||||
func (s *Store) ResumeLane(site string) error {
|
||||
if _, err := s.db.Exec(
|
||||
`UPDATE poll_lanes SET paused_until = 0 WHERE site = $1`, site); err != nil {
|
||||
return fmt.Errorf("resume lane %s: %w", site, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// PausedLanes returns Lane rows with a nonzero pause stamp. Expiry comparison
|
||||
// stays with the caller because the store is deliberately clockless.
|
||||
func (s *Store) PausedLanes() ([]LanePause, error) {
|
||||
rows, err := s.db.Query(`
|
||||
SELECT site, paused_until
|
||||
FROM poll_lanes
|
||||
WHERE paused_until > 0
|
||||
ORDER BY site`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query paused lanes: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := []LanePause{}
|
||||
for rows.Next() {
|
||||
var pause LanePause
|
||||
if err := rows.Scan(&pause.Site, &pause.PausedUntil); err != nil {
|
||||
return nil, fmt.Errorf("scan paused lane: %w", err)
|
||||
}
|
||||
out = append(out, pause)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// LaneGates reads a Site's pause and refusal stamps in one row read — the
|
||||
// top-of-pass gate the poller uses (issue #141). A missing state row is the
|
||||
// default: unpaused and not refusing.
|
||||
func (s *Store) LaneGates(site string) (pausedUntil, refuseUntil int64, err error) {
|
||||
err = s.db.QueryRow(
|
||||
`SELECT paused_until, refuse_until FROM poll_lanes WHERE site = $1`, site).
|
||||
Scan(&pausedUntil, &refuseUntil)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return 0, 0, nil
|
||||
}
|
||||
if err != nil {
|
||||
return 0, 0, fmt.Errorf("lane gates %s: %w", site, err)
|
||||
}
|
||||
return pausedUntil, refuseUntil, nil
|
||||
}
|
||||
|
||||
// DueForLatestCheck returns one Site's series whose server-side
|
||||
// latest-chapter check has aged past cutoffMs, ordered by how many bookmarks
|
||||
// reference them (descending) then least-recently-checked first. One Site per
|
||||
@@ -942,6 +1170,13 @@ func (s *Store) Delete(readerID int64, key string) error {
|
||||
// limit — the Lane's own gap paces the fetches, and the batch size that used
|
||||
// to cap this query is gone with the shared pace.
|
||||
//
|
||||
// A forced Series (force_poll_at newer than latest_checked_at, issue #146)
|
||||
// overrides exactly three gates: the rest cutoff, the Sighting-deferral
|
||||
// clause and the finished-only bucket. It never overrides an empty
|
||||
// series_url or the Bookmarks join — nothing to fetch, and no consumer for
|
||||
// the result — so those stay unconditional. Forced rows sort to the front of
|
||||
// the queue; the reader-count-then-age ordering among the rest is ADR-0003.
|
||||
//
|
||||
// The reader_count ordering is the point of the split (ADR-0003): a series
|
||||
// shared by several readers is fetched once per due cycle, and the popular
|
||||
// ones stay freshest while the long tail absorbs any shortfall. Within one
|
||||
@@ -970,19 +1205,26 @@ func (s *Store) Delete(readerID int64, key string) error {
|
||||
// allowed to defer at all was settled when the Sighting was recorded — see
|
||||
// RecordSighting.
|
||||
func (s *Store) DueForLatestCheck(site string, cutoffMs, ceilingMs int64) ([]Series, error) {
|
||||
rows, err := s.db.Query(`SELECT `+seriesColumns+`, COUNT(*) AS reader_count
|
||||
rows, err := s.db.Query(`SELECT `+seriesColumns+`,
|
||||
(s.force_poll_at > s.latest_checked_at) AS forced,
|
||||
COUNT(*) AS reader_count
|
||||
FROM series s
|
||||
JOIN bookmarks b ON b.site = s.site AND b.series_id = s.series_id
|
||||
WHERE s.site = $1
|
||||
AND s.series_url <> ''
|
||||
AND s.latest_checked_at <= $2::bigint
|
||||
AND (s.latest_checked_at <= $2::bigint
|
||||
OR s.force_poll_at > s.latest_checked_at)
|
||||
GROUP BY s.site, s.series_id, s.title, s.series_url, s.cover,
|
||||
s.kind, s.latest_chapter, s.latest_chapter_num, s.latest_checked_at
|
||||
HAVING COUNT(*) FILTER (WHERE b.status <> 'finished') > 0
|
||||
s.kind, s.latest_chapter, s.latest_chapter_num, s.latest_checked_at,
|
||||
s.force_poll_at
|
||||
HAVING (COUNT(*) FILTER (WHERE b.status <> 'finished') > 0
|
||||
OR s.force_poll_at > s.latest_checked_at)
|
||||
AND (COUNT(*) > 1
|
||||
OR s.latest_sighted_at <= $2::bigint
|
||||
OR s.latest_checked_at <= $3::bigint)
|
||||
ORDER BY reader_count DESC, s.latest_checked_at ASC`, site, cutoffMs, ceilingMs)
|
||||
OR s.latest_checked_at <= $3::bigint
|
||||
OR s.force_poll_at > s.latest_checked_at)
|
||||
ORDER BY (s.force_poll_at > s.latest_checked_at) DESC,
|
||||
reader_count DESC, s.latest_checked_at ASC`, site, cutoffMs, ceilingMs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query due series: %w", err)
|
||||
}
|
||||
@@ -1042,6 +1284,25 @@ func (s *Store) MarkLatestChecked(site, seriesID string, ts int64) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// ForceSeriesPoll stamps a Series with the owner's "check now" request
|
||||
// (issue #146): a fact about the Series the Lane's next pass reads through
|
||||
// DueForLatestCheck, never a command to the poller — so the request survives
|
||||
// a restart. Writing again overwrites the request time; the write is
|
||||
// idempotent. Touching a missing series is not an error: the row may have
|
||||
// been orphaned, and the caller's read decides what exists. The stamp never
|
||||
// expires by itself — an unanswered request keeps ageing — and pending is
|
||||
// derived as force_poll_at > latest_checked_at, which is why the poller's
|
||||
// check stamp is written before the fetch: the first attempt ends the
|
||||
// pending state whatever it returns.
|
||||
func (s *Store) ForceSeriesPoll(site, seriesID string, at int64) error {
|
||||
if _, err := s.db.Exec(
|
||||
`UPDATE series SET force_poll_at = $1 WHERE site = $2 AND series_id = $3`,
|
||||
at, site, seriesID); err != nil {
|
||||
return fmt.Errorf("force poll %s:%s: %w", site, seriesID, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// LatestCheckedAt reads the column MarkLatestChecked writes. It exists for
|
||||
// tests outside this package (the poller's own tests assert on rest
|
||||
// bookkeeping) — see MarkLatestChecked for why the field stays off the
|
||||
|
||||
Reference in New Issue
Block a user