fix: unblock multi-flow registries and expose workflow refusals

crates/registry-breg/Cargo.toml

@@ -370,6 +370,11 @@ name = "postgres_immediate_action_requirements"
 path = "tests/postgres_immediate_action_requirements.rs"
 required-features = ["postgres-test"]
 
+[[test]]
+name = "postgres_multi_reference_action_authority"
+path = "tests/postgres_multi_reference_action_authority.rs"
+required-features = ["postgres-test", "tooling"]
+
 [[test]]
 name = "postgres_action_handlers"
 path = "tests/postgres_action_handlers.rs"

crates/registry-breg/src/artifacts.rs

@@ -746,7 +746,7 @@ pub(crate) fn openapi_action_input_schema(action: &CompiledAction) -> Value {
         input_schema["required"] = json!([]);
     }
     let mut properties = Map::from_iter([("input".to_owned(), input_schema)]);
-    if !condition_inputs.is_empty() {
+    if !action.execution.is_proposal() {
         properties.insert(
             "preconditions".to_owned(),
             action_preconditions_schema(condition_inputs.iter().copied()),

crates/registry-breg/src/migration_plan.rs

@@ -946,6 +946,17 @@ impl ValidatedReviewedMigrationPlan {
     pub fn migrations(&self) -> &[ValidatedReviewedMigration] {
         &self.migrations
     }
+
+    pub(crate) fn statement_count(&self) -> usize {
+        self.migrations
+            .iter()
+            .map(|migration| {
+                migration.steps.len()
+                    + migration.pre_assertions.len()
+                    + migration.post_assertions.len()
+            })
+            .sum()
+    }
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -995,6 +1006,7 @@ pub(crate) struct ReviewedPlanBindings<'a> {
 pub(crate) struct PreparedReviewedMigrationPlan {
     pub descriptor_paths: Vec<String>,
     pub files: BTreeMap<String, Vec<u8>>,
+    pub statement_count: usize,
 }
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -1073,6 +1085,7 @@ pub(crate) fn prepare_reviewed_migration_plan(
     Ok(PreparedReviewedMigrationPlan {
         descriptor_paths,
         files,
+        statement_count: validated.statement_count(),
     })
 }
 

crates/registry-breg/src/package.rs

@@ -69,13 +69,15 @@ pub const MAX_RHAI_PLANNER_SOURCE_BYTES: u64 = crate::plan::MAX_PLAN_SOURCE_BYTE
 pub const MAX_RHAI_PLANNER_PATH_BYTES: usize = 256;
 
 const MANIFEST_PATH: &str = "package.json";
-const MAX_MANIFEST_BYTES: u64 = 4 * 1024 * 1024;
+const MAX_MANIFEST_BYTES: u64 = 8 * 1024 * 1024;
 const MAX_FILE_BYTES: u64 = MAX_PACKAGE_SOURCE_FILE_BYTES;
 pub const MAX_PACKAGE_BYTES: u64 = 64 * 1024 * 1024;
 pub const MAX_PACKAGE_FILES: usize = 1_024;
+/// Maximum number of statements represented by one bounded package migration
+/// and its durable PostgreSQL activation-ledger identity.
+pub const MAX_MIGRATION_STATEMENTS: usize = 4_096;
 const MAX_PATH_BYTES: usize = 512;
 const MAX_PATH_COMPONENTS: usize = 16;
-const MAX_MIGRATION_STATEMENTS: usize = 1_024;
 const MAX_MIGRATION_BASELINE_BYTES: usize = 4 * 1024 * 1024;
 
 #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
@@ -839,15 +841,23 @@ impl PreparedPackage {
         }
     }
 
+    fn manifest_bytes(&self) -> Result<Vec<u8>> {
+        let bytes = canonicalize_json(
+            &serde_json::to_value(self.envelope()).map_err(|_| PackageError::CanonicalJson)?,
+        )
+        .map_err(|_| PackageError::CanonicalJson)?;
+        if bytes.len() as u64 > MAX_MANIFEST_BYTES {
+            return Err(PackageError::Bounds);
+        }
+        Ok(bytes)
+    }
+
     /// The package identity this candidate has once published without a
     /// `REVISION` file: the digest of the sum file over its files and
     /// manifest. Nothing is written.
     pub fn package_digest(&self) -> Result<String> {
         let mut files = self.files.clone();
-        let manifest_bytes = canonicalize_json(
-            &serde_json::to_value(self.envelope()).map_err(|_| PackageError::CanonicalJson)?,
-        )
-        .map_err(|_| PackageError::CanonicalJson)?;
+        let manifest_bytes = self.manifest_bytes()?;
         files.insert(MANIFEST_PATH.to_owned(), manifest_bytes);
         plan_package(
             Path::new("."),
@@ -874,6 +884,7 @@ impl PreparedPackage {
         destination: &Path,
         revision: Option<&str>,
     ) -> Result<SharedVerifiedPackage> {
+        let manifest_bytes = self.manifest_bytes()?;
         reject_symlink_components(destination)?;
         if destination.exists() {
             return Err(PackageError::Closure);
@@ -895,10 +906,6 @@ impl PreparedPackage {
                 }
                 write_new_file(&full, bytes)?;
             }
-            let manifest_bytes = canonicalize_json(
-                &serde_json::to_value(self.envelope()).map_err(|_| PackageError::CanonicalJson)?,
-            )
-            .map_err(|_| PackageError::CanonicalJson)?;
             write_new_file(&destination.join(MANIFEST_PATH), &manifest_bytes)?;
             let package = write_sum_file(
                 destination,
@@ -3245,11 +3252,21 @@ pub fn prepare_package_with_project_assets(
     };
     validate_migration_plan(&manifest, &compiled)?;
     validate_source_inventory(&manifest)?;
-    Ok(PreparedPackage {
+    finish_prepared_package(manifest, compiled, files)
+}
+
+fn finish_prepared_package(
+    manifest: PackageManifest,
+    registry: CompiledRegistry,
+    files: BTreeMap<String, Vec<u8>>,
+) -> Result<PreparedPackage> {
+    let prepared = PreparedPackage {
         manifest,
-        registry: compiled,
+        registry,
         files,
-    })
+    };
+    prepared.manifest_bytes()?;
+    Ok(prepared)
 }
 
 #[cfg(feature = "tooling")]
@@ -3284,17 +3301,17 @@ fn reviewed_successor_inputs(
     let PreparedReviewedMigrationPlan {
         descriptor_paths,
         files,
+        statement_count,
     } = reviewed;
-    Ok((
-        reviewed_successor_migration_plan(
-            baseline,
-            compiled,
-            &change_set,
-            descriptor_paths,
-            prior_schema_fingerprint,
-        )?,
-        files,
-    ))
+    let plan = reviewed_successor_migration_plan(
+        baseline,
+        compiled,
+        &change_set,
+        descriptor_paths,
+        prior_schema_fingerprint,
+    )?;
+    validate_migration_plan_bounds(&plan, statement_count)?;
+    Ok((plan, files))
 }
 
 fn package_compiler_assets(
@@ -4672,6 +4689,9 @@ fn rederive(
     }
     validate_migration_plan(manifest, &compiled)?;
     let reviewed_migration_plan = rederive_reviewed_migration_plan(manifest, loaded, &compiled)?;
+    if let Some(reviewed) = &reviewed_migration_plan {
+        validate_migration_plan_bounds(&manifest.migration_plan, reviewed.statement_count())?;
+    }
     Ok((compiled, reviewed_migration_plan))
 }
 
@@ -4964,15 +4984,7 @@ fn validate_captured_bindings(
 }
 
 fn validate_migration_plan(manifest: &PackageManifest, compiled: &CompiledRegistry) -> Result<()> {
-    if manifest.migration_plan.statements.len() > MAX_MIGRATION_STATEMENTS {
-        return Err(PackageError::Bounds);
-    }
-    if manifest.migration_plan.changes.len() > MAX_MIGRATION_STATEMENTS {
-        return Err(PackageError::Bounds);
-    }
-    if manifest.migration_plan.reviewed_descriptors.len() > MAX_MIGRATION_STATEMENTS {
-        return Err(PackageError::Bounds);
-    }
+    validate_migration_plan_bounds(&manifest.migration_plan, 0)?;
     let mut prior_descriptor = None;
     for descriptor in &manifest.migration_plan.reviewed_descriptors {
         validate_relative(descriptor)?;
@@ -4991,6 +5003,30 @@ fn validate_migration_plan(manifest: &PackageManifest, compiled: &CompiledRegist
     Ok(())
 }
 
+fn validate_migration_plan_bounds(
+    plan: &MigrationPlan,
+    reviewed_statement_count: usize,
+) -> Result<()> {
+    if plan.statements.len() > MAX_MIGRATION_STATEMENTS {
+        return Err(PackageError::Bounds);
+    }
+    if plan.changes.len() > MAX_MIGRATION_STATEMENTS {
+        return Err(PackageError::Bounds);
+    }
+    if plan.reviewed_descriptors.len() > MAX_MIGRATION_STATEMENTS {
+        return Err(PackageError::Bounds);
+    }
+    if plan
+        .statements
+        .len()
+        .checked_add(reviewed_statement_count)
+        .is_none_or(|count| count > MAX_MIGRATION_STATEMENTS)
+    {
+        return Err(PackageError::Bounds);
+    }
+    Ok(())
+}
+
 fn expected_migration_plan(
     manifest: &PackageManifest,
     compiled: &CompiledRegistry,

crates/registry-breg/src/package/tests/immediate_actions.rs

@@ -1145,3 +1145,112 @@ fn reviewed_successor_installs_added_proposal_operation_policies() {
         .collect::<BTreeSet<_>>();
     assert_eq!(actual, expected);
 }
+
+#[test]
+fn package_migration_plan_accepts_the_statement_bound_and_refuses_one_more() {
+    let compiled = compile(&source());
+    let mut plan = initial_migration_plan(&compiled);
+    let statement = plan
+        .statements
+        .first()
+        .expect("the fixture compiles at least one DDL statement")
+        .clone();
+    plan.statements = vec![statement.clone(); MAX_MIGRATION_STATEMENTS];
+    validate_migration_plan_bounds(&plan, 0).expect("the exact package statement bound validates");
+
+    plan.statements.push(statement);
+    assert_eq!(
+        validate_migration_plan_bounds(&plan, 0),
+        Err(PackageError::Bounds)
+    );
+
+    plan.statements.truncate(1_025);
+    validate_migration_plan_bounds(&plan, MAX_MIGRATION_STATEMENTS - 1_025)
+        .expect("compiler and reviewed statements may fill the aggregate bound exactly");
+    assert_eq!(
+        validate_migration_plan_bounds(&plan, MAX_MIGRATION_STATEMENTS - 1_024),
+        Err(PackageError::Bounds)
+    );
+}
+
+fn package_capacity_request(source_revision: &str) -> PackageBuildRequest {
+    let mut value = source();
+    value["package"] = json!({"sourceRevision": source_revision});
+    value["manifestProjection"] = json!({
+        "accessProfile": "operator",
+        "classificationCeiling": "internal",
+        "catalog": {
+            "baseUrl": "https://action-package.example.test",
+            "title": "Action Package",
+            "publisher": {"id": "action-authority", "name": "Action Authority"}
+        },
+        "publicService": {"id": "action-service", "title": "Action Service"},
+        "datasets": [{
+            "id": "test-dataset", "title": "Test Dataset",
+            "owner": "Action Authority", "status": "active"
+        }],
+        "dataServices": [{
+            "id": "action-api", "title": "Action API",
+            "endpointUrl": "https://action-package.example.test",
+            "servesDatasets": ["test-dataset"]
+        }],
+        "distributions": []
+    });
+    value["accessProfiles"][0]["permissions"]["records"] = json!([{
+        "recordType": "item", "operations": ["get", "list"],
+        "readableFields": ["label"], "rowBoundaries": "unrestricted"
+    }]);
+    PackageBuildRequest {
+        from_package_digest: None,
+        compiler_source_revision: source_revision.to_owned(),
+        schema_fingerprint:
+            "sha256:0000000000000000000000000000000000000000000000000000000000000000".to_owned(),
+        project: PackageSourceFile {
+            path: "source/registry.json".to_owned(),
+            bytes: serde_json::to_vec(&value).expect("package fixture serializes"),
+        },
+        modules: Vec::new(),
+        fixture_journeys: PackageSourceFile {
+            path: FIXTURE_JOURNEYS_PATH.to_owned(),
+            bytes: b"capacity guard fixture".to_vec(),
+        },
+        migration_plan: PackageMigrationPlanInput::InitialCompiledDdl,
+    }
+}
+
+#[test]
+fn oversized_canonical_manifest_is_refused_before_a_package_can_be_left_behind() {
+    let mut prepared =
+        prepare_package(package_capacity_request("capacity-source")).expect("fixture prepares");
+    let oversized_revision = "x".repeat(MAX_MANIFEST_BYTES as usize);
+    prepared.manifest.compiler.source_revision = oversized_revision;
+    assert_eq!(
+        finish_prepared_package(
+            prepared.manifest.clone(),
+            prepared.registry.clone(),
+            prepared.files.clone(),
+        )
+        .err(),
+        Some(PackageError::Bounds),
+        "the preparation finalizer refuses a manifest the loader cannot read"
+    );
+
+    assert_eq!(prepared.package_digest(), Err(PackageError::Bounds));
+
+    let nanos = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .expect("clock follows epoch")
+        .as_nanos();
+    let destination = std::env::temp_dir().join(format!(
+        "breg-manifest-capacity-{}-{nanos}",
+        std::process::id()
+    ));
+    assert_eq!(
+        prepared.publish_to_directory(&destination).err(),
+        Some(PackageError::Bounds)
+    );
+    assert!(
+        !destination.exists(),
+        "refused publication does not create a package directory"
+    );
+}

crates/registry-breg/src/postgres/interlock.rs

@@ -30,7 +30,7 @@ use crate::migration_plan::{
 };
 use crate::model::{CompiledBlindIndex, CompiledEntity, CompiledField, CompiledRegistry};
 use crate::mutation::install_mutation_schema;
-use crate::package::CompiledRegistryMigrationBaseline;
+use crate::package::{CompiledRegistryMigrationBaseline, MAX_MIGRATION_STATEMENTS};
 
 use super::{
     catalog::{
@@ -40,11 +40,11 @@ use super::{
     },
     config::ConnectionTls,
     migration_ledger::{
-        in_flight_activation, migration_phase_state, record_applied, record_chunk_progress,
-        record_failed, record_postconditions_complete, record_preconditions_complete,
-        record_reverted, record_started, record_step_complete, statement_checksum, step_progress,
-        verify_resumable, ActivationPlanKind, MigrationLedgerEntry, MigrationLedgerStep,
-        MigrationLedgerStepKind,
+        ensure_migration_statement_capacity, in_flight_activation, migration_phase_state,
+        record_applied, record_chunk_progress, record_failed, record_postconditions_complete,
+        record_preconditions_complete, record_reverted, record_started, record_step_complete,
+        statement_checksum, step_progress, verify_resumable, ActivationPlanKind,
+        MigrationLedgerEntry, MigrationLedgerStep, MigrationLedgerStepKind,
     },
     schema::{
         execute_compiled_ddl_statement, is_spatial_candidate_view_drop_sql,
@@ -55,9 +55,8 @@ use super::{
     ExpectedRegistryIdentity, PostgresKernelError, Result, SqlIdentifier,
 };
 
-// These defense-in-depth bounds match the verified package manifest envelope:
-// at most 1,024 migration statements inside at most 4 MiB of manifest bytes.
-const MAX_VERIFIED_DDL_STATEMENTS: usize = 1024;
+// These defense-in-depth bounds match the verified package manifest envelope.
+const MAX_VERIFIED_DDL_STATEMENTS: usize = MAX_MIGRATION_STATEMENTS;
 const MAX_VERIFIED_DDL_STATEMENT_BYTES: usize = 4 * 1024 * 1024;
 const MAX_VERIFIED_DDL_STATEMENT_TIMEOUT: Duration = Duration::from_secs(60 * 60);
 /// Final field-encryption verification must inspect every affected row without
@@ -2606,6 +2605,12 @@ async fn begin_successor_in(
         .is_some();
     if fresh {
         verify_import_run_admission_state(transaction, current.activation_uuid()?).await?;
+        // A role change re-applies the active package, whose fingerprint
+        // covers the ledger constraint already in place; only a successor to
+        // a different package may widen it.
+        if target.package_digest != current.package_digest {
+            ensure_migration_statement_capacity(transaction).await?;
+        }
     }
     let changed = transaction
         .execute(