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

@@ -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)

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 {

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)
+}

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
 }

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

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=

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
+}

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()
+}

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)
 }