feat: one Poll Lane per Site, replacing the shared pace (#100)
Each Site runs its own poll goroutine, paced by Rest and Gap from the registry (sites.go) instead of the shared cooldown/interval/stagger/batch config. Rest is enforced by the due query's WHERE clause; the Lane sleeps its effective gap between fetches (hour/eligible, floored at 1s). Lane-local failures: two refusals stop that Site for 15m, a lost browser stops only the round's remaining browser Lanes, and cover heals moved to background goroutines so a slow CDN cannot consume a Lane's gap. The five LATEST_CHAPTER_POLL_* pace env vars are gone; only the kill switch remains.
This commit is contained in:
@@ -16,7 +16,6 @@ import (
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgtype"
|
||||
_ "github.com/jackc/pgx/v5/stdlib"
|
||||
)
|
||||
|
||||
@@ -39,7 +38,7 @@ type Bookmark struct {
|
||||
// origin once the bytes exist, and "" until they do — never a third-party
|
||||
// address and never an address that 404s (ADR-0007). A client may still
|
||||
// send this field and it is discarded on the way in; see Upsert.
|
||||
Cover string `json:"cover"`
|
||||
Cover string `json:"cover"`
|
||||
LastChapter string `json:"last_chapter"`
|
||||
LastChapterNum float64 `json:"last_chapter_num"`
|
||||
LastChapterURL string `json:"last_chapter_url"`
|
||||
@@ -897,10 +896,13 @@ func (s *Store) Delete(readerID int64, key string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// DueForLatestCheck returns series whose server-side latest-chapter check has
|
||||
// aged past the appropriate cutoff, ordered by how many bookmarks reference
|
||||
// them (descending) then least-recently-checked first, at most limit of them.
|
||||
// Browser-backed sites use browserCutoffMs; every other site uses cutoffMs.
|
||||
// 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
|
||||
// query, because each Poll Lane asks for its own list: the query carries one
|
||||
// Site and one cut-off instead of parallel lists (issue #100). There is no
|
||||
// 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.
|
||||
//
|
||||
// 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
|
||||
@@ -916,20 +918,17 @@ func (s *Store) Delete(readerID int64, key string) error {
|
||||
// burns requests. Archived bookmarks still count — knowing what a shelved
|
||||
// series is up to is the whole reason for archiving instead of deleting.
|
||||
// A series with no bookmarks at all never appears: the join excludes it.
|
||||
func (s *Store) DueForLatestCheck(cutoffMs, browserCutoffMs int64, browserSites []string, limit int) ([]Series, error) {
|
||||
func (s *Store) DueForLatestCheck(site string, cutoffMs int64) ([]Series, error) {
|
||||
rows, err := s.db.Query(`SELECT `+seriesColumns+`, COUNT(*) AS reader_count
|
||||
FROM series s
|
||||
JOIN bookmarks b ON b.site = s.site AND b.series_id = s.series_id
|
||||
WHERE s.series_url <> ''
|
||||
AND s.latest_checked_at <= CASE
|
||||
WHEN s.site = ANY($3::text[]) THEN $2::bigint
|
||||
ELSE $1::bigint
|
||||
END
|
||||
WHERE s.site = $1
|
||||
AND s.series_url <> ''
|
||||
AND s.latest_checked_at <= $2::bigint
|
||||
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
|
||||
ORDER BY reader_count DESC, s.latest_checked_at ASC
|
||||
LIMIT $4`, cutoffMs, browserCutoffMs, pgtype.FlatArray[string](browserSites), limit)
|
||||
ORDER BY reader_count DESC, s.latest_checked_at ASC`, site, cutoffMs)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("query due series: %w", err)
|
||||
}
|
||||
@@ -946,6 +945,30 @@ func (s *Store) DueForLatestCheck(cutoffMs, browserCutoffMs int64, browserSites
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// EligibleSeriesCount returns how many of a Site's Series still have at least
|
||||
// one bookmark outside the finished bucket. It is the denominator of the
|
||||
// Lane's pace (issue #100): the effective gap is the smaller of the registry
|
||||
// gap and one hour divided by this count, so Series that will never be Polled
|
||||
// do not make the Lane faster than it needs to be, and counting every eligible
|
||||
// Series rather than only those currently due keeps the pace steady — the
|
||||
// single worst moment to be fastest is startup, when everything is due at
|
||||
// once.
|
||||
func (s *Store) EligibleSeriesCount(site string) (int, error) {
|
||||
var n int
|
||||
err := s.db.QueryRow(`SELECT COUNT(*) FROM (
|
||||
SELECT 1
|
||||
FROM series s
|
||||
JOIN bookmarks b ON b.site = s.site AND b.series_id = s.series_id
|
||||
WHERE s.site = $1
|
||||
GROUP BY s.site, s.series_id
|
||||
HAVING COUNT(*) FILTER (WHERE b.status <> 'finished') > 0
|
||||
) e`, site).Scan(&n)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("count eligible series %s: %w", site, err)
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
// MarkLatestChecked records that the server looked at a series at ts, whatever
|
||||
// the look turned up. Marking a missing series is not an error: the row may
|
||||
// have been orphaned while a fetch was in flight.
|
||||
@@ -954,7 +977,7 @@ func (s *Store) DueForLatestCheck(cutoffMs, browserCutoffMs int64, browserSites
|
||||
// out of the client-visible read path on purpose. PUT /bookmarks/{key} decodes
|
||||
// a whole Bookmark from the client and Upsert writes every series column it
|
||||
// knows about, so a userscript PUT — which has no idea this field exists —
|
||||
// would write a zero and reset the cooldown, making the poller re-fetch that
|
||||
// would write a zero and reset the rest, making the poller re-fetch that
|
||||
// series every tick for as long as the user kept reading it.
|
||||
func (s *Store) MarkLatestChecked(site, seriesID string, ts int64) error {
|
||||
if _, err := s.db.Exec(
|
||||
@@ -966,7 +989,7 @@ func (s *Store) MarkLatestChecked(site, seriesID string, ts int64) error {
|
||||
}
|
||||
|
||||
// LatestCheckedAt reads the column MarkLatestChecked writes. It exists for
|
||||
// tests outside this package (the poller's own tests assert on cooldown
|
||||
// tests outside this package (the poller's own tests assert on rest
|
||||
// bookkeeping) — see MarkLatestChecked for why the field stays off the
|
||||
// client-visible row.
|
||||
func (s *Store) LatestCheckedAt(site, seriesID string) (int64, error) {
|
||||
|
||||
Reference in New Issue
Block a user