const LOGS_DB_FILENAME: &str = "logs_2.sqlite";        // 日志库：高频追加、按线程分区裁剪，单独成文件避免锁住主库const GOALS_DB_FILENAME: &str = "goals_1.sqlite";       // 目标记账库：goal 状态与续跑延迟记录const MEMORIES_DB_FILENAME: &str = "memories_1.sqlite"; // 记忆库：stage-1/stage-2 任务表 + 提取产物const QUEUE_DB_FILENAME: &str = "queue_1.sqlite";       // 持久化用户消息队列：崩溃不丢未发送的输入const STATE_DB_FILENAME: &str = "state_5.sqlite";       // 主状态库：线程元数据等，文件名后缀是 schema 世代pub async fn open_read_write_pool(&self, path: &Path) -> Result<SqlitePool, Error> { // 所有可写库的统一打开入口    let options = SqliteConnectOptions::new()          // 构造连接选项，集中一处便于审计        .filename(path)                                 // 指定目标文件路径        .create_if_missing(true)                         // 首次启动自动建文件，老用户则直接复用        .journal_mode(SqliteJournalMode::Wal)            // WAL 模式：读者不阻塞写者，多进程并发的核心选择        .synchronous(SqliteSynchronous::Normal)          // NORMAL 级别 fsync：WAL 下兼顾持久性与吞吐的平衡点        .auto_vacuum(SqliteAutoVacuum::Incremental)      // 增量 vacuum：空闲页可小批量回收，避免整库 VACUUM 卡前台        .busy_timeout(Duration::from_secs(5))            // 锁竞争时最多等 5 秒再报 busy，多进程抢写不立即失败        .log_statements(LevelFilter::Off);               // 关掉 SQL 日志：高频查询会刷爆日志盘    SqlitePoolOptions::new()                             // 连接池构建器        .max_connections(5)                               // 每库最多 5 条连接：SQLite 单写者，更多只会加剧锁竞争        .connect_with(options)                            // 用上述选项真正建池        .await                                            // 异步等待连接就绪

let has_memories_v2 = tokio::fs::try_exists(sqlite.memories_v2_db_path()).await?; // 先探测 v2 记忆库是否已存在，决定稍后要不要走损坏恢复let pool = match sqlite.open_state_db(&state_migrator, telemetry_override).await { // 打开主状态库并跑迁移（open+migrate 两阶段都带遥测）    Ok(db) => Arc::new(db),                                                        // 成功：Arc 包装，后面多个 store 共享同一池    Err(err) => {                                                                  // 失败分支：fail fast        warn!("failed to open state db at {}: {err}", state_path.display());       // 日志带完整路径，方便定位是哪个文件坏了        return Err(err);                                                            // 直接返回错误，不留半初始化状态    }};                                                                                 // （纯结构行）let logs_pool = match sqlite.open_logs_db(&logs_migrator, telemetry_override).await { // 第二个库：日志    Ok(db) => Arc::new(db),                                                        // 成功路径同上    Err(err) => {                                                                  // 失败必须回滚已打开的池        warn!("failed to open logs db at {}: {err}", logs_path.display());         // 同样带路径告警        close_sqlite_pools(&[pool.as_ref()]).await;                                // 级联关闭：主库先关，避免泄漏连接        return Err(err);                                                            // 返回错误终止初始化    }};                                                                                 // （纯结构行）let runtime = Arc::new(Self {                 // 组装 StateRuntime，整体用 Arc 共享给上层    thread_goals: GoalStore::new(Arc::clone(&goals_pool)),      // goal store 独占自己的池    memories: MemoryStore::new(Arc::clone(&memories_pool), Arc::clone(&pool)), // 记忆 store 同时持有记忆库和主库：水位线要跨库对账    memories_v2: Arc::new(tokio::sync::OnceCell::new()),        // v2 记忆懒加载：只有文件已存在才真正打开，不给新用户凭空建空库    thread_queue: SqliteQueueStore::new(queue_pool),            // 持久化队列 store    pool, logs_pool, sqlite, default_provider,                  // 池与配置直接移入结构体    thread_updated_at_millis: Arc::new(AtomicI64::new(thread_updated_at_millis)), // 启动时缓存 MAX(updated_at_ms)，后续"最近活跃"排序免查库    thread_recency_at_millis: Arc::new(AtomicI64::new(thread_recency_at_millis)) // 同理缓存 recency 水位，原子变量供并发读});

fn runtime_migrator(base: &'static Migrator) -> Migrator { // 给每个内嵌 migrator 套一层运行时包装    Migrator { // 手工构造包装后的 Migrator，逐字段复制原配置        migrations: Cow::Borrowed(base.migrations.as_ref()), // 复用编译期嵌入的迁移集，零拷贝借用        ignore_missing: true,                                 // 关键开关：容忍"库比我新"——旧版二进制能打开被新版并行迁移过的库        locking: base.locking,                                // 保持原 migrator 的锁行为不变        no_tx: base.no_tx,                                    // 保持原事务策略（是否每条迁移独立事务）        table_name: base.table_name.clone(),                  // _sqlx_migrations 记录表名沿用        create_schemas: base.create_schemas.clone(),          // schema 创建策略沿用    }}

SELECT thread_id FROM logs WHERE thread_id IN (/* 本批涉及的线程 */)GROUP BY thread_id HAVING SUM(estimated_bytes) > /* 10MiB */ OR COUNT(*) > /* 1000 */ -- 廉价预检：只挑真正超预算的线程，避免对全表跑窗口函数DELETE FROM logs                                   -- 只删超预算的行，其余原样保留WHERE id IN (                                      -- 子查询先算出要删的 id 集合    SELECT id    FROM (        SELECT id,                                 -- 行主键，外层用它定位删除目标            SUM(estimated_bytes) OVER (            -- 窗口累计字节：从最新一行往旧累加                PARTITION BY thread_id             // 预算按线程独立计算，互不挤占                ORDER BY ts DESC, ts_nanos DESC, id DESC   // 时间倒序；纳秒和主键做平局裁决，保证顺序稳定可复现            ) AS cumulative_bytes,                 -- 每行的"自最新累计字节数"            ROW_NUMBER() OVER (                    -- 同分区内的行号                PARTITION BY thread_id             // 与字节累计同一分区维度                ORDER BY ts DESC, ts_nanos DESC, id DESC   // 排序必须和累计一致，否则行数上限会误伤            ) AS row_number                        // 用于 1000 行的硬上限判断        FROM logs        WHERE thread_id IN (/* 预检出的超预算线程 */)      -- 只扫描受影响分区，不碰其他线程的数据    )    WHERE cumulative_bytes > /* 10MiB */           // 累计字节超出预算的行 → 删（即最旧的那部分）       OR row_number > /* 1000 */                  // 或者行数超上限 → 删，双保险

let row = sqlx::query( // 单条语句完成"限长检查+定序插入+回读新行"，原子性由 SQLite 写锁保证    "INSERT INTO queued_items (id, thread_id, payload_json, queue_order, created_at_ms, updated_at_ms) -- 六列：身份、归属线程、负载、序号、两个时间戳     SELECT ?, ?, ?, // SELECT 形式而非 VALUES，才能内嵌子查询算序号和限长条件            COALESCE((SELECT MAX(queue_order) FROM queued_items WHERE thread_id = ?), -1) + 1, -- 新序号=当前最大+1；空队列时从 0 开始，COALESCE 兜底 NULL            ?, ? // created_at / updated_at：初始两值相同，后续 update/reorder 只刷 updated     WHERE (SELECT COUNT(*) FROM queued_items WHERE thread_id = ?) < ?   // 限长检查内嵌在 INSERT 里：计数<上限才真正插入，单语句原子完成     RETURNING id, thread_id, payload_json",                          // 直接返回新行内容，省掉一次回查).bind(Uuid::now_v7().to_string())   // UUIDv7：时间有序，天然可排序、无中心发号器.bind(thread_id.to_string())        // 归属哪个线程的队列（按线程隔离）.bind(payload_json)                 // 用户消息负载（JSON 序列化后的原文）.bind(thread_id.to_string())        // MAX(queue_order) 子查询用的线程过滤条件.bind(now_ms)                       // created_at：入队时刻（毫秒时间戳）.bind(now_ms)                       // updated_at：初始与创建一致，后续 update/reorder 会刷新.bind(thread_id.to_string())        // COUNT(*) 限长检查的线程过滤条件.bind(i64::try_from(MAX_QUEUE_ITEMS)?)   // 上限常量 100，try_from 防溢出.fetch_one(self.pool.as_ref())      // 期望恰好一行：超限则无行返回、fetch_one 报错——失败语义清晰.await?;

let ownership_token = Uuid::new_v4().to_string(); // 随机所有权令牌：后续 mark done/failed 必须出示它，防止过期 worker 覆盖新状态let mut tx = self.pool.begin_with("BEGIN IMMEDIATE").await?; // 立即拿写锁再开事务：检查-插入必须原子，否则两个进程会同时抢到同一任务// —— 幂等短路 1：该线程已有更新或同版本的产物，直接跳过if existing_source_updated_at >= source_updated_at {          // 旧输入不覆盖新输出    return Ok(Stage1JobClaimOutcome::SkippedUpToDate);        // 明确返回"已最新"，调用方无需重试}// —— 幂等短路 2：上次成功处理的水位线已经 ≥ 本次输入水位if last_success_watermark.is_some_and(|w| w >= source_updated_at) { // 同一份输入不重复提取（崩溃重放场景）    return Ok(Stage1JobClaimOutcome::SkippedUpToDate);        // 同上，幂等返回}// —— 核心 upsert：INSERT ... ON CONFLICT(kind, job_key) DO UPDATE SET ... WHEREWHERE // upsert 的接管条件全部写在这里，逐条 AND 收窄可抢范围    (jobs.status != 'running' OR jobs.lease_until IS NULL OR jobs.lease_until <= excluded.started_at)   // 只接管"非运行中或租约已过期"的行：绝不抢活着的 worker    AND (jobs.retry_at IS NULL OR jobs.retry_at <= excluded.started_at OR excluded.input_watermark > COALESCE(jobs.input_watermark, -1))  // 尊重退避窗口；输入更新时允许提前接管    AND (jobs.retry_remaining > 0 OR excluded.input_watermark > COALESCE(jobs.input_watermark, -1))     // 重试次数耗尽后，只有"输入真的变新了"才值得再跑    AND (SELECT COUNT(*) FROM jobs AS running_jobs WHERE running_jobs.kind = excluded.kind          AND running_jobs.status = 'running' AND running_jobs.lease_until > excluded.started_at // 只统计"租约还活着"的任务，过期租约不算占位          AND running_jobs.job_key != excluded.job_key) < ?   // 全局并发上限：存活 stage-1 任务数必须小于 max_running，在 SQL 里强制

fn sqlite_database_code_is_corruption(code: Cow<'_, str>) -> bool { // 按 SQLite 错误码判定"损坏"    matches!(code.as_ref().to_ascii_lowercase().as_str(), // 错误码转小写后做集合匹配，大小写不敏感        "11" | "26" | "sqlite_corrupt" | "sqlite_notadb")  // 11=SQLITE_CORRUPT、26=SQLITE_NOTADB：两个经典损坏码，大小写不敏感匹配}pub fn sqlite_error_detail_is_corruption(detail: &str) -> bool { // 错误信息兜底判定（有些驱动只给文本不给码）    let detail = detail.to_ascii_lowercase();                    // 统一小写再比较    detail.contains("database disk image is malformed")         // 页级损坏的标准报错文案        || detail.contains("file is not a database")            // 文件头不是 SQLite magic number        || detail.contains("(code: 11)")                         // sqlx 把码拼在消息尾部的格式        || ...                                                   // 其余同族文案略}fn sqlite_paths(db_path: &Path) -> Vec<PathBuf> { // 枚举"一个库"的全部物理文件    let mut wal_path = db_path.as_os_str().to_os_string(); // 在路径末尾拼 -wal 得到 WAL 文件路径    wal_path.push("-wal");                          // WAL 日志：必须和主库一起搬，否则新库+旧WAL=二次损坏    let mut shm_path = db_path.as_os_str().to_os_string(); // 同理拼出共享内存索引文件路径    shm_path.push("-shm");                          // 共享内存索引文件：同理，是 WAL 模式的伴生文件    vec![db_path.to_path_buf(), PathBuf::from(wal_path), PathBuf::from(shm_path)] // 主库+两个副车三件套整体迁移}

启动路径（StateRuntime::init_inner）

▼

▼

▼

▼