feat!: add ObservableCardanoNode.submitTx method

implement it for OgmiosObservableCardanoNode

Author: mkazlauskas

packages/core/src/CardanoNode/types/ObservableCardanoNode.ts

@@ -1,5 +1,5 @@
 import { bufferChainSyncEvent } from '../util/bufferChainSyncEvent';
-import type { Cardano, HealthCheckResponse } from '../..';
+import type { Cardano, HealthCheckResponse, TxCBOR } from '../..';
 import type { EraSummary } from './CardanoNode';
 import type { Observable } from 'rxjs';
 
@@ -71,6 +71,12 @@ export interface ObservableCardanoNode {
    * @throws CardanoNodeErrors.UnknownCardanoNodeError on any other unexpected/unhandled errors
    */
   findIntersect(points: PointOrOrigin[]): Observable<ObservableChainSync>;
+
+  /**
+   * @param tx serialized transaction
+   * @returns transaction id
+   */
+  submitTx(tx: TxCBOR): Observable<Cardano.TransactionId>;
 }
 
 export const ObservableCardanoNode = { bufferChainSyncEvent } as const;

packages/ogmios/src/CardanoNode/OgmiosObservableCardanoNode/OgmiosObservableCardanoNode.ts

@@ -11,11 +11,13 @@ import {
   ObservableCardanoNode,
   ObservableChainSync,
   PointOrOrigin,
-  StateQueryErrorCode
+  StateQueryErrorCode,
+  TxCBOR
 } from '@cardano-sdk/core';
 import {
   ChainSynchronization,
   ConnectionConfig,
+  TransactionSubmission,
   createConnectionObject,
   createLedgerStateQueryClient,
   getServerHealth
@@ -31,6 +33,7 @@ import {
   of,
   shareReplay,
   switchMap,
+  take,
   throwError,
   timeout
 } from 'rxjs';
@@ -39,7 +42,7 @@ import { WithLogger, contextLogger } from '@cardano-sdk/util';
 import { createObservableChainSyncClient } from './createObservableChainSyncClient';
 import { ogmiosServerHealthToHealthCheckResponse } from '../../util';
 import { ogmiosToCorePointOrOrigin, ogmiosToCoreTipOrOrigin, pointOrOriginToOgmios } from './util';
-import { queryEraSummaries, queryGenesisParameters } from '../queries';
+import { queryEraSummaries, queryGenesisParameters, withCoreCardanoNodeError } from '../queries';
 import isEqual from 'lodash/isEqual';
 
 const ogmiosToCoreIntersection = (intersection: ChainSynchronization.Intersection) => ({
@@ -48,9 +51,11 @@ const ogmiosToCoreIntersection = (intersection: ChainSynchronization.Intersectio
 });
 
 export type LocalStateQueryRetryConfig = Pick<RetryBackoffConfig, 'initialInterval' | 'maxInterval'>;
+export type SubmitTxRetryConfig = Pick<RetryBackoffConfig, 'initialInterval' | 'maxInterval' | 'maxRetries'>;
 
-const DEFAULT_HEALTH_CHECK_TIMEOUT = 2000;
-const DEFAULT_LSQ_RETRY_CONFIG: LocalStateQueryRetryConfig = {
+const DEFAULT_HEALTH_CHECK_TIMEOUT = Milliseconds(2000);
+const DEFAULT_SUBMIT_MAX_RETRIES = 5;
+const DEFAULT_RETRY_CONFIG: LocalStateQueryRetryConfig = {
   initialInterval: 1000,
   maxInterval: 30_000
 };
@@ -59,16 +64,22 @@ export type OgmiosObservableCardanoNodeProps = InteractionContextProps & {
   healthCheckTimeout?: Milliseconds;
   /** Default: {initialInterval: 1000, maxInterval: 30_000} */
   localStateQueryRetryConfig?: LocalStateQueryRetryConfig;
+  /** Default: {initialInterval: 1000, maxInterval: 30_000, maxRetries: 5} */
+  submitTxQueryRetryConfig?: SubmitTxRetryConfig;
 };
 
-const retryableStateQueryErrors = new Set<number>([
+const retryableCardanoNodeErrors = new Set<number>([
   GeneralCardanoNodeErrorCode.ServerNotReady,
-  StateQueryErrorCode.UnavailableInCurrentEra,
   GeneralCardanoNodeErrorCode.ConnectionFailure
 ]);
 
+const retryableStateQueryErrors = new Set<number>([
+  ...retryableCardanoNodeErrors,
+  StateQueryErrorCode.UnavailableInCurrentEra
+]);
+
 const stateQueryRetryBackoffConfig = (
-  retryConfig: LocalStateQueryRetryConfig = DEFAULT_LSQ_RETRY_CONFIG,
+  retryConfig: LocalStateQueryRetryConfig = DEFAULT_RETRY_CONFIG,
   logger: Logger
 ): RetryBackoffConfig => ({
   ...retryConfig,
@@ -81,8 +92,24 @@ const stateQueryRetryBackoffConfig = (
   }
 });
 
+const submitTxRetryBackoffConfig = (
+  retryConfig: SubmitTxRetryConfig = DEFAULT_RETRY_CONFIG,
+  logger: Logger
+): RetryBackoffConfig => ({
+  ...retryConfig,
+  maxRetries: retryConfig.maxRetries || DEFAULT_SUBMIT_MAX_RETRIES,
+  shouldRetry: (error) => {
+    if (retryableStateQueryErrors.has(CardanoNodeUtil.asCardanoNodeError(error)?.code)) {
+      logger.warn('Failed to submitTx, will retry', error);
+      return true;
+    }
+    return false;
+  }
+});
+
 export class OgmiosObservableCardanoNode implements ObservableCardanoNode {
   readonly #connectionConfig$: Observable<ConnectionConfig>;
+  readonly #submitTxRetryBackoffConfig: RetryBackoffConfig;
   readonly #logger: Logger;
   readonly #interactionContext$;
 
@@ -93,6 +120,7 @@ export class OgmiosObservableCardanoNode implements ObservableCardanoNode {
   constructor(props: OgmiosObservableCardanoNodeProps, { logger }: WithLogger) {
     this.#connectionConfig$ = props.connectionConfig$;
     this.#logger = contextLogger(logger, 'ObservableOgmiosCardanoNode');
+    this.#submitTxRetryBackoffConfig = submitTxRetryBackoffConfig(props.submitTxQueryRetryConfig, logger);
     this.#interactionContext$ = createObservableInteractionContext(
       {
         ...props
@@ -177,4 +205,18 @@ export class OgmiosObservableCardanoNode implements ObservableCardanoNode {
       )
     );
   }
+
+  submitTx(tx: TxCBOR): Observable<Cardano.TransactionId> {
+    return this.#interactionContext$.pipe(
+      switchMap((context) =>
+        from(
+          withCoreCardanoNodeError(() =>
+            TransactionSubmission.submitTransaction(context, tx)
+          ) as Promise<Cardano.TransactionId>
+        )
+      ),
+      retryBackoff(this.#submitTxRetryBackoffConfig),
+      take(1)
+    );
+  }
 }

packages/ogmios/test/CardanoNode/ObservableOgmiosCardanoNode.test.ts

@@ -5,7 +5,10 @@ import {
   GeneralCardanoNodeErrorCode,
   Milliseconds,
   StateQueryError,
-  StateQueryErrorCode
+  StateQueryErrorCode,
+  TxCBOR,
+  TxSubmissionError,
+  TxSubmissionErrorCode
 } from '@cardano-sdk/core';
 import { Connection, InteractionContext, Mirror, createConnectionObject, safeJSON } from '@cardano-ogmios/client';
 import { HEALTH_RESPONSE_BODY } from '../mocks/util';
@@ -16,11 +19,12 @@ import {
   MockedChainSynchronization,
   MockedLedgerStateQueryClient,
   MockedSocket,
+  MockedTransactionSubmission,
   ogmiosEraSummaries
 } from './util';
-import { NextBlockResponse, RollForward } from '@cardano-ogmios/schema';
+import { NextBlockResponse, RollForward, SubmitTransactionFailureEraMismatch } from '@cardano-ogmios/schema';
 import { OgmiosObservableCardanoNode } from '../../src';
-import { combineLatest, delay as delayEmission, firstValueFrom, mergeMap, of } from 'rxjs';
+import { combineLatest, delay as delayEmission, firstValueFrom, mergeMap, of, toArray } from 'rxjs';
 import { generateRandomHexString, logger } from '@cardano-sdk/util-dev';
 import { mockGenesisShelley, mockShelleyBlock } from '../ogmiosToCore/testData';
 import delay from 'delay';
@@ -30,6 +34,9 @@ jest.mock('@cardano-ogmios/client', () => {
   return {
     ...original,
     ChainSynchronization: {},
+    TransactionSubmission: {
+      submitTransaction: jest.fn()
+    },
     createInteractionContext: jest.fn(),
     createLedgerStateQueryClient: jest.fn(),
     getServerHealth: jest.fn()
@@ -42,6 +49,7 @@ describe('ObservableOgmiosCardanoNode', () => {
   let getServerHealth: MockGetServerHealth;
   let socket: MockedSocket;
   let chainSynchronization: MockedChainSynchronization;
+  let TransactionSubmission: MockedTransactionSubmission;
   let createInteractionContext: MockCreateInteractionContext;
 
   const tip = {
@@ -56,7 +64,8 @@ describe('ObservableOgmiosCardanoNode', () => {
       createInteractionContext,
       createLedgerStateQueryClient,
       getServerHealth,
-      ChainSynchronization: chainSynchronization
+      ChainSynchronization: chainSynchronization,
+      TransactionSubmission
     } = require('@cardano-ogmios/client'));
     ledgerStateQueryClient = {
       eraSummaries: jest.fn() as MockedLedgerStateQueryClient['eraSummaries'],
@@ -104,6 +113,7 @@ describe('ObservableOgmiosCardanoNode', () => {
     createInteractionContext.mockReset();
     createLedgerStateQueryClient.mockReset();
     getServerHealth.mockReset();
+    TransactionSubmission.submitTransaction.mockReset();
   });
 
   describe('LSQs on QueryUnavailableInCurrentEra', () => {
@@ -155,7 +165,6 @@ describe('ObservableOgmiosCardanoNode', () => {
     });
   });
 
-  // TODO: this passes when run individually but not when running entire test suite
   it('opaquely reconnects when connection is refused', async () => {
     createInteractionContext.mockRejectedValueOnce({ name: 'WebSocketClosed' });
     const node = new OgmiosObservableCardanoNode({ connectionConfig$: of(connection) }, { logger });
@@ -222,4 +231,66 @@ describe('ObservableOgmiosCardanoNode', () => {
       expect(result.ok).toBe(false);
     });
   });
+
+  describe('submitTx', () => {
+    let node: OgmiosObservableCardanoNode;
+    const submitTxMaxRetries = 2;
+
+    beforeEach(() => {
+      node = new OgmiosObservableCardanoNode(
+        {
+          connectionConfig$: of(connection),
+          submitTxQueryRetryConfig: { initialInterval: 1, maxRetries: submitTxMaxRetries }
+        },
+        { logger }
+      );
+    });
+
+    describe('successful submission', () => {
+      it('emits transaction id and completes', async () => {
+        TransactionSubmission.submitTransaction.mockResolvedValueOnce('id');
+        await expect(firstValueFrom(node.submitTx('cbor' as TxCBOR).pipe(toArray()))).resolves.toEqual(['id']);
+        expect(TransactionSubmission.submitTransaction).toBeCalledTimes(1);
+      });
+    });
+
+    describe('submission error', () => {
+      it('maps error to core type', async () => {
+        TransactionSubmission.submitTransaction.mockRejectedValueOnce({
+          code: 3005,
+          data: { ledgerEra: 'shelley', queryEra: 'alonzo' },
+          message: 'Era mismatch'
+        } as SubmitTransactionFailureEraMismatch);
+        await expect(firstValueFrom(node.submitTx('cbor' as TxCBOR))).rejects.toThrowError(
+          expect.objectContaining({
+            code: TxSubmissionErrorCode.EraMismatch,
+            name: TxSubmissionError.name
+          })
+        );
+        expect(TransactionSubmission.submitTransaction).toBeCalledTimes(1);
+      });
+    });
+
+    describe('connection error', () => {
+      it('attempts to resubmit opaquely', async () => {
+        TransactionSubmission.submitTransaction
+          .mockRejectedValueOnce({ code: 'ECONNREFUSED' })
+          .mockResolvedValueOnce('id');
+        await expect(firstValueFrom(node.submitTx('cbor' as TxCBOR))).resolves.toBe('id');
+        expect(TransactionSubmission.submitTransaction).toBeCalledTimes(2);
+      });
+
+      it('rejects after maxRetries attempts to submit', async () => {
+        const error = { code: 'ECONNREFUSED' };
+        TransactionSubmission.submitTransaction.mockRejectedValue(error);
+
+        await expect(firstValueFrom(node.submitTx('cbor' as TxCBOR))).rejects.toThrowError(
+          expect.objectContaining({
+            code: GeneralCardanoNodeErrorCode.ConnectionFailure
+          })
+        );
+        expect(TransactionSubmission.submitTransaction).toBeCalledTimes(submitTxMaxRetries + 1);
+      });
+    });
+  });
 });

packages/ogmios/test/CardanoNode/util.ts

@@ -1,6 +1,7 @@
 import {
   ChainSynchronization,
   InteractionContext,
+  TransactionSubmission,
   createInteractionContext,
   createLedgerStateQueryClient,
   getServerHealth
@@ -13,6 +14,7 @@ export type MockedLedgerStateQueryClient = jest.Mocked<LedgerStateQueryClient>;
 export type MockCreateLedgerStateQuery = jest.MockedFunction<typeof createLedgerStateQueryClient>;
 export type MockGetServerHealth = jest.MockedFunction<typeof getServerHealth>;
 export type MockedChainSynchronization = jest.Mocked<typeof ChainSynchronization>;
+export type MockedTransactionSubmission = jest.Mocked<typeof TransactionSubmission>;
 export type MockCreateInteractionContext = jest.MockedFunction<typeof createInteractionContext>;
 
 export const ogmiosEraSummaries: EraSummary[] = [

packages/util-dev/src/chainSync/index.ts

@@ -6,12 +6,14 @@ import {
   ChainSyncEventType,
   ChainSyncRollBackward,
   ChainSyncRollForward,
+  GeneralCardanoNodeError,
+  GeneralCardanoNodeErrorCode,
   Intersection,
   ObservableCardanoNode,
   Point,
   PointOrOrigin
 } from '@cardano-sdk/core';
-import { Observable, of } from 'rxjs';
+import { Observable, of, throwError } from 'rxjs';
 import { fromSerializableObject } from '@cardano-sdk/util';
 import { genesisToEraSummary } from './genesisToEraSummary';
 import memoize from 'lodash/memoize';
@@ -135,7 +137,11 @@ export const chainSyncData = memoize((dataSet: ChainSyncDataSet) => {
       });
     },
     genesisParameters$: of(compactGenesis),
-    healthCheck$: new Observable()
+    healthCheck$: new Observable(),
+    submitTx: () =>
+      throwError(
+        () => new GeneralCardanoNodeError(GeneralCardanoNodeErrorCode.Unknown, null, 'submitTx is not implemented')
+      )
   };
   return {
     allEvents,