perf: switch to sharded adapter & dynamic subscription mode

Author: MartinKolarik

Diff for: .github/workflows/ci.yml

@@ -15,7 +15,7 @@ jobs:
 
     services:
       redis:
-        image: redislabs/rejson:latest
+        image: redis/redis-stack-server:latest
         ports:
           - 6379:6379
         options: >-

Diff for: package-lock.json

@@ -9,7 +9,7 @@
 		"@koa/router": "^12.0.1",
 		"@maxmind/geoip2-node": "^4.2.0",
 		"@redocly/openapi-core": "^1.6.0",
-		"@socket.io/redis-adapter": "^8.2.1",
+		"@socket.io/redis-adapter": "github:MartinKolarik/socket.io-redis-adapter#dist",
 		"adm-zip": "^0.5.10",
 		"any-ascii": "^0.3.2",
 		"bluebird": "^3.7.2",

Diff for: package.json

@@ -1,5 +1,5 @@
 import { type RemoteSocket, Server, Socket } from 'socket.io';
-import { createAdapter } from '@socket.io/redis-adapter';
+import { createShardedAdapter } from '@socket.io/redis-adapter';
 // eslint-disable-next-line n/no-missing-import
 import type { DefaultEventsMap } from 'socket.io/dist/typed-events.js';
 import type { Probe } from '../../probe/types.js';
@@ -32,8 +32,8 @@ export const initWsServer = async () => {
 		pingTimeout: 3000,
 	});
 
-	io.adapter(createAdapter(pubClient, subClient, {
-		publishOnSpecificResponseChannel: true,
+	io.adapter(createShardedAdapter(pubClient, subClient, {
+		subscriptionMode: 'dynamic',
 	}));
 };
 

Diff for: src/lib/ws/server.ts

@@ -1,7 +1,7 @@
 import config from 'config';
 import type { Server } from 'socket.io';
 import createHttpError from 'http-errors';
-import { getWsServer } from '../lib/ws/server.js';
+import { getWsServer, PROBES_NAMESPACE } from '../lib/ws/server.js';
 import { getProbeRouter, type ProbeRouter } from '../probe/router.js';
 import type { Probe } from '../probe/types.js';
 import { getMetricsAgent, type MetricsAgent } from '../lib/metrics.js';
@@ -61,7 +61,7 @@ export class MeasurementRunner {
 		const maxInProgressProbes = config.get<number>('measurement.maxInProgressProbes');
 		onlineProbesMap.forEach((probe, index) => {
 			const inProgressUpdates = request.inProgressUpdates && inProgressProbes++ < maxInProgressProbes;
-			this.io.of('probes').to(probe.client).emit('probe:measurement:request', {
+			this.io.of(PROBES_NAMESPACE).to(probe.client).emit('probe:measurement:request', {
 				measurementId,
 				testId: index.toString(),
 				measurement: {

Diff for: src/measurement/runner.ts

@@ -40,7 +40,7 @@ describe('MeasurementRunner', () => {
 	beforeEach(() => {
 		sinon.resetHistory();
 		to.returns({ emit });
-		io.of.returns({ to } as any);
+		io.of.withArgs('/probes').returns({ to } as any);
 		store.createMeasurement.resolves('measurementid');
 		testId = 0;
 	});