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

Merge branch 'main' into jazzz/account_service

Browse Source
This commit is contained in:
Jazz Turner-Baggs 2026-07-14 11:05:10 -07:00
commit 9e5abecc80
No known key found for this signature in database
50 changed files with 2126 additions and 441 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.github/scripts/mock-registry.py vendored
View File

@ -38,7 +38,7 @@ class Handler(BaseHTTPRequestHandler):
self._drain()
self._reply(404)
def log_message(self, *args):
def log_message(self, format, *args):
pass

12
.github/workflows/ci.yml vendored
View File

@ -20,12 +20,12 @@ jobs:
- run: rustup update stable && rustup default stable
# hashgraph-like-consensus's build.rs shells out to protoc via prost-build.
- run: sudo apt-get update && sudo apt-get install -y protobuf-compiler
# chat-cli pulls in components' embedded_p2p_delivery feature, whose
# build.rs links liblogosdelivery (built via Nix or LOGOS_DELIVERY_LIB_DIR).
# The smoketest job builds and exercises it under Nix; here we keep the
# toolchain-only job fast by skipping it.
- run: cargo build --verbose --workspace --exclude chat-cli
- run: cargo test --verbose --workspace --exclude chat-cli
# chat-cli and the embedded-logos-delivery crate link liblogosdelivery
# (built via Nix or LOGOS_DELIVERY_LIB_DIR). The smoketest job builds and
# exercises them under Nix; here we keep the toolchain-only job fast by
# skipping them.
- run: cargo build --verbose --workspace --exclude chat-cli --exclude embedded-logos-delivery
- run: cargo test --verbose --workspace --exclude chat-cli --exclude embedded-logos-delivery
clippy:
name: Clippy

47
Cargo.lock generated
View File

@ -1817,7 +1817,7 @@ dependencies = [
[[package]]
name = "de-mls"
version = "4.0.0"
source = "git+https://github.com/vacp2p/de-mls?branch=main#2b3eed17723d4fb12037ef8cfb1f5e8c115b09a1"
source = "git+https://github.com/vacp2p/de-mls?rev=2c7a8669c1492c749c02efd2c5ac45e93e4926a3#2c7a8669c1492c749c02efd2c5ac45e93e4926a3"
dependencies = [
"hashgraph-like-consensus",
"indexmap 2.14.0",
@ -2072,6 +2072,21 @@ dependencies = [
"zeroize",
]
[[package]]
name = "embedded-logos-delivery"
version = "0.1.0"
dependencies = [
"base64",
"crossbeam-channel",
"libchat",
"logos-delivery",
"logos-generic-chat",
"serde",
"serde_json",
"thiserror",
"tracing",
]
[[package]]
name = "enum-ordinalize"
version = "4.3.2"
@ -2529,9 +2544,9 @@ dependencies = [
[[package]]
name = "hashgraph-like-consensus"
version = "0.5.1"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "95b6328e1ba3b6ff66e24a018b40c6ea61a3105607755ac378003b57de7bc19d"
checksum = "cba53b85cc33f28262bbd9c258bbbf3f3b66501ef7cc2b7137712215fdf7cbf5"
dependencies = [
"alloy",
"alloy-signer",
@ -3599,6 +3614,32 @@ dependencies = [
[[package]]
name = "logos-chat"
version = "0.1.0"
dependencies = [
"components",
"crossbeam-channel",
"embedded-logos-delivery",
"libchat",
"logos-account",
"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"
dependencies = [
"chat-sqlite",
"components",

12
Cargo.toml
View File

@ -12,8 +12,11 @@ members = [
"core/shared-traits",
"core/sqlite",
"core/storage",
"crates/client",
"crates/logos-chat",
"crates/generic-chat",
"extensions/components",
"extensions/embedded-logos-delivery",
"extensions/logos-delivery-rust",
]
default-members = [
@ -25,7 +28,7 @@ default-members = [
"core/shared-traits",
"core/sqlite",
"core/storage",
"crates/client",
"crates/generic-chat",
]
[workspace.dependencies]
@ -34,8 +37,11 @@ logos-account = { path = "core/account" }
chat-sqlite = { path = "core/sqlite" }
components = { path = "extensions/components" }
crypto = { path = "core/crypto" }
embedded-logos-delivery = { path = "extensions/embedded-logos-delivery" }
libchat = { path = "core/conversations" }
logos-chat = { path = "crates/client" }
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/Cargo.toml
View File

@ -11,7 +11,7 @@ path = "src/main.rs"
# Workspace dependencies (sorted)
crossbeam-channel = { workspace = true }
logos-account = { workspace = true, features = ["dev"] }
logos-chat = { workspace = true, features = ["embedded-p2p-delivery"] }
logos-chat = { workspace = true }
# External dependencies (sorted)
anyhow = "1.0"

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

@ -8,11 +8,9 @@ use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use clap::{Parser, ValueEnum};
use crossbeam_channel::Receiver;
use logos_account::TestLogosAccount;
use logos_chat::{
AccountDirectory, ChatClient, ChatClientBuilder, ChatStore, DelegateSigner, Event,
HttpRegistry, LogosChatClient, NETWORK_PRESET, REGISTRY_ENDPOINT, RegistrationService,
StorageConfig, Transport,
AccountDirectory, ChatClient, ChatStore, Event, LogosConfig, P2pConfig, RegistrationService,
Transport,
};
use app::ChatApp;
@ -49,9 +47,10 @@ struct Cli {
#[arg(long)]
preset: Option<String>,
/// TCP port for the embedded logos-delivery node.
#[arg(long, default_value_t = 60000)]
port: u16,
/// TCP port for the embedded logos-delivery node. When omitted, the
/// preconfigured port is used.
#[arg(long)]
port: Option<u16>,
/// Write logs to a file instead of stderr (keeps TUI output clean).
#[arg(long)]
@ -77,54 +76,46 @@ fn main() -> Result<()> {
let db_str = db_path(&cli)?;
match cli.transport {
// logos-delivery is the transport baked into `LogosChatClient`, so the
// Logos client opens it from config rather than receiving one.
TransportKind::LogosDelivery => {
let preset = cli.preset.as_deref().unwrap_or(NETWORK_PRESET);
println!("Starting logos-delivery node (preset={preset})...");
let mut p2p_config = P2pConfig::default();
if let Some(port) = cli.port {
p2p_config.port = port;
}
if let Some(preset) = cli.preset.as_deref() {
p2p_config.preset = preset.to_string();
}
println!(
"Starting logos-delivery node (preset={})...",
p2p_config.preset
);
println!("This may take a few seconds while connecting to the network.");
let (client, events) = LogosChatClient::open(
db_str,
"chat-cli",
cli.port,
cli.preset.as_deref(),
cli.registry_url.as_deref(),
)
.map_err(|e| anyhow::anyhow!("{e:?}"))
.context("failed to open chat client")?;
let mut config = LogosConfig::new(db_str, "chat-cli");
if let Some(registry_url) = cli.registry_url.as_deref() {
config.set_registry_url(registry_url);
}
config.set_p2p_config(p2p_config);
let (client, events) = logos_chat::open(config)
.map_err(|e| anyhow::anyhow!("{e:?}"))
.context("failed to open chat client")?;
println!("Node connected.");
launch_tui(client, events, &cli)
}
// The file transport is a local-only path: it reuses the Logos service
// stack (delegate identity, HTTP registry, encrypted storage) but swaps
// the transport, so it builds a client directly instead of going through
// `LogosChatClient`.
// the transport in via `open_with_transport`.
TransportKind::File => {
let transport_dir = cli.data.join("transport");
let transport = transport::file::FileTransport::new(&transport_dir)
.context("failed to create file transport")?;
let endpoint = cli.registry_url.as_deref().unwrap_or(REGISTRY_ENDPOINT);
// A fresh dev account endorsing a fresh delegate each launch,
// mirroring `LogosChatClient::open`.
let account = TestLogosAccount::new();
let delegate = DelegateSigner::random();
let mut registry = HttpRegistry::new(endpoint);
account
.add_delegate_signer(&mut registry, delegate.public_key())
.map_err(|e| anyhow::anyhow!("{e:?}"))
.context("failed to publish the device bundle")?;
let (client, events) = ChatClientBuilder::new(account.address())
.ident(delegate)
.transport(transport)
.registration(registry)
.storage_config(StorageConfig::Encrypted {
path: db_str,
key: "chat-cli".to_string(),
})
.build()
let mut config = LogosConfig::new(db_str, "chat-cli");
if let Some(registry_url) = cli.registry_url.as_deref() {
config.set_registry_url(registry_url);
}
let (client, events) = logos_chat::open_with_transport(config, transport)
.map_err(|e| anyhow::anyhow!("{e:?}"))
.context("failed to open chat client")?;

4
core/conversations/Cargo.toml
View File

@ -18,9 +18,9 @@ storage = { workspace = true }
alloy = "2.0"
base64 = "0.22"
chat-proto = { git = "https://github.com/logos-messaging/chat_proto", rev = "37ec98a151f6d50aab2905802ac0a896477e62ea" }
de-mls = { git = "https://github.com/vacp2p/de-mls", branch = "main"}
de-mls = { git = "https://github.com/vacp2p/de-mls", rev = "2c7a8669c1492c749c02efd2c5ac45e93e4926a3"} # Expose Mls Extensions (#131)
double-ratchets = { path = "../double-ratchets" }
hashgraph-like-consensus = "0.5.1"
hashgraph-like-consensus = "0.6.0"
hex = "0.4.3"
openmls = { version = "0.8.1", features = ["libcrux-provider"] }
openmls_libcrux_crypto = "0.3.1"

12
core/conversations/src/conversation.rs
View File

@ -1,15 +1,17 @@
mod direct_v1;
pub mod group_v1;
mod group_v2;
pub mod mls_extensions;
mod privatev1;
pub use crate::errors::ChatError;
use crate::outcomes::ConvoOutcome;
use crate::proto::EncryptedPayload;
use crate::service_context::{ExternalServices, ServiceContext};
use crate::types::ConvoMetadata;
pub use direct_v1::DirectV1Convo;
pub use group_v1::GroupV1Convo;
pub use group_v2::GroupV2Convo;
pub use group_v2::{GroupV2Clock, GroupV2Convo};
pub use privatev1::PrivateV1Convo;
use shared_traits::IdentIdRef;
@ -42,6 +44,14 @@ pub(crate) trait GroupConvo<S: ExternalServices>: Convo<S> + std::fmt::Debug + S
cx: &mut ServiceContext<S>,
members: &[IdentIdRef],
) -> Result<(), ChatError>;
/// Each current member's MLS leaf-credential content (hex-encoded), self
/// included.
fn members(&self) -> Result<Vec<Vec<u8>>, ChatError>;
// All GroupConvos MUST return ConvoMetadata
// the return type is Option<_> to support legacy ConvoTypes which
// are being phased out.
fn metadata(&self) -> Option<ConvoMetadata>;
}
pub(crate) trait Identified {

13
core/conversations/src/conversation/group_v1.rs
View File

@ -16,6 +16,7 @@ use crate::conversation::ConversationIdRef;
use crate::inbox_v2::MlsProvider;
use crate::service_context::{ExternalServices, ServiceContext};
use crate::types::ConvoMetadata;
use crate::utils::{blake2b_hex, hash_size};
use crate::{
DeliveryService, IdentityProvider,
@ -341,4 +342,16 @@ impl<S: ExternalServices> GroupConvo<S> for GroupV1Convo {
self.send_payload(cx, commit.to_bytes()?)
}
fn members(&self) -> Result<Vec<Vec<u8>>, ChatError> {
Ok(self
.mls_group
.members()
.map(|m| m.credential.serialized_content().to_vec())
.collect())
}
fn metadata(&self) -> Option<ConvoMetadata> {
None
}
}

171
core/conversations/src/conversation/group_v2.rs
View File

@ -2,7 +2,10 @@
// DeMLS and Libchat have different execution models, trait definitions and ownership/lifetimes of objects.
// The easies path is to do a Spike to see what it would take, gather the friction points and then iterate.
use crate::types::AddressedEncryptedPayload;
use crate::conversation::mls_extensions::{
ConvoMetaInfo, GROUP_METADATA_EXTENSION_TYPE, capabilities_with_group_metadata,
};
use crate::types::{AddressedEncryptedPayload, ConvoMetadata};
use crate::{Content, WakeupService};
use alloy::signers::local::PrivateKeySigner;
use blake2::{Blake2b, Digest, digest::consts::U6};
@ -11,18 +14,20 @@ use de_mls::protos::de_mls::messages::v1::{
AppMessage as AppMessageProto, MemberWelcome, app_message,
};
use de_mls::{
Conversation, ConversationConfig, ConversationEvent, PeerScoringService, ScoringConfig,
Conversation, ConversationEvent, MockClock, PeerScoringService, ScoringConfig, WallClock,
default_score_deltas,
defaults::{DefaultConsensusPlugin, DefaultPeerScoring, InMemoryPeerScoreStorage},
};
use hashgraph_like_consensus::signing::EthereumConsensusSigner;
use openmls::extensions::{Extension, Extensions, UnknownExtension};
use openmls::group::MlsGroupCreateConfig;
use openmls::prelude::tls_codec::Deserialize as _;
use openmls::prelude::{KeyPackageIn, OpenMlsProvider as _, ProtocolVersion};
use openmls_traits::crypto::OpenMlsCrypto;
use prost::Message;
use shared_traits::{IdentId, IdentIdRef};
use std::sync::Arc;
use std::time::Duration;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tracing::{info, instrument, warn};
use crate::IdentityProvider;
@ -32,6 +37,29 @@ use crate::{
conversation::{ChatError, Convo, GroupConvo, Identified},
};
/// The de-mls time source: every conversation deadline (freeze windows,
/// consensus timeouts, auto-votes) and consensus wire timestamp is measured
/// against this clock. Production runs on system time; tests share one
/// `MockClock` with the harness scheduler so virtual time moves the
/// protocol's timers.
#[derive(Debug, Clone, Default)]
pub enum GroupV2Clock {
#[default]
System,
Mock(MockClock),
}
impl WallClock for GroupV2Clock {
fn now(&self) -> Duration {
match self {
GroupV2Clock::System => SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default(),
GroupV2Clock::Mock(clock) => clock.now(),
}
}
}
/// Local member id bytes — the account identity the protocol matches on,
/// shared with the MLS credential and the consensus member.
fn member_id<S: ExternalServices>(service_ctx: &ServiceContext<S>) -> Vec<u8> {
@ -58,24 +86,9 @@ fn make_consensus() -> DefaultConsensusPlugin {
DefaultConsensusPlugin::new(EthereumConsensusSigner::new(PrivateKeySigner::random()))
}
/// TEST-ONLY millisecond timers. de-mls deadlines are real wall-clock, so the
/// default 60s timers never fire under fast virtual time. Production needs a
/// real config injected from the caller, not these hardcoded values.
fn demls_config() -> ConversationConfig {
ConversationConfig {
commit_inactivity_duration: Duration::from_millis(50),
freeze_duration: Duration::from_millis(20),
voting_delay: Duration::from_millis(30),
election_voting_delay: Duration::from_millis(30),
consensus_timeout: Duration::from_millis(150),
proposal_expiration: Duration::from_millis(2000),
..ConversationConfig::default()
}
}
pub struct GroupV2Convo {
convo_id: String,
conversation: Conversation<DefaultConsensusPlugin, InMemoryPeerScoreStorage>,
conversation: Conversation<DefaultConsensusPlugin, InMemoryPeerScoreStorage, GroupV2Clock>,
/// Joiners WE invited, as `(member_id, signer_id)`: the de-mls member id
/// (the joiner's leaf credential content, read from its key package) paired
/// with the signer id its welcome is delivered to.
@ -95,28 +108,47 @@ fn rand_string(n: usize) -> String {
hex::encode(bytes)
}
fn group_config() -> MlsGroupCreateConfig {
fn group_config<S: ExternalServices>(
cx: &mut ServiceContext<S>,
name: &str,
desc: &str,
) -> MlsGroupCreateConfig {
let meta = ConvoMetaInfo::new(name, desc);
let extensions = Extensions::from_vec(vec![Extension::Unknown(
GROUP_METADATA_EXTENSION_TYPE,
UnknownExtension(meta.to_extension_bytes()),
)])
.expect("failed to create extensions");
MlsGroupCreateConfig::builder()
.use_ratchet_tree_extension(true)
.ciphersuite(cx.mls_provider.crypto().supported_ciphersuites()[0])
.capabilities(capabilities_with_group_metadata())
.use_ratchet_tree_extension(true) // Embed the ratchet tree in the Welcome so joiners can build the group
.with_group_context_extensions(extensions)
.build()
}
impl GroupV2Convo {
pub fn new<S: ExternalServices>(
service_ctx: &mut ServiceContext<S>,
name: &str,
desc: &str,
) -> Result<Self, ChatError> {
let convo_id = rand_string(5);
let group_config = group_config(service_ctx, name, desc);
let conversation = Conversation::create(
&convo_id,
&member_id(service_ctx),
&service_ctx.mls_provider,
service_ctx.mls_identity.get_credential(),
&group_config(),
&group_config,
&service_ctx.mls_identity,
&make_consensus(),
make_scoring(),
service_ctx.demls_clock.clone(),
rand_app_id(),
demls_config(),
service_ctx.demls_config.clone(),
)?;
let convo = GroupV2Convo {
convo_id,
@ -146,8 +178,9 @@ impl GroupV2Convo {
&welcome.conversation_sync_bytes,
&make_consensus(),
make_scoring(),
service_ctx.demls_clock.clone(),
rand_app_id(),
demls_config(),
service_ctx.demls_config.clone(),
)?
else {
return Err(ChatError::generic("welcome not addressed to this member"));
@ -277,14 +310,26 @@ where
service_ctx: &mut ServiceContext<S>,
members: &[IdentIdRef],
) -> Result<(), ChatError> {
// Record who WE invited before touching the conversation: after_op
// forwards a welcome only to joiners in pending_invites. Members are
// signer ids; the de-mls member id must match the id of the
// IdentityProvider that generated the key package (its MLS leaf
// credential content — de-mls matches members by credential), so it is
// read from the fetched key package rather than assumed equal to the
// signer id.
for member in members {
// Dedup the requested signers up front: an account can resolve to the
// same signer twice, or a caller can repeat one, and a duplicate would
// otherwise cost a redundant key-package fetch here and a second Add
// proposal for a member already being added in this batch.
let mut seen = std::collections::HashSet::new();
let members: Vec<IdentIdRef> = members
.iter()
.copied()
.filter(|m| seen.insert(m.as_str().to_string()))
.collect();
// Fetch and validate every key package before proposing any add, so a
// member with no key package fails the call before it opens proposals
// for the others. Members are signer ids; the de-mls member id must
// match the id of the IdentityProvider that generated the key package
// (its MLS leaf credential content — de-mls matches members by
// credential), so it is read from the fetched key package rather than
// assumed equal to the signer id.
let mut invites = Vec::with_capacity(members.len());
for member in &members {
let kp_bytes = service_ctx
.registry
.retrieve(member.as_str())
@ -298,17 +343,67 @@ where
.credential()
.serialized_content()
.to_vec();
self.pending_invites
.push((member_id.clone(), member.to_string()));
self.conversation.add_member(
invites.push((member_id, member.to_string(), kp_bytes));
}
// pending_invites drives welcome delivery: after_op forwards a welcome
// only to a joiner recorded here. Record a member only if de-mls will
// actually propose its add — recording one it silently drops strands an
// entry that a later re-join can match, firing a spurious duplicate
// welcome. de-mls drops self and members already in the group; and since
// add_member only opens a proposal, the committed roster won't reflect a
// member added earlier in this same loop, so the set tracks those too.
// Seed it with the roster and self, insert as we go, and one check
// covers all three.
let mut roster: std::collections::HashSet<Vec<u8>> =
self.conversation.members()?.into_iter().collect();
roster.insert(self.conversation.member_id_bytes().to_vec());
let mut result = Ok(());
for (member_id, signer_id, kp_bytes) in invites {
if !roster.insert(member_id.clone()) {
continue;
}
self.pending_invites.push((member_id.clone(), signer_id));
if let Err(e) = self.conversation.add_member(
&service_ctx.mls_provider,
&service_ctx.mls_identity,
&member_id,
&kp_bytes,
)?;
) {
self.pending_invites.pop();
result = Err(e.into());
break;
}
}
self.after_op(service_ctx)?;
Ok(())
// Flush even on a mid-loop failure: proposals already opened must be
// published and the wakeup re-armed, or they sit dormant until an
// unrelated frame drives the conversation.
let flushed = self.after_op(service_ctx).map(drop);
result.and(flushed)
}
fn members(&self) -> Result<Vec<Vec<u8>>, ChatError> {
// Guarantee the local member is listed so callers see the full roster.
let mut members = self.conversation.members()?;
let self_id = self.conversation.member_id_bytes().to_vec();
if !members.contains(&self_id) {
members.push(self_id);
}
Ok(members)
}
fn metadata(&self) -> Option<ConvoMetadata> {
let res = self.conversation.extensions().iter().find_map(|ext| {
if let Extension::Unknown(ext_type, UnknownExtension(bytes)) = ext
&& *ext_type == GROUP_METADATA_EXTENSION_TYPE
{
return ConvoMetaInfo::from_extension_bytes(bytes).ok();
};
None
});
res.map(Into::into)
}
// fn conversation_state(&self) -> Result<ConversationState, ChatError> {

120
core/conversations/src/conversation/mls_extensions.rs Normal file
View File

@ -0,0 +1,120 @@
use openmls::{
extensions::ExtensionType,
prelude::{
Capabilities,
tls_codec::{Deserialize, Error as TlsError, Serialize, Size, VLByteSlice, VLBytes},
},
};
use std::io::{Read, Write};
use crate::types::ConvoMetadata;
/// MLS extension type carrying our [`ConvoMetadata`]. In the private-use
/// range (0xF000–0xFFFF) reserved by RFC 9420 for non-registered extensions.
pub const GROUP_METADATA_EXTENSION_TYPE: u16 = 0xFF01;
pub fn capabilities_with_group_metadata() -> Capabilities {
Capabilities::new(
None, // default protocol versions
None, // default ciphersuites
Some(&[ExtensionType::Unknown(GROUP_METADATA_EXTENSION_TYPE)]),
None, // default proposal types
None, // default credential types
)
}
/// Wire-format version of [`ConvoMetaInfo`], encoded as a `u16`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u16)]
enum ConvoMetaInfoVersion {
V1 = 1,
}
impl TryFrom<u16> for ConvoMetaInfoVersion {
type Error = TlsError;
fn try_from(value: u16) -> Result<Self, Self::Error> {
match value {
1 => Ok(Self::V1),
other => Err(TlsError::DecodingError(format!(
"unknown ConvoMetaInfo version {other}"
))),
}
}
}
#[derive(Debug, Clone)]
pub struct ConvoMetaInfo {
version: ConvoMetaInfoVersion,
name: String,
desc: String,
}
impl ConvoMetaInfo {
pub fn new(name: impl Into<String>, desc: impl Into<String>) -> Self {
Self {
version: ConvoMetaInfoVersion::V1,
name: name.into(),
desc: desc.into(),
}
}
pub fn to_extension_bytes(&self) -> Vec<u8> {
// TLS presentation-language encoding — matches the MLS stack's wire
// format. Writing to a Vec is infallible; the only error path is a
// field exceeding tls_codec's ~1 GiB length cap, unreachable here.
self.tls_serialize_detached()
.expect("ConvoMetaInfo serialization to Vec is infallible")
}
pub fn from_extension_bytes(bytes: &[u8]) -> Result<Self, TlsError> {
Self::tls_deserialize(&mut &bytes[..])
}
}
// Each field is encoded as a variable-length opaque (`opaque <V>`); `IdentId`
// and `String` aren't `tls_codec` types, so we encode/decode their UTF-8 bytes.
impl Size for ConvoMetaInfo {
fn tls_serialized_len(&self) -> usize {
(self.version as u16).tls_serialized_len()
+ VLByteSlice(self.name.as_bytes()).tls_serialized_len()
+ VLByteSlice(self.desc.as_bytes()).tls_serialized_len()
}
}
impl Serialize for ConvoMetaInfo {
fn tls_serialize<W: Write>(&self, writer: &mut W) -> Result<usize, TlsError> {
let mut written = (self.version as u16).tls_serialize(writer)?;
written += VLByteSlice(self.name.as_bytes()).tls_serialize(writer)?;
written += VLByteSlice(self.desc.as_bytes()).tls_serialize(writer)?;
Ok(written)
}
}
impl Deserialize for ConvoMetaInfo {
fn tls_deserialize<R: Read>(bytes: &mut R) -> Result<Self, TlsError> {
let version = ConvoMetaInfoVersion::try_from(u16::tls_deserialize(bytes)?)?;
let name = vl_string(bytes)?;
let desc = vl_string(bytes)?;
Ok(Self {
version,
name,
desc,
})
}
}
fn vl_string<R: Read>(bytes: &mut R) -> Result<String, TlsError> {
let raw = VLBytes::tls_deserialize(bytes)?;
String::from_utf8(raw.into())
.map_err(|_| TlsError::DecodingError("invalid utf-8 in ConvoMetaInfo".into()))
}
impl From<ConvoMetaInfo> for ConvoMetadata {
fn from(value: ConvoMetaInfo) -> Self {
Self {
name: value.name,
desc: value.desc,
}
}
}

81
core/conversations/src/core.rs
View File

@ -3,7 +3,11 @@ use crate::conversation::{
ConversationIdRef, DirectV1Convo, GroupV1Convo, GroupV2Convo, Identified, PrivateV1Convo,
};
use crate::service_context::{ExternalServices, ServiceContext};
use crate::{DeliveryService, IdentityProvider, RegistrationService, WakeupService};
use crate::types::ConvoMetadata;
use crate::{
DeliveryService, GroupV2Clock, GroupV2Config, IdentityProvider, RegistrationService,
WakeupService,
};
use crate::{
conversation::{Convo, GroupConvo},
errors::ChatError,
@ -100,6 +104,17 @@ where
Ok(core)
}
pub fn set_group_v2_clock(&mut self, clock: GroupV2Clock) {
self.services.demls_clock = clock;
}
/// Overrides the GroupV2 (de-mls) timing/policy config. Applies to
/// conversations created/joined after the call; a creator's phase
/// durations reach joiners inside the welcome's `ConversationSync`.
pub fn set_group_v2_config(&mut self, config: GroupV2Config) {
self.services.demls_config = config;
}
/// Builds the inbox/account/MLS/causal state, subscribes both inbound
/// addresses, and assembles the service bundle — shared by both constructors.
fn assemble(
@ -140,6 +155,8 @@ where
causal,
identity,
wakeup_service,
demls_clock: GroupV2Clock::default(),
demls_config: GroupV2Config::default(),
},
inbox,
pq_inbox,
@ -224,7 +241,7 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
&mut self,
participants: &[IdentIdRef],
) -> Result<ConversationId, ChatError> {
self.create_group_convo_v2(participants)
self.create_group_convo_v2(participants, "", "")
}
pub fn create_group_convo_v1(
@ -253,11 +270,13 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
pub fn create_group_convo_v2(
&mut self,
participants: &[IdentIdRef],
name: &str,
desc: &str,
) -> Result<ConversationId, ChatError> {
// TODO: (P1) Ensure errors are handled properly. This is a high chance for
// desynchronized state: MlsGroup persistence, conversation persistence, and
// invite delivery all happen separately.
let mut convo = GroupV2Convo::new(&mut self.services)?;
let mut convo = GroupV2Convo::new(&mut self.services, name, desc)?;
convo.add_member(&mut self.services, participants)?;
let convo_id = convo.id().to_string();
@ -293,6 +312,28 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
}
}
/// Each member's MLS leaf-credential content (hex-encoded); errors if
/// `convo_id` names a direct (non-group) conversation.
pub fn group_members(&mut self, convo_id: &str) -> Result<Vec<Vec<u8>>, ChatError> {
if self.cached_convos.contains_key(convo_id) {
let convo = self
.cached_convos
.get(convo_id)
.ok_or_else(|| ChatError::NoConvo(convo_id.to_string()))?;
match convo {
ConvoTypeOwned::Group(group_convo) => group_convo.members(),
ConvoTypeOwned::Direct(convo) => Err(ChatError::UnsupportedFunction(
convo.id().into(),
"List Members".into(),
)),
}
} else {
let convo = self.load_group_convo(convo_id)?;
convo.members()
}
}
pub fn list_conversations(&self) -> Result<Vec<ConversationId>, ChatError> {
// Check Legacy load_convo store
let records = self.services.store.load_conversations()?;
@ -304,9 +345,13 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
convos.push(convo.to_string());
}
// Conversations may use both storage mechanisms.
// Remove duplicates
convos.dedup();
// A conversation can live in both the store and the in-memory cache (a
// DirectV1 join persists to the store and is also cached), so drop
// duplicates across the two. `Vec::dedup` only removes *consecutive*
// repeats and `cached_convos` iterates in nondeterministic HashMap
// order, so dedup through a set instead.
let mut seen = std::collections::HashSet::new();
convos.retain(|c| seen.insert(c.clone()));
Ok(convos)
}
@ -362,16 +407,13 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
// Dispatch encrypted payload to the post-quantum inbox.
fn dispatch_to_inbox2(&mut self, payload: &[u8]) -> Result<PayloadOutcome, ChatError> {
if let Some(convo) = self.pq_inbox.handle_frame(&mut self.services, payload)? {
if let Some((convo, class)) = self.pq_inbox.handle_frame(&mut self.services, payload)? {
let convo_id = convo.id().to_string();
// Cache convos created by InboxV2
self.register_convo(ConvoTypeOwned::Group(convo))?;
Ok(PayloadOutcome::Inbox(InboxOutcome {
new_conversation: crate::NewConversation {
convo_id,
class: crate::ConversationClass::Group,
},
new_conversation: crate::NewConversation { convo_id, class },
initial: None,
}))
} else {
@ -485,6 +527,23 @@ impl<'a, S: ExternalServices + 'static> Core<S> {
.load_conversation(convo_id)?
.ok_or_else(|| ChatError::NoConvo(convo_id.into()))
}
pub fn convo_metadata(&self, convo_id: ConversationIdRef) -> Result<ConvoMetadata, ChatError> {
match self.cached_convos.get(convo_id) {
Some(ConvoTypeOwned::Group(group_convo)) => {
group_convo
.metadata()
.ok_or(ChatError::UnsupportedConvoType(
"metadata is not available for this legacy convo_type".into(),
))
}
Some(ConvoTypeOwned::Direct(_)) => Err(ChatError::UnsupportedFunction(
convo_id.into(),
"implementation coming".into(),
)),
None => Err(ChatError::NoConvo(convo_id.into())),
}
}
}
enum ConvoTypeOwned<S: ExternalServices> {

39
core/conversations/src/inbox_v2.rs
View File

@ -20,6 +20,8 @@ use crate::conversation::GroupConvo;
use crate::conversation::GroupV1Convo;
use crate::conversation::GroupV2Convo;
use crate::conversation::Identified as _;
use crate::conversation::mls_extensions::GROUP_METADATA_EXTENSION_TYPE;
use crate::outcomes::ConversationClass;
use crate::service_context::{ExternalServices, ServiceContext};
use crate::utils::{blake2b_hex, hash_size};
use crate::{AddressedEnvelope, IdentId, IdentIdRef, IdentityProvider};
@ -70,6 +72,10 @@ pub fn invite_user_v2<DS: DeliveryService>(
.map_err(ChatError::generic)
}
/// A convo built from an InboxV2 invite, paired with the display class its
/// invite type implies.
type ClassifiedConvo<S> = (Box<dyn GroupConvo<S>>, ConversationClass);
/// A PQ focused Conversation initializer.
/// InboxV2 is signer-scoped: it receives invites under this installation's
/// signer id (the hex of the signer's verifying key), supporting PQ based
@ -95,8 +101,11 @@ impl InboxV2 {
) -> Result<(), ChatError> {
let keypackage_bytes = Self::create_keypackage(cx)?.tls_serialize_detached()?;
// TODO: (P3) Each keypackage can only be used once either enable...
// "LastResort" package or publish multiple
// TODO: publishes a single key package per installation. The intended
// design is a pool of one-time key packages (the registry pops one per
// fetch, the client replenishes) with the last-resort key package as the
// exhaustion fallback rather than the primary; that needs pop/claim
// semantics in the registry service. Tracked in #169.
cx.registry
.register(&cx.mls_identity, keypackage_bytes)
.map_err(ChatError::generic)?;
@ -112,12 +121,15 @@ impl InboxV2 {
conversation_id_for(&self.ident_id)
}
/// The convo built from an invite, paired with the display class its invite
/// type implies: `InviteType::GroupV1` carries the pairwise DirectV1 welcome,
/// so it is `Private`; `InviteType::GroupV2` is a real group.
#[instrument(name = "inboxV2.handle_frame", skip_all, fields(user_id = %service_ctx.mls_identity.display_name()))]
pub fn handle_frame<S: ExternalServices>(
&self,
service_ctx: &mut ServiceContext<S>,
payload_bytes: &[u8],
) -> Result<Option<Box<dyn GroupConvo<S>>>, ChatError> {
) -> Result<Option<ClassifiedConvo<S>>, ChatError> {
// On a broadcast transport the inbox address also receives traffic
// that isn't an invite (or that prost decodes into an empty frame).
// Treat anything we can't interpret as "not for us" and skip it,
@ -131,14 +143,15 @@ impl InboxV2 {
match payload {
InviteType::GroupV1(inv) => {
Ok(Some(Box::new(self.handle_heavy_invite(service_ctx, inv)?)))
let convo = self.handle_heavy_invite(service_ctx, inv)?;
Ok(Some((Box::new(convo), ConversationClass::Private)))
}
InviteType::GroupV2(welcome_bytes) => {
info!("Process WelcomeMessage");
let mw =
MemberWelcome::decode(welcome_bytes.as_slice()).map_err(ChatError::generic)?;
let convo = GroupV2Convo::new_from_welcome(service_ctx, &mw)?;
Ok(Some(Box::new(convo)))
Ok(Some((Box::new(convo), ConversationClass::Group)))
}
}
}
@ -183,11 +196,23 @@ impl InboxV2 {
fn create_keypackage<S: ExternalServices>(
cx: &ServiceContext<S>,
) -> Result<KeyPackage, ChatError> {
// Last-resort key package. openmls consumes (deletes) a normal key
// package's init key on the first welcome that uses it; since each
// installation publishes just one, a second group inviting it would find
// no matching key package and reject the welcome ("welcome not addressed
// to this member"). Last-resort key packages are retained, so one admits
// the installation to any number of groups. Every key-package extension
// must be advertised in the leaf capabilities, hence LastResort there.
let capabilities = Capabilities::builder()
.ciphersuites(vec![CIPHER_SUITE])
.extensions(vec![ExtensionType::ApplicationId])
.extensions(vec![
ExtensionType::ApplicationId,
ExtensionType::LastResort,
ExtensionType::Unknown(GROUP_METADATA_EXTENSION_TYPE),
])
.build();
let a = KeyPackage::builder()
.mark_as_last_resort()
.leaf_node_capabilities(capabilities)
.build(
CIPHER_SUITE,
@ -203,7 +228,7 @@ impl InboxV2 {
#[derive(Clone, PartialEq, Message)]
pub struct InboxV2Frame {
#[prost(oneof = "InviteType", tags = "1, 2")]
#[prost(oneof = "InviteType", tags = "1, 2, 3")]
pub payload: Option<InviteType>,
}

9
core/conversations/src/lib.rs
View File

@ -15,7 +15,16 @@ mod utils;
pub use causal_history::{Frontier, MissingMessage};
pub use chat_sqlite::ChatStorage;
pub use chat_sqlite::StorageConfig;
pub use conversation::GroupV2Clock;
pub use core::{ConversationId, Core, Introduction};
/// Timing/policy for GroupV2 conversations (de-mls's per-conversation config).
/// Defaults to the de-mls library defaults; inject via
/// [`Core::set_group_v2_config`]. The creator's phase durations (commit
/// inactivity, freeze, recovery, voting inactivity, proposal expiration,
/// consensus timeout) travel to joiners with the welcome and overwrite
/// theirs; vote delays and the policy fields stay local to each member.
pub use de_mls::ConversationConfig as GroupV2Config;
pub use de_mls::MockClock;
pub use errors::ChatError;
pub use outcomes::{
Content, ConversationClass, ConvoOutcome, InboxOutcome, NewConversation, PayloadOutcome,

9
core/conversations/src/service_context.rs
View File

@ -5,6 +5,7 @@ use storage::ChatStore;
use crate::IdentityProvider;
use crate::causal_history::CausalHistoryStore;
use crate::conversation::GroupV2Clock;
use crate::inbox_v2::{MlsEphemeralPqProvider, MlsIdentityProvider};
use crate::service_traits::WakeupService;
use crate::{DeliveryService, RegistrationService};
@ -44,6 +45,12 @@ pub(crate) struct ServiceContext<S: ExternalServices> {
pub(crate) causal: CausalHistoryStore,
pub(crate) identity: Identity,
pub(crate) wakeup_service: S::WS,
/// Time source for GroupV2 (de-mls) conversations.
pub(crate) demls_clock: GroupV2Clock,
/// Timing/policy for GroupV2 (de-mls) conversations, applied at
/// create/join. The creator's phase durations reach joiners inside the
/// welcome's `ConversationSync`.
pub(crate) demls_config: de_mls::ConversationConfig,
}
#[cfg(test)]
@ -110,6 +117,8 @@ mod test_support {
causal: CausalHistoryStore::new(),
identity: Identity::new(name),
wakeup_service: NoopWakeups {},
demls_clock: GroupV2Clock::default(),
demls_config: de_mls::ConversationConfig::default(),
})
}
}

6
core/conversations/src/types.rs
View File

@ -66,3 +66,9 @@ impl AddressedEncryptedPayload {
)
}
}
#[derive(Debug)]
pub struct ConvoMetadata {
pub name: String,
pub desc: String,
}

2
core/crypto/Cargo.toml
View File

@ -12,5 +12,5 @@ rand_core = { version = "0.6", features = ["getrandom"] }
sha2 = "0.10"
thiserror = "2"
x25519-dalek = { version = "2.0.1", features = ["static_secrets"] }
xeddsa = "1.0.2"
xeddsa = "=1.0.2"
zeroize = { version = "1.8.2", features = ["derive"] }

19
core/integration_tests_core/src/test_client.rs
View File

@ -1,5 +1,6 @@
use crate::test_ident::TestIdent;
use libchat::{ConversationId, Core, IdentityProvider, PayloadOutcome};
use libchat::{GroupV2Clock, GroupV2Config};
use shared_traits::IdentId;
use std::collections::HashMap;
use std::fmt::Debug;
@ -148,9 +149,11 @@ impl<const N: usize> TestHarness<N> {
let ident = TestIdent::new(Self::names(i));
addresses.insert(i, ident.id().clone());
let core_client =
let mut core_client =
ClientType::new_with_name(ident, ds.clone(), rs.clone(), wp, MemStore::new())
.unwrap();
core_client.set_group_v2_clock(GroupV2Clock::Mock(ws.clock()));
core_client.set_group_v2_config(fast_group_v2_config());
let client = TestClient::init(core_client);
@ -291,6 +294,20 @@ impl TestHarness<4> {
}
}
/// Millisecond GroupV2 timers for virtual-time tests — the production
/// defaults converge too slowly for the harness's step sizes.
fn fast_group_v2_config() -> GroupV2Config {
GroupV2Config {
commit_inactivity_duration: Duration::from_millis(50),
freeze_duration: Duration::from_millis(20),
voting_delay: Duration::from_millis(30),
election_voting_delay: Duration::from_millis(30),
consensus_timeout: Duration::from_millis(150),
proposal_expiration: Duration::from_millis(2000),
..GroupV2Config::default()
}
}
#[cfg(test)]
mod tests {
use super::*;

14
core/integration_tests_core/src/wakeup.rs
View File

@ -1,4 +1,4 @@
use libchat::{ConversationId, WakeupService};
use libchat::{ConversationId, MockClock, WakeupService};
use std::cell::RefCell;
use std::cmp::Reverse;
use std::collections::BinaryHeap;
@ -87,6 +87,7 @@ impl InnerWakeupService {
pub struct TestWakeupService {
inner: Rc<RefCell<InnerWakeupService>>,
clock: MockClock,
}
impl Debug for TestWakeupService {
@ -103,9 +104,14 @@ impl TestWakeupService {
pub fn new() -> Self {
Self {
inner: Rc::new(RefCell::new(InnerWakeupService::new())),
clock: MockClock::new(),
}
}
pub fn clock(&self) -> MockClock {
self.clock.clone()
}
pub fn new_provider(&self, id: usize) -> TestWakeupProvider {
TestWakeupProvider {
service: self.inner.clone(),
@ -117,11 +123,7 @@ impl TestWakeupService {
pub fn advance_time(&mut self, duration: Duration) -> Vec<WakeupRecord> {
let mut srv = self.inner.borrow_mut();
trace!(?duration, "Advanced");
// de-mls deadlines are real wall-clock; sleep so the millisecond-scale
// commit/consensus timers actually elapse between poll cycles
// Note: This is error prone as WakeupService tracks its own `now` variable. Does not account for processing time.
std::thread::sleep(duration);
self.clock.advance(duration);
srv.now = srv.now.checked_add(duration).unwrap();
srv.get_expired()
}

142
core/integration_tests_core/tests/test_group_v2.rs
View File

@ -18,7 +18,7 @@ fn groupv2_2way_roundtrip() {
let particpants = &[&harness.raya().addr()];
let convo_id = harness
.saro()
.create_group_convo_v2(particpants)
.create_group_convo_v2(particpants, "", "")
.expect("saro create group");
// Carry the invite through (commit, WelcomeReady, routing to Raya's inbox,
@ -56,7 +56,7 @@ fn core_client() {
let particpants = &[&harness.raya().addr()];
let convo_id = harness
.saro()
.create_group_convo_v2(particpants)
.create_group_convo_v2(particpants, "", "")
.expect("Saro create");
// Carry the invite through (commit, WelcomeReady, routing to Raya's inbox,
@ -114,7 +114,7 @@ fn core_client_batch_add() {
let particpants = &[&harness.raya().addr(), &harness.pax().addr()];
harness
.saro()
.create_group_convo_v2(particpants)
.create_group_convo_v2(particpants, "", "")
.expect("Saro create");
// Carry the invite through (commit, WelcomeReady, routing to Raya's inbox,
@ -144,7 +144,7 @@ fn core_client_four_members_two_epochs() {
let particpants = &[&harness.raya().addr(), &harness.pax().addr()];
let convo_id = harness
.saro()
.create_group_convo_v2(particpants)
.create_group_convo_v2(particpants, "", "")
.expect("Saro create");
// Carry the invite through (commit, WelcomeReady, routing to Raya's inbox,
@ -176,3 +176,137 @@ fn core_client_four_members_two_epochs() {
&& h.mira().check(&convo_id, MSG)
});
}
#[test]
fn group_name_propagation() {
let _ = tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
.with_test_writer()
.try_init();
let name = "Jankiest Friends";
let desc = "A cool group chat, with cool people";
let mut harness = TestHarness::<4>::new(|_, _| {});
let members = &[&harness.raya().addr()];
let convo_id = harness
.saro()
.create_group_convo_v2(members, name, desc)
.expect("Saro create");
// Carry the invite through (commit, WelcomeReady, routing to Raya's inbox,
// accept_welcome); settle until Raya has joined.
harness.process_until_label("Raya join", |h| h.raya().convo_count() == 1);
// Verfiy that Saro's Metadata is correct
assert_eq!(
harness.saro().convo_metadata(&convo_id).expect("meta").name,
name
);
assert_eq!(
harness.saro().convo_metadata(&convo_id).expect("meta").desc,
desc
);
// Verify that Raya has the same MetaData
assert_eq!(
harness.saro().convo_metadata(&convo_id).expect("meta").name,
harness.raya().convo_metadata(&convo_id).expect("meta").name
);
assert_eq!(
harness.saro().convo_metadata(&convo_id).expect("meta").desc,
harness.raya().convo_metadata(&convo_id).expect("meta").desc
);
// Epoch 2: Raya adds the 3rd member; settle until Pax has joined
let members = &[&harness.pax().addr()];
harness
.raya()
.group_add_member(&convo_id, members)
.expect("Add Pax");
harness.process_until_label("Pax join", |h| h.pax().convo_count() == 1);
// Verify that Pax has the same MetaData
assert_eq!(
harness.saro().convo_metadata(&convo_id).expect("meta").name,
harness.pax().convo_metadata(&convo_id).expect("meta").name
);
assert_eq!(
harness.saro().convo_metadata(&convo_id).expect("meta").desc,
harness.pax().convo_metadata(&convo_id).expect("meta").desc
);
}
#[test]
fn member_joins_two_groups() {
// The same installation is invited to two separate groups. Its single
// registered key package is consumed by the first join, so the second
// group's welcome must still admit it — otherwise it is rejected with
// "welcome not addressed to this member" and never joins. Regression for
// key-package reuse across groups.
let _ = tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
.with_test_writer()
.try_init();
let mut harness = TestHarness::<2>::new(|_, _| {});
let raya_addr = harness.raya().addr();
// Group 1: Saro invites Raya.
harness
.saro()
.create_group_convo_v2(&[&raya_addr], "", "")
.expect("saro create group 1");
harness.process_until_label("raya joins group 1", |h| h.raya().convo_count() == 1);
// Group 2: Saro invites Raya again, into a fresh group.
harness
.saro()
.create_group_convo_v2(&[&raya_addr], "", "")
.expect("saro create group 2");
harness.process_until_label("raya joins group 2", |h| h.raya().convo_count() == 2);
assert_eq!(
harness.raya().convo_count(),
2,
"raya did not join the second group"
);
}
#[test]
fn direct_v1_then_group_v2_reuses_key_package() {
// The reported flake: a DirectV1 (pairwise) conversation is opened with a
// peer first, then the same peer is invited to a GroupV2. Both fetch the
// peer's single registered key package; the DirectV1 join consumes it, so
// without a last-resort key package the group welcome finds no key package
// and is rejected ("welcome not addressed to this member").
let _ = tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
.with_test_writer()
.try_init();
let mut harness = TestHarness::<2>::new(|_, _| {});
let raya_addr = harness.raya().addr();
// 1. DirectV1 with Raya.
harness
.saro()
.create_direct_convo_v1(&[&raya_addr])
.expect("saro create direct");
harness.process_until_label("raya joins direct", |h| h.raya().convo_count() == 1);
// 2. GroupV2 inviting the same Raya.
harness
.saro()
.create_group_convo_v2(&[&raya_addr], "", "")
.expect("saro create group");
harness.process_until_label("raya joins group", |h| h.raya().convo_count() == 2);
assert_eq!(
harness.raya().convo_count(),
2,
"raya did not join the group after a direct chat"
);
}

7
crates/client/src/config.rs
View File

@ -1,7 +0,0 @@
//! Baked-in configuration for the Logos service stack.
/// The endpoint for the account and keypackage registration service.
pub const REGISTRY_ENDPOINT: &str = "https://devnet.chat-kc.logos.co";
/// The logos-delivery network preset the Logos client joins by default.
pub const NETWORK_PRESET: &str = "logos.dev";

89
crates/client/src/logos.rs
View File

@ -1,89 +0,0 @@
//! The opinionated Logos client.
//!
//! `LogosChatClient` commits to the Logos service stack so independently built
//! clients share the same production services instead of each re-deriving them:
//! a delegate identity, the HTTP keypackage + account registry, encrypted
//! on-disk storage, and — unlike the generic [`ChatClientBuilder`] — the
//! logos-delivery transport itself. logos-delivery *is* the transport for the
//! Logos client, so the caller no longer supplies one; only per-client secrets
//! (the database path and key) and the network config are passed in.
//!
//! The logos-delivery transport carries a native dependency, so this whole
//! module is gated behind the `embedded-p2p-delivery` cargo feature. The registry
//! endpoint lives in [`crate::config`] instead, so it stays available to other
//! transports when the feature is off.
use components::{EmbeddedP2pDeliveryService, HttpRegistry, P2pConfig};
use crossbeam_channel::Receiver;
use libchat::{ChatStorage, StorageConfig};
use logos_account::TestLogosAccount;
use crate::ChatClientBuilder;
use crate::client::{ChatClient, Transport};
use crate::config::{NETWORK_PRESET, REGISTRY_ENDPOINT};
use crate::delegate::DelegateSigner;
use crate::errors::ClientError;
use crate::event::Event;
// logos-delivery already implements `DeliveryService`; teaching it the inbound
// half here (in the crate that owns `Transport`) makes it a full transport, so
// callers need no wrapper newtype.
impl Transport for EmbeddedP2pDeliveryService {
fn inbound(&mut self) -> Receiver<Vec<u8>> {
self.inbound_queue()
}
}
/// A [`ChatClient`] wired to the Logos service stack: a [`DelegateSigner`]
/// identity acting for a fresh dev account, the HTTP keypackage + account
/// registry ([`HttpRegistry`], which is both the keypackage store and the
/// account → device directory), encrypted [`ChatStorage`], and the
/// logos-delivery transport.
pub type LogosChatClient = ChatClient<EmbeddedP2pDeliveryService, HttpRegistry, ChatStorage>;
impl LogosChatClient {
/// Open a client on the Logos stack, starting a logos-delivery node on
/// `tcp_port` as its transport and persisting to the encrypted database at
/// `db_path` unlocked with `db_key`. When `preset`/`registry_url` are `Some`,
/// they override the baked-in network preset/registry endpoint (e.g. a local
/// deployment); otherwise the preconfigured values are used.
///
/// `db_path` is a per-client location, `db_key` is a secret, and `tcp_port`
/// is a per-client local resource, so all three are caller-supplied — never
/// baked into the library.
pub fn open(
db_path: impl Into<String>,
db_key: impl Into<String>,
tcp_port: u16,
preset: Option<&str>,
registry_url: Option<&str>,
) -> Result<(Self, Receiver<Event>), ClientError> {
let transport = EmbeddedP2pDeliveryService::start(P2pConfig {
preset: preset.unwrap_or(NETWORK_PRESET).to_string(),
tcp_port,
..Default::default()
})
.map_err(|e| ClientError::Transport(e.to_string()))?;
let endpoint = registry_url.unwrap_or(REGISTRY_ENDPOINT);
// A fresh account endorsing a fresh delegate each open: the account
// key is dropped after publishing the bundle, so devices cannot be
// added later. A caller-supplied, custody-holding account replaces
// this once the platform provides one.
let account = TestLogosAccount::new();
let delegate = DelegateSigner::random();
let mut registry = HttpRegistry::new(endpoint);
account
.add_delegate_signer(&mut registry, delegate.public_key())
.map_err(|e| ClientError::BundlePublish(e.to_string()))?;
ChatClientBuilder::new(account.address())
.ident(delegate)
.transport(transport)
.registration(registry)
.storage_config(StorageConfig::Encrypted {
path: db_path.into(),
key: db_key.into(),
})
.build()
}
}

5
crates/client/Cargo.toml → crates/generic-chat/Cargo.toml
View File

@ -1,14 +1,11 @@
[package]
name = "logos-chat"
name = "logos-generic-chat"
version = "0.1.0"
edition = "2024"
[lib]
crate-type = ["rlib"]
[features]
embedded-p2p-delivery = ["components/embedded_p2p_delivery"]
[dependencies]
# Workspace dependencies (sorted)
chat-sqlite = { workspace = true }

4
crates/client/examples/message-exchange/README.md → crates/generic-chat/examples/message-exchange/README.md
View File

@ -1,9 +1,9 @@
# message-exchange
An example Rust application built on top of [`crates/client`](../../).
An example Rust application built on top of [`crates/generic-chat`](../../).
It demonstrates that creating a working chat client in pure Rust is trivial: depend on
`crates/client`, pick a `DeliveryService` implementation (here the in-memory
`crates/generic-chat`, pick a `DeliveryService` implementation (here the in-memory
`InProcessDelivery` shipped with the crate), and wire up `ChatClient`. No boilerplate, no FFI.
## Running

2
crates/client/examples/message-exchange/main.rs → crates/generic-chat/examples/message-exchange/main.rs
View File

@ -1,6 +1,6 @@
use components::EphemeralRegistry;
use logos_account::TestLogosAccount;
use logos_chat::{ChatClientBuilder, Event, InProcessDelivery, MessageBus};
use logos_generic_chat::{ChatClientBuilder, Event, InProcessDelivery, MessageBus};
use std::time::Duration;
fn main() {

26
crates/client/src/builder.rs → crates/generic-chat/src/builder.rs
View File

@ -1,6 +1,6 @@
use components::EphemeralRegistry;
use crossbeam_channel::Receiver;
use libchat::{ChatError, ChatStorage, RegistrationService, StorageConfig};
use libchat::{ChatError, ChatStorage, GroupV2Config, RegistrationService, StorageConfig};
use logos_account::AccountDirectory;
use storage::ChatStore;
@ -20,6 +20,7 @@ pub struct ChatClientBuilder<I = Unset, T = Unset, R = Unset, S = Unset> {
transport: T,
registration: R,
storage: S,
group_v2: Option<GroupV2Config>,
}
impl ChatClientBuilder {
@ -35,6 +36,7 @@ impl ChatClientBuilder {
transport: Unset,
registration: Unset,
storage: Unset,
group_v2: None,
}
}
}
@ -47,6 +49,7 @@ impl<I, T, R, S> ChatClientBuilder<I, T, R, S> {
transport: self.transport,
registration: self.registration,
storage: self.storage,
group_v2: self.group_v2,
}
}
@ -57,6 +60,7 @@ impl<I, T, R, S> ChatClientBuilder<I, T, R, S> {
transport,
registration: self.registration,
storage: self.storage,
group_v2: self.group_v2,
}
}
@ -67,6 +71,7 @@ impl<I, T, R, S> ChatClientBuilder<I, T, R, S> {
transport: self.transport,
registration,
storage: self.storage,
group_v2: self.group_v2,
}
}
@ -77,6 +82,7 @@ impl<I, T, R, S> ChatClientBuilder<I, T, R, S> {
transport: self.transport,
registration: self.registration,
storage,
group_v2: self.group_v2,
}
}
@ -91,8 +97,18 @@ impl<I, T, R, S> ChatClientBuilder<I, T, R, S> {
transport: self.transport,
registration: self.registration,
storage,
group_v2: self.group_v2,
}
}
/// Timing/policy for GroupV2 conversations this client creates or joins.
/// Defaults to the de-mls library defaults; the creator's phase durations
/// travel to joiners with the welcome and overwrite theirs (vote delays
/// and policy fields stay local).
pub fn group_v2_config(mut self, config: GroupV2Config) -> Self {
self.group_v2 = Some(config);
self
}
}
type Built<T, R, S> = Result<(ChatClient<T, R, S>, Receiver<Event>), ClientError>;
@ -111,6 +127,7 @@ where
self.transport,
self.registration,
self.storage,
self.group_v2,
)
}
}
@ -124,6 +141,7 @@ impl<T: Transport + Send + 'static> ChatClientBuilder<Unset, T, Unset, Unset> {
self.transport,
EphemeralRegistry::new(),
ChatStorage::in_memory(),
self.group_v2,
)
}
}
@ -140,6 +158,7 @@ where
self.transport,
EphemeralRegistry::new(),
ChatStorage::in_memory(),
self.group_v2,
)
}
}
@ -157,6 +176,7 @@ where
self.transport,
self.registration,
ChatStorage::in_memory(),
self.group_v2,
)
}
}
@ -174,6 +194,7 @@ where
self.transport,
EphemeralRegistry::new(),
self.storage,
self.group_v2,
)
}
}
@ -191,6 +212,7 @@ where
self.transport,
self.registration,
ChatStorage::in_memory(),
self.group_v2,
)
}
}
@ -209,6 +231,7 @@ where
self.transport,
self.registration,
self.storage,
self.group_v2,
)
}
}
@ -226,6 +249,7 @@ where
self.transport,
EphemeralRegistry::new(),
self.storage,
self.group_v2,
)
}
}

348
crates/client/src/client.rs → crates/generic-chat/src/client.rs
View File

@ -1,3 +1,4 @@
use std::collections::HashSet;
use std::sync::Arc;
use std::thread::{self, JoinHandle};
@ -5,8 +6,8 @@ use components::{ThreadedWakeupService, WakeupEvent};
use crossbeam_channel::{Receiver, Sender, select};
use crypto::Ed25519VerifyingKey;
use libchat::{
ConversationId, ConvoOutcome, Core, DeliveryService, IdentId, IdentIdRef, InboxOutcome,
Introduction, PayloadOutcome, RegistrationService,
ConversationId, ConvoOutcome, Core, DeliveryService, GroupV2Config, IdentId, IdentIdRef,
InboxOutcome, Introduction, PayloadOutcome, RegistrationService,
};
use logos_account::{AccountDirectory, resolve_device_ids};
use parking_lot::Mutex;
@ -20,6 +21,19 @@ type ClientCore<T, R, S> = Core<(DelegateIdentity, T, R, ThreadedWakeupService,
type AccountAddressRef<'a> = &'a str;
type LocalSignerId = IdentId;
/// A member of a group conversation's roster.
///
/// Shares [`MessageSender`]'s field semantics: `account` is set only when the
/// member's credential claimed an account *and* the directory confirmed this
/// device belongs to it. Unlike a message sender, an unconfirmable claim does
/// not hide the member: it is cryptographically in the group, so it is listed
/// by `local_identity` (its device) with `account: None`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GroupMember {
pub account: Option<IdentId>,
pub local_identity: IdentId,
}
/// The transport as the client sees it: a [`DeliveryService`] for outbound
/// publishing plus the inbound payload stream the worker drains. One object owns
/// both directions of the boundary.
@ -73,6 +87,7 @@ where
mut transport: T,
reg: R,
storage: S,
group_v2: Option<GroupV2Config>,
) -> Result<(Self, Receiver<Event>), ClientError> {
let inbound = transport.inbound();
@ -80,7 +95,10 @@ where
let wakeup_service = ThreadedWakeupService::new(wakeup_tx);
let directory = reg.clone();
let ident = DelegateIdentity::new(ident, &account);
let core = Core::new_with_name(ident, transport, reg, wakeup_service, storage)?;
let mut core = Core::new_with_name(ident, transport, reg, wakeup_service, storage)?;
if let Some(config) = group_v2 {
core.set_group_v2_config(config);
}
Ok(Self::spawn(core, directory, account, inbound, wakeup_rx))
}
@ -151,6 +169,55 @@ where
.map_err(Into::into)
}
/// Create a GroupV2 conversation with the given accounts' devices. Each
/// account resolves to the signer ids its directory bundle endorses; the
/// group invite goes to every one of them. An empty slice creates a group
/// with only this client, to grow via [`Self::add_group_members`].
pub fn create_group_conversation(
&mut self,
accounts: &[AccountAddressRef],
) -> Result<ConversationId, ClientError> {
let signers = self.signers_from_accounts(accounts)?;
let signer_refs: Vec<IdentIdRef> = signers.iter().collect();
self.core
.lock()
.create_group_convo(&signer_refs)
.map_err(Into::into)
}
/// Add accounts' devices to an existing group conversation. The add is
/// staged as an MLS proposal and merged by the group's next commit (driven
/// asynchronously by the wakeup loop); each joiner's welcome is sent when
/// that commit lands, not when this call returns.
pub fn add_group_members(
&mut self,
convo_id: &str,
accounts: &[AccountAddressRef],
) -> Result<(), ClientError> {
let signers = self.signers_from_accounts(accounts)?;
let signer_refs: Vec<IdentIdRef> = signers.iter().collect();
self.core
.lock()
.group_add_member(convo_id, &signer_refs)
.map_err(Into::into)
}
/// The group's roster, one [`GroupMember`] per account (self included). An
/// account's several devices collapse to a single entry surfacing that
/// account; a member whose account claim the directory can't confirm stays
/// on the roster individually, keyed by its device. Costs one directory
/// lookup per member that claims an account, the same per-member cost a
/// received message's sender check pays.
pub fn group_members(&mut self, convo_id: &str) -> Result<Vec<GroupMember>, ClientError> {
let credentials = self.core.lock().group_members(convo_id)?;
let members = credentials
.iter()
.filter_map(|credential| roster_member(&self.directory, credential));
Ok(dedup_members(members))
}
/// Parse intro bundle bytes and initiate a private conversation. Outbound
/// envelopes are published by the core. Returns this side's conversation ID.
///
@ -193,6 +260,19 @@ where
.map_err(|e| ClientError::AccountResolution(e.to_string()))?;
Ok(device_ids.into_iter().map(IdentId::new).collect())
}
/// Resolve each account to its signer ids and flatten them, failing on the
/// first unresolvable account.
fn signers_from_accounts(
&self,
accounts: &[AccountAddressRef],
) -> Result<Vec<LocalSignerId>, ClientError> {
let mut signers = Vec::new();
for account in accounts {
signers.extend(self.signers_from_account(account)?);
}
Ok(signers)
}
}
impl<T, R, S> Drop for ChatClient<T, R, S>
@ -298,6 +378,62 @@ enum SenderError {
Unverified,
}
/// The resolution of a credential's account claim against the directory.
enum AccountClaim {
/// The credential claimed no account.
None,
/// Confirmed: the directory lists this device under the claimed account.
Verified(IdentId),
/// An account was claimed but could not be confirmed (see [`SenderError`]).
Unverified(SenderError),
}
/// Parse a wire credential into the device it names and the resolution of any
/// account claim, checked against the account → device directory. `Err` only
/// when no device can be attributed at all (missing or unparseable credential).
///
/// The account-claim policy is left to the caller: a message drops on an
/// unconfirmable claim, a roster entry keeps the device and forgoes the account.
fn parse_credential(
directory: &impl AccountDirectory,
encoded: &[u8],
) -> Result<(IdentId, AccountClaim), SenderError> {
// No credential at all: there is no device to attribute.
if encoded.is_empty() {
return Err(SenderError::Missing);
}
let Ok(data) = hex::decode(encoded) else {
tracing::warn!("credential is not valid hex");
return Err(SenderError::NotHex);
};
let Ok(cred) = DelegateCredential::try_from(data) else {
tracing::warn!("malformed credential");
return Err(SenderError::Malformed);
};
let device = IdentId::new(hex::encode(cred.delegate_id().as_ref()));
// An unassociated delegate asserts no account → device mapping.
let Some(account_addr) = cred.account_addr() else {
return Ok((device, AccountClaim::None));
};
let Some(account_key) = account_key_from_hex(account_addr) else {
tracing::warn!(account_addr, "account address is not a verifying key");
return Ok((
device,
AccountClaim::Unverified(SenderError::AccountNotAKey),
));
};
let claim = match directory.fetch(&account_key) {
Ok(Some(set)) if set.devices.iter().any(|d| d.as_str() == device.as_str()) => {
AccountClaim::Verified(IdentId::new(account_addr.to_string()))
}
_ => {
tracing::warn!(account_addr, device = %device.as_str(), "account → device mapping is wrong or unconfirmable");
AccountClaim::Unverified(SenderError::Unverified)
}
};
Ok((device, claim))
}
/// Decode and verify a message's sender from its credential, checked against the
/// account → device directory (our account store).
///
@ -309,48 +445,62 @@ fn decode_sender(
directory: &impl AccountDirectory,
encoded: &[u8],
) -> Result<MessageSender, SenderError> {
// No credential at all: there is no sender to attribute, so drop it.
if encoded.is_empty() {
return Err(SenderError::Missing);
}
let Ok(data) = hex::decode(encoded) else {
tracing::warn!("sender credential is not valid hex; dropping message");
return Err(SenderError::NotHex);
};
let cred = match DelegateCredential::try_from(data) {
Ok(cred) => cred,
Err(_) => {
tracing::warn!("malformed sender credential; dropping message");
return Err(SenderError::Malformed);
}
};
let device = hex::encode(cred.delegate_id().as_ref());
// An unassociated delegate asserts no account → device mapping.
let Some(account_addr) = cred.account_addr() else {
return Ok(MessageSender {
let (device, claim) = parse_credential(directory, encoded)?;
match claim {
AccountClaim::None => Ok(MessageSender {
account: None,
local_identity: IdentId::new(device),
});
};
let Some(account_key) = account_key_from_hex(account_addr) else {
tracing::warn!(
account_addr,
"sender account address is not a verifying key; dropping message"
);
return Err(SenderError::AccountNotAKey);
};
match directory.fetch(&account_key) {
Ok(Some(set)) if set.devices.iter().any(|d| d == &device) => Ok(MessageSender {
account: Some(IdentId::new(account_addr.to_string())),
local_identity: IdentId::new(device),
local_identity: device,
}),
_ => {
tracing::warn!(account_addr, %device, "account → device mapping is wrong or unconfirmable; dropping message");
Err(SenderError::Unverified)
}
AccountClaim::Verified(account) => Ok(MessageSender {
account: Some(account),
local_identity: device,
}),
// An unconfirmable account claim drops the message: every delivered
// message must carry a verified sender.
AccountClaim::Unverified(err) => Err(err),
}
}
/// Map a group member's credential (as reported by MLS, in the same hex-encoded
/// form a message carries as its sender) to a roster entry, tolerating an
/// unconfirmable account claim by listing the device without an account. `None`
/// only when the credential cannot be parsed, which does not happen for a real
/// MLS leaf.
fn roster_member(directory: &impl AccountDirectory, encoded: &[u8]) -> Option<GroupMember> {
let (device, claim) = parse_credential(directory, encoded).ok()?;
let account = match claim {
AccountClaim::Verified(account) => Some(account),
AccountClaim::None | AccountClaim::Unverified(_) => None,
};
Some(GroupMember {
account,
local_identity: device,
})
}
/// The key that decides whether two roster entries are the same member: a
/// verified account, so an account's several devices count once; or, for a
/// member with no confirmed account, its device — unique per MLS leaf, so it
/// never merges with another.
fn member_key(member: &GroupMember) -> &str {
member
.account
.as_ref()
.unwrap_or(&member.local_identity)
.as_str()
}
/// Collapse a roster to one entry per account (keeping the first-seen device as
/// the account's representative) while leaving account-less members individual,
/// order preserved.
fn dedup_members(members: impl IntoIterator<Item = GroupMember>) -> Vec<GroupMember> {
let mut seen = HashSet::new();
members
.into_iter()
.filter(|member| seen.insert(member_key(member).to_owned()))
.collect()
}
fn convo_events(outcome: ConvoOutcome, directory: &impl AccountDirectory) -> Vec<Event> {
let ConvoOutcome { convo_id, content } = outcome;
content
@ -397,7 +547,10 @@ mod sender_check_tests {
use libchat::IdentId;
use logos_account::{DeviceSet, SignedDeviceBundle};
use super::{MessageSender, SenderError, decode_sender};
use super::{
GroupMember, MessageSender, SenderError, decode_sender, dedup_members, member_key,
roster_member,
};
use crate::delegate::DelegateCredential;
/// In-test account → device directory. Holds device id sets keyed by the hex
@ -567,4 +720,117 @@ mod sender_check_tests {
Err(SenderError::AccountNotAKey)
);
}
/// A verified account claim surfaces the member's account and device — the
/// same happy path as a message sender.
#[test]
fn roster_verified_member_surfaces_account() {
let account = key();
let device = key();
let dir = FakeDir::with_devices(&account, &[&device]);
let cred = DelegateCredential::associated(&device, &hex::encode(account.as_ref()));
assert_eq!(
roster_member(&dir, &encoded(cred)),
Some(GroupMember {
account: Some(local_id(&account)),
local_identity: local_id(&device),
})
);
}
/// Unlike a message sender, a spoofed account claim does not hide the
/// member: the device is cryptographically in the group, so it is listed
/// with no account rather than dropped.
#[test]
fn roster_contradicted_claim_lists_device_without_account() {
let account = key();
let endorsed = key();
let spoofer = key();
let dir = FakeDir::with_devices(&account, &[&endorsed]);
let cred = DelegateCredential::associated(&spoofer, &hex::encode(account.as_ref()));
assert_eq!(
roster_member(&dir, &encoded(cred)),
Some(GroupMember {
account: None,
local_identity: local_id(&spoofer),
})
);
}
/// A member whose credential claims no account is listed by device only.
#[test]
fn roster_unassociated_member_lists_device_without_account() {
let dir = FakeDir::default();
let device = key();
let cred = DelegateCredential::unassociated(&device);
assert_eq!(
roster_member(&dir, &encoded(cred)),
Some(GroupMember {
account: None,
local_identity: local_id(&device),
})
);
}
/// A directory outage leaves the account unconfirmed, but the member stays
/// on the roster by device (a message would drop here).
#[test]
fn roster_directory_outage_lists_device_without_account() {
let account = key();
let device = key();
let dir = FakeDir {
fail: true,
..Default::default()
};
let cred = DelegateCredential::associated(&device, &hex::encode(account.as_ref()));
assert_eq!(
roster_member(&dir, &encoded(cred)),
Some(GroupMember {
account: None,
local_identity: local_id(&device),
})
);
}
/// A non-key account address can't be confirmed, so the member is listed by
/// device without an account.
#[test]
fn roster_non_key_account_lists_device_without_account() {
let dir = FakeDir::default();
let device = key();
let cred = DelegateCredential::associated(&device, "user@example.com");
assert_eq!(
roster_member(&dir, &encoded(cred)),
Some(GroupMember {
account: None,
local_identity: local_id(&device),
})
);
}
/// The roster collapses an account's several devices into one entry (keeping
/// the first device seen) while leaving account-less members individual,
/// order preserved.
#[test]
fn dedup_collapses_account_devices_and_keeps_unknowns() {
let with_account = |account: &str, device: &str| GroupMember {
account: Some(IdentId::new(account.to_string())),
local_identity: IdentId::new(device.to_string()),
};
let device_only = |device: &str| GroupMember {
account: None,
local_identity: IdentId::new(device.to_string()),
};
let roster = dedup_members(vec![
with_account("alice", "alice-dev-1"),
with_account("alice", "alice-dev-2"),
device_only("orphan-x"),
with_account("bob", "bob-dev-1"),
device_only("orphan-y"),
]);
let keys: Vec<&str> = roster.iter().map(member_key).collect();
assert_eq!(keys, ["alice", "orphan-x", "bob", "orphan-y"]);
// Alice's collapsed entry keeps her first-seen device.
assert_eq!(roster[0].local_identity.as_str(), "alice-dev-1");
}
}

0
crates/client/src/delegate.rs → crates/generic-chat/src/delegate.rs
View File

0
crates/client/src/delivery_in_process.rs → crates/generic-chat/src/delivery_in_process.rs
View File

0
crates/client/src/errors.rs → crates/generic-chat/src/errors.rs
View File

0
crates/client/src/event.rs → crates/generic-chat/src/event.rs
View File

10
crates/client/src/lib.rs → crates/generic-chat/src/lib.rs
View File

@ -1,27 +1,21 @@
mod builder;
mod client;
mod config;
mod delegate;
mod delivery_in_process;
mod errors;
mod event;
#[cfg(feature = "embedded-p2p-delivery")]
mod logos;
pub use builder::{ChatClientBuilder, Unset};
pub use client::{ChatClient, Transport};
pub use config::{NETWORK_PRESET, REGISTRY_ENDPOINT};
pub use client::{ChatClient, GroupMember, Transport};
pub use delegate::DelegateSigner;
pub use delivery_in_process::{InProcessDelivery, MessageBus};
pub use errors::ClientError;
pub use event::{Event, MessageSender};
#[cfg(feature = "embedded-p2p-delivery")]
pub use logos::LogosChatClient;
// Re-export types callers need to interact with ChatClient.
pub use libchat::{
AddressedEnvelope, ChatStore, ConversationClass, ConversationId, DeliveryService,
IdentityProvider, RegistrationService, StorageConfig,
GroupV2Config, IdentityProvider, RegistrationService, StorageConfig,
};
// The directory trait bounds ChatClient's registry parameter, so callers
// writing code generic over ChatClient need it too.

338
crates/generic-chat/tests/group_v2.rs Normal file
View File

@ -0,0 +1,338 @@
//! GroupV2 through the threaded client: three accounts on the in-process
//! transport, driven purely over the public `ChatClient` API and its event
//! channel. Group commits and welcomes are minted asynchronously by de-mls
//! (wakeup-driven), so assertions wait for events rather than expecting a
//! fixed sequence.
use std::time::Duration;
use components::EphemeralRegistry;
use crossbeam_channel::Receiver;
use libchat::ChatStorage;
use logos_account::TestLogosAccount;
use logos_generic_chat::{
ChatClient, ChatClientBuilder, ConversationClass, DelegateSigner, Event, GroupV2Config,
InProcessDelivery, MessageBus,
};
/// Millisecond GroupV2 timers so the de-mls commit/consensus dance completes
/// in test time; the library defaults wait 60s before committing an add.
fn fast_group_v2_config() -> GroupV2Config {
GroupV2Config {
commit_inactivity_duration: Duration::from_millis(50),
freeze_duration: Duration::from_millis(20),
voting_delay: Duration::from_millis(30),
election_voting_delay: Duration::from_millis(30),
consensus_timeout: Duration::from_millis(150),
proposal_expiration: Duration::from_millis(2000),
..GroupV2Config::default()
}
}
type TestClient = ChatClient<InProcessDelivery, EphemeralRegistry, ChatStorage>;
/// A client for a fresh account: mints the account and a delegate, publishes
/// the endorsing bundle, and builds the client on the shared bus/registry with
/// the fast GroupV2 timers. Returns the account address peers invite by.
fn create_test_client(
message_bus: MessageBus,
mut reg: EphemeralRegistry,
) -> (TestClient, Receiver<Event>, String) {
let account = TestLogosAccount::new();
let delegate = DelegateSigner::random();
account
.add_delegate_signer(&mut reg, delegate.public_key())
.unwrap();
let (client, events) = ChatClientBuilder::new(account.address())
.ident(delegate)
.transport(InProcessDelivery::new(message_bus))
.registration(reg)
.group_v2_config(fast_group_v2_config())
.build()
.expect("client create");
let addr = client.addr().to_string();
(client, events, addr)
}
/// Wait until an event matching `f` arrives, skipping unrelated events (group
/// protocol traffic can interleave observations); panic after `timeout`.
fn wait_for_event<F, T>(events: &Receiver<Event>, label: &str, timeout: Duration, mut f: F) -> T
where
F: FnMut(&Event) -> Option<T>,
{
let deadline = std::time::Instant::now() + timeout;
loop {
let remaining = deadline
.checked_duration_since(std::time::Instant::now())
.unwrap_or_else(|| panic!("timed out waiting for {label}"));
match events.recv_timeout(remaining) {
Ok(event) => {
if let Some(out) = f(&event) {
return out;
}
}
Err(_) => panic!("timed out waiting for {label}"),
}
}
}
/// Wait for the group conversation to start on a joiner and return its id.
fn wait_for_group_started(events: &Receiver<Event>, label: &str) -> String {
wait_for_event(events, label, Duration::from_secs(10), |e| match e {
Event::ConversationStarted { convo_id, class } => {
assert_eq!(*class, ConversationClass::Group);
Some(convo_id.to_string())
}
_ => None,
})
}
/// Poll a client's roster for `convo_id` until its verified accounts equal
/// `expected` (order-independent), or panic after a timeout. The roster settles
/// asynchronously as each member applies the add commit, so it is polled rather
/// than snapshotted.
fn wait_for_members(client: &mut TestClient, convo_id: &str, expected: &[&str]) {
use std::collections::BTreeSet;
let want: BTreeSet<&str> = expected.iter().copied().collect();
let deadline = std::time::Instant::now() + Duration::from_secs(10);
loop {
let roster = client.group_members(convo_id).expect("group_members");
let got: BTreeSet<&str> = roster
.iter()
.filter_map(|m| m.account.as_ref().map(|a| a.as_str()))
.collect();
if got == want {
return;
}
if std::time::Instant::now() >= deadline {
panic!("roster did not converge for {convo_id}: got {got:?}, want {want:?}");
}
std::thread::sleep(Duration::from_millis(20));
}
}
/// Wait for `content` to arrive and return the sender's verified account.
fn wait_for_message(events: &Receiver<Event>, content: &[u8]) -> Option<String> {
let label = format!("MessageReceived({})", String::from_utf8_lossy(content));
wait_for_event(events, &label, Duration::from_secs(10), |e| match e {
Event::MessageReceived {
content: got,
sender,
..
} if got == content => Some(sender.account.as_ref().map(|a| a.as_str().to_string())),
_ => None,
})
}
/// A three-account group: saro creates it with raya, raya (a non-creator)
/// adds pax, and a message from each member reaches both others with a
/// directory-verified sender account.
#[test]
fn group_v2_three_members() {
let bus = MessageBus::default();
let reg = EphemeralRegistry::new();
let (mut saro, saro_events, saro_addr) = create_test_client(bus.clone(), reg.clone());
let (mut raya, raya_events, raya_addr) = create_test_client(bus.clone(), reg.clone());
let (mut pax, pax_events, pax_addr) = create_test_client(bus.clone(), reg.clone());
let convo_id = saro
.create_group_conversation(&[&raya_addr])
.expect("saro create group");
// The invite lands once saro's steward commit finalizes (wakeup-driven);
// both sides then share the de-mls conversation id.
let raya_convo_id = wait_for_group_started(&raya_events, "raya ConversationStarted");
assert_eq!(raya_convo_id, convo_id);
// Both sides see the two-account roster once the add commits.
wait_for_members(&mut saro, &convo_id, &[&saro_addr, &raya_addr]);
wait_for_members(&mut raya, &raya_convo_id, &[&saro_addr, &raya_addr]);
saro.send_message(&convo_id, b"hello raya").unwrap();
assert_eq!(
wait_for_message(&raya_events, b"hello raya").as_deref(),
Some(saro_addr.as_str())
);
raya.send_message(&raya_convo_id, b"hi saro").unwrap();
assert_eq!(
wait_for_message(&saro_events, b"hi saro").as_deref(),
Some(raya_addr.as_str())
);
// A non-creator grows the group: raya proposes pax, the steward commits,
// and raya (who holds the pending invite) routes the welcome to pax.
raya.add_group_members(&raya_convo_id, &[&pax_addr])
.expect("raya add pax");
let pax_convo_id = wait_for_group_started(&pax_events, "pax ConversationStarted");
assert_eq!(pax_convo_id, convo_id);
// Everyone is at the post-add epoch: a message from the creator reaches
// both peers, and one from the newest member reaches both elders.
saro.send_message(&convo_id, b"all three?").unwrap();
assert_eq!(
wait_for_message(&raya_events, b"all three?").as_deref(),
Some(saro_addr.as_str())
);
assert_eq!(
wait_for_message(&pax_events, b"all three?").as_deref(),
Some(saro_addr.as_str())
);
pax.send_message(&pax_convo_id, b"pax is in").unwrap();
assert_eq!(
wait_for_message(&saro_events, b"pax is in").as_deref(),
Some(pax_addr.as_str())
);
assert_eq!(
wait_for_message(&raya_events, b"pax is in").as_deref(),
Some(pax_addr.as_str())
);
// All three rosters converge on the same three accounts.
let all = [saro_addr.as_str(), raya_addr.as_str(), pax_addr.as_str()];
wait_for_members(&mut saro, &convo_id, &all);
wait_for_members(&mut raya, &raya_convo_id, &all);
wait_for_members(&mut pax, &pax_convo_id, &all);
assert_eq!(saro.list_conversations().unwrap().len(), 1);
assert_eq!(raya.list_conversations().unwrap().len(), 1);
assert_eq!(pax.list_conversations().unwrap().len(), 1);
}
/// The same two peers are invited to several groups at once. Each installation
/// registers a single key package, so admitting it to more than one group only
/// works if that key package survives a join — a regression guard for the
/// multi-group "welcome not addressed to this member" flake.
#[test]
fn peers_invited_to_many_groups() {
const GROUPS: usize = 3;
let bus = MessageBus::default();
let reg = EphemeralRegistry::new();
let (mut saro, _saro_events, saro_addr) = create_test_client(bus.clone(), reg.clone());
let (_raya, raya_events, raya_addr) = create_test_client(bus.clone(), reg.clone());
let (_pax, pax_events, pax_addr) = create_test_client(bus.clone(), reg.clone());
// Saro opens several groups, each inviting both Raya and Pax; every group
// reuses Raya's and Pax's one key package.
let mut convo_ids = Vec::new();
for _ in 0..GROUPS {
convo_ids.push(
saro.create_group_conversation(&[&raya_addr, &pax_addr])
.expect("saro create group"),
);
}
// Both peers must join all of them.
for _ in 0..GROUPS {
wait_for_group_started(&raya_events, "raya joins a group");
wait_for_group_started(&pax_events, "pax joins a group");
}
// Every group is live: a distinct message in each reaches both peers with
// the creator's verified account.
for (i, convo_id) in convo_ids.iter().enumerate() {
let msg = format!("hello group {i}").into_bytes();
saro.send_message(convo_id, &msg).unwrap();
assert_eq!(
wait_for_message(&raya_events, &msg).as_deref(),
Some(saro_addr.as_str())
);
assert_eq!(
wait_for_message(&pax_events, &msg).as_deref(),
Some(saro_addr.as_str())
);
}
assert_eq!(saro.list_conversations().unwrap().len(), GROUPS);
}
/// The creator is in its own roster from the start, with no other members: the
/// roster always includes self.
#[test]
fn group_creator_is_in_own_roster() {
let bus = MessageBus::default();
let reg = EphemeralRegistry::new();
let (mut saro, _saro_events, saro_addr) = create_test_client(bus.clone(), reg.clone());
let convo_id = saro.create_group_conversation(&[]).expect("empty group");
let roster = saro.group_members(&convo_id).expect("group_members");
let accounts: Vec<Option<&str>> = roster
.iter()
.map(|m| m.account.as_ref().map(|a| a.as_str()))
.collect();
assert_eq!(accounts, vec![Some(saro_addr.as_str())]);
}
/// A batch add is validated before any member is proposed: a member whose
/// account is endorsed in the directory but whose device registered no key
/// package fails the whole call, the resolvable member in the same batch is
/// not invited, and the group keeps working.
#[test]
fn add_batch_with_missing_key_package_invites_no_one() {
let bus = MessageBus::default();
let mut reg = EphemeralRegistry::new();
let (mut saro, _saro_events, _saro_addr) = create_test_client(bus.clone(), reg.clone());
let (_raya, raya_events, raya_addr) = create_test_client(bus.clone(), reg.clone());
let (_pax, pax_events, pax_addr) = create_test_client(bus.clone(), reg.clone());
// Ghost: its account endorses a device in the directory, but that device
// never registered a key package (no client was built for it).
let ghost_account = TestLogosAccount::new();
let ghost_delegate = DelegateSigner::random();
ghost_account
.add_delegate_signer(&mut reg, ghost_delegate.public_key())
.unwrap();
let convo_id = saro
.create_group_conversation(&[&raya_addr])
.expect("saro create group");
wait_for_group_started(&raya_events, "raya ConversationStarted");
saro.add_group_members(&convo_id, &[&ghost_account.address(), &pax_addr])
.expect_err("ghost has no key package");
// Pax was in the failed batch and must not have been invited.
assert!(
pax_events.recv_timeout(Duration::from_secs(1)).is_err(),
"pax must not join from a failed batch"
);
// The failed add left the group functional.
saro.send_message(&convo_id, b"still alive").unwrap();
wait_for_message(&raya_events, b"still alive");
}
/// Group membership is resolved through the account directory, so inviting an
/// address whose account never published a bundle fails at resolution — on
/// create and on add alike.
#[test]
fn group_invite_of_unpublished_account_is_an_error() {
let bus = MessageBus::default();
let reg = EphemeralRegistry::new();
let (mut saro, _saro_events, _saro_addr) = create_test_client(bus.clone(), reg.clone());
let unpublished = TestLogosAccount::new();
let err = saro
.create_group_conversation(&[&unpublished.address()])
.expect_err("no bundle published for the account");
assert!(matches!(
err,
logos_generic_chat::ClientError::AccountResolution(_)
));
let convo_id = saro.create_group_conversation(&[]).expect("empty group");
let err = saro
.add_group_members(&convo_id, &[&unpublished.address()])
.expect_err("no bundle published for the account");
assert!(matches!(
err,
logos_generic_chat::ClientError::AccountResolution(_)
));
}

25
crates/client/tests/saro_and_raya.rs → crates/generic-chat/tests/saro_and_raya.rs
View File

@ -4,9 +4,9 @@ use components::EphemeralRegistry;
use crossbeam_channel::{Receiver, Sender};
use crypto::Ed25519VerifyingKey;
use logos_account::TestLogosAccount;
use logos_chat::{
AddressedEnvelope, ChatClient, ChatClientBuilder, DelegateSigner, DeliveryService, Event,
InProcessDelivery, MessageBus, Transport,
use logos_generic_chat::{
AddressedEnvelope, ChatClient, ChatClientBuilder, ConversationClass, DelegateSigner,
DeliveryService, Event, InProcessDelivery, MessageBus, Transport,
};
/// Publish a signed device bundle endorsing `device` as a device of `account`,
@ -30,7 +30,7 @@ fn create_test_client(
ChatClient<InProcessDelivery, EphemeralRegistry, libchat::ChatStorage>,
Receiver<Event>,
),
logos_chat::ClientError,
logos_generic_chat::ClientError,
> {
let account = TestLogosAccount::new();
let delegate = DelegateSigner::random();
@ -166,8 +166,13 @@ fn direct_v1_by_account_address() {
assert_eq!(raya.addr(), raya_account_addr.as_str());
let convo_id = saro.create_direct_conversation(&raya_account_addr).unwrap();
// DirectV1 is the pairwise shape, so the joiner sees it classed Private even
// though its welcome rides the InboxV2 (GroupV1 invite) path.
let raya_convo_id = expect_event(&raya_events, "ConversationStarted", |e| match e {
Event::ConversationStarted { convo_id, .. } => Ok(convo_id),
Event::ConversationStarted { convo_id, class } => {
assert_eq!(class, ConversationClass::Private);
Ok(convo_id)
}
other => Err(other),
});
@ -375,10 +380,16 @@ fn unpublished_account_address_is_an_error() {
let err = saro
.create_direct_conversation(&unpublished.address())
.expect_err("no bundle published for the account");
assert!(matches!(err, logos_chat::ClientError::AccountResolution(_)));
assert!(matches!(
err,
logos_generic_chat::ClientError::AccountResolution(_)
));
let err = saro
.create_direct_conversation("not-an-account-address")
.expect_err("not an account key");
assert!(matches!(err, logos_chat::ClientError::AccountResolution(_)));
assert!(matches!(
err,
logos_generic_chat::ClientError::AccountResolution(_)
));
}

16
crates/logos-chat/Cargo.toml Normal file
View File

@ -0,0 +1,16 @@
[package]
name = "logos-chat"
version = "0.1.0"
edition = "2024"
[lib]
crate-type = ["rlib"]
[dependencies]
# Workspace dependencies (sorted)
components = { workspace = true }
crossbeam-channel = { workspace = true }
embedded-logos-delivery = { workspace = true }
libchat = { workspace = true }
logos-account = { workspace = true, features = ["dev"] }
logos-generic-chat = { workspace = true }

12
crates/logos-chat/src/lib.rs Normal file
View File

@ -0,0 +1,12 @@
mod logos;
pub use logos::{LogosChatClient, LogosConfig, REGISTRY_ENDPOINT, open, open_with_transport};
// Facade re-exports so callers need no direct dependency on the transport
// crate.
pub use embedded_logos_delivery::{
DEFAULT_NETWORK_PRESET, DEFAULT_TCP_PORT, EmbeddedLogosDelivery, P2pConfig,
};
// Re-export the transport-generic client surface so callers depend on this
// crate alone.
pub use logos_generic_chat::*;

117
crates/logos-chat/src/logos.rs Normal file
View File

@ -0,0 +1,117 @@
//! The opinionated Logos client.
//!
//! [`open`] commits to the Logos service stack so independently built clients
//! share the same production services instead of each re-deriving them: a
//! delegate identity, the HTTP keypackage + account registry, and encrypted
//! on-disk storage. The stack is generic over the transport — any
//! [`Transport`] can be injected via [`open_with_transport`] — and the
//! concrete [`LogosChatClient`] commits to the embedded logos-delivery node,
//! whose native `liblogosdelivery` link keeps this crate outside the
//! workspace's default members.
//!
//! [`LogosChatClient`] points at `ChatClient`, which lives in
//! `logos-generic-chat`, so Rust's inherent-impl rule keeps the open
//! constructors off the alias; they are crate-level functions ([`open`],
//! [`open_with_transport`]) taking the all-inclusive [`LogosConfig`] instead.
use components::HttpRegistry;
use crossbeam_channel::Receiver;
use embedded_logos_delivery::{EmbeddedLogosDelivery, P2pConfig};
use libchat::{ChatStorage, StorageConfig};
use logos_account::TestLogosAccount;
use logos_generic_chat::{
ChatClient, ChatClientBuilder, ClientError, DelegateSigner, Event, Transport,
};
/// The endpoint for the account and keypackage registration service.
pub const REGISTRY_ENDPOINT: &str = "https://devnet.chat-kc.logos.co";
/// Configuration for opening a Logos client.
///
/// `db_path` (a per-client location) and `db_key` (a secret) are required and
/// never baked into the library. Everything else defaults: the registry
/// endpoint to the baked-in Logos value and the embedded node's p2p settings
/// to [`P2pConfig::default`]; override them with
/// [`set_registry_url`](Self::set_registry_url) and
/// [`set_p2p_config`](Self::set_p2p_config).
pub struct LogosConfig {
db_path: String,
db_key: String,
registry_url: String,
p2p_config: P2pConfig,
}
impl LogosConfig {
/// Config for the required per-client `db_path` and `db_key`. The registry
/// endpoint defaults to the baked-in Logos value; override it with
/// [`set_registry_url`](Self::set_registry_url).
pub fn new(db_path: impl Into<String>, db_key: impl Into<String>) -> Self {
Self {
db_path: db_path.into(),
db_key: db_key.into(),
registry_url: REGISTRY_ENDPOINT.to_string(),
p2p_config: P2pConfig::default(),
}
}
/// Override the registry endpoint (account + keypackage store; defaults to
/// the baked-in [`REGISTRY_ENDPOINT`]).
pub fn set_registry_url(&mut self, registry_url: impl Into<String>) {
self.registry_url = registry_url.into();
}
/// Override the embedded node's p2p settings (defaults to
/// [`P2pConfig::default`]). Only [`open`] starts an embedded node, so
/// [`open_with_transport`] ignores this.
pub fn set_p2p_config(&mut self, p2p_config: P2pConfig) {
self.p2p_config = p2p_config;
}
}
/// Open a client on the Logos stack per `config`, starting an embedded
/// logos-delivery node per its p2p settings as the transport. A convenience
/// over [`open_with_transport`] that commits to the [`LogosChatClient`]
/// transport.
pub fn open(config: LogosConfig) -> Result<(LogosChatClient, Receiver<Event>), ClientError> {
let transport = EmbeddedLogosDelivery::start(config.p2p_config.clone())
.map_err(|e| ClientError::Transport(e.to_string()))?;
open_with_transport(config, transport)
}
/// Open a client on the Logos stack per `config` with the injected transport,
/// persisting to the encrypted database.
#[allow(clippy::type_complexity)]
pub fn open_with_transport<T: Transport>(
config: LogosConfig,
transport: T,
) -> Result<(ChatClient<T, HttpRegistry, ChatStorage>, Receiver<Event>), ClientError> {
// A fresh account endorsing a fresh delegate each open: the account
// key is dropped after publishing the bundle, so devices cannot be
// added later. A caller-supplied, custody-holding account replaces
// this once the platform provides one.
let account = TestLogosAccount::new();
let delegate = DelegateSigner::random();
let mut registry = HttpRegistry::new(config.registry_url);
account
.add_delegate_signer(&mut registry, delegate.public_key())
.map_err(|e| ClientError::BundlePublish(e.to_string()))?;
ChatClientBuilder::new(account.address())
.ident(delegate)
.transport(transport)
.registration(registry)
.storage_config(StorageConfig::Encrypted {
path: config.db_path,
key: config.db_key,
})
.build()
}
/// The Logos client: a [`ChatClient`] wired to the Logos service stack — a
/// [`DelegateSigner`] identity acting for a fresh dev account, the HTTP
/// keypackage + account registry ([`HttpRegistry`], which is both the
/// keypackage store and the account → device directory), and encrypted
/// [`ChatStorage`] — running an embedded logos-delivery node as its
/// transport. Open one with [`open`], or swap the transport via
/// [`open_with_transport`].
pub type LogosChatClient = ChatClient<EmbeddedLogosDelivery, HttpRegistry, ChatStorage>;

4
extensions/components/Cargo.toml
View File

@ -2,10 +2,6 @@
name = "components"
version = "0.1.0"
edition = "2024"
links = "logosdelivery"
[features]
embedded_p2p_delivery = []
[dependencies]
# Workspace dependencies (sorted)

111
extensions/components/src/contact_registry/http.rs
View File

@ -125,7 +125,7 @@ impl RegistrationService for HttpRegistry {
};
let url = format!("{}/v0/keypackage", self.base_url);
let resp = self.http.post(&url).json(&req).send()?;
let resp = send_retrying(|| self.http.post(&url).json(&req))?;
if !resp.status().is_success() {
let status = resp.status().as_u16();
let body = resp.text().unwrap_or_default();
@ -136,7 +136,7 @@ impl RegistrationService for HttpRegistry {
fn retrieve(&self, device_id: &str) -> Result<Option<Vec<u8>>, HttpRegistryError> {
let url = format!("{}/v0/keypackage/{}", self.base_url, device_id);
let resp = self.http.get(&url).send()?;
let resp = send_retrying(|| self.http.get(&url))?;
if resp.status().as_u16() == 404 {
return Ok(None);
}
@ -188,7 +188,7 @@ impl AccountDirectory for HttpRegistry {
};
let url = format!("{}/v0/account", self.base_url);
let resp = self.http.post(&url).json(&req).send()?;
let resp = send_retrying(|| self.http.post(&url).json(&req))?;
if !resp.status().is_success() {
let status = resp.status().as_u16();
let body = resp.text().unwrap_or_default();
@ -203,7 +203,7 @@ impl AccountDirectory for HttpRegistry {
self.base_url,
hex::encode(account.as_ref())
);
let resp = self.http.get(&url).send()?;
let resp = send_retrying(|| self.http.get(&url))?;
if resp.status().as_u16() == 404 {
return Ok(None);
}
@ -266,6 +266,70 @@ fn decode_payload(payload: &[u8]) -> Option<(u64, &[u8])> {
Some((timestamp_ms, &payload[8..]))
}
/// Retry budget for the registry's transient, load-induced 5xx/429 responses.
/// The service is reliable request-by-request but sheds concurrent bursts, so a
/// few backed-off retries let a request land once the burst clears. On that path
/// each retry returns fast, so the added cost is the ~3s worst-case backoff sum,
/// well inside chat_module's ~20s init IPC budget. A fully unreachable registry
/// instead costs up to MAX_RETRIES times the reqwest timeout, which no retry
/// budget can rescue.
const MAX_RETRIES: u32 = 4;
const RETRY_BASE_MS: u64 = 200;
const RETRY_MAX_BACKOFF_MS: u64 = 2000;
/// Send a request built by `build`, retrying transient failures — network errors
/// and 5xx/429 responses — with exponential backoff and full jitter. The
/// registry is reliable request-by-request but sheds concurrent bursts with a
/// 5xx, so a backed-off retry lands once the burst clears; a 4xx (and any other
/// final response) is returned to the caller unchanged. `build` is re-invoked per
/// attempt because sending consumes the builder.
fn send_retrying(
build: impl Fn() -> reqwest::blocking::RequestBuilder,
) -> Result<reqwest::blocking::Response, HttpRegistryError> {
let mut attempt = 0;
loop {
let outcome = build().send();
let transient = match &outcome {
Err(_) => true, // network error / timeout: worth another try
Ok(resp) => is_transient_status(resp.status()),
};
if !transient || attempt >= MAX_RETRIES {
return Ok(outcome?);
}
std::thread::sleep(backoff_with_jitter(attempt));
attempt += 1;
}
}
/// Whether a response status is worth retrying: 5xx (the registry sheds
/// concurrent load with these) or 429 (explicit backpressure). A 4xx is the
/// caller's fault and won't change on retry.
fn is_transient_status(status: reqwest::StatusCode) -> bool {
status.is_server_error() || status == reqwest::StatusCode::TOO_MANY_REQUESTS
}
/// Full-jitter exponential backoff: a random delay in
/// `[0, min(RETRY_MAX_BACKOFF_MS, RETRY_BASE_MS * 2^attempt)]`. The jitter
/// decorrelates concurrent publishers so their retries don't collide into the
/// same burst that failed them.
fn backoff_with_jitter(attempt: u32) -> Duration {
let exp = RETRY_BASE_MS.saturating_mul(1u64 << attempt.min(16));
Duration::from_millis(jitter_below(exp.min(RETRY_MAX_BACKOFF_MS)))
}
/// A value in `[0, max]`, seeded from the wall clock's sub-second nanos — enough
/// entropy to spread retries across processes without pulling in an RNG crate.
fn jitter_below(max: u64) -> u64 {
if max == 0 {
return 0;
}
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.subsec_nanos() as u64)
.unwrap_or(0);
nanos % (max + 1)
}
#[cfg(test)]
mod tests {
use super::*;
@ -325,4 +389,43 @@ mod tests {
.verify(&payload, &signature)
.expect("recovered key must verify the register-time signature");
}
/// Only 5xx and 429 are retried; 2xx/4xx are returned to the caller as-is.
#[test]
fn only_5xx_and_429_are_transient() {
use reqwest::StatusCode;
for s in [500u16, 502, 503, 504, 429] {
assert!(
is_transient_status(StatusCode::from_u16(s).unwrap()),
"{s} should be retried"
);
}
for s in [200u16, 201, 400, 401, 404, 409] {
assert!(
!is_transient_status(StatusCode::from_u16(s).unwrap()),
"{s} should not be retried"
);
}
}
/// Backoff never exceeds the exponential ceiling for its attempt, nor the
/// absolute cap — and the exponent shift can't overflow at high attempts.
#[test]
fn backoff_stays_within_the_cap() {
for attempt in 0..40u32 {
let ceiling = RETRY_BASE_MS
.saturating_mul(1u64 << attempt.min(16))
.min(RETRY_MAX_BACKOFF_MS);
let delay = backoff_with_jitter(attempt).as_millis() as u64;
assert!(delay <= ceiling, "attempt {attempt}: {delay} > {ceiling}");
}
}
#[test]
fn jitter_is_bounded() {
assert_eq!(jitter_below(0), 0);
for _ in 0..200 {
assert!(jitter_below(50) <= 50);
}
}
}

6
extensions/components/src/delivery.rs
View File

@ -1,9 +1,3 @@
mod local_broadcaster;
pub use local_broadcaster::LocalBroadcaster;
#[cfg(feature = "embedded_p2p_delivery")]
pub mod embedded_p2p_delivery;
#[cfg(feature = "embedded_p2p_delivery")]
pub use embedded_p2p_delivery::{EmbeddedP2pDeliveryService, P2pConfig};

4
extensions/components/src/delivery/local_broadcaster.rs
View File

@ -69,6 +69,10 @@ impl LocalBroadcaster {
/// Pulls all messages this consumer has not yet seen on `address`,
/// applying any registered filter. Advances the cursor so the same
/// messages are not returned again.
// clippy's question_mark (1.97+) wants `self.shared.borrow().read(next)?`, but
// `read` returns a reference into the RefCell `Ref`; the `?` form drops that
// guard at the `;` and `ae` would dangle. Keep the explicit match.
#[allow(clippy::question_mark)]
pub fn poll(&mut self) -> Option<Vec<u8>> {
loop {
let next = self.cursor;

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

@ -0,0 +1,18 @@
[package]
name = "embedded-logos-delivery"
version = "0.1.0"
edition = "2024"
[dependencies]
# Workspace dependencies (sorted)
crossbeam-channel = { workspace = true }
libchat = { workspace = true }
logos-delivery = { 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"

13
extensions/components/build.rs → extensions/embedded-logos-delivery/build.rs
View File

@ -5,17 +5,12 @@ use std::process::Command;
fn main() {
println!("cargo:rerun-if-env-changed=LOGOS_DELIVERY_LIB_DIR");
if std::env::var_os("CARGO_FEATURE_EMBEDDED_P2P_DELIVERY").is_none() {
return;
}
let Some(lib_dir) = locate_lib_dir() else {
println!(
"cargo:warning=embedded_p2p_delivery feature is enabled but \
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."
"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;
};

100
extensions/embedded-logos-delivery/src/lib.rs Normal file
View File

@ -0,0 +1,100 @@
//! The embedded logos-delivery transport service.
//!
//! [`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.
//!
//! 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`.
use crossbeam_channel::Receiver;
use libchat::{AddressedEnvelope, DeliveryService};
use logos_delivery::{ThreadedDeliveryWrapper, WakuEvent};
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";
/// Default TCP port for the embedded logos-delivery node.
pub const DEFAULT_TCP_PORT: u16 = 60000;
/// The content-topic prefix carrying logos-chat traffic.
const CHAT_TOPIC_PREFIX: &str = "/logos-chat/1/";
pub fn content_topic_for(delivery_address: &str) -> String {
format!("{CHAT_TOPIC_PREFIX}{delivery_address}/proto")
}
// ── EmbeddedLogosDelivery ──────────────────────────────────────────────────
/// logos-delivery backed delivery service. Cheap to clone — all clones share
/// the same background node.
#[derive(Clone, Debug)]
pub struct EmbeddedLogosDelivery {
inner: ThreadedDeliveryWrapper<Vec<u8>>,
}
impl EmbeddedLogosDelivery {
/// 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 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()
})?;
Ok(Self { inner })
}
/// Stop delivering messages addressed to `delivery_address`.
pub fn unsubscribe(&self, delivery_address: &str) -> Result<(), DeliveryError> {
self.inner.unsubscribe(&content_topic_for(delivery_address))
}
}
impl DeliveryService for EmbeddedLogosDelivery {
type Error = DeliveryError;
fn publish(&mut self, envelope: AddressedEnvelope) -> Result<(), DeliveryError> {
debug!(
topic = &content_topic_for(&envelope.delivery_address),
"Publish"
);
self.inner.publish(
&content_topic_for(&envelope.delivery_address),
&envelope.data,
)
}
fn subscribe(
&mut self,
delivery_address: &str,
) -> Result<(), <Self as DeliveryService>::Error> {
self.inner.subscribe(&content_topic_for(delivery_address))
}
}
// Teaching the service the inbound half makes it a full client transport, so
// callers need no wrapper newtype. The impl lives here (the crate owning the
// type) because the orphan rule bars it from the `logos-chat` crate, which
// owns neither the trait nor the type.
impl logos_generic_chat::Transport for EmbeddedLogosDelivery {
fn inbound(&mut self) -> Receiver<Vec<u8>> {
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/components/src/delivery/embedded_p2p_delivery/sys.rs → extensions/logos-delivery-rust/src/sys.rs
View File

310
extensions/components/src/delivery/embedded_p2p_delivery.rs → extensions/logos-delivery-rust/src/threaded.rs
View File

@ -1,17 +1,3 @@
//! logos-delivery backed [`client::DeliveryService`] implementation.
//!
//! `LogosDeliveryService` wraps an embedded logos-delivery node running on a
//! dedicated `std::thread`. All interaction is via synchronous `std::sync::mpsc`
//! channels.
//!
//! ## 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;
@ -19,14 +5,9 @@ use std::time::Duration;
use base64::Engine;
use base64::engine::general_purpose::STANDARD as BASE64;
use crossbeam_channel::{Receiver, Sender};
use libchat::{AddressedEnvelope, DeliveryService};
use tracing::{error, info, warn};
use tracing::{error, info};
use wrapper::LogosNodeCtx;
pub fn content_topic_for(delivery_address: &str) -> String {
format!("/logos-chat/1/{delivery_address}/proto")
}
use crate::wrapper::LogosNodeCtx;
// ── Error ────────────────────────────────────────────────────────────────────
@ -36,33 +17,52 @@ pub enum DeliveryError {
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 ────────────────────────────────────────────────────────────────
struct OutboundCmd {
message_json: String,
/// 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 = Arc<Mutex<Vec<Sender<Vec<u8>>>>>;
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 tcp_port: u16,
pub port: u16,
pub log_level: String,
}
impl Default for P2pConfig {
fn default() -> Self {
Self {
preset: "logos.dev".into(),
tcp_port: 60000,
preset: DEFAULT_NETWORK_PRESET.into(),
port: DEFAULT_PORT,
log_level: "ERROR".into(),
}
}
@ -81,16 +81,25 @@ struct WakuMessage {
}
/// Top-level event envelope received from the logos-delivery node callback.
#[derive(Debug, serde::Deserialize)]
struct WakuEvent {
#[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)]
struct ReceivedMessage {
#[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
@ -98,8 +107,19 @@ struct ReceivedMessage {
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)]
#[derive(Debug, serde::Deserialize, Clone)]
#[serde(untagged)]
enum WakuPayload {
Base64(String),
@ -115,28 +135,54 @@ impl WakuPayload {
}
}
// ── EmbeddedP2pDeliveryService ──────────────────────────────────────────────────
// ── ThreadedDeliveryWrapper ─────────────────────────────────────────────────
/// logos-delivery backed delivery service. Cheap to clone — all clones share
/// the same background node.
#[derive(Clone, Debug)]
pub struct EmbeddedP2pDeliveryService {
outbound: mpsc::SyncSender<OutboundCmd>,
/// 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,
inbound_rx: Option<Receiver<Vec<u8>>>,
subscribers: SubscriberList<T>,
inbound_rx: Option<Receiver<T>>,
}
impl EmbeddedP2pDeliveryService {
/// Start the embedded logos-delivery node. The client drains inbound
/// payloads via [`Transport::inbound`].
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()));
// 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::<Vec<u8>>(1024);
let (inbound_tx, inbound_rx) = crossbeam_channel::bounded::<T>(1024);
let subs_for_thread = subscribers.clone();
@ -144,7 +190,7 @@ impl EmbeddedP2pDeliveryService {
.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);
Self::node_thread(cfg, out_rx, subs_for_thread, inbound_tx, ready_tx, map);
})) {
let msg = panic
.downcast_ref::<&str>()
@ -176,13 +222,58 @@ impl EmbeddedP2pDeliveryService {
})
}
fn node_thread(
/// 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<OutboundCmd>,
subscribers: SubscriberList,
inbound_tx: Sender<Vec<u8>>,
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.
@ -190,8 +281,8 @@ impl EmbeddedP2pDeliveryService {
"logLevel": cfg.log_level,
"mode": "Core",
"preset": cfg.preset,
"tcpPort": cfg.tcp_port,
"discv5UdpPort": cfg.tcp_port,
"tcpPort": cfg.port,
"discv5UdpPort": cfg.port,
})
.to_string();
@ -210,20 +301,24 @@ impl EmbeddedP2pDeliveryService {
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,
});
}
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);
@ -241,74 +336,25 @@ impl EmbeddedP2pDeliveryService {
// 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);
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 outbound loop finished");
}
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")
}
}
impl DeliveryService for EmbeddedP2pDeliveryService {
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)?
}
fn subscribe(&mut self, _: &str) -> Result<(), <Self as DeliveryService>::Error> {
// This Service does not support filtering
Ok(())
info!("logos-node command loop finished");
}
}

30
extensions/components/src/delivery/embedded_p2p_delivery/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()));