From f2a92d9151c681d0cb8db225e30c7312018ed9b2 Mon Sep 17 00:00:00 2001 From: pikachuren Date: Mon, 9 Nov 2026 18:57:36 +0800 Subject: [PATCH] =?UTF-8?q?fix(store):=20key=20=E6=A0=BC=E5=BC=8F=E6=94=B9?= =?UTF-8?q?=E7=94=A8=E5=85=A8=E5=B1=80=E7=B4=A2=E5=BC=95=EF=BC=8C=E6=91=86?= =?UTF-8?q?=E8=84=B1=20list=20=E9=85=8D=E9=A2=9D=E5=B9=B6=E6=8F=90?= =?UTF-8?q?=E9=80=9F=20(#125)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 5 + readmes/README_en.md | 6 + .../internal/model/store/driver/kv.test.ts | 23 ++ src/backend/internal/model/store/driver/kv.ts | 102 +++-- .../internal/model/store/format/key.ts | 361 +++++++++++++++--- src/backend/internal/model/store/keycodec.ts | 15 + .../internal/model/store/store.test.ts | 160 +++++++- 7 files changed, 560 insertions(+), 112 deletions(-) diff --git a/README.md b/README.md index bca3fac3..7b9a859f 100644 --- a/README.md +++ b/README.md @@ -153,6 +153,11 @@ pnpm run deploy:worker | `DB_DRIVER` | `auto` | 可选 `auto` / `blob` / `cfkv` / `kv` / `d1` / `do` / `mysql` | | `DB_CIPHER` | `none` | 敏感字段落盘算法，如 `aes-256-gcm` | +> **KV 配额提示**：`DB_FORMAT=key` 会维护全局索引键（`openlist_key_index`）， +> 首次读取旧数据后 load/save 不再消耗 KV `list` 配额。但每次保存仍按记录数写入 +> （默认配置约 100+ 条），Cloudflare KV 免费版只有 1000 写/天，长期使用建议 +> `DB_DRIVER=d1` 或 `DB_FORMAT=map`。 + 完整环境变量，请参阅[配置变量](https://doc.oplist.org/ecosystem/official_worker/guide_env) diff --git a/readmes/README_en.md b/readmes/README_en.md index f013e124..30121000 100644 --- a/readmes/README_en.md +++ b/readmes/README_en.md @@ -152,6 +152,12 @@ After deployment, set the following variables: | `DB_DRIVER` | `auto` | `auto` / `blob` / `cfkv` / `kv` / `d1` / `do` / `mysql` | | `DB_CIPHER` | `none` | At-rest cipher for sensitive fields, e.g. `aes-256-gcm` | +> **KV quota note**: `DB_FORMAT=key` maintains a global index key (`openlist_key_index`), +> so after the first read of legacy data, load/save no longer consume KV `list` quota. +> Each save still writes one entry per record (about 100+ with the default config), and +> the Cloudflare KV free tier allows only 1000 writes/day — prefer `DB_DRIVER=d1` or +> `DB_FORMAT=map` for long-term use. + For the full variable list, see [Configuration variables](https://doc.oplist.org/ecosystem/official_worker/guide_env). ## Support diff --git a/src/backend/internal/model/store/driver/kv.test.ts b/src/backend/internal/model/store/driver/kv.test.ts index e1b7dda3..a46081d5 100644 --- a/src/backend/internal/model/store/driver/kv.test.ts +++ b/src/backend/internal/model/store/driver/kv.test.ts @@ -1,6 +1,7 @@ import assert from "node:assert/strict" import test from "node:test" import { kvDriver } from "./kv" +import { KEY_INDEX_KEY } from "../keycodec" /** * issue #125 回归：binding 模式 list() 必须同时兼容两种平台语义。 @@ -144,6 +145,28 @@ test("kv health (binding, key format) exercises list so exhausted quota is not r assert.equal(listCalls, 1) }) +test("kv health (binding, key format) skips list when the key index exists", async () => { + // 索引存在时 key 格式的稳态读写只依赖 get，健康检查不应再白白消耗 + // 1 次 list 配额（issue #125：list 配额只有 1000 次/天）。 + let listCalls = 0 + const health = await kvDriver.health({ + DB_FORMAT: "key", + KV: { + get: async (key: string) => + key === KEY_INDEX_KEY ? '{"v":1,"keys":["users_1"]}' : null, + put: async () => {}, + delete: async () => {}, + list: async () => { + listCalls += 1 + return { keys: [], list_complete: true } + }, + }, + }) + + assert.equal(health.available, true) + assert.equal(listCalls, 0) +}) + test("kv health (binding, map format) does not require an unsupported list API", async () => { // ESA EdgeKV exposes get/put/delete but no list. Map format only needs the // single openlist_config key, so it must still be considered healthy. diff --git a/src/backend/internal/model/store/driver/kv.ts b/src/backend/internal/model/store/driver/kv.ts index 6dfc3423..7a5d0d6d 100644 --- a/src/backend/internal/model/store/driver/kv.ts +++ b/src/backend/internal/model/store/driver/kv.ts @@ -9,6 +9,7 @@ */ import { sanitizeProxyOrigin } from "../proxy" import { KV_HEALTH_PROBE_KEY, readFormat } from "../config" +import { KEY_INDEX_KEY } from "../keycodec" import type { Driver, EnvContext } from "../types" /** @@ -295,6 +296,50 @@ function requireProxyBaseUrl(env?: EnvContext): string { return baseUrl } +/** + * 读取 binding 的文本值并归一化为 `string | null`。 + * + * 只从 binding 读取，不涉及代理分支。抽成独立函数的原因： + * `get()` 与 `health()` 都需要「兼容两种 get 签名」，各写一份迟早漂移。 + * + * Cloudflare KV 用 `get(key, "text")`，EdgeOne KV 用 `get(key, {type:"text"})`， + * 两者签名不兼容，需按序尝试并归一化返回值（避免拿到对象导致上游 JSON.parse 失败）。 + * + * 但要区分「签名不兼容」与「真实故障」： + * - 签名不兼容会抛 TypeError（参数类型不符），此时才应回退到第二种签名； + * - 网络/权限错误（401、超时等）必须**原样抛出**，否则会被第二次尝试 + * 掩盖成 "key not found"，让调用方误判为数据不存在而写入错误状态。 + */ +async function kvGetText(kv: any, key: string): Promise { + const isSignatureError = (e: any): boolean => + e instanceof TypeError || + /not a function|invalid|unexpected|argument/i.test(String(e?.message || "")) + + let value: any + try { + value = await kv.get(key, "text") + } catch (e) { + if (!isSignatureError(e)) throw e + value = undefined + } + + if (value === undefined || value === null) { + try { + value = await kv.get(key, { type: "text" }) + } catch (e) { + // 若第一种签名已成功执行（未抛签名错误），说明第二种只是"空结果"的回退， + // 此时真实错误同样要抛出，不得吞掉。 + if (!isSignatureError(e)) throw e + value = null + } + } + + if (value === undefined || value === null) return null + if (typeof value === "string") return value + // 绑定误返回对象时统一序列化，保持 Driver.get 的 string 契约 + return JSON.stringify(value) +} + export const kvDriver: Driver = { name: "kv", @@ -326,42 +371,7 @@ export const kvDriver: Driver = { // 模式1: Binding 模式 if (kv) { - // Cloudflare KV 用 get(key, "text")，EdgeOne KV 用 get(key, {type:"text"})， - // 两者签名不兼容，需按序尝试并归一化返回值（避免拿到对象导致上游 JSON.parse 失败）。 - // - // 但要区分「签名不兼容」与「真实故障」： - // - 签名不兼容会抛 TypeError（参数类型不符），此时才应回退到第二种签名； - // - 网络/权限错误（401、超时等）必须**原样抛出**，否则会被第二次尝试 - // 掩盖成 "key not found"，让调用方误判为数据不存在而写入错误状态。 - const isSignatureError = (e: any): boolean => - e instanceof TypeError || - /not a function|invalid|unexpected|argument/i.test( - String(e?.message || ""), - ) - - let value: any - try { - value = await kv.get(key, "text") - } catch (e) { - if (!isSignatureError(e)) throw e - value = undefined - } - - if (value === undefined || value === null) { - try { - value = await kv.get(key, { type: "text" }) - } catch (e) { - // 若第一种签名已成功执行（未抛签名错误），说明第二种只是"空结果"的回退， - // 此时真实错误同样要抛出，不得吞掉。 - if (!isSignatureError(e)) throw e - value = null - } - } - - if (value === undefined || value === null) return null - if (typeof value === "string") return value - // 绑定误返回对象时统一序列化，保持 Driver.get 的 string 契约 - return JSON.stringify(value) + return kvGetText(kv, key) } // 模式2: HTTP 代理模式（无原生 binding → 经 Edge Function 代理） @@ -564,11 +574,12 @@ export const kvDriver: Driver = { } else { await kv.get(KV_HEALTH_PROBE_KEY) } - // key 格式的 load/save 依赖 list；map 格式只读写固定的 openlist_config, - // 不要求 EdgeKV 暴露 list API（ESA EdgeKV 内置 API 仅提供 get/put/delete）。 - // 只有 key 格式才探测 list，避免 map 格式部署被不支持的 list() 误判为 - // 存储不可用（issue #159）。对 key 格式仍检查 list，防止配额耗尽/接口异常 - // 导致 env_check 误报 ready（issue #125）。使用专用前缀，取回空页只耗 1 次。 + // key 格式的稳态读写只依赖 get/put + 全局索引（issue #125）： + // - 索引键存在时：一次 load 只花 1 次索引 get + N 次记录 get，不需要 list； + // - 索引缺失时（旧部署首次升级 / 超大索引）：load/save 会回退 list， + // 此时必须确认 list 可用，否则初始化会 500。 + // 因此这里先读索引键（1 次 read，比 list 便宜两个数量级），只有索引缺失 + // 才真的调用 list 探测。map 格式不读写索引，完全不涉及 list。 // 直接复用后端的 readFormat()：健康检查与真实读写必须用同一份解析逻辑 // （env 优先级、缺省 map、DB_DRIVER=json 别名），复刻一份迟早漂移。 if (readFormat(env) === "key") { @@ -577,7 +588,14 @@ export const kvDriver: Driver = { "KV binding does not support list(), which is required by DB_FORMAT=key", ) } - await kv.list({ prefix: KV_HEALTH_PROBE_KEY, cursor: "", limit: 256 }) + const indexRaw = await kvGetText(kv, KEY_INDEX_KEY) + if (indexRaw === null) { + await kv.list({ + prefix: KV_HEALTH_PROBE_KEY, + cursor: "", + limit: 256, + }) + } } return { driver: "kv", diff --git a/src/backend/internal/model/store/format/key.ts b/src/backend/internal/model/store/format/key.ts index 6857706b..986ddb7a 100644 --- a/src/backend/internal/model/store/format/key.ts +++ b/src/backend/internal/model/store/format/key.ts @@ -16,13 +16,55 @@ * * 转义方案无需反向解析：所有读写都是"构造键名"，实体的主键值一律 * 来自记录 JSON 本身，key 只用作存储地址。 + * + * ## 全局索引（issue #125） + * + * 历史实现每次 load/save 都靠 `list()` 枚举键名，即使 6 张表合并成 1 次 + * 全局 list，Cloudflare 免费版 1000 次/天的 list 配额也只够约 1000 次 + * load/save —— 页面刷新几次就见底（且大库一次 list 还可能要翻多页）。 + * + * 现在把「当前存活的实体键」写入 `KEY_INDEX_KEY`（见 keycodec）： + * - load：1 次 `get` 索引 + N 次 `get` 记录，不再消耗 list； + * - save：写记录 → 写新索引 → 按旧索引删除陈旧键，同样不消耗 list； + * - 索引缺失（旧部署首次升级 / 索引被删）时回退到 1 次 list 并**补写索引**， + * 因此升级无需手工迁移，之后也不再重复 list； + * - 索引损坏或超过单值上限（EdgeOne KV 1 MB）时不使用它并回退 list， + * 宁可多耗一次 list，也不能让读取看不到数据。 + * + * 索引是「成员名单」而非数据本身：记录值仍以实体键为唯一权威。写入顺序 + * 刻意是「先写记录 → 再写索引 → 最后删陈旧键」，任一步失败都不会让 + * 已经成功写入的记录从索引中消失（最坏情况只是残留几个无引用的旧键）。 */ import type { FormatAdapter, Driver } from "../types" import { TABLE_NAMES, TABLE_KEY, type TableName } from "../schema" -import { entityKeyOf, tableKeyPrefix } from "../keycodec" +import { entityKeyOf, tableKeyPrefix, KEY_INDEX_KEY } from "../keycodec" const INIT_MARK = "openlist_config" +/** 索引文档版本：结构变化时递增，旧版本会按「无索引」回退到 list。 */ +const INDEX_VERSION = 1 + +/** + * 索引值的字节上限。 + * + * EdgeOne KV 单个 value 最大 1 MB，Cloudflare 为 25 MiB；取 900 KB 留出 + * 余量。索引里的键名均为 ASCII（keycodec 保证），因此 length 近似等于字节数。 + * 超过上限时放弃索引：大库继续走 list，行为与升级前一致。 + */ +const INDEX_MAX_BYTES = 900 * 1024 + +/** + * 记录读写的并发度。 + * + * key 格式一次 load 要逐条 get，默认库就有 100+ 条记录；串行执行在边缘 + * 网络上会把单次请求拖到数秒。并发 12 把往返压缩到约 1/10，同时远低于 + * Workers 的子请求上限。 + */ +const IO_CONCURRENCY = 12 + +/** 超大索引告警只打印一次，避免每次 load/save 刷屏。 */ +let oversizeWarned = false + /** 构造表前缀（使用共享键名编码，兼容 EdgeOne KV 字符集约束） */ function tablePrefix(table: string): string { return tableKeyPrefix(table) @@ -33,23 +75,142 @@ function fullKeyOf(table: string, id: string): string { return entityKeyOf(table, id) } +/** + * 以受限并发执行异步任务，结果顺序与输入一致。 + * + * 不使用 `Promise.all(items.map(fn))`：记录数可能上千，一次性并发全部 + * 子请求会撞上平台限制；也不串行，否则每次 load 的耗时随记录数线性增长。 + */ +async function runBounded( + items: readonly T[], + limit: number, + fn: (item: T, index: number) => Promise, +): Promise { + const results = new Array(items.length) + let cursor = 0 + const workers = Array.from( + { length: Math.min(Math.max(1, limit), items.length) }, + async () => { + while (true) { + const index = cursor++ + if (index >= items.length) return + results[index] = await fn(items[index], index) + } + }, + ) + await Promise.all(workers) + return results +} + +/** + * 读取全局索引。 + * + * 任何「读不到 / 解析失败 / 结构不识别 / 版本不符」都返回 null，由调用方 + * 回退到 list。索引只是加速缓存，绝不能因为它损坏就返回空数据库 —— + * 那会被上层当成「全新部署」，进而用空库覆盖真实数据。 + */ +async function readIndex(driver: Driver, env?: any): Promise { + let raw: string | null + try { + raw = await driver.get(KEY_INDEX_KEY, env) + } catch (err) { + console.warn( + "[keyFormat] failed to read key index; falling back to list:", + err, + ) + return null + } + if (!raw) return null + + try { + const parsed: any = JSON.parse(raw) + if ( + parsed && + typeof parsed === "object" && + parsed.v === INDEX_VERSION && + Array.isArray(parsed.keys) && + parsed.keys.every((k: any) => typeof k === "string" && k.length > 0) + ) { + return parsed.keys as string[] + } + console.warn( + "[keyFormat] key index has an unknown shape; falling back to list", + ) + } catch (err) { + console.warn( + "[keyFormat] key index is not valid JSON; falling back to list:", + err, + ) + } + return null +} + +/** + * 写入全局索引。 + * + * @returns true 表示索引已落盘；false 表示索引过大被跳过（此时会删除旧索引， + * 否则下一次 load 会读到「缺少新键」的旧索引而看不到新数据）。 + */ +async function writeIndex( + driver: Driver, + keys: string[], + env?: any, +): Promise { + const payload = JSON.stringify({ v: INDEX_VERSION, keys }) + + if (payload.length > INDEX_MAX_BYTES) { + if (!oversizeWarned) { + oversizeWarned = true + console.warn( + `[keyFormat] key index is ${payload.length} bytes, over the ` + + `${INDEX_MAX_BYTES}-byte limit; falling back to list() for this ` + + `deployment. Reduce record count or use DB_FORMAT=map/sql.`, + ) + } + // 关键：旧索引必须删除，否则它会成为一份「永远不会被修正」的过期名单。 + try { + await driver.delete(KEY_INDEX_KEY, env) + } catch (err) { + console.warn("[keyFormat] failed to drop stale key index:", err) + } + return false + } + + await driver.put(KEY_INDEX_KEY, payload, env) + return true +} + +/** + * 读取当前存活键：优先索引，缺失时回退 list。 + * + * `fromIndex` 用于决定是否需要补写索引；即使索引存在，返回的键也可能 + * 含已删除实体（删除阶段失败），后续 get 会得到 null 并被跳过。 + */ +async function readMembership( + driver: Driver, + env?: any, +): Promise<{ keys: string[]; fromIndex: boolean }> { + const indexed = await readIndex(driver, env) + if (indexed) return { keys: indexed, fromIndex: true } + return { keys: await driver.list("", env), fromIndex: false } +} + /** * 将一次全局 list 的结果按表前缀分组。 * - * key 格式的键名统一为 `_`（见 keycodec），因此一次空前缀 - * list 即可取回全部表的键，再按前缀分拣 —— 结果与逐表 list 完全一致， - * 但把 load/save 各自的 list 消耗从「每表 1 次（6 张表 = 6 次）」降为 - * 1 次。Cloudflare 免费版 list 配额仅 1000 次/天（issue #125）， - * key + kv 组合下这是可持续性的关键；对配额无限的平台也无害（省 5 次 - * 往返）。 + * 键名统一为 `
_`（见 keycodec），因此一次空前缀 list 即可取回 + * 全部表的键，再按前缀分拣 —— 结果与逐表 list 完全一致。排序保证记录顺序 + * 与 KV list 的字典序一致，不因索引建立/迁移而变化。 * - * 命名空间中不属于任何表前缀的键（如 INIT_MARK、其他应用的键）会被 - * 忽略，与旧实现「逐表前缀 list 天然排除它们」的行为一致。 + * 命名空间中不属于任何表前缀的键（如 INIT_MARK、索引键、其他应用的键） + * 会被忽略，与旧实现「逐表前缀 list 天然排除它们」的行为一致。 */ function groupKeysByTable(allKeys: string[]): Record { const out = {} as Record for (const table of TABLE_NAMES) out[table] = [] - for (const key of allKeys) { + + // 去重 + 排序：索引损坏/人工编辑可能产生重复键，重复 get 会返回重复记录。 + for (const key of [...new Set(allKeys)].sort()) { for (const table of TABLE_NAMES) { if (key.startsWith(tablePrefix(table))) { out[table].push(key) @@ -60,6 +221,29 @@ function groupKeysByTable(allKeys: string[]): Record { return out } +/** 并发读取并解析一批实体记录（空值跳过，损坏记录告警后跳过）。 */ +async function fetchRecords( + keys: string[], + driver: Driver, + env?: any, +): Promise { + const raws = await runBounded(keys, IO_CONCURRENCY, (key) => + driver.get(key, env), + ) + + const records: any[] = [] + for (let i = 0; i < raws.length; i++) { + const raw = raws[i] + if (!raw) continue + try { + records.push(JSON.parse(raw)) + } catch (err) { + console.warn(`[keyFormat] Failed to parse ${keys[i]}:`, err) + } + } + return records +} + export const keyFormat: FormatAdapter = { name: "key", @@ -68,25 +252,28 @@ export const keyFormat: FormatAdapter = { const mark = await driver.get(INIT_MARK, env) if (!mark) return null - // 一次全局 list 代替逐表 list（6 张表 = 6 次），见 groupKeysByTable - const grouped = groupKeysByTable(await driver.list("", env)) + const { keys, fromIndex } = await readMembership(driver, env) + const grouped = groupKeysByTable(keys) + + // 旧部署首次升级 / 索引被删：本次已付出 1 次 list，顺手把索引补上， + // 后续请求即可只走 get。补写失败不影响本次读取（下次再扫一次）。 + // 只收录实体键：索引键、INIT_MARK 与其他应用的键不进入名单。 + if (!fromIndex) { + const entityKeys = TABLE_NAMES.flatMap((table) => grouped[table]).sort() + try { + await writeIndex(driver, entityKeys, env) + } catch (err) { + console.warn( + "[keyFormat] failed to persist key index; next load will rescan:", + err, + ) + } + } const out: Record = {} for (const table of TABLE_NAMES) { - const records = [] - for (const key of grouped[table]) { - const raw = await driver.get(key, env) - if (raw) { - try { - records.push(JSON.parse(raw)) - } catch (err) { - console.warn(`[keyFormat] Failed to parse ${key}:`, err) - } - } - } - - out[table] = records + out[table] = await fetchRecords(grouped[table], driver, env) } return out @@ -102,7 +289,9 @@ export const keyFormat: FormatAdapter = { // // 改为先写后删后，失败时的最坏情况是**残留**旧键（可被下一次成功保存清理）， // 而不会丢数据；正常成功路径的最终状态与原来完全一致。 - const liveKeys = new Set() + // Map 天然去重：同一 id 出现多次时只保留最后一次，key 名也不会重复写。 + const writes = new Map() + for (const table of TABLE_NAMES) { const keyCol = TABLE_KEY[table] const records = data?.[table] || [] @@ -111,78 +300,111 @@ export const keyFormat: FormatAdapter = { const id = String(record?.[keyCol] ?? "") if (!id) continue - const key = fullKeyOf(table, id) - liveKeys.add(key) - const value = JSON.stringify(record) - await driver.put(key, value, env) + writes.set(fullKeyOf(table, id), JSON.stringify(record)) } } - // 清理本快照中已不存在的键（删除的实体 / 改名的键） - // 一次全局 list 代替逐表 list（6 张表 = 6 次），见 groupKeysByTable - const grouped = groupKeysByTable(await driver.list("", env)) + const liveKeys = new Set(writes.keys()) + + // 并发写记录：key 格式的默认库就有 100+ 条，串行写会明显拖慢保存。 + await runBounded([...writes], IO_CONCURRENCY, ([key, value]) => + driver.put(key, value, env), + ) + + // 旧成员名单：优先索引；索引缺失（首次保存/旧部署）才扫一次 list。 + const { keys: previousKeys } = await readMembership(driver, env) + + // 先写索引，再删陈旧键。顺序不能反： + // - 索引先写：并发读最多看到「新名单 + 旧残留键」，不会看到「旧名单 + 已删除键」； + // - 若索引写失败，删除旧索引并抛错，下一次 load 会 list 兜底看到全部新记录。 + try { + await writeIndex(driver, [...liveKeys].sort(), env) + } catch (err) { + try { + await driver.delete(KEY_INDEX_KEY, env) + } catch { + // 删除失败也只能作罢：下一次 load 的索引与记录不一致时仍能读到已有记录， + // 但新记录可能暂时不可见。此处不应掩盖原始写入错误。 + } + throw err + } + + // 清理本快照中已不存在的键（删除的实体 / 改名的键）。 + // 只处理属于已知表前缀的键，避免误删 INIT_MARK、索引键或其他应用的键。 + const stale: string[] = [] + const grouped = groupKeysByTable(previousKeys) for (const table of TABLE_NAMES) { for (const key of grouped[table]) { - if (liveKeys.has(key)) continue - await driver.delete(key, env) + if (!liveKeys.has(key)) stale.push(key) } } + await runBounded(stale, IO_CONCURRENCY, (key) => driver.delete(key, env)) // 标记已初始化 await driver.put(INIT_MARK, String(Date.now()), env) return true }, - + async getTable(table: string, driver: Driver, env?: any): Promise { const prefix = tablePrefix(table) - const keys = await driver.list(prefix, env) - - const records = [] - for (const key of keys) { - const raw = await driver.get(key, env) - if (raw) { - try { - records.push(JSON.parse(raw)) - } catch (err) { - console.warn(`[keyFormat] Failed to parse ${key}:`, err) - } - } - } + const indexed = await readIndex(driver, env) + // 有索引时直接过滤，不再 list（issue #125 的 list 配额问题） + const keys = indexed + ? indexed.filter((key) => key.startsWith(prefix)).sort() + : await driver.list(prefix, env) - return records + return fetchRecords(keys, driver, env) }, async saveTable( table: string, records: any[], driver: Driver, - env?: any + env?: any, ): Promise { const keyCol = TABLE_KEY[table as keyof typeof TABLE_KEY] const prefix = tablePrefix(table) - // 清空旧数据 - const oldKeys = await driver.list(prefix, env) - for (const key of oldKeys) { - await driver.delete(key, env) - } + const indexed = await readIndex(driver, env) + const oldKeys = indexed + ? indexed.filter((key) => key.startsWith(prefix)) + : await driver.list(prefix, env) - // 写入新数据 + // 同 save：先写新记录，再删改名前/已删除的旧键，避免失败时整表变空。 + const newKeys = new Set() + const writes: Array<{ key: string; value: string }> = [] for (const record of records) { const id = String(record?.[keyCol] ?? "") if (!id) continue const key = fullKeyOf(table, id) - const value = JSON.stringify(record) - await driver.put(key, value, env) + newKeys.add(key) + writes.push({ key, value: JSON.stringify(record) }) + } + + await runBounded(writes, IO_CONCURRENCY, (w) => + driver.put(w.key, w.value, env), + ) + + // 索引存在时同步替换该表的键；索引缺失就保持缺失，下一次 load 会 list 重建。 + if (indexed) { + const merged = indexed.filter((key) => !key.startsWith(prefix)) + merged.push(...newKeys) + await writeIndex(driver, merged.sort(), env) } + + await runBounded( + oldKeys.filter((key) => !newKeys.has(key)), + IO_CONCURRENCY, + (key) => driver.delete(key, env), + ) }, async getRecord( table: string, key: string, driver: Driver, - env?: any + env?: any, ): Promise { const fullKey = fullKeyOf(table, key) const raw = await driver.get(fullKey, env) @@ -201,20 +423,35 @@ export const keyFormat: FormatAdapter = { key: string, record: any, driver: Driver, - env?: any + env?: any, ): Promise { const fullKey = fullKeyOf(table, key) const value = JSON.stringify(record) await driver.put(fullKey, value, env) + + // 索引存在时必须同步加入，否则后续 load 会漏掉这条记录。 + const indexed = await readIndex(driver, env) + if (indexed && !indexed.includes(fullKey)) { + await writeIndex(driver, [...indexed, fullKey].sort(), env) + } }, async deleteRecord( table: string, key: string, driver: Driver, - env?: any + env?: any, ): Promise { const fullKey = fullKeyOf(table, key) await driver.delete(fullKey, env) + + const indexed = await readIndex(driver, env) + if (indexed?.includes(fullKey)) { + await writeIndex( + driver, + indexed.filter((k) => k !== fullKey), + env, + ) + } }, } diff --git a/src/backend/internal/model/store/keycodec.ts b/src/backend/internal/model/store/keycodec.ts index 93deea24..e27b0cdd 100644 --- a/src/backend/internal/model/store/keycodec.ts +++ b/src/backend/internal/model/store/keycodec.ts @@ -90,6 +90,21 @@ export function decodeKeyPart(input: string): string { return out } +/** + * key 格式的全局索引键。 + * + * key 格式把每张表摊平成 `
_` 多条记录，历史实现靠 `list()` 枚举键名； + * Cloudflare 免费版 list 配额只有 1000 次/天，而一次 load/save 就要扫全量键， + * 免费版很快就会耗尽（issue #125）。索引键记录当前存活记录的完整键名， + * 用 1 次 `get` 取代 1 次（可能多页的）`list`。 + * + * 约束： + * - 只含字母/数字/下划线，满足 EdgeOne KV 的键名限制； + * - 不以任何表前缀开头（`settings_` / `users_` …），不会被 `groupKeysByTable` + * 误当成实体记录，也不会被陈旧键清理误删。 + */ +export const KEY_INDEX_KEY = "openlist_key_index" + /** 构造某张表的键名前缀，如 `users_` */ export function tableKeyPrefix(table: string): string { return `${encodeKeyPart(table)}${KEY_SEP}` diff --git a/src/backend/internal/model/store/store.test.ts b/src/backend/internal/model/store/store.test.ts index 3b03ed79..0b2eea03 100644 --- a/src/backend/internal/model/store/store.test.ts +++ b/src/backend/internal/model/store/store.test.ts @@ -5,6 +5,7 @@ import { readDriver, readFormat, getStoreBackend } from "./backend" import { readPersistedSecret, writePersistedSecret } from "./json" import { mapFormat } from "./format/map" import { keyFormat } from "./format/key" +import { KEY_INDEX_KEY } from "./keycodec" import { sqlFormat } from "./format/sql" import { d1Driver } from "./driver/d1" import { @@ -228,7 +229,10 @@ test("schema: SQL table names align with Go GORM naming", () => { // 完整表名 = 前缀 + 复数名 assert.equal(tableSqlName("settings", {}), "x_setting_items") assert.equal(tableSqlName("shares", {}), "x_sharing_dbs") - assert.equal(tableSqlName("settings", { TABLE_PREFIX: "abc_" }), "x_setting_items") + assert.equal( + tableSqlName("settings", { TABLE_PREFIX: "abc_" }), + "x_setting_items", + ) // DDL 里应包含带前缀的复数表名 assert.ok(D1_SCHEMA.some((d) => d.includes("x_setting_items"))) @@ -289,9 +293,10 @@ test("key format: roundtrip via mock KV driver (per-table keys)", async () => { assert.deepEqual(await keyFormat.load(driver), SAMPLE_DB) }) -test("key format: load/save issue one global list each (issue #125 quota)", async () => { - // 回归防护：key 格式有 6 张表，旧实现 load/save 各逐表 list（6 次）。 - // Cloudflare 免费版 list 配额仅 1000 次/天，必须收敛为每次 1 次全局 list。 +test("key format: global index replaces list() in steady state (issue #125 quota)", async () => { + // 回归防护：Cloudflare 免费版 list 配额仅 1000 次/天。旧实现 load/save + // 各要一次全局 list，正常使用约 1000 次请求就见底。现在用索引键记录存活 + // 记录，稳态下 load/save 都不再调用 list —— 只有首次（索引尚不存在）会扫一次。 const base = createMockKvDriver() let listCalls = 0 const driver: Driver = { @@ -308,12 +313,149 @@ test("key format: load/save issue one global list each (issue #125 quota)", asyn health: base.health, } + // 首次保存：索引不存在，允许 1 次 list 建立索引 + assert.equal(await keyFormat.save(SAMPLE_DB, driver), true) + assert.equal(listCalls, 1) + + // 之后 load/save 完全走索引：list 次数不再增长 + assert.deepEqual(await keyFormat.load(driver), SAMPLE_DB) + assert.equal(listCalls, 1) assert.equal(await keyFormat.save(SAMPLE_DB, driver), true) - assert.equal(listCalls, 1) // save：清空旧数据只 1 次全局 list + assert.equal(listCalls, 1) +}) + +test("key format: legacy layout without index is migrated on first load", async () => { + // 升级前的部署只有实体键 + 初始化标记，没有索引。首次 load 必须仍能读到 + // 全部数据（回退 list），并顺手补写索引，之后不再消耗 list。 + const base = createMockKvDriver() + await keyFormat.save(SAMPLE_DB, base) + await base.delete(KEY_INDEX_KEY) + + let listCalls = 0 + const driver: Driver = { + name: "mock-kv-counting", + isAvailable: base.isAvailable, + init: base.init, + get: base.get, + put: base.put, + delete: base.delete, + list: async (prefix: string) => { + listCalls += 1 + return await base.list(prefix) + }, + health: base.health, + } + + assert.deepEqual(await keyFormat.load(driver), SAMPLE_DB) + assert.equal(listCalls, 1, "首次 load 回退 list 建立索引") + assert.deepEqual(await keyFormat.load(driver), SAMPLE_DB) + assert.equal(listCalls, 1, "索引补写后不再 list") +}) + +test("key format: corrupt index falls back to list and is rewritten", async () => { + const base = createMockKvDriver() + await keyFormat.save(SAMPLE_DB, base) + await base.put(KEY_INDEX_KEY, "{ this is not json") + + let listCalls = 0 + const driver: Driver = { + name: "mock-kv-counting", + isAvailable: base.isAvailable, + init: base.init, + get: base.get, + put: base.put, + delete: base.delete, + list: async (prefix: string) => { + listCalls += 1 + return await base.list(prefix) + }, + health: base.health, + } + + // 索引损坏绝不能当成空库：必须回退 list 并完整读出 + assert.deepEqual(await keyFormat.load(driver), SAMPLE_DB) + assert.equal(listCalls, 1) + assert.deepEqual(await keyFormat.load(driver), SAMPLE_DB) + assert.equal(listCalls, 1, "损坏索引已被重建") +}) + +test("key format: stale records are deleted using the index (no list)", async () => { + const base = createMockKvDriver() + await keyFormat.save(SAMPLE_DB, base) + + let listCalls = 0 + const driver: Driver = { + name: "mock-kv-counting", + isAvailable: base.isAvailable, + init: base.init, + get: base.get, + put: base.put, + delete: base.delete, + list: async (prefix: string) => { + listCalls += 1 + return await base.list(prefix) + }, + health: base.health, + } + + const next = { ...SAMPLE_DB, users: [] } + assert.equal(await keyFormat.save(next, driver), true) + assert.equal(listCalls, 0, "索引已存在时保存不 list") + assert.equal(await base.get("users_1"), null, "被删除的用户键已清理") + + const index = JSON.parse(String(await base.get(KEY_INDEX_KEY))) + assert.ok(!index.keys.includes("users_1")) + assert.deepEqual(await keyFormat.load(driver), next) +}) + +test("key format: records are fetched with bounded concurrency", async () => { + // 串行 get 会让 100+ 条记录的 load 在边缘网络上耗时数秒。这里断言读取确实 + // 并发进行（但不超过上限，避免撞平台子请求限制）。 + const many = { + ...SAMPLE_DB, + users: Array.from({ length: 30 }, (_, i) => ({ + id: i + 1, + username: `u${i + 1}`, + })), + } + const base = createMockKvDriver() + await keyFormat.save(many, base) + + let active = 0 + let maxActive = 0 + const driver: Driver = { + name: "mock-kv-concurrency", + isAvailable: base.isAvailable, + init: base.init, + put: base.put, + delete: base.delete, + list: base.list, + health: base.health, + get: async (key) => { + active += 1 + maxActive = Math.max(maxActive, active) + await new Promise((resolve) => setTimeout(resolve, 2)) + try { + return await base.get(key) + } finally { + active -= 1 + } + }, + } const loaded = await keyFormat.load(driver) - assert.deepEqual(loaded, SAMPLE_DB) - assert.equal(listCalls, 2) // load：再 1 次全局 list（行为与逐表一致） + // 键名是 `users_`，读取顺序与 KV list 一致（字典序），与写入顺序无关； + // 这里只校验全部记录都读到了（并发不能导致漏读/串行解析错误）。 + assert.equal(loaded.users.length, many.users.length) + assert.deepEqual( + loaded.users.map((u: any) => u.id).sort((a: number, b: number) => a - b), + many.users.map((u) => u.id), + ) + assert.ok(maxActive > 1, `expected concurrent reads, saw max=${maxActive}`) + assert.ok( + maxActive <= 12, + `read concurrency must stay bounded, saw ${maxActive}`, + ) }) test("sql format: roundtrip via mock SQL driver (columnar)", async () => { @@ -434,7 +576,9 @@ function createMockD1Binding() { return { results: [] } }, async run() { - if (/INSERT\s+OR\s+REPLACE\s+INTO\s+kv\s*\(key,\s*value\)/i.test(sql)) { + if ( + /INSERT\s+OR\s+REPLACE\s+INTO\s+kv\s*\(key,\s*value\)/i.test(sql) + ) { store.set(String(params[0]), String(params[1])) } else if (/DELETE\s+FROM\s+kv\s+WHERE\s+key\s*=\s*\?/i.test(sql)) { store.delete(String(params[0]))