From 8df0e4f3d627e386a0c8cf7baa833eb85d5c4632 Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Fri, 4 Sep 2026 00:07:53 -0400 Subject: [PATCH 1/3] fix: bound cloud mutation payloads --- convex/accessControl.test.ts | 70 +- convex/cloudProtocol.test.ts | 147 +++++ convex/folders.test.ts | 8 +- convex/folders.ts | 12 + convex/function_spec.json | 123 +++- convex/invites.ts | 10 + convex/lib/cloudProtocol.ts | 33 + convex/ops.ts | 76 ++- convex/pages.ts | 12 + convex/shares.ts | 10 + convex/strategies.ts | 14 + convex/syncBoundaries.test.ts | 135 +++- convex/users.ts | 9 +- lib/collab/cloud_sync_error_message.dart | 8 + lib/collab/collab_models.dart | 42 ++ lib/collab/convex_strategy_repository.dart | 38 +- lib/collab/generated/convex_models.dart | 125 +++- lib/collab/generated/icarus_convex_api.dart | 105 ++- .../collab/strategy_op_queue_provider.dart | 252 +++++++- lib/widgets/cloud_sync_status_chip.dart | 18 +- .../collab/cloud_sync_error_message_test.dart | 11 + test/strategy_op_queue_provider_test.dart | 598 ++++++++++++++++++ test/widgets/cloud_sync_status_chip_test.dart | 82 ++- tool/audit_convex_contract.mjs | 12 + tool/snapshot_convex_contract.mjs | 9 +- 25 files changed, 1860 insertions(+), 99 deletions(-) create mode 100644 convex/cloudProtocol.test.ts diff --git a/convex/accessControl.test.ts b/convex/accessControl.test.ts index 0de684cf..f1ac9a06 100644 --- a/convex/accessControl.test.ts +++ b/convex/accessControl.test.ts @@ -6,6 +6,7 @@ import { import { makeFunctionReference } from "convex/server"; import { describe, expect, test } from "vitest"; import type { DataModel } from "./_generated/dataModel"; +import { CURRENT_CLOUD_PROTOCOL_VERSION } from "./lib/cloudProtocol"; import schema from "./schema"; import { modules } from "./test.setup"; @@ -55,9 +56,15 @@ async function createHarness(): Promise<{ const b = t.withIdentity(identity("b")); const c = t.withIdentity(identity("c")); await Promise.all([ - a.mutation(ensureCurrentUser, {}), - b.mutation(ensureCurrentUser, {}), - c.mutation(ensureCurrentUser, {}), + a.mutation(ensureCurrentUser, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + }), + b.mutation(ensureCurrentUser, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + }), + c.mutation(ensureCurrentUser, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + }), ]); return { t, a, b, c }; } @@ -69,6 +76,7 @@ async function seedStrategy( folderPublicId?: string, ) { await owner.mutation(createStrategy, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, publicId: strategyPublicId, name: "A private Strategy", mapData: "ascent", @@ -100,12 +108,16 @@ describe("A/B/C access boundary", () => { ).rejects.toThrow("Forbidden"); await a.mutation(createShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, targetType: "strategy", targetPublicId: strategyPublicId, token: "strategy-viewer-token", role: "viewer", }); - await b.mutation(redeemShare, { token: "strategy-viewer-token" }); + await b.mutation(redeemShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + token: "strategy-viewer-token", + }); await expect( b.query(getStrategyShell, { strategyPublicId }), @@ -121,6 +133,7 @@ describe("A/B/C access boundary", () => { }); await expect( b.mutation(updateStrategy, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 0, name: "Viewer must not write", @@ -128,6 +141,7 @@ describe("A/B/C access boundary", () => { ).rejects.toThrow("Forbidden"); await expect( b.mutation(addPage, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 0, pagePublicId: "viewer-page", @@ -138,14 +152,19 @@ describe("A/B/C access boundary", () => { ).rejects.toThrow("Forbidden"); await a.mutation(createShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, targetType: "strategy", targetPublicId: strategyPublicId, token: "strategy-editor-token", role: "editor", }); - await b.mutation(redeemShare, { token: "strategy-editor-token" }); + await b.mutation(redeemShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + token: "strategy-editor-token", + }); await expect( b.mutation(updateStrategy, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 0, name: "Edited by B", @@ -153,6 +172,7 @@ describe("A/B/C access boundary", () => { ).resolves.toMatchObject({ ok: true, revision: 1 }); await expect( b.mutation(addPage, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 1, pagePublicId: "editor-page", @@ -172,15 +192,20 @@ describe("A/B/C access boundary", () => { }); await a.mutation(revokeShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, targetType: "strategy", targetPublicId: strategyPublicId, token: "strategy-editor-token", }); await expect( - c.mutation(redeemShare, { token: "strategy-editor-token" }), + c.mutation(redeemShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + token: "strategy-editor-token", + }), ).rejects.toThrow("Share link revoked"); await expect( b.mutation(updateStrategy, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 2, name: "B keeps redeemed access", @@ -195,10 +220,12 @@ describe("A/B/C access boundary", () => { const strategyPublicId = "nested-strategy"; await a.mutation(createFolder, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, publicId: rootFolderPublicId, name: "A root", }); await a.mutation(createFolder, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, publicId: childFolderPublicId, name: "A child", parentFolderPublicId: rootFolderPublicId, @@ -218,12 +245,16 @@ describe("A/B/C access boundary", () => { ).rejects.toThrow("Forbidden"); await a.mutation(createShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, targetType: "folder", targetPublicId: rootFolderPublicId, token: "folder-viewer-token", role: "viewer", }); - await b.mutation(redeemShare, { token: "folder-viewer-token" }); + await b.mutation(redeemShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + token: "folder-viewer-token", + }); await expect( b.query(listFolderTree, { scope: "shared" }), @@ -245,6 +276,7 @@ describe("A/B/C access boundary", () => { ]); await expect( b.mutation(updateStrategy, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 0, name: "Viewer must not write", @@ -252,20 +284,26 @@ describe("A/B/C access boundary", () => { ).rejects.toThrow("Forbidden"); await expect( b.mutation(updateFolder, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, folderPublicId: childFolderPublicId, name: "Viewer must not rename", }), ).rejects.toThrow("Forbidden"); await a.mutation(createShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, targetType: "folder", targetPublicId: rootFolderPublicId, token: "folder-editor-token", role: "editor", }); - await b.mutation(redeemShare, { token: "folder-editor-token" }); + await b.mutation(redeemShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + token: "folder-editor-token", + }); await expect( b.mutation(updateStrategy, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 0, name: "Inherited editor write", @@ -273,12 +311,14 @@ describe("A/B/C access boundary", () => { ).resolves.toMatchObject({ ok: true, revision: 1 }); await expect( b.mutation(updateFolder, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, folderPublicId: rootFolderPublicId, name: "Editors are not owners", }), ).rejects.toThrow("Forbidden"); await expect( b.mutation(createShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, targetType: "folder", targetPublicId: rootFolderPublicId, token: "editor-cannot-reshare", @@ -287,18 +327,23 @@ describe("A/B/C access boundary", () => { ).rejects.toThrow("Forbidden"); await a.mutation(revokeShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, targetType: "folder", targetPublicId: rootFolderPublicId, token: "folder-editor-token", }); await expect( - c.mutation(redeemShare, { token: "folder-editor-token" }), + c.mutation(redeemShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + token: "folder-editor-token", + }), ).rejects.toThrow("Share link revoked"); await expect( c.query(listFolderTree, { scope: "shared" }), ).resolves.toEqual([]); await expect( b.mutation(updateStrategy, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 1, name: "Redeemed editor remains durable", @@ -311,14 +356,19 @@ describe("A/B/C access boundary", () => { const strategyPublicId = "deleted-strategy"; await seedStrategy(a, strategyPublicId, "deleted-page"); await a.mutation(createShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, targetType: "strategy", targetPublicId: strategyPublicId, token: "deleted-strategy-token", role: "editor", }); - await b.mutation(redeemShare, { token: "deleted-strategy-token" }); + await b.mutation(redeemShare, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + token: "deleted-strategy-token", + }); await a.mutation(deleteStrategy, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 0, }); diff --git a/convex/cloudProtocol.test.ts b/convex/cloudProtocol.test.ts new file mode 100644 index 00000000..114264b6 --- /dev/null +++ b/convex/cloudProtocol.test.ts @@ -0,0 +1,147 @@ +import { convexTest } from "convex-test"; +import { makeFunctionReference } from "convex/server"; +import { describe, expect, test } from "vitest"; +import { CURRENT_CLOUD_PROTOCOL_VERSION } from "./lib/cloudProtocol"; +import schema from "./schema"; +import { modules } from "./test.setup"; + +const publicMutations = [ + ["folders:create", { publicId: "folder", name: "Folder" }], + ["folders:update", { folderPublicId: "folder" }], + ["folders:move", { folderPublicId: "folder" }], + ["folders:delete", { folderPublicId: "folder" }], + [ + "invites:create", + { strategyPublicId: "strategy", token: "token", role: "viewer" }, + ], + ["invites:redeem", { token: "token" }], + ["invites:revoke", { strategyPublicId: "strategy", token: "token" }], + [ + "ops:applyBatch", + { strategyPublicId: "strategy", clientId: "client", ops: [] }, + ], + [ + "pages:add", + { + strategyPublicId: "strategy", + expectedRevision: 0, + pagePublicId: "page", + name: "Page", + sortIndex: 0, + isAttack: true, + }, + ], + [ + "pages:rename", + { + strategyPublicId: "strategy", + pagePublicId: "page", + name: "Page", + expectedRevision: 0, + }, + ], + [ + "pages:delete", + { + strategyPublicId: "strategy", + pagePublicId: "page", + expectedRevision: 0, + }, + ], + [ + "pages:reorder", + { + strategyPublicId: "strategy", + orderedPagePublicIds: [], + expectedRevision: 0, + }, + ], + [ + "shares:create", + { + targetType: "strategy", + targetPublicId: "strategy", + token: "token", + role: "viewer", + }, + ], + [ + "shares:revoke", + { targetType: "strategy", targetPublicId: "strategy", token: "token" }, + ], + ["shares:redeem", { token: "token" }], + [ + "strategies:create", + { publicId: "strategy", name: "Strategy", mapData: "ascent" }, + ], + [ + "strategies:createWithInitialPage", + { + publicId: "strategy", + name: "Strategy", + mapData: "ascent", + initialPagePublicId: "page", + initialPageName: "Page 1", + initialPageIsAttack: true, + }, + ], + [ + "strategies:update", + { strategyPublicId: "strategy", expectedRevision: 0 }, + ], + [ + "strategies:move", + { strategyPublicId: "strategy", expectedRevision: 0 }, + ], + [ + "strategies:delete", + { strategyPublicId: "strategy", expectedRevision: 0 }, + ], + ["users:ensureCurrentUser", {}], +] as const; + +describe("public cloud mutation protocol gate", () => { + test.each(publicMutations)("%s rejects an old protocol canonically", async ( + identifier, + args, + ) => { + const t = convexTest(schema, modules); + const mutation = makeFunctionReference<"mutation">(identifier); + const error = await t + .mutation(mutation, { + ...args, + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION - 1, + }) + .then( + () => null, + (caught: unknown) => caught as { data?: unknown }, + ); + + expect(error).not.toBeNull(); + expect(typeof error?.data).toBe("string"); + expect(JSON.parse(error?.data as string)).toEqual({ + code: "CLIENT_UPGRADE_REQUIRED", + message: "Client upgrade required", + }); + }); + + test("a newer unknown protocol receives the same canonical error", async () => { + const t = convexTest(schema, modules); + const mutation = makeFunctionReference<"mutation">( + "users:ensureCurrentUser", + ); + const error = await t + .mutation(mutation, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION + 1, + }) + .then( + () => null, + (caught: unknown) => caught as { data?: unknown }, + ); + + expect(JSON.parse(error?.data as string)).toEqual({ + code: "CLIENT_UPGRADE_REQUIRED", + message: "Client upgrade required", + }); + }); +}); diff --git a/convex/folders.test.ts b/convex/folders.test.ts index b1013ee1..87e3a331 100644 --- a/convex/folders.test.ts +++ b/convex/folders.test.ts @@ -6,6 +6,7 @@ import { import { makeFunctionReference } from "convex/server"; import { expect, test } from "vitest"; import type { DataModel } from "./_generated/dataModel"; +import { CURRENT_CLOUD_PROTOCOL_VERSION } from "./lib/cloudProtocol"; import schema from "./schema"; import { modules } from "./test.setup"; @@ -31,7 +32,9 @@ async function createHarness(): Promise<{ }> { const t = convexTest(schema, modules); const owner = t.withIdentity(identity); - await owner.mutation(ensureCurrentUser, {}); + await owner.mutation(ensureCurrentUser, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + }); return { t, owner }; } @@ -41,6 +44,7 @@ async function seedFolder( parentFolderPublicId?: string, ) { await owner.mutation(createFolder, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, publicId, name: publicId, parentFolderPublicId, @@ -69,6 +73,7 @@ test("folder move rejects self-parent without changing the folder", async () => await expect( owner.mutation(moveFolder, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, folderPublicId: "root", parentFolderPublicId: "root", }), @@ -85,6 +90,7 @@ test("folder move rejects a descendant parent without changing the tree", async await expect( owner.mutation(moveFolder, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, folderPublicId: "root", parentFolderPublicId: "grandchild", }), diff --git a/convex/folders.ts b/convex/folders.ts index 41181bc7..d6386902 100644 --- a/convex/folders.ts +++ b/convex/folders.ts @@ -8,6 +8,10 @@ import { requireCurrentUser, } from "./lib/auth"; import { getFolderByPublicId } from "./lib/entities"; +import { + assertSupportedCloudProtocol, + cloudProtocolArgs, +} from "./lib/cloudProtocol"; import { conflictError, forbiddenError, @@ -136,6 +140,7 @@ const folderScopeValidator = v.optional( export const create = mutation({ args: { + ...cloudProtocolArgs, publicId: v.string(), name: v.string(), parentFolderPublicId: v.optional(v.string()), @@ -148,6 +153,7 @@ export const create = mutation({ }, returns: createResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const user = await requireCurrentUser(ctx); const now = Date.now(); @@ -194,6 +200,7 @@ export const create = mutation({ export const update = mutation({ args: { + ...cloudProtocolArgs, folderPublicId: v.string(), name: v.optional(v.string()), iconId: v.optional(v.number()), @@ -208,6 +215,7 @@ export const update = mutation({ }, returns: okResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const folder = await getFolderByPublicId(ctx, args.folderPublicId); const { role } = await assertFolderRole(ctx, folder, "owner"); @@ -302,11 +310,13 @@ export const listTree = query({ export const move = mutation({ args: { + ...cloudProtocolArgs, folderPublicId: v.string(), parentFolderPublicId: v.optional(v.string()), }, returns: okResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const folder = await getFolderByPublicId(ctx, args.folderPublicId); const { role } = await assertFolderRole(ctx, folder, "owner"); @@ -336,10 +346,12 @@ export const move = mutation({ const deleteFolder = mutation({ args: { + ...cloudProtocolArgs, folderPublicId: v.string(), }, returns: okResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const folder = await getFolderByPublicId(ctx, args.folderPublicId); const { role } = await assertFolderRole(ctx, folder, "owner"); diff --git a/convex/function_spec.json b/convex/function_spec.json index 25546aba..3af50a9b 100644 --- a/convex/function_spec.json +++ b/convex/function_spec.json @@ -39358,6 +39358,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "color": { "fieldType": { "type": "string" @@ -39460,6 +39466,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "folderPublicId": { "fieldType": { "type": "string" @@ -39672,6 +39684,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "folderPublicId": { "fieldType": { "type": "string" @@ -39726,6 +39744,12 @@ }, "optional": true }, + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "color": { "fieldType": { "type": "string" @@ -40712,6 +40736,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "expiresAt": { "fieldType": { "type": "number" @@ -40937,6 +40967,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "token": { "fieldType": { "type": "string" @@ -40993,6 +41029,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "strategyPublicId": { "fieldType": { "type": "string" @@ -140684,6 +140726,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "expectedRevision": { "fieldType": { "type": "number" @@ -140812,6 +140860,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "expectedRevision": { "fieldType": { "type": "number" @@ -140970,6 +141024,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "expectedRevision": { "fieldType": { "type": "number" @@ -141060,6 +141120,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "expectedRevision": { "fieldType": { "type": "number" @@ -141141,6 +141207,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "role": { "fieldType": { "type": "union", @@ -141293,6 +141365,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "token": { "fieldType": { "type": "string" @@ -141420,6 +141498,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "targetPublicId": { "fieldType": { "type": "string" @@ -141472,6 +141556,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "folderPublicId": { "fieldType": { "type": "string" @@ -141576,6 +141666,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "folderPublicId": { "fieldType": { "type": "string" @@ -141730,6 +141826,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "expectedRevision": { "fieldType": { "type": "number" @@ -142242,6 +142344,12 @@ "args": { "type": "object", "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "expectedRevision": { "fieldType": { "type": "number" @@ -142332,6 +142440,12 @@ }, "optional": true }, + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + }, "expectedRevision": { "fieldType": { "type": "number" @@ -166041,7 +166155,14 @@ { "args": { "type": "object", - "value": {} + "value": { + "clientProtocolVersion": { + "fieldType": { + "type": "number" + }, + "optional": false + } + } }, "functionType": "Mutation", "identifier": "users.js:ensureCurrentUser", diff --git a/convex/invites.ts b/convex/invites.ts index 618d8865..f908b916 100644 --- a/convex/invites.ts +++ b/convex/invites.ts @@ -8,6 +8,10 @@ import { type CollaboratorRole, } from "./lib/auth"; import { getStrategyByPublicId } from "./lib/entities"; +import { + assertSupportedCloudProtocol, + cloudProtocolArgs, +} from "./lib/cloudProtocol"; import { forbiddenError, invalidOpError, @@ -105,6 +109,7 @@ export const get = query({ export const create = mutation({ args: { + ...cloudProtocolArgs, strategyPublicId: v.string(), token: v.string(), role: v.union(v.literal("editor"), v.literal("viewer")), @@ -112,6 +117,7 @@ export const create = mutation({ }, returns: okResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const strategy = await getStrategyByPublicId(ctx, args.strategyPublicId); const { user, role } = await assertStrategyRole(ctx, strategy, "owner"); if (role !== "owner") { @@ -135,6 +141,7 @@ export const create = mutation({ export const redeem = mutation({ args: { + ...cloudProtocolArgs, token: v.string(), }, returns: v.object({ @@ -143,6 +150,7 @@ export const redeem = mutation({ role: accessRoleValidator, }), handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const user = await requireCurrentUser(ctx); const invite = await ctx.db .query("inviteTokens") @@ -213,11 +221,13 @@ export const redeem = mutation({ export const revoke = mutation({ args: { + ...cloudProtocolArgs, strategyPublicId: v.string(), token: v.string(), }, returns: okResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const strategy = await getStrategyByPublicId(ctx, args.strategyPublicId); const { role } = await assertStrategyRole(ctx, strategy, "owner"); if (role !== "owner") { diff --git a/convex/lib/cloudProtocol.ts b/convex/lib/cloudProtocol.ts index 453ef977..80d8ef24 100644 --- a/convex/lib/cloudProtocol.ts +++ b/convex/lib/cloudProtocol.ts @@ -1,9 +1,42 @@ import { clientUpgradeRequiredError } from "./errors"; +import { v } from "convex/values"; export const CURRENT_CLOUD_PROTOCOL_VERSION = 3; +export const MAX_CLOUD_OPERATION_BYTES = 900 * 1024; +export const MAX_CLOUD_ARRAY_ENTRIES = 8_000; +export const CLOUD_OPERATION_TOO_LARGE_MESSAGE = + "This saved change is too large for cloud sync. It remains saved on this device."; + +export const cloudProtocolArgs = { + clientProtocolVersion: v.number(), +} as const; export function assertSupportedCloudProtocol(clientProtocolVersion: number): void { if (clientProtocolVersion !== CURRENT_CLOUD_PROTOCOL_VERSION) { throw clientUpgradeRequiredError(); } } + +export function serializedConvexValueUtf8Bytes(value: unknown): number { + return new TextEncoder().encode(JSON.stringify(value)).byteLength; +} + +export function cloudOperationExceedsPolicy(value: unknown): boolean { + if (serializedConvexValueUtf8Bytes(value) > MAX_CLOUD_OPERATION_BYTES) { + return true; + } + return valueExceedsArrayPolicy(value); +} + +function valueExceedsArrayPolicy(value: unknown): boolean { + if (Array.isArray(value)) { + return ( + value.length > MAX_CLOUD_ARRAY_ENTRIES || + value.some(valueExceedsArrayPolicy) + ); + } + if (value !== null && typeof value === "object") { + return Object.values(value).some(valueExceedsArrayPolicy); + } + return false; +} diff --git a/convex/ops.ts b/convex/ops.ts index a86470af..c6920806 100644 --- a/convex/ops.ts +++ b/convex/ops.ts @@ -14,7 +14,12 @@ import { strategyOpValidator, type StrategyOp as WireStrategyOp, } from "./lib/opTypes"; -import { assertSupportedCloudProtocol } from "./lib/cloudProtocol"; +import { + assertSupportedCloudProtocol, + CLOUD_OPERATION_TOO_LARGE_MESSAGE, + cloudOperationExceedsPolicy, + cloudProtocolArgs, +} from "./lib/cloudProtocol"; import { valuesEqual } from "./lib/canonicalValues"; import { errorWithCode, invalidPayloadError } from "./lib/errors"; import { purgeDeletedPageOrphansRef } from "./maintenance"; @@ -1305,9 +1310,9 @@ async function applyLineupOp( export const applyBatch = mutation({ args: { + ...cloudProtocolArgs, strategyPublicId: v.string(), clientId: v.string(), - clientProtocolVersion: v.number(), ops: v.array(strategyOpValidator), }, returns: applyBatchResultValidator, @@ -1371,38 +1376,47 @@ export const applyBatch = mutation({ } let result: OperationResult; - try { - if (op.entityType === "strategy") { - const applied = await applyStrategyOp(ctx, strategy, op); - strategy = applied.strategy; - result = applied.result; - } else if (op.entityType === "page") { - const applied = await applyPageOp(ctx, strategy, op); - strategy = applied.strategy; - result = applied.result; - } else if (op.entityType === "pageContent") { - result = await applyPageContentOp(ctx, strategy, op); - } else if (op.entityType === "element") { - result = await applyElementOp(ctx, strategy, op); - } else { - result = await applyLineupOp(ctx, strategy, op); - } - } catch (error) { - if (!(error instanceof ConvexError)) throw error; - const rawCode = - typeof error.data?.code === "string" - ? error.data.code - : "INTERNAL_ERROR"; - const message = - typeof error.data?.message === "string" - ? error.data.message - : error.message; + if (cloudOperationExceedsPolicy(rawOp)) { result = { status: "failed", - code: rawCode, - rawCode, - message, + code: "INVALID_PAYLOAD", + rawCode: "INVALID_PAYLOAD", + message: CLOUD_OPERATION_TOO_LARGE_MESSAGE, }; + } else { + try { + if (op.entityType === "strategy") { + const applied = await applyStrategyOp(ctx, strategy, op); + strategy = applied.strategy; + result = applied.result; + } else if (op.entityType === "page") { + const applied = await applyPageOp(ctx, strategy, op); + strategy = applied.strategy; + result = applied.result; + } else if (op.entityType === "pageContent") { + result = await applyPageContentOp(ctx, strategy, op); + } else if (op.entityType === "element") { + result = await applyElementOp(ctx, strategy, op); + } else { + result = await applyLineupOp(ctx, strategy, op); + } + } catch (error) { + if (!(error instanceof ConvexError)) throw error; + const rawCode = + typeof error.data?.code === "string" + ? error.data.code + : "INTERNAL_ERROR"; + const message = + typeof error.data?.message === "string" + ? error.data.message + : error.message; + result = { + status: "failed", + code: rawCode, + rawCode, + message, + }; + } } if ( diff --git a/convex/pages.ts b/convex/pages.ts index 216c9e73..ea859fce 100644 --- a/convex/pages.ts +++ b/convex/pages.ts @@ -1,6 +1,10 @@ import { mutation, query } from "./_generated/server"; import { v } from "convex/values"; import { assertStrategyRole } from "./lib/auth"; +import { + assertSupportedCloudProtocol, + cloudProtocolArgs, +} from "./lib/cloudProtocol"; import { purgeDeletedPageOrphansRef } from "./maintenance"; import { clampPageIndex, @@ -41,6 +45,7 @@ export const listForStrategy = query({ export const add = mutation({ args: { + ...cloudProtocolArgs, strategyPublicId: v.string(), expectedRevision: v.number(), pagePublicId: v.string(), @@ -52,6 +57,7 @@ export const add = mutation({ }, returns: revisionResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const strategy = await getStrategyByPublicId(ctx, args.strategyPublicId); await assertStrategyRole(ctx, strategy, "editor"); const existingPage = await ctx.db @@ -147,6 +153,7 @@ export const add = mutation({ export const rename = mutation({ args: { + ...cloudProtocolArgs, strategyPublicId: v.string(), pagePublicId: v.string(), name: v.string(), @@ -155,6 +162,7 @@ export const rename = mutation({ }, returns: revisionResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const strategy = await getStrategyByPublicId(ctx, args.strategyPublicId); await assertStrategyRole(ctx, strategy, "editor"); const page = await getPageByPublicId(ctx, args.pagePublicId); @@ -185,12 +193,14 @@ export const rename = mutation({ const deletePage = mutation({ args: { + ...cloudProtocolArgs, strategyPublicId: v.string(), pagePublicId: v.string(), expectedRevision: v.number(), }, returns: revisionResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const strategy = await getStrategyByPublicId(ctx, args.strategyPublicId); await assertStrategyRole(ctx, strategy, "editor"); const pages = await ctx.db @@ -249,12 +259,14 @@ const deletePage = mutation({ export const reorder = mutation({ args: { + ...cloudProtocolArgs, strategyPublicId: v.string(), orderedPagePublicIds: v.array(v.string()), expectedRevision: v.number(), }, returns: revisionResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const strategy = await getStrategyByPublicId(ctx, args.strategyPublicId); await assertStrategyRole(ctx, strategy, "editor"); const pages = await ctx.db diff --git a/convex/shares.ts b/convex/shares.ts index 77ecde60..85945c65 100644 --- a/convex/shares.ts +++ b/convex/shares.ts @@ -9,6 +9,10 @@ import { type CollaboratorRole, } from "./lib/auth"; import { getFolderByPublicId, getStrategyByPublicId } from "./lib/entities"; +import { + assertSupportedCloudProtocol, + cloudProtocolArgs, +} from "./lib/cloudProtocol"; import { notFoundError, errorWithCode, @@ -83,6 +87,7 @@ export const list = query({ export const create = mutation({ args: { + ...cloudProtocolArgs, targetType: targetTypeValidator, targetPublicId: v.string(), token: v.string(), @@ -90,6 +95,7 @@ export const create = mutation({ }, returns: okResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const user = await requireCurrentUser(ctx); const resolved = await resolveTarget(ctx, args.targetType, args.targetPublicId); @@ -124,12 +130,14 @@ export const create = mutation({ export const revoke = mutation({ args: { + ...cloudProtocolArgs, targetType: targetTypeValidator, targetPublicId: v.string(), token: v.string(), }, returns: okResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const resolved = await resolveTarget(ctx, args.targetType, args.targetPublicId); if (resolved.strategy !== null) { @@ -165,6 +173,7 @@ export const revoke = mutation({ export const redeem = mutation({ args: { + ...cloudProtocolArgs, token: v.string(), }, returns: v.union( @@ -183,6 +192,7 @@ export const redeem = mutation({ }), ), handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const user = await requireCurrentUser(ctx); const link = await ctx.db .query("shareLinks") diff --git a/convex/strategies.ts b/convex/strategies.ts index a0031190..4513a3ca 100644 --- a/convex/strategies.ts +++ b/convex/strategies.ts @@ -12,6 +12,10 @@ import { } from "./lib/auth"; import type { StrategyRole } from "./lib/auth"; import { getFolderByPublicId, getStrategyByPublicId } from "./lib/entities"; +import { + assertSupportedCloudProtocol, + cloudProtocolArgs, +} from "./lib/cloudProtocol"; import { mapThemePaletteValidator, strategySettingsValidator, @@ -454,6 +458,7 @@ export const getHeader = query({ export const create = mutation({ args: { + ...cloudProtocolArgs, publicId: v.string(), name: v.string(), mapData: v.string(), @@ -463,6 +468,7 @@ export const create = mutation({ }, returns: createResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const user = await requireCurrentUser(ctx); return await createStrategyWithInitialPageRecord(ctx, args, user._id, { publicId: createPublicId(), @@ -475,6 +481,7 @@ export const create = mutation({ export const createWithInitialPage = mutation({ args: { + ...cloudProtocolArgs, publicId: v.string(), name: v.string(), mapData: v.string(), @@ -489,6 +496,7 @@ export const createWithInitialPage = mutation({ }, returns: createResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const user = await requireCurrentUser(ctx); return await createStrategyWithInitialPageRecord(ctx, args, user._id, { publicId: args.initialPagePublicId, @@ -502,6 +510,7 @@ export const createWithInitialPage = mutation({ export const update = mutation({ args: { + ...cloudProtocolArgs, strategyPublicId: v.string(), expectedRevision: v.number(), name: v.optional(v.string()), @@ -513,6 +522,7 @@ export const update = mutation({ }, returns: revisionResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const strategy = await getStrategyByPublicId(ctx, args.strategyPublicId); await assertStrategyRole(ctx, strategy, "editor"); @@ -563,12 +573,14 @@ export const update = mutation({ export const move = mutation({ args: { + ...cloudProtocolArgs, strategyPublicId: v.string(), expectedRevision: v.number(), folderPublicId: v.optional(v.string()), }, returns: revisionResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const strategy = await getStrategyByPublicId(ctx, args.strategyPublicId); await assertStrategyRole(ctx, strategy, "editor"); @@ -598,11 +610,13 @@ export const move = mutation({ const deleteStrategy = mutation({ args: { + ...cloudProtocolArgs, strategyPublicId: v.string(), expectedRevision: v.number(), }, returns: okResultValidator, handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const strategy = await getStrategyByPublicId(ctx, args.strategyPublicId); await assertStrategyRole(ctx, strategy, "owner"); if (args.expectedRevision !== strategy.revision) { diff --git a/convex/syncBoundaries.test.ts b/convex/syncBoundaries.test.ts index 6482d1eb..628f94bd 100644 --- a/convex/syncBoundaries.test.ts +++ b/convex/syncBoundaries.test.ts @@ -6,6 +6,13 @@ import { import { makeFunctionReference } from "convex/server"; import { beforeAll, describe, expect, test, vi } from "vitest"; import type { DataModel } from "./_generated/dataModel"; +import { + CLOUD_OPERATION_TOO_LARGE_MESSAGE, + CURRENT_CLOUD_PROTOCOL_VERSION, + MAX_CLOUD_ARRAY_ENTRIES, + MAX_CLOUD_OPERATION_BYTES, + serializedConvexValueUtf8Bytes, +} from "./lib/cloudProtocol"; import schema from "./schema"; import { modules } from "./test.setup"; @@ -81,12 +88,15 @@ async function createHarness(): Promise<{ }> { const t = convexTest(schema, modules); const owner = t.withIdentity(identity); - await owner.mutation(ensureCurrentUser, {}); + await owner.mutation(ensureCurrentUser, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + }); return { t, owner }; } async function createBaseStrategy(owner: Harness) { await owner.mutation(createStrategyWithInitialPage, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, publicId: strategyPublicId, name: "Sync boundary strategy", mapData: "ascent", @@ -105,7 +115,7 @@ async function applyOps( return (await owner.mutation(applyBatch, { strategyPublicId, clientId, - clientProtocolVersion: 3, + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, ops: ops.map(toProtocol3Op), })) as { strategyPublicId: string; @@ -436,6 +446,7 @@ describe("page-scoped read contract", () => { try { const { t, owner } = await createHarness(); await owner.mutation(createStrategyWithInitialPage, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, publicId: strategyPublicId, name: "Page names", mapData: "ascent", @@ -1195,6 +1206,7 @@ describe("record-scoped write contract", () => { const { t, owner } = await createHarness(); await createBaseStrategy(owner); const added = (await owner.mutation(addPage, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 0, pagePublicId: pageB, @@ -1206,6 +1218,7 @@ describe("record-scoped write contract", () => { expect(added.revision).toBe(1); const deleted = (await owner.mutation(deletePage, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, pagePublicId: pageB, expectedRevision: 1, @@ -1213,6 +1226,7 @@ describe("record-scoped write contract", () => { expect(deleted).toMatchObject({ revision: 2 }); const replayed = (await owner.mutation(deletePage, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, pagePublicId: pageB, expectedRevision: 1, @@ -1228,6 +1242,7 @@ describe("record-scoped write contract", () => { const { owner } = await createHarness(); await createBaseStrategy(owner); await owner.mutation(addPage, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 0, pagePublicId: pageB, @@ -1238,6 +1253,7 @@ describe("record-scoped write contract", () => { }); const replayed = (await owner.mutation(addPage, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 0, pagePublicId: pageB, @@ -1258,6 +1274,7 @@ describe("record-scoped write contract", () => { await createBaseStrategy(owner); await owner.mutation(addPage, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 0, pagePublicId: pageB, @@ -1267,6 +1284,7 @@ describe("record-scoped write contract", () => { settings: settingsB, }); await owner.mutation(addPage, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 1, pagePublicId: "page-c", @@ -1287,6 +1305,7 @@ describe("record-scoped write contract", () => { ]); const replayed = (await owner.mutation(addPage, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 1, pagePublicId: "page-c", @@ -1301,6 +1320,7 @@ describe("record-scoped write contract", () => { const { owner } = await createHarness(); await createBaseStrategy(owner); await owner.mutation(addPage, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, expectedRevision: 0, pagePublicId: pageB, @@ -1312,6 +1332,7 @@ describe("record-scoped write contract", () => { await expect( owner.mutation(reorderPages, { + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, strategyPublicId, orderedPagePublicIds: [pageA, pageA], expectedRevision: 1, @@ -1524,6 +1545,114 @@ describe("cloud protocol v3 boundary", () => { }); }); + test("a policy-oversized Unicode op fails without blocking its sibling", async () => { + const { owner } = await createHarness(); + await createBaseStrategy(owner); + const oversized = { + opId: "oversized-drawing", + type: "element.add" as const, + elementPublicId: "large-element", + pagePublicId: pageA, + payload: { + kind: "drawing" as const, + payloadVersion: 1, + data: { + elementType: "drawing", + encodedPoints: "界".repeat(310_000), + }, + }, + sortIndex: 0, + }; + const serializedBytes = serializedConvexValueUtf8Bytes(oversized); + expect(serializedBytes).toBeGreaterThan(MAX_CLOUD_OPERATION_BYTES); + expect(serializedBytes).toBeLessThan(1024 * 1024); + + const response = (await owner.mutation(applyBatch, { + strategyPublicId, + clientId: "oversized-op-client", + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + ops: [ + oversized, + { + opId: "independent-strategy-patch", + type: "strategy.patch", + payload: { name: "Independent change landed" }, + expectedStrategyRevision: 0, + }, + ], + })) as { results: Array> }; + + expect(response.results).toEqual([ + { + opId: "oversized-drawing", + status: "failed", + code: "INVALID_PAYLOAD", + rawCode: "INVALID_PAYLOAD", + message: CLOUD_OPERATION_TOO_LARGE_MESSAGE, + }, + { + opId: "independent-strategy-patch", + status: "applied", + appliedRevision: 1, + }, + ]); + await expect( + owner.query(getShell, { strategyPublicId }), + ).resolves.toMatchObject({ + header: { name: "Independent change landed", revision: 1 }, + }); + }); + + test("a policy-wide array fails per op without blocking its sibling", async () => { + const { owner } = await createHarness(); + await createBaseStrategy(owner); + const response = (await owner.mutation(applyBatch, { + strategyPublicId, + clientId: "wide-array-client", + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, + ops: [ + { + opId: "wide-drawing", + type: "element.add", + elementPublicId: "wide-element", + pagePublicId: pageA, + payload: { + kind: "drawing", + payloadVersion: 1, + data: { + points: Array.from( + { length: MAX_CLOUD_ARRAY_ENTRIES + 1 }, + () => 0, + ), + }, + }, + sortIndex: 0, + }, + { + opId: "independent-wide-array-sibling", + type: "strategy.patch", + payload: { name: "Wide sibling landed" }, + expectedStrategyRevision: 0, + }, + ], + })) as { results: Array> }; + + expect(response.results).toEqual([ + { + opId: "wide-drawing", + status: "failed", + code: "INVALID_PAYLOAD", + rawCode: "INVALID_PAYLOAD", + message: CLOUD_OPERATION_TOO_LARGE_MESSAGE, + }, + { + opId: "independent-wide-array-sibling", + status: "applied", + appliedRevision: 1, + }, + ]); + }); + test("the wire validator rejects an illegal operation discriminator", async () => { const { owner } = await createHarness(); @@ -1531,7 +1660,7 @@ describe("cloud protocol v3 boundary", () => { owner.mutation(applyBatch, { strategyPublicId, clientId: "illegal-op", - clientProtocolVersion: 3, + clientProtocolVersion: CURRENT_CLOUD_PROTOCOL_VERSION, ops: [ { opId: "illegal-page-delete", diff --git a/convex/users.ts b/convex/users.ts index 0daeda28..19000287 100644 --- a/convex/users.ts +++ b/convex/users.ts @@ -4,13 +4,18 @@ import { getCanonicalExternalId, } from "./lib/auth"; import { unauthenticatedError } from "./lib/errors"; +import { + assertSupportedCloudProtocol, + cloudProtocolArgs, +} from "./lib/cloudProtocol"; import { okResultValidator } from "./lib/publicValidators"; import { v } from "convex/values"; export const ensureCurrentUser = mutation({ - args: {}, + args: cloudProtocolArgs, returns: okResultValidator, - handler: async (ctx) => { + handler: async (ctx, args) => { + assertSupportedCloudProtocol(args.clientProtocolVersion); const identity = await ctx.auth.getUserIdentity(); if (identity === null) { throw unauthenticatedError(); diff --git a/lib/collab/cloud_sync_error_message.dart b/lib/collab/cloud_sync_error_message.dart index 95a5224e..184d7507 100644 --- a/lib/collab/cloud_sync_error_message.dart +++ b/lib/collab/cloud_sync_error_message.dart @@ -5,6 +5,10 @@ String friendlyCloudSyncError(String raw) { 'device; keep this strategy open and recover the outbox before ' 'continuing.'; } + if (lower.contains('could not be verified in the durable outbox')) { + return 'Icarus could not verify that this change was saved on this ' + 'device. Nothing was sent. Keep this strategy open and retry.'; + } if (lower.contains('forbidden')) { return 'This account does not have permission to save these changes. ' 'They remain on this device. Ask the owner for edit access, then ' @@ -14,6 +18,10 @@ String friendlyCloudSyncError(String raw) { return 'A saved cloud change is paused after repeated failures. Retry ' 'when the connection and account are healthy.'; } + if (lower.contains('too large for cloud sync')) { + return 'A saved change is too large for cloud sync. It remains saved on ' + 'this device. Reduce it, then choose Keep mine to retry.'; + } if (lower.contains('needs attention')) { return 'Another edit reached the cloud first. Your version remains ' 'saved on this device.'; diff --git a/lib/collab/collab_models.dart b/lib/collab/collab_models.dart index 8212d186..1a87ea2a 100644 --- a/lib/collab/collab_models.dart +++ b/lib/collab/collab_models.dart @@ -2,9 +2,18 @@ import 'dart:convert'; const currentCloudProtocolVersion = 3; const currentCloudPayloadVersion = 1; +const maxCloudOperationBytes = 900 * 1024; +const maxCloudBatchBytes = 15 * 1024 * 1024; +const maxCloudArrayEntries = 8000; +const cloudOperationTooLargeMessage = + 'This saved change is too large for cloud sync. It remains saved on this ' + 'device.'; typedef CloudPayload = Map; +int serializedConvexValueUtf8Bytes(Object? value) => + utf8.encode(jsonEncode(value)).length; + CloudPayload cloudElementPayload({ required String kind, required Map data, @@ -930,6 +939,39 @@ int _requiredInt(Object? value) { throw const FormatException('Op revision or index must be a number'); } +int serializedCloudOperationUtf8Bytes(StrategyOp op) => + serializedConvexValueUtf8Bytes(op.toConvexJson()); + +bool cloudOperationExceedsPolicy(StrategyOp op) { + final value = op.toConvexJson(); + return serializedConvexValueUtf8Bytes(value) > maxCloudOperationBytes || + _cloudValueExceedsArrayPolicy(value); +} + +bool _cloudValueExceedsArrayPolicy(Object? value) { + if (value is List) { + return value.length > maxCloudArrayEntries || + value.any(_cloudValueExceedsArrayPolicy); + } + if (value is Map) { + return value.values.any(_cloudValueExceedsArrayPolicy); + } + return false; +} + +int serializedCloudBatchUtf8Bytes({ + required String strategyPublicId, + required String clientId, + required Iterable ops, +}) { + return serializedConvexValueUtf8Bytes({ + 'strategyPublicId': strategyPublicId, + 'clientId': clientId, + 'clientProtocolVersion': currentCloudProtocolVersion, + 'ops': ops.map((op) => op.toConvexJson()).toList(growable: false), + }); +} + class PendingOp { const PendingOp({ required this.op, diff --git a/lib/collab/convex_strategy_repository.dart b/lib/collab/convex_strategy_repository.dart index 833d0533..32d51e0c 100644 --- a/lib/collab/convex_strategy_repository.dart +++ b/lib/collab/convex_strategy_repository.dart @@ -27,7 +27,9 @@ class ConvexStrategyRepository { final IcarusConvexApi _api; Future ensureCurrentUser() async { - await _api.users.ensureCurrentUser(); + await _api.users.ensureCurrentUser( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), + ); } Stream> watchAllFolders() { @@ -194,6 +196,19 @@ class ConvexStrategyRepository { required List ops, }) async { if (ops.isEmpty) return const []; + for (final op in ops) { + if (cloudOperationExceedsPolicy(op)) { + throw StateError(cloudOperationTooLargeMessage); + } + } + if (serializedCloudBatchUtf8Bytes( + strategyPublicId: strategyPublicId, + clientId: clientId, + ops: ops, + ) > + maxCloudBatchBytes) { + throw StateError('Cloud operation batch exceeds the payload policy.'); + } final typedOps = ops.indexed .map( @@ -224,6 +239,7 @@ class ConvexStrategyRepository { int? customColorValue, }) async { await _api.folders.create( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), publicId: publicId, name: name, parentFolderPublicId: _optional(parentFolderPublicId), @@ -250,6 +266,7 @@ class ConvexStrategyRepository { bool clearCustomColorValue = false, }) async { await _api.folders.update( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), folderPublicId: folderPublicId, name: _optional(name), iconId: _optionalNumber(iconId), @@ -265,7 +282,10 @@ class ConvexStrategyRepository { } Future deleteFolder(String folderPublicId) async { - await _api.folders.delete(folderPublicId: folderPublicId); + await _api.folders.delete( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), + folderPublicId: folderPublicId, + ); } Future moveFolder({ @@ -273,6 +293,7 @@ class ConvexStrategyRepository { String? parentFolderPublicId, }) async { await _api.folders.move( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), folderPublicId: folderPublicId, parentFolderPublicId: _optional(parentFolderPublicId), ); @@ -287,6 +308,7 @@ class ConvexStrategyRepository { Map? themeOverridePalette, }) async { await _api.strategies.create( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), publicId: publicId, name: name, mapData: mapData, @@ -310,6 +332,7 @@ class ConvexStrategyRepository { Map? initialPageSettings, }) async { await _api.strategies.createWithInitialPage( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), publicId: publicId, name: name, mapData: mapData, @@ -330,6 +353,7 @@ class ConvexStrategyRepository { required int expectedRevision, }) async { await _api.strategies.update( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), strategyPublicId: strategyPublicId, name: ConvexOptional.present(name), expectedRevision: expectedRevision.toDouble(), @@ -341,6 +365,7 @@ class ConvexStrategyRepository { required int expectedRevision, }) async { await _api.strategies.delete( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), strategyPublicId: strategyPublicId, expectedRevision: expectedRevision.toDouble(), ); @@ -352,6 +377,7 @@ class ConvexStrategyRepository { required int expectedRevision, }) async { await _api.strategies.move( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), strategyPublicId: strategyPublicId, folderPublicId: _optional(folderPublicId), expectedRevision: expectedRevision.toDouble(), @@ -369,6 +395,7 @@ class ConvexStrategyRepository { Map? settings, }) async { await _api.pages.add( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), strategyPublicId: strategyPublicId, pagePublicId: pagePublicId, name: name, @@ -410,6 +437,7 @@ class ConvexStrategyRepository { required String role, }) async { await _api.shares.create( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), targetType: _shareTargetType(targetType), targetPublicId: targetPublicId, token: token, @@ -423,6 +451,7 @@ class ConvexStrategyRepository { required String token, }) async { await _api.shares.revoke( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), targetType: _shareTargetType(targetType), targetPublicId: targetPublicId, token: token, @@ -430,7 +459,10 @@ class ConvexStrategyRepository { } Future redeemShareLink(String token) async { - final result = await _api.shares.redeem(token: token); + final result = await _api.shares.redeem( + clientProtocolVersion: currentCloudProtocolVersion.toDouble(), + token: token, + ); return switch (result) { SharesRedeemResultFolder(:final folderPublicId, :final role) => ShareRedemption( diff --git a/lib/collab/generated/convex_models.dart b/lib/collab/generated/convex_models.dart index edec1e2d..700fac6d 100644 --- a/lib/collab/generated/convex_models.dart +++ b/lib/collab/generated/convex_models.dart @@ -4549,6 +4549,7 @@ List decodeElementsListForStrategyResult( .toList(growable: false); ConvexObject encodeFoldersCreateArgs({ + required double clientProtocolVersion, ConvexOptional color = const ConvexOptional.absent(), ConvexOptional customColorValue = const ConvexOptional.absent(), ConvexOptional iconCodePoint = const ConvexOptional.absent(), @@ -4559,6 +4560,10 @@ ConvexObject encodeFoldersCreateArgs({ ConvexOptional parentFolderPublicId = const ConvexOptional.absent(), required String publicId, }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'folders.js:create.args.clientProtocolVersion', + ), if (color.isPresent) 'color': ConvexString(color.value), if (customColorValue.isPresent) 'customColorValue': _encodeNumber( @@ -4588,8 +4593,16 @@ ConvexValue decodeFoldersCreateResult(ConvexValue value) => _decodeRaw( _validateFoldersCreateResult, ); -ConvexObject encodeFoldersDeleteArgs({required String folderPublicId}) => - ConvexObject({'folderPublicId': ConvexString(folderPublicId)}); +ConvexObject encodeFoldersDeleteArgs({ + required double clientProtocolVersion, + required String folderPublicId, +}) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'folders.js:delete.args.clientProtocolVersion', + ), + 'folderPublicId': ConvexString(folderPublicId), +}); FoldersDeleteResult decodeFoldersDeleteResult(ConvexValue value) => FoldersDeleteResult.decode(value, 'folders.js:delete.returns'); @@ -4613,9 +4626,14 @@ List decodeFoldersListTreeResult( .toList(growable: false); ConvexObject encodeFoldersMoveArgs({ + required double clientProtocolVersion, required String folderPublicId, ConvexOptional parentFolderPublicId = const ConvexOptional.absent(), }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'folders.js:move.args.clientProtocolVersion', + ), 'folderPublicId': ConvexString(folderPublicId), if (parentFolderPublicId.isPresent) 'parentFolderPublicId': ConvexString(parentFolderPublicId.value), @@ -4628,6 +4646,7 @@ ConvexObject encodeFoldersUpdateArgs({ ConvexOptional clearCustomColorValue = const ConvexOptional.absent(), ConvexOptional clearIconFontFamily = const ConvexOptional.absent(), ConvexOptional clearIconFontPackage = const ConvexOptional.absent(), + required double clientProtocolVersion, ConvexOptional color = const ConvexOptional.absent(), ConvexOptional customColorValue = const ConvexOptional.absent(), required String folderPublicId, @@ -4643,6 +4662,10 @@ ConvexObject encodeFoldersUpdateArgs({ 'clearIconFontFamily': ConvexBoolean(clearIconFontFamily.value), if (clearIconFontPackage.isPresent) 'clearIconFontPackage': ConvexBoolean(clearIconFontPackage.value), + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'folders.js:update.args.clientProtocolVersion', + ), if (color.isPresent) 'color': ConvexString(color.value), if (customColorValue.isPresent) 'customColorValue': _encodeNumber( @@ -4796,11 +4819,16 @@ List decodeImagesListForStrategyResult( .toList(growable: false); ConvexObject encodeInvitesCreateArgs({ + required double clientProtocolVersion, ConvexOptional expiresAt = const ConvexOptional.absent(), required InvitesCreateArgsRole role, required String strategyPublicId, required String token, }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'invites.js:create.args.clientProtocolVersion', + ), if (expiresAt.isPresent) 'expiresAt': _encodeNumber( expiresAt.value, @@ -4826,16 +4854,29 @@ ConvexObject encodeInvitesGetArgs({ ConvexValue decodeInvitesGetResult(ConvexValue value) => _decodeRaw(value, 'invites.js:get.returns', _validateInvitesGetResult); -ConvexObject encodeInvitesRedeemArgs({required String token}) => - ConvexObject({'token': ConvexString(token)}); +ConvexObject encodeInvitesRedeemArgs({ + required double clientProtocolVersion, + required String token, +}) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'invites.js:redeem.args.clientProtocolVersion', + ), + 'token': ConvexString(token), +}); InvitesRedeemResult decodeInvitesRedeemResult(ConvexValue value) => InvitesRedeemResult.decode(value, 'invites.js:redeem.returns'); ConvexObject encodeInvitesRevokeArgs({ + required double clientProtocolVersion, required String strategyPublicId, required String token, }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'invites.js:revoke.args.clientProtocolVersion', + ), 'strategyPublicId': ConvexString(strategyPublicId), 'token': ConvexString(token), }); @@ -4915,6 +4956,7 @@ PageGetSnapshotResult decodePageGetSnapshotResult(ConvexValue value) => PageGetSnapshotResult.decode(value, 'page.js:getSnapshot.returns'); ConvexObject encodePagesAddArgs({ + required double clientProtocolVersion, required double expectedRevision, required bool isAttack, ConvexOptional isAutoNamed = const ConvexOptional.absent(), @@ -4925,6 +4967,10 @@ ConvexObject encodePagesAddArgs({ required double sortIndex, required String strategyPublicId, }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'pages.js:add.args.clientProtocolVersion', + ), 'expectedRevision': _encodeNumber( expectedRevision, 'pages.js:add.args.expectedRevision', @@ -4943,10 +4989,15 @@ ConvexValue decodePagesAddResult(ConvexValue value) => _decodeRaw(value, 'pages.js:add.returns', _validatePagesAddResult); ConvexObject encodePagesDeleteArgs({ + required double clientProtocolVersion, required double expectedRevision, required String pagePublicId, required String strategyPublicId, }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'pages.js:delete.args.clientProtocolVersion', + ), 'expectedRevision': _encodeNumber( expectedRevision, 'pages.js:delete.args.expectedRevision', @@ -4974,12 +5025,17 @@ List decodePagesListForStrategyResult( .toList(growable: false); ConvexObject encodePagesRenameArgs({ + required double clientProtocolVersion, required double expectedRevision, ConvexOptional isAutoNamed = const ConvexOptional.absent(), required String name, required String pagePublicId, required String strategyPublicId, }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'pages.js:rename.args.clientProtocolVersion', + ), 'expectedRevision': _encodeNumber( expectedRevision, 'pages.js:rename.args.expectedRevision', @@ -4994,10 +5050,15 @@ ConvexValue decodePagesRenameResult(ConvexValue value) => _decodeRaw(value, 'pages.js:rename.returns', _validatePagesAddResult); ConvexObject encodePagesReorderArgs({ + required double clientProtocolVersion, required double expectedRevision, required List orderedPagePublicIds, required String strategyPublicId, }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'pages.js:reorder.args.clientProtocolVersion', + ), 'expectedRevision': _encodeNumber( expectedRevision, 'pages.js:reorder.args.expectedRevision', @@ -5014,11 +5075,16 @@ ConvexValue decodePagesReorderResult(ConvexValue value) => _decodeRaw(value, 'pages.js:reorder.returns', _validatePagesAddResult); ConvexObject encodeSharesCreateArgs({ + required double clientProtocolVersion, required InvitesCreateArgsRole role, required String targetPublicId, required SharesCreateArgsTargetType targetType, required String token, }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'shares.js:create.args.clientProtocolVersion', + ), 'role': ConvexString(role.wireName), 'targetPublicId': ConvexString(targetPublicId), 'targetType': ConvexString(targetType.wireName), @@ -5046,17 +5112,30 @@ List decodeSharesListResult(ConvexValue value) => ) .toList(growable: false); -ConvexObject encodeSharesRedeemArgs({required String token}) => - ConvexObject({'token': ConvexString(token)}); +ConvexObject encodeSharesRedeemArgs({ + required double clientProtocolVersion, + required String token, +}) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'shares.js:redeem.args.clientProtocolVersion', + ), + 'token': ConvexString(token), +}); SharesRedeemResult decodeSharesRedeemResult(ConvexValue value) => SharesRedeemResult.decode(value, 'shares.js:redeem.returns'); ConvexObject encodeSharesRevokeArgs({ + required double clientProtocolVersion, required String targetPublicId, required SharesCreateArgsTargetType targetType, required String token, }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'shares.js:revoke.args.clientProtocolVersion', + ), 'targetPublicId': ConvexString(targetPublicId), 'targetType': ConvexString(targetType.wireName), 'token': ConvexString(token), @@ -5066,6 +5145,7 @@ FoldersDeleteResult decodeSharesRevokeResult(ConvexValue value) => FoldersDeleteResult.decode(value, 'shares.js:revoke.returns'); ConvexObject encodeStrategiesCreateArgs({ + required double clientProtocolVersion, ConvexOptional folderPublicId = const ConvexOptional.absent(), required String mapData, required String name, @@ -5077,6 +5157,10 @@ ConvexObject encodeStrategiesCreateArgs({ const ConvexOptional.absent(), ConvexOptional themeProfileId = const ConvexOptional.absent(), }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'strategies.js:create.args.clientProtocolVersion', + ), if (folderPublicId.isPresent) 'folderPublicId': ConvexString(folderPublicId.value), 'mapData': ConvexString(mapData), @@ -5097,6 +5181,7 @@ ConvexValue decodeStrategiesCreateResult(ConvexValue value) => _decodeRaw( ); ConvexObject encodeStrategiesCreateWithInitialPageArgs({ + required double clientProtocolVersion, ConvexOptional folderPublicId = const ConvexOptional.absent(), required bool initialPageIsAttack, ConvexOptional initialPageIsAutoNamed = const ConvexOptional.absent(), @@ -5115,6 +5200,10 @@ ConvexObject encodeStrategiesCreateWithInitialPageArgs({ const ConvexOptional.absent(), ConvexOptional themeProfileId = const ConvexOptional.absent(), }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'strategies.js:createWithInitialPage.args.clientProtocolVersion', + ), if (folderPublicId.isPresent) 'folderPublicId': ConvexString(folderPublicId.value), 'initialPageIsAttack': ConvexBoolean(initialPageIsAttack), @@ -5145,9 +5234,14 @@ ConvexValue decodeStrategiesCreateWithInitialPageResult(ConvexValue value) => ); ConvexObject encodeStrategiesDeleteArgs({ + required double clientProtocolVersion, required double expectedRevision, required String strategyPublicId, }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'strategies.js:delete.args.clientProtocolVersion', + ), 'expectedRevision': _encodeNumber( expectedRevision, 'strategies.js:delete.args.expectedRevision', @@ -5200,10 +5294,15 @@ List decodeStrategiesListSharedWithMeResult( .toList(growable: false); ConvexObject encodeStrategiesMoveArgs({ + required double clientProtocolVersion, required double expectedRevision, ConvexOptional folderPublicId = const ConvexOptional.absent(), required String strategyPublicId, }) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'strategies.js:move.args.clientProtocolVersion', + ), 'expectedRevision': _encodeNumber( expectedRevision, 'strategies.js:move.args.expectedRevision', @@ -5220,6 +5319,7 @@ ConvexObject encodeStrategiesUpdateArgs({ ConvexOptional clearThemeOverridePalette = const ConvexOptional.absent(), ConvexOptional clearThemeProfileId = const ConvexOptional.absent(), + required double clientProtocolVersion, required double expectedRevision, ConvexOptional mapData = const ConvexOptional.absent(), ConvexOptional name = const ConvexOptional.absent(), @@ -5235,6 +5335,10 @@ ConvexObject encodeStrategiesUpdateArgs({ 'clearThemeOverridePalette': ConvexBoolean(clearThemeOverridePalette.value), if (clearThemeProfileId.isPresent) 'clearThemeProfileId': ConvexBoolean(clearThemeProfileId.value), + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'strategies.js:update.args.clientProtocolVersion', + ), 'expectedRevision': _encodeNumber( expectedRevision, 'strategies.js:update.args.expectedRevision', @@ -5270,7 +5374,14 @@ ConvexObject encodeStrategyGetShellArgs({required String strategyPublicId}) => StrategyGetShellResult decodeStrategyGetShellResult(ConvexValue value) => StrategyGetShellResult.decode(value, 'strategy.js:getShell.returns'); -ConvexObject encodeUsersEnsureCurrentUserArgs() => ConvexObject({}); +ConvexObject encodeUsersEnsureCurrentUserArgs({ + required double clientProtocolVersion, +}) => ConvexObject({ + 'clientProtocolVersion': _encodeNumber( + clientProtocolVersion, + 'users.js:ensureCurrentUser.args.clientProtocolVersion', + ), +}); FoldersDeleteResult decodeUsersEnsureCurrentUserResult(ConvexValue value) => FoldersDeleteResult.decode(value, 'users.js:ensureCurrentUser.returns'); diff --git a/lib/collab/generated/icarus_convex_api.dart b/lib/collab/generated/icarus_convex_api.dart index 05456b11..0b175501 100644 --- a/lib/collab/generated/icarus_convex_api.dart +++ b/lib/collab/generated/icarus_convex_api.dart @@ -198,6 +198,7 @@ final class _ElementsModule implements ElementsModule { abstract interface class FoldersModule { Future create({ + required double clientProtocolVersion, ConvexOptional color = const ConvexOptional.absent(), ConvexOptional customColorValue = const ConvexOptional.absent(), ConvexOptional iconCodePoint = const ConvexOptional.absent(), @@ -208,12 +209,16 @@ abstract interface class FoldersModule { ConvexOptional parentFolderPublicId = const ConvexOptional.absent(), required String publicId, }); - Future delete({required String folderPublicId}); + Future delete({ + required double clientProtocolVersion, + required String folderPublicId, + }); ConvexQuery> listTree({ ConvexOptional scope = const ConvexOptional.absent(), }); Future move({ + required double clientProtocolVersion, required String folderPublicId, ConvexOptional parentFolderPublicId = const ConvexOptional.absent(), }); @@ -221,6 +226,7 @@ abstract interface class FoldersModule { ConvexOptional clearCustomColorValue = const ConvexOptional.absent(), ConvexOptional clearIconFontFamily = const ConvexOptional.absent(), ConvexOptional clearIconFontPackage = const ConvexOptional.absent(), + required double clientProtocolVersion, ConvexOptional color = const ConvexOptional.absent(), ConvexOptional customColorValue = const ConvexOptional.absent(), required String folderPublicId, @@ -237,6 +243,7 @@ final class _FoldersModule implements FoldersModule { final ConvexTransport _transport; @override Future create({ + required double clientProtocolVersion, ConvexOptional color = const ConvexOptional.absent(), ConvexOptional customColorValue = const ConvexOptional.absent(), ConvexOptional iconCodePoint = const ConvexOptional.absent(), @@ -248,6 +255,7 @@ final class _FoldersModule implements FoldersModule { required String publicId, }) { final args = encodeFoldersCreateArgs( + clientProtocolVersion: clientProtocolVersion, color: color, customColorValue: customColorValue, iconCodePoint: iconCodePoint, @@ -265,8 +273,14 @@ final class _FoldersModule implements FoldersModule { } @override - Future delete({required String folderPublicId}) { - final args = encodeFoldersDeleteArgs(folderPublicId: folderPublicId); + Future delete({ + required double clientProtocolVersion, + required String folderPublicId, + }) { + final args = encodeFoldersDeleteArgs( + clientProtocolVersion: clientProtocolVersion, + folderPublicId: folderPublicId, + ); return _invoke( () => _transport.mutation('folders:delete', args), decodeFoldersDeleteResult, @@ -289,10 +303,12 @@ final class _FoldersModule implements FoldersModule { @override Future move({ + required double clientProtocolVersion, required String folderPublicId, ConvexOptional parentFolderPublicId = const ConvexOptional.absent(), }) { final args = encodeFoldersMoveArgs( + clientProtocolVersion: clientProtocolVersion, folderPublicId: folderPublicId, parentFolderPublicId: parentFolderPublicId, ); @@ -307,6 +323,7 @@ final class _FoldersModule implements FoldersModule { ConvexOptional clearCustomColorValue = const ConvexOptional.absent(), ConvexOptional clearIconFontFamily = const ConvexOptional.absent(), ConvexOptional clearIconFontPackage = const ConvexOptional.absent(), + required double clientProtocolVersion, ConvexOptional color = const ConvexOptional.absent(), ConvexOptional customColorValue = const ConvexOptional.absent(), required String folderPublicId, @@ -320,6 +337,7 @@ final class _FoldersModule implements FoldersModule { clearCustomColorValue: clearCustomColorValue, clearIconFontFamily: clearIconFontFamily, clearIconFontPackage: clearIconFontPackage, + clientProtocolVersion: clientProtocolVersion, color: color, customColorValue: customColorValue, folderPublicId: folderPublicId, @@ -507,6 +525,7 @@ final class _ImagesModule implements ImagesModule { abstract interface class InvitesModule { Future create({ + required double clientProtocolVersion, ConvexOptional expiresAt = const ConvexOptional.absent(), required InvitesCreateArgsRole role, required String strategyPublicId, @@ -516,8 +535,12 @@ abstract interface class InvitesModule { ConvexOptional strategyPublicId = const ConvexOptional.absent(), ConvexOptional token = const ConvexOptional.absent(), }); - Future redeem({required String token}); + Future redeem({ + required double clientProtocolVersion, + required String token, + }); Future revoke({ + required double clientProtocolVersion, required String strategyPublicId, required String token, }); @@ -528,12 +551,14 @@ final class _InvitesModule implements InvitesModule { final ConvexTransport _transport; @override Future create({ + required double clientProtocolVersion, ConvexOptional expiresAt = const ConvexOptional.absent(), required InvitesCreateArgsRole role, required String strategyPublicId, required String token, }) { final args = encodeInvitesCreateArgs( + clientProtocolVersion: clientProtocolVersion, expiresAt: expiresAt, role: role, strategyPublicId: strategyPublicId, @@ -563,8 +588,14 @@ final class _InvitesModule implements InvitesModule { } @override - Future redeem({required String token}) { - final args = encodeInvitesRedeemArgs(token: token); + Future redeem({ + required double clientProtocolVersion, + required String token, + }) { + final args = encodeInvitesRedeemArgs( + clientProtocolVersion: clientProtocolVersion, + token: token, + ); return _invoke( () => _transport.mutation('invites:redeem', args), decodeInvitesRedeemResult, @@ -573,10 +604,12 @@ final class _InvitesModule implements InvitesModule { @override Future revoke({ + required double clientProtocolVersion, required String strategyPublicId, required String token, }) { final args = encodeInvitesRevokeArgs( + clientProtocolVersion: clientProtocolVersion, strategyPublicId: strategyPublicId, token: token, ); @@ -695,6 +728,7 @@ final class _PageModule implements PageModule { abstract interface class PagesModule { Future add({ + required double clientProtocolVersion, required double expectedRevision, required bool isAttack, ConvexOptional isAutoNamed = const ConvexOptional.absent(), @@ -706,6 +740,7 @@ abstract interface class PagesModule { required String strategyPublicId, }); Future delete({ + required double clientProtocolVersion, required double expectedRevision, required String pagePublicId, required String strategyPublicId, @@ -714,6 +749,7 @@ abstract interface class PagesModule { required String strategyPublicId, }); Future rename({ + required double clientProtocolVersion, required double expectedRevision, ConvexOptional isAutoNamed = const ConvexOptional.absent(), required String name, @@ -721,6 +757,7 @@ abstract interface class PagesModule { required String strategyPublicId, }); Future reorder({ + required double clientProtocolVersion, required double expectedRevision, required List orderedPagePublicIds, required String strategyPublicId, @@ -732,6 +769,7 @@ final class _PagesModule implements PagesModule { final ConvexTransport _transport; @override Future add({ + required double clientProtocolVersion, required double expectedRevision, required bool isAttack, ConvexOptional isAutoNamed = const ConvexOptional.absent(), @@ -743,6 +781,7 @@ final class _PagesModule implements PagesModule { required String strategyPublicId, }) { final args = encodePagesAddArgs( + clientProtocolVersion: clientProtocolVersion, expectedRevision: expectedRevision, isAttack: isAttack, isAutoNamed: isAutoNamed, @@ -760,11 +799,13 @@ final class _PagesModule implements PagesModule { @override Future delete({ + required double clientProtocolVersion, required double expectedRevision, required String pagePublicId, required String strategyPublicId, }) { final args = encodePagesDeleteArgs( + clientProtocolVersion: clientProtocolVersion, expectedRevision: expectedRevision, pagePublicId: pagePublicId, strategyPublicId: strategyPublicId, @@ -792,6 +833,7 @@ final class _PagesModule implements PagesModule { @override Future rename({ + required double clientProtocolVersion, required double expectedRevision, ConvexOptional isAutoNamed = const ConvexOptional.absent(), required String name, @@ -799,6 +841,7 @@ final class _PagesModule implements PagesModule { required String strategyPublicId, }) { final args = encodePagesRenameArgs( + clientProtocolVersion: clientProtocolVersion, expectedRevision: expectedRevision, isAutoNamed: isAutoNamed, name: name, @@ -813,11 +856,13 @@ final class _PagesModule implements PagesModule { @override Future reorder({ + required double clientProtocolVersion, required double expectedRevision, required List orderedPagePublicIds, required String strategyPublicId, }) { final args = encodePagesReorderArgs( + clientProtocolVersion: clientProtocolVersion, expectedRevision: expectedRevision, orderedPagePublicIds: orderedPagePublicIds, strategyPublicId: strategyPublicId, @@ -831,6 +876,7 @@ final class _PagesModule implements PagesModule { abstract interface class SharesModule { Future create({ + required double clientProtocolVersion, required InvitesCreateArgsRole role, required String targetPublicId, required SharesCreateArgsTargetType targetType, @@ -840,8 +886,12 @@ abstract interface class SharesModule { required String targetPublicId, required SharesCreateArgsTargetType targetType, }); - Future redeem({required String token}); + Future redeem({ + required double clientProtocolVersion, + required String token, + }); Future revoke({ + required double clientProtocolVersion, required String targetPublicId, required SharesCreateArgsTargetType targetType, required String token, @@ -853,12 +903,14 @@ final class _SharesModule implements SharesModule { final ConvexTransport _transport; @override Future create({ + required double clientProtocolVersion, required InvitesCreateArgsRole role, required String targetPublicId, required SharesCreateArgsTargetType targetType, required String token, }) { final args = encodeSharesCreateArgs( + clientProtocolVersion: clientProtocolVersion, role: role, targetPublicId: targetPublicId, targetType: targetType, @@ -888,8 +940,14 @@ final class _SharesModule implements SharesModule { } @override - Future redeem({required String token}) { - final args = encodeSharesRedeemArgs(token: token); + Future redeem({ + required double clientProtocolVersion, + required String token, + }) { + final args = encodeSharesRedeemArgs( + clientProtocolVersion: clientProtocolVersion, + token: token, + ); return _invoke( () => _transport.mutation('shares:redeem', args), decodeSharesRedeemResult, @@ -898,11 +956,13 @@ final class _SharesModule implements SharesModule { @override Future revoke({ + required double clientProtocolVersion, required String targetPublicId, required SharesCreateArgsTargetType targetType, required String token, }) { final args = encodeSharesRevokeArgs( + clientProtocolVersion: clientProtocolVersion, targetPublicId: targetPublicId, targetType: targetType, token: token, @@ -916,6 +976,7 @@ final class _SharesModule implements SharesModule { abstract interface class StrategiesModule { Future create({ + required double clientProtocolVersion, ConvexOptional folderPublicId = const ConvexOptional.absent(), required String mapData, required String name, @@ -928,6 +989,7 @@ abstract interface class StrategiesModule { ConvexOptional themeProfileId = const ConvexOptional.absent(), }); Future createWithInitialPage({ + required double clientProtocolVersion, ConvexOptional folderPublicId = const ConvexOptional.absent(), required bool initialPageIsAttack, ConvexOptional initialPageIsAutoNamed = const ConvexOptional.absent(), @@ -947,6 +1009,7 @@ abstract interface class StrategiesModule { ConvexOptional themeProfileId = const ConvexOptional.absent(), }); Future delete({ + required double clientProtocolVersion, required double expectedRevision, required String strategyPublicId, }); @@ -960,6 +1023,7 @@ abstract interface class StrategiesModule { }); ConvexQuery> listSharedWithMe(); Future move({ + required double clientProtocolVersion, required double expectedRevision, ConvexOptional folderPublicId = const ConvexOptional.absent(), required String strategyPublicId, @@ -968,6 +1032,7 @@ abstract interface class StrategiesModule { ConvexOptional clearThemeOverridePalette = const ConvexOptional.absent(), ConvexOptional clearThemeProfileId = const ConvexOptional.absent(), + required double clientProtocolVersion, required double expectedRevision, ConvexOptional mapData = const ConvexOptional.absent(), ConvexOptional name = const ConvexOptional.absent(), @@ -986,6 +1051,7 @@ final class _StrategiesModule implements StrategiesModule { final ConvexTransport _transport; @override Future create({ + required double clientProtocolVersion, ConvexOptional folderPublicId = const ConvexOptional.absent(), required String mapData, required String name, @@ -998,6 +1064,7 @@ final class _StrategiesModule implements StrategiesModule { ConvexOptional themeProfileId = const ConvexOptional.absent(), }) { final args = encodeStrategiesCreateArgs( + clientProtocolVersion: clientProtocolVersion, folderPublicId: folderPublicId, mapData: mapData, name: name, @@ -1013,6 +1080,7 @@ final class _StrategiesModule implements StrategiesModule { @override Future createWithInitialPage({ + required double clientProtocolVersion, ConvexOptional folderPublicId = const ConvexOptional.absent(), required bool initialPageIsAttack, ConvexOptional initialPageIsAutoNamed = const ConvexOptional.absent(), @@ -1032,6 +1100,7 @@ final class _StrategiesModule implements StrategiesModule { ConvexOptional themeProfileId = const ConvexOptional.absent(), }) { final args = encodeStrategiesCreateWithInitialPageArgs( + clientProtocolVersion: clientProtocolVersion, folderPublicId: folderPublicId, initialPageIsAttack: initialPageIsAttack, initialPageIsAutoNamed: initialPageIsAutoNamed, @@ -1052,10 +1121,12 @@ final class _StrategiesModule implements StrategiesModule { @override Future delete({ + required double clientProtocolVersion, required double expectedRevision, required String strategyPublicId, }) { final args = encodeStrategiesDeleteArgs( + clientProtocolVersion: clientProtocolVersion, expectedRevision: expectedRevision, strategyPublicId: strategyPublicId, ); @@ -1111,11 +1182,13 @@ final class _StrategiesModule implements StrategiesModule { @override Future move({ + required double clientProtocolVersion, required double expectedRevision, ConvexOptional folderPublicId = const ConvexOptional.absent(), required String strategyPublicId, }) { final args = encodeStrategiesMoveArgs( + clientProtocolVersion: clientProtocolVersion, expectedRevision: expectedRevision, folderPublicId: folderPublicId, strategyPublicId: strategyPublicId, @@ -1131,6 +1204,7 @@ final class _StrategiesModule implements StrategiesModule { ConvexOptional clearThemeOverridePalette = const ConvexOptional.absent(), ConvexOptional clearThemeProfileId = const ConvexOptional.absent(), + required double clientProtocolVersion, required double expectedRevision, ConvexOptional mapData = const ConvexOptional.absent(), ConvexOptional name = const ConvexOptional.absent(), @@ -1145,6 +1219,7 @@ final class _StrategiesModule implements StrategiesModule { final args = encodeStrategiesUpdateArgs( clearThemeOverridePalette: clearThemeOverridePalette, clearThemeProfileId: clearThemeProfileId, + clientProtocolVersion: clientProtocolVersion, expectedRevision: expectedRevision, mapData: mapData, name: name, @@ -1201,7 +1276,9 @@ final class _StrategyModule implements StrategyModule { } abstract interface class UsersModule { - Future ensureCurrentUser(); + Future ensureCurrentUser({ + required double clientProtocolVersion, + }); ConvexQuery me(); } @@ -1209,8 +1286,12 @@ final class _UsersModule implements UsersModule { const _UsersModule(this._transport); final ConvexTransport _transport; @override - Future ensureCurrentUser() { - final args = encodeUsersEnsureCurrentUserArgs(); + Future ensureCurrentUser({ + required double clientProtocolVersion, + }) { + final args = encodeUsersEnsureCurrentUserArgs( + clientProtocolVersion: clientProtocolVersion, + ); return _invoke( () => _transport.mutation('users:ensureCurrentUser', args), decodeUsersEnsureCurrentUserResult, diff --git a/lib/providers/collab/strategy_op_queue_provider.dart b/lib/providers/collab/strategy_op_queue_provider.dart index 5a92319b..1e735f47 100644 --- a/lib/providers/collab/strategy_op_queue_provider.dart +++ b/lib/providers/collab/strategy_op_queue_provider.dart @@ -25,6 +25,7 @@ class StrategyOpQueueState { this.attentionByEntityKey = const {}, this.loadIssues = const [], this.durableLoaded = false, + this.hasDurabilityFailure = false, this.isFlushing = false, this.lastError, this.lastFlushAt, @@ -42,6 +43,7 @@ class StrategyOpQueueState { final Map attentionByEntityKey; final List loadIssues; final bool durableLoaded; + final bool hasDurabilityFailure; final bool isFlushing; final String? lastError; final DateTime? lastFlushAt; @@ -50,9 +52,13 @@ class StrategyOpQueueState { bool get needsAttention => loadIssues.isNotEmpty || + hasDurabilityFailure || pausedByEntityKey.isNotEmpty || attentionByEntityKey.isNotEmpty; + bool get outboxIsReliable => + durableLoaded && loadIssues.isEmpty && !hasDurabilityFailure; + List get pending => [ ...queuedByEntityKey.values.map((intent) => intent.pending), ...inFlightByEntityKey.values.map((intent) => intent.pending), @@ -67,6 +73,7 @@ class StrategyOpQueueState { Map? successorByEntityKey, Map? pausedByEntityKey, Map? attentionByEntityKey, + bool? hasDurabilityFailure, bool? isFlushing, String? lastError, bool clearError = false, @@ -85,6 +92,8 @@ class StrategyOpQueueState { attentionByEntityKey: attentionByEntityKey ?? this.attentionByEntityKey, loadIssues: loadIssues, durableLoaded: durableLoaded, + hasDurabilityFailure: + hasDurabilityFailure ?? this.hasDurabilityFailure, isFlushing: isFlushing ?? this.isFlushing, lastError: clearError ? null : (lastError ?? this.lastError), lastFlushAt: lastFlushAt ?? this.lastFlushAt, @@ -110,9 +119,12 @@ class StrategyOpQueueNotifier extends Notifier { Timer? _debounceTimer; Timer? _retryTimer; int _offlineRetryCount = 0; + bool _isDisposed = false; late DurableStrategyOutboxStore _store; late Map _recordsByStorageKey; final Set _awaitingRemoteAdoption = {}; + final Set _uncertainOversizedParking = {}; + String? _uncertainOversizedParkingMessage; Future _writeTail = Future.value(); ConvexStrategyRepository get _repo => @@ -120,12 +132,14 @@ class StrategyOpQueueNotifier extends Notifier { @override StrategyOpQueueState build() { + _isDisposed = false; _store = ref.read(durableStrategyOutboxStoreProvider); final loaded = _store.load(); _recordsByStorageKey = { for (final record in loaded.records) record.storageKey: record, }; ref.onDispose(() { + _isDisposed = true; _retryTimer?.cancel(); _debounceTimer?.cancel(); }); @@ -186,6 +200,10 @@ class StrategyOpQueueNotifier extends Notifier { } final clientId = matching.firstOrNull?.pending.clientId ?? const Uuid().v4(); + final hasDurabilityFailure = _hasUncertainParkingFor( + accountId: accountId, + strategyPublicId: strategyPublicId, + ); state = StrategyOpQueueState( accountId: accountId, strategyPublicId: strategyPublicId, @@ -196,11 +214,14 @@ class StrategyOpQueueNotifier extends Notifier { attentionByEntityKey: attention, loadIssues: state.loadIssues, durableLoaded: true, - lastError: _loadedAttentionMessage( - loadIssues: state.loadIssues, - paused: paused, - attention: attention, - ), + hasDurabilityFailure: hasDurabilityFailure, + lastError: hasDurabilityFailure + ? _uncertainOversizedParkingMessage + : _loadedAttentionMessage( + loadIssues: state.loadIssues, + paused: paused, + attention: attention, + ), ); if (queued.isNotEmpty) _scheduleFlush(flushImmediately: true); } @@ -345,10 +366,24 @@ class StrategyOpQueueNotifier extends Notifier { } if (_sameIntent(attentionIntent.pending.op, desired)) { if (successorIntent != null) { + final recoveredOversizedParking = + _uncertainOversizedParking.contains(current.storageKey); await _putRecord(current.copyWith( + status: recoveredOversizedParking + ? DurableOutboxStatus.attention + : current.status, clearSuccessorPending: true, updatedAt: DateTime.now(), + lastError: recoveredOversizedParking + ? cloudOperationTooLargeMessage + : null, )); + if (recoveredOversizedParking) { + _uncertainOversizedParking.remove(current.storageKey); + if (_uncertainOversizedParking.isEmpty) { + _uncertainOversizedParkingMessage = null; + } + } successors.remove(key); changed = true; } @@ -366,11 +401,22 @@ class StrategyOpQueueNotifier extends Notifier { clientId: successorIntent?.pending.clientId ?? attentionIntent.pending.clientId, ); + final recoveredOversizedParking = + _uncertainOversizedParking.contains(current.storageKey); await _putRecord(current.copyWith( status: DurableOutboxStatus.attention, successorPending: pending, updatedAt: DateTime.now(), + lastError: recoveredOversizedParking + ? cloudOperationTooLargeMessage + : null, )); + if (recoveredOversizedParking) { + _uncertainOversizedParking.remove(current.storageKey); + if (_uncertainOversizedParking.isEmpty) { + _uncertainOversizedParkingMessage = null; + } + } successors[key] = QueuedEntityIntent( entityKey: key, pending: pending, @@ -520,6 +566,7 @@ class StrategyOpQueueNotifier extends Notifier { pausedByEntityKey: paused, attentionByEntityKey: attention, successorByEntityKey: successors, + hasDurabilityFailure: _hasUncertainParkingForActiveStrategy, lastError: attentionMessage, clearError: attentionMessage == null, ); @@ -593,18 +640,22 @@ class StrategyOpQueueNotifier extends Notifier { final rejectedOp = rejected.op; final successor = record?.successorPending; final retryOp = successor?.op ?? rejectedOp; + final isPayloadPolicyAttention = + record?.lastError == cloudOperationTooLargeMessage; final retryRevision = record?.latestServerRevision ?? rejectedOp.expectedRevision; - if (retryRevision == null) continue; + if (!isPayloadPolicyAttention && retryRevision == null) continue; final isTombstoneRestore = (retryOp is ElementAddOp || retryOp is LineupAddOp) && (record?.lastError == 'missing_expected_revision' || record?.lastError == 'revision_mismatch'); - final rebasedOp = _rebaseRejectedOp( - retryOp, - retryRevision, - preserveAdd: isTombstoneRestore, - ); + final rebasedOp = isPayloadPolicyAttention + ? retryOp.withOpId(const Uuid().v4()) + : _rebaseRejectedOp( + retryOp, + retryRevision!, + preserveAdd: isTombstoneRestore, + ); final pending = PendingOp( op: rebasedOp, clientId: successor?.clientId ?? rejected.clientId, @@ -615,6 +666,13 @@ class StrategyOpQueueNotifier extends Notifier { status: DurableOutboxStatus.queued, clearSuccessorPending: true, )); + _uncertainOversizedParking.remove( + DurableOutboxRecord.createStorageKey( + accountId: state.accountId!, + strategyPublicId: state.strategyPublicId!, + entityKey: entry.key, + ), + ); queued[entry.key] = QueuedEntityIntent( entityKey: entry.key, pending: pending, @@ -634,6 +692,9 @@ class StrategyOpQueueNotifier extends Notifier { ); return; } + if (_uncertainOversizedParking.isEmpty) { + _uncertainOversizedParkingMessage = null; + } final attentionMessage = _loadedAttentionMessage( loadIssues: state.loadIssues, paused: state.pausedByEntityKey, @@ -643,6 +704,7 @@ class StrategyOpQueueNotifier extends Notifier { queuedByEntityKey: queued, attentionByEntityKey: attention, successorByEntityKey: successors, + hasDurabilityFailure: _hasUncertainParkingForActiveStrategy, lastError: attentionMessage, clearError: attentionMessage == null, ); @@ -671,16 +733,29 @@ class StrategyOpQueueNotifier extends Notifier { for (final key in entityKeys) { final rejected = attention[key]; + final accountId = state.accountId; + final strategyPublicId = state.strategyPublicId; + if (rejected == null || accountId == null || strategyPublicId == null) { + continue; + } + final storageKey = DurableOutboxRecord.createStorageKey( + accountId: accountId, + strategyPublicId: strategyPublicId, + entityKey: key, + ); + final hasUncertainParking = + _uncertainOversizedParking.contains(storageKey); final record = _recordForActiveKey(key); - if (rejected == null || - record == null || - record.status != DurableOutboxStatus.attention || - record.pending.op.opId != rejected.pending.op.opId) { + final hasMatchingAttentionRecord = record != null && + record.status == DurableOutboxStatus.attention && + record.pending.op.opId == rejected.pending.op.opId; + if (!hasUncertainParking && !hasMatchingAttentionRecord) { continue; } try { - await _store.remove(record.storageKey); - _recordsByStorageKey.remove(record.storageKey); + await _store.remove(storageKey); + _recordsByStorageKey.remove(storageKey); + _uncertainOversizedParking.remove(storageKey); attention.remove(key); successors.remove(key); _awaitingRemoteAdoption.add(key); @@ -700,6 +775,9 @@ class StrategyOpQueueNotifier extends Notifier { stackTrace: persistenceStackTrace, ); } + if (_uncertainOversizedParking.isEmpty) { + _uncertainOversizedParkingMessage = null; + } final attentionMessage = _loadedAttentionMessage( loadIssues: state.loadIssues, paused: state.pausedByEntityKey, @@ -712,6 +790,7 @@ class StrategyOpQueueNotifier extends Notifier { state = state.copyWith( attentionByEntityKey: attention, successorByEntityKey: successors, + hasDurabilityFailure: _hasUncertainParkingForActiveStrategy, lastError: errorMessage, clearError: errorMessage == null, ); @@ -723,12 +802,97 @@ class StrategyOpQueueNotifier extends Notifier { _awaitingRemoteAdoption.removeAll(entityKeys); } + Future _parkOversizedQueuedOps() async { + if (_hasUncertainParkingForActiveStrategy) return false; + final oversized = state.queuedByEntityKey.entries + .where((entry) => cloudOperationExceedsPolicy(entry.value.pending.op)) + .toList(growable: false); + if (oversized.isEmpty) return true; + + final queued = Map.from( + state.queuedByEntityKey, + ); + final attention = Map.from( + state.attentionByEntityKey, + ); + Object? persistenceError; + StackTrace? persistenceStackTrace; + for (final entry in oversized) { + final record = _recordForActiveKey(entry.key); + if (record == null || + record.pending.op.opId != entry.value.pending.op.opId) { + persistenceError ??= + StateError('Durable queued record is missing for ${entry.key}.'); + queued.remove(entry.key); + attention[entry.key] = entry.value; + _uncertainOversizedParking.add( + DurableOutboxRecord.createStorageKey( + accountId: state.accountId!, + strategyPublicId: state.strategyPublicId!, + entityKey: entry.key, + ), + ); + continue; + } + try { + await _putRecord(record.copyWith( + status: DurableOutboxStatus.attention, + updatedAt: DateTime.now(), + lastError: cloudOperationTooLargeMessage, + )); + queued.remove(entry.key); + attention[entry.key] = entry.value; + } catch (error, stackTrace) { + persistenceError ??= error; + persistenceStackTrace ??= stackTrace; + queued.remove(entry.key); + attention[entry.key] = entry.value; + _uncertainOversizedParking.add(record.storageKey); + } + } + if (persistenceError != null) { + log( + 'Durable outbox persistence failed: $persistenceError', + name: 'strategy_outbox', + error: persistenceError, + stackTrace: persistenceStackTrace, + ); + _debounceTimer?.cancel(); + _debounceTimer = null; + _retryTimer?.cancel(); + _retryTimer = null; + _uncertainOversizedParkingMessage = + 'Cloud work could not be verified in the durable outbox. ' + 'Nothing was sent. The change still needs attention: ' + '$persistenceError'; + } + final attentionMessage = _loadedAttentionMessage( + loadIssues: state.loadIssues, + paused: state.pausedByEntityKey, + attention: attention, + ); + state = state.copyWith( + queuedByEntityKey: queued, + attentionByEntityKey: attention, + hasDurabilityFailure: _hasUncertainParkingForActiveStrategy, + lastError: persistenceError == null + ? attentionMessage + : _uncertainOversizedParkingMessage, + clearError: persistenceError == null && attentionMessage == null, + ); + return persistenceError == null; + } + Future flushNow() async { await _writeTail; + if (_isDisposed) return; if (state.isFlushing) return; final strategyPublicId = state.strategyPublicId; if (strategyPublicId == null || state.queuedByEntityKey.isEmpty) return; + if (!await _parkOversizedQueuedOps() || _isDisposed) return; + if (state.queuedByEntityKey.isEmpty) return; + final mode = ref.read(cloudCollabModeProvider); if (!mode.featureFlagEnabled || mode.forceLocalFallback) return; final auth = ref.read(authProvider); @@ -763,10 +927,25 @@ class StrategyOpQueueNotifier extends Notifier { final candidates = state.queuedByEntityKey.values.toList(growable: false); if (candidates.isEmpty) return; final batchClientId = candidates.first.pending.clientId; - final batch = candidates - .where((intent) => intent.pending.clientId == batchClientId) - .take(_maxBatchSize) - .toList(growable: false); + final batch = []; + for (final candidate in candidates) { + if (candidate.pending.clientId != batchClientId) continue; + if (batch.length >= _maxBatchSize) break; + final nextBatch = [...batch, candidate]; + final byteSize = serializedCloudBatchUtf8Bytes( + strategyPublicId: strategyPublicId, + clientId: batchClientId, + ops: nextBatch.map((intent) => intent.pending.op), + ); + if (byteSize > maxCloudBatchBytes) break; + batch.add(candidate); + } + if (batch.isEmpty) { + state = state.copyWith( + lastError: 'Cloud operation batch exceeds the payload policy.', + ); + return; + } final queued = Map.from( state.queuedByEntityKey, ); @@ -1055,6 +1234,21 @@ class StrategyOpQueueNotifier extends Notifier { return next; } + bool _hasUncertainParkingFor({ + required String? accountId, + required String? strategyPublicId, + }) { + if (accountId == null || strategyPublicId == null) return false; + final prefix = '${Uri.encodeComponent(accountId)}|' + '${Uri.encodeComponent(strategyPublicId)}|'; + return _uncertainOversizedParking.any((key) => key.startsWith(prefix)); + } + + bool get _hasUncertainParkingForActiveStrategy => _hasUncertainParkingFor( + accountId: state.accountId, + strategyPublicId: state.strategyPublicId, + ); + void _recordPersistenceFailure(Object error, StackTrace stackTrace) { log('Durable outbox persistence failed: $error', name: 'strategy_outbox', error: error, stackTrace: stackTrace); @@ -1356,7 +1550,21 @@ class StrategyOpQueueNotifier extends Notifier { if (loadIssues.isNotEmpty) { return 'The cloud outbox contains unreadable saved work.'; } - if (attention.isNotEmpty) return 'Some saved work needs attention.'; + if (_hasUncertainParkingForActiveStrategy) { + return _uncertainOversizedParkingMessage ?? + 'Cloud work could not be verified in the durable outbox. ' + 'Nothing was sent.'; + } + if (attention.isNotEmpty) { + final hasOversizedWork = attention.entries.any((entry) { + if (cloudOperationExceedsPolicy(entry.value.pending.op)) return true; + final record = _recordForActiveKey(entry.key); + return record?.pending.op.opId == entry.value.pending.op.opId && + record?.lastError == cloudOperationTooLargeMessage; + }); + if (hasOversizedWork) return cloudOperationTooLargeMessage; + return 'Some saved work needs attention.'; + } if (paused.isNotEmpty) return 'Some saved work is paused after retries.'; return null; } diff --git a/lib/widgets/cloud_sync_status_chip.dart b/lib/widgets/cloud_sync_status_chip.dart index ccec9352..0caa9871 100644 --- a/lib/widgets/cloud_sync_status_chip.dart +++ b/lib/widgets/cloud_sync_status_chip.dart @@ -455,7 +455,10 @@ class _SyncStatusPopover extends StatelessWidget { String get _attentionExplanation { final mediaErrors = saveState.mediaSyncErrorCount; final parts = []; - if (hasRejectedWork) { + final error = saveState.cloudSyncError; + final hasOversizedWork = + error?.toLowerCase().contains('too large for cloud sync') ?? false; + if (hasRejectedWork && !hasOversizedWork) { parts.add( 'Another edit reached the cloud first. Your version remains saved ' 'on this device.', @@ -466,13 +469,22 @@ class _SyncStatusPopover extends StatelessWidget { : 'Your choice applies to all $rejectedCount conflicting changes.', ); } - final error = saveState.cloudSyncError; final retryUnavailable = error?.toLowerCase().contains('cannot be retried automatically') ?? false; - if (error != null && (!hasRejectedWork || retryUnavailable)) { + if (error != null && + (!hasRejectedWork || retryUnavailable || hasOversizedWork)) { parts.add(friendlyCloudSyncError(error)); } + if (hasRejectedWork && hasOversizedWork) { + parts.add( + rejectedCount == 1 + ? 'Choose whether to keep this local change or use the cloud ' + 'version.' + : 'Your choice applies to all $rejectedCount changes that need ' + 'attention.', + ); + } if (mediaErrors > 0) { parts.add( mediaErrors == 1 diff --git a/test/collab/cloud_sync_error_message_test.dart b/test/collab/cloud_sync_error_message_test.dart index d69d4227..a006cf9f 100644 --- a/test/collab/cloud_sync_error_message_test.dart +++ b/test/collab/cloud_sync_error_message_test.dart @@ -18,4 +18,15 @@ void main() { expect(message, "Some changes haven't reached the cloud yet."); }); + + test('does not promise local durability when the outbox is uncertain', () { + final message = friendlyCloudSyncError( + 'Cloud work could not be verified in the durable outbox. ' + 'Nothing was sent.', + ); + + expect(message, contains('could not verify')); + expect(message, contains('Nothing was sent')); + expect(message, isNot(contains('remains saved'))); + }); } diff --git a/test/strategy_op_queue_provider_test.dart b/test/strategy_op_queue_provider_test.dart index 8cbab99b..01fb419f 100644 --- a/test/strategy_op_queue_provider_test.dart +++ b/test/strategy_op_queue_provider_test.dart @@ -1,4 +1,5 @@ import 'dart:async'; +import 'dart:convert'; import 'package:flutter_riverpod/flutter_riverpod.dart'; import 'package:flutter_test/flutter_test.dart'; @@ -858,6 +859,536 @@ void main() { await flush; }); + group('cloud payload policy', () { + test('serialized operation size counts Unicode UTF-8 bytes', () { + final op = _largeElementPatch( + opId: 'unicode-size', + elementId: 'element-unicode', + value: _repeat('界', 3), + ); + final serialized = jsonEncode(op.toConvexJson()); + + expect( + serializedCloudOperationUtf8Bytes(op), + utf8.encode(serialized).length, + ); + expect(utf8.encode(serialized).length, greaterThan(serialized.length)); + }); + + test('an oversized op is durably parked while independent work lands', + () async { + final store = MemoryDurableStrategyOutboxStore(); + final repository = _RecordingAckRepository(); + var container = _cloudQueueContainer( + store: store, + repository: repository, + ); + var notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + container + .read(cloudCollabModeProvider.notifier) + .setForceLocalFallback(true); + final oversized = _largeElementPatch( + opId: 'oversized', + elementId: 'element-large', + ); + const valid = ElementPatchOp( + opId: 'independent', + elementPublicId: 'element-small', + pagePublicId: 'page-1', + payload: {'value': 'safe'}, + expectedElementRevision: 1, + ); + expect( + serializedCloudOperationUtf8Bytes(oversized), + greaterThan(maxCloudOperationBytes), + ); + await notifier.enqueue(oversized, flushImmediately: false); + await notifier.enqueue(valid, flushImmediately: false); + container + .read(cloudCollabModeProvider.notifier) + .setForceLocalFallback(false); + + await notifier.flushNow(); + + expect(repository.calls, hasLength(1)); + expect(repository.calls.single.map((op) => op.opId), ['independent']); + var current = container.read(strategyOpQueueProvider); + const oversizedKey = + EntitySyncKey.element('page-1', 'element-large'); + expect(current.attentionByEntityKey, contains(oversizedKey)); + expect(current.queuedByEntityKey, isEmpty); + expect(current.lastError, cloudOperationTooLargeMessage); + var durable = store.load().records.single; + expect(durable.status, DurableOutboxStatus.attention); + expect(durable.pending.op.opId, 'oversized'); + expect(durable.lastError, cloudOperationTooLargeMessage); + + container.dispose(); + repository.calls.clear(); + container = _cloudQueueContainer( + store: store, + repository: repository, + ); + addTearDown(container.dispose); + notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + await notifier.flushNow(); + + current = container.read(strategyOpQueueProvider); + expect(current.attentionByEntityKey, contains(oversizedKey)); + expect(current.lastError, cloudOperationTooLargeMessage); + expect(repository.calls, isEmpty); + durable = store.load().records.single; + expect(durable.status, DurableOutboxStatus.attention); + expect(durable.pending.op.opId, 'oversized'); + }); + + test('an over-wide array is parked before independent transport', + () async { + final store = MemoryDurableStrategyOutboxStore(); + final repository = _RecordingAckRepository(); + final container = _cloudQueueContainer( + store: store, + repository: repository, + ); + addTearDown(container.dispose); + final notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + final wide = ElementPatchOp( + opId: 'wide-array', + elementPublicId: 'element-wide', + pagePublicId: 'page-1', + payload: { + 'kind': 'drawing', + 'payloadVersion': 1, + 'data': {'points': List.filled(maxCloudArrayEntries + 1, 0)}, + }, + expectedElementRevision: 1, + ); + expect( + serializedCloudOperationUtf8Bytes(wide), + lessThan(maxCloudOperationBytes), + ); + expect(cloudOperationExceedsPolicy(wide), isTrue); + await notifier.enqueue(wide, flushImmediately: false); + await notifier.enqueue( + const ElementPatchOp( + opId: 'wide-array-sibling', + elementPublicId: 'element-small', + pagePublicId: 'page-1', + payload: {'value': 'safe'}, + expectedElementRevision: 1, + ), + flushImmediately: false, + ); + + await notifier.flushNow(); + + expect(repository.calls, hasLength(1)); + expect(repository.calls.single.map((op) => op.opId), [ + 'wide-array-sibling', + ]); + expect( + container.read(strategyOpQueueProvider).attentionByEntityKey, + contains(const EntitySyncKey.element('page-1', 'element-wide')), + ); + expect(store.load().records.single.lastError, + cloudOperationTooLargeMessage); + }); + + test('a failed oversized parking write blocks all transport and retries', + () async { + final store = _OversizedParkingFailureStore(); + final repository = _RecordingAckRepository(); + final container = _cloudQueueContainer( + store: store, + repository: repository, + ); + addTearDown(container.dispose); + final notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + await notifier.enqueue( + _largeElementPatch( + opId: 'oversized-write-failure', + elementId: 'element-large', + ), + flushImmediately: false, + ); + await notifier.enqueue( + const ElementPatchOp( + opId: 'independent-after-write-failure', + elementPublicId: 'element-small', + pagePublicId: 'page-1', + payload: {'value': 'safe'}, + expectedElementRevision: 1, + ), + flushImmediately: false, + ); + + await notifier.flushNow(); + await Future.delayed(const Duration(milliseconds: 400)); + + final current = container.read(strategyOpQueueProvider); + expect(repository.calls, isEmpty); + expect(current.outboxIsReliable, isFalse); + expect(current.hasDurabilityFailure, isTrue); + expect( + current.attentionByEntityKey, + contains(const EntitySyncKey.element('page-1', 'element-large')), + ); + expect( + current.queuedByEntityKey, + contains(const EntitySyncKey.element('page-1', 'element-small')), + ); + expect(current.lastError, contains('Nothing was sent')); + expect(store.attentionWrites, 1); + }); + + test('a missing durable oversized record blocks all transport', () async { + final store = _OversizedParkingFailureStore(dropBeforeThrow: true); + final repository = _RecordingAckRepository(); + final container = _cloudQueueContainer( + store: store, + repository: repository, + ); + addTearDown(container.dispose); + final notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + await notifier.enqueue( + _largeElementPatch( + opId: 'oversized-missing-record', + elementId: 'element-large', + ), + flushImmediately: false, + ); + await notifier.enqueue( + const ElementPatchOp( + opId: 'independent-after-missing-record', + elementPublicId: 'element-small', + pagePublicId: 'page-1', + payload: {'value': 'safe'}, + expectedElementRevision: 1, + ), + flushImmediately: false, + ); + + await notifier.flushNow(); + await Future.delayed(const Duration(milliseconds: 400)); + + final current = container.read(strategyOpQueueProvider); + expect(repository.calls, isEmpty); + expect(current.outboxIsReliable, isFalse); + expect(current.hasDurabilityFailure, isTrue); + expect( + current.attentionByEntityKey, + contains(const EntitySyncKey.element('page-1', 'element-large')), + ); + expect( + current.queuedByEntityKey, + contains(const EntitySyncKey.element('page-1', 'element-small')), + ); + expect(current.lastError, contains('Nothing was sent')); + expect( + store.load().records.map((record) => record.pending.op.opId), + isNot(contains('oversized-missing-record')), + ); + expect(store.attentionWrites, 1); + }); + + for (final dropBeforeThrow in [false, true]) { + test( + 'Use cloud clears uncertain oversized work when the parking ' + '${dropBeforeThrow ? 'record is missing' : 'write failed'}', + () async { + final store = _OversizedParkingFailureStore( + dropBeforeThrow: dropBeforeThrow, + ); + final container = _cloudQueueContainer( + store: store, + repository: _RecordingAckRepository(), + ); + addTearDown(container.dispose); + final notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + const key = EntitySyncKey.element('page-1', 'element-large'); + await notifier.enqueue( + _largeElementPatch( + opId: 'oversized-explicit-discard', + elementId: 'element-large', + ), + flushImmediately: false, + ); + await notifier.flushNow(); + + var current = container.read(strategyOpQueueProvider); + expect(current.attentionByEntityKey, contains(key)); + expect(current.hasDurabilityFailure, isTrue); + expect(current.outboxIsReliable, isFalse); + expect( + store.load().records.isEmpty, + dropBeforeThrow, + ); + + final discarded = await notifier.discardRejected({key}); + + expect(discarded, {key}); + current = container.read(strategyOpQueueProvider); + expect(current.attentionByEntityKey, isEmpty); + expect(current.hasDurabilityFailure, isFalse); + expect(current.outboxIsReliable, isTrue); + expect(current.lastError, isNull); + expect(store.load().records, isEmpty); + expect(store.removalAttempts, 1); + }); + } + + test('Use cloud remains fail-closed when uncertain removal fails', + () async { + final store = _OversizedParkingFailureStore(failRemove: true); + final container = _cloudQueueContainer( + store: store, + repository: _RecordingAckRepository(), + ); + addTearDown(container.dispose); + final notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + const key = EntitySyncKey.element('page-1', 'element-large'); + await notifier.enqueue( + _largeElementPatch( + opId: 'oversized-failed-discard', + elementId: 'element-large', + ), + flushImmediately: false, + ); + await notifier.flushNow(); + + final discarded = await notifier.discardRejected({key}); + + expect(discarded, isEmpty); + final current = container.read(strategyOpQueueProvider); + expect(current.attentionByEntityKey, contains(key)); + expect(current.hasDurabilityFailure, isTrue); + expect(current.outboxIsReliable, isFalse); + expect(current.lastError, contains('could not be removed')); + expect(store.load().records.single.status, DurableOutboxStatus.queued); + expect(store.removalAttempts, 1); + }); + + test('batches split below the conservative argument byte cap', () async { + final store = MemoryDurableStrategyOutboxStore(); + final repository = _RecordingAckRepository(); + final container = _cloudQueueContainer( + store: store, + repository: repository, + ); + addTearDown(container.dispose); + container + .read(cloudCollabModeProvider.notifier) + .setForceLocalFallback(true); + final notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + for (var index = 0; index < 20; index += 1) { + await notifier.enqueue( + _largeElementPatch( + opId: 'batch-$index', + elementId: 'element-$index', + value: _repeat('x', 820 * 1024), + ), + flushImmediately: false, + ); + } + final allOps = container + .read(strategyOpQueueProvider) + .queuedByEntityKey + .values + .map((intent) => intent.pending.op) + .toList(growable: false); + expect( + serializedCloudBatchUtf8Bytes( + strategyPublicId: 'strategy-1', + clientId: container.read(strategyOpQueueProvider).clientId!, + ops: allOps, + ), + greaterThan(maxCloudBatchBytes), + ); + container + .read(cloudCollabModeProvider.notifier) + .setForceLocalFallback(false); + + await notifier.flushNow(); + await repository.secondCall.future; + for (var index = 0; + index < 10 && + container.read(strategyOpQueueProvider).pending.isNotEmpty; + index += 1) { + await Future.delayed(Duration.zero); + } + + expect(repository.calls, hasLength(2)); + expect(repository.calls.expand((batch) => batch), hasLength(20)); + for (final batch in repository.calls) { + expect( + serializedCloudBatchUtf8Bytes( + strategyPublicId: 'strategy-1', + clientId: container.read(strategyOpQueueProvider).clientId!, + ops: batch, + ), + lessThanOrEqualTo(maxCloudBatchBytes), + ); + } + expect(container.read(strategyOpQueueProvider).pending, isEmpty); + expect(store.values, isEmpty); + }); + + test('an oversized same-entity successor is retained in attention', + () async { + final store = MemoryDurableStrategyOutboxStore(); + final repository = _SequencedAckRepository(); + var container = _cloudQueueContainer( + store: store, + repository: repository, + ); + var notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + const key = EntitySyncKey.element('page-1', 'element-1'); + await notifier.enqueue(_elementPatch( + opId: 'safe-predecessor', + value: 'safe', + expectedRevision: 1, + )); + final firstFlush = notifier.flushNow(); + await repository.firstStarted.future; + final oversizedSuccessor = _largeElementPatch( + opId: 'oversized-successor', + elementId: 'element-1', + ); + await notifier.syncDesiredOpsForPage( + pageId: 'page-1', + desiredOpsByEntityKey: {key: oversizedSuccessor}, + ); + + repository.completeFirst(const AppliedOpAck( + opId: 'safe-predecessor', + revision: 2, + )); + await firstFlush; + for (var index = 0; + index < 10 && + container + .read(strategyOpQueueProvider) + .attentionByEntityKey + .isEmpty; + index += 1) { + await Future.delayed(Duration.zero); + } + + var current = container.read(strategyOpQueueProvider); + expect(repository.calls, hasLength(1)); + expect(current.attentionByEntityKey, contains(key)); + expect(current.successorByEntityKey, isEmpty); + expect(current.lastError, cloudOperationTooLargeMessage); + var durable = store.load().records.single; + expect(durable.status, DurableOutboxStatus.attention); + expect(durable.pending.op.payload, oversizedSuccessor.payload); + expect(durable.lastError, cloudOperationTooLargeMessage); + + container.dispose(); + container = _cloudQueueContainer( + store: store, + repository: _RecordingAckRepository(), + ); + addTearDown(container.dispose); + notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + await notifier.flushNow(); + + current = container.read(strategyOpQueueProvider); + expect(current.attentionByEntityKey, contains(key)); + durable = store.load().records.single; + expect(durable.pending.op.payload, oversizedSuccessor.payload); + }); + + test('a valid successor behind an oversized predecessor can land', + () async { + final store = MemoryDurableStrategyOutboxStore(); + final repository = _RecordingAckRepository(); + var container = _cloudQueueContainer( + store: store, + repository: repository, + ); + container + .read(cloudCollabModeProvider.notifier) + .setForceLocalFallback(true); + var notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + const key = EntitySyncKey.element('page-1', 'element-1'); + await notifier.enqueue( + ElementAddOp( + opId: 'oversized-predecessor', + elementPublicId: 'element-1', + pagePublicId: 'page-1', + payload: { + 'kind': 'drawing', + 'payloadVersion': 1, + 'data': {'encodedPoints': _repeat('界', 310000)}, + }, + sortIndex: 0, + ), + flushImmediately: false, + ); + await notifier.flushNow(); + await notifier.syncDesiredOpsForPage( + pageId: 'page-1', + desiredOpsByEntityKey: { + key: const ElementAddOp( + opId: 'valid-successor', + elementPublicId: 'element-1', + pagePublicId: 'page-1', + payload: { + 'kind': 'drawing', + 'payloadVersion': 1, + 'data': {'encodedPoints': 'reduced drawing'}, + }, + sortIndex: 0, + ), + }, + ); + + var durable = store.load().records.single; + expect(durable.status, DurableOutboxStatus.attention); + expect(durable.pending.op.opId, 'oversized-predecessor'); + expect(durable.successorPending?.op.opId, 'valid-successor'); + + container.dispose(); + container = _cloudQueueContainer( + store: store, + repository: repository, + ); + addTearDown(container.dispose); + notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + + await notifier.retryRejected(flushImmediately: false); + await notifier.flushNow(); + + expect(repository.calls, hasLength(1)); + expect(repository.calls.single, hasLength(1)); + expect(repository.calls.single.single, isA()); + expect(repository.calls.single.single.payload, { + 'kind': 'drawing', + 'payloadVersion': 1, + 'data': {'encodedPoints': 'reduced drawing'}, + }); + expect( + serializedCloudOperationUtf8Bytes(repository.calls.single.single), + lessThanOrEqualTo(maxCloudOperationBytes), + ); + expect(container.read(strategyOpQueueProvider).pending, isEmpty); + expect(store.values, isEmpty); + }); + }); + group('acknowledgement persistence recovery', () { test('accepted ack remove failure restores the batch for retry', () async { final store = _OneShotAckFailureStore(failRemove: true); @@ -1612,6 +2143,22 @@ ElementPatchOp _elementPatch({ ); } +ElementPatchOp _largeElementPatch({ + required String opId, + required String elementId, + String? value, +}) { + return ElementPatchOp( + opId: opId, + elementPublicId: elementId, + pagePublicId: 'page-1', + payload: {'value': value ?? _repeat('界', 310000)}, + expectedElementRevision: 1, + ); +} + +String _repeat(String value, int count) => List.filled(count, value).join(); + void _expectBatchRestored( ProviderContainer container, MemoryDurableStrategyOutboxStore store, @@ -1670,6 +2217,26 @@ class _AckRepository extends ConvexStrategyRepository { } } +class _RecordingAckRepository extends ConvexStrategyRepository { + _RecordingAckRepository() : super(IcarusConvexApi(_UnusedTransport())); + + final List> calls = []; + final secondCall = Completer(); + + @override + Future> applyBatch({ + required String strategyPublicId, + required String clientId, + required List ops, + }) async { + calls.add(List.from(ops)); + if (calls.length == 2 && !secondCall.isCompleted) secondCall.complete(); + return [ + for (final op in ops) AppliedOpAck(opId: op.opId, revision: 2), + ]; + } +} + class _SequencedAckRepository extends ConvexStrategyRepository { _SequencedAckRepository() : super(IcarusConvexApi(_UnusedTransport())); @@ -1730,6 +2297,37 @@ class _OneShotAckFailureStore extends MemoryDurableStrategyOutboxStore { } } +class _OversizedParkingFailureStore + extends MemoryDurableStrategyOutboxStore { + _OversizedParkingFailureStore({ + this.dropBeforeThrow = false, + this.failRemove = false, + }); + + final bool dropBeforeThrow; + final bool failRemove; + var attentionWrites = 0; + var removalAttempts = 0; + + @override + Future put(DurableOutboxRecord record) async { + if (record.status == DurableOutboxStatus.attention && + cloudOperationExceedsPolicy(record.pending.op)) { + attentionWrites += 1; + if (dropBeforeThrow) values.remove(record.storageKey); + throw StateError('oversized attention write failed'); + } + await super.put(record); + } + + @override + Future remove(String storageKey) async { + removalAttempts += 1; + if (failRemove) throw StateError('uncertain removal failed'); + await super.remove(storageKey); + } +} + class _FailingSelectedRemovalStore extends MemoryDurableStrategyOutboxStore { String? failStorageKey; diff --git a/test/widgets/cloud_sync_status_chip_test.dart b/test/widgets/cloud_sync_status_chip_test.dart index c6831c6e..5fd5a498 100644 --- a/test/widgets/cloud_sync_status_chip_test.dart +++ b/test/widgets/cloud_sync_status_chip_test.dart @@ -8,8 +8,8 @@ import 'package:icarus/providers/collab/convex_connection_provider.dart'; import 'package:icarus/providers/collab/strategy_op_queue_provider.dart'; import 'package:icarus/providers/strategy_page_session_provider.dart'; import 'package:icarus/providers/strategy_provider.dart'; +import 'package:icarus/providers/strategy_save_state_provider.dart'; import 'package:icarus/providers/text_draft_provider.dart'; -import 'package:icarus/strategy/strategy_models.dart'; import 'package:icarus/strategy/strategy_page_models.dart'; import 'package:icarus/widgets/cloud_sync_status_chip.dart'; import 'package:shadcn_ui/shadcn_ui.dart'; @@ -35,6 +35,19 @@ class _SettledOpQueue extends StrategyOpQueueNotifier { ); } +class _UnreliableOpQueue extends StrategyOpQueueNotifier { + @override + StrategyOpQueueState build() => const StrategyOpQueueState( + accountId: 'account-a', + strategyPublicId: 'cloud-strategy', + clientId: 'client-a', + durableLoaded: true, + hasDurabilityFailure: true, + lastError: 'Cloud work could not be verified in the durable outbox. ' + 'Nothing was sent.', + ); +} + class _EmptyMediaQueue extends CloudMediaUploadQueueNotifier { @override CloudMediaUploadQueueState build() => const CloudMediaUploadQueueState( @@ -201,6 +214,39 @@ void main() { expect(find.text('Synced'), findsNothing); }); + testWidgets('durability uncertainty never appears synced or reliable', + (tester) async { + final container = ProviderContainer( + overrides: [ + strategyProvider.overrideWith(_CloudStrategyProvider.new), + strategyOpQueueProvider.overrideWith(_UnreliableOpQueue.new), + cloudMediaUploadQueueProvider.overrideWith(_EmptyMediaQueue.new), + convexConnectionProvider.overrideWith((ref) => Stream.value(true)), + ], + ); + addTearDown(container.dispose); + + await tester.pumpWidget( + UncontrolledProviderScope( + container: container, + child: const ShadApp( + home: Scaffold(body: CloudSyncStatusChip()), + ), + ), + ); + await tester.pump(); + + final queue = container.read(strategyOpQueueProvider); + expect(queue.outboxIsReliable, isFalse); + expect(find.text('Needs attention'), findsOneWidget); + expect(find.text('Synced'), findsNothing); + + await tester.tap(find.text('Needs attention')); + await tester.pumpAndSettle(); + expect(find.text('Retry sync'), findsOneWidget); + expect(find.textContaining('safely stored'), findsNothing); + }); + testWidgets('conflict popover offers an explicit cloud choice', (tester) async { final queue = _AttentionOpQueue(2); @@ -270,6 +316,40 @@ void main() { expect(session.useCloudCount, 0); }); + testWidgets('oversized saved work shows its durable attention reason', + (tester) async { + final queue = _AttentionOpQueue(1); + final session = _ConflictSession(); + final container = _createConflictContainer( + queue: queue, + session: session, + ); + addTearDown(container.dispose); + container + .read(strategySaveStateProvider.notifier) + .setCloudSyncError(cloudOperationTooLargeMessage); + + await tester.pumpWidget( + UncontrolledProviderScope( + container: container, + child: const ShadApp( + home: Scaffold(body: CloudSyncStatusChip()), + ), + ), + ); + await tester.pump(); + await tester.tap(find.text('Needs attention')); + await tester.pumpAndSettle(); + + expect( + find.textContaining('too large for cloud sync'), + findsOneWidget, + ); + expect(find.textContaining('Another edit reached'), findsNothing); + expect(find.text('Use cloud'), findsOneWidget); + expect(find.text('Keep mine'), findsOneWidget); + }); + testWidgets('failed cloud load keeps attention and explains the failure', (tester) async { final queue = _AttentionOpQueue(1); diff --git a/tool/audit_convex_contract.mjs b/tool/audit_convex_contract.mjs index 4e0362b5..f4f16a49 100644 --- a/tool/audit_convex_contract.mjs +++ b/tool/audit_convex_contract.mjs @@ -125,6 +125,18 @@ if ( fail(`${functionSpecEntry.identifier} is missing an args validator`); } else { auditValidator(functionSpecEntry.args, ["args"], functionSpecEntry.identifier); + if (functionSpecEntry.functionType === "Mutation") { + const protocolField = functionSpecEntry.args.value?.clientProtocolVersion; + if ( + functionSpecEntry.args.type !== "object" || + protocolField?.optional !== false || + protocolField?.fieldType?.type !== "number" + ) { + fail( + `${functionSpecEntry.identifier} must require numeric clientProtocolVersion`, + ); + } + } } if (functionSpecEntry.returns === null) { fail(`${functionSpecEntry.identifier} is missing a return validator`); diff --git a/tool/snapshot_convex_contract.mjs b/tool/snapshot_convex_contract.mjs index 4cfcdd11..f33228fd 100644 --- a/tool/snapshot_convex_contract.mjs +++ b/tool/snapshot_convex_contract.mjs @@ -25,8 +25,11 @@ function sortObjectKeys(value) { } function readFunctionSpec() { - const executable = process.platform === "win32" ? "npx.cmd" : "npx"; - const args = ["convex", "function-spec"]; + const executable = process.execPath; + const args = [ + resolve(repositoryRoot, "node_modules", "convex", "bin", "main.js"), + "function-spec", + ]; const previewName = process.env.CONVEX_PREVIEW_NAME; if (previewName) { args.push("--preview-name", previewName); @@ -72,7 +75,7 @@ function writeOrCheck(path, contents) { return; } const committed = readFileSync(path, "utf8"); - if (committed !== contents) { + if (committed.replaceAll("\r\n", "\n") !== contents) { throw new Error(`${path} is stale; run npm run snapshot:convex-contract`); } } From 01e79d4b7f531933bcf6ed07164ddf21e3a9f439 Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Fri, 4 Sep 2026 00:26:41 -0400 Subject: [PATCH 2/3] fix: mark durable write failures unreliable --- .../collab/strategy_op_queue_provider.dart | 77 ++++++++++++++++--- test/strategy_op_queue_provider_test.dart | 53 +++++++++++++ 2 files changed, 119 insertions(+), 11 deletions(-) diff --git a/lib/providers/collab/strategy_op_queue_provider.dart b/lib/providers/collab/strategy_op_queue_provider.dart index 1e735f47..5519967e 100644 --- a/lib/providers/collab/strategy_op_queue_provider.dart +++ b/lib/providers/collab/strategy_op_queue_provider.dart @@ -124,6 +124,7 @@ class StrategyOpQueueNotifier extends Notifier { late Map _recordsByStorageKey; final Set _awaitingRemoteAdoption = {}; final Set _uncertainOversizedParking = {}; + final Set _uncertainDurableRecords = {}; String? _uncertainOversizedParkingMessage; Future _writeTail = Future.value(); @@ -200,7 +201,7 @@ class StrategyOpQueueNotifier extends Notifier { } final clientId = matching.firstOrNull?.pending.clientId ?? const Uuid().v4(); - final hasDurabilityFailure = _hasUncertainParkingFor( + final hasDurabilityFailure = _hasDurabilityFailureFor( accountId: accountId, strategyPublicId: strategyPublicId, ); @@ -566,7 +567,7 @@ class StrategyOpQueueNotifier extends Notifier { pausedByEntityKey: paused, attentionByEntityKey: attention, successorByEntityKey: successors, - hasDurabilityFailure: _hasUncertainParkingForActiveStrategy, + hasDurabilityFailure: _hasDurabilityFailureForActiveStrategy, lastError: attentionMessage, clearError: attentionMessage == null, ); @@ -704,7 +705,7 @@ class StrategyOpQueueNotifier extends Notifier { queuedByEntityKey: queued, attentionByEntityKey: attention, successorByEntityKey: successors, - hasDurabilityFailure: _hasUncertainParkingForActiveStrategy, + hasDurabilityFailure: _hasDurabilityFailureForActiveStrategy, lastError: attentionMessage, clearError: attentionMessage == null, ); @@ -753,8 +754,7 @@ class StrategyOpQueueNotifier extends Notifier { continue; } try { - await _store.remove(storageKey); - _recordsByStorageKey.remove(storageKey); + await _removeRecordByStorageKey(storageKey); _uncertainOversizedParking.remove(storageKey); attention.remove(key); successors.remove(key); @@ -790,7 +790,7 @@ class StrategyOpQueueNotifier extends Notifier { state = state.copyWith( attentionByEntityKey: attention, successorByEntityKey: successors, - hasDurabilityFailure: _hasUncertainParkingForActiveStrategy, + hasDurabilityFailure: _hasDurabilityFailureForActiveStrategy, lastError: errorMessage, clearError: errorMessage == null, ); @@ -874,7 +874,7 @@ class StrategyOpQueueNotifier extends Notifier { state = state.copyWith( queuedByEntityKey: queued, attentionByEntityKey: attention, - hasDurabilityFailure: _hasUncertainParkingForActiveStrategy, + hasDurabilityFailure: _hasDurabilityFailureForActiveStrategy, lastError: persistenceError == null ? attentionMessage : _uncertainOversizedParkingMessage, @@ -1209,8 +1209,25 @@ class StrategyOpQueueNotifier extends Notifier { } Future _putRecord(DurableOutboxRecord record) async { - await _store.put(record); - _recordsByStorageKey[record.storageKey] = record; + try { + await _store.put(record); + _recordsByStorageKey[record.storageKey] = record; + _uncertainDurableRecords.remove(record.storageKey); + } catch (_) { + _uncertainDurableRecords.add(record.storageKey); + rethrow; + } + } + + Future _removeRecordByStorageKey(String storageKey) async { + try { + await _store.remove(storageKey); + _recordsByStorageKey.remove(storageKey); + _uncertainDurableRecords.remove(storageKey); + } catch (_) { + _uncertainDurableRecords.add(storageKey); + rethrow; + } } Future _removeRecordIfCurrent( @@ -1219,8 +1236,7 @@ class StrategyOpQueueNotifier extends Notifier { ) async { final record = _recordForActiveKey(key); if (record == null || record.pending.op.opId != opId) return; - await _store.remove(record.storageKey); - _recordsByStorageKey.remove(record.storageKey); + await _removeRecordByStorageKey(record.storageKey); } Future _serializeWrite(Future Function() action) { @@ -1249,11 +1265,47 @@ class StrategyOpQueueNotifier extends Notifier { strategyPublicId: state.strategyPublicId, ); + bool _hasUncertainDurableRecordFor({ + required String? accountId, + required String? strategyPublicId, + }) { + if (accountId == null || strategyPublicId == null) return false; + final prefix = '${Uri.encodeComponent(accountId)}|' + '${Uri.encodeComponent(strategyPublicId)}|'; + return _uncertainDurableRecords.any((key) => key.startsWith(prefix)); + } + + bool get _hasUncertainDurableRecordForActiveStrategy => + _hasUncertainDurableRecordFor( + accountId: state.accountId, + strategyPublicId: state.strategyPublicId, + ); + + bool _hasDurabilityFailureFor({ + required String? accountId, + required String? strategyPublicId, + }) => + _hasUncertainParkingFor( + accountId: accountId, + strategyPublicId: strategyPublicId, + ) || + _hasUncertainDurableRecordFor( + accountId: accountId, + strategyPublicId: strategyPublicId, + ); + + bool get _hasDurabilityFailureForActiveStrategy => + _hasDurabilityFailureFor( + accountId: state.accountId, + strategyPublicId: state.strategyPublicId, + ); + void _recordPersistenceFailure(Object error, StackTrace stackTrace) { log('Durable outbox persistence failed: $error', name: 'strategy_outbox', error: error, stackTrace: stackTrace); state = state.copyWith( isFlushing: false, + hasDurabilityFailure: _hasDurabilityFailureForActiveStrategy, lastError: 'Cloud work could not be saved to the durable outbox: $error', ); } @@ -1555,6 +1607,9 @@ class StrategyOpQueueNotifier extends Notifier { 'Cloud work could not be verified in the durable outbox. ' 'Nothing was sent.'; } + if (_hasUncertainDurableRecordForActiveStrategy) { + return 'Cloud work could not be verified in the durable outbox.'; + } if (attention.isNotEmpty) { final hasOversizedWork = attention.entries.any((entry) { if (cloudOperationExceedsPolicy(entry.value.pending.op)) return true; diff --git a/test/strategy_op_queue_provider_test.dart b/test/strategy_op_queue_provider_test.dart index 01fb419f..49644946 100644 --- a/test/strategy_op_queue_provider_test.dart +++ b/test/strategy_op_queue_provider_test.dart @@ -875,6 +875,46 @@ void main() { expect(utf8.encode(serialized).length, greaterThan(serialized.length)); }); + test( + 'an initial durable enqueue failure stays unreliable until the exact ' + 'record is rewritten', () async { + final store = _FirstPutFailureStore(); + final container = _cloudQueueContainer( + store: store, + repository: _RecordingAckRepository(), + ); + addTearDown(container.dispose); + final notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + const op = ElementPatchOp( + opId: 'initial-write-failure', + elementPublicId: 'element-1', + pagePublicId: 'page-1', + payload: {'value': 'unsaved'}, + expectedElementRevision: 1, + ); + const key = EntitySyncKey.element('page-1', 'element-1'); + + await notifier.enqueue(op, flushImmediately: false); + + var current = container.read(strategyOpQueueProvider); + expect(current.pending, isEmpty); + expect(current.hasDurabilityFailure, isTrue); + expect(current.needsAttention, isTrue); + expect(current.outboxIsReliable, isFalse); + expect(current.lastError, contains('could not be saved')); + expect(store.values, isEmpty); + + await notifier.enqueue(op, flushImmediately: false); + + current = container.read(strategyOpQueueProvider); + expect(current.queuedByEntityKey, contains(key)); + expect(current.hasDurabilityFailure, isFalse); + expect(current.outboxIsReliable, isTrue); + expect(current.lastError, isNull); + expect(store.load().records.single.pending.op.opId, op.opId); + }); + test('an oversized op is durably parked while independent work lands', () async { final store = MemoryDurableStrategyOutboxStore(); @@ -2297,6 +2337,19 @@ class _OneShotAckFailureStore extends MemoryDurableStrategyOutboxStore { } } +class _FirstPutFailureStore extends MemoryDurableStrategyOutboxStore { + var failNextPut = true; + + @override + Future put(DurableOutboxRecord record) async { + if (failNextPut) { + failNextPut = false; + throw StateError('initial write failed'); + } + await super.put(record); + } +} + class _OversizedParkingFailureStore extends MemoryDurableStrategyOutboxStore { _OversizedParkingFailureStore({ From f2d38a7d8706f17db03c44def2ff6e13a1872e5c Mon Sep 17 00:00:00 2001 From: Dara Adedeji Date: Fri, 4 Sep 2026 00:40:37 -0400 Subject: [PATCH 3/3] fix: park oversized work without overwrite --- .../collab/strategy_op_queue_provider.dart | 200 ++++++++++++++---- test/strategy_op_queue_provider_test.dart | 138 ++++++++---- 2 files changed, 252 insertions(+), 86 deletions(-) diff --git a/lib/providers/collab/strategy_op_queue_provider.dart b/lib/providers/collab/strategy_op_queue_provider.dart index 5519967e..166ab37b 100644 --- a/lib/providers/collab/strategy_op_queue_provider.dart +++ b/lib/providers/collab/strategy_op_queue_provider.dart @@ -125,6 +125,7 @@ class StrategyOpQueueNotifier extends Notifier { final Set _awaitingRemoteAdoption = {}; final Set _uncertainOversizedParking = {}; final Set _uncertainDurableRecords = {}; + final Map _uncertainDurableIntents = {}; String? _uncertainOversizedParkingMessage; Future _writeTail = Future.value(); @@ -192,13 +193,31 @@ class StrategyOpQueueNotifier extends Notifier { case DurableOutboxStatus.queued: case DurableOutboxStatus.inFlight: // An interrupted request is replayed with its original op/client id. - queued[record.entityKey] = intent; + if (cloudOperationExceedsPolicy(record.pending.op)) { + attention[record.entityKey] = intent; + } else { + queued[record.entityKey] = intent; + } case DurableOutboxStatus.paused: - paused[record.entityKey] = intent; + if (cloudOperationExceedsPolicy(record.pending.op)) { + attention[record.entityKey] = intent; + } else { + paused[record.entityKey] = intent; + } case DurableOutboxStatus.attention: attention[record.entityKey] = intent; } } + for (final record in _uncertainDurableIntents.values.where((record) => + record.accountId == accountId && + record.strategyPublicId == strategyPublicId)) { + attention[record.entityKey] = QueuedEntityIntent( + entityKey: record.entityKey, + pending: record.pending, + ); + queued.remove(record.entityKey); + paused.remove(record.entityKey); + } final clientId = matching.firstOrNull?.pending.clientId ?? const Uuid().v4(); final hasDurabilityFailure = _hasDurabilityFailureFor( @@ -217,7 +236,8 @@ class StrategyOpQueueNotifier extends Notifier { durableLoaded: true, hasDurabilityFailure: hasDurabilityFailure, lastError: hasDurabilityFailure - ? _uncertainOversizedParkingMessage + ? (_uncertainOversizedParkingMessage ?? + 'Cloud work could not be verified in the durable outbox.') : _loadedAttentionMessage( loadIssues: state.loadIssues, paused: paused, @@ -361,31 +381,48 @@ class StrategyOpQueueNotifier extends Notifier { // it must never make rejected work eligible for an automatic flush. if (attentionIntent != null) { if (desired == null) continue; - final current = _recordForActiveKey(key); + final storageKey = DurableOutboxRecord.createStorageKey( + accountId: accountId, + strategyPublicId: strategyPublicId, + entityKey: key, + ); + final current = + _recordForActiveKey(key) ?? _uncertainDurableIntents[storageKey]; if (current == null) { throw StateError('Durable attention record is missing for $key.'); } if (_sameIntent(attentionIntent.pending.op, desired)) { - if (successorIntent != null) { + final hasUncertainDurableRecord = + _uncertainDurableRecords.contains(storageKey); + if (successorIntent != null || hasUncertainDurableRecord) { final recoveredOversizedParking = - _uncertainOversizedParking.contains(current.storageKey); + _uncertainOversizedParking.contains(storageKey); + final isOversized = + cloudOperationExceedsPolicy(current.pending.op); await _putRecord(current.copyWith( - status: recoveredOversizedParking + status: isOversized ? DurableOutboxStatus.attention - : current.status, + : (hasUncertainDurableRecord + ? DurableOutboxStatus.queued + : current.status), clearSuccessorPending: true, updatedAt: DateTime.now(), - lastError: recoveredOversizedParking + lastError: isOversized ? cloudOperationTooLargeMessage : null, + clearError: !isOversized, )); if (recoveredOversizedParking) { - _uncertainOversizedParking.remove(current.storageKey); + _uncertainOversizedParking.remove(storageKey); if (_uncertainOversizedParking.isEmpty) { _uncertainOversizedParkingMessage = null; } } successors.remove(key); + if (hasUncertainDurableRecord && !isOversized) { + attention.remove(key); + queued[key] = attentionIntent; + } changed = true; } continue; @@ -544,12 +581,23 @@ class StrategyOpQueueNotifier extends Notifier { final record = _recordFor( key: key, pending: pending, - status: DurableOutboxStatus.queued, + status: cloudOperationExceedsPolicy(pending.op) + ? DurableOutboxStatus.attention + : DurableOutboxStatus.queued, + lastError: cloudOperationExceedsPolicy(pending.op) + ? cloudOperationTooLargeMessage + : null, ); await _putRecord(record); - queued[key] = QueuedEntityIntent(entityKey: key, pending: pending); + final intent = QueuedEntityIntent(entityKey: key, pending: pending); + if (record.status == DurableOutboxStatus.attention) { + attention[key] = intent; + queued.remove(key); + } else { + queued[key] = intent; + attention.remove(key); + } paused.remove(key); - attention.remove(key); changed = true; } } catch (error, stackTrace) { @@ -588,19 +636,32 @@ class StrategyOpQueueNotifier extends Notifier { final queued = Map.from( state.queuedByEntityKey, ); + final attention = Map.from( + state.attentionByEntityKey, + ); try { for (final entry in state.pausedByEntityKey.entries) { final pending = PendingOp( op: entry.value.pending.op, clientId: entry.value.pending.clientId, ); + final isOversized = cloudOperationExceedsPolicy(pending.op); await _putRecord(_recordFor( key: entry.key, pending: pending, - status: DurableOutboxStatus.queued, + status: isOversized + ? DurableOutboxStatus.attention + : DurableOutboxStatus.queued, + lastError: isOversized ? cloudOperationTooLargeMessage : null, )); - queued[entry.key] = + final intent = QueuedEntityIntent(entityKey: entry.key, pending: pending); + if (isOversized) { + attention[entry.key] = intent; + queued.remove(entry.key); + } else { + queued[entry.key] = intent; + } } } catch (error, stackTrace) { _recordPersistenceFailure(error, stackTrace); @@ -609,8 +670,10 @@ class StrategyOpQueueNotifier extends Notifier { state = state.copyWith( queuedByEntityKey: queued, pausedByEntityKey: const {}, + attentionByEntityKey: attention, + hasDurabilityFailure: _hasDurabilityFailureForActiveStrategy, clearError: - state.attentionByEntityKey.isEmpty && state.loadIssues.isEmpty, + attention.isEmpty && state.loadIssues.isEmpty, ); _scheduleFlush(flushImmediately: flushImmediately); }); @@ -642,7 +705,8 @@ class StrategyOpQueueNotifier extends Notifier { final successor = record?.successorPending; final retryOp = successor?.op ?? rejectedOp; final isPayloadPolicyAttention = - record?.lastError == cloudOperationTooLargeMessage; + record?.lastError == cloudOperationTooLargeMessage || + cloudOperationExceedsPolicy(rejectedOp); final retryRevision = record?.latestServerRevision ?? rejectedOp.expectedRevision; if (!isPayloadPolicyAttention && retryRevision == null) continue; @@ -661,11 +725,16 @@ class StrategyOpQueueNotifier extends Notifier { op: rebasedOp, clientId: successor?.clientId ?? rejected.clientId, ); + final isRetryOversized = cloudOperationExceedsPolicy(pending.op); await _putRecord(_recordFor( key: entry.key, pending: pending, - status: DurableOutboxStatus.queued, + status: isRetryOversized + ? DurableOutboxStatus.attention + : DurableOutboxStatus.queued, clearSuccessorPending: true, + lastError: + isRetryOversized ? cloudOperationTooLargeMessage : null, )); _uncertainOversizedParking.remove( DurableOutboxRecord.createStorageKey( @@ -674,11 +743,17 @@ class StrategyOpQueueNotifier extends Notifier { entityKey: entry.key, ), ); - queued[entry.key] = QueuedEntityIntent( + final intent = QueuedEntityIntent( entityKey: entry.key, pending: pending, ); - attention.remove(entry.key); + if (isRetryOversized) { + attention[entry.key] = intent; + queued.remove(entry.key); + } else { + queued[entry.key] = intent; + attention.remove(entry.key); + } successors.remove(entry.key); changed = true; } @@ -746,11 +821,16 @@ class StrategyOpQueueNotifier extends Notifier { ); final hasUncertainParking = _uncertainOversizedParking.contains(storageKey); + final hasUncertainDurableRecord = + _uncertainDurableRecords.contains(storageKey); final record = _recordForActiveKey(key); final hasMatchingAttentionRecord = record != null && - record.status == DurableOutboxStatus.attention && + (record.status == DurableOutboxStatus.attention || + cloudOperationExceedsPolicy(record.pending.op)) && record.pending.op.opId == rejected.pending.op.opId; - if (!hasUncertainParking && !hasMatchingAttentionRecord) { + if (!hasUncertainParking && + !hasUncertainDurableRecord && + !hasMatchingAttentionRecord) { continue; } try { @@ -834,21 +914,11 @@ class StrategyOpQueueNotifier extends Notifier { ); continue; } - try { - await _putRecord(record.copyWith( - status: DurableOutboxStatus.attention, - updatedAt: DateTime.now(), - lastError: cloudOperationTooLargeMessage, - )); - queued.remove(entry.key); - attention[entry.key] = entry.value; - } catch (error, stackTrace) { - persistenceError ??= error; - persistenceStackTrace ??= stackTrace; - queued.remove(entry.key); - attention[entry.key] = entry.value; - _uncertainOversizedParking.add(record.storageKey); - } + // Keep an already-durable queued record untouched. The in-memory view + // and restart loader both classify it as attention, so parking never + // risks destroying the only recoverable copy with an overwrite. + queued.remove(entry.key); + attention[entry.key] = entry.value; } if (persistenceError != null) { log( @@ -1054,20 +1124,32 @@ class StrategyOpQueueNotifier extends Notifier { ), clientId: successor.clientId, ); + final isPromotedOversized = + cloudOperationExceedsPolicy(promoted.op); await _putRecord(current.copyWith( pending: promoted, - status: DurableOutboxStatus.queued, + status: isPromotedOversized + ? DurableOutboxStatus.attention + : DurableOutboxStatus.queued, updatedAt: DateTime.now(), clearSuccessorPending: true, - clearError: true, + lastError: + isPromotedOversized ? cloudOperationTooLargeMessage : null, + clearError: !isPromotedOversized, clearLatestServerRevision: true, )); - queued[sent.entityKey] = QueuedEntityIntent( + final promotedIntent = QueuedEntityIntent( entityKey: sent.entityKey, pending: promoted, ); successors.remove(sent.entityKey); - attention.remove(sent.entityKey); + if (isPromotedOversized) { + queued.remove(sent.entityKey); + attention[sent.entityKey] = promotedIntent; + } else { + queued[sent.entityKey] = promotedIntent; + attention.remove(sent.entityKey); + } } else if (successor != null) { final retained = current.copyWith( status: DurableOutboxStatus.attention, @@ -1213,8 +1295,16 @@ class StrategyOpQueueNotifier extends Notifier { await _store.put(record); _recordsByStorageKey[record.storageKey] = record; _uncertainDurableRecords.remove(record.storageKey); + _uncertainDurableIntents.remove(record.storageKey); } catch (_) { _uncertainDurableRecords.add(record.storageKey); + _uncertainDurableIntents[record.storageKey] = record; + if (cloudOperationExceedsPolicy(record.pending.op)) { + _uncertainOversizedParking.add(record.storageKey); + _uncertainOversizedParkingMessage = + 'Cloud work could not be verified in the durable outbox. ' + 'Nothing was sent.'; + } rethrow; } } @@ -1224,6 +1314,7 @@ class StrategyOpQueueNotifier extends Notifier { await _store.remove(storageKey); _recordsByStorageKey.remove(storageKey); _uncertainDurableRecords.remove(storageKey); + _uncertainDurableIntents.remove(storageKey); } catch (_) { _uncertainDurableRecords.add(storageKey); rethrow; @@ -1303,7 +1394,34 @@ class StrategyOpQueueNotifier extends Notifier { void _recordPersistenceFailure(Object error, StackTrace stackTrace) { log('Durable outbox persistence failed: $error', name: 'strategy_outbox', error: error, stackTrace: stackTrace); + final queued = Map.from( + state.queuedByEntityKey, + ); + final inFlight = Map.from( + state.inFlightByEntityKey, + ); + final paused = Map.from( + state.pausedByEntityKey, + ); + final attention = Map.from( + state.attentionByEntityKey, + ); + for (final record in _uncertainDurableIntents.values.where((record) => + record.accountId == state.accountId && + record.strategyPublicId == state.strategyPublicId)) { + attention[record.entityKey] = QueuedEntityIntent( + entityKey: record.entityKey, + pending: record.pending, + ); + queued.remove(record.entityKey); + inFlight.remove(record.entityKey); + paused.remove(record.entityKey); + } state = state.copyWith( + queuedByEntityKey: queued, + inFlightByEntityKey: inFlight, + pausedByEntityKey: paused, + attentionByEntityKey: attention, isFlushing: false, hasDurabilityFailure: _hasDurabilityFailureForActiveStrategy, lastError: 'Cloud work could not be saved to the durable outbox: $error', diff --git a/test/strategy_op_queue_provider_test.dart b/test/strategy_op_queue_provider_test.dart index 49644946..db75b1b0 100644 --- a/test/strategy_op_queue_provider_test.dart +++ b/test/strategy_op_queue_provider_test.dart @@ -898,7 +898,7 @@ void main() { await notifier.enqueue(op, flushImmediately: false); var current = container.read(strategyOpQueueProvider); - expect(current.pending, isEmpty); + expect(current.attentionByEntityKey[key]!.pending.op, op); expect(current.hasDurabilityFailure, isTrue); expect(current.needsAttention, isTrue); expect(current.outboxIsReliable, isFalse); @@ -1136,52 +1136,100 @@ void main() { expect(store.attentionWrites, 1); }); - for (final dropBeforeThrow in [false, true]) { - test( - 'Use cloud clears uncertain oversized work when the parking ' - '${dropBeforeThrow ? 'record is missing' : 'write failed'}', - () async { - final store = _OversizedParkingFailureStore( - dropBeforeThrow: dropBeforeThrow, - ); - final container = _cloudQueueContainer( - store: store, - repository: _RecordingAckRepository(), - ); - addTearDown(container.dispose); - final notifier = container.read(strategyOpQueueProvider.notifier) - ..setActiveStrategy('strategy-1', accountId: 'account-a'); - const key = EntitySyncKey.element('page-1', 'element-large'); - await notifier.enqueue( - _largeElementPatch( - opId: 'oversized-explicit-discard', - elementId: 'element-large', - ), - flushImmediately: false, - ); - await notifier.flushNow(); + test( + 'a legacy queued oversized record parks without overwrite and survives ' + 'restart', () async { + final store = _OversizedParkingFailureStore(dropBeforeThrow: true); + final oversized = _largeElementPatch( + opId: 'legacy-oversized', + elementId: 'element-large', + ); + const key = EntitySyncKey.element('page-1', 'element-large'); + await store.put(DurableOutboxRecord( + accountId: 'account-a', + strategyPublicId: 'strategy-1', + entityKey: key, + pending: PendingOp(op: oversized, clientId: 'stable-client'), + status: DurableOutboxStatus.queued, + createdAt: DateTime(2026), + updatedAt: DateTime(2026), + )); + final repository = _RecordingAckRepository(); + var container = _cloudQueueContainer( + store: store, + repository: repository, + ); + var notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); - var current = container.read(strategyOpQueueProvider); - expect(current.attentionByEntityKey, contains(key)); - expect(current.hasDurabilityFailure, isTrue); - expect(current.outboxIsReliable, isFalse); - expect( - store.load().records.isEmpty, - dropBeforeThrow, - ); + await notifier.flushNow(); - final discarded = await notifier.discardRejected({key}); + var current = container.read(strategyOpQueueProvider); + expect(current.attentionByEntityKey, contains(key)); + expect(current.queuedByEntityKey, isEmpty); + expect(repository.calls, isEmpty); + expect(store.attentionWrites, 0); + expect(store.load().records.single.status, DurableOutboxStatus.queued); - expect(discarded, {key}); - current = container.read(strategyOpQueueProvider); - expect(current.attentionByEntityKey, isEmpty); - expect(current.hasDurabilityFailure, isFalse); - expect(current.outboxIsReliable, isTrue); - expect(current.lastError, isNull); - expect(store.load().records, isEmpty); - expect(store.removalAttempts, 1); - }); - } + container.dispose(); + container = _cloudQueueContainer( + store: store, + repository: repository, + ); + addTearDown(container.dispose); + notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + await notifier.flushNow(); + + current = container.read(strategyOpQueueProvider); + expect(current.attentionByEntityKey, contains(key)); + expect(current.lastError, cloudOperationTooLargeMessage); + expect(repository.calls, isEmpty); + expect(store.attentionWrites, 0); + expect(store.load().records.single.pending.op.opId, oversized.opId); + + final discarded = await notifier.discardRejected({key}); + expect(discarded, {key}); + expect(store.load().records, isEmpty); + }); + + test('Use cloud clears an uncertain oversized first-write failure', + () async { + final store = _OversizedParkingFailureStore(dropBeforeThrow: true); + final container = _cloudQueueContainer( + store: store, + repository: _RecordingAckRepository(), + ); + addTearDown(container.dispose); + final notifier = container.read(strategyOpQueueProvider.notifier) + ..setActiveStrategy('strategy-1', accountId: 'account-a'); + const key = EntitySyncKey.element('page-1', 'element-large'); + await notifier.enqueue( + _largeElementPatch( + opId: 'oversized-explicit-discard', + elementId: 'element-large', + ), + flushImmediately: false, + ); + await notifier.flushNow(); + + var current = container.read(strategyOpQueueProvider); + expect(current.attentionByEntityKey, contains(key)); + expect(current.hasDurabilityFailure, isTrue); + expect(current.outboxIsReliable, isFalse); + expect(store.load().records, isEmpty); + + final discarded = await notifier.discardRejected({key}); + + expect(discarded, {key}); + current = container.read(strategyOpQueueProvider); + expect(current.attentionByEntityKey, isEmpty); + expect(current.hasDurabilityFailure, isFalse); + expect(current.outboxIsReliable, isTrue); + expect(current.lastError, isNull); + expect(store.load().records, isEmpty); + expect(store.removalAttempts, 1); + }); test('Use cloud remains fail-closed when uncertain removal fails', () async { @@ -1211,7 +1259,7 @@ void main() { expect(current.hasDurabilityFailure, isTrue); expect(current.outboxIsReliable, isFalse); expect(current.lastError, contains('could not be removed')); - expect(store.load().records.single.status, DurableOutboxStatus.queued); + expect(store.load().records, isEmpty); expect(store.removalAttempts, 1); });