From 11f6910d51898158e1366470aaca765b89a86468 Mon Sep 17 00:00:00 2001 From: ZacharyZcR Date: Sun, 18 Oct 2026 08:02:56 +0800 Subject: [PATCH] fix(sync): keep the changes cursor within the 32-bit seq column (#1665) The desktop learned the server's entity types by asking for changes after Number.MAX_SAFE_INTEGER. sync_records.seq is a 32-bit integer on Postgres, which rejects that bound value, so every pass needing the probe failed with 22003 out of range. The server now clamps the cursor to the largest seq the column holds and answers 400 for a cursor or limit that is not a non-negative integer. The desktop asks for a one-row page from its saved cursor instead. --- src/backend/sync/client/engine.ts | 25 +++++-- src/backend/sync/records.ts | 6 +- src/backend/sync/server/routes.ts | 24 ++++-- .../repositories/sync-records-changes.test.ts | 55 ++++++++++++++ src/backend/tests/sync/server/changes.test.ts | 75 +++++++++++++++++++ src/backend/tests/sync/two-way.test.ts | 19 +++++ 6 files changed, 191 insertions(+), 13 deletions(-) create mode 100644 src/backend/tests/database/repositories/sync-records-changes.test.ts create mode 100644 src/backend/tests/sync/server/changes.test.ts diff --git a/src/backend/sync/client/engine.ts b/src/backend/sync/client/engine.ts index 8ccd562b3..34b0c46ff 100644 --- a/src/backend/sync/client/engine.ts +++ b/src/backend/sync/client/engine.ts @@ -499,17 +499,21 @@ export class SyncEngine { ); all.push(...result.changes); next = result.nextCursor; - this.serverTypes = new Set(result.entityTypes.map((entry) => entry.type)); - this.state.serverTypes = result.entityTypes.map((entry) => ({ - type: entry.type, - owner: entry.owner, - readOnly: entry.readOnly, - })); + this.learnTypes(result); if (!result.hasMore) break; } return { changes: all, cursor: next }; } + private learnTypes(page: ChangesPage): void { + this.serverTypes = new Set(page.entityTypes.map((entry) => entry.type)); + this.state.serverTypes = page.entityTypes.map((entry) => ({ + type: entry.type, + owner: entry.owner, + readOnly: entry.readOnly, + })); + } + private async pull(link: SyncLink): Promise { const { changes, cursor } = await this.fetchChanges(link, link.cursor); const known = new Set(link.knownTypes); @@ -528,7 +532,14 @@ export class SyncEngine { /** Pulls every row of types this device has not pulled before. */ private async backfill(link: SyncLink): Promise { if (!this.serverTypes) { - await this.fetchChanges(link, Number.MAX_SAFE_INTEGER); + // Only the entity types are wanted, so a one-row page is enough. + this.learnTypes( + await remoteJson( + link, + `/sync/v2/changes?cursor=${link.cursor}&limit=1`, + { timeoutMs: 120_000 }, + ), + ); } const fresh = listEntities() .map((entity) => entity.type) diff --git a/src/backend/sync/records.ts b/src/backend/sync/records.ts index c32883644..71377b295 100644 --- a/src/backend/sync/records.ts +++ b/src/backend/sync/records.ts @@ -140,6 +140,9 @@ export async function listRecordErrors(userId: string): Promise { return rows.filter((row) => !!row.error); } +/** The largest seq the column holds: a 32-bit integer on Postgres and MySQL. */ +const MAX_SEQ = 2_147_483_647; + export async function listChanges( userId: string, cursor: number, @@ -148,7 +151,8 @@ export async function listChanges( ): Promise { const conditions = [ eq(syncRecords.userId, userId), - gt(syncRecords.seq, cursor), + // Postgres rejects a bound value the column cannot hold. + gt(syncRecords.seq, Math.min(cursor, MAX_SEQ)), ]; if (types && types.length > 0) { conditions.push(inArray(syncRecords.entityType, types)); diff --git a/src/backend/sync/server/routes.ts b/src/backend/sync/server/routes.ts index 1c3ce4923..fb1639d54 100644 --- a/src/backend/sync/server/routes.ts +++ b/src/backend/sync/server/routes.ts @@ -37,6 +37,17 @@ function userOf(req: Request): string { return (req as AuthenticatedRequest).userId; } +/** A non-negative integer query value, the fallback when absent, null when malformed. */ +function queryInt(value: unknown, fallback: number): number | null { + if (value === undefined) return fallback; + const parsed = Number(value); + return typeof value === "string" && + /^\d+$/.test(value) && + Number.isSafeInteger(parsed) + ? parsed + : null; +} + /** * @openapi * /sync/v2/info: @@ -224,6 +235,8 @@ router.post( * responses: * 200: * description: A page of changes, the next cursor and whether more remain. + * 400: + * description: The cursor or limit is not a non-negative integer. * 500: * description: Failed to read changes. */ @@ -232,11 +245,12 @@ router.get( authenticateJWT, async (req: Request, res: Response) => { const userId = userOf(req); - const cursor = Math.max(0, Number(req.query.cursor) || 0); - const limit = Math.min( - MAX_PAGE, - Math.max(1, Number(req.query.limit) || MAX_PAGE), - ); + const cursor = queryInt(req.query.cursor, 0); + const pageSize = queryInt(req.query.limit, MAX_PAGE); + if (cursor === null || pageSize === null) { + return res.status(400).json({ error: "Invalid cursor or limit" }); + } + const limit = Math.min(MAX_PAGE, Math.max(1, pageSize)); const types = typeof req.query.types === "string" && req.query.types ? req.query.types.split(",").filter(Boolean) diff --git a/src/backend/tests/database/repositories/sync-records-changes.test.ts b/src/backend/tests/database/repositories/sync-records-changes.test.ts new file mode 100644 index 000000000..11b2973b5 --- /dev/null +++ b/src/backend/tests/database/repositories/sync-records-changes.test.ts @@ -0,0 +1,55 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { sql } from "drizzle-orm"; +import { TestSqliteDatabase } from "./test-support.js"; +import type { DatabaseContext } from "../../../database/repositories/database-context.js"; + +const state = vi.hoisted(() => ({ + context: null as DatabaseContext | null, +})); + +vi.mock("../../../database/repositories/factory.js", () => ({ + createCurrentRepositoryContext: () => state.context, +})); +vi.mock("../../../utils/database-save-trigger.js", () => ({ + DatabaseSaveTrigger: { triggerSave: vi.fn() }, +})); + +const { listChanges } = await import("../../../sync/records.js"); + +// Runs against Postgres and MySQL too in the database-dialects CI job, where +// sync_records.seq is a 32-bit integer. +describe("listChanges", () => { + let adapter: TestSqliteDatabase | null = null; + + afterEach(async () => { + if (adapter) { + await adapter.close(); + adapter = null; + } + state.context = null; + }); + + async function seed(): Promise { + adapter = new TestSqliteDatabase(); + state.context = await adapter.connect(); + await adapter.run( + sql`INSERT INTO users (id, username, password_hash) VALUES ('user-1', 'user-1', '')`, + ); + await adapter.run( + sql`INSERT INTO sync_records (user_id, entity_type, sync_id, revision, seq) VALUES ('user-1', 'hosts', 'a', 1, 1), ('user-1', 'hosts', 'b', 1, 2)`, + ); + } + + it("lists records after the cursor in seq order", async () => { + await seed(); + const records = await listChanges("user-1", 1, 10); + expect(records.map((record) => record.syncId)).toEqual(["b"]); + }); + + it("answers no records for a cursor past what seq can hold", async () => { + await seed(); + expect(await listChanges("user-1", Number.MAX_SAFE_INTEGER, 10)).toEqual( + [], + ); + }); +}); diff --git a/src/backend/tests/sync/server/changes.test.ts b/src/backend/tests/sync/server/changes.test.ts new file mode 100644 index 000000000..75199a71a --- /dev/null +++ b/src/backend/tests/sync/server/changes.test.ts @@ -0,0 +1,75 @@ +import { beforeEach, expect, it, vi } from "vitest"; + +const { listChanges } = vi.hoisted(() => ({ listChanges: vi.fn() })); +vi.mock("../../../utils/auth-manager.js", () => ({ + AuthManager: { getInstance: () => ({ createAuthMiddleware: () => vi.fn() }) }, +})); +vi.mock("../../../utils/logger.js", () => ({ syncLogger: { error: vi.fn() } })); +vi.mock("../../../utils/app-version.js", () => ({ getLocalVersion: vi.fn() })); +vi.mock("../../../plugins/index.js", () => ({ getPluginRuntime: vi.fn() })); +vi.mock("../../../plugins/sync-registry.js", () => ({ + getEntity: vi.fn(), + listEntities: () => [{ type: "hosts", owner: "core", order: 10 }], +})); +vi.mock("../../../database/repositories/factory.js", () => ({})); +vi.mock("../../../sync/entities.js", () => ({ + registerCoreSyncEntities: vi.fn(), +})); +vi.mock("../../../sync/records.js", () => ({ + listChanges, + recordKey: vi.fn(), +})); +vi.mock("../../../sync/server/feed.js", () => ({ + withUserLock: (_userId: string, fn: () => unknown) => fn(), + reconcileUser: async () => ({ wire: new Map() }), +})); +vi.mock("../../../sync/server/push.js", () => ({})); +vi.mock("../../../database/routes/branding-settings.js", () => ({})); + +import router from "../../../sync/server/routes.js"; + +const route = router.stack.find( + (layer) => layer.route?.path === "/v2/changes", +)!.route!; +const handler = route.stack[route.stack.length - 1].handle; + +beforeEach(() => { + vi.clearAllMocks(); + listChanges.mockResolvedValue([]); +}); + +async function changes(query: Record) { + const res = { status: vi.fn().mockReturnThis(), json: vi.fn() }; + await handler({ userId: "user-1", query } as never, res as never, vi.fn()); + return res; +} + +it("answers an empty page with the entity types past the newest change", async () => { + const cursor = String(Number.MAX_SAFE_INTEGER); + const res = await changes({ cursor, limit: "1000" }); + expect(res.status).not.toHaveBeenCalled(); + expect(res.json).toHaveBeenCalledWith({ + changes: [], + nextCursor: Number.MAX_SAFE_INTEGER, + hasMore: false, + entityTypes: [{ type: "hosts", owner: "core", order: 10, readOnly: false }], + }); +}); + +it("defaults a missing cursor and limit", async () => { + await changes({}); + expect(listChanges).toHaveBeenCalledWith("user-1", 0, 1001, undefined); +}); + +it.each([ + { cursor: "abc" }, + { cursor: "-1" }, + { cursor: "1.5" }, + { cursor: "Infinity" }, + { limit: "1.5" }, + { limit: "-5" }, +])("rejects a malformed query %o", async (query) => { + const res = await changes(query); + expect(res.status).toHaveBeenCalledWith(400); + expect(listChanges).not.toHaveBeenCalled(); +}); diff --git a/src/backend/tests/sync/two-way.test.ts b/src/backend/tests/sync/two-way.test.ts index 5524eb464..15795f8ff 100644 --- a/src/backend/tests/sync/two-way.test.ts +++ b/src/backend/tests/sync/two-way.test.ts @@ -15,6 +15,7 @@ const state = vi.hoisted(() => ({ server: null as null | { drizzle: unknown }, allow: true, sharedHostIds: new Set(), + paths: [] as string[], })); const SERVER_USER = "srv-user"; @@ -123,6 +124,7 @@ vi.mock("../../sync/client/http.js", async (importOriginal) => { path: string, init: { method?: string; body?: unknown } = {}, ) => { + state.paths.push(path); const previous = state.current; state.current = state.server; try { @@ -249,6 +251,7 @@ beforeEach(async () => { resetSeqAllocator(); state.allow = true; state.sharedHostIds.clear(); + state.paths = []; server = await side(SERVER_USER); state.server = server.ctx; desktops = [await side("desk-a"), await side("desk-b")]; @@ -317,6 +320,22 @@ describe("linking a desktop", () => { expect(host.ip).toBe("new"); expect(await on(desk, () => listConflicts(desk.userId))).toEqual([]); }); + + it("never asks for a cursor past what the server's seq column holds", async () => { + const [desk] = desktops; + await on(server, () => addHost(server, { name: "srv" })); + + const engine = await linkDesktop(desk); + await pass(desk, engine); + + const cursors = state.paths + .filter((path) => path.startsWith("/sync/v2/changes?")) + .map((path) => + Number(new URLSearchParams(path.split("?")[1]).get("cursor")), + ); + expect(cursors.length).toBeGreaterThan(0); + expect(Math.max(...cursors)).toBeLessThanOrEqual(2_147_483_647); + }); }); describe("after linking", () => {