logos-messaging/libchat
1
0
Fork 0
mirror of https://github.com/logos-messaging/libchat.git synced 2026-08-03 13:43:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Isolate delivery library (#178)

Browse Source 
* Isolate logos-delivery

* Update cargo.toml + fixups

* clippy fixes

* fix: topic

* remove tcp prefix from port
This commit is contained in:
Jazz Turner-Baggs 2026-07-13 08:51:17 -07:00 committed by GitHub
parent c089144dc9
commit 225b0fab14
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 636 additions and 280 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
Cargo.lock generated
View File

@ -2079,6 +2079,7 @@ dependencies = [
"base64",
"crossbeam-channel",
"libchat",
"logos-delivery",
"logos-generic-chat",
"serde",
"serde_json",
@ -3622,6 +3623,20 @@ dependencies = [
"logos-generic-chat",
]
[[package]]
name = "logos-delivery"
version = "0.1.0"
dependencies = [
"base64",
"crossbeam-channel",
"libchat",
"logos-generic-chat",
"serde",
"serde_json",
"thiserror",
"tracing",
]
[[package]]
name = "logos-generic-chat"
version = "0.1.0"

2
Cargo.toml
View File

@ -16,6 +16,7 @@ members = [
"crates/generic-chat",
"extensions/components",
"extensions/embedded-logos-delivery",
"extensions/logos-delivery-rust",
]
default-members = [
@ -39,6 +40,7 @@ crypto = { path = "core/crypto" }
embedded-logos-delivery = { path = "extensions/embedded-logos-delivery" }
libchat = { path = "core/conversations" }
logos-chat = { path = "crates/logos-chat" }
logos-delivery = { path = "extensions/logos-delivery-rust"}
logos-generic-chat = { path = "crates/generic-chat" }
shared-traits = { path = "core/shared-traits" }
storage = { path = "core/storage" }

2
bin/chat-cli/src/main.rs
View File

@ -79,7 +79,7 @@ fn main() -> Result<()> {
TransportKind::LogosDelivery => {
let mut p2p_config = P2pConfig::default();
if let Some(port) = cli.port {
p2p_config.tcp_port = port;
p2p_config.port = port;
}
if let Some(preset) = cli.preset.as_deref() {
p2p_config.preset = preset.to_string();

2
extensions/embedded-logos-delivery/Cargo.toml
View File

@ -2,12 +2,12 @@
name = "embedded-logos-delivery"
version = "0.1.0"
edition = "2024"
links = "logosdelivery"
[dependencies]
# Workspace dependencies (sorted)
crossbeam-channel = { workspace = true }
libchat = { workspace = true }
logos-delivery = { workspace = true }
logos-generic-chat = { workspace = true }
# External dependencies (sorted)

322
extensions/embedded-logos-delivery/src/lib.rs
View File

@ -1,59 +1,27 @@
//! The embedded logos-delivery transport service.
//!
//! [`EmbeddedLogosDelivery`] implements [`DeliveryService`] by wrapping an
//! embedded logos-delivery node running on a dedicated `std::thread`. All
//! interaction is via synchronous `std::sync::mpsc` channels.
//! [`EmbeddedLogosDelivery`] implements [`DeliveryService`] over the embedded
//! node owned by [`ThreadedDeliveryWrapper`]. The wrapper handles the node
//! thread and hands back raw [`WakuEvent`]s; this crate supplies the
//! delivery-specific mapping — content topics, `/logos-chat/1/…` filtering, and
//! payload decoding.
//!
//! This crate links the native `liblogosdelivery` library, so it lives outside
//! the workspace's default members; depend on it (e.g. via the `logos-chat`
//! crate) only when shipping the embedded node.
//! The native node is linked transitively via the `logos-delivery-rust` crate,
//! so this crate lives outside the workspace's default members; depend on it
//! (e.g. via the `logos-chat` crate) only when shipping the embedded node.
//!
//! ## Content topic mapping
//!
//! `AddressedEnvelope::delivery_address` maps to logos-delivery content topic
//! `/logos-chat/1/{delivery_address}/proto`.
pub(crate) mod sys;
pub(crate) mod wrapper;
use std::sync::{Arc, Mutex, mpsc};
use std::thread;
use std::time::Duration;
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64;
use crossbeam_channel::{Receiver, Sender};
use crossbeam_channel::Receiver;
use libchat::{AddressedEnvelope, DeliveryService};
use tracing::{error, info, warn};
use wrapper::LogosNodeCtx;
use logos_delivery::{ThreadedDeliveryWrapper, WakuEvent};
pub fn content_topic_for(delivery_address: &str) -> String {
format!("/logos-chat/1/{delivery_address}/proto")
}
// ── Error ────────────────────────────────────────────────────────────────────
#[derive(Debug, thiserror::Error)]
pub enum DeliveryError {
#[error("node startup failed: {0}")]
StartupFailed(String),
#[error("publish failed: {0}")]
PublishFailed(String),
#[error("send channel closed")]
ChannelClosed,
}
// ── Internals ────────────────────────────────────────────────────────────────
struct OutboundCmd {
message_json: String,
reply: mpsc::SyncSender<Result<(), DeliveryError>>,
}
type SubscriberList = Arc<Mutex<Vec<Sender<Vec<u8>>>>>;
// ── P2pConfig ───────────────────────────────────────────────────────────────────
pub use logos_delivery::{DeliveryError, P2pConfig};
use tracing::debug;
/// The logos-delivery network preset joined by default.
pub const DEFAULT_NETWORK_PRESET: &str = "logos.dev";
@ -61,68 +29,11 @@ pub const DEFAULT_NETWORK_PRESET: &str = "logos.dev";
/// Default TCP port for the embedded logos-delivery node.
pub const DEFAULT_TCP_PORT: u16 = 60000;
#[derive(Debug, Clone)]
pub struct P2pConfig {
pub preset: String,
pub tcp_port: u16,
pub log_level: String,
}
/// The content-topic prefix carrying logos-chat traffic.
const CHAT_TOPIC_PREFIX: &str = "/logos-chat/1/";
impl Default for P2pConfig {
fn default() -> Self {
Self {
preset: DEFAULT_NETWORK_PRESET.into(),
tcp_port: DEFAULT_TCP_PORT,
log_level: "ERROR".into(),
}
}
}
// ── Wire types ──────────────────────────────────────────────────────────────
/// Outbound message sent to the logos-delivery node.
#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct WakuMessage {
#[serde(rename = "contentTopic")]
content_topic: String,
/// Base64-encoded payload.
payload: String,
ephemeral: bool,
}
/// Top-level event envelope received from the logos-delivery node callback.
#[derive(Debug, serde::Deserialize)]
struct WakuEvent {
#[serde(rename = "eventType")]
event_type: String,
message: Option<ReceivedMessage>,
}
/// Message payload from a `message_received` event.
#[derive(Debug, serde::Deserialize)]
struct ReceivedMessage {
#[serde(rename = "contentTopic")]
content_topic: String,
/// The node may deliver the payload as either a base64 string or a JSON
/// array of byte values.
payload: WakuPayload,
}
/// Untagged union that handles both payload representations.
#[derive(Debug, serde::Deserialize)]
#[serde(untagged)]
enum WakuPayload {
Base64(String),
Bytes(Vec<u8>),
}
impl WakuPayload {
fn decode(self) -> Option<Vec<u8>> {
match self {
WakuPayload::Base64(s) => BASE64.decode(s).ok(),
WakuPayload::Bytes(b) => Some(b),
}
}
pub fn content_topic_for(delivery_address: &str) -> String {
format!("{CHAT_TOPIC_PREFIX}{delivery_address}/proto")
}
// ── EmbeddedLogosDelivery ──────────────────────────────────────────────────
@ -131,166 +42,28 @@ impl WakuPayload {
/// the same background node.
#[derive(Clone, Debug)]
pub struct EmbeddedLogosDelivery {
outbound: mpsc::SyncSender<OutboundCmd>,
#[allow(dead_code)]
subscribers: SubscriberList,
inbound_rx: Option<Receiver<Vec<u8>>>,
inner: ThreadedDeliveryWrapper<Vec<u8>>,
}
impl EmbeddedLogosDelivery {
/// Start the embedded logos-delivery node. The client drains inbound
/// payloads via [`Self::inbound_queue`].
/// Start the embedded logos-delivery node. Only chat payloads (on a
/// `/logos-chat/1/…` content topic) are kept on the inbound queue, decoded
/// to raw bytes.
pub fn start(cfg: P2pConfig) -> Result<Self, DeliveryError> {
let (out_tx, out_rx) = mpsc::sync_channel::<OutboundCmd>(256);
let subscribers: SubscriberList = Arc::new(Mutex::new(Vec::new()));
let (ready_tx, ready_rx) = mpsc::channel::<Result<(), DeliveryError>>();
// Create the inbound channel before spawning so the receiver is
// registered inside the thread, before any event callback fires.
let (inbound_tx, inbound_rx) = crossbeam_channel::bounded::<Vec<u8>>(1024);
let subs_for_thread = subscribers.clone();
let handle = thread::Builder::new()
.name("logos-node".into())
.spawn(move || {
if let Err(panic) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
Self::node_thread(cfg, out_rx, subs_for_thread, inbound_tx, ready_tx);
})) {
let msg = panic
.downcast_ref::<&str>()
.map(|s| s.to_string())
.or_else(|| panic.downcast_ref::<String>().cloned())
.unwrap_or_else(|| "unknown panic".into());
error!("logos-node thread panicked: {msg}");
}
})
.map_err(|e| DeliveryError::StartupFailed(e.to_string()))?;
// On failure, the node thread drops LogosNodeCtx (stop+destroy against
// a half-initialized Nim node). Join it so the process doesn't begin
// teardown mid-destroy — that race SIGSEGVs inside the Nim async loop.
let ready = ready_rx.recv().unwrap_or_else(|_| {
Err(DeliveryError::StartupFailed(
"node thread exited before ready".into(),
))
});
if let Err(e) = ready {
let _ = handle.join();
return Err(e);
}
Ok(Self {
outbound: out_tx,
subscribers,
inbound_rx: Some(inbound_rx),
})
}
fn node_thread(
cfg: P2pConfig,
out_rx: mpsc::Receiver<OutboundCmd>,
subscribers: SubscriberList,
inbound_tx: Sender<Vec<u8>>,
ready_tx: mpsc::Sender<Result<(), DeliveryError>>,
) {
// discv5UdpPort defaults to 9000 in libwaku, so a second instance with
// a distinct --port still collides on UDP. Bind it to tcp_port so a
// single --port knob keeps both ports distinct across instances.
let config_json = serde_json::json!({
"logLevel": cfg.log_level,
"mode": "Core",
"preset": cfg.preset,
"tcpPort": cfg.tcp_port,
"discv5UdpPort": cfg.tcp_port,
})
.to_string();
let mut node = match LogosNodeCtx::new(&config_json) {
Ok(n) => n,
Err(e) => {
let _ = ready_tx.send(Err(DeliveryError::StartupFailed(e)));
return;
let inner = ThreadedDeliveryWrapper::start(cfg, |event: WakuEvent| {
let msg = event.into_received()?;
if !msg.content_topic().starts_with(CHAT_TOPIC_PREFIX) {
return None;
}
};
msg.into_payload()
})?;
// Register the inbound sender before installing the event callback so
// there is no window where the callback is live but the channel is not
// yet in the subscriber list.
subscribers.lock().unwrap().push(inbound_tx);
let subs_for_cb = subscribers.clone();
let event_closure = move |_ret: i32, data: &str| {
if let Some(payload) = Self::parse_message_received(data) {
let mut guard = match subs_for_cb.lock() {
Ok(g) => g,
Err(e) => {
error!("subscriber mutex poisoned: {e}");
return;
}
};
guard.retain(|tx| match tx.try_send(payload.clone()) {
Ok(()) => true,
Err(crossbeam_channel::TrySendError::Full(_)) => true,
Err(crossbeam_channel::TrySendError::Disconnected(_)) => false,
});
}
};
node.set_event_callback(event_closure);
if let Err(e) = node.start() {
let _ = ready_tx.send(Err(DeliveryError::StartupFailed(e)));
return;
}
info!("logos-delivery node started (preset={})", cfg.preset);
// FIXME: This unconditional sleep is a stand-in for proper
// peer-connectivity detection. The right approach is to listen for a
// `peer_connected` (or equivalent status-change) event from the node
// callback and only proceed once at least one peer is reachable,
// falling back to a configurable timeout. logos-delivery would need to
// surface such an event via its callback mechanism for this to work.
thread::sleep(Duration::from_secs(3));
let default_topic = content_topic_for("delivery_address");
if let Err(e) = node.subscribe(&default_topic) {
warn!("subscribe to {default_topic}: {e}");
} else {
info!("subscribed to {default_topic}");
}
let _ = ready_tx.send(Ok(()));
while let Ok(cmd) = out_rx.recv() {
let result = node
.send(&cmd.message_json)
.map(|_| ())
.map_err(DeliveryError::PublishFailed);
let _ = cmd.reply.try_send(result);
}
info!("logos-node outbound loop finished");
Ok(Self { inner })
}
fn parse_message_received(data: &str) -> Option<Vec<u8>> {
let event: WakuEvent = serde_json::from_str(data).ok()?;
if event.event_type != "message_received" {
return None;
}
let msg = event.message?;
if !msg.content_topic.starts_with("/logos-chat/1/") {
return None;
}
msg.payload.decode()
}
pub fn inbound_queue(&mut self) -> Receiver<Vec<u8>> {
self.inbound_rx
.take()
.expect("inbound_queue called more than once")
/// Stop delivering messages addressed to `delivery_address`.
pub fn unsubscribe(&self, delivery_address: &str) -> Result<(), DeliveryError> {
self.inner.unsubscribe(&content_topic_for(delivery_address))
}
}
@ -298,28 +71,21 @@ impl DeliveryService for EmbeddedLogosDelivery {
type Error = DeliveryError;
fn publish(&mut self, envelope: AddressedEnvelope) -> Result<(), DeliveryError> {
let msg = WakuMessage {
content_topic: content_topic_for(&envelope.delivery_address),
payload: BASE64.encode(&envelope.data),
ephemeral: false,
};
let message_json =
serde_json::to_string(&msg).map_err(|e| DeliveryError::PublishFailed(e.to_string()))?;
let (reply_tx, reply_rx) = mpsc::sync_channel(1);
self.outbound
.send(OutboundCmd {
message_json,
reply: reply_tx,
})
.map_err(|_| DeliveryError::ChannelClosed)?;
reply_rx.recv().map_err(|_| DeliveryError::ChannelClosed)?
debug!(
topic = &content_topic_for(&envelope.delivery_address),
"Publish"
);
self.inner.publish(
&content_topic_for(&envelope.delivery_address),
&envelope.data,
)
}
fn subscribe(&mut self, _: &str) -> Result<(), <Self as DeliveryService>::Error> {
// This Service does not support filtering
Ok(())
fn subscribe(
&mut self,
delivery_address: &str,
) -> Result<(), <Self as DeliveryService>::Error> {
self.inner.subscribe(&content_topic_for(delivery_address))
}
}
@ -329,6 +95,6 @@ impl DeliveryService for EmbeddedLogosDelivery {
// owns neither the trait nor the type.
impl logos_generic_chat::Transport for EmbeddedLogosDelivery {
fn inbound(&mut self) -> Receiver<Vec<u8>> {
self.inbound_queue()
self.inner.inbound_queue()
}
}

18
extensions/logos-delivery-rust/Cargo.toml Normal file
View File

@ -0,0 +1,18 @@
[package]
name = "logos-delivery"
version = "0.1.0"
edition = "2024"
links = "logosdelivery"
[dependencies]
# Workspace dependencies (sorted)
crossbeam-channel = { workspace = true }
libchat = { workspace = true }
logos-generic-chat = { workspace = true }
# External dependencies (sorted)
base64 = "0.22"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
thiserror = "2"
tracing = "0.1"

160
extensions/logos-delivery-rust/build.rs Normal file
View File

@ -0,0 +1,160 @@
use std::fs;
use std::path::{Path, PathBuf};
use std::process::Command;
fn main() {
println!("cargo:rerun-if-env-changed=LOGOS_DELIVERY_LIB_DIR");
let Some(lib_dir) = locate_lib_dir() else {
println!(
"cargo:warning=liblogosdelivery could not be located; `cargo check`/\
`clippy` will pass, but building or testing will fail at link. Enter \
the dev shell with `nix develop` or set LOGOS_DELIVERY_LIB_DIR to \
the directory containing the library."
);
return;
};
let out_dir = std::env::var("OUT_DIR").expect("OUT_DIR not set");
let target_os = std::env::var("CARGO_CFG_TARGET_OS").unwrap_or_default();
// The shipped library carries a relocatable install name (@rpath on macOS,
// $ORIGIN soname on Linux), which would force every downstream BINARY to
// inject its own RPATH. Cargo propagates `rustc-link-search` and
// `rustc-link-lib` across crates, but NOT `rustc-link-arg` (the rpath) — so
// that relocatable name is exactly what makes consumers need their own
// build.rs. Instead, stamp a private copy with an ABSOLUTE install name;
// the propagating search + lib directives are then sufficient and consumers
// need zero build-script glue.
match target_os.as_str() {
"macos" => stamp_absolute_macos(&lib_dir, &out_dir),
"linux" => stamp_absolute_linux(&lib_dir, &out_dir),
other => panic!("unsupported OS for logos-delivery transport: {other}"),
}
println!("cargo:rustc-link-search=native={out_dir}");
println!("cargo:rustc-link-lib=dylib=logosdelivery");
}
/// Locate the native library directory as an ABSOLUTE, canonical path. Prefers
/// `LOGOS_DELIVERY_LIB_DIR`, then falls back to building it via nix. Returns
/// `None` when neither is available (e.g. `cargo check` without nix).
fn locate_lib_dir() -> Option<PathBuf> {
if let Ok(dir) = std::env::var("LOGOS_DELIVERY_LIB_DIR") {
if let Some(resolved) = resolve_lib_dir(&dir) {
return Some(resolved);
}
println!(
"cargo:warning=LOGOS_DELIVERY_LIB_DIR='{dir}' could not be resolved; \
falling back to `nix build`"
);
}
resolve_lib_dir(&nix_build_logos_delivery()?)
}
/// Resolve a lib dir to an absolute, canonical path. Cargo runs build scripts
/// with the cwd set to the crate dir, but a relative value (e.g. CI's
/// `./result/lib`) is anchored at the flake/workspace root where `nix build`
/// drops `result`. Canonicalizing also follows the `result` symlink to the
/// immutable store path, so the stamped install name / soname stays stable.
fn resolve_lib_dir(dir: &str) -> Option<PathBuf> {
let path = Path::new(dir);
let anchored = if path.is_absolute() {
path.to_path_buf()
} else {
let manifest = std::env::var("CARGO_MANIFEST_DIR").ok()?;
Path::new(&find_flake_root(&manifest)?).join(path)
};
anchored.canonicalize().ok()
}
/// Copy `liblogosdelivery.dylib` into `OUT_DIR` and rewrite its install name to
/// the absolute store path. The consumer records that absolute path, so dyld
/// loads the original file directly — whose own `@loader_path` RPATH resolves
/// `librln.dylib` beside it — with no RPATH needed on the consumer.
fn stamp_absolute_macos(lib_dir: &Path, out_dir: &str) {
let src = lib_dir.join("liblogosdelivery.dylib");
let dst = format!("{out_dir}/liblogosdelivery.dylib");
copy_writable(&src, Path::new(&dst));
run("install_name_tool", &["-id", path_str(&src), &dst]);
println!("cargo:rerun-if-changed={}", src.display());
}
/// Linux equivalent: an absolute `DT_SONAME` is recorded verbatim in the
/// consumer's `DT_NEEDED`, so `ld.so` loads it by path with no RPATH. Requires
/// `patchelf` at build time (provided by the nix devshell).
fn stamp_absolute_linux(lib_dir: &Path, out_dir: &str) {
let src = lib_dir.join("liblogosdelivery.so");
let dst = format!("{out_dir}/liblogosdelivery.so");
copy_writable(&src, Path::new(&dst));
run("patchelf", &["--set-soname", path_str(&src), &dst]);
println!("cargo:rerun-if-changed={}", src.display());
}
fn path_str(p: &Path) -> &str {
p.to_str()
.unwrap_or_else(|| panic!("non-UTF-8 path: {}", p.display()))
}
fn copy_writable(src: &Path, dst: &Path) {
use std::os::unix::fs::PermissionsExt;
fs::copy(src, dst)
.unwrap_or_else(|e| panic!("copy {} -> {}: {e}", src.display(), dst.display()));
// Store-sourced files are read-only; restore owner write so the install
// name / soname can be rewritten.
fs::set_permissions(dst, fs::Permissions::from_mode(0o644)).unwrap();
}
fn run(cmd: &str, args: &[&str]) {
let status = Command::new(cmd)
.args(args)
.status()
.unwrap_or_else(|e| panic!("failed to run `{cmd}`: {e}"));
assert!(status.success(), "`{cmd} {args:?}` failed with {status}");
}
fn nix_build_logos_delivery() -> Option<String> {
let manifest_dir = std::env::var("CARGO_MANIFEST_DIR").ok()?;
let flake_root = find_flake_root(&manifest_dir)?;
println!("cargo:rerun-if-changed={flake_root}/flake.lock");
let output = Command::new("nix")
.args([
"build",
".#logos-delivery",
"--no-link",
"--print-out-paths",
])
.current_dir(&flake_root)
.output()
.ok()?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
println!("cargo:warning=nix build .#logos-delivery failed: {stderr}");
return None;
}
let store_path = String::from_utf8(output.stdout).ok()?;
let lib_dir = format!("{}/lib", store_path.trim());
if std::path::Path::new(&lib_dir).exists() {
Some(lib_dir)
} else {
None
}
}
fn find_flake_root(start: &str) -> Option<String> {
let mut path = std::path::PathBuf::from(start);
loop {
if path.join("flake.nix").exists() {
return Some(path.to_string_lossy().into_owned());
}
if !path.pop() {
return None;
}
}
}

5
extensions/logos-delivery-rust/src/lib.rs Normal file
View File

@ -0,0 +1,5 @@
mod sys;
mod threaded;
mod wrapper;
pub use threaded::{DeliveryError, P2pConfig, ReceivedMessage, ThreadedDeliveryWrapper, WakuEvent};

0
extensions/embedded-logos-delivery/src/sys.rs → extensions/logos-delivery-rust/src/sys.rs
View File

360
extensions/logos-delivery-rust/src/threaded.rs Normal file
View File

@ -0,0 +1,360 @@
use std::sync::{Arc, Mutex, mpsc};
use std::thread;
use std::time::Duration;
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64;
use crossbeam_channel::{Receiver, Sender};
use tracing::{error, info};
use crate::wrapper::LogosNodeCtx;
// ── Error ────────────────────────────────────────────────────────────────────
#[derive(Debug, thiserror::Error)]
pub enum DeliveryError {
#[error("node startup failed: {0}")]
StartupFailed(String),
#[error("publish failed: {0}")]
PublishFailed(String),
#[error("subscribe failed: {0}")]
SubscribeFailed(String),
#[error("unsubscribe failed: {0}")]
UnsubscribeFailed(String),
#[error("send channel closed")]
ChannelClosed,
}
// ── Internals ────────────────────────────────────────────────────────────────
/// A node operation to run on the serialized node thread.
#[derive(Debug)]
enum NodeOp {
Publish(String), // message_json
Subscribe(String), // content_topic
Unsubscribe(String), // content_topic
}
#[derive(Debug)]
struct NodeCmd {
op: NodeOp,
reply: mpsc::SyncSender<Result<(), DeliveryError>>,
}
type SubscriberList<T> = Arc<Mutex<Vec<Sender<T>>>>;
// ── P2pConfig ───────────────────────────────────────────────────────────────────
/// The logos-delivery network preset joined by default.
pub const DEFAULT_NETWORK_PRESET: &str = "logos.dev";
/// Default TCP port for the embedded logos-delivery node.
pub const DEFAULT_PORT: u16 = 60000;
#[derive(Debug, Clone)]
pub struct P2pConfig {
pub preset: String,
pub port: u16,
pub log_level: String,
}
impl Default for P2pConfig {
fn default() -> Self {
Self {
preset: DEFAULT_NETWORK_PRESET.into(),
port: DEFAULT_PORT,
log_level: "ERROR".into(),
}
}
}
// ── Wire types ──────────────────────────────────────────────────────────────
/// Outbound message sent to the logos-delivery node.
#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct WakuMessage {
#[serde(rename = "contentTopic")]
content_topic: String,
/// Base64-encoded payload.
payload: String,
ephemeral: bool,
}
/// Top-level event envelope received from the logos-delivery node callback.
#[derive(Debug, serde::Deserialize, Clone)]
pub struct WakuEvent {
#[serde(rename = "eventType")]
event_type: String,
message: Option<ReceivedMessage>,
}
impl WakuEvent {
/// The received message iff this is a `message_received` event.
pub fn into_received(self) -> Option<ReceivedMessage> {
(self.event_type == "message_received")
.then_some(self.message)
.flatten()
}
}
/// Message payload from a `message_received` event.
#[derive(Debug, serde::Deserialize, Clone)]
pub struct ReceivedMessage {
#[serde(rename = "contentTopic")]
content_topic: String,
/// The node may deliver the payload as either a base64 string or a JSON
/// array of byte values.
payload: WakuPayload,
}
impl ReceivedMessage {
pub fn content_topic(&self) -> &str {
&self.content_topic
}
/// Decode the payload to raw bytes, whichever wire form the node used.
pub fn into_payload(self) -> Option<Vec<u8>> {
self.payload.decode()
}
}
/// Untagged union that handles both payload representations.
#[derive(Debug, serde::Deserialize, Clone)]
#[serde(untagged)]
enum WakuPayload {
Base64(String),
Bytes(Vec<u8>),
}
impl WakuPayload {
fn decode(self) -> Option<Vec<u8>> {
match self {
WakuPayload::Base64(s) => BASE64.decode(s).ok(),
WakuPayload::Bytes(b) => Some(b),
}
}
}
// ── ThreadedDeliveryWrapper ─────────────────────────────────────────────────
/// Owns the embedded node on a dedicated thread. Generic over the inbound item
/// type `T`: a caller-supplied mapper turns each raw [`WakuEvent`] into an
/// `Option<T>` on the callback thread, so filtering and decoding happen inline
/// with no relay thread. Cheap to clone — all clones share the same node.
pub struct ThreadedDeliveryWrapper<T = WakuEvent> {
outbound: mpsc::SyncSender<NodeCmd>,
#[allow(dead_code)]
subscribers: SubscriberList<T>,
inbound_rx: Option<Receiver<T>>,
}
// Manual impls so `T` carries no `Clone`/`Debug` bound at the struct level —
// `Sender<T>`/`Receiver<T>` are `Clone` for every `T`.
impl<T> Clone for ThreadedDeliveryWrapper<T> {
fn clone(&self) -> Self {
Self {
outbound: self.outbound.clone(),
subscribers: self.subscribers.clone(),
inbound_rx: self.inbound_rx.clone(),
}
}
}
impl<T> std::fmt::Debug for ThreadedDeliveryWrapper<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ThreadedDeliveryWrapper")
.field("has_inbound", &self.inbound_rx.is_some())
.finish_non_exhaustive()
}
}
impl<T> ThreadedDeliveryWrapper<T> {
/// Start the embedded logos-delivery node. `map` runs on the node's event
/// callback for every received event; return `Some(item)` to enqueue it for
/// [`Self::inbound_queue`], or `None` to drop it. It must be non-blocking.
pub fn start<F>(cfg: P2pConfig, map: F) -> Result<Self, DeliveryError>
where
T: Clone + Send + 'static,
F: FnMut(WakuEvent) -> Option<T> + Send + 'static,
{
let (out_tx, out_rx) = mpsc::sync_channel::<NodeCmd>(256);
let subscribers: SubscriberList<T> = Arc::new(Mutex::new(Vec::new()));
let (ready_tx, ready_rx) = mpsc::channel::<Result<(), DeliveryError>>();
// Create the inbound channel before spawning so the receiver is
// registered inside the thread, before any event callback fires.
let (inbound_tx, inbound_rx) = crossbeam_channel::bounded::<T>(1024);
let subs_for_thread = subscribers.clone();
let handle = thread::Builder::new()
.name("logos-node".into())
.spawn(move || {
if let Err(panic) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
Self::node_thread(cfg, out_rx, subs_for_thread, inbound_tx, ready_tx, map);
})) {
let msg = panic
.downcast_ref::<&str>()
.map(|s| s.to_string())
.or_else(|| panic.downcast_ref::<String>().cloned())
.unwrap_or_else(|| "unknown panic".into());
error!("logos-node thread panicked: {msg}");
}
})
.map_err(|e| DeliveryError::StartupFailed(e.to_string()))?;
// On failure, the node thread drops LogosNodeCtx (stop+destroy against
// a half-initialized Nim node). Join it so the process doesn't begin
// teardown mid-destroy — that race SIGSEGVs inside the Nim async loop.
let ready = ready_rx.recv().unwrap_or_else(|_| {
Err(DeliveryError::StartupFailed(
"node thread exited before ready".into(),
))
});
if let Err(e) = ready {
let _ = handle.join();
return Err(e);
}
Ok(Self {
outbound: out_tx,
subscribers,
inbound_rx: Some(inbound_rx),
})
}
/// Queue `op` on the node thread and block until it acknowledges.
fn send_cmd(&self, op: NodeOp) -> Result<(), DeliveryError> {
let (reply_tx, reply_rx) = mpsc::sync_channel(1);
self.outbound
.send(NodeCmd {
op,
reply: reply_tx,
})
.map_err(|_| DeliveryError::ChannelClosed)?;
reply_rx.recv().map_err(|_| DeliveryError::ChannelClosed)?
}
/// Start delivering messages on `content_topic`. Blocks until acknowledged.
pub fn subscribe(&self, content_topic: &str) -> Result<(), DeliveryError> {
self.send_cmd(NodeOp::Subscribe(content_topic.to_string()))
}
/// Stop delivering messages on `content_topic`. Blocks until acknowledged.
pub fn unsubscribe(&self, content_topic: &str) -> Result<(), DeliveryError> {
self.send_cmd(NodeOp::Unsubscribe(content_topic.to_string()))
}
/// Publish `payload` on `content_topic`. Blocks until the node acknowledges.
pub fn publish(&self, content_topic: &str, payload: &[u8]) -> Result<(), DeliveryError> {
let msg = WakuMessage {
content_topic: content_topic.to_string(),
payload: BASE64.encode(payload),
ephemeral: false,
};
let message_json =
serde_json::to_string(&msg).map_err(|e| DeliveryError::PublishFailed(e.to_string()))?;
self.send_cmd(NodeOp::Publish(message_json))
}
/// Take the inbound queue of mapped items. Callable once.
pub fn inbound_queue(&mut self) -> Receiver<T> {
self.inbound_rx
.take()
.expect("inbound_queue called more than once")
}
fn node_thread<F>(
cfg: P2pConfig,
out_rx: mpsc::Receiver<NodeCmd>,
subscribers: SubscriberList<T>,
inbound_tx: Sender<T>,
ready_tx: mpsc::Sender<Result<(), DeliveryError>>,
mut map: F,
) where
T: Clone + Send + 'static,
F: FnMut(WakuEvent) -> Option<T> + Send + 'static,
{
// discv5UdpPort defaults to 9000 in libwaku, so a second instance with
// a distinct --port still collides on UDP. Bind it to tcp_port so a
// single --port knob keeps both ports distinct across instances.
let config_json = serde_json::json!({
"logLevel": cfg.log_level,
"mode": "Core",
"preset": cfg.preset,
"tcpPort": cfg.port,
"discv5UdpPort": cfg.port,
})
.to_string();
let mut node = match LogosNodeCtx::new(&config_json) {
Ok(n) => n,
Err(e) => {
let _ = ready_tx.send(Err(DeliveryError::StartupFailed(e)));
return;
}
};
// Register the inbound sender before installing the event callback so
// there is no window where the callback is live but the channel is not
// yet in the subscriber list.
subscribers.lock().unwrap().push(inbound_tx);
let subs_for_cb = subscribers.clone();
let event_closure = move |_ret: i32, data: &str| {
let Ok(event) = serde_json::from_str::<WakuEvent>(data) else {
return;
};
let Some(item) = map(event) else {
return;
};
let mut guard = match subs_for_cb.lock() {
Ok(g) => g,
Err(e) => {
error!("subscriber mutex poisoned: {e}");
return;
}
};
guard.retain(|tx| match tx.try_send(item.clone()) {
Ok(()) => true,
Err(crossbeam_channel::TrySendError::Full(_)) => true,
Err(crossbeam_channel::TrySendError::Disconnected(_)) => false,
});
};
node.set_event_callback(event_closure);
if let Err(e) = node.start() {
let _ = ready_tx.send(Err(DeliveryError::StartupFailed(e)));
return;
}
info!("logos-delivery node started (preset={})", cfg.preset);
// FIXME: This unconditional sleep is a stand-in for proper
// peer-connectivity detection. The right approach is to listen for a
// `peer_connected` (or equivalent status-change) event from the node
// callback and only proceed once at least one peer is reachable,
// falling back to a configurable timeout. logos-delivery would need to
// surface such an event via its callback mechanism for this to work.
thread::sleep(Duration::from_secs(3));
let _ = ready_tx.send(Ok(()));
while let Ok(cmd) = out_rx.recv() {
info!(">>>>> {:?} ", cmd);
let result = match cmd.op {
NodeOp::Publish(msg) => node
.send(&msg)
.map(|_| ())
.map_err(DeliveryError::PublishFailed),
NodeOp::Subscribe(topic) => node
.subscribe(&topic)
.map_err(DeliveryError::SubscribeFailed),
NodeOp::Unsubscribe(topic) => node
.unsubscribe(&topic)
.map_err(DeliveryError::UnsubscribeFailed),
};
let _ = cmd.reply.try_send(result);
}
info!("logos-node command loop finished");
}
}

30
extensions/embedded-logos-delivery/src/wrapper.rs → extensions/logos-delivery-rust/src/wrapper.rs
View File

@ -126,6 +126,36 @@ impl LogosNodeCtx {
drop(unsafe { Box::from_raw(raw) });
return Err(format!("logosdelivery_subscribe returned {ret}"));
}
let result = rx
.recv()
.unwrap_or(Err("callback channel disconnected".into()));
drop(unsafe { Box::from_raw(raw) });
result
}
pub fn unsubscribe(&self, content_topic: &str) -> Result<(), String> {
let topic_cstr = CString::new(content_topic).map_err(|e| e.to_string())?;
let (tx, rx) = mpsc::sync_channel::<Result<(), String>>(1);
let closure = move |ret: i32, data: &str| {
let _ = tx.send(if ret == RET_OK {
Ok(())
} else {
Err(data.to_string())
});
};
let raw = Box::into_raw(Box::new(closure));
let cb = get_trampoline(unsafe { &*raw });
let ret = unsafe {
ffi::logosdelivery_unsubscribe(self.ctx, cb, raw as *const c_void, topic_cstr.as_ptr())
};
if ret != RET_OK {
drop(unsafe { Box::from_raw(raw) });
return Err(format!("logosdelivery_unsubscribe returned {ret}"));
}
let result = rx
.recv()
.unwrap_or(Err("callback channel disconnected".into()));