From f8e482e2a9bd536848eb39a7f99d91692ccf8584 Mon Sep 17 00:00:00 2001 From: digua Date: Sat, 29 Aug 2026 09:15:35 +0800 Subject: [PATCH] fix(relationships): release session databases during graph compute --- .../people/relationships/compute.test.ts | 59 +++++++++++++++++++ .../services/people/relationships/compute.ts | 5 +- .../people/relationships/service.test.ts | 21 ++++++- .../services/people/relationships/service.ts | 3 +- .../people/relationships/signature.test.ts | 50 ++++++++++++++++ .../people/relationships/signature.ts | 3 +- 6 files changed, 135 insertions(+), 6 deletions(-) create mode 100644 packages/node-runtime/src/services/people/relationships/signature.test.ts diff --git a/packages/node-runtime/src/services/people/relationships/compute.test.ts b/packages/node-runtime/src/services/people/relationships/compute.test.ts index a9207b9e..060f99de 100644 --- a/packages/node-runtime/src/services/people/relationships/compute.test.ts +++ b/packages/node-runtime/src/services/people/relationships/compute.test.ts @@ -72,6 +72,7 @@ class TestEnv { this.adapter = { listSessionIds: () => [...this.dbPaths.keys()], + listSessionCandidateIds: () => [...this.dbPaths.keys()], openReadonly: (id) => open(id, true), openWritable: (id) => open(id, false), closeSession: () => {}, @@ -189,6 +190,64 @@ function makeGraphEdge(sourceKey: string, targetKey: string, weight: number): Pe } } +test('processes database candidates without retaining all session connections', (t) => { + const env = new TestEnv() + t.after(() => env.cleanup()) + + for (const id of ['private-a', 'private-b']) { + env.seed({ + id, + platform: 'weixin', + type: 'private', + ownerId: 'owner', + members: [ + { id: 1, platformId: 'owner', accountName: 'Me' }, + { id: 2, platformId: `friend-${id}`, accountName: `Friend ${id}` }, + ], + messages: [ + { id: 1, senderId: 1, ts: 1704103200 }, + { id: 2, senderId: 2, ts: 1704103201 }, + ], + }) + } + + const baseAdapter = env.adapter + const activeDbs = new Map() + const closedSessionIds = new Set() + let maxActiveDbs = 0 + const adapter: SessionRuntimeAdapter = { + ...baseAdapter, + listSessionIds: () => { + throw new Error('relationship computation must use database candidates') + }, + openReadonly: (sessionId) => { + const db = baseAdapter.openReadonly(sessionId) + if (db) { + activeDbs.set(sessionId, db) + maxActiveDbs = Math.max(maxActiveDbs, activeDbs.size) + } + return db + }, + closeSession: (sessionId) => { + const db = activeDbs.get(sessionId) + if (db) db.close() + activeDbs.delete(sessionId) + closedSessionIds.add(sessionId) + }, + } + + const snapshot = computePeopleRelationshipsSnapshot({ + adapter, + signature: 'sig-candidates', + timeRangePreset: 'all', + }) + + assert.equal(snapshot.workerStats.totalSessions, 2) + assert.equal(maxActiveDbs, 1) + assert.equal(activeDbs.size, 0) + assert.deepEqual([...closedSessionIds].sort(), ['private-a', 'private-b']) +}) + test('computes a cropped relationship galaxy from private contacts and group interaction edges', (t) => { const env = new TestEnv() t.after(() => env.cleanup()) diff --git a/packages/node-runtime/src/services/people/relationships/compute.ts b/packages/node-runtime/src/services/people/relationships/compute.ts index 83c77e0c..68c9283c 100644 --- a/packages/node-runtime/src/services/people/relationships/compute.ts +++ b/packages/node-runtime/src/services/people/relationships/compute.ts @@ -193,7 +193,7 @@ export function computePeopleRelationshipsSnapshot( options: ComputePeopleRelationshipsSnapshotOptions ): PeopleRelationshipsSnapshot { const startedAt = options.now?.() ?? Date.now() - const sessionIds = options.adapter.listSessionIds() + const sessionIds = options.adapter.listSessionCandidateIds?.() ?? options.adapter.listSessionIds() const limits = normalizeLimits(options.limits) const factsCache = options.factsCacheDir ? createFactsCacheContext(options.factsCacheDir) : null const timeRange = resolvePeopleRelationshipsTimeRange( @@ -437,6 +437,7 @@ function computePeopleRelationships(options: { diagnostics.skippedFailedSessions++ appLogger.error('people-relationships', `failed to process people relationship session: ${sessionId}`, error) } finally { + options.adapter.closeSession(sessionId) processedSessions++ options.onProgress?.({ processedSessions, totalSessions: options.sessionIds.length, currentSessionId: sessionId }) } @@ -477,6 +478,8 @@ function findGlobalLatestMessageTs( `failed to inspect people relationship session range: ${sessionId}`, error ) + } finally { + adapter.closeSession(sessionId) } } return latest diff --git a/packages/node-runtime/src/services/people/relationships/service.test.ts b/packages/node-runtime/src/services/people/relationships/service.test.ts index 1b6e1ec9..60896fdc 100644 --- a/packages/node-runtime/src/services/people/relationships/service.test.ts +++ b/packages/node-runtime/src/services/people/relationships/service.test.ts @@ -599,11 +599,22 @@ test('starts a background task when snapshot is missing and returns task state i try { let runnerCalls = 0 let releaseRunner!: (snapshot: PeopleRelationshipsSnapshot) => void + let runnerSignature = '' + const baseAdapter = makeAdapter() + const adapter: SessionRuntimeAdapter = { + ...baseAdapter, + listSessionIds: () => { + throw new Error('relationship service must use database candidates') + }, + listSessionCandidateIds: () => ['session-a', 'session-b'], + getDbPath: (sessionId) => path.join('/tmp', `chatlab-${sessionId}.db`), + } const service = createPeopleRelationshipsService({ - adapter: makeAdapter(), + adapter, systemDir: dir, - runner: () => { + runner: ({ signature }) => { runnerCalls++ + runnerSignature = signature return new Promise((resolve) => { releaseRunner = resolve }) @@ -615,8 +626,12 @@ test('starts a background task when snapshot is missing and returns task state i assert.equal(response.cache.status, 'missing') assert.equal(response.task?.status, 'running') + assert.equal(response.task?.totalSessions, 2) assert.equal(runnerCalls, 1) - releaseRunner(makeSnapshot(makeFreshSignature())) + releaseRunner({ + ...makeSnapshot(runnerSignature), + workerStats: { durationMs: 1, totalSessions: 2, processedSessions: 2, skippedFailedSessions: 0 }, + }) await new Promise((resolve) => setTimeout(resolve, 0)) await service.close() } finally { diff --git a/packages/node-runtime/src/services/people/relationships/service.ts b/packages/node-runtime/src/services/people/relationships/service.ts index 6221d936..14a35b23 100644 --- a/packages/node-runtime/src/services/people/relationships/service.ts +++ b/packages/node-runtime/src/services/people/relationships/service.ts @@ -186,6 +186,7 @@ class DefaultPeopleRelationshipsService implements PeopleRelationshipsService { private ensureTaskStarted(signature: string, timeRangePreset: ContactsTimeRangePreset): void { if (this.inFlight) return + const sessionIds = this.deps.adapter.listSessionCandidateIds?.() ?? this.deps.adapter.listSessionIds() const taskId = `people_relationships_${this.now()}_${Math.random().toString(36).slice(2)}` this.task = { id: taskId, @@ -193,7 +194,7 @@ class DefaultPeopleRelationshipsService implements PeopleRelationshipsService { startedAt: this.now(), finishedAt: null, processedSessions: 0, - totalSessions: this.deps.adapter.listSessionIds().length, + totalSessions: sessionIds.length, timeRangePreset, } diff --git a/packages/node-runtime/src/services/people/relationships/signature.test.ts b/packages/node-runtime/src/services/people/relationships/signature.test.ts new file mode 100644 index 00000000..39f5f85f --- /dev/null +++ b/packages/node-runtime/src/services/people/relationships/signature.test.ts @@ -0,0 +1,50 @@ +/** + * Run: pnpm test -- packages/node-runtime/src/services/people/relationships/signature.test.ts + */ + +import assert from 'node:assert/strict' +import fs from 'node:fs' +import os from 'node:os' +import path from 'node:path' +import test from 'node:test' +import type { SessionRuntimeAdapter } from '../../adapters' +import { PEOPLE_RELATIONSHIPS_ALGORITHM_VERSION } from './compute' +import { buildPeopleRelationshipsSignature } from './signature' + +test('builds the signature from database candidates without opening every session', (t) => { + const baseDir = process.env.CHATLAB_TEST_TMPDIR ?? (fs.existsSync('/private/tmp') ? '/private/tmp' : os.tmpdir()) + const dir = fs.mkdtempSync(path.join(baseDir, 'chatlab-relationships-signature-')) + t.after(() => fs.rmSync(dir, { recursive: true, force: true })) + + const pathsById = new Map([ + ['session-b', path.join(dir, 'b.db')], + ['session-a', path.join(dir, 'a.db')], + ]) + for (const dbPath of pathsById.values()) fs.writeFileSync(dbPath, 'db') + + const adapter: SessionRuntimeAdapter = { + listSessionIds: () => { + throw new Error('signature must not validate or open every database') + }, + listSessionCandidateIds: () => [...pathsById.keys()], + openReadonly: () => { + throw new Error('signature must not open databases') + }, + openWritable: () => null, + closeSession: () => {}, + getDbPath: (sessionId) => pathsById.get(sessionId) ?? '', + deleteSessionFile: () => false, + ensureReadonly: () => { + throw new Error('signature must not open databases') + }, + ensureWritable: () => { + throw new Error('signature must not open databases') + }, + } + + const signature = buildPeopleRelationshipsSignature(adapter) + + assert.match(signature, new RegExp(`algorithm:${PEOPLE_RELATIONSHIPS_ALGORITHM_VERSION}`)) + assert.match(signature, /range:1y/) + assert.ok(signature.indexOf('session-a:') < signature.indexOf('session-b:')) +}) diff --git a/packages/node-runtime/src/services/people/relationships/signature.ts b/packages/node-runtime/src/services/people/relationships/signature.ts index 201157a6..7fa62a0a 100644 --- a/packages/node-runtime/src/services/people/relationships/signature.ts +++ b/packages/node-runtime/src/services/people/relationships/signature.ts @@ -12,7 +12,8 @@ export function buildPeopleRelationshipsSignature( `algorithm:${PEOPLE_RELATIONSHIPS_ALGORITHM_VERSION}`, `range:${normalizePeopleRelationshipsTimeRangePreset(timeRangePreset)}`, ] - for (const sessionId of [...adapter.listSessionIds()].sort()) { + const sessionIds = adapter.listSessionCandidateIds?.() ?? adapter.listSessionIds() + for (const sessionId of [...sessionIds].sort()) { const dbPath = adapter.getDbPath(sessionId) parts.push(`${sessionId}:${getDbFileVersion(dbPath)}`) }