Merge pull request #483 from gre/sync-priorities

Improve how Sync happen

package.json

@@ -43,6 +43,7 @@
     "@ledgerhq/hw-transport-node-hid": "^4.13.0",
     "@ledgerhq/ledger-core": "1.4.5",
     "@ledgerhq/live-common": "2.29.0",
+    "async": "^2.6.1",
     "axios": "^0.18.0",
     "babel-runtime": "^6.26.0",
     "bcryptjs": "^2.4.3",

src/actions/bridgeSync.js

@@ -6,9 +6,3 @@ export const setAccountSyncState = (accountId: string, state: AsyncState) => ({
   accountId,
   state,
 })
-
-export const setAccountPullMoreState = (accountId: string, state: AsyncState) => ({
-  type: 'SET_ACCOUNT_PULL_MORE_STATE',
-  accountId,
-  state,
-})

src/bridge/BridgeSyncContext.js

@@ -1,28 +1,23 @@
 // @flow
+// Unify the synchronization management for bridges with the redux store
+// it handles automatically re-calling synchronize
+// this is an even high abstraction than the bridge
+
 import logger from 'logger'
+import shuffle from 'lodash/shuffle'
 import React, { Component } from 'react'
+import priorityQueue from 'async/priorityQueue'
 import { connect } from 'react-redux'
 import type { Account } from '@ledgerhq/live-common/lib/types'
 import { createStructuredSelector } from 'reselect'
 import { updateAccountWithUpdater } from 'actions/accounts'
-import { setAccountSyncState, setAccountPullMoreState } from 'actions/bridgeSync'
+import { setAccountSyncState } from 'actions/bridgeSync'
-import {
+import { bridgeSyncSelector, syncStateLocalSelector } from 'reducers/bridgeSync'
-  bridgeSyncSelector,
-  syncStateLocalSelector,
-  pullMoreStateLocalSelector,
-} from 'reducers/bridgeSync'
 import type { BridgeSyncState } from 'reducers/bridgeSync'
 import { accountsSelector } from 'reducers/accounts'
-import { SYNC_BOOT_DELAY, SYNC_INTERVAL } from 'config/constants'
+import { SYNC_BOOT_DELAY, SYNC_ALL_INTERVAL } from 'config/constants'
 import { getBridgeForCurrency } from '.'
 
-// Unify the synchronization management for bridges with the redux store
-// it handles automatically re-calling synchronize
-// this is an even high abstraction than the bridge
-
-// $FlowFixMe can't wait flow implement createContext
-const BridgeSyncContext = React.createContext(() => {})
-
 type BridgeSyncProviderProps = {
   children: *,
 }
@@ -32,7 +27,6 @@ type BridgeSyncProviderOwnProps = BridgeSyncProviderProps & {
   accounts: Account[],
   updateAccountWithUpdater: (string, (Account) => Account) => void,
   setAccountSyncState: (string, AsyncState) => *,
-  setAccountPullMoreState: (string, AsyncState) => *,
 }
 
 type AsyncState = {
@@ -40,15 +34,15 @@ type AsyncState = {
   error: ?Error,
 }
 
-type BridgeSync = {
+export type BehaviorAction =
-  synchronize: (accountId: string) => Promise<void>,
+  | { type: 'BACKGROUND_TICK' }
+  | { type: 'SET_SKIP_UNDER_PRIORITY', priority: number }
+  | { type: 'SYNC_ONE_ACCOUNT', accountId: string, priority: number }
+  | { type: 'SYNC_ALL_ACCOUNTS', priority: number }
 
-  // sync for all accounts (if there were errors it stopped)
+export type Sync = (action: BehaviorAction) => void
-  syncAll: () => {},
 
-  //
+const BridgeSyncContext = React.createContext((_: BehaviorAction) => {})
-  pullMoreOperations: (accountId: string, count: number) => Promise<void>,
-}
 
 const mapStateToProps = createStructuredSelector({
   accounts: accountsSelector,
@@ -58,110 +52,104 @@ const mapStateToProps = createStructuredSelector({
 const actions = {
   updateAccountWithUpdater,
   setAccountSyncState,
-  setAccountPullMoreState,
 }
 
-class Provider extends Component<BridgeSyncProviderOwnProps, BridgeSync> {
+class Provider extends Component<BridgeSyncProviderOwnProps, Sync> {
   constructor() {
     super()
-    const syncPromises = {}
-    const syncSubs = {}
-    const pullMorePromises = {}
 
-    const getSyncState = accountId => syncStateLocalSelector(this.props.bridgeSync, { accountId })
+    const synchronize = (accountId: string, next: () => void) => {
-
+      const state = syncStateLocalSelector(this.props.bridgeSync, { accountId })
-    const getPullMoreOperationsState = accountId =>
+      if (state.pending) {
-      pullMoreStateLocalSelector(this.props.bridgeSync, { accountId })
+        next()
+        return
+      }
+      const account = this.props.accounts.find(a => a.id === accountId)
+      if (!account) throw new Error('account not found')
 
-    const getAccountById = accountId => {
+      const bridge = getBridgeForCurrency(account.currency)
-      const a = this.props.accounts.find(a => a.id === accountId)
-      if (!a) throw new Error('account not found')
-      return a
-    }
 
-    const getBridgeForAccountId = accountId =>
+      this.props.setAccountSyncState(accountId, { pending: true, error: null })
-      getBridgeForCurrency(getAccountById(accountId).currency)
 
-    const pullMoreOperations = (accountId, count) => {
+      // TODO use Subscription to unsubscribe at relevant time
-      const state = getPullMoreOperationsState(accountId)
+      bridge.synchronize(account).subscribe({
-      if (state.pending) {
+        next: accountUpdater => {
-        return (
-          pullMorePromises[accountId] || Promise.reject(new Error('no pullMore started. (bug)'))
-        )
-      }
-      this.props.setAccountPullMoreState(accountId, { pending: true, error: null })
-      const bridge = getBridgeForAccountId(accountId)
-      const p = bridge.pullMoreOperations(getAccountById(accountId), count).then(
-        accountUpdater => {
-          this.props.setAccountPullMoreState(accountId, {
-            pending: false,
-            error: null,
-          })
           this.props.updateAccountWithUpdater(accountId, accountUpdater)
         },
-        error => {
+        complete: () => {
-          this.props.setAccountPullMoreState(accountId, {
+          this.props.setAccountSyncState(accountId, { pending: false, error: null })
-            pending: false,
+          next()
-            error,
-          })
         },
-      )
+        error: error => {
-      pullMorePromises[accountId] = p
+          this.props.setAccountSyncState(accountId, { pending: false, error })
-      return p
+          next()
+        },
+      })
     }
 
-    const synchronize = accountId => {
+    const syncQueue = priorityQueue(synchronize, 2)
-      const state = getSyncState(accountId)
-      if (state.pending) {
-        return syncPromises[accountId] || Promise.reject(new Error('no sync started. (bug)'))
-      }
 
-      this.props.setAccountSyncState(accountId, { pending: true, error: null })
+    let skipUnderPriority: number = -1
-      const bridge = getBridgeForAccountId(accountId)
+
-      const p = new Promise((resolve, reject) => {
+    const schedule = (ids: string[], priority: number) => {
-        const subscription = bridge.synchronize(getAccountById(accountId), {
+      if (priority < skipUnderPriority) return
-          next: accountUpdater => {
+      // by convention we remove concurrent tasks with same priority
-            this.props.updateAccountWithUpdater(accountId, accountUpdater)
+      syncQueue.remove(o => priority === o.priority)
-          },
+      syncQueue.push(ids, -priority)
-          complete: () => {
-            this.props.setAccountSyncState(accountId, { pending: false, error: null })
-            resolve()
-          },
-          error: error => {
-            this.props.setAccountSyncState(accountId, { pending: false, error })
-            reject(error)
-          },
-        })
-        syncSubs[accountId] = subscription
-      })
-      syncPromises[accountId] = p
-      return p
     }
 
-    const syncAll = () => Promise.all(this.props.accounts.map(account => synchronize(account.id)))
+    // don't always sync in same order to avoid potential "never account never reached"
+    const shuffledAccountIds = () => shuffle(this.props.accounts.map(a => a.id))
+
+    const handlers = {
+      BACKGROUND_TICK: () => {
+        if (syncQueue.idle()) {
+          schedule(shuffledAccountIds(), -1)
+        }
+      },
+
+      SET_SKIP_UNDER_PRIORITY: ({ priority }) => {
+        if (priority === skipUnderPriority) return
+        skipUnderPriority = priority
+        syncQueue.remove(({ priority }) => priority < skipUnderPriority)
+      },
+
+      SYNC_ALL_ACCOUNTS: ({ priority }) => {
+        schedule(shuffledAccountIds(), priority)
+      },
+
+      SYNC_ONE_ACCOUNT: ({ accountId, priority }) => {
+        schedule([accountId], priority)
+      },
+    }
 
-    this.api = {
+    const sync = (action: BehaviorAction) => {
-      synchronize,
+      const handler = handlers[action.type]
-      syncAll,
+      if (handler) {
-      pullMoreOperations,
+        // $FlowFixMe
+        handler(action)
+      } else {
+        logger.warn('BridgeSyncContext unsupported action', action)
+      }
     }
+
+    this.api = sync
   }
 
   componentDidMount() {
     const syncLoop = async () => {
-      try {
+      this.api({ type: 'BACKGROUND_TICK' })
-        await this.api.syncAll()
+      this.syncTimeout = setTimeout(syncLoop, SYNC_ALL_INTERVAL)
-      } catch (e) {
-        logger.error('sync issues', e)
-      }
-      setTimeout(syncLoop, SYNC_INTERVAL)
     }
-    setTimeout(syncLoop, SYNC_BOOT_DELAY)
+    this.syncTimeout = setTimeout(syncLoop, SYNC_BOOT_DELAY)
   }
 
-  // TODO we might want to call sync straight away when new accounts got added (it will happen every 10s anyway)
+  componentWillUnmount() {
+    clearTimeout(this.syncTimeout)
+  }
 
-  api: BridgeSync
+  syncTimeout: *
+  api: Sync
 
   render() {
     return (

src/bridge/EthereumJSBridge.js

@@ -268,69 +268,68 @@ const EthereumBridge: WalletBridge<Transaction> = {
     return { unsubscribe }
   },
 
-  synchronize({ freshAddress, blockHeight, currency, operations }, { next, complete, error }) {
+  synchronize: ({ freshAddress, blockHeight, currency, operations }) =>
-    let unsubscribed = false
+    Observable.create(o => {
-    const api = apiForCurrency(currency)
+      let unsubscribed = false
-    async function main() {
+      const api = apiForCurrency(currency)
-      try {
+      async function main() {
-        const block = await fetchCurrentBlock(currency)
+        try {
-        if (unsubscribed) return
+          const block = await fetchCurrentBlock(currency)
-        if (block.height === blockHeight) {
-          complete()
-        } else {
-          const filterConfirmedOperations = o =>
-            o.blockHeight && blockHeight - o.blockHeight > SAFE_REORG_THRESHOLD
-
-          operations = operations.filter(filterConfirmedOperations)
-          const blockHash = operations.length > 0 ? operations[0].blockHash : undefined
-          const { txs } = await api.getTransactions(freshAddress, blockHash)
-          if (unsubscribed) return
-          if (txs.length === 0) {
-            next(a => ({
-              ...a,
-              blockHeight: block.height,
-              lastSyncDate: new Date(),
-            }))
-            complete()
-            return
-          }
-          const balance = await api.getAccountBalance(freshAddress)
-          if (unsubscribed) return
-          const nonce = await api.getAccountNonce(freshAddress)
           if (unsubscribed) return
-          next(a => {
+          if (block.height === blockHeight) {
-            const currentOps = a.operations.filter(filterConfirmedOperations)
+            o.complete()
-            const newOps = flatMap(txs, txToOps(a))
+          } else {
-            const operations = mergeOps(currentOps, newOps)
+            const filterConfirmedOperations = o =>
-            const pendingOperations = a.pendingOperations.filter(
+              o.blockHeight && blockHeight - o.blockHeight > SAFE_REORG_THRESHOLD
-              o =>
+
-                o.transactionSequenceNumber &&
+            operations = operations.filter(filterConfirmedOperations)
-                o.transactionSequenceNumber >= nonce &&
+            const blockHash = operations.length > 0 ? operations[0].blockHash : undefined
-                !operations.some(op => o.hash === op.hash),
+            const { txs } = await api.getTransactions(freshAddress, blockHash)
-            )
+            if (unsubscribed) return
-            return {
+            if (txs.length === 0) {
-              ...a,
+              o.next(a => ({
-              pendingOperations,
+                ...a,
-              operations,
+                blockHeight: block.height,
-              balance,
+                lastSyncDate: new Date(),
-              blockHeight: block.height,
+              }))
-              lastSyncDate: new Date(),
+              o.complete()
+              return
             }
-          })
+            const balance = await api.getAccountBalance(freshAddress)
-          complete()
+            if (unsubscribed) return
+            const nonce = await api.getAccountNonce(freshAddress)
+            if (unsubscribed) return
+            o.next(a => {
+              const currentOps = a.operations.filter(filterConfirmedOperations)
+              const newOps = flatMap(txs, txToOps(a))
+              const operations = mergeOps(currentOps, newOps)
+              const pendingOperations = a.pendingOperations.filter(
+                o =>
+                  o.transactionSequenceNumber &&
+                  o.transactionSequenceNumber >= nonce &&
+                  !operations.some(op => o.hash === op.hash),
+              )
+              return {
+                ...a,
+                pendingOperations,
+                operations,
+                balance,
+                blockHeight: block.height,
+                lastSyncDate: new Date(),
+              }
+            })
+            o.complete()
+          }
+        } catch (e) {
+          o.error(e)
         }
-      } catch (e) {
-        error(e)
       }
-    }
+      main()
-    main()
 
-    return {
+      return () => {
-      unsubscribe() {
         unsubscribed = true
-      },
+      }
-    }
+    }),
-  },
 
   pullMoreOperations: () => Promise.resolve(a => a), // NOT IMPLEMENTED
 
@@ -359,10 +358,8 @@ const EthereumBridge: WalletBridge<Transaction> = {
 
   isValidTransaction: (a, t) => (t.amount > 0 && t.recipient && true) || false,
 
-  // $FlowFixMe
   EditFees,
 
-  // $FlowFixMe
   EditAdvancedOptions,
 
   canBeSpent: (a, t) => Promise.resolve(t.amount <= a.balance),

src/bridge/LibcoreBridge.js

@@ -1,6 +1,7 @@
 // @flow
 import logger from 'logger'
 import React from 'react'
+import { Observable } from 'rxjs'
 import { map } from 'rxjs/operators'
 import type { Account } from '@ledgerhq/live-common/lib/types'
 import { decodeAccount, encodeAccount } from 'reducers/accounts'
@@ -54,53 +55,56 @@ const LibcoreBridge: WalletBridge<Transaction> = {
       .subscribe(observer)
   },
 
-  synchronize(account, { next, complete, error }) {
+  synchronize: account =>
-    // FIXME TODO:
+    Observable.create(o => {
-    // - when you implement addPendingOperation you also here need to:
+      // FIXME TODO:
-    //   - if there were pendingOperations that are now in operations, remove them as well.
+      // - when you implement addPendingOperation you also here need to:
-    //   - if there are pendingOperations that is older than a threshold (that depends on blockchain speed typically)
+      //   - if there were pendingOperations that are now in operations, remove them as well.
-    //     then we probably should trash them out? it's a complex question for UI
+      //   - if there are pendingOperations that is older than a threshold (that depends on blockchain speed typically)
-    ;(async () => {
+      //     then we probably should trash them out? it's a complex question for UI
-      try {
+      ;(async () => {
-        const rawAccount = encodeAccount(account)
+        try {
-        const rawSyncedAccount = await libcoreSyncAccount.send({ rawAccount }).toPromise()
+          const rawAccount = encodeAccount(account)
-        const syncedAccount = decodeAccount(rawSyncedAccount)
+          const rawSyncedAccount = await libcoreSyncAccount.send({ rawAccount }).toPromise()
-        next(account => {
+          const syncedAccount = decodeAccount(rawSyncedAccount)
-          const accountOps = account.operations
+          o.next(account => {
-          const syncedOps = syncedAccount.operations
+            const accountOps = account.operations
-          const patch: $Shape<Account> = {
+            const syncedOps = syncedAccount.operations
-            freshAddress: syncedAccount.freshAddress,
+            const patch: $Shape<Account> = {
-            freshAddressPath: syncedAccount.freshAddressPath,
+              freshAddress: syncedAccount.freshAddress,
-            balance: syncedAccount.balance,
+              freshAddressPath: syncedAccount.freshAddressPath,
-            blockHeight: syncedAccount.blockHeight,
+              balance: syncedAccount.balance,
-            lastSyncDate: new Date(),
+              blockHeight: syncedAccount.blockHeight,
-          }
+              lastSyncDate: new Date(),
-
+            }
-          const hasChanged =
+
-            accountOps.length !== syncedOps.length || // size change, we do a full refresh for now...
+            const hasChanged =
-            (accountOps.length > 0 && syncedOps.length > 0 && accountOps[0].id !== syncedOps[0].id) // if same size, only check if the last item has changed.
+              accountOps.length !== syncedOps.length || // size change, we do a full refresh for now...
-
+              (accountOps.length > 0 &&
-          if (hasChanged) {
+                syncedOps.length > 0 &&
-            patch.operations = syncedAccount.operations
+                accountOps[0].id !== syncedOps[0].id) // if same size, only check if the last item has changed.
-            patch.pendingOperations = [] // For now, we assume a change will clean the pendings.
+
-          }
+            if (hasChanged) {
-
+              patch.operations = syncedAccount.operations
-          return {
+              patch.pendingOperations = [] // For now, we assume a change will clean the pendings.
-            ...account,
+            }
-            ...patch,
+
-          }
+            return {
-        })
+              ...account,
-        complete()
+              ...patch,
-      } catch (e) {
+            }
-        error(e)
+          })
+          o.complete()
+        } catch (e) {
+          o.error(e)
+        }
+      })()
+      return {
+        unsubscribe() {
+          logger.warn('LibcoreBridge: unsub sync not implemented')
+        },
       }
-    })()
+    }),
-    return {
-      unsubscribe() {
-        logger.warn('LibcoreBridge: unsub sync not implemented')
-      },
-    }
-  },
 
   pullMoreOperations: () => Promise.reject(notImplemented),
 

src/bridge/RippleJSBridge.js

@@ -340,85 +340,86 @@ const RippleJSBridge: WalletBridge<Transaction> = {
     return { unsubscribe }
   },
 
-  synchronize({ currency, freshAddress, blockHeight }, { next, error, complete }) {
+  synchronize: ({ currency, freshAddress, blockHeight }) =>
-    let finished = false
+    Observable.create(o => {
-    const unsubscribe = () => {
+      let finished = false
-      finished = true
+      const unsubscribe = () => {
-    }
+        finished = true
-
+      }
-    async function main() {
-      const api = apiForCurrency(currency)
-      try {
-        await api.connect()
-        if (finished) return
-        const serverInfo = await getServerInfo(currency)
-        if (finished) return
-        const ledgers = serverInfo.completeLedgers.split('-')
-        const minLedgerVersion = Number(ledgers[0])
-        const maxLedgerVersion = Number(ledgers[1])
 
-        let info
+      async function main() {
+        const api = apiForCurrency(currency)
         try {
-          info = await api.getAccountInfo(freshAddress)
+          await api.connect()
-        } catch (e) {
+          if (finished) return
-          if (e.message !== 'actNotFound') {
+          const serverInfo = await getServerInfo(currency)
-            throw e
+          if (finished) return
+          const ledgers = serverInfo.completeLedgers.split('-')
+          const minLedgerVersion = Number(ledgers[0])
+          const maxLedgerVersion = Number(ledgers[1])
+
+          let info
+          try {
+            info = await api.getAccountInfo(freshAddress)
+          } catch (e) {
+            if (e.message !== 'actNotFound') {
+              throw e
+            }
           }
-        }
+          if (finished) return
-        if (finished) return
-
-        if (!info) {
-          // account does not exist, we have nothing to sync
-          complete()
-          return
-        }
 
-        const balance = parseAPIValue(info.xrpBalance)
+          if (!info) {
-        if (isNaN(balance) || !isFinite(balance)) {
+            // account does not exist, we have nothing to sync
-          throw new Error(`Ripple: invalid balance=${balance} for address ${freshAddress}`)
+            o.complete()
-        }
+            return
+          }
 
-        next(a => ({ ...a, balance }))
+          const balance = parseAPIValue(info.xrpBalance)
-
+          if (isNaN(balance) || !isFinite(balance)) {
-        const transactions = await api.getTransactions(freshAddress, {
+            throw new Error(`Ripple: invalid balance=${balance} for address ${freshAddress}`)
-          minLedgerVersion: Math.max(blockHeight, minLedgerVersion),
-          maxLedgerVersion,
-        })
-
-        if (finished) return
-
-        next(a => {
-          const newOps = transactions.map(txToOperation(a))
-          const operations = mergeOps(a.operations, newOps)
-          const [last] = operations
-          const pendingOperations = a.pendingOperations.filter(
-            o =>
-              last &&
-              last.transactionSequenceNumber &&
-              o.transactionSequenceNumber &&
-              o.transactionSequenceNumber > last.transactionSequenceNumber,
-          )
-          return {
-            ...a,
-            operations,
-            pendingOperations,
-            blockHeight: maxLedgerVersion,
-            lastSyncDate: new Date(),
           }
-        })
 
-        complete()
+          o.next(a => ({ ...a, balance }))
-      } catch (e) {
+
-        error(e)
+          const transactions = await api.getTransactions(freshAddress, {
-      } finally {
+            minLedgerVersion: Math.max(blockHeight, minLedgerVersion),
-        api.disconnect()
+            maxLedgerVersion,
+          })
+
+          if (finished) return
+
+          o.next(a => {
+            const newOps = transactions.map(txToOperation(a))
+            const operations = mergeOps(a.operations, newOps)
+            const [last] = operations
+            const pendingOperations = a.pendingOperations.filter(
+              o =>
+                last &&
+                last.transactionSequenceNumber &&
+                o.transactionSequenceNumber &&
+                o.transactionSequenceNumber > last.transactionSequenceNumber,
+            )
+            return {
+              ...a,
+              operations,
+              pendingOperations,
+              blockHeight: maxLedgerVersion,
+              lastSyncDate: new Date(),
+            }
+          })
+
+          o.complete()
+        } catch (e) {
+          o.error(e)
+        } finally {
+          api.disconnect()
+        }
       }
-    }
 
-    main()
+      main()
 
-    return { unsubscribe }
+      return unsubscribe
-  },
+    }),
 
   pullMoreOperations: () => Promise.resolve(a => a), // FIXME not implemented
 

src/bridge/UnsupportedBridge.js

@@ -5,10 +5,10 @@ import type { WalletBridge } from './types'
 const genericError = new Error('UnsupportedBridge')
 
 const UnsupportedBridge: WalletBridge<*> = {
-  synchronize(initialAccount, { error }) {
+  synchronize: () =>
-    Promise.resolve(genericError).then(error)
+    Observable.create(o => {
-    return { unsubscribe() {} }
+      o.error(genericError)
-  },
+    }),
 
   scanAccountsOnDevice(currency, deviceId, { error }) {
     Promise.resolve(genericError).then(error)

src/bridge/makeMockBridge.js

@@ -43,38 +43,37 @@ function makeMockBridge(opts?: Opts): WalletBridge<*> {
   const syncTimeouts = {}
 
   return {
-    synchronize(initialAccount, { error, next, complete }) {
+    synchronize: initialAccount =>
-      const accountId = initialAccount.id
+      Observable.create(o => {
-      if (syncTimeouts[accountId]) {
+        const accountId = initialAccount.id
-        // this is just for tests. we'll assume impl don't need to handle race condition on this function.
+        if (syncTimeouts[accountId]) {
-        logger.warn('synchronize was called multiple pending time for same accounts!!!')
+          // this is just for tests. we'll assume impl don't need to handle race condition on this function.
-      }
+          logger.warn('synchronize was called multiple pending time for same accounts!!!')
-      syncTimeouts[accountId] = setTimeout(() => {
-        if (Math.random() < syncSuccessRate) {
-          const ops = broadcasted[accountId] || []
-          broadcasted[accountId] = []
-          next(account => {
-            account = { ...account }
-            account.blockHeight++
-            for (const op of ops) {
-              account.balance += getOperationAmountNumber(op)
-            }
-            return account
-          })
-          complete()
-        } else {
-          error(new Error('Sync Failed'))
         }
-        syncTimeouts[accountId] = null
+        syncTimeouts[accountId] = setTimeout(() => {
-      }, 20000)
+          if (Math.random() < syncSuccessRate) {
+            const ops = broadcasted[accountId] || []
+            broadcasted[accountId] = []
+            o.next(account => {
+              account = { ...account }
+              account.blockHeight++
+              for (const op of ops) {
+                account.balance += getOperationAmountNumber(op)
+              }
+              return account
+            })
+            o.complete()
+          } else {
+            o.error(new Error('Sync Failed'))
+          }
+          syncTimeouts[accountId] = null
+        }, 20000)
 
-      return {
+        return () => {
-        unsubscribe() {
           clearTimeout(syncTimeouts[accountId])
           syncTimeouts[accountId] = null
-        },
+        }
-      }
+      }),
-    },
 
     scanAccountsOnDevice(currency, deviceId, { next, complete, error }) {
       let unsubscribed = false

src/bridge/types.js

@@ -32,6 +32,7 @@ export interface WalletBridge<Transaction> {
   // observer is an Observer of Account object. Account are expected to be `archived` by default because we want to import all and opt-in on what account to use.
   // the scan can stop once all accounts are discovered.
   // the function returns a Subscription and you MUST stop everything if it is unsubscribed.
+  // TODO return Observable
   scanAccountsOnDevice(
     currency: Currency,
     deviceId: DeviceId,
@@ -46,7 +47,7 @@ export interface WalletBridge<Transaction> {
   // operations if there are new ones (prepended), balance, blockHeight, ...
   // the synchronize can stop once everything is up to date. it is the user side responsability to start it again.
   // we should be able to interrupt the Subscription but we'll leave this usecase for later. if you don't support interruption, please `console.warn`
-  synchronize(initialAccount: Account, observer: Observer<(Account) => Account>): Subscription;
+  synchronize(initialAccount: Account): Observable<(Account) => Account>;
 
   // for a given account, UI wants to load more operations in the account.operations
   // if you can't do it or there is no more things to load, just return account,

src/components/AccountPage/index.js

@@ -7,6 +7,7 @@ import { translate } from 'react-i18next'
 import { Redirect } from 'react-router'
 import styled from 'styled-components'
 import type { Currency, Account } from '@ledgerhq/live-common/lib/types'
+import SyncOneAccountOnMount from 'components/SyncOneAccountOnMount'
 
 import { MODAL_SEND, MODAL_RECEIVE, MODAL_SETTINGS_ACCOUNT } from 'config/constants'
 
@@ -97,6 +98,7 @@ class AccountPage extends PureComponent<Props, State> {
     return (
       // Force re-render account page, for avoid animation
       <Box key={account.id}>
+        <SyncOneAccountOnMount priority={10} accountId={account.id} />
         <Box horizontal mb={5}>
           <AccountHeader account={account} />
           <Box horizontal alignItems="center" justifyContent="flex-end" grow flow={2}>

src/components/PollCounterValuesOnMount.js

@@ -0,0 +1,21 @@
+// @flow
+
+import React, { Component } from 'react'
+import CounterValues from 'helpers/countervalues'
+
+class Effect extends Component<{ cvPolling: * }> {
+  componentDidMount() {
+    this.props.cvPolling.poll()
+  }
+  render() {
+    return null
+  }
+}
+
+const PollCounterValuesOnMount = () => (
+  <CounterValues.PollingConsumer>
+    {cvPolling => <Effect cvPolling={cvPolling} />}
+  </CounterValues.PollingConsumer>
+)
+
+export default PollCounterValuesOnMount

src/components/SyncOneAccountOnMount.js

@@ -0,0 +1,39 @@
+// @flow
+
+import React, { Component } from 'react'
+import { BridgeSyncConsumer } from 'bridge/BridgeSyncContext'
+import type { Sync } from 'bridge/BridgeSyncContext'
+
+export class Effect extends Component<{
+  sync: Sync,
+  accountId: string,
+  priority: number,
+}> {
+  componentDidMount() {
+    const { sync, accountId, priority } = this.props
+    sync({ type: 'SYNC_ONE_ACCOUNT', accountId, priority })
+  }
+  componentDidUpdate(prevProps: *) {
+    const { sync, accountId, priority } = this.props
+    if (accountId !== prevProps.accountId) {
+      sync({ type: 'SYNC_ONE_ACCOUNT', accountId, priority })
+    }
+  }
+  render() {
+    return null
+  }
+}
+
+const SyncOneAccountOnMount = ({
+  accountId,
+  priority,
+}: {
+  accountId: string,
+  priority: number,
+}) => (
+  <BridgeSyncConsumer>
+    {sync => <Effect sync={sync} accountId={accountId} priority={priority} />}
+  </BridgeSyncConsumer>
+)
+
+export default SyncOneAccountOnMount

src/components/SyncSkipUnderPriority.js

@@ -0,0 +1,41 @@
+// @flow
+
+import React, { PureComponent } from 'react'
+import { BridgeSyncConsumer } from 'bridge/BridgeSyncContext'
+import type { Sync } from 'bridge/BridgeSyncContext'
+
+const instances = []
+
+export class Effect extends PureComponent<{
+  sync: Sync,
+  priority: number, // eslint-disable-line
+}> {
+  componentDidMount() {
+    instances.push(this)
+    this.check()
+  }
+  componentDidUpdate() {
+    this.check()
+  }
+  componentWillUnmount() {
+    const i = instances.indexOf(this)
+    if (i !== -1) {
+      instances.splice(i, 1)
+      this.check()
+    }
+  }
+  check() {
+    const { sync } = this.props
+    const priority = instances.length === 0 ? -1 : Math.max(...instances.map(i => i.props.priority))
+    sync({ type: 'SET_SKIP_UNDER_PRIORITY', priority })
+  }
+  render() {
+    return null
+  }
+}
+
+const SyncSkipUnderPriority = ({ priority }: { priority: number }) => (
+  <BridgeSyncConsumer>{sync => <Effect sync={sync} priority={priority} />}</BridgeSyncConsumer>
+)
+
+export default SyncSkipUnderPriority

src/components/TopBar/ActivityIndicator.js

@@ -124,7 +124,7 @@ class ActivityIndicatorInner extends Component<Props, State> {
 
 const ActivityIndicator = ({ globalSyncState, t }: { globalSyncState: AsyncState, t: T }) => (
   <BridgeSyncConsumer>
-    {bridgeSync => (
+    {setSyncBehavior => (
       <CounterValues.PollingConsumer>
         {cvPolling => {
           const isPending = cvPolling.pending || globalSyncState.pending
@@ -137,7 +137,7 @@ const ActivityIndicator = ({ globalSyncState, t }: { globalSyncState: AsyncState
               isError={!!isError && !isPending}
               onClick={() => {
                 cvPolling.poll()
-                bridgeSync.syncAll()
+                setSyncBehavior({ type: 'SYNC_ALL_ACCOUNTS', priority: 5 })
               }}
             />
           )

src/components/modals/ImportAccounts/index.js

@@ -6,6 +6,8 @@ import { connect } from 'react-redux'
 import { translate } from 'react-i18next'
 import { createStructuredSelector } from 'reselect'
 
+import SyncSkipUnderPriority from 'components/SyncSkipUnderPriority'
+
 import type { Currency, Account } from '@ledgerhq/live-common/lib/types'
 
 import type { T, Device } from 'types/common'
@@ -199,6 +201,7 @@ class ImportAccounts extends PureComponent<Props, State> {
         onHide={() => this.setState({ ...INITIAL_STATE })}
         render={({ onClose }) => (
           <ModalBody onClose={onClose}>
+            <SyncSkipUnderPriority priority={100} />
             <ModalTitle onBack={onBack ? () => onBack(stepProps) : void 0}>
               {t('importAccounts:title')}
             </ModalTitle>

src/components/modals/Receive/index.js

@@ -9,6 +9,8 @@ import type { T, Device } from 'types/common'
 import { MODAL_RECEIVE } from 'config/constants'
 
 import getAddress from 'commands/getAddress'
+import SyncSkipUnderPriority from 'components/SyncSkipUnderPriority'
+import SyncOneAccountOnMount from 'components/SyncOneAccountOnMount'
 
 import Box from 'components/base/Box'
 import Breadcrumb from 'components/Breadcrumb'
@@ -300,7 +302,7 @@ class ReceiveModal extends PureComponent<Props, State> {
 
   render() {
     const { t } = this.props
-    const { stepsErrors, stepsDisabled, stepIndex } = this.state
+    const { stepsErrors, stepsDisabled, stepIndex, account } = this.state
 
     const canClose = this.canClose()
     const canPrev = this.canPrev()
@@ -313,6 +315,8 @@ class ReceiveModal extends PureComponent<Props, State> {
         preventBackdropClick={!canClose}
         render={({ onClose }) => (
           <ModalBody onClose={canClose ? onClose : undefined}>
+            <SyncSkipUnderPriority priority={9} />
+            {account && <SyncOneAccountOnMount priority={10} accountId={account.id} />}
             <ModalTitle>
               {canPrev && <PrevButton onClick={this.handlePrevStep} />}
               {t('receive:title')}

src/components/modals/Send/SendModalBody.js

@@ -16,11 +16,15 @@ import { getBridgeForCurrency } from 'bridge'
 import { accountsSelector } from 'reducers/accounts'
 import { updateAccountWithUpdater } from 'actions/accounts'
 
+import PollCounterValuesOnMount from 'components/PollCounterValuesOnMount'
+
 import Breadcrumb from 'components/Breadcrumb'
 import { ModalBody, ModalTitle, ModalContent } from 'components/base/Modal'
 import PrevButton from 'components/modals/PrevButton'
 import StepConnectDevice from 'components/modals/StepConnectDevice'
 import ChildSwitch from 'components/base/ChildSwitch'
+import SyncSkipUnderPriority from 'components/SyncSkipUnderPriority'
+import SyncOneAccountOnMount from 'components/SyncOneAccountOnMount'
 
 import Footer from './Footer'
 import ConfirmationFooter from './ConfirmationFooter'
@@ -271,6 +275,10 @@ class SendModalBody extends PureComponent<Props, State<*>> {
 
     return (
       <ModalBody onClose={onClose}>
+        <PollCounterValuesOnMount />
+        <SyncSkipUnderPriority priority={80} />
+        {account && <SyncOneAccountOnMount priority={81} accountId={account.id} />}
+
         <ModalTitle>
           {canPrev && <PrevButton onClick={this.onPrevStep} />}
           {t('send:title')}

src/config/constants.js

@@ -1,7 +1,7 @@
 // @flow
 
 export const SYNC_BOOT_DELAY = 2 * 1000
-export const SYNC_INTERVAL = 30 * 1000
+export const SYNC_ALL_INTERVAL = 60 * 1000
 export const CHECK_APP_INTERVAL_WHEN_INVALID = 600
 export const CHECK_APP_INTERVAL_WHEN_VALID = 1200
 export const CHECK_UPDATE_DELAY = 5e3

src/reducers/bridgeSync.js

@@ -12,12 +12,10 @@ export type AsyncState = {
 
 export type BridgeSyncState = {
   syncs: { [accountId: string]: AsyncState },
-  pullMores: { [accountId: string]: AsyncState },
 }
 
-const state: BridgeSyncState = {
+const initialState: BridgeSyncState = {
   syncs: {},
-  pullMores: {},
 }
 
 const handlers: Object = {
@@ -33,16 +31,6 @@ const handlers: Object = {
       [action.accountId]: action.state,
     },
   }),
-
-  SET_ACCOUNT_PULL_MORE_STATE: (
-    state: BridgeSyncState,
-    action: { accountId: string, state: AsyncState },
-  ) => ({
-    pullMores: {
-      ...state.pullMores,
-      [action.accountId]: action.state,
-    },
-  }),
 }
 
 // Selectors
@@ -56,11 +44,6 @@ export const syncStateLocalSelector = (
   { accountId }: { accountId: string },
 ) => bridgeSync.syncs[accountId] || nothingState
 
-export const pullMoreStateLocalSelector = (
-  bridgeSync: BridgeSyncState,
-  { accountId }: { accountId: string },
-) => bridgeSync.pullMores[accountId] || nothingState
-
 export const globalSyncStateSelector = createSelector(
   accountsSelector,
   bridgeSyncSelector,
@@ -78,4 +61,4 @@ export const globalSyncStateSelector = createSelector(
   },
 )
 
-export default handleActions(handlers, state)
+export default handleActions(handlers, initialState)

yarn.lock

@@ -2352,7 +2352,7 @@ async@^1.4.0, async@^1.5.0, async@^1.5.2:
   version "1.5.2"
   resolved "https://registry.yarnpkg.com/async/-/async-1.5.2.tgz#ec6a61ae56480c0c3cb241c95618e20892f9672a"
 
-async@^2.1.2, async@^2.1.4, async@^2.6.0:
+async@^2.1.2, async@^2.1.4, async@^2.6.0, async@^2.6.1:
   version "2.6.1"
   resolved "https://registry.yarnpkg.com/async/-/async-2.6.1.tgz#b245a23ca71930044ec53fa46aa00a3e87c6a610"
   dependencies: