From be3737379166bbfa40cc94d650caa29479aafbc1 Mon Sep 17 00:00:00 2001 From: GitHub Actions <41898282+github-actions[bot]@users.noreply.github.com> Date: Mon, 11 Dec 2023 20:28:46 +0000 Subject: [PATCH 1/7] Bumping sdk dependency to 2.52.0-beta.1 --- package.json | 2 +- yarn.lock | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/package.json b/package.json index 7d308f3..bc4794b 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,7 @@ "license": "Apache-2.0", "dependencies": { "@coral-xyz/anchor": "^0.29.0", - "@drift-labs/sdk": "2.52.0-beta.0", + "@drift-labs/sdk": "2.52.0-beta.1", "@opentelemetry/api": "^1.1.0", "@opentelemetry/auto-instrumentations-node": "^0.31.1", "@opentelemetry/exporter-prometheus": "^0.31.0", diff --git a/yarn.lock b/yarn.lock index cc463b0..a64b3b4 100644 --- a/yarn.lock +++ b/yarn.lock @@ -115,10 +115,10 @@ enabled "2.0.x" kuler "^2.0.0" -"@drift-labs/sdk@2.52.0-beta.0": - version "2.52.0-beta.0" - resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.52.0-beta.0.tgz#03da2e192e3f2180238e5e464336e0fc1697c0e0" - integrity sha512-SnuXh06oJFzAvTQZ/dOjcWduhKC8rRMlyTE5Xqb3LfCcEkQ/B+XvxTItCSzAPh33qq0MMhZKjxGdqKezUxnV1w== +"@drift-labs/sdk@2.52.0-beta.1": + version "2.52.0-beta.1" + resolved "https://registry.yarnpkg.com/@drift-labs/sdk/-/sdk-2.52.0-beta.1.tgz#6b26c434edf9e796010b0be358c732cacc251a24" + integrity sha512-qlfMvN9Fc7TOtRXtkQbTgok7nN0pObosfL1oh7uiEvgQrMeQY2lJoUzSHi4U8djxR68MkVnLnwVGABEgKEBdRQ== dependencies: "@coral-xyz/anchor" "0.28.1-beta.2" "@ellipsis-labs/phoenix-sdk" "^1.4.2" From 54d463f84ea472d60dd2672ade08bf20339a45cf Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Mon, 11 Dec 2023 16:15:17 -0800 Subject: [PATCH 2/7] additional ws connection logging --- src/wsConnectionManager.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/wsConnectionManager.ts b/src/wsConnectionManager.ts index bb37b5c..a4e74fe 100644 --- a/src/wsConnectionManager.ts +++ b/src/wsConnectionManager.ts @@ -110,6 +110,7 @@ async function main() { wss.on('connection', (ws: WebSocket) => { console.log('Client connected'); + console.log('Number of active connections:', wss.clients.size); wsConnectionsGauge.inc(); ws.on('message', async (msg) => { @@ -235,6 +236,7 @@ async function main() { // Handle disconnection ws.on('close', () => { + console.log('Client disconnected'); // Clear any existing intervals and timeouts channelSubscribers.forEach((subscribers, channel) => { if (subscribers.delete(ws) && subscribers.size === 0) { @@ -243,6 +245,7 @@ async function main() { subscribedChannels.delete(channel); } }); + console.log('Number of active connections:', wss.clients.size); wsConnectionsGauge.dec(); }); From c1a5afe3d7426fd2a969f79dda596c4a8ce49ec1 Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Mon, 11 Dec 2023 18:52:46 -0800 Subject: [PATCH 3/7] possible memory leak fixes --- package.json | 1 + src/utils/redisClient.ts | 1 + src/wsConnectionManager.ts | 7 ++++--- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/package.json b/package.json index bc4794b..df9c1ce 100644 --- a/package.json +++ b/package.json @@ -56,6 +56,7 @@ "dlob-publish": "ts-node src/publishers/dlobPublisher.ts", "trades-publish": "ts-node src/publishers/tradesPublisher.ts", "ws-manager": "ts-node src/wsConnectionManager.ts", + "ws-manager:inspect": "yarn build && node --inspect ./lib/wsConnectionManager.js", "dev:inspect": "yarn build && node --inspect ./lib/index.js", "dev:debug": "yarn build && node --inspect-brk --inspect=2230 ./lib/index.js", "example": "ts-node example/client.ts", diff --git a/src/utils/redisClient.ts b/src/utils/redisClient.ts index 1bdf392..875fe89 100644 --- a/src/utils/redisClient.ts +++ b/src/utils/redisClient.ts @@ -34,6 +34,7 @@ export const getRedisClient = ( return false; }, maxRetriesPerRequest: null, // unlimited retries + enableOfflineQueue: false, }); redisClient.on('connect', () => { diff --git a/src/wsConnectionManager.ts b/src/wsConnectionManager.ts index a4e74fe..4f98c72 100644 --- a/src/wsConnectionManager.ts +++ b/src/wsConnectionManager.ts @@ -99,7 +99,10 @@ async function main() { const subscribers = channelSubscribers.get(subscribedChannel); if (subscribers) { subscribers.forEach((ws) => { - ws.send(JSON.stringify({ channel: subscribedChannel, data: message })); + if (ws.readyState === WebSocket.OPEN) + ws.send( + JSON.stringify({ channel: subscribedChannel, data: message }) + ); }); } }); @@ -110,7 +113,6 @@ async function main() { wss.on('connection', (ws: WebSocket) => { console.log('Client connected'); - console.log('Number of active connections:', wss.clients.size); wsConnectionsGauge.inc(); ws.on('message', async (msg) => { @@ -245,7 +247,6 @@ async function main() { subscribedChannels.delete(channel); } }); - console.log('Number of active connections:', wss.clients.size); wsConnectionsGauge.dec(); }); From 15579e1b2a76d0c66dda4fc9d8ec5daee756b41f Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Mon, 11 Dec 2023 18:56:19 -0800 Subject: [PATCH 4/7] remove redis config --- src/utils/redisClient.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/src/utils/redisClient.ts b/src/utils/redisClient.ts index 875fe89..1bdf392 100644 --- a/src/utils/redisClient.ts +++ b/src/utils/redisClient.ts @@ -34,7 +34,6 @@ export const getRedisClient = ( return false; }, maxRetriesPerRequest: null, // unlimited retries - enableOfflineQueue: false, }); redisClient.on('connect', () => { From f3a1e9e22db70630291679d10b333c0ef6d1973c Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Tue, 12 Dec 2023 06:32:30 -0800 Subject: [PATCH 5/7] per message deflate false --- src/wsConnectionManager.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/wsConnectionManager.ts b/src/wsConnectionManager.ts index 4f98c72..dab7c22 100644 --- a/src/wsConnectionManager.ts +++ b/src/wsConnectionManager.ts @@ -23,7 +23,11 @@ const wsConnectionsGauge = new Gauge({ }); const server = http.createServer(app); -const wss = new WebSocketServer({ server, path: '/ws' }); +const wss = new WebSocketServer({ + server, + path: '/ws', + perMessageDeflate: false, +}); const REDIS_HOST = process.env.REDIS_HOST || 'localhost'; const REDIS_PORT = process.env.REDIS_PORT || '6379'; From 43dfc3a865a1f7c2eb370980fbeb4ce402ce3a2a Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Tue, 12 Dec 2023 07:36:08 -0800 Subject: [PATCH 6/7] remove unhandled promise process --- src/wsConnectionManager.ts | 4 ---- 1 file changed, 4 deletions(-) diff --git a/src/wsConnectionManager.ts b/src/wsConnectionManager.ts index dab7c22..0f73c25 100644 --- a/src/wsConnectionManager.ts +++ b/src/wsConnectionManager.ts @@ -273,10 +273,6 @@ async function main() { }); } -process.on('unhandledRejection', (reason, promise) => { - console.error('Unhandled Rejection at:', promise, 'reason:', reason); -}); - async function recursiveTryCatch(f: () => void) { try { await f(); From 436e63dbd52ea12316979e23560ea0bb96176c25 Mon Sep 17 00:00:00 2001 From: Nour Alharithi Date: Tue, 12 Dec 2023 09:11:02 -0800 Subject: [PATCH 7/7] proper memory management --- src/wsConnectionManager.ts | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/src/wsConnectionManager.ts b/src/wsConnectionManager.ts index 0f73c25..26eb4b3 100644 --- a/src/wsConnectionManager.ts +++ b/src/wsConnectionManager.ts @@ -103,7 +103,7 @@ async function main() { const subscribers = channelSubscribers.get(subscribedChannel); if (subscribers) { subscribers.forEach((ws) => { - if (ws.readyState === WebSocket.OPEN) + if (ws.readyState === WebSocket.OPEN && ws.bufferedAmount < 20) ws.send( JSON.stringify({ channel: subscribedChannel, data: message }) ); @@ -271,6 +271,19 @@ async function main() { server.on('error', (error) => { console.error('Server error:', error); }); + + // Periodic check for bad clients and disconnect them to alleviate backpressure + setInterval(() => { + let set = new Set(); + for (const [_channel, wsSet] of channelSubscribers) { + set = new Set([...set, ...wsSet]); + } + for (const ws of set) { + if (ws.bufferedAmount > 500) { + ws.close(); + } + } + }, 5000); } async function recursiveTryCatch(f: () => void) {