fix(liquidity): recover Scrypt trade orders from transient WS errors

src/integration/exchange/services/scrypt-websocket-connection.ts

@@ -49,6 +49,15 @@ export function isTransientWsError(e: Error): boolean {
   return TRANSIENT_WS_ERROR_MARKERS.some((m) => e.message?.toLowerCase().includes(m.toLowerCase()));
 }
 
+// Matches the timeout errors thrown by requestWithId and requestAndWaitForUpdate below.
+// Kept separate from TRANSIENT_WS_ERROR_MARKERS: a timed-out request may still be processed
+// by Scrypt, so it must not trigger the automatic re-send in retryOnTransientWsError.
+export const WS_TIMEOUT_ERROR_MARKERS = ['Timeout waiting for', 'Request timeout after'];
+
+export function isWsTimeoutError(e: Error): boolean {
+  return WS_TIMEOUT_ERROR_MARKERS.some((m) => e.message?.toLowerCase().includes(m.toLowerCase()));
+}
+
 interface ScryptRequest {
   reqid?: number;
   type: ScryptRequestType | ScryptMessageType;

src/integration/exchange/services/scrypt.service.ts

@@ -221,7 +221,8 @@ export class ScryptService extends PricingProvider {
     const filters: Record<string, unknown> = {};
     if (since) filters.StartDate = since.toISOString();
 
-    return this.connection.fetch<ScryptExecutionReport>(ScryptMessageType.EXECUTION_REPORT, filters);
+    // fetchAll: the report may be on a later page, a single page would miss existing orders
+    return this.connection.fetchAll<ScryptExecutionReport>(ScryptMessageType.EXECUTION_REPORT, filters);
   }
 
   async getTrades(since?: Date): Promise<ScryptTrade[]> {
@@ -247,7 +248,7 @@ export class ScryptService extends PricingProvider {
     return side === ScryptOrderSide.BUY ? price : 1 / price;
   }
 
-  async sell(from: string, to: string, amount: number): Promise<string> {
+  async sell(from: string, to: string, amount: number, clOrdId?: string): Promise<string> {
     const { symbol, side } = await this.getTradePair(from, to);
     const price = await this.getOrderBookPrice(symbol, side);
     const sizeIncrement = await this.getSizeIncrement(symbol);
@@ -258,10 +259,10 @@ export class ScryptService extends PricingProvider {
     const rawQty = side === ScryptOrderSide.SELL ? amount : amount / price;
     const orderQty = Util.floorToValue(rawQty, sizeIncrement);
 
-    return this.placeAndReturnId(symbol, side, orderQty, price);
+    return this.placeAndReturnId(symbol, side, orderQty, price, clOrdId);
   }
 
-  async buy(from: string, to: string, amount: number): Promise<string> {
+  async buy(from: string, to: string, amount: number, clOrdId?: string): Promise<string> {
     const { symbol, side } = await this.getTradePair(from, to);
     const price = await this.getOrderBookPrice(symbol, side);
     const sizeIncrement = await this.getSizeIncrement(symbol);
@@ -272,7 +273,7 @@ export class ScryptService extends PricingProvider {
     const rawQty = side === ScryptOrderSide.BUY ? amount : amount / price;
     const orderQty = Util.floorToValue(rawQty, sizeIncrement);
 
-    return this.placeAndReturnId(symbol, side, orderQty, price);
+    return this.placeAndReturnId(symbol, side, orderQty, price, clOrdId);
   }
 
   private async getSizeIncrement(symbol: string): Promise<number> {
@@ -285,6 +286,7 @@ export class ScryptService extends PricingProvider {
     side: ScryptOrderSide,
     orderQty: number,
     price: number,
+    clOrdId?: string,
   ): Promise<string> {
     const response = await this.placeOrder(
       symbol,
@@ -293,6 +295,7 @@ export class ScryptService extends PricingProvider {
       ScryptOrderType.LIMIT,
       ScryptTimeInForce.GOOD_TILL_CANCEL,
       price,
+      clOrdId,
     );
     return response.id;
   }
@@ -438,8 +441,10 @@ export class ScryptService extends PricingProvider {
     orderType: ScryptOrderType = ScryptOrderType.LIMIT,
     timeInForce: ScryptTimeInForce = ScryptTimeInForce.GOOD_TILL_CANCEL,
     price?: number,
+    presetClOrdId?: string,
   ): Promise<ScryptOrderResponse> {
-    const clOrdId = randomUUID();
+    // a caller-provided ClOrdID allows recovering the order if the connection drops after sending
+    const clOrdId = presetClOrdId ?? randomUUID();
 
     // Price is required for LIMIT orders
     if (orderType === ScryptOrderType.LIMIT && price === undefined) {

src/subdomains/core/liquidity-management/adapters/actions/__tests__/scrypt.adapter.spec.ts

@@ -0,0 +1,232 @@
+import { createMock, DeepMocked } from '@golevelup/ts-jest';
+import { Test, TestingModule } from '@nestjs/testing';
+import { ScryptOrderInfo, ScryptOrderSide, ScryptOrderStatus } from 'src/integration/exchange/dto/scrypt.dto';
+import { ScryptService } from 'src/integration/exchange/services/scrypt.service';
+import { createCustomAsset } from 'src/shared/models/asset/__mocks__/asset.entity.mock';
+import { AssetService } from 'src/shared/models/asset/asset.service';
+import { TestSharedModule } from 'src/shared/utils/test.shared.module';
+import { DexService } from 'src/subdomains/supporting/dex/services/dex.service';
+import { Price } from 'src/subdomains/supporting/pricing/domain/entities/price';
+import { PricingService } from 'src/subdomains/supporting/pricing/services/pricing.service';
+import { LiquidityManagementAction } from '../../../entities/liquidity-management-action.entity';
+import { LiquidityManagementOrder } from '../../../entities/liquidity-management-order.entity';
+import { LiquidityManagementPipeline } from '../../../entities/liquidity-management-pipeline.entity';
+import { LiquidityManagementRule } from '../../../entities/liquidity-management-rule.entity';
+import { LiquidityManagementOrderStatus, LiquidityManagementSystem } from '../../../enums';
+import { OrderFailedException } from '../../../exceptions/order-failed.exception';
+import { OrderNotProcessableException } from '../../../exceptions/order-not-processable.exception';
+import { LiquidityManagementOrderRepository } from '../../../repositories/liquidity-management-order.repository';
+import { ScryptAdapter, ScryptAdapterCommands } from '../scrypt.adapter';
+
+describe('ScryptAdapter', () => {
+  let adapter: ScryptAdapter;
+
+  let scryptService: DeepMocked<ScryptService>;
+  let dexService: DeepMocked<DexService>;
+  let orderRepo: DeepMocked<LiquidityManagementOrderRepository>;
+  let pricingService: DeepMocked<PricingService>;
+  let assetService: DeepMocked<AssetService>;
+
+  beforeEach(async () => {
+    scryptService = createMock<ScryptService>();
+    dexService = createMock<DexService>();
+    orderRepo = createMock<LiquidityManagementOrderRepository>();
+    pricingService = createMock<PricingService>();
+    assetService = createMock<AssetService>();
+
+    const module: TestingModule = await Test.createTestingModule({
+      imports: [TestSharedModule],
+      providers: [
+        ScryptAdapter,
+        { provide: ScryptService, useValue: scryptService },
+        { provide: DexService, useValue: dexService },
+        { provide: LiquidityManagementOrderRepository, useValue: orderRepo },
+        { provide: PricingService, useValue: pricingService },
+        { provide: AssetService, useValue: assetService },
+      ],
+    }).compile();
+
+    adapter = module.get<ScryptAdapter>(ScryptAdapter);
+
+    // happy-path defaults for the sell command (USDT -> CHF)
+    assetService.getAssetByUniqueName.mockResolvedValue(createCustomAsset({ name: 'CHF', dexName: 'CHF' }));
+    scryptService.getCurrentPrice.mockResolvedValue(1);
+    pricingService.getPrice.mockResolvedValue(Price.create('USDT', 'CHF', 1));
+    scryptService.getAvailableBalance.mockResolvedValue(1000);
+    scryptService.getOrderStatus.mockResolvedValue(null);
+    orderRepo.save.mockImplementation(async (order) => order as LiquidityManagementOrder);
+  });
+
+  function createOrder(
+    command: ScryptAdapterCommands,
+    customValues: Partial<LiquidityManagementOrder> = {},
+  ): LiquidityManagementOrder {
+    const rule = Object.assign(new LiquidityManagementRule(), {
+      targetAsset: createCustomAsset({ name: 'USDT', dexName: 'USDT' }),
+    });
+    const pipeline = Object.assign(new LiquidityManagementPipeline(), { rule });
+    const action = Object.assign(new LiquidityManagementAction(), {
+      system: LiquidityManagementSystem.SCRYPT,
+      command,
+      params: JSON.stringify({ tradeAsset: 'CHF' }),
+    });
+
+    return Object.assign(new LiquidityManagementOrder(), {
+      id: 1,
+      created: new Date(),
+      status: LiquidityManagementOrderStatus.CREATED,
+      minAmount: 10,
+      maxAmount: 100,
+      pipeline,
+      action,
+      ...customValues,
+    });
+  }
+
+  function createSellOrder(customValues: Partial<LiquidityManagementOrder> = {}): LiquidityManagementOrder {
+    return createOrder(ScryptAdapterCommands.SELL, customValues);
+  }
+
+  function createBuyOrder(customValues: Partial<LiquidityManagementOrder> = {}): LiquidityManagementOrder {
+    return createOrder(ScryptAdapterCommands.BUY, customValues);
+  }
+
+  function createOrderInfo(id: string): ScryptOrderInfo {
+    return {
+      id,
+      orderId: 'order-1',
+      symbol: 'USDT-CHF',
+      side: ScryptOrderSide.SELL,
+      status: ScryptOrderStatus.FILLED,
+      quantity: 100,
+      filledQuantity: 100,
+      remainingQuantity: 0,
+    };
+  }
+
+  it('should return the persisted ClOrdID on a transient WS error during sell', async () => {
+    const order = createSellOrder();
+    scryptService.sell.mockRejectedValue(new Error('Connection closed'));
+
+    const correlationId = await adapter.executeOrder(order);
+
+    expect(correlationId).toEqual(order.correlationId);
+    expect(scryptService.sell).toHaveBeenCalledWith('USDT', 'CHF', 100, correlationId);
+
+    // ClOrdID must be persisted before the order is sent
+    expect(orderRepo.save).toHaveBeenCalledWith(order);
+    expect(orderRepo.save.mock.invocationCallOrder[0]).toBeLessThan(scryptService.sell.mock.invocationCallOrder[0]);
+  });
+
+  it('should fail the order on a Scrypt rejection during sell', async () => {
+    const order = createSellOrder();
+    scryptService.sell.mockRejectedValue(new Error('Scrypt order rejected: invalid order'));
+
+    await expect(adapter.executeOrder(order)).rejects.toThrow(OrderFailedException);
+  });
+
+  it('should mark the order as not processable on insufficient funds during sell', async () => {
+    const order = createSellOrder();
+    scryptService.sell.mockRejectedValue(new Error('Insufficient funds'));
+
+    await expect(adapter.executeOrder(order)).rejects.toThrow(OrderNotProcessableException);
+  });
+
+  it('should not place a new order, if the existing ClOrdID is found at Scrypt', async () => {
+    const order = createSellOrder({ correlationId: 'existing-id' });
+    scryptService.getOrderStatus.mockResolvedValue(createOrderInfo('existing-id'));
+
+    const correlationId = await adapter.executeOrder(order);
+
+    expect(correlationId).toEqual('existing-id');
+    expect(scryptService.getOrderStatus).toHaveBeenCalledWith('existing-id');
+    expect(scryptService.sell).not.toHaveBeenCalled();
+  });
+
+  it('should place a new order and chain the ClOrdID, if the existing ClOrdID is not found at Scrypt', async () => {
+    const order = createSellOrder({ correlationId: 'lost-id' });
+    scryptService.getOrderStatus.mockResolvedValue(null);
+    scryptService.sell.mockImplementation(async (_from, _to, _amount, clOrdId) => clOrdId);
+
+    const correlationId = await adapter.executeOrder(order);
+
+    expect(correlationId).not.toEqual('lost-id');
+    expect(correlationId).toEqual(order.correlationId);
+    expect(order.previousCorrelationIds).toContain('lost-id');
+    expect(scryptService.sell).toHaveBeenCalledWith('USDT', 'CHF', 100, correlationId);
+  });
+
+  it('should return the existing trade during sell, even if the balance is locked by the open order', async () => {
+    const order = createSellOrder({ correlationId: 'existing-id' });
+    scryptService.getOrderStatus.mockResolvedValue(createOrderInfo('existing-id'));
+    scryptService.getAvailableBalance.mockResolvedValue(0);
+
+    const correlationId = await adapter.executeOrder(order);
+
+    expect(correlationId).toEqual('existing-id');
+    expect(scryptService.sell).not.toHaveBeenCalled();
+  });
+
+  it('should return the persisted ClOrdID on a transient WS error during buy', async () => {
+    const order = createBuyOrder();
+    scryptService.getTradePair.mockResolvedValue({ symbol: 'USDT-CHF', side: ScryptOrderSide.BUY });
+    scryptService.sell.mockRejectedValue(new Error('Connection closed'));
+
+    const correlationId = await adapter.executeOrder(order);
+
+    expect(correlationId).toEqual(order.correlationId);
+    expect(scryptService.sell).toHaveBeenCalledWith('CHF', 'USDT', 100, correlationId);
+  });
+
+  it('should return the persisted ClOrdID on a request timeout during sell', async () => {
+    const order = createSellOrder();
+    scryptService.sell.mockRejectedValue(new Error('Timeout waiting for ExecutionReport update after 60000ms'));
+
+    const correlationId = await adapter.executeOrder(order);
+
+    expect(correlationId).toEqual(order.correlationId);
+    expect(scryptService.sell).toHaveBeenCalledWith('USDT', 'CHF', 100, correlationId);
+  });
+
+  it('should return the persisted ClOrdID on a request timeout without update wait during sell', async () => {
+    const order = createSellOrder();
+    scryptService.sell.mockRejectedValue(new Error('Request timeout after 30000ms'));
+
+    const correlationId = await adapter.executeOrder(order);
+
+    expect(correlationId).toEqual(order.correlationId);
+    expect(scryptService.sell).toHaveBeenCalledWith('USDT', 'CHF', 100, correlationId);
+  });
+
+  it('should retry the completion check on a request timeout instead of failing the order', async () => {
+    const order = createSellOrder({
+      status: LiquidityManagementOrderStatus.IN_PROGRESS,
+      correlationId: 'existing-id',
+      updated: new Date(),
+    });
+    scryptService.checkTrade.mockRejectedValue(new Error('Request timeout after 30000ms'));
+
+    await expect(adapter.checkCompletion(order)).resolves.toBe(false);
+  });
+
+  it('should return the existing ClOrdID, if the re-execution guard fails with a transient error', async () => {
+    const order = createSellOrder({ correlationId: 'existing-id' });
+    scryptService.getOrderStatus.mockRejectedValue(new Error('Connection closed'));
+
+    const correlationId = await adapter.executeOrder(order);
+
+    expect(correlationId).toEqual('existing-id');
+    expect(scryptService.sell).not.toHaveBeenCalled();
+  });
+
+  it('should place a new order on a transient guard error, if no ClOrdID exists yet', async () => {
+    const order = createSellOrder();
+    scryptService.getOrderStatus.mockRejectedValue(new Error('Connection closed'));
+    scryptService.sell.mockImplementation(async (_from, _to, _amount, clOrdId) => clOrdId);
+
+    const correlationId = await adapter.executeOrder(order);
+
+    expect(correlationId).toEqual(order.correlationId);
+    expect(scryptService.sell).toHaveBeenCalledWith('USDT', 'CHF', 100, correlationId);
+  });
+});