feat(analysis): stage bounded JSON codec and iterative expression ownership

This commit is contained in:
Codex
2026-09-11 09:14:16 -07:00
parent cdebb49d89
commit 847b1caa7e
11 changed files with 1971 additions and 110 deletions
+2 -1
View File
@@ -7,7 +7,8 @@ rust-version.workspace = true
[dependencies]
serde.workspace = true
serde_json.workspace = true
serde_json = { workspace = true, features = ["unbounded_depth"] }
stacker = "0.1.25"
uuid.workspace = true
thiserror.workspace = true
tempfile = "3"
+2
View File
@@ -1,5 +1,7 @@
//! Schema 8 engineering definitions and static admission. No expression evaluation.
mod admission;
mod expression_lifecycle;
pub(crate) mod expression_serde;
pub mod references;
mod resolve;
mod typecheck;
@@ -0,0 +1,377 @@
//! Ordinary expression ownership never consumes stack proportional to tree depth.
//! This also covers a partially decoded expression and rejected programmatic input.
use super::Expr;
use std::{fmt, mem};
impl Expr {
fn clone_shell(&self) -> Self {
let leaf = || Box::new(Self::Boolean(false));
match self {
Self::Boolean(value) => Self::Boolean(*value),
Self::Number(value) => Self::Number(value.clone()),
Self::Quantity { magnitude, unit } => Self::Quantity {
magnitude: magnitude.clone(),
unit: *unit,
},
Self::Property(id) => Self::Property(*id),
Self::Parameter(id) => Self::Parameter(*id),
Self::List(values) => {
Self::List((0..values.len()).map(|_| Self::Boolean(false)).collect())
}
Self::Unary { op, .. } => Self::Unary {
op: *op,
operand: leaf(),
},
Self::Binary { op, .. } => Self::Binary {
op: *op,
left: leaf(),
right: leaf(),
},
Self::If { .. } => Self::If {
condition: leaf(),
when_true: leaf(),
when_false: leaf(),
},
Self::Call {
definition,
arguments,
} => Self::Call {
definition: *definition,
arguments: arguments
.keys()
.map(|id| (*id, Self::Boolean(false)))
.collect(),
},
Self::Sum { .. } => Self::Sum {
values: leaf(),
initial: leaf(),
},
Self::Power { exponent, .. } => Self::Power {
base: leaf(),
exponent: *exponent,
},
Self::AsKind { kind, .. } => Self::AsKind {
kind: *kind,
value: leaf(),
},
}
}
fn detach_children(&mut self, pending: &mut Vec<Self>) {
let mut take = |child: &mut Box<Self>| {
pending.push(mem::replace(child.as_mut(), Self::Boolean(false)));
};
match self {
Self::Unary { operand, .. } => take(operand),
Self::Binary { left, right, .. } => {
take(left);
take(right);
}
Self::If {
condition,
when_true,
when_false,
} => {
take(condition);
take(when_true);
take(when_false);
}
Self::Sum { values, initial } => {
take(values);
take(initial);
}
Self::Power { base, .. } => take(base),
Self::AsKind { value, .. } => take(value),
Self::List(values) => pending.append(values),
Self::Call { arguments, .. } => {
pending.extend(mem::take(arguments).into_values());
}
Self::Boolean(_)
| Self::Number(_)
| Self::Quantity { .. }
| Self::Property(_)
| Self::Parameter(_) => {}
}
}
}
impl Drop for Expr {
fn drop(&mut self) {
let mut pending = Vec::new();
self.detach_children(&mut pending);
while let Some(mut node) = pending.pop() {
node.detach_children(&mut pending);
// node now owns only leaves/empty collections. Its ordinary drop
// reenters this method at most once, independently of original depth.
}
}
}
impl Clone for Expr {
fn clone(&self) -> Self {
let mut result = Self::Boolean(false);
let mut pending = vec![(self, &mut result)];
while let Some((source, destination)) = pending.pop() {
*destination = source.clone_shell();
match (source, destination) {
(Self::List(a), Self::List(b)) => pending.extend(a.iter().zip(b.iter_mut())),
(Self::Unary { operand: a, .. }, Self::Unary { operand: b, .. })
| (Self::Power { base: a, .. }, Self::Power { base: b, .. })
| (Self::AsKind { value: a, .. }, Self::AsKind { value: b, .. }) => {
pending.push((a, b));
}
(
Self::Binary {
left: a, right: b, ..
},
Self::Binary {
left: c, right: d, ..
},
)
| (
Self::Sum {
values: a,
initial: b,
},
Self::Sum {
values: c,
initial: d,
},
) => {
pending.push((a, c));
pending.push((b, d));
}
(
Self::If {
condition: a,
when_true: b,
when_false: c,
},
Self::If {
condition: d,
when_true: e,
when_false: f,
},
) => {
pending.push((a, d));
pending.push((b, e));
pending.push((c, f));
}
(Self::Call { arguments: a, .. }, Self::Call { arguments: b, .. }) => {
pending.extend(a.values().zip(b.values_mut()));
}
_ => {}
}
}
result
}
}
impl PartialEq for Expr {
fn eq(&self, other: &Self) -> bool {
let mut pending = vec![(self, other)];
while let Some((a, b)) = pending.pop() {
match (a, b) {
(Self::Boolean(a), Self::Boolean(b)) if a == b => {}
(Self::Number(a), Self::Number(b)) if a == b => {}
(
Self::Quantity {
magnitude: a,
unit: b,
},
Self::Quantity {
magnitude: c,
unit: d,
},
) if a == c && b == d => {}
(Self::Property(a), Self::Property(b)) if a == b => {}
(Self::Parameter(a), Self::Parameter(b)) if a == b => {}
(Self::List(a), Self::List(b)) if a.len() == b.len() => {
pending.extend(a.iter().zip(b.iter()));
}
(Self::Unary { op: a, operand: b }, Self::Unary { op: c, operand: d })
if a == c =>
{
pending.push((b, d));
}
(
Self::Binary {
op: a,
left: b,
right: c,
},
Self::Binary {
op: d,
left: e,
right: f,
},
) if a == d => {
pending.push((b, e));
pending.push((c, f));
}
(
Self::If {
condition: a,
when_true: b,
when_false: c,
},
Self::If {
condition: d,
when_true: e,
when_false: f,
},
) => {
pending.push((a, d));
pending.push((b, e));
pending.push((c, f));
}
(
Self::Call {
definition: a,
arguments: b,
},
Self::Call {
definition: c,
arguments: d,
},
) if a == c && b.len() == d.len() => {
for ((key_a, value_a), (key_b, value_b)) in b.iter().zip(d.iter()) {
if key_a != key_b {
return false;
}
pending.push((value_a, value_b));
}
}
(
Self::Sum {
values: a,
initial: b,
},
Self::Sum {
values: c,
initial: d,
},
) => {
pending.push((a, c));
pending.push((b, d));
}
(
Self::Power {
base: a,
exponent: b,
},
Self::Power {
base: c,
exponent: d,
},
) if b == d => {
pending.push((a, c));
}
(Self::AsKind { kind: a, value: b }, Self::AsKind { kind: c, value: d })
if a == c =>
{
pending.push((b, d));
}
_ => return false,
}
}
true
}
}
impl Eq for Expr {}
impl fmt::Debug for Expr {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
enum Task<'a> {
Node(&'a Expr),
Text(&'static str),
Scalar(&'a dyn fmt::Debug),
}
use Task::*;
let mut pending = vec![Node(self)];
while let Some(task) = pending.pop() {
match task {
Text(text) => f.write_str(text)?,
Scalar(value) => write!(f, "{value:?}")?,
Node(node) => match node {
Self::Boolean(value) => write!(f, "Boolean({value:?})")?,
Self::Number(value) => write!(f, "Number({value:?})")?,
Self::Quantity { magnitude, unit } => {
write!(f, "Quantity {{ magnitude: {magnitude:?}, unit: {unit:?} }}")?;
}
Self::Property(id) => write!(f, "Property({id:?})")?,
Self::Parameter(id) => write!(f, "Parameter({id:?})")?,
Self::List(values) => {
f.write_str("List([")?;
pending.push(Text("])"));
for (index, value) in values.iter().enumerate().rev() {
pending.push(Node(value));
if index > 0 {
pending.push(Text(", "));
}
}
}
Self::Unary { op, operand } => {
write!(f, "Unary {{ op: {op:?}, operand: ")?;
pending.extend([Text(" }"), Node(operand)]);
}
Self::Binary { op, left, right } => {
write!(f, "Binary {{ op: {op:?}, left: ")?;
pending.extend([Text(" }"), Node(right), Text(", right: "), Node(left)]);
}
Self::If {
condition,
when_true,
when_false,
} => {
f.write_str("If { condition: ")?;
pending.extend([
Text(" }"),
Node(when_false),
Text(", when_false: "),
Node(when_true),
Text(", when_true: "),
Node(condition),
]);
}
Self::Call {
definition,
arguments,
} => {
write!(f, "Call {{ definition: {definition:?}, arguments: {{")?;
pending.push(Text("} }"));
for (index, (id, value)) in arguments.iter().enumerate().rev() {
pending.extend([Node(value), Text(": "), Scalar(id)]);
if index > 0 {
pending.push(Text(", "));
}
}
}
Self::Sum { values, initial } => {
f.write_str("Sum { values: ")?;
pending.extend([
Text(" }"),
Node(initial),
Text(", initial: "),
Node(values),
]);
}
Self::Power { base, exponent } => {
f.write_str("Power { base: ")?;
pending.extend([
Text(" }"),
Scalar(exponent),
Text(", exponent: "),
Node(base),
]);
}
Self::AsKind { kind, value } => {
write!(f, "AsKind {{ kind: {kind:?}, value: ")?;
pending.extend([Text(" }"), Node(value)]);
}
},
}
}
Ok(())
}
}
@@ -0,0 +1,261 @@
//! The existing externally tagged Serde schema, with admission at each Expr edge.
//! The private remote derive is the sole wire grammar: it still rejects unknown
//! fields/variants, duplicate fields and malformed UUIDs/Exact values.
use super::{
BinaryOp, DefinitionRef, Expr, MAX_CLOSURE_NODES, MAX_EXPRESSION_DEPTH, MAX_LIST_LENGTH,
ParameterId, UnaryOp,
};
use crate::{Id, analysis_numeric::Exact};
use serde::{
Deserialize, Serialize,
de::{DeserializeSeed, MapAccess, SeqAccess, Visitor},
};
use std::{cell::Cell, collections::BTreeMap, fmt};
pub(crate) const MAX_JSON_EXPRESSION_NODES: usize = 262_144;
// These are requested segment capacities, not an allocator or RSS guarantee.
const RED_ZONE: usize = 64 * 1024;
const STACK_SEGMENT: usize = 512 * 1024;
pub(crate) fn with_stack<R>(operation: impl FnOnce() -> R) -> R {
stacker::maybe_grow(RED_ZONE, STACK_SEGMENT, operation)
}
#[derive(Clone, Copy)]
struct Meter {
active: bool,
depth: usize,
nodes: usize,
tree_nodes: usize,
maximum: usize,
}
impl Meter {
const DEFAULT: Self = Self {
active: false,
depth: 0,
nodes: 0,
tree_nodes: 0,
maximum: MAX_JSON_EXPRESSION_NODES,
};
}
thread_local! {
static DECODING: Cell<Meter> = const { Cell::new(Meter::DEFAULT) };
static ENCODING: Cell<Meter> = const { Cell::new(Meter::DEFAULT) };
}
type LocalMeter = &'static std::thread::LocalKey<Cell<Meter>>;
/// One codec operation owns the aggregate meter across all disjoint bindings.
/// Scope restoration also runs on errors and unwinding; no mutable TLS borrow is
/// held across Serde/user code or a stack switch.
pub(crate) struct Scope {
local: LocalMeter,
previous: Meter,
}
impl Scope {
pub(crate) fn decoding(maximum: usize) -> Self {
Self::new(&DECODING, maximum)
}
pub(crate) fn encoding(maximum: usize) -> Self {
Self::new(&ENCODING, maximum)
}
fn new(local: LocalMeter, maximum: usize) -> Self {
let next = Meter {
active: true,
maximum: maximum.min(MAX_JSON_EXPRESSION_NODES),
..Meter::DEFAULT
};
let previous = local.with(|meter| meter.replace(next));
Self { local, previous }
}
}
impl Drop for Scope {
fn drop(&mut self) {
self.local.with(|meter| meter.set(self.previous));
}
}
struct NodeGuard {
local: LocalMeter,
root: bool,
}
impl NodeGuard {
fn enter(local: LocalMeter) -> Result<Self, &'static str> {
local.with(|slot| {
let mut meter = slot.get();
let root = !meter.active;
if root {
meter = Meter {
active: true,
..Meter::DEFAULT
};
}
if meter.depth == 0 {
meter.tree_nodes = 0;
}
if meter.depth >= MAX_EXPRESSION_DEPTH {
return Err("Expression logical depth exceeds 128.");
}
if meter.nodes >= meter.maximum {
return Err("JSON expression allocation count exceeded.");
}
if meter.tree_nodes >= MAX_CLOSURE_NODES {
return Err("Stored expression node count exceeds 65536.");
}
meter.depth += 1;
meter.nodes += 1;
meter.tree_nodes += 1;
slot.set(meter);
Ok(Self { local, root })
})
}
}
impl Drop for NodeGuard {
fn drop(&mut self) {
self.local.with(|slot| {
let mut meter = slot.get();
meter.depth -= 1;
if self.root {
meter.active = false;
}
slot.set(meter);
});
}
}
impl<'de> Deserialize<'de> for Expr {
fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
let _node = NodeGuard::enter(&DECODING).map_err(serde::de::Error::custom)?;
with_stack(|| WireExpr::deserialize(deserializer))
}
}
impl Serialize for Expr {
fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
let _node = NodeGuard::enter(&ENCODING).map_err(serde::ser::Error::custom)?;
match self {
Self::List(values) if values.len() > MAX_LIST_LENGTH => {
return Err(serde::ser::Error::custom(
"Expression List exceeds 1024 items.",
));
}
Self::Call { arguments, .. } if arguments.len() > MAX_CLOSURE_NODES => {
return Err(serde::ser::Error::custom(
"Call argument count exceeds 65536.",
));
}
_ => {}
}
with_stack(|| WireExpr::serialize(self, serializer))
}
}
// A private remote derive avoids exposing an unguarded inherent deserialize on
// public Expr. Children remain Expr, so every recursive edge enters the guard.
#[derive(Serialize, Deserialize)]
#[serde(remote = "Expr", deny_unknown_fields)]
enum WireExpr {
Boolean(bool),
Number(Exact),
Quantity {
magnitude: Exact,
unit: DefinitionRef,
},
Property(Id),
Parameter(ParameterId),
List(#[serde(deserialize_with = "bounded_list")] Vec<Expr>),
Unary {
op: UnaryOp,
operand: Box<Expr>,
},
Binary {
op: BinaryOp,
left: Box<Expr>,
right: Box<Expr>,
},
If {
condition: Box<Expr>,
when_true: Box<Expr>,
when_false: Box<Expr>,
},
Call {
definition: DefinitionRef,
#[serde(deserialize_with = "bounded_arguments")]
arguments: BTreeMap<ParameterId, Expr>,
},
Sum {
values: Box<Expr>,
initial: Box<Expr>,
},
Power {
base: Box<Expr>,
exponent: i8,
},
AsKind {
kind: DefinitionRef,
value: Box<Expr>,
},
}
struct ListItem {
allowed: bool,
}
impl<'de> DeserializeSeed<'de> for ListItem {
type Value = Expr;
fn deserialize<D: serde::Deserializer<'de>>(self, deserializer: D) -> Result<Expr, D::Error> {
if !self.allowed {
return Err(serde::de::Error::custom(
"Expression List exceeds 1024 items.",
));
}
Expr::deserialize(deserializer)
}
}
fn bounded_list<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Result<Vec<Expr>, D::Error> {
struct ListVisitor;
impl<'de> Visitor<'de> for ListVisitor {
type Value = Vec<Expr>;
fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("at most 1024 expression items")
}
fn visit_seq<A: SeqAccess<'de>>(self, mut access: A) -> Result<Self::Value, A::Error> {
let mut values = Vec::new();
while let Some(value) = access.next_element_seed(ListItem {
allowed: values.len() < MAX_LIST_LENGTH,
})? {
values.push(value);
}
Ok(values)
}
}
deserializer.deserialize_seq(ListVisitor)
}
fn bounded_arguments<'de, D: serde::Deserializer<'de>>(
deserializer: D,
) -> Result<BTreeMap<ParameterId, Expr>, D::Error> {
struct Arguments;
impl<'de> Visitor<'de> for Arguments {
type Value = BTreeMap<ParameterId, Expr>;
fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("unique scoped parameter UUID keys")
}
fn visit_map<A: MapAccess<'de>>(self, mut access: A) -> Result<Self::Value, A::Error> {
let mut values = BTreeMap::new();
while let Some(id) = access.next_key::<ParameterId>()? {
if values.contains_key(&id) {
return Err(serde::de::Error::custom(format!(
"Duplicate parameter ID {}",
id.0
)));
}
if values.len() >= MAX_CLOSURE_NODES {
return Err(serde::de::Error::custom(
"Call argument count exceeds 65536.",
));
}
// Count/duplicate checks precede construction of the next subtree.
values.insert(id, access.next_value()?);
}
Ok(values)
}
}
deserializer.deserialize_map(Arguments)
}
-13
View File
@@ -114,8 +114,6 @@ pub enum BinaryOp {
And,
Or,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub enum Expr {
Boolean(bool),
Number(Exact),
@@ -142,7 +140,6 @@ pub enum Expr {
},
Call {
definition: DefinitionRef,
#[serde(deserialize_with = "unique_parameter_map")]
arguments: BTreeMap<ParameterId, Expr>,
},
Sum {
@@ -159,16 +156,6 @@ pub enum Expr {
},
}
fn unique_parameter_map<'de, D: serde::Deserializer<'de>>(
deserializer: D,
) -> Result<BTreeMap<ParameterId, Expr>, D::Error> {
crate::unique_id_map(deserializer).map(|map| {
map.into_iter()
.map(|(id, expression)| (ParameterId(id), expression))
.collect()
})
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub enum AnalysisSubject {
+150 -96
View File
@@ -2,6 +2,7 @@ use crate::*;
use serde::{Deserialize, Serialize};
use std::{io::Write, path::Path};
use thiserror::Error;
mod json_codec;
#[path = "migration_v2/mod.rs"]
pub(crate) mod v2;
pub(crate) use v2::adapter::migrate as migrate_v2_limited;
@@ -132,6 +133,19 @@ pub fn decode_limited(
pub fn decode_with_report_limited(
bytes: &[u8],
limits: crate::language::ParseLimits,
) -> Result<DecodeResult, FileError> {
decode_with_report_profile(
bytes,
limits,
json_codec::Profile::with_byte_limit(limits.max_source_bytes),
json_codec::DEPTH_128_VALIDATED,
)
}
fn decode_with_report_profile(
bytes: &[u8],
limits: crate::language::ParseLimits,
profile: json_codec::Profile,
depth_128: bool,
) -> Result<DecodeResult, FileError> {
if bytes.len() > limits.max_source_bytes {
return Err(FileError::Limit(format!(
@@ -140,66 +154,43 @@ pub fn decode_with_report_limited(
)));
}
let decoded = if bytes.iter().find(|b| !b.is_ascii_whitespace()) == Some(&b'{') {
let header: serde_json::Value = serde_json::from_slice(bytes)?;
if header
.get("schema_version")
.and_then(serde_json::Value::as_u64)
.is_some_and(|v| v <= 6)
{
reject_source7_fields(&header).map_err(FileError::Invalid)?;
json_codec::preflight(bytes, profile)?;
let version = json_codec::version(bytes)?;
if version.is_some_and(|version| version <= 6) {
let header = json_codec::legacy(bytes)?;
let result = check_legacy_header(&header, version);
json_codec::discard_value(header);
result?;
}
if header
.get("schema_version")
.and_then(serde_json::Value::as_u64)
.is_some_and(|v| v <= 5)
&& let Some(diagrams) = header
.get("diagrams")
.and_then(serde_json::Value::as_object)
{
for diagram in diagrams.values().filter_map(serde_json::Value::as_object) {
for key in diagram.keys() {
if key == "view_options" {
return Err(FileError::Invalid("Diagram.view_options requires schema/source 6, including an explicitly supplied default value.".into()));
}
if !matches!(
key.as_str(),
"id" | "alias"
| "name"
| "description"
| "layer"
| "projection"
| "nodes"
| "routes"
) {
return Err(FileError::Invalid(format!(
"Unknown diagram field '{key}' in the frozen pre6 contract."
)));
}
}
}
}
match header
.get("schema_version")
.and_then(serde_json::Value::as_u64)
{
match version {
Some(1) => {
let old = serde_json::from_slice(bytes)?;
let old = json_codec::legacy(bytes)?;
check_v1_limits(&old, limits).map_err(FileError::Limit)?;
migrate_v1_with_report_limited(old, limits)?
}
Some(2) => migrate_v2_limited(serde_json::from_slice(bytes)?, limits)?,
Some(3) => v3::migrate(serde_json::from_slice(bytes)?, limits)?,
Some(4) => v4::migrate(serde_json::from_slice(bytes)?, limits)?,
Some(5) => migrate_v5(serde_json::from_slice::<v7::Project7>(bytes)?.into_current()?, limits)?,
Some(6) => migrate_v6(serde_json::from_slice::<v7::Project7>(bytes)?.into_current()?, limits)?,
Some(7) => migrate_v7(serde_json::from_slice::<v7::Project7>(bytes)?.into_current()?, limits)?,
Some(2) => migrate_v2_limited(json_codec::legacy(bytes)?, limits)?,
Some(3) => v3::migrate(json_codec::legacy(bytes)?, limits)?,
Some(4) => v4::migrate(json_codec::legacy(bytes)?, limits)?,
Some(5) => migrate_v5(
json_codec::legacy::<v7::Project7>(bytes)?.into_current()?,
limits,
)?,
Some(6) => migrate_v6(
json_codec::legacy::<v7::Project7>(bytes)?.into_current()?,
limits,
)?,
Some(7) => migrate_v7(
json_codec::legacy::<v7::Project7>(bytes)?.into_current()?,
limits,
)?,
Some(8) => DecodeResult {
project: serde_json::from_slice(bytes)?,
project: json_codec::current(bytes, profile, depth_128)?,
migration: None,
},
_ => {
return Err(FileError::Invalid(
"Unsupported or missing schema_version; expected 1, 2, 3, 4, 5, 6, 7 or 8.".into(),
"Unsupported or missing schema_version; expected 1, 2, 3, 4, 5, 6, 7 or 8."
.into(),
));
}
}
@@ -228,6 +219,38 @@ pub fn decode_with_report_limited(
Ok(decoded)
}
fn check_legacy_header(header: &serde_json::Value, version: Option<u64>) -> Result<(), FileError> {
reject_source7_fields(header).map_err(FileError::Invalid)?;
if version.is_some_and(|version| version <= 5)
&& let Some(diagrams) = header
.get("diagrams")
.and_then(serde_json::Value::as_object)
{
for diagram in diagrams.values().filter_map(serde_json::Value::as_object) {
for key in diagram.keys() {
if key == "view_options" {
return Err(FileError::Invalid("Diagram.view_options requires schema/source 6, including an explicitly supplied default value.".into()));
}
if !matches!(
key.as_str(),
"id" | "alias"
| "name"
| "description"
| "layer"
| "projection"
| "nodes"
| "routes"
) {
return Err(FileError::Invalid(format!(
"Unknown diagram field '{key}' in the frozen pre6 contract."
)));
}
}
}
}
Ok(())
}
/// Versions 4 and 5 share unchanged typed records, but their constructor set
/// remains frozen. Apply this before normalizing the advertised version.
pub(crate) fn pre6_relation_kind(kind: RelationKind) -> bool {
@@ -372,48 +395,49 @@ pub(crate) fn migrate_v5(
/// Reject explicit new fields before serde defaults can erase their presence.
/// This is also applied to source's resolved structured values.
pub(crate) fn reject_source7_fields(value: &serde_json::Value) -> Result<(), String> {
match value {
serde_json::Value::Object(fields) => {
if fields.contains_key("ContextualLink") {
return Err("ContextualLink requires schema/source 7.".into());
let mut pending = vec![value];
while let Some(value) = pending.pop() {
match value {
serde_json::Value::Object(fields) => {
if fields.contains_key("ContextualLink") {
return Err("ContextualLink requires schema/source 7.".into());
}
if fields.get("Relation").and_then(serde_json::Value::as_str)
== Some("ContextualLink")
{
return Err("ContextualLink view selectors require schema/source 7.".into());
}
if fields.keys().any(|key| {
matches!(
key.as_str(),
"Operation"
| "ContextOwner"
| "ContextSource"
| "ContextTarget"
| "ContextPurpose"
)
}) {
return Err(
"Operation/contextual-link semantic facets require schema/source 7.".into(),
);
}
if fields
.get("Exchange")
.and_then(serde_json::Value::as_object)
.is_some_and(|body| body.contains_key("operation"))
{
return Err(
"Exchange.operation requires schema/source 7, including explicit null."
.into(),
);
}
pending.extend(fields.values().rev());
}
if fields.get("Relation").and_then(serde_json::Value::as_str) == Some("ContextualLink")
{
return Err("ContextualLink view selectors require schema/source 7.".into());
}
if fields.keys().any(|key| {
matches!(
key.as_str(),
"Operation"
| "ContextOwner"
| "ContextSource"
| "ContextTarget"
| "ContextPurpose"
)
}) {
return Err(
"Operation/contextual-link semantic facets require schema/source 7.".into(),
);
}
if fields
.get("Exchange")
.and_then(serde_json::Value::as_object)
.is_some_and(|body| body.contains_key("operation"))
{
return Err(
"Exchange.operation requires schema/source 7, including explicit null.".into(),
);
}
for nested in fields.values() {
reject_source7_fields(nested)?;
serde_json::Value::Array(values) => {
pending.extend(values.iter().rev());
}
_ => {}
}
serde_json::Value::Array(values) => {
for nested in values {
reject_source7_fields(nested)?;
}
}
_ => {}
}
Ok(())
}
@@ -438,14 +462,23 @@ pub(crate) fn migrate_v7(
mut project: Project,
limits: crate::language::ParseLimits,
) -> Result<DecodeResult, FileError> {
if project.schema_version != 7 { return Err(FileError::Invalid("Expected schema_version 7.".into())); }
if project.schema_version != 7 {
return Err(FileError::Invalid("Expected schema_version 7.".into()));
}
reject_pre8_project(&project)?;
limits.check_model(&project).map_err(FileError::Limit)?;
project.schema_version = SCHEMA_VERSION;
Ok(DecodeResult { project, migration: Some(MigrationReport::data_upgrade(7)) })
Ok(DecodeResult {
project,
migration: Some(MigrationReport::data_upgrade(7)),
})
}
fn reject_pre8_project(project: &Project) -> Result<(), FileError> {
if !project.definitions.is_empty() { return Err(FileError::Invalid("Definitions require schema/source 8.".into())); }
if !project.definitions.is_empty() {
return Err(FileError::Invalid(
"Definitions require schema/source 8.".into(),
));
}
v7::reject_elements(project.elements.values())
}
@@ -866,6 +899,9 @@ pub(crate) mod v1 {
}
fn validate_project(p: &Project) -> Result<(), FileError> {
json_codec::document_stack(|| validate_project_inner(p))
}
fn validate_project_inner(p: &Project) -> Result<(), FileError> {
if let Some(i) = validation::validate(p)
.into_iter()
.find(|i| i.severity == validation::Severity::Error)
@@ -881,7 +917,10 @@ pub fn load(path: &Path) -> Result<Project, FileError> {
load_with_report(path).map(|decoded| decoded.project)
}
pub fn load_with_report(path: &Path) -> Result<DecodeResult, FileError> {
decode_with_report(&std::fs::read(path)?)
decode_with_report(&json_codec::read_bounded(
path,
json_codec::MAX_DOCUMENT_BYTES,
)?)
}
/// Serialize to a sibling temporary file, flush, then atomically replace the destination.
pub fn save(path: &Path, p: &Project) -> Result<(), FileError> {
@@ -894,6 +933,21 @@ pub fn save_checked(
path: &Path,
p: &Project,
expected: Option<&[u8]>,
) -> Result<Vec<u8>, FileError> {
save_checked_profile(
path,
p,
expected,
json_codec::Profile::DOCUMENT,
json_codec::DEPTH_128_VALIDATED,
)
}
fn save_checked_profile(
path: &Path,
p: &Project,
expected: Option<&[u8]>,
profile: json_codec::Profile,
depth_128: bool,
) -> Result<Vec<u8>, FileError> {
validate_project(p)?;
let mut bytes = if path
@@ -902,7 +956,7 @@ pub fn save_checked(
{
crate::language::format(p).into_bytes()
} else {
serde_json::to_vec_pretty(p)?
json_codec::encode(p, profile, depth_128)?
};
if !bytes.ends_with(b"\n") {
bytes.push(b'\n');
@@ -915,8 +969,8 @@ pub fn save_checked(
temp.write_all(&bytes)?;
temp.as_file().sync_all()?;
if let Some(expected) = expected {
match std::fs::read(path) {
Ok(current) if current == expected => {}
match json_codec::matches_file(path, expected) {
Ok(true) => {}
Ok(_) => return Err(FileError::Conflict),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Err(FileError::Conflict),
Err(e) => return Err(FileError::Io(e)),
@@ -0,0 +1,427 @@
//! Private persistence JSON boundary. Scan original bytes before Serde allocates;
//! the scanner is a resource meter, never a second accepting JSON grammar.
use super::FileError;
use crate::{
Project,
analysis::expression_serde::{self, Scope},
};
use serde::{
Deserialize,
de::{IgnoredAny, MapAccess, Visitor},
};
use std::{
fmt,
io::{self, Read, Write},
};
pub(super) const MAX_DOCUMENT_BYTES: usize = 64 * 1024 * 1024;
const MAX_CONTAINERS: usize = 512;
// This switch stays false until Windows/Linux debug/release codec + ordinary
// lifecycle gates pass. Tests exercise the candidate through the same private
// entry point with depth_128=true. It is not a user-controlled parser option.
pub(super) const DEPTH_128_VALIDATED: bool = false;
/// Allocation cardinalities, independent of graph admission and numeric work.
/// The default deliberately admits the existing CLI's 100000 graph objects,
/// 250000 records and 500000 ordered references within its 64 MiB envelope.
/// Strings have no new 1 MiB restriction: any span fitting that envelope fits.
/// These are explicit slot/text bounds, not an assertion of process RSS.
#[derive(Clone, Copy, Debug)]
pub(super) struct Profile {
pub bytes: usize,
pub tokens: usize,
pub entries: usize,
pub container_entries: usize,
pub string_bytes: usize,
pub total_string_bytes: usize,
pub expression_nodes: usize,
}
impl Profile {
pub const DOCUMENT: Self = Self {
bytes: MAX_DOCUMENT_BYTES,
tokens: 16 * 1024 * 1024,
entries: 8 * 1024 * 1024,
container_entries: 500_000,
string_bytes: MAX_DOCUMENT_BYTES,
total_string_bytes: MAX_DOCUMENT_BYTES,
expression_nodes: expression_serde::MAX_JSON_EXPRESSION_NODES,
};
pub fn with_byte_limit(maximum: usize) -> Self {
Self {
bytes: maximum.min(MAX_DOCUMENT_BYTES),
..Self::DOCUMENT
}
}
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(super) struct Usage {
pub tokens: usize,
pub entries: usize,
pub string_bytes: usize,
pub deepest_container: usize,
}
#[derive(Clone, Copy)]
struct Container {
delimiter: u8,
entries: usize,
first: bool,
}
const EMPTY: Container = Container {
delimiter: 0,
entries: 0,
first: true,
};
fn limit(message: &str) -> FileError {
FileError::Limit(message.into())
}
fn charge(used: &mut usize, amount: usize, maximum: usize, message: &str) -> Result<(), FileError> {
if amount > maximum.saturating_sub(*used) {
return Err(limit(message));
}
*used += amount;
Ok(())
}
pub(super) fn preflight(bytes: &[u8], profile: Profile) -> Result<Usage, FileError> {
if bytes.len() > profile.bytes {
return Err(limit("JSON input byte limit exceeded."));
}
// Fixed storage: no Vec/Value/String and no recursive calls during preflight.
let mut containers = [EMPTY; MAX_CONTAINERS];
let mut depth = 0usize;
let mut usage = Usage::default();
let mut cursor = 0usize;
while cursor < bytes.len() {
let byte = bytes[cursor];
if byte.is_ascii_whitespace() {
cursor += 1;
continue;
}
charge(
&mut usage.tokens,
1,
profile.tokens,
"JSON token limit exceeded.",
)?;
if !matches!(byte, b',' | b':' | b'}' | b']') && depth > 0 {
let frame = &mut containers[depth - 1];
if frame.first {
charge(
&mut frame.entries,
1,
profile.container_entries,
"JSON container entry limit exceeded.",
)?;
charge(
&mut usage.entries,
1,
profile.entries,
"JSON aggregate entry limit exceeded.",
)?;
frame.first = false;
}
}
match byte {
b'{' | b'[' => {
if depth == MAX_CONTAINERS {
return Err(limit("JSON container depth exceeds 512."));
}
containers[depth] = Container {
delimiter: byte,
..EMPTY
};
depth += 1;
usage.deepest_container = usage.deepest_container.max(depth);
cursor += 1;
}
b'}' | b']' => {
if depth == 0
|| !matches!(
(containers[depth - 1].delimiter, byte),
(b'{', b'}') | (b'[', b']')
)
{
return Err(FileError::Invalid(
"Mismatched JSON container delimiter.".into(),
));
}
depth -= 1;
cursor += 1;
}
b',' => {
if depth > 0 {
containers[depth - 1].first = true;
}
cursor += 1;
}
b':' => cursor += 1,
b'"' => {
cursor += 1;
let start = cursor;
let mut closed = false;
while cursor < bytes.len() {
match bytes[cursor] {
b'"' => {
closed = true;
break;
}
b'\\' => {
// Skip exactly one escaped byte, including an escaped
// backslash. Serde validates actual escapes/UTF-8.
cursor += 1;
if cursor < bytes.len() {
cursor += 1;
}
}
_ => cursor += 1,
}
}
let span = cursor - start;
if span > profile.string_bytes {
return Err(limit("JSON string byte limit exceeded."));
}
charge(
&mut usage.string_bytes,
span,
profile.total_string_bytes,
"JSON aggregate string byte limit exceeded.",
)?;
if !closed {
return Err(FileError::Invalid("Unterminated JSON string.".into()));
}
cursor += 1;
}
_ => {
// Count a scalar token without interpreting its value. Invalid
// tokens and invalid whitespace must still fail the Serde pass.
cursor += 1;
while cursor < bytes.len()
&& !bytes[cursor].is_ascii_whitespace()
&& !matches!(
bytes[cursor],
b'{' | b'[' | b'}' | b']' | b',' | b':' | b'"'
)
{
cursor += 1;
}
}
}
}
if depth != 0 {
return Err(FileError::Invalid("Unclosed JSON container.".into()));
}
Ok(usage)
}
struct Version(Option<u64>);
impl<'de> Deserialize<'de> for Version {
fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
struct Header;
impl<'de> Visitor<'de> for Header {
type Value = Version;
fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("a project header")
}
fn visit_map<A: MapAccess<'de>>(self, mut access: A) -> Result<Version, A::Error> {
#[derive(Deserialize)]
#[serde(field_identifier)]
enum Key {
#[serde(rename = "schema_version")]
Schema,
#[serde(other)]
Other,
}
let mut version = None;
while let Some(key) = access.next_key::<Key>()? {
match key {
Key::Schema => {
if version.is_some() {
return Err(serde::de::Error::duplicate_field("schema_version"));
}
version = Some(access.next_value()?);
}
Key::Other => {
access.next_value::<IgnoredAny>()?;
}
}
}
Ok(Version(version))
}
}
deserializer.deserialize_map(Header)
}
}
/// IgnoredAny in serde_json walks arbitrary ignored containers iteratively.
/// This probe does not materialize ignored subtrees. Serde's scratch buffer for
/// container markers/escaped keys is bounded by the preceding raw scan. It
/// consumes the entire original stream and rejects duplicate/escaped versions.
pub(super) fn version(bytes: &[u8]) -> Result<Option<u64>, FileError> {
document_stack(|| {
let mut deserializer = serde_json::Deserializer::from_slice(bytes);
let version = Version::deserialize(&mut deserializer)?;
deserializer.end()?;
Ok(version.0)
})
}
pub(super) fn document_stack<R>(operation: impl FnOnce() -> R) -> R {
// Non-Expr Project and historical schemas are finitely nested by their
// concrete Rust types. Reserve for that outer generated Serde frame chain;
// only recursive Expr edges use the smaller repeated boundary guard.
stacker::maybe_grow(512 * 1024, 2 * 1024 * 1024, operation)
}
pub(super) fn current(
bytes: &[u8],
profile: Profile,
depth_128: bool,
) -> Result<Project, FileError> {
preflight(bytes, profile)?;
let _scope = Scope::decoding(profile.expression_nodes);
document_stack(|| {
let mut deserializer = serde_json::Deserializer::from_slice(bytes);
if depth_128 {
deserializer.disable_recursion_limit();
}
let project = Project::deserialize(&mut deserializer)?;
deserializer.end()?;
Ok(project)
})
}
pub(super) fn legacy<T: serde::de::DeserializeOwned>(bytes: &[u8]) -> Result<T, FileError> {
// Never disable the legacy recursion guard, including its temporary Value.
document_stack(|| Ok(serde_json::from_slice(bytes)?))
}
pub(super) fn discard_value(value: serde_json::Value) {
let mut pending = vec![value];
while let Some(value) = pending.pop() {
match value {
serde_json::Value::Array(values) => pending.extend(values),
serde_json::Value::Object(values) => pending.extend(values.into_values()),
_ => {}
}
}
}
fn reserve_bounded(bytes: &mut Vec<u8>, additional: usize, maximum: usize) -> io::Result<()> {
let required = bytes.len() + additional;
if required > bytes.capacity() {
let next = bytes
.capacity()
.max(4096)
.saturating_mul(2)
.max(required)
.min(maximum);
bytes
.try_reserve_exact(next - bytes.len())
.map_err(io::Error::other)?;
}
Ok(())
}
struct CappedWriter {
bytes: Vec<u8>,
maximum: usize,
exceeded: bool,
}
impl CappedWriter {
fn new(maximum: usize) -> Self {
Self {
bytes: Vec::new(),
maximum,
exceeded: false,
}
}
}
impl Write for CappedWriter {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
if bytes.len() > self.maximum.saturating_sub(self.bytes.len()) {
self.exceeded = true;
return Err(io::Error::other("JSON output byte limit exceeded."));
}
// Check before reserve/extend; failed output never creates a giant Vec.
reserve_bounded(&mut self.bytes, bytes.len(), self.maximum)?;
self.bytes.extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
/// Compact output avoids indentation multiplying the cost of legal depth128.
/// The returned bytes have already passed the identical reader resource profile.
pub(super) fn encode(
project: &Project,
profile: Profile,
depth_128: bool,
) -> Result<Vec<u8>, FileError> {
let _scope = Scope::encoding(profile.expression_nodes);
let mut writer = CappedWriter::new(profile.bytes);
let result = document_stack(|| serde_json::to_writer(&mut writer, project));
if writer.exceeded {
return Err(limit("JSON output byte limit exceeded."));
}
result?;
writer.write_all(b"\n").map_err(|error| {
if writer.exceeded {
limit("JSON output byte limit exceeded.")
} else {
FileError::Io(error)
}
})?;
// Establish save/reopen symmetry before filesystem mutation, including the
// conservative production guard until the deeper profile is validated.
let reopened = current(&writer.bytes, profile, depth_128)?;
if reopened != *project {
return Err(FileError::Invalid(
"JSON serialization did not preserve the project value.".into(),
));
}
Ok(writer.bytes)
}
pub(super) fn read_bounded(path: &std::path::Path, maximum: usize) -> Result<Vec<u8>, FileError> {
let mut file = std::fs::File::open(path)?;
if file.metadata()?.len() > maximum as u64 {
return Err(limit("Input file byte limit exceeded."));
}
let mut bytes = Vec::new();
let mut chunk = [0u8; 8192];
loop {
let count = file.read(&mut chunk)?;
if count == 0 {
return Ok(bytes);
}
if count > maximum.saturating_sub(bytes.len()) {
return Err(limit("Input file byte limit exceeded."));
}
reserve_bounded(&mut bytes, count, maximum)?;
bytes.extend_from_slice(&chunk[..count]);
}
}
pub(super) fn matches_file(path: &std::path::Path, expected: &[u8]) -> io::Result<bool> {
let mut file = std::fs::File::open(path)?;
if file.metadata()?.len() != expected.len() as u64 {
return Ok(false);
}
let mut compared = 0usize;
let mut chunk = [0u8; 8192];
loop {
let count = file.read(&mut chunk)?;
if count == 0 {
return Ok(compared == expected.len());
}
if expected.get(compared..compared + count) != Some(&chunk[..count]) {
return Ok(false);
}
compared += count;
}
}
#[cfg(test)]
mod tests;
@@ -0,0 +1,612 @@
use super::*;
use crate::{
Element, ElementBody, Id, OaBody,
analysis::{
AnalysisType, Binding, CalculationDefinition, Definition, DefinitionBody, Expr, Parameter,
ParameterId, ScalarType, UnaryOp,
},
language::ParseLimits,
persistence,
};
use std::{
fmt::Write as _,
process::{Command, Stdio},
time::{Duration, Instant},
};
const IDENTITY: &str = "00000000-0000-0000-0000-000000000001";
const PARAMETER: &str = "00000000-0000-0000-0000-000000000002";
const MAIN: &str = "00000000-0000-0000-0000-000000000003";
fn id(value: &str) -> Id {
Id::parse_str(value).unwrap()
}
fn boolean_type() -> AnalysisType {
AnalysisType::Scalar(ScalarType::Boolean)
}
fn definition(id: Id, alias: &str, parameters: Vec<Parameter>, body: Expr) -> Definition {
Definition {
id,
alias: alias.into(),
name: alias.into(),
description: String::new(),
body: DefinitionBody::Calculation(CalculationDefinition {
parameters,
result: boolean_type(),
body: Binding::Expression(body),
}),
}
}
fn project() -> Project {
let mut project = Project::empty();
project.id = Id::from_u128(1000);
project.definitions.insert(
id(IDENTITY),
definition(
id(IDENTITY),
"identity",
vec![Parameter {
id: ParameterId(id(PARAMETER)),
name: "value".into(),
value_type: boolean_type(),
}],
Expr::Parameter(ParameterId(id(PARAMETER))),
),
);
project.definitions.insert(
id(MAIN),
definition(id(MAIN), "main", vec![], Expr::Boolean(true)),
);
project
}
fn wire(expression: &str) -> Vec<u8> {
let shallow = serde_json::to_string(&project()).unwrap();
assert_eq!(shallow.matches(r#"{"Boolean":true}"#).count(), 1);
shallow
.replace(r#"{"Boolean":true}"#, expression)
.into_bytes()
}
fn chain(depth: usize, call: bool) -> String {
assert!(depth > 0);
let prefix = if call {
format!(r#"{{"Call":{{"definition":{{"Local":"{IDENTITY}"}},"arguments":{{"{PARAMETER}":"#)
} else {
r#"{"Unary":{"op":"Not","operand":"#.into()
};
let suffix = if call { "}}}" } else { "}}" };
format!(
"{}{}{}",
prefix.repeat(depth - 1),
r#"{"Boolean":true}"#,
suffix.repeat(depth - 1)
)
}
fn candidate(bytes: &[u8], profile: Profile) -> Result<Project, FileError> {
persistence::decode_with_report_profile(bytes, ParseLimits::UNLIMITED, profile, true)
.map(|decoded| decoded.project)
}
#[test]
fn resource_preflight_has_exact_lowerable_boundaries() {
let bytes = br#"{"key":["a\\\"[{}]",0,true],"other":"\u0061"}"#;
let usage = preflight(bytes, Profile::DOCUMENT).unwrap();
let exact = Profile {
bytes: bytes.len(),
tokens: usage.tokens,
entries: usage.entries,
container_entries: 3,
string_bytes: 9,
total_string_bytes: usage.string_bytes,
..Profile::DOCUMENT
};
preflight(bytes, exact).unwrap();
for lower in [
Profile {
bytes: exact.bytes - 1,
..exact
},
Profile {
tokens: exact.tokens - 1,
..exact
},
Profile {
entries: exact.entries - 1,
..exact
},
Profile {
container_entries: 2,
..exact
},
Profile {
total_string_bytes: exact.total_string_bytes - 1,
..exact
},
Profile {
string_bytes: exact.string_bytes - 1,
..exact
},
] {
assert!(matches!(preflight(bytes, lower), Err(FileError::Limit(_))));
}
let deepest = format!("{}0{}", "[".repeat(512), "]".repeat(512));
assert_eq!(
preflight(deepest.as_bytes(), Profile::DOCUMENT)
.unwrap()
.deepest_container,
512
);
assert!(matches!(
preflight(format!("[{deepest}]").as_bytes(), Profile::DOCUMENT),
Err(FileError::Limit(_))
));
// A huge ignored scalar array exhausts preflight before a header is decoded.
assert!(matches!(
preflight(
br#"{"schema_version":8,"junk":[0,0,0,0]}"#,
Profile {
container_entries: 3,
..Profile::DOCUMENT
}
),
Err(FileError::Limit(_))
));
}
#[test]
fn scanner_is_not_a_parallel_json_grammar() {
for bytes in [
br#"{"schema_version":8}{}"#.as_slice(),
br#"{"schema_version":8,}"#,
br#"{"schema_version":8 "junk":0}"#,
br#"{"schema_version":8,"schema_version":8}"#,
br#"{"schema_version":8,"\u0073chema_version":8}"#,
br#"{"schema_version":8,"junk":"\q"}"#,
br#"{"schema_version":8,"junk":"unfinished}"#,
br#"{"schema_version":8,"junk":[}"#,
] {
assert!(candidate(bytes, Profile::DOCUMENT).is_err(), "{bytes:?}");
}
let mut invalid_utf8 = wire(r#"{"Boolean":true}"#);
let offset = invalid_utf8
.windows(7)
.position(|window| window == b"mission")
.unwrap();
invalid_utf8[offset] = 0xff;
assert!(candidate(&invalid_utf8, Profile::DOCUMENT).is_err());
let escaped = String::from_utf8(wire(r#"{"Boolean":true}"#))
.unwrap()
.replace(r#""schema_version""#, r#""\u0073chema_version""#);
assert_eq!(
candidate(escaped.as_bytes(), Profile::DOCUMENT).unwrap(),
project()
);
}
#[test]
fn remote_serde_keeps_each_existing_expression_variant() {
let n = r#"{"Number":{"numerator":"1","denominator":"1"}}"#;
let b = r#"{"Boolean":true}"#;
let expressions = [
b.to_string(),
n.to_string(),
format!(
r#"{{"Quantity":{{"magnitude":{{"numerator":"1","denominator":"1"}},"unit":{{"Local":"{IDENTITY}"}}}}}}"#
),
format!(r#"{{"Property":"{IDENTITY}"}}"#),
format!(r#"{{"Parameter":"{PARAMETER}"}}"#),
format!(r#"{{"List":[{n},{n}]}}"#),
format!(r#"{{"Unary":{{"op":"Not","operand":{b}}}}}"#),
format!(r#"{{"Binary":{{"op":"Add","left":{n},"right":{n}}}}}"#),
format!(r#"{{"If":{{"condition":{b},"when_true":{n},"when_false":{n}}}}}"#),
format!(
r#"{{"Call":{{"definition":{{"Local":"{IDENTITY}"}},"arguments":{{"{PARAMETER}":{b}}}}}}}"#
),
format!(r#"{{"Sum":{{"values":{{"List":[{n}]}},"initial":{n}}}}}"#),
format!(r#"{{"Power":{{"base":{n},"exponent":2}}}}"#),
format!(r#"{{"AsKind":{{"kind":{{"Local":"{IDENTITY}"}},"value":{n}}}}}"#),
];
for original in expressions {
let value: Expr = serde_json::from_str(&original).unwrap();
let copy = value.clone();
assert!(value == copy);
assert_eq!(serde_json::to_string(&value).unwrap(), original);
assert_eq!(format!("{value:?}"), format!("{copy:?}"));
}
let list = Expr::List(vec![Expr::Boolean(true), Expr::Boolean(false)]);
assert_eq!(format!("{list:?}"), "List([Boolean(true), Boolean(false)])");
assert!(list != Expr::List(vec![Expr::Boolean(false), Expr::Boolean(true)]));
}
#[test]
fn json_profile_does_not_contract_declared_cli_collection_counts() {
// Exercise original-byte typed construction at the existing graph ceiling.
// This gate checks codec/count compatibility, not 100000-object UI speed.
let mut model = Project::empty();
model.id = Id::from_u128(1_000_000);
for value in 1..=100_000 {
let id = Id::from_u128(value);
model.elements.insert(
id,
Element {
id,
alias: format!("activity_{value}"),
name: String::new(),
description: String::new(),
parent: None,
body: ElementBody::Oa(OaBody::Activity),
},
);
}
let bytes = encode(&model, Profile::DOCUMENT, true).unwrap();
let decoded = current(&bytes, Profile::DOCUMENT, true).unwrap();
assert_eq!(decoded.elements.len(), 100_000);
let cli = ParseLimits {
max_source_bytes: 64 * 1024 * 1024,
max_graph_objects: 100_000,
max_records: 250_000,
max_step_references: 500_000,
};
cli.check_model(&decoded).unwrap();
let references = format!(
"[{}\"{IDENTITY}\"]",
format!("\"{IDENTITY}\",").repeat(499_999)
);
preflight(references.as_bytes(), Profile::DOCUMENT).unwrap();
assert!(matches!(
preflight(
format!("[\"{IDENTITY}\",{}]", &references[1..references.len() - 1]).as_bytes(),
Profile::DOCUMENT
),
Err(FileError::Limit(_))
));
// Metadata over the initially proposed 1 MiB limit remains admissible.
let mut text = Project::empty();
text.description = "x".repeat(1024 * 1024 + 1);
let bytes = encode(&text, Profile::DOCUMENT, true).unwrap();
assert_eq!(current(&bytes, Profile::DOCUMENT, true).unwrap(), text);
eprintln!(
"JSON profile: graph bytes={}, ordered-reference bytes={}, max container entries={}",
serde_json::to_string(&model).unwrap().len(),
references.len(),
Profile::DOCUMENT.container_entries
);
}
#[test]
fn production_save_always_reopens_with_its_reader_profile() {
let deep = candidate(&wire(&chain(128, true)), Profile::DOCUMENT).unwrap();
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("project.json");
std::fs::write(&path, b"previous").unwrap();
let saved = persistence::save_checked(&path, &deep, Some(b"previous"));
if DEPTH_128_VALIDATED {
let bytes = saved.unwrap();
assert_eq!(persistence::decode(&bytes).unwrap(), deep);
} else {
assert!(saved.is_err());
assert_eq!(std::fs::read(&path).unwrap(), b"previous");
}
}
fn accepted_phase() {
for call in [false, true] {
let bytes = wire(&chain(128, call));
let usage = preflight(&bytes, Profile::DOCUMENT).unwrap();
let model = candidate(&bytes, Profile::DOCUMENT).unwrap();
let copy = model.clone();
assert!(model == copy);
let debug = format!("{model:?}");
assert!(!debug.is_empty());
let output = encode(&model, Profile::DOCUMENT, true).unwrap();
assert!(candidate(&output, Profile::DOCUMENT).unwrap() == model);
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("deep.json");
let saved = persistence::save_checked_profile(&path, &model, None, Profile::DOCUMENT, true)
.unwrap();
assert!(
candidate(
&read_bounded(&path, MAX_DOCUMENT_BYTES).unwrap(),
Profile::DOCUMENT
)
.unwrap()
== model
);
assert_eq!(saved, output);
eprintln!(
"accepted call={call} depth128 input={} output={} tokens={} entries={} raw strings={} containers={} debug={}",
bytes.len(),
output.len(),
usage.tokens,
usage.entries,
usage.string_bytes,
usage.deepest_container,
debug.len()
);
}
}
fn refused_phase() {
for call in [false, true] {
let error = candidate(&wire(&chain(129, call)), Profile::DOCUMENT).unwrap_err();
assert!(error.to_string().contains("logical depth"), "{error}");
}
let wide = format!(
r#"{{"List":[{}null]}}"#,
r#"{"Boolean":true},"#.repeat(1024)
);
let error = candidate(&wire(&wide), Profile::DOCUMENT).unwrap_err();
assert!(error.to_string().contains("1024"), "{error}");
let bytes = wire(&chain(128, true));
let exact_model = candidate(
&bytes,
Profile {
expression_nodes: 129,
..Profile::DOCUMENT
},
)
.unwrap();
let error = candidate(
&bytes,
Profile {
expression_nodes: 128,
..Profile::DOCUMENT
},
)
.unwrap_err();
// Identity's Parameter plus main's 128-node expression exceed this total.
assert!(error.to_string().contains("allocation count"), "{error}");
let scan = preflight(&bytes, Profile::DOCUMENT).unwrap();
assert!(matches!(
candidate(
&bytes,
Profile {
tokens: scan.tokens - 1,
..Profile::DOCUMENT
}
),
Err(FileError::Limit(_))
));
document_stack(|| {
use crate::analysis::{AdmissionLimits, admit};
let events = admit(&exact_model, AdmissionLimits::REFERENCE)
.unwrap()
.admission_usage()
.static_events;
admit(
&exact_model,
AdmissionLimits {
max_static_events: events,
},
)
.unwrap();
let error = match admit(
&exact_model,
AdmissionLimits {
max_static_events: events - 1,
},
) {
Ok(_) => panic!("Static work below the measured exact count must refuse."),
Err(error) => error,
};
assert_eq!(error.code, "analysis_limit");
});
let mut tree = r#"{"Boolean":true}"#.to_owned();
for _ in 0..15 {
tree = format!(r#"{{"Binary":{{"op":"And","left":{tree},"right":{tree}}}}}"#);
}
// 65535 balanced nodes plus one unary node is the stored-tree boundary.
tree = format!(r#"{{"Unary":{{"op":"Not","operand":{tree}}}}}"#);
let at_tree_limit = wire(&tree);
current(&at_tree_limit, Profile::DOCUMENT, true).unwrap();
let beyond = wire(&format!(r#"{{"Unary":{{"op":"Not","operand":{tree}}}}}"#));
assert!(
current(&beyond, Profile::DOCUMENT, true)
.unwrap_err()
.to_string()
.contains("65536")
);
// No downstream work after refusal poisons the next ordinary operation.
candidate(&wire(&chain(128, true)), Profile::DOCUMENT).unwrap();
}
fn malformed_phase() {
let deep = chain(127, false);
for (label, expression) in [
(
"duplicate operand",
format!(r#"{{"Unary":{{"op":"Not","operand":{deep},"operand":null}}}}"#),
),
(
"escaped duplicate",
format!(r#"{{"Unary":{{"op":"Not","operand":{deep},"\u006fperand":null}}}}"#),
),
(
"unknown field",
format!(r#"{{"Unary":{{"op":"Not","operand":{deep},"surprise":null}}}}"#),
),
(
"duplicate enum",
format!(r#"{{"Unary":{{"op":"Not","operand":{deep}}},"Boolean":true}}"#),
),
(
"duplicate parameter",
format!(
r#"{{"Call":{{"definition":{{"Local":"{IDENTITY}"}},"arguments":{{"{PARAMETER}":{deep},"\u0030{}":null}}}}}}"#,
&PARAMETER[1..]
),
),
] {
let error = candidate(&wire(&expression), Profile::DOCUMENT).unwrap_err();
assert!(
!error.to_string().contains("logical depth"),
"{label}: {error}"
);
eprintln!("{label}: {error}");
}
let complete = String::from_utf8(wire(&chain(128, true))).unwrap();
let duplicate_late = format!(
r#"{},"name":"late duplicate"}}"#,
&complete[..complete.len() - 1]
);
assert!(candidate(duplicate_late.as_bytes(), Profile::DOCUMENT).is_err());
assert!(candidate(format!("{complete} null").as_bytes(), Profile::DOCUMENT).is_err());
let invalid_surrogate = complete.replace(r#""description":"""#, r#""description":"\uD800""#);
assert!(candidate(invalid_surrogate.as_bytes(), Profile::DOCUMENT).is_err());
// Fully rejected raw depth also tears down without a recursive JSON Value.
let excessive = format!("{}0{}", "[".repeat(100_000), "]".repeat(100_000));
assert!(matches!(
preflight(excessive.as_bytes(), Profile::DOCUMENT),
Err(FileError::Limit(_))
));
}
fn ownership_phase() {
let mut value = Expr::Boolean(true);
for _ in 1..100_000 {
value = Expr::Unary {
op: UnaryOp::Not,
operand: Box::new(value),
};
}
let mut copy = value.clone();
assert!(value == copy);
let mut cursor = &mut copy;
loop {
match cursor {
Expr::Unary { operand, .. } => cursor = operand,
Expr::Boolean(leaf) => {
*leaf = false;
break;
}
_ => unreachable!(),
}
}
assert!(value != copy);
// A streaming formatter avoids retaining a 100000-node Debug string.
struct Counter(usize);
impl fmt::Write for Counter {
fn write_str(&mut self, value: &str) -> fmt::Result {
self.0 += value.len();
Ok(())
}
}
let mut debug = Counter(0);
write!(&mut debug, "{value:?}").unwrap();
assert!(debug.0 > 100_000);
assert!(serde_json::to_writer(io::sink(), &value).is_err());
drop(value);
drop(copy);
eprintln!(
"iterative manual100000 ownership/debug/drop completed; debug bytes={}",
debug.0
);
}
fn atomic_phase() {
let model = project();
let bytes = encode(&model, Profile::DOCUMENT, true).unwrap();
let exact = Profile {
bytes: bytes.len(),
..Profile::DOCUMENT
};
assert_eq!(encode(&model, exact, true).unwrap(), bytes);
let directory = tempfile::tempdir().unwrap();
let path = directory.path().join("preserved.json");
std::fs::write(&path, b"original").unwrap();
let error = persistence::save_checked_profile(
&path,
&model,
Some(b"original"),
Profile {
bytes: bytes.len() - 1,
..Profile::DOCUMENT
},
true,
)
.unwrap_err();
assert!(matches!(error, FileError::Limit(_)));
assert_eq!(std::fs::read(&path).unwrap(), b"original");
assert_eq!(std::fs::read_dir(directory.path()).unwrap().count(), 1);
let error = persistence::save_checked_profile(
&path,
&model,
Some(b"different"),
Profile::DOCUMENT,
true,
)
.unwrap_err();
assert!(matches!(error, FileError::Conflict));
assert_eq!(std::fs::read(&path).unwrap(), b"original");
assert_eq!(std::fs::read_dir(directory.path()).unwrap().count(), 1);
assert!(read_bounded(&path, 7).is_err());
assert_eq!(read_bounded(&path, 8).unwrap(), b"original");
let mut writer = CappedWriter::new(3);
writer.write_all(b"abc").unwrap();
let capacity = writer.bytes.capacity();
assert!(writer.write_all(b"d").is_err());
assert_eq!(writer.bytes, b"abc");
assert_eq!(writer.bytes.capacity(), capacity);
}
/// Root runs this exact test in each approved target/profile. Each phase runs in
/// another process so a stack overflow is an explicit failed phase, and inside
/// a 128 KiB thread (smaller than the ordinary test stack). No RUST_MIN_STACK
/// override, ignored test, release-only path or increased test stack.
#[test]
fn codec_lifecycle_subprocess() {
const PHASE: &str = "ARCADIA_CODEC_LIFECYCLE_PHASE";
if let Ok(phase) = std::env::var(PHASE) {
eprintln!("codec child phase={phase}, configured thread stack=131072");
std::thread::Builder::new()
.name(format!("codec-{phase}"))
.stack_size(128 * 1024)
.spawn(move || match phase.as_str() {
"accepted" => accepted_phase(),
"refused" => refused_phase(),
"malformed" => malformed_phase(),
"ownership" => ownership_phase(),
"atomic" => atomic_phase(),
_ => panic!("Unknown codec phase"),
})
.unwrap()
.join()
.unwrap();
return;
}
for phase in ["accepted", "refused", "malformed", "ownership", "atomic"] {
let mut child = Command::new(std::env::current_exe().unwrap())
.args([
"--exact",
"persistence::json_codec::tests::codec_lifecycle_subprocess",
"--test-threads=1",
"--nocapture",
])
.env(PHASE, phase)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.unwrap();
let start = Instant::now();
loop {
if child.try_wait().unwrap().is_some() {
break;
}
if start.elapsed() > Duration::from_secs(120) {
child.kill().unwrap();
let output = child.wait_with_output().unwrap();
panic!(
"codec {phase} timed out: {}",
String::from_utf8_lossy(&output.stderr)
);
}
std::thread::sleep(Duration::from_millis(20));
}
let output = child.wait_with_output().unwrap();
eprintln!(
"codec {phase}: status={} elapsed={:?}\n{}{}",
output.status,
start.elapsed(),
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
assert!(
output.status.success(),
"codec phase {phase} aborted or failed"
);
}
}