From 53708b541dcbc649c207a94a3f583d2dfa33696d Mon Sep 17 00:00:00 2001 From: d-klotz Date: Fri, 1 May 2026 07:45:45 -0300 Subject: [PATCH 1/5] feat(devtools): expose Integration.Definition.queue tuning knobs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an optional `Definition.queue` block on each integration that lets consumers express SQS + queue-worker tuning declaratively in their app definition, instead of patching it in via post-deploy `aws lambda update-event-source-mapping` / `aws sqs set-queue-attributes` scripts. Knobs (all optional, defaults preserved): queue.visibilityTimeout SQS VisibilityTimeout [0..43200] default 1800 queue.messageRetentionPeriod SQS MessageRetentionPeriod [60..1209600] default 345600 queue.maxReceiveCount DLQ redrive maxReceiveCount [1..1000] default 3 queue.worker.batchSize SQS event batchSize [1..10000] default 1 queue.worker.maximumBatchingWindow SQS MaximumBatchingWindowInSeconds [0..300] default unset queue.worker.maximumConcurrency ESM ScalingConfig.MaximumConcurrency [2..1000] default unset queue.worker.reservedConcurrency Lambda ReservedConcurrency [0..1000] default 20 queue.worker.timeout Lambda execution timeout (s) [1..900] default 900 Implementation notes: - All knobs optional; falls through to current defaults via `??`. Apps that don't set `Definition.queue` produce identical output (verified by a backward-compat test). - maximumBatchingWindow / maximumConcurrency only emitted to the serverless template when defined — keeps the rendered template stable for non-opt-in consumers. - Uses the serverless-framework input key `maximumBatchingWindow`, NOT the longer CFN property name `maximumBatchingWindowInSeconds`. osls and serverless@3+ reject the longer key with a schema error. - Bounds-checked at template-generation time (consistent with aurora-builder's `minCapacity`/`maxCapacity` validation), so out-of-range values surface as clear errors instead of opaque CloudFormation rejections post-package. Downstream prerequisite for `worker.batchSize > 1`: the queue worker handler must return `{ batchItemFailures: [...] }` for partial-batch failures (the framework already wires `functionResponseType: 'ReportBatchItemFailures'` — same as before). With batchSize=1 this was moot; with batchSize>1 it isn't. Tests: 97 pass (89 existing + 8 new covering override paths, empty-config edge cases, conditional-spread invariants, and bounds-check failures). Co-Authored-By: Claude Opus 4.7 (1M context) --- .../integration/integration-builder.js | 104 +++++++++--- .../integration/integration-builder.test.js | 158 ++++++++++++++++++ .../domains/shared/types/app-definition.js | 18 ++ 3 files changed, 259 insertions(+), 21 deletions(-) diff --git a/packages/devtools/infrastructure/domains/integration/integration-builder.js b/packages/devtools/infrastructure/domains/integration/integration-builder.js index a27cd6dc7..11bc00c37 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.js @@ -30,6 +30,51 @@ const { getAdminFunctionNames, } = require('../shared/function-environments'); +/** + * Bounds-check Definition.queue knobs against AWS SQS limits. + * Throws on out-of-range values so we fail at template-generation time + * instead of waiting for an opaque CloudFormation rejection. + */ +function validateQueueConfig(integrationName, queueConfig) { + const inRange = (val, min, max) => + val === undefined || (Number.isFinite(val) && val >= min && val <= max); + const checks = [ + ['visibilityTimeout', queueConfig.visibilityTimeout, 0, 43200], + ['messageRetentionPeriod', queueConfig.messageRetentionPeriod, 60, 1209600], + ['maxReceiveCount', queueConfig.maxReceiveCount, 1, 1000], + ]; + for (const [key, val, min, max] of checks) { + if (!inRange(val, min, max)) { + throw new Error( + `Integration '${integrationName}': queue.${key}=${val} is out of range [${min}, ${max}]` + ); + } + } +} + +/** + * Bounds-check Definition.queue.worker knobs against the serverless + * framework + AWS Lambda+SQS limits. Mirrors osls's own input schema. + */ +function validateWorkerConfig(integrationName, workerConfig) { + const inRange = (val, min, max) => + val === undefined || (Number.isFinite(val) && val >= min && val <= max); + const checks = [ + ['batchSize', workerConfig.batchSize, 1, 10000], + ['maximumBatchingWindow', workerConfig.maximumBatchingWindow, 0, 300], + ['maximumConcurrency', workerConfig.maximumConcurrency, 2, 1000], + ['reservedConcurrency', workerConfig.reservedConcurrency, 0, 1000], + ['timeout', workerConfig.timeout, 1, 900], + ]; + for (const [key, val, min, max] of checks) { + if (!inRange(val, min, max)) { + throw new Error( + `Integration '${integrationName}': queue.worker.${key}=${val} is out of range [${min}, ${max}]` + ); + } + } +} + class IntegrationBuilder extends InfrastructureBuilder { constructor() { super(); @@ -226,7 +271,8 @@ class IntegrationBuilder extends InfrastructureBuilder { this.createIntegrationQueue( integrationName, result, - appDefinition + appDefinition, + integration.Definition.queue ); } else { console.log(` ✓ Using external ${integrationName}Queue`); @@ -425,27 +471,36 @@ class IntegrationBuilder extends InfrastructureBuilder { // Create Queue Worker function const queueWorkerName = `${integrationName}QueueWorker`; + const workerConfig = integration.Definition.queue?.worker || {}; + validateWorkerConfig(integrationName, workerConfig); + const sqsEvent = { + arn: { + 'Fn::GetAtt': [ + `${this.capitalizeFirst(integrationName)}Queue`, + 'Arn', + ], + }, + batchSize: workerConfig.batchSize ?? 1, + functionResponseType: 'ReportBatchItemFailures', + }; + // The serverless framework SQS event accepts `maximumBatchingWindow` + // (not `...InSeconds` — that's the AWS-side CFN property name). osls + // and serverless@3+ both reject the longer key with a schema error. + if (workerConfig.maximumBatchingWindow !== undefined) { + sqsEvent.maximumBatchingWindow = + workerConfig.maximumBatchingWindow; + } + if (workerConfig.maximumConcurrency !== undefined) { + sqsEvent.maximumConcurrency = workerConfig.maximumConcurrency; + } result.functions[queueWorkerName] = { handler: `node_modules/@friggframework/core/handlers/workers/integration-defined-workers.handlers.${integrationName}.queueWorker`, skipEsbuild: true, // Nested exports in node_modules - skip esbuild bundling package: functionPackageConfig, ...(usePrismaLayer && { layers: [{ Ref: 'PrismaLambdaLayer' }] }), // Queue workers need Prisma for database operations - reservedConcurrency: 20, - events: [ - { - sqs: { - arn: { - 'Fn::GetAtt': [ - `${this.capitalizeFirst(integrationName)}Queue`, - 'Arn', - ], - }, - batchSize: 1, - functionResponseType: 'ReportBatchItemFailures', - }, - }, - ], - timeout: 900, // 15 minutes max for queue workers (Lambda maximum) + reservedConcurrency: workerConfig.reservedConcurrency ?? 20, + events: [{ sqs: sqsEvent }], + timeout: workerConfig.timeout ?? 900, // 15 minutes max for queue workers (Lambda maximum) }; console.log(` ✓ Queue worker function defined`); } @@ -554,7 +609,13 @@ class IntegrationBuilder extends InfrastructureBuilder { /** * Create integration-specific SQS queue CloudFormation resource */ - createIntegrationQueue(integrationName, result, appDefinition) { + createIntegrationQueue( + integrationName, + result, + appDefinition, + queueConfig = {} + ) { + validateQueueConfig(integrationName, queueConfig); const queueReference = `${this.capitalizeFirst(integrationName)}Queue`; const queueName = `\${self:service}--\${self:provider.stage}-${queueReference}`; @@ -562,10 +623,11 @@ class IntegrationBuilder extends InfrastructureBuilder { Type: 'AWS::SQS::Queue', Properties: { QueueName: `\${self:custom.${queueReference}}`, - MessageRetentionPeriod: 345600, // 4 days (SQS default) - VisibilityTimeout: 1800, + MessageRetentionPeriod: + queueConfig.messageRetentionPeriod ?? 345600, // 4 days (SQS default) + VisibilityTimeout: queueConfig.visibilityTimeout ?? 1800, RedrivePolicy: { - maxReceiveCount: 3, + maxReceiveCount: queueConfig.maxReceiveCount ?? 3, deadLetterTargetArn: { 'Fn::GetAtt': ['InternalErrorQueue', 'Arn'], }, diff --git a/packages/devtools/infrastructure/domains/integration/integration-builder.test.js b/packages/devtools/infrastructure/domains/integration/integration-builder.test.js index 7ddabef10..b4e0b2fc8 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.test.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.test.js @@ -397,6 +397,164 @@ describe('IntegrationBuilder', () => { }); }); + describe('Integration.Definition.queue tuning knobs', () => { + it('overrides VisibilityTimeout, MessageRetentionPeriod, and maxReceiveCount on the queue', async () => { + const appDefinition = { + integrations: [ + { + Definition: { + name: 'tuned', + queue: { + visibilityTimeout: 960, + messageRetentionPeriod: 86400, + maxReceiveCount: 5, + }, + }, + }, + ], + }; + + const result = await integrationBuilder.build(appDefinition, {}); + const props = result.resources.TunedQueue.Properties; + + expect(props.VisibilityTimeout).toBe(960); + expect(props.MessageRetentionPeriod).toBe(86400); + expect(props.RedrivePolicy.maxReceiveCount).toBe(5); + }); + + it('overrides worker batchSize, maximumBatchingWindow, maximumConcurrency, reservedConcurrency, and timeout', async () => { + const appDefinition = { + integrations: [ + { + Definition: { + name: 'tuned', + queue: { + worker: { + batchSize: 5, + maximumBatchingWindow: 2, + maximumConcurrency: 20, + reservedConcurrency: 10, + timeout: 120, + }, + }, + }, + }, + ], + }; + + const result = await integrationBuilder.build(appDefinition, {}); + const fn = result.functions.tunedQueueWorker; + const sqsEvent = fn.events[0].sqs; + + expect(sqsEvent.batchSize).toBe(5); + expect(sqsEvent.maximumBatchingWindow).toBe(2); + expect(sqsEvent.maximumConcurrency).toBe(20); + expect(fn.reservedConcurrency).toBe(10); + expect(fn.timeout).toBe(120); + }); + + it('preserves defaults when queue is omitted (backward compat)', async () => { + const appDefinition = { + integrations: [{ Definition: { name: 'untouched' } }], + }; + + const result = await integrationBuilder.build(appDefinition, {}); + const props = result.resources.UntouchedQueue.Properties; + const fn = result.functions.untouchedQueueWorker; + const sqsEvent = fn.events[0].sqs; + + expect(props.VisibilityTimeout).toBe(1800); + expect(props.MessageRetentionPeriod).toBe(345600); + expect(props.RedrivePolicy.maxReceiveCount).toBe(3); + expect(sqsEvent.batchSize).toBe(1); + expect(sqsEvent.maximumBatchingWindow).toBeUndefined(); + expect(sqsEvent.maximumConcurrency).toBeUndefined(); + expect(fn.reservedConcurrency).toBe(20); + expect(fn.timeout).toBe(900); + }); + + it('handles queue: {} and queue.worker: {} as no-op', async () => { + const appDefinition = { + integrations: [ + { Definition: { name: 'empty', queue: {} } }, + { Definition: { name: 'emptyworker', queue: { worker: {} } } }, + ], + }; + + const result = await integrationBuilder.build(appDefinition, {}); + + expect(result.resources.EmptyQueue.Properties.VisibilityTimeout).toBe(1800); + expect(result.functions.emptyQueueWorker.events[0].sqs.batchSize).toBe(1); + expect(result.functions.emptyworkerQueueWorker.events[0].sqs.batchSize).toBe(1); + }); + + it('does NOT emit maximumBatchingWindow / maximumConcurrency keys when unset', async () => { + // Conditional spread keeps the emitted serverless template stable + // for consumers who don't opt in. Important — adding undefined + // keys would change the serialized template hash. + const appDefinition = { + integrations: [{ Definition: { name: 'notuned' } }], + }; + + const result = await integrationBuilder.build(appDefinition, {}); + const sqsEvent = result.functions.notunedQueueWorker.events[0].sqs; + + expect('maximumBatchingWindow' in sqsEvent).toBe(false); + expect('maximumConcurrency' in sqsEvent).toBe(false); + }); + + it('rejects out-of-range queue.visibilityTimeout (>43200)', async () => { + const appDefinition = { + integrations: [ + { + Definition: { + name: 'bad', + queue: { visibilityTimeout: 99999 }, + }, + }, + ], + }; + + await expect( + integrationBuilder.build(appDefinition, {}) + ).rejects.toThrow(/visibilityTimeout=99999 is out of range/); + }); + + it('rejects out-of-range queue.worker.maximumConcurrency (<2)', async () => { + const appDefinition = { + integrations: [ + { + Definition: { + name: 'bad', + queue: { worker: { maximumConcurrency: 1 } }, + }, + }, + ], + }; + + await expect( + integrationBuilder.build(appDefinition, {}) + ).rejects.toThrow(/worker\.maximumConcurrency=1 is out of range/); + }); + + it('rejects out-of-range queue.worker.timeout (>900)', async () => { + const appDefinition = { + integrations: [ + { + Definition: { + name: 'bad', + queue: { worker: { timeout: 901 } }, + }, + }, + ], + }; + + await expect( + integrationBuilder.build(appDefinition, {}) + ).rejects.toThrow(/worker\.timeout=901 is out of range/); + }); + }); + describe('DLQ Observability', () => { it('should create a CloudWatch alarm for DLQ message depth', async () => { const appDefinition = { diff --git a/packages/devtools/infrastructure/domains/shared/types/app-definition.js b/packages/devtools/infrastructure/domains/shared/types/app-definition.js index 55426bb3b..f1b794fb0 100644 --- a/packages/devtools/infrastructure/domains/shared/types/app-definition.js +++ b/packages/devtools/infrastructure/domains/shared/types/app-definition.js @@ -99,11 +99,29 @@ * @property {boolean} enable - Whether WebSocket API is enabled */ +/** + * Integration queue tuning knobs (optional, all defaults preserved when unset). + * Bounds enforced at template-generation time — see validateQueueConfig / + * validateWorkerConfig in integration-builder.js. + * + * @typedef {Object} IntegrationQueueDefinition + * @property {number} [visibilityTimeout] - SQS VisibilityTimeout seconds. Range 0..43200. Default 1800. + * @property {number} [messageRetentionPeriod] - SQS MessageRetentionPeriod seconds. Range 60..1209600. Default 345600 (4 days). + * @property {number} [maxReceiveCount] - DLQ redrive policy maxReceiveCount. Range 1..1000. Default 3. + * @property {Object} [worker] - Queue-worker Lambda + event-source-mapping tuning + * @property {number} [worker.batchSize] - SQS event batchSize. Range 1..10000. Default 1. + * @property {number} [worker.maximumBatchingWindow] - SQS event MaximumBatchingWindowInSeconds. Range 0..300. Default unset. NOTE: serverless-framework key, not the longer CFN property name. + * @property {number} [worker.maximumConcurrency] - SQS ESM ScalingConfig.MaximumConcurrency. Range 2..1000. Default unset. + * @property {number} [worker.reservedConcurrency] - Lambda ReservedConcurrency. Range 0..1000. Default 20. + * @property {number} [worker.timeout] - Queue-worker Lambda execution timeout (seconds). Range 1..900. Default 900. + */ + /** * Integration configuration * @typedef {Object} IntegrationDefinition * @property {Object} Definition - Integration definition object * @property {string} Definition.name - Integration name + * @property {IntegrationQueueDefinition} [Definition.queue] - Optional SQS + queue-worker tuning */ /** From 1c1ebe82c171dc843ce172cde0ee1c408ff95379 Mon Sep 17 00:00:00 2001 From: d-klotz Date: Fri, 31 Jul 2026 18:28:58 -0300 Subject: [PATCH 2/5] fix(devtools): make the DLQ alarm capable of firing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The generated DLQMessageAlarm watched ApproximateNumberOfMessagesVisible with Threshold 500, while createDLQObservability also attaches dlqProcessor as an SQS consumer that drains the same queue continuously. Depth therefore never accumulates and the alarm could never fire, despite its comment claiming it "fires when any message lands in the DLQ". Observed in a production Frigg app: 146 messages landed in the DLQ in one day while ApproximateNumberOfMessages read 0 throughout, and no alarm fired. Every Frigg app with a DLQ has the same blind spot. Alarm on arrival instead. The metric is NumberOfMessagesDeleted rather than NumberOfMessagesSent because SQS does not increment Sent for messages arriving via redrive — verified in production, where Sent read 0 for seven consecutive days while Received and Deleted tracked the real inflow. Co-Authored-By: Claude Opus 5 (1M context) --- .../integration/integration-builder.js | 20 +++++++++++--- .../integration/integration-builder.test.js | 27 +++++++++++++++---- 2 files changed, 38 insertions(+), 9 deletions(-) diff --git a/packages/devtools/infrastructure/domains/integration/integration-builder.js b/packages/devtools/infrastructure/domains/integration/integration-builder.js index 11bc00c37..f35b5d153 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.js @@ -565,19 +565,31 @@ class IntegrationBuilder extends InfrastructureBuilder { * Called for both stack-owned and external InternalErrorQueues. */ createDLQObservability(result, functionPackageConfig, queueArn, queueName) { - // CloudWatch Alarm: fires when any message lands in the DLQ + // CloudWatch Alarm: fires when any message lands in the DLQ. + // + // This alarms on arrival rate, not on standing depth. dlqProcessor + // (below) consumes this queue continuously, so ApproximateNumberOf- + // MessagesVisible sits at ~0 no matter how many messages arrive and a + // depth-based alarm can never fire. + // + // The metric is NumberOfMessagesDeleted rather than the more obvious + // NumberOfMessagesSent because SQS does not increment Sent when a + // message arrives via redrive from a source queue — which is the only + // way messages reach a DLQ. Deleted increments as dlqProcessor drains + // them, making it the reliable arrival signal. result.resources.DLQMessageAlarm = { Type: 'AWS::CloudWatch::Alarm', Properties: { AlarmDescription: 'Messages in dead-letter queue — integration queue processing failures', Namespace: 'AWS/SQS', - MetricName: 'ApproximateNumberOfMessagesVisible', - Statistic: 'Maximum', - Threshold: 500, + MetricName: 'NumberOfMessagesDeleted', + Statistic: 'Sum', + Threshold: 0, ComparisonOperator: 'GreaterThanThreshold', EvaluationPeriods: 1, Period: 300, + TreatMissingData: 'notBreaching', AlarmActions: [{ Ref: 'InternalErrorBridgeTopic' }], Dimensions: [{ Name: 'QueueName', Value: queueName }], }, diff --git a/packages/devtools/infrastructure/domains/integration/integration-builder.test.js b/packages/devtools/infrastructure/domains/integration/integration-builder.test.js index b4e0b2fc8..5ce70ee79 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.test.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.test.js @@ -556,18 +556,35 @@ describe('IntegrationBuilder', () => { }); describe('DLQ Observability', () => { - it('should create a CloudWatch alarm for DLQ message depth', async () => { + it('should alarm on messages arriving in the DLQ, not on standing depth', async () => { const appDefinition = { integrations: [{ Definition: { name: 'test' } }], }; const result = await integrationBuilder.build(appDefinition, {}); + const props = result.resources.DLQMessageAlarm.Properties; - expect(result.resources.DLQMessageAlarm).toBeDefined(); expect(result.resources.DLQMessageAlarm.Type).toBe('AWS::CloudWatch::Alarm'); - expect(result.resources.DLQMessageAlarm.Properties.MetricName).toBe('ApproximateNumberOfMessagesVisible'); - expect(result.resources.DLQMessageAlarm.Properties.ComparisonOperator).toBe('GreaterThanThreshold'); - expect(result.resources.DLQMessageAlarm.Properties.Threshold).toBe(500); + // dlqProcessor consumes this queue continuously, so a depth metric + // sits at zero no matter how many messages arrive. + expect(props.MetricName).not.toBe('ApproximateNumberOfMessagesVisible'); + expect(props.MetricName).toBe('NumberOfMessagesDeleted'); + expect(props.Statistic).toBe('Sum'); + expect(props.Threshold).toBe(0); + expect(props.ComparisonOperator).toBe('GreaterThanThreshold'); + expect(props.TreatMissingData).toBe('notBreaching'); + }); + + it('should not alarm on NumberOfMessagesSent, which SQS never increments on redrive', async () => { + const appDefinition = { + integrations: [{ Definition: { name: 'test' } }], + }; + + const result = await integrationBuilder.build(appDefinition, {}); + + expect(result.resources.DLQMessageAlarm.Properties.MetricName).not.toBe( + 'NumberOfMessagesSent' + ); }); it('should wire alarm to InternalErrorBridgeTopic for notifications', async () => { From 662f90007b3abf9b8da76d454988abbbcc8c25d8 Mon Sep 17 00:00:00 2001 From: d-klotz Date: Fri, 31 Jul 2026 18:31:44 -0300 Subject: [PATCH 3/5] style(devtools): prettier-format the changed integration-builder lines Co-Authored-By: Claude Opus 5 (1M context) --- .../domains/integration/integration-builder.js | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/packages/devtools/infrastructure/domains/integration/integration-builder.js b/packages/devtools/infrastructure/domains/integration/integration-builder.js index f35b5d153..266a3bacd 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.js @@ -40,7 +40,12 @@ function validateQueueConfig(integrationName, queueConfig) { val === undefined || (Number.isFinite(val) && val >= min && val <= max); const checks = [ ['visibilityTimeout', queueConfig.visibilityTimeout, 0, 43200], - ['messageRetentionPeriod', queueConfig.messageRetentionPeriod, 60, 1209600], + [ + 'messageRetentionPeriod', + queueConfig.messageRetentionPeriod, + 60, + 1209600, + ], ['maxReceiveCount', queueConfig.maxReceiveCount, 1, 1000], ]; for (const [key, val, min, max] of checks) { @@ -487,8 +492,7 @@ class IntegrationBuilder extends InfrastructureBuilder { // (not `...InSeconds` — that's the AWS-side CFN property name). osls // and serverless@3+ both reject the longer key with a schema error. if (workerConfig.maximumBatchingWindow !== undefined) { - sqsEvent.maximumBatchingWindow = - workerConfig.maximumBatchingWindow; + sqsEvent.maximumBatchingWindow = workerConfig.maximumBatchingWindow; } if (workerConfig.maximumConcurrency !== undefined) { sqsEvent.maximumConcurrency = workerConfig.maximumConcurrency; From 489fc5b8eddb366385a58855dc9483e7d9111411 Mon Sep 17 00:00:00 2001 From: d-klotz Date: Fri, 31 Jul 2026 18:57:05 -0300 Subject: [PATCH 4/5] fix(devtools): also alarm when the DLQ stops draining MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NumberOfMessagesDeleted only increments while dlqProcessor is healthy. If it is throttled or erroring — plausible at reservedConcurrency 1 under a burst — messages accumulate in the DLQ and the arrival alarm stays silent, which is the worst case to miss. Add DLQBacklogAlarm on ApproximateNumberOfMessagesVisible > 0. That is also the metric AWS recommends for DLQs; their guidance assumes nothing consumes the queue, which is why it needs pairing with an arrival alarm here rather than replacing it. Co-Authored-By: Claude Opus 5 (1M context) --- .../integration/integration-builder.js | 44 ++++++++++++++----- .../integration/integration-builder.test.js | 20 +++++++++ 2 files changed, 54 insertions(+), 10 deletions(-) diff --git a/packages/devtools/infrastructure/domains/integration/integration-builder.js b/packages/devtools/infrastructure/domains/integration/integration-builder.js index 266a3bacd..a7f51e105 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.js @@ -569,18 +569,24 @@ class IntegrationBuilder extends InfrastructureBuilder { * Called for both stack-owned and external InternalErrorQueues. */ createDLQObservability(result, functionPackageConfig, queueArn, queueName) { - // CloudWatch Alarm: fires when any message lands in the DLQ. + // Two alarms, because this DLQ has a consumer and so needs both an + // arrival signal and a backlog signal. // - // This alarms on arrival rate, not on standing depth. dlqProcessor - // (below) consumes this queue continuously, so ApproximateNumberOf- - // MessagesVisible sits at ~0 no matter how many messages arrive and a - // depth-based alarm can never fire. + // AWS recommends ApproximateNumberOfMessagesVisible for DLQs, which + // assumes the usual shape: nothing consumes the DLQ, so depth + // accumulates. dlqProcessor (below) drains this one continuously, so + // depth returns to ~0 between CloudWatch samples and a depth alarm + // alone misses arrivals entirely. // - // The metric is NumberOfMessagesDeleted rather than the more obvious - // NumberOfMessagesSent because SQS does not increment Sent when a - // message arrives via redrive from a source queue — which is the only - // way messages reach a DLQ. Deleted increments as dlqProcessor drains - // them, making it the reliable arrival signal. + // DLQMessageAlarm — arrivals. NumberOfMessagesDeleted rather than the + // more obvious NumberOfMessagesSent: per the SQS docs, messages moved + // to a DLQ by redrive are NOT counted by Sent (only manual sends are), + // and redrive is the only way messages get here. Deleted increments as + // dlqProcessor drains, making it the arrival signal. + // + // DLQBacklogAlarm — backlog. Deleted only increments while + // dlqProcessor is healthy; if it is throttled or erroring, messages + // pile up and the arrival alarm stays silent. Depth > 0 catches that. result.resources.DLQMessageAlarm = { Type: 'AWS::CloudWatch::Alarm', Properties: { @@ -599,6 +605,24 @@ class IntegrationBuilder extends InfrastructureBuilder { }, }; + result.resources.DLQBacklogAlarm = { + Type: 'AWS::CloudWatch::Alarm', + Properties: { + AlarmDescription: + 'Dead-letter queue is not draining — dlqProcessor may be throttled or failing', + Namespace: 'AWS/SQS', + MetricName: 'ApproximateNumberOfMessagesVisible', + Statistic: 'Maximum', + Threshold: 0, + ComparisonOperator: 'GreaterThanThreshold', + EvaluationPeriods: 1, + Period: 300, + TreatMissingData: 'notBreaching', + AlarmActions: [{ Ref: 'InternalErrorBridgeTopic' }], + Dimensions: [{ Name: 'QueueName', Value: queueName }], + }, + }; + // DLQ processor Lambda: logs failed messages with structured context result.functions.dlqProcessor = { handler: diff --git a/packages/devtools/infrastructure/domains/integration/integration-builder.test.js b/packages/devtools/infrastructure/domains/integration/integration-builder.test.js index 5ce70ee79..dfe58e53d 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.test.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.test.js @@ -587,6 +587,26 @@ describe('IntegrationBuilder', () => { ); }); + it('should also alarm on standing depth, to catch a stalled dlqProcessor', async () => { + const appDefinition = { + integrations: [{ Definition: { name: 'test' } }], + }; + + const result = await integrationBuilder.build(appDefinition, {}); + const props = result.resources.DLQBacklogAlarm.Properties; + + // NumberOfMessagesDeleted only increments while dlqProcessor is + // draining. If it is throttled or erroring, messages pile up and + // the arrival alarm stays silent — depth catches that case. + expect(props.MetricName).toBe('ApproximateNumberOfMessagesVisible'); + expect(props.Statistic).toBe('Maximum'); + expect(props.Threshold).toBe(0); + expect(props.ComparisonOperator).toBe('GreaterThanThreshold'); + expect(props.AlarmActions).toEqual([ + { Ref: 'InternalErrorBridgeTopic' }, + ]); + }); + it('should wire alarm to InternalErrorBridgeTopic for notifications', async () => { const appDefinition = { integrations: [{ Definition: { name: 'test' } }], From 71688884b05d4d8df04446ea2da98d2304c01c2e Mon Sep 17 00:00:00 2001 From: d-klotz Date: Fri, 31 Jul 2026 19:17:59 -0300 Subject: [PATCH 5/5] =?UTF-8?q?fix(devtools):=20address=20review=20?= =?UTF-8?q?=E2=80=94=20alarm=20flapping,=20batchSize=20cross-check,=20sile?= =?UTF-8?q?nt=20knobs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three fixes from review: DLQBacklogAlarm used Statistic Maximum, which breaches on any burst larger than one dlqProcessor batch (concurrency 1, batchSize 10). It would have fired alongside DLQMessageAlarm on every ordinary burst, training people to ignore the alarm that means "consumer stalled". Minimum only breaches when the queue was never empty across the period. SQS event sources reject batchSize > 10 unless a batching window is set. The validator allowed it, so the template packaged fine and failed at deploy — exactly what the validator exists to prevent. Queue-level knobs silently did nothing when the queue is externally owned, while queue.worker.* still applied. Not mutating a queue we do not own is correct, but dropping declared config without a word is the worst failure mode for a tuning API. Warn, and run the same bounds checks on that path. Co-Authored-By: Claude Opus 5 (1M context) --- .../integration/integration-builder.js | 49 ++++++++++++++++++- .../integration/integration-builder.test.js | 47 +++++++++++++++++- 2 files changed, 94 insertions(+), 2 deletions(-) diff --git a/packages/devtools/infrastructure/domains/integration/integration-builder.js b/packages/devtools/infrastructure/domains/integration/integration-builder.js index a7f51e105..ef591b17e 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.js @@ -78,6 +78,43 @@ function validateWorkerConfig(integrationName, workerConfig) { ); } } + + // SQS event sources only accept batchSize > 10 when a batching window is + // set. Without this check the template packages fine and fails at deploy. + if ( + workerConfig.batchSize > 10 && + !(workerConfig.maximumBatchingWindow >= 1) + ) { + throw new Error( + `Integration '${integrationName}': queue.worker.batchSize=${workerConfig.batchSize} requires queue.worker.maximumBatchingWindow >= 1` + ); + } +} + +const QUEUE_CONFIG_KEYS = [ + 'visibilityTimeout', + 'messageRetentionPeriod', + 'maxReceiveCount', +]; + +/** + * Queue-level knobs only apply to a queue this stack owns. When the queue is + * external we must not mutate it — but silently dropping the config the app + * declared is the worst failure mode for a tuning API, so say so. + */ +function warnIgnoredQueueConfig(integrationName, queueConfig) { + if (!queueConfig) return; + validateQueueConfig(integrationName, queueConfig); + const ignored = QUEUE_CONFIG_KEYS.filter( + (key) => queueConfig[key] !== undefined + ); + if (ignored.length > 0) { + console.warn( + ` ⚠ Integration '${integrationName}': queue.${ignored.join( + ', queue.' + )} ignored — the queue is externally owned. queue.worker.* still applies.` + ); + } } class IntegrationBuilder extends InfrastructureBuilder { @@ -281,6 +318,10 @@ class IntegrationBuilder extends InfrastructureBuilder { ); } else { console.log(` ✓ Using external ${integrationName}Queue`); + warnIgnoredQueueConfig( + integrationName, + integration.Definition.queue + ); this.useExternalIntegrationQueue( integrationName, queueDecision, @@ -612,7 +653,13 @@ class IntegrationBuilder extends InfrastructureBuilder { 'Dead-letter queue is not draining — dlqProcessor may be throttled or failing', Namespace: 'AWS/SQS', MetricName: 'ApproximateNumberOfMessagesVisible', - Statistic: 'Maximum', + // Minimum, not Maximum: dlqProcessor runs at concurrency 1 + // with batchSize 10, so any burst larger than that leaves + // depth briefly above zero. Maximum would fire on every + // ordinary burst alongside DLQMessageAlarm and train people to + // ignore it. Minimum only breaches when the queue was never + // empty across the whole period — i.e. genuinely not draining. + Statistic: 'Minimum', Threshold: 0, ComparisonOperator: 'GreaterThanThreshold', EvaluationPeriods: 1, diff --git a/packages/devtools/infrastructure/domains/integration/integration-builder.test.js b/packages/devtools/infrastructure/domains/integration/integration-builder.test.js index dfe58e53d..42ec2d203 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.test.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.test.js @@ -553,6 +553,49 @@ describe('IntegrationBuilder', () => { integrationBuilder.build(appDefinition, {}) ).rejects.toThrow(/worker\.timeout=901 is out of range/); }); + + it('rejects batchSize > 10 without a batching window', async () => { + const appDefinition = { + integrations: [ + { + Definition: { + name: 'bad', + queue: { worker: { batchSize: 500 } }, + }, + }, + ], + }; + + await expect( + integrationBuilder.build(appDefinition, {}) + ).rejects.toThrow( + /batchSize=500 requires queue\.worker\.maximumBatchingWindow >= 1/ + ); + }); + + it('allows batchSize > 10 when a batching window is set', async () => { + const appDefinition = { + integrations: [ + { + Definition: { + name: 'ok', + queue: { + worker: { + batchSize: 500, + maximumBatchingWindow: 5, + }, + }, + }, + }, + ], + }; + + const result = await integrationBuilder.build(appDefinition, {}); + + expect(result.functions.okQueueWorker.events[0].sqs.batchSize).toBe( + 500 + ); + }); }); describe('DLQ Observability', () => { @@ -599,7 +642,9 @@ describe('IntegrationBuilder', () => { // draining. If it is throttled or erroring, messages pile up and // the arrival alarm stays silent — depth catches that case. expect(props.MetricName).toBe('ApproximateNumberOfMessagesVisible'); - expect(props.Statistic).toBe('Maximum'); + // Maximum would breach on any burst bigger than one dlqProcessor + // batch, duplicating DLQMessageAlarm. Minimum means "never empty". + expect(props.Statistic).toBe('Minimum'); expect(props.Threshold).toBe(0); expect(props.ComparisonOperator).toBe('GreaterThanThreshold'); expect(props.AlarmActions).toEqual([