peoplelib/src/_test/range-sessions.test.js

const assert = require('node:assert');
const test = require('node:test');
const fs = require('fs');
const os = require('os');
const path = require('path');

const modulePath = require.resolve('../reader/range-sessions');
const dirs = [];

function fixture() {
  const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'peoplelib-range-sessions-'));
  dirs.push(dir);
  const file = path.join(dir, 'fixture.pdf');
  const bytes = Buffer.alloc(6 * 1024 * 1024);
  for (let index = 0; index < bytes.length; index++) bytes[index] = index % 251;
  fs.writeFileSync(file, bytes);
  delete require.cache[modulePath];
  const sessions = require(modulePath);
  sessions.init((_entryId, fileIndex) => ({
    abs: file,
    format: 'pdf',
    fileIndex: Number.isInteger(fileIndex) ? fileIndex : 0
  }));
  return { sessions, file, bytes };
}

test.after(async () => {
  try {
    const sessions = require(modulePath);
    await sessions.closeAll();
  } catch (error) { /* ignore */ }
  for (const dir of dirs) {
    try { fs.rmSync(dir, { recursive: true, force: true }); } catch (error) { /* ignore */ }
  }
});

test('PDF 分段会话只返回请求范围并可显式关闭', async () => {
  const { sessions, bytes } = fixture();
  const opened = await sessions.open(10, 'entry', 0);
  assert.match(opened.sessionId, /^[a-f0-9-]{36}$/);
  assert.strictEqual(opened.size, bytes.length);
  assert.strictEqual(opened.chunkSize, sessions.RANGE_CHUNK_BYTES);
  const begin = 1024 * 1024 + 137;
  const end = begin + 256 * 1024;
  const result = await sessions.read(10, opened.sessionId, begin, end);
  assert.deepStrictEqual(result, bytes.subarray(begin, end));
  assert.strictEqual(await sessions.close(10, opened.sessionId), true);
  assert.strictEqual(await sessions.close(10, opened.sessionId), false);
  assert.strictEqual(sessions.status().sessions, 0);
});

test('PDF 分段会话绑定发送者并限制范围与并发会话数量', async () => {
  const { sessions, bytes } = fixture();
  const first = await sessions.open(20, 'entry', 0);
  await assert.rejects(() => sessions.read(21, first.sessionId, 0, 1024), /无效或已关闭/);
  await assert.rejects(() => sessions.read(20, first.sessionId, -1, 1024), /范围无效/);
  await assert.rejects(
    () => sessions.read(20, first.sessionId, 0, sessions.MAX_RANGE_BYTES + 1),
    /不能超过 4 MB/
  );
  await assert.rejects(
    () => sessions.read(20, first.sessionId, bytes.length - 10, bytes.length + 1),
    /范围无效/
  );

  const ids = [first.sessionId];
  for (let index = 0; index < sessions.MAX_SESSIONS_PER_SENDER; index++) {
    ids.push((await sessions.open(20, 'entry', 0)).sessionId);
  }
  assert.strictEqual(sessions.status().sessions, sessions.MAX_SESSIONS_PER_SENDER);
  await assert.rejects(() => sessions.read(20, ids[0], 0, 1024), /无效或已关闭/);
  assert.strictEqual(await sessions.closeSender(20), sessions.MAX_SESSIONS_PER_SENDER);
  assert.strictEqual(sessions.status().sessions, 0);
});

test('PDF 在阅读期间发生变化时拒绝继续提供旧会话数据', async () => {
  const { sessions, file } = fixture();
  const opened = await sessions.open(30, 'entry', 0);
  fs.appendFileSync(file, Buffer.from([1]));
  await assert.rejects(
    () => sessions.read(30, opened.sessionId, 0, 1024),
    /发生变化/
  );
  await sessions.closeSender(30);
});

test('40 GB 文件使用安全整数偏移按需读取而不分配整文件缓冲区', async () => {
  delete require.cache[modulePath];
  const sessions = require(modulePath);
  const size = 40 * 1024 * 1024 * 1024;
  let closed = false;
  const stat = { size, mtimeMs: 1, ctimeMs: 1, isFile: () => true };
  const handle = {
    stat: async () => stat,
    read: async (buffer, offset, length, position) => {
      for (let index = 0; index < length; index++) {
        buffer[offset + index] = (position + index) % 251;
      }
      return { bytesRead: length, buffer };
    },
    close: async () => { closed = true; }
  };
  sessions.init(
    () => ({ abs: 'virtual-40gb.pdf', format: 'pdf', fileIndex: 0 }),
    { promises: { open: async () => handle } }
  );
  const opened = await sessions.open(40, 'huge', 0);
  assert.strictEqual(opened.size, size);
  const begin = size - 8192;
  const result = await sessions.read(40, opened.sessionId, begin, begin + 4096);
  assert.strictEqual(result.length, 4096);
  assert.strictEqual(result[0], begin % 251);
  assert.strictEqual(result[4095], (begin + 4095) % 251);
  await sessions.closeSender(40);
  assert.strictEqual(closed, true);
});

test('发送者销毁与会话创建竞态不会遗留文件句柄', async () => {
  delete require.cache[modulePath];
  const sessions = require(modulePath);
  const stat = { size: 4096, mtimeMs: 1, ctimeMs: 1, isFile: () => true };
  let releaseOpen;
  let closed = false;
  sessions.init(
    () => ({ abs: 'delayed.pdf', format: 'pdf', fileIndex: 0 }),
    {
      promises: {
        open: () => new Promise((resolve) => {
          releaseOpen = () => resolve({
            stat: async () => stat,
            read: async () => ({ bytesRead: 0 }),
            close: async () => { closed = true; }
          });
        })
      }
    }
  );
  const opening = sessions.open(50, 'entry', 0);
  while (!releaseOpen) await new Promise((resolve) => setImmediate(resolve));
  await sessions.closeSender(50);
  releaseOpen();
  await assert.rejects(opening, /窗口已关闭/);
  assert.strictEqual(closed, true);
  assert.strictEqual(sessions.status().sessions, 0);
});

test('范围读取完成后再次校验文件签名', async () => {
  delete require.cache[modulePath];
  const sessions = require(modulePath);
  let changed = false;
  let closed = false;
  const handle = {
    stat: async () => ({
      size: 4096,
      mtimeMs: changed ? 2 : 1,
      ctimeMs: 1,
      isFile: () => true
    }),
    read: async (buffer, offset, length) => {
      buffer.fill(1, offset, offset + length);
      changed = true;
      return { bytesRead: length, buffer };
    },
    close: async () => { closed = true; }
  };
  sessions.init(
    () => ({ abs: 'changing.pdf', format: 'pdf', fileIndex: 0 }),
    { promises: { open: async () => handle } }
  );
  const opened = await sessions.open(60, 'entry', 0);
  await assert.rejects(
    () => sessions.read(60, opened.sessionId, 0, 1024),
    /发生变化/
  );
  while (!closed) await new Promise((resolve) => setImmediate(resolve));
  assert.strictEqual(sessions.status().sessions, 0);
});