Merge pull request #841 from sputn1ck/better_ntfn_stream_l402

Notifications: Improve L402 handling

cmd/loop/l402.go

@@ -73,3 +73,31 @@ func listAuth(ctx *cli.Context) error {
 	printJSON(tokens)
 	return nil
 }
+
+var fetchL402Command = cli.Command{
+	Name:  "fetchl402",
+	Usage: "fetches a new L402 authentication token from the server",
+	Description: "Fetches a new L402 authentication token from the server. " +
+		"This token is required to listen to notifications from the server, " +
+		"such as reservation notifications. If a L402 is already present in " +
+		"the store, this command is a no-op.",
+	Action: fetchL402,
+}
+
+func fetchL402(ctx *cli.Context) error {
+	client, cleanup, err := getClient(ctx)
+	if err != nil {
+		return err
+	}
+	defer cleanup()
+
+	res, err := client.FetchL402Token(
+		context.Background(), &looprpc.FetchL402TokenRequest{},
+	)
+	if err != nil {
+		return err
+	}
+
+	printRespJSON(res)
+	return nil
+}

cmd/loop/main.go

@@ -144,7 +144,7 @@ func main() {
 	}
 	app.Commands = []cli.Command{
 		loopOutCommand, loopInCommand, termsCommand,
-		monitorCommand, quoteCommand, listAuthCommand,
+		monitorCommand, quoteCommand, listAuthCommand, fetchL402Command,
 		listSwapsCommand, swapInfoCommand, getLiquidityParamsCommand,
 		setLiquidityRuleCommand, suggestSwapCommand, setParamsCommand,
 		getInfoCommand, abandonSwapCommand, reservationsCommands,

loopd/daemon.go

@@ -504,8 +504,8 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
 
 	// Start the notification manager.
 	notificationCfg := &notifications.Config{
-		Client:    loop_swaprpc.NewSwapServerClient(swapClient.Conn),
-		FetchL402: swapClient.Server.FetchL402,
+		Client:       loop_swaprpc.NewSwapServerClient(swapClient.Conn),
+		CurrentToken: swapClient.L402Store.CurrentToken,
 	}
 	notificationManager := notifications.NewManager(notificationCfg)
 

loopd/perms/perms.go

@@ -77,6 +77,10 @@ var RequiredPermissions = map[string][]bakery.Op{
 		Entity: "auth",
 		Action: "read",
 	}},
+	"/looprpc.SwapClient/FetchL402Token": {{
+		Entity: "auth",
+		Action: "write",
+	}},
 	"/looprpc.SwapClient/SuggestSwaps": {{
 		Entity: "suggestions",
 		Action: "read",

loopd/swapclient_server.go

@@ -991,6 +991,21 @@ func (s *swapClientServer) GetLsatTokens(ctx context.Context,
 	return s.GetL402Tokens(ctx, req)
 }
 
+// FetchL402Token fetches a L402 Token from the server. This is required to
+// listen for server notifications such as reservations. If a token is already
+// in the local L402, nothing will happen.
+func (s *swapClientServer) FetchL402Token(ctx context.Context,
+	_ *looprpc.FetchL402TokenRequest) (*looprpc.FetchL402TokenResponse,
+	error) {
+
+	err := s.impl.Server.FetchL402(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return &looprpc.FetchL402TokenResponse{}, nil
+}
+
 // GetInfo returns basic information about the loop daemon and details to swaps
 // from the swap store.
 func (s *swapClientServer) GetInfo(ctx context.Context,

looprpc/client.proto

@@ -89,6 +89,12 @@ service SwapClient {
     */
     rpc GetLsatTokens (TokensRequest) returns (TokensResponse);
 
+    /* loop: `fetchl402`
+    FetchL402Token fetches an L402 token from the server, this is required in
+    order to receive reservation notifications from the server.
+    */
+    rpc FetchL402Token (FetchL402TokenRequest) returns (FetchL402TokenResponse);
+
     /* loop: `getinfo`
     GetInfo gets basic information about the loop daemon.
     */
@@ -830,6 +836,12 @@ message TokensResponse {
     repeated L402Token tokens = 1;
 }
 
+message FetchL402TokenRequest {
+}
+
+message FetchL402TokenResponse {
+}
+
 message L402Token {
     /*
     The base macaroon that was baked by the auth server.

looprpc/client.swagger.json

@@ -677,6 +677,9 @@
       "default": "FAILURE_REASON_NONE",
       "description": " - FAILURE_REASON_NONE: FAILURE_REASON_NONE is set when the swap did not fail, it is either in\nprogress or succeeded.\n - FAILURE_REASON_OFFCHAIN: FAILURE_REASON_OFFCHAIN indicates that a loop out failed because it wasn't\npossible to find a route for one or both off chain payments that met the fee\nand timelock limits required.\n - FAILURE_REASON_TIMEOUT: FAILURE_REASON_TIMEOUT indicates that the swap failed because on chain htlc\ndid not confirm before its expiry, or it confirmed too late for us to reveal\nour preimage and claim.\n - FAILURE_REASON_SWEEP_TIMEOUT: FAILURE_REASON_SWEEP_TIMEOUT indicates that a loop out permanently failed\nbecause the on chain htlc wasn't swept before the server revoked the\nhtlc.\n - FAILURE_REASON_INSUFFICIENT_VALUE: FAILURE_REASON_INSUFFICIENT_VALUE indicates that a loop out has failed\nbecause the on chain htlc had a lower value than requested.\n - FAILURE_REASON_TEMPORARY: FAILURE_REASON_TEMPORARY indicates that a swap cannot continue due to an\ninternal error. Manual intervention such as a restart is required.\n - FAILURE_REASON_INCORRECT_AMOUNT: FAILURE_REASON_INCORRECT_AMOUNT indicates that a loop in permanently failed\nbecause the amount extended by an external loop in htlc is insufficient.\n - FAILURE_REASON_ABANDONED: FAILURE_REASON_ABANDONED indicates that a swap permanently failed because\nthe client manually abandoned the swap.\n - FAILURE_REASON_INSUFFICIENT_CONFIRMED_BALANCE: FAILURE_REASON_INSUFFICIENT_CONFIRMED_BALANCE indicates that a swap\nwasn't published due to insufficient confirmed balance.\n - FAILURE_REASON_INCORRECT_HTLC_AMT_SWEPT: FAILURE_REASON_INCORRECT_HTLC_AMT_SWEPT indicates that a swap\nwasn't published due to insufficient confirmed balance."
     },
+    "looprpcFetchL402TokenResponse": {
+      "type": "object"
+    },
     "looprpcGetInfoResponse": {
       "type": "object",
       "properties": {

looprpc/client_grpc.pb.go

@@ -68,6 +68,10 @@ type SwapClientClient interface {
 	// Type LsatToken used by GetLsatTokens in the past was renamed to L402Token,
 	// but this does not affect binary encoding, so we can use type L402Token here.
 	GetLsatTokens(ctx context.Context, in *TokensRequest, opts ...grpc.CallOption) (*TokensResponse, error)
+	// loop: `fetchl402`
+	// FetchL402Token fetches an L402 token from the server, this is required in
+	// order to receive reservation notifications from the server.
+	FetchL402Token(ctx context.Context, in *FetchL402TokenRequest, opts ...grpc.CallOption) (*FetchL402TokenResponse, error)
 	// loop: `getinfo`
 	// GetInfo gets basic information about the loop daemon.
 	GetInfo(ctx context.Context, in *GetInfoRequest, opts ...grpc.CallOption) (*GetInfoResponse, error)
@@ -252,6 +256,15 @@ func (c *swapClientClient) GetLsatTokens(ctx context.Context, in *TokensRequest,
 	return out, nil
 }
 
+func (c *swapClientClient) FetchL402Token(ctx context.Context, in *FetchL402TokenRequest, opts ...grpc.CallOption) (*FetchL402TokenResponse, error) {
+	out := new(FetchL402TokenResponse)
+	err := c.cc.Invoke(ctx, "/looprpc.SwapClient/FetchL402Token", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 func (c *swapClientClient) GetInfo(ctx context.Context, in *GetInfoRequest, opts ...grpc.CallOption) (*GetInfoResponse, error) {
 	out := new(GetInfoResponse)
 	err := c.cc.Invoke(ctx, "/looprpc.SwapClient/GetInfo", in, out, opts...)
@@ -378,6 +391,10 @@ type SwapClientServer interface {
 	// Type LsatToken used by GetLsatTokens in the past was renamed to L402Token,
 	// but this does not affect binary encoding, so we can use type L402Token here.
 	GetLsatTokens(context.Context, *TokensRequest) (*TokensResponse, error)
+	// loop: `fetchl402`
+	// FetchL402Token fetches an L402 token from the server, this is required in
+	// order to receive reservation notifications from the server.
+	FetchL402Token(context.Context, *FetchL402TokenRequest) (*FetchL402TokenResponse, error)
 	// loop: `getinfo`
 	// GetInfo gets basic information about the loop daemon.
 	GetInfo(context.Context, *GetInfoRequest) (*GetInfoResponse, error)
@@ -458,6 +475,9 @@ func (UnimplementedSwapClientServer) GetL402Tokens(context.Context, *TokensReque
 func (UnimplementedSwapClientServer) GetLsatTokens(context.Context, *TokensRequest) (*TokensResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method GetLsatTokens not implemented")
 }
+func (UnimplementedSwapClientServer) FetchL402Token(context.Context, *FetchL402TokenRequest) (*FetchL402TokenResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method FetchL402Token not implemented")
+}
 func (UnimplementedSwapClientServer) GetInfo(context.Context, *GetInfoRequest) (*GetInfoResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method GetInfo not implemented")
 }
@@ -732,6 +752,24 @@ func _SwapClient_GetLsatTokens_Handler(srv interface{}, ctx context.Context, dec
 	return interceptor(ctx, in, info, handler)
 }
 
+func _SwapClient_FetchL402Token_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(FetchL402TokenRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(SwapClientServer).FetchL402Token(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/looprpc.SwapClient/FetchL402Token",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(SwapClientServer).FetchL402Token(ctx, req.(*FetchL402TokenRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 func _SwapClient_GetInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(GetInfoRequest)
 	if err := dec(in); err != nil {
@@ -931,6 +969,10 @@ var SwapClient_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "GetLsatTokens",
 			Handler:    _SwapClient_GetLsatTokens_Handler,
 		},
+		{
+			MethodName: "FetchL402Token",
+			Handler:    _SwapClient_FetchL402Token_Handler,
+		},
 		{
 			MethodName: "GetInfo",
 			Handler:    _SwapClient_GetInfo_Handler,

looprpc/swapclient.pb.json.go

@@ -363,6 +363,31 @@ func RegisterSwapClientJSONCallbacks(registry map[string]func(ctx context.Contex
 		callback(string(respBytes), nil)
 	}
 
+	registry["looprpc.SwapClient.FetchL402Token"] = func(ctx context.Context,
+		conn *grpc.ClientConn, reqJSON string, callback func(string, error)) {
+
+		req := &FetchL402TokenRequest{}
+		err := marshaler.Unmarshal([]byte(reqJSON), req)
+		if err != nil {
+			callback("", err)
+			return
+		}
+
+		client := NewSwapClientClient(conn)
+		resp, err := client.FetchL402Token(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.GetInfo"] = func(ctx context.Context,
 		conn *grpc.ClientConn, reqJSON string, callback func(string, error)) {
 

notifications/manager.go

@@ -5,6 +5,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/lightninglabs/aperture/l402"
 	"github.com/lightninglabs/loop/swapserverrpc"
 	"google.golang.org/grpc"
 )
@@ -37,8 +38,9 @@ type Config struct {
 	// Client is the client used to communicate with the swap server.
 	Client Client
 
-	// FetchL402 is the function used to fetch the l402 token.
-	FetchL402 func(context.Context) error
+	// CurrentToken returns the token that is currently contained in the
+	// store or an l402.ErrNoToken error if there is none.
+	CurrentToken func() (*l402.Token, error)
 }
 
 // Manager is a manager for notifications that the swap server sends to the
@@ -113,9 +115,13 @@ func (m *Manager) Run(ctx context.Context) error {
 		// the FetchL402 method. As a client might not have outbound capacity
 		// yet, we'll retry until we get a valid response.
 		if !m.hasL402 {
-			err := m.cfg.FetchL402(ctx)
+			_, err := m.cfg.CurrentToken()
 			if err != nil {
-				log.Errorf("Error fetching L402: %v", err)
+				// We only log the error if it's not the case that we
+				// don't have a token yet to avoid spamming the logs.
+				if err != l402.ErrNoToken {
+					log.Errorf("Error getting L402 from store: %v", err)
+				}
 				continue
 			}
 			m.hasL402 = true

notifications/manager_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/lightninglabs/aperture/l402"
 	"github.com/lightninglabs/loop/swapserverrpc"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc"
@@ -101,9 +102,9 @@ func TestManager_ReservationNotification(t *testing.T) {
 	// Create a Manager with the mock client
 	mgr := NewManager(&Config{
 		Client: mockClient,
-		FetchL402: func(ctx context.Context) error {
+		CurrentToken: func() (*l402.Token, error) {
 			// Simulate successful fetching of L402
-			return nil
+			return nil, nil
 		},
 	})