Fix unreliable SSE event stream updates during workflow form transitions (#20242)

Before - workflow run not up to date, needs refresh to see created company in some cases https://github.com/user-attachments/assets/28517e97-2404-4f75-8bce-cc33e3cbea20 After https://github.com/user-attachments/assets/60f930cb-1265-4c50-8ec5-aa4f978b1873 ## Summary - Split `SSEQuerySubscribeEffect`'s single debounced `updateQueryListeners` into separate `syncAdditions` (leading edge, 1s debounce) and `syncRemovals` (trailing edge, 200ms debounce) callbacks. This prevents query unregistrations during component mount/unmount transitions from creating gaps where events are missed, while keeping new registrations immediate. - Each sync path now updates `activeQueryListenersState` granularly (append-only for additions, filter-only for removals) instead of overwriting the entire state, eliminating a race condition where removals could mark unregistered queries as active. - Mount `WorkflowRunSSESubscribeEffect` inside `WorkflowEditActionFormFiller` so the workflow-run query subscription stays active during form steps. - Extract `buildSortedConnectionEdges` util that builds the resulting edge list of a cached record connection after new records are created. Position placeholders (`'first'` / `'last'`) bypass orderBy and are pinned to the front/back; sortable positions (numeric or undefined) are merged into existing edges and sorted by the connection's actual `orderBy`. This replaces the broken `length * position` insertion logic in `triggerCreateRecordsOptimisticEffect` that treated the sortable `position` field as a 0-1 ratio, causing new records from SSE to land at invisible indices in the cached list. Also fixes `totalCount` increment for batched creates, derives `pageInfo` cursors from the final array, and gracefully skips records whose `toReference` returns null. ## Test plan - [x] Run a workflow with a form step — verify the workflow status updates live after form submission (no stuck "running" state) - [x] Run the same workflow multiple times — verify company creation events appear live on the record index page for every run, not just the first - [x] Click the "+" button to create a record in first position — verify it appears immediately at the top - [x] Verify other SSE-backed live updates (record creation, deletion, updates) still work correctly --------- Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-05 20:14:25 +02:00
parent 6854dc549b
commit a03c2647cf
4 changed files with 229 additions and 145 deletions
@@ -0,0 +1,53 @@
+import { type FieldFunctionOptions } from '@apollo/client/cache';
+
+import { sortCachedObjectEdges } from '@/apollo/optimistic-effect/utils/sortCachedObjectEdges';
+import { type RecordGqlRefEdge } from '@/object-record/cache/types/RecordGqlRefEdge';
+import { type RecordGqlNode } from '@/object-record/graphql/types/RecordGqlNode';
+import { type RecordGqlOperationOrderBy } from 'twenty-shared/types';
+
+type NewEntry = {
+  edge: RecordGqlRefEdge;
+  record: RecordGqlNode;
+};
+
+type BuildSortedConnectionEdgesArgs = {
+  currentEdges: readonly RecordGqlRefEdge[];
+  newEntries: readonly NewEntry[];
+  orderBy: RecordGqlOperationOrderBy | undefined;
+  readField: FieldFunctionOptions['readField'];
+};
+
+export const buildSortedConnectionEdges = ({
+  currentEdges,
+  newEntries,
+  orderBy,
+  readField,
+}: BuildSortedConnectionEdgesArgs): RecordGqlRefEdge[] => {
+  const firstEdges: RecordGqlRefEdge[] = [];
+  const lastEdges: RecordGqlRefEdge[] = [];
+  const sortableEdges: RecordGqlRefEdge[] = [];
+
+  for (const { edge, record } of newEntries) {
+    if (record.position === 'first') {
+      firstEdges.push(edge);
+    } else if (record.position === 'last') {
+      lastEdges.push(edge);
+    } else {
+      sortableEdges.push(edge);
+    }
+  }
+
+  let middleEdges: RecordGqlRefEdge[];
+
+  if (Array.isArray(orderBy) && orderBy.length > 0) {
+    middleEdges = sortCachedObjectEdges({
+      edges: [...currentEdges, ...sortableEdges],
+      orderBy,
+      readCacheField: readField,
+    });
+  } else {
+    middleEdges = [...sortableEdges, ...currentEdges];
+  }
+
+  return [...firstEdges, ...middleEdges, ...lastEdges];
+};
@@ -1,6 +1,7 @@
 import { type ApolloCache, type StoreObject } from '@apollo/client';
 import { isNonEmptyString } from '@sniptt/guards';

+import { buildSortedConnectionEdges } from '@/apollo/optimistic-effect/utils/buildSortedConnectionEdges';
 import { triggerUpdateRelationsOptimisticEffect } from '@/apollo/optimistic-effect/utils/triggerUpdateRelationsOptimisticEffect';
 import { type EnrichedObjectMetadataItem } from '@/object-metadata/types/EnrichedObjectMetadataItem';
 import { type RecordGqlRefEdge } from '@/object-record/cache/types/RecordGqlRefEdge';
@@ -119,120 +120,78 @@ export const triggerCreateRecordsOptimisticEffect = ({
          hasPreviousPage?: boolean;
        }>('pageInfo', rootQueryCachedObjectRecordConnection);

-        const nextRootQueryCachedRecordEdges = rootQueryCachedRecordEdges
-          ? [...rootQueryCachedRecordEdges]
-          : [];
+        const newEntries = recordsToCreate.flatMap<{
+          edge: RecordGqlRefEdge;
+          record: RecordGqlNode;
+        }>((recordToCreate) => {
+          if (!isNonEmptyString(recordToCreate.id)) {
+            return [];
+          }

-        const nextQueryCachedPageInfo = isDefined(rootQueryCachedPageInfo)
-          ? { ...rootQueryCachedPageInfo }
-          : {};
+          if (
+            isDefined(rootQueryFilter) &&
+            shouldMatchRootQueryFilter === true &&
+            !isRecordMatchingFilter({
+              record: recordToCreate,
+              filter: rootQueryFilter,
+              objectMetadataItem,
+            })
+          ) {
+            return [];
+          }

-        const hasAddedRecords = recordsToCreate
-          .map((recordToCreate) => {
-            if (isNonEmptyString(recordToCreate.id)) {
-              if (
-                isDefined(rootQueryFilter) &&
-                shouldMatchRootQueryFilter === true
-              ) {
-                const recordToCreateMatchesThisRootQueryFilter =
-                  isRecordMatchingFilter({
-                    record: recordToCreate,
-                    filter: rootQueryFilter,
-                    objectMetadataItem,
-                  });
+          const node = toReference(recordToCreate);

-                if (!recordToCreateMatchesThisRootQueryFilter) {
-                  return false;
-                }
-              }
+          if (!isDefined(node)) {
+            return [];
+          }

-              const recordToCreateReference = toReference(recordToCreate);
+          const recordAlreadyInCache = rootQueryCachedRecordEdges?.some(
+            (cachedEdge) =>
+              cache.identify(node) === cache.identify(cachedEdge.node),
+          );

-              if (!recordToCreateReference) {
-                throw new Error(
-                  `Failed to create reference for record with id: ${recordToCreate.id}`,
-                );
-              }
+          if (recordAlreadyInCache === true) {
+            return [];
+          }

-              const recordAlreadyInCache = rootQueryCachedRecordEdges?.some(
-                (cachedEdge) => {
-                  return (
-                    cache.identify(recordToCreateReference) ===
-                    cache.identify(cachedEdge.node)
-                  );
-                },
-              );
+          return [
+            {
+              edge: {
+                __typename: getEdgeTypename(objectMetadataItem.nameSingular),
+                node,
+                cursor: encodeCursor(recordToCreate),
+              },
+              record: recordToCreate,
+            },
+          ];
+        });

-              if (isDefined(recordToCreateReference) && !recordAlreadyInCache) {
-                const cursor = encodeCursor(recordToCreate);
-
-                const edge = {
-                  __typename: getEdgeTypename(objectMetadataItem.nameSingular),
-                  node: recordToCreateReference,
-                  cursor,
-                };
-
-                if (
-                  !isDefined(recordToCreate.position) ||
-                  recordToCreate.position === 'first'
-                ) {
-                  nextRootQueryCachedRecordEdges.unshift(edge);
-                  nextQueryCachedPageInfo.startCursor = cursor;
-                } else if (recordToCreate.position === 'last') {
-                  nextRootQueryCachedRecordEdges.push(edge);
-                  nextQueryCachedPageInfo.endCursor = cursor;
-                } else if (typeof recordToCreate.position === 'number') {
-                  let index = Math.round(
-                    nextRootQueryCachedRecordEdges.length *
-                      recordToCreate.position,
-                  );
-
-                  if (recordToCreate.position < 0) {
-                    index = Math.max(
-                      0,
-                      nextRootQueryCachedRecordEdges.length +
-                        Math.round(recordToCreate.position),
-                    );
-                  } else if (recordToCreate.position > 1) {
-                    index = nextRootQueryCachedRecordEdges.length;
-                  }
-
-                  index = Math.max(
-                    0,
-                    Math.min(index, nextRootQueryCachedRecordEdges.length),
-                  );
-
-                  nextRootQueryCachedRecordEdges.splice(index, 0, edge);
-
-                  if (index === 0) {
-                    nextQueryCachedPageInfo.startCursor = cursor;
-                  } else if (
-                    index ===
-                    nextRootQueryCachedRecordEdges.length - 1
-                  ) {
-                    nextQueryCachedPageInfo.endCursor = cursor;
-                  }
-                }
-
-                return true;
-              }
-            }
-
-            return false;
-          })
-          .some((hasAddedRecord) => hasAddedRecord);
-
-        if (!hasAddedRecords) {
+        if (newEntries.length === 0) {
          return rootQueryCachedObjectRecordConnection;
        }

+        const sortedEdges = buildSortedConnectionEdges({
+          currentEdges: rootQueryCachedRecordEdges ?? [],
+          newEntries,
+          orderBy: rootQueryVariables?.orderBy,
+          readField,
+        });
+
        return {
          ...rootQueryCachedObjectRecordConnection,
-          edges: nextRootQueryCachedRecordEdges,
+          edges: sortedEdges,
          totalCount: isDefined(rootQueryCachedRecordTotalCount)
-            ? rootQueryCachedRecordTotalCount + 1
+            ? rootQueryCachedRecordTotalCount + newEntries.length
            : undefined,
-          pageInfo: nextQueryCachedPageInfo,
+          pageInfo: {
+            ...(rootQueryCachedPageInfo ?? {}),
+            startCursor:
+              sortedEdges[0]?.cursor ?? rootQueryCachedPageInfo?.startCursor,
+            endCursor:
+              sortedEdges[sortedEdges.length - 1]?.cursor ??
+              rootQueryCachedPageInfo?.endCursor,
+          },
        };
      },
    },
@@ -41,13 +41,35 @@ export const SSEQuerySubscribeEffect = () => {
  const requiredQueryListeners = useAtomStateValue(requiredQueryListenersState);
  const activeQueryListeners = useAtomStateValue(activeQueryListenersState);

-  const updateQueryListeners = useCallback(async () => {
+  const handleError = useCallback(
+    (error: unknown) => {
+      if (CombinedGraphQLErrors.is(error)) {
+        const extensions = getGraphqlErrorExtensionsFromError(error);
+
+        if (
+          isGracefullyHandledEventStreamError({
+            subCode: extensions?.subCode,
+            code: extensions?.code,
+          })
+        ) {
+          store.set(activeQueryListenersState.atom, []);
+          store.set(shouldDestroyEventStreamState.atom, true);
+
+          return;
+        }
+
+        throw new Error(`Unhandled error for event stream: ${error.message}`);
+      }
+    },
+    [store],
+  );
+
+  const syncAdditions = useCallback(async () => {
    if (!isDefined(sseEventStreamId)) {
      return;
    }

    const requiredQueryListeners = store.get(requiredQueryListenersState.atom);
-
    const activeQueryListeners = store.get(activeQueryListenersState.atom);

    const queryListenersToAdd = requiredQueryListeners.filter(
@@ -57,12 +79,9 @@ export const SSEQuerySubscribeEffect = () => {
        ),
    );

-    const queryListenersToRemove = activeQueryListeners.filter(
-      (listener) =>
-        !requiredQueryListeners.some(
-          (requiredListener) => requiredListener.queryId === listener.queryId,
-        ),
-    );
+    if (queryListenersToAdd.length === 0) {
+      return;
+    }

    try {
      for (const queryListenerToAdd of queryListenersToAdd) {
@@ -83,7 +102,46 @@ export const SSEQuerySubscribeEffect = () => {
          return;
        }
      }
+    } catch (error) {
+      handleError(error);

+      return;
+    }
+
+    const currentActive = store.get(activeQueryListenersState.atom);
+
+    store.set(activeQueryListenersState.atom, [
+      ...currentActive,
+      ...queryListenersToAdd,
+    ]);
+  }, [addQueryToEventStream, handleError, sseEventStreamId, store]);
+
+  const syncRemovals = useCallback(async () => {
+    if (!isDefined(sseEventStreamId)) {
+      return;
+    }
+
+    const freshRequiredQueryListeners = store.get(
+      requiredQueryListenersState.atom,
+    );
+    const activeQueryListeners = store.get(activeQueryListenersState.atom);
+
+    const queryListenersToRemove = activeQueryListeners.filter(
+      (listener) =>
+        !freshRequiredQueryListeners.some(
+          (requiredListener) => requiredListener.queryId === listener.queryId,
+        ),
+    );
+
+    if (queryListenersToRemove.length === 0) {
+      return;
+    }
+
+    const removedQueryIds = new Set(
+      queryListenersToRemove.map((listener) => listener.queryId),
+    );
+
+    try {
      for (const queryListenerToRemove of queryListenersToRemove) {
        const result = await removeQueryFromEventStream({
          variables: {
@@ -102,60 +160,72 @@ export const SSEQuerySubscribeEffect = () => {
        }
      }
    } catch (error) {
-      if (CombinedGraphQLErrors.is(error)) {
-        const extensions = getGraphqlErrorExtensionsFromError(error);
+      handleError(error);

-        if (
-          isGracefullyHandledEventStreamError({
-            subCode: extensions?.subCode,
-            code: extensions?.code,
-          })
-        ) {
-          store.set(activeQueryListenersState.atom, []);
-          store.set(shouldDestroyEventStreamState.atom, true);
-
-          return;
-        }
-
-        throw new Error(`Unhandled error for event stream: ${error.message}`);
-      }
+      return;
    }

-    store.set(activeQueryListenersState.atom, requiredQueryListeners);
-  }, [
-    addQueryToEventStream,
-    removeQueryFromEventStream,
-    sseEventStreamId,
-    store,
-  ]);
+    const currentActive = store.get(activeQueryListenersState.atom);

-  const debouncedUpdateQueryListeners = useDebouncedCallback(
-    updateQueryListeners,
-    1000,
-    { leading: true },
-  );
+    store.set(
+      activeQueryListenersState.atom,
+      currentActive.filter(
+        (listener) => !removedQueryIds.has(listener.queryId),
+      ),
+    );
+  }, [handleError, removeQueryFromEventStream, sseEventStreamId, store]);
+
+  const debouncedSyncAdditions = useDebouncedCallback(syncAdditions, 1000, {
+    leading: true,
+  });
+
+  const debouncedSyncRemovals = useDebouncedCallback(syncRemovals, 200, {
+    leading: false,
+  });

  useEffect(() => {
    if (!isNonEmptyString(sseEventStreamId) || !sseEventStreamReady) {
      return;
    }

-    const areRequiredQueryListenersDifferentFromActiveQueryListeners =
-      compareArraysOfObjectsByProperty(
-        requiredQueryListeners,
-        activeQueryListeners,
-        'queryId',
-      );
+    const areDifferent = compareArraysOfObjectsByProperty(
+      requiredQueryListeners,
+      activeQueryListeners,
+      'queryId',
+    );

-    if (areRequiredQueryListenersDifferentFromActiveQueryListeners) {
-      debouncedUpdateQueryListeners();
+    if (!areDifferent) {
+      return;
+    }
+
+    const hasAdditions = requiredQueryListeners.some(
+      (listener) =>
+        !activeQueryListeners.some(
+          (activeListener) => activeListener.queryId === listener.queryId,
+        ),
+    );
+
+    const hasRemovals = activeQueryListeners.some(
+      (listener) =>
+        !requiredQueryListeners.some(
+          (requiredListener) => requiredListener.queryId === listener.queryId,
+        ),
+    );
+
+    if (hasAdditions) {
+      debouncedSyncAdditions();
+    }
+
+    if (hasRemovals) {
+      debouncedSyncRemovals();
    }
  }, [
    sseEventStreamId,
    sseEventStreamReady,
    requiredQueryListeners,
    activeQueryListeners,
-    debouncedUpdateQueryListeners,
+    debouncedSyncAdditions,
+    debouncedSyncRemovals,
  ]);

  return null;
@@ -6,6 +6,7 @@ import { type FieldMetadata } from '@/object-record/record-field/ui/types/FieldM
 import { SidePanelFooter } from '@/ui/layout/side-panel/components/SidePanelFooter';
 import { useWorkflowRunIdOrThrow } from '@/workflow/hooks/useWorkflowRunIdOrThrow';
 import { type WorkflowFormAction } from '@/workflow/types/Workflow';
+import { WorkflowRunSSESubscribeEffect } from '@/workflow/workflow-diagram/components/WorkflowRunSSESubscribeEffect';
 import { WorkflowStepBody } from '@/workflow/workflow-steps/components/WorkflowStepBody';
 import { useUpdateWorkflowRunStep } from '@/workflow/workflow-steps/hooks/useUpdateWorkflowRunStep';
 import { WorkflowFormFieldInput } from '@/workflow/workflow-steps/workflow-actions/components/WorkflowFormFieldInput';
@@ -100,6 +101,7 @@ export const WorkflowEditActionFormFiller = ({

  return (
    <>
+      <WorkflowRunSSESubscribeEffect workflowRunId={workflowRunId} />
      <WorkflowStepBody>
        {formData.map((field) => {
          if (field.type === 'RECORD') {