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

@@ -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<void> {
     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<void> {
     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<ChangesPage>(
+          link,
+          `/sync/v2/changes?cursor=${link.cursor}&limit=1`,
+          { timeoutMs: 120_000 },
+        ),
+      );
     }
     const fresh = listEntities()
       .map((entity) => entity.type)

src/backend/sync/records.ts

@@ -140,6 +140,9 @@ export async function listRecordErrors(userId: string): Promise<SyncRecord[]> {
   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<SyncRecord[]> {
   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));

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)

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<void> {
+    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(
+      [],
+    );
+  });
+});

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<string, string>) {
+  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();
+});

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<number>(),
+  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", () => {