Implement tx confirmation for channelOpen

.circleci/config.yml

@@ -113,7 +113,7 @@ jobs:
       - attach_workspace: *attach_options
       - run:
           name: Run unit tests
-          command: npm run test -- --ci --runInBand --reporters=default --reporters=jest-junit --detectOpenHandles
+          command: npm run test -- --ci --runInBand --reporters=default --reporters=jest-junit
           environment:
             JEST_JUNIT_OUTPUT_DIR: "reports/junit"
       - store_test_results:

raiden-ts/src/actions.ts

@@ -57,3 +57,25 @@ export const RaidenEvents = [
 ];
 /* Tagged union of RaidenEvents actions */
 export type RaidenEvent = ActionType<typeof RaidenEvents>;
+
+/**
+ * Set of [serializable] actions which are first emitted with
+ * payload.confirmed=undefined, then, after confirmation blocks, either with confirmed=true if tx
+ * is still present on blockchain, or confirmed=false if it got removed by a reorg.
+ *
+ * These actions must comply with the following type:
+ * {
+ *   payload: {
+ *     txHash: Hash;
+ *     txBlock: number;
+ *     confirmed: undefined | boolean;
+ *   };
+ *   meta: any;
+ * }
+ */
+export const ConfirmableActions = [ChannelsActions.channelOpen.success];
+/**
+ * Union of codecs of actions above
+ */
+export const ConfirmableAction = ChannelsActions.channelOpen.success.codec;
+export type ConfirmableAction = ChannelsActions.channelOpen.success;

raiden-ts/src/channels/actions.ts

@@ -46,9 +46,10 @@ export const channelOpen = createAsyncAction(
   t.type({
     id: t.number,
     settleTimeout: t.number,
-    openBlock: t.number,
     isFirstParticipant: t.boolean,
     txHash: Hash,
+    txBlock: t.number,
+    confirmed: t.union([t.undefined, t.boolean]),
   }),
 );
 export namespace channelOpen {

raiden-ts/src/channels/epics.ts

@@ -1,4 +1,14 @@
-import { Observable, from, of, EMPTY, merge, interval, defer, concat as concat$ } from 'rxjs';
+import {
+  Observable,
+  from,
+  of,
+  EMPTY,
+  merge,
+  interval,
+  defer,
+  concat as concat$,
+  combineLatest,
+} from 'rxjs';
 import {
   catchError,
   filter,
@@ -13,6 +23,7 @@ import {
   first,
   take,
   mapTo,
+  pluck,
 } from 'rxjs/operators';
 import { findKey, get, isEmpty, negate } from 'lodash';
 
@@ -22,13 +33,14 @@ import { HashZero, Zero } from 'ethers/constants';
 import { Filter } from 'ethers/providers';
 
 import { RaidenEpicDeps } from '../types';
-import { RaidenAction, raidenShutdown } from '../actions';
+import { RaidenAction, raidenShutdown, ConfirmableAction } from '../actions';
 import { Channel, ChannelState } from '../channels';
 import { RaidenState } from '../state';
 import { SignatureZero, ShutdownReason } from '../constants';
 import { chooseOnchainAccount, getContractWithSigner } from '../helpers';
-import { Address, Hash, UInt, Signature } from '../utils/types';
+import { Address, Hash, UInt, Signature, isntNil } from '../utils/types';
 import { isActionOf } from '../utils/actions';
+import { pluckDistinct } from '../utils/rx';
 import { fromEthersEvent, getEventsStream, getNetwork } from '../utils/ethers';
 import { encode } from '../utils/data';
 import {
@@ -256,9 +268,10 @@ export const tokenMonitoredEpic = (
                 {
                   id: id.toNumber(),
                   settleTimeout: settleTimeout.toNumber(),
-                  openBlock: event.blockNumber!,
                   isFirstParticipant: address === p1,
                   txHash: event.transactionHash! as Hash,
+                  txBlock: event.blockNumber!,
+                  confirmed: undefined,
                 },
                 {
                   tokenNetwork: tokenNetworkContract.address as Address,
@@ -518,17 +531,16 @@ export const channelOpenedEpic = (
     withLatestFrom(state$),
     // proceed only if channel is in 'open' state and a deposit is required
     filter(([action, state]) => {
-      const channel: Channel | undefined = get(state.channels, [
-        action.meta.tokenNetwork,
-        action.meta.partner,
-      ]);
-      return !!channel && channel.state === ChannelState.open;
+      const channel = state.channels[action.meta.tokenNetwork]?.[action.meta.partner];
+      // filter only after tx confirmed & channel persisted in state
+      return !!action.payload.confirmed && !!channel && channel.state === ChannelState.open;
     }),
     map(([action]) =>
       channelMonitor(
         {
           id: action.payload.id,
-          fromBlock: action.payload.openBlock, // fetch past events as well, if needed
+          // fetch past events as well, if needed, including events before confirmation
+          fromBlock: action.payload.txBlock,
         },
         action.meta,
       ),
@@ -878,3 +890,60 @@ export const channelSettleableEpic = (
       }
     }),
   );
+
+function checkPendingAction(
+  action: ConfirmableAction,
+  provider: RaidenEpicDeps['provider'],
+  blockNumber: number,
+  confirmationBlocks: number,
+): Observable<ConfirmableAction> {
+  return defer(() => provider.getTransactionReceipt(action.payload.txHash)).pipe(
+    map(receipt => {
+      if (receipt?.confirmations !== undefined && receipt.confirmations >= confirmationBlocks) {
+        return {
+          ...action,
+          // beyond setting confirmed, also re-set blockNumber,
+          // which may have changed on a reorg
+          payload: { ...action.payload, txBlock: receipt.blockNumber!, confirmed: true },
+        };
+      } else if (action.payload.txBlock + 2 * confirmationBlocks < blockNumber) {
+        // if this txs didn't get confirmed for more than 2*confirmationBlocks, it was removed
+        return {
+          ...action,
+          payload: { ...action.payload, confirmed: false },
+        };
+      } // else, it seems removed, but give it twice confirmationBlocks to be picked up again
+    }),
+    filter(isntNil),
+  );
+}
+
+/**
+ * Process new blocks and re-emit confirmed or removed actions
+ *
+ * @param action$ - Observable of channelSettle actions
+ * @param state$ - Observable of RaidenStates
+ * @param deps - RaidenEpicDeps
+ * @param deps.config$,deps.provider - RaidenEpicDeps members
+ * @returns Observable of confirmed or removed actions
+ */
+export const confirmationEpic = (
+  {}: Observable<RaidenAction>,
+  state$: Observable<RaidenState>,
+  { config$, provider }: RaidenEpicDeps,
+): Observable<ConfirmableAction> =>
+  combineLatest(
+    state$.pipe(pluckDistinct('blockNumber')),
+    state$.pipe(pluck('pendingTxs')),
+    config$.pipe(pluckDistinct('confirmationBlocks')),
+  ).pipe(
+    // exhaust will ignore blocks while concat$ is busy
+    exhaustMap(([blockNumber, pendingTxs, confirmationBlocks]) =>
+      concat$(
+        ...pendingTxs
+          // only txs/confirmable actions which are more than confirmationBlocks in the past
+          .filter(a => a.payload.txBlock + confirmationBlocks <= blockNumber)
+          .map(action => checkPendingAction(action, provider, blockNumber, confirmationBlocks)),
+      ),
+    ),
+  );

raiden-ts/src/channels/reducer.ts

@@ -1,12 +1,12 @@
-import { get, set, unset } from 'lodash/fp';
+import { get, set, unset, getOr } from 'lodash/fp';
 import { Zero } from 'ethers/constants';
 import { Reducer } from 'redux';
 
 import { UInt } from '../utils/types';
-import { createReducer } from '../utils/actions';
+import { createReducer, matchMeta, isActionOf } from '../utils/actions';
 import { partialCombineReducers } from '../utils/redux';
 import { RaidenState, initialState } from '../state';
-import { RaidenAction } from '../actions';
+import { RaidenAction, ConfirmableActions } from '../actions';
 import {
   channelClose,
   channelDeposit,
@@ -50,17 +50,22 @@ function channelOpenSuccessReducer(
   state: RaidenState['channels'],
   action: channelOpen.success,
 ): RaidenState['channels'] {
-  const path = [action.meta.tokenNetwork, action.meta.partner],
-    channel: Channel = {
-      state: ChannelState.open,
-      own: { deposit: Zero as UInt<32> },
-      partner: { deposit: Zero as UInt<32> },
-      id: action.payload.id,
-      settleTimeout: action.payload.settleTimeout,
-      openBlock: action.payload.openBlock,
-      isFirstParticipant: action.payload.isFirstParticipant,
-      /* txHash: action.txHash, */ // not needed in state for now, but comes in action
-    };
+  const path = [action.meta.tokenNetwork, action.meta.partner];
+  // ignore if older than currently set channel, or unconfirmed or removed
+  if (
+    getOr(0, [...path, 'openBlock'], state) >= action.payload.txBlock ||
+    !action.payload.confirmed
+  )
+    return state;
+  const channel: Channel = {
+    state: ChannelState.open,
+    own: { deposit: Zero as UInt<32> },
+    partner: { deposit: Zero as UInt<32> },
+    id: action.payload.id,
+    settleTimeout: action.payload.settleTimeout,
+    isFirstParticipant: action.payload.isFirstParticipant,
+    openBlock: action.payload.txBlock,
+  };
   return set(path, channel, state);
 }
 
@@ -169,13 +174,25 @@ const channels: Reducer<RaidenState['channels'], RaidenAction> = createReducer(
   .handle(channelClose.success, channelCloseSuccessReducer)
   .handle(channelSettle.success, channelSettleSuccessReducer);
 
+const pendingTxs: Reducer<RaidenState['pendingTxs'], RaidenAction> = (
+  state = initialState.pendingTxs,
+  action: RaidenAction,
+): RaidenState['pendingTxs'] => {
+  // filter out non-ConfirmableActions's
+  if (!isActionOf(ConfirmableActions, action)) return state;
+  // if confirmed==undefined, add action to state
+  else if (action.payload.confirmed === undefined) return [...state, action];
+  // else (either confirmed or removed), remove from state
+  else return state.filter(a => a.type !== action.type || !matchMeta(action.meta, a));
+};
+
 /**
  * Nested/combined reducer for channels
  * blockNumber, tokens & channels reducers get its own slice of the state, corresponding to the
  * name of the reducer. channels root reducer instead must be handled the complete state instead,
  * so it compose the output with each key/nested/combined state.
  */
 export const channelsReducer = partialCombineReducers(
-  { blockNumber, tokens, channels },
+  { blockNumber, tokens, channels, pendingTxs },
   initialState,
 );

raiden-ts/src/config.ts

@@ -4,6 +4,15 @@ import { Network } from 'ethers/utils';
 import { Address } from './utils/types';
 import { getNetworkName } from './utils/ethers';
 
+const logLevels = t.keyof({
+  ['']: null,
+  trace: null,
+  debug: null,
+  info: null,
+  warn: null,
+  error: null,
+});
+
 /**
  * A Raiden configuration object with required parameters and
  * optional parameters from [[PartialRaidenConfig]].
@@ -18,22 +27,15 @@ import { getNetworkName } from './utils/ethers';
  * - pfsSafetyMargin - Safety margin to be added to fees received from PFS. Use `1.1` to add a 10% safety margin.
  * - matrixExcessRooms - Keep this much rooms for a single user of interest (partner, target).
  *                       Leave LRU beyond this threshold.
+ * - confirmationBlocks - How many blocks to wait before considering a transaction as confirmed
  * - matrixServer? - Specify a matrix server to use.
  * - logger? - String specifying the console log level of redux-logger. Use '' to disable.
  *             Defaults to 'debug' if undefined and process.env.NODE_ENV === 'development'
- * - pfs - Path Finding Service URL or Address. Set to null to disable, or leave undefined to
- *             enable automatic fetching from ServiceRegistry.
- * - subkey - When using subkey, this sets the behavior when { subkey } option isn't explicitly set
- *            in on-chain method calls. false (default) = use main key; true = use subkey
+ * - pfs? - Path Finding Service URL or Address. Set to null to disable, or leave undefined to
+ *          enable automatic fetching from ServiceRegistry.
+ * - subkey? - When using subkey, this sets the behavior when { subkey } option isn't explicitly set
+ *             in on-chain method calls. false (default) = use main key; true = use subkey
  */
-const logLevels = t.keyof({
-  ['']: null,
-  trace: null,
-  debug: null,
-  info: null,
-  warn: null,
-  error: null,
-});
 export const RaidenConfig = t.readonly(
   t.intersection([
     t.type({
@@ -45,6 +47,7 @@ export const RaidenConfig = t.readonly(
       pfsRoom: t.union([t.string, t.null]),
       pfsSafetyMargin: t.number,
       matrixExcessRooms: t.number,
+      confirmationBlocks: t.number,
     }),
     t.partial({
       matrixServer: t.string,
@@ -87,5 +90,6 @@ export function makeDefaultConfig({ network }: { network: Network }): RaidenConf
     pfsRoom: `raiden_${getNetworkName(network)}_path_finding`,
     matrixExcessRooms: 3,
     pfsSafetyMargin: 1.0,
+    confirmationBlocks: 5,
   };
 }

raiden-ts/src/path/utils.ts

@@ -41,7 +41,7 @@ export function channelCanRoute(
     return `path: channel with "${partner}" in state "${channel.state}" instead of "${ChannelState.open}"`;
   const { ownCapacity: capacity } = channelAmounts(channel);
   if (capacity.lt(value))
-    return `path: channel with "${partner}" don't have enough capacity=${capacity.toString()}`;
+    return `path: channel with "${partner}" doesn't have enough capacity=${capacity.toString()}`;
   return true;
 }
 

raiden-ts/src/raiden.ts

@@ -520,7 +520,8 @@ export class Raiden {
     assert(!options.subkey || this.deps.main, "Can't send tx from subkey if not set");
 
     const meta = { tokenNetwork, partner };
-    const promise = asyncActionToPromise(channelOpen, meta, this.action$).then(
+    // wait for confirmation
+    const promise = asyncActionToPromise(channelOpen, meta, this.action$, true).then(
       ({ txHash }) => txHash, // pluck txHash
     );
     this.store.dispatch(channelOpen.request(options, meta));

raiden-ts/src/state.ts

@@ -5,6 +5,7 @@ import { debounce, merge as _merge } from 'lodash';
 
 import { PartialRaidenConfig } from './config';
 import { ContractsInfo } from './types';
+import { ConfirmableAction } from './actions';
 import { losslessParse, losslessStringify } from './utils/data';
 import { Address, Secret, decode, Signed, Storage } from './utils/types';
 import { Channel } from './channels/state';
@@ -61,6 +62,7 @@ export const RaidenState = t.readonly(
         ),
       ),
     }),
+    pendingTxs: t.readonlyArray(ConfirmableAction),
   }),
 );
 
@@ -131,6 +133,7 @@ export function makeInitialState(
     path: {
       iou: {},
     },
+    pendingTxs: [],
   };
 }