schedule exchange updating

packages/taler-wallet-core/src/db.ts

@@ -8,7 +8,7 @@ import { IDBFactory, IDBDatabase } from "idb-bridge";
  * with each major change.  When incrementing the major version,
  * the wallet should import data from the previous version.
  */
-const TALER_DB_NAME = "taler-walletdb-v8";
+const TALER_DB_NAME = "taler-walletdb-v9";
 
 /**
  * Current database minor version, should be incremented

packages/taler-wallet-core/src/operations/exchanges.ts

@@ -30,6 +30,8 @@ import {
   WireFee,
   ExchangeUpdateReason,
   ExchangeUpdatedEventRecord,
+  initRetryInfo,
+  updateRetryInfoTimeout,
 } from "../types/dbTypes";
 import { canonicalizeBaseUrl } from "../util/helpers";
 import * as Amounts from "../util/amounts";
@@ -43,7 +45,12 @@ import {
   WALLET_CACHE_BREAKER_CLIENT_VERSION,
   WALLET_EXCHANGE_PROTOCOL_VERSION,
 } from "./versions";
-import { getTimestampNow, Duration, isTimestampExpired } from "../util/time";
+import {
+  getTimestampNow,
+  Duration,
+  isTimestampExpired,
+  durationFromSpec,
+} from "../util/time";
 import { compare } from "../util/libtoolVersion";
 import { createRecoupGroup, processRecoupGroup } from "./recoup";
 import { TalerErrorCode } from "../TalerErrorCode";
@@ -56,6 +63,7 @@ import { Logger } from "../util/logging";
 import { URL } from "../util/url";
 import { reconcileReserveHistory } from "../util/reserveHistoryUtil";
 import { checkDbInvariant } from "../util/invariants";
+import { NotificationType } from "../types/notifications";
 
 const logger = new Logger("exchanges.ts");
 
@@ -86,17 +94,23 @@ async function denominationRecordFromKeys(
   return d;
 }
 
-async function setExchangeError(
+async function handleExchangeUpdateError(
   ws: InternalWalletState,
   baseUrl: string,
   err: TalerErrorDetails,
 ): Promise<void> {
-  logger.warn(`last error for exchange ${baseUrl}:`, err);
+  await ws.db.runWithWriteTransaction([Stores.exchanges], async (tx) => {
-  const mut = (exchange: ExchangeRecord): ExchangeRecord => {
+    const exchange = await tx.get(Stores.exchanges, baseUrl);
+    if (!exchange) {
+      return;
+    }
+    exchange.retryInfo.retryCounter++;
+    updateRetryInfoTimeout(exchange.retryInfo);
     exchange.lastError = err;
-    return exchange;
+  });
-  };
+  if (err) {
-  await ws.db.mutate(Stores.exchanges, baseUrl, mut);
+    ws.notify({ type: NotificationType.ExchangeOperationError, error: err });
+  }
 }
 
 function getExchangeRequestTimeout(e: ExchangeRecord): Duration {
@@ -142,7 +156,7 @@ async function updateExchangeWithKeys(
         exchangeBaseUrl: baseUrl,
       },
     );
-    await setExchangeError(ws, baseUrl, opErr);
+    await handleExchangeUpdateError(ws, baseUrl, opErr);
     throw new OperationFailedAndReportedError(opErr);
   }
 
@@ -158,7 +172,7 @@ async function updateExchangeWithKeys(
         walletProtocolVersion: WALLET_EXCHANGE_PROTOCOL_VERSION,
       },
     );
-    await setExchangeError(ws, baseUrl, opErr);
+    await handleExchangeUpdateError(ws, baseUrl, opErr);
     throw new OperationFailedAndReportedError(opErr);
   }
 
@@ -198,10 +212,13 @@ async function updateExchangeWithKeys(
         masterPublicKey: exchangeKeysJson.master_public_key,
         protocolVersion: protocolVersion,
         signingKeys: exchangeKeysJson.signkeys,
-        nextUpdateTime: getExpiryTimestamp(resp),
+        nextUpdateTime: getExpiryTimestamp(resp, {
+          minDuration: durationFromSpec({ hours: 1 }),
+        }),
       };
       r.updateStatus = ExchangeUpdateStatus.FetchWire;
       r.lastError = undefined;
+      r.retryInfo = initRetryInfo(false);
       await tx.put(Stores.exchanges, r);
 
       for (const newDenom of newDenominations) {
@@ -433,6 +450,7 @@ async function updateExchangeWithWireInfo(
     };
     r.updateStatus = ExchangeUpdateStatus.FetchTerms;
     r.lastError = undefined;
+    r.retryInfo = initRetryInfo(false);
     await tx.put(Stores.exchanges, r);
   });
 }
@@ -443,7 +461,7 @@ export async function updateExchangeFromUrl(
   forceNow = false,
 ): Promise<ExchangeRecord> {
   const onOpErr = (e: TalerErrorDetails): Promise<void> =>
-    setExchangeError(ws, baseUrl, e);
+    handleExchangeUpdateError(ws, baseUrl, e);
   return await guardOperationException(
     () => updateExchangeFromUrlImpl(ws, baseUrl, forceNow),
     onOpErr,
@@ -460,6 +478,7 @@ async function updateExchangeFromUrlImpl(
   baseUrl: string,
   forceNow = false,
 ): Promise<ExchangeRecord> {
+  logger.trace(`updating exchange info for ${baseUrl}`);
   const now = getTimestampNow();
   baseUrl = canonicalizeBaseUrl(baseUrl);
 
@@ -480,6 +499,7 @@ async function updateExchangeFromUrlImpl(
       termsOfServiceAcceptedTimestamp: undefined,
       termsOfServiceLastEtag: undefined,
       termsOfServiceText: undefined,
+      retryInfo: initRetryInfo(false),
     };
     await ws.db.put(Stores.exchanges, newExchangeRecord);
   } else {
@@ -488,40 +508,30 @@ async function updateExchangeFromUrlImpl(
       if (!rec) {
         return;
       }
-      if (rec.updateStatus != ExchangeUpdateStatus.FetchKeys && !forceNow) {
+      if (rec.updateStatus != ExchangeUpdateStatus.FetchKeys) {
+        const t = rec.details?.nextUpdateTime;
+        if (!forceNow && t && !isTimestampExpired(t)) {
           return;
         }
+      }
       if (rec.updateStatus != ExchangeUpdateStatus.FetchKeys && forceNow) {
         rec.updateReason = ExchangeUpdateReason.Forced;
       }
       rec.updateStarted = now;
       rec.updateStatus = ExchangeUpdateStatus.FetchKeys;
       rec.lastError = undefined;
+      rec.retryInfo = initRetryInfo(false);
       t.put(Stores.exchanges, rec);
     });
   }
 
-  r = await ws.db.get(Stores.exchanges, baseUrl);
-  checkDbInvariant(!!r);
-
-
-  const t = r.details?.nextUpdateTime;
-  if (!forceNow && t && !isTimestampExpired(t)) {
-    logger.trace("using cached exchange info");
-    return r;
-  }
-
   await updateExchangeWithKeys(ws, baseUrl);
   await updateExchangeWithWireInfo(ws, baseUrl);
   await updateExchangeWithTermsOfService(ws, baseUrl);
   await updateExchangeFinalize(ws, baseUrl);
 
   const updatedExchange = await ws.db.get(Stores.exchanges, baseUrl);
-
+  checkDbInvariant(!!updatedExchange);
-  if (!updatedExchange) {
-    // This should practically never happen
-    throw Error("exchange not found");
-  }
   return updatedExchange;
 }
 

packages/taler-wallet-core/src/operations/pending.ts

@@ -56,10 +56,6 @@ async function gatherExchangePending(
   resp: PendingOperationsResponse,
   onlyDue = false,
 ): Promise<void> {
-  if (onlyDue) {
-    // FIXME: exchanges should also be updated regularly
-    return;
-  }
   await tx.iter(Stores.exchanges).forEach((e) => {
     switch (e.updateStatus) {
       case ExchangeUpdateStatus.Finished:
@@ -79,7 +75,7 @@ async function gatherExchangePending(
             type: PendingOperationType.Bug,
             givesLifeness: false,
             message:
-              "Exchange record does not have details, but no update in progress.",
+              "Exchange record does not have details, but no update finished.",
             details: {
               exchangeBaseUrl: e.baseUrl,
             },
@@ -90,14 +86,28 @@ async function gatherExchangePending(
             type: PendingOperationType.Bug,
             givesLifeness: false,
             message:
-              "Exchange record does not have wire info, but no update in progress.",
+              "Exchange record does not have wire info, but no update finished.",
             details: {
               exchangeBaseUrl: e.baseUrl,
             },
           });
         }
+        if (e.details && e.details.nextUpdateTime.t_ms < now.t_ms) {
+          resp.pendingOperations.push({
+            type: PendingOperationType.ExchangeUpdate,
+            givesLifeness: false,
+            stage: ExchangeUpdateOperationStage.FetchKeys,
+            exchangeBaseUrl: e.baseUrl,
+            lastError: e.lastError,
+            reason: "scheduled",
+          });
+          break;
+        }
         break;
       case ExchangeUpdateStatus.FetchKeys:
+        if (onlyDue && e.retryInfo.nextRetry.t_ms > now.t_ms) {
+          return;
+        }
         resp.pendingOperations.push({
           type: PendingOperationType.ExchangeUpdate,
           givesLifeness: false,
@@ -108,6 +118,9 @@ async function gatherExchangePending(
         });
         break;
       case ExchangeUpdateStatus.FetchWire:
+        if (onlyDue && e.retryInfo.nextRetry.t_ms > now.t_ms) {
+          return;
+        }
         resp.pendingOperations.push({
           type: PendingOperationType.ExchangeUpdate,
           givesLifeness: false,
@@ -118,6 +131,9 @@ async function gatherExchangePending(
         });
         break;
       case ExchangeUpdateStatus.FinalizeUpdate:
+        if (onlyDue && e.retryInfo.nextRetry.t_ms > now.t_ms) {
+          return;
+        }
         resp.pendingOperations.push({
           type: PendingOperationType.ExchangeUpdate,
           givesLifeness: false,

packages/taler-wallet-core/src/types/dbTypes.ts

@@ -113,6 +113,7 @@ export function updateRetryInfoTimeout(
     r.nextRetry = { t_ms: "never" };
     return;
   }
+  r.active = true;
   const t =
     now.t_ms + p.backoffDelta.d_ms * Math.pow(p.backoffBase, r.retryCounter);
   r.nextRetry = { t_ms: t };
@@ -642,6 +643,11 @@ export interface ExchangeRecord {
   updateReason?: ExchangeUpdateReason;
 
   lastError?: TalerErrorDetails;
+
+  /**
+   * Retry status for fetching updated information about the exchange.
+   */
+  retryInfo: RetryInfo;
 }
 
 

packages/taler-wallet-core/src/util/http.ts

@@ -26,7 +26,7 @@ import { Codec } from "./codec";
 import { OperationFailedError, makeErrorDetails } from "../operations/errors";
 import { TalerErrorCode } from "../TalerErrorCode";
 import { Logger } from "./logging";
-import { Duration, Timestamp, getTimestampNow } from "./time";
+import { Duration, Timestamp, getTimestampNow, timestampAddDuration, timestampMin, timestampMax } from "./time";
 
 const logger = new Logger("http.ts");
 
@@ -257,15 +257,24 @@ export async function readSuccessResponseTextOrThrow<T>(
 /**
  * Get the timestamp at which the response's content is considered expired.
  */
-export function getExpiryTimestamp(httpResponse: HttpResponse): Timestamp {
+export function getExpiryTimestamp(
+  httpResponse: HttpResponse,
+  opt: { minDuration?: Duration },
+): Timestamp {
   const expiryDateMs = new Date(
     httpResponse.headers.get("expiry") ?? "",
   ).getTime();
+  let t: Timestamp;
   if (Number.isNaN(expiryDateMs)) {
-    return getTimestampNow();
+    t = getTimestampNow();
   } else {
-    return {
+    t = {
       t_ms: expiryDateMs,
+    };
   }
+  if (opt.minDuration) {
+    const t2 = timestampAddDuration(getTimestampNow(), opt.minDuration);
+    return timestampMax(t, t2);
   }
+  return t;
 }

packages/taler-wallet-core/src/util/time.ts

@@ -76,6 +76,38 @@ export function timestampMin(t1: Timestamp, t2: Timestamp): Timestamp {
   return { t_ms: Math.min(t1.t_ms, t2.t_ms) };
 }
 
+export function timestampMax(t1: Timestamp, t2: Timestamp): Timestamp {
+  if (t1.t_ms === "never") {
+    return { t_ms: "never" };
+  }
+  if (t2.t_ms === "never") {
+    return { t_ms: "never" };
+  }
+  return { t_ms: Math.max(t1.t_ms, t2.t_ms) };
+}
+
+const SECONDS = 1000
+const MINUTES = SECONDS * 60;
+const HOURS = MINUTES * 60;
+
+export function durationFromSpec(spec: {
+  seconds?: number,
+  hours?: number,
+  minutes?: number,
+}): Duration {
+  let d_ms = 0;
+  if (spec.seconds) {
+    d_ms += spec.seconds * SECONDS;
+  }
+  if (spec.minutes) {
+    d_ms += spec.minutes * MINUTES;
+  }
+  if (spec.hours) {
+    d_ms += spec.hours * HOURS;
+  }
+  return { d_ms };
+}
+
 /**
  * Truncate a timestamp so that that it represents a multiple
  * of seconds.  The timestamp is always rounded down.