logos-blockchain/logos-sql-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-sql-zone.git synced 2026-07-24 09:13:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

updated to work with latest zone-sdk

Browse Source
This commit is contained in:
kashepavadan 2026-06-19 17:04:12 -04:00
parent b6c991ccb3
commit ce09c5ba0e
5 changed files with 422 additions and 286 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

615
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

19
Cargo.toml
View File

@ -90,6 +90,11 @@ lb-tools = { default-features = false, package = "logo
lb-tui-zone = { default-features = false, package = "logos-blockchain-tui-zone", path = "./logos-blockchain/testnet/tui-zone" }
lbp-error = { default-features = false, package = "logos-blockchain-proofs-error", path = "./logos-blockchain/zk/proofs/error" }
lb_network = { default-features = false, package = "logos-blockchain-network-service", path = "./logos-blockchain/services/network" }
lbc-poc-sys = { default-features = false, git = "https://github.com/logos-blockchain/logos-blockchain-circuits.git", package = "logos-blockchain-circuits-poc-sys", tag = "v0.5.3" }
lbc-pol-sys = { default-features = false, git = "https://github.com/logos-blockchain/logos-blockchain-circuits.git", package = "logos-blockchain-circuits-pol-sys", tag = "v0.5.3" }
lbc-poq-sys = { default-features = false, git = "https://github.com/logos-blockchain/logos-blockchain-circuits.git", package = "logos-blockchain-circuits-poq-sys", tag = "v0.5.3" }
lbc-signature-sys = { default-features = false, git = "https://github.com/logos-blockchain/logos-blockchain-circuits.git", package = "logos-blockchain-circuits-signature-sys", tag = "v0.5.3" }
lbc-types = { default-features = false, git = "https://github.com/logos-blockchain/logos-blockchain-circuits.git", package = "logos-blockchain-circuits-types", tag = "v0.5.3" }
testing-framework-core = { default-features = false, git = "https://github.com/logos-blockchain/logos-blockchain-testing.git", rev = "f731791" }
testing-framework-runner-compose = { default-features = false, git = "https://github.com/logos-blockchain/logos-blockchain-testing.git", rev = "f731791" }
testing-framework-runner-k8s = { default-features = false, git = "https://github.com/logos-blockchain/logos-blockchain-testing.git", rev = "f731791" }
@ -114,11 +119,11 @@ log = { default-features = false, version = "0.4" }
overwatch = { default-features = false, git = "https://github.com/logos-co/Overwatch", rev = "448c192" }
overwatch-derive = { default-features = false, git = "https://github.com/logos-co/Overwatch", rev = "448c192" }
anyhow = { default-features = false, version = "1" }
ark-bn254 = { default-features = false, version = "0.4" }
ark-ec = { default-features = false, version = "0.4" }
ark-ff = { default-features = false, version = "0.4" }
ark-groth16 = { default-features = false, version = "0.4" }
ark-serialize = { default-features = false, version = "0.4.2" }
ark-bn254 = { default-features = false, version = "0.5" }
ark-ec = { default-features = false, version = "0.5" }
ark-ff = { default-features = false, version = "0.5" }
ark-groth16 = { default-features = false, version = "0.5" }
ark-serialize = { default-features = false, version = "0.5" }
astro-float = { default-features = false, version = "0.9" }
async-ctrlc = { default-features = false, version = "1.2" }
async-stream = { default-features = false, version = "0.3.6" }
@ -146,7 +151,8 @@ humantime = { default-features = false, version = "2.1"
hyper = { default-features = false, version = "1.0" }
igd-next = { default-features = false, version = "0.16.1" }
itertools = { default-features = false, version = "0.14" }
jf-poseidon2 = { default-features = false, git = "https://github.com/EspressoSystems/jellyfish.git", rev = "dc166cf0f803c3e5067f9dfcc21e3dade986a447" }
jf-poseidon2 = { default-features = false, git = "https://github.com/EspressoSystems/jellyfish.git", rev = "8d80230358e900f8d63765a937f63f4978ca1daa" }
rust-rapidsnark = { default-features = false, git = "https://github.com/logos-blockchain/logos-blockchain-rust-rapidsnark.git", rev = "e91187f8ccb5bbfc7bb00dac88169112428da78f" }
libp2p-swarm-test = { default-features = false, version = "0.5" }
multiaddr = { default-features = false, version = "0.18" }
natpmp = { default-features = false, version = "0.5" }
@ -178,6 +184,7 @@ rs-merkle-tree = { default-features = false, version = "0.1"
serde_arrays = { default-features = false, version = "0.2.0" }
serde_path_to_error = { default-features = false, version = "0.1" }
serde_yml = { default-features = false, version = "0.0.12" }
serde_urlencoded = { default-features = false, version = "0.7.1" }
serial_test = { default-features = false, version = "3.1.1" }
sntpc = { default-features = false, version = "0.5" }
strum = { default-features = false, version = "0.27" }

2
common/src/state.rs
View File

@ -1,5 +1,5 @@
use lb_core::mantle::ops::channel::MsgId;
use lb_zone_sdk::state::InscriptionInfo;
use lb_zone_sdk::sequencer::InscriptionInfo;
use crate::message::Msg;

2
indexer/src/indexer.rs
View File

@ -85,7 +85,7 @@ impl Indexer {
let logos_blockchain_zone_sdk::ZoneMessage::Block(zone_block) = zone_msg else {
continue;
};
let sql_text = match String::from_utf8(zone_block.data) {
let sql_text = match String::from_utf8(Vec::from(zone_block.data)) {
Ok(s) => s,
Err(e) => {
error!("Zone block data is not valid UTF-8: {e}");

70
sequencer/src/sequencer.rs
View File

@ -10,11 +10,11 @@ use std::{
use demo_sqlite_common::state::{InMemoryZoneState, ZoneState as _};
use fs2::FileExt as _;
use lb_common_http_client::{BasicAuthCredentials, CommonHttpClient};
use lb_core::mantle::ops::channel::ChannelId;
use lb_core::mantle::ops::channel::{ChannelId, inscribe::Inscription};
use lb_key_management_system_service::keys::{ED25519_SECRET_KEY_SIZE, Ed25519Key};
use logos_blockchain_zone_sdk::{
adapter::NodeHttpClient,
sequencer::{Event, SequencerCheckpoint, SequencerHandle, ZoneSequencer},
sequencer::{Event, FinalizedOp, OrphanedTx, SequencerCheckpoint, SequencerClient, ZoneSequencer},
};
use reqwest::Url;
use thiserror::Error;
@ -26,13 +26,15 @@ pub enum SequencerError {
Url(String),
#[error("IO error: {0}")]
Io(#[from] io::Error),
#[error("Inscription too large: {0}")]
InscriptionTooLarge(String),
}
pub type Result<T> = std::result::Result<T, SequencerError>;
pub struct Sequencer {
sequencer: ZoneSequencer<NodeHttpClient>,
handle: SequencerHandle<NodeHttpClient>,
client: SequencerClient,
state: InMemoryZoneState,
queue_file: String,
checkpoint_path: String,
@ -103,11 +105,12 @@ impl Sequencer {
.expect("failed to write channel id");
let node = NodeHttpClient::new(CommonHttpClient::new(basic_auth), node_url);
let (sequencer, handle) = ZoneSequencer::init(channel_id, signing_key, node, checkpoint);
let sequencer = ZoneSequencer::init(channel_id, signing_key, node, checkpoint);
let client = sequencer.client();
Ok(Self {
sequencer,
handle,
client,
state: InMemoryZoneState::default(),
queue_file: queue_file.to_owned(),
checkpoint_path: checkpoint_path.to_owned(),
@ -115,30 +118,33 @@ impl Sequencer {
}
pub async fn run(self) {
let Self { mut sequencer, handle, mut state, queue_file, checkpoint_path } = self;
let Self { mut sequencer, client, mut state, queue_file, checkpoint_path } = self;
let mut batch_handle = handle.clone();
let batch_client = client;
tokio::spawn(async move {
// Wait until the sequencer completes cold-start backfill before publishing.
let mut ready_rx = batch_client.subscribe_ready();
let _ = ready_rx.wait_for(|r| *r).await;
let mut interval = tokio::time::interval(Duration::from_millis(100));
loop {
interval.tick().await;
batch_handle.wait_ready().await;
if let Err(e) = process_pending_batch(&queue_file, &batch_handle).await {
if let Err(e) = process_pending_batch(&queue_file, &batch_client).await {
error!("Batch processing failed: {e}");
}
}
});
loop {
let Some(event) = sequencer.next_event().await else { continue; };
handle_event(event, &handle, &mut state, &checkpoint_path).await;
let event = sequencer.next_event().await;
handle_event(event, &mut sequencer, &mut state, &checkpoint_path);
}
}
}
async fn handle_event(
fn handle_event(
event: Event,
handle: &SequencerHandle<NodeHttpClient>,
sequencer: &mut ZoneSequencer<NodeHttpClient>,
state: &mut InMemoryZoneState,
checkpoint_path: &str,
) {
@ -146,34 +152,32 @@ async fn handle_event(
Event::Ready => {
info!("Sequencer ready");
}
Event::ChannelUpdate { orphaned, adopted } => {
state.on_adopted(&adopted);
for info in &orphaned {
Event::BlocksProcessed { checkpoint, channel_update, finalized } => {
state.on_adopted(&channel_update.adopted);
for orphan in &channel_update.orphaned {
let OrphanedTx::Inscription(info) = orphan else { continue };
state.on_orphaned(&info.this_msg);
debug!(msg_id = %hex::encode(info.this_msg.as_ref()), "Auto-republishing orphan");
if let Err(e) = handle.publish_message(info.payload.clone()).await {
if let Err(e) = sequencer.handle().publish(info.payload.clone()) {
error!("failed to auto-republish: {e}");
}
}
}
Event::TxsFinalized { inscriptions, .. } => {
let inscriptions: Vec<_> = finalized
.into_iter()
.flat_map(|tx| tx.ops.into_iter())
.filter_map(|op| match op {
FinalizedOp::Inscription(info) => Some(info),
_ => None,
})
.collect();
state.on_finalized(&inscriptions);
}
Event::Published { info, checkpoint } => {
debug!(msg_id = %hex::encode(info.this_msg.as_ref()), "Published");
state.on_published(&info);
save_checkpoint(Path::new(checkpoint_path), &checkpoint);
}
Event::FinalizedInscriptions { inscriptions } => {
state.on_finalized(&inscriptions);
}
Event::MempoolPending(_) | Event::TurnNotification { .. } => {}
}
}
async fn process_pending_batch(
queue_file: &str,
handle: &SequencerHandle<NodeHttpClient>,
) -> Result<()> {
async fn process_pending_batch(queue_file: &str, client: &SequencerClient) -> Result<()> {
let pending = queue_drain(queue_file)?;
if pending.is_empty() {
return Ok(());
@ -182,8 +186,10 @@ async fn process_pending_batch(
let count = pending.len();
debug!("Processing batch of {} queries", count);
let sql_text = pending.join("\n").as_bytes().to_vec();
if let Err(e) = handle.publish_message(sql_text).await {
let sql_bytes = pending.join("\n").into_bytes();
let inscription = Inscription::try_from(sql_bytes)
.map_err(|e| SequencerError::InscriptionTooLarge(e.to_string()))?;
if let Err(e) = client.publish(inscription).await {
error!("failed to publish batch: {e}");
} else {
info!("Submitted batch of {} statement(s)", count);