Support reliable CI readiness and multiple host-network nodes

.github/actions/compilation-cache/action.yml

@@ -5,23 +5,31 @@ inputs:
     description: Compilation owner whose cache should be reused
     required: false
     default: ${{ github.job }}
+  family:
+    description: Compilation cache family, including fallback isolation
+    required: false
+    default: release-ci-v1
+  directories:
+    description: Compilation output directories owned by this job
+    required: false
+    default: |
+      target/debug
+      target/release
+      sdk/target/release
+      sdk/target/sdist/release
+      target/iceberg-sdk/release
+      app/crowdb-diskio/build
+      lib/crowdb-tree/build
+      lib/crowdb-rpc/build
 runs:
   using: composite
   steps:
     - uses: Swatinem/rust-cache@v2
       with:
-        shared-key: release-ci-v1-${{ hashFiles('pixi.lock') }}
+        shared-key: ${{ inputs.family }}-${{ hashFiles('pixi.lock') }}
         key: ${{ inputs.key }}
         cache-targets: false
-        cache-directories: |
-          target/debug
-          target/release
-          sdk/target/release
-          sdk/target/sdist/release
-          target/iceberg-sdk/release
-          app/crowdb-diskio/build
-          lib/crowdb-tree/build
-          lib/crowdb-rpc/build
+        cache-directories: ${{ inputs.directories }}
         cmd-format: "pixi run {0}"
         cache-workspace-crates: true
         env-vars: "CARGO CC CXX CFLAGS CXXFLAGS CMAKE RUST CROWDB_KEEP_DEBUG_INFO"

.github/workflows/ci.yml

@@ -63,8 +63,6 @@ jobs:
       uses: ./.github/actions/compilation-cache
     - name: Clean env
       run: pixi run clean --runtime-only
-    - name: Build C++ libs
-      run: pixi run build
     - name: Run C++ and FFI tests
       run: pixi run test-cpp
     - name: Capture runner state on failure
@@ -171,8 +169,6 @@ jobs:
       uses: ./.github/actions/compilation-cache
     - name: Clean env
       run: pixi run clean --runtime-only
-    - name: Build C++ libs (crowdb-diskio subprocess)
-      run: pixi run build
     - name: Run storage component tests
       run: pixi run test-storage
     - name: Capture runner state on failure
@@ -212,6 +208,11 @@ jobs:
   AccessTests:
     timeout-minutes: 90
     runs-on: ubuntu-24.04
+    env:
+      # Keep function symbols and panic locations without duplicating native
+      # DWARF across hundreds of linked test executables.
+      CARGO_PROFILE_TEST_STRIP: debuginfo
+      CARGO_BUILD_JOBS: "4"
     steps:
     - uses: actions/checkout@v4
       with:
@@ -227,6 +228,12 @@ jobs:
       with:
         pixi-version: v0.77.0
     - uses: ./.github/actions/compilation-cache
+      with:
+        family: access-ci-v2
+        directories: |
+          target/debug
+          sdk/target/release
+          sdk/target/sdist/release
     - name: Clean env
       run: pixi run clean --runtime-only
     - name: Run access component tests
@@ -658,6 +665,11 @@ jobs:
     - uses: prefix-dev/setup-pixi@v0.8.1
       with:
         pixi-version: v0.77.0
+    - name: Install owned network fault tools
+      if: matrix.partition == 'console-api'
+      run: |
+        pixi run sudo apt-get update
+        pixi run sudo apt-get install -y iptables
     - uses: ./.github/actions/restore-compiled
       with:
         artifact: ${{ needs.BuildClients.outputs.artifact }}

app/crowdb-access-server/src/iceberg/file_http/injection.rs

@@ -57,18 +57,28 @@ impl FileHttp {
             )
             .await
             .map_err(|error| {
+                tracing::warn!(file = %owner.file, %error,
+                    "native injection receive failed; unpublished blocks retained for reclamation");
                 body.failure()
                     .map_or_else(|| admission_error(error), multipart::encoding_error)
             })?;
         crowdb_access_iceberg::file::FileSealer::new(self.blocks.clone(), self.limits.max_file_bytes)
             .map_err(|_| FileS3ErrorCode::InternalError)?
             .seal_uploaded(owner, file_request.location.clone(), tree.clone())
             .await
-            .map_err(multipart::seal_error)?;
+            .map_err(|error| {
+                tracing::warn!(file = %owner.file, %error,
+                    "native injection seal failed; unpublished blocks retained for reclamation");
+                multipart::seal_error(error)
+            })?;
         let file = FileRepository::new(self.store.clone())
             .finalize_native_parquet(admission.context(), owner, &stem, tree, locations)
             .await
-            .map_err(catalog_error)?;
+            .map_err(|error| {
+                tracing::warn!(file = %owner.file, %error,
+                    "native injection publication failed; candidate retained for reconciliation");
+                catalog_error(error)
+            })?;
         let bytes = serde_json::to_vec(&serde_json::json!({"location": file.location.to_string()}))
             .map_err(|_| FileS3ErrorCode::InternalError)?;
         let length = bytes.len();

app/crowdb-access-server/src/s3/dependencies.rs

@@ -9,6 +9,8 @@ use crowdb_access_s3::{metadata::MetadataKey, metrics::DependencyHealth};
 
 use super::operations::ProductionS3Operations;
 
+pub(super) const PROBE_BUDGET: Duration = Duration::from_secs(2);
+
 pub type DependencyProbeFuture = Pin<Box<dyn Future<Output = Option<StorageReadiness>> + Send + 'static>>;
 
 pub struct StorageReadiness {
@@ -18,8 +20,8 @@ pub struct StorageReadiness {
 
 impl ProductionS3Operations {
     pub(super) async fn observe_storage(&self) -> StorageReadiness {
-        let (metadata, chunks) = tokio::join!(
-            bounded("metadata", async {
+        observe_storage_dependencies(
+            async {
                 self.storage
                     .metadata
                     .get(
@@ -29,24 +31,50 @@ impl ProductionS3Operations {
                     .await
                     .map_err(|error| error.to_string())?;
                 Ok("Routed Chunk-KV bucket namespace read succeeded".into())
-            }),
-            async {
-                Box::pin(self.storage.chunks.observe_readiness(Duration::from_millis(750)))
+            },
+            |budget| async move {
+                Box::pin(self.storage.chunks.observe_readiness(budget))
                     .await
                     .map_err(|error| error.to_string())
             },
-        );
-        StorageReadiness { metadata, chunks }
+        )
+        .await
     }
+
+    #[cfg(feature = "test-util")]
+    /// Exercise the production dependency join and caller budget with controlled reads.
+    pub async fn observe_storage_dependencies_for_tests<M, C, F>(metadata: M, chunks: C) -> StorageReadiness
+    where
+        M: Future<Output = Result<String, String>>,
+        C: FnOnce(Duration) -> F,
+        F: Future<Output = Result<String, String>>,
+    {
+        observe_storage_dependencies(metadata, chunks).await
+    }
+}
+
+async fn observe_storage_dependencies<M, C, F>(metadata: M, chunks: C) -> StorageReadiness
+where
+    M: Future<Output = Result<String, String>>,
+    C: FnOnce(Duration) -> F,
+    F: Future<Output = Result<String, String>>,
+{
+    let (metadata, chunks) = tokio::join!(bounded("metadata", metadata), chunks(PROBE_BUDGET));
+    StorageReadiness { metadata, chunks }
 }
 
 pub(super) async fn bounded(
     dependency: &str,
     probe: impl Future<Output = Result<String, String>>,
 ) -> Result<String, String> {
-    tokio::time::timeout(Duration::from_millis(750), probe)
+    tokio::time::timeout(PROBE_BUDGET, probe)
         .await
-        .unwrap_or_else(|_| Err(format!("{dependency} read exceeded its 750 ms deadline")))
+        .unwrap_or_else(|_| {
+            Err(format!(
+                "{dependency} read exceeded its {} ms deadline",
+                PROBE_BUDGET.as_millis()
+            ))
+        })
 }
 
 pub(super) fn health(result: &Result<String, String>) -> DependencyHealth {

app/crowdb-access-server/src/s3/health.rs

@@ -41,6 +41,16 @@ impl S3HttpHandler for AccessHealthHandler {
     }
 }
 
+impl S3Dispatcher {
+    #[cfg(feature = "test-util")]
+    /// Read the production cached response without transport or dependency work.
+    #[must_use]
+    pub fn health_response_for_tests(&self) -> Response<ResponseBody> {
+        let observed = self.observations.latest.load_full();
+        readiness_response(observed.as_deref(), self.health.is_live(), false)
+    }
+}
+
 fn readiness_response(
     observed: Option<&HealthObservation>,
     live: bool,

app/crowdb-access-server/src/s3/health/observation.rs

@@ -17,6 +17,8 @@ use tokio::time::{Instant, MissedTickBehavior};
 
 use super::super::{dependencies, S3Dispatcher, StorageReadiness};
 
+const OBSERVATION_BUDGET: Duration = dependencies::PROBE_BUDGET.saturating_add(Duration::from_millis(100));
+
 pub(super) const MAX_AGE: Duration = Duration::from_secs(3);
 
 #[derive(Default)]
@@ -82,16 +84,43 @@ impl S3Dispatcher {
                 u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
             });
         let observed = tokio::time::timeout(
-            Duration::from_millis(800),
+            OBSERVATION_BUDGET,
             Arc::clone(&self.operations).observe_dependencies(),
         )
         .await
         .unwrap_or_else(|_| {
             Some(StorageReadiness {
-                metadata: Err("Dependency observation exceeded its 800 ms deadline".into()),
-                chunks: Err("Dependency observation exceeded its 800 ms deadline".into()),
+                metadata: Err(format!(
+                    "Dependency observation exceeded its {} ms deadline",
+                    OBSERVATION_BUDGET.as_millis()
+                )),
+                chunks: Err(format!(
+                    "Dependency observation exceeded its {} ms deadline",
+                    OBSERVATION_BUDGET.as_millis()
+                )),
             })
         });
+        let generation = self
+            .observations
+            .latest
+            .load()
+            .as_ref()
+            .map_or(1, |last| last.generation + 1);
+        if let Some(failed) = observed
+            .as_ref()
+            .filter(|value| value.metadata.is_err() || value.chunks.is_err())
+        {
+            tracing::warn!(
+                generation,
+                configured_budget_ms = u64::try_from(dependencies::PROBE_BUDGET.as_millis()).unwrap_or(u64::MAX),
+                observation_budget_ms = u64::try_from(OBSERVATION_BUDGET.as_millis()).unwrap_or(u64::MAX),
+                total_elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
+                checked_at_unix_ms,
+                metadata_error = ?failed.metadata.as_ref().err(),
+                chunks_error = ?failed.chunks.as_ref().err(),
+                "Access dependency observation failed; background sampling continues"
+            );
+        }
         let mut health = self
             .health
             .snapshot(&self.metrics, self.body_allocator.as_deref());
@@ -113,12 +142,6 @@ impl S3Dispatcher {
                 DependencyHealth::Unavailable
             };
         }
-        let generation = self
-            .observations
-            .latest
-            .load()
-            .as_ref()
-            .map_or(1, |last| last.generation + 1);
         HealthObservation {
             health,
             details,