{"record":{"id":"a82c1889d951481b","repo":"nestjs/nest","slug":"rmq-broker-has-blocked-the-connection-flow-contro","errorCode":null,"errorMessage":"RMQ broker has blocked the connection (flow control). Reason: ${reason}","messagePattern":"RMQ broker has blocked the connection \\(flow control\\)\\. Reason: (.+?)","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"packages/microservices/client/client-rmq.ts","lineNumber":311,"sourceCode":"\n      if (this.isInitialConnect) {\n        this.isInitialConnect = false;\n\n        if (!this.channel) {\n          this.connectionPromise = this.createChannel();\n        }\n      } else {\n        this.connectionPromise = Promise.resolve();\n      }\n    });\n  }\n\n  public registerBlockedListener(client: AmqpConnectionManager): void {\n    client.addListener(\n      RmqEventsMap.BLOCKED,\n      ({ reason }: { reason: string }) => {\n        this._status$.next(RmqStatus.BLOCKED);\n        this.logger.warn(BLOCKED_RMQ_MESSAGE(reason));\n      },\n    );\n  }\n\n  public registerUnblockedListener(client: AmqpConnectionManager): void {\n    client.addListener(RmqEventsMap.UNBLOCKED, () => {\n      this._status$.next(RmqStatus.UNBLOCKED);\n      this.logger.log(UNBLOCKED_RMQ_MESSAGE);\n    });\n  }\n\n  public on<\n    EventKey extends keyof RmqEvents = keyof RmqEvents,\n    EventCallback extends RmqEvents[EventKey] = RmqEvents[EventKey],\n  >(event: EventKey, callback: EventCallback) {\n    if (this.client) {\n      this.client.addListener(event, callback);\n    } else {","sourceCodeStart":293,"sourceCodeEnd":329,"githubUrl":"https://github.com/nestjs/nest/blob/dd75d7bd8c5e88048587e6768d36eb695f3e7a25/packages/microservices/client/client-rmq.ts#L293-L329","documentation":"RabbitMQ blocks publishing connections when the node raises a resource alarm (memory above vm_memory_high_watermark or free disk below disk_free_limit). The amqp-connection-manager emits a BLOCKED event carrying the broker's reason; ClientRMQ's registerBlockedListener records it, pushes RmqStatus.BLOCKED into the client's status stream and logs this warning. While blocked, the broker holds back all publishes on that connection, so client.send()/emit() calls silently stall until the matching UNBLOCKED event.","triggerScenarios":"A ClientRMQ (ClientsModule.register / ClientProxyFactory with Transport.RMQ) publishing while the broker trips a memory or disk alarm — the broker sends connection.blocked with reason 'memory' or 'disk'. Commonly caused by vm_memory_high_watermark breach, disk_free_limit breach, or a node-wide alarm raised by another noisy client on the same broker node.","commonSituations":"RabbitMQ in docker-compose/Kubernetes with tight memory limits; producers outpacing consumers so queues grow into the memory alarm; large payloads (media, batch uploads); shared brokers where another team's flood blocks every publishing connection including yours; broker node disk filling up.","solutions":["Clear the alarm on the broker: check `rabbitmqctl status` (alarms section), free disk above disk_free_limit, and give the node more memory or raise vm_memory_high_watermark (e.g. rabbitmq.conf: vm_memory_high_watermark.relative = 0.6, disk_free_limit.absolute = 2GB).","Fix the backpressure root cause: scale up consumers and bound prefetch so queues drain instead of growing into the alarm.","Subscribe to the client's `client.status` observable and pause/throttle publishing while it emits RmqStatus.BLOCKED, resuming on UNBLOCKED/CONNECTED.","Reduce publish rate or batch size at the producer and enable publisher confirms so stalled publishes are detectable."],"exampleFix":"// before: keeps bursting through broker flow control\nsetInterval(() => client.emit('telemetry', sample()), 1);\n\n// after: buffer while the broker has blocked the connection\nimport { RmqStatus } from '@nestjs/microservices';\nlet blocked = false;\nconst buffer: any[] = [];\nclient.status.subscribe((status) => {\n  const wasBlocked = blocked;\n  blocked = status === RmqStatus.BLOCKED;\n  if (wasBlocked && !blocked) {\n    while (buffer.length) client.emit('telemetry', buffer.shift());\n  }\n});\nfunction publish(msg: any) {\n  blocked ? buffer.push(msg) : client.emit('telemetry', msg);\n}","handlingStrategy":"fallback","validationCode":"import { firstValueFrom, filter } from 'rxjs';\nimport { RmqStatus } from '@nestjs/microservices';\n\n// don't burst while the broker has the connection blocked\nasync function waitForUnblocked(client: ClientRMQ) {\n  const status = await firstValueFrom(client.status);\n  if (status === RmqStatus.BLOCKED) {\n    await firstValueFrom(\n      client.status.pipe(filter((s) => s === RmqStatus.UNBLOCKED || s === RmqStatus.CONNECTED)),\n    );\n  }\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Set vm_memory_high_watermark and disk_free_limit with real headroom for the broker container/node","Bound consumer lag: scale consumers and use finite prefetch so queues don't grow into memory alarms","Subscribe to client.status and pause/throttle publishing on RmqStatus.BLOCKED instead of pushing blindly","Alert on broker alarms (rabbitmq-diagnostics alarms) and on BLOCKED client status","Load-test publish throughput against a broker holding production-sized queues"],"tags":["rabbitmq","amqp","flow-control","backpressure","resource-alarm"],"backgroundTag":"rabbitmq-connection-blocked-flow-control","analyzedSha":"dd75d7bd8c5e88048587e6768d36eb695f3e7a25","analyzedAt":"2026-08-21T19:39:39.867Z","contentChangedAt":"2026-08-21T19:39:39.867Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}