From 71c9351727e5cca9ec2de4aecb0eb6b2a86dfb7c Mon Sep 17 00:00:00 2001 From: moudyellaz Date: Sat, 4 Jul 2026 00:47:20 +0200 Subject: [PATCH] feat(cross-zone): add interactive two-zone chat demo app --- Cargo.lock | 21 + Cargo.toml | 2 + Justfile | 7 + lez/sequencer/core/src/cross_zone_watcher.rs | 7 +- tools/cross_zone_chat/Cargo.toml | 25 + tools/cross_zone_chat/src/index.html | 234 ++++++++ tools/cross_zone_chat/src/main.rs | 533 +++++++++++++++++++ 7 files changed, 828 insertions(+), 1 deletion(-) create mode 100644 tools/cross_zone_chat/Cargo.toml create mode 100644 tools/cross_zone_chat/src/index.html create mode 100644 tools/cross_zone_chat/src/main.rs diff --git a/Cargo.lock b/Cargo.lock index 21aaaafd..e5b096e8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1865,6 +1865,27 @@ dependencies = [ "risc0-zkvm", ] +[[package]] +name = "cross_zone_chat" +version = "0.1.0" +dependencies = [ + "anyhow", + "axum 0.8.9", + "common", + "cross_zone_inbox_core", + "cross_zone_outbox_core", + "env_logger", + "lee", + "log", + "ping_core", + "programs", + "risc0-zkvm", + "sequencer_service_rpc", + "serde", + "test_fixtures", + "tokio", +] + [[package]] name = "cross_zone_inbox_core" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 932325b4..99ad6879 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -65,6 +65,7 @@ members = [ "tools/cycle_bench", "tools/crypto_primitives_bench", "tools/integration_bench", + "tools/cross_zone_chat", ] [workspace.dependencies] @@ -194,6 +195,7 @@ elliptic-curve = { version = "0.13.8", features = ["arithmetic"] } actix-web = { version = "4.13.0", default-features = false, features = [ "macros", ] } +axum = "0.8.4" clap = { version = "4.5.42", features = ["derive", "env"] } reqwest = { version = "0.12", features = ["json", "rustls-tls", "stream"] } pyo3 = { version = "0.29", features = ["auto-initialize"] } diff --git a/Justfile b/Justfile index 86ba677c..8b5501aa 100644 --- a/Justfile +++ b/Justfile @@ -112,6 +112,13 @@ demo-cross-zone-bridge: @echo "🌉 Cross-zone bridge demo (lock on A, mint on B)" {{DEMO_ENV}} RISC0_DEV_MODE=1 cargo test -p integration_tests --release --test cross_zone_bridge -- --nocapture +# Demo: interactive cross-zone chat. Boots two zones on one Bedrock and serves a +# local two-column web UI; type in one zone and watch the message cross into the +# other. Two people can chat across the zones. Dev mode, no proving. +cross-zone-chat: + @echo "💬 Cross-zone chat demo — open the printed localhost URL" + {{DEMO_ENV}} RISC0_DEV_MODE=1 cargo run -p cross_zone_chat --release + # Clean runtime data clean: @echo "🧹 Cleaning run artifacts" diff --git a/lez/sequencer/core/src/cross_zone_watcher.rs b/lez/sequencer/core/src/cross_zone_watcher.rs index 560b1eee..ffa238c4 100644 --- a/lez/sequencer/core/src/cross_zone_watcher.rs +++ b/lez/sequencer/core/src/cross_zone_watcher.rs @@ -5,7 +5,7 @@ use cross_zone::{build_dispatch_from_emission, extract_emission}; use futures::StreamExt as _; use lee::PublicKey; use lee_core::program::ProgramId; -use log::{error, info, warn}; +use log::{debug, error, info, warn}; use logos_blockchain_core::mantle::ops::channel::ChannelId; use logos_blockchain_zone_sdk::{ CommonHttpClient, ZoneMessage, adapter::NodeHttpClient, indexer::ZoneIndexer, @@ -90,6 +90,11 @@ async fn watch_peer( }; match borsh::from_slice::(&zone_block.data) { Ok(block) => { + debug!( + "Watcher observed finalized peer {} block {}", + hex::encode(peer_zone), + block.header.block_id + ); // Reject blocks not signed by the pinned peer key (equivocation): // the channel signer is authenticated by the zone-sdk, but that // does not prove the peer's honest sequencer produced the block. diff --git a/tools/cross_zone_chat/Cargo.toml b/tools/cross_zone_chat/Cargo.toml new file mode 100644 index 00000000..2c74e6df --- /dev/null +++ b/tools/cross_zone_chat/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "cross_zone_chat" +version = "0.1.0" +edition = "2024" +license = { workspace = true } + +[lints] +workspace = true + +[dependencies] +test_fixtures.workspace = true +sequencer_service_rpc = { workspace = true, features = ["client"] } +programs.workspace = true +ping_core.workspace = true +cross_zone_outbox_core.workspace = true +cross_zone_inbox_core.workspace = true +lee.workspace = true +common.workspace = true +risc0-zkvm.workspace = true +tokio = { workspace = true, features = ["rt-multi-thread", "macros"] } +anyhow.workspace = true +serde = { workspace = true, features = ["derive"] } +axum.workspace = true +log.workspace = true +env_logger.workspace = true diff --git a/tools/cross_zone_chat/src/index.html b/tools/cross_zone_chat/src/index.html new file mode 100644 index 00000000..8dc5e82a --- /dev/null +++ b/tools/cross_zone_chat/src/index.html @@ -0,0 +1,234 @@ + + + + + +Cross-zone chat + + + +
+ Cross-zone chat — type in a column; each message shows its journey: sent into a source block, that block reaching Bedrock finality, then delivery into the other zone. The finality wait is the bulk of the latency. +
+
+
+

Zone A

+
+
+ + +
+
+
+

Zone B

+
+
+ + +
+
+
+ + + diff --git a/tools/cross_zone_chat/src/main.rs b/tools/cross_zone_chat/src/main.rs new file mode 100644 index 00000000..5a066e9e --- /dev/null +++ b/tools/cross_zone_chat/src/main.rs @@ -0,0 +1,533 @@ +//! Interactive cross-zone chat demo. +//! +//! Boots one Bedrock node hosting two zones (A and B), each running a sequencer +//! whose cross-zone watcher points at the other zone, then serves a local +//! two-column web UI. Type a message in one column and watch it cross Bedrock +//! and appear in the other; two people can chat across the zones. +//! +//! This is the `cross_zone_ping` integration-test topology made interactive: +//! a send builds a `ping_sender::Send` on the sender zone targeting the other +//! zone's `ping_receiver`, the other zone's watcher reads the finalized source +//! block and injects the inbox dispatch, and a per-zone block tailer decodes the +//! delivered `cross_zone_inbox::Dispatch` payloads back into chat text. +//! +//! To make the cross-zone machinery visible, each message is tracked through its +//! pipeline stages and surfaced to the page: submitted into a source block on the +//! sender zone, that block's Bedrock finality (pending -> finalized), and final +//! delivery into a block on the receiver zone. The Bedrock-finality wait is what +//! dominates the latency, so it is shown live. +//! +//! Prerequisite: a working local Docker daemon (Bedrock comes up via +//! `bedrock/docker-compose.yml`). Run with `RISC0_DEV_MODE=1` for fast, +//! proving-free latency; on macOS run under the `just cross-zone-chat` recipe so +//! the DYLD framework shim is set. + +#![allow( + clippy::arithmetic_side_effects, + clippy::print_stdout, + clippy::print_stderr, + clippy::unused_async, + clippy::needless_pass_by_value, + clippy::infinite_loop, + reason = "Demo binary: stdout banner is the deliverable; ordinal/elapsed arithmetic is \ + bounded at chat scale; the per-zone scanners and finality poller are daemon \ + loops that run for the process lifetime; axum handlers must be `async` and take \ + their extractors (State/Json/Query) by value to satisfy the framework's bounds." +)] + +use std::{ + collections::BTreeSet, + net::SocketAddr, + sync::{ + Arc, Mutex, + atomic::{AtomicU32, AtomicU64, Ordering}, + }, + time::{Duration, Instant}, +}; + +use anyhow::{Context as _, Result}; +use axum::{ + Json, Router, + extract::State, + http::StatusCode, + response::Html, + routing::{get, post}, +}; +use common::{block::BedrockStatus, transaction::LeeTransaction}; +use cross_zone_inbox_core::{CrossZoneConfig, CrossZonePeer, Instruction, ZoneId}; +use cross_zone_outbox_core::outbox_pda; +use lee::{ + ProgramId, PublicTransaction, + public_transaction::{Message, WitnessSet}, +}; +use log::{info, warn}; +use ping_core::{ReceiverInstruction, SenderInstruction, ping_record_pda}; +use sequencer_service_rpc::{RpcClient as _, SequencerClient, SequencerClientBuilder}; +use serde::{Deserialize, Serialize}; +use test_fixtures::{ + config::{self, SequencerPartialConfig, UrlProtocol, bedrock_channel_id, bedrock_channel_id_b}, + setup::{setup_bedrock_node, setup_sequencer}, +}; + +const HTTP_PORT: u16 = 8088; +const POLL_INTERVAL: Duration = Duration::from_secs(1); + +/// One chat message tracked through its cross-zone pipeline. Displayed in the +/// receiving zone's column; the stage fields drive the on-page timeline. +struct TrackedMessage { + id: u64, + /// Label ("A" / "B") of the sender zone. + source_label: &'static str, + /// Label of the receiver zone; the column this message shows in. + dest_label: &'static str, + text: String, + /// Outbox ordinal of this send, used to match the source-zone block. + ordinal: u32, + /// Block on the sender zone that included the `ping_sender` tx. + source_block: Option, + /// Whether `source_block` has reached Bedrock finality. + finalized: bool, + /// Block on the receiver zone that delivered the inbox dispatch. + delivered_block: Option, + created: Instant, + /// Wall-clock seconds from submit to delivery, set once delivered. + delivered_secs: Option, +} + +/// A single send to perform. +#[derive(Deserialize)] +struct SendRequest { + zone: String, + text: String, +} + +/// The id assigned to a freshly submitted message. +#[derive(Serialize)] +struct SendResponse { + id: u64, +} + +/// One message's current pipeline state, as served to the page. +#[derive(Serialize)] +struct TimelineEntry { + id: u64, + /// Receiver column the message belongs to. + zone: &'static str, + source_zone: &'static str, + text: String, + source_block: Option, + finalized: bool, + delivered_block: Option, + /// Seconds elapsed (live until delivered, then frozen at delivery time). + elapsed: u64, +} + +/// Everything one zone needs to send to its peer. +struct ZoneRuntime { + /// The peer zone's channel id; the target of sends from this zone. + other_zone: ZoneId, + client: SequencerClient, + /// Monotonic outbox ordinal per (this zone -> peer); each send must use a + /// fresh value because the outbox PDA is claimed only when default. + ordinal: AtomicU32, +} + +struct AppState { + zone_a: ZoneRuntime, + zone_b: ZoneRuntime, + /// Monotonic id assigned to every message. + next_id: AtomicU64, + messages: Mutex>, +} + +impl AppState { + fn zone(&self, label: &str) -> Option<&ZoneRuntime> { + match label { + "A" => Some(&self.zone_a), + "B" => Some(&self.zone_b), + _ => None, + } + } + + /// Records a freshly submitted outbound message and returns its id. + fn record_outbound( + &self, + source_label: &'static str, + dest_label: &'static str, + ordinal: u32, + text: String, + ) -> u64 { + let id = self.next_id.fetch_add(1, Ordering::SeqCst); + self.messages + .lock() + .expect("messages mutex poisoned") + .push(TrackedMessage { + id, + source_label, + dest_label, + text, + ordinal, + source_block: None, + finalized: false, + delivered_block: None, + created: Instant::now(), + delivered_secs: None, + }); + info!("[stage] msg {id} submitted {source_label}->{dest_label}"); + id + } + + /// Marks the source-zone block that included a send, matched by ordinal. + fn mark_source_block(&self, source_label: &'static str, ordinal: u32, block_id: u64) { + let mut messages = self.messages.lock().expect("messages mutex poisoned"); + if let Some(message) = messages.iter_mut().find(|m| { + m.source_label == source_label && m.ordinal == ordinal && m.source_block.is_none() + }) { + message.source_block = Some(block_id); + info!( + "[stage] msg {} in source block {source_label}#{block_id} (+{}s)", + message.id, + message.created.elapsed().as_secs() + ); + } + } + + /// Marks delivery on the receiver zone. Prefers the oldest undelivered + /// message whose text matches; falls back to the oldest undelivered for the + /// zone (delivery order is preserved end to end). + fn mark_delivered(&self, dest_label: &'static str, text: &str, block_id: u64) { + let mut messages = self.messages.lock().expect("messages mutex poisoned"); + let undelivered = + |m: &TrackedMessage| m.dest_label == dest_label && m.delivered_block.is_none(); + let index = messages + .iter() + .position(|m| undelivered(m) && m.text == text) + .or_else(|| messages.iter().position(undelivered)); + if let Some(index) = index { + let message = &mut messages[index]; + message.delivered_block = Some(block_id); + let secs = message.created.elapsed().as_secs(); + message.delivered_secs = Some(secs); + info!( + "[stage] msg {} delivered {dest_label}#{block_id} (+{secs}s total)", + message.id + ); + } + } + + /// Snapshot of (source zone, block) pairs whose finality is still unknown. + fn pending_finality(&self) -> BTreeSet<(&'static str, u64)> { + self.messages + .lock() + .expect("messages mutex poisoned") + .iter() + .filter(|m| !m.finalized) + .filter_map(|m| m.source_block.map(|block| (m.source_label, block))) + .collect() + } + + /// Marks every message whose source block has reached Bedrock finality. + fn mark_finalized(&self, source_label: &'static str, block_id: u64) { + for message in self + .messages + .lock() + .expect("messages mutex poisoned") + .iter_mut() + { + if message.source_label == source_label + && message.source_block == Some(block_id) + && !message.finalized + { + message.finalized = true; + info!( + "[stage] msg {} source block {source_label}#{block_id} finalized on Bedrock (+{}s)", + message.id, + message.created.elapsed().as_secs() + ); + } + } + } + + fn timeline(&self) -> Vec { + self.messages + .lock() + .expect("messages mutex poisoned") + .iter() + .map(|m| TimelineEntry { + id: m.id, + zone: m.dest_label, + source_zone: m.source_label, + text: m.text.clone(), + source_block: m.source_block, + finalized: m.finalized, + delivered_block: m.delivered_block, + elapsed: m + .delivered_secs + .unwrap_or_else(|| m.created.elapsed().as_secs()), + }) + .collect() + } +} + +#[tokio::main] +async fn main() -> Result<()> { + env_logger::init(); + + // Declared first so Bedrock outlives both zones (drops run in reverse order). + let (_bedrock, bedrock_addr) = setup_bedrock_node() + .await + .context("Failed to set up shared Bedrock node")?; + + let channel_a = bedrock_channel_id(); + let channel_b = bedrock_channel_id_b(); + let zone_a: ZoneId = *channel_a.as_ref(); + let zone_b: ZoneId = *channel_b.as_ref(); + let receiver_id = programs::ping_receiver().id(); + + // Each zone watches the other and may deliver only to ping_receiver. + let cross_zone_a = watch_peer(zone_b, receiver_id); + let cross_zone_b = watch_peer(zone_a, receiver_id); + + let partial = SequencerPartialConfig::default(); + let (seq_a, _home_a) = + setup_sequencer(partial, bedrock_addr, vec![], channel_a, Some(cross_zone_a)) + .await + .context("Failed to set up zone A sequencer")?; + let (seq_b, _home_b) = + setup_sequencer(partial, bedrock_addr, vec![], channel_b, Some(cross_zone_b)) + .await + .context("Failed to set up zone B sequencer")?; + + let state = Arc::new(AppState { + zone_a: ZoneRuntime { + other_zone: zone_b, + client: sequencer_client(seq_a.addr())?, + ordinal: AtomicU32::new(0), + }, + zone_b: ZoneRuntime { + other_zone: zone_a, + client: sequencer_client(seq_b.addr())?, + ordinal: AtomicU32::new(0), + }, + next_id: AtomicU64::new(1), + messages: Mutex::new(Vec::new()), + }); + + tokio::spawn(scan_zone(Arc::clone(&state), "A")); + tokio::spawn(scan_zone(Arc::clone(&state), "B")); + tokio::spawn(poll_finality(Arc::clone(&state))); + + let app = Router::new() + .route("/", get(index)) + .route("/send", post(send_handler)) + .route("/timeline", get(timeline_handler)) + .with_state(state); + + let addr = SocketAddr::from(([127, 0, 0, 1], HTTP_PORT)); + let listener = tokio::net::TcpListener::bind(addr) + .await + .with_context(|| format!("Failed to bind {addr}"))?; + + println!("\n 💬 Cross-zone chat ready — open http://127.0.0.1:{HTTP_PORT} in your browser\n"); + + axum::serve(listener, app) + .await + .context("HTTP server error")?; + Ok(()) +} + +/// A cross-zone config whose single peer is `peer`, allowed to deliver only to +/// `receiver_id`. +fn watch_peer(peer: ZoneId, receiver_id: ProgramId) -> CrossZoneConfig { + CrossZoneConfig { + peers: vec![CrossZonePeer { + channel_id: peer, + allowed_targets: vec![receiver_id], + expected_block_signing_pubkey: None, + }], + } +} + +fn sequencer_client(addr: SocketAddr) -> Result { + let url = + config::addr_to_url(UrlProtocol::Http, addr).context("Failed to build sequencer URL")?; + SequencerClientBuilder::default() + .build(url) + .context("Failed to build sequencer client") +} + +/// Scans one zone's new blocks. A `ping_sender` tx marks the message's source +/// block (its outbound leg); an inbox dispatch marks delivery on this zone. +/// Runs forever; transient RPC errors are logged and retried. +async fn scan_zone(state: Arc, label: &'static str) { + let inbox_id = programs::cross_zone_inbox().id(); + let sender_id = programs::ping_sender().id(); + let client = &state.zone(label).expect("zone runtime exists").client; + + // Start from the current tip so genesis/boot blocks are skipped. + let mut cursor = client.get_last_block_id().await.unwrap_or(0); + + loop { + tokio::time::sleep(POLL_INTERVAL).await; + + let last = match client.get_last_block_id().await { + Ok(last) => last, + Err(err) => { + warn!("[{label}] get_last_block_id failed: {err:#}"); + continue; + } + }; + + while cursor < last { + let next = cursor.saturating_add(1); + match client.get_block(next).await { + Ok(Some(block)) => { + for tx in &block.body.transactions { + let LeeTransaction::Public(public) = tx else { + continue; + }; + let program_id = public.message.program_id; + let data = &public.message.instruction_data; + // A tx targets at most one of these programs; check both + // independently rather than chaining (avoids an empty else). + if program_id == inbox_id + && let Some(text) = decode_inbox_text(data) + { + state.mark_delivered(label, &text, next); + } + if program_id == sender_id + && let Some(ordinal) = decode_send_ordinal(data) + { + state.mark_source_block(label, ordinal, next); + } + } + } + Ok(None) => {} + Err(err) => { + warn!("[{label}] get_block({next}) failed: {err:#}"); + break; + } + } + cursor = next; + } + } +} + +/// Polls each tracked source block's Bedrock finality until it is finalized. +async fn poll_finality(state: Arc) { + loop { + tokio::time::sleep(POLL_INTERVAL).await; + for (label, block_id) in state.pending_finality() { + let client = &state.zone(label).expect("zone runtime exists").client; + match client.get_block(block_id).await { + Ok(Some(block)) => { + if matches!(block.bedrock_status, BedrockStatus::Finalized) { + state.mark_finalized(label, block_id); + } + } + Ok(None) => {} + Err(err) => warn!("[{label}] finality get_block({block_id}) failed: {err:#}"), + } + } + } +} + +/// Recovers the chat text from an inbox dispatch tx's instruction data. +fn decode_inbox_text(instruction_data: &[u32]) -> Option { + let instruction: Instruction = + risc0_zkvm::serde::from_slice::(instruction_data).ok()?; + let Instruction::Dispatch(message) = instruction; + decode_payload(&message.payload) +} + +/// Recovers the outbox ordinal from a `ping_sender::Send` tx's instruction data. +fn decode_send_ordinal(instruction_data: &[u32]) -> Option { + let instruction: SenderInstruction = + risc0_zkvm::serde::from_slice::(instruction_data).ok()?; + let SenderInstruction::Send { ordinal, .. } = instruction; + Some(ordinal) +} + +/// Decodes a `ping_receiver::Record` payload (risc0 words in LE bytes) to text. +fn decode_payload(payload: &[u8]) -> Option { + let chunks = payload.chunks_exact(4); + if !chunks.remainder().is_empty() { + return None; + } + let words: Vec = chunks + .map(|chunk| u32::from_le_bytes(chunk.try_into().expect("chunks_exact(4) yields 4 bytes"))) + .collect(); + let instruction: ReceiverInstruction = + risc0_zkvm::serde::from_slice::(&words).ok()?; + let ReceiverInstruction::Record { payload: bytes } = instruction; + Some(String::from_utf8_lossy(&bytes).into_owned()) +} + +/// Builds the unsigned `ping_sender::Send` that carries `text` to `other_zone`, +/// mirroring `integration_tests/tests/cross_zone_ping.rs`. +fn build_send_tx(other_zone: ZoneId, ordinal: u32, text: &str) -> LeeTransaction { + let receiver_id = programs::ping_receiver().id(); + let outbox_id = programs::cross_zone_outbox().id(); + + let words = risc0_zkvm::serde::to_vec(&ReceiverInstruction::Record { + payload: text.as_bytes().to_vec(), + }) + .expect("serialize record instruction"); + let payload: Vec = words.iter().flat_map(|word| word.to_le_bytes()).collect(); + + let send = SenderInstruction::Send { + outbox_program_id: outbox_id, + target_zone: other_zone, + target_program_id: receiver_id, + target_accounts: vec![ping_record_pda(receiver_id).into_value()], + payload, + ordinal, + }; + + let outbox_account = outbox_pda(outbox_id, &other_zone, ordinal); + let message = Message::try_new( + programs::ping_sender().id(), + vec![outbox_account], + vec![], + send, + ) + .expect("build ping_sender message"); + + LeeTransaction::Public(PublicTransaction::new( + message, + WitnessSet::from_raw_parts(vec![]), + )) +} + +async fn index() -> Html<&'static str> { + Html(include_str!("index.html")) +} + +async fn send_handler( + State(state): State>, + Json(request): Json, +) -> Result, (StatusCode, String)> { + let text = request.text.trim(); + if text.is_empty() { + return Err((StatusCode::BAD_REQUEST, "empty message".to_owned())); + } + let (source_label, dest_label) = match request.zone.as_str() { + "A" => ("A", "B"), + "B" => ("B", "A"), + other => return Err((StatusCode::BAD_REQUEST, format!("unknown zone {other}"))), + }; + let zone = state.zone(source_label).expect("zone runtime exists"); + + let ordinal = zone.ordinal.fetch_add(1, Ordering::SeqCst); + let tx = build_send_tx(zone.other_zone, ordinal, text); + let id = state.record_outbound(source_label, dest_label, ordinal, text.to_owned()); + zone.client + .send_transaction(tx) + .await + .map_err(|err| (StatusCode::BAD_GATEWAY, format!("submit failed: {err:#}")))?; + Ok(Json(SendResponse { id })) +} + +async fn timeline_handler(State(state): State>) -> Json> { + Json(state.timeline()) +}