feat(swaps): recover crashed swap deals

This commit attempts to recover swap deals that were interrupted due to
a system or `xud` crash. In the case where we are the maker and have
attempted to send payment for the second leg of the swap, we attempt to
query the swap client for the preimage of that payment in case it went
through. We can then use that preimage to try to claim the payment from
the first leg of the swap. In case the payment is knkown to have failed,
we simply attempt to close any open invoices and mark the swap deal as
having errored.

If an outgoing payment is still in flight and we do not have the
preimage for it, we add it to a set of "pending" swaps and check on it
on a scheduled interval until we can determine whether it has failed
or succeeded.

A new `SwapRecovery` class is introduced to contain the logic for
recovering interrupted swap deals and for tracking swaps that are
still pending.

Raiden currently does not expose an API call to push a preimage to claim
an incoming payment or to reject an incoming payment, instead we print
a warning to the log for now.

The recovery attempts happen on `xud` startup by looking for any swap
deals in the database that have an `Active` state.

Closes #1079.

lib/Xud.ts

@@ -205,7 +205,9 @@ class Xud extends EventEmitter {
     }
     if (this.grpcAPIProxy) {
       closePromises.push(this.grpcAPIProxy.close());
-      await this.grpcAPIProxy.close();
+    }
+    if (this.swaps) {
+      this.swaps.close();
     }
     await Promise.all(closePromises);
 

lib/constants/enums.ts

@@ -81,6 +81,12 @@ export enum SwapState {
   Active = 0,
   Error = 1,
   Completed = 2,
+  /**
+   * A swap that was executed but wasn't formally completed. This may occur as a result of xud
+   * crashing late in the swap process, after htlcs for both legs of the swap are set up but
+   * before the swap is formally complete.
+   */
+  Recovered = 3,
 }
 
 export enum ReputationEvent {
@@ -125,6 +131,8 @@ export enum SwapFailureReason {
   DealTimedOut = 11,
   /** The swap failed due to an unrecognized error. */
   UnknownError = 12,
+  /** The swap failed because of a system or xud crash while the swap was being executed. */
+  Crash = 12,
 }
 
 export enum DisconnectionReason {

lib/lndclient/LndClient.ts

@@ -1,6 +1,6 @@
 import grpc, { ChannelCredentials, ClientReadableStream } from 'grpc';
 import Logger from '../Logger';
-import SwapClient, { ClientStatus, SwapClientInfo } from '../swaps/SwapClient';
+import SwapClient, { ClientStatus, SwapClientInfo, PaymentState } from '../swaps/SwapClient';
 import errors from './errors';
 import { LightningClient, WalletUnlockerClient } from '../proto/lndrpc_grpc_pb';
 import { InvoicesClient } from '../proto/lndinvoices_grpc_pb';
@@ -596,6 +596,38 @@ class LndClient extends SwapClient {
     }
   }
 
+  public lookupPayment = async (rHash: string) => {
+    const payments = await this.listPayments(true);
+    for (const payment of payments.getPaymentsList()) {
+      if (payment.getPaymentHash() === rHash) {
+        const paymentStatus = payment.getStatus();
+        if (payment.getStatus() === lndrpc.Payment.PaymentStatus.SUCCEEDED) {
+          const preimage = payment.getPaymentPreimage();
+          return { preimage, state: PaymentState.Succeeded };
+        } else if (paymentStatus === lndrpc.Payment.PaymentStatus.FAILED) {
+          return { state: PaymentState.Failed };
+        } else if (paymentStatus === lndrpc.Payment.PaymentStatus.IN_FLIGHT) {
+          return { state: PaymentState.Pending };
+        } else {
+          // unexpected payment status
+          this.logger.warn(`unexpected payment state for payment with hash ${rHash}`);
+          return { state: PaymentState.Failed };
+        }
+      }
+    }
+
+    // if no payment is found, we assume that the payment was never attempted by lnd
+    return { state: PaymentState.Failed };
+  }
+
+  private listPayments = (includeIncomplete?: boolean): Promise<lndrpc.ListPaymentsResponse> => {
+    const request = new lndrpc.ListPaymentsRequest();
+    if (includeIncomplete) {
+      request.setIncludeIncomplete(includeIncomplete);
+    }
+    return this.unaryInvoiceCall<lndrpc.ListPaymentsRequest, lndrpc.ListPaymentsResponse>('listPayments', request);
+  }
+
   private addHoldInvoice = (request: lndinvoices.AddHoldInvoiceRequest): Promise<lndinvoices.AddHoldInvoiceResp> => {
     return this.unaryInvoiceCall<lndinvoices.AddHoldInvoiceRequest, lndinvoices.AddHoldInvoiceResp>('addHoldInvoice', request);
   }

lib/raidenclient/RaidenClient.ts

@@ -1,6 +1,6 @@
 import http from 'http';
 import Logger from '../Logger';
-import SwapClient, { ClientStatus, ChannelBalance } from '../swaps/SwapClient';
+import SwapClient, { ClientStatus, ChannelBalance, PaymentState } from '../swaps/SwapClient';
 import errors from './errors';
 import { SwapDeal } from '../swaps/types';
 import { SwapClientType, SwapState, SwapRole } from '../constants/enums';
@@ -18,6 +18,30 @@ import { CurrencyInstance } from '../db/types';
 
 type RaidenErrorResponse = { errors: string };
 
+type PaymentEvent = {
+  event: string;
+  payment_network_address: string;
+  token_network_address: string
+  identifier: number;
+  amount?: number;
+  target?: string
+  initiator?: string;
+  secret?: string;
+  route?: string[];
+  reason?: string;
+  log_time: string;
+};
+
+type PendingTransfer = {
+  initiator: string;
+  locked_amount: string;
+  payment_identifier: string;
+  role: string;
+  target: string;
+  token_address: string;
+  transferred_amount: string;
+};
+
 /**
  * A utility function to parse the payload from an http response.
  */
@@ -70,6 +94,14 @@ class RaidenClient extends SwapClient {
     this.directChannelChecks = directChannelChecks;
   }
 
+  /**
+   * Derives an integer identifier using the first 4 bytes of a provided payment hash in hex.
+   * @param rHash a payment hash in hex
+   */
+  private static getIdentifier(rHash: string) {
+    return parseInt(rHash.substr(0, 8), 16);
+  }
+
   /**
    * Checks for connectivity and gets our Raiden account address
    */
@@ -203,6 +235,57 @@ class RaidenClient extends SwapClient {
     // not implemented, raiden does not use invoices
   }
 
+  public lookupPayment = async (rHash: string, currency?: string, destination?: string) => {
+    const identifier = RaidenClient.getIdentifier(rHash);
+
+    // first check if the payment is pending
+    const pendingTransfers = await this.getPendingTransfers(currency, destination);
+    for (const pendingTransfer of pendingTransfers) {
+      if (identifier === Number(pendingTransfer.payment_identifier)) {
+        return { state: PaymentState.Pending };
+      }
+    }
+
+    // if the payment isn't pending, check if it has succeeded or failed
+    let endpoint = 'payments';
+    if (currency) {
+      const tokenAddress = this.tokenAddresses.get(currency);
+      endpoint += `/${tokenAddress}`;
+      if (destination) {
+        endpoint += `/${destination}`;
+      }
+    }
+    const res = await this.sendRequest(endpoint, 'GET');
+    const paymentEvents = await parseResponseBody<PaymentEvent[]>(res);
+    for (const paymentEvent of paymentEvents) {
+      if (paymentEvent.identifier === identifier) {
+        const success = paymentEvent.event === 'EventPaymentSentSuccess';
+        if (success) {
+          const preimage = paymentEvent.secret;
+          return { preimage, state: PaymentState.Succeeded };
+        } else {
+          return { state: PaymentState.Failed };
+        }
+      }
+    }
+
+    // if there is no pending payment or event found, we assume that the payment was never attempted by raiden
+    return { state: PaymentState.Failed };
+  }
+
+  private getPendingTransfers = async (currency?: string, destination?: string) => {
+    let endpoint = 'pending_transfers';
+    if (currency) {
+      const tokenAddress = this.tokenAddresses.get(currency);
+      endpoint += `/${tokenAddress}`;
+      if (destination) {
+        endpoint += `/${destination}`;
+      }
+    }
+    const res = await this.sendRequest(endpoint, 'GET');
+    return parseResponseBody<PendingTransfer[]>(res);
+  }
+
   public getRoutes = async (amount: number, destination: string, currency: string) => {
     // a query routes call is not currently provided by raiden
 
@@ -417,8 +500,8 @@ class RaidenClient extends SwapClient {
    */
   private tokenPayment = async (payload: TokenPaymentRequest): Promise<TokenPaymentResponse> => {
     const endpoint = `payments/${payload.token_address}/${payload.target_address}`;
-    payload.identifier = Math.round(Math.random() * (Number.MAX_SAFE_INTEGER - 1) + 1);
     if (payload.secret_hash) {
+      payload.identifier = RaidenClient.getIdentifier(payload.secret_hash);
       payload.secret_hash = `0x${payload.secret_hash}`;
     }
 

lib/swaps/SwapClient.ts

@@ -24,6 +24,17 @@ export type SwapClientInfo = {
   newUris?: string[];
 };
 
+export enum PaymentState {
+  Succeeded,
+  Failed,
+  Pending,
+}
+
+export type PaymentStatus = {
+  state: PaymentState,
+  preimage?: string,
+};
+
 interface SwapClient {
   on(event: 'connectionVerified', listener: (swapClientInfo: SwapClientInfo) => void): this;
   emit(event: 'connectionVerified', swapClientInfo: SwapClientInfo): boolean;
@@ -122,6 +133,12 @@ abstract class SwapClient extends EventEmitter {
 
   public abstract async removeInvoice(rHash: string): Promise<void>;
 
+  /**
+   * Checks to see whether we've made a payment using a given rHash.
+   * @returns the preimage for the payment, or `undefined` if no payment was made
+   */
+  public abstract async lookupPayment(rHash: string, currency?: string, destination?: string): Promise<PaymentStatus>;
+
   /**
    * Gets the block height of the chain backing this swap client.
    */

lib/swaps/SwapRecovery.ts

@@ -0,0 +1,118 @@
+import SwapClientManager from './SwapClientManager';
+import { SwapDealInstance } from '../db/types';
+import Logger from '../Logger';
+import { SwapPhase, SwapState, SwapFailureReason, SwapRole, SwapClientType } from '../constants/enums';
+import SwapClient, { PaymentState } from './SwapClient';
+
+/**
+ * A class that's responsible for recovering swap deals that were interrupted due to a system or xud crash,
+ * ensuring that we do not lose funds on a partially completed swap.
+ */
+class SwapRecovery {
+  /** A set of swaps where we have a pending outgoing payment for swaps where we don't know the preimage. */
+  public pendingSwaps: Set<SwapDealInstance> = new Set();
+  private pendingSwapsTimer?: NodeJS.Timeout;
+
+  constructor(private swapClientManager: SwapClientManager, private logger: Logger) { }
+
+  public beginTimer = () => {
+    if (!this.pendingSwapsTimer) {
+      this.pendingSwapsTimer = setInterval(this.checkPendingSwaps, 300000);
+    }
+  }
+
+  private checkPendingSwaps = () => {
+    this.pendingSwaps.forEach(pendingSwap => this.recoverDeal(pendingSwap).catch(this.logger.error));
+  }
+
+  public stopTimer = () => {
+    if (this.pendingSwapsTimer) {
+      clearInterval(this.pendingSwapsTimer);
+      this.pendingSwapsTimer = undefined;
+    }
+  }
+
+  private failDeal = async (deal: SwapDealInstance, receivingSwapClient?: SwapClient) => {
+    if (receivingSwapClient) {
+      try {
+        await receivingSwapClient.removeInvoice(deal.rHash);
+      } catch (err) {}
+    }
+    deal.state = SwapState.Error;
+    deal.failureReason = SwapFailureReason.Crash;
+    this.pendingSwaps.delete(deal);
+    await deal.save();
+  }
+
+  public recoverDeal = async (deal: SwapDealInstance) => {
+    const makerSwapClient = this.swapClientManager.get(deal.makerCurrency);
+    const takerSwapClient = this.swapClientManager.get(deal.takerCurrency);
+    if (!makerSwapClient || !makerSwapClient.isConnected()) {
+      this.logger.warn(`could not recover deal ${deal.rHash} because ${deal.makerCurrency} swap client is offline`);
+      this.pendingSwaps.add(deal);
+      return;
+    }
+    if (!takerSwapClient || !takerSwapClient.isConnected()) {
+      this.logger.warn(`could not recover deal ${deal.rHash} because ${deal.takerCurrency} swap client is offline`);
+      this.pendingSwaps.add(deal);
+      return;
+    }
+
+    this.logger.info(`recovering swap deal ${deal.rHash}`);
+    switch (deal.phase) {
+      case SwapPhase.SwapAccepted:
+        // we accepted the deal but stopped before sending payment
+        // cancel the open invoice if we have one
+        await this.failDeal(deal, makerSwapClient);
+        break;
+      case SwapPhase.SendingPayment:
+        // we started sending payment but didn't claim our payment
+        if (deal.role === SwapRole.Maker) {
+          // we should check to see if our payment went through
+          // if it did, we can claim payment with the preimage for our side of the swap
+          const paymentStatus = await takerSwapClient.lookupPayment(deal.rHash);
+          if (paymentStatus.state === PaymentState.Succeeded) {
+            try {
+              deal.rPreimage = paymentStatus.preimage!;
+              if (makerSwapClient.type === SwapClientType.Raiden) {
+                // tslint:disable-next-line: max-line-length
+                this.logger.warn(`cannot claim payment on Raiden for swap ${deal.rHash} using preimage ${deal.rPreimage}, this should be investigated manually`);
+              } else {
+                await makerSwapClient.settleInvoice(deal.rHash, deal.rPreimage);
+                this.logger.info(`recovered ${deal.makerCurrency} swap payment of ${deal.makerAmount} using preimage ${deal.rPreimage}`);
+              }
+              deal.state = SwapState.Recovered;
+              this.pendingSwaps.delete(deal);
+              await deal.save();
+              // TODO: update order and trade in database to indicate they were executed
+            } catch (err) {
+              // tslint:disable-next-line: max-line-length
+              this.logger.error(`could not settle ${deal.makerCurrency} invoice for payment ${deal.rHash} and preimage ${deal.rPreimage}, this should be investigated manually`, err);
+              await this.failDeal(deal);
+            }
+          } else if (paymentStatus.state === PaymentState.Failed) {
+            // the payment failed, so cancel the open invoice if we have one
+            await this.failDeal(deal, makerSwapClient);
+          } else {
+            // the payment is pending, we will need to follow up on this
+            this.pendingSwaps.add(deal);
+          }
+        } else if (deal.role === SwapRole.Taker) {
+          // we are not at risk of losing funds, but we should cancel any open invoices
+          await this.failDeal(deal, takerSwapClient);
+        }
+        break;
+      case SwapPhase.PaymentReceived:
+        // we've claimed our payment
+        // TODO: send a swap completed packet? it may be too late to do so
+        deal.state = SwapState.Recovered;
+        await deal.save();
+        break;
+      default:
+        break;
+    }
+
+  }
+}
+
+export default SwapRecovery;