fix(zero-swift): harden the client against crashes, data loss, and JS divergences (#38)

Amp-Thread-ID: https://ampcode.com/threads/T-01a17664-8fd0-76be-8fd5-75ac6e7aa0d8

Author: tmm

.changeset/harden-client.md

@@ -0,0 +1,5 @@
+---
+'zero-swift': patch
+---
+
+A mutate from inside a mutator runs after it, as in JS, instead of sharing its mutation ID and dropping it from the push. A poke's `desiredQueriesPatches` apply in the server's order. Closing a connecting socket aborts its handshake, so a timed-out attempt no longer leaks a socket. App errors drop falsy `details`, as JS does.

.changeset/harden-queries.md

@@ -0,0 +1,5 @@
+---
+'zero-swift': patch
+---
+
+A flipped `whereExists` over a subquery with a `limit` no longer crashes the app: `materialize` and `liveQuery` throw "Partitioned take does not allow unpartitioned fetches", and `run` rejects with it, as in JS. View listeners are notified as in JS when one of them changes the client: a `mutate` from a listener runs once the notification is done, so listeners after it no longer end on stale rows and other views see every change; a listener removed during a notification is skipped, one added is called in the same pass, and destroying a view from its own listener still lets its other listeners hear that change. A change a view can't take where JS throws, such as two rows for a `one` relationship or a new row reaching a flipped `whereExists` over a `limit`, no longer crashes the app either: as in JS, the poke fails and the connection goes to the `error` state, or the mutation's `client` outcome is the error, and later mutations, `materialize`, and `run` fail with it.

.changeset/harden-storage.md

@@ -0,0 +1,5 @@
+---
+'zero-swift': patch
+---
+
+Storage hardening. Collecting old databases no longer deletes one a client opens while the collection runs, files of a newer layout or format, or files that aren't Zero databases. A client of a group that is behind another no longer writes an older snapshot over a newer one. `zero.delete()` and `dropAllDatabases` also drop the memory databases clients fell back to after their SQLite store failed. Stored data that doesn't decode now makes the client delete the database and start a new one, calling `onClientStateNotFound` if set, instead of running on memory at every launch. Writes kept because the database was busy are written by `flush()` and `close()`, or reported as a storage failure, instead of being lost. Dropping a database twice no longer deletes the new one a replaced client started.

.changeset/harden-untrusted-input.md

@@ -0,0 +1,5 @@
+---
+'zero-swift': patch
+---
+
+Bad server data, storage contents, and options no longer crash the app. A poke whose last mutation ID isn't a safe integer, or whose row update moves a row off its primary key, is rejected and the client disconnects; stored mutations with invalid IDs are dropped at launch. Huge or non-finite timer delays fire at once, as `setTimeout` does. A query built with a negative or out-of-range `limit`, a `limit` or `orderBy` in a junction relationship, or `start` or `related` inside `exists` throws JS's message when materialized, run, or preloaded. An `Int` column holding -2^63 fails to read instead of trapping, and `minPageSize` is clamped.

packages/zero-conformance/src/client/app.ts

@@ -34,6 +34,28 @@ export const queries = defineQueries({
       zql.issue.where('id', args).related('owner').one(),
     ),
     firstTwo: defineQuery(() => zql.issue.orderBy('title', 'asc').limit(2)),
+    // A flipped EXISTS over a limited subquery: fetching it throws
+    // `Partitioned take does not allow unpartitioned fetches`.
+    // Hydrates, but adding an issue then fetches the limited flipped
+    // subquery without a constraint, which throws from the push.
+    withAuthoredComments: defineQuery(() =>
+      zql.issue
+        .whereExists('comments', q =>
+          q.whereExists('author', a => a.limit(3), {flip: true}),
+        )
+        .orderBy('id', 'asc'),
+    ),
+    // `oneComment` is a `one` relationship that several comments can match.
+    withOneComment: defineQuery(() =>
+      zql.issue.related('oneComment').orderBy('id', 'asc'),
+    ),
+    withFirstComment: defineQuery(() =>
+      zql.issue
+        .whereExists('comments', q => q.orderBy('id', 'asc').limit(1), {
+          flip: true,
+        })
+        .orderBy('id', 'asc'),
+    ),
   },
   users: {
     all: defineQuery(() => zql.user.orderBy('id', 'asc')),
@@ -118,8 +140,46 @@ const issue = {
   ),
 };
 
+type CRUDOp = {
+  op: 'insert' | 'upsert' | 'update' | 'delete';
+  table: string;
+  value: JSONObject;
+};
+
+const harden = {
+  /**
+   * Runs `ops` (CRUD operations on any table, values as given), then throws
+   * `fail` if set, or reads the `read` tables back through the transaction
+   * and returns them with the mutation ID, or returns `return`. With
+   * `failOnRebase` it throws before writing when replayed.
+   */
+  ops: defineMutator(async ({tx, args}: {tx: Tx; args: JSONObject}) => {
+    if (args['failOnRebase'] === true && tx.reason === 'rebase') {
+      throw new ApplicationError('fails on rebase');
+    }
+    for (const o of (args['ops'] ?? []) as CRUDOp[]) {
+      await tx.mutate[o.table][o.op](o.value);
+    }
+    if (typeof args['fail'] === 'string') {
+      throw new ApplicationError(args['fail'], {details: args['details']});
+    }
+    const read = args['read'] as string[] | undefined;
+    if (read) {
+      const out: Record<string, unknown> = {};
+      for (const t of read) {
+        out[t] = await tx.run(
+          (zql as unknown as Record<string, typeof zql.issue>)[t],
+        );
+      }
+      return {mutationID: tx.mutationID, read: out} as ReadonlyJSONValue;
+    }
+    return args['return'];
+  }),
+};
+
 export const mutators = defineMutators({
   issue,
+  harden,
   comment: {
     add: defineMutator(async ({tx, args}: {tx: Tx; args: JSONObject}) => {
       await tx.mutate.comment.insert({

packages/zero-conformance/src/client/harden-connection.test.ts

@@ -0,0 +1,26 @@
+import {test} from 'vitest';
+import {emitVectors} from '../vectors.ts';
+import {
+  runScenario,
+  START_TIME,
+  type Scenario,
+  type StepResult,
+} from './harness.ts';
+import {hardenConnectionScenarios} from './scenarios/harden-connection.ts';
+
+type Case = Scenario & {expect: StepResult[]};
+
+test('client hardening scenarios: connection', {timeout: 600_000}, async () => {
+  const cases: Case[] = [];
+  for (const scenario of hardenConnectionScenarios) {
+    cases.push({...scenario, expect: await runScenario(scenario)});
+  }
+  emitVectors(
+    'client-harden-connection',
+    'client-scenarios',
+    'Hardening scenarios for the connection and the sync protocol, run against the JS zero-client like vectors/client.json. See src/client/scenarios/harden-connection.ts.',
+    cases,
+    {startTime: START_TIME},
+    {compact: true},
+  );
+});

packages/zero-conformance/src/client/harden-mutations.test.ts

@@ -0,0 +1,30 @@
+import {test} from 'vitest';
+import {emitVectors} from '../vectors.ts';
+import {
+  runScenario,
+  START_TIME,
+  type Scenario,
+  type StepResult,
+} from './harness.ts';
+import {hardenMutationScenarios} from './scenarios/harden-mutations.ts';
+
+type Case = Scenario & {expect: StepResult[]};
+
+test(
+  'client hardening scenarios for mutations',
+  {timeout: 600_000},
+  async () => {
+    const cases: Case[] = [];
+    for (const scenario of hardenMutationScenarios) {
+      cases.push({...scenario, expect: await runScenario(scenario)});
+    }
+    emitVectors(
+      'client-harden-mutations',
+      'client-scenarios',
+      'Hardening scenarios for mutations, run against the JS zero-client like client.json (see src/client/harness.ts and src/client/scenarios/harden-mutations.ts).',
+      cases,
+      {startTime: START_TIME},
+      {compact: true},
+    );
+  },
+);

packages/zero-conformance/src/client/harden-queries.test.ts

@@ -0,0 +1,29 @@
+// Hardening client scenarios for views and queries, kept apart from
+// `client.json`. `swift/Tests/ConformanceTests/HardenQueryTests.swift`
+// replays them.
+import {test} from 'vitest';
+import {emitVectors} from '../vectors.ts';
+import {
+  runScenario,
+  START_TIME,
+  type Scenario,
+  type StepResult,
+} from './harness.ts';
+import {hardenQueryScenarios} from './scenarios/harden-queries.ts';
+
+type Case = Scenario & {expect: StepResult[]};
+
+test('harden query scenarios', {timeout: 600_000}, async () => {
+  const cases: Case[] = [];
+  for (const scenario of hardenQueryScenarios) {
+    cases.push({...scenario, expect: await runScenario(scenario)});
+  }
+  emitVectors(
+    'harden-queries',
+    'client-scenarios',
+    'Hardening client scenarios: re-entrant listeners, odd TTLs, query change throttling, pokes that leave views unchanged. See src/client/harness.ts.',
+    cases,
+    {startTime: START_TIME},
+    {compact: true},
+  );
+});

packages/zero-conformance/src/client/harness.ts

@@ -81,6 +81,10 @@ export type Step =
       bytes?: number[];
       socket?: number;
     }
+  /** A text frame of exactly `text`, which need not be JSON. */
+  | {op: 'receiveText'; text: string; socket?: number}
+  /** A binary frame of exactly `bytes`, the message type byte included. */
+  | {op: 'receiveBinary'; bytes: number[]; socket?: number}
   /** The socket closes from the server or network side. */
   | {
       op: 'socketClose';
@@ -95,6 +99,15 @@ export type Step =
    * tab does (`document.visibilityState`).
    */
   | {op: 'visibility'; state: 'visible' | 'hidden'}
+  /**
+   * Once `connection.state` next becomes `state`, runs `then` (a `connect`
+   * or `close` step) synchronously from inside the state listener.
+   */
+  | {
+      op: 'onState';
+      state: string;
+      then: {op: 'connect'; auth?: string} | {op: 'close'};
+    }
   /** `zero.connection.connect()`. */
   | {op: 'connect'; auth?: string}
   | {op: 'close'}
@@ -113,7 +126,22 @@ export type Step =
       advance?: number;
       options?: ScenarioOptions;
       mutators?: 'all' | 'issues';
-    };
+    }
+  /**
+   * Adds the listener `listener` to `view`. Its calls are reported in
+   * `listeners`. Once it has been called more than `after` times (default
+   * 1: the call `addListener` makes), its next call performs `actions`,
+   * from inside the notification, once.
+   */
+  | {
+      op: 'listen';
+      view: string;
+      listener: string;
+      after?: number;
+      actions?: Step[];
+    }
+  /** Removes a listener added by `listen`. */
+  | {op: 'unlisten'; listener: string};
 
 type Args = ReadonlyJSONValue | undefined;
 
@@ -172,6 +200,14 @@ export type StepResult = {
   preloads?: Record<string, {cached: Settled; complete: Settled}>;
   /** `onUpdateNeeded` and `onClientStateNotFound` calls. */
   events?: ReadonlyJSONValue[];
+  /**
+   * Listeners added by `listen`, in order: the data and result type of
+   * each call during the step.
+   */
+  listeners?: Record<
+    string,
+    {calls: {data: ReadonlyJSONValue; resultType: string}[]; threw?: string}
+  >;
 };
 
 class CloseEventPolyfill extends Event {
@@ -355,6 +391,14 @@ export async function runScenario(scenario: Scenario): Promise<StepResult[]> {
   >();
   const sentCounts: number[] = [];
   const closedBefore: boolean[] = [];
+  const listeners = new Map<
+    string,
+    {
+      remove: () => void;
+      calls: {data: ReadonlyJSONValue; resultType: string}[];
+      threw?: string | undefined;
+    }
+  >();
 
   const targetSocket = (index: number | undefined) => {
     const socket = index === undefined ? sockets.at(-1) : sockets[index];
@@ -474,6 +518,18 @@ export async function runScenario(scenario: Scenario): Promise<StepResult[]> {
         );
         return;
       }
+      case 'receiveText':
+        targetSocket(step.socket).dispatchEvent(
+          new MessageEvent('message', {data: step.text}),
+        );
+        return;
+      case 'receiveBinary':
+        targetSocket(step.socket).dispatchEvent(
+          new MessageEvent('message', {
+            data: Uint8Array.from(step.bytes).buffer,
+          }),
+        );
+        return;
       case 'socketClose':
         targetSocket(step.socket).dispatchEvent(
           new CloseEvent('close', {
@@ -489,6 +545,21 @@ export async function runScenario(scenario: Scenario): Promise<StepResult[]> {
       case 'visibility':
         document.setVisibility(step.state);
         return;
+      case 'onState': {
+        const {then} = step;
+        const unsubscribe = z.connection.state.subscribe(state => {
+          if (state.name !== step.state) return;
+          unsubscribe();
+          if (then.op === 'close') {
+            void z.close();
+          } else {
+            void z.connection.connect(
+              then.auth === undefined ? undefined : {auth: then.auth},
+            );
+          }
+        });
+        return;
+      }
       case 'connect':
         void z.connection.connect(
           step.auth === undefined ? undefined : {auth: step.auth},
@@ -497,6 +568,36 @@ export async function runScenario(scenario: Scenario): Promise<StepResult[]> {
       case 'close':
         void z.close();
         return;
+      case 'listen': {
+        const view = views.get(step.view)?.view;
+        if (!view) throw new Error(`No view ${step.view}`);
+        const after = step.after ?? 1;
+        let pending = step.actions;
+        const entry: {
+          remove: () => void;
+          calls: {data: ReadonlyJSONValue; resultType: string}[];
+          threw?: string | undefined;
+        } = {remove: () => {}, calls: []};
+        listeners.set(step.listener, entry);
+        let count = 0;
+        entry.remove = view.addListener((data, resultType) => {
+          entry.calls.push({data: json(data), resultType});
+          if (++count > after && pending) {
+            const actions = pending;
+            pending = undefined;
+            for (const action of actions) {
+              // The synchronous part of each step runs inside the listener.
+              perform(action).catch(e => {
+                entry.threw ??= errorMessage(e);
+              });
+            }
+          }
+        });
+        return;
+      }
+      case 'unlisten':
+        listeners.get(step.listener)?.remove();
+        return;
       case 'restart': {
         await z.close();
         await settle();
@@ -517,6 +618,7 @@ export async function runScenario(scenario: Scenario): Promise<StepResult[]> {
         placeholders.set(clientGroupID, '$clientGroupID');
         ids.set('$clientID', clientID);
         views.clear();
+        listeners.clear();
         mutations.clear();
         runs.clear();
         preloads.clear();
@@ -583,6 +685,15 @@ export async function runScenario(scenario: Scenario): Promise<StepResult[]> {
       if (events.length > 0) {
         result.events = events.splice(0);
       }
+      if (listeners.size > 0) {
+        result.listeners = Object.fromEntries(
+          Array.from(listeners, ([k, v]) => [
+            k,
+            {calls: v.calls.splice(0), ...(v.threw ? {threw: v.threw} : {})},
+          ]),
+        );
+        for (const v of listeners.values()) v.threw = undefined;
+      }
       results.push(result as StepResult);
     }
   } finally {