mirror of
https://github.com/logos-blockchain/logos-execution-zone.git
synced 2026-07-11 16:29:45 +00:00
feat(cross-zone): add interactive two-zone chat demo app
This commit is contained in:
moudyellaz
parent
4f75e29a3c
commit
71c9351727
21
Cargo.lock
generated
21
Cargo.lock
generated
@ -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"
|
||||
|
||||
@ -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"] }
|
||||
|
||||
7
Justfile
7
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"
|
||||
|
||||
@ -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::<Block>(&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.
|
||||
|
||||
25
tools/cross_zone_chat/Cargo.toml
Normal file
25
tools/cross_zone_chat/Cargo.toml
Normal file
@ -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
|
||||
234
tools/cross_zone_chat/src/index.html
Normal file
234
tools/cross_zone_chat/src/index.html
Normal file
@ -0,0 +1,234 @@
|
||||
<!DOCTYPE html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="utf-8" />
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1" />
|
||||
<title>Cross-zone chat</title>
|
||||
<style>
|
||||
:root { color-scheme: light dark; }
|
||||
* { box-sizing: border-box; }
|
||||
body {
|
||||
margin: 0;
|
||||
font-family: system-ui, -apple-system, Segoe UI, Roboto, sans-serif;
|
||||
background: #0e1116;
|
||||
color: #e6edf3;
|
||||
}
|
||||
header {
|
||||
padding: 14px 20px;
|
||||
border-bottom: 1px solid #232a33;
|
||||
font-size: 15px;
|
||||
color: #9aa7b4;
|
||||
}
|
||||
header b { color: #e6edf3; }
|
||||
.columns { display: flex; height: calc(100vh - 51px); }
|
||||
.zone {
|
||||
flex: 1 1 0;
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
min-width: 0;
|
||||
}
|
||||
.zone + .zone { border-left: 1px solid #232a33; }
|
||||
.zone h2 {
|
||||
margin: 0;
|
||||
padding: 12px 18px;
|
||||
font-size: 14px;
|
||||
letter-spacing: .04em;
|
||||
text-transform: uppercase;
|
||||
color: #9aa7b4;
|
||||
border-bottom: 1px solid #232a33;
|
||||
}
|
||||
.zone.a h2 { color: #6cb6ff; }
|
||||
.zone.b h2 { color: #6bd968; }
|
||||
.log {
|
||||
flex: 1 1 auto;
|
||||
overflow-y: auto;
|
||||
padding: 14px 18px;
|
||||
display: flex;
|
||||
flex-direction: column;
|
||||
gap: 8px;
|
||||
}
|
||||
.msg {
|
||||
background: #161b22;
|
||||
border: 1px solid #232a33;
|
||||
border-radius: 10px;
|
||||
padding: 8px 12px;
|
||||
max-width: 90%;
|
||||
word-wrap: break-word;
|
||||
white-space: pre-wrap;
|
||||
}
|
||||
.msg.inflight { border-color: #3a3320; }
|
||||
.msg .text { margin-bottom: 6px; }
|
||||
.stages {
|
||||
display: flex;
|
||||
flex-wrap: wrap;
|
||||
gap: 6px;
|
||||
font-size: 11px;
|
||||
line-height: 1.4;
|
||||
color: #9aa7b4;
|
||||
}
|
||||
.stage {
|
||||
display: inline-flex;
|
||||
align-items: center;
|
||||
gap: 4px;
|
||||
padding: 1px 7px;
|
||||
border-radius: 999px;
|
||||
border: 1px solid #2d3640;
|
||||
background: #0e1116;
|
||||
white-space: nowrap;
|
||||
}
|
||||
.stage.todo { opacity: .4; }
|
||||
.stage.wait { border-color: #6b5a1f; color: #e3b341; }
|
||||
.stage.done { border-color: #2c5d33; color: #6bd968; }
|
||||
.stage .blk { color: #6cb6ff; font-variant-numeric: tabular-nums; }
|
||||
.elapsed { color: #7d8893; font-variant-numeric: tabular-nums; }
|
||||
.composer {
|
||||
display: flex;
|
||||
gap: 8px;
|
||||
padding: 12px 16px;
|
||||
border-top: 1px solid #232a33;
|
||||
}
|
||||
.composer input {
|
||||
flex: 1 1 auto;
|
||||
padding: 9px 12px;
|
||||
border-radius: 8px;
|
||||
border: 1px solid #2d3640;
|
||||
background: #0e1116;
|
||||
color: #e6edf3;
|
||||
font-size: 14px;
|
||||
}
|
||||
.composer button {
|
||||
padding: 9px 16px;
|
||||
border-radius: 8px;
|
||||
border: 0;
|
||||
background: #2563eb;
|
||||
color: #fff;
|
||||
font-size: 14px;
|
||||
cursor: pointer;
|
||||
}
|
||||
.composer button:disabled { opacity: .5; cursor: default; }
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<header>
|
||||
<b>Cross-zone chat</b> — type in a column; each message shows its journey: sent into a source block, that block reaching <span style="color:#e3b341">Bedrock finality</span>, then <span style="color:#6bd968">delivery</span> into the other zone. The finality wait is the bulk of the latency.
|
||||
</header>
|
||||
<div class="columns">
|
||||
<section class="zone a" data-zone="A">
|
||||
<h2>Zone A</h2>
|
||||
<div class="log" id="log-A"></div>
|
||||
<form class="composer" data-zone="A">
|
||||
<input type="text" placeholder="Message from Zone A → B" autocomplete="off" />
|
||||
<button type="submit">Send</button>
|
||||
</form>
|
||||
</section>
|
||||
<section class="zone b" data-zone="B">
|
||||
<h2>Zone B</h2>
|
||||
<div class="log" id="log-B"></div>
|
||||
<form class="composer" data-zone="B">
|
||||
<input type="text" placeholder="Message from Zone B → A" autocomplete="off" />
|
||||
<button type="submit">Send</button>
|
||||
</form>
|
||||
</section>
|
||||
</div>
|
||||
<script>
|
||||
// id -> { el, textEl, stagesEl } for in-place updates.
|
||||
const bubbles = new Map();
|
||||
|
||||
function stage(cls, label, blockZone, block) {
|
||||
const span = '<span class="stage ' + cls + '">' + label +
|
||||
(block != null ? ' <span class="blk">' + blockZone + '#' + block + '</span>' : '') +
|
||||
'</span>';
|
||||
return span;
|
||||
}
|
||||
|
||||
function renderStages(m) {
|
||||
const parts = [];
|
||||
// 1. submitted into a source block on the sender zone
|
||||
if (m.source_block != null) {
|
||||
parts.push(stage('done', 'sent', m.source_zone, m.source_block));
|
||||
} else {
|
||||
parts.push(stage('wait', 'sending…'));
|
||||
}
|
||||
// 2. Bedrock finality of that source block
|
||||
if (m.finalized) {
|
||||
parts.push(stage('done', 'Bedrock ✓ finalized'));
|
||||
} else if (m.source_block != null) {
|
||||
parts.push(stage('wait', 'Bedrock ⏳ pending'));
|
||||
} else {
|
||||
parts.push(stage('todo', 'Bedrock'));
|
||||
}
|
||||
// 3. delivery on the receiver zone
|
||||
if (m.delivered_block != null) {
|
||||
parts.push(stage('done', 'delivered', m.zone, m.delivered_block));
|
||||
} else {
|
||||
parts.push(stage(m.finalized ? 'wait' : 'todo', 'delivering…'));
|
||||
}
|
||||
// elapsed
|
||||
const t = m.delivered_block != null ? (m.elapsed + 's ✓') : (m.elapsed + 's');
|
||||
parts.push('<span class="elapsed">' + t + '</span>');
|
||||
return parts.join('');
|
||||
}
|
||||
|
||||
function upsert(m) {
|
||||
let b = bubbles.get(m.id);
|
||||
if (!b) {
|
||||
const log = document.getElementById('log-' + m.zone);
|
||||
const el = document.createElement('div');
|
||||
el.className = 'msg inflight';
|
||||
const textEl = document.createElement('div');
|
||||
textEl.className = 'text';
|
||||
textEl.textContent = m.text;
|
||||
const stagesEl = document.createElement('div');
|
||||
stagesEl.className = 'stages';
|
||||
el.appendChild(textEl);
|
||||
el.appendChild(stagesEl);
|
||||
log.appendChild(el);
|
||||
log.scrollTop = log.scrollHeight;
|
||||
b = { el, stagesEl };
|
||||
bubbles.set(m.id, b);
|
||||
}
|
||||
b.el.className = 'msg' + (m.delivered_block != null ? '' : ' inflight');
|
||||
b.stagesEl.innerHTML = renderStages(m);
|
||||
}
|
||||
|
||||
async function poll() {
|
||||
try {
|
||||
const res = await fetch('/timeline');
|
||||
if (!res.ok) return;
|
||||
const entries = await res.json();
|
||||
for (const m of entries) upsert(m);
|
||||
} catch (_) { /* transient; retry next tick */ }
|
||||
}
|
||||
|
||||
for (const form of document.querySelectorAll('form.composer')) {
|
||||
form.addEventListener('submit', async (ev) => {
|
||||
ev.preventDefault();
|
||||
const zone = form.dataset.zone;
|
||||
const input = form.querySelector('input');
|
||||
const text = input.value.trim();
|
||||
if (!text) return;
|
||||
input.value = '';
|
||||
try {
|
||||
const res = await fetch('/send', {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ zone, text }),
|
||||
});
|
||||
if (!res.ok) {
|
||||
input.value = text;
|
||||
input.placeholder = 'send failed: ' + (await res.text());
|
||||
} else {
|
||||
poll();
|
||||
}
|
||||
} catch (err) {
|
||||
input.value = text;
|
||||
input.placeholder = 'send error: ' + err;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
setInterval(poll, 1000);
|
||||
poll();
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
533
tools/cross_zone_chat/src/main.rs
Normal file
533
tools/cross_zone_chat/src/main.rs
Normal file
@ -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<u64>,
|
||||
/// Whether `source_block` has reached Bedrock finality.
|
||||
finalized: bool,
|
||||
/// Block on the receiver zone that delivered the inbox dispatch.
|
||||
delivered_block: Option<u64>,
|
||||
created: Instant,
|
||||
/// Wall-clock seconds from submit to delivery, set once delivered.
|
||||
delivered_secs: Option<u64>,
|
||||
}
|
||||
|
||||
/// 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<u64>,
|
||||
finalized: bool,
|
||||
delivered_block: Option<u64>,
|
||||
/// 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<Vec<TrackedMessage>>,
|
||||
}
|
||||
|
||||
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<TimelineEntry> {
|
||||
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<SequencerClient> {
|
||||
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<AppState>, 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<AppState>) {
|
||||
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<String> {
|
||||
let instruction: Instruction =
|
||||
risc0_zkvm::serde::from_slice::<Instruction, u32>(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<u32> {
|
||||
let instruction: SenderInstruction =
|
||||
risc0_zkvm::serde::from_slice::<SenderInstruction, u32>(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<String> {
|
||||
let chunks = payload.chunks_exact(4);
|
||||
if !chunks.remainder().is_empty() {
|
||||
return None;
|
||||
}
|
||||
let words: Vec<u32> = 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::<ReceiverInstruction, u32>(&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<u8> = 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<Arc<AppState>>,
|
||||
Json(request): Json<SendRequest>,
|
||||
) -> Result<Json<SendResponse>, (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<Arc<AppState>>) -> Json<Vec<TimelineEntry>> {
|
||||
Json(state.timeline())
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user