add a version to the JSON output

Author: mdroidian

apps/roam/src/utils/syncDgNodesToSupabase.ts

@@ -37,6 +37,7 @@ const SYNC_TIMEOUT = "60s"; // must be less than half the SYNC_INTERVAL.
 const BATCH_SIZE = 200;
 const CONCEPT_BATCH_SIZE = 200;
 const DEFAULT_TIME = new Date("1970-01-01");
+const END_SYNC_TASK_RESULT_VERSION = 1;
 
 type SyncPhaseDurations = Record<string, number>;
 
@@ -49,6 +50,7 @@ type SyncTaskInfo = {
 type LocalConceptDataInput = Partial<CompositeTypes<"concept_local_input">>;
 
 type EndSyncTaskRpcResult = {
+  version?: number;
   ok: boolean;
   stale: boolean;
   reason?: string;
@@ -99,15 +101,39 @@ const getSyncWorkerId = (): string => {
   return syncWorkerId;
 };
 
-const isEndSyncTaskRpcResult = (
+const getJsonObject = (
   data: Json | undefined,
-): data is EndSyncTaskRpcResult => {
+): Record<string, unknown> | null => {
   if (typeof data !== "object" || data === null || Array.isArray(data)) {
-    return false;
+    return null;
+  }
+
+  return data as Record<string, unknown>;
+};
+
+const getEndSyncTaskResultVersion = (data: Json | undefined): number => {
+  const result = getJsonObject(data);
+  if (result === null || typeof result.version !== "number") {
+    return 0;
   }
 
-  const result = data as { ok?: unknown; stale?: unknown };
-  return typeof result.ok === "boolean" && typeof result.stale === "boolean";
+  return result.version;
+};
+
+const isEndSyncTaskRpcResult = (
+  data: Json | undefined,
+): data is EndSyncTaskRpcResult => {
+  const result = getJsonObject(data);
+  if (result === null) return false;
+
+  const versionIsSupported =
+    result.version === undefined || typeof result.version === "number";
+
+  return (
+    versionIsSupported &&
+    typeof result.ok === "boolean" &&
+    typeof result.stale === "boolean"
+  );
 };
 
 const measureSyncPhase = async <T>({
@@ -319,6 +345,27 @@ export const endSyncTask = async ({
     }
 
     if (!isEndSyncTaskRpcResult(data)) {
+      const resultVersion = getEndSyncTaskResultVersion(data);
+      if (resultVersion > END_SYNC_TASK_RESULT_VERSION) {
+        const rpcResult: EndSyncTaskRpcResult = {
+          version: resultVersion,
+          ok: true,
+          stale: false,
+          reason: "unsupported_future_result_version",
+        };
+        posthog.capture("Sync end task future result", {
+          ...telemetryContext,
+          endSyncPayload: data,
+          endSyncResult: rpcResult,
+        });
+
+        return {
+          ok: true,
+          stale: false,
+          rpcResult,
+        };
+      }
+
       const reason = "Supabase end_sync_task returned unexpected payload";
       notifyEndSyncFailure({
         status,

packages/database/doc/sync_functions.md

@@ -16,4 +16,4 @@ When a worker calls `propose_sync_task`, it will receive either:
 3. Null, meaning the task was not executed successfully before, and your worker is tasked with starting from scratch.
 
 When a worker finishes the task, it should clean up with `end_sync_task`, giving the same identifying arguments, a status (`complete` or `failed`), and the time at which the task was claimed.
-The function returns a JSON object with `ok` and `stale` fields plus task metadata. A stale response means another attempt claimed a newer task while this attempt was still running, so the caller should not report it as a successful completion.
+The function returns a versioned JSON object with `ok` and `stale` fields plus task metadata. A stale response means another attempt claimed a newer task while this attempt was still running, so the caller should not report it as a successful completion.

packages/database/supabase/migrations/20260528033000_end_sync_task_result.sql

@@ -32,6 +32,7 @@ BEGIN
     IF t_worker != s_worker AND COALESCE(s_started_at, t_last_task_start) < t_last_task_start THEN
         -- we probably took too long. Let the other task have priority.
         RETURN jsonb_build_object(
+            'version', 1,
             'ok', false,
             'stale', true,
             'reason', 'completed_by_newer_task',
@@ -50,6 +51,7 @@ BEGIN
     IF t_worker != s_worker THEN
         RAISE EXCEPTION 'Wrong worker'
             USING DETAIL = jsonb_build_object(
+                'version', 1,
                 'requestedStatus', s_status,
                 'callerWorker', s_worker,
                 'currentWorker', t_worker,
@@ -83,6 +85,7 @@ BEGIN
         WHERE id=t_id;
 
     RETURN jsonb_build_object(
+        'version', 1,
         'ok', true,
         'stale', false,
         'requestedStatus', s_status,

packages/database/supabase/schemas/sync.sql

@@ -73,6 +73,7 @@ BEGIN
     IF t_worker != s_worker AND COALESCE(s_started_at, t_last_task_start) < t_last_task_start THEN
         -- we probably took too long. Let the other task have priority.
         RETURN jsonb_build_object(
+            'version', 1,
             'ok', false,
             'stale', true,
             'reason', 'completed_by_newer_task',
@@ -91,6 +92,7 @@ BEGIN
     IF t_worker != s_worker THEN
         RAISE EXCEPTION 'Wrong worker'
             USING DETAIL = jsonb_build_object(
+                'version', 1,
                 'requestedStatus', s_status,
                 'callerWorker', s_worker,
                 'currentWorker', t_worker,
@@ -124,6 +126,7 @@ BEGIN
         WHERE id=t_id;
 
     RETURN jsonb_build_object(
+        'version', 1,
         'ok', true,
         'stale', false,
         'requestedStatus', s_status,