fix: SSE event stream reconnection after idle connection death (#21061)
The SSE event stream could silently die from network partitions, NAT
table flushes, browser tab throttling, or server restarts. When this
happened:
1. The `error` callback only called `captureException` — no reconnection
was triggered
2. The `complete` callback was `() => {}` — a cleanly terminated stream
left the client permanently broken
3. No mechanism existed to detect a silently dead connection where no
FIN/RST was received
## Summary
- **Fix `error`/`complete` callbacks**: The `graphql-sse` subscription's
`error` callback only reported to Sentry, and `complete` was a no-op.
Both now set `shouldDestroyEventStreamState = true` to trigger the
destroy-recreate lifecycle, ensuring detected transport failures and
clean stream terminations lead to automatic reconnection.
- **Add server-side keepalive**: The existing heartbeat timer now runs
every 30s (instead of 6min) and publishes empty events through the Redis
pub/sub channel in addition to refreshing the Redis TTL (throttled to
~6min). Unlike GraphQL Yoga's opaque SSE comment pings, these are real
subscription events that flow through the client's `next`/`message`
handlers.
- **Add client-side keepalive monitor (`SSEKeepAliveEffect`)**: Tracks
the timestamp of the last received event. If no event arrives within 90
seconds (3x the keepalive interval), it clears query listeners and
triggers a stream destroy-recreate cycle.
## Test plan
- [x] Start the app, verify SSE events flow normally (workflow runs
update in real-time)
- [x] Leave the app idle for >90 seconds, then trigger a workflow run —
verify the stream auto-reconnects and events are delivered
- [x] Kill the server, restart it, verify the frontend recovers its
event stream
- [x] Verify keepalive events (empty
`objectRecordEventsWithQueryIds`/`metadataEvents`) appear in browser
network tab every ~30s
- [x] Verify no regressions in SSE-dependent features (record updates,
metadata changes, workflow run visualization)
This commit is contained in:
+1
@@ -0,0 +1 @@
|
||||
export const APPLICATION_KEEPALIVE_INTERVAL_MS = 30 * 1_000; // 30 seconds
|
||||
@@ -16,6 +16,7 @@ import { AuthWorkspace } from 'src/engine/decorators/auth/auth-workspace.decorat
|
||||
import { NoPermissionGuard } from 'src/engine/guards/no-permission.guard';
|
||||
import { UserAuthGuard } from 'src/engine/guards/user-auth.guard';
|
||||
import { WorkspaceAuthGuard } from 'src/engine/guards/workspace-auth.guard';
|
||||
import { APPLICATION_KEEPALIVE_INTERVAL_MS } from 'src/engine/subscriptions/constants/application-keepalive-interval-ms.constant';
|
||||
import { EVENT_STREAM_TTL_MS } from 'src/engine/subscriptions/constants/event-stream-ttl.constant';
|
||||
import { AddQuerySubscriptionInput } from 'src/engine/subscriptions/dtos/add-query-subscription.input';
|
||||
import { EventSubscriptionDTO } from 'src/engine/subscriptions/dtos/event-subscription.dto';
|
||||
@@ -116,17 +117,36 @@ export class EventStreamResolver {
|
||||
throw error;
|
||||
}
|
||||
|
||||
let lastTtlRefreshAt = 0;
|
||||
|
||||
return wrapAsyncIteratorWithLifecycle(iterator, {
|
||||
initialValue: {
|
||||
objectRecordEventsWithQueryIds: [],
|
||||
metadataEvents: [],
|
||||
},
|
||||
onHeartbeat: () =>
|
||||
this.eventStreamService.refreshEventStreamTTL({
|
||||
onHeartbeat: async () => {
|
||||
const now = Date.now();
|
||||
|
||||
if (now - lastTtlRefreshAt > EVENT_STREAM_TTL_MS / 5) {
|
||||
lastTtlRefreshAt = now;
|
||||
await this.eventStreamService.refreshEventStreamTTL({
|
||||
workspaceId: workspace.id,
|
||||
eventStreamChannelId,
|
||||
});
|
||||
}
|
||||
|
||||
await this.subscriptionService.publishToEventStream({
|
||||
workspaceId: workspace.id,
|
||||
eventStreamChannelId,
|
||||
}),
|
||||
heartbeatIntervalMs: EVENT_STREAM_TTL_MS / 5,
|
||||
payload: {
|
||||
objectRecordEventsWithQueryIds: [],
|
||||
metadataEvents: [],
|
||||
},
|
||||
});
|
||||
|
||||
return true;
|
||||
},
|
||||
heartbeatIntervalMs: APPLICATION_KEEPALIVE_INTERVAL_MS,
|
||||
onCleanup: () =>
|
||||
this.eventStreamService.destroyEventStream({
|
||||
workspaceId: workspace.id,
|
||||
|
||||
Reference in New Issue
Block a user