perf: switch to sharded redis adapter

.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: >-

package.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",

src/lib/ws/server.ts

@@ -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: 'static',
 	}));
 };
 

src/measurement/runner.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: {

test/tests/unit/measurement/runner.test.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;
 	});