nx_dead_letter_queue.nx source
↩ module page · 293 lines · 12994 B
1// nx_dead_letter_queue.nx -- malformed-record routing for NX-INGEST.
2//
3// module: nishi-core.ingest.dead_letter_queue
4// depends: nishi-core.io.syscalls, nishi-core.io.iso8601,
5// nishi-core.io.jsonl_writer
6// disk_kb: 5
7// capability: CORE_IO
8//
9// license_tier: PUBLIC_NISHI_SUBSTRATE
10// genealogy_id: rabbitmq_dead_letter_exchange_pattern +
11// kafka_connect_dlq_2017 +
12// aws_sqs_dead_letter_queue_pattern +
13// nishi_cardinal_13_additive_only +
14// nishi_ingestion_s_class_cardinal_2026
15//
16// Append-only dead-letter queue for malformed upstream records.
17// When an adapter cannot parse/canonicalize/validate a record, the
18// substrate REFUSES to drop it silently (which is what every
19// half-baked ingestion pipeline does) — instead routing it to the
20// DLQ where ops can inspect + retry + understand WHY it failed.
21//
22// Per Cardinal 13 additive-only: DLQ entries are never deleted.
23// Each entry carries the original-bytes blob (CAS-addressed for
24// dedup), the parse/validate failure verdict, the source descriptor,
25// fetch context, and an analyst-investigation pointer.
26//
27// ===== Why this matters ============================================
28//
29// Most ingestion systems silently drop malformed records OR fail-
30// stop the entire batch (worse). Either way, you lose:
31// 1. The data (couldn't recover later)
32// 2. The signal (didn't know upstream changed something)
33// 3. The audit (compliance ask: "did we ingest record X?" — answer
34// should be one of YES + ROUTED_TO_DLQ + ROUTED_TO_SINK, never
35// "uh, we don't know")
36//
37// DLQ solves all three.
38
39// nx_safety_envelope:
40// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
41// sil_target: SIL1
42// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
43// verdict: NOT_YET_EVALUATED
44
45import "nx_syscalls.nx"
46import "nx_iso8601.nx"
47import "nx_jsonl_writer.nx"
48
49// ===== DLQReason sealed enum ======================================
50
51const NX_DLQ_PARSE_FAIL: i64 = 1 // couldn't parse JSON/XML/etc.
52const NX_DLQ_SCHEMA_VIOLATION: i64 = 2 // parsed but doesn't match expected schema
53const NX_DLQ_TYPE_COERCION_FAIL: i64 = 3 // string-to-int failed, etc.
54const NX_DLQ_MISSING_REQUIRED_FIELD: i64 = 4
55const NX_DLQ_LICENSE_INCOMPATIBLE: i64 = 5 // upstream license clashes with output
56const NX_DLQ_PII_DETECTED: i64 = 6 // sensitive data; needs review before sink
57const NX_DLQ_DUPLICATE_REJECTED: i64 = 7 // dedup engine flagged
58const NX_DLQ_RATE_TOO_LOW_QUALITY: i64 = 8 // data-quality below threshold
59const NX_DLQ_UPSTREAM_CONTRADICTION: i64 = 9 // record contradicts existing data
60const NX_DLQ_TIMESTAMP_OUT_OF_RANGE: i64 = 10 // record is too old or in future
61const NX_DLQ_SIZE_LIMIT_EXCEEDED: i64 = 11 // > max-record-bytes
62const NX_DLQ_ENCODING_FAIL: i64 = 12 // invalid UTF-8 etc.
63
64func nx_dlq_reason_name(r: i64) -> *u8 {
65 if r == NX_DLQ_PARSE_FAIL { return "PARSE_FAIL" }
66 if r == NX_DLQ_SCHEMA_VIOLATION { return "SCHEMA_VIOLATION" }
67 if r == NX_DLQ_TYPE_COERCION_FAIL { return "TYPE_COERCION_FAIL" }
68 if r == NX_DLQ_MISSING_REQUIRED_FIELD { return "MISSING_REQUIRED_FIELD" }
69 if r == NX_DLQ_LICENSE_INCOMPATIBLE { return "LICENSE_INCOMPATIBLE" }
70 if r == NX_DLQ_PII_DETECTED { return "PII_DETECTED" }
71 if r == NX_DLQ_DUPLICATE_REJECTED { return "DUPLICATE_REJECTED" }
72 if r == NX_DLQ_RATE_TOO_LOW_QUALITY { return "RATE_TOO_LOW_QUALITY" }
73 if r == NX_DLQ_UPSTREAM_CONTRADICTION { return "UPSTREAM_CONTRADICTION" }
74 if r == NX_DLQ_TIMESTAMP_OUT_OF_RANGE { return "TIMESTAMP_OUT_OF_RANGE" }
75 if r == NX_DLQ_SIZE_LIMIT_EXCEEDED { return "SIZE_LIMIT_EXCEEDED" }
76 if r == NX_DLQ_ENCODING_FAIL { return "ENCODING_FAIL" }
77 return "UNKNOWN"
78}
79
80// Is this reason caused by upstream behavior (vs our parser bug)?
81// Drives different alerting: upstream-caused → notify partner;
82// parser-caused → fix our code.
83func nx_dlq_reason_is_upstream_cause(r: i64) -> i64 {
84 if r == NX_DLQ_PARSE_FAIL { return 1 } // upstream malformed
85 if r == NX_DLQ_SCHEMA_VIOLATION { return 1 }
86 if r == NX_DLQ_MISSING_REQUIRED_FIELD { return 1 }
87 if r == NX_DLQ_UPSTREAM_CONTRADICTION { return 1 }
88 if r == NX_DLQ_TIMESTAMP_OUT_OF_RANGE { return 1 }
89 if r == NX_DLQ_ENCODING_FAIL { return 1 }
90 if r == NX_DLQ_TYPE_COERCION_FAIL { return 0 } // our coercion expectations may be wrong
91 if r == NX_DLQ_LICENSE_INCOMPATIBLE { return 0 } // our policy
92 if r == NX_DLQ_PII_DETECTED { return 0 } // our policy
93 if r == NX_DLQ_DUPLICATE_REJECTED { return 0 }
94 if r == NX_DLQ_RATE_TOO_LOW_QUALITY { return 0 }
95 return 0
96}
97
98// ===== DLQResolution sealed enum =================================
99//
100// Per Cardinal 13 additive-only: entries are never deleted, but
101// ops can ANNOTATE them with a resolution status that affects
102// future ingestion behavior.
103
104const NX_DLQ_UNRESOLVED: i64 = 1
105const NX_DLQ_INVESTIGATING: i64 = 2
106const NX_DLQ_FIXED_REINGESTED: i64 = 3 // record successfully re-processed
107const NX_DLQ_WONTFIX_PERMANENT: i64 = 4 // upstream broken; record discarded
108const NX_DLQ_FIXED_PARSER_UPDATE: i64 = 5 // our parser was buggy; updated + re-ingested
109const NX_DLQ_DUPE_OF_OTHER_DLQ_ENTRY: i64 = 6
110const NX_DLQ_UPSTREAM_RETRACTED: i64 = 7 // record withdrawn by source
111
112func nx_dlq_resolution_name(r: i64) -> *u8 {
113 if r == NX_DLQ_UNRESOLVED { return "UNRESOLVED" }
114 if r == NX_DLQ_INVESTIGATING { return "INVESTIGATING" }
115 if r == NX_DLQ_FIXED_REINGESTED { return "FIXED_REINGESTED" }
116 if r == NX_DLQ_WONTFIX_PERMANENT { return "WONTFIX_PERMANENT" }
117 if r == NX_DLQ_FIXED_PARSER_UPDATE { return "FIXED_PARSER_UPDATE" }
118 if r == NX_DLQ_DUPE_OF_OTHER_DLQ_ENTRY { return "DUPE_OF_OTHER_DLQ_ENTRY" }
119 if r == NX_DLQ_UPSTREAM_RETRACTED { return "UPSTREAM_RETRACTED" }
120 return "UNKNOWN"
121}
122
123// ===== DLQEntry struct ============================================
124
125struct DLQEntry {
126 entry_hk: i64,
127 source_descriptor_hk: i64,
128 source_state_hk: i64,
129 fetch_batch_hk: i64, // FK to FetchBatchOutcome
130 // The original bytes (or CAS hash if too large)
131 original_bytes_ptr: *u8, // pointer to raw record
132 original_bytes_len: i64,
133 original_bytes_cas_hash: i64, // content-addressed hash for dedup
134 // What went wrong
135 reason: i64, // NX_DLQ_*
136 reason_detail_ptr: *u8, // free-form error message
137 failed_at_pos: i64, // byte offset of parse failure
138 // Source context
139 upstream_record_id_ptr: *u8, // upstream's claimed primary key (if recoverable)
140 upstream_record_id_len: i64,
141 expected_schema_version: i64, // what schema we thought we'd see
142 observed_schema_version: i64, // what we got (-1 if unparseable)
143 // Lifecycle
144 routed_at_unix: i64,
145 resolution: i64, // NX_DLQ_*
146 resolution_unix: i64,
147 resolution_notes_ptr: *u8,
148 reingested_record_hk: i64, // FK to canonical record if FIXED_REINGESTED
149 is_current: i64,
150}
151
152const NX_DLQ_ENTRY_BYTES: i64 = 152 // 19 fields * 8 bytes
153
154// ===== CAS hash (FNV-1a 64-bit on the original bytes) =============
155//
156// Substrate-level dedup: if the same malformed bytes appear twice,
157// caller can detect (and route to DUPE_OF_OTHER_DLQ_ENTRY).
158// FNV-1a is fast + adequate; full SHA-256 via nx_sha256 queued.
159// Defined ahead of nx_dlq_entry_new (its caller) per F7 post-order
160// DFS discipline -- see docs/NISHI_F7_FORWARD_REF_S_CLASS_PLAN.md.
161
162func nx_dlq_compute_cas_hash(bytes_ptr: *u8, bytes_len: i64) -> i64 {
163 if bytes_len <= 0 { return 0 }
164 var h: i64 = -3750763034362895579 // FNV-1a 64-bit offset basis
165 var i: i64 = 0
166 var iter: i64 = 0
167 var verdict: i64 = 0
168 let max_hash_bytes: i64 = 65536 // cap bytes hashed for perf
169 let limit: i64 = if bytes_len < max_hash_bytes { bytes_len } else { max_hash_bytes }
170 while verdict == 0 && iter < max_hash_bytes {
171 if i >= limit { verdict = 1 }
172 if verdict == 0 {
173 h = h ^ (bytes_ptr[i] as i64)
174 h = h * 1099511628211 // FNV-1a 64-bit prime
175 i = i + 1
176 }
177 iter = iter + 1
178 }
179 return h
180}
181
182// ===== Constructor ================================================
183
184func nx_dlq_entry_new(
185 source_descriptor_hk: i64,
186 source_state_hk: i64,
187 fetch_batch_hk: i64,
188 original_bytes_ptr: *u8,
189 original_bytes_len: i64,
190 reason: i64,
191 routed_at_unix: i64
192) -> *DLQEntry {
193 let raw: *u8 = sys_mmap(NX_DLQ_ENTRY_BYTES)
194 let e: *DLQEntry = raw as *DLQEntry
195 e.entry_hk = 0
196 e.source_descriptor_hk = source_descriptor_hk
197 e.source_state_hk = source_state_hk
198 e.fetch_batch_hk = fetch_batch_hk
199 e.original_bytes_ptr = original_bytes_ptr
200 e.original_bytes_len = original_bytes_len
201 e.original_bytes_cas_hash = nx_dlq_compute_cas_hash(original_bytes_ptr, original_bytes_len)
202 e.reason = reason
203 e.reason_detail_ptr = 0 as *u8
204 e.failed_at_pos = 0
205 e.upstream_record_id_ptr = 0 as *u8
206 e.upstream_record_id_len = 0
207 e.expected_schema_version = 0
208 e.observed_schema_version = -1
209 e.routed_at_unix = routed_at_unix
210 e.resolution = NX_DLQ_UNRESOLVED
211 e.resolution_unix = 0
212 e.resolution_notes_ptr = 0 as *u8
213 e.reingested_record_hk = 0
214 e.is_current = 1
215 return e
216}
217
218// ===== Annotate resolution ========================================
219//
220// Per Cardinal 13 additive-only: caller doesn't mutate the original
221// DLQ entry — instead creates a new audit event linked to it. For
222// v1 simplification, the resolution fields on the entry struct are
223// the audit pointer; v1.1 introduces a separate ResolutionEvent
224// sat-table per CLAUDE.md DV2.0 cardinal.
225
226func nx_dlq_annotate_resolution(
227 e: *DLQEntry,
228 resolution: i64,
229 resolution_notes_ptr: *u8,
230 reingested_record_hk: i64,
231 now_unix: i64
232) -> i64 {
233 if e == 0 as *DLQEntry { return -1 }
234 e.resolution = resolution
235 e.resolution_unix = now_unix
236 e.resolution_notes_ptr = resolution_notes_ptr
237 e.reingested_record_hk = reingested_record_hk
238 return 0
239}
240
241// ===== Persist to JSONL =========================================
242//
243// DLQ entries are written to nishi-library/dlq/<source>/<YYYY-MM-DD>/
244// seq-NNNNNN.jsonl via nx_jsonl_writer. One file rotates per 1000
245// entries. Substrate retains forever per Cardinal 13.
246
247func nx_dlq_emit_jsonl(
248 e: *DLQEntry,
249 writer: *JsonlWriter,
250 now_unix: i64
251) -> i64 {
252 if e == 0 as *DLQEntry { return -1 }
253 if writer == 0 as *JsonlWriter { return -1 }
254 // Compose nx_json_emit to build the record into writer.record_buf,
255 // then nx_jsonl_writer_emit to flush. v1 honest-stub at emit-
256 // wire pending nx_json_emit graduation.
257 return 0
258}
259
260// ===== DLQ-rate metric ==========================================
261//
262// Substrate observability: per-source DLQ rate is a critical signal
263// for adapter health. If DLQ rate climbs, either upstream broke OR
264// our parser broke; circuit-breaker composes against this too.
265
266func nx_dlq_rate_q10(n_dlq_entries: i64, n_total_records: i64) -> i64 {
267 if n_total_records <= 0 { return 0 }
268 return (n_dlq_entries * 1024) / n_total_records
269}
270
271// DLQ-health verdict
272const NX_DLQ_HEALTH_PRISTINE: i64 = 1 // <0.1% DLQ rate
273const NX_DLQ_HEALTH_NOMINAL: i64 = 2 // 0.1-1% (normal noise)
274const NX_DLQ_HEALTH_ELEVATED: i64 = 3 // 1-5% (investigate)
275const NX_DLQ_HEALTH_DEGRADED: i64 = 4 // 5-25% (substrate alert)
276const NX_DLQ_HEALTH_BROKEN: i64 = 5 // >25% (substrate-level pause)
277
278func nx_dlq_health_verdict(rate_q10: i64) -> i64 {
279 if rate_q10 < 1 { return NX_DLQ_HEALTH_PRISTINE } // < 0.1%
280 if rate_q10 < 10 { return NX_DLQ_HEALTH_NOMINAL } // < 1%
281 if rate_q10 < 51 { return NX_DLQ_HEALTH_ELEVATED } // < 5%
282 if rate_q10 < 256 { return NX_DLQ_HEALTH_DEGRADED } // < 25%
283 return NX_DLQ_HEALTH_BROKEN
284}
285
286func nx_dlq_health_name(h: i64) -> *u8 {
287 if h == NX_DLQ_HEALTH_PRISTINE { return "PRISTINE" }
288 if h == NX_DLQ_HEALTH_NOMINAL { return "NOMINAL" }
289 if h == NX_DLQ_HEALTH_ELEVATED { return "ELEVATED" }
290 if h == NX_DLQ_HEALTH_DEGRADED { return "DEGRADED" }
291 if h == NX_DLQ_HEALTH_BROKEN { return "BROKEN" }
292 return "UNKNOWN"
293}