diff --git a/deploy/plugin-stats-worker/README.md b/deploy/plugin-stats-worker/README.md index b2e2a5cd98..646b34efdb 100644 --- a/deploy/plugin-stats-worker/README.md +++ b/deploy/plugin-stats-worker/README.md @@ -41,9 +41,9 @@ Analytics Engine writes are effectively free at this scale. ## Daily aggregation (permanent archive) -AE retains only three months, so a cron trigger (`30 0 * * *` UTC) aggregates the -trailing window into the `plugin_stats_archive` KV namespace — the permanent -layer future UI reads: +AE retains only three months, so cron triggers aggregate the trailing window +into the `plugin_stats_archive` KV namespace — the permanent layer future UI +reads: - `total:{kind}:{pluginId}:{version}` — all-time cumulative events - `summary` — one JSON blob `{dl: {pluginId: n}, inst: {...}, updt: {...}}` for @@ -53,6 +53,24 @@ layer future UI reads: classified inst/updt events, distinct downloaders for dl) - `meta:last-success` — window marker; only advanced after all writes land, so a failed run retries the same window on the next cron +- `agg:state` / `agg:tuning` / `agg:lock` — resumable-pass cursor, adaptive + chunk size, and best-effort mutex (see below) + +**Chunked resumable passes** (added 2026-10-20). The original single-shot loop +needed one KV get+put pair per aggregated key; once a failed window grew the +retry backlog past the Workers per-invocation subrequest ceiling (1000 paid / +50 free), every run died mid-loop and the watermark never advanced — the +aggregation stalled silently for three weeks (2026-09-28 → 2026-10-20) while +raw AE data kept flowing. Now every invocation processes at most `chunk` keys +(tuning halves on failure, creeps up on success, 8–400), persists its cursor in +`agg:state`, and completes the summary write + watermark advance only when the +pass finishes. Crons: `30 0 * * *` opens the day's window, `*/10 * * * *` +resumes any in-flight pass. AE row order is normalized with a JS sort because +the offset cursor must index identically across invocations. Required +subrequests per invocation: `chunk`-bounded (~2·chunk + 6) — never the whole +backlog. After deploying schema-affecting changes, or to backfill from scratch, +pass `x-archive-from: reset` (totals are additive — only safe when the +archive is known-empty). Required secret (SQL API access; OAuth login does not carry analytics scope): diff --git a/deploy/plugin-stats-worker/worker.ts b/deploy/plugin-stats-worker/worker.ts index 427a327391..69cf63ce7e 100644 --- a/deploy/plugin-stats-worker/worker.ts +++ b/deploy/plugin-stats-worker/worker.ts @@ -37,7 +37,8 @@ type AnalyticsEngineBinding = { type KvBinding = { get(key: string): Promise; - put(key: string, value: string): Promise; + put(key: string, value: string, options?: { expirationTtl?: number }): Promise; + delete(key: string): Promise; }; type Env = { @@ -173,10 +174,14 @@ async function fetchDailyUniques(env: Env, from: Date, to: Date): Promise { - const [kind, plugin, day] = key.split(":"); - return { kind, plugin, day, uniq }; - }); + // Stable order across invocations: the resumable cursor is an offset into + // this array, and the SQL dialect cannot be trusted with ORDER BY. + return [...byDay] + .map(([key, uniq]) => { + const [kind, plugin, day] = key.split(":"); + return { kind, plugin, day, uniq }; + }) + .sort((a, b) => `${a.kind}:${a.plugin}:${a.day}`.localeCompare(`${b.kind}:${b.plugin}:${b.day}`)); } async function fetchWindowCounts(env: Env, from: Date, to: Date): Promise { @@ -196,6 +201,8 @@ async function fetchWindowCounts(env: Env, from: Date, to: Date): Promise ({ kind: String(row.kind ?? ""), @@ -203,49 +210,140 @@ async function fetchWindowCounts(env: Env, from: Date, to: Date): Promise COUNTED_KINDS.has(row.kind) && PLUGIN_ID_PATTERN.test(row.plugin) && VERSION_PATTERN.test(row.version)); + .filter((row) => COUNTED_KINDS.has(row.kind) && PLUGIN_ID_PATTERN.test(row.plugin) && VERSION_PATTERN.test(row.version)) + .sort((a, b) => `${a.kind}:${a.plugin}:${a.version}`.localeCompare(`${b.kind}:${b.plugin}:${b.version}`)); } -// Aggregates events in (last-success, now] into permanent KV totals. The window -// marker is only advanced after every write lands, so a failed run retries the -// same window on the next cron; a crash mid-write can double-count a window, -// which is acceptable for decorative stats. -async function runAggregation(env: Env, fromOverride?: string): Promise<{ from: string; to: string; activeKeys: number; uniqueDays: number }> { - const to = new Date(); - // `x-archive-from: reset` re-aggregates from the epoch (backfill after a fix); - // totals are additive, so only use it when the archived window is known-empty. - const override = fromOverride?.trim().toLowerCase() === "reset" ? new Date(0) : undefined; - const lastSuccess = await env.PLUGIN_ARCHIVE.get(ARCHIVE_META_KEY); - const from = override ?? (lastSuccess ? new Date(`${lastSuccess.replace(" ", "T")}Z`) : new Date(0)); - if (Number.isNaN(from.getTime())) throw new Error(`Corrupted ${ARCHIVE_META_KEY}: ${lastSuccess}`); +// Aggregates events in (last-success, now] into permanent KV totals, chunked +// and resumable. The pre-2026-10 unchunked loop died mid-run once the window +// grew past the Workers per-invocation subrequest ceiling (1000 paid / 50 +// free; every versioned total costs a get+put pair). A single failed run then +// left the watermark behind, the next window grew larger, and the aggregation +// stalled permanently (2026-09-28 → 2026-10-20). Now every trigger processes +// at most `chunk` keys, persists a cursor in `agg:state`, and the next cron +// (daily start + */10 continue) resumes until the pass completes; only then +// are the summary written and the watermark advanced. `agg:tuning` carries +// the chunk size across passes — it halves on failure and creeps up on +// progress, so the worker adapts to whatever subrequest/KV limits the +// account has. A crash mid-chunk can double-count that chunk's additive +// totals; acceptable for decorative stats (same stance as v1). +type AggPhase = "counts" | "pluginTotals" | "uniques"; +type AggState = { windowFrom: string; windowTo: string; phase: AggPhase; offset: number; chunk: number }; - const [counts, uniques] = await Promise.all([fetchWindowCounts(env, from, to), fetchDailyUniques(env, from, to)]); - const summary = JSON.parse((await env.PLUGIN_ARCHIVE.get(ARCHIVE_SUMMARY_KEY)) ?? "{}") as Record>; +const AGG_STATE_KEY = "agg:state"; +const AGG_LOCK_KEY = "agg:lock"; +const AGG_TUNING_KEY = "agg:tuning"; +const AGG_MIN_CHUNK = 8; // keeps get+put pairs + AE fetches under the free plan's 50 subrequests +const AGG_MAX_CHUNK = 400; // 400×2 + fetches + state writes stays under the paid plan's 1000 +const AGG_LOCK_TTL_SECONDS = 300; - const pluginTotals = new Map(); - for (const count of counts) { - const totalKey = `total:${count.kind}:${count.plugin}:${count.version}`; - const current = Number.parseInt((await env.PLUGIN_ARCHIVE.get(totalKey)) ?? "0", 10) || 0; - await env.PLUGIN_ARCHIVE.put(totalKey, String(current + count.events)); - pluginTotals.set(`${count.kind}:${count.plugin}`, (pluginTotals.get(`${count.kind}:${count.plugin}`) ?? 0) + count.events); - const byKind = (summary[count.kind] ??= {}); - byKind[count.plugin] = (byKind[count.plugin] ?? 0) + count.events; - } - for (const [pair, delta] of pluginTotals) { - const [kind, plugin] = pair.split(":"); - const pluginKey = `total:${kind}:${plugin}`; - const current = Number.parseInt((await env.PLUGIN_ARCHIVE.get(pluginKey)) ?? "0", 10) || 0; - await env.PLUGIN_ARCHIVE.put(pluginKey, String(current + delta)); - } - // Daily uniques overwrite (not add): reruns converge to the full-day value, - // and multi-version releases cannot inflate the per-plugin downloader count. - for (const unique of uniques) { - await env.PLUGIN_ARCHIVE.put(`uniqd:${unique.kind}:${unique.plugin}:${unique.day}`, String(unique.uniq)); - } +async function readChunkSize(env: Env): Promise { + const chunk = Number.parseInt((await env.PLUGIN_ARCHIVE.get(AGG_TUNING_KEY)) ?? "", 10); + return Number.isFinite(chunk) && chunk >= AGG_MIN_CHUNK && chunk <= AGG_MAX_CHUNK ? chunk : AGG_MAX_CHUNK; +} - await env.PLUGIN_ARCHIVE.put(ARCHIVE_SUMMARY_KEY, JSON.stringify(summary)); - await env.PLUGIN_ARCHIVE.put(ARCHIVE_META_KEY, sqlTimestamp(to)); - return { from: sqlTimestamp(from), to: sqlTimestamp(to), activeKeys: counts.length, uniqueDays: uniques.length }; +async function runAggregation(env: Env, fromOverride?: string): Promise< + | { skipped: "locked" } + | { completed: true; from: string; to: string; activeKeys: number; uniqueDays: number; writes: number } + | { completed: false; phase: AggPhase; offset: number; of: number; writes: number } +> { + // Best-effort mutex: cron triggers may overlap and additive totals race. + if (await env.PLUGIN_ARCHIVE.get(AGG_LOCK_KEY)) return { skipped: "locked" }; + await env.PLUGIN_ARCHIVE.put(AGG_LOCK_KEY, "1", { expirationTtl: AGG_LOCK_TTL_SECONDS }); + try { + let state: AggState | null = null; + const rawState = await env.PLUGIN_ARCHIVE.get(AGG_STATE_KEY); + if (rawState) { + try { + state = JSON.parse(rawState) as AggState; + } catch { + state = null; // corrupted cursor: start over rather than stay stuck + } + } + if (!state) { + // `x-archive-from: reset` re-aggregates from the epoch (backfill after + // a fix); totals are additive, so only use it when the archived window + // is known-empty. + const fromOverrideTs = fromOverride?.trim().toLowerCase() === "reset" ? "1970-01-01 00:00:00" : fromOverride?.trim(); + const windowFrom = fromOverrideTs ?? (await env.PLUGIN_ARCHIVE.get(ARCHIVE_META_KEY)) ?? "1970-01-01 00:00:00"; + state = { windowFrom, windowTo: sqlTimestamp(new Date()), phase: "counts", offset: 0, chunk: await readChunkSize(env) }; + } + const from = new Date(`${state.windowFrom.replace(" ", "T")}Z`); + const to = new Date(`${state.windowTo.replace(" ", "T")}Z`); + if (Number.isNaN(from.getTime()) || Number.isNaN(to.getTime())) throw new Error(`Corrupted aggregation state: ${JSON.stringify(state)}`); + + // Re-fetched every chunk (2 subrequests) so the persisted cursor stays a + // few bytes and AE remains the single source of truth for the window. + const [counts, uniques] = await Promise.all([fetchWindowCounts(env, from, to), fetchDailyUniques(env, from, to)]); + const pluginTotals = new Map(); + for (const count of counts) { + pluginTotals.set(`${count.kind}:${count.plugin}`, (pluginTotals.get(`${count.kind}:${count.plugin}`) ?? 0) + count.events); + } + const pluginTotalEntries = [...pluginTotals.entries()]; + + let writes = 0; + if (state.phase === "counts") { + for (const count of counts.slice(state.offset, state.offset + state.chunk)) { + const totalKey = `total:${count.kind}:${count.plugin}:${count.version}`; + const current = Number.parseInt((await env.PLUGIN_ARCHIVE.get(totalKey)) ?? "0", 10) || 0; + await env.PLUGIN_ARCHIVE.put(totalKey, String(current + count.events)); + writes += 2; + } + state.offset += Math.min(state.chunk, Math.max(0, counts.length - state.offset)); + if (state.offset >= counts.length) state = { ...state, phase: "pluginTotals", offset: 0 }; + } else if (state.phase === "pluginTotals") { + for (const [pair, delta] of pluginTotalEntries.slice(state.offset, state.offset + state.chunk)) { + const [kind, plugin] = pair.split(":"); + const pluginKey = `total:${kind}:${plugin}`; + const current = Number.parseInt((await env.PLUGIN_ARCHIVE.get(pluginKey)) ?? "0", 10) || 0; + await env.PLUGIN_ARCHIVE.put(pluginKey, String(current + delta)); + writes += 2; + } + state.offset += Math.min(state.chunk, Math.max(0, pluginTotalEntries.length - state.offset)); + if (state.offset >= pluginTotalEntries.length) state = { ...state, phase: "uniques", offset: 0 }; + } else { + // Daily uniques overwrite (not add): reruns converge to the full-day + // value, and multi-version releases cannot inflate the downloader count. + for (const unique of uniques.slice(state.offset, state.offset + state.chunk)) { + await env.PLUGIN_ARCHIVE.put(`uniqd:${unique.kind}:${unique.plugin}:${unique.day}`, String(unique.uniq)); + writes += 1; + } + state.offset += Math.min(state.chunk, Math.max(0, uniques.length - state.offset)); + if (state.offset >= uniques.length) { + const summary = JSON.parse((await env.PLUGIN_ARCHIVE.get(ARCHIVE_SUMMARY_KEY)) ?? "{}") as Record>; + for (const count of counts) { + const byKind = (summary[count.kind] ??= {}); + byKind[count.plugin] = (byKind[count.plugin] ?? 0) + count.events; + } + await env.PLUGIN_ARCHIVE.put(ARCHIVE_SUMMARY_KEY, JSON.stringify(summary)); + await env.PLUGIN_ARCHIVE.put(ARCHIVE_META_KEY, state.windowTo); + await env.PLUGIN_ARCHIVE.delete(AGG_STATE_KEY); + await env.PLUGIN_ARCHIVE.put(AGG_TUNING_KEY, String(Math.min(AGG_MAX_CHUNK, Math.ceil(state.chunk * 1.25)))); + return { completed: true, from: state.windowFrom, to: state.windowTo, activeKeys: counts.length, uniqueDays: uniques.length, writes }; + } + } + + state.chunk = Math.min(AGG_MAX_CHUNK, state.chunk + 1); + await env.PLUGIN_ARCHIVE.put(AGG_STATE_KEY, JSON.stringify(state)); + const of = state.phase === "counts" ? counts.length : state.phase === "pluginTotals" ? pluginTotalEntries.length : uniques.length; + return { completed: false, phase: state.phase, offset: state.offset, of, writes }; + } catch (error) { + // Halve the persisted chunk so the next trigger retries smaller; the + // cursor itself stays put (idempotent enough for decorative counters). + try { + const rawState = await env.PLUGIN_ARCHIVE.get(AGG_STATE_KEY); + if (rawState) { + const st = JSON.parse(rawState) as AggState; + st.chunk = Math.max(AGG_MIN_CHUNK, Math.floor(st.chunk / 2)); + await env.PLUGIN_ARCHIVE.put(AGG_STATE_KEY, JSON.stringify(st)); + } + } catch { + // best-effort tuning write; the error below is the real signal + } + throw error; + } finally { + await env.PLUGIN_ARCHIVE.delete(AGG_LOCK_KEY); + } } async function handleInstallBeacon(request: Request, env: Env): Promise { diff --git a/deploy/plugin-stats-worker/wrangler.json b/deploy/plugin-stats-worker/wrangler.json index e5dbe4a9f5..bd877910bd 100644 --- a/deploy/plugin-stats-worker/wrangler.json +++ b/deploy/plugin-stats-worker/wrangler.json @@ -29,7 +29,8 @@ }, "triggers": { "crons": [ - "30 0 * * *" + "30 0 * * *", + "*/10 * * * *" ] } }