diff --git a/package-lock.json b/package-lock.json index 9147068..e01a25d 100644 --- a/package-lock.json +++ b/package-lock.json @@ -8570,7 +8570,7 @@ }, "packages/plugin": { "name": "@harperfast/prerender", - "version": "0.70.0", + "version": "0.71.0", "license": "Apache-2.0", "dependencies": { "fast-xml-parser": "^5.0.9", diff --git a/packages/plugin/METRICS.md b/packages/plugin/METRICS.md index 86c5597..df31f71 100644 --- a/packages/plugin/METRICS.md +++ b/packages/plugin/METRICS.md @@ -121,17 +121,17 @@ PK drives the scan (an open range can make the planner walk a metric's entire hi One-line summaries; `src/metrics.js` carries the full description of every dimension value and the reasoning behind it. -| Metric | Kind | `path` | `method` | `type` | What it's for | -| ---------------- | ------- | ---------- | ----------- | ---------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `bot_request` | counter | host | botName | deviceType | Raw crawl volume and mix at ingress. The denominator for every serve-side ratio. | -| `bot_serve` | counter | source | cacheStatus | botName | **Origin offload** and **cache hit rate** — the two rollout numbers. | -| `route_serve` | counter | route | cacheStatus | deviceType | The same outcome per route: which route's `renderInterval` needs to move. | -| `page_age` | ms | botName | deviceType | — | Freshness as delivered: ms since the served snapshot rendered (cache serves only). | -| `route_page_age` | ms | route | cacheStatus | deviceType | Served age per route, split by freshness state — the "should this TTL move" number. | -| `render` | value | series | per-series | per-series | The render fleet in one scan: `time_ms` (duration by statusCode × candidacy, one sample per device variant — renders/hour = concurrency ÷ time_ms) and `outcome` (counter by outcome × detail, exactly one per posted result — and a result is one URL, every device in it, since v0.66.0 — the render-failure alert). | -| `origin_fetch` | ms | statusCode | reason | — | Cost of every non-cache serve: origin latency + status, by why the cache didn't answer (miss/stale/skip/invalidated/bypass/blob-missing/blob-timeout/render-timeout). | -| `prerender_ops` | value | series | detail | context | Every low-volume ops signal in one scan: `unrouted` (class, bucket), `sitemap_*` (refresh-run counters: sitemaps/created/updated/skipped/removed/failed, plus `sitemap_departure_*` — one series per outcome of the post-walk sitemap-departure check, `departure_render`/`departure_expire` for actions taken, `departure_would_render`/`departure_would_expire` under `sitemap.departure.dryRun`, and `departure_reattached`/`departure_suppressed`/`departure_route_opted_out`/`departure_target_gone`/`departure_capped` for the candidates nothing happened to. **`departure_reattached` is the one to watch**: it counts URLs that only LOOKED departed because they shifted across a paginated sitemap's child boundary, so a large share means the corpus is shearing and the raw `sitemap_removed` count is not a departure count. `departure_capped` means `maxActions` bound and some departed URLs were left for the next walk), `serve_error`, `config_warnings`, `page_age_negative` (bot, device), `demand_*` (ladder decisions + `fast_fraction`/`fill`), `invalidation_error` (kind), `invalidation_reenqueue` (outcome, scope — including the cross-node outcomes `forwarded`/`forward-failed`; `forwarded` means this node handed the heal to the key's owner, which counts its OWN verdict in this same series, so the two are deliberately not double-counted), `page_verification` (outcome: `written`/`read-error`/`write-error` — per-page invalidation exemptions being recorded; the exemptions actually GRANTED are `bot_serve` cacheStatus `verified`, not this), `probe_*` (change-probe pass counters: probed/seeded/rebaselined/changed/triggered/deferred/failed per pass, plus `probe_canary_trip` and `probe_invalidated`; `probe_rebaselined` counts URLs whose stored baseline was taken under a different rule fingerprint and were re-seeded without comparison — expect one pass of them after a rule edit, and treat a steady count as a rule that keeps changing; `probe_changed`/`probe_probed` is the measured change rate, a rising `probe_failed` share is the endpoint-changed-shape alarm), `discovery_gated` (gate, bot: cacheable misses the discovery gate held out of target creation — the corpus growth being prevented, not denied mints), `probe_fresh` (probes skipped because a baseline was younger than `reprobeAfter` — the work a restarted sweep skipped), `probe_throttled` (probes the origin refused with pushback — **alert on this**: it is the only signal that the probe is loading an origin that cannot take it), `probe_unreadable` (registry rows whose key failed to decode, skipped by the sweep's walk — a nonzero count means the table holds rows the application layer cannot address; escalate to the database layer), `probe_page_mismatch` (cached pages that disagreed with the origin — the round-trip-blindness class `pageCheck` catches; a rising share means renders are landing on transient states, and each one is a served page carrying wrong price/availability until it re-renders), `probe_cycle_behind` (CONTINUOUS MODE: batches that needed more than `ratePerSecond` to hit `cycleTarget` — the pass is flat out against its agreed origin ceiling and still losing ground. **Alert on a sustained count**: it is the explicit replacement for the interval model's silently skipped pass, and it means the corpus has outgrown the rate, so either `cycleTarget` is too ambitious or the ceiling needs renegotiating. Zero in interval mode, where no target is set). | -| `queue_health` | value | series | result | — | Every queue signal in one scan: the snapshot gauges (`overdue`, `lease_occupancy`, `below_floor`, `below_floor_age_ms`, `floor_pin_age_ms`, `paused`), `claim_scan_ms` (per pass, method = granted/empty/capped), `claim_granted` (per claim, method = ready/index), `ready_sweep_ms` (per sweep, method = complete/capped), `ready_published`, `ready_cadence` (per sweep, method = carried/resolved), `reconcile_restored`/`reconcile_missing` (per sweep). | +| Metric | Kind | `path` | `method` | `type` | What it's for | +| ---------------- | ------- | ---------- | ----------- | ---------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `bot_request` | counter | host | botName | deviceType | Raw crawl volume and mix at ingress. The denominator for every serve-side ratio. | +| `bot_serve` | counter | source | cacheStatus | botName | **Origin offload** and **cache hit rate** — the two rollout numbers. | +| `route_serve` | counter | route | cacheStatus | deviceType | The same outcome per route: which route's `renderInterval` needs to move. | +| `page_age` | ms | botName | deviceType | — | Freshness as delivered: ms since the served snapshot rendered (cache serves only). | +| `route_page_age` | ms | route | cacheStatus | deviceType | Served age per route, split by freshness state — the "should this TTL move" number. | +| `render` | value | series | per-series | per-series | The render fleet in one scan: `time_ms` (duration by statusCode × candidacy, one sample per device variant — renders/hour = concurrency ÷ time_ms) and `outcome` (counter by outcome × detail, exactly one per posted result — and a result is one URL, every device in it, since v0.66.0 — the render-failure alert). | +| `origin_fetch` | ms | statusCode | reason | — | Cost of every non-cache serve: origin latency + status, by why the cache didn't answer (miss/stale/skip/invalidated/bypass/blob-missing/blob-timeout/render-timeout). | +| `prerender_ops` | value | series | detail | context | Every low-volume ops signal in one scan: `unrouted` (class, bucket), `sitemap_*` (refresh-run counters: sitemaps/created/updated/skipped/removed/failed, plus `sitemap_departure_*` — one series per outcome of the post-walk sitemap-departure check, `departure_render`/`departure_expire` for actions taken, `departure_would_render`/`departure_would_expire` under `sitemap.departure.dryRun`, and `departure_reattached`/`departure_suppressed`/`departure_route_opted_out`/`departure_target_gone`/`departure_capped` for the candidates nothing happened to. **`departure_reattached` is the one to watch**: it counts URLs that only LOOKED departed because they shifted across a paginated sitemap's child boundary, so a large share means the corpus is shearing and the raw `sitemap_removed` count is not a departure count. `departure_capped` means `maxActions` bound and some departed URLs were left for the next walk), `serve_error`, `config_warnings`, `page_age_negative` (bot, device), `demand_*` (ladder decisions + `fast_fraction`/`fill`), `invalidation_error` (kind), `invalidation_reenqueue` (outcome, scope — including the cross-node outcomes `forwarded`/`forward-failed`; `forwarded` means this node handed the heal to the key's owner, which counts its OWN verdict in this same series, so the two are deliberately not double-counted), `page_verification` (outcome: `written`/`read-error`/`write-error` — per-page invalidation exemptions being recorded; the exemptions actually GRANTED are `bot_serve` cacheStatus `verified`, not this), `probe_*` (change-probe pass counters: probed/seeded/rebaselined/changed/triggered/deferred/failed per pass, plus `probe_canary_trip` and `probe_invalidated`; `probe_rebaselined` counts URLs whose stored baseline was taken under a different rule fingerprint and were re-seeded without comparison — expect one pass of them after a rule edit, and treat a steady count as a rule that keeps changing; `probe_changed`/`probe_probed` is the measured change rate, a rising `probe_failed` share is the endpoint-changed-shape alarm), `discovery_gated` (gate, bot: cacheable misses the discovery gate held out of target creation — the corpus growth being prevented, not denied mints), `probe_fresh` (probes skipped because a baseline was younger than `reprobeAfter` — the work a restarted sweep skipped), `probe_throttled` (probes the origin refused with pushback — **alert on this**: it is the only signal that the probe is loading an origin that cannot take it), `probe_unreadable` (registry rows whose key failed to decode, skipped by the sweep's walk — a nonzero count means the table holds rows the application layer cannot address; escalate to the database layer), `probe_page_mismatch` (cached pages that disagreed with the origin — the round-trip-blindness class `pageCheck` catches; a rising share means renders are landing on transient states, and each one is a served page carrying wrong price/availability until it re-renders), `probe_trigger_queue_depth` (high-water depth of the trigger queue during the pass — triggers are submitted to a bounded queue that drains beside the walk, so a value steadily at `changeProbe.trigger.maxPending` means the drain rate is behind the detection rate and changes are being deferred for want of QUEUE rather than of budget; those two are indistinguishable in `probe_deferred` alone), `probe_cycle_behind` (CONTINUOUS MODE: batches that needed more than `ratePerSecond` to hit `cycleTarget` — the pass is flat out against its agreed origin ceiling and still losing ground. **Alert on a sustained count**: it is the explicit replacement for the interval model's silently skipped pass, and it means the corpus has outgrown the rate, so either `cycleTarget` is too ambitious or the ceiling needs renegotiating. Zero in interval mode, where no target is set). | +| `queue_health` | value | series | result | — | Every queue signal in one scan: the snapshot gauges (`overdue`, `lease_occupancy`, `below_floor`, `below_floor_age_ms`, `floor_pin_age_ms`, `paused`), `claim_scan_ms` (per pass, method = granted/empty/capped), `claim_granted` (per claim, method = ready/index), `ready_sweep_ms` (per sweep, method = complete/capped), `ready_published`, `ready_cadence` (per sweep, method = carried/resolved), `reconcile_restored`/`reconcile_missing` (per sweep). | Notes that bite: diff --git a/packages/plugin/package.json b/packages/plugin/package.json index a6e2500..6732f9d 100644 --- a/packages/plugin/package.json +++ b/packages/plugin/package.json @@ -1,6 +1,6 @@ { "name": "@harperfast/prerender", - "version": "0.70.0", + "version": "0.71.0", "type": "module", "description": "Configurable Harper plugin for prerendering pages for bots and crawlers", "license": "Apache-2.0", diff --git a/packages/plugin/src/configSchema.js b/packages/plugin/src/configSchema.js index 3a38c42..7209a9a 100644 --- a/packages/plugin/src/configSchema.js +++ b/packages/plugin/src/configSchema.js @@ -1162,6 +1162,52 @@ export const configSchema = group('Prerender plugin configuration.', { 'transaction may live anywhere near that long (see the scan group).', { min: 10 } ), + trigger: group( + 'How detected changes are turned into re-renders. Submitted to a bounded queue that drains ' + + 'BESIDE the walk rather than inside it, so a pass runs at its probe-rate floor whatever the ' + + 'change rate.\n\n' + + 'WHY THAT MATTERS. Triggering is six database operations, and when it ran in-line in the row ' + + 'handler it shared the pass\u2019s concurrency with probing \u2014 so trigger volume set PASS ' + + 'DURATION, and pass duration is detection latency, because the gap between two probes of one ' + + 'URL is one pass. That closes a loop: more change \u2192 more triggers \u2192 longer pass ' + + '\u2192 a longer window in which each URL can change \u2192 more change. Measured on one ' + + 'deployment, arming the probe took a pass from 9.2h to a projected ~21h with bot traffic flat ' + + 'across both windows, and every knob traded deferrals against latency instead of escaping the ' + + 'loop. Submitting makes pass duration max(probe time, drain time) rather than the sum, and ' + + 'makes the meaningful limit triggers per SECOND \u2014 what the render fleet experiences.\n\n' + + 'A full queue is reported as `deferred`, exactly like exhausting `maxTriggersPerSweep`: the ' + + 'signature is left stale and the next pass re-detects. Nothing is lost by dropping the queue, ' + + 'which is why it is in memory and why an aborted pass simply abandons it.', + { + ratePerSecond: option( + 5, + 'Triggers started per second. This is the rate the RENDER QUEUE sees, not the origin: a ' + + 'trigger writes, it does not fetch. Size it against SPARE RENDER CAPACITY and the claim ' + + 'floor \u2014 not against the origin ceiling that `changeProbe.ratePerSecond` respects, ' + + 'and not against how fast the queue could go.\n\n' + + 'HOW TO SIZE IT. Aim for a drain that finishes INSIDE the pass: past that, the queue ' + + 'backs up and changes defer for want of queue rather than of budget. Take ' + + '`maxTriggersPerSweep` over the pass length you expect \u2014 90,000 triggers across a ' + + '9h pass is ~2.8/s, so the default leaves headroom without being able to outrun a ' + + 'fleet.\n\n' + + 'GOING MUCH HIGHER IS THE ONE WAY THIS CHANGE CAN HURT, because it is something the ' + + 'old in-line path could never do: at 20/s a 90,000-trigger budget drains in ~1.25h, ' + + 'which on a four-node cluster injects renders several times faster than the fleet can ' + + 'claim them \u2014 deepening the ready set and starving its lowest-priority class. ' + + 'Raise it only against a measured render rate that sits below the fleet ceiling. ' + + '0 or less drains unpaced.', + { min: 0 } + ), + concurrency: option(4, 'Triggers in flight at once.', { min: 1 }), + maxPending: option( + 5000, + 'Queue depth before submissions are refused and counted as `deferred`. Bounds memory ' + + 'across a pass that can detect hundreds of thousands of changes; it is NOT the ' + + 'per-pass budget, which stays `maxTriggersPerSweep`.', + { min: 1 } + ), + } + ), maxTriggersPerSweep: option( 5000, 'Ceiling on re-renders one sweep pass may file (per node). Changes past it stay detected but ' + diff --git a/packages/plugin/src/util/changeProbe.js b/packages/plugin/src/util/changeProbe.js index 3ccd3a8..85e8a94 100644 --- a/packages/plugin/src/util/changeProbe.js +++ b/packages/plugin/src/util/changeProbe.js @@ -54,6 +54,7 @@ import { epochMsOf, currentMinuteMs, getNextTimeOfDay, DAY, MINUTE, SECOND } fro import { getResidencyByUrl } from './residency.js'; import { resolveEffectiveInterval } from './routeClass.js'; import { writeSchedule } from './renderSchedule.js'; +import { createInlineTrigger, createTriggerQueue, SUBMIT_FULL } from './triggerQueue.js'; import { recordInvalidation, isScopeResolvable, resolveInvalidation } from './invalidation.js'; import { dispatcherFor, configuredStagingIp } from './upstream.js'; import { cacheKeysOf } from '../resources/Target.js'; @@ -140,8 +141,9 @@ const newStats = () => ({ rebaselined: 0, // baseline was taken under a DIFFERENT rule fingerprint: observation stored, nothing compared or triggered (a rule edit, not a content change) unchanged: 0, changed: 0, - triggered: 0, // changes that scheduled a re-render - deferred: 0, // changes past maxTriggersPerSweep — signature kept stale so the next pass retries + queued: 0, // changes handed to the trigger queue (accepted, not necessarily settled yet) + triggered: 0, // changes that scheduled a re-render — merged from the queue after it drains + deferred: 0, // changes past maxTriggersPerSweep, or refused by a full queue — signature kept stale so the next pass retries failed: 0, // fetch/parse/extraction failures — signature untouched, nothing triggered errors: 0, // trigger writes that threw fresh: 0, // skipped: baseline younger than reprobeAfter (a pass already covered it) @@ -489,7 +491,11 @@ export const runProbePass = async ({ probe, read, write, - trigger, + // Hands one detected change to whatever performs the re-render — the paced queue for a sweep, + // the immediate shape for the canary (util/triggerQueue.js). It RETURNS rather than awaiting + // the trigger: the sweep must not pay trigger latency inside the row handler, because that is + // what made pass duration a function of the change rate. + submitTrigger, // Injected like `write`/`trigger` so a pass can be exercised without Harper, and DEFAULTED INERT: // a caller that does not wire verification gets exactly the pre-feature behaviour. `isArmed` is // resolved once per rule per pass (see below), never per row. @@ -697,27 +703,26 @@ export const runProbePass = async ({ await write(row.url, observed, { rowExists: stored !== null, fingerprint: rule.fingerprint }); return; } - if (stats.triggered >= maxTriggers) { + if (stats.queued >= maxTriggers) { // Budget spent: leave the signature STALE so the next pass re-detects and retries. // Bounds how much queue injection one pass can do (a mass change is the canary's job). + // Counted on ACCEPTANCE, not completion: the budget has to be decided synchronously here + // or a burst would race past it while earlier triggers were still settling. stats.deferred++; return; } - try { - await trigger(row); - stats.triggered++; - // The page's claim is CLEARED on EVERY acted trip, not just a page disagreement: the - // trip hard-expired the page, so whatever the claim described is no longer served — and - // a preserved claim would re-detect against the NEW baseline on the next pass (a price - // drift's old claim disagrees with the new price by construction) and re-spend the - // trigger budget on a page already expired and already filed. The next render writes a - // fresh claim; until then there is nothing to compare, which is the correct "I don't - // know" state. Folded into this write so it costs no second round trip. - await write(row.url, observed, { rowExists: stored !== null, clearClaim: true, fingerprint: rule.fingerprint }); - } catch (e) { - stats.errors++; - globalThis.logger?.error?.(e, `[prerender] change-probe trigger failed for ${row.url}`); + // Submit, never await the trigger itself. A refusal means the queue is at its depth limit, + // which is the same statement as the budget above — leave the signature stale and let the + // next pass retry. The BASELINE WRITE MOVED WITH THE TRIGGER, into the queue, because it + // must happen only after the trigger succeeds; that ordering is the whole retry story. + if ( + (await submitTrigger({ row, observed, rowExists: stored !== null, fingerprint: rule.fingerprint })) === + SUBMIT_FULL + ) { + stats.deferred++; + return; } + stats.queued++; }; // Pacing: batches of `concurrency`, each batch held to the window `ratePerSecond` implies for @@ -865,6 +870,10 @@ const emitStats = (stats, kind) => { metrics.changeProbe(stats.throttled, 'throttled'); metrics.changeProbe(stats.pageMismatch, 'page_mismatch'); metrics.changeProbe(stats.behindBatches, 'cycle_behind'); + // The trigger queue's high-water depth. Steadily at `trigger.maxPending` means the drain + // rate is behind the detection rate and changes are being deferred for want of queue, not + // for want of budget — the two look identical in `deferred` alone. + if (Number.isFinite(stats.triggerQueueDepth)) metrics.changeProbe(stats.triggerQueueDepth, 'trigger_queue_depth'); } catch (e) { logger.warn(`[prerender] change-probe ${kind} metrics not recorded: ${e?.message ?? String(e)}`); } @@ -1060,6 +1069,17 @@ export const runProbeSweepOnce = async ({ dryRun, label = null, reseed = false } const collectors = new Map(rules.map((rule) => [rule.label, cohortCollector(count)])); let unreadable = 0; let yields = 0; + // Triggers drain BESIDE the walk, not inside it. `submit` returns immediately, so the pass + // runs at its probe-rate floor whatever the change rate — see util/triggerQueue.js for the + // feedback loop this breaks. + const triggers = createTriggerQueue({ + trigger: triggerRevalidate, + write: writeSignature, + maxPending: config.changeProbe.trigger.maxPending, + ratePerSecond: config.changeProbe.trigger.ratePerSecond, + concurrency: config.changeProbe.trigger.concurrency, + onError: (e, item) => logger.error(e, `[prerender] change-probe trigger failed for ${item.row.url}`), + }); const stats = await runProbePass({ rows: walkTargets(config.changeProbe.chunkSize, () => { unreadable++; @@ -1071,7 +1091,7 @@ export const runProbeSweepOnce = async ({ dryRun, label = null, reseed = false } probe: probeOnce, read: readSignature, write: writeSignature, - trigger: triggerRevalidate, + submitTrigger: triggers.submit, verify: writeVerification, isArmed: verificationArmedFor, ...limits, @@ -1092,6 +1112,14 @@ export const runProbeSweepOnce = async ({ dryRun, label = null, reseed = false } isCanceled: () => !config.changeProbe.enabled || sweepInterrupt !== null, collectCohort: (rule, url) => collectors.get(rule.label).add(url), }); + // A cancelled pass abandons what is still queued: those rows never had their baseline + // written, so the next pass re-detects them. Otherwise wait the queue out — the pass is not + // finished while re-renders it decided on are still unfiled, and `triggered` would under-report. + if (stats.aborted) triggers.stop(); + await triggers.drain(); + stats.triggered = triggers.stats.triggered; + stats.errors = triggers.stats.errors; + stats.triggerQueueDepth = triggers.stats.maxDepth; stats.unreadable = unreadable; // An interrupted pass keeps the OLD cohorts — a partial walk's sample covers only the key // range it reached, and the chained reseed rebuilds them properly. @@ -1286,6 +1314,11 @@ export const runProbeCanaryOnce = async ({ dryRun } = {}) => { perRule.push({ rule: rule.label, cohort: 0, skipped: 'empty cohort' }); continue; } + const canaryTriggers = createInlineTrigger({ + trigger: triggerRevalidate, + write: writeSignature, + onError: (e, item) => logger.error(e, `[prerender] change-probe trigger failed for ${item.row.url}`), + }); const stats = await runProbePass({ rows: readCohortRows(urls), rules: [rule], @@ -1295,7 +1328,9 @@ export const runProbeCanaryOnce = async ({ dryRun } = {}) => { probe: probeOnce, read: readSignature, write: writeSignature, - trigger: triggerRevalidate, + // IMMEDIATE, not queued: the cohort is a few hundred URLs and the canary's whole value + // is being fast, so there is nothing for a paced queue to spread. + submitTrigger: canaryTriggers.submit, ...limits, // NEVER skips on baseline age. The cohort is small and deliberately probed on a // cadence far tighter than `reprobeAfter` — freshness-skipping here would silence @@ -1303,6 +1338,9 @@ export const runProbeCanaryOnce = async ({ dryRun } = {}) => { reprobeAfter: 0, isCanceled: () => !config.changeProbe.enabled, }); + await canaryTriggers.drain(); + stats.triggered = canaryTriggers.stats.triggered; + stats.errors = canaryTriggers.stats.errors; emitStats(stats, 'canary'); const verdict = canaryVerdict(stats, { threshold: canary.threshold, minSample: canary.minSample }); let action = null; diff --git a/packages/plugin/src/util/triggerQueue.js b/packages/plugin/src/util/triggerQueue.js new file mode 100644 index 0000000..a34c967 --- /dev/null +++ b/packages/plugin/src/util/triggerQueue.js @@ -0,0 +1,225 @@ +/** + * A bounded, separately-paced queue for change-probe triggers. + * + * ── WHY THIS EXISTS ────────────────────────────────────────────────────────────────────────── + * + * The sweep used to run `await trigger(row)` inside the row handler. Triggering is six database + * operations (a read and a patch per device page, plus the schedule write), so with the handler's + * concurrency shared between probing and triggering, TRIGGER VOLUME SET PASS DURATION — and pass + * duration is detection latency, because the interval between two probes of the same URL is one + * pass. + * + * That closes a loop: more change -> more triggers -> longer pass -> a longer window in which each + * URL can change -> more change detected per pass -> more triggers. Measured on one deployment, + * arming the probe took a pass from 9.2h to a projected ~21h with bot traffic flat across both + * windows (153,728/hr vs 150,743/hr, so contention was ruled out), and every available config knob + * traded deferrals against latency rather than escaping the loop. Raising the trigger budget made + * the pass longer; lowering it shed more change. Neither is a fix. + * + * Submitting instead of awaiting breaks the loop: the sweep runs at its probe-rate floor whatever + * the change rate, and triggers drain beside it at their own pace. Pass duration becomes + * `max(probe time, drain time)` instead of the sum, and the knob that matters becomes triggers per + * SECOND — what the render fleet actually experiences — rather than triggers per pass. + * + * ── WHAT IS PRESERVED EXACTLY ──────────────────────────────────────────────────────────────── + * + * The baseline write still happens ONLY after a successful trigger, and it happens HERE rather + * than in the caller, because that ordering is the whole retry story: a trigger that fails, or one + * that never ran because the queue was cleared, leaves the stored signature stale, so the next + * pass re-detects the same change and tries again. Writing the baseline at submit time would lose + * the change outright on any failure. `test/triggerQueue.test.js` pins this. + * + * A full queue is reported to the caller as a refusal, which the sweep counts as `deferred` — the + * same accounting the per-pass budget already produced, and with the same stale-signature + * semantics, so nothing downstream has to learn a new state. + * + * ── DELIBERATELY IN MEMORY, AND DELIBERATELY NOT DURABLE ───────────────────────────────────── + * + * Losing the queue costs nothing but a repeat detection: every unsettled item still has its old + * signature stored, so the next pass finds it again. That is why `stop()` may simply drop pending + * work, and why a restart mid-pass needs no recovery path. Durability here would buy nothing and + * add a table on the probe's write path, which is the one thing `ProbeState`'s design avoids. + */ + +import { MAX_TIMER_MS } from './probePacer.js'; + +/** The outcome of a submit, as the caller's stats understand it. */ +export const SUBMIT_QUEUED = 'queued'; +export const SUBMIT_FULL = 'full'; + +/** + * @param {object} ports + * @param {(row: object) => Promise} ports.trigger performs the re-render trigger + * @param {(url: string, observed: string, opts: object) => Promise} ports.write baseline write + * @param {number} ports.maxPending queue depth before submissions are refused + * @param {number} ports.ratePerSecond drain pace; 0 or less means unpaced + * @param {number} ports.concurrency triggers in flight at once + * @param {(ms: number) => Promise} [ports.sleep] + * @param {() => number} [ports.now] + * @param {(error: unknown, item: object) => void} [ports.onError] + */ +export const createTriggerQueue = ({ + trigger, + write, + maxPending = 5000, + ratePerSecond = 10, + concurrency = 4, + sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)), + now = () => Date.now(), + onError, +} = {}) => { + const pending = []; + const stats = { triggered: 0, errors: 0, refused: 0, maxDepth: 0 }; + let stopped = false; + let inFlight = 0; + let pumping = false; + // The next instant a trigger may start, advanced by one slot per dispatch. Pacing the START of + // each trigger rather than sleeping between completions keeps the rate honest when triggers + // have uneven latency, which they do: a page miss is one read, a hit is a read and a patch. + let nextSlotAt = 0; + let idleResolvers = []; + + const slotMs = ratePerSecond > 0 ? 1000 / ratePerSecond : 0; + + const settleIdle = () => { + if (pending.length || inFlight) return; + const waiting = idleResolvers; + idleResolvers = []; + for (const resolve of waiting) resolve(); + }; + + const runOne = async (item) => { + try { + await trigger(item.row); + stats.triggered++; + // AFTER the trigger, never before — see the module comment. `clearClaim` goes with it + // because the trigger hard-expired the page, so whatever the stored page claim described + // is no longer being served. + await write(item.row.url, item.observed, { + rowExists: item.rowExists, + clearClaim: true, + fingerprint: item.fingerprint, + }); + } catch (e) { + stats.errors++; + // Swallowed on purpose: the signature stays stale, so the next pass retries this URL. + // A trigger failure must never take down the pass that submitted it. + onError?.(e, item); + } + }; + + const pump = async () => { + if (pumping) return; + pumping = true; + try { + while (!stopped && pending.length && inFlight < concurrency) { + if (slotMs > 0) { + const at = now(); + const startAt = Math.max(at, nextSlotAt); + nextSlotAt = startAt + slotMs; + // CLAMPED, because `sleep` wraps `setTimeout`: past a signed 32-bit delay it fires + // after 1ms instead of waiting, which would turn a deliberately slow drain into an + // unpaced one — the exact inversion `probePacer`'s own clamp exists to prevent. An + // absurdly small `ratePerSecond` is the way there (0.0001/s is a ~2.8h slot), and the + // resulting early wake just dispatches sooner, which at that configured rate is + // indistinguishable from the intent. + const wait = Math.min(startAt - at, MAX_TIMER_MS); + if (wait > 0) await sleep(wait); + // `stopped` can flip while we were asleep. + if (stopped) break; + } + const item = pending.shift(); + if (!item) break; + inFlight++; + runOne(item).finally(() => { + inFlight--; + settleIdle(); + // Re-enter rather than recurse: `pump` is single-flight, so this just restarts the + // loop if it has already exited. + void pump(); + }); + } + } finally { + pumping = false; + settleIdle(); + } + }; + + return { + stats, + + /** + * Offer one detected change. Returns `SUBMIT_QUEUED` or `SUBMIT_FULL`; never throws and + * never waits on the trigger itself, which is the entire point. + */ + submit(item) { + if (stopped || pending.length >= maxPending) { + stats.refused++; + return SUBMIT_FULL; + } + pending.push(item); + if (pending.length > stats.maxDepth) stats.maxDepth = pending.length; + void pump(); + return SUBMIT_QUEUED; + }, + + /** Resolves once everything submitted so far has settled. */ + async drain() { + void pump(); + if (!pending.length && !inFlight) return; + await new Promise((resolve) => idleResolvers.push(resolve)); + }, + + /** + * Stop accepting and abandon what is pending. Safe by construction: an abandoned item never + * had its baseline written, so the next pass re-detects it. + */ + stop() { + stopped = true; + const dropped = pending.length; + pending.length = 0; + settleIdle(); + return dropped; + }, + + get depth() { + return pending.length + inFlight; + }, + }; +}; + +/** + * The synchronous shape, for callers that should not decouple: the canary. + * + * Its cohort is a few hundred URLs and its whole value is being fast and immediate, so queuing + * would add machinery to a path with nothing to gain from it. Same contract, so `runProbePass` + * has one code path rather than a branch. + */ +export const createInlineTrigger = ({ trigger, write, onError } = {}) => { + const stats = { triggered: 0, errors: 0, refused: 0, maxDepth: 0 }; + return { + stats, + async submit(item) { + try { + await trigger(item.row); + stats.triggered++; + await write(item.row.url, item.observed, { + rowExists: item.rowExists, + clearClaim: true, + fingerprint: item.fingerprint, + }); + } catch (e) { + stats.errors++; + onError?.(e, item); + } + return SUBMIT_QUEUED; + }, + async drain() {}, + stop() { + return 0; + }, + get depth() { + return 0; + }, + }; +}; diff --git a/packages/plugin/test/changeProbe.test.js b/packages/plugin/test/changeProbe.test.js index b3e1ae4..d30cd6f 100644 --- a/packages/plugin/test/changeProbe.test.js +++ b/packages/plugin/test/changeProbe.test.js @@ -122,6 +122,16 @@ const runPass = async ({ const written = []; const writeOptions = []; const triggered = []; + const { createInlineTrigger } = await import('../src/util/triggerQueue.js'); + const triggers = createInlineTrigger({ + trigger: async (row) => { + triggered.push(row.url); + }, + write: async (url, signature, options) => { + written.push({ url, signature }); + writeOptions.push({ url, ...options }); + }, + }); const stats = await changeProbe.runProbePass({ rows: stream(rows), rules, @@ -145,7 +155,10 @@ const runPass = async ({ written.push({ url, signature }); writeOptions.push({ url, ...options }); }, - trigger: async (target) => triggered.push(target.url), + // The REAL inline shape, not a stub: the trigger-then-write ordering now lives in + // util/triggerQueue.js, and a stub re-implementing it here would keep passing while the + // shipped path regressed. This is also exactly how the canary wires itself. + submitTrigger: triggers.submit, dryRun, maxTriggers, concurrency: 2, @@ -153,6 +166,10 @@ const runPass = async ({ pause: async () => {}, ...overrides, }); + // What the sweep and the canary both do once the pass returns. + await triggers.drain(); + stats.triggered = triggers.stats.triggered; + stats.errors = triggers.stats.errors; return { stats, written, writeOptions, triggered, rules }; }; @@ -284,9 +301,20 @@ test('past the trigger budget a change DEFERS: signature left stale so the next assert.deepEqual(written, [{ url: URL_A, signature: '[2]' }]); }); -test('a failed trigger write keeps the signature stale too', async () => { +test('a failed trigger keeps the signature stale too', async () => { + // The property is unchanged by the move to a submitted trigger; only the seam moved. Driven + // through the REAL inline shape rather than a stub, because the ordering under test — baseline + // written only after the trigger succeeds — now lives in util/triggerQueue.js, and a stub that + // re-implemented it would pass while the shipped path regressed. const { compileProbeRules } = await import('../src/util/changeProbeSpec.js'); + const { createInlineTrigger } = await import('../src/util/triggerQueue.js'); const written = []; + const triggers = createInlineTrigger({ + trigger: async () => { + throw new Error('write refused'); + }, + write: async (url, signature) => written.push({ url, signature }), + }); const stats = await changeProbe.runProbePass({ rows: stream([row(URL_A)]), rules: compileProbeRules(RULES_RAW), @@ -295,18 +323,18 @@ test('a failed trigger write keeps the signature stale too', async () => { probe: async () => '[2]', read: async () => ({ signature: '[1]', probedAt: NaN }), write: async (url, signature) => written.push({ url, signature }), - trigger: async () => { - throw new Error('write refused'); - }, + submitTrigger: triggers.submit, dryRun: false, maxTriggers: 10, concurrency: 1, ratePerSecond: 1000, pause: async () => {}, }); - assert.equal(stats.errors, 1); - assert.equal(stats.triggered, 0); - assert.deepEqual(written, []); + await triggers.drain(); + assert.equal(triggers.stats.errors, 1); + assert.equal(triggers.stats.triggered, 0); + assert.equal(stats.queued, 1, 'the change was accepted for triggering'); + assert.deepEqual(written, [], 'no baseline may be written when the trigger threw'); }); test('pacing sleeps out the remainder of each batch window', async () => { @@ -447,7 +475,7 @@ test('freshness skip: a baseline younger than reprobeAfter is not re-probed', as }, read: async (url) => ({ signature: '[1]', probedAt: url === URL_A ? T - 60_000 : T - 20 * HOUR }), write: async () => {}, - trigger: async () => {}, + submitTrigger: async () => 'queued', dryRun: false, maxTriggers: 10, concurrency: 1, @@ -476,7 +504,7 @@ test('freshness skip: an unparseable or missing probedAt probes rather than skip // A row with no timestamp, and one that never had a baseline at all. read: async (url) => (url === URL_A ? { signature: '[1]', probedAt: NaN } : null), write: async () => {}, - trigger: async () => {}, + submitTrigger: async () => 'queued', dryRun: false, maxTriggers: 10, concurrency: 1, @@ -505,7 +533,7 @@ test('origin backoff: a pushback response stretches the pacing window, a clean b }, read: async () => null, write: async () => {}, - trigger: async () => {}, + submitTrigger: async () => 'queued', dryRun: false, maxTriggers: 10, concurrency: 1, @@ -533,7 +561,7 @@ test('origin backoff: an explicit Retry-After outranks the computed window', asy }, read: async () => null, write: async () => {}, - trigger: async () => {}, + submitTrigger: async () => 'queued', dryRun: false, maxTriggers: 10, concurrency: 1, @@ -559,7 +587,7 @@ test('origin backoff: a fully refusing origin ends the pass instead of crawling' }, read: async () => null, write: async () => {}, - trigger: async () => {}, + submitTrigger: async () => 'queued', dryRun: false, maxTriggers: 10, concurrency: 1, @@ -589,7 +617,7 @@ test('origin backoff: a rule/product failure is NOT distress and must not thrott }, read: async () => null, write: async () => {}, - trigger: async () => {}, + submitTrigger: async () => 'queued', dryRun: false, maxTriggers: 10, concurrency: 1, @@ -622,7 +650,7 @@ test('freshness skip: a BigInt probedAt is coerced, not thrown on', async () => }, read: async () => ({ signature: '[1]', probedAt: Number(BigInt(T - 60_000)) }), write: async () => {}, - trigger: async () => {}, + submitTrigger: async () => 'queued', dryRun: false, maxTriggers: 10, concurrency: 1, @@ -651,7 +679,7 @@ test('origin backoff: the pacing wait can never exceed setTimeout’s 32-bit cap }, read: async () => null, write: async () => {}, - trigger: async () => {}, + submitTrigger: async () => 'queued', dryRun: false, maxTriggers: 10, concurrency: 1, @@ -679,6 +707,13 @@ test('a trip hard-expires the page PAST the swr window — a known-wrong page is patched.push({ id, ...fields }); } }; + // Capture the schedule write too: the trigger must file ONE row, keyed by the URL. + const scheduled = []; + globalThis.databases.render_schedule.RenderSchedule = class extends FakeTable { + static async put(id, fields) { + scheduled.push({ id, ...fields }); + } + }; const before = Date.now(); await changeProbe.triggerRevalidate(row('https://example.com/product/prd-a/')); // Bound against a clock read taken AFTER the call: the trigger reads Date.now() itself, so @@ -734,6 +769,17 @@ const runPageCheckPass = async ({ rows, answers, stored = {}, ...overrides }) => const { compileProbeRules } = await import('../src/util/changeProbeSpec.js'); const written = []; const triggered = []; + const write = async (url, signature, opts = {}) => + written.push({ url, signature, rowExists: opts.rowExists === true, clearClaim: opts.clearClaim === true }); + const { createInlineTrigger } = await import('../src/util/triggerQueue.js'); + // The real inline shape — `clearClaim` is set by the trigger path itself, and these tests are + // precisely the ones asserting on it. + const triggers = createInlineTrigger({ + trigger: async (row) => { + triggered.push(row.url); + }, + write, + }); const stats = await changeProbe.runProbePass({ rows: stream(rows), rules: compileProbeRules(PAGECHECK_RULES), @@ -741,9 +787,8 @@ const runPageCheckPass = async ({ rows, answers, stored = {}, ...overrides }) => hostname: 'node-a', probe: async (rule, url) => answers[url] ?? null, read: async (url) => stored[url] ?? null, - write: async (url, signature, opts = {}) => - written.push({ url, signature, rowExists: opts.rowExists === true, clearClaim: opts.clearClaim === true }), - trigger: async (target) => triggered.push(target.url), + write, + submitTrigger: triggers.submit, dryRun: false, maxTriggers: 100, concurrency: 1, @@ -751,6 +796,9 @@ const runPageCheckPass = async ({ rows, answers, stored = {}, ...overrides }) => pause: async () => {}, ...overrides, }); + await triggers.drain(); + stats.triggered = triggers.stats.triggered; + stats.errors = triggers.stats.errors; return { stats, written, triggered }; }; @@ -1053,7 +1101,7 @@ const runPaced = async ({ rows, answers = {}, clockStep = 0, ...overrides }) => }, read: async () => null, write: async () => {}, - trigger: async () => {}, + submitTrigger: async () => 'queued', dryRun: true, maxTriggers: 1000, concurrency: 2, @@ -1538,7 +1586,7 @@ const runVerifyPass = async ({ rows, answers, stored = {}, armed = true, ...over probe: async (rule, url) => answers[url] ?? null, read: async (url) => stored[url] ?? null, write: async () => {}, - trigger: async () => {}, + submitTrigger: async () => 'queued', verify: async (url, basisAt) => verified.push({ url, basisAt }), isArmed: async (scope) => { armedCalls.push(scope); @@ -1661,7 +1709,7 @@ test('with no verify/isArmed wired, the pass behaves exactly as before', async ( probe: async () => AGREE_SIG, read: async () => ({ signature: AGREE_SIG, probedAt: NaN, pageSignature: AGREE_CLAIM, pageClaimAt: CLAIM_AT }), write: async () => {}, - trigger: async () => {}, + submitTrigger: async () => 'queued', dryRun: false, maxTriggers: 100, concurrency: 1, diff --git a/packages/plugin/test/triggerQueue.test.js b/packages/plugin/test/triggerQueue.test.js new file mode 100644 index 0000000..490155c --- /dev/null +++ b/packages/plugin/test/triggerQueue.test.js @@ -0,0 +1,279 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createInlineTrigger, createTriggerQueue, SUBMIT_FULL, SUBMIT_QUEUED } from '../src/util/triggerQueue.js'; + +globalThis.logger ??= { debug() {}, info() {}, warn() {}, error() {} }; + +const item = (url, over = {}) => ({ + row: { url, sitemapUrl: null }, + observed: '[1]', + rowExists: true, + fingerprint: 'fp', + ...over, +}); + +const recorder = () => { + const triggered = []; + const written = []; + return { + triggered, + written, + trigger: async (row) => { + triggered.push(row.url); + }, + write: async (url, observed, opts) => { + written.push({ url, observed, ...opts }); + }, + }; +}; + +// The property the whole design rests on: submit RETURNS, it does not wait for the trigger. If it +// awaited, pass duration would still be a function of the change rate, which is the loop this +// exists to break. +test('submit returns without waiting for the trigger to run', async () => { + let released; + const gate = new Promise((resolve) => { + released = resolve; + }); + const q = createTriggerQueue({ + trigger: () => gate, + write: async () => {}, + ratePerSecond: 0, + }); + + const outcome = q.submit(item('https://example.com/a')); + assert.equal(outcome, SUBMIT_QUEUED, 'submit resolved while the trigger is still blocked'); + assert.equal(q.stats.triggered, 0, 'nothing has settled yet'); + + released(); + await q.drain(); + assert.equal(q.stats.triggered, 1); +}); + +test('the baseline is written only AFTER the trigger succeeds, and carries clearClaim', async () => { + const r = recorder(); + const q = createTriggerQueue({ ...r, ratePerSecond: 0 }); + q.submit(item('https://example.com/a')); + await q.drain(); + + assert.deepEqual(r.triggered, ['https://example.com/a']); + assert.equal(r.written.length, 1); + assert.equal(r.written[0].url, 'https://example.com/a'); + assert.equal(r.written[0].clearClaim, true, 'the trip hard-expired the page, so the stored claim must go'); + assert.equal(r.written[0].fingerprint, 'fp'); + assert.equal(r.written[0].rowExists, true); +}); + +// The retry story: a failed trigger must leave the signature stale so the next pass re-detects. +test('a throwing trigger writes NO baseline and is counted as an error', async () => { + const written = []; + const q = createTriggerQueue({ + trigger: async () => { + throw new Error('refused'); + }, + write: async (url) => written.push(url), + ratePerSecond: 0, + }); + q.submit(item('https://example.com/a')); + await q.drain(); + + assert.deepEqual(written, [], 'writing here would lose the change outright'); + assert.equal(q.stats.errors, 1); + assert.equal(q.stats.triggered, 0); +}); + +test('one failure does not stop the queue', async () => { + const r = recorder(); + let calls = 0; + const q = createTriggerQueue({ + trigger: async (row) => { + calls++; + if (calls === 1) throw new Error('first one fails'); + r.triggered.push(row.url); + }, + write: r.write, + ratePerSecond: 0, + }); + q.submit(item('https://example.com/a')); + q.submit(item('https://example.com/b')); + await q.drain(); + + assert.equal(q.stats.errors, 1); + assert.equal(q.stats.triggered, 1); + assert.deepEqual(r.triggered, ['https://example.com/b']); +}); + +test('a full queue refuses, which the caller counts as deferred', async () => { + let released; + const gate = new Promise((resolve) => { + released = resolve; + }); + const q = createTriggerQueue({ + trigger: () => gate, + write: async () => {}, + maxPending: 2, + concurrency: 1, + ratePerSecond: 0, + }); + + // One is taken in flight immediately; the next two fill the pending list. + q.submit(item('https://example.com/a')); + const outcomes = [ + q.submit(item('https://example.com/b')), + q.submit(item('https://example.com/c')), + q.submit(item('https://example.com/d')), + ]; + assert.deepEqual(outcomes.slice(-1), [SUBMIT_FULL], 'the last one is past maxPending'); + assert.ok(q.stats.refused >= 1); + + released(); + await q.drain(); +}); + +// Safe by construction: an abandoned item never had its baseline written, so the next pass finds +// the same change again. +test('stop abandons pending work and writes nothing for it', async () => { + const r = recorder(); + let released; + const gate = new Promise((resolve) => { + released = resolve; + }); + const q = createTriggerQueue({ + trigger: async (row) => { + if (row.url.endsWith('/a')) return gate; + r.triggered.push(row.url); + }, + write: r.write, + concurrency: 1, + ratePerSecond: 0, + }); + q.submit(item('https://example.com/a')); + q.submit(item('https://example.com/b')); + q.submit(item('https://example.com/c')); + + const dropped = q.stop(); + assert.equal(dropped, 2, 'b and c were abandoned'); + + released(); + await q.drain(); + // The IN-FLIGHT item still completes and is still baselined, which is right: its trigger + // actually succeeded, and a write you have already issued cannot be un-issued. Only PENDING + // work is abandoned — and abandoning it is safe precisely because its baseline was never + // written, so the next pass re-detects the same change. + assert.deepEqual(r.triggered, [], 'neither abandoned item ran'); + assert.deepEqual( + r.written.map((w) => w.url), + ['https://example.com/a'], + 'only the in-flight item was baselined' + ); +}); + +test('submissions after stop are refused rather than silently dropped', async () => { + const q = createTriggerQueue({ trigger: async () => {}, write: async () => {}, ratePerSecond: 0 }); + q.stop(); + assert.equal(q.submit(item('https://example.com/a')), SUBMIT_FULL); +}); + +test('drain resolves immediately when nothing was submitted', async () => { + const q = createTriggerQueue({ trigger: async () => {}, write: async () => {}, ratePerSecond: 0 }); + await q.drain(); + assert.equal(q.stats.triggered, 0); +}); + +test('pacing spaces trigger starts by the configured rate', async () => { + const waits = []; + let clock = 0; + const q = createTriggerQueue({ + trigger: async () => {}, + write: async () => {}, + ratePerSecond: 4, // one slot every 250ms + concurrency: 1, + now: () => clock, + sleep: async (ms) => { + waits.push(ms); + clock += ms; + }, + }); + for (const url of ['a', 'b', 'c']) q.submit(item(`https://example.com/${url}`)); + await q.drain(); + + assert.equal(q.stats.triggered, 3); + // The first starts immediately; the next two each wait out a slot on a clock that never + // advances on its own. + assert.deepEqual( + waits.filter((w) => w > 0), + [250, 250] + ); +}); + +test('maxDepth reports the high-water mark, for sizing maxPending', async () => { + let released; + const gate = new Promise((resolve) => { + released = resolve; + }); + const q = createTriggerQueue({ + trigger: () => gate, + write: async () => {}, + concurrency: 1, + ratePerSecond: 0, + }); + q.submit(item('https://example.com/a')); + q.submit(item('https://example.com/b')); + q.submit(item('https://example.com/c')); + assert.ok(q.stats.maxDepth >= 2, `maxDepth was ${q.stats.maxDepth}`); + released(); + await q.drain(); +}); + +// ---- the inline shape the canary uses ---- + +test('the inline trigger settles synchronously and keeps the same ordering', async () => { + const r = recorder(); + const inline = createInlineTrigger(r); + const outcome = await inline.submit(item('https://example.com/a')); + + assert.equal(outcome, SUBMIT_QUEUED); + assert.equal(inline.stats.triggered, 1, 'already settled by the time submit resolves'); + assert.deepEqual(r.triggered, ['https://example.com/a']); + assert.equal(r.written[0].clearClaim, true); +}); + +test('the inline trigger also withholds the baseline when the trigger throws', async () => { + const written = []; + const inline = createInlineTrigger({ + trigger: async () => { + throw new Error('refused'); + }, + write: async (url) => written.push(url), + }); + await inline.submit(item('https://example.com/a')); + + assert.deepEqual(written, []); + assert.equal(inline.stats.errors, 1); + assert.equal(inline.stats.triggered, 0); +}); + +test('an absurdly slow rate clamps the sleep instead of overflowing setTimeout', async () => { + // Past a signed 32-bit delay `setTimeout` fires after 1ms rather than waiting, so an unclamped + // wait would turn the slowest possible drain into an unpaced one. + const waits = []; + const q = createTriggerQueue({ + trigger: async () => {}, + write: async () => {}, + ratePerSecond: 1e-7, // a 10,000,000,000 ms slot — past 2^31-1 on the very first wait + concurrency: 1, + now: () => 0, // a clock that never advances, so the backlog of slots keeps growing + sleep: async (ms) => { + waits.push(ms); + }, + }); + for (const url of ['a', 'b', 'c']) q.submit(item(`https://example.com/${url}`)); + await q.drain(); + + assert.equal(q.stats.triggered, 3); + for (const w of waits) assert.ok(w <= 2147483647, `wait ${w} exceeds the max timer`); + assert.ok( + waits.some((w) => w === 2147483647), + 'the pathological slot should have been clamped, not passed through' + ); +});