修复完整历史与运行态内存放大并重建发布包
This commit is contained in:
shiyue
314
scripts/session-history-memory-unit.js
Normal file
314
scripts/session-history-memory-unit.js
Normal file
@@ -0,0 +1,314 @@
|
||||
#!/usr/bin/env node
|
||||
'use strict';
|
||||
|
||||
const assert = require('node:assert/strict');
|
||||
const fs = require('node:fs');
|
||||
const os = require('node:os');
|
||||
const path = require('node:path');
|
||||
const crypto = require('node:crypto');
|
||||
const { spawnSync } = require('node:child_process');
|
||||
const { createSessionHistoryStore } = require('../lib/session-history-store');
|
||||
|
||||
const ID = 'synthetic-memory-history';
|
||||
const TARGET_BYTES = Math.round(70.5 * 1024 * 1024);
|
||||
const TAIL_COUNT = 12;
|
||||
const PAGE_COUNT = 17;
|
||||
const SAVE_COUNT = 6;
|
||||
const ORIGINAL_CONTENT = '完整历史：中文、引号"、反斜线\\、换行\n、表情🙂\n' + 'x'.repeat(62 * 1024);
|
||||
const ORIGINAL_RESULT = '完整工具输出：' + 't'.repeat(2 * 1024);
|
||||
const startedAt = Date.now();
|
||||
|
||||
function stage(message) {
|
||||
process.stderr.write(`[历史内存回归 +${((Date.now() - startedAt) / 1000).toFixed(2)} 秒] ${message}\n`);
|
||||
}
|
||||
|
||||
function temporaryRoot() {
|
||||
// 内存回归优先使用真实 tmpfs；慢盘逐块 fsync 可能耗尽 60 秒测试预算。
|
||||
// 文件写入、校验和 fsync 都保持真实调用，原子失败语义由存储单测覆盖。
|
||||
if (process.platform === 'linux' && typeof fs.statfsSync === 'function') {
|
||||
try {
|
||||
const stat = fs.statfsSync('/dev/shm');
|
||||
fs.accessSync('/dev/shm', fs.constants.W_OK);
|
||||
if (Number(stat.bavail) * Number(stat.bsize) >= 512 * 1024 * 1024) return '/dev/shm';
|
||||
} catch {}
|
||||
}
|
||||
return os.tmpdir();
|
||||
}
|
||||
|
||||
function originalMessage(index) {
|
||||
return {
|
||||
id: `original-${index}`,
|
||||
role: index % 2 ? 'assistant' : 'user',
|
||||
content: `${index}:${ORIGINAL_CONTENT}`,
|
||||
timestamp: new Date(1800000000000 + index * 1000).toISOString(),
|
||||
toolCalls: [{
|
||||
id: `tool-${index}`,
|
||||
name: 'synthetic-tool',
|
||||
input: { text: '输入完整保存', nested: { array: [index, { value: '\\"\n🙂' }] } },
|
||||
result: ORIGINAL_RESULT,
|
||||
}],
|
||||
};
|
||||
}
|
||||
|
||||
function addedMessage(index) {
|
||||
return { id: `added-${index}`, role: 'assistant', content: `新增消息 ${index}：` + 'a'.repeat(4096) };
|
||||
}
|
||||
|
||||
function digest(value) {
|
||||
return crypto.createHash('sha256').update(JSON.stringify(value)).digest('hex');
|
||||
}
|
||||
|
||||
// 测试资料由父进程逐条写入，避免测试自身先建立整份大对象。
|
||||
function writeLegacyFixture(sessionsDir) {
|
||||
const historyDir = path.join(sessionsDir, '_history');
|
||||
fs.mkdirSync(historyDir, { recursive: true });
|
||||
const archivePath = path.join(historyDir, `${ID}.json`);
|
||||
const revision = crypto.randomUUID();
|
||||
const metadata = JSON.parse('{"__proto__":{"saved":"原值"},"constructor":"保留键","deep":{"a":{"b":{"c":{"value":"完整值"}}}}}');
|
||||
const session = { id: ID, agent: 'codexapp', codexAppThreadId: 'synthetic-thread', title: '内存回归合成资料', metadata };
|
||||
const head = JSON.stringify({ version: 1, revision, complete: true, resetRevision: null, session }).slice(0, -2) + ',"messages":[';
|
||||
const fd = fs.openSync(archivePath, 'wx', 0o600);
|
||||
let bytes = Buffer.byteLength(head);
|
||||
let count = 0;
|
||||
const paddingHead = '],"padding":"';
|
||||
const ending = '"}}';
|
||||
const paddingOverhead = Buffer.byteLength(paddingHead + ending);
|
||||
let padding;
|
||||
let batch = [];
|
||||
let batchBytes = 0;
|
||||
function flushBatch() {
|
||||
if (!batch.length) return;
|
||||
fs.writeSync(fd, batch.join(''));
|
||||
batch = [];
|
||||
batchBytes = 0;
|
||||
}
|
||||
try {
|
||||
fs.writeSync(fd, head);
|
||||
while (true) {
|
||||
const json = `${count ? ',' : ''}${JSON.stringify(originalMessage(count))}`;
|
||||
if (bytes + Buffer.byteLength(json) + paddingOverhead > TARGET_BYTES) break;
|
||||
batch.push(json);
|
||||
batchBytes += Buffer.byteLength(json);
|
||||
if (batchBytes >= 1024 * 1024) flushBatch();
|
||||
bytes += Buffer.byteLength(json);
|
||||
count += 1;
|
||||
}
|
||||
padding = 'p'.repeat(TARGET_BYTES - bytes - paddingOverhead);
|
||||
flushBatch();
|
||||
fs.writeSync(fd, paddingHead + padding + ending);
|
||||
} finally {
|
||||
fs.closeSync(fd);
|
||||
}
|
||||
return { archivePath, count, bytes: fs.statSync(archivePath).size, metadataHash: digest(metadata), paddingHash: digest(padding) };
|
||||
}
|
||||
|
||||
function createTrackedFilesystem(archivePath) {
|
||||
const opened = new Map();
|
||||
let metrics;
|
||||
function reset(label) {
|
||||
metrics = { label, chunkFiles: new Set(), chunkBytes: 0, legacyBytes: 0, legacyMaxRead: 0,
|
||||
chunkWrites: 0, chunkFsyncMs: 0, chunkFsyncs: 0, nextLegacyProgress: 8 * 1024 * 1024 };
|
||||
}
|
||||
reset('初始化');
|
||||
function resolveFile(value) {
|
||||
if (typeof value === 'number') return opened.get(value) || '';
|
||||
return value ? path.resolve(String(value)) : '';
|
||||
}
|
||||
function recordRead(file, bytes) {
|
||||
if (file === archivePath) {
|
||||
metrics.legacyBytes += bytes;
|
||||
metrics.legacyMaxRead = Math.max(metrics.legacyMaxRead, bytes);
|
||||
if (metrics.legacyBytes >= metrics.nextLegacyProgress) {
|
||||
stage(`已扫描旧归档 ${(metrics.legacyBytes / 1024 / 1024).toFixed(1)} MiB，写块 ${metrics.chunkWrites} 次，块 fsync 累计 ${(metrics.chunkFsyncMs / 1000).toFixed(2)} 秒`);
|
||||
metrics.nextLegacyProgress += 8 * 1024 * 1024;
|
||||
}
|
||||
}
|
||||
if (file.includes(`${path.sep}chunks${path.sep}`)) {
|
||||
metrics.chunkFiles.add(file);
|
||||
metrics.chunkBytes += bytes;
|
||||
}
|
||||
}
|
||||
const filesystem = new Proxy(fs, {
|
||||
get(target, key) {
|
||||
if (key === 'openSync') return (file, ...args) => {
|
||||
const fd = target.openSync(file, ...args);
|
||||
const resolved = resolveFile(file);
|
||||
opened.set(fd, resolved);
|
||||
if (resolved.includes(`${path.sep}chunks${path.sep}`) && String(args[0]).includes('w')) metrics.chunkWrites += 1;
|
||||
return fd;
|
||||
};
|
||||
if (key === 'closeSync') return fd => {
|
||||
const result = target.closeSync(fd);
|
||||
opened.delete(fd);
|
||||
return result;
|
||||
};
|
||||
if (key === 'readFileSync') return (file, ...args) => {
|
||||
const resolved = resolveFile(file);
|
||||
assert.notEqual(resolved, archivePath, '旧完整归档必须流式读取，禁止整文件 readFileSync');
|
||||
const value = target.readFileSync(file, ...args);
|
||||
recordRead(resolved, typeof value === 'string' ? Buffer.byteLength(value) : value.length);
|
||||
return value;
|
||||
};
|
||||
if (key === 'readSync') return (fd, ...args) => {
|
||||
const file = resolveFile(fd);
|
||||
const requested = typeof args[1] === 'object' ? args[1].length ?? args[0].length : args[2] ?? args[0].length;
|
||||
if (file === archivePath) assert(requested <= 256 * 1024, '迁移旧归档的单次读取必须有界');
|
||||
const bytes = target.readSync(fd, ...args);
|
||||
recordRead(file, bytes);
|
||||
return bytes;
|
||||
};
|
||||
if (key === 'fsyncSync') return fd => {
|
||||
const chunk = resolveFile(fd).includes(`${path.sep}chunks${path.sep}`);
|
||||
const started = Date.now();
|
||||
const result = target.fsyncSync(fd);
|
||||
if (chunk) {
|
||||
metrics.chunkFsyncs += 1;
|
||||
metrics.chunkFsyncMs += Date.now() - started;
|
||||
if (metrics.chunkFsyncs % 16 === 0) stage(`块 fsync ${metrics.chunkFsyncs} 次，累计 ${(metrics.chunkFsyncMs / 1000).toFixed(2)} 秒`);
|
||||
}
|
||||
return result;
|
||||
};
|
||||
return target[key];
|
||||
},
|
||||
});
|
||||
return { fs: filesystem, reset, snapshot: () => ({ ...metrics, chunkFiles: metrics.chunkFiles.size }) };
|
||||
}
|
||||
|
||||
function runChild(sessionsDir, fixture) {
|
||||
const tracked = createTrackedFilesystem(fixture.archivePath);
|
||||
const store = createSessionHistoryStore({ sessionsDir, fs: tracked.fs });
|
||||
for (const name of ['loadTail', 'loadMetadata', 'loadHistoryPage']) {
|
||||
assert.equal(typeof store[name], 'function', `分块归档缺少 ${name} 接口`);
|
||||
}
|
||||
const observations = [];
|
||||
stage('开始迁移旧归档并加载首个尾页');
|
||||
let tail = store.loadTail(ID, TAIL_COUNT);
|
||||
stage('旧归档迁移完成');
|
||||
assert(tail, '旧归档迁移后必须可加载尾部');
|
||||
assert.equal(tail.messages.length, TAIL_COUNT);
|
||||
assert.equal(tail.messages[0].id, `original-${fixture.count - TAIL_COUNT}`);
|
||||
assert.equal(digest(tail.messages.at(-1)), digest(originalMessage(fixture.count - 1)));
|
||||
observations.push(tracked.snapshot());
|
||||
const manifestPath = path.join(sessionsDir, '_history', ID, 'manifest.json');
|
||||
const chunkDir = path.join(sessionsDir, '_history', ID, 'chunks');
|
||||
assert(fs.existsSync(manifestPath), '旧归档应发布版本 2 清单');
|
||||
assert.equal(JSON.parse(fs.readFileSync(manifestPath, 'utf8')).version, 2);
|
||||
const initialChunkFiles = fs.readdirSync(chunkDir).filter(name => name.endsWith('.json'));
|
||||
assert(initialChunkFiles.length > 2, '大归档应拆成多个独立消息块');
|
||||
|
||||
function assertBoundedRead(label, callback, metadataOnly = false) {
|
||||
tracked.reset(label);
|
||||
const result = callback();
|
||||
const observation = tracked.snapshot();
|
||||
assert.equal(observation.legacyBytes, 0, `${label} 不应重新扫描旧完整归档`);
|
||||
assert(observation.chunkFiles < initialChunkFiles.length, `${label} 不应加载全部消息块`);
|
||||
assert(observation.chunkWrites < initialChunkFiles.length, `${label} 不应重写全部消息块`);
|
||||
if (metadataOnly) assert.equal(observation.chunkFiles, 0, `${label} 不应读取消息块`);
|
||||
observations.push(observation);
|
||||
return result;
|
||||
}
|
||||
|
||||
const metadata = assertBoundedRead('只加载元数据', () => store.loadMetadata(ID), true);
|
||||
assert.equal(metadata.id, ID);
|
||||
assert.equal(digest(metadata.metadata), fixture.metadataHash);
|
||||
assert.equal(digest(metadata.padding), fixture.paddingHash);
|
||||
assert(!metadata.messages || metadata.messages.length === 0, '元数据接口不应携带消息正文');
|
||||
|
||||
for (let index = 0; index < SAVE_COUNT; index += 1) {
|
||||
stage(`开始第 ${index + 1} 次尾部加载和保存`);
|
||||
tail = assertBoundedRead(`第 ${index + 1} 次加载尾部`, () => store.loadTail(ID, TAIL_COUNT));
|
||||
const lastOriginal = tail.messages.find(message => message.id === `original-${fixture.count - 1}`);
|
||||
assert(lastOriginal, '尾页应该覆盖待更新的最后一条原始消息');
|
||||
lastOriginal.content = originalMessage(fixture.count - 1).content + `|第 ${index + 1} 次完整修改`;
|
||||
lastOriginal.toolCalls[0].result = ORIGINAL_RESULT + `|第 ${index + 1} 次工具输出`;
|
||||
tail.messages.push(addedMessage(index));
|
||||
tail.title = `已提交 ${index + 1} 次尾部更新`;
|
||||
assertBoundedRead(`第 ${index + 1} 次保存尾部`, () => store.save(tail));
|
||||
stage(`第 ${index + 1} 次尾部保存完成`);
|
||||
}
|
||||
|
||||
const finalMetadata = assertBoundedRead('重新加载最新元数据', () => store.loadMetadata(ID), true);
|
||||
assert.equal(finalMetadata.title, `已提交 ${SAVE_COUNT} 次尾部更新`);
|
||||
assert.equal(digest(finalMetadata.metadata), fixture.metadataHash);
|
||||
assert.equal(digest(finalMetadata.padding), fixture.paddingHash);
|
||||
const reopened = createSessionHistoryStore({ sessionsDir, fs: tracked.fs });
|
||||
let before = fixture.count + SAVE_COUNT;
|
||||
let checked = 0;
|
||||
let pageCount = 0;
|
||||
let revision;
|
||||
stage('开始重开存储并逐页校验全部历史');
|
||||
while (before > 0) {
|
||||
const page = assertBoundedRead(`分页 ${pageCount + 1}`, () => reopened.loadHistoryPage(ID, before, PAGE_COUNT));
|
||||
assert.equal(page.total, fixture.count + SAVE_COUNT, '尾页保存必须保留未加载的全部历史');
|
||||
assert.equal(page.end, before);
|
||||
assert.equal(page.start, Math.max(0, before - PAGE_COUNT));
|
||||
assert.equal(page.messages.length, page.end - page.start);
|
||||
assert.equal(page.complete, true);
|
||||
if (revision) assert.equal(page.revision, revision, '分页期间应保持同一归档修订');
|
||||
revision = page.revision;
|
||||
page.messages.forEach((message, offset) => {
|
||||
const position = page.start + offset;
|
||||
const expected = position < fixture.count ? originalMessage(position) : addedMessage(position - fixture.count);
|
||||
if (position === fixture.count - 1) {
|
||||
expected.content += `|第 ${SAVE_COUNT} 次完整修改`;
|
||||
expected.toolCalls[0].result += `|第 ${SAVE_COUNT} 次工具输出`;
|
||||
}
|
||||
assert.equal(message.id, expected.id, `第 ${position} 条消息的顺序或稳定 ID 错误`);
|
||||
assert.equal(digest(message), digest(expected), `第 ${position} 条正文、工具或元数据出现截断或变化`);
|
||||
checked += 1;
|
||||
});
|
||||
before = page.start;
|
||||
pageCount += 1;
|
||||
if (pageCount % 16 === 0) stage(`已校验 ${pageCount} 页、${checked} 条消息`);
|
||||
}
|
||||
assert.equal(checked, fixture.count + SAVE_COUNT);
|
||||
const currentManifest = JSON.parse(fs.readFileSync(manifestPath, 'utf8'));
|
||||
const currentChunkFiles = new Set(currentManifest.entries.map(entry => entry.file));
|
||||
const reusedChunks = initialChunkFiles.filter(name => currentChunkFiles.has(name)).length;
|
||||
assert(reusedChunks > 0, '尾部更新必须复用未变化的消息块');
|
||||
stage(`全部 ${pageCount} 页、${checked} 条历史校验完成`);
|
||||
console.log(JSON.stringify({
|
||||
archiveMiB: fixture.bytes / 1024 / 1024,
|
||||
originalMessages: fixture.count,
|
||||
checkedMessages: checked,
|
||||
pages: pageCount,
|
||||
saves: SAVE_COUNT,
|
||||
peakRssMiB: process.resourceUsage().maxRSS / 1024,
|
||||
initialChunks: initialChunkFiles.length,
|
||||
reusedChunks,
|
||||
maxChunksPerBoundedOperation: Math.max(...observations.slice(1).map(item => item.chunkFiles)),
|
||||
legacyMaxReadKiB: observations[0].legacyMaxRead / 1024,
|
||||
migrationFsyncSeconds: observations[0].chunkFsyncMs / 1000,
|
||||
}));
|
||||
}
|
||||
|
||||
function runTests() {
|
||||
const sessionsDir = fs.mkdtempSync(path.join(temporaryRoot(), 'ccweb-history-memory-'));
|
||||
try {
|
||||
stage(`开始流式写入 70.5 MiB 合成夹具（${path.dirname(sessionsDir)}）`);
|
||||
const fixture = writeLegacyFixture(sessionsDir);
|
||||
stage(`夹具写入完成，共 ${fixture.count} 条原始消息`);
|
||||
assert.equal(fixture.bytes, TARGET_BYTES);
|
||||
const child = spawnSync(process.execPath, ['--max-old-space-size=256', __filename, '--child', sessionsDir, JSON.stringify(fixture)], {
|
||||
encoding: 'utf8', timeout: 60000, maxBuffer: 512 * 1024,
|
||||
stdio: ['ignore', 'pipe', process.env.CCWEB_HISTORY_TEST_PROGRESS === '1' ? 'inherit' : 'pipe'],
|
||||
});
|
||||
if (child.error || child.status !== 0) process.stderr.write(child.stderr || '');
|
||||
assert.ifError(child.error);
|
||||
assert.equal(child.status, 0, `256 MiB 堆子进程失败（信号：${child.signal || '无'}）：\n${child.stderr || child.stdout}`);
|
||||
const result = JSON.parse(child.stdout.trim());
|
||||
console.log(`通过：${result.archiveMiB.toFixed(1)} MiB 旧归档流式迁移，256 MiB 堆内逐页校验 ${result.checkedMessages} 条完整历史`);
|
||||
console.log(`通过：${result.saves} 次尾部保存、${result.pages} 页读取，单次最多触达 ${result.maxChunksPerBoundedOperation}/${result.initialChunks} 个原始块`);
|
||||
console.log(`内存：子进程 RSS 峰值 ${result.peakRssMiB.toFixed(1)} MiB；旧归档单次读取最大 ${result.legacyMaxReadKiB.toFixed(1)} KiB；复用 ${result.reusedChunks} 个原始块`);
|
||||
return result;
|
||||
} finally {
|
||||
fs.rmSync(sessionsDir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
if (require.main === module) {
|
||||
if (process.argv[2] === '--child') runChild(process.argv[3], JSON.parse(process.argv[4]));
|
||||
else runTests();
|
||||
}
|
||||
|
||||
module.exports = { runTests };
|
||||
Reference in New Issue
Block a user