From 18afedcb7168545448009eda3a035506e5048897 Mon Sep 17 00:00:00 2001 From: knqiufan Date: Fri, 23 Oct 2026 11:26:41 +0800 Subject: [PATCH 1/2] perf(archive): batch target compatibility checks --- docs/en/docs/operate/portable-archive.md | 47 +++ docs/zh/docs/operate/portable-archive.md | 38 +++ scripts/benchmark_archive_validation.py | 164 +++++++++++ .../builtin/portability/bundle.py | 95 +++++-- tests/builtin/portability/test_bundle.py | 268 +++++++++++++++++- tests/e2e/test_portable_archive.py | 70 +++++ 6 files changed, 662 insertions(+), 20 deletions(-) create mode 100644 scripts/benchmark_archive_validation.py diff --git a/docs/en/docs/operate/portable-archive.md b/docs/en/docs/operate/portable-archive.md index d53274f32..03329bea4 100644 --- a/docs/en/docs/operate/portable-archive.md +++ b/docs/en/docs/operate/portable-archive.md @@ -237,6 +237,53 @@ Logical export is not a substitute for a crash-consistent deployment backup: ## Current boundaries +Target compatibility checks group records by type, with at most 100 records per SQL statement, 900 bound +parameters across the pending groups, and 1 MiB of encoded lookup keys and expected digests. Wider composite +keys can produce smaller batches. Pending groups retain keys and digests, not record bodies. The result cursor +fetches one target row at a time; peak memory still includes a single decoded archive record, a target row, +canonical encoding buffers, and database-driver overhead. A large existing target body therefore does not +multiply by the batch size. This is a working-set bound, not a fixed process-memory limit. + +Each batch preserves the target database's key comparison rules and checks both logical identities and +alternate physical unique keys. Any differing match makes that archive record a conflict, counted once. +Nullable unique keys retain SQL null semantics. Dry-run results are not cached for a later restore: restore +checks the current target again, and write-time conflicts still roll back the transaction. Publication-chain +and combined-tag-set checks run after the record checks. + +For the representative archive containing one Scope, one journal head, and N small Sources, the compatibility +query budget against initialized tables is `ceil(N / 100) + 2`: 3, 12, and 52 statements for 100, 1,000, and +5,000 Sources. Missing tables require no record compatibility queries. Schema introspection, local dependency +index queries, and publication/tag integrity queries are separate from this budget. + +Run `uv run python scripts/benchmark_archive_validation.py --sources 100 1000 5000 --runs 3` to measure full +validation times and query counts on disposable file-backed SQLite targets. Fixture generation and export +are excluded. The separately reported restore duration includes revalidation, staging, authoritative writes, +and the receipt, with no projection rebuilder configured; it is not an isolated write or search-rebuild timing. +Compare the same script and environment across checkouts. Fewer SQL executions do not imply the same reduction +in complete validation time: archive parsing, dependency checks, decoding, and digests remain per-record work. +Embedded SeekDB is local; remote OceanBase costs depend on the deployment topology and need real-backend measurements. + +On 2026-10-23, the script was run against baseline `4a21da08` (PR #1561 plus current master) and the batched +implementation on Windows 11 build 26200, CPython 3.12.13, SQLite 3.50.4, with default SQLiteProfile settings, +256-byte Source content, and no injected network delay. Each cell below is the median of three full validations; +query counts were identical in all repetitions and exclude schema introspection. + +| Records | Target | Queries: baseline → batched | Seconds: baseline → batched | +| ---: | --- | ---: | ---: | +| 102 | Empty | 102 → 3 | 0.3469 → 0.2828 | +| 102 | Identical | 102 → 3 | 0.3755 → 0.3039 | +| 1,002 | Empty | 1,002 → 12 | 2.6401 → 1.9954 | +| 1,002 | Identical | 1,002 → 12 | 3.2584 → 2.3064 | +| 5,002 | Empty | 5,002 → 52 | 16.3445 → 9.1713 | +| 5,002 | Identical | 5,002 → 52 | 16.9955 → 10.7552 | +| 5,002 | Tables absent | 0 → 0 | 7.9935 → 7.5444 | + +The missing-table control also varies despite executing no compatibility queries; these are local diagnostic +measurements, not a throughput guarantee. One 5,002-record restore without projections took 17.6685 seconds on +the baseline and 10.5197 seconds with batching, including internal validation and staging. Projection rebuilding +was not measured. Native SeekDB and OceanBase were not executed in this environment; their parameterized archive +acceptance tests require actual backend runs before those configurations are qualified for this optimization. + Export streams database rows to a temporary NDJSON file. Validation stores record identities and Candidate versions in a temporary on-disk index to verify each head against its exact version. Restore replays dependency levels as streams, so aggregate archive payloads are not retained in diff --git a/docs/zh/docs/operate/portable-archive.md b/docs/zh/docs/operate/portable-archive.md index 77eeecf32..c4eeeec3d 100644 --- a/docs/zh/docs/operate/portable-archive.md +++ b/docs/zh/docs/operate/portable-archive.md @@ -205,6 +205,44 @@ reader 会拒绝这些记录不完整的 schema version `1` 表示，而不是 ## 当前边界 +目标兼容性检查按记录类型分组，每条 SQL 最多检查 100 条记录。待处理分组总计最多使用 900 个绑定参数， +编码后的查询键和预期摘要总计最多 1 MiB；较宽的复合键会使批次提前拆分。缓存只保留键和摘要，不保留记录正文。 +结果游标每次读取一条目标记录，内存峰值仍包含单条已解码归档记录、单条目标记录、规范编码缓冲和数据库驱动开销。 +因此，目标中的大正文不会按批次条数成倍驻留内存。这是工作集边界，不是固定的进程内存上限。 + +批量查询保留目标数据库的键比较规则，同时检查逻辑身份和其他物理唯一键。任何命中项内容不同，都将该归档记录 +计为一次冲突；可空唯一键继续遵循 SQL 的 null 语义。dry-run 结果不会缓存给后续恢复使用：restore 会重新检查 +当前目标，写入阶段的冲突仍会回滚整个事务。记录检查完成后，继续执行发布链和合并标签集合的完整性检查。 + +对于一个 Scope、一个日志头和 N 条小 Source 组成的代表性归档，目标表已初始化时，兼容性查询预算为 +`ceil(N / 100) + 2`：100、1,000、5,000 条 Source 分别最多执行 3、12、52 条查询。 +目标表不存在时，不执行对应记录的兼容性查询。表结构探测、本地依赖索引查询、发布链和标签完整性查询单独计算。 + +运行 `uv run python scripts/benchmark_archive_validation.py --sources 100 1000 5000 --runs 3`， +可在临时文件型 SQLite 数据库上测量完整校验耗时和查询次数。计时不包含数据准备和导出。 +单独报告的恢复耗时包含重新校验、记录暂存、权威数据写入和回执，未配置投影重建器，不能解释成纯写入或搜索重建耗时。 +跨提交比较时使用同一脚本和环境。SQL 执行次数减少不代表完整校验耗时同比下降，归档解析、依赖检查、解码和摘要计算 +仍需要逐条完成。嵌入式 SeekDB 在本地运行；远程 OceanBase 的成本与部署拓扑有关，需要在真实后端上测量。 + +2026-10-23 在 Windows 11 build 26200、CPython 3.12.13、SQLite 3.50.4 上，对基线 `4a21da08` +（PR #1561 与当前 master 的合并代码）和批处理实现运行上述脚本。使用默认 SQLiteProfile 配置，Source 正文为 +256 字节，未注入网络延迟。以下耗时为三次完整校验的中位数，每次查询计数一致，不包含表结构探测。 + +| 记录数 | 目标状态 | 查询数：基线 → 批处理 | 秒：基线 → 批处理 | +| ---: | --- | ---: | ---: | +| 102 | 空表 | 102 → 3 | 0.3469 → 0.2828 | +| 102 | 已有相同数据 | 102 → 3 | 0.3755 → 0.3039 | +| 1,002 | 空表 | 1,002 → 12 | 2.6401 → 1.9954 | +| 1,002 | 已有相同数据 | 1,002 → 12 | 3.2584 → 2.3064 | +| 5,002 | 空表 | 5,002 → 52 | 16.3445 → 9.1713 | +| 5,002 | 已有相同数据 | 5,002 → 52 | 16.9955 → 10.7552 | +| 5,002 | 表不存在 | 0 → 0 | 7.9935 → 7.5444 | + +不执行兼容性查询的缺表对照也有耗时波动，因此这些本地诊断数据不能作为吞吐量保证。 +5,002 条记录的一次不含投影重建的恢复，基线耗时 17.6685 秒，批处理耗时 10.5197 秒，均包含内部校验和暂存。 +本次未测量投影重建，也未在该环境执行原生 SeekDB 或 OceanBase。针对这些配置，仍需运行参数化归档验收测试， +取得真实后端执行结果后才能确认本优化的适用性。 + 导出会把数据库 row 流式写到临时 NDJSON 文件。验证在临时磁盘 index 中保存 record identity 和 Candidate version， 用于校验每个 head 对应的精确版本。恢复则按依赖层流式 回放，因此归档的聚合 payload 不会常驻进程内存；Scope metadata 由 manifest scope list 约束。命令不是远程 Server diff --git a/scripts/benchmark_archive_validation.py b/scripts/benchmark_archive_validation.py new file mode 100644 index 000000000..f9ae4d417 --- /dev/null +++ b/scripts/benchmark_archive_validation.py @@ -0,0 +1,164 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Measure public archive validation on disposable, file-backed SQLite targets. + +Run the same script against each checkout with its src directory on PYTHONPATH. +Fixture creation/export and restore population are outside validation timings. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import platform +import sqlite3 +import statistics +import tempfile +import time +from pathlib import Path + +from sqlalchemy import event, insert + +from powercontext.builtin.persistence.sources import SourceRepository +from powercontext.builtin.persistence.sqlite import SQLiteConfig, SQLiteProfile +from powercontext.builtin.persistence.tables import ( + BUILTIN_TABLES, + SCOPES_TABLE, + SOURCE_JOURNAL_HEADS_TABLE, + SOURCES_TABLE, +) +from powercontext.builtin.portability import PortableBundleService +from powercontext.builtin.sources import BUILTIN_SOURCE_REGISTRY, ContentSource +from powercontext.paths import sqlite_url +from powercontext.sources import SourceMaterialization + + +async def measure(directory: Path, count: int, runs: int) -> None: + archive = directory / "sources.pcb" + + async def authorize(_scopes: tuple[str, ...]) -> None: + pass + + async with SQLiteProfile.open( + SQLiteConfig(url=sqlite_url(directory / "source.db")), tables=BUILTIN_TABLES + ) as source: + async with source.database.transaction() as connection: + await connection.execute( + insert(SCOPES_TABLE).values( + scope_id="benchmark", + title="", + summary="", + scope_id_search="benchmark", + title_search="", + summary_search="", + version=1, + ) + ) + await connection.execute(insert(SOURCE_JOURNAL_HEADS_TABLE).values(scope_id="benchmark", position=count)) + for start in range(0, count, 100): + rows = [] + for number in range(start, min(start + 100, count)): + identity = f"source-{number:06}" + payload = ( + ContentSource(name=identity, materialization=SourceMaterialization.CAPTURED, content="x" * 256) + .model_dump_json() + .encode() + ) + rows.append({ + "scope_id": "benchmark", + "source_type": "content", + "source_id": identity, + "payload": payload, + "journal_position": number + 1, + }) + await connection.execute(insert(SOURCES_TABLE), rows) + await PortableBundleService(source.database).export(["benchmark"], archive, authorize=authorize) + + for state in ("empty", "identical", "absent"): + async with SQLiteProfile.open( + SQLiteConfig(url=sqlite_url(directory / f"{state}.db")), + tables=() if state == "absent" else BUILTIN_TABLES, + ) as target: + service = PortableBundleService( + target.database, source_decoder=SourceRepository(BUILTIN_SOURCE_REGISTRY).decode_source + ) + restore_seconds = None + if state == "identical": + started = time.perf_counter() + await service.restore(archive) + restore_seconds = time.perf_counter() - started + executions = [] + + def observe(_connection, _cursor, statement, parameters, _context, _many, executions=executions): + if statement.startswith("SELECT") and any( + f"FROM {table.name}" in statement + for table in (SCOPES_TABLE, SOURCE_JOURNAL_HEADS_TABLE, SOURCES_TABLE) + ): + executions.append(len(parameters)) + + event.listen(target.database.engine.sync_engine, "before_cursor_execute", observe) + durations, queries = [], [] + max_parameters = 0 + for _ in range(runs): + executions.clear() + started = time.perf_counter() + result = await service.validate(archive) + durations.append(round(time.perf_counter() - started, 4)) + if ( + not result.compatible + or result.conflicts + or result.already_present != (count + 2 if state == "identical" else 0) + ): + message = "unexpected validation classification" + raise RuntimeError(message) + queries.append(len(executions)) + max_parameters = max(max_parameters, max(executions, default=0)) + print( + json.dumps({ + "records": count + 2, + "state": state, + "queries": queries, + "seconds": durations, + "median_seconds": statistics.median(durations), + "max_parameters": max_parameters, + "restore_without_projections_seconds": restore_seconds, + }), + flush=True, + ) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--sources", type=int, nargs="+", default=[100, 1_000, 5_000]) + parser.add_argument("--runs", type=int, default=3) + args = parser.parse_args() + if args.runs < 1 or any(count < 1 for count in args.sources): + parser.error("source counts and runs must be positive") + print( + json.dumps({ + "python": platform.python_version(), + "sqlite": sqlite3.sqlite_version, + "platform": platform.platform(), + }), + flush=True, + ) + for count in args.sources: + with tempfile.TemporaryDirectory(prefix="pc-archive-benchmark-") as temporary: + asyncio.run(measure(Path(temporary), count, args.runs)) + + +if __name__ == "__main__": + main() diff --git a/src/powercontext/builtin/portability/bundle.py b/src/powercontext/builtin/portability/bundle.py index 54bb08b82..873aff7ac 100644 --- a/src/powercontext/builtin/portability/bundle.py +++ b/src/powercontext/builtin/portability/bundle.py @@ -54,9 +54,11 @@ UniqueConstraint, and_, insert, + literal_column, or_, select, tuple_, + union_all, update, ) from sqlalchemy import inspect as sqlalchemy_inspect @@ -126,6 +128,9 @@ _MAX_RECORDS = 2_000_000 _MAX_LINE_BYTES = 8 * 1024 * 1024 _MAX_MANIFEST_BYTES = 256 * 1024 +_TARGET_BATCH_RECORDS = 100 +_TARGET_BATCH_PARAMETERS = 900 +_TARGET_BATCH_KEY_BYTES = 1024 * 1024 _HEX_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") _ABSOLUTE_HOST_PATH = re.compile(r"^(?:/|file:/+|[A-Za-z]:[\\/]|\\\\)") _SENSITIVE_KEYS = frozenset({ @@ -263,6 +268,14 @@ class _Record: digest: str +@dataclass(frozen=True, slots=True) +class _TargetLookup: + """Only keys and the expected digest survive between streamed archive rows.""" + + identities: tuple[Mapping[str, Any], ...] + digest: str + + @dataclass(frozen=True, slots=True) class _RecordSpec: """Stable logical fields and their current relational persistence adapter.""" @@ -764,11 +777,25 @@ async def _target_compatibility( } ) checked_types: set[RecordType] = set() - for processed, record in enumerate(_iter_records(source), start=1): + pending: dict[RecordType, list[_TargetLookup]] = {} + pending_parameters = pending_bytes = processed = 0 + + async def flush() -> None: + nonlocal already_present, conflicts, processed + for kind, lookups in pending.items(): + present, different = await _check_target_batch(connection, kind, lookups) + already_present += present + conflicts += different + processed += len(lookups) + _report_progress(progress, "target", processed, total) + pending.clear() + + for record in _iter_records(source): self._validate_target_content(record, supported_source_types=supported_source_types) spec = _SPECS[record.record_type] table = spec.table if table.name not in existing_tables: + processed += 1 _report_progress(progress, "target", processed, total) continue if record.record_type not in checked_types: @@ -783,23 +810,19 @@ async def _target_compatibility( ) checked_types.add(record.record_type) keys = (record.identity, *_unique_identities(record)) - where = or_( - *( - and_(*(table.c[key] == _database_value(value, table.c[key]) for key, value in identity.items())) - for identity in keys - ) - ) - statement = select(*(table.c[field] for field in (*spec.identity, *spec.payload))).where(where) - rows = (await connection.execute(statement)).mappings().all() - if rows: - if all( - _record_from_row(record.record_type, cast(Mapping[str, Any], row)).digest == record.digest - for row in rows - ): - already_present += 1 - else: - conflicts += 1 - _report_progress(progress, "target", processed, total) + parameters = sum(len(identity) for identity in keys) + size = len(_canonical_json(keys)) + len(record.digest) + if pending and ( + len(pending.get(record.record_type, ())) >= _TARGET_BATCH_RECORDS + or pending_parameters + parameters > _TARGET_BATCH_PARAMETERS + or pending_bytes + size > _TARGET_BATCH_KEY_BYTES + ): + await flush() + pending_parameters = pending_bytes = 0 + pending.setdefault(record.record_type, []).append(_TargetLookup(keys, record.digest)) + pending_parameters += parameters + pending_bytes += size + await flush() if ( conflicts == 0 and ARTIFACT_PUBLICATIONS_TABLE.name in existing_tables @@ -937,6 +960,42 @@ async def _stream_export( return count, records_by_type, digest.value() +async def _check_target_batch( + connection: AsyncConnection, kind: RecordType, lookups: list[_TargetLookup] +) -> tuple[int, int]: + spec = _SPECS[kind] + table = spec.table + statements = [] + for ordinal, lookup in enumerate(lookups): + predicate = or_( + *( + and_(*(table.c[key] == _database_value(value, table.c[key]) for key, value in identity.items())) + for identity in lookup.identities + ) + ) + # Carry the archive ordinal through SQL so matching retains the target's + # comparison semantics (including collation), not Python key equality. + statements.append( + select( + literal_column(str(ordinal)).label("_archive_ordinal"), + *(table.c[field] for field in (*spec.identity, *spec.payload)), + ).where(predicate) + ) + present = [False] * len(lookups) + different = [False] * len(lookups) + # execute() buffers rows in some drivers even without .all(). Bound the + # cursor to one target row: its payload can dwarf the incoming archive row. + async with connection.stream(union_all(*statements), execution_options={"yield_per": 1}) as rows: + async for row in rows.mappings(): + ordinal = int(row["_archive_ordinal"]) + present[ordinal] = True + if not different[ordinal]: + different[ordinal] = ( + _record_from_row(kind, cast(Mapping[str, Any], row)).digest != lookups[ordinal].digest + ) + return sum(found and not conflict for found, conflict in zip(present, different, strict=True)), sum(different) + + async def _export_rows( connection: AsyncConnection, kind: RecordType, scopes: tuple[str, ...] ) -> AsyncIterator[Mapping[str, Any]]: diff --git a/tests/builtin/portability/test_bundle.py b/tests/builtin/portability/test_bundle.py index 41f7f6de8..e0742f501 100644 --- a/tests/builtin/portability/test_bundle.py +++ b/tests/builtin/portability/test_bundle.py @@ -23,12 +23,13 @@ import sqlite3 import tracemalloc import zipfile +from dataclasses import replace from datetime import UTC, datetime from pathlib import Path -from typing import cast +from typing import Any, cast import pytest -from sqlalchemy import insert, select +from sqlalchemy import MetaData, UniqueConstraint, event, insert, select, update from powercontext.builtin.artifacts.skill import capture_skill_directory from powercontext.builtin.persistence.database import AsyncDatabase @@ -87,6 +88,269 @@ async def _insert_scope(connection, scope_id: str = "project:one") -> None: ) +async def _source_archive(tmp_path: Path, count: int = 205) -> Path: + archive = tmp_path / "sources.pcb" + async with SQLiteProfile.open(SQLiteConfig(), tables=BUILTIN_TABLES) as source: + async with source.database.transaction() as connection: + await _insert_scope(connection) + await connection.execute(insert(SOURCE_JOURNAL_HEADS_TABLE).values(scope_id="project:one", position=count)) + await connection.execute(insert(SOURCES_TABLE), [_source_row(number) for number in range(1, count + 1)]) + await PortableBundleService(source.database).export(["project:one"], archive, authorize=_authorize_export) + return archive + + +def _source_row(number: int) -> dict[str, Any]: + source_id = f"source-{number:05}" + return { + "scope_id": "project:one", + "source_type": "content", + "source_id": source_id, + "payload": _content_payload(source_id, "x" * 256), + "journal_position": number, + } + + +@pytest.mark.parametrize("count", [100, 1_000]) +@pytest.mark.parametrize("state", ["absent", "empty", "identical", "partial", "conflicts"]) +def test_validation_batches_target_queries_without_changing_classification( + tmp_path: Path, count: int, state: str +) -> None: + async def scenario() -> None: + archive = await _source_archive(tmp_path, count) + tables = () if state == "absent" else BUILTIN_TABLES + async with SQLiteProfile.open(SQLiteConfig(), tables=tables) as target: + service = PortableBundleService(target.database, source_decoder=_SOURCES.decode_source) + present = conflicts = 0 + if state == "identical": + await service.restore(archive) + present = count + 2 + elif state in {"partial", "conflicts"}: + async with target.database.transaction() as connection: + await _insert_scope(connection) + await connection.execute( + insert(SOURCE_JOURNAL_HEADS_TABLE).values(scope_id="project:one", position=count) + ) + rows = [_source_row(number) for number in range(1, count // 2 + 1)] + if state == "conflicts": + # A primary match and an alternate match may be different + # rows. Each archive record still contributes one conflict. + rows[0]["journal_position"], rows[-1]["journal_position"] = ( + rows[-1]["journal_position"], + rows[0]["journal_position"], + ) + rows[1]["payload"] = _content_payload(rows[1]["source_id"], "changed") + rows[2]["source_id"] = "another-source" + rows[2]["payload"] = _content_payload("another-source", "different identity") + conflicts = 4 + await connection.execute(insert(SOURCES_TABLE), rows) + present = count // 2 + 2 - conflicts + statements = [] + + def observe(_connection, _cursor, statement, parameters, _context, _many): + if statement.startswith("SELECT") and any( + f"FROM {table.name}" in statement + for table in (SCOPES_TABLE, SOURCE_JOURNAL_HEADS_TABLE, SOURCES_TABLE) + ): + statements.append(len(parameters)) + + event.listen(target.database.engine.sync_engine, "before_cursor_execute", observe) + validation = await service.validate(archive) + assert validation.already_present == present + assert validation.conflicts == conflicts + assert validation.compatible is (conflicts == 0) + assert len(statements) <= (0 if state == "absent" else (count + 99) // 100 + 2) + assert max(statements, default=0) <= 900 + if conflicts: + with pytest.raises(BundleConflictError): + await service.restore(archive) + async with target.database.transaction() as connection: + assert len((await connection.execute(select(SOURCES_TABLE))).all()) == count // 2 + + asyncio.run(scenario()) + + +def test_restore_rechecks_target_after_dry_run_across_batch_boundaries(tmp_path: Path) -> None: + async def scenario() -> None: + archive = await _source_archive(tmp_path) + async with SQLiteProfile.open(SQLiteConfig(), tables=BUILTIN_TABLES) as target: + service = PortableBundleService(target.database, source_decoder=_SOURCES.decode_source) + assert (await service.validate(archive)).compatible + async with target.database.transaction() as connection: + left, right = _source_row(100), _source_row(101) + left["journal_position"], right["journal_position"] = 101, 100 + await connection.execute(insert(SOURCES_TABLE), [left, right]) + validation = await service.validate(archive) + assert validation.conflicts == 2 + assert validation.already_present == 0 + with pytest.raises(BundleConflictError): + await service.restore(archive) + async with target.database.transaction() as connection: + assert (await connection.execute(select(SCOPES_TABLE))).all() == [] + assert len((await connection.execute(select(SOURCES_TABLE))).all()) == 2 + + asyncio.run(scenario()) + + +def test_batch_matching_retains_database_collation_semantics(tmp_path: Path) -> None: + async def scenario() -> None: + archive = await _source_archive(tmp_path, 1) + async with SQLiteProfile.open(SQLiteConfig(), tables=()) as target: + async with target.database.transaction() as connection: + # Compatibility checks must respect the existing target's SQL + # equality, even if its identity spelling differs from the archive. + await connection.exec_driver_sql( + "CREATE TABLE pc_sources (scope_id TEXT, source_type TEXT, source_id TEXT COLLATE NOCASE, " + "payload BLOB, journal_position INTEGER, PRIMARY KEY(scope_id, source_type, source_id), " + "UNIQUE(scope_id, journal_position))" + ) + row = _source_row(1) + row.update(source_id="SOURCE-00001", journal_position=10) + await connection.execute(insert(SOURCES_TABLE), row) + validation = await PortableBundleService(target.database, source_decoder=_SOURCES.decode_source).validate( + archive + ) + assert validation.conflicts == 1 + assert not validation.compatible + + asyncio.run(scenario()) + + +def test_batch_validation_streams_large_target_payloads(tmp_path: Path) -> None: + async def scenario() -> None: + archive = await _source_archive(tmp_path, 100) + config = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'large-target.db'}") + async with SQLiteProfile.open(config, tables=BUILTIN_TABLES) as target: + service = PortableBundleService(target.database, source_decoder=_SOURCES.decode_source) + await service.restore(archive) + async with target.database.transaction() as connection: + for number in range(1, 101): + row = _source_row(number) + await connection.execute( + update(SOURCES_TABLE) + .where(SOURCES_TABLE.c.source_id == row["source_id"]) + .values(payload=_content_payload(row["source_id"], "x" * (512 * 1024))) + ) + tracemalloc.start() + try: + validation = await service.validate(archive) + _, peak = tracemalloc.get_traced_memory() + finally: + tracemalloc.stop() + assert validation.conflicts == 100 + assert validation.already_present == 2 + # The target has 50 MiB of payload; a batch must not buffer it all. + assert peak < 16 * 1024 * 1024 + + asyncio.run(scenario()) + + +def test_cancelled_target_stream_can_be_retried(tmp_path: Path) -> None: + async def scenario() -> None: + archive = await _source_archive(tmp_path) + config = SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'cancelled.db'}") + async with SQLiteProfile.open(config, tables=BUILTIN_TABLES) as target: + service = PortableBundleService(target.database, source_decoder=_SOURCES.decode_source) + await service.restore(archive) + + def cancel_read(_connection, _cursor, statement, _parameters, _context, _many): + if statement.startswith("SELECT") and "FROM pc_sources" in statement: + task = asyncio.current_task() + assert task is not None + asyncio.get_running_loop().call_soon(task.cancel) + + event.listen(target.database.engine.sync_engine, "after_cursor_execute", cancel_read) + try: + with pytest.raises(asyncio.CancelledError): + await service.validate(archive) + finally: + event.remove(target.database.engine.sync_engine, "after_cursor_execute", cancel_read) + validation = await service.validate(archive) + assert validation.compatible and validation.already_present == 207 + assert (await service.restore(archive)).inserted == 0 + + asyncio.run(scenario()) + + +def test_validation_bounds_parameters_for_wide_tag_identities(tmp_path: Path) -> None: + async def scenario() -> None: + archive = tmp_path / "wide.pcb" + async with SQLiteProfile.open(SQLiteConfig(), tables=BUILTIN_TABLES) as source: + async with source.database.transaction() as connection: + await _insert_scope(connection) + for number in range(100): + identity = {"scope_id": "project:one", "family": "profile", "artifact_id": f"profile-{number}"} + await connection.execute(insert(ARTIFACTS_TABLE).values(**identity, revision=1, content=b"{}")) + await connection.execute(insert(ARTIFACT_HEADS_TABLE).values(**identity, revision=1)) + await connection.execute( + insert(ARTIFACT_TAGS_TABLE).values( + **identity, + target_type="artifact", + target_id=identity["artifact_id"], + tag_key="portable", + tag_key_hash=hashlib.sha256(b"portable").digest(), + tag="portable", + assigned_at=datetime(2026, 1, 1, tzinfo=UTC), + ) + ) + await PortableBundleService(source.database).export(["project:one"], archive, authorize=_authorize_export) + async with SQLiteProfile.open(SQLiteConfig(), tables=BUILTIN_TABLES) as target: + parameter_counts = [] + + def observe(_connection, _cursor, statement, parameters, _context, _many): + if statement.startswith("SELECT") and "FROM pc_artifact_tags" in statement: + parameter_counts.append(len(parameters)) + + event.listen(target.database.engine.sync_engine, "before_cursor_execute", observe) + service = PortableBundleService(target.database) + assert (await service.validate(archive)).compatible + assert parameter_counts and max(parameter_counts) <= 900 + # Includes the separate tag-set integrity lookup. + assert len(parameter_counts) <= 4 + event.remove(target.database.engine.sync_engine, "before_cursor_execute", observe) + assert (await service.restore(archive)).inserted == 301 + validation = await service.validate(archive) + assert validation.compatible and validation.already_present == 301 + + asyncio.run(scenario()) + + +def test_validation_does_not_match_null_alternate_unique_keys(tmp_path: Path, monkeypatch) -> None: + # Exercise the generic nullable-key contract with a nullable unique column. + # Current built-in alternate keys are non-nullable; this does not change + # the production schema or claim qualification of a new supported schema. + spec = bundle_module._SPECS["artifact_revision"] + table = spec.table.to_metadata(MetaData()) + table.append_constraint(UniqueConstraint("created_by")) + monkeypatch.setitem(bundle_module._SPECS, "artifact_revision", replace(spec, table=table)) + + async def scenario() -> None: + archive = tmp_path / "nullable.pcb" + async with SQLiteProfile.open(SQLiteConfig(), tables=BUILTIN_TABLES) as source: + async with source.database.transaction() as connection: + await _insert_scope(connection) + for artifact_id in ("first", "second"): + await connection.execute( + insert(ARTIFACTS_TABLE).values( + scope_id="project:one", family="profile", artifact_id=artifact_id, revision=1, content=b"{}" + ) + ) + await PortableBundleService(source.database).export(["project:one"], archive, authorize=_authorize_export) + async with SQLiteProfile.open(SQLiteConfig(), tables=BUILTIN_TABLES) as target: + async with target.database.transaction() as connection: + await connection.execute( + insert(ARTIFACTS_TABLE).values( + scope_id="project:one", family="profile", artifact_id="unrelated", revision=1, content=b"{}" + ) + ) + service = PortableBundleService(target.database) + validation = await service.validate(archive) + assert validation.compatible and validation.already_present == 0 + assert (await service.restore(archive)).inserted == 3 + assert (await service.validate(archive)).already_present == 3 + + asyncio.run(scenario()) + + @pytest.mark.parametrize("suffix", ["", "-wal", "-shm", "-journal"]) @pytest.mark.parametrize("output_name", ["source.db", "SOURCE.DB"]) def test_export_cannot_replace_its_sqlite_database_or_journals(tmp_path: Path, suffix: str, output_name: str) -> None: diff --git a/tests/e2e/test_portable_archive.py b/tests/e2e/test_portable_archive.py index a4101aace..d01ca7b89 100644 --- a/tests/e2e/test_portable_archive.py +++ b/tests/e2e/test_portable_archive.py @@ -1063,6 +1063,76 @@ def target_config(tmp_path: Path, short_tmp_path: Path, request: pytest.FixtureR yield BuiltinConfig(database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'target.db'}")) +def test_archive_batches_validation_and_rechecks_after_dry_run(tmp_path: Path, target_config: BuiltinConfig) -> None: + async def authorize(_scopes: tuple[str, ...]) -> None: + pass + + async def scenario() -> None: + archive = tmp_path / "batched.pcb" + async with open_builtin_contexts(BuiltinConfig()) as source: + scope = await source.scopes.create(ScopeDraft(title="Batch", summary="Batch", idempotency_key="batch")) + context = await source.get(scope.scope_id) + for number in range(205): + await context.sources.capture(ContentCapture(source_id=f"source-{number:04}", content="portable")) + exported = await source.portability.export([scope.scope_id], archive, authorize=authorize) + + async with open_builtin_contexts(target_config) as target: + query_parameters = [] + + def observe(_connection, _cursor, statement, parameters, _context, _many): + if statement.startswith("SELECT") and "FROM pc_sources" in statement: + query_parameters.append(len(parameters)) + + event.listen(target.database.engine.sync_engine, "before_cursor_execute", observe) + try: + validation = await target.portability.validate(archive) + assert validation.compatible and validation.already_present == 0 + assert 0 < len(query_parameters) <= 3 + assert max(query_parameters) <= 900 + finally: + event.remove(target.database.engine.sync_engine, "before_cursor_execute", observe) + restored = await target.portability.restore(archive) + assert restored.inserted == exported.record_count and restored.projections_ready + query_parameters.clear() + event.listen(target.database.engine.sync_engine, "before_cursor_execute", observe) + try: + validation = await target.portability.validate(archive) + assert validation.compatible and validation.already_present == exported.record_count + assert 0 < len(query_parameters) <= 3 + finally: + event.remove(target.database.engine.sync_engine, "before_cursor_execute", observe) + + # Change a real target after a successful dry-run, then ensure the + # fresh restore-time check rejects it without overwriting anything. + async with target.database.transaction() as connection: + original = await connection.scalar( + select(SOURCES_TABLE.c.payload).where( + SOURCES_TABLE.c.scope_id == scope.scope_id, SOURCES_TABLE.c.source_id == "source-0100" + ) + ) + assert original is not None + await connection.execute( + update(SOURCES_TABLE) + .where(SOURCES_TABLE.c.scope_id == scope.scope_id, SOURCES_TABLE.c.source_id == "source-0100") + .values(payload=bytes(original).replace(b"portable", b"modified")) + ) + with pytest.raises(BundleConflictError): + await target.portability.restore(archive) + assert (await target.portability.validate(archive)).conflicts == 1 + async with target.database.transaction() as connection: + await connection.execute( + update(SOURCES_TABLE) + .where(SOURCES_TABLE.c.scope_id == scope.scope_id, SOURCES_TABLE.c.source_id == "source-0100") + .values(payload=original) + ) + retry = await target.portability.restore(archive) + assert retry.inserted == 0 and retry.already_present == exported.record_count + async with open_builtin_contexts(target_config) as restarted: + assert (await restarted.portability.validate(archive)).already_present == exported.record_count + + asyncio.run(scenario()) + + @pytest.mark.parametrize("legacy", [False, True]) @pytest.mark.parametrize("target_definition", ["absent", "identical", "conflicting"]) def test_archive_preserves_remote_observations_without_worker_adapter( From 4b5c8a7a1038caa48020b9d7bcc60385fb43b268 Mon Sep 17 00:00:00 2001 From: knqiufan Date: Sun, 25 Oct 2026 14:00:10 +0800 Subject: [PATCH 2/2] fix(archive): report progress after target batches --- docs/en/docs/operate/portable-archive.md | 3 +++ docs/zh/docs/operate/portable-archive.md | 2 ++ .../builtin/portability/bundle.py | 6 +++-- tests/builtin/portability/test_bundle.py | 26 +++++++++++++++++++ 4 files changed, 35 insertions(+), 2 deletions(-) diff --git a/docs/en/docs/operate/portable-archive.md b/docs/en/docs/operate/portable-archive.md index 03329bea4..8a31d1e97 100644 --- a/docs/en/docs/operate/portable-archive.md +++ b/docs/en/docs/operate/portable-archive.md @@ -250,6 +250,9 @@ Nullable unique keys retain SQL null semantics. Dry-run results are not cached f checks the current target again, and write-time conflicts still roll back the transaction. Publication-chain and combined-tag-set checks run after the record checks. +Target-check progress is reported after each completed batch, even when its processed count skips a +1,000-record reporting threshold. Records whose target tables are absent retain periodic per-record reporting. + For the representative archive containing one Scope, one journal head, and N small Sources, the compatibility query budget against initialized tables is `ceil(N / 100) + 2`: 3, 12, and 52 statements for 100, 1,000, and 5,000 Sources. Missing tables require no record compatibility queries. Schema introspection, local dependency diff --git a/docs/zh/docs/operate/portable-archive.md b/docs/zh/docs/operate/portable-archive.md index c4eeeec3d..8ce6c4707 100644 --- a/docs/zh/docs/operate/portable-archive.md +++ b/docs/zh/docs/operate/portable-archive.md @@ -214,6 +214,8 @@ reader 会拒绝这些记录不完整的 schema version `1` 表示，而不是 计为一次冲突；可空唯一键继续遵循 SQL 的 null 语义。dry-run 结果不会缓存给后续恢复使用：restore 会重新检查 当前目标，写入阶段的冲突仍会回滚整个事务。记录检查完成后，继续执行发布链和合并标签集合的完整性检查。 +目标检查每完成一批就报告一次进度，即使累计记录数跨过了 1,000 条记录的报告阈值。目标表不存在的记录仍按逐条处理的周期报告进度。 + 对于一个 Scope、一个日志头和 N 条小 Source 组成的代表性归档，目标表已初始化时，兼容性查询预算为 `ceil(N / 100) + 2`：100、1,000、5,000 条 Source 分别最多执行 3、12、52 条查询。 目标表不存在时，不执行对应记录的兼容性查询。表结构探测、本地依赖索引查询、发布链和标签完整性查询单独计算。 diff --git a/src/powercontext/builtin/portability/bundle.py b/src/powercontext/builtin/portability/bundle.py index 873aff7ac..853da1b1a 100644 --- a/src/powercontext/builtin/portability/bundle.py +++ b/src/powercontext/builtin/portability/bundle.py @@ -787,7 +787,7 @@ async def flush() -> None: already_present += present conflicts += different processed += len(lookups) - _report_progress(progress, "target", processed, total) + _report_progress(progress, "target", processed, total, force=True) pending.clear() for record in _iter_records(source): @@ -2626,8 +2626,10 @@ def _report_progress( phase: Literal["export", "inspect", "validate", "target", "restore", "projections"], processed: int, total: int | None, + *, + force: bool = False, ) -> None: - if observer is not None and (processed in {0, 1, total} or processed % 1_000 == 0): + if observer is not None and (force or processed in {0, 1, total} or processed % 1_000 == 0): observer(BundleProgress(phase=phase, processed=processed, total=total)) diff --git a/tests/builtin/portability/test_bundle.py b/tests/builtin/portability/test_bundle.py index e0742f501..0b8307a19 100644 --- a/tests/builtin/portability/test_bundle.py +++ b/tests/builtin/portability/test_bundle.py @@ -25,6 +25,7 @@ import zipfile from dataclasses import replace from datetime import UTC, datetime +from itertools import pairwise from pathlib import Path from typing import Any, cast @@ -169,6 +170,31 @@ def observe(_connection, _cursor, statement, parameters, _context, _many): asyncio.run(scenario()) +@pytest.mark.parametrize("state", ["absent", "empty", "identical"]) +def test_validation_reports_intermediate_target_progress(tmp_path: Path, state: str) -> None: + async def scenario() -> None: + count = 2_000 + archive = await _source_archive(tmp_path, count) + tables = () if state == "absent" else BUILTIN_TABLES + async with SQLiteProfile.open(SQLiteConfig(), tables=tables) as target: + service = PortableBundleService(target.database, source_decoder=_SOURCES.decode_source) + if state == "identical": + await service.restore(archive) + progress = [] + validation = await service.validate(archive, progress=progress.append) + assert validation.compatible + target_progress = [event for event in progress if event.phase == "target"] + assert target_progress[0].processed == 1 + assert target_progress[-1].processed == count + 2 + assert all(event.total == count + 2 for event in target_progress) + # Batches can cross reporting thresholds without landing on them. + assert all( + 0 < current.processed - previous.processed <= 1_000 for previous, current in pairwise(target_progress) + ) + + asyncio.run(scenario()) + + def test_restore_rechecks_target_after_dry_run_across_batch_boundaries(tmp_path: Path) -> None: async def scenario() -> None: archive = await _source_archive(tmp_path)