Skip to content

Reduce complexity of WSS #132

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Yostra wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Reduce complexity of WSS #132

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 1 addition & 89 deletions packages/sync-engine/src/tests/unit/websocket-client.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,19 +270,6 @@ describe('websocket-client', () => {
})

it('should detect stale connection when no pong received within PONG_WAIT', async () => {
// Use a longer reconnect delay so stale detection can occur before proactive reconnect
vi.stubGlobal(
'fetch',
vi.fn().mockResolvedValue({
ok: true,
json: () =>
Promise.resolve({
...mockSessionResponse,
reconnect_delay: 60, // 60s reconnect delay
}),
})
)

const { createStripeWebSocketClient } = await import('../../websocket-client')

const onError = vi.fn()
Expand Down Expand Up @@ -310,82 +297,7 @@ describe('websocket-client', () => {
})
})

describe('Proactive reconnection', () => {
it('should use server-provided reconnect_delay for reconnect interval', async () => {
const { createStripeWebSocketClient } = await import('../../websocket-client')

const onReady = vi.fn()
await createStripeWebSocketClient({
stripeApiKey: 'sk_test_123',
onEvent: vi.fn(),
onReady,
})

// Wait for runLoop to create WebSocket
await vi.advanceTimersByTimeAsync(0)

const wsInstance = wsInstances[0]
wsInstance._triggerOpen()
wsInstance._triggerPong() // Keep connection alive

expect(onReady).toHaveBeenCalledTimes(1)

// Advance to just before reconnect interval (5s from session)
await vi.advanceTimersByTimeAsync(4900)
wsInstance._triggerPong()

// Should not have reconnected yet
expect(wsInstances.length).toBe(1)

// Advance past reconnect interval
await vi.advanceTimersByTimeAsync(200)

// Should have created a new WebSocket for reconnection
expect(wsInstances.length).toBe(2)
})

it('should use default 60s reconnect interval when server does not provide one', async () => {
vi.stubGlobal(
'fetch',
vi.fn().mockResolvedValue({
ok: true,
json: () =>
Promise.resolve({
...mockSessionResponse,
reconnect_delay: 0, // No server-provided delay
}),
})
)

const { createStripeWebSocketClient } = await import('../../websocket-client')

await createStripeWebSocketClient({
stripeApiKey: 'sk_test_123',
onEvent: vi.fn(),
})

// Wait for runLoop to create WebSocket
await vi.advanceTimersByTimeAsync(0)

const wsInstance = wsInstances[0]
wsInstance._triggerOpen()

// Keep connection alive with pongs
for (let i = 0; i < 6; i++) {
await vi.advanceTimersByTimeAsync(9000)
wsInstance._triggerPong()
}

// Should not have reconnected yet (only 54s passed)
expect(wsInstances.length).toBe(1)

// Advance to trigger 60s reconnect
await vi.advanceTimersByTimeAsync(7000)

// Should have created a new WebSocket
expect(wsInstances.length).toBe(2)
})

describe('Reconnection', () => {
it('should reconnect immediately on unexpected disconnect', async () => {
const { createStripeWebSocketClient } = await import('../../websocket-client')

Expand Down
33 changes: 3 additions & 30 deletions packages/sync-engine/src/websocket-client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ const CLI_VERSION = '1.33.0'
const PONG_WAIT = 10 * 1000 // 10 seconds - max time to wait for pong
const PING_PERIOD = (PONG_WAIT * 2) / 10 // 2 seconds - send ping before pong timeout
const CONNECT_ATTEMPT_WAIT = 10 * 1000 // 10 seconds - retry interval on connection failure
const DEFAULT_RECONNECT_INTERVAL = 60 * 1000 // 60 seconds - proactive reconnect interval

export interface WebhookProcessingResult {
status: number
Expand Down Expand Up @@ -117,14 +116,8 @@ export async function createStripeWebSocketClient(
// Create session
const session = await createCliSession(stripeApiKey)

// Server-controlled reconnect interval (default 60s)
const reconnectInterval = session.reconnect_delay
? session.reconnect_delay * 1000
: DEFAULT_RECONNECT_INTERVAL

let ws: WebSocket | null = null
let pingInterval: NodeJS.Timeout | null = null
let reconnectTimer: NodeJS.Timeout | null = null
let connected = false
let shouldRun = true
let lastPongReceived: number = Date.now()
Expand All @@ -138,10 +131,6 @@ export async function createStripeWebSocketClient(
clearInterval(pingInterval)
pingInterval = null
}
if (reconnectTimer) {
clearTimeout(reconnectTimer)
reconnectTimer = null
}
if (ws) {
ws.removeAllListeners()
if (ws.readyState === WebSocket.OPEN || ws.readyState === WebSocket.CONNECTING) {
Expand Down Expand Up @@ -331,30 +320,14 @@ export async function createStripeWebSocketClient(

if (!shouldRun) break

// 2. Connection established - wait for one of these events:
// - stop() called
// - unexpected disconnect (notifyClose)
// - proactive reconnect timer fires
// 2. Connection established - wait for disconnect or stop().
// Ping/pong heartbeat detects stale connections and terminates them,
// which triggers the close handler and unblocks this promise.
await new Promise<void>((resolve) => {
// Set up notifyClose signal
notifyCloseResolve = resolve

// Set up stop signal
stopResolve = resolve

// Set up proactive reconnect timer
reconnectTimer = setTimeout(() => {
// Proactive reconnection to prevent stale connections
cleanupConnection()
resolve()
}, reconnectInterval)
})

// Clean up before next iteration or exit
if (reconnectTimer) {
clearTimeout(reconnectTimer)
reconnectTimer = null
}
notifyCloseResolve = null
stopResolve = null
}
Expand Down