diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 74dd1e4ce..369bb0ac9 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -1252,7 +1252,7 @@ jobs: //:resource_inputs \ //:release_bins 2>&1 | tee dist/bazel-representative-build.log - - name: Real tiny progressive lifecycle producer + - name: Real tiny and ownership-growth lifecycle producer shell: bash run: | set -euo pipefail @@ -1271,6 +1271,13 @@ jobs: --certify benchmarks/target/debug/graphforge-benchmark-certify \ --generator benchmarks/target/debug/graphforge-benchmark-graph500-generator \ --workspace-root "$RUNNER_TEMP" + uv run --project benchmarks --locked python \ + benchmarks/scripts/test-tiny-lifecycle-certification.py \ + --growth \ + --gf "$gf_bin" \ + --certify benchmarks/target/debug/graphforge-benchmark-certify \ + --generator benchmarks/target/debug/graphforge-benchmark-graph500-generator \ + --workspace-root "$RUNNER_TEMP" - name: Bazel binding cdylibs + packaging handoff run: | diff --git a/benchmarks/scripts/test-tiny-lifecycle-certification.py b/benchmarks/scripts/test-tiny-lifecycle-certification.py index 0a65e217e..dcf1cf593 100644 --- a/benchmarks/scripts/test-tiny-lifecycle-certification.py +++ b/benchmarks/scripts/test-tiny-lifecycle-certification.py @@ -1,10 +1,12 @@ #!/usr/bin/env python3 -"""Run the real scale-1 lifecycle through the public certification binaries.""" +"""Run tiny and ownership-growth lifecycles through public certification binaries.""" from __future__ import annotations import argparse +import hashlib import importlib +import itertools import json import os from pathlib import Path @@ -35,7 +37,7 @@ def decoded_objects(text: str) -> list[dict[str, object]]: return objects -def storage_totals(root: Path) -> dict[str, int]: +def storage_totals(root: Path, selected: set[Path] | None = None) -> dict[str, int]: """Independent regular-file oracle, used only after every child has exited. Directories are not data objects. Hardlinked references count once by the @@ -52,6 +54,8 @@ def storage_totals(root: Path) -> dict[str, int]: raise SystemExit("allocation oracle encountered a linked directory") for name in files: path = Path(directory) / name + if selected is not None and path not in selected: + continue before = path.lstat() if not stat.S_ISREG(before.st_mode): raise SystemExit("allocation oracle encountered a non-regular file") @@ -90,14 +94,194 @@ def storage_totals(root: Path) -> dict[str, int]: } +OWNER_SUFFIXES = ("construction", "import", "transactions", "locks", "admission_lock", "published") +OWNER_NAMES = { + f"{project}-project-{suffix}" for project in ("source", "imported") for suffix in OWNER_SUFFIXES +} | {"generated-inputs", "query-results", "portable-package"} + + +def owner_inventories(workspace: Path) -> dict[str, set[Path]]: + """Classify every quiescent route independently of the producer's owner map.""" + owners = {name: set() for name in OWNER_NAMES} + admissions = {} + for project in ("source", "imported"): + digest = hashlib.sha256( + b"graphforge-project-lifecycle-lock/v1\0" + + os.fsencode(workspace) + + b"\0" + + os.fsencode(project) + ).hexdigest() + admissions[f".graphforge-admission-{digest}.lock"] = f"{project}-project-admission_lock" + for directory, directories, files in os.walk(workspace, followlinks=False): + for name in directories: + if not stat.S_ISDIR((Path(directory) / name).lstat().st_mode): + raise SystemExit("owner oracle encountered linked directory") + for name in files: + path = Path(directory) / name + parts = path.relative_to(workspace).parts + if len(parts) == 1 and name in admissions: + owner = admissions[name] + elif len(parts) == 1 and name in {"nodes.parquet", "edges.parquet"}: + owner = "generated-inputs" + elif len(parts) == 1 and name in { + "node-count.arrow", + "edge-count.arrow", + "one-hop.arrow", + "two-hop.arrow", + "imported-node-count.arrow", + "imported-edge-count.arrow", + "imported-one-hop.arrow", + "imported-two-hop.arrow", + }: + owner = "query-results" + elif parts[0] == "portable": + owner = "portable-package" + elif parts[0] in {"source", "imported"} and len(parts) > 1: + suffix = { + ".graphforge-construction": "construction", + "import-sessions": "import", + "transactions": "transactions", + "locks": "locks", + }.get(parts[1]) + if parts[1] in {"FORMAT", "CURRENT", "generations", "graph-objects"}: + suffix = ( + "locks" + if len(parts) == 4 + and parts[1] == "generations" + and parts[3] == "lease.lock" + else "published" + ) + if suffix is None: + raise SystemExit("unclassified project owner route") + owner = f"{parts[0]}-project-{suffix}" + else: + raise SystemExit("unclassified workspace owner route") + owners[owner].add(path) + return owners + + +DATA_OWNERS = { + "generated-inputs", + "query-results", + "portable-package", + "source-project-construction", + "source-project-import", + "source-project-published", + "imported-project-published", +} +ZERO_OWNERS = {"imported-project-construction", "imported-project-import"} +BYTE_FIELDS = ("logical_bytes", "physical_logical_bytes", "allocated_bytes") + + +def positive_slopes(name: str, values: list[int], work: list[int]) -> None: + deltas = [values[i + 1] - values[i] for i in (0, 1)] + if min(deltas) <= 0: + raise ValueError(f"{name}: data growth is flat or decreasing") + left = deltas[0] * (work[2] - work[1]) + right = deltas[1] * (work[1] - work[0]) + if left > 2 * right or right > 2 * left: + raise ValueError(f"{name}: adjacent normalized slopes differ by more than factor2") + + +def normalized_ceiling(name: str, values: list[int], work: list[int]) -> None: + if values[0] <= 0 or any(b < a for a, b in itertools.pairwise(values)): + raise ValueError(f"{name}: positive nondecreasing evidence required") + if any(values[i] * work[0] > 2 * values[0] * work[i] for i in (1, 2)): + raise ValueError(f"{name}: exceeds existing factor2 normalized ceiling") + + +def validate_growth(observations: list[dict[str, object]]) -> None: + """Existing factor2 ceiling, plus strict EOF slopes; no output-fitted constants. + + Allocated owner bytes may plateau due to per-file allocation quanta. They + retain the existing quantized normalized ceiling, not an EOF upper bound. + Fixed empty locks and absent owners are checked separately. Complete current + and peak must actually grow, using the same adjacent slope bound as EOF. + """ + if [o["scale"] for o in observations] != [6, 7, 8]: + raise ValueError("growth requires exactly scale6/7/8") + sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "harness")) + consumer = importlib.import_module("graphforge_bench.progressive_run") + work = [] + for o in observations: + try: + consumer.validate_lifecycle_storage_receipt(o["receipt"]) + except consumer.ControllerError as error: + raise ValueError(str(error)) from error + n, e = o["live_nodes"], o["live_edges"] + if n != 1 << o["scale"] or e != 16 * n: + raise ValueError("invalid authoritative live denominator") + if o["query_rows"] != [n, e] or o["imported_query_rows"] != [n, e]: + raise ValueError("growth query output was truncated or duplicated") + work.append(n + e) + ratios = o["ratios"] + for field in ("retained_storage_bytes", "transient_peak_storage_bytes"): + if ratios[field] != { + "per_live_node": [o["receipt"][field], n], + "per_live_edge": [o["receipt"][field], e], + }: + raise ValueError("fabricated normalized ratio") + owners = o["receipt"]["retained_owners"] + if set(owners) != OWNER_NAMES: + raise ValueError("missing or unknown growth owner") + for owner in sorted(OWNER_NAMES): + totals = [o["receipt"]["retained_owners"][owner]["totals"] for o in observations] + if any( + set(t) + != { + "logical_references", + "logical_bytes", + "physical_objects", + "physical_logical_bytes", + "allocated_bytes", + } + for t in totals + ): + raise ValueError("missing owner metric") + for field in totals[0]: + values = [t[field] for t in totals] + if any(type(v) is not int or v < 0 for v in values): + raise ValueError("invalid owner numerator") + if owner in ZERO_OWNERS: + if any(values): + raise ValueError(f"{owner}: absent owner became nonzero") + elif owner.endswith(("-locks", "-admission_lock")): + if field in BYTE_FIELDS and any(values): + raise ValueError(f"{owner}: empty lock gained payload") + if len(set(values)) != 1: + raise ValueError(f"{owner}: fixed lock inventory changed") + elif owner in DATA_OWNERS: + normalized_ceiling(f"{owner}.{field}", values, work) + if field in ("logical_bytes", "physical_logical_bytes"): + positive_slopes(f"{owner}.{field}", values, work) + # Transaction journals have fixed protocol shape and fixed-width + # UUID/digest fields; no graph payload or numeric row inventory. + elif len(set(values)) != 1: + raise ValueError(f"{owner}: fixed transaction protocol changed") + for field in ("retained_storage_bytes", "transient_peak_storage_bytes"): + values = [o["receipt"][field] for o in observations] + normalized_ceiling(field, values, work) + positive_slopes(field, values, work) + + def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--gf", required=True, type=executable) parser.add_argument("--certify", required=True, type=executable) parser.add_argument("--generator", required=True, type=executable) parser.add_argument("--workspace-root", required=True, type=Path) + parser.add_argument("--growth", action="store_true") args = parser.parse_args() + if args.growth: + observations = [run(args, scale) for scale in (6, 7, 8)] + validate_growth(observations) + print(json.dumps({"lifecycle_growth": observations}, sort_keys=True)) + else: + run(args, 1) + + +def run(args: argparse.Namespace, scale: int) -> dict[str, object]: root = Path(__file__).resolve().parents[1] workspace_root = args.workspace_root.resolve(strict=True) filesystem = subprocess.run( @@ -125,11 +309,26 @@ def main() -> None: runtime_tmp.mkdir() environment["TMPDIR"] = str(runtime_tmp) evidence_path = work / "evidence.json" + profile = json.loads((root / "fixtures/progressive/tiny-executable.json").read_text()) + generator_args = profile["phases"][1]["action"]["args"] + generator_args[generator_args.index("--scale") + 1] = str(scale) + if args.growth: + payloads = [ + "MATCH (n) RETURN n.node_uuid AS id ORDER BY id", + "MATCH ()-[r]->() RETURN r.edge_uuid AS id ORDER BY id", + ] + for phase_name, indices in (("query", (0, 1)), ("reopen_proof", (2, 3))): + phase = next(item for item in profile["phases"] if item["phase"] == phase_name) + for index, query in zip(indices, payloads, strict=True): + command = phase["action"]["commands"][index] + command[command.index("--cypher") + 1] = query + profile_path = work / "profile.json" + profile_path.write_text(json.dumps(profile)) completed = subprocess.run( [ str(args.certify), "run", - str(root / "fixtures/progressive/tiny-executable.json"), + str(profile_path), str(evidence_path), ], cwd=work, @@ -207,12 +406,12 @@ def main() -> None: } if [len(queries[name]) for name in ("recount", "query", "reopen_proof")] != [2, 2, 4]: raise SystemExit("tiny lifecycle omitted ordinary source/imported query receipts") - for index, expected in enumerate((2, 32)): + for index, expected in enumerate((1 << scale, (1 << scale) * 16)): if ( queries["recount"][index].get("scalar_u64") != expected or queries["reopen_proof"][index].get("scalar_u64") != expected ): - raise SystemExit("tiny stored/imported counts differ from SCALE1 raw input") + raise SystemExit("tiny stored/imported counts differ from deterministic raw input") for source, imported in zip( queries["recount"] + queries["query"], queries["reopen_proof"], strict=True ): @@ -264,17 +463,35 @@ def main() -> None: f"lifecycle retained allocation {retained} " f"differs from independent union {measured}" ) - for name, private in { - "source-project-construction": workspace / "source" / ".graphforge-construction", - "source-project-import": workspace / "source" / "import-sessions", - "source-project-transactions": workspace / "source" / "transactions", - "imported-project-transactions": workspace / "imported" / "transactions", - }.items(): - measured_owner = storage_totals(private) - if measured_owner["allocated_bytes"] <= 0: - raise SystemExit("tiny lifecycle omitted live private/control allocation") + for name, paths in owner_inventories(workspace).items(): + measured_owner = storage_totals(workspace, paths) if receipt["retained_owners"][name]["totals"] != measured_owner: raise SystemExit(f"lifecycle owner {name} differs from independent raw facts") + for name in ( + "source-project-construction", + "source-project-import", + "source-project-transactions", + "imported-project-transactions", + ): + if receipt["retained_owners"][name]["totals"]["allocated_bytes"] <= 0: + raise SystemExit("tiny lifecycle omitted live private/control allocation") + + return { + "scale": scale, + "live_nodes": queries["recount"][0]["scalar_u64"], + "live_edges": queries["recount"][1]["scalar_u64"], + "query_rows": [q["rows"] for q in queries["query"]], + "imported_query_rows": [q["rows"] for q in queries["reopen_proof"][2:]], + "receipt": receipt, + "allocation_quantum": os.statvfs(workspace).f_frsize, + "ratios": { + field: { + "per_live_node": [receipt[field], queries["recount"][0]["scalar_u64"]], + "per_live_edge": [receipt[field], queries["recount"][1]["scalar_u64"]], + } + for field in ("retained_storage_bytes", "transient_peak_storage_bytes") + }, + } if __name__ == "__main__": diff --git a/benchmarks/tests/test_lifecycle_growth.py b/benchmarks/tests/test_lifecycle_growth.py new file mode 100644 index 000000000..9f5e90c6c --- /dev/null +++ b/benchmarks/tests/test_lifecycle_growth.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import copy +import importlib.util +from pathlib import Path +import unittest + +SPEC = importlib.util.spec_from_file_location( + "tiny_growth", Path(__file__).parents[1] / "scripts/test-tiny-lifecycle-certification.py" +) +assert SPEC and SPEC.loader +GROWTH = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(GROWTH) + + +def observations(): + result = [] + for scale, factor in zip((6, 7, 8), (1, 2, 4), strict=True): + owners = {} + for owner in GROWTH.OWNER_NAMES: + if owner in GROWTH.ZERO_OWNERS: + metrics = [0, 0, 0, 0, 0] + elif owner.endswith(("-locks", "-admission_lock")): + metrics = [1, 0, 1, 0, 0] + elif owner in GROWTH.DATA_OWNERS: + metrics = [2, 8192 * factor, 2, 8192 * factor, 8192 * factor] + else: + metrics = [1, 628, 1, 628, 4096] + owners[owner] = { + "totals": dict( + zip( + ( + "logical_references", + "logical_bytes", + "physical_objects", + "physical_logical_bytes", + "allocated_bytes", + ), + metrics, + strict=True, + ) + ) + } + receipt = { + "contract": "graphforge-lifecycle-storage/2", + "source_project_current_allocated_bytes": 8192 * factor, + "retained_owners": owners, + "retained_storage_bytes": 50000 * factor, + "transient_peak_storage_bytes": 60000 * factor, + } + n = 1 << scale + result.append( + { + "scale": scale, + "live_nodes": n, + "live_edges": n * 16, + "query_rows": [n, n * 16], + "imported_query_rows": [n, n * 16], + "receipt": receipt, + "ratios": { + field: { + "per_live_node": [receipt[field], n], + "per_live_edge": [receipt[field], n * 16], + } + for field in ("retained_storage_bytes", "transient_peak_storage_bytes") + }, + } + ) + return result + + +class LifecycleGrowthTests(unittest.TestCase): + def test_declared_linear_and_fixed_policies(self): + GROWTH.validate_growth(observations()) + + def test_each_data_owner_rejects_flat_underreported_and_inflated_eof(self): + for owner in GROWTH.DATA_OWNERS: + for field in ("logical_bytes", "physical_logical_bytes"): + for value in (8192, 17000, 200000): + with self.subTest(owner=owner, field=field, value=value): + changed = observations() + changed[2]["receipt"]["retained_owners"][owner]["totals"][field] = value + with self.assertRaises(ValueError): + GROWTH.validate_growth(changed) + + def test_missing_owner_metric_and_fabricated_ratio_refused(self): + for mutate in ( + lambda o: o[1]["receipt"]["retained_owners"].pop("source-project-import"), + lambda o: o[1]["receipt"]["retained_owners"]["generated-inputs"]["totals"].pop( + "physical_objects" + ), + lambda o: o[1]["ratios"]["retained_storage_bytes"]["per_live_edge"].__setitem__(1, 1), + lambda o: o[1].__setitem__("live_nodes", 0), + lambda o: o[1].__setitem__("query_rows", [128, 1000]), + lambda o: o[1].__setitem__("imported_query_rows", [128, 1000]), + ): + changed = copy.deepcopy(observations()) + mutate(changed) + with self.assertRaises(ValueError): + GROWTH.validate_growth(changed) + + def test_quantized_owner_plateau_allowed_but_complete_peak_flat_refused(self): + changed = observations() + for o in changed: + o["receipt"]["retained_owners"]["generated-inputs"]["totals"]["allocated_bytes"] = 32768 + GROWTH.validate_growth(changed) + changed = observations() + for o in changed: + o["receipt"]["transient_peak_storage_bytes"] = 120000 + o["ratios"]["transient_peak_storage_bytes"]["per_live_node"][0] = 120000 + o["ratios"]["transient_peak_storage_bytes"]["per_live_edge"][0] = 120000 + with self.assertRaises(ValueError): + GROWTH.validate_growth(changed) diff --git a/crates/graphforge-api/tests/scale_g500_ladder.rs b/crates/graphforge-api/tests/scale_g500_ladder.rs index 07dadb1fd..eb0b943cf 100644 --- a/crates/graphforge-api/tests/scale_g500_ladder.rs +++ b/crates/graphforge-api/tests/scale_g500_ladder.rs @@ -7298,6 +7298,9 @@ fn lifecycle_metric_policy_accepts_bounded_fixed_protocol_and_rejects_false_grow #[test] fn equivalent_full_lifecycle_1x_2x_4x_has_bounded_metric_policies() { + // This fixture proves phase/category policies and recovery drills. Its + // PhaseJournal allocation estimate is not the complete lifecycle/2 owner + // proof; that proof runs through the public certification binaries. let mut failures = Vec::new(); for axis in [LinearityAxis::Nodes, LinearityAxis::Edges] { let mut observations = Vec::new(); diff --git a/docs/development/perf-g500-ladder.md b/docs/development/perf-g500-ladder.md index f810f4e15..81ce27097 100644 --- a/docs/development/perf-g500-ladder.md +++ b/docs/development/perf-g500-ladder.md @@ -311,6 +311,35 @@ CSR, reopen/query, export/verify, clean import/reopen/query, corruption, cancellation, resource-limit, and interrupted-finalization drills. It validates the qualification phase inventory and bounds bytes, calls, objects, blocks, and fsyncs per phase rather than relying only on aggregate I/O. +Its test-local allocation estimate does not prove the complete `lifecycle/2` +ownership contract. The public certifier's ownership-growth regression runs +SCALE6/7/8 fixtures through import, commit, reopen, export, verification, clean +import, and reopen. Reopened counts must be exactly 64/128/256 nodes and +1024/2048/4096 edges. These fixtures retain every generated edge as a distinct +UUID-bearing record; their counts are not unique endpoint-pair counts. +Ordered node and edge outputs keep the growth workload linear. The default +SCALE1 regression retains its original one-hop and two-hop queries. +At each size, the regression reconciles all 15 retained owner views and the +complete retained allocation against an independent, quiescent filesystem +inventory. Observed transient peak remains a separate writer measurement. +Data-bearing logical-byte totals and complete retained/peak allocation must +grow at both steps; adjacent slopes, normalized by live nodes plus live edges, +must stay within a factor of two. The existing factor-of-two normalized upper +ceiling also applies. Per-owner allocated bytes and object/reference counts may +plateau at allocation or file boundaries. Empty owners stay empty, while lock +and transaction inventories retain their fixed protocol costs. Comparisons use +integer cross-products, not rounded ratios or constants fitted to the run. +These small admission fixtures do not qualify a canonical S20 or S22 rung. + +Run the complete ownership-growth admission with the existing binaries: + +```bash +python benchmarks/scripts/test-tiny-lifecycle-certification.py --growth \ + --gf /path/to/gf --certify /path/to/graphforge-benchmark-certify \ + --generator /path/to/graphforge-benchmark-graph500-generator \ + --workspace-root /admitted-volume +``` + Node canonical cost uses reopened live nodes; edge canonical, authoritative project, and lifecycle peak costs use reopened live edges. Ratios preserve raw integer numerators and denominators; rounded decimals are not evidence.