Fix forward compatibility by quarantining messages

Author: steida

.changeset/three-buses-play.md

@@ -0,0 +1,7 @@
+---
+"@evolu/common": patch
+---
+
+Fix forward compatibility by quarantining messages with unknown schema
+
+Messages with unknown tables or columns are now stored in `evolu_message_quarantine` table instead of being discarded. This fixes an issue where apps had to be updated to receive messages from newer versions. The quarantine table is queryable via `createQuery` and quarantined messages are automatically applied when the schema is updated.

packages/common/src/local-first/Db.ts

@@ -32,6 +32,7 @@ import {
 } from "../Worker.js";
 import {
   AppOwner,
+  AppOwnerDep,
   createAppOwner,
   createOwnerSecret,
   createOwnerWebSocketTransport,
@@ -63,6 +64,7 @@ import {
   createSync,
   SyncDep,
   SyncOwner,
+  tryApplyQuarantinedMessages,
 } from "./Sync.js";
 import {
   Timestamp,
@@ -320,10 +322,6 @@ type DbWorkerDeps = Omit<
   SqliteDep &
   SyncDep;
 
-export interface AppOwnerDep {
-  readonly appOwner: AppOwner;
-}
-
 export interface PostMessageDep {
   readonly postMessage: (message: DbWorkerOutput) => void;
 }
@@ -372,9 +370,7 @@ const createDbWorkerDeps = async (
     const dbSchema = getDbSchema(deps)();
     if (!dbSchema.ok) return dbSchema;
 
-    const dbIsInitialized = dbSchema.value.tables.some(
-      (table) => table.name === "evolu_version",
-    );
+    const dbIsInitialized = "evolu_version" in dbSchema.value.tables;
 
     let appOwner: AppOwner;
     let clock: Clock;
@@ -426,15 +422,21 @@ const createDbWorkerDeps = async (
       if (!result.ok) return result;
     }
 
-    const result1 = ensureDbSchema(deps)(initMessage.dbSchema, dbSchema.value);
-    if (!result1.ok) return result1;
+    {
+      const result = ensureDbSchema(deps)(initMessage.dbSchema, dbSchema.value);
+      if (!result.ok) return result;
+    }
+
+    {
+      const result = ensureMessageQuarantineTable(deps);
+      if (!result.ok) return result;
+    }
 
     const sync = createSync({
       ...deps,
       clock,
       symmetricCrypto: createSymmetricCrypto(platformDeps),
       timestampConfig: initMessage.config,
-      postMessage,
       dbSchema: initMessage.dbSchema,
     })({
       appOwner,
@@ -448,6 +450,14 @@ const createDbWorkerDeps = async (
     });
     if (!sync.ok) return sync;
 
+    {
+      const result = tryApplyQuarantinedMessages({
+        ...deps,
+        dbSchema: initMessage.dbSchema,
+      })();
+      if (!result.ok) return result;
+    }
+
     sync.value.useOwner(true, appOwner);
 
     return ok({
@@ -557,6 +567,41 @@ const initializeDb =
     return ok();
   };
 
+/**
+ * Ensures the quarantine table exists for storing messages with unknown schema.
+ *
+ * When a device receives sync messages containing tables or columns that don't
+ * exist in its current schema (e.g., from a newer app version), those messages
+ * are stored here instead of being discarded. This enables forward
+ * compatibility:
+ *
+ * 1. Unknown data is preserved and can be applied when the app is updated
+ * 2. Messages are still propagated to other devices that may understand them
+ * 3. Partial messages work - known columns go to app tables, unknown to quarantine
+ *
+ * The `union all` query in `readDbChange` combines `evolu_history` and this
+ * table, ensuring all data (known and unknown) is included when syncing to
+ * other devices.
+ */
+const ensureMessageQuarantineTable = (
+  deps: SqliteDep,
+): Result<void, SqliteError> => {
+  const result = deps.sqlite.exec(sql`
+    create table if not exists evolu_message_quarantine (
+      "ownerId" blob not null,
+      "timestamp" blob not null,
+      "table" text not null,
+      "id" blob not null,
+      "column" text not null,
+      "value" any,
+      primary key ("ownerId", "timestamp", "table", "id", "column")
+    )
+    strict;
+  `);
+  if (!result.ok) return result;
+  return ok();
+};
+
 const handlers: Omit<MessageHandlers<DbWorkerInput, DbWorkerDeps>, "init"> = {
   getAppOwner: (deps) => () => {
     deps.postMessage({
@@ -628,19 +673,19 @@ const handlers: Omit<MessageHandlers<DbWorkerInput, DbWorkerDeps>, "init"> = {
   },
 
   reset: (deps) => (message) => {
-    const resetResult = deps.sqlite.transaction(() => {
+    const result = deps.sqlite.transaction(() => {
       const dbSchema = getDbSchema(deps)();
       if (!dbSchema.ok) return dbSchema;
 
-      for (const table of dbSchema.value.tables) {
+      for (const tableName in dbSchema.value.tables) {
         /**
          * The dropped table is completely removed from the database schema and
          * the disk file. The table can not be recovered. All indices and
          * triggers associated with the table are also deleted.
          * https://sqlite.org/lang_droptable.html
          */
         const result = deps.sqlite.exec(sql`
-          drop table ${sql.identifier(table.name)};
+          drop table ${sql.identifier(tableName)};
         `);
         if (!result.ok) return result;
       }
@@ -659,8 +704,8 @@ const handlers: Omit<MessageHandlers<DbWorkerInput, DbWorkerDeps>, "init"> = {
       return ok();
     });
 
-    if (!resetResult.ok) {
-      deps.postMessage({ type: "onError", error: resetResult.error });
+    if (!result.ok) {
+      deps.postMessage({ type: "onError", error: result.error });
       return;
     }
 

packages/common/src/local-first/Evolu.ts

@@ -101,6 +101,9 @@ export interface EvoluConfig extends Partial<DbConfig> {
    * URL to reload browser tabs after reset or restore.
    *
    * The default value is `/`.
+   *
+   * Note: This option will be moved to web platform deps in the next major
+   * version.
    */
   readonly reloadUrl?: string;
 }
@@ -836,15 +839,15 @@ const createEvoluInstance =
 
       subscribeQuery: (query) => (listener) => {
         // Call the listener only if the result has been changed.
-        let previousResult: unknown = null;
+        let previousRows: unknown = null;
         const unsubscribe = subscribedQueries.subscribe(query)(() => {
-          const result = evolu.getQueryRows(query);
-          if (previousResult === result) return;
-          previousResult = result;
+          const rows = evolu.getQueryRows(query);
+          if (previousRows === rows) return;
+          previousRows = rows;
           listener();
         });
         return () => {
-          previousResult = null;
+          previousRows = null;
           unsubscribe();
         };
       },

packages/common/src/local-first/Owner.ts

@@ -188,6 +188,10 @@ export interface AppOwner extends Owner {
   readonly mnemonic?: Mnemonic | null;
 }
 
+export interface AppOwnerDep {
+  readonly appOwner: AppOwner;
+}
+
 /** Creates an {@link AppOwner} from an {@link OwnerSecret}. */
 export const createAppOwner = (secret: OwnerSecret): AppOwner => ({
   ...createOwner(secret),

packages/common/src/local-first/Schema.ts

@@ -1,5 +1,10 @@
 import * as Kysely from "kysely";
-import { mapObject, objectToEntries, ReadonlyRecord } from "../Object.js";
+import {
+  createRecord,
+  getProperty,
+  mapObject,
+  ReadonlyRecord,
+} from "../Object.js";
 import { ok, Result } from "../Result.js";
 import {
   SafeSql,
@@ -30,6 +35,8 @@ import {
   omit,
   optional,
   OptionalType,
+  record,
+  set,
   String,
   TableId,
   Type,
@@ -42,6 +49,7 @@ import { AppOwner, OwnerId } from "./Owner.js";
 import { Query, Row } from "./Query.js";
 import type { CrdtMessage, DbChange } from "./Storage.js";
 import { Timestamp, TimestampBytes } from "./Timestamp.js";
+import { readonly } from "../Function.js";
 
 /**
  * Defines the schema of an Evolu database.
@@ -169,12 +177,10 @@ export const evoluSchemaToDbSchema = (
   schema: EvoluSchema,
   indexesConfig?: IndexesConfig,
 ): DbSchema => {
-  const tables = objectToEntries(schema).map(([tableName, table]) => ({
-    name: tableName,
-    columns: objectToEntries(table)
-      .filter(([k]) => k !== "id")
-      .map(([k]) => k),
-  }));
+  const tables = mapObject(
+    schema,
+    (table) => new Set(Object.keys(table).filter((k) => k !== "id")),
+  );
 
   const indexes = indexesConfig
     ? indexesConfig(createIndex).map(
@@ -206,6 +212,13 @@ export type CreateQuery<S extends EvoluSchema> = <R extends Row>(
             readonly column: string;
             readonly value: SqliteValue;
           };
+          readonly evolu_message_quarantine: {
+            readonly timestamp: TimestampBytes;
+            readonly table: string;
+            readonly id: IdBytes;
+            readonly column: string;
+            readonly value: SqliteValue;
+          };
         }
       >,
       "selectFrom" | "fn" | "with" | "withRecursive"
@@ -231,7 +244,11 @@ export const SystemColumns = object({
 });
 export type SystemColumns = typeof SystemColumns.Type;
 
-export const systemColumns = Object.keys(SystemColumns.props);
+export const systemColumns = readonly(
+  new Set(Object.keys(SystemColumns.props)),
+);
+
+export const systemColumnsWithId = readonly([...systemColumns, "id"]);
 
 export type MutationKind = "insert" | "update" | "upsert";
 
@@ -443,21 +460,17 @@ export type InferColumnErrors<
   >;
 }[keyof MutationMapping<T, M>];
 
-export const DbTable = object({
-  name: String,
-  columns: array(String),
-});
-export type DbTable = typeof DbTable.Type;
-
 export const DbIndex = object({ name: String, sql: String });
 export type DbIndex = typeof DbIndex.Type;
 
 export const DbSchema = object({
-  tables: array(DbTable),
+  tables: record(String, set(String)),
   indexes: array(DbIndex),
 });
 export type DbSchema = typeof DbSchema.Type;
 
+// TODO: Use a ref and update dbSchema on hot reloading to support
+// development workflows where schema changes without full app restart.
 export interface DbSchemaDep {
   readonly dbSchema: DbSchema;
 }
@@ -469,7 +482,7 @@ export const getDbSchema =
     DbSchema,
     SqliteError
   > => {
-    const map = new Map<string, Array<string>>();
+    const tables = createRecord<string, Set<string>>();
 
     const tableAndColumnInfoRows = deps.sqlite.exec(sql`
       select
@@ -487,12 +500,9 @@ export const getDbSchema =
         tableName: string;
         columnName: string;
       };
-      if (!map.has(tableName)) map.set(tableName, []);
-      map.get(tableName)?.push(columnName);
+      (tables[tableName] ??= new Set()).add(columnName);
     });
 
-    const tables = Array.from(map, ([name, columns]) => ({ name, columns }));
-
     const indexesRows = deps.sqlite.exec(
       allIndexes
         ? sql`
@@ -546,23 +556,19 @@ export const ensureDbSchema =
       currentSchema = dbSchema.value;
     }
 
-    newSchema.tables.forEach((newTable) => {
-      const currentTable = currentSchema.tables.find(
-        (t) => t.name === newTable.name,
-      );
-      if (!currentTable) {
-        queries.push(createAppTable(newTable));
+    for (const [tableName, newColumns] of Object.entries(newSchema.tables)) {
+      const currentColumns = getProperty(currentSchema.tables, tableName);
+      if (!currentColumns) {
+        queries.push(createAppTable(tableName, newColumns));
       } else {
-        newTable.columns
-          .filter((newColumn) => !currentTable.columns.includes(newColumn))
-          .forEach((newColumn) => {
-            queries.push(sql`
-              alter table ${sql.identifier(newTable.name)}
-              add column ${sql.identifier(newColumn)} blob;
-            `);
-          });
+        for (const newColumn of newColumns.difference(currentColumns)) {
+          queries.push(sql`
+            alter table ${sql.identifier(tableName)}
+            add column ${sql.identifier(newColumn)} any;
+          `);
+        }
       }
-    });
+    }
 
     // Remove current indexes that are not in the newSchema.
     currentSchema.indexes
@@ -595,19 +601,19 @@ export const ensureDbSchema =
     return ok();
   };
 
-const createAppTable = (table: DbTable) => sql`
-  create table ${sql.identifier(table.name)} (
+const createAppTable = (tableName: string, columns: ReadonlySet<string>) => sql`
+  create table ${sql.identifier(tableName)} (
     "id" text,
     ${sql.raw(
-      `${systemColumns
-        .concat(table.columns)
-        .filter((c) => c !== "id")
+      `${[...systemColumns, ...columns]
         // With strict tables and any type, data is preserved exactly as received
         // without any type affinity coercion. This allows storing any data type
         // while maintaining strict null enforcement for primary key columns.
+        // TODO: Use proper SQLite types for system columns (text for createdAt,
+        // updatedAt, ownerId, integer for isDeleted) instead of "any".
         .map((name) => `${sql.identifier(name).sql} any`)
-        .join(", ")}`,
-    )},
+        .join(", ")}, `,
+    )}
     primary key ("ownerId", "id")
   )
   without rowid, strict;