From 7f8e125a8a7e2d1759a8a475d44398ecd77ce8e7 Mon Sep 17 00:00:00 2001 From: bruwbird Date: Wed, 12 Jun 2024 18:49:10 +0900 Subject: [PATCH] cln+lnd: liveness check add a plugin hook in the case of core lightning and an interceptor in the case of lnd, and pre-install a dead/alive monitor for each daemon. --- clightning/clightning.go | 73 +++++++++++++++++++++++++++++++ cmd/peerswaplnd/peerswapd/main.go | 19 +++++++- electrum/client.go | 4 ++ electrum/electrum.go | 1 + electrum/mock/electrum.go | 14 ++++++ go.mod | 2 +- go.sum | 2 - lwk/lwkwallet.go | 11 +++++ wallet/elementsrpcwallet.go | 5 +++ wallet/wallet.go | 1 + 10 files changed, 128 insertions(+), 4 deletions(-) diff --git a/clightning/clightning.go b/clightning/clightning.go index ea0d9ec2..be05a291 100644 --- a/clightning/clightning.go +++ b/clightning/clightning.go @@ -4,6 +4,7 @@ import ( "context" "crypto/rand" "encoding/hex" + "encoding/json" "errors" "fmt" log2 "log" @@ -131,6 +132,7 @@ func NewClightningClient(ctx context.Context) (*ClightningClient, <-chan interfa cl.Plugin = glightning.NewPlugin(cl.onInit) err := cl.Plugin.RegisterHooks(&glightning.Hooks{ CustomMsgReceived: cl.OnCustomMsg, + RpcCommand: cl.OnRPCCommand, }) if err != nil { return nil, nil, err @@ -408,6 +410,77 @@ func (cl *ClightningClient) OnCustomMsg(event *glightning.CustomMsgReceivedEvent return event.Continue(), nil } +type Message struct { + Message string `json:"message"` +} + +func (cl *ClightningClient) OnRPCCommand(event *glightning.RpcCommandEvent) (*glightning.RpcCommandResponse, error) { + if cl.gbitcoin != nil { + ok, err := cl.gbitcoin.Ping() + if err != nil || !ok { + return event.ReturnResult(&Message{Message: "bitcoin daemon unavailable"}) + } + } + + if cl.liquidWallet != nil { + ok, err := cl.liquidWallet.Ping() + if err != nil || !ok { + return event.ReturnError("liquid unavailable", -1) + } + } + return event.Continue(), nil +} + +type LiquidBackendUnavailableResponse struct { + Result struct { + Message struct { + Liquid string `json:"liquid"` + } `json:"message"` + } `json:"result"` +} + +func ReturnMessage(message string) *glightning.RpcCommandResponse { + // type Resp struct { + // Message string `json:"message"` + // } + // result := &struct { + // Result Resp `json:"result"` + // }{ + // Result: Resp{ + // Message: message, + // }, + // } + type ResultMessage struct { + Result struct { + Message struct { + Liquid string `json:"liquid"` + } `json:"message"` + } `json:"result"` + } + + var resultMessage = ResultMessage{ + Result: struct { + Message struct { + Liquid string `json:"liquid"` + } `json:"message"` + }{ + Message: struct { + Liquid string `json:"liquid"` + }{ + Liquid: "liquid backend is not available", + }, + }, + } + marshaled, err := json.Marshal(resultMessage) + if err != nil { + log.Debugf("could not marshal message: %v", err) + } + log.Debugf("returning message: %s", marshaled) + return &glightning.RpcCommandResponse{ + ReturnObj: marshaled, + } +} + // AddMessageHandler adds a listener for incoming peermessages func (cl *ClightningClient) AddMessageHandler(f func(peerId string, msgType string, payload []byte) error) { cl.msgHandlers = append(cl.msgHandlers, f) diff --git a/cmd/peerswaplnd/peerswapd/main.go b/cmd/peerswaplnd/peerswapd/main.go index 04554373..018901b5 100644 --- a/cmd/peerswaplnd/peerswapd/main.go +++ b/cmd/peerswaplnd/peerswapd/main.go @@ -397,7 +397,9 @@ func run() error { } defer lis.Close() - grpcSrv := grpc.NewServer() + grpcSrv := grpc.NewServer( + grpc.UnaryInterceptor(livenessCheckInterceptor(liquidRpcWallet)), + ) peerswaprpc.RegisterPeerSwapServer(grpcSrv, peerswaprpcServer) @@ -439,6 +441,21 @@ func run() error { return nil } +func livenessCheckInterceptor(liquidWallet wallet.Wallet) grpc.UnaryServerInterceptor { + return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { + if liquidWallet != nil { + ok, err := liquidWallet.Ping() + if err != nil { + return nil, fmt.Errorf("liquid backend not reachable: %v", err) + } + if !ok { + return nil, errors.New("liquid backend not reachable") + } + } + return handler(ctx, req) + } +} + func getBitcoinChain(ctx context.Context, li lnrpc.LightningClient) (*chaincfg.Params, error) { gi, err := li.GetInfo(ctx, &lnrpc.GetInfoRequest{}) if err != nil { diff --git a/electrum/client.go b/electrum/client.go index fac23ed9..6e35557e 100644 --- a/electrum/client.go +++ b/electrum/client.go @@ -84,3 +84,7 @@ func (c *electrumClient) GetFee(ctx context.Context, target uint32) (float32, er } return c.client.GetFee(ctx, target) } + +func (c *electrumClient) Ping(ctx context.Context) error { + return c.reconnect(ctx) +} diff --git a/electrum/electrum.go b/electrum/electrum.go index 43dd0c60..332f7208 100644 --- a/electrum/electrum.go +++ b/electrum/electrum.go @@ -12,4 +12,5 @@ type RPC interface { GetRawTransaction(ctx context.Context, txHash string) (string, error) BroadcastTransaction(ctx context.Context, rawTx string) (string, error) GetFee(ctx context.Context, target uint32) (float32, error) + Ping(ctx context.Context) error } diff --git a/electrum/mock/electrum.go b/electrum/mock/electrum.go index b6ff9a0a..44313eea 100644 --- a/electrum/mock/electrum.go +++ b/electrum/mock/electrum.go @@ -100,6 +100,20 @@ func (mr *MockRPCMockRecorder) GetRawTransaction(ctx, txHash any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetRawTransaction", reflect.TypeOf((*MockRPC)(nil).GetRawTransaction), ctx, txHash) } +// Ping mocks base method. +func (m *MockRPC) Ping(ctx context.Context) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Ping", ctx) + ret0, _ := ret[0].(error) + return ret0 +} + +// Ping indicates an expected call of Ping. +func (mr *MockRPCMockRecorder) Ping(ctx any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Ping", reflect.TypeOf((*MockRPC)(nil).Ping), ctx) +} + // SubscribeHeaders mocks base method. func (m *MockRPC) SubscribeHeaders(ctx context.Context) (<-chan *electrum.SubscribeHeadersResult, error) { m.ctrl.T.Helper() diff --git a/go.mod b/go.mod index 8c59c00c..90540f96 100644 --- a/go.mod +++ b/go.mod @@ -38,7 +38,7 @@ require ( github.com/btcsuite/go-socks v0.0.0-20170105172521-4720035b7bfd // indirect github.com/btcsuite/websocket v0.0.0-20150119174127-31079b680792 // indirect github.com/btcsuite/winsvc v1.0.0 // indirect - github.com/cenkalti/backoff/v4 v4.3.0 // indirect + github.com/cenkalti/backoff/v4 v4.3.0 github.com/cespare/xxhash/v2 v2.1.2 // indirect github.com/coreos/go-semver v0.3.0 // indirect github.com/coreos/go-systemd/v22 v22.4.0 // indirect diff --git a/go.sum b/go.sum index c12ba272..f8289c28 100644 --- a/go.sum +++ b/go.sum @@ -121,8 +121,6 @@ github.com/btcsuite/websocket v0.0.0-20150119174127-31079b680792/go.mod h1:ghJtE github.com/btcsuite/winsvc v1.0.0 h1:J9B4L7e3oqhXOcm+2IuNApwzQec85lE+QaikUcCs+dk= github.com/btcsuite/winsvc v1.0.0/go.mod h1:jsenWakMcC0zFBFurPLEAyrnc/teJEM1O46fmI40EZs= github.com/cenkalti/backoff/v4 v4.1.1/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw= -github.com/cenkalti/backoff/v4 v4.1.3 h1:cFAlzYUlVYDysBEH2T5hyJZMh3+5+WCBvSnK6Q8UtC4= -github.com/cenkalti/backoff/v4 v4.1.3/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw= github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= diff --git a/lwk/lwkwallet.go b/lwk/lwkwallet.go index e84afd38..6009ae13 100644 --- a/lwk/lwkwallet.go +++ b/lwk/lwkwallet.go @@ -280,3 +280,14 @@ func (r *LWKRpcWallet) SetLabel(txID, address, label string) error { // TODO: call set label return nil } + +func (r *LWKRpcWallet) Ping() (bool, error) { + ctx, cancel := context.WithTimeout(context.Background(), defaultContextTimeout) + defer cancel() + _, err := r.lwkClient.version(ctx) + if err != nil { + return false, errors.New("lwk connection failed: " + err.Error()) + } + err = r.electrumClient.Ping(ctx) + return err == nil, err +} diff --git a/wallet/elementsrpcwallet.go b/wallet/elementsrpcwallet.go index 06bf11ee..6f36cf41 100644 --- a/wallet/elementsrpcwallet.go +++ b/wallet/elementsrpcwallet.go @@ -31,6 +31,7 @@ type RpcClient interface { SendRawTx(txHex string) (string, error) EstimateFee(blocks uint32, mode string) (*gelements.FeeResponse, error) SetLabel(address, label string) error + Ping() (bool, error) } // ElementsRpcWallet uses the elementsd rpc wallet @@ -191,3 +192,7 @@ func satsToAmountString(sats uint64) string { bitcoinAmt := float64(sats) / 100000000 return fmt.Sprintf("%f", bitcoinAmt) } + +func (r *ElementsRpcWallet) Ping() (bool, error) { + return r.rpcClient.Ping() +} diff --git a/wallet/wallet.go b/wallet/wallet.go index b0a016b3..9c2ed177 100644 --- a/wallet/wallet.go +++ b/wallet/wallet.go @@ -22,4 +22,5 @@ type Wallet interface { SendRawTx(rawTx string) (txid string, err error) GetFee(txSize int64) (uint64, error) SetLabel(txID, address, label string) error + Ping() (bool, error) }