nx_trace_query.nx source
↩ module page · 221 lines · 9107 B
1// nx_trace_query.nx -- filter / count / project over CBOR span streams.
2//
3// Layer 3 of the sovereign-log roadmap. Operates on a raw CBOR span
4// buffer (n bytes of back-to-back nx_cbor_emit_span output). The
5// caller decides the source: live in-memory chunk_buf, a flushed
6// nx_log_chunked chunk's bytes, a memory-mapped on-disk file, or a
7// concatenated multi-chunk read. Query layer doesn't care about
8// storage layout -- it walks bytes.
9//
10// Design rationale: stream-walk per query keeps the primitive small
11// (no internal indexing) and avoids the LSM-tree compaction problem
12// for v1. When a query becomes hot, an index can land at Layer 4
13// (bloom filter + segment summary) without changing this API.
14//
15// Per the build-the-engine-not-instances cardinal: this primitive
16// scales any future projection without per-projection code.
17//
18// genealogy_id: cqrs_event_sourcing + opentelemetry_query
19// lineage_id: substrate_trace_query_v1
20
21// nx_safety_envelope:
22// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
23// sil_target: SIL1
24// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
25// verdict: NOT_YET_EVALUATED
26
27import "nx_syscalls.nx"
28import "nx_tier.nx"
29import "nx_cbor.nx"
30import "nx_cbor_decode.nx"
31import "nx_trace_emit.nx"
32
33// ===== Count spans per kind =======================================
34//
35// Single pass over the buffer. counts_out is caller-sized to
36// NX_SPAN_N_KINDS i64 slots; we bump counts_out[kind] for every
37// span encountered. Returns total spans walked.
38
39const NX_TQ_MAX_ATTRS_PER_SPAN: nx_int = 32
40
41func nx_tq_count_kinds(buf: *u8, n_bytes: nx_int, counts_out: *i64) -> nx_int {
42 // Zero the output
43 var z: nx_int = 0
44 while z < NX_SPAN_N_KINDS {
45 counts_out[z] = 0
46 z = z + 1
47 }
48
49 let np: *i64 = (sys_mmap(8)) as *i64
50 let trace_o: *i64 = (sys_mmap(8)) as *i64
51 let span_o: *i64 = (sys_mmap(8)) as *i64
52 let parent_o: *i64 = (sys_mmap(8)) as *i64
53 let kind_o: *i64 = (sys_mmap(8)) as *i64
54 let start_o: *i64 = (sys_mmap(8)) as *i64
55 let end_o: *i64 = (sys_mmap(8)) as *i64
56 let attrs_o: *i64 = (sys_mmap(NX_TQ_MAX_ATTRS_PER_SPAN * 16)) as *i64
57 let n_attrs_o: *i64 = (sys_mmap(8)) as *i64
58
59 var pos: nx_int = 0
60 var n_total: nx_int = 0
61 while pos < n_bytes {
62 let rc: nx_int = nx_cbor_read_span(buf, pos, np,
63 trace_o, span_o, parent_o, kind_o,
64 start_o, end_o,
65 attrs_o, NX_TQ_MAX_ATTRS_PER_SPAN,
66 n_attrs_o)
67 if rc != 0 { return n_total } // malformed; stop where we are
68 let k: nx_int = kind_o[0]
69 if nx_span_kind_is_valid(k) == 1 {
70 counts_out[k] = counts_out[k] + 1
71 }
72 n_total = n_total + 1
73 pos = np[0]
74 }
75 return n_total
76}
77
78// ===== Count spans for one trace_id ===============================
79
80func nx_tq_count_for_trace(buf: *u8, n_bytes: nx_int, trace_id: nx_int) -> nx_int {
81 let np: *i64 = (sys_mmap(8)) as *i64
82 let trace_o: *i64 = (sys_mmap(8)) as *i64
83 let span_o: *i64 = (sys_mmap(8)) as *i64
84 let parent_o: *i64 = (sys_mmap(8)) as *i64
85 let kind_o: *i64 = (sys_mmap(8)) as *i64
86 let start_o: *i64 = (sys_mmap(8)) as *i64
87 let end_o: *i64 = (sys_mmap(8)) as *i64
88 let attrs_o: *i64 = (sys_mmap(NX_TQ_MAX_ATTRS_PER_SPAN * 16)) as *i64
89 let n_attrs_o: *i64 = (sys_mmap(8)) as *i64
90
91 var pos: nx_int = 0
92 var count: nx_int = 0
93 while pos < n_bytes {
94 let rc: nx_int = nx_cbor_read_span(buf, pos, np,
95 trace_o, span_o, parent_o, kind_o,
96 start_o, end_o,
97 attrs_o, NX_TQ_MAX_ATTRS_PER_SPAN,
98 n_attrs_o)
99 if rc != 0 { return count }
100 if trace_o[0] == trace_id { count = count + 1 }
101 pos = np[0]
102 }
103 return count
104}
105
106// ===== Batch summary projection ===================================
107//
108// Walks the buffer; for every BATCH_CLOSE span found, copies its
109// attrs into the summaries_out buffer. Layout per summary slot:
110// [trace_id, n_attrs, attr_k0, attr_v0, attr_k1, attr_v1, ...]
111// Each slot occupies (2 + 2 * NX_TQ_MAX_ATTRS_PER_SPAN) i64 cells.
112// summaries_cap is the number of slots the caller allocated.
113// Returns the count of BATCH_CLOSE spans written (may be < cap if
114// few batches; may be == cap if buffer too small).
115
116const NX_TQ_SUMMARY_HDR_FIELDS: nx_int = 2 // trace_id + n_attrs
117const NX_TQ_SUMMARY_STRIDE: nx_int = 66 // 2 + 2 * 32
118
119func nx_tq_extract_batch_summaries(buf: *u8, n_bytes: nx_int,
120 summaries_out: *i64,
121 summaries_cap: nx_int) -> nx_int {
122 let np: *i64 = (sys_mmap(8)) as *i64
123 let trace_o: *i64 = (sys_mmap(8)) as *i64
124 let span_o: *i64 = (sys_mmap(8)) as *i64
125 let parent_o: *i64 = (sys_mmap(8)) as *i64
126 let kind_o: *i64 = (sys_mmap(8)) as *i64
127 let start_o: *i64 = (sys_mmap(8)) as *i64
128 let end_o: *i64 = (sys_mmap(8)) as *i64
129 let attrs_o: *i64 = (sys_mmap(NX_TQ_MAX_ATTRS_PER_SPAN * 16)) as *i64
130 let n_attrs_o: *i64 = (sys_mmap(8)) as *i64
131
132 var pos: nx_int = 0
133 var n_summaries: nx_int = 0
134 while pos < n_bytes {
135 let rc: nx_int = nx_cbor_read_span(buf, pos, np,
136 trace_o, span_o, parent_o, kind_o,
137 start_o, end_o,
138 attrs_o, NX_TQ_MAX_ATTRS_PER_SPAN,
139 n_attrs_o)
140 if rc != 0 { return n_summaries }
141
142 if kind_o[0] == NX_SPAN_BATCH_CLOSE {
143 if n_summaries < summaries_cap {
144 let slot_base: nx_int = n_summaries * NX_TQ_SUMMARY_STRIDE
145 summaries_out[slot_base + 0] = trace_o[0]
146 summaries_out[slot_base + 1] = n_attrs_o[0]
147 var i: nx_int = 0
148 while i < n_attrs_o[0] {
149 if i < NX_TQ_MAX_ATTRS_PER_SPAN {
150 summaries_out[slot_base + 2 + 2 * i] = attrs_o[2 * i]
151 summaries_out[slot_base + 2 + 2 * i + 1] = attrs_o[2 * i + 1]
152 }
153 i = i + 1
154 }
155 n_summaries = n_summaries + 1
156 }
157 }
158 pos = np[0]
159 }
160 return n_summaries
161}
162
163// ===== Distinct-attr-value extractor ==============================
164//
165// For all spans of `kind` in the buffer, find every distinct VALUE
166// at attr_key. Caller supplies a flat output buffer; we treat it
167// as a hash-table-free unique-list (linear append, dedup on insert).
168// Cheaper than a full hash table for small N (which is the common
169// case for "how many distinct outfit IDs were picked this batch").
170// Returns count of distinct values.
171
172func nx_tq_distinct_attr_values(buf: *u8, n_bytes: nx_int,
173 kind: nx_int, attr_key: nx_int,
174 distinct_out: *i64,
175 distinct_cap: nx_int) -> nx_int {
176 let np: *i64 = (sys_mmap(8)) as *i64
177 let trace_o: *i64 = (sys_mmap(8)) as *i64
178 let span_o: *i64 = (sys_mmap(8)) as *i64
179 let parent_o: *i64 = (sys_mmap(8)) as *i64
180 let kind_o: *i64 = (sys_mmap(8)) as *i64
181 let start_o: *i64 = (sys_mmap(8)) as *i64
182 let end_o: *i64 = (sys_mmap(8)) as *i64
183 let attrs_o: *i64 = (sys_mmap(NX_TQ_MAX_ATTRS_PER_SPAN * 16)) as *i64
184 let n_attrs_o: *i64 = (sys_mmap(8)) as *i64
185
186 var pos: nx_int = 0
187 var n_distinct: nx_int = 0
188 while pos < n_bytes {
189 let rc: nx_int = nx_cbor_read_span(buf, pos, np,
190 trace_o, span_o, parent_o, kind_o,
191 start_o, end_o,
192 attrs_o, NX_TQ_MAX_ATTRS_PER_SPAN,
193 n_attrs_o)
194 if rc != 0 { return n_distinct }
195
196 if kind_o[0] == kind {
197 var ai: nx_int = 0
198 while ai < n_attrs_o[0] {
199 if attrs_o[2 * ai] == attr_key {
200 let val: nx_int = attrs_o[2 * ai + 1]
201 // Linear search for val in distinct_out[0..n_distinct]
202 var found: nx_int = 0
203 var di: nx_int = 0
204 while di < n_distinct {
205 if distinct_out[di] == val { found = 1 }
206 di = di + 1
207 }
208 if found == 0 {
209 if n_distinct < distinct_cap {
210 distinct_out[n_distinct] = val
211 n_distinct = n_distinct + 1
212 }
213 }
214 }
215 ai = ai + 1
216 }
217 }
218 pos = np[0]
219 }
220 return n_distinct
221}