Revert "update client_view to make progress if many tabs sync at once"

tantaman · tantaman · commit 7e2bf4193326 · 2023-12-01T16:13:20.000-05:00
This reverts commit 9f1cb55.
diff --git a/client/src/app.tsx b/client/src/app.tsx
@@ -59,12 +59,17 @@ const App = ({rep, undoManager}: AppProps) => {
     'NOT_RECEIVED_FROM_SERVER',
   );
   const partialSyncComplete = partialSync === 'COMPLETE';
-  useEffect(() => {
+  function pull() {
     console.log('partialSync', partialSync);
     if (!partialSyncComplete) {
-      rep.pull();
+      if (document.hidden) {
+        setTimeout(pull, 1000);
+      } else {
+        rep.pull();
+      }
     }
-  }, [rep, partialSync, partialSyncComplete]);
+  }
+  useEffect(pull, [rep, partialSync, partialSyncComplete]);
 
   useEffect(() => {
     const ev = new EventSource(`/api/replicache/poke?channel=poke`);
diff --git a/server/src/pull/cvr.ts b/server/src/pull/cvr.ts
@@ -7,29 +7,6 @@ export type CVR = {
   order: number;
 };
 
-const VERSION_TREE_SQL = /*sql*/ `WITH RECURSIVE "version_tree" AS (
-  SELECT
-    "client_group_id",
-    "version",
-    "parent_version"
-  FROM
-    "client_view"
-  WHERE
-    "client_group_id" = $1 AND
-    "version" = $2
-
-  UNION ALL
-
-  SELECT
-    cv."client_group_id",
-    cv."version",
-    cv."parent_version"
-  FROM
-    "client_view" cv
-    JOIN "version_tree" vt ON vt."parent_version" = cv."version"
-    WHERE cv."client_group_id" = vt."client_group_id"
-)`;
-
 export async function getCVR(
   executor: Executor,
   clientGroupID: string,
@@ -49,68 +26,61 @@ export async function getCVR(
   };
 }
 
-export function putCVR(executor: Executor, cvr: CVR, parentVersion: number) {
+export function putCVR(executor: Executor, cvr: CVR) {
   return executor(
     /*sql*/ `INSERT INTO client_view (
       "client_group_id",
       "client_version",
-      "version",
-      "parent_version"
-    ) VALUES ($1, $2, $3, $4) ON CONFLICT ("client_group_id", "version") DO UPDATE SET
+      "version"
+    ) VALUES ($1, $2, $3) ON CONFLICT ("client_group_id", "version") DO UPDATE SET
       client_version = excluded.client_version
     `,
-    [cvr.clientGroupID, cvr.clientVersion, cvr.order, parentVersion],
+    [cvr.clientGroupID, cvr.clientVersion, cvr.order],
   );
 }
 
 /**
  * Find all rows in a table that are not in our CVR table.
+ * - either the id is not present
+ * - or the version is different
  *
- * Multiple tabs could be trying to sync at the same time with different CVRs.
- * We path back through our CVR tree to ensure we send the correct data to each tab.
- *
- * We can prune branches of the client view tree that are no longer needed.
- *
- * If we receive a request for a pruned branch we'd track back to find the
- * greatest common ancestor.
+ * We could do cursoring to speed this along.
+ * I.e., For a given pull sequence, we can keep track of the last id and version.
+ * A "reset pull" brings cursor back to 0 to deal with privacy or query changes.
  *
- * We could & should also compact:
- * 1. Deletion entries in the client_view_entry table
- * 2. Old entity_versions in the client_view_entry table
+ * But also, if the frontend drives the queries then we'll never actually be fetching
+ * so much as the frontend queries will be reasonably sized. The frontend queries would also
+ * be cursored.
  *
- * Compared to the old approach, the client_view_entry table now grows
- * without bound (and requires compaction) since we store a record for each
- * entity_version.
+ * E.g., `SELECT * FROM issue WHERE modified > x OR (modified = y AND id > z) ORDER BY modified, id LIMIT 100`
+ * Cursored on modified and id.
  *
- * We wouldn't have to introduce all of this machinery if
- * tabs coordinated sync.
+ * This means our compare against the CVR would not be a full table scan of `issue`.
  */
 export function findUnsentItems(
   executor: Executor,
   table: keyof typeof TableOrdinal,
   clientGroupID: string,
-  version: number,
+  order: number,
   limit: number,
 ) {
   // The query used below is the fastest.
   // Other query forms that were tried:
   // 1. 10x slower: SELECT * FROM table WHERE NOT EXISTS (SELECT 1 FROM client_view_entry ...)
   // 2. 2x slower: SELECT id, version FROM table EXCEPT SELECT entity_id, entity_version FROM client_view_entry ...
   // 3. SELECT * FROM table LEFT JOIN client_view_entry ...
-  const sql = /*sql*/ `
-  ${VERSION_TREE_SQL}
-  SELECT *
-  FROM "${table}" t
-  WHERE (t."id", t."version") NOT IN (
-    SELECT "entity_id", "entity_version"
-    FROM "client_view_entry"
-    WHERE "client_view_version" IN (SELECT "version" FROM "version_tree") AND
-    "client_group_id" = $1 AND
-    "entity" = $3
-  )
-  LIMIT $4;`;
+  const sql = /*sql*/ `SELECT *
+      FROM "${table}" t
+      WHERE (t."id", t."version") NOT IN (
+        SELECT "entity_id", "entity_version"
+        FROM "client_view_entry"
+        WHERE "client_group_id" = $1
+          AND "client_view_version" <= $2
+          AND "entity" = $3
+      )
+      LIMIT $4;`;
 
-  const params = [clientGroupID, version, TableOrdinal[table], limit];
+  const params = [clientGroupID, order, TableOrdinal[table], limit];
   return executor(sql, params);
 }
 
@@ -124,40 +94,45 @@ export function findDeletions(
   executor: Executor,
   table: keyof typeof TableOrdinal,
   clientGroupID: string,
-  version: number,
+  order: number,
   limit: number,
 ) {
-  const sql = /*sql*/ `${VERSION_TREE_SQL}
-  SELECT DISTINCT("entity_id") FROM "client_view_entry" as cve WHERE 
-    cve."client_group_id" = $1 AND
-    cve."entity" = $3 AND
-    cve."client_view_version" IN (SELECT "version" FROM "version_tree") AND
-    -- check that the entity is missing from the base table
-    NOT EXISTS (
-      SELECT 1 FROM "${table}" WHERE id = cve."entity_id"
+  // Find rows that are in the CVR table but not in the actual table.
+  // Exclude rows that have a delete entry already.
+  // TODO: Maybe we can remove the second `NOT EXISTS` if we prune the CVR table.
+  // This pruning would mean that we need to record the delete against the
+  // current CVR rather than next CVR. If a request comes in for that prior CVR,
+  // we return the stored delete records and do not compute deletes.
+  return executor(
+    /*sql*/ `SELECT "entity_id" FROM "client_view_entry"
+    WHERE "client_view_entry"."entity" = $1 AND NOT EXISTS (
+      SELECT 1 FROM "${table}" WHERE id = "client_view_entry"."entity_id"
     ) AND
-    -- check that we didn't already record this deletion
-    -- and that the row wasn't later resurrected and sent if we did record a deletion
-    NOT EXISTS (
-      SELECT 1
-      FROM client_view_entry as cve2 LEFT JOIN client_view_entry as cve3
-        ON (
-          cve2.entity_id = cve3.entity_id AND
-          cve2.client_view_version < cve3.client_view_version AND
-          cve2.client_group_id = cve3.client_group_id AND
-          cve2.entity = cve3.entity
-        )
-      WHERE
-        cve3.entity_id IS NULL AND
-        cve2.entity_id = cve.entity_id AND
-        cve2.entity_version IS NULL AND
-        cve2.client_group_id = $1 AND
-        cve2.entity = $3 AND
-        cve2.client_view_version IN (SELECT "version" FROM "version_tree")
-    )
-  LIMIT $4`;
-  const vars = [clientGroupID, version, TableOrdinal[table], limit];
-  return executor(sql, vars);
+    "client_view_entry"."client_group_id" = $2 AND
+    "client_view_entry"."client_view_version" <= $3 
+    AND NOT EXISTS (
+      SELECT 1 FROM "client_view_delete_entry" WHERE "client_view_delete_entry"."entity" = $1 AND "client_view_delete_entry"."entity_id" = "client_view_entry"."entity_id"
+      AND "client_view_delete_entry"."client_group_id" = $2 AND "client_view_delete_entry"."client_view_version" <= $3
+    ) LIMIT $4`,
+    [TableOrdinal[table], clientGroupID, order, limit],
+  );
+}
+
+export async function dropCVREntries(
+  executor: Executor,
+  clientGroupID: string,
+  order: number,
+) {
+  await Promise.all([
+    executor(
+      /*sql*/ `DELETE FROM "client_view_entry" WHERE "client_group_id" = $1 AND "client_view_version" > $2`,
+      [clientGroupID, order],
+    ),
+    executor(
+      /*sql*/ `DELETE FROM "client_view_delete_entry" WHERE "client_group_id" = $1 AND "client_view_version" > $2`,
+      [clientGroupID, order],
+    ),
+  ]);
 }
 
 export async function recordUpdates(
@@ -196,7 +171,12 @@ export async function recordUpdates(
     "entity",
     "entity_id",
     "entity_version"
-  ) VALUES ${placeholders.join(', ')}`,
+  ) VALUES ${placeholders.join(
+    ', ',
+  )} ON CONFLICT ("client_group_id", "entity", "entity_id") DO UPDATE SET
+    "entity_version" = excluded."entity_version",
+    "client_view_version" = excluded."client_view_version"
+  `,
     values,
   );
 }
@@ -225,22 +205,21 @@ export async function recordDeletes(
   }
   const placeholders = [];
   const values = [];
-  const stride = 5;
+  const stride = 4;
   for (let i = 0; i < ids.length; i++) {
     placeholders.push(
       `($${i * stride + 1}, $${i * stride + 2}, $${i * stride + 3}, $${
         i * stride + 4
-      }, $${i * stride + 5})`,
+      })`,
     );
-    values.push(clientGroupID, order, TableOrdinal[table], ids[i], null);
+    values.push(clientGroupID, order, TableOrdinal[table], ids[i]);
   }
   await executor(
-    /*sql*/ `INSERT INTO client_view_entry (
+    /*sql*/ `INSERT INTO client_view_delete_entry (
     "client_group_id",
     "client_view_version",
     "entity",
-    "entity_id",
-    "entity_version"
+    "entity_id"
   ) VALUES ${placeholders.join(', ')}`,
     values,
   );
diff --git a/server/src/pull/pull.ts b/server/src/pull/pull.ts
@@ -4,7 +4,14 @@ import {transact} from '../pg';
 import {getClientGroupForUpdate, putClientGroup, searchClients} from '../data';
 import type Express from 'express';
 import {hasNextPage, isPageEmpty, readNextPage} from './next-page';
-import {CVR, getCVR, putCVR, recordDeletes, recordUpdates} from './cvr';
+import {
+  CVR,
+  dropCVREntries,
+  getCVR,
+  putCVR,
+  recordDeletes,
+  recordUpdates,
+} from './cvr';
 import {PartialSyncState, PARTIAL_SYNC_STATE_KEY} from 'shared';
 
 const cookieSchema = z.object({
@@ -44,6 +51,16 @@ export async function pull(
         order: 0,
       };
 
+      // Drop any cvr entries greater than the order we received.
+      // Getting an old order from the client means that the client is possibly missing the data
+      // from future orders.
+      //
+      // Why do we delete later CVRs?
+      // Since we are sharing CVR data across orders (CVR_n = CVR_n-1 + current_pull).
+      // To keep greater CVRs around, which may have never been received, means that the next CVR would
+      // indicate that the data was received when it was not.
+      await dropCVREntries(executor, clientGroupID, baseCVR.order);
+
       const [clientChanges, nextPage] = await Promise.all([
         searchClients(executor, {
           clientGroupID,
@@ -115,7 +132,7 @@ export async function pull(
 
       await Promise.all([
         putClientGroup(executor, nextClientGroupRecord),
-        putCVR(executor, nextCVR, baseCVR.order),
+        putCVR(executor, nextCVR),
       ]);
 
       await Promise.all([
diff --git a/server/src/schema.ts b/server/src/schema.ts
@@ -73,31 +73,22 @@ export async function createSchemaVersion1(executor: Executor) {
   await executor(/*sql*/ `CREATE TABLE "client_view" (
     "client_group_id" VARCHAR(36) NOT NULL,
     "version" INTEGER NOT NULL,
-    "parent_version" INTEGER NOT NULL,
     "client_version" INTEGER NOT NULL,
     PRIMARY KEY ("client_group_id", "version")
   )`);
 
-  await executor(
-    /*sql*/ `CREATE INDEX client_view_parent ON client_view (client_group_id, parent_version)`,
-  );
-
   await executor(/*sql*/ `CREATE TABLE "client_view_entry" (
     "client_group_id" VARCHAR(36) NOT NULL,
     "client_view_version" INTEGER NOT NULL,
     "entity" INTEGER NOT NULL,
     "entity_id" VARCHAR(36) NOT NULL,
-    "entity_version" INTEGER
+    "entity_version" INTEGER NOT NULL,
+    -- unique by client_group_id, entity, entity_id
+    -- 1. A missing row version is semantically the same as a behind row version
+    -- 2. Our CVR is recursive. CVR_n = CVR_n-1 + (changes since CVR_n-1)
+    PRIMARY KEY ("client_group_id", "entity", "entity_id")
   )`);
 
-  // This index is for quick lookup of everything in a client view.
-  // - get entries for the specific client
-  // - for the specific table
-  // - at the specific client_view version
-  await executor(
-    /*sql*/ `CREATE INDEX client_view_entry_lookup ON client_view_entry (client_group_id, entity, client_view_version)`,
-  );
-
   await executor(/*sql*/ `CREATE TABLE "client_view_delete_entry" (
     "client_group_id" VARCHAR(36) NOT NULL,
     "client_view_version" INTEGER NOT NULL,
