diff --git a/cmd/loop/debug.go b/cmd/loop/debug.go index e2f7318c..84b24dae 100644 --- a/cmd/loop/debug.go +++ b/cmd/loop/debug.go @@ -43,12 +43,7 @@ func forceAutoloop(ctx context.Context, cmd *cli.Command) error { } func getDebugClient(ctx context.Context, cmd *cli.Command) (looprpc.DebugClient, func(), error) { - rpcServer := cmd.String("rpcserver") - tlsCertPath, macaroonPath, err := extractPathArgs(cmd) - if err != nil { - return nil, nil, err - } - conn, cleanup, err := getClientConn(rpcServer, tlsCertPath, macaroonPath) + conn, cleanup, err := sessionTransport.Dial(cmd) if err != nil { return nil, nil, err } diff --git a/cmd/loop/main.go b/cmd/loop/main.go index f990f913..f6726b1d 100644 --- a/cmd/loop/main.go +++ b/cmd/loop/main.go @@ -252,14 +252,9 @@ func getClient(cmd *cli.Command) (looprpc.SwapClientClient, // getClientWithConn returns both the SwapClient RPC client and the underlying // gRPC connection so callers can perform connection-aware actions. func getClientWithConn(cmd *cli.Command) (looprpc.SwapClientClient, - *grpc.ClientConn, func(), error) { + daemonConn, func(), error) { - rpcServer := cmd.String("rpcserver") - tlsCertPath, macaroonPath, err := extractPathArgs(cmd) - if err != nil { - return nil, nil, nil, err - } - conn, cleanup, err := getClientConn(rpcServer, tlsCertPath, macaroonPath) + conn, cleanup, err := sessionTransport.Dial(cmd) if err != nil { return nil, nil, nil, err } @@ -481,7 +476,8 @@ func logSwap(swap *looprpc.SwapStatus) { fmt.Println() } -func getClientConn(address, tlsCertPath, macaroonPath string) (*grpc.ClientConn, +// getClientConn dials the loopd gRPC server with TLS and macaroon auth. +func getClientConn(address, tlsCertPath, macaroonPath string) (daemonConn, func(), error) { // We always need to send a macaroon. @@ -495,6 +491,14 @@ func getClientConn(address, tlsCertPath, macaroonPath string) (*grpc.ClientConn, macOption, } + // Install gRPC interceptors for session recording if needed. + if unary := sessionTransport.UnaryInterceptor(); unary != nil { + opts = append(opts, grpc.WithChainUnaryInterceptor(unary)) + } + if stream := sessionTransport.StreamInterceptor(); stream != nil { + opts = append(opts, grpc.WithChainStreamInterceptor(stream)) + } + // Since TLS cannot be disabled, we'll always have a cert file to read. creds, err := credentials.NewClientTLSFromFile(tlsCertPath, "") if err != nil { diff --git a/cmd/loop/session_transport.go b/cmd/loop/session_transport.go new file mode 100644 index 00000000..5f06d244 --- /dev/null +++ b/cmd/loop/session_transport.go @@ -0,0 +1,83 @@ +package main + +import ( + "context" + + "github.com/urfave/cli/v3" + "google.golang.org/grpc" + "google.golang.org/grpc/connectivity" +) + +// grpcTransport customizes gRPC dialing and interceptors for sessions. +type grpcTransport interface { + // Dial returns a gRPC connection for the CLI. + Dial(cmd *cli.Command) (daemonConn, func(), error) + + // UnaryInterceptor returns the unary interceptor to apply for session + // flows. + UnaryInterceptor() grpc.UnaryClientInterceptor + + // StreamInterceptor returns the stream interceptor to apply for session + // flows. + StreamInterceptor() grpc.StreamClientInterceptor +} + +// daemonConn is the client connection interface required by stop/wait flows. +type daemonConn interface { + grpc.ClientConnInterface + + // GetState reports the current connectivity state of the client + // channel. + GetState() connectivity.State + + // WaitForStateChange blocks until the state changes or the context + // expires. + WaitForStateChange(ctx context.Context, + sourceState connectivity.State) bool + + // Connect forces the channel out of idle mode. + Connect() +} + +// directGrpcTransport establishes real gRPC connections to loopd. +type directGrpcTransport struct{} + +// Dial opens a direct gRPC connection. +func (t *directGrpcTransport) Dial( + cmd *cli.Command) (daemonConn, func(), error) { + + return dialDirectConn(cmd) +} + +// UnaryInterceptor returns nil because direct connections do not need wrapping. +func (t *directGrpcTransport) UnaryInterceptor() grpc.UnaryClientInterceptor { + return nil +} + +// StreamInterceptor returns nil because direct connections do not need +// wrapping. +func (t *directGrpcTransport) StreamInterceptor() grpc.StreamClientInterceptor { + return nil +} + +// sessionTransport defines the active gRPC transport for CLI commands. +var sessionTransport grpcTransport = &directGrpcTransport{} + +// hookGrpc installs the active gRPC session transport hook. +func hookGrpc(transport grpcTransport) func() { + prev := sessionTransport + sessionTransport = transport + + return func() { sessionTransport = prev } +} + +// dialDirectConn returns the standard CLI gRPC connection. +func dialDirectConn(cmd *cli.Command) (daemonConn, func(), error) { + rpcServer := cmd.String("rpcserver") + tlsCertPath, macaroonPath, err := extractPathArgs(cmd) + if err != nil { + return nil, nil, err + } + + return getClientConn(rpcServer, tlsCertPath, macaroonPath) +} diff --git a/cmd/loop/stop.go b/cmd/loop/stop.go index 293c8c6b..75fbd387 100644 --- a/cmd/loop/stop.go +++ b/cmd/loop/stop.go @@ -6,7 +6,6 @@ import ( "github.com/lightninglabs/loop/looprpc" "github.com/urfave/cli/v3" - "google.golang.org/grpc" "google.golang.org/grpc/connectivity" ) @@ -63,7 +62,7 @@ func stopDaemon(ctx context.Context, cmd *cli.Command) error { // 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 { +func waitForDaemonShutdown(ctx context.Context, conn daemonConn) error { for { state := conn.GetState()