Rewrite durable task lifecycle, gallery indexing and subscription publication

This commit is contained in:
root
2026-10-07 17:46:32 +08:00
parent 776eb8cf53
commit e6c7dab12e
72 changed files with 7291 additions and 551 deletions
+151
View File
@@ -0,0 +1,151 @@
#!/usr/bin/env bash
# 图片索引(cache.db)迁移:为缓存表补唯一约束与查询索引。
#
# 设计约束:
# * 默认 dry-run,只读检查;必须显式 --apply 才执行写入。
# * 必须显式 --db 指定数据库文件,不读取任何运行时配置,也不推断生产路径。
# * 写库用 sqlite3 -cmd '.bail on',任一语句出错立即以非零码退出,事务不提交。
# * 只以读写模式打开目标库(文件必须已存在,不会新建);dry-run 全用 -readonly。
# * 备份用 SQLite 自身的 .backup(online backup API),而不是 cp:
# cp 会漏掉 WAL 里尚未 checkpoint 的已提交内容。
# * 备份后立即校验备份可读且含两张缓存表,失败则中止并删除无效备份。
# * 不健康(不完整/冲突)的 gid 整组索引会被删除,由下一次在线阅读重新抓取。
# * 唯一约束建在唯一索引上:若清理后仍有重复,建索引会失败并整体回滚。
#
# 用法:
# scripts/migrate_image_index_cache.sh --db /tmp/cache-copy.db # dry-run(只读)
# scripts/migrate_image_index_cache.sh --db /tmp/cache-copy.db --apply # 执行迁移
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
SQL_FILE="$SCRIPT_DIR/migrate_image_index_cache.sql"
db_path=""
apply="no"
usage() {
sed -n '2,30p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//'
}
while [[ $# -gt 0 ]]; do
case "$1" in
--db) db_path="${2:-}"; shift 2 ;;
--apply) apply="yes"; shift ;;
-h|--help) usage; exit 0 ;;
*)
echo "未知参数: $1" >&2
exit 2 ;;
esac
done
if [[ -z "$db_path" ]]; then
echo "必须显式指定缓存库: --db <cache.db>;本脚本不会自动推断数据库位置。" >&2
exit 2
fi
if [[ ! -f "$db_path" ]]; then
echo "数据库不存在: $db_path" >&2
exit 2
fi
if [[ ! -f "$SQL_FILE" ]]; then
echo "缺少 SQL: $SQL_FILE" >&2
exit 2
fi
if ! command -v sqlite3 >/dev/null 2>&1; then
echo "缺少 sqlite3 命令行工具" >&2
exit 2
fi
if [[ "$(sqlite3 -readonly "$db_path" \
"select count(*) from sqlite_master where type='table' and name in ('gidToKey','ImageKeyCache');")" != "2" ]]; then
echo "缓存表缺失:$db_path 中找不到 gidToKey 与 ImageKeyCache。" >&2
exit 2
fi
# 体检统计,与 SQL 里实际清理的范围一致:
# bad_gids 将被整组删除的 gid 个数(去重),覆盖:
# 重复头部 / 行数不符 / 页码去重后不足 / 页码区间异常 /
# 空 imgkey / 空 key / 非法 pages
# bad_gid_rows 这些 gid 占用的头部行数(重复头部的 gid 会多于 1 行)
# duplicate_pages 重复的 (gid,page) 组数,仅用于展示,会随所属 gid 一并清理
# orphan_gids 有页 key 但无头部记录的 gid 个数
read_stats() {
sqlite3 -readonly "$db_path" "
select 'header_rows', count(*) from gidToKey;
select 'page_rows', count(*) from ImageKeyCache;
select 'duplicate_gids', count(*) from (select gid from gidToKey group by gid having count(*) > 1);
select 'duplicate_pages', count(*) from (select gid,page from ImageKeyCache group by gid,page having count(*) > 1);
select 'bad_gids', count(distinct g.gid) from gidToKey g
where g.pages is null or g.pages <= 0
or g.key is null or trim(g.key) = ''
or (select count(*) from ImageKeyCache i where i.gid = g.gid) <> g.pages
or (select count(distinct i.page) from ImageKeyCache i where i.gid = g.gid) <> g.pages
or (select min(i.page) from ImageKeyCache i where i.gid = g.gid) <> 1
or (select max(i.page) from ImageKeyCache i where i.gid = g.gid) <> g.pages
or exists (select 1 from ImageKeyCache i
where i.gid = g.gid and (i.imgkey is null or trim(i.imgkey) = ''));
select 'bad_gid_rows', coalesce(sum(c), 0) from (
select count(*) as c from gidToKey g where g.gid in (
select g2.gid from gidToKey g2 where g2.pages is null or g2.pages <= 0
or g2.key is null or trim(g2.key) = ''
or (select count(*) from ImageKeyCache i where i.gid = g2.gid) <> g2.pages
or (select count(distinct i.page) from ImageKeyCache i where i.gid = g2.gid) <> g2.pages
or (select min(i.page) from ImageKeyCache i where i.gid = g2.gid) <> 1
or (select max(i.page) from ImageKeyCache i where i.gid = g2.gid) <> g2.pages
or exists (select 1 from ImageKeyCache i
where i.gid = g2.gid and (i.imgkey is null or trim(i.imgkey) = '')))
group by g.gid);
select 'orphan_gids', count(distinct gid) from ImageKeyCache
where gid not in (select gid from gidToKey);
select 'strict_index', count(*) from sqlite_master
where type='index' and name='ux_ImageKeyCache_gid_page';
"
}
mode_label="dry-run(只读)"
if [[ "$apply" == "yes" ]]; then mode_label="apply(读写)"; fi
echo "=== 缓存库图片索引迁移($mode_label) ==="
echo "数据库: $db_path"
echo "--- 迁移前体检 ---"
read_stats | while IFS='|' read -r name value; do
printf ' %-16s %s\n' "$name" "$value"
done
if [[ "$apply" != "yes" ]]; then
echo
echo "dry-run:未做任何写入。确认无误后加 --apply 执行迁移。"
echo "提示:请先在 SQLite 副本上演练,副本应这样生成(含 WAL 内容):"
echo " sqlite3 -readonly <生产库> \".backup '<副本路径>'\""
exit 0
fi
# ---- 备份:SQLite online backup,包含 WAL 中已提交但未 checkpoint 的内容 ----
backup="$db_path.before-image-index.$(date -u +%Y%m%dT%H%M%SZ)"
if ! sqlite3 "$db_path" ".backup '$backup'"; then
echo "备份失败,未做任何修改: $backup" >&2
exit 3
fi
if [[ ! -s "$backup" ]]; then
echo "备份文件为空,未做任何修改: $backup" >&2
exit 3
fi
if [[ "$(sqlite3 -readonly "$backup" \
"select count(*) from sqlite_master where type='table' and name in ('gidToKey','ImageKeyCache');")" != "2" ]]; then
echo "备份校验失败(缺少缓存表),未做任何修改,已删除无效备份。" >&2
rm -f -- "$backup"
exit 3
fi
echo "已备份(SQLite .backup): $backup"
# ---- 执行迁移:.bail on 保证任一语句失败即退出,事务不提交 ----
sqlite3 -cmd '.bail on' "$db_path" < "$SQL_FILE"
echo "--- 迁移后体检 ---"
read_stats | while IFS='|' read -r name value; do
printf ' %-16s %s\n' "$name" "$value"
done
echo "图片索引迁移完成: $db_path"
+67
View File
@@ -0,0 +1,67 @@
-- 图片索引缓存迁移:为 cache.db 的两张索引表补唯一约束与查询索引。
--
-- 目标(对应本次重写的三条不变量):
-- 1. gid 唯一 —— 同一本画廊只能有一条 gidToKey 记录;
-- 2. (gid,page) 唯一 —— 同一页只能有一个 imgkey,同时天然满足按页查询的索引需求;
-- 3. 只保留「完整」的索引 —— 半截或冲突的 gid 整组删除,让下一次在线阅读重新抓取。
--
-- 为什么用唯一索引而不是重建表:
-- * 唯一索引对 SQLite 与唯一约束等价,都会拒绝重复写入;
-- * 不需要搬表、改列,旧版本二进制仍可读写同一份数据;
-- * 回滚只需 drop index,最坏情况也只需回到脚本自动生成的备份文件。
--
-- 幂等:所有重建索引都带 IF NOT EXISTS;清理语句只按集合删除,重复执行结果相同。
-- 安全:本脚本只做 DELETE 与 CREATE INDEX,不建表、不改列、不碰 gidToKey/ImageKeyCache
-- 之外的任何对象。必须在事务中整体提交,调用方使用 sqlite3 -bail 保证任一语句
-- 失败即终止、事务回滚。
--
-- 执行前请务必先备份数据库文件(迁移脚本 migrate_image_index_cache.sh --apply 会自动备份)。
BEGIN IMMEDIATE;
-- ---------- 1. 冲突 gid:同一 gid 有多条头部记录 ----------
-- 无法判断哪一条正确,整组移除,由下次访问重抓。
CREATE TEMP TABLE __conflict_gid AS
SELECT gid FROM gidToKey GROUP BY gid HAVING COUNT(*) > 1;
DELETE FROM ImageKeyCache WHERE gid IN (SELECT gid FROM __conflict_gid);
DELETE FROM gidToKey WHERE gid IN (SELECT gid FROM __conflict_gid);
-- ---------- 2. 半截或非法 gid:头部与页 key 不一致 ----------
-- 判定同时覆盖:页码数量不符、页码不连续(跳号/重复)、区间不是 1..pages、
-- 头部字段为空、以及存在空的 imgkey 行。
CREATE TEMP TABLE __broken_gid AS
SELECT g.gid
FROM gidToKey g
WHERE g.pages IS NULL OR g.pages <= 0
OR g.key IS NULL OR trim(g.key) = ''
OR (SELECT COUNT(*) FROM ImageKeyCache i WHERE i.gid = g.gid) <> g.pages
OR (SELECT COUNT(DISTINCT i.page) FROM ImageKeyCache i WHERE i.gid = g.gid) <> g.pages
OR (SELECT MIN(i.page) FROM ImageKeyCache i WHERE i.gid = g.gid) <> 1
OR (SELECT MAX(i.page) FROM ImageKeyCache i WHERE i.gid = g.gid) <> g.pages
OR EXISTS (SELECT 1 FROM ImageKeyCache i
WHERE i.gid = g.gid AND (i.imgkey IS NULL OR trim(i.imgkey) = ''));
DELETE FROM ImageKeyCache WHERE gid IN (SELECT gid FROM __broken_gid);
DELETE FROM gidToKey WHERE gid IN (SELECT gid FROM __broken_gid);
-- ---------- 3. 孤儿页 key:没有对应头部记录 ----------
DELETE FROM ImageKeyCache WHERE gid NOT IN (SELECT gid FROM gidToKey);
-- ---------- 4. 唯一约束 ----------
-- 到这里若仍有重复,说明上面的判定有遗漏;CREATE UNIQUE INDEX 会失败,
-- 配合 -bail 让整个事务回滚,不会留下「一半约束」的库。
CREATE UNIQUE INDEX IF NOT EXISTS ux_gidToKey_gid
ON gidToKey (gid);
CREATE UNIQUE INDEX IF NOT EXISTS ux_ImageKeyCache_gid_page
ON ImageKeyCache (gid, page);
-- 按 gid 统计页数会用到 (gid,page) 前缀,按页取 key 也走同一索引,
-- 因此不再单独建 page 索引,避免重复占用写放大。
DROP TABLE __conflict_gid;
DROP TABLE __broken_gid;
COMMIT;
+31
View File
@@ -0,0 +1,31 @@
-- 任务操作记录表(协议 v2:先持久化意图,再产生外部副作用)
--
-- 这是显式迁移脚本,不会被应用自动执行,也不对生产库运行。
-- 执行前请备份:sqlite3 LionWebsite.db ".backup LionWebsite.db.before-task-operation"
--
-- 幂等:可重复执行。
create table if not exists task_operation
(
gid integer not null
constraint task_operation_pk
primary key,
operationId varchar(64),
deleteOperationId varchar(64),
state varchar(20) not null,
lastRevision integer not null default 0,
lastReportId varchar(64),
updateTime long not null default 0
);
-- 节点上报消费只按 gid 命中主键,这里额外为「按状态找待处理意图」提供查询入口。
create index if not exists idx_task_operation_state on task_operation (state);
-- gallery.gid 目前只有 table 级主键约束(部分历史库甚至是自增主键的衍生约束),
-- 这里显式建立唯一索引,保证同一 gid 不会出现两条任务行:
-- 「同一 gid 只有一个有效意图」是 v2 生命周期成立的前提。
-- 若该语句因历史重复数据失败,请先人工确认重复行来源再处理,不要直接删除。
create unique index if not exists idx_gallery_gid_unique on gallery (gid);
-- 旧任务行没有操作记录:启动/查询时按 gid 惰性补建为 SUBMITTED(已在上游提交过),
-- 不需要在迁移里批量写入,避免把正在进行的任务判成 UNKNOWN。
+44
View File
@@ -0,0 +1,44 @@
#!/usr/bin/env python3
"""Explicit task-v2 schema migration. Default is read-only; --apply backs up first."""
import argparse
import datetime
import json
from pathlib import Path
import sqlite3
def migrate(database: Path, apply: bool = False) -> dict:
if not database.is_file():
raise ValueError("database must already exist")
with sqlite3.connect(database.resolve().as_uri() + ("?mode=rw" if apply else "?mode=ro"), uri=True) as db:
if not db.execute("pragma table_info(gallery)").fetchall():
raise ValueError("required table missing: gallery")
duplicates = db.execute("select count(*) from (select gid from gallery group by gid having count(*)>1)").fetchone()[0]
if duplicates:
raise ValueError("duplicate gallery gid; reconcile before migration")
present = bool(db.execute("pragma table_info(task_operation)").fetchall())
result = {"apply": apply, "create_operation_table": not present}
if not apply:
return result
stamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
backup = database.with_name(database.name + ".before-task-v2." + stamp)
with sqlite3.connect(backup) as copy:
db.backup(copy)
backup.chmod(0o600)
sql = Path(__file__).with_name("migrate_task_operation_v2.sql").read_text()
try:
db.executescript("BEGIN IMMEDIATE;\n" + sql + "\nCOMMIT;")
except BaseException:
db.rollback()
raise
result["backup"] = str(backup)
result["operation_rows"] = db.execute("select count(*) from task_operation").fetchone()[0]
return result
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("database", type=Path)
parser.add_argument("--apply", action="store_true")
arguments = parser.parse_args()
print(json.dumps(migrate(arguments.database, arguments.apply), ensure_ascii=False))
+64
View File
@@ -0,0 +1,64 @@
#!/usr/bin/env python3
"""Add stable user identity to subscription bindings and access records.
The default invocation only reports schema changes. --apply creates a consistent
SQLite backup before the transaction. Run against an isolated database first.
"""
import argparse
import datetime
import json
from pathlib import Path
import sqlite3
def migrate(db_path: Path, apply: bool = False) -> dict:
if not db_path.is_file():
raise ValueError("database must already exist")
uri = db_path.resolve().as_uri() + ("?mode=rw" if apply else "?mode=ro")
with sqlite3.connect(uri, uri=True) as db:
changes = []
for table in ("sub_bind", "sub_update_record"):
columns = {row[1] for row in db.execute(f'PRAGMA table_info("{table}")')}
if not columns:
raise ValueError(f"required table missing: {table}")
if "user_id" not in columns:
changes.append(f"add {table}.user_id")
duplicates = db.execute(
"select count(*) from (select username from User group by username having count(*) > 1)"
).fetchone()[0]
if duplicates:
raise ValueError("ambiguous duplicate usernames; resolve identities before migration")
result = {"apply": apply, "schema_changes": changes}
if not apply:
return result
stamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
backup = db_path.with_name(db_path.name + ".before-user-links." + stamp)
with sqlite3.connect(backup) as snapshot:
db.backup(snapshot)
backup.chmod(0o600)
db.execute("BEGIN IMMEDIATE")
try:
for change in changes:
table = change.split()[1].split(".")[0]
db.execute(f'ALTER TABLE "{table}" ADD COLUMN user_id INTEGER REFERENCES User(id)')
for table in ("sub_bind", "sub_update_record"):
db.execute(f'UPDATE "{table}" SET user_id=(SELECT id FROM User WHERE username="{table}".user) WHERE user_id IS NULL')
db.execute("CREATE UNIQUE INDEX IF NOT EXISTS User_username_uindex ON User(username)")
db.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_sub_bind_user_id ON sub_bind(user_id) WHERE user_id IS NOT NULL")
db.execute("CREATE INDEX IF NOT EXISTS idx_sub_record_user_id ON sub_update_record(user_id, id)")
db.commit()
except BaseException:
db.rollback()
raise
result["backup"] = str(backup)
result["unresolved_bindings"] = db.execute("select count(*) from sub_bind where user_id is null").fetchone()[0]
result["unresolved_records"] = db.execute("select count(*) from sub_update_record where user_id is null").fetchone()[0]
return result
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("database", type=Path)
parser.add_argument("--apply", action="store_true")
args = parser.parse_args()
print(json.dumps(migrate(args.database, args.apply), ensure_ascii=False))
+55
View File
@@ -0,0 +1,55 @@
import importlib.util
from pathlib import Path
import sqlite3
import tempfile
import unittest
spec = importlib.util.spec_from_file_location("migration", Path(__file__).with_name("migrate_user_links.py"))
migration = importlib.util.module_from_spec(spec)
spec.loader.exec_module(migration)
class MigrationTest(unittest.TestCase):
def test_dry_run_idempotent_migration_and_rename(self):
with tempfile.TemporaryDirectory() as temporary:
path = Path(temporary) / "test.db"
with sqlite3.connect(path) as db:
db.executescript("""
CREATE TABLE User(id INTEGER PRIMARY KEY, username TEXT);
CREATE TABLE sub_bind(key TEXT PRIMARY KEY, user TEXT, subscription_account_id INTEGER);
CREATE TABLE sub_update_record(id INTEGER PRIMARY KEY, user TEXT, location TEXT);
INSERT INTO User VALUES(1,'old-name');
INSERT INTO sub_bind VALUES('example-key','old-name',2);
INSERT INTO sub_update_record VALUES(10,'old-name','unknown');
INSERT INTO sub_update_record VALUES(11,'old-name','unknown');
""")
before = path.read_bytes()
self.assertEqual(2, len(migration.migrate(path)["schema_changes"]))
self.assertEqual(before, path.read_bytes())
self.assertEqual(0, migration.migrate(path, True)["unresolved_bindings"])
self.assertEqual([], migration.migrate(path, True)["schema_changes"])
with sqlite3.connect(path) as db:
db.execute("update User set username='new-name' where id=1")
self.assertEqual((1,), db.execute("select user_id from sub_bind").fetchone())
# A slow first lookup must not overwrite the second request's row.
db.execute("update sub_update_record set location='first-location' where id=10")
self.assertEqual(("unknown",), db.execute("select location from sub_update_record where id=11").fetchone())
def test_ambiguous_names_fail_without_modifying_database(self):
with tempfile.TemporaryDirectory() as temporary:
path = Path(temporary) / "test.db"
with sqlite3.connect(path) as db:
db.executescript("""
CREATE TABLE User(id INTEGER PRIMARY KEY, username TEXT);
CREATE TABLE sub_bind(user TEXT);
CREATE TABLE sub_update_record(user TEXT);
INSERT INTO User VALUES(1,'same'),(2,'same');
""")
before = path.read_bytes()
with self.assertRaises(ValueError):
migration.migrate(path, True)
self.assertEqual(before, path.read_bytes())
if __name__ == "__main__":
unittest.main()