feat!: move probe logs to timescale db

migrations/time-series-1/20260815120000_create-probe-logs.js

@@ -0,0 +1,23 @@
+export const up = async (db) => {
+	await db.schema.createTable('probe_log_counter', (table) => {
+		table.uuid('probeUuid').primary();
+		table.bigInteger('lastAllocatedId').notNullable();
+	});
+
+	await db.schema.createTable('probe_log', (table) => {
+		table.uuid('probeUuid').notNullable();
+		table.bigInteger('probeLogId').notNullable();
+		table.timestamp('timestamp', { useTz: true });
+		table.timestamp('receivedAt', { useTz: true }).notNullable();
+		table.string('level', 8);
+		table.string('scope', 64);
+		table.string('message', 8192).notNullable();
+	});
+
+	await db.raw(`SELECT create_hypertable('probe_log', by_range('receivedAt', INTERVAL '1 day'))`);
+	await db.raw(`SELECT add_retention_policy('probe_log', INTERVAL '30 days')`);
+	await db.raw(`CREATE INDEX probe_log_probe_cursor_idx ON probe_log ("probeUuid", "probeLogId" DESC)`);
+	await db.raw(`CREATE INDEX probe_log_probe_scope_cursor_idx ON probe_log ("probeUuid", "scope", "probeLogId" DESC)`);
+};
+
+export const down = () => {};

src/lib/http/server.ts

@@ -38,6 +38,7 @@ import { registerAlternativeIpRoute } from '../../alternative-ip/route/alternati
 import { registerLimitsRoute } from '../../limits/route/get-limits.js';
 import { blacklist } from './middleware/blacklist.js';
 import { registerGetProbeLogsRoute } from '../../probe/route/get-probe-logs.js';
+import { registerGetProbeLogScopesRoute } from '../../probe/route/get-probe-log-scopes.js';
 import { registerRestartProbeRoute } from '../../probe/route/restart-probe.js';
 import { captureMiddlewareChainSpan, captureMiddlewareSpan } from '../metrics.js';
 import type { IoContext } from '../server.js';
@@ -84,6 +85,8 @@ export const getHttpServer = (ioContext: IoContext) => {
 	registerGetProbesRoute(apiRouter, ioContext);
 	// GET /probes/:id/logs
 	registerGetProbeLogsRoute(apiRouter, ioContext);
+	// GET /probes/log-scopes
+	registerGetProbeLogScopesRoute(apiRouter);
 	// POST /probes/:id/restart
 	registerRestartProbeRoute(apiRouter, ioContext);
 	// POST /send-code

src/lib/redis/scripts.ts

@@ -226,5 +226,125 @@ const claimTimedOutMeasurements = defineScript({
 	},
 });
 
-export const scripts = { recordProgress, recordProgressAppend, recordResult, markFinishedByTimeout, claimTimedOutMeasurements };
+const registerProbeLogScopes = defineScript({
+	SCRIPT: `
+	local reporterKey = KEYS[1]
+	local knownScopesKey = KEYS[2]
+	local activeWindow = tonumber(ARGV[1])
+	local maxScopes = tonumber(ARGV[2])
+	local minReporters = tonumber(ARGV[3])
+	local reporterIdentity = ARGV[4]
+	local now = tonumber(redis.call('TIME')[1])
+	local cutoff = now - activeWindow
+
+	redis.call('ZREMRANGEBYSCORE', reporterKey, '-inf', cutoff)
+
+	local scores = redis.call('ZMSCORE', reporterKey, unpack(ARGV, 5))
+	local newScopes = 0
+
+	for index = 1, #scores do
+		if not scores[index] then
+			newScopes = newScopes + 1
+		end
+	end
+
+	if redis.call('ZCARD', reporterKey) + newScopes > maxScopes then
+		return 0
+	end
+
+	local zaddArguments = {}
+
+	for index = 5, #ARGV do
+		table.insert(zaddArguments, now)
+		table.insert(zaddArguments, ARGV[index])
+	end
+
+	redis.call('ZADD', reporterKey, 'GT', unpack(zaddArguments))
+	redis.call('EXPIRE', reporterKey, activeWindow)
+
+	local knownScopes = redis.call('SMISMEMBER', knownScopesKey, unpack(ARGV, 5))
+	local newKnownScopes = {}
+
+	for index = 3, #KEYS do
+		local scopeKey = KEYS[index]
+		local scopeIndex = index - 2
+
+		redis.call('ZADD', scopeKey, 'GT', now, reporterIdentity)
+		redis.call('ZREMRANGEBYSCORE', scopeKey, '-inf', cutoff)
+		redis.call('EXPIRE', scopeKey, activeWindow)
+
+		if knownScopes[scopeIndex] == 0 and redis.call('ZCARD', scopeKey) >= minReporters then
+			table.insert(newKnownScopes, ARGV[index + 2])
+		end
+	end
+
+	if #newKnownScopes > 0 then
+		redis.call('SADD', knownScopesKey, unpack(newKnownScopes))
+	end
+
+	return 1
+	`,
+	parseCommand (
+		parser: CommandParser,
+		reporterKey: string,
+		knownScopesKey: string,
+		scopeKeys: string[],
+		reporterIdentity: string,
+		activeWindow: number,
+		maxScopes: number,
+		minReporters: number,
+		scopes: string[],
+	) {
+		parser.push((scopeKeys.length + 2).toString());
+		parser.pushKey(reporterKey);
+		parser.pushKey(knownScopesKey);
+		scopeKeys.forEach(scopeKey => parser.pushKey(scopeKey));
+
+		parser.push(
+			activeWindow.toString(),
+			maxScopes.toString(),
+			minReporters.toString(),
+			reporterIdentity,
+			...scopes,
+		);
+	},
+	transformReply (reply: number) {
+		return reply === 1;
+	},
+});
+
+const countProbeLogScopeReporters = defineScript({
+	NUMBER_OF_KEYS: 2,
+	SCRIPT: `
+	local now = tonumber(redis.call('TIME')[1])
+	local cutoff = now - tonumber(ARGV[2])
+
+	redis.call('ZREMRANGEBYSCORE', KEYS[2], '-inf', cutoff)
+	local count = redis.call('ZCARD', KEYS[2])
+
+	if count < tonumber(ARGV[3]) then
+		redis.call('SREM', KEYS[1], ARGV[1])
+	end
+
+	return count
+	`,
+	parseCommand (parser: CommandParser, knownScopesKey: string, scopeKey: string, scope: string, activeWindow: number, minReporters: number) {
+		parser.pushKey(knownScopesKey);
+		parser.pushKey(scopeKey);
+		parser.push(scope, activeWindow.toString(), minReporters.toString());
+	},
+	transformReply (reply: number) {
+		return reply;
+	},
+});
+
+export const scripts = {
+	recordProgress,
+	recordProgressAppend,
+	recordResult,
+	markFinishedByTimeout,
+	claimTimedOutMeasurements,
+	registerProbeLogScopes,
+	countProbeLogScopeReporters,
+};
 export type RedisScripts = typeof scripts;

src/lib/ws/gateway.ts

@@ -14,6 +14,7 @@ import { handleIsIPv4SupportedUpdate, handleIsIPv6SupportedUpdate } from '../../
 import { handleNewLogs } from '../../probe/handler/logs.js';
 import { handleAltIps } from '../../probe/handler/alt-ips.js';
 import { handleAdoptionServerStart } from '../../probe/handler/local-adoption-server.js';
+import { handleLogScopes } from '../../probe/handler/log-scopes.js';
 import { handleSettingsUpdate } from '../../probe/handler/settings.js';
 import type { IoContext } from '../server.js';
 import { metricsAgent } from '../metrics.js';
@@ -45,6 +46,7 @@ export const initGateway = (ioContext: IoContext) => {
 			// Handlers
 			subscribeWithHandler(socket, 'probe:status:update', handleStatusUpdate(probe));
 			subscribeWithAckHandler(socket, 'probe:logs', handleNewLogs(probe));
+			subscribeWithHandler(socket, 'probe:log-scopes', handleLogScopes(probe));
 			subscribeWithAckHandler(socket, 'probe:alt-ips', handleAltIps(probe, altIpsClient));
 			subscribeWithHandler(socket, 'probe:isIPv6Supported:update', handleIsIPv6SupportedUpdate(probe));
 			subscribeWithHandler(socket, 'probe:isIPv4Supported:update', handleIsIPv4SupportedUpdate(probe));

src/probe/handler/log-scopes.ts

@@ -0,0 +1,18 @@
+import { logScopesSchema } from '../schema/probe-response-schema.js';
+import { getProbeLogScopesStorage } from '../log-scopes-storage.js';
+import { ProbeValidatorError } from '../../lib/probe-validator.js';
+import type { SocketProbe } from '../types.js';
+
+const probeLogScopesStorage = getProbeLogScopesStorage();
+
+export const handleLogScopes = (probe: SocketProbe) => async (scopes: string[]): Promise<void> => {
+	const validation = logScopesSchema.validate(scopes);
+
+	if (validation.error) {
+		throw validation.error;
+	}
+
+	if (!await probeLogScopesStorage.writeScopes(probe.ipAddress, validation.value)) {
+		throw new ProbeValidatorError('Probe log scope limit exceeded.');
+	}
+};

src/probe/log-scopes-storage.ts

@@ -0,0 +1,78 @@
+import { getPersistentRedisClient, type RedisClient } from '../lib/redis/persistent-client.js';
+import { getIpKey } from '../lib/ws/helper/probe-ip-limit.js';
+
+export const KNOWN_SCOPES_KEY = 'gp:log-scopes';
+export const SCOPE_KEY_PREFIX = 'gp:log-scope:';
+export const REPORTER_SCOPES_KEY_PREFIX = 'gp:log-scopes:reporter:';
+export const SCOPE_ACTIVE_WINDOW = 864_000; // 10 days in seconds
+export const SCOPE_READ_CACHE_TTL = 3_600_000; // 1 hour in ms
+export const MIN_SCOPE_REPORTERS = 10;
+export const SCOPE_FLEET_SHARE = 0.5;
+export const MAX_SCOPES_PER_REPORTER = 64;
+
+export class ProbeLogScopesStorage {
+	private cachedScopes: string[] | undefined;
+	private cacheExpiresAt = 0;
+
+	constructor (private readonly redis: RedisClient) {}
+
+	async writeScopes (ipAddress: string, scopes: string[]): Promise<boolean> {
+		const uniqueScopes = [ ...new Set(scopes) ];
+
+		if (uniqueScopes.length === 0) {
+			return true;
+		}
+
+		const reporterIdentity = getIpKey(ipAddress);
+		const scopeKeys = uniqueScopes.map(scope => `${SCOPE_KEY_PREFIX}${scope}`);
+
+		// Keep the reporter cap, forward indexes, and known-scope publication atomic across concurrent reports and reads.
+		return this.redis.registerProbeLogScopes(
+			`${REPORTER_SCOPES_KEY_PREFIX}${reporterIdentity}`,
+			KNOWN_SCOPES_KEY,
+			scopeKeys,
+			reporterIdentity,
+			SCOPE_ACTIVE_WINDOW,
+			MAX_SCOPES_PER_REPORTER,
+			MIN_SCOPE_REPORTERS,
+			uniqueScopes,
+		);
+	}
+
+	async readScopes (): Promise<string[]> {
+		if (this.cachedScopes !== undefined && Date.now() < this.cacheExpiresAt) {
+			return this.cachedScopes;
+		}
+
+		const knownScopes = await this.redis.sMembers(KNOWN_SCOPES_KEY);
+
+		const counts = await Promise.all(knownScopes.map(scope => this.redis.countProbeLogScopeReporters(
+			KNOWN_SCOPES_KEY,
+			`${SCOPE_KEY_PREFIX}${scope}`,
+			scope,
+			SCOPE_ACTIVE_WINDOW,
+			MIN_SCOPE_REPORTERS,
+		)));
+
+		const totalReporters = counts.reduce((max, count) => Math.max(max, count), 0);
+		const threshold = Math.max(totalReporters * SCOPE_FLEET_SHARE, MIN_SCOPE_REPORTERS);
+
+		this.cachedScopes = knownScopes
+			.filter((_, index) => counts[index]! >= threshold)
+			.sort();
+
+		this.cacheExpiresAt = Date.now() + SCOPE_READ_CACHE_TTL;
+
+		return this.cachedScopes;
+	}
+}
+
+let probeLogScopesStorage: ProbeLogScopesStorage;
+
+export const getProbeLogScopesStorage = () => {
+	if (!probeLogScopesStorage) {
+		probeLogScopesStorage = new ProbeLogScopesStorage(getPersistentRedisClient());
+	}
+
+	return probeLogScopesStorage;
+};