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.
ElementsProject · Jun 12, 2024 · b09f793 · b09f793
1 parent 3d45569
commit b09f793
Show file tree

Hide file tree

Showing 11 changed files with 126 additions and 4 deletions.
diff --git a/clightning/clightning.go b/clightning/clightning.go
@@ -131,6 +131,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 +409,23 @@ func (cl *ClightningClient) OnCustomMsg(event *glightning.CustomMsgReceivedEvent
 	return event.Continue(), nil
 }
 
+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.ReplaceWith(&Available{}), nil
+		}
+	}
+	if cl.liquidWallet != nil {
+		ok, err := cl.liquidWallet.Ping()
+		if err != nil || !ok {
+			return event.ReplaceWith(&Available{}), nil
+		}
+	}
+	return event.Continue(), nil
+}
+
 // 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/clightning/clightning_commands.go b/clightning/clightning_commands.go
@@ -1109,6 +1109,59 @@ func (c ListConfig) LongDescription() string {
 	return c.Description()
 }
 
+type Available struct {
+	cl *ClightningClient
+}
+
+func (c *Available) Name() string {
+	return "peerswap-available"
+}
+
+func (c *Available) New() interface{} {
+	return &Available{
+		cl: c.cl,
+	}
+}
+
+func (c *Available) Call() (jrpc2.Result, error) {
+	if !c.cl.isReady {
+		return nil, ErrWaitingForReady
+	}
+	if c.cl.gbitcoin != nil {
+		ok, err := c.cl.gbitcoin.Ping()
+		if err != nil {
+			return nil, errors.New("bitcoin daemon not reachable")
+		}
+		if !ok {
+			return nil, errors.New("bitcoin daemon not reachable")
+		}
+	}
+	if c.cl.liquidWallet != nil {
+		ok, err := c.cl.liquidWallet.Ping()
+		if err != nil {
+			return nil, errors.New("bitcoin daemon not reachable")
+		}
+		if !ok {
+			return nil, errors.New("bitcoin daemon not reachable")
+		}
+	}
+	return nil, nil
+}
+
+func (c *Available) Get(client *ClightningClient) jrpc2.ServerMethod {
+	return &Available{
+		cl: client,
+	}
+}
+
+func (c Available) Description() string {
+	return "Returns if the peerswap plugin is available"
+}
+
+func (c Available) LongDescription() string {
+	return c.Description()
+}
+
 type PeerSwapPeerChannel struct {
 	ChannelId     string `json:"short_channel_id"`
 	LocalBalance  uint64 `json:"local_balance"`

diff --git 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 rpc not reachable: %v", err)
+			}
+			if !ok {
+				return nil, errors.New("liquid rpc 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
@@ -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
@@ -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
diff --git 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
@@ -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
@@ -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
@@ -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
@@ -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)
 }