fix(address-comments): derive the Atlas event_id and bound the SQS publish

- src/models/atlas/atlas-event-id.ts — derive event_id as a UUIDv5 of
  swap_id, event_type and the rsk tx hash, so the MessageDeduplicationId
  has something to deduplicate on (per @Dominikkq)
- src/services/atlas/{pegin,pegout}-atlas-event.builder.ts — use it in
  place of randomUUID
- src/services/atlas/sqs-atlas-event-publisher.ts — bound the publish
  with connection/request timeouts and maxAttempts, so a hung queue
  cannot strand the rest of a block (per @Dominikkq); requestTimeout
  needs throwOnRequestTimeout to abort rather than only warn
- ENV_VARIABLES.md — document the three new variables and both fixes

Author: ronaldsg20

.env.test

@@ -5,7 +5,7 @@ RSK_PEGOUT_MINIMUM_CONFIRMATIONS=10
 FAST_MINING_BLOCK=1
 AVERAGE_MINING_BLOCK=6
 LOW_MINING_BLOCK=12
-BLOCKBOOK_URL='https://blockbook-01.testnet.2wp.iovlabs.net:19130/'
+BLOCKBOOK_URL='https://blockbook.testnet.2wp.iovlabs.net:19130/'
 MAX_AMOUNT_ALLOWED_IN_SATOSHI=100000000
 
 # Federation Addresses history

ENV_VARIABLES.md

@@ -41,6 +41,9 @@ This table was created to guide and centralize the **environment variables** nec
 |ATLAS_SQS_QUEUE_URL           |                               |'URL of the SQS FIFO queue the Atlas events are published to. Required when `ATLAS_EVENTS_ENABLED=true`; the daemon aborts at startup without it'|
 |AWS_REGION                    |`us-east-1`                    |'AWS region of the Atlas SQS queue'                      |
 |ATLAS_SQS_ENDPOINT            |`http://localhost:4566`        |'Custom SQS endpoint. Local development and tests only (LocalStack); leave empty in deployments'|
+|ATLAS_SQS_REQUEST_TIMEOUT_MS  |`5000`                         |'How long to wait for the queue to answer a publish before giving up. Values of zero or less are rejected and fall back to the default'|
+|ATLAS_SQS_CONNECTION_TIMEOUT_MS|`2000`                        |'How long to wait for the connection to the queue to be established'|
+|ATLAS_SQS_MAX_ATTEMPTS        |`3`                            |'Attempts per publish, retries included'                 |
 
 ### Atlas SWAP events
 
@@ -85,6 +88,15 @@ ordered while different swaps are processed in parallel. The queue must have
 content based deduplication **disabled**: `MessageDeduplicationId` is the
 `event_id`.
 
+That only works because `event_id` is **derived, never random**: it is the
+UUIDv5 of `swap_id`, `event_type` and the Rootstock transaction the transition
+was read from (`models/atlas/atlas-event-id.ts`). A reprocessed block — a
+restart, a re-scan — therefore derives the id it derived the first time and the
+queue drops the duplicate. Note the limit: the FIFO deduplication window is five
+minutes, so this is not idempotency by itself. What it is, is the precondition
+for it — an `event_id` that means the same thing on every emission is what lets
+Atlas deduplicate on the consumer side, where there is no window.
+
 The network travels in the chain ids (`rootstock_testnet` / `bitcoin_testnet`),
 derived from `NETWORK`. Because a wrong network would silently contaminate the
 analytics database, `NETWORK` is validated when the daemon starts and the daemon
@@ -99,6 +111,22 @@ Publication happens after the status has been written to Mongo and never fails
 the caller: if SQS is unreachable the failure is logged at error level and block
 processing continues. Events lost in that window are not recovered.
 
+Every publish is bounded in time. The AWS SDK sets no request timeout by
+default, so a queue endpoint that accepts the connection and then never answers
+would leave the publish pending forever — and because `RskChainSyncService`
+commits the block pointer *before* it notifies its subscribers, and does not
+await them, that hang would not stall the daemon anywhere visible: it would
+strand the rest of that block's Bridge transactions, unwritten and unlogged, on
+a block already recorded as processed. `ATLAS_SQS_REQUEST_TIMEOUT_MS` and
+`ATLAS_SQS_CONNECTION_TIMEOUT_MS` close that, and `ATLAS_SQS_MAX_ATTEMPTS` caps
+the retries, so the worst case is roughly `maxAttempts * requestTimeout` plus
+the SDK's backoff and a hang arrives at the error branch above as a timeout.
+
+One SDK detail worth keeping in mind if these are ever retuned: `requestTimeout`
+on its own only **logs a warning** and lets the request hang anyway. The
+publisher passes `throwOnRequestTimeout: true` alongside it, which is what makes
+the timeout actually abort.
+
 Every publication, successful or not, logs one line carrying
 `metric: 'atlas_events_published_total'` with `status`, `flow`, `eventType` and
 the running `total`. **That field name is the contract with the log aggregator**

src/__tests__/integration/atlas-pegout-events.integration.ts

@@ -230,6 +230,51 @@ describe('Integration: Atlas peg events over SQS', function () {
     expect(messages[0].Attributes?.MessageDeduplicationId).to.equal(event.event_id);
   });
 
+  // The test above re-sends one event object, which would pass even if the id
+  // were random. This is the case that actually matters: a reprocessed block
+  // rebuilds the event from scratch, and only a derived event_id gives the
+  // queue the same MessageDeduplicationId to drop it on.
+  it('deduplicates a transition rebuilt from scratch, as a reprocessed block would', async () => {
+    const swapId = givenSwapId();
+
+    await publisher.publish(
+      PegoutAtlasEventBuilder.build(givenPegout(swapId, PegoutStatuses.RECEIVED))!,
+    );
+    await publisher.publish(
+      PegoutAtlasEventBuilder.build(givenPegout(swapId, PegoutStatuses.RECEIVED))!,
+    );
+
+    const messages = await drain(10);
+    expect(messages).to.have.length(1);
+    expectValid(parse(messages[0]));
+  });
+
+  // The other half of the contract: deduplication must not swallow the genuine
+  // transitions that follow, which share the swap_id and differ by event type.
+  it('keeps the later transitions of the same peg-out', async () => {
+    const swapId = givenSwapId();
+
+    await publisher.publish(
+      PegoutAtlasEventBuilder.build(givenPegout(swapId, PegoutStatuses.RECEIVED))!,
+    );
+    await publisher.publish(
+      PegoutAtlasEventBuilder.build(givenPegout(swapId, PegoutStatuses.RECEIVED))!,
+    );
+    await publisher.publish(
+      PegoutAtlasEventBuilder.build(
+        givenPegout(swapId, PegoutStatuses.WAITING_FOR_CONFIRMATION),
+      )!,
+    );
+
+    const messages = await drain(10, 2);
+    const received = messages.map(parse);
+    received.forEach(expectValid);
+    expect(received.map(event => event.event_type)).to.eql([
+      AtlasEventType.SWAP_CREATED,
+      AtlasEventType.SWAP_PENDING,
+    ]);
+  });
+
   it('delivers a rejected peg-out as a single message', async () => {
     const swapId = givenSwapId();
     const rejected = PegoutAtlasEventBuilder.build(

src/__tests__/unit/models/atlas/atlas-event-id.unit.ts

@@ -0,0 +1,78 @@
+import {expect} from '@loopback/testlab';
+import {atlasEventId} from '../../../../models/atlas/atlas-event-id';
+import {AtlasEventType} from '../../../../models/atlas/atlas-event.model';
+
+/** The `event_id` pattern of schemas/atlas-swap-event.schema.json. */
+const UUID_PATTERN = /^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$/;
+/** Version 5 in the 13th nibble, RFC 4122 variant in the 17th. */
+const UUID_V5_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-5[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/;
+
+const SWAP_ID = '0x8e0b47b0c60f7e02b41ee1b7d4f0d4e3f9a1c2b3d4e5f60718293a4b5c6d7e8f';
+const RSK_TX_HASH = '0x5b2f1a0c9d8e7f60514233a4b5c6d7e8f90a1b2c3d4e5f60718293a4b5c6d7e8';
+
+describe('Model: atlasEventId', () => {
+
+  it('derives the same id for the same event, which is what makes deduplication work', () => {
+    const first = atlasEventId(SWAP_ID, AtlasEventType.SWAP_CREATED, RSK_TX_HASH);
+    const second = atlasEventId(SWAP_ID, AtlasEventType.SWAP_CREATED, RSK_TX_HASH);
+
+    expect(first).to.equal(second);
+  });
+
+  it('produces an id the schema accepts', () => {
+    const id = atlasEventId(SWAP_ID, AtlasEventType.SWAP_COMPLETED, RSK_TX_HASH);
+
+    expect(id).to.match(UUID_PATTERN);
+  });
+
+  // A consumer reading the version nibble is entitled to find a real one.
+  it('produces a well formed version 5 uuid, not a hash cut into shape', () => {
+    const id = atlasEventId(SWAP_ID, AtlasEventType.SWAP_COMPLETED, RSK_TX_HASH);
+
+    expect(id).to.match(UUID_V5_PATTERN);
+  });
+
+  it('separates the transitions of one swap', () => {
+    const ids = [
+      AtlasEventType.SWAP_CREATED,
+      AtlasEventType.SWAP_PENDING,
+      AtlasEventType.SWAP_COMPLETED,
+      AtlasEventType.SWAP_REJECTED,
+    ].map(eventType => atlasEventId(SWAP_ID, eventType, RSK_TX_HASH));
+
+    expect(new Set(ids).size).to.equal(ids.length);
+  });
+
+  it('separates two swaps reporting the same transition', () => {
+    const other = `${SWAP_ID.slice(0, -1)}0`;
+
+    expect(atlasEventId(SWAP_ID, AtlasEventType.SWAP_CREATED, RSK_TX_HASH))
+      .to.not.equal(atlasEventId(other, AtlasEventType.SWAP_CREATED, RSK_TX_HASH));
+  });
+
+  // The batched peg-outs of one Bridge transaction differ only by the index the
+  // processor appends to `rskTxHash`, so that part has to reach the digest.
+  it('separates two transitions differing only by the rsk transaction', () => {
+    const batched = `${RSK_TX_HASH}___1`;
+
+    expect(atlasEventId(SWAP_ID, AtlasEventType.SWAP_COMPLETED, RSK_TX_HASH))
+      .to.not.equal(atlasEventId(SWAP_ID, AtlasEventType.SWAP_COMPLETED, batched));
+  });
+
+  // Same reasoning as normalizeSwapId: one transaction spelled two ways is one
+  // event, and must not be published twice under two ids.
+  it('ignores the spelling of the hashes it is given', () => {
+    const id = atlasEventId(SWAP_ID, AtlasEventType.SWAP_CREATED, RSK_TX_HASH);
+
+    expect(atlasEventId(SWAP_ID.toUpperCase().replace('0X', '0x'), AtlasEventType.SWAP_CREATED, ` ${RSK_TX_HASH} `))
+      .to.equal(id);
+  });
+
+  it('still derives an id when there is no rsk transaction to name', () => {
+    const id = atlasEventId(SWAP_ID, AtlasEventType.SWAP_CREATED, undefined);
+
+    expect(id).to.match(UUID_PATTERN);
+    expect(atlasEventId(SWAP_ID, AtlasEventType.SWAP_CREATED, null)).to.equal(id);
+  });
+
+});

src/__tests__/unit/services/atlas/pegin-atlas-event.builder.unit.ts

@@ -344,11 +344,34 @@ describe('Service: PeginAtlasEventBuilder', () => {
     expect(PeginAtlasEventBuilder.build(pegin, {})).to.be.empty();
   });
 
-  it('generates a distinct event_id per event', () => {
-    const first = PeginAtlasEventBuilder.build(givenPegin(PeginStatus.LOCKED), {});
-    const second = PeginAtlasEventBuilder.build(givenPegin(PeginStatus.LOCKED), {});
+  describe('event_id', () => {
 
-    expect(first[0].event_id).to.not.equal(second[0].event_id);
+    // Same contract as the peg-out builder: the id is the deduplication key, so
+    // rebuilding the same peg-in has to derive the same ids in the same order.
+    it('derives the same ids when the same peg-in is rebuilt', () => {
+      const first = PeginAtlasEventBuilder.build(givenPegin(PeginStatus.LOCKED), {});
+      const second = PeginAtlasEventBuilder.build(givenPegin(PeginStatus.LOCKED), {});
+
+      expect(first.map(event => event.event_id)).to.eql(second.map(event => event.event_id));
+    });
+
+    // Both events of a peg-in come from one Rootstock transaction, which leaves
+    // the event type as the only thing separating them.
+    it('gives the two events of one peg-in distinct ids', () => {
+      const [created, completed] = PeginAtlasEventBuilder.build(givenPegin(PeginStatus.LOCKED), {});
+
+      expect(created.event_id).to.not.equal(completed.event_id);
+    });
+
+    it('gives two peg-ins reporting the same transition distinct ids', () => {
+      const other = givenPegin(PeginStatus.LOCKED);
+      other.btcTxId = `${btcTxId.slice(0, -1)}1`;
+
+      const [created] = PeginAtlasEventBuilder.build(givenPegin(PeginStatus.LOCKED), {});
+      const [otherCreated] = PeginAtlasEventBuilder.build(other, {});
+
+      expect(created.event_id).to.not.equal(otherCreated.event_id);
+    });
   });
 
   it('throws when NETWORK is not configured instead of guessing the network', () => {

src/__tests__/unit/services/atlas/pegout-atlas-event.builder.unit.ts

@@ -353,9 +353,48 @@ describe('Service: PegoutAtlasEventBuilder', () => {
     });
   });
 
-  it('generates a distinct event_id per event', () => {
-    const first = PegoutAtlasEventBuilder.build(givenPegout({status: PegoutStatuses.RECEIVED}))!;
-    const second = PegoutAtlasEventBuilder.build(givenPegout({status: PegoutStatuses.RECEIVED}))!;
-    expect(first.event_id).to.not.equal(second.event_id);
+  describe('event_id', () => {
+
+    // The publisher sends it as the MessageDeduplicationId, so a reprocessed
+    // block has to derive the id it derived the first time or the queue has
+    // nothing to deduplicate on.
+    it('derives the same id when the same transition is rebuilt', () => {
+      const first = PegoutAtlasEventBuilder.build(givenPegout({status: PegoutStatuses.RECEIVED}))!;
+      const second = PegoutAtlasEventBuilder.build(givenPegout({status: PegoutStatuses.RECEIVED}))!;
+
+      expect(first.event_id).to.equal(second.event_id);
+    });
+
+    it('gives each transition of one peg-out its own id', () => {
+      const received = PegoutAtlasEventBuilder.build(givenPegout({
+        status: PegoutStatuses.RECEIVED,
+      }))!;
+      const waiting = PegoutAtlasEventBuilder.build(givenPegout({
+        status: PegoutStatuses.WAITING_FOR_CONFIRMATION,
+      }))!;
+
+      expect(received.event_id).to.not.equal(waiting.event_id);
+    });
+
+    // Batched peg-outs share the Bridge transaction and are told apart by the
+    // index the processor appends to rskTxHash.
+    it('gives the batched peg-outs of one Bridge transaction their own ids', () => {
+      const first = PegoutAtlasEventBuilder.build(givenPegout({
+        status: PegoutStatuses.WAITING_FOR_CONFIRMATION,
+        rskTxHash: `${originatingRskTxHash}___0`,
+      }))!;
+      const second = PegoutAtlasEventBuilder.build(givenPegout({
+        status: PegoutStatuses.WAITING_FOR_CONFIRMATION,
+        rskTxHash: `${originatingRskTxHash}___1`,
+      }))!;
+
+      expect(first.event_id).to.not.equal(second.event_id);
+    });
+
+    it('derives an id the schema accepts', () => {
+      expectValidAgainstSchema(
+        PegoutAtlasEventBuilder.build(givenPegout({status: PegoutStatuses.RECEIVED})),
+      );
+    });
   });
 });

src/__tests__/unit/services/atlas/sqs-atlas-event-publisher.unit.ts

@@ -1,3 +1,4 @@
+import * as net from 'net';
 import {Application} from '@loopback/core';
 import {expect, sinon} from '@loopback/testlab';
 import {SQSClient} from '@aws-sdk/client-sqs';
@@ -12,6 +13,7 @@ import {isAtlasEventsEnabled} from '../../../../services/atlas/atlas-event-publi
 import {
   SqsAtlasEventPublisher,
   assertQueueUrlConfigured,
+  positiveIntFromEnv,
 } from '../../../../services/atlas/sqs-atlas-event-publisher';
 import {NoopAtlasEventPublisher} from '../../../../services/atlas/noop-atlas-event-publisher';
 import {DependencyInjectionHandler} from '../../../../dependency-injection-handler';
@@ -103,6 +105,87 @@ describe('Service: SqsAtlasEventPublisher', () => {
     await publisher.publish(event);
   });
 
+  describe('request timeout', () => {
+    const originalCredentials = {
+      keyId: process.env.AWS_ACCESS_KEY_ID,
+      secret: process.env.AWS_SECRET_ACCESS_KEY,
+      requestTimeout: process.env.ATLAS_SQS_REQUEST_TIMEOUT_MS,
+      maxAttempts: process.env.ATLAS_SQS_MAX_ATTEMPTS,
+    };
+    let server: net.Server;
+    const connections: net.Socket[] = [];
+
+    /**
+     * Accepts the connection and never answers, which is the failure a
+     * connection timeout alone does not catch.
+     */
+    before(async () => {
+      server = net.createServer(socket => {
+        // Kept only so the teardown can destroy them: the SDK leaves the socket
+        // alive, and `close()` waits for every connection before it fires.
+        connections.push(socket);
+      });
+      await new Promise<void>(resolve => server.listen(0, '127.0.0.1', resolve));
+    });
+
+    after(async () => {
+      connections.forEach(socket => socket.destroy());
+      await new Promise<void>(resolve => { server.close(() => resolve()); });
+      restore('AWS_ACCESS_KEY_ID', originalCredentials.keyId);
+      restore('AWS_SECRET_ACCESS_KEY', originalCredentials.secret);
+      restore('ATLAS_SQS_REQUEST_TIMEOUT_MS', originalCredentials.requestTimeout);
+      restore('ATLAS_SQS_MAX_ATTEMPTS', originalCredentials.maxAttempts);
+    });
+
+    // Without a request timeout this publish never settles. The sync service
+    // commits the block pointer before it notifies its subscribers and does not
+    // await them, so the hang would strand the rest of the block silently
+    // rather than stall the daemon somewhere visible.
+    it('gives up on a queue that accepts the connection and never answers', async () => {
+      const {port} = server.address() as net.AddressInfo;
+      process.env.ATLAS_SQS_ENDPOINT = `http://127.0.0.1:${port}`;
+      process.env.ATLAS_SQS_REQUEST_TIMEOUT_MS = '150';
+      process.env.ATLAS_SQS_MAX_ATTEMPTS = '1';
+      process.env.AWS_ACCESS_KEY_ID = 'test';
+      process.env.AWS_SECRET_ACCESS_KEY = 'test';
+      const publisher = new SqsAtlasEventPublisher();
+
+      const startedAt = Date.now();
+      await publisher.publish(event, 'pegout');
+      const elapsed = Date.now() - startedAt;
+
+      expect(elapsed).to.be.lessThan(5000);
+      expect(publisher.metrics.total('failure', event.event_type, 'pegout')).to.equal(1);
+      expect(publisher.metrics.total('success', event.event_type, 'pegout')).to.equal(0);
+      publisher.destroy();
+    });
+  });
+
+  describe('timeout configuration', () => {
+    const NAME = 'ATLAS_SQS_TEST_VALUE';
+
+    afterEach(() => delete process.env[NAME]);
+
+    it('takes a positive value from the environment', () => {
+      process.env[NAME] = '750';
+
+      expect(positiveIntFromEnv(NAME, 5000)).to.equal(750);
+    });
+
+    it('falls back when the variable is absent', () => {
+      expect(positiveIntFromEnv(NAME, 5000)).to.equal(5000);
+    });
+
+    // Zero and negative are how the SDK spells "wait forever", which is the
+    // failure the timeout exists to prevent: they are not valid settings.
+    it('falls back rather than accepting a value that disables the timeout', () => {
+      for (const value of ['0', '-1', '-4000', '', 'soon', 'NaN']) {
+        process.env[NAME] = value;
+        expect(positiveIntFromEnv(NAME, 5000)).to.equal(5000);
+      }
+    });
+  });
+
   describe('publication metric', () => {
     it('records a success when SQS accepts the message', async () => {
       sandbox.stub(SQSClient.prototype, 'send').resolves({MessageId: 'id'} as never);

src/__tests__/unit/services/btc-last-block.service.unit.ts

@@ -50,7 +50,7 @@ describe('Service: BitcoinService', () => {
 
     it('Verify ${process.env.BLOCKBOOK_URL} configuration', async () => {
         const nodeHost = process.env.BLOCKBOOK_URL;
-        sinon.assert.match(nodeHost, 'https://blockbook-01.testnet.2wp.iovlabs.net:19130/');
+        sinon.assert.match(nodeHost, 'https://blockbook.testnet.2wp.iovlabs.net:19130/');
     });
 
 });