feat: set expire to gp:test-to-probe hSet

src/lib/probe-validator.ts

@@ -1,8 +1,11 @@
+import config from 'config';
 import TTLCache from '@isaacs/ttlcache';
 import { type RedisCluster, getMeasurementRedisClient } from './redis/measurement-client.js';
 
 export class ProbeValidator {
-	private readonly testIdToProbeId = new TTLCache<string, string>({ ttl: 60 * 1000 });
+	private readonly testIdToProbeId = new TTLCache<string, string>({
+		ttl: (config.get<number>('measurement.timeout') + 30) * 1000,
+	});
 
 	constructor (private readonly redis: RedisCluster) {}
 
@@ -35,4 +38,12 @@ export class ProbeValidator {
 	}
 }
 
-export const probeValidator = new ProbeValidator(getMeasurementRedisClient());
+let probeValidator: ProbeValidator;
+
+export const getProbeValidator = () => {
+	if (!probeValidator) {
+		probeValidator = new ProbeValidator(getMeasurementRedisClient());
+	}
+
+	return probeValidator;
+};

src/measurement/handler/progress.ts

@@ -1,11 +1,11 @@
 import type { Probe } from '../../probe/types.js';
 import type { MeasurementProgressMessage } from '../types.js';
 import { getMeasurementRunner } from '../runner.js';
-import { probeValidator } from '../../lib/probe-validator.js';
+import { getProbeValidator } from '../../lib/probe-validator.js';
 
 const runner = getMeasurementRunner();
 
 export const handleMeasurementProgress = (probe: Probe) => async (data: MeasurementProgressMessage): Promise<void> => {
-	await probeValidator.validateProbe(data.measurementId, data.testId, probe.uuid);
+	await getProbeValidator().validateProbe(data.measurementId, data.testId, probe.uuid);
 	await runner.recordProgress(data);
 };

src/measurement/handler/request.ts

@@ -1,5 +1,5 @@
 import type { Probe } from '../../probe/types.js';
-import { probeValidator } from '../../lib/probe-validator.js';
+import { getProbeValidator } from '../../lib/probe-validator.js';
 import { MeasurementRequestMessage } from '../types.js';
 
 export const listenMeasurementRequest = (probe: Probe) => (event: string, data: unknown) => {
@@ -8,5 +8,5 @@ export const listenMeasurementRequest = (probe: Probe) => (event: string, data:
 	}
 
 	const message = data as MeasurementRequestMessage;
-	probeValidator.addValidIds(message.measurementId, message.testId, probe.uuid);
+	getProbeValidator().addValidIds(message.measurementId, message.testId, probe.uuid);
 };

src/measurement/handler/result.ts

@@ -1,11 +1,11 @@
 import type { Probe } from '../../probe/types.js';
 import type { MeasurementResultMessage } from '../types.js';
 import { getMeasurementRunner } from '../runner.js';
-import { probeValidator } from '../../lib/probe-validator.js';
+import { getProbeValidator } from '../../lib/probe-validator.js';
 
 const runner = getMeasurementRunner();
 
 export const handleMeasurementResult = (probe: Probe) => async (data: MeasurementResultMessage): Promise<void> => {
-	await probeValidator.validateProbe(data.measurementId, data.testId, probe.uuid);
+	await getProbeValidator().validateProbe(data.measurementId, data.testId, probe.uuid);
 	await runner.recordResult(data);
 };

src/measurement/store.ts

@@ -54,8 +54,6 @@ export class MeasurementStore {
 	async createMeasurement (request: MeasurementRequest, onlineProbesMap: Map<number, Probe>, allProbes: (Probe | OfflineProbe)[]): Promise<string> {
 		const id = cryptoRandomString({ length: 16, type: 'alphanumeric' });
 		const key = getMeasurementKey(id);
-
-		const probesAwaitingTtl = config.get<number>('measurement.timeout') + 5;
 		const startTime = new Date();
 		const results = this.probesToResults(allProbes, request.type);
 
@@ -76,13 +74,14 @@ export class MeasurementStore {
 		const testsToProbes = Object.fromEntries(Array.from(onlineProbesMap, ([ testId, probe ]) => [ `${id}_${testId}`, probe.uuid ]));
 
 		await Promise.all([
-			!_.isEmpty(testsToProbes) && this.redis.hSet('gp:test-to-probe', testsToProbes),
 			this.redis.hSet('gp:in-progress', id, startTime.getTime()),
-			this.redis.set(getMeasurementKey(id, 'probes_awaiting'), onlineProbesMap.size, { EX: probesAwaitingTtl }),
+			this.redis.set(getMeasurementKey(id, 'probes_awaiting'), onlineProbesMap.size, { EX: config.get<number>('measurement.timeout') + 30 }),
 			this.redis.json.set(key, '$', measurementWithoutDefaults),
 			this.redis.json.set(getMeasurementKey(id, 'ips'), '$', allProbes.map(probe => probe.ipAddress)),
 			this.redis.expire(key, config.get<number>('measurement.resultTTL')),
 			this.redis.expire(getMeasurementKey(id, 'ips'), config.get<number>('measurement.resultTTL')),
+			!_.isEmpty(testsToProbes) && this.redis.hSet('gp:test-to-probe', testsToProbes),
+			!_.isEmpty(testsToProbes) && this.redis.hExpire('gp:test-to-probe', Object.keys(testsToProbes), config.get<number>('measurement.timeout') + 30),
 		]);
 
 		return id;

test/tests/unit/measurement/store.test.ts

@@ -38,6 +38,7 @@ describe('measurement store', () => {
 		hScan: sandbox.stub(),
 		hDel: sandbox.stub(),
 		hSet: sandbox.stub(),
+		hExpire: sandbox.stub(),
 		set: sandbox.stub(),
 		expire: sandbox.stub(),
 		del: sandbox.stub(),
@@ -152,16 +153,9 @@ describe('measurement store', () => {
 
 		expect(redisMock.hSet.callCount).to.equal(2);
 
-		expect(redisMock.hSet.args[0]).to.deep.equal([ 'gp:test-to-probe', {
-			measurementid_0: 'z-z-z-z-z',
-			measurementid_1: '10-10-10-10-10',
-			measurementid_2: 'x-x-x-x-x',
-			measurementid_3: '0-0-0-0-0',
-		}]);
-
-		expect(redisMock.hSet.args[1]).to.deep.equal([ 'gp:in-progress', 'measurementid', now ]);
+		expect(redisMock.hSet.args[0]).to.deep.equal([ 'gp:in-progress', 'measurementid', now ]);
 		expect(redisMock.set.callCount).to.equal(1);
-		expect(redisMock.set.args[0]).to.deep.equal([ 'gp:m:{measurementid}:probes_awaiting', 4, { EX: 35 }]);
+		expect(redisMock.set.args[0]).to.deep.equal([ 'gp:m:{measurementid}:probes_awaiting', 4, { EX: 60 }]);
 		expect(redisMock.json.set.callCount).to.equal(2);
 
 		expect(redisMock.json.set.args[0]).to.deep.equal([ 'gp:m:{measurementid}:results', '$', {
@@ -244,6 +238,26 @@ describe('measurement store', () => {
 		expect(redisMock.json.set.args[1]).to.deep.equal([ 'gp:m:{measurementid}:ips', '$', [ '1.1.1.1', '2.2.2.2', '3.3.3.3', '4.4.4.4' ] ]);
 
 		expect(redisMock.expire.args[1]).to.deep.equal([ 'gp:m:{measurementid}:ips', 604800 ]);
+
+		expect(redisMock.hSet.args[1]).to.deep.equal([ 'gp:test-to-probe', {
+			measurementid_0: 'z-z-z-z-z',
+			measurementid_1: '10-10-10-10-10',
+			measurementid_2: 'x-x-x-x-x',
+			measurementid_3: '0-0-0-0-0',
+		}]);
+
+		expect(redisMock.hExpire.callCount).to.equal(1);
+
+		expect(redisMock.hExpire.args[0]).to.deep.equal([
+			'gp:test-to-probe',
+			[
+				'measurementid_0',
+				'measurementid_1',
+				'measurementid_2',
+				'measurementid_3',
+			],
+			60,
+		]);
 	});
 
 	it('should initialize measurement object with the proper default values', async () => {
@@ -409,7 +423,7 @@ describe('measurement store', () => {
 			},
 		]);
 
-		expect(redisMock.set.args[0]).to.deep.equal([ 'gp:m:{measurementid}:probes_awaiting', 0, { EX: 35 }]);
+		expect(redisMock.set.args[0]).to.deep.equal([ 'gp:m:{measurementid}:probes_awaiting', 0, { EX: 60 }]);
 	});
 
 	it('should store non-default fields of the measurement request', async () => {

test/tests/unit/probe-validator.test.ts

@@ -1,18 +1,41 @@
 import { expect } from 'chai';
-import { probeValidator } from '../../../src/lib/probe-validator.js';
+import * as sinon from 'sinon';
+import { ProbeValidator } from '../../../src/lib/probe-validator.js';
+import { RedisCluster } from '../../../src/lib/redis/shared.js';
 
 describe('ProbeValidator', () => {
+	const sandbox = sinon.createSandbox();
+	const redis = { hGet: sandbox.stub() };
+	const probeValidator = new ProbeValidator(redis as unknown as RedisCluster);
+
+	beforeEach(() => {
+		redis.hGet.resolves(undefined);
+	});
+
 	it('should pass through valid probe id', async () => {
 		probeValidator.addValidIds('measurement-id', 'test-id', 'probe-uuid');
-		probeValidator.validateProbe('measurement-id', 'test-id', 'probe-uuid');
+		await probeValidator.validateProbe('measurement-id', 'test-id', 'probe-uuid');
 	});
 
 	it('should throw for invalid probe id', async () => {
 		probeValidator.addValidIds('measurement-id', 'test-id', 'probe-uuid');
-		expect(() => probeValidator.validateProbe('measurement-id', 'test-id', 'invalid-probe-uuid')).to.throw();
+		const error = await probeValidator.validateProbe('measurement-id', 'test-id', 'invalid-probe-uuid').catch(err => err);
+		expect(error.message).to.equal('Probe ID is wrong for key measurement-id_test-id. Expected: probe-uuid, actual: invalid-probe-uuid');
 	});
 
 	it('should throw for missing key', async () => {
-		expect(() => probeValidator.validateProbe('missing-measurement-id', 'test-id', 'probe-uuid')).to.throw();
+		const error = await probeValidator.validateProbe('missing-measurement-id', 'test-id', 'probe-uuid').catch(err => err);
+		expect(error.message).to.equal('Probe ID not found for key missing-measurement-id_test-id');
+	});
+
+	it('should search key in redis if not found locally', async () => {
+		redis.hGet.resolves('probe-uuid');
+		await probeValidator.validateProbe('only-redis-measurement-id', 'test-id', 'probe-uuid');
+	});
+
+	it('should throw if redis probe id is different', async () => {
+		redis.hGet.resolves('different-probe-uuid');
+		const error = await probeValidator.validateProbe('only-redis-measurement-id', 'test-id', 'probe-uuid').catch(err => err);
+		expect(error.message).to.equal('Probe ID is wrong for key only-redis-measurement-id_test-id. Expected: different-probe-uuid, actual: probe-uuid');
 	});
 });