Closes #100. Each Site runs its own Poll Lane: an independent goroutine with its own rest and pace from the registry (`backend/internal/latest/sites.go`), replacing the shared cooldown/interval/stagger/batch configuration. Rest (1h, all six Sites including the browser trio) is enforced by the due query's WHERE clause; the Lane sleeps its effective gap between fetches — the registry 10s, or rest/eligible when a Site holds enough Series, floored at 1s with a Site-naming warning when the floor engages. Lane-local failure handling: - Two challenge-held results stop that Site's Lane for 15m; the probes keep their stamp, untried Series stay due. - A lost browser sets a shared Poller flag: the other browser Lanes skip their passes for the same 15m (no stamp-per-pass-per-Lane on a dead tab), then decay and probe again. - Browser wake gate preserved (5 due, or one waiting 15m, ADR-0005); one tab shared by the three browser Sites; "browser lane behind by X" logged every pass. - Cover work (healing a stored source URL and filling a blank from the series page) runs in the background so a slow CDN cannot consume a Lane's gap. Removed: `LATEST_CHAPTER_POLL_{COOLDOWN,BROWSER_COOLDOWN,INTERVAL,BATCH,STAGGER}` and the 6h browser rest. Only `LATEST_CHAPTER_POLL_ENABLED` remains; DEPLOY.md documents the exact `.env` edit. ADR-0010 records the decisions. Reviewed-on: #106 Co-authored-by: Sulthan Zaki <sultankiki05@gmail.com> Co-committed-by: Sulthan Zaki <sultankiki05@gmail.com>
This commit was merged in pull request #106.
This commit is contained in:
@@ -120,7 +120,7 @@ func (a *Acquirer) acquire(ctx context.Context, sr store.Series) {
|
||||
// Stamped after success — the reverse of the poller, which stamps before
|
||||
// the fetch: the Reader is here, watching the Series they just created, so
|
||||
// a failed acquisition must leave the row due for a fast retry rather than
|
||||
// consuming the cooldown. The stamp happens even when the page read
|
||||
// consuming the rest. The stamp happens even when the page read
|
||||
// succeeded but produced no facts to persist.
|
||||
if err := a.Store.MarkLatestChecked(sr.Site, sr.SeriesID, time.Now().UnixMilli()); err != nil {
|
||||
log.Printf("acquire %q: mark checked: %v", sr.Key(), err)
|
||||
|
||||
@@ -18,7 +18,7 @@ import (
|
||||
|
||||
// challengeTimeout bounds one navigate-and-solve. A Cloudflare managed
|
||||
// challenge clears in a few seconds when it clears at all; anything longer is a
|
||||
// challenge that is not going to pass, and the caller's cooldown was already
|
||||
// challenge that is not going to pass, and the caller's rest was already
|
||||
// stamped before this ran.
|
||||
const challengeTimeout = 45 * time.Second
|
||||
|
||||
@@ -213,7 +213,7 @@ func (f *BrowserFetcher) Image(ctx context.Context, imageURL string) ([]byte, st
|
||||
|
||||
// errChallengeHeld reports that the budget ran out with the interstitial still
|
||||
// up. Distinct from a transport failure: it means "this site said no", which
|
||||
// the poller answers with a 403 and its ordinary cooldown.
|
||||
// the poller answers with a refusal backoff for that Site's Lane (issue #100).
|
||||
var errChallengeHeld = errors.New("challenge held")
|
||||
|
||||
// errBrowserInterrupted distinguishes a remote Chrome restart from the
|
||||
|
||||
@@ -5,6 +5,8 @@ import (
|
||||
"errors"
|
||||
"log"
|
||||
"net/url"
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"bookmarkmanager/backend/internal/store"
|
||||
@@ -28,15 +30,11 @@ type BrowserCoverFetcher interface {
|
||||
// in parallel and report the same observable fact, so whichever writes last wins
|
||||
// and neither needs to know about the other.
|
||||
//
|
||||
// Two clocks, deliberately independent:
|
||||
//
|
||||
// - Interval is how often this goroutine wakes up and looks.
|
||||
// - Cooldowns are how long a series rests since its own last check. Browser-
|
||||
// backed sites use the longer BrowserCooldown.
|
||||
//
|
||||
// Cooldowns are enforced by the WHERE clause in DueForLatestCheck rather than
|
||||
// by any timer. Shortening Interval therefore cannot shorten anyone's cooldown;
|
||||
// it only makes the poller wake up and find nothing due more often.
|
||||
// Every Site gets its own Poll Lane: one independent stream of Polls with its
|
||||
// own pace, running concurrently with every other Site's (issue #100). Rest
|
||||
// time and gap live in the Site registry, not here — see sites.go. Rest is
|
||||
// enforced by the WHERE clause in DueForLatestCheck rather than by any timer;
|
||||
// the gap is enforced by the Lane sleeping between fetches.
|
||||
type Poller struct {
|
||||
Store *store.Store
|
||||
Fetch Fetcher
|
||||
@@ -50,11 +48,19 @@ type Poller struct {
|
||||
// same failure-isolated prefetch path.
|
||||
CoverBytesFetch CoverBytesFetcher
|
||||
Now func() time.Time // injected so tests can freeze it
|
||||
Cooldown time.Duration
|
||||
BrowserCooldown time.Duration
|
||||
Interval time.Duration
|
||||
Stagger time.Duration
|
||||
Batch int
|
||||
|
||||
// refuseUntil gates a Site's Lane after it refused twice in one run: no
|
||||
// Series of that Site is attempted again before this time (issue #100).
|
||||
// browserDownAt is when a browser Lane last lost the sidecar; the other
|
||||
// browser Lanes skip their passes for the next refuseBackoff, so a
|
||||
// restarting Chrome does not stamp one Series per pass per Lane (story 20).
|
||||
mu sync.Mutex
|
||||
refuseUntil map[string]time.Time
|
||||
browserDownAt time.Time
|
||||
// coverWG tracks in-flight cover work. Covers heal in the background so a
|
||||
// slow cover host cannot delay the next Series-page Poll; tests join it
|
||||
// before asserting on cover fetches.
|
||||
coverWG sync.WaitGroup
|
||||
}
|
||||
|
||||
// fillBlankCover gives a Series its Cover when it has none. The blank state is
|
||||
@@ -76,7 +82,14 @@ func (p *Poller) fillBlankCover(ctx context.Context, sr store.Series, cover stri
|
||||
if cover == "" {
|
||||
return
|
||||
}
|
||||
p.storeCover(ctx, sr, cover)
|
||||
// Like healCover, the fill runs in the background: a large import of
|
||||
// blanks would otherwise pay one og:image fetch per Series against the
|
||||
// Lane's gap (issue #100, story 12).
|
||||
p.coverWG.Add(1)
|
||||
go func() {
|
||||
defer p.coverWG.Done()
|
||||
p.storeCover(ctx, sr, cover)
|
||||
}()
|
||||
}
|
||||
|
||||
// prefetchCover heals Series that already carry a third-party source URL but
|
||||
@@ -143,72 +156,248 @@ func fetcherFor(site string, browser, tls Fetcher) Fetcher {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Run polls until ctx is cancelled.
|
||||
//
|
||||
// runOnce is called synchronously, so a batch that overruns the tick delays the
|
||||
// next one instead of stacking a second batch on top of it. That is the intended
|
||||
// failure mode for a misconfigured batch x stagger: a slower cadence, never
|
||||
// concurrent fetch storms.
|
||||
// Run polls until ctx is cancelled: one goroutine per Site Lane, each pacing
|
||||
// itself by the Site's effective gap. Lanes share nothing but the store and
|
||||
// the browser fetcher's single tab (BrowserFetcher serializes itself), so one
|
||||
// hostile Site burns only its own budget.
|
||||
// laneNames returns every registry Site in the deterministic order both Run
|
||||
// and runOnce iterate: sorted, so lane behaviour and its tests agree on who
|
||||
// runs first.
|
||||
func laneNames() []string {
|
||||
names := make([]string, 0, len(sites))
|
||||
for name := range sites {
|
||||
names = append(names, name)
|
||||
}
|
||||
sort.Strings(names)
|
||||
return names
|
||||
}
|
||||
|
||||
func (p *Poller) Run(ctx context.Context) {
|
||||
log.Printf("latest-chapter poller: interval=%s cooldown=%s browser-cooldown=%s batch=%d stagger=%s",
|
||||
p.Interval, p.Cooldown, p.BrowserCooldown, p.Batch, p.Stagger)
|
||||
t := time.NewTicker(p.Interval)
|
||||
defer t.Stop()
|
||||
names := laneNames()
|
||||
log.Printf("latest-chapter poller: %d lanes, rest=%s gap=%s", len(names), defaultRest, defaultGap)
|
||||
for _, name := range names {
|
||||
go p.lane(ctx, name)
|
||||
}
|
||||
<-ctx.Done()
|
||||
log.Println("latest-chapter poller: stopped")
|
||||
}
|
||||
|
||||
// lane is one Site's Poll Lane: one pass, then sleep the pace the pass
|
||||
// reported, then another pass, until ctx is cancelled. The sleep is the whole
|
||||
// pace discipline — a pass that fetched nothing still reports its gap so the
|
||||
// Lane wakes often enough to notice Series as they become due. The pass shares
|
||||
// the Poller's browser-down state, so a sidecar loss is noticed once and the
|
||||
// other browser Lanes skip passes until the backoff window decays.
|
||||
func (p *Poller) lane(ctx context.Context, name string) {
|
||||
for {
|
||||
pace := p.runLanePass(ctx, name, true)
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
log.Println("latest-chapter poller: stopped")
|
||||
return
|
||||
case <-t.C:
|
||||
p.runOnce(ctx)
|
||||
case <-time.After(pace):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// runOnce processes one batch of due series.
|
||||
// runOnce processes one round: one pass of every Lane, back to back, no real
|
||||
// time passing. This is the deterministic entry point the test suite drives a
|
||||
// round at a time. The production Run loop does the same work paced by its own
|
||||
// sleeps; pacing is the only difference.
|
||||
func (p *Poller) runOnce(ctx context.Context) {
|
||||
for _, name := range laneNames() {
|
||||
p.runLanePass(ctx, name, false)
|
||||
}
|
||||
}
|
||||
|
||||
// runLanePass processes one pass of one Site's Lane: select the due Series,
|
||||
// pace through them, and report how long the Lane should wait before its next
|
||||
// pass. paced spaces consecutive fetches by the Site's effective gap — the
|
||||
// production Lane's rate limit; the deterministic test entry runs back to back.
|
||||
func (p *Poller) runLanePass(ctx context.Context, name string, paced bool) time.Duration {
|
||||
now := p.Now()
|
||||
cutoff := now.Add(-p.Cooldown).UnixMilli()
|
||||
browserCutoff := now.Add(-p.BrowserCooldown).UnixMilli()
|
||||
due, err := p.Store.DueForLatestCheck(cutoff, browserCutoff, browserBackedSites(), p.Batch)
|
||||
if err != nil {
|
||||
log.Printf("latest poll: due query: %v", err)
|
||||
return
|
||||
if until := p.refusalBackoff(name); now.Before(until) {
|
||||
// Cooling down after a refusal: do not attempt this Site at all.
|
||||
return until.Sub(now)
|
||||
}
|
||||
if isBrowserSite(name) {
|
||||
if downFor, down := p.browserDownFor(now); down && downFor < refuseBackoff {
|
||||
// A sibling browser Lane lost the sidecar within the backoff
|
||||
// window: skip this pass, so a restarting Chrome does not stamp
|
||||
// this Site's Series one pass at a time. After refuseBackoff the
|
||||
// flag decays and the Lane probes again (issue #100, story 20).
|
||||
log.Printf("latest poll %s: browser lane skipping pass (sidecar down %s ago)", name, downFor)
|
||||
return refuseBackoff - downFor
|
||||
}
|
||||
}
|
||||
s := sites[name]
|
||||
f := fetcherFor(name, p.BrowserFetch, p.Fetch)
|
||||
if f == nil {
|
||||
// No fetcher at all right now (browser absent, no fallback): every
|
||||
// Series stays unstamped and due, so a browser that appears after a
|
||||
// restart finds its full queue waiting (issue #100).
|
||||
return defaultGap
|
||||
}
|
||||
|
||||
due, err := p.Store.DueForLatestCheck(name, now.Add(-s.Rest).UnixMilli())
|
||||
if err != nil {
|
||||
log.Printf("latest poll %s: due query: %v", name, err)
|
||||
return defaultGap
|
||||
}
|
||||
if s.Browser != nil && f == p.BrowserFetch && !browserWakeDue(due, now, s.Rest) {
|
||||
// Below both thresholds Chrome stays asleep (ADR-0005 on-demand
|
||||
// browser): waking it for a single Poll would cost a challenge solve
|
||||
// per request.
|
||||
return defaultGap
|
||||
}
|
||||
if s.Browser != nil {
|
||||
// Browser Lanes share one tab, so their combined ceiling is about 360
|
||||
// Polls an hour. When they cannot keep up, the wait past the rest time
|
||||
// grows — log by how much, every pass, so the decision to give them
|
||||
// more pages is made from a measurement rather than a guess.
|
||||
if behind := maxSeriesWait(due, now, s.Rest) - s.Rest; behind > 0 {
|
||||
log.Printf("latest poll %s: browser lane behind by %s (browser Sites cannot keep up with the hour)", name, behind)
|
||||
}
|
||||
}
|
||||
|
||||
eligible, err := p.Store.EligibleSeriesCount(name)
|
||||
if err != nil {
|
||||
log.Printf("latest poll %s: eligible count: %v", name, err)
|
||||
return defaultGap
|
||||
}
|
||||
gap, clamped := effectiveGap(s, eligible)
|
||||
if clamped {
|
||||
log.Printf("latest poll %s: gap clamped to %s floor (eligible series=%d)", name, minGap, eligible)
|
||||
}
|
||||
if eligible == 0 {
|
||||
// Nothing to poll for the foreseeable future; sleep a full rest instead
|
||||
// of re-querying every gap.
|
||||
return s.Rest
|
||||
}
|
||||
|
||||
refusals := 0
|
||||
checked := 0
|
||||
for i, sr := range due {
|
||||
if ctx.Err() != nil {
|
||||
break
|
||||
}
|
||||
// Staggered rather than fired together: a burst of simultaneous requests
|
||||
// from one server IP is the traffic shape most likely to move that IP's
|
||||
// bot score. This is the server-side analogue of the userscript's "one
|
||||
// series per navigation ... indistinguishable from browsing" (L455-456).
|
||||
stopped := false
|
||||
if i > 0 && p.Stagger > 0 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
stopped = true
|
||||
case <-time.After(p.Stagger):
|
||||
}
|
||||
}
|
||||
if stopped {
|
||||
if refusals >= 2 {
|
||||
// This Site refused twice in a row: the remaining Series are left
|
||||
// unstamped and due, and the Lane waits refuseBackoff before
|
||||
// trying it again.
|
||||
break
|
||||
}
|
||||
p.checkOne(ctx, sr)
|
||||
if paced && i > 0 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
break
|
||||
case <-time.After(gap):
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
break
|
||||
}
|
||||
}
|
||||
if err := p.checkOne(ctx, sr); err != nil {
|
||||
switch {
|
||||
case errors.Is(err, errChallengeHeld):
|
||||
refusals++
|
||||
case errors.Is(err, errBrowserInterrupted):
|
||||
p.setBrowserDown(now)
|
||||
log.Printf("latest poll %s: browser unreachable, browser lanes skipping passes for %s", name, refuseBackoff)
|
||||
return gap
|
||||
default:
|
||||
refusals = 0
|
||||
}
|
||||
} else {
|
||||
refusals = 0
|
||||
}
|
||||
checked++
|
||||
}
|
||||
// due vs checked is how you tell which constraint is binding: ticks that
|
||||
// report due=0 mean the cooldown is the limit, ticks that report due==batch
|
||||
// every time mean throughput is.
|
||||
log.Printf("latest poll: due=%d checked=%d", len(due), checked)
|
||||
if checked > 0 {
|
||||
log.Printf("latest poll %s: due=%d checked=%d", name, len(due), checked)
|
||||
}
|
||||
if refusals >= 2 {
|
||||
p.setRefusalBackoff(name, now.Add(refuseBackoff))
|
||||
log.Printf("latest poll %s: refused twice this run, waiting %s", name, refuseBackoff)
|
||||
return refuseBackoff
|
||||
}
|
||||
return gap
|
||||
}
|
||||
|
||||
func (p *Poller) refusalBackoff(name string) time.Time {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return p.refuseUntil[name]
|
||||
}
|
||||
|
||||
func (p *Poller) setRefusalBackoff(name string, until time.Time) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
if p.refuseUntil == nil {
|
||||
p.refuseUntil = make(map[string]time.Time)
|
||||
}
|
||||
p.refuseUntil[name] = until
|
||||
}
|
||||
|
||||
// setBrowserDown records when a browser Lane lost the sidecar. It is Poller
|
||||
// state rather than pass state so the other browser Lanes see it too.
|
||||
func (p *Poller) setBrowserDown(now time.Time) {
|
||||
p.mu.Lock()
|
||||
p.browserDownAt = now
|
||||
p.mu.Unlock()
|
||||
}
|
||||
|
||||
// browserDownFor reports how long the sidecar has been down and that it is
|
||||
// down at all — the zero time means never down, which must not read as a
|
||||
// zero-duration loss. The window decays: once refuseBackoff passes without a
|
||||
// fresh loss, Lanes probe again.
|
||||
func (p *Poller) browserDownFor(now time.Time) (time.Duration, bool) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
if p.browserDownAt.IsZero() {
|
||||
return 0, false
|
||||
}
|
||||
return now.Sub(p.browserDownAt), true
|
||||
}
|
||||
|
||||
// isBrowserSite reports whether the registry routes this Site's page through
|
||||
// the browser sidecar.
|
||||
func isBrowserSite(name string) bool {
|
||||
return sites[name].Browser != nil
|
||||
}
|
||||
|
||||
// browserWakeDue reports whether a browser Lane may start a run: five or more
|
||||
// of its Series are due, or any one of them has been due for browserWakeAge.
|
||||
// Below both thresholds the Lane leaves Chrome asleep — Series Polled together
|
||||
// become due together, so the group naturally stays clustered, and the age
|
||||
// rule exists to stop a Series that drifted out of the group from starving.
|
||||
func browserWakeDue(due []store.Series, now time.Time, rest time.Duration) bool {
|
||||
if len(due) >= browserWakeCount {
|
||||
return true
|
||||
}
|
||||
return maxSeriesWait(due, now, rest) >= browserWakeAge
|
||||
}
|
||||
|
||||
// maxSeriesWait returns how long the most-overdue of the due Series has been
|
||||
// waiting past its due moment (0 when due is empty).
|
||||
func maxSeriesWait(due []store.Series, now time.Time, rest time.Duration) time.Duration {
|
||||
var oldest time.Duration
|
||||
for _, sr := range due {
|
||||
if w := now.Sub(time.UnixMilli(sr.LatestCheckedAt).Add(rest)); w > oldest {
|
||||
oldest = w
|
||||
}
|
||||
}
|
||||
return oldest
|
||||
}
|
||||
|
||||
// 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
|
||||
// batch or take down the process.
|
||||
func (p *Poller) checkOne(ctx context.Context, sr store.Series) {
|
||||
// Lane or take down the process. The returned error is the page read's
|
||||
// classified outcome so the Lane can tell a refusal from a loss of the
|
||||
// browser; non-classified failures still return nil-equivalent behaviour.
|
||||
func (p *Poller) checkOne(ctx context.Context, sr store.Series) error {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
log.Printf("latest poll %q: recovered from panic: %v", sr.Key(), r)
|
||||
@@ -216,44 +405,46 @@ func (p *Poller) checkOne(ctx context.Context, sr store.Series) {
|
||||
}()
|
||||
|
||||
// Stamped before the fetch, not after, so an error, a timeout, or a shutdown
|
||||
// mid-request still consumes the cooldown. Otherwise a renamed or deleted
|
||||
// series would be retried on every single tick forever. The userscript
|
||||
// stamps in the same order and for the same reason (L471-473).
|
||||
// mid-request still consumes the rest. Otherwise a renamed or deleted
|
||||
// series would be retried on every single pass forever. The userscript
|
||||
// stamps in the same order and for the same reason (L471-473). A Series
|
||||
// never reaches checkOne without a fetcher — runLanePass skips those — so
|
||||
// the stamp means "attempted", and an untried Series stays due.
|
||||
if err := p.Store.MarkLatestChecked(sr.Site, sr.SeriesID, p.Now().UnixMilli()); err != nil {
|
||||
log.Printf("latest poll %q: mark checked: %v", sr.Key(), err)
|
||||
return
|
||||
return nil
|
||||
}
|
||||
|
||||
facts, err := readSeriesPage(ctx, sr.Site, sr.SeriesURL, p.BrowserFetch, p.Fetch)
|
||||
if err != nil {
|
||||
switch {
|
||||
case errors.Is(err, errNotFetchable):
|
||||
// The cooldown above is already consumed, so a row that never
|
||||
// passes the gate is retried at cooldown pace rather than
|
||||
// The rest above is already consumed, so a row that never
|
||||
// passes the gate is retried at rest pace rather than
|
||||
// hot-looping.
|
||||
log.Printf("latest poll %q: not fetchable: site=%q url=%q", sr.Key(), sr.Site, sr.SeriesURL)
|
||||
return
|
||||
return err
|
||||
case errors.Is(err, errNoFetcher):
|
||||
log.Printf("latest poll %q: no fetcher for site %q", sr.Key(), sr.Site)
|
||||
return
|
||||
return err
|
||||
}
|
||||
// A legacy cover heals independently of the page read: its source may
|
||||
// answer — a CDN — while the origin does not, so a fetch failure does
|
||||
// not skip the heal, matching the order the shared read replaced.
|
||||
p.prefetchCover(ctx, sr)
|
||||
p.healCover(ctx, sr)
|
||||
log.Printf("latest poll %q: %v", sr.Key(), err)
|
||||
return
|
||||
return err
|
||||
}
|
||||
// A legacy cover source is healed independently of the page read.
|
||||
p.prefetchCover(ctx, sr)
|
||||
p.healCover(ctx, sr)
|
||||
// Cover fill is independent of the chapter signal: a page that lost its
|
||||
// chapter list may keep its og:image, and a blank Series heals either way.
|
||||
p.fillBlankCover(ctx, sr, facts.Cover)
|
||||
if !facts.HasLatest {
|
||||
// Most likely a challenge page or a layout change. Either way the row is
|
||||
// already stamped, so this waits out a cooldown instead of hot-looping.
|
||||
// already stamped, so this waits out a rest instead of hot-looping.
|
||||
log.Printf("latest poll %q: no chapter links in %d bytes", sr.Key(), facts.BodyLen)
|
||||
return
|
||||
return nil
|
||||
}
|
||||
|
||||
// Equality, not >, mirroring the userscript (L427): a site that retracts a
|
||||
@@ -261,7 +452,7 @@ func (p *Poller) checkOne(ctx context.Context, sr store.Series) {
|
||||
// against the due-query snapshot; a concurrent write in between only costs
|
||||
// one redundant UPDATE of the same absolute value, never a wrong one.
|
||||
if sr.LatestChapterNum != nil && *sr.LatestChapterNum == facts.Latest.Num {
|
||||
return
|
||||
return nil
|
||||
}
|
||||
|
||||
// Series-level write: the row is shared, so one update refreshes every
|
||||
@@ -270,9 +461,29 @@ func (p *Poller) checkOne(ctx context.Context, sr store.Series) {
|
||||
// the list.
|
||||
if err := p.Store.SetLatestChapter(sr.Site, sr.SeriesID, facts.Latest.Label, facts.Latest.Num); err != nil {
|
||||
log.Printf("latest poll %q: set latest chapter: %v", sr.Key(), err)
|
||||
return
|
||||
return nil
|
||||
}
|
||||
log.Printf("latest poll %q: latest is now %s", sr.Key(), facts.Latest.Label)
|
||||
return nil
|
||||
}
|
||||
|
||||
// healCover runs prefetchCover in the background. Cover bytes come from a
|
||||
// different host — often a CDN — and heal once in a Series's life, so they
|
||||
// must not consume a Lane's gap: a large import with many blanks would
|
||||
// otherwise make every Latest Chapter go stale behind a slow image host
|
||||
// (issue #100).
|
||||
func (p *Poller) healCover(ctx context.Context, sr store.Series) {
|
||||
p.coverWG.Add(1)
|
||||
go func() {
|
||||
defer p.coverWG.Done()
|
||||
p.prefetchCover(ctx, sr)
|
||||
}()
|
||||
}
|
||||
|
||||
// waitCovers blocks until every in-flight cover heal finishes. Tests call it
|
||||
// after a round before asserting on cover fetches.
|
||||
func (p *Poller) waitCovers() {
|
||||
p.coverWG.Wait()
|
||||
}
|
||||
|
||||
// fetchableSeriesURL reports whether site is a Site the registry knows and
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"strings"
|
||||
@@ -156,19 +157,16 @@ func (f *fakeBytesCoverFetcher) callCount() int {
|
||||
return len(f.calls)
|
||||
}
|
||||
|
||||
// newTestPoller wires a poller with a frozen clock and no stagger, so tests run
|
||||
// instantly and deterministically.
|
||||
// newTestPoller wires a poller with a frozen clock. Rest and gap come from the
|
||||
// Site registry, so tests seed checked_at relative to the one-hour rest; the
|
||||
// round entry point (runOnce) runs every Lane back to back with no real
|
||||
// pacing, so tests stay instant and deterministic.
|
||||
func newTestPoller(t *testing.T, s *store.Store, f Fetcher, at time.Time) *Poller {
|
||||
t.Helper()
|
||||
return &Poller{
|
||||
Store: s,
|
||||
Fetch: f,
|
||||
Now: func() time.Time { return at },
|
||||
Cooldown: time.Hour,
|
||||
BrowserCooldown: 6 * time.Hour,
|
||||
Interval: 10 * time.Minute,
|
||||
Stagger: 0,
|
||||
Batch: 14,
|
||||
Store: s,
|
||||
Fetch: f,
|
||||
Now: func() time.Time { return at },
|
||||
}
|
||||
}
|
||||
|
||||
@@ -208,9 +206,10 @@ func TestRunOncePrefetchesPublicCover(t *testing.T) {
|
||||
covers := &fakeBytesCoverFetcher{body: []byte("cover-bytes"), contentType: "image/jpeg"}
|
||||
p := &Poller{
|
||||
Store: s, Fetch: &fakeFetcher{body: asuraSeriesFixture, status: 200}, CoverBytesFetch: covers,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) }, Cooldown: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
|
||||
if got := covers.callCount(); got != 1 {
|
||||
t.Fatalf("cover fetch calls = %d, want 1", got)
|
||||
@@ -241,9 +240,10 @@ func TestRunOnceDoesNotStoreNonImagePublicCover(t *testing.T) {
|
||||
p := &Poller{
|
||||
Store: s, Fetch: &fakeFetcher{body: asuraSeriesFixture, status: 200},
|
||||
CoverBytesFetch: &fakeBytesCoverFetcher{body: []byte("challenge"), contentType: "text/html"},
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) }, Cooldown: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
|
||||
if _, _, found, err := s.GetCover(coverURL); err != nil || found {
|
||||
t.Fatalf("non-image cover = found %v, err %v; want missing", found, err)
|
||||
@@ -350,23 +350,6 @@ func TestRunOnceMarksCheckedOnFailure(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunOnceRespectsBatchLimit(t *testing.T) {
|
||||
s, _ := newTestStore(t)
|
||||
for i := 0; i < 20; i++ {
|
||||
key := "asura:s" + string(rune('a'+i))
|
||||
seedForCheck(t, s, key, "https://asurascans.com/comics/"+key, 0)
|
||||
}
|
||||
|
||||
f := &fakeFetcher{body: "", status: 200}
|
||||
p := newTestPoller(t, s, f, time.UnixMilli(5_000_000))
|
||||
p.Batch = 5
|
||||
p.runOnce(context.Background())
|
||||
|
||||
if got := f.callCount(); got != 5 {
|
||||
t.Fatalf("fetched %d series, want 5 (batch limit)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// The point of the split (ADR-0003): a series referenced by several bookmarks
|
||||
// is fetched once per due cycle, not once per bookmark. Two bookmarks share a
|
||||
// series when two readers track it (issue #22).
|
||||
@@ -409,8 +392,8 @@ func TestRunOnceFetchesSharedSeriesOnce(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// One unreachable series must not abandon the rest of the batch.
|
||||
func TestRunOnceOneBadSeriesDoesNotStallBatch(t *testing.T) {
|
||||
// One unreachable series must not abandon the rest of the Lane.
|
||||
func TestRunOnceOneBadSeriesDoesNotStallLane(t *testing.T) {
|
||||
s, _ := newTestStore(t)
|
||||
keys := []string{"asura:a", "asura:b", "asura:c", "asura:d", "asura:e"}
|
||||
for _, k := range keys {
|
||||
@@ -436,7 +419,9 @@ func TestRunOnceOneBadSeriesDoesNotStallBatch(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunLogsCooldowns(t *testing.T) {
|
||||
// Pace now lives in the registry, not config: Run logs the lane defaults so a
|
||||
// deployment can see what the poller is doing without reading the source.
|
||||
func TestRunLogsLaneDefaults(t *testing.T) {
|
||||
var logs strings.Builder
|
||||
previous := log.Writer()
|
||||
log.SetOutput(&logs)
|
||||
@@ -444,21 +429,19 @@ func TestRunLogsCooldowns(t *testing.T) {
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
(&Poller{
|
||||
Cooldown: time.Hour,
|
||||
BrowserCooldown: 6 * time.Hour,
|
||||
Interval: time.Hour,
|
||||
}).Run(ctx)
|
||||
(&Poller{Now: func() time.Time { return time.Now() }}).Run(ctx)
|
||||
|
||||
if got := logs.String(); !strings.Contains(got, "cooldown=1h") ||
|
||||
!strings.Contains(got, "browser-cooldown=6h") {
|
||||
t.Fatalf("startup log = %q, want both cooldowns", got)
|
||||
got := logs.String()
|
||||
for _, want := range []string{"6 lanes", "rest=1h0m0s", "gap=10s"} {
|
||||
if !strings.Contains(got, want) {
|
||||
t.Fatalf("startup log = %q, want %q", got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The cooldown is enforced by the due query, so a second immediate pass must do
|
||||
// nothing at all — this is what makes the tick interval independent of it.
|
||||
func TestRunOnceHonoursCooldownAcrossPasses(t *testing.T) {
|
||||
// The rest is enforced by the due query, so a second immediate pass must do
|
||||
// nothing at all — this is what makes the Lane's sleep independent of it.
|
||||
func TestRunOnceHonoursRestAcrossPasses(t *testing.T) {
|
||||
s, _ := newTestStore(t)
|
||||
const url = "https://asurascans.com/comics/x"
|
||||
seedForCheck(t, s, "asura:x", url, 0)
|
||||
@@ -471,56 +454,18 @@ func TestRunOnceHonoursCooldownAcrossPasses(t *testing.T) {
|
||||
if got := f.callCount(); got != 1 {
|
||||
t.Fatalf("first pass fetched %d, want 1", got)
|
||||
}
|
||||
// Same instant, and again 59 minutes later: both inside the 1h cooldown.
|
||||
// Same instant, and again 59 minutes later: both inside the 1h rest.
|
||||
p.runOnce(context.Background())
|
||||
p.Now = func() time.Time { return now.Add(59 * time.Minute) }
|
||||
p.runOnce(context.Background())
|
||||
if got := f.callCount(); got != 1 {
|
||||
t.Fatalf("fetched %d times inside the cooldown, want 1", got)
|
||||
t.Fatalf("fetched %d times inside the rest, want 1", got)
|
||||
}
|
||||
// Past the cooldown, it is due again.
|
||||
// Past the rest, it is due again.
|
||||
p.Now = func() time.Time { return now.Add(61 * time.Minute) }
|
||||
p.runOnce(context.Background())
|
||||
if got := f.callCount(); got != 2 {
|
||||
t.Fatalf("fetched %d times after the cooldown, want 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunOnceUsesBrowserCooldown(t *testing.T) {
|
||||
s, _ := newTestStore(t)
|
||||
const browserKey = "kagane:019f84bc-9ba0-7ed9-86f5-8b905ec7c28b"
|
||||
seedForCheck(t, s, "asura:plain", "https://asurascans.com/comics/plain", 0)
|
||||
seedForCheck(t, s, browserKey, "https://kagane.to/series/019f84bc-9ba0-7ed9-86f5-8b905ec7c28b", 0)
|
||||
|
||||
hour := time.Hour
|
||||
now := time.Unix(2*int64(hour/time.Second), 0)
|
||||
tls := &fakeFetcher{status: 200}
|
||||
browser := &fakeFetcher{status: 200}
|
||||
p := &Poller{
|
||||
Store: s,
|
||||
Fetch: tls,
|
||||
BrowserFetch: browser,
|
||||
Now: func() time.Time { return now },
|
||||
Cooldown: hour,
|
||||
BrowserCooldown: 6 * hour,
|
||||
Batch: 10,
|
||||
}
|
||||
|
||||
p.runOnce(context.Background())
|
||||
if got := tls.callCount(); got != 1 {
|
||||
t.Fatalf("plain-TLS fetches after 2h = %d, want 1", got)
|
||||
}
|
||||
if got := browser.callCount(); got != 0 {
|
||||
t.Fatalf("browser fetches after 2h = %d, want 0", got)
|
||||
}
|
||||
|
||||
now = time.Unix(7*int64(hour/time.Second), 0)
|
||||
p.runOnce(context.Background())
|
||||
if got := tls.callCount(); got != 2 {
|
||||
t.Fatalf("plain-TLS fetches after 7h = %d, want 2", got)
|
||||
}
|
||||
if got := browser.callCount(); got != 1 {
|
||||
t.Fatalf("browser fetches after 7h = %d, want 1", got)
|
||||
t.Fatalf("fetched %d times after the rest, want 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -665,17 +610,18 @@ func TestKaganeSkippedWhenNoBrowserFetcher(t *testing.T) {
|
||||
f := &fakeFetcher{body: kaganeAPIFixture, status: 200}
|
||||
|
||||
p := &Poller{
|
||||
Store: s,
|
||||
Fetch: f,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
Cooldown: time.Hour, BrowserCooldown: time.Hour,
|
||||
Interval: time.Hour, Batch: 10,
|
||||
Store: s,
|
||||
Fetch: f,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
|
||||
if len(f.calls) != 0 {
|
||||
t.Errorf("TLS fetcher was called for kagane: %v", f.calls)
|
||||
}
|
||||
if got := readLatestCheckedAt(t, s, "kagane:019f84bc-9ba0-7ed9-86f5-8b905ec7c28b"); got != 0 {
|
||||
t.Errorf("latest_checked_at = %d, want 0 (untried stays due until a browser appears)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// novelfull without a browser is not skipped outright: its challenge is a
|
||||
@@ -699,11 +645,10 @@ func TestNovelfullUsesTLSWhenNoBrowserFetcher(t *testing.T) {
|
||||
covers := &fakeBytesCoverFetcher{body: []byte("cover-bytes"), contentType: "image/webp"}
|
||||
p := &Poller{
|
||||
Store: s, Fetch: tlsF, CoverBytesFetch: covers,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
Cooldown: time.Hour, BrowserCooldown: time.Hour,
|
||||
Interval: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
|
||||
if len(tlsF.calls) != 1 {
|
||||
t.Fatalf("TLS fetcher calls = %d, want 1", len(tlsF.calls))
|
||||
@@ -741,9 +686,7 @@ func TestKaganeUsesBrowserFetcher(t *testing.T) {
|
||||
browserF := &fakeFetcher{body: kaganeAPIFixture, status: 200}
|
||||
p := &Poller{
|
||||
Store: s, Fetch: tlsF, BrowserFetch: browserF,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
Cooldown: time.Hour, BrowserCooldown: time.Hour,
|
||||
Interval: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
|
||||
@@ -789,11 +732,10 @@ func TestComixUsesBrowserFetcher(t *testing.T) {
|
||||
p := &Poller{
|
||||
Store: s, Fetch: tlsF, BrowserFetch: browserF,
|
||||
CoverFetch: covers, CoverBytesFetch: tlsCovers,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
Cooldown: time.Hour, BrowserCooldown: time.Hour,
|
||||
Interval: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
|
||||
if len(tlsF.calls) != 0 {
|
||||
t.Errorf("TLS fetcher was called for comix: %v", tlsF.calls)
|
||||
@@ -831,15 +773,16 @@ func TestComixSkippedWhenNoBrowserFetcher(t *testing.T) {
|
||||
f := &fakeFetcher{body: comixSeriesFixture, status: 200}
|
||||
p := &Poller{
|
||||
Store: s, Fetch: f,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
Cooldown: time.Hour, BrowserCooldown: time.Hour,
|
||||
Interval: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
|
||||
if len(f.calls) != 0 {
|
||||
t.Errorf("TLS fetcher was called for comix: %v", f.calls)
|
||||
}
|
||||
if got := readLatestCheckedAt(t, s, "comix:n8we-dungeons-and-crayons"); got != 0 {
|
||||
t.Errorf("latest_checked_at = %d, want 0 (untried stays due until a browser appears)", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunOncePrefetchesKaganeCover(t *testing.T) {
|
||||
@@ -860,10 +803,10 @@ func TestRunOncePrefetchesKaganeCover(t *testing.T) {
|
||||
p := &Poller{
|
||||
Store: s, Fetch: &fakeFetcher{body: kaganeAPIFixture, status: 200},
|
||||
BrowserFetch: &fakeFetcher{body: kaganeAPIFixture, status: 200}, CoverFetch: covers,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
Cooldown: time.Hour, BrowserCooldown: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
|
||||
body, contentType, ok, err := s.CoverByAddress(store.CoverAddress(coverURL))
|
||||
if err != nil || !ok {
|
||||
@@ -898,11 +841,13 @@ func TestRunOnceDoesNotRefetchKaganeCover(t *testing.T) {
|
||||
covers := &fakeCoverFetcher{body: []byte("cover-bytes"), contentType: "image/webp"}
|
||||
p := &Poller{
|
||||
Store: s, BrowserFetch: &fakeFetcher{body: kaganeAPIFixture, status: 200}, CoverFetch: covers,
|
||||
Now: func() time.Time { return at }, Cooldown: time.Hour, BrowserCooldown: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return at },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
at = at.Add(2 * time.Hour)
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
|
||||
if got := covers.callCount(); got != 1 {
|
||||
t.Fatalf("cover fetch calls = %d, want 1 after two due cycles", got)
|
||||
@@ -927,9 +872,10 @@ func TestRunOnceCoverFailureDoesNotBlockChapter(t *testing.T) {
|
||||
p := &Poller{
|
||||
Store: s, BrowserFetch: &fakeFetcher{body: kaganeAPIFixture, status: 200},
|
||||
CoverFetch: &fakeCoverFetcher{err: errors.New("browser unavailable")},
|
||||
Now: func() time.Time { return now }, Cooldown: time.Hour, BrowserCooldown: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return now },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
|
||||
got, found, err := s.Get(s.OwnerID(), key)
|
||||
if err != nil || !found {
|
||||
@@ -960,9 +906,10 @@ func TestRunOnceRejectsInvalidKaganeCover(t *testing.T) {
|
||||
p := &Poller{
|
||||
Store: s, BrowserFetch: &fakeFetcher{body: kaganeAPIFixture, status: 200},
|
||||
CoverFetch: &fakeCoverFetcher{body: []byte("not an image"), contentType: "text/html"},
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) }, Cooldown: time.Hour, BrowserCooldown: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
|
||||
if _, _, found, err := s.CoverByAddress(store.CoverAddress(coverURL)); err != nil || found {
|
||||
t.Fatalf("invalid cover persisted = %v, err %v; want missing", found, err)
|
||||
@@ -985,9 +932,10 @@ func TestRunOnceWithoutCoverFetcherStillPollsKagane(t *testing.T) {
|
||||
seedCoverSource(t, dbURL, "kagane", seriesID, coverURL)
|
||||
p := &Poller{
|
||||
Store: s, BrowserFetch: &fakeFetcher{body: kaganeAPIFixture, status: 200},
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) }, Cooldown: time.Hour, BrowserCooldown: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
|
||||
if _, _, found, err := s.CoverByAddress(store.CoverAddress(coverURL)); err != nil || found {
|
||||
t.Fatalf("cover after nil CoverFetch = found %v, err %v; want missing", found, err)
|
||||
@@ -1010,9 +958,10 @@ func TestRunOnceRoutesNonKaganeCoverToPublicFetcher(t *testing.T) {
|
||||
p := &Poller{
|
||||
Store: s, Fetch: &fakeFetcher{body: asuraSeriesFixture, status: 200}, CoverFetch: browserCovers,
|
||||
CoverBytesFetch: publicCovers,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) }, Cooldown: time.Hour, BrowserCooldown: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
|
||||
if got := publicCovers.callCount(); got != 1 {
|
||||
t.Fatalf("public cover fetch calls = %d, want 1", got)
|
||||
@@ -1128,10 +1077,10 @@ func TestRunOnceFillsBlankCoverFromSeriesPage(t *testing.T) {
|
||||
p := &Poller{
|
||||
Store: s, Fetch: page, BrowserFetch: page,
|
||||
CoverBytesFetch: public, CoverFetch: browser,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
Cooldown: time.Hour, BrowserCooldown: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
|
||||
got := readBookmark(t, s, tc.key)
|
||||
wantWire := testCoverBaseURL + "/covers/" + store.CoverAddress(tc.wantCover)
|
||||
@@ -1181,7 +1130,7 @@ func TestRunOnceDoesNotReplaceExistingCover(t *testing.T) {
|
||||
p := &Poller{
|
||||
Store: s, Fetch: &fakeFetcher{body: asuraSeriesFixture + asuraCoverFixture, status: 200},
|
||||
CoverBytesFetch: public,
|
||||
Now: func() time.Time { return at }, Cooldown: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return at },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
at = at.Add(2 * time.Hour)
|
||||
@@ -1216,9 +1165,10 @@ func TestRunOnceRetriesFailedBlankCoverOnNextPoll(t *testing.T) {
|
||||
p := &Poller{
|
||||
Store: s, Fetch: &fakeFetcher{body: asuraSeriesFixture + asuraCoverFixture, status: 200},
|
||||
CoverBytesFetch: public,
|
||||
Now: func() time.Time { return at }, Cooldown: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return at },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
if got := readBookmark(t, s, key); got.Cover != "" {
|
||||
t.Fatalf("Cover after failed fetch = %q, want blank", got.Cover)
|
||||
}
|
||||
@@ -1231,6 +1181,7 @@ func TestRunOnceRetriesFailedBlankCoverOnNextPoll(t *testing.T) {
|
||||
public.contentType = "image/jpeg"
|
||||
at = at.Add(2 * time.Hour)
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
|
||||
if got := public.callCount(); got != 2 {
|
||||
t.Fatalf("cover fetch calls after retry = %d, want 2", got)
|
||||
@@ -1264,9 +1215,10 @@ func TestRunOnceBlankCoverFailureDoesNotBlockChapter(t *testing.T) {
|
||||
p := &Poller{
|
||||
Store: s, Fetch: &fakeFetcher{body: asuraSeriesFixture + asuraCoverFixture, status: 200},
|
||||
CoverBytesFetch: &fakeBytesCoverFetcher{err: errors.New("cdn down")},
|
||||
Now: func() time.Time { return now }, Cooldown: time.Hour, Batch: 10,
|
||||
Now: func() time.Time { return now },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
p.waitCovers()
|
||||
|
||||
got := readBookmark(t, s, key)
|
||||
if got.LatestChapterNum == nil || *got.LatestChapterNum != 181 {
|
||||
@@ -1289,3 +1241,342 @@ const kaganeAPIFixtureWithCover = `
|
||||
{"book_id":"b","title":"Episode 41","chapter_no":"41","sort_no":41},
|
||||
{"book_id":"c","title":"Episode 40.5","chapter_no":"40.5","sort_no":40}]}
|
||||
`
|
||||
|
||||
func TestEffectiveGap(t *testing.T) {
|
||||
asura := sites["asura"]
|
||||
tests := []struct {
|
||||
eligible int
|
||||
want time.Duration
|
||||
clamped bool
|
||||
}{
|
||||
{0, 10 * time.Second, false},
|
||||
{5, 10 * time.Second, false},
|
||||
{360, 10 * time.Second, false},
|
||||
{720, 5 * time.Second, false},
|
||||
{3600, time.Second, false},
|
||||
{4000, time.Second, true},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
got, clamped := effectiveGap(asura, tt.eligible)
|
||||
if got != tt.want || clamped != tt.clamped {
|
||||
t.Errorf("effectiveGap(asura, %d) = (%s, %v), want (%s, %v)",
|
||||
tt.eligible, got, clamped, tt.want, tt.clamped)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The due query orders by sharedness first, then age: a series two readers
|
||||
// track is polled before a single-reader series that has waited far longer
|
||||
// (ADR-0003, issue #100).
|
||||
func TestRunOnceOrdersBySharednessThenAge(t *testing.T) {
|
||||
s, url := newTestStore(t)
|
||||
other, err := store.Open(url, store.Owner{DiscordID: "second-reader", TokenHash: sha256.Sum256([]byte("second-token-hash"))}, t.TempDir(), testCoverBaseURL)
|
||||
if err != nil {
|
||||
t.Fatalf("Open second reader: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { other.Close() })
|
||||
|
||||
now := time.UnixMilli(5_000_000)
|
||||
// popular: two readers, due for 2 minutes. loner: one reader, due for 23
|
||||
// minutes. Popularity must win — the loner waited far longer.
|
||||
const (
|
||||
popularKey = "asura:popular"
|
||||
lonerKey = "asura:loner"
|
||||
popularURL = "https://asurascans.com/comics/popular"
|
||||
lonerURL = "https://asurascans.com/comics/loner"
|
||||
)
|
||||
seedForCheck(t, s, popularKey, popularURL, now.Add(-62*time.Minute).UnixMilli())
|
||||
seedForCheck(t, s, lonerKey, lonerURL, 0)
|
||||
if _, err := s.Upsert(other.OwnerID(), store.Bookmark{
|
||||
Key: popularKey, Site: "asura", SeriesID: "popular", UpdatedAt: 2000,
|
||||
}); err != nil {
|
||||
t.Fatalf("seed second reader: %v", err)
|
||||
}
|
||||
|
||||
f := &fakeFetcher{body: asuraSeriesFixture, status: 200}
|
||||
newTestPoller(t, s, f, now).runOnce(context.Background())
|
||||
|
||||
if len(f.calls) != 2 {
|
||||
t.Fatalf("fetches = %d, want 2", len(f.calls))
|
||||
}
|
||||
if f.calls[0] != popularURL {
|
||||
t.Fatalf("first fetch = %q, want the shared series %q", f.calls[0], popularURL)
|
||||
}
|
||||
}
|
||||
|
||||
// Two refusals in one pass stop the Lane: the remaining Series stay unstamped
|
||||
// and due, and the Lane backs off for refuseBackoff before trying the Site
|
||||
// again. One hostile Site burns only its own Lane's budget (issue #100).
|
||||
func TestRunOnceSiteRefusalSkipsRestOfLaneAndBacksOff(t *testing.T) {
|
||||
s, _ := newTestStore(t)
|
||||
now := time.UnixMilli(5_000_000)
|
||||
const n = 4
|
||||
for i := 0; i < n; i++ {
|
||||
key := fmt.Sprintf("kagane:s%d", i)
|
||||
seedForCheck(t, s, key, "https://kagane.to/series/"+key[7:], 0)
|
||||
}
|
||||
|
||||
browser := &fakeFetcher{status: 403}
|
||||
tls := &fakeFetcher{status: 200}
|
||||
// An asura Series sits on its own Lane: it must still be polled while
|
||||
// kagane's Lane burns its budget on refusals (story 4).
|
||||
seedForCheck(t, s, "asura:still-polls", "https://asurascans.com/series/still-polls", 0)
|
||||
p := &Poller{
|
||||
Store: s, Fetch: tls, BrowserFetch: browser,
|
||||
Now: func() time.Time { return now },
|
||||
}
|
||||
|
||||
p.runOnce(context.Background())
|
||||
if got := browser.callCount(); got != 2 {
|
||||
t.Fatalf("browser fetches after first pass = %d, want 2 (refused twice)", got)
|
||||
}
|
||||
if got := tls.callCount(); got != 1 {
|
||||
t.Fatalf("asura fetches after first pass = %d, want 1 (its Lane is independent)", got)
|
||||
}
|
||||
stamped := 0
|
||||
for i := 0; i < n; i++ {
|
||||
if readLatestCheckedAt(t, s, fmt.Sprintf("kagane:s%d", i)) == now.UnixMilli() {
|
||||
stamped++
|
||||
}
|
||||
}
|
||||
if stamped != 2 {
|
||||
t.Fatalf("stamped series = %d, want 2; the remaining two stay due", stamped)
|
||||
}
|
||||
|
||||
// Inside the backoff window nothing is attempted.
|
||||
p.Now = func() time.Time { return now.Add(14 * time.Minute) }
|
||||
p.runOnce(context.Background())
|
||||
if got := browser.callCount(); got != 2 {
|
||||
t.Fatalf("browser fetches inside backoff = %d, want still 2", got)
|
||||
}
|
||||
|
||||
// Past the backoff the Lane resumes and the two untried Series are polled.
|
||||
p.Now = func() time.Time { return now.Add(16 * time.Minute) }
|
||||
p.runOnce(context.Background())
|
||||
if got := browser.callCount(); got != 4 {
|
||||
t.Fatalf("browser fetches after backoff = %d, want 4", got)
|
||||
}
|
||||
for i := 0; i < n; i++ {
|
||||
if got := readLatestCheckedAt(t, s, fmt.Sprintf("kagane:s%d", i)); got == 0 {
|
||||
t.Fatalf("kagane:s%d still untried after backoff", i)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Below five due Series with none waiting long, a browser Lane leaves Chrome
|
||||
// asleep; five due, or one waiting browserWakeAge, wakes it (ADR-0005).
|
||||
func TestBrowserLaneWakeThresholds(t *testing.T) {
|
||||
s, _ := newTestStore(t)
|
||||
now := time.UnixMilli(5_000_000)
|
||||
// Freshly due: checked two minutes before the rest elapses, so the wait
|
||||
// is far below browserWakeAge.
|
||||
seed := func(i int) {
|
||||
key := fmt.Sprintf("kagane:w%d", i)
|
||||
seedForCheck(t, s, key, "https://kagane.to/series/"+key[7:], now.Add(-62*time.Minute).UnixMilli())
|
||||
}
|
||||
for i := 0; i < 3; i++ {
|
||||
seed(i)
|
||||
}
|
||||
|
||||
browser := &fakeFetcher{body: kaganeAPIFixture, status: 200}
|
||||
p := &Poller{
|
||||
Store: s, Fetch: &fakeFetcher{status: 200}, BrowserFetch: browser,
|
||||
Now: func() time.Time { return now },
|
||||
}
|
||||
|
||||
p.runOnce(context.Background())
|
||||
if got := browser.callCount(); got != 0 {
|
||||
t.Fatalf("browser fetches with 3 freshly-due series = %d, want 0 (Chrome stays asleep)", got)
|
||||
}
|
||||
// 5 due crosses the count threshold.
|
||||
for i := 3; i < 5; i++ {
|
||||
seed(i)
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
if got := browser.callCount(); got != 5 {
|
||||
t.Fatalf("browser fetches with 5 due series = %d, want 5", got)
|
||||
}
|
||||
// A single long-neglected series wakes the browser by age alone.
|
||||
seedForCheck(t, s, "kagane:ancient", "https://kagane.to/series/ancient", 0)
|
||||
p.runOnce(context.Background())
|
||||
if got := browser.callCount(); got != 6 {
|
||||
t.Fatalf("browser fetches with one ancient series = %d, want 6", got)
|
||||
}
|
||||
}
|
||||
|
||||
// When one browser Lane loses the sidecar, the round's remaining browser
|
||||
// Lanes are skipped: every fetch would fail anyway, and their Series must not
|
||||
// burn their stamps on a dead Chrome (issue #100).
|
||||
func TestRunOnceUnreachableBrowserStopsBrowserLanes(t *testing.T) {
|
||||
s, _ := newTestStore(t)
|
||||
now := time.UnixMilli(5_000_000)
|
||||
seedForCheck(t, s, "comix:c", "https://comix.to/title/c", 0)
|
||||
seedForCheck(t, s, "kagane:k", "https://kagane.to/series/k", 0)
|
||||
seedForCheck(t, s, "novelfull:n", "https://novelfull.com/n.html", 0)
|
||||
|
||||
interrupted := fmt.Errorf("%w: %w", errBrowserInterrupted, errors.New("restart"))
|
||||
browser := &fakeFetcher{status: 200, err: interrupted}
|
||||
p := &Poller{
|
||||
Store: s, Fetch: &fakeFetcher{status: 200}, BrowserFetch: browser,
|
||||
Now: func() time.Time { return now },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
|
||||
// Sorted lane order: comix, kagane, novelfull. Only comix attempted.
|
||||
if got := browser.callCount(); got != 1 {
|
||||
t.Fatalf("browser fetches = %d, want 1 (only the first browser lane)", got)
|
||||
}
|
||||
if got := readLatestCheckedAt(t, s, "comix:c"); got != now.UnixMilli() {
|
||||
t.Fatalf("comix stamp = %d, want %d", got, now.UnixMilli())
|
||||
}
|
||||
for _, key := range []string{"kagane:k", "novelfull:n"} {
|
||||
if got := readLatestCheckedAt(t, s, key); got != 0 {
|
||||
t.Fatalf("%s stamp = %d, want 0 (untried)", key, got)
|
||||
}
|
||||
}
|
||||
|
||||
// The shared flag decays after refuseBackoff: the next round probes
|
||||
// again. comix's Series is resting (stamped last round), so the probe
|
||||
// falls to kagane — the only Lane with something due — and its fresh
|
||||
// loss re-gates the Lanes behind it.
|
||||
p.Now = func() time.Time { return now.Add(16 * time.Minute) }
|
||||
p.runOnce(context.Background())
|
||||
if got := browser.callCount(); got != 2 {
|
||||
t.Fatalf("browser fetches after backoff decay = %d, want 2 (kagane probes again)", got)
|
||||
}
|
||||
if got := readLatestCheckedAt(t, s, "comix:c"); got != now.UnixMilli() {
|
||||
t.Fatalf("comix stamp after decay = %d, want %d (resting, untouched)", got, now.UnixMilli())
|
||||
}
|
||||
if got := readLatestCheckedAt(t, s, "kagane:k"); got != now.Add(16*time.Minute).UnixMilli() {
|
||||
t.Fatalf("kagane stamp after decay = %d, want %d (the probe)", got, now.Add(16*time.Minute).UnixMilli())
|
||||
}
|
||||
if got := readLatestCheckedAt(t, s, "novelfull:n"); got != 0 {
|
||||
t.Fatalf("novelfull stamp after decay = %d, want 0 (gated by kagane's fresh loss)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// A browser Lane that cannot keep up logs the backlog every pass, so the
|
||||
// decision to give browser Sites more pages is measured, not guessed.
|
||||
func TestRunOnceLogsBrowserLaneBehind(t *testing.T) {
|
||||
s, _ := newTestStore(t)
|
||||
now := time.UnixMilli(5_000_000)
|
||||
// Checked three rests ago: two rests' worth of wait past the due moment.
|
||||
seedForCheck(t, s, "kagane:old", "https://kagane.to/series/old", now.Add(-3*time.Hour).UnixMilli())
|
||||
|
||||
var logs strings.Builder
|
||||
prev := log.Writer()
|
||||
log.SetOutput(&logs)
|
||||
t.Cleanup(func() { log.SetOutput(prev) })
|
||||
|
||||
p := &Poller{
|
||||
Store: s, Fetch: &fakeFetcher{status: 200},
|
||||
BrowserFetch: &fakeFetcher{body: kaganeAPIFixture, status: 200},
|
||||
Now: func() time.Time { return now },
|
||||
}
|
||||
p.runOnce(context.Background())
|
||||
|
||||
if got := logs.String(); !strings.Contains(got, "latest poll kagane: browser lane behind by 1h0m0s") {
|
||||
t.Fatalf("behind log = %q, want it to name kagane and the backlog", got)
|
||||
}
|
||||
}
|
||||
|
||||
// gatedCoverFetcher blocks every Fetch on a gate, so a test can hold a cover
|
||||
// heal in flight and prove a Lane does not wait for it.
|
||||
type gatedCoverFetcher struct {
|
||||
inner *fakeBytesCoverFetcher
|
||||
gate chan struct{}
|
||||
started chan struct{}
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
func (g *gatedCoverFetcher) Fetch(ctx context.Context, sourceURL string) ([]byte, string, error) {
|
||||
g.once.Do(func() { g.started <- struct{}{} })
|
||||
<-g.gate
|
||||
return g.inner.Fetch(ctx, sourceURL)
|
||||
}
|
||||
|
||||
// Cover heals run in the background: a heal stuck on a slow CDN must not
|
||||
// delay the Lane's next Series-page Poll, or a large import with many blanks
|
||||
// would make every Latest Chapter go stale (issue #100).
|
||||
func TestRunOnceCoverFetchDoesNotDelayNextPoll(t *testing.T) {
|
||||
s, dbURL := newTestStore(t)
|
||||
// Two Series whose legacy cover sources still need healing.
|
||||
for i := 0; i < 2; i++ {
|
||||
key := fmt.Sprintf("asura:cover-%d", i)
|
||||
seriesID := fmt.Sprintf("cover-%d", i)
|
||||
if _, err := s.Upsert(s.OwnerID(), store.Bookmark{
|
||||
Key: key, Site: "asura", SeriesID: seriesID,
|
||||
SeriesURL: "https://asurascans.com/comics/" + seriesID, UpdatedAt: 1000,
|
||||
}); err != nil {
|
||||
t.Fatalf("seed: %v", err)
|
||||
}
|
||||
seedCoverSource(t, dbURL, "asura", seriesID, fmt.Sprintf("https://cdn.example/covers/%d.jpg", i))
|
||||
}
|
||||
|
||||
pages := &fakeFetcher{body: asuraSeriesFixture, status: 200}
|
||||
gated := &gatedCoverFetcher{
|
||||
inner: &fakeBytesCoverFetcher{body: []byte("cover-bytes"), contentType: "image/jpeg"},
|
||||
gate: make(chan struct{}),
|
||||
started: make(chan struct{}, 1),
|
||||
}
|
||||
p := &Poller{
|
||||
Store: s, Fetch: pages, CoverBytesFetch: gated,
|
||||
Now: func() time.Time { return time.UnixMilli(5_000_000) },
|
||||
}
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
p.runOnce(context.Background())
|
||||
close(done)
|
||||
}()
|
||||
<-gated.started // the first cover heal is now stuck on its CDN
|
||||
select {
|
||||
case <-done:
|
||||
// The Lane finished its page fetches without waiting for the cover.
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("runOnce blocked on an in-flight cover fetch")
|
||||
}
|
||||
close(gated.gate)
|
||||
p.waitCovers()
|
||||
|
||||
if got := pages.callCount(); got != 2 {
|
||||
t.Fatalf("series page fetches = %d, want 2", got)
|
||||
}
|
||||
if got := gated.inner.callCount(); got != 2 {
|
||||
t.Fatalf("cover fetches = %d, want 2", got)
|
||||
}
|
||||
}
|
||||
|
||||
// At 3601 eligible Series the effective gap falls below the one-second floor;
|
||||
// the clamp warning must name the Site so the operator knows which Lane is
|
||||
// outrunning its plan.
|
||||
func TestRunOnceClampWarningNamesTheSite(t *testing.T) {
|
||||
s, dbURL := newTestStore(t)
|
||||
db, err := sql.Open("pgx", dbURL)
|
||||
if err != nil {
|
||||
t.Fatalf("open %s: %v", dbURL, err)
|
||||
}
|
||||
defer db.Close()
|
||||
if _, err := db.Exec(`INSERT INTO series (site, series_id, series_url, latest_checked_at)
|
||||
SELECT 'asura', 'bulk-' || g, 'https://asurascans.com/comics/bulk-' || g, 0
|
||||
FROM generate_series(1, 3601) AS g`); err != nil {
|
||||
t.Fatalf("bulk seed series: %v", err)
|
||||
}
|
||||
if _, err := db.Exec(`INSERT INTO bookmarks (reader_id, site, series_id, updated_at)
|
||||
SELECT (SELECT id FROM readers ORDER BY id LIMIT 1), 'asura', 'bulk-' || g, 1000
|
||||
FROM generate_series(1, 3601) AS g`); err != nil {
|
||||
t.Fatalf("bulk seed bookmarks: %v", err)
|
||||
}
|
||||
|
||||
var logs strings.Builder
|
||||
prev := log.Writer()
|
||||
log.SetOutput(&logs)
|
||||
t.Cleanup(func() { log.SetOutput(prev) })
|
||||
|
||||
newTestPoller(t, s, &fakeFetcher{body: "<html></html>", status: 200}, time.UnixMilli(5_000_000)).
|
||||
runOnce(context.Background())
|
||||
|
||||
if got := logs.String(); !strings.Contains(got, "latest poll asura: gap clamped to 1s floor (eligible series=3601)") {
|
||||
t.Fatalf("clamp warning = %q, want it to name asura and 3601", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
// seriesRead carries the two facts the poll and the acquirer both extract
|
||||
// from a series page. Persistence, stamps and scheduling stay with the
|
||||
// callers, so the policies that keep the two flows distinct (stamp order,
|
||||
// cooldowns) are not swallowed by the module.
|
||||
// rests) are not swallowed by the module.
|
||||
type seriesRead struct {
|
||||
Latest latestChapter
|
||||
HasLatest bool
|
||||
@@ -22,6 +22,9 @@ type seriesRead struct {
|
||||
|
||||
// errNotFetchable and errNoFetcher separate the gate and the route from fetch
|
||||
// failures so each caller keeps its own distinct log line for all three.
|
||||
// errChallengeHeld (browser.go) is the outcome of a Site that answered with
|
||||
// its interstitial — status 403 (cf-mitigated) or a challenge page body — and
|
||||
// is how a Lane tells a refusal from an ordinary failure (issue #100).
|
||||
var (
|
||||
errNotFetchable = errors.New("series url not fetchable")
|
||||
errNoFetcher = errors.New("no fetcher for site")
|
||||
@@ -49,9 +52,21 @@ func readSeriesPage(ctx context.Context, site, seriesURL string, browser, tls Fe
|
||||
if err != nil {
|
||||
return seriesRead{}, fmt.Errorf("fetch %s: %w", seriesURL, err)
|
||||
}
|
||||
if status == 403 {
|
||||
// Cloudflare's challenge response for these Sites (cf-mitigated). The
|
||||
// browser fetcher returns exactly this on a held interstitial, and a
|
||||
// plain-TLS 403 means the same: the Site is refusing.
|
||||
return seriesRead{}, fmt.Errorf("%w: fetch %s: status %d", errChallengeHeld, seriesURL, status)
|
||||
}
|
||||
if status != 200 {
|
||||
return seriesRead{}, fmt.Errorf("fetch %s: status %d", seriesURL, status)
|
||||
}
|
||||
if isInterstitial(body) {
|
||||
// A 200 that is the challenge page, not the payload: the TLS route can
|
||||
// receive this where the browser would have kept re-reading. Same
|
||||
// refusal as the 403.
|
||||
return seriesRead{}, fmt.Errorf("%w: fetch %s: interstitial body", errChallengeHeld, seriesURL)
|
||||
}
|
||||
latest, hasLatest := latestChapterFrom(site, seriesURL, body)
|
||||
cover, hasCover := coverFrom(site, seriesURL, body)
|
||||
return seriesRead{Latest: latest, HasLatest: hasLatest, Cover: cover, HasCover: hasCover, BodyLen: len(body)}, nil
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/chromedp/chromedp"
|
||||
)
|
||||
@@ -32,6 +33,11 @@ type site struct {
|
||||
LatestChapter func(seriesURL, body string) (latestChapter, bool)
|
||||
// Cover finds the Cover address in a fetched body.
|
||||
Cover func(seriesURL, body string) (string, bool)
|
||||
// Rest is how long a Series of this Site rests between Polls.
|
||||
Rest time.Duration
|
||||
// Gap is the Lane's strictest pace: at least one second must pass between
|
||||
// two consecutive Series-page Polls of this Site (issue #100).
|
||||
Gap time.Duration
|
||||
// Browser reads this Site's payload from a cleared browser tab; nil
|
||||
// means the page is fetched over plain TLS.
|
||||
Browser *browserRead
|
||||
@@ -380,6 +386,50 @@ func publishedCoverURL(value string) string {
|
||||
return strings.ReplaceAll(value, " ", "%20")
|
||||
}
|
||||
|
||||
// Poll Lane constants (issue #100). The per-Site structure is deliberately
|
||||
// uniform at first — every Site rests an hour and gaps ten seconds — but it
|
||||
// exists so a single Site can be slowed if it turns hostile, and the numbers
|
||||
// stay in the registry so the structure has a place to differ.
|
||||
const (
|
||||
// defaultRest is how long every Series rests between Polls.
|
||||
defaultRest = time.Hour
|
||||
// defaultGap is the strictest pace of every Lane unless the eligible
|
||||
// Series count forces it tighter.
|
||||
defaultGap = 10 * time.Second
|
||||
// minGap floors the effective gap. One request per second is already an
|
||||
// order of magnitude past the strictest rate rule a free-plan Site can
|
||||
// express (docs/research/cloudflare-bot-scoring-and-poll-cadence.md);
|
||||
// below it the Lane is outrunning its own plan and says so loudly.
|
||||
minGap = time.Second
|
||||
// refuseBackoff is how long a Lane waits after its Site refused twice in
|
||||
// one run before attempting it again.
|
||||
refuseBackoff = 15 * time.Minute
|
||||
// browserWakeCount and browserWakeAge gate a browser Lane's run: five or
|
||||
// more due Series, or any one of them waiting this long, or Chrome stays
|
||||
// asleep (ADR-0005 on-demand browser).
|
||||
browserWakeCount = 5
|
||||
browserWakeAge = 15 * time.Minute
|
||||
)
|
||||
|
||||
// effectiveGap is a Site's pace: the registry gap, or one rest divided by the
|
||||
// eligible Series count when that is smaller, never below one second. The
|
||||
// denominator follows defaultRest rather than a literal hour so a Site whose
|
||||
// rest is ever changed keeps its per-Series pace in step. The second return is
|
||||
// true when the one-second floor engaged (and the Lane logs a warning naming
|
||||
// the Site, every round it does).
|
||||
func effectiveGap(s site, eligible int) (time.Duration, bool) {
|
||||
gap := s.Gap
|
||||
if eligible > 0 {
|
||||
if perSeries := defaultRest / time.Duration(eligible); perSeries < gap {
|
||||
gap = perSeries
|
||||
}
|
||||
}
|
||||
if gap < minGap {
|
||||
return minGap, true
|
||||
}
|
||||
return gap, false
|
||||
}
|
||||
|
||||
// sites is the registry: one entry per Site, keyed by the stored site string.
|
||||
// Adding a Site means adding an entry here and nowhere else — the dispatch
|
||||
// functions above and the poller's route list are lookups into this map. An
|
||||
@@ -390,16 +440,22 @@ var sites = map[string]site{
|
||||
Host: "asurascans.com",
|
||||
LatestChapter: asuraLatestChapter,
|
||||
Cover: ogImageCover,
|
||||
Rest: defaultRest,
|
||||
Gap: defaultGap,
|
||||
},
|
||||
"demonic": {
|
||||
Host: "demonicscans.org",
|
||||
LatestChapter: demonicLatestChapter,
|
||||
Cover: ogImageCover,
|
||||
Rest: defaultRest,
|
||||
Gap: defaultGap,
|
||||
},
|
||||
"comix": {
|
||||
Host: "comix.to",
|
||||
LatestChapter: comixLatestChapter,
|
||||
Cover: comixCoverEntry,
|
||||
Rest: defaultRest,
|
||||
Gap: defaultGap,
|
||||
Browser: &browserRead{
|
||||
Read: comixRead,
|
||||
// The interstitial is served in place of the page, so "arrived"
|
||||
@@ -414,6 +470,8 @@ var sites = map[string]site{
|
||||
Host: "kagane.to",
|
||||
LatestChapter: kaganeLatestChapter,
|
||||
Cover: kaganeCoverEntry,
|
||||
Rest: defaultRest,
|
||||
Gap: defaultGap,
|
||||
Browser: &browserRead{
|
||||
Read: kaganeRead,
|
||||
Done: func(body string) bool { return body != "" },
|
||||
@@ -426,6 +484,8 @@ var sites = map[string]site{
|
||||
Host: "novelfull.com",
|
||||
LatestChapter: novelfullLatestChapter,
|
||||
Cover: novelfullCoverEntry,
|
||||
Rest: defaultRest,
|
||||
Gap: defaultGap,
|
||||
Browser: &browserRead{
|
||||
Read: novelfullRead,
|
||||
// The interstitial has a DOM too, so "the payload arrived" has to
|
||||
@@ -438,13 +498,15 @@ var sites = map[string]site{
|
||||
Host: "lightnovelworld.net",
|
||||
LatestChapter: lnwLatestChapter,
|
||||
Cover: ogImageCover,
|
||||
Rest: defaultRest,
|
||||
Gap: defaultGap,
|
||||
},
|
||||
}
|
||||
|
||||
// browserBackedSites is derived from the registry: the Sites whose pages are
|
||||
// read through the browser sidecar, which are also the ones granted the longer
|
||||
// cooldown. Sorted so callers that range it (the due query, the browser
|
||||
// fetcher's dispatch) see a stable order instead of map-iteration noise.
|
||||
// read through the browser sidecar. Sorted so callers that range it (the
|
||||
// browser fetcher's dispatch) see a stable order instead of map-iteration
|
||||
// noise.
|
||||
func browserBackedSites() []string {
|
||||
out := make([]string, 0, len(sites))
|
||||
for name, s := range sites {
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -298,7 +298,7 @@ func TestDueForLatestCheck(t *testing.T) {
|
||||
s := newTestStore(t)
|
||||
seedForCheck(t, s, "asura:x", tt.seriesURL, tt.checkedAt)
|
||||
|
||||
due, err := s.DueForLatestCheck(now-hour, now-hour, nil, 10)
|
||||
due, err := s.DueForLatestCheck("asura", now-hour)
|
||||
if err != nil {
|
||||
t.Fatalf("DueForLatestCheck: %v", err)
|
||||
}
|
||||
@@ -309,22 +309,25 @@ func TestDueForLatestCheck(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestDueForLatestCheckOldestFirstAndLimited(t *testing.T) {
|
||||
func TestDueForLatestCheckOldestFirstAndScopedToSite(t *testing.T) {
|
||||
s := newTestStore(t)
|
||||
// Insert newest-checked first so a correct ORDER BY has to reverse it.
|
||||
seedForCheck(t, s, "asura:c", "https://asurascans.com/comics/c", 300)
|
||||
seedForCheck(t, s, "asura:b", "https://asurascans.com/comics/b", 200)
|
||||
seedForCheck(t, s, "asura:a", "https://asurascans.com/comics/a", 100)
|
||||
// A second Site's due series must not appear in asura's list: each Lane
|
||||
// asks for one Site, and no Lane may see another's queue.
|
||||
seedForCheck(t, s, "demonic:z", "https://demonicscans.org/manga/z", 0)
|
||||
|
||||
due, err := s.DueForLatestCheck(1000, 1000, nil, 2)
|
||||
due, err := s.DueForLatestCheck("asura", 1000)
|
||||
if err != nil {
|
||||
t.Fatalf("DueForLatestCheck: %v", err)
|
||||
}
|
||||
if len(due) != 2 {
|
||||
t.Fatalf("got %d rows, want 2 (limit)", len(due))
|
||||
if len(due) != 3 {
|
||||
t.Fatalf("got %d rows, want 3 (all of asura's, none of demonic's)", len(due))
|
||||
}
|
||||
if due[0].Key() != "asura:a" || due[1].Key() != "asura:b" {
|
||||
t.Fatalf("got %q,%q; want asura:a,asura:b (oldest first)", due[0].Key(), due[1].Key())
|
||||
if due[0].Key() != "asura:a" || due[1].Key() != "asura:b" || due[2].Key() != "asura:c" {
|
||||
t.Fatalf("got %q,%q,%q; want asura:a,asura:b,asura:c (oldest first)", due[0].Key(), due[1].Key(), due[2].Key())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -496,7 +499,7 @@ func TestDueForLatestCheckSkipsFinishedKeepsArchived(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
due, err := store.DueForLatestCheck(time.Now().UnixMilli(), time.Now().UnixMilli(), nil, 10)
|
||||
due, err := store.DueForLatestCheck("asura", time.Now().UnixMilli())
|
||||
if err != nil {
|
||||
t.Fatalf("DueForLatestCheck: %v", err)
|
||||
}
|
||||
@@ -512,6 +515,38 @@ func TestDueForLatestCheckSkipsFinishedKeepsArchived(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// The gap's denominator counts every Series the Lane will ever Poll: a
|
||||
// finished Series must not make the Lane faster than it needs to be, and
|
||||
// another Site's Series must not leak into this Site's count.
|
||||
func TestEligibleSeriesCount(t *testing.T) {
|
||||
store := newTestStore(t)
|
||||
seedForCheck(t, store, "asura:reading", "https://asurascans.com/comics/reading", 0)
|
||||
seedForCheck(t, store, "asura:archived", "https://asurascans.com/comics/archived", 0)
|
||||
if _, err := store.Upsert(store.OwnerID(), Bookmark{
|
||||
Key: "asura:finished", Site: "asura", SeriesID: "finished",
|
||||
SeriesURL: "https://asurascans.com/comics/finished",
|
||||
Status: StatusFinished, UpdatedAt: 1000,
|
||||
}); err != nil {
|
||||
t.Fatalf("seed finished: %v", err)
|
||||
}
|
||||
seedForCheck(t, store, "demonic:z", "https://demonicscans.org/manga/z", 0)
|
||||
|
||||
n, err := store.EligibleSeriesCount("asura")
|
||||
if err != nil {
|
||||
t.Fatalf("EligibleSeriesCount: %v", err)
|
||||
}
|
||||
if n != 2 {
|
||||
t.Fatalf("eligible = %d, want 2 (finished excluded, demonic excluded)", n)
|
||||
}
|
||||
n, err = store.EligibleSeriesCount("demonic")
|
||||
if err != nil {
|
||||
t.Fatalf("EligibleSeriesCount(demonic): %v", err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("eligible(demonic) = %d, want 1", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDisplayChapter(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
@@ -970,7 +1005,7 @@ func TestDueForLatestCheckOrdersByReaderCountThenAge(t *testing.T) {
|
||||
seedSecondReader(t, s, "asura:pop:2", "asura", "pop", 1001)
|
||||
seedForCheck(t, s, "asura:solo", "https://asurascans.com/comics/solo", 100)
|
||||
|
||||
due, err := s.DueForLatestCheck(1000, 1000, nil, 10)
|
||||
due, err := s.DueForLatestCheck("asura", 1000)
|
||||
if err != nil {
|
||||
t.Fatalf("DueForLatestCheck: %v", err)
|
||||
}
|
||||
@@ -996,7 +1031,7 @@ func TestDueForLatestCheckExcludesOrphanSeries(t *testing.T) {
|
||||
t.Fatalf("seed orphan series: %v", err)
|
||||
}
|
||||
|
||||
due, err := s.DueForLatestCheck(1000, 1000, nil, 10)
|
||||
due, err := s.DueForLatestCheck("asura", 1000)
|
||||
if err != nil {
|
||||
t.Fatalf("DueForLatestCheck: %v", err)
|
||||
}
|
||||
@@ -1334,7 +1369,7 @@ func TestTwoReadersShareOneSeriesWithIndependentProgress(t *testing.T) {
|
||||
t.Fatalf("series rows = %d, want 1 shared row for two bookmarks", series)
|
||||
}
|
||||
|
||||
due, err := s.DueForLatestCheck(time.Now().UnixMilli(), time.Now().UnixMilli(), nil, 10)
|
||||
due, err := s.DueForLatestCheck("asura", time.Now().UnixMilli())
|
||||
if err != nil {
|
||||
t.Fatalf("DueForLatestCheck: %v", err)
|
||||
}
|
||||
@@ -1350,7 +1385,7 @@ func TestTwoReadersShareOneSeriesWithIndependentProgress(t *testing.T) {
|
||||
if b, ok, err := s.Get(s.OwnerID(), "asura:solo"); err != nil || !ok || b.LastChapterNum != 200 {
|
||||
t.Fatalf("owner's bookmark after the other's delete = %+v ok=%v err=%v, want it intact", b, ok, err)
|
||||
}
|
||||
due, err = s.DueForLatestCheck(time.Now().UnixMilli(), time.Now().UnixMilli(), nil, 10)
|
||||
due, err = s.DueForLatestCheck("asura", time.Now().UnixMilli())
|
||||
if err != nil {
|
||||
t.Fatalf("DueForLatestCheck after delete: %v", err)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user