perf(runtime): batch durable Cell enrollment

crates/cellule-axum/examples/application-builder-service/Cargo.toml

@@ -18,7 +18,7 @@ cellule-cookbook-support = { path = "../../../../cookbook/support" }
 cellule-runtime = { path = "../../../cellule-runtime" }
 cellule-store = { path = "../../../cellule-store" }
 object_store = { version = "0.14.1", default-features = false }
-rusqlite = { version = "0.34", features = ["bundled"] }
+rusqlite = { version = "0.39", features = ["bundled", "fallible_uint"] }
 serde_json = "1"
 tempfile = "3"
 tokio = { version = "1.49", features = ["macros", "net", "rt-multi-thread", "signal"] }

crates/cellule-axum/examples/fleet/authority/mod.rs

@@ -233,23 +233,67 @@ impl Authority {
     }
 }
 
+impl Authority {
+    async fn bind_cells(
+        &self,
+        cells: &[(
+            &cellule_runtime::control::authority::CellAuthority,
+            &cellule_runtime::control::authority::VersionedControl,
+        )],
+    ) -> Result<Vec<cellule_runtime::control::authority::VersionedControl>> {
+        let _catalog = self.catalog.lock().await;
+        let mut state = self.live_state().await?;
+        let current = self.current(&state, 1).await?;
+        let (next, pinned) = self
+            .directory
+            .bind_bundle_cells(&current, cells, clock()?)
+            .await?;
+        state.observed = next;
+        self.lease.check()?;
+        Ok(pinned)
+    }
+}
+
 impl NodeBundleAuthority for Authority {
+    fn bind_batch<'a>(
+        &'a self,
+        cells: &'a [(
+            &'a cellule_runtime::control::authority::CellAuthority,
+            &'a cellule_runtime::control::authority::VersionedControl,
+        )],
+    ) -> BoxFuture<'a, Option<Vec<Result<cellule_runtime::control::authority::VersionedControl>>>>
+    {
+        Box::pin(async move {
+            match self.bind_cells(cells).await {
+                // Declining releases both locks before individual callbacks;
+                // queued cells remain admissible under the original 4-MiB cap.
+                Err(Error::Capacity(
+                    "bundle selected shard bytes" | "bundle enrollment pin metadata",
+                )) => None,
+                Ok(pinned) => Some(pinned.into_iter().map(Ok).collect()),
+                Err(source) if cells.len() == 1 => Some(vec![Err(source)]),
+                Err(source) => {
+                    let source = Arc::new(source);
+                    Some(
+                        cells
+                            .iter()
+                            .map(|_| Err(Error::Shared(source.clone())))
+                            .collect(),
+                    )
+                }
+            }
+        })
+    }
     fn bind<'a>(
         &'a self,
         authority: &'a cellule_runtime::control::authority::CellAuthority,
         observed: &'a cellule_runtime::control::authority::VersionedControl,
     ) -> BoxFuture<'a, Result<cellule_runtime::control::authority::VersionedControl>> {
         Box::pin(async move {
-            let _catalog = self.catalog.lock().await;
-            let mut state = self.live_state().await?;
-            let current = self.current(&state, 1).await?;
-            let (next, pinned) = self
-                .directory
-                .bind_bundle_cell(&current, authority, observed, clock()?)
-                .await?;
-            state.observed = next;
-            self.lease.check()?;
-            Ok(pinned)
+            self.bind_cells(&[(authority, observed)])
+                .await?
+                .pop()
+                .ok_or(Error::Node("bundle enrollment lacks Cell"))
         })
     }
 

crates/cellule-axum/examples/fleet/authority/renewal_tests.rs

@@ -22,25 +22,49 @@ async fn authority() -> Arc<Authority> {
         code,
         code,
     );
-    let enrollment = Enrollment::start(
+    // Keep the production 30-second advertisement lifetime, but make its
+    // issue time unambiguously older than the queued renewal. Creating and
+    // renewing in the same wall-clock millisecond cannot advance authority.
+    let now = clock().unwrap();
+    let observed = directory
+        .create(
+            NodeAdvertisement::sign(
+                node(0),
+                SessionId::from_bytes([11; 16]),
+                "https://localhost:8081".into(),
+                tls.fleet(),
+                tls.certificate(),
+                code,
+                code,
+                tls.signing_key(),
+                1,
+                now - 1_000,
+                now + 29_000,
+                vec![code],
+                vec![1],
+                NodeFailureDomain::default(),
+                capacity(None),
+            )
+            .unwrap(),
+            now,
+        )
+        .await
+        .unwrap();
+    // A shorter local deadline remains conservative relative to that signed
+    // advertisement. The 20 queued jobs must still cross this one-second bound.
+    Arc::new(Authority {
         directory,
         tls,
-        0,
-        SessionId::from_bytes([11; 16]),
-        "https://localhost:8081".into(),
         code,
-        None,
-    )
-    .await
-    .unwrap();
-    enrollment.stop.send(true).unwrap();
-    enrollment.heartbeat.await.unwrap().unwrap();
-    // A shorter local deadline is conservative relative to the signed
-    // advertisement. It reproduces a nearly exhausted renewal window.
-    let mut authority = Arc::try_unwrap(enrollment.authority).ok().unwrap();
-    let now = clock().unwrap();
-    authority.lease = NodeLeaseGuard::new(now, now + 1_000).unwrap();
-    Arc::new(authority)
+        lease: NodeLeaseGuard::new(now, now + 1_000).unwrap(),
+        catalog: Mutex::new(()),
+        state: Mutex::new(State {
+            observed,
+            closed: None,
+        }),
+        follower: None,
+        shutdown_error: std::sync::OnceLock::new(),
+    })
 }
 
 #[tokio::test]