From 48f4c1661b3f977dccf5512562aa58c654442d75 Mon Sep 17 00:00:00 2001 From: Thomas Trompette Date: Mon, 27 Jul 2026 18:21:08 +0200 Subject: [PATCH] fix(sse): resync records on reconnect and recover from silent query listener errors (#23357) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Context Live updates stop working in prod in a way that only a page refresh fixes. Two independent holes in the SSE self-healing path, both silent. ## 1. Reconnecting restored the stream but never the data Events emitted while the stream was down are not replayed, so any record change during the gap stayed missing from the UI indefinitely. On reconnect `SSEClientEffect` called `resyncMetadataStore()` and dispatched `SSE_CLIENT_RECONNECTED_EVENT_NAME`, but the only listeners were `AgentChatMessagesFetchEffect` and `AgentChatStreamKeepAliveEffect`. No record surface listened. Metadata self-healed, records did not. This fires on every deploy, laptop sleep and network blip, and the reconnect backoff is a uniform random draw up to 2 minutes, so the gap is routinely long. The event was also incomplete. It was dispatched on graphql-sse transport reconnects only. When the keep-alive watchdog or an error set `shouldDestroyEventStream` and `SSEEventStreamEffect` built a replacement stream, nothing was dispatched at all. `useTriggerEventStreamCreation` now dispatches it from the creation path too, for every stream that replaces an earlier one in the tab. Each surface is wired to the resync path it already uses, through an optional `onSseReconnected` on `useListenToEventsForQuery`. That hook is the single funnel every SSE subscriber already goes through, so the tab reloads exactly what it declared an interest in and nothing else: - Record table: reset virtualization, plus `useRefetchAggregateQueriesForObjectMetadataItem` for the header count, which is served by a separate aggregate query that the row reset does not touch. - Record board: `triggerRecordBoardInitialQuery({ shouldResetScroll: false })`. Scroll position preserved. - Workflow versions: `shouldWorkflowRefetchRequest`. These are the same resets each component already runs on every record event, so the only new thing is the trigger. `SSEClientEffect` keeps the metadata store resync, which is genuinely global; that also fixes the matching gap on the metadata side, since `resyncMetadataStore()` used to be called straight from the graphql-sse `connected` callback and so never ran for a watchdog- or error-driven stream re-creation. **Known gap:** the record show page and workflow run detail are not covered. Both read through `useFindOneRecord`, but their subscription lives in a sibling component with no access to `refetch`. Wiring them needs either `refetch` exposed from `RecordShowEffect` / `useWorkflowRun`, or the subscription moved into the data owner — the latter changes subscription lifetime, and `useListenToEventsForQuery` unregisters by `queryId` on unmount regardless of other consumers. Left out pending a decision. ## 2. A network error on `addQueryToEventStream` silently unsubscribed a view forever `handleError` in `SSEQuerySubscribeEffect` only reacted to `CombinedGraphQLErrors`. On Apollo Client v4 a network failure or a 5xx from a rolling pod surfaces as `ServerError` or a plain `Error`, so the handler was a complete no-op: no Sentry capture, no stream teardown, and `syncAdditions` returned before recording the listener as active. Since neither `requiredQueryListeners` nor `activeQueryListeners` changed, the driving effect never re-ran. That query stayed unregistered server-side for the rest of the session while the stream looked healthy and every other view kept updating live. Recovery required remounting the component or refreshing. The recovery now runs for every error type. `getGraphqlErrorExtensionsFromError` is called unconditionally: it accepts `unknown` and reads `extensions` off any object-shaped error, so an error carrying a gracefully-handled `code` is still recognised as one whether or not it is a `CombinedGraphQLErrors`. Note: errors without extensions now reach Sentry, since `isGracefullyHandledEventStreamError` returns false for them. That is new noise during outages, but this failure class is currently completely invisible. ## Testing - `nx typecheck twenty-front` and oxlint `--type-aware` + oxfmt pass. - Verified on a local instance with an A/B/A run. A row inserted straight into Postgres emits no SSE event, which is exactly the state after a disconnect; restarting the server then forces a reconnect. With the fix the row appears and the count updates with no page reload; reverted to `main` it stays invisible indefinitely. Redis confirmed the stream had reconnected and re-registered its queries in the negative run, so that result is the missing resync rather than a dead stream. - Fix 2 is **not** exercised at runtime — it needs a network-level failure on the `addQueryToEventStream` mutation specifically. Reasoned about only. - There are no existing tests for the `sse-db-event` module. ## Known gap A tab's very first stream is not treated as a reconnection, since `isRecreatedEventStream` is derived from `lastSseEventReceivedTimestampState` already being set. If that first stream connects but never receives its first message and is then replaced, no resync is dispatched. That is the separate issue of `SSEKeepAliveEffect` being gated on `sseEventStreamReady`, which is itself only set by the first message: a stream that never becomes ready is never watched and never torn down. Not addressed here. Review in cubic --- .../RecordBoardSSESubscribeEffect.tsx | 10 ++++- ...cordTableVirtualizedSSESubscribeEffect.tsx | 24 ++++++++++- .../components/SSEClientEffect.tsx | 17 +++++++- .../components/SSEQuerySubscribeEffect.tsx | 36 ++++++++-------- .../constants/SseResyncDebounceTimeInMs.ts | 1 + .../hooks/useListenToEventsForQuery.ts | 41 ++++++++++++++++++- .../hooks/useTriggerEventStreamCreation.ts | 29 +++++++++---- .../components/WorkflowSSESubscribeEffect.tsx | 11 ++--- 8 files changed, 133 insertions(+), 36 deletions(-) create mode 100644 packages/twenty-front/src/modules/sse-db-event/constants/SseResyncDebounceTimeInMs.ts diff --git a/packages/twenty-front/src/modules/object-record/record-board/components/RecordBoardSSESubscribeEffect.tsx b/packages/twenty-front/src/modules/object-record/record-board/components/RecordBoardSSESubscribeEffect.tsx index f47de6b6f2..59583df0f2 100644 --- a/packages/twenty-front/src/modules/object-record/record-board/components/RecordBoardSSESubscribeEffect.tsx +++ b/packages/twenty-front/src/modules/object-record/record-board/components/RecordBoardSSESubscribeEffect.tsx @@ -1,6 +1,7 @@ -import { useContext } from 'react'; +import { useCallback, useContext } from 'react'; import { RecordBoardContext } from '@/object-record/record-board/contexts/RecordBoardContext'; +import { useTriggerRecordBoardInitialQuery } from '@/object-record/record-board/hooks/useTriggerRecordBoardInitialQuery'; import { useRecordIndexContextOrThrow } from '@/object-record/record-index/contexts/RecordIndexContext'; import { useRecordIndexGroupCommonQueryVariables } from '@/object-record/record-index/hooks/useRecordIndexGroupCommonQueryVariables'; import { useListenToEventsForQuery } from '@/sse-db-event/hooks/useListenToEventsForQuery'; @@ -10,9 +11,15 @@ export const RecordBoardSSESubscribeEffect = () => { const { objectMetadataItem } = useRecordIndexContextOrThrow(); const { combinedFilters, orderBy } = useRecordIndexGroupCommonQueryVariables(); + const { triggerRecordBoardInitialQuery } = + useTriggerRecordBoardInitialQuery(); const queryId = `record-board-${recordBoardId}`; + const reloadBoard = useCallback(() => { + triggerRecordBoardInitialQuery({ shouldResetScroll: false }); + }, [triggerRecordBoardInitialQuery]); + useListenToEventsForQuery({ queryId, operationSignature: { @@ -22,6 +29,7 @@ export const RecordBoardSSESubscribeEffect = () => { orderBy, }, }, + onSseReconnected: reloadBoard, }); return null; diff --git a/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedSSESubscribeEffect.tsx b/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedSSESubscribeEffect.tsx index a128ade281..97a7928c12 100644 --- a/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedSSESubscribeEffect.tsx +++ b/packages/twenty-front/src/modules/object-record/record-table/virtualization/components/RecordTableVirtualizedSSESubscribeEffect.tsx @@ -1,12 +1,15 @@ -import { useMemo } from 'react'; +import { useCallback, useMemo } from 'react'; import { flattenedFieldMetadataItemsSelector } from '@/object-metadata/states/flattenedFieldMetadataItemsSelector'; import { turnSortsIntoOrderBy } from '@/object-record/object-sort-dropdown/utils/turnSortsIntoOrderBy'; import { currentRecordFilterGroupsComponentState } from '@/object-record/record-filter-group/states/currentRecordFilterGroupsComponentState'; import { useFilterValueDependencies } from '@/object-record/record-filter/hooks/useFilterValueDependencies'; import { currentRecordFiltersComponentState } from '@/object-record/record-filter/states/currentRecordFiltersComponentState'; +import { useRefetchAggregateQueriesForObjectMetadataItem } from '@/object-record/hooks/useRefetchAggregateQueriesForObjectMetadataItem'; import { useRecordIndexContextOrThrow } from '@/object-record/record-index/contexts/RecordIndexContext'; import { currentRecordSortsComponentState } from '@/object-record/record-sort/states/currentRecordSortsComponentState'; +import { useRecordTableContextOrThrow } from '@/object-record/record-table/contexts/RecordTableContext'; +import { useResetVirtualizationBecauseDataChanged } from '@/object-record/record-table/virtualization/hooks/useResetVirtualizationBecauseDataChanged'; import { useListenToEventsForQuery } from '@/sse-db-event/hooks/useListenToEventsForQuery'; import { useAtomComponentStateValue } from '@/ui/utilities/state/jotai/hooks/useAtomComponentStateValue'; import { useAtomStateValue } from '@/ui/utilities/state/jotai/hooks/useAtomStateValue'; @@ -14,8 +17,26 @@ import { computeRecordGqlOperationFilter } from 'twenty-shared/utils'; export const RecordTableVirtualizedSSESubscribeEffect = () => { const { objectMetadataItem } = useRecordIndexContextOrThrow(); + const { objectNameSingular } = useRecordTableContextOrThrow(); const { filterValueDependencies } = useFilterValueDependencies(); + const { resetVirtualizationBecauseDataChanged } = + useResetVirtualizationBecauseDataChanged(objectNameSingular); + + const { refetchAggregateQueriesForObjectMetadataItem } = + useRefetchAggregateQueriesForObjectMetadataItem(); + + const reloadTable = useCallback(async () => { + await Promise.all([ + resetVirtualizationBecauseDataChanged(), + refetchAggregateQueriesForObjectMetadataItem({ objectMetadataItem }), + ]); + }, [ + objectMetadataItem, + refetchAggregateQueriesForObjectMetadataItem, + resetVirtualizationBecauseDataChanged, + ]); + const flattenedFieldMetadataItems = useAtomStateValue( flattenedFieldMetadataItemsSelector, ); @@ -60,6 +81,7 @@ export const RecordTableVirtualizedSSESubscribeEffect = () => { useListenToEventsForQuery({ queryId, operationSignature, + onSseReconnected: reloadTable, }); return null; diff --git a/packages/twenty-front/src/modules/sse-db-event/components/SSEClientEffect.tsx b/packages/twenty-front/src/modules/sse-db-event/components/SSEClientEffect.tsx index c19105ab36..edefa2dbdd 100644 --- a/packages/twenty-front/src/modules/sse-db-event/components/SSEClientEffect.tsx +++ b/packages/twenty-front/src/modules/sse-db-event/components/SSEClientEffect.tsx @@ -1,8 +1,10 @@ import { useHasAccessTokenPair } from '@/auth/hooks/useHasAccessTokenPair'; import { tokenPairState } from '@/auth/states/tokenPairState'; +import { useListenToBrowserEvent } from '@/browser-event/hooks/useListenToBrowserEvent'; import { dispatchBrowserEvent } from '@/browser-event/utils/dispatchBrowserEvent'; import { useResyncMetadataStore } from '@/metadata-store/hooks/useResyncMetadataStore'; import { SSE_CLIENT_RECONNECTED_EVENT_NAME } from '@/sse-db-event/constants/SseClientReconnectedEventName'; +import { SSE_RESYNC_DEBOUNCE_TIME_IN_MS } from '@/sse-db-event/constants/SseResyncDebounceTimeInMs'; import { useHandleSseClientConnectionRetry } from '@/sse-db-event/hooks/useHandleSseClientConnectionRetry'; import { activeQueryListenersState } from '@/sse-db-event/states/activeQueryListenersState'; import { sseClientState } from '@/sse-db-event/states/sseClientState'; @@ -12,6 +14,7 @@ import { isNonEmptyArray } from '@sniptt/guards'; import { createClient } from 'graphql-sse'; import { useCallback, useEffect } from 'react'; import { isDefined } from 'twenty-shared/utils'; +import { useDebouncedCallback } from 'use-debounce'; import { REACT_APP_SERVER_BASE_URL } from '~/config'; import { useStore } from 'jotai'; @@ -22,6 +25,17 @@ export const SSEClientEffect = () => { const tokenPair = useAtomStateValue(tokenPairState); const { resyncMetadataStore } = useResyncMetadataStore(); + const debouncedResyncMetadataStore = useDebouncedCallback( + resyncMetadataStore, + SSE_RESYNC_DEBOUNCE_TIME_IN_MS, + { leading: false }, + ); + + useListenToBrowserEvent({ + eventName: SSE_CLIENT_RECONNECTED_EVENT_NAME, + onBrowserEvent: debouncedResyncMetadataStore, + }); + const handleSSEClientConnected = useCallback( (reconnected: boolean) => { const currentActiveQueryListeners = store.get( @@ -33,11 +47,10 @@ export const SSEClientEffect = () => { } if (reconnected) { - resyncMetadataStore(); dispatchBrowserEvent(SSE_CLIENT_RECONNECTED_EVENT_NAME); } }, - [store, resyncMetadataStore], + [store], ); const { handleSseClientConnectionRetry } = diff --git a/packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx b/packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx index 3f5e046ee0..00693210ed 100644 --- a/packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx +++ b/packages/twenty-front/src/modules/sse-db-event/components/SSEQuerySubscribeEffect.tsx @@ -7,7 +7,6 @@ import { sseEventStreamIdState } from '@/sse-db-event/states/sseEventStreamIdSta import { sseEventStreamReadyState } from '@/sse-db-event/states/sseEventStreamReadyState'; import { isGracefullyHandledEventStreamError } from '@/sse-db-event/utils/isGracefullyHandledEventStreamError'; import { useAtomStateValue } from '@/ui/utilities/state/jotai/hooks/useAtomStateValue'; -import { CombinedGraphQLErrors } from '@apollo/client/errors'; import { captureException } from '@sentry/react'; import { useMutation } from '@apollo/client/react'; import { isNonEmptyString } from '@sniptt/guards'; @@ -44,25 +43,26 @@ export const SSEQuerySubscribeEffect = () => { const handleError = useCallback( (error: unknown) => { - if (CombinedGraphQLErrors.is(error)) { - const extensions = getGraphqlErrorExtensionsFromError(error); + const extensions = getGraphqlErrorExtensionsFromError(error); - if ( - !isGracefullyHandledEventStreamError({ - subCode: extensions?.subCode, - code: extensions?.code, - }) - ) { - captureException( - new Error(`Unhandled error for event stream: ${error.message}`, { - cause: error, - }), - ); - } - - store.set(activeQueryListenersState.atom, []); - store.set(shouldDestroyEventStreamState.atom, true); + if ( + !isGracefullyHandledEventStreamError({ + subCode: extensions?.subCode, + code: extensions?.code, + }) + ) { + captureException( + new Error( + `Unhandled error for event stream: ${ + error instanceof Error ? error.message : String(error) + }`, + { cause: error }, + ), + ); } + + store.set(activeQueryListenersState.atom, []); + store.set(shouldDestroyEventStreamState.atom, true); }, [store], ); diff --git a/packages/twenty-front/src/modules/sse-db-event/constants/SseResyncDebounceTimeInMs.ts b/packages/twenty-front/src/modules/sse-db-event/constants/SseResyncDebounceTimeInMs.ts new file mode 100644 index 0000000000..93f0244889 --- /dev/null +++ b/packages/twenty-front/src/modules/sse-db-event/constants/SseResyncDebounceTimeInMs.ts @@ -0,0 +1 @@ +export const SSE_RESYNC_DEBOUNCE_TIME_IN_MS = 1_000; diff --git a/packages/twenty-front/src/modules/sse-db-event/hooks/useListenToEventsForQuery.ts b/packages/twenty-front/src/modules/sse-db-event/hooks/useListenToEventsForQuery.ts index 399bb2b605..00648be15c 100644 --- a/packages/twenty-front/src/modules/sse-db-event/hooks/useListenToEventsForQuery.ts +++ b/packages/twenty-front/src/modules/sse-db-event/hooks/useListenToEventsForQuery.ts @@ -1,20 +1,28 @@ +import { useListenToBrowserEvent } from '@/browser-event/hooks/useListenToBrowserEvent'; +import { SSE_CLIENT_RECONNECTED_EVENT_NAME } from '@/sse-db-event/constants/SseClientReconnectedEventName'; +import { SSE_RESYNC_DEBOUNCE_TIME_IN_MS } from '@/sse-db-event/constants/SseResyncDebounceTimeInMs'; import { useChangeQueryListenState } from '@/sse-db-event/hooks/useChangeQueryListenState'; -import { useEffect } from 'react'; +import { captureException } from '@sentry/react'; +import { useCallback, useEffect } from 'react'; import { type MetadataGqlOperationSignature, type RecordGqlOperationSignature, } from 'twenty-shared/types'; +import { isDefined } from 'twenty-shared/utils'; +import { useDebouncedCallback } from 'use-debounce'; export const useListenToEventsForQuery = ({ queryId, operationSignature, skip = false, + onSseReconnected, }: { queryId: string; operationSignature: | RecordGqlOperationSignature | MetadataGqlOperationSignature; skip?: boolean; + onSseReconnected?: () => void | Promise; }) => { const { changeQueryIdListenState } = useChangeQueryListenState(); @@ -29,4 +37,35 @@ export const useListenToEventsForQuery = ({ changeQueryIdListenState(false, queryId, operationSignature); }; }, [changeQueryIdListenState, queryId, operationSignature, skip]); + + const handleSseReconnected = useCallback(() => { + if (skip || !isDefined(onSseReconnected)) { + return; + } + + const captureResyncError = (error: unknown) => { + captureException( + new Error(`Failed to resync "${queryId}" after SSE reconnection`, { + cause: error, + }), + ); + }; + + try { + void Promise.resolve(onSseReconnected()).catch(captureResyncError); + } catch (error) { + captureResyncError(error); + } + }, [onSseReconnected, queryId, skip]); + + const debouncedHandleSseReconnected = useDebouncedCallback( + handleSseReconnected, + SSE_RESYNC_DEBOUNCE_TIME_IN_MS, + { leading: false }, + ); + + useListenToBrowserEvent({ + eventName: SSE_CLIENT_RECONNECTED_EVENT_NAME, + onBrowserEvent: debouncedHandleSseReconnected, + }); }; diff --git a/packages/twenty-front/src/modules/sse-db-event/hooks/useTriggerEventStreamCreation.ts b/packages/twenty-front/src/modules/sse-db-event/hooks/useTriggerEventStreamCreation.ts index d15e0a0d92..09bb58a46b 100644 --- a/packages/twenty-front/src/modules/sse-db-event/hooks/useTriggerEventStreamCreation.ts +++ b/packages/twenty-front/src/modules/sse-db-event/hooks/useTriggerEventStreamCreation.ts @@ -1,3 +1,5 @@ +import { dispatchBrowserEvent } from '@/browser-event/utils/dispatchBrowserEvent'; +import { SSE_CLIENT_RECONNECTED_EVENT_NAME } from '@/sse-db-event/constants/SseClientReconnectedEventName'; import { ON_EVENT_SUBSCRIPTION } from '@/sse-db-event/graphql/subscriptions/OnEventSubscription'; import { useDispatchMetadataEventsFromSseToBrowserEvents } from '@/sse-db-event/hooks/useDispatchMetadataEventsFromSseToBrowserEvents'; import { useDispatchObjectRecordEventsFromSseToBrowserEvents } from '@/sse-db-event/hooks/useDispatchObjectRecordEventsFromSseToBrowserEvents'; @@ -67,8 +69,25 @@ export const useTriggerEventStreamCreation = () => { store.set(sseEventStreamIdState.atom, newSseEventStreamId); store.set(sseEventStreamReadyState.atom, false); + const isRecreatedEventStream = isDefined( + store.get(lastSseEventReceivedTimestampState.atom), + ); + let hasReceivedFirstEvent = false; + const handleFirstEventReceived = () => { + if (hasReceivedFirstEvent) { + return; + } + + hasReceivedFirstEvent = true; + store.set(sseEventStreamReadyState.atom, true); + + if (isRecreatedEventStream) { + dispatchBrowserEvent(SSE_CLIENT_RECONNECTED_EVENT_NAME); + } + }; + const dispose = sseClient.subscribe( { query: print(ON_EVENT_SUBSCRIPTION), @@ -107,10 +126,7 @@ export const useTriggerEventStreamCreation = () => { return; } - if (!hasReceivedFirstEvent) { - hasReceivedFirstEvent = true; - store.set(sseEventStreamReadyState.atom, true); - } + handleFirstEventReceived(); const eventSubscription = value?.data?.onEventSubscription; @@ -169,10 +185,7 @@ export const useTriggerEventStreamCreation = () => { store.set(shouldDestroyEventStreamState.atom, true); } else { - if (!hasReceivedFirstEvent) { - hasReceivedFirstEvent = true; - store.set(sseEventStreamReadyState.atom, true); - } + handleFirstEventReceived(); const objectRecordEventsWithQueryIds = result?.data?.onEventSubscription diff --git a/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowSSESubscribeEffect.tsx b/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowSSESubscribeEffect.tsx index 13e48fc304..4b0d969c75 100644 --- a/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowSSESubscribeEffect.tsx +++ b/packages/twenty-front/src/modules/workflow/workflow-diagram/components/WorkflowSSESubscribeEffect.tsx @@ -24,6 +24,10 @@ export const WorkflowSSESubscribeEffect = ({ objectNameSingular: CoreObjectNameSingular.WorkflowVersion, }); + const requestWorkflowRefetch = useCallback(() => { + setShouldWorkflowRefetchRequest(true); + }, [setShouldWorkflowRefetchRequest]); + useListenToEventsForQuery({ queryId, operationSignature: { @@ -34,14 +38,11 @@ export const WorkflowSSESubscribeEffect = ({ }, }, }, + onSseReconnected: requestWorkflowRefetch, }); - const handleWorkflowVersionCreateOne = useCallback(() => { - setShouldWorkflowRefetchRequest(true); - }, [setShouldWorkflowRefetchRequest]); - useListenToObjectRecordOperationBrowserEvent({ - onObjectRecordOperationBrowserEvent: handleWorkflowVersionCreateOne, + onObjectRecordOperationBrowserEvent: requestWorkflowRefetch, objectMetadataItemId: workflowVersionMetadataItem.id, operationTypes: ['create-one'], });