feat(gwv2): shared-crate groundwork for the gatewayv2 gateway

Everything the gatewaydv2 daemon needs from existing crates, kept
additive so v1 is untouched at runtime:

- fedimint-eventlog/client: per-operation event-log index and
  replay+tail subscription, bridged so the entry encoding is unchanged
- fedimint-gwv2-client: daemon-driven send/receive primitives
  (start_receive, finalize_send with failure reason, per-operation
  event emission), richer payment events (preimage, realized ln fee,
  forfeit signature, destination node) and an optional daemon callback
  so the v2 daemon can orchestrate payments without the v1 state
  machines
- fedimint-lightning: migrate the LDK client to ldk-node 0.7, pinned to
  a fork backporting upstream's probing service (lightningdevkit/ldk-node#815)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: joschisan

Cargo.lock

@@ -273,11 +273,14 @@ jsonrpsee-core = "0.24.9"
 jsonrpsee-types = "0.24.8"
 jsonrpsee-wasm-client = "0.24.9"
 jsonrpsee-ws-client = { version = "0.24.9", default-features = false }
-ldk-node = { version = "0.6.1", package = "fedimint-ldk-node" }
+# fedimint-0.7.0 plus a backport of upstream ldk-node's probing service
+# (lightningdevkit/ldk-node#815); switch back to crates.io once a release
+# containing the probing service ships.
+ldk-node = { package = "fedimint-ldk-node", git = "https://github.com/joschisan/ldk-node", rev = "adf392e4a82923f52e824a9ce5fd27964029bcfc" }
 leptos = { version = "0.7.8", default-features = false }
-lightning = "0.1.3"
-lightning-invoice = { version = "0.33.2", features = ["std"] }
-lightning-types = "0.2.0"
+lightning = "0.2.0"
+lightning-invoice = { version = "0.34.0", features = ["std"] }
+lightning-types = "0.3.0"
 lnurl-rs = { version = "0.9.0", default-features = false }
 lockable = "0.1.1"
 lru = "0.16.3"

Cargo.toml

@@ -130,6 +130,7 @@ pub trait ClientContextIface: MaybeSend + MaybeSync {
         dbtx: &mut DatabaseTransaction<'_, NonCommittable>,
         module_kind: Option<ModuleKind>,
         module_id: ModuleInstanceId,
+        operation: Option<OperationId>,
         kind: EventKind,
         payload: serde_json::Value,
         persist: EventPersistence,
@@ -824,6 +825,34 @@ where
     where
         E: Event + Send,
         Cap: Send,
+    {
+        self.log_event_inner(dbtx, None, event).await;
+    }
+
+    /// Like [`Self::log_event`], but additionally tags the event with
+    /// `operation`, indexing it into the per-operation secondary log so it can
+    /// be read/streamed via [`crate::ClientContextIface`]-backed
+    /// `read_operation_events` / `subscribe_operation_events`.
+    pub async fn log_event_for_operation<E, Cap>(
+        &self,
+        dbtx: &mut DatabaseTransaction<'_, Cap>,
+        operation: OperationId,
+        event: E,
+    ) where
+        E: Event + Send,
+        Cap: Send,
+    {
+        self.log_event_inner(dbtx, Some(operation), event).await;
+    }
+
+    async fn log_event_inner<E, Cap>(
+        &self,
+        dbtx: &mut DatabaseTransaction<'_, Cap>,
+        operation: Option<OperationId>,
+        event: E,
+    ) where
+        E: Event + Send,
+        Cap: Send,
     {
         if <E as Event>::MODULE != Some(<M as ClientModule>::kind()) {
             warn!(
@@ -839,6 +868,7 @@ where
                 &mut dbtx.global_dbtx(self.global_dbtx_access_token).to_ref_nc(),
                 <E as Event>::MODULE,
                 self.module_instance_id,
+                operation,
                 <E as Event>::KIND,
                 serde_json::to_value(event).expect("Can't fail"),
                 <E as Event>::PERSISTENCE,

fedimint-client-module/src/module/mod.rs

@@ -7,7 +7,7 @@ use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use anyhow::{Context as _, anyhow, bail, format_err};
-use async_stream::try_stream;
+use async_stream::{stream, try_stream};
 use bitcoin::key::Secp256k1;
 use bitcoin::key::rand::thread_rng;
 use bitcoin::secp256k1::{self, PublicKey};
@@ -2347,6 +2347,7 @@ impl Client {
         dbtx: &mut DatabaseTransaction<'_, Cap>,
         kind: EventKind,
         module: Option<(ModuleKind, ModuleInstanceId)>,
+        operation: Option<OperationId>,
         payload: Vec<u8>,
         persist: EventPersistence,
     ) where
@@ -2359,6 +2360,7 @@ impl Client {
             kind,
             module_kind,
             module_id,
+            operation,
             payload,
             persist,
         )
@@ -2483,6 +2485,51 @@ impl Client {
             .await
     }
 
+    /// One-shot snapshot of every event currently logged for `operation`, in
+    /// event-log order, read from the per-operation secondary index.
+    pub async fn read_operation_events(&self, operation: OperationId) -> Vec<PersistedLogEntry> {
+        self.db
+            .begin_transaction_nc()
+            .await
+            .get_operation_event_log(operation)
+            .await
+    }
+
+    /// Stream every event logged for `operation`, in event-log order: history
+    /// is replayed first, then new events are yielded as the ordering task
+    /// commits them. The stream ends only when the client's event-log ordering
+    /// task stops (i.e. on client shutdown).
+    pub fn subscribe_operation_events(
+        &self,
+        operation: OperationId,
+    ) -> BoxStream<'static, PersistedLogEntry> {
+        let db = self.db.clone();
+        let mut log_event_added = self.log_event_added_rx.clone();
+
+        Box::pin(stream! {
+            let mut next = EventLogId::LOG_START;
+
+            loop {
+                let entries = db
+                    .begin_transaction_nc()
+                    .await
+                    .get_operation_event_log(operation)
+                    .await;
+
+                for entry in entries {
+                    if next <= entry.id() {
+                        next = entry.id().next();
+                        yield entry;
+                    }
+                }
+
+                if log_event_added.changed().await.is_err() {
+                    break;
+                }
+            }
+        })
+    }
+
     pub async fn get_event_log_trimable(
         &self,
         pos: Option<EventLogTrimableId>,
@@ -2707,6 +2754,7 @@ impl ClientContextIface for Client {
         dbtx: &mut DatabaseTransaction<'_, NonCommittable>,
         module_kind: Option<ModuleKind>,
         module_id: ModuleInstanceId,
+        operation: Option<OperationId>,
         kind: EventKind,
         payload: serde_json::Value,
         persist: EventPersistence,
@@ -2717,6 +2765,7 @@ impl ClientContextIface for Client {
             dbtx,
             kind,
             module_kind.map(|kind| (kind, module_id)),
+            operation,
             serde_json::to_vec(&payload).expect("Serialization can't fail"),
             persist,
         )

fedimint-client/src/client.rs

@@ -109,6 +109,7 @@ impl IGlobalClientContext for ModuleGlobalClientContext {
                 dbtx.global_tx(),
                 kind,
                 module,
+                None,
                 serde_json::to_vec(&payload).expect("Serialization can't fail"),
                 persist,
             )

fedimint-client/src/client/global_ctx.rs

@@ -21,7 +21,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
 use std::{fmt, ops};
 
-use fedimint_core::core::{ModuleInstanceId, ModuleKind};
+use fedimint_core::core::{ModuleInstanceId, ModuleKind, OperationId};
 use fedimint_core::db::{
     Database, DatabaseTransaction, IDatabaseTransactionOpsCoreTyped, NonCommittable,
 };
@@ -45,6 +45,16 @@ use tracing::{debug, trace};
 pub const DB_KEY_PREFIX_UNORDERED_EVENT_LOG: u8 = 0x3a;
 pub const DB_KEY_PREFIX_EVENT_LOG: u8 = 0x39;
 pub const DB_KEY_PREFIX_EVENT_LOG_TRIMABLE: u8 = 0x41;
+/// Transient bridge, written in the same dbtx as an operation-tagged unordered
+/// event: maps the (write-time) [`UnordedEventLogId`] to its `OperationId`.
+/// The ordering task consumes it to build
+/// [`DB_KEY_PREFIX_EVENT_LOG_BY_OPERATION`] once the final [`EventLogId`] is
+/// known, then deletes it.
+pub const DB_KEY_PREFIX_UNORDERED_EVENT_OPERATION: u8 = 0x3d;
+/// Secondary index for operation-scoped tailing: `(OperationId, EventLogId)` ->
+/// the full [`EventLogEntry`], so `subscribe_operation_events` is a cheap range
+/// scan without dereferencing back into the main ordered log.
+pub const DB_KEY_PREFIX_EVENT_LOG_BY_OPERATION: u8 = 0x3e;
 
 /// Minimum age in ID count for trimable events to be deleted
 const TRIMABLE_EVENTLOG_MIN_ID_AGE: u64 = 10_000;
@@ -103,7 +113,7 @@ static UNORDEREDED_EVENT_LOG_ID_COUNTER: AtomicU64 = AtomicU64::new(0);
 /// conflicts due the ID allocation. Instead they are picked based on
 /// a time and a counter, so they are mostly but not strictly ordered and
 /// monotonic, and even more importantly: not contiguous.
-#[derive(Debug, Encodable, Decodable)]
+#[derive(Debug, Clone, Copy, Encodable, Decodable)]
 pub struct UnordedEventLogId {
     ts_usecs: u64,
     counter: u64,
@@ -446,6 +456,40 @@ impl_db_lookup!(
     query_prefix = UnorderedEventLogIdPrefixAll
 );
 
+/// Bridge key: the write-time [`UnordedEventLogId`] of an operation-tagged
+/// event -> its [`OperationId`]. Written by the operation-aware `log_event`,
+/// consumed and deleted by the ordering task.
+#[derive(Clone, Debug, Encodable, Decodable)]
+pub struct UnorderedEventOperationKey(pub UnordedEventLogId);
+
+impl_db_record!(
+    key = UnorderedEventOperationKey,
+    value = OperationId,
+    db_prefix = DB_KEY_PREFIX_UNORDERED_EVENT_OPERATION,
+);
+
+/// Secondary per-operation index key. Ordered by `(operation, event_id)`, so a
+/// range scan over one operation yields its events in event-log order.
+#[derive(Clone, Debug, Encodable, Decodable)]
+pub struct EventLogByOperationKey {
+    pub operation: OperationId,
+    pub event_id: EventLogId,
+}
+
+#[derive(Clone, Debug, Encodable, Decodable)]
+pub struct EventLogByOperationKeyPrefix(pub OperationId);
+
+impl_db_record!(
+    key = EventLogByOperationKey,
+    value = EventLogEntry,
+    db_prefix = DB_KEY_PREFIX_EVENT_LOG_BY_OPERATION,
+);
+
+impl_db_lookup!(
+    key = EventLogByOperationKey,
+    query_prefix = EventLogByOperationKeyPrefix
+);
+
 #[derive(Clone, Debug, Encodable, Decodable)]
 pub struct EventLogIdPrefixAll;
 
@@ -525,6 +569,7 @@ pub trait DBTransactionEventLogExt {
         kind: EventKind,
         module_kind: Option<ModuleKind>,
         module_id: Option<ModuleInstanceId>,
+        operation: Option<OperationId>,
         payload: Vec<u8>,
         persist: EventPersistence,
     );
@@ -546,12 +591,18 @@ pub trait DBTransactionEventLogExt {
             E::KIND,
             E::MODULE,
             module_id,
+            None,
             serde_json::to_vec(&event).expect("Serialization can't fail"),
             <E as Event>::PERSISTENCE,
         )
         .await;
     }
 
+    /// All events logged for `operation`, in event-log order, read from the
+    /// per-operation secondary index. Each returned [`PersistedLogEntry`] is
+    /// keyed by the entry's ordered [`EventLogId`].
+    async fn get_operation_event_log(&mut self, operation: OperationId) -> Vec<PersistedLogEntry>;
+
     /// Next [`EventLogId`] to use for new ordered events.
     ///
     /// Used by ordering task, though might be
@@ -586,6 +637,7 @@ where
         kind: EventKind,
         module_kind: Option<ModuleKind>,
         module_id: Option<ModuleInstanceId>,
+        operation: Option<OperationId>,
         payload: Vec<u8>,
         persist: EventPersistence,
     ) {
@@ -623,11 +675,30 @@ where
         {
             panic!("Trying to overwrite event in the client event log");
         }
+
+        // Record the operation association so the ordering task can build the
+        // per-operation secondary index once it assigns the final `EventLogId`.
+        if let Some(operation) = operation {
+            self.insert_entry(&UnorderedEventOperationKey(unordered_id), &operation)
+                .await;
+        }
+
         self.on_commit(move || {
             log_ordering_wakeup_tx.send_replace(());
         });
     }
 
+    async fn get_operation_event_log(&mut self, operation: OperationId) -> Vec<PersistedLogEntry> {
+        self.find_by_prefix(&EventLogByOperationKeyPrefix(operation))
+            .await
+            .map(|(k, v)| PersistedLogEntry {
+                id: k.event_id,
+                inner: v,
+            })
+            .collect()
+            .await
+    }
+
     async fn get_next_event_log_id(&mut self) -> EventLogId {
         self.find_by_prefix_sorted_descending(&EventLogIdPrefixAll)
             .await
@@ -747,6 +818,14 @@ pub async fn run_event_log_ordering_task(
                 dbtx.remove_entry(unordered_id).await.is_some(),
                 "Must never fail to remove entry"
             );
+
+            // Consume the operation bridge (if any) written alongside this event.
+            // The per-operation index is keyed by the default log's `EventLogId`,
+            // so it is only built for non-trimable persisted events below.
+            let operation = dbtx
+                .remove_entry(&UnorderedEventOperationKey(*unordered_id))
+                .await;
+
             if entry.persist() {
                 // Non-trimable events get persisted in both the default event log
                 // and trimable event log
@@ -757,6 +836,20 @@ pub async fn run_event_log_ordering_task(
                             .is_none(),
                         "Must never overwrite existing event"
                     );
+                    if let Some(operation) = operation {
+                        assert!(
+                            dbtx.insert_entry(
+                                &EventLogByOperationKey {
+                                    operation,
+                                    event_id: next_entry_id,
+                                },
+                                &entry.inner,
+                            )
+                            .await
+                            .is_none(),
+                            "Must never overwrite existing per-operation event"
+                        );
+                    }
                     trace!(target: LOG_CLIENT_EVENT_LOG, ?unordered_id, id=?next_entry_id, "Ordered event log event");
                     next_entry_id = next_entry_id.next();
                 }