Share verified Cell publication and pipeline native replication

.github/workflows/coordination-model.yml

@@ -6,6 +6,10 @@ on:
       - "crates/cellule-runtime/model/**"
       - "crates/cellule-runtime/src/coordination/mod.rs"
       - "crates/cellule-runtime/src/coordination/**"
+      - "crates/cellule-runtime/src/control/**"
+      - "crates/cellule-runtime/src/follower/**"
+      - "crates/cellule-runtime/src/node/**"
+      - "crates/cellule-runtime/src/publication/**"
       - ".github/workflows/coordination-model.yml"
   schedule:
     - cron: "17 3 * * 1"
@@ -32,6 +36,9 @@ jobs:
           chmod +x crates/cellule-runtime/model/check.sh
           crates/cellule-runtime/model/check.sh fast
           crates/cellule-runtime/model/check.sh negative
+          crates/cellule-runtime/model/check.sh write-proofs
+          crates/cellule-runtime/model/check.sh bundle-coverage
+          crates/cellule-runtime/model/check.sh binding-drain
 
   broad:
     if: github.event_name != 'pull_request'

crates/cellule-axum/Cargo.toml

@@ -38,6 +38,7 @@ tempfile.workspace = true
 tokio = { workspace = true, features = ["macros", "net", "rt-multi-thread", "signal"] }
 tokio-util = { workspace = true, features = ["rt"] }
 tower = { version = "0.5", features = ["util"] }
+tracing-subscriber = "0.3"
 
 [[example]]
 name = "typed-api-service"

crates/cellule-axum/README.md

@@ -120,6 +120,10 @@ For REST routes `POST /orders` and `GET /orders/{id}`, run the
 cargo run -p cellule-axum --example sql --locked
 ```
 
+The SQL service writes runtime warnings, including the original cause when a
+command fences its Cell, to stderr. Logging stays in the embedding application;
+HTTP clients receive the same outcome-aware error envelope.
+
 ## Write a manual handler
 
 Add the adapter alongside Axum and your Cellule application crates:

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

@@ -1,6 +1,9 @@
 //! One serialized directory view for heartbeats and durability CAS callbacks.
 use super::*;
 use cellule_runtime::node::durability::NodeLogAuthority;
+use cellule_runtime::node::durability::{
+    BundleCheckpoint, NodeBundleAuthority, NodeBundlePublicationAuthority,
+};
 use cellule_runtime::node::log::{NodeLogRetirementObservation, NodeLogRotationBarrier};
 use cellule_runtime::node::{
     NodeAdvertisement, NodeCapacity, NodeFailureDomain, VersionedNodeAdvertisement,
@@ -151,6 +154,10 @@ impl Authority {
             .directory
             .recruit_log(&state.observed, 1, 16 << 20, 3, clock()?)
             .await?;
+        state.observed = self
+            .directory
+            .initialize_bundle_lane(&state.observed, 1, clock()?)
+            .await?;
         let members = state
             .observed
             .advertisement()
@@ -199,6 +206,139 @@ impl Authority {
     }
 }
 
+impl NodeBundleAuthority for Authority {
+    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 mut state = self.state.lock().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)
+        })
+    }
+
+    fn close<'a>(
+        &'a self,
+        authority: &'a cellule_runtime::control::authority::CellAuthority,
+        observed: &'a cellule_runtime::control::authority::VersionedControl,
+        issued: cellule_runtime::node::log::CellIssuedRange,
+    ) -> BoxFuture<'a, Result<()>> {
+        Box::pin(async move {
+            // Runtime joins the complete issued producer prefix before entering
+            // this mutex, including captures whose Fleet ACK preceded selection.
+            let mut state = self.state.lock().await;
+            let mut current = self.current(&state, issued.log_epoch()).await?;
+            let proof = self
+                .directory
+                .load_bundle_coverage(authority, observed, cellule_ltx::Limits::default())
+                .await?;
+            current = self
+                .directory
+                .checkpoint_bundle_cell(
+                    &current,
+                    authority,
+                    &proof,
+                    cellule_ltx::Limits::default(),
+                    clock()?,
+                )
+                .await?;
+            state.observed = current;
+            state.observed = self
+                .directory
+                .begin_bundle_close(&state.observed, proof.binding(), issued, clock()?)
+                .await?;
+            state.observed = self
+                .directory
+                .finish_bundle_close(&state.observed, proof.binding(), issued, clock()?)
+                .await?;
+            self.lease.check()
+        })
+    }
+}
+
+impl NodeBundlePublicationAuthority for Authority {
+    fn select<'a>(
+        &'a self,
+        captures: &'a [cellule_runtime::node::log_shipper::AssignedCapture],
+        lease: &'a NodeLeaseGuard,
+    ) -> BoxFuture<'a, Result<Vec<cellule_runtime::node::bundle::BundleCoverageProof>>> {
+        Box::pin(async move {
+            let mut state = self.state.lock().await;
+            let current = self.current(&state, 1).await?;
+            let frames = captures
+                .iter()
+                .flat_map(|c| c.frames().iter().cloned())
+                .collect::<Vec<_>>();
+            let assignments = captures.iter().map(|c| c.assignment()).collect::<Vec<_>>();
+            let prepared = self
+                .directory
+                .prepare_node_bundle(&current, &frames, &assignments, clock()?)
+                .await?;
+            let (next, proofs) = self
+                .directory
+                .select_node_bundle(
+                    &current,
+                    &prepared,
+                    lease,
+                    cellule_ltx::Limits::default(),
+                    clock()?,
+                )
+                .await?;
+            state.observed = next;
+            self.lease.check()?;
+            Ok(proofs)
+        })
+    }
+
+    fn checkpoint<'a>(&'a self, checkpoints: &'a [BundleCheckpoint]) -> BoxFuture<'a, Result<()>> {
+        Box::pin(async move {
+            let mut state = self.state.lock().await;
+            let current = self.current(&state, 1).await?;
+            let mut ready = Vec::new();
+            for checkpoint in checkpoints {
+                let root = checkpoint.root();
+                let observed = checkpoint
+                    .authority()
+                    .load(cellule_runtime::CellId::from_bytes(root.cell))
+                    .await?
+                    .ok_or(Error::Fenced)?;
+                let actual = observed.value().ltx_root().ok_or(Error::Fenced)?;
+                if observed.value().bundle_binding == Some(checkpoint.proof().binding())
+                    && actual.cell == root.cell
+                    && actual.incarnation == root.incarnation
+                    && actual.commit_sequence > root.commit_sequence
+                {
+                    // The later root's original notification is retained by its
+                    // joined publisher. This observation releases no locators.
+                    continue;
+                }
+                ready.push((checkpoint.authority(), checkpoint.proof()));
+            }
+            state.observed = if ready.is_empty() {
+                current
+            } else {
+                self.directory
+                    .checkpoint_bundle_cells(
+                        &current,
+                        &ready,
+                        cellule_ltx::Limits::default(),
+                        clock()?,
+                    )
+                    .await?
+            };
+            self.lease.check()
+        })
+    }
+}
+
 fn capacity(follower: Option<&cellule_runtime::follower::FollowerStore>) -> NodeCapacity {
     // Memory/disk hints are fixture admission ceilings, not measured physical
     // headroom. Follower retention and free budget come from the canonical store.

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

@@ -136,7 +136,9 @@ impl Config {
                 cellule_ltx::Limits::default(),
                 runtime.telemetry_handle(),
             )?;
-            runtime.install_node_durability(application, config.build()?)?;
+            let durability = config.build()?;
+            runtime.install_node_durability(application, durability.clone())?;
+            durability.start_bundle_publication(enrollment.authority.clone())?;
             Ok(())
         }
         .await;

crates/cellule-axum/examples/fleet/server.rs

@@ -1,4 +1,4 @@
-//! Private follower process: mTLS, fresh enrollment and canonical fsynced storage.
+//! Private follower process: signed grants, mTLS and canonical fsynced storage.
 use super::*;
 use axum::{
     Extension, Router,
@@ -9,7 +9,9 @@ use axum::{
 };
 use bytes::Bytes;
 use cellule_peer_http::PeerTlsIdentity;
-use cellule_runtime::follower::FollowerStore;
+use cellule_runtime::follower::{
+    AppendGrantIssuer, AppendGrantPeer, FollowerStore, GrantedFollowerAppend,
+};
 use ed25519_dalek::VerifyingKey;
 use prost::Message;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
@@ -41,6 +43,7 @@ pub(super) async fn serve(
         cellule_ltx::Limits::default(),
         cellule_ltx::DiskBudget::new(1 << 30),
     )?
+    .with_append_grant_receiver(session)
     .with_telemetry(
         cellule_runtime::fleet::telemetry::CellTelemetryHandle::from_sink(metrics.clone()),
     );
@@ -165,21 +168,30 @@ async fn handle_inner(server: Server, peer: PeerTlsIdentity, encoded: Bytes) ->
     if request.member != server.member.as_bytes() {
         return Err(Error::PeerAuthorization("capacity log member differs"));
     }
-    let enrolled = server
-        .metrics
-        .enrollment(
-            true,
+    let enrolled = if matches!(request.operation, 2..=4) {
+        Some(
             server
-                .directory
-                .peer_verifier(sender, peer.certificate(), peer.public_key(), clock()?),
+                .metrics
+                .enrollment(
+                    true,
+                    server.directory.peer_verifier(
+                        sender,
+                        peer.certificate(),
+                        peer.public_key(),
+                        clock()?,
+                    ),
+                )
+                .await?,
         )
-        .await;
-    let enrolled = enrolled?;
+    } else {
+        None
+    };
     // Directory I/O consumed time. Preserve the accepted horizon across wall
     // clock rollback, while monotonic elapsed time still expires the request.
     let now = wire::request_time(started_ms, clock()?, started_at.elapsed())?;
     validate_request(&request, now, "after directory verification")?;
     server.lease.check()?;
+    let deadline = request.deadline_ms;
     let mut reply = wire::Reply {
         member: server.member.as_bytes().to_vec(),
         request_digest: blake3::hash(&body).as_bytes().to_vec(),
@@ -190,20 +202,21 @@ async fn handle_inner(server: Server, peer: PeerTlsIdentity, encoded: Bytes) ->
             if sender != leader {
                 return Err(Error::PeerAuthorization("capacity append sender differs"));
             }
-            enrolled.authorize_log_append(
-                server.member,
-                request.epoch,
-                request.covered_through,
-                clock()?,
-            )?;
             let phase = std::time::Instant::now();
             let result = server
                 .store
-                .append(
-                    leader,
-                    request.epoch,
-                    request.frames.into_iter().map(Bytes::from).collect(),
-                    request.covered_through,
+                .append_granted(
+                    GrantedFollowerAppend {
+                        peer: AppendGrantPeer {
+                            session: sender,
+                            certificate: peer.certificate(),
+                            public_key: peer.public_key(),
+                        },
+                        log_epoch: request.epoch,
+                        grant: Digest::try_from(request.grant.as_slice())?,
+                        frames: request.frames.into_iter().map(Bytes::from).collect(),
+                    },
+                    server.lease.clone(),
                 )
                 .await;
             server
@@ -213,6 +226,33 @@ async fn handle_inner(server: Server, peer: PeerTlsIdentity, encoded: Bytes) ->
             reply.base_sequence = receipt.base_sequence;
             reply.durable_through = receipt.durable_through;
         }
+        5 => {
+            if sender != leader {
+                return Err(Error::PeerAuthorization("capacity grant sender differs"));
+            }
+            let grant = server
+                .metrics
+                .enrollment(
+                    true,
+                    server.store.open_append_grant(
+                        AppendGrantPeer {
+                            session: sender,
+                            certificate: peer.certificate(),
+                            public_key: peer.public_key(),
+                        },
+                        request.epoch,
+                        request.first_sequence,
+                        AppendGrantIssuer {
+                            directory: &server.directory,
+                            signing_key: server.tls.signing_key(),
+                            lease: &server.lease,
+                            now_ms: now,
+                        },
+                    ),
+                )
+                .await?;
+            reply.grant = grant.encode();
+        }
         3 => {
             if sender != leader {
                 return Err(Error::PeerAuthorization("capacity retire sender differs"));
@@ -262,7 +302,13 @@ async fn handle_inner(server: Server, peer: PeerTlsIdentity, encoded: Bytes) ->
         }
         _ => return Err(Error::PeerAuthorization("capacity operation differs")),
     }
+    drop(enrolled);
     server.lease.check()?;
+    if wire::request_time(started_ms, clock()?, started_at.elapsed())? >= deadline {
+        return Err(Error::PeerAuthorization(
+            "capacity reply exceeded request horizon",
+        ));
+    }
     let reply = wire::sign(
         reply.encode_to_vec(),
         server.tls.signing_key(),