Add workaround for failed exports of queries: create a fallback table

For each query to export, optionally provide a fallback `CREATE TABLE`
query which will run if the export job for the query fails. Implement
this by calling an API route `/api/create-fallback-table-after-failed-export`
after an export for a query fails for any reason.

This works around the bug where queries with an empty result fail
to export to Seafowl, see: splitgraph/seafowl#423

Author: milesrichardson

examples/nextjs-import-airbyte-github-export-seafowl/.env.test.local

@@ -19,3 +19,13 @@ GITHUB_PAT_SECRET="github_pat_**********************_***************************
 # e.g. To intercept requests to Splitgraph API sent from madatdata libraries in API routes
 # You can also set this by running: yarn dev-mitm (see package.json)
 # MITMPROXY_ADDRESS="http://localhost:7979"
+
+# OPTIONAL: Set Seafowl environment variables to use for creating fallback tables when exports fail
+# NOTE 1: At the moment the instance URL must be https://demo.seafowl.cloud because that's where
+# the Splitgraph export API exports tables to when no instance URL is specified, and we are
+# currently not specifying the instance URL when starting exports, and only use it when creating fallback tables.
+# NOTE 2: The dbname (SEAFOWL_INSTANCE_DATABASE) MUST match NEXT_PUBLIC_SPLITGRAPH_GITHUB_ANALYTICS_META_NAMESPACE
+#
+# SEAFOWL_INSTANCE_URL="https://demo.seafowl.cloud"
+# SEAFOWL_INSTANCE_SECRET="********************************"
+# SEAFOWL_INSTANCE_DATABASE="**********"

examples/nextjs-import-airbyte-github-export-seafowl/components/ImportExportStepper/DebugPanel.tsx

@@ -8,7 +8,15 @@ export const DebugPanel = () => {
       <pre style={{ minWidth: "80%", minHeight: "300px" }}>
         {JSON.stringify(
           state,
-          (_key, value) => (value instanceof Set ? Array.from(value) : value),
+          (_key, value) => {
+            if (value instanceof Set) {
+              return Array.from(value);
+            } else if (value instanceof Map) {
+              return Object.fromEntries(value);
+            } else {
+              return value;
+            }
+          },
           2
         )}
       </pre>

examples/nextjs-import-airbyte-github-export-seafowl/components/ImportExportStepper/ExportPanel.tsx

@@ -87,6 +87,11 @@ export const ExportPanel = () => {
               destinationTable,
               destinationSchema,
               sourceQuery,
+              fallbackCreateTableQuery: queriesToExport.find(
+                (q) =>
+                  q.destinationSchema === destinationSchema &&
+                  q.destinationTable === destinationTable
+              )?.fallbackCreateTableQuery,
             })
           ),
           ...data["tables"].map(

examples/nextjs-import-airbyte-github-export-seafowl/components/ImportExportStepper/export-hooks.tsx

@@ -46,7 +46,10 @@ export const useFindMatchingExportTable = (
 };
 
 export const usePollExportTasks = () => {
-  const [{ stepperState, loadingExportTasks }, dispatch] = useStepper();
+  const [
+    { stepperState, loadingExportTasks, exportedTablesLoading },
+    dispatch,
+  ] = useStepper();
 
   useEffect(() => {
     if (stepperState !== "awaiting_export") {
@@ -71,11 +74,50 @@ export const usePollExportTasks = () => {
                 type: "export_task_complete",
                 completedTask: { taskId },
               }),
-            onError: ({ taskId, error }) =>
+            onError: async ({ taskId, error }) => {
+              // If the task failed but we're not going to retry, then check if
+              // there is a fallback query to create the table, and if so,
+              // create it before marking the task as complete.
+              if (!error.retryable) {
+                // NOTE: There is an implicit assumption that `exportedTablesLoading`
+                // and `loadingExportTasks` are updated at the same time, which they
+                // are, by the reducer that handles the `export_task_start` and
+                // `export_task_complete` actions.
+                const maybeExportedQueryWithCreateTableFallback = Array.from(
+                  exportedTablesLoading
+                ).find(
+                  (t) => t.taskId === taskId && t.fallbackCreateTableQuery
+                );
+
+                if (maybeExportedQueryWithCreateTableFallback) {
+                  await createFallbackTableAfterFailedExport({
+                    destinationSchema:
+                      maybeExportedQueryWithCreateTableFallback.destinationSchema,
+                    destinationTable:
+                      maybeExportedQueryWithCreateTableFallback.destinationTable,
+                    fallbackCreateTableQuery:
+                      maybeExportedQueryWithCreateTableFallback.fallbackCreateTableQuery,
+
+                    // On error or success, we mutate the error variable which
+                    // will be passed by `dispatch` outside of this conditional.
+                    onError: (errorCreatingFallbackTable) => {
+                      error.message = `${error.message} (and also error creating fallback: ${errorCreatingFallbackTable.message})`;
+                    },
+                    onSuccess: () => {
+                      error = undefined; // No error because we consider the task complete after creating the fallback table.
+                    },
+                  });
+                }
+              }
+
               dispatch({
-                type: "export_error",
-                error: `Error exporting ${taskId}: ${error.message}`,
-              }),
+                type: "export_task_complete",
+                completedTask: {
+                  taskId,
+                  error,
+                },
+              });
+            },
             abortSignal: abortController.signal,
           })
         )
@@ -86,7 +128,7 @@ export const usePollExportTasks = () => {
       clearInterval(interval);
       abortController.abort();
     };
-  }, [loadingExportTasks, stepperState, dispatch]);
+  }, [loadingExportTasks, exportedTablesLoading, stepperState, dispatch]);
 };
 
 const pollExportTaskOnce = async ({
@@ -97,7 +139,13 @@ const pollExportTaskOnce = async ({
 }: {
   taskId: string;
   onSuccess: ({ taskId }: { taskId: string }) => void;
-  onError: ({ taskId, error }: { taskId: string; error: any }) => void;
+  onError: ({
+    taskId,
+    error,
+  }: {
+    taskId: string;
+    error: { message: string; retryable: boolean };
+  }) => void;
   abortSignal: AbortSignal;
 }) => {
   try {
@@ -118,6 +166,7 @@ const pollExportTaskOnce = async ({
     } else if (data.error) {
       if (!data.completed) {
         console.log("WARN: Failed status, not completed:", data.error);
+        onError({ taskId, error: { message: data.error, retryable: false } });
       } else {
         throw new Error(data.error);
       }
@@ -127,6 +176,71 @@ const pollExportTaskOnce = async ({
       return;
     }
 
-    onError({ taskId, error });
+    onError({
+      taskId,
+      error: {
+        message: `Error exporting ${taskId}: ${
+          error.message ?? error.name ?? "unknown"
+        }`,
+        retryable: true,
+      },
+    });
+  }
+};
+
+/**
+ * Call the API route to create a fallback table after a failed export.
+ *
+ * Note that both `destinationTable` and `destinationSchema` should already
+ * be included in the `fallbackCreateTableQuery`, but we need them so that
+ * the endpoint can separately `CREATE SCHEMA` and `DROP TABLE` in case the
+ * schema does not yet exist, or the table already exists (we overwrite it to
+ * be consistent with behavior of Splitgraph export API).
+ */
+const createFallbackTableAfterFailedExport = async ({
+  destinationSchema,
+  destinationTable,
+  fallbackCreateTableQuery,
+  onSuccess,
+  onError,
+}: Required<
+  Pick<
+    ExportTable,
+    "destinationSchema" | "destinationTable" | "fallbackCreateTableQuery"
+  >
+> & {
+  onSuccess: () => void;
+  onError: (error: { message: string }) => void;
+}) => {
+  try {
+    const response = await fetch(
+      "/api/create-fallback-table-after-failed-export",
+      {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+        },
+        body: JSON.stringify({
+          destinationSchema,
+          destinationTable,
+          fallbackCreateTableQuery,
+        }),
+      }
+    );
+    const data = await response.json();
+    if (data.error || !data.success) {
+      console.log(
+        `FAIL: error from endpoint creating fallback table: ${data.error}`
+      );
+      onError({ message: data.error ?? "unknown" });
+    } else {
+      console.log("SUCCESS: created fallback table");
+      onSuccess();
+    }
+  } catch (error) {
+    console.log(`FAIL: caught error while creating fallback table: ${error}`);
+    onError({
+      message: `${error.message ?? error.name ?? "unknown"}`,
+    });
   }
 };

examples/nextjs-import-airbyte-github-export-seafowl/components/ImportExportStepper/stepper-states.ts

@@ -8,12 +8,14 @@ export type ExportTable = {
   destinationTable: string;
   taskId: string;
   sourceQuery?: string;
+  fallbackCreateTableQuery?: string;
 };
 
 // NOTE: Multiple tables can have the same taskId, so we track them separately
 // in order to not need to redundantly poll the API for each table individually
 export type ExportTask = {
   taskId: string;
+  error?: { message: string; retryable: boolean };
 };
 
 export type StepperState = {
@@ -35,6 +37,7 @@ export type StepperState = {
   debug?: string | null;
   loadingExportTasks?: Set<ExportTask>;
   completedExportTasks?: Set<ExportTask>;
+  tasksWithError?: Map<string, string[]>; // taskId -> errors
 };
 
 export type StepperAction =
@@ -65,6 +68,7 @@ const initialState: StepperState = {
   exportedTablesCompleted: new Set<ExportTable>(),
   loadingExportTasks: new Set<ExportTask>(),
   completedExportTasks: new Set<ExportTask>(),
+  tasksWithError: new Map<string, string[]>(),
   importError: null,
   exportError: null,
   debug: null,
@@ -217,12 +221,14 @@ const stepperReducer = (
         destinationTable,
         destinationSchema,
         sourceQuery,
+        fallbackCreateTableQuery,
         taskId,
       } of tables) {
         exportedTablesLoading.add({
           destinationTable,
           destinationSchema,
           sourceQuery,
+          fallbackCreateTableQuery,
           taskId,
         });
       }
@@ -248,11 +254,47 @@ const stepperReducer = (
         stepperState: "awaiting_export",
       };
 
+    /**
+     * NOTE: A task is "completed" even if it received an error, in which case
+     * we will retry it up to maxRetryCount if `error.retryable` is `true`
+     *
+     * That is, _all tasks_ will eventually "complete," whether successfully or not.
+     */
     case "export_task_complete":
       const {
-        completedTask: { taskId: completedTaskId },
+        completedTask: { taskId: completedTaskId, error: maybeError },
       } = action;
 
+      const maxRetryCount = 3;
+
+      const updatedTasksWithError = new Map(state.tasksWithError);
+      const previousErrors = updatedTasksWithError.get(completedTaskId) ?? [];
+      const hadPreviousError = previousErrors.length > 0;
+
+      if (!maybeError && hadPreviousError) {
+        updatedTasksWithError.delete(completedTaskId);
+      } else if (maybeError) {
+        updatedTasksWithError.set(completedTaskId, [
+          ...previousErrors,
+          maybeError.message,
+        ]);
+        const numAttempts = updatedTasksWithError.get(completedTaskId).length;
+
+        if (maybeError.retryable && numAttempts < maxRetryCount) {
+          console.log("RETRY: ", completedTaskId, `(${numAttempts} so far)`);
+          return {
+            ...state,
+            tasksWithError: updatedTasksWithError,
+          };
+        } else {
+          console.log(
+            "FAIL: ",
+            completedTaskId,
+            `(${numAttempts} reached max ${maxRetryCount})`
+          );
+        }
+      }
+
       // One taskId could match multiple tables, so find reference to each of them
       // and then use that reference to delete them from loading set and add them to completed set
       const completedTables = Array.from(state.exportedTablesLoading).filter(
@@ -394,7 +436,7 @@ const useMarkAsComplete = (
 
         if (!data.status) {
           throw new Error(
-            "Got unexpected resposne shape when marking import/export complete"
+            "Got unexpected response shape when marking import/export complete"
           );
         }
 

examples/nextjs-import-airbyte-github-export-seafowl/env-vars.d.ts

@@ -38,6 +38,30 @@ namespace NodeJS {
      */
     MITMPROXY_ADDRESS?: string;
 
+    /**
+     * Optionally provide the SEAFOWL_INSTANCE_URL to use for creating fallback tables
+     * when an export fails.
+     *
+     * Note that at the moment, this must only be set to https://demo.seafowl.cloud
+     * because that's where Splitgraph exports to by default, and we are not currently
+     * passing any instance URL to the Splitgraph export API.
+     */
+    SEAFOWL_INSTANCE_URL?: "https://demo.seafowl.cloud";
+
+    /**
+     * Optionally provide the SEAFOWL_INSTANCE_SECRET to use for creating fallback tables
+     * when an export fails.
+     */
+    SEAFOWL_INSTANCE_SECRET?: string;
+
+    /**
+     * Optionally provide the dbname to use for creating fallback tables
+     * when an export fails.
+     *
+     * Note this MUST match the NEXT_PUBLIC_SPLITGRAPH_GITHUB_ANALYTICS_META_NAMESPACE
+     */
+    SEAFOWL_INSTANCE_DATABASE?: string;
+
     /**
      * The namespace of the repository in Splitgraph where metadata is stored
      * containing the state of imported GitHub repositories, which should contain