feat: inject the GroupV2 time source and timing config (#168)

* add WallClock into groupv2

* switch to main branch, remove custom type

---------

Co-authored-by: Jazz Turner-Baggs <473256+jazzz@users.noreply.github.com>

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",

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"

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;
 

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<S: ExternalServices>(service_ctx: &ServiceContext<S>) -> Vec<u8> {
@@ -58,7 +83,7 @@ fn make_consensus() -> DefaultConsensusPlugin {
 
 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.
@@ -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"));

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<S> {
         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()
     }

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,

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<S: ExternalServices> {
     pub(crate) causal: CausalHistoryStore,
     pub(crate) identity: Identity,
     pub(crate) wakeup_service: S::WS,
-    /// Timing/policy applied to GroupV2 conversations created or joined by
+    /// Time source for GroupV2 (de-mls) conversations.
-    /// this core. Read at conversation construction; a joiner's phase
+    pub(crate) demls_clock: GroupV2Clock,
-    /// durations are then overwritten by the creator's, carried with the
+    /// Timing/policy for GroupV2 (de-mls) conversations, applied at
-    /// welcome (vote delays and policy fields stay local).
+    /// create/join. The creator's phase durations reach joiners inside the
-    pub(crate) group_v2_config: de_mls::ConversationConfig,
+    /// 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(),
             })
         }
     }

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<TestIdent, LocalBroadcaster, EphemeralRegistry, WP, MemStore>;
 type ClientType = Core<(TestIdent, LocalBroadcaster, EphemeralRegistry, WP, MemStore)>;
 
@@ -166,6 +152,7 @@ impl<const N: usize> TestHarness<N> {
             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::*;

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<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
+        self.clock.advance(duration);
-        // 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);
-
         srv.now = srv.now.checked_add(duration).unwrap();
         srv.get_expired()
     }