Merge pull request #661 from hieblmi/abandon-swaps

Abandon API for pending Loop-in swaps

Author: hieblmi

client.go

@@ -17,6 +17,7 @@ import (
 	"github.com/lightninglabs/loop/loopdb"
 	"github.com/lightninglabs/loop/swap"
 	"github.com/lightninglabs/loop/sweep"
+	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/routing/route"
 	"google.golang.org/grpc/status"
 )
@@ -68,6 +69,11 @@ type Client struct {
 	started uint32 // To be used atomically.
 	errChan chan error
 
+	// abandonChans allows for accessing a swap's abandon channel by
+	// providing its swap hash. This map is used to look up the abandon
+	// channel of a swap if the client requests to abandon it.
+	abandonChans map[lntypes.Hash]chan struct{}
+
 	lndServices *lndclient.LndServices
 	sweeper     *sweep.Sweeper
 	executor    *executor
@@ -179,6 +185,7 @@ func NewClient(dbDir string, loopDB loopdb.SwapStore,
 		sweeper:      sweeper,
 		executor:     executor,
 		resumeReady:  make(chan struct{}),
+		abandonChans: make(map[lntypes.Hash]chan struct{}),
 	}
 
 	cleanup := func() {
@@ -317,10 +324,10 @@ func (s *Client) Run(ctx context.Context, statusChan chan<- SwapInfo) error {
 	}()
 
 	// Main event loop.
-	err = s.executor.run(mainCtx, statusChan)
+	err = s.executor.run(mainCtx, statusChan, s.abandonChans)
 
 	// Consider canceled as happy flow.
-	if err == context.Canceled {
+	if errors.Is(err, context.Canceled) {
 		err = nil
 	}
 
@@ -374,6 +381,12 @@ func (s *Client) resumeSwaps(ctx context.Context,
 			continue
 		}
 
+		// Store the swap's abandon channel so that the client can
+		// abandon the swap by providing the swap hash.
+		s.executor.Lock()
+		s.abandonChans[swap.hash] = swap.abandonChan
+		s.executor.Unlock()
+
 		s.executor.initiateSwap(ctx, swap)
 	}
 }
@@ -578,6 +591,10 @@ func (s *Client) LoopIn(globalCtx context.Context,
 	}
 	swap := initResult.swap
 
+	s.executor.Lock()
+	s.abandonChans[swap.hash] = swap.abandonChan
+	s.executor.Unlock()
+
 	// Post swap to the main loop.
 	s.executor.initiateSwap(globalCtx, swap)
 
@@ -753,3 +770,26 @@ func (s *Client) Probe(ctx context.Context, req *ProbeRequest) error {
 		req.RouteHints,
 	)
 }
+
+// AbandonSwap sends a signal on the abandon channel of the swap identified by
+// the passed swap hash. This will cause the swap to abandon itself.
+func (s *Client) AbandonSwap(ctx context.Context,
+	req *AbandonSwapRequest) error {
+
+	if req == nil {
+		return errors.New("no request provided")
+	}
+
+	s.executor.Lock()
+	defer s.executor.Unlock()
+
+	select {
+	case s.abandonChans[req.SwapHash] <- struct{}{}:
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+		// This is to avoid writing to a full channel.
+	}
+
+	return nil
+}

cmd/loop/main.go

@@ -147,7 +147,7 @@ func main() {
 		monitorCommand, quoteCommand, listAuthCommand,
 		listSwapsCommand, swapInfoCommand, getLiquidityParamsCommand,
 		setLiquidityRuleCommand, suggestSwapCommand, setParamsCommand,
-		getInfoCommand,
+		getInfoCommand, abandonSwapCommand,
 	}
 
 	err := app.Run(os.Args)

cmd/loop/swaps.go

@@ -90,3 +90,73 @@ func swapInfo(ctx *cli.Context) error {
 	printRespJSON(resp)
 	return nil
 }
+
+var abandonSwapCommand = cli.Command{
+	Name:  "abandonswap",
+	Usage: "abandon a swap with a given swap hash",
+	Description: "This command overrides the database and abandons a " +
+		"swap with a given swap hash.\n\n" +
+		"!!! This command might potentially lead to loss of funds if " +
+		"it is applied to swaps that are still waiting for pending " +
+		"user funds. Before executing this command make sure that " +
+		"no funds are locked by the swap.",
+	ArgsUsage: "ID",
+	Flags: []cli.Flag{
+		cli.BoolFlag{
+			Name: "i_know_what_i_am_doing",
+			Usage: "Specify this flag if you made sure that you " +
+				"read and understood the following " +
+				"consequence of applying this command.",
+		},
+	},
+	Action: abandonSwap,
+}
+
+func abandonSwap(ctx *cli.Context) error {
+	args := ctx.Args()
+
+	var id string
+	switch {
+	case ctx.IsSet("id"):
+		id = ctx.String("id")
+
+	case ctx.NArg() > 0:
+		id = args[0]
+		args = args.Tail() // nolint:wastedassign
+
+	default:
+		// Show command help if no arguments and flags were provided.
+		return cli.ShowCommandHelp(ctx, "abandonswap")
+	}
+
+	if len(id) != hex.EncodedLen(lntypes.HashSize) {
+		return fmt.Errorf("invalid swap ID")
+	}
+	idBytes, err := hex.DecodeString(id)
+	if err != nil {
+		return fmt.Errorf("cannot hex decode id: %v", err)
+	}
+
+	client, cleanup, err := getClient(ctx)
+	if err != nil {
+		return err
+	}
+	defer cleanup()
+
+	if !ctx.Bool("i_know_what_i_am_doing") {
+		return cli.ShowCommandHelp(ctx, "abandonswap")
+	}
+
+	resp, err := client.AbandonSwap(
+		context.Background(), &looprpc.AbandonSwapRequest{
+			Id:                idBytes,
+			IKnowWhatIAmDoing: ctx.Bool("i_know_what_i_am_doing"),
+		},
+	)
+	if err != nil {
+		return err
+	}
+
+	printRespJSON(resp)
+	return nil
+}

executor.go

@@ -13,6 +13,7 @@ import (
 	"github.com/lightninglabs/lndclient"
 	"github.com/lightninglabs/loop/loopdb"
 	"github.com/lightninglabs/loop/sweep"
+	"github.com/lightningnetwork/lnd/lntypes"
 	"github.com/lightningnetwork/lnd/queue"
 )
 
@@ -46,6 +47,8 @@ type executor struct {
 	currentHeight uint32
 	ready         chan struct{}
 
+	sync.Mutex
+
 	executorConfig
 }
 
@@ -61,7 +64,8 @@ func newExecutor(cfg *executorConfig) *executor {
 // run starts the executor event loop. It accepts and executes new swaps,
 // providing them with required config data.
 func (s *executor) run(mainCtx context.Context,
-	statusChan chan<- SwapInfo) error {
+	statusChan chan<- SwapInfo,
+	abandonChans map[lntypes.Hash]chan struct{}) error {
 
 	var (
 		err            error
@@ -167,6 +171,15 @@ func (s *executor) run(mainCtx context.Context,
 					log.Errorf("Execute error: %v", err)
 				}
 
+				// If a loop-in ended we have to remove its
+				// abandon channel from our abandonChans map
+				// since the swap finalized.
+				if swap, ok := newSwap.(*loopInSwap); ok {
+					s.Lock()
+					delete(abandonChans, swap.hash)
+					s.Unlock()
+				}
+
 				select {
 				case swapDoneChan <- swapID:
 				case <-mainCtx.Done():

interface.go

@@ -394,3 +394,9 @@ type ProbeRequest struct {
 	// Optional hop hints.
 	RouteHints [][]zpay32.HopHint
 }
+
+// AbandonSwapRequest specifies the swap to abandon. It is identified by its
+// swap hash.
+type AbandonSwapRequest struct {
+	SwapHash lntypes.Hash
+}

loopd/perms/perms.go

@@ -31,6 +31,16 @@ var RequiredPermissions = map[string][]bakery.Op{
 		Entity: "swap",
 		Action: "read",
 	}},
+	"/looprpc.SwapClient/AbandonSwap": {{
+		Entity: "swap",
+		Action: "execute",
+	}, {
+		Entity: "loop",
+		Action: "in",
+	}, {
+		Entity: "loop",
+		Action: "out",
+	}},
 	"/looprpc.SwapClient/LoopOutTerms": {{
 		Entity: "terms",
 		Action: "read",

loopd/swapclient_server.go

@@ -288,6 +288,9 @@ func (s *swapClientServer) marshallSwap(loopSwap *loop.SwapInfo) (
 	case loopdb.StateFailIncorrectHtlcAmt:
 		failureReason = clientrpc.FailureReason_FAILURE_REASON_INCORRECT_AMOUNT
 
+	case loopdb.StateFailAbandoned:
+		failureReason = clientrpc.FailureReason_FAILURE_REASON_ABANDONED
+
 	default:
 		return nil, fmt.Errorf("unknown swap state: %v", loopSwap.State)
 	}
@@ -508,6 +511,49 @@ func (s *swapClientServer) SwapInfo(_ context.Context,
 	return s.marshallSwap(&swp)
 }
 
+// AbandonSwap requests the server to abandon a swap with the given hash.
+func (s *swapClientServer) AbandonSwap(ctx context.Context,
+	req *clientrpc.AbandonSwapRequest) (*clientrpc.AbandonSwapResponse,
+	error) {
+
+	if !req.IKnowWhatIAmDoing {
+		return nil, fmt.Errorf("please read the AbandonSwap API " +
+			"documentation")
+	}
+
+	swapHash, err := lntypes.MakeHash(req.Id)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing swap hash: %v", err)
+	}
+
+	s.swapsLock.Lock()
+	swap, ok := s.swaps[swapHash]
+	s.swapsLock.Unlock()
+	if !ok {
+		return nil, fmt.Errorf("swap with hash %s not found", req.Id)
+	}
+
+	if swap.SwapType.IsOut() {
+		return nil, fmt.Errorf("abandoning loop out swaps is not " +
+			"supported yet")
+	}
+
+	// If the swap is in a final state, we cannot abandon it.
+	if swap.State.IsFinal() {
+		return nil, fmt.Errorf("cannot abandon swap in final state, "+
+			"state = %s, hash = %s", swap.State.String(), swapHash)
+	}
+
+	err = s.impl.AbandonSwap(ctx, &loop.AbandonSwapRequest{
+		SwapHash: swapHash,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("error abandoning swap: %v", err)
+	}
+
+	return &clientrpc.AbandonSwapResponse{}, nil
+}
+
 // LoopOutTerms returns the terms that the server enforces for loop out swaps.
 func (s *swapClientServer) LoopOutTerms(ctx context.Context,
 	_ *clientrpc.TermsRequest) (*clientrpc.OutTermsResponse, error) {

loopdb/swapstate.go

@@ -64,6 +64,10 @@ const (
 	// StateFailIncorrectHtlcAmt indicates that the amount of an externally
 	// published loop in htlc didn't match the swap amount.
 	StateFailIncorrectHtlcAmt SwapState = 10
+
+	// StateFailAbandoned indicates that a swap has been abandoned. Its
+	// execution has been canceled. It won't further be processed.
+	StateFailAbandoned SwapState = 11
 )
 
 // SwapStateType defines the types of swap states that exist. Every swap state
@@ -98,6 +102,18 @@ func (s SwapState) Type() SwapStateType {
 	return StateTypeFail
 }
 
+// IsPending returns true if the swap is in a pending state.
+func (s SwapState) IsPending() bool {
+	return s == StateInitiated || s == StateHtlcPublished ||
+		s == StatePreimageRevealed || s == StateFailTemporary ||
+		s == StateInvoiceSettled
+}
+
+// IsFinal returns true if the swap is in a final state.
+func (s SwapState) IsFinal() bool {
+	return !s.IsPending()
+}
+
 // String returns a string representation of the swap's state.
 func (s SwapState) String() string {
 	switch s {
@@ -134,6 +150,9 @@ func (s SwapState) String() string {
 	case StateFailIncorrectHtlcAmt:
 		return "IncorrectHtlcAmt"
 
+	case StateFailAbandoned:
+		return "FailAbandoned"
+
 	default:
 		return "Unknown"
 	}