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