From 54dbee765623483e7bd14f2b6aec90d99305547d Mon Sep 17 00:00:00 2001 From: Travis James Date: Sun, 30 Aug 2026 12:52:29 -0500 Subject: [PATCH] fix: enforce graph limit on canonical stream --- CHANGELOG.md | 7 + PERFORMANCE.md | 48 +++++ .../tests/canonical_publication_limit.rs | 127 ++++++++++++ crates/compass-core/src/cluster_existing.rs | 13 +- crates/compass-core/src/pipeline.rs | 65 ++++--- crates/compass-graph/src/lib.rs | 8 +- crates/compass-graph/src/snapshot.rs | 105 +++++++++- crates/compass-store/src/lib.rs | 32 ++++ scripts/qualify_graph_size_ratios.py | 180 ++++++++++++++++++ 9 files changed, 545 insertions(+), 40 deletions(-) create mode 100644 crates/compass-cli/tests/canonical_publication_limit.rs create mode 100755 scripts/qualify_graph_size_ratios.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e6bd828f4..18986b062 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ ## Unreleased +- Enforce `COMPASS_MAX_GRAPH_BYTES` against the actual canonical bytes streamed + into atomic staging for full, SQLite-backed, and fact-neutral delta + publications. A real overrun leaves the previous graph and store reference + active. Add a five-estate qualification command and measured expansion + distribution; source-to-graph ratios remain diagnostic evidence and are not + an admission rule. + - Reset all 14 registered universal-evidence producer versions to v1 and keep them `Qualified` under the refreshed release decision at `tests/qualification/universal-evidence-promotion.json`. Cached evidence diff --git a/PERFORMANCE.md b/PERFORMANCE.md index 184d43e19..652cf887e 100644 --- a/PERFORMANCE.md +++ b/PERFORMANCE.md @@ -24,6 +24,54 @@ Compare a proposed change with a previously approved Compass result captured on the same runner and corpus. A median regression above 10% requires explicit review and evidence explaining the tradeoff. +## Canonical graph size qualification + +`COMPASS_MAX_GRAPH_BYTES` is enforced against the canonical JSON bytes emitted +into atomic staging. Compass does not predict canonical graph size from source +bytes: measured expansion varies with language, structural density, inventory +coverage, and extraction evidence, so a single linear multiplier is not a safe +admission model. If the emitted stream crosses the bound, publication aborts +and the previously active artifact set remains intact. + +Measure the five-estate corpus after producing one completed `graph.json` per +estate. The command reads only the bounded metadata prefix, sums the persisted +`byteSize` values for admitted files, and reports per-estate ratios plus the +minimum, median, and maximum distribution: + +```bash +python3 scripts/qualify_graph_size_ratios.py \ + --estate ara-scanworks-ui /path/to/ara-scanworks-ui/compass-out/graph.json \ + --estate ara-pm /path/to/ara-pm/compass-out/graph.json \ + --estate captivebrowser-133 /path/to/captivebrowser-133/compass-out/graph.json \ + --estate venus /path/to/venus/compass-out/graph.json \ + --estate solaris-platform /path/to/solaris-platform/compass-out/graph.json \ + --json-output target/qualification/graph-size-ratios.json \ + --markdown-output target/qualification/graph-size-ratios.md +``` + +The report schema is `compass.qualification.graph-size-ratios/1`. Generated +reports stay under `target/`; a retained observation must identify the Compass +commit, corpus revision, and host separately. Ratios are sizing evidence only +and must never be converted into a fatal preflight estimate. + +The 2026-08-30 qualification used the release binary from this change, JSON +storage, and identical `--no-viz --no-cluster --no-program` settings for every +estate. Admitted source bytes are the persisted `byteSize` values in each +completed artifact, not a filesystem-wide estimate: + +| Estate | Admitted files | Admitted source bytes | Canonical graph bytes | Ratio | +| --- | ---: | ---: | ---: | ---: | +| `ara-pm` | 952 | 39,468,272 | 14,274,716 | 0.361676× | +| `ara-scanworks-ui` | 494 | 6,688,344 | 3,721,796 | 0.556460× | +| `captivebrowser-133` | 1,602 | 52,006,357 | 403,028,839 | 7.749607× | +| `solaris-platform` | 5,102 | 112,938,563 | 87,429,961 | 0.774137× | +| `venus` | 2,410 | 140,482,350 | 6,832,605 | 0.048637× | + +The observed distribution is **0.048637× minimum**, **0.556460× median**, and +**7.749607× maximum**. The roughly 159-fold spread between the minimum and +maximum is direct evidence that source bytes alone are not a safe fatal +admission signal. + ## Markdown graph-v1 quality qualification Markdown tables intentionally retain their table, header, row, and cell nodes. diff --git a/crates/compass-cli/tests/canonical_publication_limit.rs b/crates/compass-cli/tests/canonical_publication_limit.rs new file mode 100644 index 000000000..ea6841f13 --- /dev/null +++ b/crates/compass-cli/tests/canonical_publication_limit.rs @@ -0,0 +1,127 @@ +use std::error::Error; +use std::fs; +use std::process::Command; + +use compass_files::BuildGuard; + +#[test] +fn actual_delta_overrun_preserves_the_active_snapshot() -> Result<(), Box> { + let root = tempfile::tempdir()?; + let source = root.path().join("sample.rs"); + fs::write(&source, "pub fn sample() -> u64 { 1 }\n// before\n")?; + let initial = Command::new(env!("CARGO_BIN_EXE_compass")) + .args([ + "update", + ".", + "--force", + "--store", + "json", + "--no-cluster", + "--no-viz", + "--no-program", + ]) + .current_dir(root.path()) + .env_remove("COMPASS_OUT") + .env_remove("COMPASS_MAX_GRAPH_BYTES") + .output()?; + assert!( + initial.status.success(), + "initial publication failed: {}", + String::from_utf8_lossy(&initial.stderr) + ); + + let output = root.path().join("compass-out"); + let active_graph = BuildGuard::resolve_artifact(&output, "graph.json")?; + let prior_bytes = fs::read(&active_graph)?; + fs::write( + &source, + format!( + "pub fn sample() -> u64 {{ 1 }}\n// {}\n", + "after".repeat(200) + ), + )?; + + let rejected = Command::new(env!("CARGO_BIN_EXE_compass")) + .args([ + "update", + ".", + "--store", + "json", + "--no-cluster", + "--no-viz", + "--no-program", + ]) + .current_dir(root.path()) + .env_remove("COMPASS_OUT") + .env("COMPASS_MAX_GRAPH_BYTES", prior_bytes.len().to_string()) + .output()?; + assert_eq!(rejected.status.code(), Some(1)); + let stderr = String::from_utf8(rejected.stderr)?; + assert!(stderr.contains("canonical graph exceeds"), "{stderr}"); + assert!(stderr.contains("COMPASS_MAX_GRAPH_BYTES"), "{stderr}"); + + let still_active = BuildGuard::resolve_artifact(&output, "graph.json")?; + assert_eq!(fs::read(still_active)?, prior_bytes); + Ok(()) +} + +#[test] +fn actual_full_overrun_preserves_the_active_sqlite_snapshot() -> Result<(), Box> { + let root = tempfile::tempdir()?; + let source = root.path().join("sample.rs"); + fs::write(&source, "pub fn sample() -> u64 { 1 }\n")?; + let initial = Command::new(env!("CARGO_BIN_EXE_compass")) + .args([ + "update", + ".", + "--force", + "--store", + "sqlite", + "--no-cluster", + "--no-viz", + "--no-program", + ]) + .current_dir(root.path()) + .env_remove("COMPASS_OUT") + .env_remove("COMPASS_MAX_GRAPH_BYTES") + .output()?; + assert!( + initial.status.success(), + "initial SQLite publication failed: {}", + String::from_utf8_lossy(&initial.stderr) + ); + + let output = root.path().join("compass-out"); + let active_graph = BuildGuard::resolve_artifact(&output, "graph.json")?; + let prior_graph = fs::read(&active_graph)?; + let active_store_ref = BuildGuard::resolve_artifact(&output, "store.ref")?; + let prior_store_ref = fs::read(&active_store_ref)?; + let expanded = (0..200) + .map(|index| format!("pub fn added_{index}() -> usize {{ {index} }}\n")) + .collect::(); + fs::write(&source, expanded)?; + + let rejected = Command::new(env!("CARGO_BIN_EXE_compass")) + .args([ + "update", + ".", + "--store", + "sqlite", + "--no-cluster", + "--no-viz", + "--no-program", + ]) + .current_dir(root.path()) + .env_remove("COMPASS_OUT") + .env("COMPASS_MAX_GRAPH_BYTES", prior_graph.len().to_string()) + .output()?; + assert_eq!(rejected.status.code(), Some(1)); + let stderr = String::from_utf8(rejected.stderr)?; + assert!(stderr.contains("canonical graph exceeds"), "{stderr}"); + + let still_active_graph = BuildGuard::resolve_artifact(&output, "graph.json")?; + let still_active_ref = BuildGuard::resolve_artifact(&output, "store.ref")?; + assert_eq!(fs::read(still_active_graph)?, prior_graph); + assert_eq!(fs::read(still_active_ref)?, prior_store_ref); + Ok(()) +} diff --git a/crates/compass-core/src/cluster_existing.rs b/crates/compass-core/src/cluster_existing.rs index 9ac37bb7c..c78f5c485 100644 --- a/crates/compass-core/src/cluster_existing.rs +++ b/crates/compass-core/src/cluster_existing.rs @@ -6,8 +6,9 @@ use std::time::{Duration, Instant}; use compass_files::{BuildGuard, write_atomic_with_digest, write_json_atomic, write_text_atomic}; use compass_graph::{ ClusterOptions, Communities, GodNode, blind_spot_report, cluster, community_member_signatures, - god_nodes, label_communities_by_hub, remap_communities_to_previous, score_communities, - suggest_questions, surprising_connections, write_canonical_graph_json, + god_nodes, label_communities_by_hub, max_canonical_graph_bytes, remap_communities_to_previous, + score_communities, suggest_questions, surprising_connections, + write_canonical_graph_json_bounded, }; use compass_model::GraphDocument; use compass_model::GraphError; @@ -280,12 +281,12 @@ where let graph_path = staging.join("graph.json"); let graph_identity = if let Some(typed) = typed_document { let receipt = write_atomic_with_digest(&graph_path, |writer| { - write_canonical_graph_json(&typed, writer).map_err(|source| { - compass_files::FileError::Io { + write_canonical_graph_json_bounded(&typed, writer, max_canonical_graph_bytes()).map_err( + |source| compass_files::FileError::Io { path: graph_path.clone(), source, - } - }) + }, + ) })?; format!("sha256:{}", receipt.sha256) } else { diff --git a/crates/compass-core/src/pipeline.rs b/crates/compass-core/src/pipeline.rs index 98b139145..0adf643ee 100644 --- a/crates/compass-core/src/pipeline.rs +++ b/crates/compass-core/src/pipeline.rs @@ -23,10 +23,11 @@ use compass_graph::{ build_owned_with_tiebreaker_at_inference as build_document, canonical_edge_kind, canonical_raw_edge_sites, cluster_incremental, deduped_node_count, extraction_from_v1, garbage_collect_graph_snapshots, graph_insights_with_blind_spots, graph_snapshot_needs_gc, - label_communities_by_hub, normalize_document_v1_with_evidence_best_effort_owned_at_inference, + label_communities_by_hub, max_canonical_graph_bytes, + normalize_document_v1_with_evidence_best_effort_owned_at_inference, normalize_document_v1_with_inventory_and_source_digests_best_effort_owned_at_inference, normalize_document_v1_with_inventory_best_effort_at_inference, score_communities, - write_canonical_graph_json, write_fact_neutral_graph_json_delta_prevalidated, + write_canonical_graph_json_bounded, write_fact_neutral_graph_json_delta_prevalidated_bounded, }; use compass_languages::{ BindingFact, DeclarationFact, EXTRACTION_QUALITY_EXTENSION, EXTRACTION_QUALITY_PARTIAL, @@ -1974,11 +1975,12 @@ fn publish_fact_neutral_incremental( let receipt = write_atomic_with_digest(&graph_path, |writer| { let delta_started = Instant::now(); let used_delta = if let Some(bytes) = previous_bytes.as_deref() { - write_fact_neutral_graph_json_delta_prevalidated( + write_fact_neutral_graph_json_delta_prevalidated_bounded( bytes, current, changed_node_ids, writer, + max_canonical_graph_bytes(), ) .map_err(|source| compass_files::FileError::Io { path: graph_path.clone(), @@ -1998,12 +2000,11 @@ fn publish_fact_neutral_incremental( if used_delta { Ok(()) } else { - write_canonical_graph_json(current, writer).map_err(|source| { - compass_files::FileError::Io { + write_canonical_graph_json_bounded(current, writer, max_canonical_graph_bytes()) + .map_err(|source| compass_files::FileError::Io { path: graph_path.clone(), source, - } - }) + }) } })?; ( @@ -3629,11 +3630,14 @@ fn build_graph_inner_unscoped( } else { let graph_path = output_dir.join("graph.json"); let receipt = write_atomic_with_digest(&graph_path, |writer| { - write_canonical_graph_json(&published.document, writer).map_err(|source| { - compass_files::FileError::Io { - path: graph_path.clone(), - source, - } + write_canonical_graph_json_bounded( + &published.document, + writer, + max_canonical_graph_bytes(), + ) + .map_err(|source| compass_files::FileError::Io { + path: graph_path.clone(), + source, }) })?; ( @@ -4192,11 +4196,14 @@ fn build_graph_inner_unscoped( } else { let graph_path = output_dir.join("graph.json"); let receipt = write_atomic_with_digest(&graph_path, |writer| { - write_canonical_graph_json(&published_document, writer).map_err(|source| { - compass_files::FileError::Io { - path: graph_path.clone(), - source, - } + write_canonical_graph_json_bounded( + &published_document, + writer, + max_canonical_graph_bytes(), + ) + .map_err(|source| compass_files::FileError::Io { + path: graph_path.clone(), + source, }) })?; ( @@ -4879,12 +4886,11 @@ fn publish_graph_and_store_from_canonical( let (graph_receipt, content) = rayon::join( || { write_atomic_with_digest(&graph_path, |writer| { - write_canonical_graph_json(graph, writer).map_err(|source| { - compass_files::FileError::Io { + write_canonical_graph_json_bounded(graph, writer, max_canonical_graph_bytes()) + .map_err(|source| compass_files::FileError::Io { path: graph_path.clone(), source, - } - }) + }) }) }, || builder.prepare_content(&store, graph), @@ -4922,8 +4928,12 @@ fn publish_graph_and_store_delta( let result = write_atomic_with_digest(&graph_path, |writer| { let used_delta = match (previous_bytes.as_deref(), changed_node_ids) { (Some(bytes), Some(changed)) => { - write_fact_neutral_graph_json_delta_prevalidated( - bytes, graph, changed, writer, + write_fact_neutral_graph_json_delta_prevalidated_bounded( + bytes, + graph, + changed, + writer, + max_canonical_graph_bytes(), ) .map_err(|source| { compass_files::FileError::Io { @@ -4937,12 +4947,11 @@ fn publish_graph_and_store_delta( if used_delta { Ok(()) } else { - write_canonical_graph_json(graph, writer).map_err(|source| { - compass_files::FileError::Io { + write_canonical_graph_json_bounded(graph, writer, max_canonical_graph_bytes()) + .map_err(|source| compass_files::FileError::Io { path: graph_path.clone(), source, - } - }) + }) } }); profile_internal_duration("graph JSON delta publication", started.elapsed()); @@ -8773,7 +8782,7 @@ mod tests { let changed_graph = V1GraphDocument::load(&changed.output_dir.join("graph.json"))?; let mut canonical_changed = Vec::new(); - write_canonical_graph_json(&changed_graph, &mut canonical_changed)?; + compass_graph::write_canonical_graph_json(&changed_graph, &mut canonical_changed)?; assert_eq!( fs::read(changed.output_dir.join("graph.json"))?, canonical_changed, diff --git a/crates/compass-graph/src/lib.rs b/crates/compass-graph/src/lib.rs index aac7169aa..7737b6fc3 100644 --- a/crates/compass-graph/src/lib.rs +++ b/crates/compass-graph/src/lib.rs @@ -36,12 +36,14 @@ pub use snapshot::{ GRAPH_SNAPSHOT_MAX_ITEMS, GRAPH_SNAPSHOT_MAX_OBJECTS, GRAPH_SNAPSHOT_SELECTOR_SCHEMA_V1, GRAPH_TERM_POSTING_CHUNK_ITEMS, GraphSnapshotBuilder, GraphSnapshotGcStats, GraphSnapshotManifest, GraphSnapshotMetadata, GraphSnapshotReader, IndexKind, - PreparedGraphSnapshot, PreparedGraphSnapshotContent, SnapshotError, SnapshotReadLimits, - SnapshotRoot, SnapshotSelector, TermPostingWork, activate_graph_snapshot, + MAX_CANONICAL_GRAPH_BYTES, PreparedGraphSnapshot, PreparedGraphSnapshotContent, SnapshotError, + SnapshotReadLimits, SnapshotRoot, SnapshotSelector, TermPostingWork, activate_graph_snapshot, active_graph_snapshot, canonical_graph_document, canonical_graph_document_presorted, canonical_graph_json, encode_graph_index_key, garbage_collect_graph_snapshots, - graph_snapshot_needs_gc, prepare_graph_snapshot, write_canonical_graph_json, + graph_snapshot_needs_gc, max_canonical_graph_bytes, prepare_graph_snapshot, + write_canonical_graph_json, write_canonical_graph_json_bounded, write_fact_neutral_graph_json_delta, write_fact_neutral_graph_json_delta_prevalidated, + write_fact_neutral_graph_json_delta_prevalidated_bounded, }; pub use v1::{ BuildEvidence, InventoryEvidence, SourceDigest, V1_PUBLICATION_SEMANTICS_VERSION, diff --git a/crates/compass-graph/src/snapshot.rs b/crates/compass-graph/src/snapshot.rs index b5a7a5549..65080d6f6 100644 --- a/crates/compass-graph/src/snapshot.rs +++ b/crates/compass-graph/src/snapshot.rs @@ -19,9 +19,9 @@ use compass_model::code_graph::{ }; use compass_model::validate_code_graph; use compass_store::{ - ImmutableWrite, Key, MAX_IMMUTABLE_BATCH_BYTES, MAX_IMMUTABLE_BATCH_ITEMS, MAX_KEY_SEGMENTS, - MAX_SCAN_BYTES, MAX_SCAN_ITEMS, MAX_VALUE_BYTES, NamespaceId, PartitionKey, Store, StoreError, - WriteCondition, decode_key_segments, encode_key_segments, + ImmutableWrite, Key, MAX_GRAPH_BYTES, MAX_IMMUTABLE_BATCH_BYTES, MAX_IMMUTABLE_BATCH_ITEMS, + MAX_KEY_SEGMENTS, MAX_SCAN_BYTES, MAX_SCAN_ITEMS, MAX_VALUE_BYTES, NamespaceId, PartitionKey, + Store, StoreError, WriteCondition, decode_key_segments, encode_key_segments, max_graph_bytes, }; use rayon::prelude::*; use serde::{Deserialize, Serialize}; @@ -53,6 +53,14 @@ pub const GRAPH_SNAPSHOT_MAX_OBJECTS: usize = 100_000; pub const GRAPH_SNAPSHOT_MAX_ITEMS: usize = 5_000_000; pub const GRAPH_SNAPSHOT_MAX_FANOUT: usize = 32; pub const GRAPH_SNAPSHOT_MAX_LEAF_ENTRIES: usize = 128; +/// Default maximum size of the canonical graph published by a snapshot. +pub const MAX_CANONICAL_GRAPH_BYTES: u64 = MAX_GRAPH_BYTES as u64; +/// Effective canonical publication bound, including the explicit opt-in +/// `COMPASS_MAX_GRAPH_BYTES` override. +#[must_use] +pub fn max_canonical_graph_bytes() -> u64 { + max_graph_bytes() as u64 +} /// Maximum previous JSON artifact retained while attempting a byte-preserving /// fact-neutral publication. Larger artifacts use the bounded streaming /// serializer instead of adding another resident graph-sized buffer. @@ -113,6 +121,17 @@ pub enum SnapshotError { CapabilityUnavailable(String), } +impl SnapshotError { + /// Construct the stable, actionable failure for canonical graph + /// publication above a byte limit. + #[must_use] + pub fn canonical_graph_too_large(maximum: u64) -> Self { + Self::Limit(format!( + "canonical graph exceeds the {maximum}-byte limit; retry or rebuild with a smaller scope using --exclude or persistent patterns in .compassignore, or explicitly raise the bound with COMPASS_MAX_GRAPH_BYTES=" + )) + } +} + #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum IndexKind { @@ -1108,6 +1127,22 @@ pub fn write_canonical_graph_json( writer.write_all(b"}") } +/// Stream canonical graph JSON while enforcing the configured publication +/// limit against the bytes that are actually emitted. +/// +/// The limit is checked before each write reaches the destination. Callers +/// can therefore use this function inside an atomic staging callback without +/// ever publishing an oversized artifact or retaining a guessed source-size +/// multiplier. +pub fn write_canonical_graph_json_bounded( + graph: &GraphDocument, + writer: &mut W, + maximum: u64, +) -> io::Result<()> { + let mut bounded = CanonicalGraphLimitWriter::new(writer, maximum); + write_canonical_graph_json(graph, &mut bounded) +} + /// Publish a fact-neutral graph edit by reusing the previous canonical node /// and link bytes. Fact-neutral edits only update file-node metadata and graph /// inventory; the semantic node IDs and every relationship remain unchanged. @@ -1146,6 +1181,70 @@ pub fn write_fact_neutral_graph_json_delta_prevalidated( ) } +/// Publish a prevalidated fact-neutral delta while enforcing the canonical +/// graph byte limit against the actual emitted artifact. +pub fn write_fact_neutral_graph_json_delta_prevalidated_bounded( + previous_bytes: &[u8], + graph: &GraphDocument, + changed_node_ids: &BTreeSet, + writer: &mut W, + maximum: u64, +) -> io::Result { + let mut bounded = CanonicalGraphLimitWriter::new(writer, maximum); + write_fact_neutral_graph_json_delta_inner( + previous_bytes, + graph, + changed_node_ids, + false, + &mut bounded, + ) +} + +struct CanonicalGraphLimitWriter<'a, W: Write + ?Sized> { + inner: &'a mut W, + bytes: u64, + maximum: u64, +} + +impl<'a, W: Write + ?Sized> CanonicalGraphLimitWriter<'a, W> { + fn new(inner: &'a mut W, maximum: u64) -> Self { + Self { + inner, + bytes: 0, + maximum, + } + } +} + +impl Write for CanonicalGraphLimitWriter<'_, W> { + fn write(&mut self, buffer: &[u8]) -> io::Result { + let buffer_len = u64::try_from(buffer.len()) + .map_err(|_| io::Error::other("canonical graph byte count does not fit u64"))?; + let next = self + .bytes + .checked_add(buffer_len) + .ok_or_else(|| io::Error::other("canonical graph byte count exceeds u64"))?; + if next > self.maximum { + return Err(io::Error::new( + io::ErrorKind::FileTooLarge, + SnapshotError::canonical_graph_too_large(self.maximum), + )); + } + let written = self.inner.write(buffer)?; + self.bytes = self + .bytes + .checked_add(u64::try_from(written).map_err(|_| { + io::Error::other("canonical graph written byte count does not fit u64") + })?) + .ok_or_else(|| io::Error::other("canonical graph byte count exceeds u64"))?; + Ok(written) + } + + fn flush(&mut self) -> io::Result<()> { + self.inner.flush() + } +} + fn write_fact_neutral_graph_json_delta_inner( previous_bytes: &[u8], graph: &GraphDocument, diff --git a/crates/compass-store/src/lib.rs b/crates/compass-store/src/lib.rs index 637310529..5a76fc69a 100644 --- a/crates/compass-store/src/lib.rs +++ b/crates/compass-store/src/lib.rs @@ -48,6 +48,38 @@ pub const MAX_IMMUTABLE_BATCH_BYTES: usize = 16 * 1024 * 1024; /// `publish_snapshot`/`read_snapshot` path because that API materializes the /// canonical payload in one allocation. pub const MAX_GRAPH_BYTES: usize = 2 * 1024 * 1024 * 1024; + +/// Effective opt-in graph byte bound used by snapshot publication and reads. +/// Invalid, zero, overflowing, or unrepresentable values fail closed to the +/// 2 GiB default. Accepted forms match Compass graph readers: raw bytes, `MB`, +/// and `GB`, with optional underscores in the numeric component. +#[must_use] +pub fn max_graph_bytes() -> usize { + let raw = std::env::var("COMPASS_MAX_GRAPH_BYTES").ok(); + parse_max_graph_bytes(raw.as_deref()) +} + +fn parse_max_graph_bytes(raw: Option<&str>) -> usize { + let Some(raw) = raw else { + return MAX_GRAPH_BYTES; + }; + let upper = raw.trim().to_ascii_uppercase(); + let (number, multiplier) = if let Some(number) = upper.strip_suffix("GB") { + (number.trim(), 1024_u128 * 1024 * 1024) + } else if let Some(number) = upper.strip_suffix("MB") { + (number.trim(), 1024_u128 * 1024) + } else { + (upper.as_str(), 1) + }; + number + .replace('_', "") + .parse::() + .ok() + .filter(|value| *value > 0) + .and_then(|value| value.checked_mul(multiplier)) + .and_then(|value| usize::try_from(value).ok()) + .unwrap_or(MAX_GRAPH_BYTES) +} const GRAPH_NAMESPACE: &[u8] = b"compass.current.graph.v1"; const CATALOG_PARTITION: &[u8] = b"catalog"; const OBJECT_PARTITION: &[u8] = b"object"; diff --git a/scripts/qualify_graph_size_ratios.py b/scripts/qualify_graph_size_ratios.py new file mode 100755 index 000000000..8854d3c11 --- /dev/null +++ b/scripts/qualify_graph_size_ratios.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +"""Measure admitted-source to canonical-graph expansion without loading graphs.""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import statistics +import sys +import tempfile +from dataclasses import dataclass +from pathlib import Path + + +SCHEMA = "compass.qualification.graph-size-ratios/1" +DEFAULT_METADATA_LIMIT = 256 * 1024 * 1024 +NODES_MARKER = b',"nodes":[' +BYTE_SIZE = re.compile(rb'"byteSize":([0-9]+)') + + +@dataclass(frozen=True) +class Estate: + name: str + graph: Path + + +def parse_estate(values: list[str]) -> Estate: + name, graph = values + if not name or any(character.isspace() for character in name): + raise argparse.ArgumentTypeError("estate NAME must be non-empty and contain no whitespace") + return Estate(name=name, graph=Path(graph).resolve()) + + +def read_metadata_prefix(path: Path, maximum: int) -> bytes: + prefix = bytearray() + with path.open("rb") as stream: + while len(prefix) <= maximum: + block = stream.read(min(1024 * 1024, maximum + 1 - len(prefix))) + if not block: + break + prefix.extend(block) + marker = prefix.find(NODES_MARKER) + if marker >= 0: + return bytes(prefix[:marker]) + raise ValueError( + f"{path}: canonical graph metadata did not end within the " + f"{maximum}-byte qualification bound" + ) + + +def measure(estate: Estate, maximum: int) -> dict[str, object]: + if not estate.graph.is_file(): + raise ValueError(f"{estate.graph}: graph artifact does not exist") + prefix = read_metadata_prefix(estate.graph, maximum) + sizes = [int(match.group(1)) for match in BYTE_SIZE.finditer(prefix)] + if not sizes: + raise ValueError(f"{estate.graph}: canonical graph metadata has no admitted files") + source_bytes = sum(sizes) + graph_bytes = estate.graph.stat().st_size + if source_bytes <= 0 or graph_bytes <= 0: + raise ValueError(f"{estate.graph}: byte measurements must be positive") + return { + "estate": estate.name, + "graph": str(estate.graph), + "admittedFiles": len(sizes), + "admittedSourceBytes": source_bytes, + "canonicalGraphBytes": graph_bytes, + "graphToSourceRatio": graph_bytes / source_bytes, + } + + +def render_markdown(report: dict[str, object]) -> str: + rows = report["estates"] + summary = report["summary"] + lines = [ + "# Canonical graph size qualification", + "", + "The ratio is measured from each canonical artifact's admitted file inventory; " + "it is diagnostic evidence, not a publication predictor.", + "", + "| Estate | Files | Admitted source bytes | Canonical graph bytes | Ratio |", + "| --- | ---: | ---: | ---: | ---: |", + ] + for row in rows: + lines.append( + f"| {row['estate']} | {row['admittedFiles']:,} | " + f"{row['admittedSourceBytes']:,} | {row['canonicalGraphBytes']:,} | " + f"{row['graphToSourceRatio']:.6f}x |" + ) + lines.extend( + [ + "", + f"Distribution: minimum **{summary['minimumRatio']:.6f}x**, " + f"median **{summary['medianRatio']:.6f}x**, and " + f"maximum **{summary['maximumRatio']:.6f}x**.", + "", + ] + ) + return "\n".join(lines) + + +def write_atomic(path: Path, payload: bytes) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + try: + with os.fdopen(descriptor, "wb") as stream: + stream.write(payload) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + except BaseException: + try: + os.unlink(temporary) + except FileNotFoundError: + pass + raise + + +def main() -> int: + parser = argparse.ArgumentParser( + description="measure canonical graph bytes per admitted source byte" + ) + parser.add_argument( + "--estate", + nargs=2, + action="append", + metavar=("NAME", "GRAPH_JSON"), + required=True, + help="estate label and its completed canonical graph artifact; repeat exactly five times", + ) + parser.add_argument("--json-output", type=Path) + parser.add_argument("--markdown-output", type=Path) + parser.add_argument( + "--max-metadata-bytes", type=int, default=DEFAULT_METADATA_LIMIT + ) + arguments = parser.parse_args() + + estates = [parse_estate(values) for values in arguments.estate] + if len(estates) != 5: + parser.error(f"expected exactly five --estate entries, found {len(estates)}") + names = [estate.name for estate in estates] + if len(set(names)) != len(names): + parser.error("estate names must be unique") + if arguments.max_metadata_bytes <= 0: + parser.error("--max-metadata-bytes must be positive") + + try: + rows = sorted( + (measure(estate, arguments.max_metadata_bytes) for estate in estates), + key=lambda row: str(row["estate"]), + ) + except (OSError, ValueError) as error: + print(f"error: {error}", file=sys.stderr) + return 2 + + ratios = [float(row["graphToSourceRatio"]) for row in rows] + report: dict[str, object] = { + "schema": SCHEMA, + "estates": rows, + "summary": { + "minimumRatio": min(ratios), + "medianRatio": statistics.median(ratios), + "maximumRatio": max(ratios), + }, + } + json_payload = (json.dumps(report, indent=2, sort_keys=False) + "\n").encode() + markdown_payload = render_markdown(report).encode() + if arguments.json_output: + write_atomic(arguments.json_output.resolve(), json_payload) + if arguments.markdown_output: + write_atomic(arguments.markdown_output.resolve(), markdown_payload) + if not arguments.json_output and not arguments.markdown_output: + sys.stdout.buffer.write(markdown_payload) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())