diff --git a/Cargo.lock b/Cargo.lock index b2d6311..caa1ced 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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?branch=main#e7d2726fd005e56329f00f91e8c07672a3d3639b" dependencies = [ "hashgraph-like-consensus", "indexmap 2.14.0", @@ -2543,9 +2543,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", diff --git a/core/conversations/Cargo.toml b/core/conversations/Cargo.toml index ac681ee..a2114a2 100644 --- a/core/conversations/Cargo.toml +++ b/core/conversations/Cargo.toml @@ -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", branch = "main" } 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" diff --git a/core/conversations/src/conversation.rs b/core/conversations/src/conversation.rs index 7774ff4..37edb61 100644 --- a/core/conversations/src/conversation.rs +++ b/core/conversations/src/conversation.rs @@ -9,7 +9,7 @@ use crate::proto::EncryptedPayload; use crate::service_context::{ExternalServices, ServiceContext}; 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; diff --git a/core/conversations/src/conversation/group_v2.rs b/core/conversations/src/conversation/group_v2.rs index d2b91d9..6b42f70 100644 --- a/core/conversations/src/conversation/group_v2.rs +++ b/core/conversations/src/conversation/group_v2.rs @@ -11,7 +11,8 @@ use de_mls::protos::de_mls::messages::v1::{ AppMessage as AppMessageProto, MemberWelcome, app_message, }; use de_mls::{ - Conversation, ConversationEvent, PeerScoringService, ScoringConfig, default_score_deltas, + Conversation, ConversationEvent, MockClock, PeerScoringService, ScoringConfig, WallClock, + default_score_deltas, defaults::{DefaultConsensusPlugin, DefaultPeerScoring, InMemoryPeerScoreStorage}, }; use hashgraph_like_consensus::signing::EthereumConsensusSigner; @@ -21,6 +22,7 @@ use openmls::prelude::{KeyPackageIn, OpenMlsProvider as _, ProtocolVersion}; use prost::Message; use shared_traits::{IdentId, IdentIdRef}; use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tracing::{info, instrument, warn}; use crate::IdentityProvider; @@ -30,6 +32,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(service_ctx: &ServiceContext) -> Vec { @@ -58,7 +83,7 @@ fn make_consensus() -> DefaultConsensusPlugin { pub struct GroupV2Convo { convo_id: String, - conversation: Conversation, + conversation: Conversation, /// 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. @@ -98,8 +123,9 @@ impl GroupV2Convo { &service_ctx.mls_identity, &make_consensus(), make_scoring(), + service_ctx.demls_clock.clone(), rand_app_id(), - service_ctx.group_v2_config.clone(), + service_ctx.demls_config.clone(), )?; let convo = GroupV2Convo { convo_id, @@ -129,8 +155,9 @@ impl GroupV2Convo { &welcome.conversation_sync_bytes, &make_consensus(), make_scoring(), + service_ctx.demls_clock.clone(), rand_app_id(), - service_ctx.group_v2_config.clone(), + service_ctx.demls_config.clone(), )? else { return Err(ChatError::generic("welcome not addressed to this member")); diff --git a/core/conversations/src/core.rs b/core/conversations/src/core.rs index bab9fad..f0f48d1 100644 --- a/core/conversations/src/core.rs +++ b/core/conversations/src/core.rs @@ -3,7 +3,10 @@ use crate::conversation::{ ConversationIdRef, DirectV1Convo, GroupV1Convo, GroupV2Convo, Identified, PrivateV1Convo, }; use crate::service_context::{ExternalServices, ServiceContext}; -use crate::{DeliveryService, GroupV2Config, IdentityProvider, RegistrationService, WakeupService}; +use crate::{ + DeliveryService, GroupV2Clock, GroupV2Config, IdentityProvider, RegistrationService, + WakeupService, +}; use crate::{ conversation::{Convo, GroupConvo}, errors::ChatError, @@ -100,6 +103,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,7 +154,8 @@ where causal, identity, wakeup_service, - group_v2_config: GroupV2Config::default(), + demls_clock: GroupV2Clock::default(), + demls_config: GroupV2Config::default(), }, inbox, pq_inbox, @@ -175,12 +190,6 @@ impl<'a, S: ExternalServices + 'static> Core { self.pq_inbox.register(&mut self.services) } - /// Timing/policy for GroupV2 conversations created or joined after this - /// call. Existing conversations keep the config they were built with. - pub fn set_group_v2_config(&mut self, config: GroupV2Config) { - self.services.group_v2_config = config; - } - pub fn installation_name(&self) -> &str { self.services.identity.get_name() } diff --git a/core/conversations/src/lib.rs b/core/conversations/src/lib.rs index 9eac46c..4cfa0b6 100644 --- a/core/conversations/src/lib.rs +++ b/core/conversations/src/lib.rs @@ -15,6 +15,7 @@ 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 @@ -23,6 +24,7 @@ pub use core::{ConversationId, Core, Introduction}; /// 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, diff --git a/core/conversations/src/service_context.rs b/core/conversations/src/service_context.rs index 3f7372d..c698faf 100644 --- a/core/conversations/src/service_context.rs +++ b/core/conversations/src/service_context.rs @@ -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,11 +45,12 @@ pub(crate) struct ServiceContext { pub(crate) causal: CausalHistoryStore, pub(crate) identity: Identity, pub(crate) wakeup_service: S::WS, - /// Timing/policy applied to GroupV2 conversations created or joined by - /// this core. Read at conversation construction; a joiner's phase - /// durations are then overwritten by the creator's, carried with the - /// welcome (vote delays and policy fields stay local). - pub(crate) group_v2_config: de_mls::ConversationConfig, + /// 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)] @@ -115,7 +117,8 @@ mod test_support { causal: CausalHistoryStore::new(), identity: Identity::new(name), wakeup_service: NoopWakeups {}, - group_v2_config: de_mls::ConversationConfig::default(), + demls_clock: GroupV2Clock::default(), + demls_config: de_mls::ConversationConfig::default(), }) } } diff --git a/core/integration_tests_core/src/test_client.rs b/core/integration_tests_core/src/test_client.rs index 9b75852..c13fe1e 100644 --- a/core/integration_tests_core/src/test_client.rs +++ b/core/integration_tests_core/src/test_client.rs @@ -1,5 +1,6 @@ use crate::test_ident::TestIdent; -use libchat::{ConversationId, Core, GroupV2Config, IdentityProvider, PayloadOutcome}; +use libchat::{ConversationId, Core, IdentityProvider, PayloadOutcome}; +use libchat::{GroupV2Clock, GroupV2Config}; use shared_traits::IdentId; use std::collections::HashMap; use std::fmt::Debug; @@ -21,21 +22,6 @@ const RAYA: usize = 1; const PAX: usize = 2; const MIRA: usize = 3; -/// Millisecond GroupV2 timers for the harness. de-mls deadlines are real -/// wall-clock, so the library defaults (60s commit inactivity) would stall -/// `process_until`, which settles in 50ms steps. -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 ClientType = CoreClient; type ClientType = Core<(TestIdent, LocalBroadcaster, EphemeralRegistry, WP, MemStore)>; @@ -166,6 +152,7 @@ impl TestHarness { 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); @@ -307,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::*; diff --git a/core/integration_tests_core/src/wakeup.rs b/core/integration_tests_core/src/wakeup.rs index abcb125..c3d693e 100644 --- a/core/integration_tests_core/src/wakeup.rs +++ b/core/integration_tests_core/src/wakeup.rs @@ -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>, + 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 { 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() }