From f5f8e611fcb3adb13e6ac924e209abe4577a1cb6 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Fri, 10 Jul 2026 06:40:33 +0000 Subject: [PATCH 1/3] fix: dead-letter parse and handler failures instead of silently dropping (#9) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Parse failures in consumer.py were ACK'd (silent data loss). Handler failures were already nacked but the DLQ binding on knowledge.dlx was missing, so messages vanished. This change: - Switches parse failure path from basic_ack to basic_nack(requeue=False) so poison payloads route to knowledge.ingest.dlq via the DLX - Adds the missing Binding bean in IngestQueueConfig (knowledge.dlx → knowledge.ingest.dlq) and changes the DLX to FanoutExchange - Adds a Testcontainers integration test asserting malformed payloads land on the DLQ rather than disappearing - Updates unit tests to assert nack (not ack) on parse failures - Updates stale skeleton docs in README.md and __main__.py docstring MCP API gaps (link_note / project-vocabulary tools) are out of scope for this fix and tracked separately on the issue. --- .../knowledge/queue/IngestQueueConfig.kt | 26 ++++- ingest-worker/README.md | 24 ++++- .../src/knowledge_worker/__main__.py | 12 ++- .../src/knowledge_worker/consumer.py | 30 +++--- .../tests/integration/test_rabbit_consumer.py | 94 ++++++++++++++++++- ingest-worker/tests/unit/test_consumer.py | 13 ++- 6 files changed, 166 insertions(+), 33 deletions(-) diff --git a/api/src/main/kotlin/com/jorisjonkers/personalstack/knowledge/queue/IngestQueueConfig.kt b/api/src/main/kotlin/com/jorisjonkers/personalstack/knowledge/queue/IngestQueueConfig.kt index 3033e0f..c02bf0a 100644 --- a/api/src/main/kotlin/com/jorisjonkers/personalstack/knowledge/queue/IngestQueueConfig.kt +++ b/api/src/main/kotlin/com/jorisjonkers/personalstack/knowledge/queue/IngestQueueConfig.kt @@ -1,4 +1,3 @@ - package com.jorisjonkers.personalstack.knowledge.queue import org.springframework.amqp.core.Binding @@ -6,14 +5,15 @@ import org.springframework.amqp.core.BindingBuilder import org.springframework.amqp.core.Queue import org.springframework.amqp.core.QueueBuilder import org.springframework.amqp.core.TopicExchange +import org.springframework.amqp.core.FanoutExchange import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration /** * RabbitMQ topology owned by knowledge-api's write path. The Python - * ingest worker (Phase 5) is the eventual consumer; declaring the - * exchange + queues here means the worker can come up to a pre-bound - * topology rather than racing to declare it on first boot. + * ingest worker is the eventual consumer; declaring the exchange + queues + * here means the worker can come up to a pre-bound topology rather than + * racing to declare it on first boot. * * Routing keys (topic exchange): * knowledge.lesson — capture_lesson tool @@ -24,6 +24,14 @@ import org.springframework.context.annotation.Configuration * `knowledge.*` namespace via wildcard binding. The worker can split * into per-type queues later if throughput or backoff semantics * diverge — the routing key is already on the message envelope. + * + * Dead-letter topology: + * knowledge.ingest → x-dead-letter-exchange: knowledge.dlx + * x-dead-letter-routing-key: knowledge.ingest.dlq + * knowledge.dlx → knowledge.ingest.dlq (fanout binding) + * + * Both parse failures (nacked by the Python worker) and handler errors + * route to `knowledge.ingest.dlq` for visibility and manual replay. */ @Configuration class IngestQueueConfig { @@ -40,8 +48,9 @@ class IngestQueueConfig { @Bean fun knowledgeExchange(): TopicExchange = TopicExchange(EXCHANGE, true, false) + /** Fanout DLX — any nacked message from the ingest queue lands here. */ @Bean - fun knowledgeDlxExchange(): TopicExchange = TopicExchange(DLX, true, false) + fun knowledgeDlxExchange(): FanoutExchange = FanoutExchange(DLX, true, false) @Bean fun knowledgeIngestQueue(): Queue = @@ -59,4 +68,11 @@ class IngestQueueConfig { knowledgeIngestQueue: Queue, knowledgeExchange: TopicExchange, ): Binding = BindingBuilder.bind(knowledgeIngestQueue).to(knowledgeExchange).with("knowledge.*") + + /** Bind the DLQ to the DLX so nacked messages are routed and stored. */ + @Bean + fun knowledgeDlqBinding( + knowledgeIngestDlq: Queue, + knowledgeDlxExchange: FanoutExchange, + ): Binding = BindingBuilder.bind(knowledgeIngestDlq).to(knowledgeDlxExchange) } diff --git a/ingest-worker/README.md b/ingest-worker/README.md index bd26e52..e3d5ab1 100644 --- a/ingest-worker/README.md +++ b/ingest-worker/README.md @@ -1,8 +1,16 @@ # knowledge-ingest-worker -Python consumer that reads `knowledge.ingest`-bound messages from RabbitMQ and (in stacked follow-ups) commits canonical notes to the `knowledge-vault` git repo + populates LightRAG chunks/entities/relations in `knowledge_db`. +Python consumer that reads `knowledge.ingest`-bound messages from RabbitMQ, commits canonical notes to the `knowledge-vault` git repo, and writes vault pointers back to `kb_notes` in Postgres. The LightRAG chunking + embedding pipeline (Ollama embeddings, pgvector, entity/relation extraction) layers on top of that in stacked follow-ups. -Today the worker ships the skeleton: connect, consume, log each delivery as a structured JSON line, ACK. The git-vault writer lands next; the LightRAG ingest pipeline (Ollama embeddings + pgvector + entity/relation extraction) layers on top of that. +## Error handling + +| Failure mode | Behaviour | +|---|---| +| Malformed JSON / schema violation / bad encoding | `basic_nack(requeue=False)` → routed to `knowledge.ingest.dlq` via DLX | +| Handler error (vault write, DB error) | `basic_nack(requeue=False)` → same DLQ | +| Clean dispatch | `basic_ack` | + +The DLX topology (`knowledge.dlx` exchange + `knowledge.ingest.dlq` queue) is declared by the knowledge-api service on startup. The worker only consumes — it never needs to declare queues. ## Local development @@ -26,3 +34,15 @@ uv run mypy # type check | `INGEST_PREFETCH` | `4` | Bounded by Ollama / LightRAG latency, not AMQP | | `LOG_LEVEL` | `INFO` | | | `SERVICE_VERSION` | `unknown` | Stamped onto each log line (baked into the image at build) | +| `VAULT_ENABLED` | `false` | Set `true` in production; falls back to `LoggingHandler` | +| `VAULT_CLONE_URL` | `git@github.com:…/knowledge-vault.git` | SSH URL for the vault repo | +| `VAULT_CLONE_DIR` | `/var/lib/knowledge-vault` | Persistent volume mount point | +| `VAULT_BRANCH` | `main` | | +| `VAULT_SSH_KEY_PATH` | `/etc/git-secrets/id_ed25519` | Vault Agent injects deploy key in production | +| `VAULT_AUTHOR_NAME` | `knowledge-ingest-worker` | | +| `VAULT_AUTHOR_EMAIL` | `worker@knowledge.local` | | +| `KB_PERSIST_ENABLED` | `false` | Set `true` to write vault pointers back to `kb_notes` | +| `DB_HOST` | `postgres.data-system.svc.cluster.local` | k8s service DNS | +| `DB_PORT` | `5432` | | +| `DB_NAME` | `knowledge_db` | | +| `DB_USER` / `DB_PASSWORD` | `kb_user` / `kb_password` | Vault-projected in production | diff --git a/ingest-worker/src/knowledge_worker/__main__.py b/ingest-worker/src/knowledge_worker/__main__.py index a63cc45..158f616 100644 --- a/ingest-worker/src/knowledge_worker/__main__.py +++ b/ingest-worker/src/knowledge_worker/__main__.py @@ -1,10 +1,14 @@ """Entry point: ``python -m knowledge_worker`` (or via ``opentelemetry-instrument``). -Today the worker only ships the consumer skeleton + the -`LoggingHandler` — each delivery emits one structured log and is -ACKed. The real ingest flow (LightRAG, Ollama embeddings, -knowledge-vault git commits) lands in stacked follow-ups. +Wires the ``VaultHandler`` (when ``VAULT_ENABLED=true``) or falls +back to ``LoggingHandler`` for local smoke runs. ``VaultHandler`` +clones the knowledge-vault git repo, writes one markdown file per +delivery, commits + pushes, and calls back to ``kb_notes`` via +``PostgresNoteStore`` when ``KB_PERSIST_ENABLED=true``. + +Parse failures and handler errors are nacked without requeue so they +route to ``knowledge.ingest.dlq`` via the DLX declared by knowledge-api. """ from __future__ import annotations diff --git a/ingest-worker/src/knowledge_worker/consumer.py b/ingest-worker/src/knowledge_worker/consumer.py index 0cf3253..d01f30b 100644 --- a/ingest-worker/src/knowledge_worker/consumer.py +++ b/ingest-worker/src/knowledge_worker/consumer.py @@ -1,10 +1,15 @@ """Blocking pika consumer. -Acks on a clean `handler.handle` return. Reraises everything else -so pika's default behaviour (`requeue=True`) nacks the delivery and -returns it to the broker. The knowledge-api side already declares -the DLX, so terminal failures eventually land on -`knowledge.ingest.dlq` after the broker's redelivery count trips. +Acks on a clean ``handler.handle`` return. + +Parse failures (malformed JSON, schema violations, bad encoding) are +immediately dead-lettered: ``basic_nack(requeue=False)`` routes the +delivery through the DLX that knowledge-api declares on the ingest +queue so the poison payload lands on ``knowledge.ingest.dlq`` and +stays recoverable. + +Handler failures (vault write errors, DB errors) are also nacked +without requeue so they flow to the same DLQ rather than vanishing. """ from __future__ import annotations @@ -35,7 +40,7 @@ class _Delivery: class Consumer: - """Wraps a pika `BlockingConnection` + per-delivery dispatch.""" + """Wraps a pika ``BlockingConnection`` + per-delivery dispatch.""" def __init__( self, @@ -106,18 +111,15 @@ def _on_message( try: note = self._parse(delivery) except (ValidationError, json.JSONDecodeError, UnicodeDecodeError) as exc: - # Bad payloads can't get fixed by retrying — ack so the - # broker doesn't redeliver the same broken message - # forever. A future change can route these to the - # dead-letter queue instead by switching to nack + - # requeue=false; today the broker's redelivery limit - # handles that path well enough. + # Poison payloads cannot be fixed by retrying. Dead-letter + # them immediately so they land on knowledge.ingest.dlq via + # the DLX declared by knowledge-api on this queue. self._log.error( "consumer.parse_failed", routing_key=delivery.routing_key, error=str(exc), ) - channel.basic_ack(delivery_tag=method.delivery_tag) + channel.basic_nack(delivery_tag=method.delivery_tag, requeue=False) return try: @@ -140,6 +142,6 @@ def _parse(delivery: _Delivery) -> CapturedNote: def silence_pika_warning_logs() -> None: - """pika emits a noisy WARNING per `basic_qos` round-trip. Demote to INFO.""" + """pika emits a noisy WARNING per ``basic_qos`` round-trip. Demote to INFO.""" logging.getLogger("pika").setLevel(logging.INFO) diff --git a/ingest-worker/tests/integration/test_rabbit_consumer.py b/ingest-worker/tests/integration/test_rabbit_consumer.py index fd0d8f9..d985f33 100644 --- a/ingest-worker/tests/integration/test_rabbit_consumer.py +++ b/ingest-worker/tests/integration/test_rabbit_consumer.py @@ -1,12 +1,17 @@ -"""End-to-end smoke test: publish a `CapturedNote` payload onto the +"""End-to-end smoke test: publish a ``CapturedNote`` payload onto the same topic exchange the knowledge-api side declares, then verify the worker consumes + dispatches to its handler. Spins up RabbitMQ via Testcontainers (Docker required). Declares the -`knowledge` topic exchange + `knowledge.ingest` queue on the test's +``knowledge`` topic exchange + ``knowledge.ingest`` queue on the test's own connection rather than relying on knowledge-api to do it — keeps the worker's start-up free of declaration assumptions, which is the production posture too. + +DLX topology mirrors ``IngestQueueConfig`` on the knowledge-api side: + knowledge.ingest → x-dead-letter-exchange: knowledge.dlx + x-dead-letter-routing-key: knowledge.ingest.dlq + knowledge.dlx → knowledge.ingest.dlq (queue, durable) """ from __future__ import annotations @@ -30,6 +35,9 @@ pytestmark = pytest.mark.integration +DLX = "knowledge.dlx" +DLQ = "knowledge.ingest.dlq" + @pytest.fixture(scope="module") def rabbit() -> Iterator[dict[str, object]]: @@ -60,6 +68,7 @@ def _settings(rabbit: dict[str, object]) -> Settings: def _declare_topology(settings: Settings) -> pika.BlockingConnection: + """Declare the full DLX topology that knowledge-api owns in production.""" conn = pika.BlockingConnection( pika.ConnectionParameters( host=settings.rabbitmq_host, @@ -71,8 +80,22 @@ def _declare_topology(settings: Settings) -> pika.BlockingConnection: ) ) ch = conn.channel() + # Main exchange ch.exchange_declare(exchange="knowledge", exchange_type="topic", durable=True) - ch.queue_declare(queue=settings.queue, durable=True) + # Dead-letter exchange (fanout so any routing key lands on the DLQ) + ch.exchange_declare(exchange=DLX, exchange_type="fanout", durable=True) + # DLQ — must be declared before the main queue so the DLX target exists + ch.queue_declare(queue=DLQ, durable=True) + ch.queue_bind(queue=DLQ, exchange=DLX, routing_key=DLQ) + # Main ingest queue wired to the DLX + ch.queue_declare( + queue=settings.queue, + durable=True, + arguments={ + "x-dead-letter-exchange": DLX, + "x-dead-letter-routing-key": DLQ, + }, + ) ch.queue_bind(queue=settings.queue, exchange="knowledge", routing_key="knowledge.*") return conn @@ -100,6 +123,22 @@ def _publish( conn.close() +def _publish_raw(settings: Settings, routing_key: str, body: bytes) -> None: + """Publish a raw (potentially malformed) body without JSON encoding.""" + conn = pika.BlockingConnection( + pika.ConnectionParameters( + host=settings.rabbitmq_host, + port=settings.rabbitmq_port, + credentials=pika.PlainCredentials( + settings.rabbitmq_user, settings.rabbitmq_password + ), + ) + ) + ch = conn.channel() + ch.basic_publish(exchange="knowledge", routing_key=routing_key, body=body) + conn.close() + + def _await_delivery(handler: RecordingHandler, expected: int, timeout_s: float = 10.0) -> None: deadline = time.monotonic() + timeout_s while time.monotonic() < deadline: @@ -111,6 +150,32 @@ def _await_delivery(handler: RecordingHandler, expected: int, timeout_s: float = ) +def _await_dlq_message( + settings: Settings, timeout_s: float = 10.0 +) -> bytes | None: + """Poll the DLQ queue and return the body of the first message found.""" + conn = pika.BlockingConnection( + pika.ConnectionParameters( + host=settings.rabbitmq_host, + port=settings.rabbitmq_port, + credentials=pika.PlainCredentials( + settings.rabbitmq_user, settings.rabbitmq_password + ), + ) + ) + ch = conn.channel() + deadline = time.monotonic() + timeout_s + try: + while time.monotonic() < deadline: + method, _, body = ch.basic_get(queue=DLQ, auto_ack=True) + if method is not None: + return body + time.sleep(0.05) + return None + finally: + conn.close() + + def _payload(routing_id: str) -> dict[str, object]: return { "id": routing_id, @@ -171,3 +236,26 @@ def test_handles_three_messages_in_order(rabbit: dict[str, object]) -> None: finally: consumer.stop() runner.join(timeout=5) + + +def test_malformed_payload_routes_to_dlq(rabbit: dict[str, object]) -> None: + """A poison payload must land on the DLQ, not be silently dropped.""" + settings = _settings(rabbit) + _declare_topology(settings).close() + + handler = RecordingHandler() + consumer = Consumer(settings, handler) + consumer.start() + runner = threading.Thread(target=consumer.run_forever, daemon=True) + runner.start() + + try: + _publish_raw(settings, "knowledge.lesson", b"{not valid json") + dlq_body = _await_dlq_message(settings) + assert dlq_body == b"{not valid json", ( + "malformed payload was not routed to the DLQ" + ) + assert handler.deliveries == [], "handler must not have been called for a poison payload" + finally: + consumer.stop() + runner.join(timeout=5) diff --git a/ingest-worker/tests/unit/test_consumer.py b/ingest-worker/tests/unit/test_consumer.py index 41dfa48..e3b9541 100644 --- a/ingest-worker/tests/unit/test_consumer.py +++ b/ingest-worker/tests/unit/test_consumer.py @@ -75,19 +75,21 @@ def test_valid_message_is_acked_and_handed_to_handler() -> None: assert isinstance(note, CapturedNote) -def test_invalid_payload_is_acked_to_avoid_redelivery_loops() -> None: +def test_invalid_payload_is_nacked_to_dlq() -> None: + """Malformed JSON must be dead-lettered, not silently ACK'd.""" handler = RecordingHandler() consumer = Consumer(_settings(), handler) channel = _FakeChannel() consumer._on_message(channel, _FakeMethod(), object(), b"{not valid json") # type: ignore[arg-type] - assert channel.acks == [1] - assert channel.nacks == [] + assert channel.acks == [] + assert channel.nacks == [(1, False)] assert handler.deliveries == [] -def test_payload_missing_required_field_is_acked_and_skipped() -> None: +def test_payload_missing_required_field_is_nacked_to_dlq() -> None: + """Schema violations must be dead-lettered, not silently ACK'd.""" handler = RecordingHandler() consumer = Consumer(_settings(), handler) channel = _FakeChannel() @@ -97,7 +99,8 @@ def test_payload_missing_required_field_is_acked_and_skipped() -> None: body = json.dumps(payload).encode() consumer._on_message(channel, _FakeMethod(), object(), body) # type: ignore[arg-type] - assert channel.acks == [1] + assert channel.acks == [] + assert channel.nacks == [(1, False)] assert handler.deliveries == [] From 0182e54783f34eb5424ce49a376c95b204bcdcd9 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Fri, 10 Jul 2026 06:45:44 +0000 Subject: [PATCH 2/3] fix: sort imports in IngestQueueConfig to satisfy ktlint --- .../personalstack/knowledge/queue/IngestQueueConfig.kt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api/src/main/kotlin/com/jorisjonkers/personalstack/knowledge/queue/IngestQueueConfig.kt b/api/src/main/kotlin/com/jorisjonkers/personalstack/knowledge/queue/IngestQueueConfig.kt index c02bf0a..5bd4abf 100644 --- a/api/src/main/kotlin/com/jorisjonkers/personalstack/knowledge/queue/IngestQueueConfig.kt +++ b/api/src/main/kotlin/com/jorisjonkers/personalstack/knowledge/queue/IngestQueueConfig.kt @@ -2,10 +2,10 @@ package com.jorisjonkers.personalstack.knowledge.queue import org.springframework.amqp.core.Binding import org.springframework.amqp.core.BindingBuilder +import org.springframework.amqp.core.FanoutExchange import org.springframework.amqp.core.Queue import org.springframework.amqp.core.QueueBuilder import org.springframework.amqp.core.TopicExchange -import org.springframework.amqp.core.FanoutExchange import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration From ff136dd9f7ae7490a8d1204d4fa2642b3fa44402 Mon Sep 17 00:00:00 2001 From: JorisJonkers Agent Date: Fri, 10 Jul 2026 06:51:09 +0000 Subject: [PATCH 3/3] fix: keep DLX as TopicExchange to avoid PRECONDITION_FAILED on existing broker Per codex review: switching knowledge.dlx from TopicExchange to FanoutExchange would cause PRECONDITION_FAILED on any broker that already has it declared. Retain TopicExchange and bind the DLQ with routing key knowledge.ingest.dlq (which the main queue already stamps as x-dead-letter-routing-key). Update the integration test topology to match. --- .../knowledge/queue/IngestQueueConfig.kt | 21 ++++++++++++------- .../tests/integration/test_rabbit_consumer.py | 4 ++-- 2 files changed, 16 insertions(+), 9 deletions(-) diff --git a/api/src/main/kotlin/com/jorisjonkers/personalstack/knowledge/queue/IngestQueueConfig.kt b/api/src/main/kotlin/com/jorisjonkers/personalstack/knowledge/queue/IngestQueueConfig.kt index 5bd4abf..2a0b991 100644 --- a/api/src/main/kotlin/com/jorisjonkers/personalstack/knowledge/queue/IngestQueueConfig.kt +++ b/api/src/main/kotlin/com/jorisjonkers/personalstack/knowledge/queue/IngestQueueConfig.kt @@ -2,7 +2,6 @@ package com.jorisjonkers.personalstack.knowledge.queue import org.springframework.amqp.core.Binding import org.springframework.amqp.core.BindingBuilder -import org.springframework.amqp.core.FanoutExchange import org.springframework.amqp.core.Queue import org.springframework.amqp.core.QueueBuilder import org.springframework.amqp.core.TopicExchange @@ -28,7 +27,12 @@ import org.springframework.context.annotation.Configuration * Dead-letter topology: * knowledge.ingest → x-dead-letter-exchange: knowledge.dlx * x-dead-letter-routing-key: knowledge.ingest.dlq - * knowledge.dlx → knowledge.ingest.dlq (fanout binding) + * knowledge.dlx (TopicExchange) → knowledge.ingest.dlq + * + * The main queue sets `x-dead-letter-routing-key` to `knowledge.ingest.dlq`, + * so the DLQ binding uses that exact key. Keeping DLX as a TopicExchange + * avoids a PRECONDITION_FAILED error when rolling out onto a broker that + * already has `knowledge.dlx` declared as a topic exchange. * * Both parse failures (nacked by the Python worker) and handler errors * route to `knowledge.ingest.dlq` for visibility and manual replay. @@ -48,9 +52,8 @@ class IngestQueueConfig { @Bean fun knowledgeExchange(): TopicExchange = TopicExchange(EXCHANGE, true, false) - /** Fanout DLX — any nacked message from the ingest queue lands here. */ @Bean - fun knowledgeDlxExchange(): FanoutExchange = FanoutExchange(DLX, true, false) + fun knowledgeDlxExchange(): TopicExchange = TopicExchange(DLX, true, false) @Bean fun knowledgeIngestQueue(): Queue = @@ -69,10 +72,14 @@ class IngestQueueConfig { knowledgeExchange: TopicExchange, ): Binding = BindingBuilder.bind(knowledgeIngestQueue).to(knowledgeExchange).with("knowledge.*") - /** Bind the DLQ to the DLX so nacked messages are routed and stored. */ + /** + * Bind the DLQ to the DLX using the exact routing key the main queue + * stamps on dead-lettered messages, so nacked deliveries are stored + * rather than dropped by the broker. + */ @Bean fun knowledgeDlqBinding( knowledgeIngestDlq: Queue, - knowledgeDlxExchange: FanoutExchange, - ): Binding = BindingBuilder.bind(knowledgeIngestDlq).to(knowledgeDlxExchange) + knowledgeDlxExchange: TopicExchange, + ): Binding = BindingBuilder.bind(knowledgeIngestDlq).to(knowledgeDlxExchange).with(DLQ) } diff --git a/ingest-worker/tests/integration/test_rabbit_consumer.py b/ingest-worker/tests/integration/test_rabbit_consumer.py index d985f33..ec86079 100644 --- a/ingest-worker/tests/integration/test_rabbit_consumer.py +++ b/ingest-worker/tests/integration/test_rabbit_consumer.py @@ -82,8 +82,8 @@ def _declare_topology(settings: Settings) -> pika.BlockingConnection: ch = conn.channel() # Main exchange ch.exchange_declare(exchange="knowledge", exchange_type="topic", durable=True) - # Dead-letter exchange (fanout so any routing key lands on the DLQ) - ch.exchange_declare(exchange=DLX, exchange_type="fanout", durable=True) + # Dead-letter exchange — topic to match the production IngestQueueConfig declaration + ch.exchange_declare(exchange=DLX, exchange_type="topic", durable=True) # DLQ — must be declared before the main queue so the DLX target exists ch.queue_declare(queue=DLQ, durable=True) ch.queue_bind(queue=DLQ, exchange=DLX, routing_key=DLQ)