Merge pull request #1048 from starius/loop-stop

cli: add command "loop stop"

Author: starius

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) {
+// 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, err
+		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 {