Use fetch queue for SSE reconnects & tx creation requests (#1127)

* Use fetch queue on SSE reconnects & for tx creation requests

* Make balances load quicker

* Replace more (last?) instances of requests without queue

src/Generic/lib/stellar.ts

@@ -3,40 +3,6 @@ import fetch from "isomorphic-fetch"
 import { xdr, Asset, Horizon, Keypair, NotFoundError, Server, ServerApi, Transaction } from "stellar-sdk"
 import { AssetRecord } from "../hooks/stellar-ecosystem"
 import { AccountData } from "./account"
-import { joinURL } from "./url"
-import { CustomError } from "./errors"
-
-export interface SmartFeePreset {
-  capacityTrigger: number
-  maxFee: number
-  percentile: number
-}
-
-interface FeeStatsDetails {
-  max: string
-  min: string
-  mode: string
-  p10: string
-  p20: string
-  p30: string
-  p40: string
-  p50: string
-  p60: string
-  p70: string
-  p80: string
-  p90: string
-  p95: string
-  p99: string
-}
-
-// See <https://www.stellar.org/developers/horizon/reference/endpoints/fee-stats.html>
-interface FeeStats {
-  last_ledger: string
-  last_ledger_base_fee: string
-  ledger_capacity_usage: string
-  fee_charged: FeeStatsDetails
-  max_fee: FeeStatsDetails
-}
 
 const MAX_INT64 = "9223372036854775807"
 
@@ -91,32 +57,6 @@ export function stringifyAsset(assetOrTrustline: Asset | Horizon.BalanceLine) {
   }
 }
 
-async function fetchFeeStats(horizon: Server): Promise<FeeStats> {
-  const url = joinURL(getHorizonURL(horizon), "/fee_stats")
-  const response = await fetch(url)
-
-  if (!response.ok) {
-    throw CustomError("RequestFailedError", `Request to ${url} failed with status code ${response.status}`, {
-      target: url,
-      status: response.status
-    })
-  }
-  return response.json()
-}
-
-export async function selectSmartTransactionFee(horizon: Server, preset: SmartFeePreset): Promise<number> {
-  const feeStats = await fetchFeeStats(horizon)
-  const capacityUsage = Number.parseFloat(feeStats.ledger_capacity_usage)
-  const percentileFees = feeStats.fee_charged
-
-  const smartFee =
-    capacityUsage > preset.capacityTrigger
-      ? Number.parseInt((percentileFees as any)[`p${preset.percentile}`] || feeStats.fee_charged.mode, 10)
-      : Number.parseInt(feeStats.fee_charged.min, 10)
-
-  return Math.min(smartFee, preset.maxFee)
-}
-
 export async function friendbotTopup(horizonURL: string, publicKey: string) {
   const horizonMetadata = await (await fetch(horizonURL)).json()
   const friendBotHref = horizonMetadata._links.friendbot.href.replace(/\{\?.*/, "")

src/Generic/lib/third-party-security.ts

@@ -2,6 +2,7 @@ import { Server, Transaction, Horizon } from "stellar-sdk"
 import { CustomError } from "./errors"
 import StellarGuardIcon from "~Icons/components/StellarGuard"
 import LobstrVaultIcon from "~Icons/components/LobstrVault"
+import { workers } from "~Workers/worker-controller"
 
 export interface ThirdPartySecurityService {
   endpoints: {
@@ -34,8 +35,11 @@ const services: ThirdPartySecurityService[] = [
 ]
 
 export async function isThirdPartyProtected(horizon: Server, accountPubKey: string) {
-  const account = await horizon.loadAccount(accountPubKey)
-  const signerKeys = account.signers.map(signer => signer.key)
+  const { netWorker } = await workers
+  const horizonURL = horizon.serverURL.toString()
+
+  const account = await netWorker.fetchAccountData(horizonURL, accountPubKey)
+  const signerKeys = (account?.signers || []).map(signer => signer.key)
 
   const enabledService = services.find(service => signerKeys.includes(service.publicKey))
   return enabledService

src/Generic/lib/transaction.ts

@@ -1,4 +1,5 @@
 import {
+  Account as StellarAccount,
   Asset,
   Keypair,
   Memo,
@@ -13,7 +14,14 @@ import {
 import { Account } from "~App/contexts/accounts"
 import { WrongPasswordError, CustomError } from "./errors"
 import { applyTimeout } from "./promise"
-import { getAllSources, isNotFoundError, isSignedByAnyOf, selectSmartTransactionFee, SmartFeePreset } from "./stellar"
+import { getAllSources, isNotFoundError, isSignedByAnyOf } from "./stellar"
+import { workers } from "~Workers/worker-controller"
+
+interface SmartFeePreset {
+  capacityTrigger: number
+  maxFee: number
+  percentile: number
+}
 
 // See <https://github.com/stellar/go/issues/926>
 const highFeePreset: SmartFeePreset = {
@@ -72,9 +80,20 @@ async function accountExists(horizon: Server, publicKey: string) {
   }
 }
 
-async function selectTransactionFeeWithFallback(horizon: Server, fallbackFee: number) {
+async function selectTransactionFeeWithFallback(horizonURL: string, preset: SmartFeePreset, fallbackFee: number) {
   try {
-    return await selectSmartTransactionFee(horizon, highFeePreset)
+    const { netWorker } = await workers
+    const feeStats = await netWorker.fetchFeeStats(horizonURL)
+
+    const capacityUsage = Number.parseFloat(feeStats.ledger_capacity_usage)
+    const percentileFees = feeStats.fee_charged
+
+    const smartFee =
+      capacityUsage > preset.capacityTrigger
+        ? Number.parseInt((percentileFees as any)[`p${preset.percentile}`] || feeStats.fee_charged.mode, 10)
+        : Number.parseInt(feeStats.fee_charged.min, 10)
+
+    return Math.min(smartFee, preset.maxFee)
   } catch (error) {
     // Don't show error notification, since our horizon's endpoint is non-functional anyway
     // tslint:disable-next-line no-console
@@ -98,17 +117,27 @@ interface TxBlueprint {
 
 export async function createTransaction(operations: Array<xdr.Operation<any>>, options: TxBlueprint) {
   const { horizon, walletAccount } = options
+  const { netWorker } = await workers
+
   const fallbackFee = 10000
+  const horizonURL = horizon.serverURL.toString()
   const timeout = selectTransactionTimeout(options.accountData)
 
-  const [account, smartTxFee, timebounds] = await Promise.all([
-    applyTimeout(horizon.loadAccount(walletAccount.publicKey), 10000, () =>
+  const [accountMetadata, smartTxFee, timebounds] = await Promise.all([
+    applyTimeout(netWorker.fetchAccountData(horizonURL, walletAccount.publicKey), 10000, () =>
       fail(`Fetching source account data timed out`)
     ),
-    applyTimeout(selectTransactionFeeWithFallback(horizon, fallbackFee), 5000, () => fallbackFee),
-    applyTimeout(horizon.fetchTimebounds(timeout), 10000, () => fail(`Syncing time bounds with horizon timed out`))
-  ])
+    applyTimeout(selectTransactionFeeWithFallback(horizonURL, highFeePreset, fallbackFee), 5000, () => fallbackFee),
+    applyTimeout(netWorker.fetchTimebounds(horizonURL, timeout), 10000, () =>
+      fail(`Syncing time bounds with horizon timed out`)
+    )
+  ] as const)
 
+  if (!accountMetadata) {
+    throw Error(`Failed to query account from horizon server: ${walletAccount.publicKey}`)
+  }
+
+  const account = new StellarAccount(accountMetadata.id, accountMetadata.sequence)
   const networkPassphrase = walletAccount.testnet ? Networks.TESTNET : Networks.PUBLIC
   const txFee = Math.max(smartTxFee, options.minTransactionFee || 0)
 
@@ -163,13 +192,23 @@ export async function signTransaction(transaction: Transaction, walletAccount: A
 }
 
 export async function requiresRemoteSignatures(horizon: Server, transaction: Transaction, walletPublicKey: string) {
+  const { netWorker } = await workers
+  const horizonURL = horizon.serverURL.toString()
   const sources = getAllSources(transaction)
 
   if (sources.length > 1) {
     return true
   }
 
-  const accounts = await Promise.all(sources.map(sourcePublicKey => horizon.loadAccount(sourcePublicKey)))
+  const accounts = await Promise.all(
+    sources.map(async sourcePublicKey => {
+      const account = await netWorker.fetchAccountData(horizonURL, sourcePublicKey)
+      if (!account) {
+        throw Error(`Could not fetch account metadata from horizon server: ${sourcePublicKey}`)
+      }
+      return account
+    })
+  )
 
   return accounts.some(account => {
     const thisWalletSigner = account.signers.find(signer => signer.key === walletPublicKey)

src/Workers/lib/event-source.ts

@@ -2,7 +2,8 @@ import { handleConnectionState } from "./connection"
 
 const watchdogIntervalTime = 15_000
 
-function createReconnectDelay(options: { delay: number }): () => Promise<void> {
+function createReconnectDelay(options: { initialDelay: number }): () => Promise<void> {
+  let delay = options.initialDelay
   let lastConnectionAttemptTime = 0
 
   const networkBackOnline = () => {
@@ -21,14 +22,17 @@ function createReconnectDelay(options: { delay: number }): () => Promise<void> {
   }
 
   return async function delayReconnect() {
-    const justConnectedBefore = Date.now() - lastConnectionAttemptTime < options.delay
-    const waitUntil = Date.now() + options.delay
+    const justConnectedBefore = Date.now() - lastConnectionAttemptTime < delay
+    const waitUntil = Date.now() + delay
 
     await networkBackOnline()
 
     if (justConnectedBefore) {
-      // Reconnect immediately (skip await) if last reconnection is long ago
       await timeReached(waitUntil)
+      delay = Math.min(delay * 1.5, options.initialDelay * 8)
+    } else {
+      // Reconnect immediately (skip await) if last reconnection is long ago
+      delay = options.initialDelay
     }
 
     lastConnectionAttemptTime = Date.now()
@@ -41,7 +45,11 @@ interface SSEHandlers {
   onMessage?(event: MessageEvent): void
 }
 
-export function createReconnectingSSE(createURL: () => string, handlers: SSEHandlers) {
+export function createReconnectingSSE(
+  createURL: () => string,
+  handlers: SSEHandlers,
+  queueRequest: (task: () => any) => Promise<any>
+) {
   let currentlySubscribed = false
   let delayReconnect: () => Promise<void>
 
@@ -97,21 +105,20 @@ export function createReconnectingSSE(createURL: () => string, handlers: SSEHand
         handlers.onStreamError(error)
       }
 
-      delayReconnect().then(
-        () => subscribe(),
-        unexpectedError => {
+      delayReconnect()
+        .then(() => queueRequest(() => subscribe()))
+        .catch(unexpectedError => {
           if (handlers.onUnexpectedError) {
             handlers.onUnexpectedError(unexpectedError)
           }
-        }
-      )
+        })
     }
 
     currentlySubscribed = true
   }
 
   const setup = async () => {
-    delayReconnect = createReconnectDelay({ delay: 1000 })
+    delayReconnect = createReconnectDelay({ initialDelay: 1000 })
     subscribe()
   }