fix(mst2): finish chunk-map admission and isolate native observers

src/api/router/snapshot_persisted_metadata_tests.rs

@@ -685,14 +685,17 @@ async fn mst2_persisted_meta_reader_holds_protection_until_all_route_bytes_are_o
                 }
             })
         };
-        let observer = fixture
-            .state
-            .storage
-            .mono_storage()
-            .get_connection()
-            .begin()
-            .await
-            .unwrap();
+        // The two service connections belong to the reader and competitor.
+        // Observe their lock wait without competing for either connection.
+        let mut observer_config = fixture.state.storage.config().database.clone();
+        assert_eq!(observer_config.max_connection, 2);
+        observer_config.max_connection = 1;
+        observer_config.min_connection = 0;
+        let observer_connection =
+            crate::jupiter::storage::init::database_connection(&observer_config)
+                .await
+                .unwrap();
+        let observer = observer_connection.begin().await.unwrap();
         tokio::select! {
             () = wait_retention_waiter(&observer) => {},
             outcome = &mut competing => {

src/ceres/snapshot/chunk_map_gate.rs

@@ -25,7 +25,7 @@ pub(crate) struct InstallFlight {
 impl InstallFlight {
     pub(crate) fn acquire(key: [u8; 32]) -> Result<Self, SnapshotError> {
         static REGISTRY: OnceLock<Arc<Mutex<Registry>>> = OnceLock::new();
-        Self::from_registry(REGISTRY.get_or_init(|| Arc::default()).clone(), key)
+        Self::from_registry(REGISTRY.get_or_init(Arc::default).clone(), key)
     }
 
     fn from_registry(registry: Arc<Mutex<Registry>>, key: [u8; 32]) -> Result<Self, SnapshotError> {

src/contract/vault/pki.rs

@@ -385,8 +385,16 @@ mod tests_raw {
         .unwrap()
         .clone();
 
+        let issuance_started = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_secs();
         // issue cert
         let resp = test_write_api(core, "pki/issue/tls/test", true, Some(issue_data)).await;
+        let issuance_completed = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_secs();
         assert!(resp.is_ok());
         let resp_body = resp.unwrap();
         assert!(resp_body.is_some());
@@ -429,15 +437,10 @@ mod tests_raw {
         let ttl_compare = cert.not_after().compare(&expiration_time);
         assert!(ttl_compare.is_ok());
         assert_eq!(ttl_compare.unwrap(), std::cmp::Ordering::Equal);
-        let now_timestamp = SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .unwrap()
-            .as_secs();
         let expiration_ttl = cert_data["expiration"].as_u64().unwrap();
-        let ttl = expiration_ttl - now_timestamp;
         let expect_ttl = 10 * 24 * 60 * 60;
-        assert!(ttl <= expect_ttl);
-        assert!((ttl + 10) > expect_ttl);
+        assert!(expiration_ttl >= issuance_started + expect_ttl);
+        assert!(expiration_ttl <= issuance_completed + expect_ttl);
 
         let authority_key_id = cert.authority_key_id();
         assert!(authority_key_id.is_some());

src/jupiter/tests.rs

@@ -379,6 +379,7 @@ pub async fn test_storage_with_config(temp_dir: impl AsRef<Path>, config: Config
         app_service: Arc::new(svc),
         native_projection_cache: Arc::default(),
         native_snapshot_sessions: Arc::default(),
+        native_chunk_maps: Arc::default(),
         projection_observation_sink: None,
         cl_service: CLService::mock(),
         push_queue_service: PushQueueService::new(