fix(#707): scope session GC retention (#720)
Some checks failed
ci/woodpecker/push/ci Pipeline was successful
ci/woodpecker/push/publish Pipeline was canceled

This commit was merged in pull request #720.
This commit is contained in:
2026-07-13 00:44:19 +00:00
parent e92186d768
commit 753a360517
10 changed files with 163 additions and 175 deletions

View File

@@ -3,6 +3,7 @@ import { Logger } from '@nestjs/common';
import type { QueueHandle } from '@mosaicstack/queue';
import type { LogService } from '@mosaicstack/log';
import { SessionGCService } from './session-gc.service.js';
import { CommandAuthorizationService } from '../commands/command-authorization.service.js';
type MockRedis = {
scan: ReturnType<typeof vi.fn>;
@@ -12,7 +13,12 @@ type MockRedis = {
describe('SessionGCService', () => {
let service: SessionGCService;
let mockRedis: MockRedis;
let mockLogService: { logs: { promoteToWarm: ReturnType<typeof vi.fn> } };
let mockLogService: {
logs: {
promoteSessionToWarm: ReturnType<typeof vi.fn>;
promoteToWarm: ReturnType<typeof vi.fn>;
};
};
/**
* Helper: build a scan mock that returns all provided keys in a single
@@ -30,6 +36,7 @@ describe('SessionGCService', () => {
mockLogService = {
logs: {
promoteSessionToWarm: vi.fn().mockResolvedValue(0),
promoteToWarm: vi.fn().mockResolvedValue(0),
},
};
@@ -59,54 +66,76 @@ describe('SessionGCService', () => {
expect(result.cleaned.valkeyKeys).toBeUndefined();
});
it('escapes glob metacharacters in a session identifier', async () => {
await service.collect('abc*?[tenant]\\escape');
expect(mockRedis.scan).toHaveBeenCalledWith(
'0',
'MATCH',
'mosaic:session:abc\\*\\?\\[tenant\\]\\\\escape:*',
'COUNT',
100,
);
});
it('preserves a valid durable approval after session GC', async () => {
const entries = new Map<string, string>();
const redis = {
scan: vi.fn().mockResolvedValue(['0', ['mosaic:session:owned:state']]),
get: vi.fn(async (key: string) => entries.get(key) ?? null),
set: vi.fn(async (key: string, value: string) => entries.set(key, value)),
del: vi.fn(async (...keys: string[]) => {
let deleted = 0;
for (const key of keys) deleted += Number(entries.delete(key));
return deleted;
}),
};
const authorization = new CommandAuthorizationService(
{
select: () => ({
from: () => ({ where: () => ({ limit: async () => [{ role: 'admin' }] }) }),
}),
} as never,
redis,
);
const command = {
name: 'gc',
description: 'System-wide garbage collection',
aliases: [],
scope: 'admin',
execution: 'socket',
available: true,
} as never;
const payload = { command: 'gc', conversationId: 'owned' };
const approval = await authorization.createApproval(command, payload, 'admin-1');
const approvalKey = `tess:command-approval:${approval!.approvalId}`;
const gc = new SessionGCService(redis as never, mockLogService as unknown as LogService);
await gc.collect('owned');
expect(entries.has(approvalKey)).toBe(true);
await expect(
authorization.authorize(command, payload, 'admin-1', approval!.approvalId),
).resolves.toEqual({ allowed: true });
});
it('collect() returns sessionId in result', async () => {
const result = await service.collect('test-session-id');
expect(result.sessionId).toBe('test-session-id');
});
it('fullCollect() deletes all session keys', async () => {
mockRedis.scan = makeScanMock(['mosaic:session:abc:system', 'mosaic:session:xyz:foo']);
const result = await service.fullCollect();
expect(mockRedis.del).toHaveBeenCalled();
expect(result.valkeyKeys).toBe(2);
it('collect() demotes logs only for the requested session', async () => {
await service.collect('owned-session');
expect(mockLogService.logs.promoteSessionToWarm).toHaveBeenCalledWith(
'owned-session',
expect.any(Date),
);
expect(mockLogService.logs.promoteToWarm).not.toHaveBeenCalled();
});
it('fullCollect() with no keys returns 0 valkeyKeys', async () => {
mockRedis.scan = makeScanMock([]);
const result = await service.fullCollect();
expect(result.valkeyKeys).toBe(0);
expect(mockRedis.del).not.toHaveBeenCalled();
});
it('fullCollect() returns duration', async () => {
const result = await service.fullCollect();
expect(result.duration).toBeGreaterThanOrEqual(0);
});
it('sweepOrphans() extracts unique session IDs and collects them', async () => {
// First scan call returns the global session list; subsequent calls return
// per-session keys during collect().
mockRedis.scan = vi
.fn()
.mockResolvedValueOnce([
'0',
['mosaic:session:abc:system', 'mosaic:session:abc:messages', 'mosaic:session:xyz:system'],
])
// collect('abc') scan
.mockResolvedValueOnce(['0', ['mosaic:session:abc:system', 'mosaic:session:abc:messages']])
// collect('xyz') scan
.mockResolvedValueOnce(['0', ['mosaic:session:xyz:system']]);
mockRedis.del.mockResolvedValue(1);
const result = await service.sweepOrphans();
expect(result.orphanedSessions).toBeGreaterThanOrEqual(0);
expect(result.duration).toBeGreaterThanOrEqual(0);
});
it('sweepOrphans() returns empty when no session keys', async () => {
mockRedis.scan = makeScanMock([]);
const result = await service.sweepOrphans();
expect(result.orphanedSessions).toBe(0);
expect(result.totalCleaned).toHaveLength(0);
it('does not expose automatic global GC entry points', () => {
expect('fullCollect' in service).toBe(false);
expect('sweepOrphans' in service).toBe(false);
});
});

View File

@@ -1,4 +1,4 @@
import { Inject, Injectable, Logger, type OnModuleInit } from '@nestjs/common';
import { Inject, Injectable } from '@nestjs/common';
import type { QueueHandle } from '@mosaicstack/queue';
import type { LogService } from '@mosaicstack/log';
import { LOG_SERVICE } from '../log/log.tokens.js';
@@ -13,49 +13,18 @@ export interface GCResult {
};
}
export interface GCSweepResult {
orphanedSessions: number;
totalCleaned: GCResult[];
duration: number;
}
export interface FullGCResult {
valkeyKeys: number;
logsDemoted: number;
jobsPurged: number;
tempFilesRemoved: number;
duration: number;
/** Escape Redis glob metacharacters so a session identifier is always literal. */
function escapeRedisGlobLiteral(value: string): string {
return value.replace(/[\\*?\[\]]/g, '\\$&');
}
@Injectable()
export class SessionGCService implements OnModuleInit {
private readonly logger = new Logger(SessionGCService.name);
export class SessionGCService {
constructor(
@Inject(REDIS) private readonly redis: QueueHandle['redis'],
@Inject(LOG_SERVICE) private readonly logService: LogService,
) {}
onModuleInit(): void {
// Fire-and-forget: run full GC asynchronously so it does not block the
// NestJS bootstrap chain. Cold-start GC typically takes 100500 ms
// depending on Valkey key count; deferring it removes that latency from
// the TTFB of the first HTTP request.
this.fullCollect()
.then((result) => {
this.logger.log(
`Full GC complete: ${result.valkeyKeys} Valkey keys, ` +
`${result.logsDemoted} logs demoted, ` +
`${result.jobsPurged} jobs purged, ` +
`${result.tempFilesRemoved} temp dirs removed ` +
`(${result.duration}ms)`,
);
})
.catch((err: unknown) => {
this.logger.error('Cold-start GC failed', err instanceof Error ? err.stack : String(err));
});
}
/**
* Scan Valkey for all keys matching a pattern using SCAN (non-blocking).
* KEYS is avoided because it blocks the Valkey event loop for the full scan
@@ -79,86 +48,20 @@ export class SessionGCService implements OnModuleInit {
const result: GCResult = { sessionId, cleaned: {} };
// 1. Valkey: delete all session-scoped keys
const pattern = `mosaic:session:${sessionId}:*`;
const pattern = `mosaic:session:${escapeRedisGlobLiteral(sessionId)}:*`;
const valkeyKeys = await this.scanKeys(pattern);
if (valkeyKeys.length > 0) {
await this.redis.del(...valkeyKeys);
result.cleaned.valkeyKeys = valkeyKeys.length;
}
// 2. PG: demote hot-tier agent_logs for this session to warm
const cutoff = new Date(); // demote all hot logs for this session
const logsDemoted = await this.logService.logs.promoteToWarm(cutoff);
// 2. PG: demote hot-tier agent logs for this session only.
const cutoff = new Date();
const logsDemoted = await this.logService.logs.promoteSessionToWarm(sessionId, cutoff);
if (logsDemoted > 0) {
result.cleaned.logsDemoted = logsDemoted;
}
return result;
}
/**
* Sweep GC — find orphaned artifacts from dead sessions.
* System-wide operation: only call from admin-authorized paths or internal
* scheduled jobs. Individual session cleanup is handled by collect().
*/
async sweepOrphans(): Promise<GCSweepResult> {
const start = Date.now();
const cleaned: GCResult[] = [];
// 1. Find all session-scoped Valkey keys (non-blocking SCAN)
const allSessionKeys = await this.scanKeys('mosaic:session:*');
// Extract unique session IDs from keys
const sessionIds = new Set<string>();
for (const key of allSessionKeys) {
const match = key.match(/^mosaic:session:([^:]+):/);
if (match) sessionIds.add(match[1]!);
}
// 2. For each session ID, collect stale keys
for (const sessionId of sessionIds) {
const gcResult = await this.collect(sessionId);
if (Object.keys(gcResult.cleaned).length > 0) {
cleaned.push(gcResult);
}
}
return {
orphanedSessions: cleaned.length,
totalCleaned: cleaned,
duration: Date.now() - start,
};
}
/**
* Full GC — aggressive collection for cold start.
* Assumes no sessions survived the restart.
*/
async fullCollect(): Promise<FullGCResult> {
const start = Date.now();
// 1. Valkey: delete ALL session-scoped keys (non-blocking SCAN)
const sessionKeys = await this.scanKeys('mosaic:session:*');
if (sessionKeys.length > 0) {
await this.redis.del(...sessionKeys);
}
// 2. NOTE: channel keys are NOT collected on cold start
// (discord/telegram plugins may reconnect and resume)
// 3. PG: demote stale hot-tier logs older than 24h to warm
const hotCutoff = new Date(Date.now() - 24 * 60 * 60 * 1000);
const logsDemoted = await this.logService.logs.promoteToWarm(hotCutoff);
// 4. No summarization job purge API available yet
const jobsPurged = 0;
return {
valkeyKeys: sessionKeys.length,
logsDemoted,
jobsPurged,
tempFilesRemoved: 0,
duration: Date.now() - start,
};
}
}