diff --git a/packages/devtools/infrastructure/domains/integration/integration-builder.js b/packages/devtools/infrastructure/domains/integration/integration-builder.js index a27cd6dc7..ef591b17e 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.js @@ -30,6 +30,93 @@ 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}]` + ); + } + } + + // 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 { constructor() { super(); @@ -226,10 +313,15 @@ class IntegrationBuilder extends InfrastructureBuilder { this.createIntegrationQueue( integrationName, result, - appDefinition + appDefinition, + integration.Definition.queue ); } else { console.log(` ✓ Using external ${integrationName}Queue`); + warnIgnoredQueueConfig( + integrationName, + integration.Definition.queue + ); this.useExternalIntegrationQueue( integrationName, queueDecision, @@ -425,27 +517,35 @@ 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`); } @@ -510,19 +610,61 @@ 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. + // + // 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. + // + // 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: { AlarmDescription: 'Messages in dead-letter queue — integration queue processing failures', Namespace: 'AWS/SQS', + MetricName: 'NumberOfMessagesDeleted', + Statistic: 'Sum', + Threshold: 0, + ComparisonOperator: 'GreaterThanThreshold', + EvaluationPeriods: 1, + Period: 300, + TreatMissingData: 'notBreaching', + AlarmActions: [{ Ref: 'InternalErrorBridgeTopic' }], + Dimensions: [{ Name: 'QueueName', Value: queueName }], + }, + }; + + 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: 500, + // 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, Period: 300, + TreatMissingData: 'notBreaching', AlarmActions: [{ Ref: 'InternalErrorBridgeTopic' }], Dimensions: [{ Name: 'QueueName', Value: queueName }], }, @@ -554,7 +696,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 +710,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..42ec2d203 100644 --- a/packages/devtools/infrastructure/domains/integration/integration-builder.test.js +++ b/packages/devtools/infrastructure/domains/integration/integration-builder.test.js @@ -397,19 +397,259 @@ 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/); + }); + + 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', () => { - 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 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'); + // 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([ + { Ref: 'InternalErrorBridgeTopic' }, + ]); }); it('should wire alarm to InternalErrorBridgeTopic for notifications', async () => { 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 */ /**