mirror of
https://github.com/Darkatse/TauriTavern.git
synced 2026-10-19 07:35:14 +00:00
perf(files): bound stream reads by the opened file size
Stat each opened file once and request its remaining bytes in chunks up to 4 MiB, avoiding padded tail responses and an extra EOF read. Reject truncated or malformed responses and retain both read and cleanup failures. Decode the trailer with DataView, declare ES2022 library types in the shared type-check configuration, and include the fs permission, cancellation coverage, navigation fixture, and contract documentation.
This commit is contained in:
Darkatse
parent
d84e5a36ec
commit
5ca99a87a1
6 changed files with 246 additions and 161 deletions
|
|
@ -30,6 +30,8 @@ transport 解析完整 JSONL 后直接把同一对象数组交给核心调用方
|
|||
|
||||
所有平台通过共享的 Tauri FileHandle pull stream 有界读取 JSONL。每次加载始终复用同一个文件 handle，并在 EOF、取消或失败时关闭资源；桌面标准模式、portable 模式及自定义数据目录使用同一个已解析 `data_root` runtime scope。
|
||||
|
||||
读取前对已打开的 handle 调用 `fstat`，之后按剩余字节数请求，每次不超过共享 reader 的块上限，读满声明长度即止，不额外发 EOF 空读。正数短读继续读取，提前 EOF 或非法响应长度直接报错并关闭资源；读取和关闭同时失败时保留两个错误。
|
||||
|
||||
`power_user.chat_truncation` 只限制首次挂载的 DOM 数量，不裁剪 `chat[]`。`Show more messages` 从完整数组中补挂更早楼层，不发起历史 I/O，也不改变数组索引。后续 DOM virtualization 若实施，也只能替换渲染层，不能改变 canonical data contract。
|
||||
|
||||
## 3. 完整保存
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@
|
|||
"notification:allow-is-permission-granted",
|
||||
"fs:allow-open",
|
||||
"fs:allow-read",
|
||||
"fs:allow-fstat",
|
||||
"fs:allow-read-text-file",
|
||||
"fs:allow-resource-read-recursive",
|
||||
"fs:default",
|
||||
|
|
|
|||
|
|
@ -1,20 +1,7 @@
|
|||
// @ts-check
|
||||
|
||||
const FS_READ_CHUNK_BYTES = 512 * 1024;
|
||||
|
||||
/** @param {Uint8Array} bytes */
|
||||
function readBigEndianUint64(bytes) {
|
||||
let value = 0;
|
||||
for (let i = 0; i < bytes.length; i += 1) {
|
||||
const byte = bytes[i];
|
||||
if (byte === undefined) {
|
||||
throw new Error('Unexpected fs read trailer byte');
|
||||
}
|
||||
value *= 0x100;
|
||||
value += byte;
|
||||
}
|
||||
return value;
|
||||
}
|
||||
// Amortize IPC roundtrips while bounding each response and decode burst.
|
||||
const FS_READ_MAX_CHUNK_BYTES = 4 * 1024 * 1024;
|
||||
|
||||
/** @param {any} data */
|
||||
function normalizeReadResponse(data) {
|
||||
|
|
@ -57,6 +44,9 @@ export function createReadableFileStreamService({ invoke }) {
|
|||
return new ReadableStream({
|
||||
async pull(controller) {
|
||||
const rid = await ridPromise;
|
||||
if (closed) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const bytes = await readChunk(rid);
|
||||
|
|
@ -68,7 +58,11 @@ export function createReadableFileStreamService({ invoke }) {
|
|||
|
||||
controller.enqueue(bytes);
|
||||
} catch (error) {
|
||||
await closeOnce();
|
||||
try {
|
||||
await closeOnce();
|
||||
} catch (closeError) {
|
||||
throw new AggregateError([error, closeError], 'Failed to read and close resource');
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
|
|
@ -80,18 +74,42 @@ export function createReadableFileStreamService({ invoke }) {
|
|||
|
||||
/** @param {string} filePath */
|
||||
function createReadableFileStream(filePath) {
|
||||
/** @type {number | undefined} */
|
||||
let remaining;
|
||||
return createReadableResourceStream(
|
||||
invoke('plugin:fs|open', {
|
||||
path: filePath,
|
||||
options: { read: true },
|
||||
}),
|
||||
async (rid) => {
|
||||
// Keep stat inside the resource owner's error handling: open may succeed even if stat fails.
|
||||
if (remaining === undefined) {
|
||||
/** @type {{ size: number }} */
|
||||
const { size } = await invoke('plugin:fs|fstat', { rid });
|
||||
if (!Number.isSafeInteger(size) || size < 0) {
|
||||
throw new Error(`Invalid file size for ${filePath}: ${size}`);
|
||||
}
|
||||
remaining = size;
|
||||
}
|
||||
if (remaining === 0) {
|
||||
return new Uint8Array(0);
|
||||
}
|
||||
|
||||
// plugin-fs returns the requested length even on a short read, plus its 8-byte trailer.
|
||||
const len = Math.min(FS_READ_MAX_CHUNK_BYTES, remaining);
|
||||
const data = await invoke('plugin:fs|read', {
|
||||
rid,
|
||||
len: FS_READ_CHUNK_BYTES,
|
||||
len,
|
||||
});
|
||||
const bytes = normalizeReadResponse(data);
|
||||
const bytesRead = readBigEndianUint64(bytes.subarray(bytes.byteLength - 8));
|
||||
const bytesRead = Number(new DataView(bytes.buffer, bytes.byteOffset + bytes.byteLength - 8, 8).getBigUint64(0));
|
||||
if (bytesRead > len || bytesRead > bytes.byteLength - 8) {
|
||||
throw new Error(`Invalid fs read length for ${filePath}: ${bytesRead}`);
|
||||
}
|
||||
if (bytesRead === 0) {
|
||||
throw new Error(`File shorter than its declared size: ${filePath} (${remaining} bytes remaining)`);
|
||||
}
|
||||
remaining -= bytesRead;
|
||||
return bytes.subarray(0, bytesRead);
|
||||
},
|
||||
);
|
||||
|
|
|
|||
|
|
@ -90,12 +90,14 @@ test('chat persistence and navigation', async (context) => {
|
|||
handles.set(rid, new TextEncoder().encode(payloads.get(args.path)));
|
||||
return rid;
|
||||
}
|
||||
case 'plugin:fs|fstat': return { size: handles.get(args.rid).length };
|
||||
case 'plugin:fs|read': {
|
||||
const content = handles.get(args.rid);
|
||||
const bytes = new window.Uint8Array(content.length + 8);
|
||||
bytes.set(content);
|
||||
new window.DataView(bytes.buffer).setBigUint64(content.length, BigInt(content.length));
|
||||
handles.set(args.rid, new Uint8Array());
|
||||
const chunk = content.subarray(0, args.len);
|
||||
const bytes = new window.Uint8Array(args.len + 8);
|
||||
bytes.set(chunk);
|
||||
new window.DataView(bytes.buffer).setBigUint64(args.len, BigInt(chunk.length));
|
||||
handles.set(args.rid, content.subarray(chunk.length));
|
||||
return bytes;
|
||||
}
|
||||
case 'plugin:resources|close':
|
||||
|
|
|
|||
|
|
@ -3,165 +3,227 @@ import test from 'node:test';
|
|||
|
||||
import { createReadableFileStreamService } from '../src/tauri/main/services/files/readable-file-stream-service.js';
|
||||
|
||||
function createFsReadResponse(payload) {
|
||||
const trailer = new Uint8Array(8);
|
||||
let length = payload.byteLength;
|
||||
for (let index = trailer.length - 1; index >= 0; index -= 1) {
|
||||
trailer[index] = length & 0xff;
|
||||
length = Math.floor(length / 0x100);
|
||||
}
|
||||
|
||||
const response = new Uint8Array(payload.byteLength + trailer.byteLength);
|
||||
response.set(payload, 0);
|
||||
response.set(trailer, payload.byteLength);
|
||||
function createFsReadResponse(payload, requestedLength = payload.byteLength) {
|
||||
const response = new Uint8Array(requestedLength + 8);
|
||||
response.set(payload);
|
||||
new DataView(response.buffer).setBigUint64(requestedLength, BigInt(payload.byteLength));
|
||||
return response;
|
||||
}
|
||||
|
||||
async function readStreamBytes(stream) {
|
||||
const reader = stream.getReader();
|
||||
const chunks = [];
|
||||
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) {
|
||||
break;
|
||||
}
|
||||
chunks.push(...value);
|
||||
}
|
||||
|
||||
return chunks;
|
||||
return new Uint8Array(await new Response(stream).arrayBuffer());
|
||||
}
|
||||
|
||||
test('readable file stream service reads Tauri fs chunks and closes the resource', async () => {
|
||||
function mockFileService(overrides = {}) {
|
||||
const calls = [];
|
||||
const readResponses = [
|
||||
createFsReadResponse(Uint8Array.from([0x50, 0x4b])),
|
||||
createFsReadResponse(Uint8Array.from([0x03, 0x04])),
|
||||
createFsReadResponse(new Uint8Array(0)),
|
||||
];
|
||||
const commands = {
|
||||
'plugin:fs|open': () => 7,
|
||||
'plugin:fs|fstat': () => ({ size: 4 }),
|
||||
'plugin:resources|close': () => null,
|
||||
...overrides,
|
||||
};
|
||||
const service = createReadableFileStreamService({
|
||||
invoke: async (command, args) => {
|
||||
calls.push({ command, args });
|
||||
if (command === 'plugin:fs|open') {
|
||||
return 7;
|
||||
}
|
||||
if (command === 'plugin:fs|read') {
|
||||
return readResponses.shift();
|
||||
}
|
||||
if (command === 'plugin:resources|close') {
|
||||
return null;
|
||||
}
|
||||
throw new Error(`Unexpected command: ${command}`);
|
||||
assert.ok(commands[command], 'Unexpected command: ' + command);
|
||||
return commands[command](args);
|
||||
},
|
||||
});
|
||||
return { service, calls };
|
||||
}
|
||||
|
||||
test('file stream caps requests and reads the exact remaining bytes without an EOF call', async () => {
|
||||
const cap = 4 * 1024 * 1024;
|
||||
const payload = new Uint8Array(cap + 2).fill(0x61);
|
||||
payload.set([0x62, 0x63], cap);
|
||||
let offset = 0;
|
||||
const { service, calls } = mockFileService({
|
||||
'plugin:fs|fstat': () => ({ size: payload.length }),
|
||||
'plugin:fs|read': ({ len }) => {
|
||||
const chunk = payload.subarray(offset, offset + len);
|
||||
offset += chunk.length;
|
||||
return createFsReadResponse(chunk, len).buffer;
|
||||
},
|
||||
});
|
||||
|
||||
const bytes = await readStreamBytes(service.createReadableFileStream('/tmp/archive.zip'));
|
||||
|
||||
assert.deepEqual(bytes, [0x50, 0x4b, 0x03, 0x04]);
|
||||
assert.deepEqual(await readStreamBytes(service.createReadableFileStream('/tmp/archive.zip')), payload);
|
||||
assert.deepEqual(calls, [
|
||||
{
|
||||
command: 'plugin:fs|open',
|
||||
args: {
|
||||
path: '/tmp/archive.zip',
|
||||
options: { read: true },
|
||||
},
|
||||
},
|
||||
{
|
||||
command: 'plugin:fs|read',
|
||||
args: {
|
||||
rid: 7,
|
||||
len: 512 * 1024,
|
||||
},
|
||||
},
|
||||
{
|
||||
command: 'plugin:fs|read',
|
||||
args: {
|
||||
rid: 7,
|
||||
len: 512 * 1024,
|
||||
},
|
||||
},
|
||||
{
|
||||
command: 'plugin:fs|read',
|
||||
args: {
|
||||
rid: 7,
|
||||
len: 512 * 1024,
|
||||
},
|
||||
},
|
||||
{
|
||||
command: 'plugin:resources|close',
|
||||
args: {
|
||||
rid: 7,
|
||||
},
|
||||
},
|
||||
{ command: 'plugin:fs|open', args: { path: '/tmp/archive.zip', options: { read: true } } },
|
||||
{ command: 'plugin:fs|fstat', args: { rid: 7 } },
|
||||
{ command: 'plugin:fs|read', args: { rid: 7, len: cap } },
|
||||
{ command: 'plugin:fs|read', args: { rid: 7, len: 2 } },
|
||||
{ command: 'plugin:resources|close', args: { rid: 7 } },
|
||||
]);
|
||||
});
|
||||
|
||||
test('readable file stream service reads chat backup chunks and closes the resource', async () => {
|
||||
const calls = [];
|
||||
const readResponses = [
|
||||
Uint8Array.from([1, 2]),
|
||||
Uint8Array.from([3]),
|
||||
new Uint8Array(0),
|
||||
];
|
||||
const service = createReadableFileStreamService({
|
||||
invoke: async (command, args) => {
|
||||
calls.push({ command, args });
|
||||
if (command === 'open_chat_backup_download') {
|
||||
return 8;
|
||||
}
|
||||
if (command === 'read_chat_backup_download') {
|
||||
return readResponses.shift();
|
||||
}
|
||||
if (command === 'plugin:resources|close') {
|
||||
return null;
|
||||
}
|
||||
throw new Error(`Unexpected command: ${command}`);
|
||||
},
|
||||
test('file stream continues positive short reads without delivering padding', async () => {
|
||||
const chunks = [Uint8Array.of(1, 2), Uint8Array.of(3, 4, 5)];
|
||||
const { service, calls } = mockFileService({
|
||||
'plugin:fs|fstat': () => ({ size: 5 }),
|
||||
'plugin:fs|read': ({ len }) => createFsReadResponse(chunks.shift(), len),
|
||||
});
|
||||
|
||||
const stream = await service.createChatBackupDownloadStream('chat_alice.jsonl');
|
||||
assert.deepEqual(await readStreamBytes(stream), [1, 2, 3]);
|
||||
assert.deepEqual(calls, [
|
||||
{
|
||||
command: 'open_chat_backup_download',
|
||||
args: { name: 'chat_alice.jsonl' },
|
||||
},
|
||||
...Array.from({ length: 3 }, () => ({
|
||||
command: 'read_chat_backup_download',
|
||||
args: { rid: 8 },
|
||||
assert.deepEqual(await readStreamBytes(service.createReadableFileStream('/tmp/file')), Uint8Array.of(1, 2, 3, 4, 5));
|
||||
assert.deepEqual(calls.filter(call => call.command === 'plugin:fs|read').map(call => call.args.len), [5, 3]);
|
||||
});
|
||||
|
||||
test('empty file closes without a read', async () => {
|
||||
const { service, calls } = mockFileService({
|
||||
'plugin:fs|fstat': () => ({ size: 0 }),
|
||||
});
|
||||
|
||||
assert.deepEqual(await readStreamBytes(service.createReadableFileStream('/tmp/empty')), new Uint8Array(0));
|
||||
assert.deepEqual(calls.map(call => call.command), ['plugin:fs|open', 'plugin:fs|fstat', 'plugin:resources|close']);
|
||||
});
|
||||
|
||||
test('file stream fails instead of completing a truncated file', async () => {
|
||||
const chunks = [Uint8Array.of(1, 2), new Uint8Array(0)];
|
||||
const { service, calls } = mockFileService({
|
||||
'plugin:fs|read': ({ len }) => createFsReadResponse(chunks.shift(), len),
|
||||
});
|
||||
|
||||
await assert.rejects(
|
||||
readStreamBytes(service.createReadableFileStream('/tmp/truncated')),
|
||||
/shorter than its declared size: \/tmp\/truncated \(2 bytes remaining\)/,
|
||||
);
|
||||
assert.equal(calls.filter(call => call.command === 'plugin:resources|close').length, 1);
|
||||
});
|
||||
|
||||
test('file stream closes an opened resource when stat or read fails', async t => {
|
||||
const cases = [
|
||||
...[undefined, null, '4', -1, 0.5, Number.MAX_SAFE_INTEGER + 1].map(size => ({
|
||||
name: 'invalid size ' + size,
|
||||
commands: { 'plugin:fs|fstat': () => ({ size }) },
|
||||
error: /Invalid file size/,
|
||||
})),
|
||||
{
|
||||
command: 'plugin:resources|close',
|
||||
args: { rid: 8 },
|
||||
name: 'stat rejection',
|
||||
commands: { 'plugin:fs|fstat': () => { throw new Error('stat denied'); } },
|
||||
error: /stat denied/,
|
||||
},
|
||||
{
|
||||
name: 'read rejection',
|
||||
commands: { 'plugin:fs|read': () => { throw new Error('read failed'); } },
|
||||
error: /read failed/,
|
||||
},
|
||||
{
|
||||
name: 'non-binary response',
|
||||
commands: { 'plugin:fs|read': () => ({ bytes: [] }) },
|
||||
error: /Unexpected resource read response/,
|
||||
},
|
||||
{
|
||||
name: 'missing trailer',
|
||||
commands: { 'plugin:fs|read': () => Uint8Array.of(65, 66) },
|
||||
error: RangeError,
|
||||
},
|
||||
{
|
||||
name: 'read count exceeds request',
|
||||
commands: { 'plugin:fs|read': () => createFsReadResponse(Uint8Array.of(1, 2, 3, 4, 5)) },
|
||||
error: /Invalid fs read length/,
|
||||
},
|
||||
{
|
||||
name: 'read count exceeds response body',
|
||||
commands: { 'plugin:fs|read': () => createFsReadResponse(Uint8Array.of(1, 2, 3, 4)).subarray(1) },
|
||||
error: /Invalid fs read length/,
|
||||
},
|
||||
];
|
||||
|
||||
for (const entry of cases) {
|
||||
await t.test(entry.name, async () => {
|
||||
const { service, calls } = mockFileService(entry.commands);
|
||||
await assert.rejects(readStreamBytes(service.createReadableFileStream('/tmp/file')), entry.error);
|
||||
assert.deepEqual(calls.filter(call => call.command === 'plugin:resources|close'), [
|
||||
{ command: 'plugin:resources|close', args: { rid: 7 } },
|
||||
]);
|
||||
assert.equal(calls.at(-1).command, 'plugin:resources|close');
|
||||
if (entry.commands['plugin:fs|fstat']) {
|
||||
assert.equal(calls.some(call => call.command === 'plugin:fs|read'), false);
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
test('open failure propagates without closing an unallocated resource', async () => {
|
||||
const failure = new Error('open denied');
|
||||
const { service, calls } = mockFileService({
|
||||
'plugin:fs|open': () => { throw failure; },
|
||||
});
|
||||
await assert.rejects(readStreamBytes(service.createReadableFileStream('/tmp/file')), error => error === failure);
|
||||
assert.deepEqual(calls.map(call => call.command), ['plugin:fs|open']);
|
||||
});
|
||||
|
||||
test('read failure retains both the original and cleanup errors', async t => {
|
||||
const failure = new Error('operation failed');
|
||||
const closeFailure = new Error('close failed');
|
||||
for (const command of ['plugin:fs|fstat', 'plugin:fs|read']) {
|
||||
await t.test(command, async () => {
|
||||
const { service, calls } = mockFileService({
|
||||
[command]: () => { throw failure; },
|
||||
'plugin:resources|close': () => { throw closeFailure; },
|
||||
});
|
||||
await assert.rejects(readStreamBytes(service.createReadableFileStream('/tmp/file')), error => {
|
||||
assert.ok(error instanceof AggregateError);
|
||||
assert.deepEqual(error.errors, [failure, closeFailure]);
|
||||
return true;
|
||||
});
|
||||
assert.equal(calls.filter(call => call.command === 'plugin:resources|close').length, 1);
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
test('close failure is not a successful end of stream', async () => {
|
||||
const failure = new Error('close failed');
|
||||
const { service, calls } = mockFileService({
|
||||
'plugin:fs|fstat': () => ({ size: 0 }),
|
||||
'plugin:resources|close': () => { throw failure; },
|
||||
});
|
||||
await assert.rejects(readStreamBytes(service.createReadableFileStream('/tmp/file')), error => error === failure);
|
||||
assert.equal(calls.filter(call => call.command === 'plugin:resources|close').length, 1);
|
||||
});
|
||||
|
||||
test('cancelling during open, stat or read closes once and delivers no data', async t => {
|
||||
for (const [command, result] of [
|
||||
['plugin:fs|open', 7],
|
||||
['plugin:fs|fstat', { size: 4 }],
|
||||
['plugin:fs|read', createFsReadResponse(Uint8Array.of(1, 2, 3, 4))],
|
||||
]) {
|
||||
await t.test(command, async () => {
|
||||
const entered = Promise.withResolvers();
|
||||
const pending = Promise.withResolvers();
|
||||
const { service, calls } = mockFileService({
|
||||
'plugin:fs|read': () => createFsReadResponse(Uint8Array.of(1, 2, 3, 4)),
|
||||
[command]: () => {
|
||||
entered.resolve();
|
||||
return pending.promise;
|
||||
},
|
||||
});
|
||||
const reader = service.createReadableFileStream('/tmp/file').getReader();
|
||||
const reading = reader.read();
|
||||
await entered.promise;
|
||||
const cancelled = reader.cancel();
|
||||
pending.resolve(result);
|
||||
await cancelled;
|
||||
|
||||
assert.deepEqual(await reading, { value: undefined, done: true });
|
||||
assert.deepEqual(calls.filter(call => call.command === 'plugin:resources|close'), [
|
||||
{ command: 'plugin:resources|close', args: { rid: 7 } },
|
||||
]);
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
test('chat backup stream retains host-owned chunking without fs stat', async () => {
|
||||
const chunks = [Uint8Array.of(1, 2), Uint8Array.of(3), new Uint8Array(0)];
|
||||
const { service, calls } = mockFileService({
|
||||
'open_chat_backup_download': () => 8,
|
||||
'read_chat_backup_download': () => chunks.shift(),
|
||||
});
|
||||
const stream = await service.createChatBackupDownloadStream('chat_alice.jsonl');
|
||||
|
||||
assert.deepEqual(await readStreamBytes(stream), Uint8Array.of(1, 2, 3));
|
||||
assert.deepEqual(calls, [
|
||||
{ command: 'open_chat_backup_download', args: { name: 'chat_alice.jsonl' } },
|
||||
...Array.from({ length: 3 }, () => ({ command: 'read_chat_backup_download', args: { rid: 8 } })),
|
||||
{ command: 'plugin:resources|close', args: { rid: 8 } },
|
||||
]);
|
||||
});
|
||||
|
||||
test('readable resource stream closes on read failure', async () => {
|
||||
const calls = [];
|
||||
const service = createReadableFileStreamService({
|
||||
invoke: async (command, args) => {
|
||||
calls.push({ command, args });
|
||||
if (command === 'plugin:fs|open') {
|
||||
return 9;
|
||||
}
|
||||
if (command === 'plugin:fs|read') {
|
||||
return { bytes: [] };
|
||||
}
|
||||
if (command === 'plugin:resources|close') {
|
||||
return null;
|
||||
}
|
||||
throw new Error(`Unexpected command: ${command}`);
|
||||
},
|
||||
});
|
||||
|
||||
const reader = service.createReadableFileStream('/tmp/archive.zip').getReader();
|
||||
|
||||
await assert.rejects(() => reader.read(), /Unexpected resource read response/);
|
||||
assert.deepEqual(calls.at(-1), {
|
||||
command: 'plugin:resources|close',
|
||||
args: {
|
||||
rid: 9,
|
||||
},
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@
|
|||
"target": "ES2020",
|
||||
"module": "ES2020",
|
||||
"moduleResolution": "Bundler",
|
||||
"lib": ["ES2020", "DOM", "DOM.Iterable"],
|
||||
"lib": ["ES2022", "DOM", "DOM.Iterable"],
|
||||
|
||||
"allowJs": true,
|
||||
"checkJs": false,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue