From fa559e44ab923ac952ccb1072f885d2c25ade63c Mon Sep 17 00:00:00 2001 From: Xetera Date: Mon, 26 Jan 2026 22:14:54 +0300 Subject: [PATCH 1/2] feat: return deltas for the source schema from GET request It's not super clear if this is the final approach we want to take for this feature. This schema is massive and it can be very costly to constantly query this from the source. --- src/remote/remote-controller.ts | 29 ++++++++++++++++++++-------- src/remote/remote.ts | 34 +++++++++++++++++++++++++++++---- src/remote/schema-loader.ts | 24 +++++++++++++++++++++++ 3 files changed, 75 insertions(+), 12 deletions(-) create mode 100644 src/remote/schema-loader.ts diff --git a/src/remote/remote-controller.ts b/src/remote/remote-controller.ts index fc9de27..8a239ab 100644 --- a/src/remote/remote-controller.ts +++ b/src/remote/remote-controller.ts @@ -44,7 +44,7 @@ export class RemoteController { } else if (request.method === "POST") { return await this.onFullSync(request); } else if (request.method === "GET") { - return this.getStatus(); + return await this.getStatus(); } return methodNotAllowed(); } @@ -102,23 +102,25 @@ export class RemoteController { } } - private getStatus(): Response { + private async getStatus(): Promise { if (!this.syncResponse || this.syncStatus !== SyncStatus.COMPLETED) { return Response.json({ status: this.syncStatus }); } const { schema, meta } = this.syncResponse; - const queries = this.remote.optimizer.getQueries(); - const disabledIndexes = this.remote.optimizer.getDisabledIndexes(); - this.remote.pollQueriesOnce().catch((error) => { - log.error("Failed to poll queries", "remote-controller"); - console.error(error); - }); + const { queries, diffs, disabledIndexes } = await this.remote.getStatus(); + let deltas: DeltasResult; + if (diffs.status === "fulfilled") { + deltas = { type: "ok", value: diffs.value }; + } else { + deltas = { type: "error", value: String(diffs.reason) }; + } return Response.json({ status: this.syncStatus, meta, schema, queries: { type: "ok", value: queries }, disabledIndexes: { type: "ok", value: disabledIndexes }, + deltas, }); } @@ -225,3 +227,14 @@ export class RemoteController { function methodNotAllowed(): Response { return Response.json("Method not allowed", { status: 405 }); } + +type DeltasResult = { + type: "ok"; + // the type of this is not super important + // currently the frontend only cares whether + // or not this array is empty + value: unknown[]; +} | { + type: "error"; + value: string; +}; diff --git a/src/remote/remote.ts b/src/remote/remote.ts index aeb5327..993e568 100644 --- a/src/remote/remote.ts +++ b/src/remote/remote.ts @@ -9,12 +9,14 @@ import { type Connectable } from "../sync/connectable.ts"; import { DumpCommand, RestoreCommand } from "../sync/schema-link.ts"; import { ConnectionManager } from "../sync/connection-manager.ts"; import { type RecentQuery } from "../sql/recent-query.ts"; -import { type FullSchema, SchemaDiffer } from "../sync/schema_differ.ts"; +import { type Op } from "jsondiffpatch/formatters/jsonpatch"; +import { type FullSchema } from "../sync/schema_differ.ts"; import { type RemoteSyncFullSchemaResponse } from "./remote.dto.ts"; import { QueryOptimizer } from "./query-optimizer.ts"; import { EventEmitter } from "node:events"; import { log } from "../log.ts"; import { QueryLoader } from "./query-loader.ts"; +import { SchemaLoader } from "./schema-loader.ts"; type RemoteEvents = { dumpLog: [line: string]; @@ -40,7 +42,6 @@ export class Remote extends EventEmitter { */ private static readonly STATS_ROWS_THRESHOLD = 5_000; - private readonly differ = new SchemaDiffer(); readonly optimizer: QueryOptimizer; /** @@ -58,6 +59,7 @@ export class Remote extends EventEmitter { private isPolling = false; private queryLoader?: QueryLoader; + private schemaLoader?: SchemaLoader; constructor( /** This has to be a local url. Very bad things will happen if this is a remote URL */ @@ -100,7 +102,7 @@ export class Remote extends EventEmitter { ]); if (fullSchema.status === "fulfilled") { - this.differ.put(source, fullSchema.value); + this.schemaLoader?.update(fullSchema.value); } // Second: resolve stats strategy using table list from schema @@ -147,11 +149,34 @@ export class Remote extends EventEmitter { }; } + async getStatus() { + const queries = this.optimizer.getQueries(); + const disabledIndexes = this.optimizer.getDisabledIndexes(); + const [diffs] = await Promise.allSettled([ + this.schemaLoader?.poll().then( + (results) => results.diffs, + (error) => { + log.error("Failed to poll schema", "remote"); + console.error(error); + throw error; + }, + ) ?? + [] as Op[], /* no panic in case schemaLoader has not loaded in yet */ + this.pollQueriesOnce().catch((error) => { + log.error("Failed to poll queries", "remote"); + console.error(error); + throw error; + }), + ]); + + return { queries, diffs, disabledIndexes }; + } + /** * Runs a single poll of pg_stat_statements if * there isn't already an in-flight request */ - async pollQueriesOnce() { + private async pollQueriesOnce() { if (this.queryLoader && !this.isPolling) { try { this.isPolling = true; @@ -299,6 +324,7 @@ export class Remote extends EventEmitter { this.queryLoader.stop(); } this.queryLoader = new QueryLoader(this.sourceManager, source); + this.schemaLoader = new SchemaLoader(this.sourceManager, source); this.queryLoader.on("pollError", (error) => { log.error("Failed to poll queries", "remote"); console.error(error); diff --git a/src/remote/schema-loader.ts b/src/remote/schema-loader.ts new file mode 100644 index 0000000..e2407a3 --- /dev/null +++ b/src/remote/schema-loader.ts @@ -0,0 +1,24 @@ +import { Connectable } from "../sync/connectable.ts"; +import { ConnectionManager } from "../sync/connection-manager.ts"; +import { FullSchema, SchemaDiffer } from "../sync/schema_differ.ts"; + +export class SchemaLoader { + constructor( + private readonly sourceManager: ConnectionManager, + private readonly connectable: Connectable, + ) {} + + private readonly differ = new SchemaDiffer(); + + async poll() { + const connector = this.sourceManager.getConnectorFor(this.connectable); + const schema = await connector.getSchema(); + + const diffs = this.update(schema) ?? []; + return { diffs }; + } + + update(fullSchema: FullSchema) { + return this.differ.put(this.connectable, fullSchema); + } +} From 48641fda316ac26a848278bd4ceddce2ac307371 Mon Sep 17 00:00:00 2001 From: Xetera Date: Mon, 26 Jan 2026 22:29:51 +0300 Subject: [PATCH 2/2] chore: add test for schema drift detection --- src/remote/remote.test.ts | 93 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 93 insertions(+) diff --git a/src/remote/remote.test.ts b/src/remote/remote.test.ts index 3f9d56a..f818255 100644 --- a/src/remote/remote.test.ts +++ b/src/remote/remote.test.ts @@ -6,6 +6,7 @@ import { assertEquals } from "@std/assert/equals"; import { ConnectionManager } from "../sync/connection-manager.ts"; import { assertArrayIncludes } from "@std/assert"; import { PgIdentifier } from "@query-doctor/core"; +import { type Op } from "jsondiffpatch/formatters/jsonpatch"; const TEST_TARGET_CONTAINER_NAME = "postgres:17"; const TEST_TARGET_CONTAINER_TIMESCALEDB_NAME = @@ -423,3 +424,95 @@ Deno.test({ } }, }); + +Deno.test({ + name: "schema loader detects changes after database modification", + sanitizeOps: false, + sanitizeResources: false, + fn: async () => { + const [sourceDb, targetDb] = await Promise.all([ + new PostgreSqlContainer("postgres:17") + .withCopyContentToContainer([ + { + content: ` + create extension pg_stat_statements; + create table testing(a int, b text); + insert into testing values (1, 'test'); + create index "testing_b_idx" on testing(b); + select * from testing where a = 1; + `, + target: "/docker-entrypoint-initdb.d/init.sql", + }, + ]) + .withCommand(["-c", "shared_preload_libraries=pg_stat_statements"]) + .start(), + testSpawnTarget(), + ]); + + try { + const target = Connectable.fromString(targetDb.getConnectionUri()); + const source = Connectable.fromString(sourceDb.getConnectionUri()); + + const manager = ConnectionManager.forLocalDatabase(); + await using remote = new Remote(target, manager); + + const sourcePg = postgres(source.toString()); + + await remote.syncFrom(source); + await remote.optimizer.finish; + + const initialStatus = await remote.getStatus(); + const initialDiffsResult = initialStatus.diffs; + assertEquals( + initialDiffsResult.status, + "fulfilled", + "Schema poll should succeed", + ); + const initialDiffs = initialDiffsResult.status === "fulfilled" + ? initialDiffsResult.value + : []; + assertEquals( + initialDiffs.length, + 0, + "Should have no diffs initially after sync", + ); + + await sourcePg.unsafe(` + alter table testing add column c int; + create index "testing_c_idx" on testing(c); + `); + + const statusAfterChange = await remote.getStatus(); + const diffsResult = statusAfterChange.diffs; + + assertEquals( + diffsResult.status, + "fulfilled", + "Schema poll should succeed", + ); + const diffs = diffsResult.status === "fulfilled" + ? diffsResult.value + : []; + + assertEquals( + diffs.length, + 2, + "Should detect 2 schema changes (added column and index)", + ); + + const addedColumnDiff = diffs.find((diff: Op) => + typeof diff.path === "string" && diff.path.includes("columns") + ); + assertEquals(addedColumnDiff?.op, "add", "Should detect column addition"); + + const addedIndexDiff = diffs.find((diff: Op) => + typeof diff.path === "string" && diff.path.includes("indexes") + ); + assertEquals(addedIndexDiff?.op, "add", "Should detect index addition"); + + await sourcePg.end(); + } finally { + await Promise.all([sourceDb.stop(), targetDb.stop()]); + } + }, +});