explicit flush arguments

ponder-sh · Jun 12, 2024 · b5a0c0e · b5a0c0e
1 parent b60df84
commit b5a0c0e
Show file tree

Hide file tree

Showing 5 changed files with 30 additions and 35 deletions.
diff --git a/packages/core/src/bin/utils/run.ts b/packages/core/src/bin/utils/run.ts
@@ -266,7 +266,7 @@ export async function run({
       }
     }
 
-    await historicalStore.flush();
+    await historicalStore.flush({ isFullFlush: true });
 
     // Become healthy
     common.logger.info({

diff --git a/packages/core/src/indexing-store/historical.test.ts b/packages/core/src/indexing-store/historical.test.ts
@@ -71,7 +71,7 @@ test("findUnique() w/ cache miss", async (context) => {
     data: { name: "Skip", age: 12 },
   });
 
-  await (indexingStore as HistoricalStore).flush();
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   const instance = await indexingStore.findUnique({
     tableName: "Pet",
@@ -465,7 +465,7 @@ test("update() w/ cache miss", async (context) => {
     data: { name: "Skip", bigAge: 100n },
   });
 
-  await (indexingStore as HistoricalStore).flush();
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   const updatedInstance = await indexingStore.update({
     tableName: "Pet",
@@ -490,7 +490,7 @@ test("update() w/ find cache", async (context) => {
     data: { name: "Skip", bigAge: 100n },
   });
 
-  await (indexingStore as HistoricalStore).flush();
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   await indexingStore.findUnique({ tableName: "Pet", id: "id1" });
 
@@ -509,8 +509,7 @@ test("update() w/ find cache", async (context) => {
 
   expect(findInstance).toMatchObject({ id: "id1", name: "Peanut Butter" });
 
-  // @ts-ignore
-  await indexingStore.flush();
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   const rows = await database.indexingDb
     .withSchema(namespaceInfo.userNamespace)
@@ -700,7 +699,7 @@ test("upsert() w/ cache miss", async (context) => {
     data: { name: "Skip", age: 12 },
   });
 
-  await (indexingStore as HistoricalStore).flush();
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   const updatedInstance = await indexingStore.upsert({
     tableName: "Pet",
@@ -722,8 +721,7 @@ test("upsert() w/ find cache", async (context) => {
 
   // add pet.id1 to find cache
 
-  // @ts-ignore
-  await indexingStore.flush();
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   await indexingStore.findUnique({
     tableName: "Pet",
@@ -748,8 +746,7 @@ test("upsert() w/ find cache", async (context) => {
 
   // add pet.id1 to find cache, remove from create cache
 
-  // @ts-ignore
-  await indexingStore.flush();
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   await indexingStore.findUnique({
     tableName: "Pet",
@@ -772,8 +769,7 @@ test("upsert() w/ find cache", async (context) => {
 
   expect(findInstance).toMatchObject({ id: "id1", name: "Kevin" });
 
-  // @ts-ignore
-  await indexingStore.flush();
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   const rows = await database.indexingDb
     .withSchema(namespaceInfo.userNamespace)
@@ -832,7 +828,7 @@ test("delete() w/ find cache", async (context) => {
     data: { name: "Skip", age: 12 },
   });
 
-  await (indexingStore as HistoricalStore).flush();
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   await indexingStore.findUnique({ tableName: "Pet", id: "id1" });
 
@@ -849,8 +845,7 @@ test("delete() w/ find cache", async (context) => {
   });
   expect(deletedInstance).toBe(null);
 
-  // @ts-ignore
-  await indexingStore.flush();
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   const rows = await database.indexingDb
     .withSchema(namespaceInfo.userNamespace)
@@ -1065,7 +1060,7 @@ test("flush() insert", async (context) => {
     data: { name: "Skip", age: 12 },
   });
 
-  await (indexingStore as HistoricalStore).flush();
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   const rows = await database.indexingDb
     .withSchema(namespaceInfo.userNamespace)
@@ -1095,15 +1090,15 @@ test("flush() update", async (context) => {
     data: { name: "Skip", age: 12 },
   });
 
-  await (indexingStore as HistoricalStore).flush();
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   await indexingStore.update({
     tableName: "Pet",
     id: "id1",
     data: { name: "Peanut Butter" },
   });
 
-  await (indexingStore as HistoricalStore).flush();
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   const rows = await database.indexingDb
     .withSchema(namespaceInfo.userNamespace)
@@ -1143,8 +1138,7 @@ test("flush() partial", async (context) => {
     ],
   });
 
-  // @ts-ignore
-  await indexingStore.flush(false);
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: false });
 
   const rows = await database.indexingDb
     .withSchema(namespaceInfo.userNamespace)

diff --git a/packages/core/src/indexing-store/historical.ts b/packages/core/src/indexing-store/historical.ts
@@ -162,11 +162,11 @@ export const getHistoricalStore = ({
 
   const shouldFlush = () => cacheSizeBytes > maxSizeBytes;
 
-  const flush = createQueue<void, boolean | void>({
+  const flush = createQueue<void, { isFullFlush: boolean }>({
     concurrency: 1,
     initialStart: true,
     browser: false,
-    worker: async (fullFlush = true) => {
+    worker: async ({ isFullFlush }: { isFullFlush: boolean }) => {
       const flushIndex = totalCacheOps - cacheSize * (1 - CACHE_FLUSH);
 
       await Promise.all(
@@ -179,7 +179,7 @@ export const getHistoricalStore = ({
 
             const cacheEntries = Object.values(tableStoreCache);
 
-            if (fullFlush) {
+            if (isFullFlush) {
               insertRecords = cacheEntries
                 .filter(({ type }) => type === "insert")
                 .map(({ record }) => record!);
@@ -302,7 +302,7 @@ export const getHistoricalStore = ({
         ),
       );
 
-      if (fullFlush) {
+      if (isFullFlush) {
         for (const tableName of Object.keys(tables)) {
           storeCache[tableName] = {};
         }
@@ -361,7 +361,7 @@ export const getHistoricalStore = ({
       cacheSizeBytes += bytes;
       cacheSize++;
 
-      if (shouldFlush()) await flush(false);
+      if (shouldFlush()) await flush({ isFullFlush: false });
 
       return structuredClone(record);
     },
@@ -373,7 +373,7 @@ export const getHistoricalStore = ({
       after?: string | null;
       limit?: number;
     }) => {
-      await flush(true);
+      await flush({ isFullFlush: true });
 
       return readonlyStore.findMany(arg);
     },
@@ -416,7 +416,7 @@ export const getHistoricalStore = ({
       cacheSizeBytes += bytes;
       cacheSize++;
 
-      if (shouldFlush()) await flush(false);
+      if (shouldFlush()) await flush({ isFullFlush: false });
 
       return structuredClone(record);
     },
@@ -458,7 +458,7 @@ export const getHistoricalStore = ({
 
       cacheSize += data.length;
 
-      if (shouldFlush()) await flush(false);
+      if (shouldFlush()) await flush({ isFullFlush: false });
 
       const returnData = structuredClone(data);
       for (const record of data) {
@@ -544,7 +544,7 @@ export const getHistoricalStore = ({
         | Partial<Omit<UserRecord, "id">>
         | ((args: { current: UserRecord }) => Partial<Omit<UserRecord, "id">>);
     }) => {
-      await flush(true);
+      await flush({ isFullFlush: true });
       const table = (schema[tableName] as { table: Table }).table;
 
       if (typeof data === "function") {
@@ -749,7 +749,7 @@ export const getHistoricalStore = ({
         cacheSize++;
         cacheSizeBytes += bytes;
 
-        if (shouldFlush()) await flush(false);
+        if (shouldFlush()) await flush({ isFullFlush: false });
 
         return structuredClone(record);
       }

diff --git a/packages/core/src/indexing-store/store.ts b/packages/core/src/indexing-store/store.ts
@@ -95,7 +95,9 @@ export type WriteStore<
 export type RealtimeStore = ReadonlyStore & WriteStore<"realtime">;
 
 export type HistoricalStore = ReadonlyStore &
-  WriteStore<"historical"> & { flush: () => Promise<void> };
+  WriteStore<"historical"> & {
+    flush: (arg: { isFullFlush: boolean }) => Promise<void>;
+  };
 
 export type IndexingStore<
   env extends "historical" | "realtime" = "historical" | "realtime",

diff --git a/packages/core/src/server/service.test.ts b/packages/core/src/server/service.test.ts
@@ -3,7 +3,7 @@ import {
   setupDatabaseServices,
   setupIsolatedDatabase,
 } from "@/_test/setup.js";
-import type { ReadonlyStore } from "@/indexing-store/store.js";
+import type { HistoricalStore, ReadonlyStore } from "@/indexing-store/store.js";
 import { createSchema } from "@/schema/schema.js";
 import { encodeCheckpoint, zeroCheckpoint } from "@/utils/checkpoint.js";
 import type { GraphQLSchema } from "graphql";
@@ -161,8 +161,7 @@ test("graphql", async (context) => {
     },
   });
 
-  // @ts-ignore
-  await indexingStore.flush(true);
+  await (indexingStore as HistoricalStore).flush({ isFullFlush: true });
 
   const graphqlSchema = buildGraphqlSchema(schema);