nx_dead_letter_queue.nx
buildroot/runtime/nx_dead_letter_queue.nx
about
nx_dead_letter_queue.nx -- malformed-record routing for NX-INGEST.
module: nishi-core.ingest.dead_letter_queue
depends: nishi-core.io.syscalls, nishi-core.io.iso8601,
nishi-core.io.jsonl_writer
disk_kb: 5
capability: CORE_IO
license_tier: PUBLIC_NISHI_SUBSTRATE
genealogy_id: rabbitmq_dead_letter_exchange_pattern +
kafka_connect_dlq_2017 +
aws_sqs_dead_letter_queue_pattern +
nishi_cardinal_13_additive_only +
nishi_ingestion_s_class_cardinal_2026
Append-only dead-letter queue for malformed upstream records.
When an adapter cannot parse/canonicalize/validate a record, the
substrate REFUSES to drop it silently (which is what every
half-baked ingestion pipeline does) — instead routing it to the
DLQ where ops can inspect + retry + understand WHY it failed.
Per Cardinal 13 additive-only: DLQ entries are never deleted.
Each entry carries the original-bytes blob (CAS-addressed for
dedup), the parse/validate failure verdict, the source descriptor,
fetch context, and an analyst-investigation pointer.
===== Why this matters ============================================
Most ingestion systems silently drop malformed records OR fail-
stop the entire batch (worse). Either way, you lose:
1. The data (couldn't recover later)
2. The signal (didn't know upstream changed something)
3. The audit (compliance ask: "did we ingest record X?" — answer
should be one of YES + ROUTED_TO_DLQ + ROUTED_TO_SINK, never
"uh, we don't know")
DLQ solves all three.
dependencies 3 imports · 0 importers
imports: nx_syscalls.nxnx_iso8601.nxnx_jsonl_writer.nx
imported by: nobody (leaf or entry point)
structs
| 125 | struct DLQEntry |
consts
| 51 | const NX_DLQ_PARSE_FAIL: i64 = 1 // couldn't parse JSON/XML/etc. |
| 52 | const NX_DLQ_SCHEMA_VIOLATION: i64 = 2 // parsed but doesn't match expected schema |
| 53 | const NX_DLQ_TYPE_COERCION_FAIL: i64 = 3 // string-to-int failed, etc. |
| 54 | const NX_DLQ_MISSING_REQUIRED_FIELD: i64 = 4 |
| 55 | const NX_DLQ_LICENSE_INCOMPATIBLE: i64 = 5 // upstream license clashes with output |
| 56 | const NX_DLQ_PII_DETECTED: i64 = 6 // sensitive data; needs review before sink |
| 57 | const NX_DLQ_DUPLICATE_REJECTED: i64 = 7 // dedup engine flagged |
| 58 | const NX_DLQ_RATE_TOO_LOW_QUALITY: i64 = 8 // data-quality below threshold |
| 59 | const NX_DLQ_UPSTREAM_CONTRADICTION: i64 = 9 // record contradicts existing data |
| 60 | const NX_DLQ_TIMESTAMP_OUT_OF_RANGE: i64 = 10 // record is too old or in future |
| 61 | const NX_DLQ_SIZE_LIMIT_EXCEEDED: i64 = 11 // > max-record-bytes |
| 62 | const NX_DLQ_ENCODING_FAIL: i64 = 12 // invalid UTF-8 etc. |
| 104 | const NX_DLQ_UNRESOLVED: i64 = 1 |
| 105 | const NX_DLQ_INVESTIGATING: i64 = 2 |
| 106 | const NX_DLQ_FIXED_REINGESTED: i64 = 3 // record successfully re-processed |
| 107 | const NX_DLQ_WONTFIX_PERMANENT: i64 = 4 // upstream broken; record discarded |
| 108 | const NX_DLQ_FIXED_PARSER_UPDATE: i64 = 5 // our parser was buggy; updated + re-ingested |
| 109 | const NX_DLQ_DUPE_OF_OTHER_DLQ_ENTRY: i64 = 6 |
| 110 | const NX_DLQ_UPSTREAM_RETRACTED: i64 = 7 // record withdrawn by source |
| 152 | const NX_DLQ_ENTRY_BYTES: i64 = 152 // 19 fields * 8 bytes |
| 272 | const NX_DLQ_HEALTH_PRISTINE: i64 = 1 // <0.1% DLQ rate |
| 273 | const NX_DLQ_HEALTH_NOMINAL: i64 = 2 // 0.1-1% (normal noise) |
| 274 | const NX_DLQ_HEALTH_ELEVATED: i64 = 3 // 1-5% (investigate) |
| 275 | const NX_DLQ_HEALTH_DEGRADED: i64 = 4 // 5-25% (substrate alert) |
| 276 | const NX_DLQ_HEALTH_BROKEN: i64 = 5 // >25% (substrate-level pause) |
functions
| 64 | func nx_dlq_reason_name(r: i64) -> *u8 |
| 83 | func nx_dlq_reason_is_upstream_cause(r: i64) -> i64 |
| 112 | func nx_dlq_resolution_name(r: i64) -> *u8 |
| 162 | func nx_dlq_compute_cas_hash(bytes_ptr: *u8, bytes_len: i64) -> i64 called by 1: nx_dlq_entry_new |
| 184 | func nx_dlq_entry_new( |
| 226 | func nx_dlq_annotate_resolution( |
| 247 | func nx_dlq_emit_jsonl( |
| 266 | func nx_dlq_rate_q10(n_dlq_entries: i64, n_total_records: i64) -> i64 |
| 278 | func nx_dlq_health_verdict(rate_q10: i64) -> i64 |
| 286 | func nx_dlq_health_name(h: i64) -> *u8 |