feat(plugin): crawl breadth — distinct URLs per bot per day (HLL); v0.27.0 #65
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,211 @@ | ||
| /** | ||
| * Crawl breadth: distinct URLs crawled per bot per UTC day, via per-thread HyperLogLog | ||
| * sketches (util/hll.js — see there for why a sketch and why per-thread shards merge into | ||
| * an exact global union). | ||
| * | ||
| * Shape: each worker thread keeps one in-memory sketch per bot for the current day and, on | ||
| * a timer, merges it into this node's `crawl_stats.CrawlSketch` row (`day|bot|node`) — | ||
| * read-merge-write under the cross-worker mutex (util/coordination.js), so concurrent | ||
| * workers never lose each other's registers. One row per node (not per thread) keeps the | ||
| * read side inside the admin console's query-cost rules: a week is days × bots × nodes | ||
| * small rows, not × threads. Cross-node there is no contention at all — the node is in the | ||
| * key. The read side (PrerenderAdmin's crawl-breadth route) merges a day's rows per bot. | ||
| * | ||
| * Hot-path cost (recordCrawl): one number compare for day rollover, one Map lookup, one | ||
| * 53-bit string hash + one byte max. No allocation, no await, no storage touch. Everything | ||
| * async (flush, retention sweep) happens on the timer or via setImmediate. | ||
| * | ||
| * Loss model: a thread dying loses at most flushInterval worth of that thread's unmerged | ||
| * observations — an undercount, never a double count. Thread sketches are cumulative for | ||
| * the day and merging is idempotent, so a failed flush is simply retried whole next cycle. | ||
| * Rows replicate (table default), so any single node can answer for the cluster. | ||
| */ | ||
|
|
||
| import { config } from '../config.js'; | ||
| import { getMutex } from './coordination.js'; | ||
| import { createSketch, addToSketch, mergeSketch, estimateSketch } from './hll.js'; | ||
|
|
||
| // When a thread sees more distinct bot names than the cap in one day (a UA-derivation | ||
| // flood — registry bots can't exceed it), the overflow shares this bucket. Distinct from | ||
| // the registry's 'other' so a capped day is visible as such rather than blending in. | ||
| export const OVERFLOW_BUCKET = '~overflow'; | ||
|
|
||
| let sketches = new Map(); // botName -> Uint8Array registers | ||
| const dirty = new Set(); // botNames with unflushed observations | ||
| let day = null; // 'YYYY-MM-DD' (UTC) the current sketches belong to | ||
| let dayEndMs = 0; // rollover boundary, so the hot path compares one number | ||
| let flushTimer = null; | ||
|
|
||
| const utcDayOf = (ms) => new Date(ms).toISOString().slice(0, 10); | ||
| const table = () => databases.crawl_stats.CrawlSketch; | ||
|
|
||
| /** | ||
| * Observe one crawled URL. Called on the bot serving path (behind the analytics gate) — | ||
| * everything above about hot-path cost is about this function. Synchronous by design. | ||
| */ | ||
| export function recordCrawl(botName, url) { | ||
| if (!config.crawlStats.enabled) return; | ||
|
|
||
| const now = Date.now(); | ||
| if (now >= dayEndMs) rollover(now); | ||
|
|
||
| let sketch = sketches.get(botName); | ||
| if (!sketch) { | ||
| if (sketches.size >= config.crawlStats.maxBotsPerThread) { | ||
| botName = OVERFLOW_BUCKET; | ||
| sketch = sketches.get(botName); | ||
| } | ||
| if (!sketch) { | ||
| sketch = createSketch(); | ||
| sketches.set(botName, sketch); | ||
| } | ||
| } | ||
| addToSketch(sketch, url); | ||
| dirty.add(botName); | ||
|
|
||
| // Lazily started so only threads that actually serve bot traffic run a timer. | ||
| if (!flushTimer) { | ||
| flushTimer = setInterval(() => flushSketches().catch((e) => logger.error(e)), config.crawlStats.flushInterval); | ||
| flushTimer.unref?.(); | ||
| } | ||
| } | ||
|
|
||
| // Close out the old day and start the new one. The old sketches are captured and persisted | ||
| // off the hot path via setImmediate; the maps are swapped synchronously so no observation | ||
| // lands in the wrong day. Also the retention hook: one worker per node sweeps expired rows. | ||
| function rollover(now) { | ||
| const previous = day ? { day, sketches, dirty: [...dirty] } : null; | ||
| day = utcDayOf(now); | ||
| dayEndMs = Date.parse(day) + 24 * 60 * 60 * 1000; | ||
| sketches = new Map(); | ||
| dirty.clear(); | ||
|
|
||
| if (previous?.dirty.length) { | ||
| setImmediate(() => persist(previous.day, previous.sketches, previous.dirty).catch((e) => logger.error(e))); | ||
| } | ||
| if (previous && server.workerIndex === 0) { | ||
| setImmediate(() => sweepExpired().catch((e) => logger.error(e))); | ||
| } | ||
| } | ||
|
|
||
| /** Persist this thread's dirty sketches for the current day. Exported for tests. */ | ||
| export async function flushSketches() { | ||
| if (!dirty.size) return; | ||
| const bots = [...dirty]; | ||
| dirty.clear(); | ||
| try { | ||
| await persist(day, sketches, bots); | ||
| } catch (e) { | ||
| // Sketches are cumulative for the day and merging is idempotent, so retrying the | ||
| // whole flush next cycle is safe — re-mark rather than wait for new traffic. | ||
| for (const bot of bots) dirty.add(bot); | ||
| throw e; | ||
| } | ||
| } | ||
|
|
||
| async function persist(forDay, forSketches, bots) { | ||
| const CrawlSketch = table(); | ||
| for (const bot of bots) { | ||
| const mine = forSketches.get(bot); | ||
| if (!mine) continue; | ||
| // Read-merge-write of this node's row, serialized against the other workers by the | ||
| // cross-worker mutex. The row is this node's own (node is in the key), so the read | ||
| // is local — no cross-node fetch to time out on. Explicit lock/finally rather than | ||
| // `withLock` because this is a one-shot section — the repo's withLock idiom wraps a | ||
| // REUSABLE function (see RenderQueue.claim), and an inline `withLock(fn)()` reads | ||
| // like an accidental double call. | ||
| const id = `${forDay}|${bot}|${server.hostname}`; | ||
| const mutex = getMutex(`crawlSketch/${bot}`); | ||
| await mutex.lock(); | ||
| try { | ||
| const existing = await CrawlSketch.get(id); | ||
| // Merge into a copy: merging the row INTO the thread sketch would fold other | ||
| // workers' registers into thread-local state, and a later failed write would | ||
| // then re-contribute them as if they were this thread's own. | ||
| const merged = new Uint8Array(mine); | ||
| if (existing?.registers) mergeSketch(merged, existing.registers); | ||
| // `estimate` is this NODE's count, stored for eyeballing a raw row — the real | ||
| // per-day number requires merging all nodes first (distinct counts don't add). | ||
| await CrawlSketch.put(id, { | ||
| day: forDay, | ||
| bot, | ||
| node: server.hostname, | ||
| registers: merged, | ||
| estimate: estimateSketch(merged), | ||
| updatedAt: Date.now(), | ||
| }); | ||
| } finally { | ||
| mutex.unlock(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Delete rows past retention. Runs once per day-rollover on one worker per node; deletes | ||
| // are idempotent so concurrent nodes sweeping the same replicated rows is harmless. Bounded | ||
| // per pass — anything left over is caught the next day. | ||
| async function sweepExpired() { | ||
| const CrawlSketch = table(); | ||
| const cutoff = utcDayOf(Date.now() - config.crawlStats.retentionDays * 24 * 60 * 60 * 1000); | ||
| const expired = await CrawlSketch.search({ | ||
| conditions: [{ attribute: 'day', comparator: 'less_than', value: cutoff }], | ||
| select: ['id'], | ||
| limit: 1000, | ||
| }); | ||
| for await (const row of expired) { | ||
| await CrawlSketch.delete(row.id); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Streaming accumulator for the admin crawl-breadth route: fold one raw sketch row into | ||
| * `byDay` (a Map of day -> Map of bot -> { registers, shards }). Split this way so the | ||
| * route can `for await` a search cursor row by row — merging each 16 KB row into the | ||
| * accumulator and releasing it — instead of buffering the whole result set; the resident | ||
| * set is one sketch per (day, bot), not one per row. | ||
| */ | ||
| export function mergeBreadthRow(byDay, row) { | ||
| if (!row?.registers || !row.bot) return; | ||
| let bots = byDay.get(row.day); | ||
| if (!bots) byDay.set(row.day, (bots = new Map())); | ||
| let entry = bots.get(row.bot); | ||
| if (!entry) bots.set(row.bot, (entry = { registers: createSketch(), shards: 0 })); | ||
| mergeSketch(entry.registers, row.registers); | ||
| entry.shards++; | ||
| } | ||
|
|
||
| /** | ||
| * Turn the accumulator into per-day breadth estimates: | ||
| * [{ day, total, bots: [{ bot, distinctUrls, shards }] }] sorted by day desc, where | ||
| * `total` is the distinct-URL count of the UNION across bots (not a sum). | ||
| */ | ||
| export function finalizeBreadth(byDay) { | ||
| return [...byDay] | ||
| .sort(([a], [b]) => (a < b ? 1 : -1)) | ||
| .map(([forDay, bots]) => { | ||
| const union = createSketch(); | ||
| const perBot = [...bots] | ||
| .map(([bot, { registers, shards }]) => { | ||
| mergeSketch(union, registers); | ||
| return { bot, distinctUrls: estimateSketch(registers), shards }; | ||
| }) | ||
| .sort((a, b) => b.distinctUrls - a.distinctUrls); | ||
| return { day: forDay, total: estimateSketch(union), bots: perBot }; | ||
| }); | ||
| } | ||
|
|
||
| /** Convenience over the two halves above, for in-memory row sets (and tests). */ | ||
| export function computeBreadth(rows) { | ||
| const byDay = new Map(); | ||
| for (const row of rows) mergeBreadthRow(byDay, row); | ||
| return finalizeBreadth(byDay); | ||
| } | ||
|
|
||
| /** Test hook: reset module state between tests. */ | ||
| export function resetCrawlStats() { | ||
| sketches = new Map(); | ||
| dirty.clear(); | ||
| day = null; | ||
| dayEndMs = 0; | ||
| if (flushTimer) clearInterval(flushTimer); | ||
| flushTimer = null; | ||
| } |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The documentation for
recordUnmatchedandbotsappears to be incorrect. Insrc/config.js, these are properties of theanalyticsobject. However, in this README example, they are shown at the top level, alongsidecrawlStats.To avoid confusion, the example configuration in the README should accurately reflect the structure defined in the code. It should likely be structured as follows: