code wiki / _hdl_build / nx_analyst_store_scale_gate.nx
nx_analyst_store_scale_gate.nx source
↩ module page · 191 lines · 8948 B
1// nx_analyst_store_scale_gate.nx -- F1001: prove the analyst survives a dataset LARGER than one seg_store
2// version window, and that when it cannot, it SAYS SO instead of quietly analysing a recency slice.
3//
4// THE DEFECT THIS PINS (found 2026-07-23 reading nx_seg_store for F1001): ss_scan keeps at most
5// SS_VER_WINDOW=256 versions of one key and the window SLIDES -- oldest out, newest kept. So 300 rows
6// stored under one key read back as the NEWEST 256, and nx_analyst_store returned them with no signal at
7// all. Every number in the resulting report described a slice nobody asked for. T1/T2 reproduce that on
8// real committed data; T3-T5 prove the sharded path reads all 300; T6-T8 pin the fail-closed edges.
9// D001-compliant: verdict via the shared nx_gate_verdict lib. license_tier: ORIGINAL expect_exit: 0
10import "nx_gate_verdict.nx"
11import "_hdl_build/nx_analyst_store.nx"
12
13const SG_ROWS: i64 = 300 // > SS_VER_WINDOW on purpose
14const SG_SHARD: i64 = 150 // rows per shard in the sharded store
15const SG_MAXSH: i64 = 8
16const SG_OUTCAP: i64 = 32768
17
18func sg_lcg(s: i64) -> i64 { return (s * 1103515245 + 12345) & 0x7fffffff }
19func sg_catn(buf: *u8, o: i64, v: i64) -> i64 {
20 var oo: i64 = o
21 var m: i64 = v
22 if m < 0 { buf[oo] = 45 as u8; oo = oo + 1; m = 0 - m }
23 let t: *u8 = sys_mmap(24)
24 var k: i64 = 0
25 if m == 0 { t[0] = 48 as u8; k = 1 }
26 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
27 var i: i64 = k - 1
28 while i >= 0 { buf[oo] = t[i]; oo = oo + 1; i = i - 1 }
29 return oo
30}
31func sg_has(hay: *u8, n: i64, needle: *u8) -> i64 {
32 var nl: i64 = 0
33 while needle[nl] != (0 as u8) { nl = nl + 1 }
34 var i: i64 = 0
35 while i + nl <= n {
36 var j: i64 = 0
37 var eq: i64 = 1
38 while j < nl { if hay[i+j] != needle[j] { eq = 0 } j = j + 1 }
39 if eq == 1 { return 1 }
40 i = i + 1
41 }
42 return 0
43}
44func sg_eq(a: i64, b: i64) -> i64 { if a == b { return 1 } return 0 }
45// build "<pfx><n>" into out
46func sg_prefix(out: *u8, tag: *u8, n: i64) -> i64 {
47 var o: i64 = 0
48 let base: *u8 = "/tmp/anscale_" as *u8
49 while base[o] != (0 as u8) { out[o] = base[o]; o = o + 1 }
50 var j: i64 = 0
51 while tag[j] != (0 as u8) { out[o] = tag[j]; o = o + 1; j = j + 1 }
52 o = sg_catn(out, o, n)
53 out[o] = 0 as u8
54 return o
55}
56// one dataset row: id <TAB> size <TAB> age <TAB> price, price = 5*size - 2*age + noise
57func sg_row(row: *u8, id: i64, seed: i64) -> i64 {
58 var s: i64 = sg_lcg(seed)
59 let size: i64 = 50 + (s % 150)
60 s = sg_lcg(s)
61 let age: i64 = s % 40
62 s = sg_lcg(s)
63 let price: i64 = 5 * size - 2 * age + (s % 20)
64 var ro: i64 = 0
65 ro = sg_catn(row, ro, id)
66 row[ro] = 9 as u8; ro = ro + 1
67 ro = sg_catn(row, ro, size)
68 row[ro] = 9 as u8; ro = ro + 1
69 ro = sg_catn(row, ro, age)
70 row[ro] = 9 as u8; ro = ro + 1
71 ro = sg_catn(row, ro, price)
72 return ro
73}
74
75func main() -> i64 {
76 let ctr: *i64 = gv_ctr()
77 gv_head("nx_analyst_store_scale_gate -- 300 rows through a 256-version window: caught, or sharded" as *u8)
78 let now: i64 = sys_now_us()
79 let row: *u8 = sys_mmap(96)
80 let fl: *i64 = sys_mmap(8) as *i64
81
82 // ---- STORE A: all SG_ROWS under ONE key -> must be DETECTED as windowed -------------------
83 let pa: *u8 = sys_mmap(128)
84 sg_prefix(pa, "a" as *u8, now)
85 let wa: *i64 = ss_begin()
86 var i: i64 = 0
87 while i < SG_ROWS { let rl: i64 = sg_row(row, i, 11 + i * 7); ss_add(wa, 1, "dsrow" as *u8, row, rl); i = i + 1 }
88 ss_commit(pa, wa, now)
89
90 let colA: *i64 = sys_mmap(8 * (SG_ROWS + 64)) as *i64
91 fl[0] = 0
92 let gotA: i64 = asr_load_field_win(pa, "dsrow" as *u8, 0, colA, SG_ROWS + 32, fl)
93 gv_check("T1a 300 rows under one key read back as exactly SS_VER_WINDOW (the window SLID)" as *u8, sg_eq(gotA, SS_VER_WINDOW), ctr)
94 gv_check("T1b the truncation is DETECTED (flag set), not returned silently" as *u8, sg_eq(fl[0], 1), ctr)
95 // the surviving rows are the NEWEST: ids 44..299, so the first id back must NOT be 0
96 gv_check("T1c the surviving slice is the NEWEST rows (first id != 0) -- proves data was DROPPED" as *u8, sg_eq(colA[0], SG_ROWS - SS_VER_WINDOW), ctr)
97
98 let names: *i64 = sys_mmap(8 * 4) as *i64
99 names[0] = "id" as *u8 as i64
100 names[1] = "size" as *u8 as i64
101 names[2] = "age" as *u8 as i64
102 names[3] = "price" as *u8 as i64
103 let outA: *u8 = sys_mmap(SG_OUTCAP)
104 let rwA: i64 = asr_analyze_stored(pa, "dsrow" as *u8, names, 4, 3, SG_ROWS + 32, outA, SG_OUTCAP)
105 gv_check("T2 the REPORT leads with the TRUNCATED-LOAD banner (silence here is the whole bug)" as *u8, sg_has(outA, rwA, "!! TRUNCATED LOAD" as *u8), ctr)
106
107 // ---- STORE B: the SAME rows sharded dsrow:0 / dsrow:1 -> must read ALL of them -------------
108 let pb: *u8 = sys_mmap(128)
109 sg_prefix(pb, "b" as *u8, now)
110 let wb: *i64 = ss_begin()
111 let skey: *u8 = sys_mmap(64)
112 i = 0
113 while i < SG_ROWS {
114 asr_shard_key("dsrow" as *u8, i / SG_SHARD, skey)
115 let rl: i64 = sg_row(row, i, 11 + i * 7)
116 ss_add(wb, 1, skey, row, rl)
117 i = i + 1
118 }
119 ss_commit(pb, wb, now)
120
121 let colB: *i64 = sys_mmap(8 * (SG_ROWS + 64)) as *i64
122 fl[0] = 0
123 let gotB: i64 = asr_load_field_sharded(pb, "dsrow" as *u8, 0, colB, SG_ROWS + 32, SG_MAXSH, fl)
124 gv_check("T3a sharded load returns ALL 300 rows (past the single-key window)" as *u8, sg_eq(gotB, SG_ROWS), ctr)
125 gv_check("T3b and reports NO truncation (no shard touched the window)" as *u8, sg_eq(fl[0], 0), ctr)
126 // shard concatenation must be IN ORDER: ids ascend 0..299 straight across the shard boundary
127 var ordok: i64 = 1
128 if colB[0] != 0 { ordok = 0 }
129 if colB[SG_SHARD - 1] != SG_SHARD - 1 { ordok = 0 }
130 if colB[SG_SHARD] != SG_SHARD { ordok = 0 }
131 if colB[SG_ROWS - 1] != SG_ROWS - 1 { ordok = 0 }
132 gv_check("T4 shards concatenate IN ORDER across the boundary (ids 0,149,150,299 land exactly)" as *u8, ordok, ctr)
133
134 let outB: *u8 = sys_mmap(SG_OUTCAP)
135 let rwB: i64 = asr_analyze_stored_sharded(pb, "dsrow" as *u8, names, 4, 3, SG_ROWS + 32, SG_MAXSH, outB, SG_OUTCAP)
136 var t5: i64 = 1
137 if rwB <= 0 { t5 = 0 }
138 if sg_has(outB, rwB, "rows=300" as *u8) == 0 { t5 = 0 }
139 if sg_has(outB, rwB, "size has the strongest association" as *u8) == 0 { t5 = 0 }
140 if sg_has(outB, rwB, "!! " as *u8) == 1 { t5 = 0 }
141 gv_check("T5 analysis over all 300 sharded rows finds the planted driver, with NO warning banner" as *u8, t5, ctr)
142
143 // ---- fail-closed edges --------------------------------------------------------------------
144 let colC: *i64 = sys_mmap(8 * 64) as *i64
145 fl[0] = 0
146 let gotC: i64 = asr_load_field_sharded(pb, "nosuchkey" as *u8, 0, colC, 64, SG_MAXSH, fl)
147 gv_check("T6 ADVERSARY sharded load of an absent keybase -> 0 rows, no false success" as *u8, sg_eq(gotC, 0), ctr)
148
149 // shard budget exhausted with rows still unread -> MUST flag
150 let colD: *i64 = sys_mmap(8 * (SG_ROWS + 64)) as *i64
151 fl[0] = 0
152 let gotD: i64 = asr_load_field_sharded(pb, "dsrow" as *u8, 0, colD, SG_ROWS + 32, 1, fl)
153 var t7: i64 = 1
154 if gotD != SG_SHARD { t7 = 0 }
155 if fl[0] != 1 { t7 = 0 }
156 gv_check("T7 maxshards=1 stops after shard 0 AND flags that rows remain unread" as *u8, t7, ctr)
157
158 // STORE C: exactly SS_VER_WINDOW rows in one shard -- indistinguishable from truncated, so FLAG it
159 let pc: *u8 = sys_mmap(128)
160 sg_prefix(pc, "c" as *u8, now)
161 let wc: *i64 = ss_begin()
162 i = 0
163 while i < SS_VER_WINDOW { let rl: i64 = sg_row(row, i, 11 + i * 7); ss_add(wc, 1, "dsrow:0" as *u8, row, rl); i = i + 1 }
164 ss_commit(pc, wc, now)
165 let colE: *i64 = sys_mmap(8 * (SS_VER_WINDOW + 64)) as *i64
166 fl[0] = 0
167 let gotE: i64 = asr_load_field_sharded(pc, "dsrow" as *u8, 0, colE, SS_VER_WINDOW + 32, SG_MAXSH, fl)
168 var t8: i64 = 1
169 if gotE != SS_VER_WINDOW { t8 = 0 }
170 if fl[0] != 1 { t8 = 0 }
171 gv_check("T8 a shard holding EXACTLY the window flags anyway -- cannot be told apart from truncated" as *u8, t8, ctr)
172
173 // the un-windowed control: a small shard must NOT flag (the guard is not always-on)
174 let pd: *u8 = sys_mmap(128)
175 sg_prefix(pd, "d" as *u8, now)
176 let wd: *i64 = ss_begin()
177 i = 0
178 while i < 40 { let rl: i64 = sg_row(row, i, 11 + i * 7); ss_add(wd, 1, "dsrow:0" as *u8, row, rl); i = i + 1 }
179 ss_commit(pd, wd, now)
180 let colF: *i64 = sys_mmap(8 * 128) as *i64
181 fl[0] = 0
182 let gotF: i64 = asr_load_field_sharded(pd, "dsrow" as *u8, 0, colF, 128, SG_MAXSH, fl)
183 var t9: i64 = 1
184 if gotF != 40 { t9 = 0 }
185 if fl[0] != 0 { t9 = 0 }
186 gv_check("T9 CONTROL a 40-row shard loads clean and does NOT flag (guard is discriminating)" as *u8, t9, ctr)
187
188 let rc: i64 = gv_verdict("ANALYST-STORE-SCALE-GATE" as *u8, ctr, "datasets past one version window are read whole, or the report says they were not" as *u8)
189 sys_exit(rc)
190 return rc
191}