Release: develop -> main

migration/1784885000000-AddLiquidityOrderNotSentRecheckDue.js

@@ -0,0 +1,49 @@
+/**
+ * @typedef {import('typeorm').MigrationInterface} MigrationInterface
+ * @typedef {import('typeorm').QueryRunner} QueryRunner
+ */
+
+/**
+ * Marks a liquidity management order somebody has released as never sent, until the venue has been asked
+ * once more. While the column is set the order stays quarantined: the release is accepted, not yet in effect.
+ *
+ * Concluding that a request never reached the venue is the one judgement nothing here can verify from the
+ * outside, and it can be made at the very moment reconciliation is watching the venue confirm that same
+ * order. Were the release to take effect at once, the order would be terminal — its rule free to plan
+ * against funds that are in fact committed — before anything could contradict it. Waiting for one machine
+ * answer costs a single reconciliation pass, normally seconds, and closes that window entirely.
+ *
+ * Two exceptions end the wait without an answer, both about liveness rather than evidence: an order no
+ * integration can look up any more, and a venue that has answered nothing for long enough.
+ *
+ * The column records work outstanding, not when the release was asked for: that goes into the order's own
+ * reason, which nothing clears. Indexed so the wait never turns into a scan.
+ *
+ * Purely additive and nullable, no backfill: existing rows read NULL, which is exactly right — they predate
+ * this path entirely and have no release awaiting confirmation. The index is the deterministic TypeORM name
+ * for a single-column index on this table, matching the entity's `@Index()`.
+ *
+ * @class
+ * @implements {MigrationInterface}
+ */
+module.exports = class AddLiquidityOrderNotSentRecheckDue1784885000000 {
+  name = 'AddLiquidityOrderNotSentRecheckDue1784885000000';
+
+  /**
+   * @param {QueryRunner} queryRunner
+   */
+  async up(queryRunner) {
+    await queryRunner.query(`ALTER TABLE "liquidity_management_order" ADD "notSentRecheckDue" TIMESTAMP`);
+    await queryRunner.query(
+      `CREATE INDEX "IDX_cfc953689f0268e33cf14c1cc0" ON "liquidity_management_order" ("notSentRecheckDue")`,
+    );
+  }
+
+  /**
+   * @param {QueryRunner} queryRunner
+   */
+  async down(queryRunner) {
+    await queryRunner.query(`DROP INDEX "public"."IDX_cfc953689f0268e33cf14c1cc0"`);
+    await queryRunner.query(`ALTER TABLE "liquidity_management_order" DROP COLUMN "notSentRecheckDue"`);
+  }
+};

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

@@ -1,6 +1,10 @@
 import { EventEmitter as MockEventEmitter } from 'events';
 import Ws from 'ws';
-import { ScryptMessageType, ScryptWebSocketConnection } from '../scrypt-websocket-connection';
+import {
+  ScryptMessageType,
+  ScryptRequestTimeoutError,
+  ScryptWebSocketConnection,
+} from '../scrypt-websocket-connection';
 
 type MockWebSocketInstance = MockEventEmitter & {
   url: string;
@@ -999,4 +1003,57 @@ describe('ScryptWebSocketConnection', () => {
 
     expect(cb).toHaveBeenCalledTimes(1);
   });
+
+  describe('unanswered requests', () => {
+    const REQUEST_TIMEOUT_MS = 30000;
+
+    function subscribeReqIds(ws: MockWebSocketInstance, streamName: ScryptMessageType): number[] {
+      return ws.send.mock.calls
+        .map(([payload]) => JSON.parse(payload as string))
+        .filter((msg) => msg.type === 'subscribe' && msg.streams?.[0]?.name === streamName)
+        .map((msg) => msg.reqid as number);
+    }
+
+    it('retries a read once when the venue never answers, instead of failing the caller', async () => {
+      const ws = await firstConnectWithStream();
+      const streamName = ScryptMessageType.EXECUTION_REPORT;
+
+      const fetchPromise = connection.fetch(streamName);
+      await flushPromises();
+      expect(subscribeReqIds(ws, streamName)).toHaveLength(1);
+
+      // silence for the whole deadline — the venue simply does not reply
+      jest.advanceTimersByTime(REQUEST_TIMEOUT_MS);
+      await flushPromises();
+
+      const reqIds = subscribeReqIds(ws, streamName);
+      expect(reqIds).toHaveLength(2);
+
+      ws.emit(
+        'message',
+        JSON.stringify({ reqid: reqIds[1], type: streamName, initial: true, data: [{ ClOrdID: 'ord-after-retry' }] }),
+      );
+
+      await expect(fetchPromise).resolves.toEqual([{ ClOrdID: 'ord-after-retry' }]);
+      expect(loggerWarn).toHaveBeenCalledWith(expect.stringContaining(`Retrying fetch ${streamName}`));
+    });
+
+    it('surfaces an unanswered request as ScryptRequestTimeoutError, not a plain Error', async () => {
+      const ws = await firstConnectWithStream();
+      const streamName = ScryptMessageType.EXECUTION_REPORT;
+
+      const fetchPromise = connection.fetch(streamName);
+      const assertion = expect(fetchPromise).rejects.toBeInstanceOf(ScryptRequestTimeoutError);
+      await flushPromises();
+
+      // both the first attempt and its retry go unanswered
+      jest.advanceTimersByTime(REQUEST_TIMEOUT_MS);
+      await flushPromises();
+      jest.advanceTimersByTime(REQUEST_TIMEOUT_MS);
+      await flushPromises();
+
+      await assertion;
+      expect(subscribeReqIds(ws, streamName)).toHaveLength(2);
+    });
+  });
 });

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

@@ -4,7 +4,14 @@ import {
   ScryptTransactionStatus,
   ScryptTransactionType,
 } from '../../dto/scrypt.dto';
-import { ScryptMessageType, ScryptWebSocketConnection } from '../scrypt-websocket-connection';
+import {
+  ScryptAmendRejectedError,
+  ScryptMessageType,
+  ScryptRequestTimeoutError,
+  ScryptUnconfirmedWriteError,
+  ScryptVenueRejectionError,
+  ScryptWebSocketConnection,
+} from '../scrypt-websocket-connection';
 import { ScryptService } from '../scrypt.service';
 
 jest.mock('src/config/config', () => {
@@ -546,4 +553,97 @@ describe('ScryptService', () => {
       ]);
     });
   });
+  describe('checkTrade — the amend write boundary', () => {
+    function stubAmendPath(editOutcome: Error): void {
+      jest.spyOn(service as any, 'getOrderStatus').mockResolvedValue({
+        id: 'dfx-lm-7',
+        status: ScryptOrderStatus.PARTIALLY_FILLED,
+        price: 1,
+        remainingQuantity: 5,
+      });
+      jest.spyOn(service as any, 'getTradePrice').mockResolvedValue(2);
+      jest.spyOn(service as any, 'editOrder').mockRejectedValue(editOutcome);
+      jest.spyOn(service as any, 'cancelOrder').mockResolvedValue(undefined);
+    }
+
+    it('propagates an unconfirmed amend instead of swallowing it', async () => {
+      // Regression guard: the amend used to be wrapped in a catch that cancelled and returned false, so the
+      // caller never learned that a replacement order might be live at the venue under the reserved id.
+      stubAmendPath(new ScryptRequestTimeoutError('Timeout waiting for ExecutionReport update after 60000ms'));
+
+      await expect(service.checkTrade('dfx-lm-7', 'EUR', 'USDT', new Date(), 'dfx-lm-7-1')).rejects.toBeInstanceOf(
+        ScryptUnconfirmedWriteError,
+      );
+      expect((service as any).cancelOrder).not.toHaveBeenCalled();
+    });
+
+    it('carries the reserved replacement reference on the raised error', async () => {
+      stubAmendPath(new ScryptRequestTimeoutError('Timeout waiting for ExecutionReport update after 60000ms'));
+
+      await expect(service.checkTrade('dfx-lm-7', 'EUR', 'USDT', new Date(), 'dfx-lm-7-1')).rejects.toMatchObject({
+        reference: 'dfx-lm-7-1',
+      });
+    });
+
+    it('forgets a cached open order when the follow-up cancel goes unconfirmed', async () => {
+      // the cancel is a write as well: unconfirmed, it may have taken effect while the cached report still
+      // shows the order open — and a non-terminal entry is never refreshed, so every later check would wait
+      // on a picture that cannot change
+      stubAmendPath(new ScryptVenueRejectionError('Scrypt order edit rejected: price out of band'));
+      jest.spyOn(service as any, 'cancelOrder').mockRejectedValue(new ScryptRequestTimeoutError('Request timeout'));
+      (service as any).executionReports.set('dfx-lm-7', { ClOrdID: 'dfx-lm-7', OrdStatus: ScryptOrderStatus.NEW });
+
+      await expect(service.checkTrade('dfx-lm-7', 'EUR', 'USDT', new Date(), 'dfx-lm-7-1')).rejects.toMatchObject({
+        message: expect.stringContaining('cancel went unconfirmed'),
+      });
+
+      expect((service as any).executionReports.has('dfx-lm-7')).toBe(false);
+    });
+
+    it('keeps the cached order when the cancel is confirmed', async () => {
+      stubAmendPath(new ScryptVenueRejectionError('Scrypt order edit rejected: price out of band'));
+      (service as any).executionReports.set('dfx-lm-7', { ClOrdID: 'dfx-lm-7', OrdStatus: ScryptOrderStatus.NEW });
+
+      await expect(service.checkTrade('dfx-lm-7', 'EUR', 'USDT', new Date(), 'dfx-lm-7-1')).rejects.toBeInstanceOf(
+        ScryptAmendRejectedError,
+      );
+
+      expect((service as any).executionReports.has('dfx-lm-7')).toBe(true);
+    });
+
+    it('keeps waiting on a pending order however old it is — pending is observed, not unknown', async () => {
+      // quarantining it would make reconciliation find the reference, hand the order back, and the next
+      // completion check quarantine it again: a loop, not a resolution
+      jest.spyOn(service as any, 'getOrderStatus').mockResolvedValue({
+        id: 'dfx-lm-7',
+        status: ScryptOrderStatus.PENDING_NEW,
+        remainingQuantity: 5,
+      });
+
+      await expect(service.checkTrade('dfx-lm-7', 'EUR', 'USDT', new Date(Date.now() - 120 * 60 * 1000))).resolves.toBe(
+        false,
+      );
+    });
+
+    it('keeps waiting on a pending order that is still young', async () => {
+      jest.spyOn(service as any, 'getOrderStatus').mockResolvedValue({
+        id: 'dfx-lm-7',
+        status: ScryptOrderStatus.PENDING_NEW,
+        remainingQuantity: 5,
+      });
+
+      await expect(service.checkTrade('dfx-lm-7', 'EUR', 'USDT', new Date())).resolves.toBe(false);
+    });
+
+    it('cancels on an explicit rejection, but reports the refusal and the spent reference', async () => {
+      // A rejection is a reply: nothing was created, so cancelling is safe. The caller still has to learn
+      // about it — the replacement reference is burnt at the venue and must not be derived again.
+      stubAmendPath(new ScryptVenueRejectionError('Scrypt order edit rejected: price out of band'));
+
+      await expect(service.checkTrade('dfx-lm-7', 'EUR', 'USDT', new Date(), 'dfx-lm-7-1')).rejects.toMatchObject({
+        spentReference: 'dfx-lm-7-1',
+      });
+      expect((service as any).cancelOrder).toHaveBeenCalled();
+    });
+  });
 });

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

@@ -50,6 +50,81 @@ export function isTransientWsError(e: Error): boolean {
   return TRANSIENT_WS_ERROR_MARKERS.some((m) => e.message?.toLowerCase().includes(m.toLowerCase()));
 }
 
+/**
+ * A request was sent but no answer arrived within its deadline.
+ *
+ * Deliberately its own type rather than another entry in TRANSIENT_WS_ERROR_MARKERS: those markers describe
+ * a socket that demonstrably dropped the request, so retrying is safe for anything. A timeout describes
+ * silence — the venue may or may not have acted. Only idempotent reads may retry it; every write path must
+ * translate it into an unknown outcome. Matching on the message text instead would make that distinction
+ * impossible to enforce, because both kinds of timeout would read the same.
+ */
+export class ScryptRequestTimeoutError extends Error {}
+
+/**
+ * A write that may or may not have taken effect at the venue — raised where an order was created, amended or
+ * restarted and no reply confirmed the outcome.
+ *
+ * Distinct from {@link ScryptRequestTimeoutError}, which describes only *how* the call ended: the same
+ * dropped socket is harmless on a read and unresolved on a write, so the distinction that matters to the
+ * caller is the side effect, not the transport. Anything carrying this type must be quarantined and
+ * reconciled, never repeated.
+ */
+export class ScryptUnconfirmedWriteError extends Error {
+  constructor(
+    message: string,
+    readonly reference: string | undefined,
+  ) {
+    super(message);
+  }
+}
+
+/**
+ * An order the venue once acknowledged can no longer be found in its state.
+ *
+ * Not a failure: the order may have completed or been cancelled outside our view, and we cannot tell which.
+ * Treating it as failed would release the rule to open a second position against the same funds.
+ */
+export class ScryptOrderNotFoundError extends Error {}
+
+/**
+ * An amend the venue refused. The replacement was never created, so the ORIGINAL order is still live — and
+ * its reference is spent, because the venue requires references to be unique. Carries it so the caller can
+ * record it and derive a fresh one next time instead of reusing a burnt reference forever.
+ */
+export class ScryptAmendRejectedError extends Error {
+  constructor(
+    message: string,
+    readonly spentReference: string | undefined,
+  ) {
+    super(message);
+  }
+}
+
+/**
+ * The venue sent an explicit error in reply to one specific request.
+ *
+ * Deliberately NOT a rejection: `unknown reqid` arrives the same way and means the venue lost our request
+ * context, which for a mutation is as open as silence. Callers that can tell the two apart narrow it; callers
+ * that cannot must keep treating it as an unresolved outcome.
+ */
+export class ScryptErrorResponseError extends Error {}
+
+/**
+ * The venue replied and refused the request. This is the ONLY evidence that a write did not take effect —
+ * everything else leaves the outcome open.
+ *
+ * A type rather than a set of message patterns: a rejection is now impossible to miss by phrasing a message
+ * differently, and impossible to fake by a transport error that happens to contain the word. Every path that
+ * turns a venue refusal into an exception must use this type, or the caller will retry a settled outcome
+ * forever.
+ */
+export class ScryptVenueRejectionError extends Error {}
+
+export function isVenueRejection(e: Error): boolean {
+  return e instanceof ScryptVenueRejectionError;
+}
+
 interface ScryptRequest {
   reqid?: number;
   type: ScryptRequestType | ScryptMessageType;
@@ -120,7 +195,7 @@ export class ScryptWebSocketConnection {
       }
     };
 
-    return this.retryOnTransientWsError(doFetch, `fetch ${streamName}`);
+    return this.retryIdempotentRead(doFetch, `fetch ${streamName}`);
   }
 
   async fetchAll<T>(streamName: ScryptMessageType, filters?: Record<string, unknown>): Promise<T[]> {
@@ -155,7 +230,7 @@ export class ScryptWebSocketConnection {
       }
     };
 
-    return this.retryOnTransientWsError(doFetch, `fetchAll ${streamName}`);
+    return this.retryIdempotentRead(doFetch, `fetchAll ${streamName}`);
   }
 
   // Register a callback fired after a successful RE-connect (not the first connect). Used to re-fetch state that
@@ -164,11 +239,19 @@ export class ScryptWebSocketConnection {
     this.reconnectCallbacks.push(callback);
   }
 
-  private async retryOnTransientWsError<T>(operation: () => Promise<T>, label: string): Promise<T> {
+  /**
+   * Retry wrapper for IDEMPOTENT READS ONLY — `fetch` and `fetchAll`. Never widen this to a call that can
+   * create, amend or cancel an order, or move funds: it retries on timeout, and a timed-out write may
+   * already have been executed by the venue. Write paths must surface the timeout so the caller can treat
+   * the outcome as unknown (see OrderOutcomeUnknownException in the liquidity-management subdomain).
+   */
+  private async retryIdempotentRead<T>(operation: () => Promise<T>, label: string): Promise<T> {
     try {
       return await operation();
     } catch (error) {
-      if (isTransientWsError(error)) {
+      // A read that went unanswered is safe to repeat: re-subscribing to a snapshot stream has no side
+      // effect at the venue. Without this, a single silent 30s window ends the whole liquidity order.
+      if (isTransientWsError(error) || error instanceof ScryptRequestTimeoutError) {
         this.logger.warn(`Retrying ${label} after transient error: ${error.message}`);
         return operation();
       }
@@ -186,7 +269,7 @@ export class ScryptWebSocketConnection {
     return new Promise((resolve, reject) => {
       const timeoutId = setTimeout(() => {
         unsubscribe();
-        reject(new Error(`Timeout waiting for ${streamName} update after ${timeoutMs}ms`));
+        reject(new ScryptRequestTimeoutError(`Timeout waiting for ${streamName} update after ${timeoutMs}ms`));
       }, timeoutMs);
 
       const unsubscribe = this.subscribe(streamName, (data) => {
@@ -440,7 +523,7 @@ export class ScryptWebSocketConnection {
     return new Promise((resolve, reject) => {
       const timeout = setTimeout(() => {
         this.pendingRequests.delete(reqId);
-        reject(new Error(`Request timeout after ${timeoutMs}ms`));
+        reject(new ScryptRequestTimeoutError(`Request timeout after ${timeoutMs}ms`));
       }, timeoutMs);
 
       this.pendingRequests.set(reqId, { resolve, reject, timeout });
@@ -459,7 +542,11 @@ export class ScryptWebSocketConnection {
 
     if (message.type === ScryptMessageType.ERROR) {
       const errorMsg = typeof message.error === 'object' ? JSON.stringify(message.error) : message.error;
-      request.reject(new Error(`Scrypt error: ${errorMsg}`));
+      // The venue answered this specific request negatively. Whether that settles the outcome depends on the
+      // reason — a malformed order is settled, a lost session is not — and Scrypt does not document its
+      // codes, so this stays a distinct type and the caller decides. Never silently a plain Error: that is
+      // what let a refusal look like a transport hiccup.
+      request.reject(new ScryptErrorResponseError(`Scrypt error: ${errorMsg}`));
     } else {
       request.resolve(message);
     }