diff --git a/.env.example b/.env.example index d123fa5..b1cd003 100644 --- a/.env.example +++ b/.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 diff --git a/README.md b/README.md index 0614054..2944910 100644 --- a/README.md +++ b/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' diff --git a/package.json b/package.json index d5ab41b..ca47516 100644 --- a/package.json +++ b/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", diff --git a/src/cache/redis-cache.mjs b/src/cache/redis-cache.mjs index 8629a9a..7b6a4ec 100644 --- a/src/cache/redis-cache.mjs +++ b/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(); } }; diff --git a/src/client/api.mjs b/src/client/api.mjs index 3dfbed4..40599a4 100644 --- a/src/client/api.mjs +++ b/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; +} diff --git a/src/database/mysql-adapter.mjs b/src/database/mysql-adapter.mjs index 5c98b43..60c714b 100644 --- a/src/database/mysql-adapter.mjs +++ b/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() }); diff --git a/src/database/sqlite-adapter.mjs b/src/database/sqlite-adapter.mjs index bc8d405..14b4884 100644 --- a/src/database/sqlite-adapter.mjs +++ b/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() }); diff --git a/src/database/state-cache.mjs b/src/database/state-cache.mjs new file mode 100644 index 0000000..c71bdbe --- /dev/null +++ b/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 }; +} diff --git a/tests/api-dedup.test.mjs b/tests/api-dedup.test.mjs new file mode 100644 index 0000000..0536a7d --- /dev/null +++ b/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'); diff --git a/tests/cache.test.mjs b/tests/cache.test.mjs index 595020d..225f0ee 100644 --- a/tests/cache.test.mjs +++ b/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 缓存测试通过'); diff --git a/tests/state-cache.test.mjs b/tests/state-cache.test.mjs new file mode 100644 index 0000000..a8e66fd --- /dev/null +++ b/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');