gl-plugin: Upgrade cln_grpc and cln_rpc

Upgrade to the current version and, adjust the wrapper and other
messages to the changes. Need to fill out the streams after we
made NotificationStream - respectively the server module public.

Signed-off-by: Peter Neuroth <pet.v.ne@gmail.com>

Author: nepet

libs/gl-plugin/Cargo.toml

@@ -2,10 +2,7 @@
 name = "gl-plugin"
 version = "0.3.0"
 edition = "2021"
-authors = [
-        "Christian Decker",
-        "The Greenlight Team"
-]
+authors = ["Christian Decker", "The Greenlight Team"]
 description = "Python bindings for the Greenlight client library."
 repository = "https://github.com/Blockstream/greenlight"
 license = "MIT"
@@ -33,17 +30,17 @@ lazy_static = "^1.4.0"
 linemux = "0.2"
 log = "0.4"
 nix = "^0"
-prost = "0.11"
+prost = "0.12"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1"
 sled = "0.34"
 thiserror = "1"
 tokio = { version = "1", features = ["full"] }
 tokio-stream = { version = "0.1", features = ["net"] }
 tokio-util = { version = "0.7", features = ["codec"] }
-tonic = { version = "^0.8", features = ["tls", "transport"] }
+tonic = { version = "0.11", features = ["tls", "transport"] }
 tower = { version = "0.4" }
 vls-protocol = { workspace = true }
 
 [build-dependencies]
-tonic-build = "^0.8"
+tonic-build = "0.11"

libs/gl-plugin/src/awaitables.rs

@@ -1,9 +1,6 @@
 use cln_rpc::{
     model::{
-        requests::{
-            ConnectRequest, GetinfoRequest, GetrouteRequest, ListpeerchannelsRequest,
-            ListpeersRequest,
-        },
+        requests::{ConnectRequest, GetinfoRequest, GetrouteRequest, ListpeerchannelsRequest},
         responses::GetrouteResponse,
     },
     primitives::{Amount, PublicKey, ShortChannelId},
@@ -311,7 +308,6 @@ async fn billboard(
             .await
             .map_err(|e| Error::Rpc(e))?
             .channels
-            .ok_or(Error::Channel("No channels found"))?
             .into_iter()
             .filter(|c| {
                 c.short_channel_id == Some(scid)
@@ -324,28 +320,10 @@ async fn billboard(
             .status
             .ok_or(Error::Channel("Status not found"))?)
     } else {
-        #[allow(deprecated)]
-        Ok(rpc
-            .call_typed(&ListpeersRequest {
-                id: Some(peer_id),
-                level: None,
-            })
-            .await
-            .map_err(|e| Error::Rpc(e))?
-            .peers
-            .into_iter()
-            .nth(0)
-            .ok_or(Error::Channel("Has no peers list"))?
-            .channels
-            .into_iter()
-            .nth(0)
-            .ok_or(Error::Channel("Has no channels list"))?
-            .into_iter()
-            .filter(|c| c.short_channel_id == Some(scid))
-            .nth(0)
-            .ok_or(Error::Channel("No channel with scid"))?
-            .status
-            .ok_or(Error::Channel("No amount found"))?)
+        return Err(Error::Service(format!(
+            "Not supported in this version of core-lightning: {}, need at least v23.05gl1",
+            version,
+        )));
     }
 }
 
@@ -381,7 +359,6 @@ async fn spendable_msat(
             .await
             .map_err(|e| Error::Rpc(e))?
             .channels
-            .ok_or(Error::Channel("No channels found"))?
             .into_iter()
             .filter(|c| {
                 c.short_channel_id == Some(scid)
@@ -394,28 +371,10 @@ async fn spendable_msat(
             .spendable_msat
             .ok_or(Error::Channel("No amount found"))?)
     } else {
-        #[allow(deprecated)]
-        Ok(rpc
-            .call_typed(&ListpeersRequest {
-                id: Some(peer_id),
-                level: None,
-            })
-            .await
-            .map_err(|e| Error::Rpc(e))?
-            .peers
-            .into_iter()
-            .nth(0)
-            .ok_or(Error::Channel("Has no peers list"))?
-            .channels
-            .into_iter()
-            .nth(0)
-            .ok_or(Error::Channel("Has no channels list"))?
-            .into_iter()
-            .filter(|c| c.short_channel_id == Some(scid))
-            .nth(0)
-            .ok_or(Error::Channel("No channel with scid"))?
-            .spendable_msat
-            .ok_or(Error::Channel("No amount found"))?)
+        return Err(Error::Service(format!(
+            "Not supported in this version of core-lightning: {}, need at least v23.05gl1",
+            version,
+        )));
     }
 }
 

libs/gl-plugin/src/bin/plugin.rs

@@ -21,10 +21,18 @@ async fn main() -> Result<(), Error> {
     let config = Config::new().context("loading config")?;
     let stage = Arc::new(Stage::new());
     let (events, _) = tokio::sync::broadcast::channel(16);
+    let (notifications, _) = tokio::sync::broadcast::channel(16);
     let state_store = get_signer_store().await?;
 
     start_hsm_server(config.clone(), stage.clone())?;
-    start_node_server(config, stage.clone(), events.clone(), state_store).await?;
+    start_node_server(
+        config,
+        stage.clone(),
+        events.clone(),
+        notifications,
+        state_store,
+    )
+    .await?;
 
     let plugin = gl_plugin::init(stage, events).await?;
     if let Some(plugin) = plugin.start().await? {
@@ -38,6 +46,7 @@ async fn start_node_server(
     config: Config,
     stage: Arc<Stage>,
     events: tokio::sync::broadcast::Sender<Event>,
+    notifications: tokio::sync::broadcast::Sender<cln_rpc::notifications::Notification>,
     signer_state_store: Box<dyn StateStore>,
 ) -> Result<(), Error> {
     let addr: SocketAddr = config
@@ -62,6 +71,7 @@ async fn start_node_server(
         stage.clone(),
         config.clone(),
         events.clone(),
+        notifications.clone(),
         signer_state_store,
     )
     .await?;

libs/gl-plugin/src/node/mod.rs

@@ -6,6 +6,7 @@ use crate::{stager, tramp};
 use anyhow::{Context, Error, Result};
 use base64::{engine::general_purpose, Engine as _};
 use bytes::BufMut;
+use cln_rpc::Notification;
 use gl_client::persist::State;
 use governor::{
     clock::MonotonicClock, state::direct::NotKeyed, state::InMemoryState, Quota, RateLimiter,
@@ -85,13 +86,15 @@ pub struct PluginNodeServer {
     grpc_binding: String,
     signer_state_store: Arc<Mutex<Box<dyn StateStore>>>,
     pub ctx: crate::context::Context,
+    notifications: tokio::sync::broadcast::Sender<Notification>,
 }
 
 impl PluginNodeServer {
     pub async fn new(
         stage: Arc<stager::Stage>,
         config: Config,
         events: tokio::sync::broadcast::Sender<super::Event>,
+        notifications: tokio::sync::broadcast::Sender<Notification>,
         signer_state_store: Box<dyn StateStore>,
     ) -> Result<Self, Error> {
         let tls = ServerTlsConfig::new()
@@ -126,6 +129,7 @@ impl PluginNodeServer {
             signer_state: Arc::new(Mutex::new(signer_state)),
             signer_state_store: Arc::new(Mutex::new(signer_state_store)),
             grpc_binding: config.node_grpc_binding,
+            notifications,
         };
 
         tokio::spawn(async move {

libs/gl-plugin/src/node/rpcwait.rs

@@ -1,5 +1,5 @@
-use tonic::transport::NamedService;
 use log::warn;
+use tonic::server::NamedService;
 use tower::Service;
 
 /// The RPC socket will not be available right away, so we wrap the