fix: Refactor archive_to_dlq logic and fix minor issues

Author: aidankhogg

migrations/001_initial.sql

@@ -104,6 +104,26 @@ BEGIN
 END;
 $$;
 
+-- pgmq_read_by_id(queue_name, msg_id)
+CREATE OR REPLACE FUNCTION pgmq_read_by_id(queue_name text, msg_id bigint)
+RETURNS JSONB
+LANGUAGE plpgsql
+AS $$
+DECLARE
+    msg RECORD;
+BEGIN
+    SELECT * FROM pgmq.read(queue_name, msg_id) INTO msg;
+    RETURN CASE WHEN msg IS NULL THEN NULL ELSE json_build_object(
+        'msg_id', msg.msg_id,
+        'message', msg.message,
+        'read_ct', msg.read_ct,
+        'enqueued_at', msg.enqueued_at,
+        'first_received_at', msg.first_received_at,
+        'next_msg_scheduled_for', msg.next_msg_scheduled_for
+    ) END;
+END;
+$$;
+
 -- App deployments (tracks deployed applications)
 CREATE TABLE IF NOT EXISTS app_deployments (
     id              BIGSERIAL PRIMARY KEY,

netengine/core/pgmq_client.py

@@ -52,19 +52,29 @@ async def archive_to_dlq(self, queue_name: str, msg_id: int, reason: str) -> Non
             return
 
         envelope = EventEnvelope(**json.loads(msg["message"]))
-        updated = EventEnvelope(
-            event_id=envelope.event_id,
-            correlation_id=envelope.correlation_id,
-            event_type=envelope.event_type,
-            emitted_by=envelope.emitted_by,
-            emitted_at=envelope.emitted_at,
-            payload={**envelope.payload, "dlq_reason": reason},
-            parent_event_id=envelope.parent_event_id,
-            retry_count=envelope.retry_count + 1,
-        )
-
         await self.delete(queue_name, msg_id)
-        if updated.retry_count >= MAX_RETRIES:
-            await self.send(f"{queue_name}_dlq", updated)
+
+        if envelope.retry_count + 1 >= MAX_RETRIES:
+            dlq_envelope = EventEnvelope(
+                event_id=envelope.event_id,
+                correlation_id=envelope.correlation_id,
+                event_type=envelope.event_type,
+                emitted_by=envelope.emitted_by,
+                emitted_at=envelope.emitted_at,
+                payload={**envelope.payload, "dlq_reason": reason},
+                parent_event_id=envelope.parent_event_id,
+                retry_count=envelope.retry_count + 1,
+            )
+            await self.send(f"{queue_name}_dlq", dlq_envelope)
         else:
-            await self.send(queue_name, updated)
+            requeue_envelope = EventEnvelope(
+                event_id=envelope.event_id,
+                correlation_id=envelope.correlation_id,
+                event_type=envelope.event_type,
+                emitted_by=envelope.emitted_by,
+                emitted_at=envelope.emitted_at,
+                payload=envelope.payload,
+                parent_event_id=envelope.parent_event_id,
+                retry_count=envelope.retry_count + 1,
+            )
+            await self.send(queue_name, requeue_envelope)

netengine/errors.py

@@ -1,7 +1,5 @@
 from typing import Any
 
-from omegaconf import DictConfig as Config
-
 
 class BaseNetEngineException(Exception):
     def __init__(
@@ -12,7 +10,7 @@ def __init__(
     ) -> None:
         self._msg = message
         self._code: int | str | None = None
-        self._log_rules: Config | dict[str, Any] = {
+        self._log_rules: dict[str, Any] = {
             "log_on_init": True,
             "at_lvl": "TRACE",
             "with_msg": message,

netengine/handlers/dns.py

@@ -792,6 +792,10 @@ async def add_zone_record(
             logger.debug(f"Zone file flushed to disk: {zone_file_path}")
             # Signal CoreDNS to reload zones (SIGUSR1)
             await self._reload_coredns(context)
+        else:
+            logger.warning(
+                f"Zone directory does not exist, disk flush skipped: {zone_file_path.parent}"
+            )
 
         logger.info(f"Zone record updated: {zone} {record_type} {name} -> {value} (TTL: {ttl})")
 

tests/conftest.py

@@ -61,7 +61,7 @@ def dev_sandbox_spec() -> NetEngineSpec:
 @pytest.fixture(autouse=True)
 def isolated_runtime_state_file(tmp_path, monkeypatch):
     """Keep tests from reading or writing the repository-root runtime state file."""
-    monkeypatch.setenv("NETENGINE_STATE_FILE", str(tmp_path / "netengines_state.json"))
+    monkeypatch.setenv("NETENGINE_STATE_FILE", str(tmp_path / "netengine_state.json"))
 
 
 @pytest.fixture