Merge pull request #1550 from input-output-hk/fix/fatal-provider-errors

fix: fatal provider errors

packages/tx-construction/src/tsconfig.json

@@ -7,6 +7,9 @@
     {
       "path": "../../util/src"
     },
+    {
+      "path": "../../util-rxjs/src"
+    },
     {
       "path": "../../core/src"
     },

packages/tx-construction/src/tx-builder/TxBuilder.ts

@@ -39,11 +39,11 @@ import {
   validateValidityInterval
 } from './utils';
 import { SelectionSkeleton } from '@cardano-sdk/input-selection';
-import { coldObservableProvider } from '@cardano-sdk/util-rxjs';
 import { contextLogger, deepEquals } from '@cardano-sdk/util';
 import { createOutputValidator } from '../output-validation';
 import { initializeTx } from './initializeTx';
 import { lastValueFrom } from 'rxjs';
+import { poll } from '@cardano-sdk/util-rxjs';
 import omit from 'lodash/omit.js';
 import uniq from 'lodash/uniq.js';
 
@@ -499,12 +499,12 @@ export class GenericTxBuilder implements TxBuilder {
       allRewardAccounts = uniq([...knownAddresses, ...newAddresses]).map(({ rewardAccount }) => rewardAccount);
     }
 
-    const rewardAccounts$ = coldObservableProvider({
+    const rewardAccounts$ = poll({
       logger: contextLogger(this.#logger, 'getOrCreateRewardAccounts'),
       pollUntil: (rewardAccounts) =>
         allRewardAccounts.every((newAccount) => rewardAccounts.some((acct) => acct.address === newAccount)),
-      provider: this.#dependencies.txBuilderProviders.rewardAccounts,
-      retryBackoffConfig: { initialInterval: 10, maxInterval: 100, maxRetries: 10 }
+      retryBackoffConfig: { initialInterval: 10, maxInterval: 100, maxRetries: 10 },
+      sample: this.#dependencies.txBuilderProviders.rewardAccounts
     });
 
     try {

packages/util-rxjs/src/index.ts

@@ -5,5 +5,5 @@ export * from './passthrough';
 export * from './finalizeWithLatest';
 export * from './concatAndCombineLatest';
 export * from './shareRetryBackoff';
-export * from './coldObservableProvider';
+export * from './poll';
 export * from './types';

packages/util-rxjs/src/poll.ts

@@ -0,0 +1,79 @@
+import { Logger } from 'ts-log';
+import {
+  NEVER,
+  Observable,
+  concat,
+  defer,
+  distinctUntilChanged,
+  from,
+  mergeMap,
+  of,
+  switchMap,
+  takeUntil,
+  throwError
+} from 'rxjs';
+import { RetryBackoffConfig, retryBackoff } from 'backoff-rxjs';
+import { strictEquals } from '@cardano-sdk/util';
+
+const POLL_UNTIL_RETRY = Symbol('POLL_UNTIL_RETRY');
+
+export interface PollProps<T> {
+  sample: () => Promise<T>;
+  retryBackoffConfig: RetryBackoffConfig;
+  trigger$?: Observable<unknown>;
+  equals?: (t1: T, t2: T) => boolean;
+  combinator?: typeof switchMap;
+  cancel$?: Observable<unknown>;
+  pollUntil?: (v: T) => boolean;
+  logger: Logger;
+}
+
+export const poll = <T>({
+  sample,
+  retryBackoffConfig,
+  trigger$ = of(true),
+  equals = strictEquals,
+  combinator = switchMap,
+  cancel$ = NEVER,
+  pollUntil = () => true,
+  logger
+}: PollProps<T>) =>
+  trigger$.pipe(
+    combinator(() =>
+      defer(() =>
+        from(sample()).pipe(
+          mergeMap((v) =>
+            pollUntil(v)
+              ? of(v)
+              : // Emit value, but also throw error to force retryBackoff to kick in
+                concat(
+                  of(v),
+                  throwError(() => POLL_UNTIL_RETRY)
+                )
+          )
+        )
+      ).pipe(
+        retryBackoff({
+          ...retryBackoffConfig,
+          shouldRetry: (error) => {
+            if (error === POLL_UNTIL_RETRY) {
+              logger.warn('"pollUntil" condition not met, will retry');
+              return true;
+            }
+
+            logger.error(error);
+
+            if (retryBackoffConfig.shouldRetry) {
+              const shouldRetry = retryBackoffConfig.shouldRetry(error);
+              logger.debug(`Should retry: ${shouldRetry}`);
+              return shouldRetry;
+            }
+
+            return true;
+          }
+        })
+      )
+    ),
+    distinctUntilChanged(equals),
+    takeUntil(cancel$)
+  );