wallet-core: put refresh sessions into own store

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

@@ -107,6 +107,8 @@ import { RetryInfo, TaskIdentifiers } from "./operations/common.js";
    full contract terms from the DB quite often.
    Instead, we should probably extract what we need into a separate object
    store.
+ - More object stores should have an "id" primary key,
+   as this makes referencing less expensive.
  */
 
 /**
@@ -943,9 +945,6 @@ export interface RefreshReasonDetails {
 export interface RefreshGroupRecord {
   operationStatus: RefreshOperationStatus;
 
-  // FIXME: Put this into a different object store?
-  lastErrorPerCoin: { [coinIndex: number]: TalerErrorDetail };
-
   /**
    * Unique, randomly generated identifier for this group of
    * refresh operations.
@@ -969,13 +968,9 @@ export interface RefreshGroupRecord {
 
   oldCoinPubs: string[];
 
-  // FIXME:  Should this go into a separate
-  // object store for faster updates?
-  refreshSessionPerCoin: (RefreshSessionRecord | undefined)[];
-
   inputPerCoin: AmountString[];
 
-  estimatedOutputPerCoin: AmountString[];
+  expectedOutputPerCoin: AmountString[];
 
   /**
    * Flag for each coin whether refreshing finished.
@@ -997,6 +992,13 @@ export interface RefreshGroupRecord {
  * Ongoing refresh
  */
 export interface RefreshSessionRecord {
+  refreshGroupId: string;
+
+  /**
+   * Index of the coin in the refresh group.
+   */
+  coinIndex: number;
+
   /**
    * 512-bit secret that can be used to derive
    * the other cryptographic material for the refresh session.
@@ -1021,6 +1023,8 @@ export interface RefreshSessionRecord {
    * The no-reveal-index after we've done the melting.
    */
   norevealIndex?: number;
+
+  lastError?: TalerErrorDetail;
 }
 
 export enum RefundReason {
@@ -2372,6 +2376,13 @@ export const WalletStoresV1 = {
       byStatus: describeIndex("byStatus", "operationStatus"),
     },
   ),
+  refreshSessions: describeStore(
+    "refreshSessions",
+    describeContents<RefreshSessionRecord>({
+      keyPath: ["refreshGroupId", "coinIndex"],
+    }),
+    {},
+  ),
   recoupGroups: describeStore(
     "recoupGroups",
     describeContents<RecoupGroupRecord>({

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

@@ -95,14 +95,7 @@ function computeRefreshGroupAvailableAmount(r: RefreshGroupRecord): AmountJson {
     return available;
   }
   for (let i = 0; i < r.oldCoinPubs.length; i++) {
-    const session = r.refreshSessionPerCoin[i];
+    available = Amounts.add(available, r.expectedOutputPerCoin[i]).amount;
-    if (session) {
-      // We are always assuming the refresh will succeed, thus we
-      // report the output as available balance.
-      available = Amounts.add(available, session.amountRefreshOutput).amount;
-    } else {
-      available = Amounts.add(available, r.estimatedOutputPerCoin[i]).amount;
-    }
   }
   return available;
 }

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

@@ -76,7 +76,11 @@ import {
   RefreshReasonDetails,
   WalletStoresV1,
 } from "../db.js";
-import { isWithdrawableDenom, PendingTaskType } from "../index.js";
+import {
+  isWithdrawableDenom,
+  PendingTaskType,
+  RefreshSessionRecord,
+} from "../index.js";
 import {
   EXCHANGE_COINS_LOCK,
   InternalWalletState,
@@ -170,18 +174,23 @@ function updateGroupStatus(rg: RefreshGroupRecord): { final: boolean } {
 
 /**
  * Create a refresh session for one particular coin inside a refresh group.
+ *
+ * If the session already exists, return the existing one.
+ *
+ * If the session doesn't need to be created (refresh group gone or session already
+ * finished), return undefined.
  */
-async function refreshCreateSession(
+async function provideRefreshSession(
   ws: InternalWalletState,
   refreshGroupId: string,
   coinIndex: number,
-): Promise<void> {
+): Promise<RefreshSessionRecord | undefined> {
   logger.trace(
     `creating refresh session for coin ${coinIndex} in refresh group ${refreshGroupId}`,
   );
 
   const d = await ws.db
-    .mktx((x) => [x.refreshGroups, x.coins])
+    .mktx((x) => [x.refreshGroups, x.coins, x.refreshSessions])
     .runReadWrite(async (tx) => {
       const refreshGroup = await tx.refreshGroups.get(refreshGroupId);
       if (!refreshGroup) {
@@ -192,21 +201,24 @@ async function refreshCreateSession(
       ) {
         return;
       }
-      const existingRefreshSession =
+      const existingRefreshSession = await tx.refreshSessions.get([
-        refreshGroup.refreshSessionPerCoin[coinIndex];
+        refreshGroupId,
-      if (existingRefreshSession) {
+        coinIndex,
-        return;
+      ]);
-      }
       const oldCoinPub = refreshGroup.oldCoinPubs[coinIndex];
       const coin = await tx.coins.get(oldCoinPub);
       if (!coin) {
         throw Error("Can't refresh, coin not found");
       }
-      return { refreshGroup, coin };
+      return { refreshGroup, coin, existingRefreshSession };
     });
 
   if (!d) {
-    return;
+    return undefined;
+  }
+
+  if (d.existingRefreshSession) {
+    return d.existingRefreshSession;
   }
 
   const { refreshGroup, coin } = d;
@@ -288,17 +300,23 @@ async function refreshCreateSession(
   const sessionSecretSeed = encodeCrock(getRandomBytes(64));
 
   // Store refresh session for this coin in the database.
-  await ws.db
+  const newSession = await ws.db
-    .mktx((x) => [x.refreshGroups, x.coins])
+    .mktx((x) => [x.refreshGroups, x.coins, x.refreshSessions])
     .runReadWrite(async (tx) => {
       const rg = await tx.refreshGroups.get(refreshGroupId);
       if (!rg) {
         return;
       }
-      if (rg.refreshSessionPerCoin[coinIndex]) {
+      const existingSession = await tx.refreshSessions.get([
+        refreshGroupId,
+        coinIndex,
+      ]);
+      if (existingSession) {
         return;
       }
-      rg.refreshSessionPerCoin[coinIndex] = {
+      const newSession: RefreshSessionRecord = {
+        coinIndex,
+        refreshGroupId,
         norevealIndex: undefined,
         sessionSecretSeed: sessionSecretSeed,
         newDenoms: newCoinDenoms.selectedDenoms.map((x) => ({
@@ -307,11 +325,13 @@ async function refreshCreateSession(
         })),
         amountRefreshOutput: Amounts.stringify(newCoinDenoms.totalCoinValue),
       };
-      await tx.refreshGroups.put(rg);
+      await tx.refreshSessions.put(newSession);
+      return newSession;
     });
   logger.trace(
     `created refresh session for coin #${coinIndex} in ${refreshGroupId}`,
   );
+  return newSession;
 }
 
 function getRefreshRequestTimeout(rg: RefreshGroupRecord): Duration {
@@ -326,13 +346,16 @@ async function refreshMelt(
   coinIndex: number,
 ): Promise<void> {
   const d = await ws.db
-    .mktx((x) => [x.refreshGroups, x.coins, x.denominations])
+    .mktx((x) => [x.refreshGroups, x.refreshSessions, x.coins, x.denominations])
     .runReadWrite(async (tx) => {
       const refreshGroup = await tx.refreshGroups.get(refreshGroupId);
       if (!refreshGroup) {
         return;
       }
-      const refreshSession = refreshGroup.refreshSessionPerCoin[coinIndex];
+      const refreshSession = await tx.refreshSessions.get([
+        refreshGroupId,
+        coinIndex,
+      ]);
       if (!refreshSession) {
         return;
       }
@@ -442,7 +465,12 @@ async function refreshMelt(
   if (resp.status === HttpStatusCode.NotFound) {
     const errDetails = await readUnexpectedResponseDetails(resp);
     const transitionInfo = await ws.db
-      .mktx((x) => [x.refreshGroups, x.coins, x.coinAvailability])
+      .mktx((x) => [
+        x.refreshGroups,
+        x.refreshSessions,
+        x.coins,
+        x.coinAvailability,
+      ])
       .runReadWrite(async (tx) => {
         const rg = await tx.refreshGroups.get(refreshGroupId);
         if (!rg) {
@@ -456,12 +484,22 @@ async function refreshMelt(
         }
         const oldTxState = computeRefreshTransactionState(rg);
         rg.statusPerCoin[coinIndex] = RefreshCoinStatus.Failed;
-        rg.lastErrorPerCoin[coinIndex] = errDetails;
+        const refreshSession = await tx.refreshSessions.get([
+          refreshGroupId,
+          coinIndex,
+        ]);
+        if (!refreshSession) {
+          throw Error(
+            "db invariant failed: missing refresh session in database",
+          );
+        }
+        refreshSession.lastError = errDetails;
         const updateRes = updateGroupStatus(rg);
         if (updateRes.final) {
           await makeCoinsVisible(ws, tx, transactionId);
         }
         await tx.refreshGroups.put(rg);
+        await tx.refreshSessions.put(refreshSession);
         const newTxState = computeRefreshTransactionState(rg);
         return {
           oldTxState,
@@ -493,7 +531,7 @@ async function refreshMelt(
   refreshSession.norevealIndex = norevealIndex;
 
   await ws.db
-    .mktx((x) => [x.refreshGroups])
+    .mktx((x) => [x.refreshGroups, x.refreshSessions])
     .runReadWrite(async (tx) => {
       const rg = await tx.refreshGroups.get(refreshGroupId);
       if (!rg) {
@@ -502,7 +540,7 @@ async function refreshMelt(
       if (rg.timestampFinished) {
         return;
       }
-      const rs = rg.refreshSessionPerCoin[coinIndex];
+      const rs = await tx.refreshSessions.get([refreshGroupId, coinIndex]);
       if (!rs) {
         return;
       }
@@ -510,7 +548,7 @@ async function refreshMelt(
         return;
       }
       rs.norevealIndex = norevealIndex;
-      await tx.refreshGroups.put(rg);
+      await tx.refreshSessions.put(rs);
     });
 }
 
@@ -581,13 +619,16 @@ async function refreshReveal(
     `doing refresh reveal for ${refreshGroupId} (old coin ${coinIndex})`,
   );
   const d = await ws.db
-    .mktx((x) => [x.refreshGroups, x.coins, x.denominations])
+    .mktx((x) => [x.refreshGroups, x.refreshSessions, x.coins, x.denominations])
     .runReadOnly(async (tx) => {
       const refreshGroup = await tx.refreshGroups.get(refreshGroupId);
       if (!refreshGroup) {
         return;
       }
-      const refreshSession = refreshGroup.refreshSessionPerCoin[coinIndex];
+      const refreshSession = await tx.refreshSessions.get([
+        refreshGroupId,
+        coinIndex,
+      ]);
       if (!refreshSession) {
         return;
       }
@@ -755,6 +796,7 @@ async function refreshReveal(
       x.denominations,
       x.coinAvailability,
       x.refreshGroups,
+      x.refreshSessions,
     ])
     .runReadWrite(async (tx) => {
       const rg = await tx.refreshGroups.get(refreshGroupId);
@@ -762,7 +804,7 @@ async function refreshReveal(
         logger.warn("no refresh session found");
         return;
       }
-      const rs = rg.refreshSessionPerCoin[coinIndex];
+      const rs = await tx.refreshSessions.get([refreshGroupId, coinIndex]);
       if (!rs) {
         return;
       }
@@ -858,10 +900,15 @@ async function processRefreshSession(
   logger.trace(
     `processing refresh session for coin ${coinIndex} of group ${refreshGroupId}`,
   );
-  let refreshGroup = await ws.db
+  let { refreshGroup, refreshSession } = await ws.db
-    .mktx((x) => [x.refreshGroups])
+    .mktx((x) => [x.refreshGroups, x.refreshSessions])
     .runReadOnly(async (tx) => {
-      return tx.refreshGroups.get(refreshGroupId);
+      const rg = await tx.refreshGroups.get(refreshGroupId);
+      const rs = await tx.refreshSessions.get([refreshGroupId, coinIndex]);
+      return {
+        refreshGroup: rg,
+        refreshSession: rs,
+      };
     });
   if (!refreshGroup) {
     return;
@@ -869,18 +916,9 @@ async function processRefreshSession(
   if (refreshGroup.statusPerCoin[coinIndex] === RefreshCoinStatus.Finished) {
     return;
   }
-  if (!refreshGroup.refreshSessionPerCoin[coinIndex]) {
+  if (!refreshSession) {
-    await refreshCreateSession(ws, refreshGroupId, coinIndex);
+    refreshSession = await provideRefreshSession(ws, refreshGroupId, coinIndex);
-    refreshGroup = await ws.db
-      .mktx((x) => [x.refreshGroups])
-      .runReadOnly(async (tx) => {
-        return tx.refreshGroups.get(refreshGroupId);
-      });
-    if (!refreshGroup) {
-      return;
   }
-  }
-  const refreshSession = refreshGroup.refreshSessionPerCoin[coinIndex];
   if (!refreshSession) {
     if (refreshGroup.statusPerCoin[coinIndex] !== RefreshCoinStatus.Finished) {
       throw Error(
@@ -1058,13 +1096,11 @@ export async function createRefreshGroup(
     timestampFinished: undefined,
     statusPerCoin: oldCoinPubs.map(() => RefreshCoinStatus.Pending),
     oldCoinPubs: oldCoinPubs.map((x) => x.coinPub),
-    lastErrorPerCoin: {},
     reasonDetails,
     reason,
     refreshGroupId,
-    refreshSessionPerCoin: oldCoinPubs.map(() => undefined),
     inputPerCoin: oldCoinPubs.map((x) => x.amount),
-    estimatedOutputPerCoin: estimatedOutputPerCoin.map((x) =>
+    expectedOutputPerCoin: estimatedOutputPerCoin.map((x) =>
       Amounts.stringify(x),
     ),
     timestampCreated: TalerPreciseTimestamp.now(),

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

@@ -760,7 +760,7 @@ function buildTransactionForRefresh(
   ).amount;
   const outputAmount = Amounts.sumOrZero(
     refreshGroupRecord.currency,
-    refreshGroupRecord.estimatedOutputPerCoin,
+    refreshGroupRecord.expectedOutputPerCoin,
   ).amount;
   return {
     type: TransactionType.Refresh,