From d5f28d6f83542e4940af7145bdce7390ad32303e Mon Sep 17 00:00:00 2001 From: Armando Dutra Date: Mon, 6 Feb 2023 14:58:39 -0300 Subject: [PATCH] fix open channel states --- Cargo.lock | 119 ++++++++++++++-------------- Cargo.toml | 26 +++---- cli/Cargo.toml | 14 ++-- rpc/Cargo.toml | 14 ++-- src/bus/ctl.rs | 7 +- src/channeld/automata/accept.rs | 130 +++++++++++++++++++++++++++---- src/channeld/automata/mod.rs | 3 + src/channeld/automata/propose.rs | 33 ++++---- src/channeld/runtime.rs | 13 +++- src/lnpd/runtime.rs | 5 ++ src/routed/runtime.rs | 8 +- src/watchd/runtime.rs | 43 ++++++++-- src/watchd/worker.rs | 65 +++++++++++++++- 13 files changed, 343 insertions(+), 137 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3daac60..0a5e60b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -28,9 +28,9 @@ dependencies = [ [[package]] name = "amplify" -version = "3.13.0" +version = "3.14.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "116019a174e912931d5b19ca7ab6a22596d12cdb1320358fad3368f0aba135a9" +checksum = "af87bda25a5d87be1b9aedb1516eba82b4b4ad930a595916afc8bb39bf84a851" dependencies = [ "amplify_derive", "amplify_num", @@ -39,9 +39,10 @@ dependencies = [ "rand 0.8.5", "serde 1.0.139", "serde_json", - "serde_yaml 0.8.25", + "serde_yaml 0.9.16", "stringly_conversions", "toml", + "wasm-bindgen", ] [[package]] @@ -141,9 +142,9 @@ dependencies = [ [[package]] name = "bitcoin_blockchain" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d1047b4204cfc9a3e0e765794c06750e2abf0978f07d28bc2aae3f0839971a13" +checksum = "edb38270e9c10c1858bf6c939a700ea249e7dd0e8b36e3258b55ce7c9bdd2499" dependencies = [ "amplify", "chrono", @@ -162,9 +163,9 @@ dependencies = [ [[package]] name = "bitcoin_hd" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aa6d0e54e948e5a6f3c2e1fdfa343859ba5bb5b66b205ffcfa273c38c6e29064" +checksum = "aae39db6b04c44c76c50b8f8dd09799adb71ae7807a1de2efe3d127169c04160" dependencies = [ "amplify", "bitcoin", @@ -177,9 +178,9 @@ dependencies = [ [[package]] name = "bitcoin_onchain" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6c4e6a2c413c6d03bbdcdf864e8bccf51760b9a96710e93ca30491aaac00ba7c" +checksum = "5e4a6744365c862b8c74cb33a21532f91f1f32f4082361fd3aab5d22c1afd482" dependencies = [ "amplify", "bitcoin", @@ -194,9 +195,9 @@ dependencies = [ [[package]] name = "bitcoin_scripts" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b6512b98b415f5a1147c521f2989bd3cbbc77378c9f3777382835859119729a" +checksum = "5f8b08389e5391cf8311fd4de09a340ed98b4b7c2f87c956125c22341b5d14fb" dependencies = [ "amplify", "bitcoin", @@ -231,9 +232,9 @@ checksum = "8d696c370c750c948ada61c69a0ee2cbbb9c50b1019ddb86d9317157a99c2cae" [[package]] name = "bp-core" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1fdb7483fc93def8767e100a5b3370365461fb361ec9002d9652575e4920a647" +checksum = "42196466c9978cce3e24a0aba98497827755df3fdae3d55a743e8cf378cf5155" dependencies = [ "amplify", "bitcoin", @@ -246,9 +247,9 @@ dependencies = [ [[package]] name = "bp-dbc" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6acaeeb2876ef6a73d4de6172313b254ab2d04fde131e1909c082cae7e122f02" +checksum = "c3ada76cb0b56cd87810adc5df0cc577d6ba048655a0898a241b0839f45a222d" dependencies = [ "amplify", "bitcoin", @@ -260,9 +261,9 @@ dependencies = [ [[package]] name = "bp-seals" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e1bfe4757003dd3de6d459172a50a8e055619ee1f3ffb2ba1ccbeb3db4b0f112" +checksum = "73bb2e2050af42af6c0ce1cd58b5e2f3b13d5105effc5a5dd1339551c4ed11ec" dependencies = [ "amplify", "bitcoin", @@ -468,9 +469,9 @@ dependencies = [ [[package]] name = "commit_verify" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "79b115427218c894907f013305353b18fe35347d5daf9eb6b59d8df59a67bf92" +checksum = "c7f422e4e3f14f628b3cc3a241c0ea53b6457d100dda2980e2d7e719f1e41cf7" dependencies = [ "amplify", "bitcoin_hashes", @@ -748,9 +749,9 @@ dependencies = [ [[package]] name = "descriptor-wallet" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c9f75d1d7c5121865731a84abc014be3a8751f541b7ce0384e703e466507ceca" +checksum = "5d5dc9100428d2b492e0f50ce0261c856fcd80f4e998f513cdb5eed9ac5ed9d8" dependencies = [ "amplify", "bitcoin", @@ -768,9 +769,9 @@ dependencies = [ [[package]] name = "descriptors" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0ba6d96c752fe6046fc2ba9e6bc753e90efe979c7cc62a8ee7837a61392f2ad0" +checksum = "2882b380c1be2129610b934abbfbdfa8b2ba13d934c948f0922e8d33bfc78b08" dependencies = [ "amplify", "bitcoin", @@ -1035,9 +1036,9 @@ dependencies = [ [[package]] name = "inet2_addr" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3fee84136eb3deacdbdd7618e3b4cce4a3fc922103eb91c2d2811aa3a16df2d8" +checksum = "f598ff3f1d304119896e65d1d2c8e422ea09fb9d24f68582e636aacecf18a969" dependencies = [ "amplify", "lightning_encoding", @@ -1054,9 +1055,9 @@ dependencies = [ [[package]] name = "inet2_derive" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2058efb9e12c8c8b7807b5a233575062defc81953a3b91823b4a557c9bc59ea8" +checksum = "61f3d87919e38693a04e95fbc65cfce0da5a0a094fde4a40f19ca7455f602739" dependencies = [ "amplify", "proc-macro2", @@ -1084,9 +1085,9 @@ dependencies = [ [[package]] name = "internet2" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0afb751e9427eb26c527a3176570d0bba847fecfa31600695cb406c09180a1ef" +checksum = "c7b39dd3f6f8003a69a4778afae0bcc3865c3da5e3329cc045d3725145ea62e4" dependencies = [ "amplify", "bitcoin_hashes", @@ -1191,9 +1192,9 @@ dependencies = [ [[package]] name = "lightning_encoding" -version = "0.9.0-rc.1" +version = "0.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0bab704ca4f5eb4b567ecc1c2ff35dd4ddea89cc2ab9966d3aa1b49eae42c7db" +checksum = "a56f420c81ea9f113a2ccefffc55124394feddf19f5b10ebbad81edee4763d28" dependencies = [ "amplify", "bitcoin", @@ -1206,9 +1207,9 @@ dependencies = [ [[package]] name = "lightning_encoding_derive" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8816470f7f5fe634a8e370c6faa92e70e144e4996a140b9d23c1bdcb16ecfea8" +checksum = "1d0f5261aacd87fa76a12ff80b5d0e6782ce969c0301ab7cfbabe20f4492881d" dependencies = [ "amplify_syn", "encoding_derive_helpers", @@ -1261,9 +1262,9 @@ dependencies = [ [[package]] name = "lnp-core" -version = "0.9.0-rc.1" +version = "0.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "eb157e66f06e267575e1c8468e61946a959d58c9f73d46784ebc54651e6525f3" +checksum = "238ce14de0c3ceccef8a31317d06964949881c6d8e4898df3076e86bae58ad67" dependencies = [ "amplify", "bitcoin", @@ -1281,9 +1282,9 @@ dependencies = [ [[package]] name = "lnp2p" -version = "0.9.0-rc.1" +version = "0.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9f2779e6016a334f8868e7cf501caff6a0f9204cae6b1b7af3111ca8ea5eec17" +checksum = "b27ed14a42f7d90e3a16ff602b877e73a9af9a255416e92841ba839c7d138390" dependencies = [ "amplify", "bitcoin", @@ -1363,9 +1364,9 @@ dependencies = [ [[package]] name = "lnpbp" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c7d5b4a2069d6cb4c3b2d622f7f94960db5f9d9310bb63bc8ac8e6eb4036572" +checksum = "fba6ec36bc2269cf7af59b078d290ab4e985e17903b22568107ef1ef0b783d92" dependencies = [ "amplify", "lnpbp_bech32", @@ -1377,9 +1378,9 @@ dependencies = [ [[package]] name = "lnpbp_bech32" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "445c736fa2f17fa943ce5778a518d8bd6efea75b4cbe9db775d628f8fc772ad6" +checksum = "ecdbd10fe8d9d53febf413b4f8bbaa1911492597b19bfbb84a598f37062eab96" dependencies = [ "amplify", "bech32", @@ -1393,9 +1394,9 @@ dependencies = [ [[package]] name = "lnpbp_chain" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b07e48078661864f577ce6165faf9165182a2028d9b6c9f7387245da3c90b6ab" +checksum = "8cf3cde4b87fef3ba34e31e9a02fa9957d638fecccc3e12d83eaa096351fc0e0" dependencies = [ "amplify", "bitcoin", @@ -1441,9 +1442,9 @@ dependencies = [ [[package]] name = "microservices" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "33f89fc3ff96408a9309d66c2301e6a1a94d294c89a31975f83eb60d3e67ebb0" +checksum = "910f98fe7bad020ca7221beab2f4605068f9f64079f8a3c0c302373a1797c15c" dependencies = [ "amplify", "colored", @@ -1618,18 +1619,18 @@ dependencies = [ [[package]] name = "proc-macro2" -version = "1.0.40" +version = "1.0.50" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dd96a1e8ed2596c337f8eae5f24924ec83f5ad5ab21ea8e455d3566c69fbcaf7" +checksum = "6ef7d57beacfaf2d8aee5937dab7b7f28de3cb8b1828479bb5de2a7106f2bae2" dependencies = [ "unicode-ident", ] [[package]] name = "psbt" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c1c5fbee319d037ed8827d4374efabeb7dd31ef0dfc5d280c9cd8639a0f6dbf" +checksum = "e53ad01c5e6d2e9375b5594ce166a0374182562b1ec82318a7f039ed2efda759" dependencies = [ "amplify", "bitcoin", @@ -2073,18 +2074,18 @@ checksum = "f054c6c1a6e95179d6f23ed974060dcefb2d9388bb7256900badad682c499de4" [[package]] name = "single_use_seals" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cc2d580c144fbe83cbca5a2897382e2d7abfb8cf8eef8f70e493582126a96734" +checksum = "e092f1d5411486816ec67cd7d0280acb0e05a1bd509721bc3dde3b1ffa5bfa0e" dependencies = [ "amplify_derive", ] [[package]] name = "slip132" -version = "0.9.0-rc.1" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "612a082863f686f8d44ca682ce40aeb690faf0a714576f55daac0716d1085d43" +checksum = "41a2947cb179006a73896fca01015ee5255c05b8b83e74c5e9d623ed4480abe2" dependencies = [ "amplify", "bitcoin", @@ -2123,9 +2124,9 @@ checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" [[package]] name = "strict_encoding" -version = "0.9.0-rc.2" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "efe081f3a730420b6ceb5086644df4ec64c0deec54eb134adfcc351da01b9500" +checksum = "0be7060b49729cd0b9b2391114632ef64c363a4055d91de049f5555b466193bb" dependencies = [ "amplify", "bitcoin", @@ -2137,9 +2138,9 @@ dependencies = [ [[package]] name = "strict_encoding_derive" -version = "0.8.0" +version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fd6a1540913b4033307dc0c09e5497ed33d940941f926e2d5e6e676521fbe353" +checksum = "34c9cabafb397fc1144463228ad4ba57c3c670a0117505fe59b15d8c74449716" dependencies = [ "amplify_syn", "encoding_derive_helpers", @@ -2181,9 +2182,9 @@ checksum = "6bdef32e8150c2a081110b42772ffe7d7c9032b606bc226c8260fd97e0976601" [[package]] name = "syn" -version = "1.0.98" +version = "1.0.107" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c50aef8a904de4c23c788f104b7dddc7d6f79c647c7c8ce4cc8f73eb0ca773dd" +checksum = "1f4064b5b16e03ae50984a5a8ed5d4f8803e6bc1fd170a3cda91a1be4b18e3f5" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index 1076cf0..259405e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -44,15 +44,15 @@ required-features = ["server"] [dependencies] # LNP/BP crates -amplify = "3.13.0" +amplify = "3.14.1" strict_encoding = { version = "0.9.0-rc.2", features = ["miniscript"] } -bitcoin_scripts = "0.9.0-rc.1" -bitcoin_blockchain = "0.9.0-rc.1" -descriptor-wallet = { version = "0.9.0-rc.1", features = ["keygen", "miniscript", "electrum", "sign", "construct"] } -lnpbp = "0.9.0-rc.1" -lnp-core = "0.9.0-rc.1" +bitcoin_scripts = "0.9.0" +bitcoin_blockchain = "0.9.0" +descriptor-wallet = { version = "0.9.0", features = ["keygen", "miniscript", "electrum", "sign", "construct"] } +lnpbp = "0.9.0" +lnp-core = "0.9.0" lnp_rpc = { version = "0.9.0-rc.1", path = "./rpc" } -internet2 = { version = "0.9.0-rc.1", features = ["keygen"] } +internet2 = { version = "0.9.0", features = ["keygen"] } microservices = { version = "0.9.0-rc.1", default-features = false, features = ["node", "peer"] } # Bitcoin bitcoin = { version = "0.29.2", features = ["rand"] } @@ -80,14 +80,14 @@ zmq = { package = "zmq2", version = "0.5.0" } strict_encoding_test = "0.9.0-rc.1" [build-dependencies] -amplify = "3.13.0" -lnpbp = "0.9.0-rc.1" +amplify = "3.14.1" +lnpbp = "0.9.0" bitcoin = "0.29.2" lightning-invoice = "0.21.0" -lnp-core = "0.9.0-rc.1" +lnp-core = "0.9.0" lnp_rpc = { version = "0.9.0-rc.1", path = "./rpc" } -internet2 = "0.9.0-rc.1" -microservices = { version = "0.9.0-rc.1", default-features = false, features = ["peer", "shell"] } +internet2 = "0.9.0" +microservices = { version = "0.9.0", default-features = false, features = ["peer", "shell"] } clap = { version = "~3.2.23", features = ["env", "derive"] } clap_complete = "~3.2.5" log = { version = "0.4", features = ["max_level_trace", "release_max_level_debug"] } @@ -119,4 +119,4 @@ embedded = ["microservices/embedded"] tor = ["microservices/tor", "internet2/tor"] [package.metadata.configure_me] -spec = "config_spec.toml" +spec = "config_spec.toml" \ No newline at end of file diff --git a/cli/Cargo.toml b/cli/Cargo.toml index 67bce97..0a7a7b2 100644 --- a/cli/Cargo.toml +++ b/cli/Cargo.toml @@ -16,23 +16,23 @@ path = "src/main.rs" name = "lnp-cli" [dependencies] -amplify = "3.13.0" -lnp-core = { version = "0.9.0-rc.1", default-features = false } +amplify = "3.14.1" +lnp-core = { version = "0.9.0", default-features = false } lnp_rpc = { version = "0.9.0-rc.1", path = "../rpc" } lightning-invoice = { version = "0.21.0", optional = true } -internet2 = "0.9.0-rc.1" -microservices = { version = "0.9.0-rc.1", default-features = false, features = ["cli"] } +internet2 = "0.9.0" +microservices = { version = "0.9.0", default-features = false, features = ["cli"] } shellexpand = "2.1" clap = { version = "~3.2.23", features = ["derive", "env"] } log = "0.4.14" [build-dependencies] -amplify = "3.13.0" +amplify = "3.14.1" clap = { version = "~3.2.23", features = ["derive", "env"] } clap_complete = "~3.2.5" lightning-invoice = "0.21.0" -internet2 = "0.9.0-rc.1" -lnp-core = { version = "0.9.0-rc.1", default-features = false } +internet2 = "0.9.0" +lnp-core = { version = "0.9.0", default-features = false } lnp_rpc = { version = "0.9.0-rc.1", path = "../rpc" } configure_me_codegen = "0.4" diff --git a/rpc/Cargo.toml b/rpc/Cargo.toml index 70cdd3b..d7635a9 100644 --- a/rpc/Cargo.toml +++ b/rpc/Cargo.toml @@ -14,16 +14,16 @@ readme = "../README.md" name = "lnp_rpc" [dependencies] -amplify = "3.13.0" -strict_encoding = "0.9.0-rc.2" -bitcoin_scripts = "0.9.0-rc.1" -lnp-core = { version = "0.9.0-rc.1", default-features = false } -lnpbp = "0.9.0-rc.1" +amplify = "3.14.1" +strict_encoding = "0.9.0" +bitcoin_scripts = "0.9.0" +lnp-core = { version = "0.9.0", default-features = false } +lnpbp = "0.9.0" bitcoin = { version = "0.29.2", features = ["rand"] } lightning-invoice = { version = "0.21.0", optional = true } -internet2 = "0.9.0-rc.1" +internet2 = "0.9.0" microservices = { version = "0.9.0-rc.1", default-features = false, features = ["client"] } -descriptor-wallet = "0.9.0-rc.1" +descriptor-wallet = "0.9.0" serde_crate = { package = "serde", version = "1", features = ["derive"], optional = true } serde_with = { version = "1.14", optional = true } serde_yaml = { version = "0.8", optional = true } diff --git a/src/bus/ctl.rs b/src/bus/ctl.rs index abb7440..066e3cc 100644 --- a/src/bus/ctl.rs +++ b/src/bus/ctl.rs @@ -18,7 +18,7 @@ use bitcoin_scripts::PubkeyScript; use internet2::addr::{NodeAddr, NodeId}; use internet2::presentation::sphinx::Hop; use lnp::channel::bolt::{CommonParams, LocalKeyset, PeerParams, Policy}; -use lnp::p2p::bolt::{ChannelId, OpenChannel, PaymentOnion}; +use lnp::p2p::bolt::{ChannelId, OpenChannel, PaymentOnion, TempChannelId}; use lnp::router::gossip::LocalChannelInfo; use lnp_rpc::{ChannelInfo, Failure, PeerInfo}; use microservices::esb::ClientId; @@ -141,6 +141,11 @@ pub enum CtlMsg { #[display("channel_info({0})", alt = "{0:#}")] ChannelInfo(ChannelInfo), + + // Channel tasks + // ----------------- + #[display("channel_update({old_id}, {new_id})")] + ChannelUpdate { old_id: TempChannelId, new_id: ChannelId }, } impl CtlMsg { diff --git a/src/channeld/automata/accept.rs b/src/channeld/automata/accept.rs index 9a99c6d..0ff70f6 100644 --- a/src/channeld/automata/accept.rs +++ b/src/channeld/automata/accept.rs @@ -12,15 +12,17 @@ // If not, see . use lnp::channel::bolt::Lifecycle; -use lnp::p2p::bolt::{ActiveChannelId, Messages}; +use lnp::p2p::bolt::{ActiveChannelId, ChannelId, FundingSigned, Messages as LnMsg}; use lnp::Extension; +use lnp_rpc::ServiceId; use microservices::cli::LogStyle; +use microservices::esb::Handler; use super::Error; use crate::automata::{Event, StateMachine}; -use crate::bus::{AcceptChannelFrom, BusMsg}; +use crate::bus::{AcceptChannelFrom, BusMsg, CtlMsg}; use crate::channeld::runtime::Runtime; -use crate::Endpoints; +use crate::{Endpoints, Responder}; /// Channel proposal workflow #[derive(Copy, Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Debug, Display)] @@ -59,7 +61,7 @@ impl StateMachine for ChannelAccept { ChannelAccept::Funded => finish_funded(event, runtime), ChannelAccept::Locked => { finish_locked(event, runtime)?; - info!("ChannelPropose {} has completed its work", channel_id); + info!("ChannelAccept {} has completed its work", channel_id); return Ok(None); } }?; @@ -85,15 +87,18 @@ impl ChannelAccept { impl ChannelAccept { /// Constructs channel acceptance state machine pub fn with( - endpoints: &mut Endpoints, - accept_channel_from: AcceptChannelFrom, runtime: &mut Runtime, + endpoints: &mut Endpoints, + request: AcceptChannelFrom, ) -> Result { - let open_channel = Messages::OpenChannel(accept_channel_from.channel_req); + let open_channel = LnMsg::OpenChannel(request.channel_req.clone()); runtime.state.channel.update_from_peer(&open_channel)?; - runtime.send_p2p(endpoints, open_channel)?; - + let _ = runtime.send_ctl( + endpoints, + ServiceId::Signer, + CtlMsg::DeriveKeyset(request.channel_req.temporary_channel_id.into()), + ); Ok(ChannelAccept::Accepted) } @@ -105,21 +110,112 @@ impl ChannelAccept { "Accepted".ended(), channel_id.actor(), ), - _ => todo!(), + ChannelAccept::Signed => { + format!("{} channel {:#} from a remote peer", "Signed".ended(), channel_id.actor(),) + } + ChannelAccept::Funded => { + format!("{} channel {:#} from a remote peer", "Funded".ended(), channel_id.actor(),) + } + ChannelAccept::Locked => { + format!("{} channel {:#} from a remote peer", "Locked".ended(), channel_id.actor(),) + } } } } -fn finish_accepted(_event: Event, _runtime: &mut Runtime) -> Result { - todo!() +fn finish_accepted(event: Event, runtime: &mut Runtime) -> Result { + let accept_event = match event.message { + BusMsg::Ctl(CtlMsg::Keyset(_, keys)) => { + runtime.state.channel.constructor_mut().set_local_keys(keys); + + let accept_channel = runtime.state.channel.compose_accept_channel()?; + let accept_channel = LnMsg::AcceptChannel(accept_channel); + + trace!("Notifying remote peer about channel creation"); + runtime.send_p2p(event.endpoints, accept_channel)?; + Ok(ChannelAccept::Signed) + } + wrong_msg => { + return Err(Error::UnexpectedMessage(wrong_msg, Lifecycle::Accepted, event.source)) + } + }; + accept_event } -fn finish_signed(_event: Event, _runtime: &mut Runtime) -> Result { - todo!() +fn finish_signed(event: Event, runtime: &mut Runtime) -> Result { + let signed_event = match event.message { + BusMsg::Bolt(LnMsg::FundingCreated(funding)) => { + let old_id = runtime + .state + .channel + .temp_channel_id() + .expect("temporary channel id always known at this stage"); + + let channel_id = ChannelId::with(funding.funding_txid, funding.funding_output_index); + debug!("Changing channel id from {} to {}", runtime.identity(), channel_id); + runtime + .set_identity(event.endpoints, channel_id) + .expect("unable to change ZMQ channel identity"); + runtime.state.channel.update_from_peer(&LnMsg::FundingCreated(funding.clone()))?; + + runtime.send_ctl(event.endpoints, ServiceId::LnpBroker, CtlMsg::ChannelUpdate { + old_id, + new_id: channel_id, + })?; + + runtime.send_p2p( + event.endpoints, + LnMsg::FundingSigned(FundingSigned { channel_id, signature: funding.signature }), + )?; + Ok(ChannelAccept::Funded) + } + wrong_msg => { + return Err(Error::UnexpectedMessage(wrong_msg, Lifecycle::Signed, event.source)) + } + }; + + signed_event } -fn finish_funded(_event: Event, _runtime: &mut Runtime) -> Result { - todo!() +fn finish_funded(event: Event, runtime: &mut Runtime) -> Result { + let funded_event = match event.message { + BusMsg::Bolt(LnMsg::FundingLocked(funding)) => { + // Save next per commitment point + runtime.state.channel.update_from_peer(&LnMsg::FundingLocked(funding.clone()))?; + trace!("Notifying runtime about channel creation"); + let _ = runtime.send_ctl( + event.endpoints, + ServiceId::Router, + CtlMsg::ChannelCreated( + runtime.state.channel.channel_info(runtime.state.remote_id()), + ), + ); + + // TODO: find the alternative to this. The hello is calling to force running + // finish_locked method + runtime.send_ctl(event.endpoints, ServiceId::Router, CtlMsg::Hello)?; + + Ok(ChannelAccept::Locked) + } + wrong_msg => { + return Err(Error::UnexpectedMessage(wrong_msg, Lifecycle::Locked, event.source)) + } + }; + funded_event } -fn finish_locked(_event: Event, _runtime: &mut Runtime) -> Result<(), Error> { todo!() } +fn finish_locked(event: Event, runtime: &mut Runtime) -> Result<(), Error> { + let locked_event = match event.message { + BusMsg::Ctl(CtlMsg::Hello) => { + debug!("Funding transaction mined, notifying remote peer"); + let funding_locked = runtime.state.channel.compose_funding_locked(); + runtime.send_p2p(event.endpoints, LnMsg::FundingLocked(funding_locked))?; + + Ok(()) + } + wrong_msg => { + return Err(Error::UnexpectedMessage(wrong_msg, Lifecycle::Active, event.source)) + } + }; + locked_event +} diff --git a/src/channeld/automata/mod.rs b/src/channeld/automata/mod.rs index f62eb7f..cb57ba1 100644 --- a/src/channeld/automata/mod.rs +++ b/src/channeld/automata/mod.rs @@ -295,6 +295,9 @@ impl Runtime { BusMsg::Ctl(CtlMsg::OpenChannelWith(open_channel_with)) => { ChannelPropose::with(self, endpoints, open_channel_with)?.into() } + BusMsg::Ctl(CtlMsg::AcceptChannelFrom(accept_channel_from)) => { + ChannelAccept::with(self, endpoints, accept_channel_from)?.into() + } wrong_msg => { return Err(Error::UnexpectedMessage(wrong_msg, Lifecycle::Initial, source)) } diff --git a/src/channeld/automata/propose.rs b/src/channeld/automata/propose.rs index 05708ba..b8b5d7e 100644 --- a/src/channeld/automata/propose.rs +++ b/src/channeld/automata/propose.rs @@ -256,7 +256,9 @@ fn complete_signing( let channel_id = ChannelId::with(funding_txid, funding_output_index); debug!("Changing channel id from {} to {}", runtime.identity(), channel_id); - runtime.set_identity(event.endpoints, channel_id).expect("unrecoverable ZMQ failure"); + runtime + .set_identity(event.endpoints, channel_id) + .expect("unable to change ZMQ channel identity"); // needed to update ESB routing map runtime.send_ctl(event.endpoints, ServiceId::LnpBroker, CtlMsg::Hello)?; @@ -282,8 +284,7 @@ fn complete_funding( let txid = runtime.state.channel.funding().txid(); debug!("Waiting for funding transaction {} to be mined", txid); - // TODO: Uncomment once watching daemon will be running - // runtime.send_ctl(&mut event.endpoints, ServiceId::Watch, CtlMsg::Track(txid))?; + runtime.send_ctl(event.endpoints, ServiceId::Watch, CtlMsg::Track { txid, depth: 0 })?; Ok(ChannelPropose::Published) } @@ -292,22 +293,16 @@ fn complete_published( event: Event, runtime: &mut Runtime, ) -> Result, automata::Error> { - if !matches!( - event.message, - BusMsg::Ctl(CtlMsg::TxFound(_)) | BusMsg::Bolt(LnMsg::FundingLocked(_)) - ) { - return Err(Error::UnexpectedMessage(event.message, Lifecycle::Signed, event.source)); - } - - debug!("Funding transaction mined, notifying remote peer"); - let funding_locked = runtime.state.channel.compose_funding_locked(); - runtime.send_p2p(event.endpoints, LnMsg::FundingLocked(funding_locked))?; - - if let BusMsg::Bolt(LnMsg::FundingLocked(_)) = event.message { - complete_locked(event, runtime)?; - } - - Ok(Some(ChannelPropose::Locked)) + let published_event = match event.message { + BusMsg::Ctl(CtlMsg::TxFound(_)) => { + debug!("Funding transaction mined, notifying remote peer"); + let funding_locked = runtime.state.channel.compose_funding_locked(); + runtime.send_p2p(event.endpoints, LnMsg::FundingLocked(funding_locked))?; + Ok(Some(ChannelPropose::Locked)) + } + wrong_msg => Err(Error::UnexpectedMessage(wrong_msg, Lifecycle::Funded, event.source)), + }; + published_event } fn complete_locked(event: Event, runtime: &mut Runtime) -> Result<(), automata::Error> { diff --git a/src/channeld/runtime.rs b/src/channeld/runtime.rs index ffe07e5..e85a5ea 100644 --- a/src/channeld/runtime.rs +++ b/src/channeld/runtime.rs @@ -156,6 +156,11 @@ impl Runtime { endpoints.set_identity(ServiceBus::Rpc, identity.clone())?; self.identity = identity; + // TODO: Remove after fix update identity + // the set_identity method causes the lose reference between services, the thread::sleep is + // a workaround to that. + std::thread::sleep(core::time::Duration::from_secs(2)); + fs::remove_file(self.config.channel_file(prev_id))?; Ok(()) @@ -197,11 +202,11 @@ impl Runtime { LnMsg::ChannelReestablish(_) | LnMsg::AcceptChannel(_) + | LnMsg::FundingCreated(_) | LnMsg::FundingSigned(_) | LnMsg::FundingLocked(_) => { self.process(endpoints, ServiceId::PeerBolt(remote_id), BusMsg::Bolt(message))?; } - _ => { // Ignore the rest of LN peer messages } @@ -230,7 +235,7 @@ impl Runtime { // Processing remote request to open a channel CtlMsg::AcceptChannelFrom(bus::AcceptChannelFrom { remote_id, .. }) => { self.enquirer = None; - let remote_id = remote_id.clone(); + self.state.remote_id = Some(remote_id); if self.process(endpoints, source, BusMsg::Ctl(request))? { // Updating state only if the request was processed self.state.remote_id = Some(remote_id); @@ -240,8 +245,10 @@ impl Runtime { CtlMsg::FundingConstructed(_) | CtlMsg::TxFound(_) | CtlMsg::Signed(_) + | CtlMsg::Keyset(..) | CtlMsg::Error { .. } - | CtlMsg::EsbError { .. } => { + | CtlMsg::EsbError { .. } + | CtlMsg::Hello => { self.process(endpoints, source, BusMsg::Ctl(request))?; } diff --git a/src/lnpd/runtime.rs b/src/lnpd/runtime.rs index b797150..361335e 100644 --- a/src/lnpd/runtime.rs +++ b/src/lnpd/runtime.rs @@ -463,6 +463,11 @@ impl Runtime { } } + CtlMsg::ChannelUpdate { old_id, new_id } => { + self.update_chanel_id(old_id.to_owned(), new_id.to_owned()); + return Ok(()); + } + wrong_msg => { error!("Request is not supported by the CTL interface"); return Err(Error::wrong_esb_msg(ServiceBus::Ctl, wrong_msg)); diff --git a/src/routed/runtime.rs b/src/routed/runtime.rs index 14f1d79..ee1b699 100644 --- a/src/routed/runtime.rs +++ b/src/routed/runtime.rs @@ -130,11 +130,15 @@ impl Runtime { fn handle_ctl( &mut self, - _: &mut Endpoints, - _: ServiceId, + endpoints: &mut Endpoints, + source: ServiceId, message: CtlMsg, ) -> Result<(), Error> { match message { + CtlMsg::Hello => { + self.send_ctl(endpoints, source, message)?; + } + CtlMsg::ChannelCreated(channel_info) => { debug!("Adding local channel {} to the routing table", channel_info.channel_id); self.router.update_from_local(&UpdateMsg::DirectChannelAdd(channel_info))?; diff --git a/src/watchd/runtime.rs b/src/watchd/runtime.rs index bdc4acc..c0798d3 100644 --- a/src/watchd/runtime.rs +++ b/src/watchd/runtime.rs @@ -76,7 +76,7 @@ impl WatcherRuntime { pub(self) fn send_over_bridge(&mut self, req: BusMsg) -> Result<(), Error> { debug!("Forwarding electrum update message over BRIDGE interface to the runtime"); - self.bridge.send_to(ServiceBus::Bridge, self.identity.clone(), req)?; + self.bridge.send_to(ServiceBus::Bridge, ServiceId::Watch, req)?; Ok(()) } @@ -85,8 +85,28 @@ impl WatcherRuntime { let msg = self.receiver.recv()?; debug!("Processing message {}", msg); trace!("Message details: {:?}", msg); - // TODO: Forward electrum notifications over the bridge + // TODO: Forward all electrum notifications over the bridge // self.send_over_bridge(msg.into()).expect("watcher bridge is halted"); + match msg { + ElectrumUpdate::TxBatch(transactions, _) => { + for transaction in transactions { + self.send_over_bridge(BusMsg::Ctl(CtlMsg::TxFound(crate::bus::TxStatus { + txid: transaction.txid(), + block_pos: None, + }))) + .expect("unable forward electrum notifications over the bridge"); + } + } + ElectrumUpdate::Connecting + | ElectrumUpdate::Connected + | ElectrumUpdate::Complete + | ElectrumUpdate::FeeEstimate(..) + | ElectrumUpdate::LastBlock(_) + | ElectrumUpdate::LastBlockUpdate(_) + | ElectrumUpdate::ChannelDisconnected + | ElectrumUpdate::Error(_) => { /* nothing to do here */ } + } + Ok(()) } } @@ -153,6 +173,10 @@ impl Runtime { if *required_height >= tx_status.block_pos.map(|b| b.pos).unwrap_or_default() { let service_id = service_id.clone(); self.untrack(tx_status.txid); + match self.electrum_worker.untrack_transaction(tx_status.txid) { + Ok(_) => debug!("Untracking tx {}", tx_status.txid), + _ => error!("Unable untrack transaction in electrum worker"), + } endpoints.send_to( ServiceBus::Ctl, ServiceId::Watch, @@ -194,14 +218,19 @@ impl Runtime { ) -> Result<(), Error> { match message { CtlMsg::Track { txid, depth } => { - debug!("Tracking status for tx {}", txid); + debug!("Tracking status for tx {txid}"); self.track_list.insert(txid, (depth, source)); - // TODO: Request worker + match self.electrum_worker.track_transaction(txid) { + Ok(_) => debug!("Tracking status for tx {txid}"), + _ => error!("Unable track transaction in electrum worker"), + } } - CtlMsg::Untrack(txid) => { - // TODO: Request worker self.untrack(txid); + match self.electrum_worker.untrack_transaction(txid) { + Ok(_) => debug!("Untracking tx {txid}"), + _ => error!("Unable untrack transaction in electrum worker"), + } } wrong_msg => { @@ -214,7 +243,7 @@ impl Runtime { } fn untrack(&mut self, txid: Txid) { - debug!("Stopping tracking tx {}", txid); + debug!("Stopping tracking tx {txid}"); if self.track_list.remove(&txid).is_none() { warn!("Transaction {} was not tracked before", txid); } diff --git a/src/watchd/worker.rs b/src/watchd/worker.rs index ac44b38..2fea972 100644 --- a/src/watchd/worker.rs +++ b/src/watchd/worker.rs @@ -17,7 +17,7 @@ use std::sync::mpsc; use std::thread::{self, JoinHandle}; use std::time::Duration; -use bitcoin::Transaction; +use bitcoin::{Transaction, Txid}; use electrum_client::{Client as ElectrumClient, ElectrumApi, HeaderNotification}; #[derive(Copy, Clone, Ord, PartialOrd, Eq, PartialEq, Hash, Debug, Display, Error, From)] @@ -70,7 +70,7 @@ impl ElectrumWorker { let client = connect_electrum(electrum_url)?; let (tx, rx) = mpsc::channel::(); - let processor = ElectrumProcessor { client, sender, rx }; + let processor = ElectrumProcessor::with(client, sender, rx)?; let worker_thread = thread::Builder::new() .name(s!("electrum_watcher")) .spawn(move || processor.run()) @@ -81,6 +81,7 @@ impl ElectrumWorker { .name(s!("electrum_pacemaker")) .spawn(move || loop { thread::sleep(Duration::from_secs(interval)); + sender.send(ElectrumCmd::GetTrasactions).expect("Electrum thread is dead"); sender.send(ElectrumCmd::PopHeader).expect("Electrum thread is dead") }) .expect("unable to start blockchain watcher pacemaker thread"); @@ -96,6 +97,16 @@ impl ElectrumWorker { pub fn reconnect(&self, electrum_url: String) -> Result<(), WatcherChannelFailure> { self.cmd(ElectrumCmd::Reconnect(electrum_url)) } + + #[inline] + pub fn track_transaction(&self, txid: Txid) -> Result<(), WatcherChannelFailure> { + self.cmd(ElectrumCmd::TrackTransaction(txid)) + } + + #[inline] + pub fn untrack_transaction(&self, txid: Txid) -> Result<(), WatcherChannelFailure> { + self.cmd(ElectrumCmd::UntrackTransaction(txid)) + } } fn connect_electrum(electrum_url: &str) -> Result { @@ -104,18 +115,31 @@ fn connect_electrum(electrum_url: &str) -> Result, rx: mpsc::Receiver, + tracks: Vec, } impl ElectrumProcessor { + pub fn with( + client: ElectrumClient, + sender: mpsc::Sender, + rx: mpsc::Receiver, + ) -> Result { + Ok(ElectrumProcessor { client, sender, rx, tracks: vec![] }) + } + pub fn run(mut self) { loop { match self.rx.recv() { @@ -133,6 +157,12 @@ impl ElectrumProcessor { let resp = match cmd { ElectrumCmd::Reconnect(electrum_url) => self.reconnect(&electrum_url).map(|_| None), ElectrumCmd::PopHeader => self.pop_header(), + ElectrumCmd::GetTrasactions => { + let txs = &self.tracks.clone(); + self.get_transactions(txs) + } + ElectrumCmd::TrackTransaction(txid) => self.track_transaction(txid), + ElectrumCmd::UntrackTransaction(txid) => self.untrack_transaction(txid), }; match resp { Ok(Some(msg)) => { @@ -155,4 +185,35 @@ impl ElectrumProcessor { fn pop_header(&self) -> Result, electrum_client::Error> { self.client.block_headers_pop().map(|res| res.map(ElectrumUpdate::LastBlockUpdate)) } + + fn get_transactions( + &mut self, + txids: &Vec, + ) -> Result, electrum_client::Error> { + if self.tracks.is_empty() { + return Ok(None); + } + self.client.batch_transaction_get(txids).map(|res| Some(ElectrumUpdate::TxBatch(res, 0.0))) + } + + fn track_transaction( + &mut self, + txid: Txid, + ) -> Result, electrum_client::Error> { + self.tracks.push(txid); + self.client + .transaction_get(&txid.clone()) + .map(|res| Some(ElectrumUpdate::TxBatch([res].to_vec(), 0.0))) + } + + fn untrack_transaction( + &mut self, + txid: Txid, + ) -> Result, electrum_client::Error> { + let index = self.tracks.iter().position(|x| *x == txid).unwrap(); + self.tracks.remove(index); + self.client + .transaction_get(&txid.clone()) + .map(|res| Some(ElectrumUpdate::TxBatch([res].to_vec(), 0.0))) + } }