diff --git a/README.md b/README.md new file mode 100644 index 0000000..0a08e7d --- /dev/null +++ b/README.md @@ -0,0 +1,145 @@ +# Transaction processor controller + +This service listens for wallet bootstrap requests and creates one +`txn-processor-dynamic` Kubernetes pod per grantee wallet. + +It does not receive individual blockchain messages. Those messages are already +waiting in the wallet-specific queue consumed by the dynamic pod. + +## Responsibilities + +1. Consume a bootstrap request from `GLOBAL_TXN_CONTROLLER_QUEUE`. +2. Build the tenant-specific MongoDB URL. +3. Check whether the wallet's processor pod already exists. +4. Create the pod when needed and pass the bootstrap fields as environment + variables. +5. Monitor the pod and delete it after completion, failure, or a prolonged + pending state. +6. Move bootstrap requests that cannot be processed to the controller DLQ. + +The Kubernetes namespace is currently `hypermine-development`. Dynamic pod +images use: + +```text +ghcr.io/hypersign-protocol/txn-processor-dynamic: +``` + +## RabbitMQ queues + +| Queue | Default | Durability | Producer/consumer | +| ------------------------ | ----------------------------- | ----------- | ----------------------------------- | +| Controller queue | `GLOBAL_TXN_CONTROLLER_QUEUE` | Non-durable | Entity API -> controller | +| Controller DLQ | `GLOBAL_TXN_CONTROLLER_DLQ` | Durable | Controller failures and retry drain | +| Wallet transaction queue | `TXN_QUEUE_` | Non-durable | Entity API -> dynamic processor | + +The controller DLQ preserves the original bootstrap message and adds: + +- `x-dlq-retry-count` +- `x-dlq-reason` +- `x-dlq-entered-at` + +The DLQ is periodically drained back into the controller queue. Messages are +discarded after `MAX_DLQ_RETRIES`. + +## Bootstrap message + +The entity API publishes a JSON object to `GLOBAL_TXN_CONTROLLER_QUEUE`. +The controller copies every field into the dynamic pod environment and adds the +calculated `DB_URL`. + +```json +{ + "RMQ_URL": "amqp://rabbitmq:5672", + "QUEUE_NAME": "TXN_QUEUE_hid1abc...", + "NODE_RPC_URL": "https://rpc.example", + "GRANTEE_MNEMONIC": "", + "GRANTER_ADDRESS": "hid1granter...", + "DID_REGISTER_FIXED_FEE": "4000", + "DID_UPDATE_FIXED_FEE": "1000", + "DID_DEACTIVATE_FIXED_FEE": "1000", + "CRED_REGISTER_FIXED_FEE": "2000", + "CRED_UPDATE_FIXED_FEE": "2000", + "SCHEMA_CREATE_FIXED_FEE": "2000", + "SCHEMA_UPDATE_FIXED_FEE": "2000", + "ESTIMATE_GAS_PRICE": "155303", + "podName": "txn-dynamic", + "granteeWalletAddress": "hid1abc...", + "tenent": "tenant-subdomain", + "Tx_Query_API": "https://api.example/cosmos/tx/v1beta1/txs/", + "SSI_TXN_RESULT_EXCHANGE": "ssi.txn.results" +} +``` + +`tenent` is intentionally shown with the existing misspelling. The controller +currently reads that exact property when constructing the MongoDB URL. Renaming +it requires a coordinated entity API and controller change. + +`Tx_Query_API` is also the spelling currently sent by the entity API. The +dynamic worker reads `TX_QUERY_API`, so a custom value is not applied today and +the worker uses its default endpoint. Correcting the name requires a coordinated +entity API deployment. + +The dynamic pod name is: + +```text +- +``` + +If that pod is already running or pending, the controller acknowledges the +bootstrap request without creating another pod. + +## Controller environment variables + +| Variable | Required | Default | Purpose | +| ----------------------------- | -------- | ----------------------------- | ------------------------------------------ | +| `AMQ_URL` | Yes | None | RabbitMQ connection used by the controller | +| `GLOBAL_TXN_CONTROLLER_QUEUE` | No | `GLOBAL_TXN_CONTROLLER_QUEUE` | Bootstrap queue name | +| `GLOBAL_TXN_CONTROLLER_DLQ` | No | `GLOBAL_TXN_CONTROLLER_DLQ` | Failed-bootstrap queue | +| `MAX_DLQ_RETRIES` | No | `5` | Maximum DLQ processing attempts | +| `DLQ_DRAIN_INTERVAL_MS` | No | `300000` | Delay between DLQ drain passes | +| `TXN_PROCESSOR_DYNAMIC_TAG` | Yes | None | Dynamic processor container tag | +| `DB_URL` | Yes | None | MongoDB server/base connection string | +| `PREFIX` | Yes | None | Prefix added before the tenant identifier | +| `DB_CONFIG` | Yes | None | MongoDB connection suffix/options | +| `LOG_LEVEL` | No | `info` | `error`, `warn`, `info`, or `debug` | + +The controller uses the cluster's default Kubernetes configuration. Its runtime +identity needs permission to read, create, and delete pods in +`hypermine-development`, and read the `mongo` secret mounted into dynamic +pods. + +## Settlement configuration + +The controller does not publish settlement events, but it must pass +`SSI_TXN_RESULT_EXCHANGE` from the entity API bootstrap message to the dynamic +pod. The default exchange name across the services is: + +```text +ssi.txn.results +``` + +The Developer Dashboard expects: + +| Setting | Default | +| ------------------------- | ------------------------------------- | +| `RABBIT_MQ_URI` | Required; no default | +| `SSI_TXN_RESULT_EXCHANGE` | `ssi.txn.results` | +| `SSI_TXN_RESULT_QUEUE` | `developer-dashboard.ssi.txn-results` | +| `SSI_TXN_UNKNOWN_QUEUE` | `developer-dashboard.ssi.txn-unknown` | + +The normal result queue binds `ssi.txn.succeeded` and `ssi.txn.failed`. +The unknown queue binds `ssi.txn.unknown` for reconciliation. + +All participating applications must use the same RabbitMQ broker and exchange +name. + +## Local start + +```bash +npm install +npm start +``` + +The local process also needs working Kubernetes credentials. Starting the +controller without cluster access will connect to RabbitMQ but fail when it +tries to inspect or create a pod. diff --git a/src/index.js b/src/index.js index 82fb9da..6c797f6 100644 --- a/src/index.js +++ b/src/index.js @@ -181,12 +181,62 @@ const deploy = async (name, // console.log(data.body.status.phase); } catch (err) { log('error', err); + throw err; // let the consumer catch block handle it } }; const queueName = process.env.GLOBAL_TXN_CONTROLLER_QUEUE || 'GLOBAL_TXN_CONTROLLER_QUEUE'; +const dlqName = process.env.GLOBAL_TXN_CONTROLLER_DLQ || 'GLOBAL_TXN_CONTROLLER_DLQ'; +const MAX_DLQ_RETRIES = parseInt(process.env.MAX_DLQ_RETRIES || '5'); +const DLQ_DRAIN_INTERVAL_MS = parseInt(process.env.DLQ_DRAIN_INTERVAL_MS || '300000'); // 5 min + +// Send message to DLQ, preserving original content. Tracks retry count in headers. +const sendToDLQ = (channel, message, errorReason) => { + const retryCount = message.properties.headers?.['x-dlq-retry-count'] || 0; + if (retryCount >= MAX_DLQ_RETRIES) { + log('error', `Message permanently discarded after ${MAX_DLQ_RETRIES} DLQ retries. Reason: ${errorReason}`); + return; + } + channel.sendToQueue(dlqName, message.content, { + persistent: true, + headers: { + ...message.properties.headers, + 'x-dlq-retry-count': retryCount + 1, + 'x-dlq-reason': String(errorReason).slice(0, 500), + 'x-dlq-entered-at': new Date().toISOString() + } + }); + log('warn', `Message sent to DLQ (attempt ${retryCount + 1}/${MAX_DLQ_RETRIES}): ${errorReason}`); +}; + +// Drain DLQ by republishing messages back to the main queue in their original format. +const drainDLQ = async (channel) => { + let count = 0; + try { + while (true) { + const msg = await channel.get(dlqName, { noAck: false }); + if (!msg) break; + const retryCount = msg.properties.headers?.['x-dlq-retry-count'] || 0; + if (retryCount >= MAX_DLQ_RETRIES) { + log('error', `Permanently discarding DLQ message after ${retryCount} retries`); + channel.ack(msg); + continue; + } + // Republish original content back to main queue — consumer will process it normally + channel.sendToQueue(queueName, msg.content, { + persistent: false, + headers: msg.properties.headers + }); + channel.ack(msg); + count++; + } + } catch (err) { + log('error', `DLQ drain error: ${err.message}`); + } + if (count > 0) log('info', `DLQ drained: ${count} message(s) republished to main queue`); +}; (async () => { try { @@ -197,11 +247,42 @@ const queueName = process.env.GLOBAL_TXN_CONTROLLER_QUEUE || 'GLOBAL_TXN_CONTROL const connection = await amqp.connect(process.env.AMQ_URL, { heartbeat: 30 }) + + let channelOpen = true; + + connection.on('error', (err) => { + log('error', `AMQP connection error: ${err.message}`); + }); + connection.on('close', () => { + log('error', 'AMQP connection closed unexpectedly. Exiting for restart...'); + channelOpen = false; + clearInterval(drainInterval); + process.exit(1); + }); + const channel = await connection.createChannel(); - await channel.assertQueue(queueName, { - durable: false, - }) + channel.on('error', (err) => { + log('error', `AMQP channel error: ${err.message}`); + channelOpen = false; + }); + channel.on('close', () => { + log('error', 'AMQP channel closed. Exiting for restart...'); + channelOpen = false; + clearInterval(drainInterval); + process.exit(1); + }); + + await channel.assertQueue(queueName, { durable: false }); + await channel.assertQueue(dlqName, { durable: true }); + const drainInterval = setInterval(() => { + if (!channelOpen) { + log('warn', 'Skipping DLQ drain: channel is not open'); + return; + } + drainDLQ(channel); + }, DLQ_DRAIN_INTERVAL_MS); + log('info', `DLQ drain scheduled every ${DLQ_DRAIN_INTERVAL_MS / 1000}s`); await channel.consume(queueName, async (message) => { let queueMsg; log('debug', 'Trying to consume') @@ -241,18 +322,25 @@ const queueName = process.env.GLOBAL_TXN_CONTROLLER_QUEUE || 'GLOBAL_TXN_CONTROL } catch (error) { log('error', error.message); - channel.nack(message, false, false) - + if (channelOpen) { + sendToDLQ(channel, message, error.message); + channel.ack(message); + } else { + log('warn', 'Channel closed during message processing; message will be requeued on restart'); + } } }) - process.on('SIGINT', async () => { - log('info', 'Closing RabbitMQ connection...'); - await channel.close(); - await connection.close(); + const shutdown = async (signal) => { + log('info', `${signal} received, shutting down gracefully...`); + clearInterval(drainInterval); + try { await channel.close(); } catch (_) {} + try { await connection.close(); } catch (_) {} process.exit(0); - }); + }; + process.on('SIGTERM', () => shutdown('SIGTERM')); // k8s sends this + process.on('SIGINT', () => shutdown('SIGINT')); // local dev Ctrl+C } catch (error) { log('error', error.message) }