Adopt authority-safe checked mdbase APIs

.github/workflows/server-ci.yml

@@ -356,6 +356,7 @@ jobs:
       - run: pnpm install --frozen-lockfile
       - run: pnpm build
       - run: cargo fmt --all --check
+      - run: pnpm check:cargo-features
       - run: cargo clippy --locked --workspace --all-targets -- -D warnings
       # Compile and test every Rust crate against the exact SDK revision used
       # by the production provider image. This keeps path-only developer

Cargo.toml

@@ -37,7 +37,7 @@ directories = "6"
 futures-util = "0.3"
 hkdf = "0.12"
 hmac = "0.12"
-mdbase = { path = "../mdbase-rs" }
+mdbase = { path = "../mdbase-rs", default-features = false }
 mdbase-interop = { version = "0.1.0-rc.2", git = "https://github.com/mdbase-dev/mdbase-spec.git", rev = "5aa34fd14de2e0ae9e565425b724b306046b64ca" }
 mdbase-runtime = { path = "../mdbase-rs/crates/mdbase-runtime" }
 notify = "8"

config/architecture-budgets.json

@@ -35,6 +35,13 @@
     "crates/connect-hosted-storage-benchmark/src/main/legacy_query.rs": 1,
     "crates/connect-hosted-storage-benchmark/src/main/query.rs": 1
   },
+  "externalGuards": {
+    "directWireOnlyConstructors": 0,
+    "directCanonicalOutcomeStructs": 0,
+    "privateCanonicalResultFieldCalls": 0,
+    "legacyCrudFeatures": 0,
+    "semanticProjectionFormatVersion": 6
+  },
   "reviewBudgets": {
     "productionFiles": 663,
     "relativeImports": 1375,

crates/connect-agent/src/loopback/tests.rs

@@ -549,6 +549,20 @@ async fn encrypted_full_collection_operations_bound_and_repair_invalid_records()
         .await;
     assert_eq!(repaired["result"]["valid"], true);
     assert_eq!(fs::read(&utf8_path).unwrap(), repaired_utf8);
+    let repaired_changes = fixture
+        .direct(&app, "changes", json!({ "after": baseline + 1 }), 9)
+        .await;
+    assert!(
+        repaired_changes["result"]["events"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .any(|event| {
+                event["type"] == "mdbase.record.created"
+                    && event["payload"]["path"] == "invalid-utf8.md"
+            }),
+        "an update repair of a non-indexed record must retain its canonical create effect"
+    );
 
     let non_mapping_path = collection.join("non-mapping.md");
     let newer = b"---\ntitle: Newer external bytes\n---\nPreserve me\n";
@@ -563,7 +577,7 @@ async fn encrypted_full_collection_operations_bound_and_repair_invalid_records()
                 "document": "---\ntitle: Stale replacement\n---\nMust not apply\n",
                 "if_revision": revisions["non-mapping.md"]
             }),
-            9,
+            10,
         )
         .await;
     assert_eq!(stale["result"]["valid"], false);

crates/connect-core/src/registry.rs

@@ -59,6 +59,21 @@ const CONNECT_COLLECTION_ID: &str = "collection_id";
 const MIRROR_MARKER_DIRECTORY: &str = ".mdbase";
 const MIRROR_MARKER_FILE: &str = "connect-role.json";
 
+/// Connect-owned local authority capture policy. The entry and retained-byte
+/// ceilings align with Connect's 100k-record / 4 GiB admitted authority shape;
+/// individual exact documents retain mdbase's 64 MiB ceiling, while resource
+/// discovery is separately capped at 10k entries.
+fn local_capture_limits() -> mdbase::runtime::CaptureLimits {
+    mdbase::runtime::CaptureLimits::builder()
+        .max_entries(100_000)
+        .max_file_bytes(64 * 1024 * 1024)
+        .max_aggregate_bytes(4 * 1024 * 1024 * 1024)
+        .max_depth(128)
+        .max_resource_entries(10_000)
+        .max_retained_bytes(4 * 1024 * 1024 * 1024)
+        .build()
+}
+
 mod agent_state;
 mod authority;
 mod authority_store;

crates/connect-core/src/registry/collections.rs

@@ -230,11 +230,17 @@ impl CollectionRegistry {
             ));
         }
 
-        let context = mdbase::runtime::OperationContext::new(
+        let context = mdbase::runtime::OperationContext::with_capture_limits(
             &mdbase::OperationCancellation::new(),
             mdbase::runtime::OperationDeadline::after(std::time::Duration::from_secs(30)),
+            local_capture_limits(),
         );
-        provider.reset_runtime_support_for_fork(&context)?;
+        // A collection that has never opened a coordinated runtime has no
+        // runtime support to carry across the fork. Once support exists, let
+        // mdbase recover and clear it atomically before changing identity.
+        if path.join(".mdbase/transactions").is_dir() {
+            provider.reset_runtime_support_for_fork(&context)?;
+        }
 
         let independent_id = Uuid::new_v4();
         write_collection_id(&path, independent_id)?;

crates/connect-core/src/registry/operations.rs

@@ -322,7 +322,7 @@ impl CollectionRegistry {
             let execution =
                 execute_runtime_request(require_runtime(runtime)?, &request, None, &context)?;
             let result = v03_operation_result(&execution.operation);
-            if !execution.operation.valid {
+            if !execution.operation.is_valid() {
                 return Err(type_pack_setup_error(&result));
             }
             Ok(result.result)

crates/connect-core/src/registry/runtime_changes.rs

@@ -358,10 +358,7 @@ fn merge_payload(mut left: Value, right: Value) -> Value {
 fn runtime_context(
     cancellation: &mdbase::OperationCancellation,
 ) -> mdbase::runtime::OperationContext {
-    mdbase::runtime::OperationContext::new(
-        cancellation,
-        mdbase::runtime::OperationDeadline::after(Duration::from_secs(24 * 60 * 60)),
-    )
+    operation_context(cancellation)
 }
 
 #[cfg(test)]

crates/connect-core/src/registry/runtime_executor.rs

@@ -60,9 +60,10 @@ impl CollectionExecutor {
     ) -> Result<Self, ConnectError> {
         let (runtime, provider, feed) = if coordinated {
             let runtime = Arc::new(FilesystemRuntime::open(root, Duration::from_millis(120))?);
-            let context = OperationContext::new(
+            let context = OperationContext::with_capture_limits(
                 &mdbase::OperationCancellation::new(),
                 mdbase::runtime::OperationDeadline::after(Duration::from_secs(30)),
+                local_capture_limits(),
             );
             let feed = runtime.open_change_feed(owner, &context)?;
             runtime.establish_change_feed_baseline(&feed, &context)?;
@@ -461,9 +462,10 @@ mod tests {
     use super::*;
 
     fn context(duration: Duration) -> OperationContext {
-        OperationContext::new(
+        OperationContext::with_capture_limits(
             &mdbase::OperationCancellation::new(),
             mdbase::runtime::OperationDeadline::after(duration),
+            local_capture_limits(),
         )
     }
 

crates/connect-core/src/registry/runtime_operations.rs

@@ -62,9 +62,10 @@ enum QueryCursorAction<'a> {
 pub(super) fn operation_context(
     cancellation: &mdbase::OperationCancellation,
 ) -> mdbase::runtime::OperationContext {
-    mdbase::runtime::OperationContext::new(
+    mdbase::runtime::OperationContext::with_capture_limits(
         cancellation,
-        mdbase::runtime::OperationDeadline::after(std::time::Duration::from_secs(24 * 60 * 60)),
+        mdbase::runtime::OperationDeadline::after(std::time::Duration::from_secs(30)),
+        local_capture_limits(),
     )
 }
 
@@ -142,17 +143,9 @@ pub(super) fn execute_runtime_read(
         QueryCursorAction::Release(cursor) => {
             executor.release_read(cursor, scope_binding, context)?;
             Ok(RuntimeExecution {
-                operation: mdbase::runtime::CanonicalOperationOutcome {
-                    valid: true,
-                    value: mdbase::runtime::CanonicalOperationValue::WireOnly(
-                        mdbase::runtime::WireOnlyOperationValue::Validation(json!({
-                            "released": true,
-                            "results": [],
-                            "meta": {"total_count": 0, "has_more": false}
-                        })),
-                    ),
-                    diagnostics: Vec::new(),
-                },
+                operation: mdbase::runtime::CanonicalOperationOutcome::cursor_release(
+                    mdbase::runtime::CursorReleaseOutcome { released: true },
+                ),
                 outcome: None,
             })
         }
@@ -189,7 +182,7 @@ fn read_page_operation(
     mut operation: mdbase::runtime::CanonicalOperationOutcome,
     next: Option<String>,
 ) -> mdbase::runtime::CanonicalOperationOutcome {
-    if let mdbase::runtime::CanonicalOperationValue::Query(Some(query)) = &mut operation.value {
+    if let Some(query) = operation.query_value_mut() {
         if let Some(meta) = query.meta.as_object_mut() {
             match next {
                 Some(cursor) => {
@@ -890,11 +883,29 @@ pub(super) fn require_runtime(
 mod typed_boundary_tests {
     use super::*;
 
+    #[test]
+    fn typed_cursor_release_serializes_to_the_exact_existing_connect_envelope() {
+        let operation = mdbase::runtime::CanonicalOperationOutcome::cursor_release(
+            mdbase::runtime::CursorReleaseOutcome { released: true },
+        );
+        assert_eq!(
+            operation_response_value(&operation).unwrap(),
+            json!({
+                "valid": true,
+                "result": {
+                    "released": true,
+                    "results": [],
+                    "meta": {"total_count": 0, "has_more": false}
+                },
+                "diagnostics": []
+            })
+        );
+    }
+
     #[test]
     fn canonical_query_serializes_to_the_exact_existing_connect_envelope() {
-        let operation = mdbase::runtime::CanonicalOperationOutcome {
-            valid: true,
-            value: mdbase::runtime::CanonicalOperationValue::Query(Some(
+        let operation = mdbase::runtime::CanonicalOperationOutcome::try_completed(
+            mdbase::runtime::CanonicalOperationValue::Query(Some(
                 mdbase::runtime::CanonicalQueryValue {
                     records: vec![mdbase::api::ProjectedValue::new(json!({
                         "path": "tasks/one.md",
@@ -906,8 +917,9 @@ mod typed_boundary_tests {
                     embedded_diagnostics: Vec::new(),
                 },
             )),
-            diagnostics: Vec::new(),
-        };
+            Vec::new(),
+        )
+        .unwrap();
 
         assert_eq!(
             operation_response_value(&operation).unwrap(),

crates/connect-core/src/registry/scope.rs

@@ -218,10 +218,10 @@ pub(super) fn ensure_operation_in_scope(
 ) -> Result<(), ConnectError> {
     use mdbase::runtime::CanonicalOperationValue;
 
-    if !operation.valid {
+    if !operation.is_valid() {
         return Ok(());
     }
-    match &operation.value {
+    match operation.value() {
         CanonicalOperationValue::Read(Some(record))
         | CanonicalOperationValue::Create(Some(record))
         | CanonicalOperationValue::Update(Some(record)) => {
@@ -251,7 +251,7 @@ pub(super) fn operation_record(
     operation: &mdbase::runtime::CanonicalOperationOutcome,
 ) -> Option<&mdbase::api::RecordDocument> {
     use mdbase::runtime::CanonicalOperationValue;
-    match &operation.value {
+    match operation.value() {
         CanonicalOperationValue::Read(Some(record))
         | CanonicalOperationValue::Create(Some(record))
         | CanonicalOperationValue::Update(Some(record)) => Some(record),

crates/connect-hosted-provider/src/provider.rs

@@ -42,6 +42,12 @@ use subtle::ConstantTimeEq;
 use tokio::sync::{oneshot, Mutex, OwnedSemaphorePermit, RwLock, Semaphore};
 use uuid::Uuid;
 
+const CONNECT_SEMANTIC_PROJECTION_FORMAT_VERSION: u32 = 6;
+const _: () = assert!(
+    mdbase::runtime::SEMANTIC_PROJECTION_FORMAT_VERSION
+        == CONNECT_SEMANTIC_PROJECTION_FORMAT_VERSION
+);
+
 use crate::{
     backup_admin::lock_blob_deletion,
     blob_store::BlobStore,

crates/connect-hosted-provider/src/provider/operation_dispatch.rs

@@ -638,7 +638,7 @@ impl HostedProvider {
                     )
                     .await?;
                 let assessment_value = type_pack_assessment(&assessment)?;
-                if !assessment.valid || !assessment_value.applicable {
+                if !assessment.is_valid() || !assessment_value.applicable {
                     return Err(type_pack_provision_error(&assessment.to_v03()));
                 }
                 let expected_assessment_digest = assessment_value.assessment_digest.clone();
@@ -806,7 +806,7 @@ impl HostedProvider {
             }
         }
         let assessment_value = collection_setup_assessment(&assessment)?;
-        if !assessment.valid || !assessment_value.applicable {
+        if !assessment.is_valid() || !assessment_value.applicable {
             return Err(type_pack_provision_error(&assessment.to_v03()));
         }
         let applied = self
@@ -894,8 +894,8 @@ impl HostedProvider {
 fn type_pack_assessment(
     operation: &mdbase::runtime::CanonicalOperationOutcome,
 ) -> ApiResult<&mdbase::runtime::CanonicalTypePackValue> {
-    match &operation.value {
-        mdbase::runtime::CanonicalOperationValue::TypePack(Some(value)) => Ok(value),
+    match operation.value() {
+        mdbase::runtime::CanonicalOperationValue::AssessTypePack(Some(value)) => Ok(value),
         _ => Err(ApiError::internal(
             "Canonical type-pack assessment returned the wrong typed operation family.",
         )),
@@ -905,15 +905,15 @@ fn type_pack_assessment(
 fn collection_setup_assessment(
     operation: &mdbase::runtime::CanonicalOperationOutcome,
 ) -> ApiResult<&mdbase::v03::CollectionSetupAssessment> {
-    match &operation.value {
-        mdbase::runtime::CanonicalOperationValue::CollectionSetup(Some(value)) => {
-            match value.as_ref() {
-                mdbase::runtime::CanonicalCollectionSetupValue::Assessment(value) => Ok(value),
-                _ => Err(ApiError::internal(
-                    "Canonical collection-setup assessment returned the wrong typed operation family.",
-                )),
-            }
-        }
+    match operation.value() {
+        mdbase::runtime::CanonicalOperationValue::AssessCollectionSetup(Some(value)) => match value
+            .as_ref()
+        {
+            mdbase::runtime::CanonicalCollectionSetupValue::Assessment(value) => Ok(value),
+            _ => Err(ApiError::internal(
+                "Canonical collection-setup assessment returned the wrong typed operation family.",
+            )),
+        },
         _ => Err(ApiError::internal(
             "Canonical collection-setup assessment returned the wrong typed operation family.",
         )),

crates/connect-hosted-provider/src/provider/operation_resource_mutations.rs

@@ -94,7 +94,7 @@ impl HostedProvider {
         let result = serde_json::to_value(plan.operation.to_v03()).map_err(|error| {
             ApiError::internal(format!("Hosted operation could not serialize: {error}"))
         })?;
-        if !plan.operation.valid {
+        if !plan.operation.is_valid() {
             transaction.commit().await?;
             return Ok(result);
         }

crates/connect-hosted-provider/src/provider/operation_types.rs

@@ -209,7 +209,7 @@ impl HostedProvider {
             "definition mutation",
         )?;
         let envelope = plan.operation.to_v03();
-        if !plan.operation.valid {
+        if !plan.operation.is_valid() {
             return serde_json::to_value(envelope).map_err(|error| {
                 ApiError::internal(format!("Hosted definition could not serialize: {error}"))
             });

crates/connect-hosted-provider/src/provider/policy.rs

@@ -223,10 +223,10 @@ pub(super) fn ensure_canonical_read_visible(
     operation: &mdbase::runtime::CanonicalOperationOutcome,
     allowed_types: &[String],
 ) -> ApiResult<()> {
-    if allowed_types.is_empty() || !operation.valid {
+    if allowed_types.is_empty() || !operation.is_valid() {
         return Ok(());
     }
-    let visible = match &operation.value {
+    let visible = match operation.value() {
         mdbase::runtime::CanonicalOperationValue::Read(Some(record)) => record
             .types
             .iter()

crates/connect-hosted-provider/src/provider/tests.rs

@@ -138,8 +138,8 @@ fn legacy_record_replay_never_uses_ambient_state_or_an_empty_success() {
 #[test]
 fn definition_decisions_match_closed_typed_fields() {
     let source = include_str!("operation_dispatch.rs");
-    assert!(source.contains("CanonicalOperationValue::TypePack(Some(value))"));
-    assert!(source.contains("CanonicalOperationValue::CollectionSetup(Some("));
+    assert!(source.contains("CanonicalOperationValue::AssessTypePack(Some(value))"));
+    assert!(source.contains("CanonicalOperationValue::AssessCollectionSetup(Some("));
     assert!(!source.contains("WireOnlyOperationValue::TypePack"));
     assert!(!source.contains("WireOnlyOperationValue::CollectionSetup"));
 }