Merge pull request #1048 from starius/loop-stop

cli: add command "loop stop"

cmd/loop/main.go

@@ -89,7 +89,7 @@ var (
 		listSwapsCommand, swapInfoCommand, getLiquidityParamsCommand,
 		setLiquidityRuleCommand, suggestSwapCommand, setParamsCommand,
 		getInfoCommand, abandonSwapCommand, reservationsCommands,
-		instantOutCommand, listInstantOutsCommand,
+		instantOutCommand, listInstantOutsCommand, stopCommand,
 		printManCommand, printMarkdownCommand,
 	}
 )
@@ -190,20 +190,37 @@ func main() {
 	}
 }
 
-func getClient(ctx context.Context, cmd *cli.Command) (looprpc.SwapClientClient, func(), error) {
-	rpcServer := cmd.String("rpcserver")
-	tlsCertPath, macaroonPath, err := extractPathArgs(cmd)
+// getClient establishes a SwapClient RPC connection and returns the client and
+// a cleanup handler.
+func getClient(ctx context.Context, cmd *cli.Command) (looprpc.SwapClientClient,
+	func(), error) {
+
+	client, _, cleanup, err := getClientWithConn(ctx, cmd)
 	if err != nil {
 		return nil, nil, err
 	}
+
+	return client, cleanup, nil
+}
+
+// getClientWithConn returns both the SwapClient RPC client and the underlying
+// gRPC connection so callers can perform connection-aware actions.
+func getClientWithConn(ctx context.Context, cmd *cli.Command) (
+	looprpc.SwapClientClient, *grpc.ClientConn, func(), error) {
+
+	rpcServer := cmd.String("rpcserver")
+	tlsCertPath, macaroonPath, err := extractPathArgs(cmd)
+	if err != nil {
+		return nil, nil, nil, err
+	}
 	conn, err := getClientConn(ctx, rpcServer, tlsCertPath, macaroonPath)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
 	}
 	cleanup := func() { conn.Close() }
 
 	loopClient := looprpc.NewSwapClientClient(conn)
-	return loopClient, cleanup, nil
+	return loopClient, conn, cleanup, nil
 }
 
 func getMaxRoutingFee(amt btcutil.Amount) btcutil.Amount {

cmd/loop/stop.go

@@ -0,0 +1,88 @@
+package main
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/lightninglabs/loop/looprpc"
+	"github.com/urfave/cli/v3"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/connectivity"
+)
+
+var stopCommand = &cli.Command{
+	Name:        "stop",
+	Usage:       "stop the loop daemon",
+	Description: "Requests loopd to perform a graceful shutdown.",
+	Flags: []cli.Flag{
+		&cli.BoolFlag{
+			Name:  "wait",
+			Usage: "wait until loopd fully shuts down",
+		},
+	},
+	Action: stopDaemon,
+}
+
+// stopDaemon requests the daemon to shut down gracefully and optionally waits
+// for the gRPC connection to terminate.
+func stopDaemon(ctx context.Context, cmd *cli.Command) error {
+	waitForShutdown := cmd.Bool("wait")
+
+	// Establish a client connection to loopd.
+	client, conn, cleanup, err := getClientWithConn(ctx, cmd)
+	if err != nil {
+		return err
+	}
+	defer cleanup()
+
+	// Request loopd to shut down.
+	_, err = client.StopDaemon(ctx, &looprpc.StopDaemonRequest{})
+	if err != nil {
+		return err
+	}
+
+	fmt.Println("Shutting down loopd")
+
+	// Optionally wait for the gRPC connection to terminate.
+	if !waitForShutdown {
+		return nil
+	}
+
+	fmt.Println("Waiting for loopd to exit...")
+
+	err = waitForDaemonShutdown(ctx, conn)
+	if err != nil {
+		return err
+	}
+
+	fmt.Println("Loopd shut down")
+
+	return nil
+}
+
+// waitForDaemonShutdown monitors the gRPC connectivity state until the daemon
+// disappears. To avoid getting stuck in idle mode we nudge the connection to
+// reconnect when needed.
+func waitForDaemonShutdown(ctx context.Context, conn *grpc.ClientConn) error {
+	for {
+		state := conn.GetState()
+
+		switch state {
+		case connectivity.Shutdown:
+			return nil
+
+		// Connection attempts now fail which means loopd is offline.
+		case connectivity.TransientFailure:
+			return nil
+
+		// Force the channel out of Idle so we'll see failure states
+		// once loopd stops serving.
+		case connectivity.Idle:
+			conn.Connect()
+		}
+
+		if !conn.WaitForStateChange(ctx, state) {
+			return ctx.Err()
+		}
+	}
+}

docs/loop.1

@@ -431,6 +431,16 @@ list all instant out swaps
 .PP
 \fB--help, -h\fP: show help
 
+.SH stop
+.PP
+stop the loop daemon
+
+.PP
+\fB--help, -h\fP: show help
+
+.PP
+\fB--wait\fP: wait until loopd fully shuts down
+
 .SH static, s
 .PP
 perform on-chain to off-chain swaps using static addresses.

docs/loop.md

@@ -468,6 +468,25 @@ The following flags are supported:
 |-----------------|-------------|------|:-------------:|
 | `--help` (`-h`) | show help   | bool |    `false`    |
 
+### `stop` command
+
+stop the loop daemon.
+
+Requests loopd to perform a graceful shutdown.
+
+Usage:
+
+```bash
+$ loop [GLOBAL FLAGS] stop [COMMAND FLAGS] [ARGUMENTS...]
+```
+
+The following flags are supported:
+
+| Name            | Description                       | Type | Default value |
+|-----------------|-----------------------------------|------|:-------------:|
+| `--wait`        | wait until loopd fully shuts down | bool |    `false`    |
+| `--help` (`-h`) | show help                         | bool |    `false`    |
+
 ### `static` command (aliases: `s`)
 
 perform on-chain to off-chain swaps using static addresses.

loopd/daemon.go

@@ -746,6 +746,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 		withdrawalManager:    withdrawalManager,
 		staticLoopInManager:  staticLoopInManager,
 		assetClient:          d.assetClient,
+		stopDaemon:           d.Stop,
 	}
 
 	// Retrieve all currently existing swaps from the database.

loopd/swapclient_server.go

@@ -104,6 +104,9 @@ type swapClientServer struct {
 	nextSubscriberID     int
 	swapsLock            sync.Mutex
 	mainCtx              context.Context
+
+	// stopDaemon is invoked to trigger a graceful shutdown of the daemon.
+	stopDaemon func()
 }
 
 // LoopOut initiates a loop out swap with the given parameters. The call returns
@@ -1346,6 +1349,22 @@ func (s *swapClientServer) GetInfo(ctx context.Context,
 	}, nil
 }
 
+// StopDaemon triggers a graceful shutdown of the daemon process.
+func (s *swapClientServer) StopDaemon(ctx context.Context,
+	_ *looprpc.StopDaemonRequest) (*looprpc.StopDaemonResponse, error) {
+
+	// Ensure we have a shutdown handler to invoke.
+	if s.stopDaemon == nil {
+		return nil, status.Error(codes.Unimplemented,
+			"stop daemon not supported")
+	}
+
+	// Initiate the shutdown sequence.
+	s.stopDaemon()
+
+	return &looprpc.StopDaemonResponse{}, nil
+}
+
 // GetLiquidityParams gets our current liquidity manager's parameters.
 func (s *swapClientServer) GetLiquidityParams(_ context.Context,
 	_ *looprpc.GetLiquidityParamsRequest) (*looprpc.LiquidityParameters,

loopd/swapclient_server_test.go

@@ -261,6 +261,29 @@ func TestValidateLoopInRequest(t *testing.T) {
 	}
 }
 
+// TestSwapClientServerStopDaemon ensures that calling StopDaemon triggers the
+// daemon shutdown.
+func TestSwapClientServerStopDaemon(t *testing.T) {
+	t.Parallel()
+
+	// Prepare a server instance that tracks whether shutdown is requested.
+	var stopCalled bool
+	server := &swapClientServer{
+		stopDaemon: func() {
+			stopCalled = true
+		},
+	}
+
+	// Request the daemon to stop and assert the callback executed.
+	_, err := server.StopDaemon(
+		context.Background(), &looprpc.StopDaemonRequest{},
+	)
+	require.NoError(t, err)
+
+	// Ensure our shutdown callback executed.
+	require.True(t, stopCalled)
+}
+
 // TestValidateLoopOutRequest tests validation of loop out requests.
 func TestValidateLoopOutRequest(t *testing.T) {
 	tests := []struct {

looprpc/client.pb.gw.go

@@ -487,6 +487,24 @@ func local_request_SwapClient_GetInfo_0(ctx context.Context, marshaler runtime.M
 
 }
 
+func request_SwapClient_StopDaemon_0(ctx context.Context, marshaler runtime.Marshaler, client SwapClientClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var protoReq StopDaemonRequest
+	var metadata runtime.ServerMetadata
+
+	msg, err := client.StopDaemon(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
+	return msg, metadata, err
+
+}
+
+func local_request_SwapClient_StopDaemon_0(ctx context.Context, marshaler runtime.Marshaler, server SwapClientServer, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
+	var protoReq StopDaemonRequest
+	var metadata runtime.ServerMetadata
+
+	msg, err := server.StopDaemon(ctx, &protoReq)
+	return msg, metadata, err
+
+}
+
 func request_SwapClient_GetLiquidityParams_0(ctx context.Context, marshaler runtime.Marshaler, client SwapClientClient, req *http.Request, pathParams map[string]string) (proto.Message, runtime.ServerMetadata, error) {
 	var protoReq GetLiquidityParamsRequest
 	var metadata runtime.ServerMetadata
@@ -1195,6 +1213,31 @@ func RegisterSwapClientHandlerServer(ctx context.Context, mux *runtime.ServeMux,
 
 	})
 
+	mux.Handle("POST", pattern_SwapClient_StopDaemon_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		var stream runtime.ServerTransportStream
+		ctx = grpc.NewContextWithServerTransportStream(ctx, &stream)
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		var err error
+		var annotatedContext context.Context
+		annotatedContext, err = runtime.AnnotateIncomingContext(ctx, mux, req, "/looprpc.SwapClient/StopDaemon", runtime.WithHTTPPathPattern("/v1/daemon/stop"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := local_request_SwapClient_StopDaemon_0(annotatedContext, inboundMarshaler, server, req, pathParams)
+		md.HeaderMD, md.TrailerMD = metadata.Join(md.HeaderMD, stream.Header()), metadata.Join(md.TrailerMD, stream.Trailer())
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+
+		forward_SwapClient_StopDaemon_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+
+	})
+
 	mux.Handle("GET", pattern_SwapClient_GetLiquidityParams_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
 		ctx, cancel := context.WithCancel(req.Context())
 		defer cancel()
@@ -1875,6 +1918,28 @@ func RegisterSwapClientHandlerClient(ctx context.Context, mux *runtime.ServeMux,
 
 	})
 
+	mux.Handle("POST", pattern_SwapClient_StopDaemon_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
+		ctx, cancel := context.WithCancel(req.Context())
+		defer cancel()
+		inboundMarshaler, outboundMarshaler := runtime.MarshalerForRequest(mux, req)
+		var err error
+		var annotatedContext context.Context
+		annotatedContext, err = runtime.AnnotateContext(ctx, mux, req, "/looprpc.SwapClient/StopDaemon", runtime.WithHTTPPathPattern("/v1/daemon/stop"))
+		if err != nil {
+			runtime.HTTPError(ctx, mux, outboundMarshaler, w, req, err)
+			return
+		}
+		resp, md, err := request_SwapClient_StopDaemon_0(annotatedContext, inboundMarshaler, client, req, pathParams)
+		annotatedContext = runtime.NewServerMetadataContext(annotatedContext, md)
+		if err != nil {
+			runtime.HTTPError(annotatedContext, mux, outboundMarshaler, w, req, err)
+			return
+		}
+
+		forward_SwapClient_StopDaemon_0(annotatedContext, mux, outboundMarshaler, w, req, resp, mux.GetForwardResponseOptions()...)
+
+	})
+
 	mux.Handle("GET", pattern_SwapClient_GetLiquidityParams_0, func(w http.ResponseWriter, req *http.Request, pathParams map[string]string) {
 		ctx, cancel := context.WithCancel(req.Context())
 		defer cancel()
@@ -2233,6 +2298,8 @@ var (
 
 	pattern_SwapClient_GetInfo_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"v1", "loop", "info"}, ""))
 
+	pattern_SwapClient_StopDaemon_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"v1", "daemon", "stop"}, ""))
+
 	pattern_SwapClient_GetLiquidityParams_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"v1", "liquidity", "params"}, ""))
 
 	pattern_SwapClient_SetLiquidityParams_0 = runtime.MustPattern(runtime.NewPattern(1, []int{2, 0, 2, 1, 2, 2}, []string{"v1", "liquidity", "params"}, ""))
@@ -2289,6 +2356,8 @@ var (
 
 	forward_SwapClient_GetInfo_0 = runtime.ForwardResponseMessage
 
+	forward_SwapClient_StopDaemon_0 = runtime.ForwardResponseMessage
+
 	forward_SwapClient_GetLiquidityParams_0 = runtime.ForwardResponseMessage
 
 	forward_SwapClient_SetLiquidityParams_0 = runtime.ForwardResponseMessage

looprpc/client.proto

@@ -100,6 +100,11 @@ service SwapClient {
     */
     rpc GetInfo (GetInfoRequest) returns (GetInfoResponse);
 
+    /* loop: `stop`
+    StopDaemon instructs the daemon to shut down gracefully.
+    */
+    rpc StopDaemon (StopDaemonRequest) returns (StopDaemonResponse);
+
     /* loop: `getparams`
     GetLiquidityParams gets the parameters that the daemon's liquidity manager
     is currently configured with. This may be nil if nothing is configured.
@@ -202,6 +207,12 @@ service SwapClient {
         returns (StaticAddressLoopInResponse);
 }
 
+message StopDaemonRequest {
+}
+
+message StopDaemonResponse {
+}
+
 message LoopOutRequest {
     /*
     Requested swap amount in sat. This does not include the swap and miner fee.

looprpc/client.swagger.json

@@ -39,6 +39,29 @@
         ]
       }
     },
+    "/v1/daemon/stop": {
+      "post": {
+        "summary": "loop: `stop`\nStopDaemon instructs the daemon to shut down gracefully.",
+        "operationId": "SwapClient_StopDaemon",
+        "responses": {
+          "200": {
+            "description": "A successful response.",
+            "schema": {
+              "$ref": "#/definitions/looprpcStopDaemonResponse"
+            }
+          },
+          "default": {
+            "description": "An unexpected error response.",
+            "schema": {
+              "$ref": "#/definitions/rpcStatus"
+            }
+          }
+        },
+        "tags": [
+          "SwapClient"
+        ]
+      }
+    },
     "/v1/instantout": {
       "post": {
         "summary": "loop: `instantout`\nInstantOut initiates an instant out swap with the given parameters.",
@@ -2532,6 +2555,9 @@
         }
       }
     },
+    "looprpcStopDaemonResponse": {
+      "type": "object"
+    },
     "looprpcSuggestSwapsResponse": {
       "type": "object",
       "properties": {

looprpc/client.yaml

@@ -66,3 +66,5 @@ http:
     - selector: looprpc.SwapClient.StaticAddressLoopIn
       post: "/v1/staticaddr/loopin"
       body: "*"
+    - selector: looprpc.SwapClient.StopDaemon
+      post: "/v1/daemon/stop"

looprpc/client_grpc.pb.go

@@ -75,6 +75,9 @@ type SwapClientClient interface {
 	// loop: `getinfo`
 	// GetInfo gets basic information about the loop daemon.
 	GetInfo(ctx context.Context, in *GetInfoRequest, opts ...grpc.CallOption) (*GetInfoResponse, error)
+	// loop: `stop`
+	// StopDaemon instructs the daemon to shut down gracefully.
+	StopDaemon(ctx context.Context, in *StopDaemonRequest, opts ...grpc.CallOption) (*StopDaemonResponse, error)
 	// loop: `getparams`
 	// GetLiquidityParams gets the parameters that the daemon's liquidity manager
 	// is currently configured with. This may be nil if nothing is configured.
@@ -301,6 +304,15 @@ func (c *swapClientClient) GetInfo(ctx context.Context, in *GetInfoRequest, opts
 	return out, nil
 }
 
+func (c *swapClientClient) StopDaemon(ctx context.Context, in *StopDaemonRequest, opts ...grpc.CallOption) (*StopDaemonResponse, error) {
+	out := new(StopDaemonResponse)
+	err := c.cc.Invoke(ctx, "/looprpc.SwapClient/StopDaemon", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *swapClientClient) GetLiquidityParams(ctx context.Context, in *GetLiquidityParamsRequest, opts ...grpc.CallOption) (*LiquidityParameters, error) {
 	out := new(LiquidityParameters)
 	err := c.cc.Invoke(ctx, "/looprpc.SwapClient/GetLiquidityParams", in, out, opts...)
@@ -497,6 +509,9 @@ type SwapClientServer interface {
 	// loop: `getinfo`
 	// GetInfo gets basic information about the loop daemon.
 	GetInfo(context.Context, *GetInfoRequest) (*GetInfoResponse, error)
+	// loop: `stop`
+	// StopDaemon instructs the daemon to shut down gracefully.
+	StopDaemon(context.Context, *StopDaemonRequest) (*StopDaemonResponse, error)
 	// loop: `getparams`
 	// GetLiquidityParams gets the parameters that the daemon's liquidity manager
 	// is currently configured with. This may be nil if nothing is configured.
@@ -607,6 +622,9 @@ func (UnimplementedSwapClientServer) FetchL402Token(context.Context, *FetchL402T
 func (UnimplementedSwapClientServer) GetInfo(context.Context, *GetInfoRequest) (*GetInfoResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method GetInfo not implemented")
 }
+func (UnimplementedSwapClientServer) StopDaemon(context.Context, *StopDaemonRequest) (*StopDaemonResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method StopDaemon not implemented")
+}
 func (UnimplementedSwapClientServer) GetLiquidityParams(context.Context, *GetLiquidityParamsRequest) (*LiquidityParameters, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method GetLiquidityParams not implemented")
 }
@@ -938,6 +956,24 @@ func _SwapClient_GetInfo_Handler(srv interface{}, ctx context.Context, dec func(
 	return interceptor(ctx, in, info, handler)
 }
 
+func _SwapClient_StopDaemon_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(StopDaemonRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(SwapClientServer).StopDaemon(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/looprpc.SwapClient/StopDaemon",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(SwapClientServer).StopDaemon(ctx, req.(*StopDaemonRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _SwapClient_GetLiquidityParams_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(GetLiquidityParamsRequest)
 	if err := dec(in); err != nil {
@@ -1271,6 +1307,10 @@ var SwapClient_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "GetInfo",
 			Handler:    _SwapClient_GetInfo_Handler,
 		},
+		{
+			MethodName: "StopDaemon",
+			Handler:    _SwapClient_StopDaemon_Handler,
+		},
 		{
 			MethodName: "GetLiquidityParams",
 			Handler:    _SwapClient_GetLiquidityParams_Handler,

looprpc/perms.go

@@ -176,4 +176,8 @@ var RequiredPermissions = map[string][]bakery.Op{
 		Entity: "swap",
 		Action: "read",
 	}},
+	"/looprpc.SwapClient/StopDaemon": {{
+		Entity: "loop",
+		Action: "admin",
+	}},
 }

looprpc/swapclient.pb.json.go

@@ -413,6 +413,31 @@ func RegisterSwapClientJSONCallbacks(registry map[string]func(ctx context.Contex
 		callback(string(respBytes), nil)
 	}
 
+	registry["looprpc.SwapClient.StopDaemon"] = func(ctx context.Context,
+		conn *grpc.ClientConn, reqJSON string, callback func(string, error)) {
+
+		req := &StopDaemonRequest{}
+		err := marshaler.Unmarshal([]byte(reqJSON), req)
+		if err != nil {
+			callback("", err)
+			return
+		}
+
+		client := NewSwapClientClient(conn)
+		resp, err := client.StopDaemon(ctx, req)
+		if err != nil {
+			callback("", err)
+			return
+		}
+
+		respBytes, err := marshaler.Marshal(resp)
+		if err != nil {
+			callback("", err)
+			return
+		}
+		callback(string(respBytes), nil)
+	}
+
 	registry["looprpc.SwapClient.GetLiquidityParams"] = func(ctx context.Context,
 		conn *grpc.ClientConn, reqJSON string, callback func(string, error)) {