优化数据库读写

.env.example

@@ -48,3 +48,7 @@ PUBLIC_SITE_FOOTER_NOTICE=本平台展示数据仅用于系统演示
 
 # 也可以用单个连接地址替代全部 MYSQL_* 连接参数：
 # DATABASE_URL=mysql://exam_app:password@127.0.0.1:3306/exam_information
+# 全量数据库状态快照的最长复用时间（毫秒）。应用内写入会立即失效；MySQL 外部直写最长延迟此时间可见。
+DATABASE_STATE_CACHE_TTL_MS=30000
+# Redis 未配置或暂不可用时，本机响应缓存的最大条目数。
+LOCAL_CACHE_MAX_ENTRIES=200

README.md

@@ -231,7 +231,9 @@ npm start
 
 ### Redis 缓存（可选）
 
-配置 `REDIS_URL` 后，应用会缓存公开首页、已发布公告详情和每名考生的已发布成绩查询。公开数据默认 TTL 为 60 秒，成绩默认 TTL 为 24 小时；成绩录入/发布、批量导入、考试归档和成绩复议会自动使成绩缓存失效，超级管理员也可以在成绩管理中心手动刷新全部成绩缓存。Redis 在启动或运行期间不可用时，接口会自动回源数据库，不影响登录、报名和管理功能。
+配置 `REDIS_URL` 后，应用会缓存公开首页、已发布公告详情和每名考生的已发布成绩查询。公开数据默认 TTL 为 60 秒，成绩默认 TTL 为 24 小时；成绩录入/发布、批量导入、考试归档和成绩复议会自动使成绩缓存失效，超级管理员也可以在成绩管理中心手动刷新全部成绩缓存。Redis 未配置或暂时不可用时，应用会自动切换到有界本机缓存（默认最多 200 条），避免热点接口反复回源；写入后的失效语义保持不变。
+
+数据库关系表会按写入版本复用进程内只读快照，避免每个请求重复扫描并转换全部业务表。应用自身写入会立即使快照失效；SQLite 还通过 `PRAGMA data_version` 检测其他连接的提交，MySQL 外部直写默认最多延迟 30 秒可见。可通过 `DATABASE_STATE_CACHE_TTL_MS` 调整 MySQL 快照时间，通过 `LOCAL_CACHE_MAX_ENTRIES` 控制本机接口缓存上限。
 
 ```powershell
 $env:REDIS_URL = 'redis://127.0.0.1:6379/0'

package.json

@@ -5,7 +5,7 @@
   "type": "module",
   "scripts": {
     "start": "node server.mjs",
-    "test": "node tests/client-auth.test.mjs && node tests/cache.test.mjs && node tests/document-verification.test.mjs && node tests/admission.test.mjs && node tests/seed.test.mjs && node tests/system.test.mjs",
+    "test": "node tests/client-auth.test.mjs && node tests/api-dedup.test.mjs && node tests/cache.test.mjs && node tests/state-cache.test.mjs && node tests/document-verification.test.mjs && node tests/admission.test.mjs && node tests/seed.test.mjs && node tests/system.test.mjs",
     "test:cache": "node tests/cache.test.mjs",
     "reset-db": "node scripts/reset-dev-database.mjs",
     "seed-test-data": "node scripts/import-test-data.mjs",

src/cache/redis-cache.mjs

@@ -5,26 +5,69 @@ function positiveInteger(value, fallback, maximum = Number.MAX_SAFE_INTEGER) {
   return Number.isInteger(parsed) && parsed > 0 ? Math.min(parsed, maximum) : fallback;
 }
 
-function disabledCache(status = 'disabled') {
+function disabledCache(status = 'disabled', { ttlSeconds = 60, maxEntries = 200 } = {}) {
+  const values = new Map();
+  const pending = new Map();
+  const generations = new Map();
+
+  const cacheKey = (namespace, key) => `${namespace}:${key}`;
+  const generation = namespace => generations.get(namespace) || 0;
+  const prune = () => {
+    const now = Date.now();
+    for (const [key, entry] of values) if (entry.expiresAt <= now) values.delete(key);
+    while (values.size > maxEntries) values.delete(values.keys().next().value);
+  };
+
   return {
     enabled: false,
     status,
-    async remember(_namespace, _key, loader) {
-      return loader();
+    async remember(namespace, key, loader, options = {}) {
+      const fullKey = cacheKey(namespace, key);
+      const cached = values.get(fullKey);
+      if (cached && cached.expiresAt > Date.now()) return cached.value;
+      if (cached) values.delete(fullKey);
+      const startedGeneration = generation(namespace);
+      const active = pending.get(fullKey);
+      if (active?.generation === startedGeneration) return active.promise;
+
+      const loading = Promise.resolve(loader()).then(value => {
+        if (generation(namespace) === startedGeneration) {
+          const lifetime = positiveInteger(options.ttlSeconds, ttlSeconds, 86400);
+          values.set(fullKey, { value, expiresAt: Date.now() + lifetime * 1000 });
+          prune();
+        }
+        return value;
+      }).finally(() => {
+        if (pending.get(fullKey)?.promise === loading) pending.delete(fullKey);
+      });
+      pending.set(fullKey, { generation: startedGeneration, promise: loading });
+      return loading;
     },
-    async invalidate() { return false; },
-    async close() {}
+    async invalidate(namespace) {
+      generations.set(namespace, generation(namespace) + 1);
+      const prefix = `${namespace}:`;
+      for (const key of values.keys()) if (key.startsWith(prefix)) values.delete(key);
+      // Preserve the public meaning of this return value: no Redis namespace
+      // was refreshed, even though the local fallback was invalidated.
+      return false;
+    },
+    async close() {
+      values.clear();
+      pending.clear();
+    }
   };
 }
 
 export async function createRedisCache({ env = process.env, logger = console, clientFactory = createClient } = {}) {
   const url = String(env.REDIS_URL || '').trim();
-  if (!url) return disabledCache();
+  const defaultTtlSeconds = positiveInteger(env.REDIS_CACHE_TTL_SECONDS, 60, 86400);
+  const localMaxEntries = positiveInteger(env.LOCAL_CACHE_MAX_ENTRIES, 200, 5000);
+  if (!url) return disabledCache('disabled', { ttlSeconds: defaultTtlSeconds, maxEntries: localMaxEntries });
+  const fallback = disabledCache('unavailable', { ttlSeconds: defaultTtlSeconds, maxEntries: localMaxEntries });
 
   const prefix = String(env.REDIS_CACHE_PREFIX || 'exam-information')
     .trim()
     .replace(/[^a-zA-Z0-9:_-]/g, '-') || 'exam-information';
-  const defaultTtlSeconds = positiveInteger(env.REDIS_CACHE_TTL_SECONDS, 60, 86400);
   const connectTimeout = positiveInteger(env.REDIS_CONNECT_TIMEOUT_MS, 1500, 30000);
   const pending = new Map();
   let warningReported = false;
@@ -54,7 +97,7 @@ export async function createRedisCache({ env = process.env, logger = console, cl
   } catch (error) {
     warn(error);
     if (client.isOpen) client.destroy();
-    return disabledCache('unavailable');
+    return fallback;
   }
 
   const versionKey = namespace => `${prefix}:namespace:${namespace}`;
@@ -75,7 +118,7 @@ export async function createRedisCache({ env = process.env, logger = console, cl
       return client.isReady ? 'ready' : 'unavailable';
     },
     async remember(namespace, key, loader, { ttlSeconds = defaultTtlSeconds } = {}) {
-      if (!client.isReady) return loader();
+      if (!client.isReady) return fallback.remember(namespace, key, loader, { ttlSeconds });
       try {
         const version = await namespaceVersion(namespace);
         const cacheKey = `${prefix}:${namespace}:${version}:${key}`;
@@ -99,10 +142,11 @@ export async function createRedisCache({ env = process.env, logger = console, cl
         return loading;
       } catch (error) {
         warn(error);
-        return loader();
+        return fallback.remember(namespace, key, loader, { ttlSeconds });
       }
     },
     async invalidate(namespace) {
+      await fallback.invalidate(namespace);
       if (!client.isReady) return false;
       try {
         await client.incr(versionKey(namespace));
@@ -113,6 +157,7 @@ export async function createRedisCache({ env = process.env, logger = console, cl
       }
     },
     async close() {
+      await fallback.close();
       if (client.isOpen) await client.quit();
     }
   };

src/client/api.mjs

@@ -1,4 +1,6 @@
-export async function api(path, options = {}) {
+const pendingReads = new Map();
+
+async function request(path, options) {
   const binaryBody = options.body instanceof ArrayBuffer || options.body instanceof Blob || options.body instanceof FormData;
   const response = await fetch(path, {
     credentials: 'same-origin',
@@ -15,3 +17,18 @@ export async function api(path, options = {}) {
   }
   return data;
 }
+
+export function api(path, options = {}) {
+  const method = String(options.method || 'GET').toUpperCase();
+  if (method !== 'GET' || options.body != null || options.signal) return request(path, options);
+
+  // A quick double click or repeated render must not download and parse the
+  // same large JSON response more than once while the first request is active.
+  const key = String(path);
+  if (pendingReads.has(key)) return pendingReads.get(key);
+  const loading = request(path, options).finally(() => {
+    if (pendingReads.get(key) === loading) pendingReads.delete(key);
+  });
+  pendingReads.set(key, loading);
+  return loading;
+}

src/database/mysql-adapter.mjs

@@ -1,5 +1,7 @@
 import { synchronizeMysqlPartitions } from './partition-storage.mjs';
 
+import { createStateCache } from './state-cache.mjs';
+
 export function createMysqlAdapter(context) {
   const {
     mkdir,
@@ -408,7 +410,9 @@ export function createMysqlAdapter(context) {
       partitionConnection.release();
     }
   
+    let stateCache;
     const transaction = async operations => {
+      stateCache?.invalidate();
       const connection = await pool.getConnection();
       try {
         await connection.beginTransaction();
@@ -419,10 +423,11 @@ export function createMysqlAdapter(context) {
         await connection.rollback();
         throw error;
       } finally {
+        stateCache?.invalidate();
         connection.release();
       }
     };
-    const read = async () => {
+    const loadState = async () => {
       const connection = await pool.getConnection();
       try {
         await connection.beginTransaction();
@@ -436,10 +441,11 @@ export function createMysqlAdapter(context) {
         connection.release();
       }
     };
+    stateCache = createStateCache({ load: loadState });
     return createRepository({
       client: 'mysql',
       location: connectionUrl ? 'DATABASE_URL' : `${process.env.MYSQL_HOST}:${process.env.MYSQL_PORT || 3306}/${database}`,
-      read,
+      read: stateCache.read,
       transaction,
       close: async () => pool.end()
     });

src/database/sqlite-adapter.mjs

@@ -1,4 +1,5 @@
 import { synchronizeSqlitePartitions } from './partition-storage.mjs';
+import { createStateCache } from './state-cache.mjs';
 
 export function createSqliteAdapter(context) {
   const {
@@ -19,8 +20,14 @@ export function createSqliteAdapter(context) {
     await mkdir(dirname(path), { recursive: true });
   
     const connection = new DatabaseSync(path, { timeout: 5000 });
-    connection.exec('PRAGMA journal_mode = WAL;');
-    connection.exec('PRAGMA synchronous = NORMAL;');
+    connection.exec(`
+      PRAGMA journal_mode = WAL;
+      PRAGMA synchronous = NORMAL;
+      PRAGMA temp_store = MEMORY;
+      PRAGMA cache_size = -65536;
+      PRAGMA mmap_size = 268435456;
+      PRAGMA wal_autocheckpoint = 1000;
+    `);
     const tableExists = name => Boolean(connection.prepare("SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?").get(name));
     const ensureColumns = (table, columns) => {
       if (!tableExists(table)) return;
@@ -484,7 +491,16 @@ export function createSqliteAdapter(context) {
 
     synchronizeSqlitePartitions(connection);
   
+    const dataVersion = connection.prepare('PRAGMA data_version');
+    const stateCache = createStateCache({
+      load: () => stateFromRows(readSqliteRows(connection)),
+      version: () => Number(dataVersion.get().data_version)
+    });
+
     const transaction = async operations => {
+      // Call this before the first possible await so request-local mutations of
+      // the previous snapshot can never be observed by another request.
+      stateCache.invalidate();
       connection.exec('BEGIN IMMEDIATE');
       try {
         for (const item of operations) connection.prepare(item.sql).run(...item.params);
@@ -493,12 +509,14 @@ export function createSqliteAdapter(context) {
       } catch (error) {
         connection.exec('ROLLBACK');
         throw error;
+      } finally {
+        stateCache.invalidate();
       }
     };
     return createRepository({
       client: 'sqlite',
       location: path,
-      read: async () => stateFromRows(readSqliteRows(connection)),
+      read: stateCache.read,
       transaction,
       close: async () => connection.close()
     });

src/database/state-cache.mjs

@@ -0,0 +1,55 @@
+function cacheDuration(value, fallback = 30000) {
+  const parsed = Number(value);
+  return Number.isFinite(parsed) && parsed >= 0 ? Math.min(Math.trunc(parsed), 3600000) : fallback;
+}
+
+/**
+ * Keeps the materialized application state in process instead of rebuilding it
+ * from every relational table for every HTTP request. Writes explicitly
+ * invalidate the snapshot; an optional version reader also detects changes made
+ * by another database connection.
+ */
+export function createStateCache({ load, version, maxAgeMs = process.env.DATABASE_STATE_CACHE_TTL_MS }) {
+  // A database-native version token is stronger than a timer, so SQLite can
+  // keep the snapshot indefinitely while still observing external commits.
+  const ttlMs = version ? 0 : cacheDuration(maxAgeMs);
+  let snapshot = null;
+  let snapshotVersion;
+  let loadedAt = 0;
+  let generation = 0;
+  let pending = null;
+
+  function invalidate() {
+    generation += 1;
+    snapshot = null;
+    snapshotVersion = undefined;
+    loadedAt = 0;
+    pending = null;
+  }
+
+  async function read() {
+    const currentVersion = version ? await version() : undefined;
+    const freshByAge = !ttlMs || Date.now() - loadedAt < ttlMs;
+    if (snapshot && freshByAge && (!version || currentVersion === snapshotVersion)) return snapshot;
+
+    if (pending && (!version || pending.version === currentVersion)) return pending.promise;
+
+    const startedGeneration = generation;
+    const loading = Promise.resolve().then(load).then(state => {
+      if (generation === startedGeneration) {
+        snapshot = state;
+        snapshotVersion = currentVersion;
+        loadedAt = Date.now();
+      }
+      return state;
+    });
+    pending = { version: currentVersion, promise: loading };
+    try {
+      return await loading;
+    } finally {
+      if (pending?.promise === loading) pending = null;
+    }
+  }
+
+  return { read, invalidate };
+}

tests/api-dedup.test.mjs

@@ -0,0 +1,25 @@
+import assert from 'node:assert/strict';
+import { api } from '../src/client/api.mjs';
+
+const originalFetch = globalThis.fetch;
+let calls = 0;
+let release;
+globalThis.fetch = async () => {
+  calls += 1;
+  await new Promise(resolve => { release = resolve; });
+  return new Response(JSON.stringify({ ok: true, calls }), { status: 200, headers: { 'Content-Type': 'application/json' } });
+};
+
+try {
+  const first = api('/api/large-ledger');
+  const second = api('/api/large-ledger');
+  await Promise.resolve();
+  assert.equal(calls, 1, '并发的相同 GET 请求应只发送一次');
+  release();
+  assert.deepEqual(await first, { ok: true, calls: 1 });
+  assert.deepEqual(await second, { ok: true, calls: 1 });
+} finally {
+  globalThis.fetch = originalFetch;
+}
+
+console.log('API read deduplication tests passed');

tests/cache.test.mjs

@@ -83,16 +83,35 @@ const silentLogger = { warn() {} };
   await cache.invalidate('results');
   assert.deepEqual(await cache.remember('results', 'candidate:1', loadResults), { version: 3 });
   assert.equal(resultLoads, 3, '后台手动刷新后应重新生成成绩缓存');
+
+  client.isReady = false;
+  let fallbackLoads = 0;
+  assert.equal(await cache.remember('public', 'runtime-fallback', async () => ++fallbackLoads), 1);
+  assert.equal(await cache.remember('public', 'runtime-fallback', async () => ++fallbackLoads), 1);
+  assert.equal(fallbackLoads, 1, 'Redis 运行中断开后，相同热点读取应由本机缓存合并');
+  client.isReady = true;
   await cache.close();
 }
 
 {
   const cache = await createRedisCache({ env: {}, logger: silentLogger });
   let loads = 0;
-  await cache.remember('public', 'home', async () => ++loads);
-  await cache.remember('public', 'home', async () => ++loads);
-  assert.equal(loads, 2, '未配置 Redis 时应始终回源');
+  assert.equal(await cache.remember('public', 'home', async () => ++loads), 1);
+  assert.equal(await cache.remember('public', 'home', async () => ++loads), 1);
+  assert.equal(loads, 1, '未配置 Redis 时应使用有界本机缓存，避免重复回源');
+  await cache.invalidate('public');
+  assert.equal(await cache.remember('public', 'home', async () => ++loads), 2);
+  assert.equal(loads, 2, '本机缓存应在数据库写入后立即失效');
   assert.equal(cache.status, 'disabled');
+
+  let releaseOld;
+  const oldLoad = cache.remember('public', 'race', () => new Promise(resolve => { releaseOld = () => resolve('old'); }));
+  await cache.invalidate('public');
+  const newLoad = cache.remember('public', 'race', async () => 'new');
+  releaseOld();
+  assert.equal(await oldLoad, 'old');
+  assert.equal(await newLoad, 'new', '失效后不得等待失效前仍在运行的加载');
+  assert.equal(await cache.remember('public', 'race', async () => 'unexpected'), 'new', '旧加载完成后不得覆盖新缓存');
 }
 
 {
@@ -104,6 +123,7 @@ const silentLogger = { warn() {} };
   });
   assert.equal(cache.status, 'unavailable');
   assert.equal(await cache.remember('public', 'home', async () => 'database'), 'database');
+  assert.equal(await cache.remember('public', 'home', async () => 'unexpected'), 'database', 'Redis 故障时本机缓存应继续承接重复读取');
 }
 
 console.log('Redis 缓存测试通过');

tests/state-cache.test.mjs

@@ -0,0 +1,41 @@
+import assert from 'node:assert/strict';
+import { createStateCache } from '../src/database/state-cache.mjs';
+
+let loads = 0;
+let release;
+const gate = new Promise(resolve => { release = resolve; });
+const coalesced = createStateCache({
+  maxAgeMs: 30000,
+  async load() {
+    loads += 1;
+    await gate;
+    return { load: loads };
+  }
+});
+const firstPending = coalesced.read();
+const secondPending = coalesced.read();
+release();
+const [first, second] = await Promise.all([firstPending, secondPending]);
+assert.equal(loads, 1, '并发冷读取应合并为一次全量加载');
+assert.strictEqual(first, second, '并发读取应共享同一份快照');
+assert.strictEqual(await coalesced.read(), first, '有效期内应直接复用内存快照');
+
+coalesced.invalidate();
+const afterInvalidation = await coalesced.read();
+assert.equal(loads, 2, '应用写入失效后应重新加载');
+assert.notStrictEqual(afterInvalidation, first, '失效后不得继续返回旧快照');
+
+let version = 1;
+let versionLoads = 0;
+const versioned = createStateCache({
+  version: () => version,
+  load: () => ({ load: ++versionLoads })
+});
+const versionOne = await versioned.read();
+assert.strictEqual(await versioned.read(), versionOne, '数据库版本未变化时应复用快照');
+version += 1;
+const versionTwo = await versioned.read();
+assert.equal(versionLoads, 2, '外部数据库版本变化后应重新加载');
+assert.notStrictEqual(versionTwo, versionOne, '外部写入后不得返回旧快照');
+
+console.log('State cache tests passed');