diff --git a/src/remote/api-client.ts b/src/remote/api-client.ts index 60080a6..2f75e6e 100644 --- a/src/remote/api-client.ts +++ b/src/remote/api-client.ts @@ -29,9 +29,16 @@ export function hookUpApiReporter(api: RpcStub, remote: Remote): () = }); }; const onStatsApplied = (stats: Parameters[0]) => { - api.pushStats(stats).catch((err) => { - log.error(`Failed to push stats: ${err}`, "api-client"); - }); + // Report the outcome back. The daily floor is what stops a project's + // statistics going stale, and only this call knows whether the server + // actually took the dump. + api.pushStats(stats).then( + () => remote.markStatsPushed(), + (err) => { + log.error(`Failed to push stats: ${err}`, "api-client"); + remote.markStatsPushFailed(); + }, + ); }; const onQueriesPolled = (queries: RecentQuery[]) => { api.pushQuery(JSON.parse(JSON.stringify(queries))).catch((err) => { diff --git a/src/remote/remote.ts b/src/remote/remote.ts index 7caab03..df8df42 100644 --- a/src/remote/remote.ts +++ b/src/remote/remote.ts @@ -110,7 +110,9 @@ export class Remote extends EventEmitter { private lastSkippedRefresh?: { reason: string; loggedAt: number }; private static readonly SKIP_LOG_INTERVAL_MS = 30 * 60 * 1000; /** - * When this analyzer last pushed a dump, for the daily floor. A drift- + * When the server last *accepted* a dump, for the daily floor. Set by + * {@link Remote.markStatsPushed}, not by the emit that starts the push — a + * dump that never crossed the wire must not hold the floor down. A drift- * triggered push updates it too, so the two triggers can't dump twice in * quick succession. */ @@ -481,9 +483,11 @@ export class Remote extends EventEmitter { "remote", ); // applyStatistics records the new baseline and emits `statsApplied`, - // which is what carries the dump back to the server. + // which is what carries the dump back to the server. Clearing the backoff + // is markStatsPushed's job, not this one's: the push it starts settles on + // a microtask, so clearing here would wipe the backoff a push that failed + // in the meantime had just set, and re-dump on the very next poll. await this.applyStatistics(await this.dumpSourceStats(source)); - this.retryStatsAfter = undefined; } catch (error) { this.retryStatsAfter = Date.now() + Remote.STATS_RETRY_BACKOFF_MS; throw error; @@ -580,14 +584,41 @@ export class Remote extends EventEmitter { // `fromAssumption` carries no real numbers at all, so it pushes nothing — // synthetic defaults are not this project's production statistics. if (statsMode.kind === "fromStatisticsExport" && statsMode.stats.length > 0) { + // The baseline records what this analyzer dumped, so it moves here even + // if the push is later lost — rolling it back would re-trigger Size Drift + // on every poll and dump the source over and over. The floor is different: + // it means "the server holds these numbers", so only delivery can arm it. this.statsBaseline = baselineFromDump(statsMode.stats); - this.lastStatsPushAt = Date.now(); this.emit("statsApplied", statsMode.stats); } // don't block the reply by awaiting all optimizations this.optimizer.restart(); } + /** + * The server accepted a pushed dump. + * + * Arming the floor is deferred to here rather than done at the emit, because + * the emit only says the dump left this process. `pushStats` is fire-and- + * forget over a socket that can die mid-flight, so arming on the attempt made + * a lost push look like a delivered one and shelved the retry for a full day. + */ + markStatsPushed(): void { + this.lastStatsPushAt = Date.now(); + this.retryStatsAfter = undefined; + } + + /** + * The dump was built but never reached the server. + * + * Backs off rather than leaving the floor past due, which would earn a fresh + * source dump on every 60s schema poll for as long as the connection stays + * broken — the case this is most likely to be called in. + */ + markStatsPushFailed(): void { + this.retryStatsAfter = Date.now() + Remote.STATS_RETRY_BACKOFF_MS; + } + async resetPgStatStatements(source: Connectable): Promise { const connector = this.sourceManager.getConnectorFor(source); await connector.resetPgStatStatements(); diff --git a/src/remote/stats-push-delivery.test.ts b/src/remote/stats-push-delivery.test.ts new file mode 100644 index 0000000..ae67ee1 --- /dev/null +++ b/src/remote/stats-push-delivery.test.ts @@ -0,0 +1,205 @@ +import { describe, expect, it, vi } from "vitest"; +import { + Statistics, + type ExportedStats, + type ServerApi, + type StatisticsMode, +} from "@query-doctor/core"; +import type { RpcStub } from "capnweb"; +import { Connectable } from "../sync/connectable.ts"; +import { ConnectionManager } from "../sync/connection-manager.ts"; +import { Remote } from "./remote.ts"; +import { hookUpApiReporter } from "./api-client.ts"; +import { DEFAULT_REFRESH_FLOOR_MS } from "./stats-drift.ts"; + +function makeRemote(): Remote { + const remote = new Remote( + Connectable.fromString("postgresql://postgres@localhost:5432/postgres"), + ConnectionManager.forRemoteDatabase(), + ); + vi.spyOn(remote.optimizer, "setStatistics").mockResolvedValue(undefined); + vi.spyOn(remote.optimizer, "restart").mockResolvedValue(undefined); + return remote; +} + +function table(name: string, reltuples: number): ExportedStats { + return { + schemaName: "public", + tableName: name, + reltuples, + relpages: 1, + relallvisible: 0, + columns: [], + indexes: [], + } as unknown as ExportedStats; +} + +/** Both fields are private; read them the way the seeding tests do. */ +function floorArmedAt(remote: Remote): number | undefined { + return (remote as unknown as { lastStatsPushAt?: number }).lastStatsPushAt; +} +function backoffUntil(remote: Remote): number | undefined { + return (remote as unknown as { retryStatsAfter?: number }).retryStatsAfter; +} + +/** Lets the fire-and-forget push settle before the assertion reads the floor. */ +function flush(): Promise { + return new Promise((resolve) => setImmediate(resolve)); +} + +/** The connection manager the drift check reads row counts through. */ +function sourceManagerOf(remote: Remote): ConnectionManager { + return (remote as unknown as { sourceManager: ConnectionManager }).sourceManager; +} + +/** The private dump the refresh calls, exposed so a poll can be counted. */ +function dumperOf( + remote: Remote, +): { dumpSourceStats(source: Connectable): Promise } { + return remote as unknown as { + dumpSourceStats(source: Connectable): Promise; + }; +} + +/** One schema-poll tick. Private, like the tick that drives it in production. */ +function poll(remote: Remote, source: Connectable): Promise { + return ( + remote as unknown as { + refreshStatsIfStale(source: Connectable): Promise; + } + ).refreshStatsIfStale(source); +} + +function apiWithPushStats( + pushStats: (stats: ExportedStats[]) => Promise, +): RpcStub { + return { pushStats } as unknown as RpcStub; +} + +/** + * The daily floor exists so a project's statistics can't go stale unnoticed. + * `pushStats` is fire-and-forget over a socket that dies mid-flight, so arming + * the floor when the dump is emitted rather than when the server accepts it + * makes a lost push indistinguishable from a delivered one — and shelves the + * retry for 24 hours. That is how a project went five days without a capture. + */ +describe("statistics push delivery", () => { + it("does not arm the daily floor on the emit alone", async () => { + const remote = makeRemote(); + + await remote.applyStatistics( + Statistics.statsModeFromExport([table("users", 10_000)]), + ); + + expect(floorArmedAt(remote)).toBeUndefined(); + }); + + it("arms the daily floor once the server accepts the push, and lifts the backoff", async () => { + const remote = makeRemote(); + hookUpApiReporter( + apiWithPushStats(() => Promise.resolve()), + remote, + ); + // A previous attempt had failed, so there is a live backoff to clear. + remote.markStatsPushFailed(); + expect(backoffUntil(remote)).toBeTypeOf("number"); + + await remote.applyStatistics( + Statistics.statsModeFromExport([table("users", 10_000)]), + ); + await flush(); + + expect(floorArmedAt(remote)).toBeTypeOf("number"); + expect(backoffUntil(remote)).toBeUndefined(); + }); + + it("backs off for minutes, not a day, when the push fails", async () => { + // The socket dropping mid-push must leave the floor where it was, so the + // next poll re-dumps rather than believing the stale numbers were stored. + const remote = makeRemote(); + hookUpApiReporter( + apiWithPushStats(() => Promise.reject(new Error("WebSocket connection failed."))), + remote, + ); + + await remote.applyStatistics( + Statistics.statsModeFromExport([table("users", 10_000)]), + ); + await flush(); + + expect(floorArmedAt(remote)).toBeUndefined(); + // The bug was a lost push buying a full day of silence. Anything on the + // order of the daily floor would reintroduce it, so pin the magnitude + // rather than the exact constant. + const backoff = backoffUntil(remote)! - Date.now(); + expect(backoff).toBeGreaterThan(60_000); + expect(backoff).toBeLessThan(DEFAULT_REFRESH_FLOOR_MS / 10); + }); + + it("re-dumps on a later poll when a push was lost, and not before the backoff", async () => { + // The regression test for the reported bug. Asserting on the floor field + // alone would have passed while the analyzer still sat silent for a day — + // what matters is whether the next poll actually dumps again. + // + // Only Date is faked: the push settles on a real microtask turn. + vi.useFakeTimers({ toFake: ["Date"] }); + try { + const remote = makeRemote(); + hookUpApiReporter( + apiWithPushStats(() => Promise.reject(new Error("WebSocket connection failed."))), + remote, + ); + const source = Connectable.fromString( + "postgresql://postgres@localhost:5432/source", + ); + // Row counts match the baseline, so drift stays quiet and the daily floor + // is the only thing that can earn a dump. That is the path that broke. + vi.spyOn(sourceManagerOf(remote), "getConnectorFor").mockReturnValue({ + getReltuplesByTable: async () => new Map([["public.users", 10_000]]), + } as never); + const dump = vi + .spyOn(dumperOf(remote), "dumpSourceStats") + .mockResolvedValue( + Statistics.statsModeFromExport([table("users", 10_000)]), + ); + + remote.seedStatsBaseline([table("users", 10_000)]); + vi.setSystemTime(Date.now() + DEFAULT_REFRESH_FLOOR_MS + 1_000); + + await poll(remote, source); + expect(dump).toHaveBeenCalledTimes(1); + await flush(); + + // Immediately after the lost push: backed off, so no second dump. + await poll(remote, source); + expect(dump).toHaveBeenCalledTimes(1); + + // Once the backoff lapses the analyzer tries again. Before this change the + // floor had been armed by the attempt, and this call dumped nothing. + vi.setSystemTime(backoffUntil(remote)! + 1_000); + await poll(remote, source); + expect(dump).toHaveBeenCalledTimes(2); + } finally { + vi.useRealTimers(); + } + }); + + it("keeps the local drift baseline even when the push is lost", async () => { + // The baseline describes what this analyzer dumped, not what the server + // stored. Rolling it back on a failed push would re-trigger Size Drift on + // every poll and dump the source over and over. + const remote = makeRemote(); + hookUpApiReporter( + apiWithPushStats(() => Promise.reject(new Error("WebSocket connection failed."))), + remote, + ); + + await remote.applyStatistics( + Statistics.statsModeFromExport([table("users", 10_000)]), + ); + await flush(); + + const baseline = (remote as unknown as { statsBaseline?: unknown }).statsBaseline; + expect(baseline).toBeDefined(); + }); +});