nx_srcreach.nx source

1// nx_srcreach.nx -- THE SOURCE REGISTRY READER AND THE PACE BUDGET, for /compare/researchreach rungs RR1 and RR3. 2// 3// Operator 2026-09-06: "we need to measure and make sure we never have any unreachable data whether by search or api 4// we need to be able to gather it all" and "a regular cadence for all domains without smashing our nas resources". 5// 6// nx_srcreach load <conf> -- read and classify the registry, print the partition 7// nx_srcreach plan <conf> <reqs_per_source> <max_requests> -- ALSO print the request budget, and REFUSE over the ceiling 8// 9// WHY plan PRINTS BEFORE IT WOULD FETCH: a sweep that cannot state its own request count is a load incident waiting to 10// happen. The budget is computed and announced with NO network touched, so the decision to go wide is made on an 11// arithmetic a person can check rather than discovered from the array afterwards. 12// 13// RR2 LANDED 2026-09-06 AND THE PROBE IS REAL. An earlier draft of this header explained why the probe entry point was 14// deliberately absent; that note is retired because the capability now exists rather than the name. It COMPOSES the 15// estate's fetcher over sovereign TLS instead of re-implementing a client, probes into ONE fixed scratch name so a 16// forty-source sweep leaves one file rather than forty mirrors, paces per host, and reports a partition over EVERY row. 17// A HISTORICAL NOTE WORTH KEEPING: while that paragraph existed, nx_domain_admit reported rule-disagree wordboundary=1 18// declared=0 against this very file -- the naive scan found the symbol inside the comment explaining its absence, which 19// is the estate's own prose-is-source-bytes trap firing on the organ that documents it. 20// 21// BITE-PROVEN LOCALLY 2026-09-06 on a fixture assembled at runtime with FOUR planted defects: each fired, each was 22// NAMED by its key, and the partition summed (rows=6 ok=2 bad_fields=1 bad_kind=1 bad_proto=1 bad_pace=1 sum=6, 23// verdict=REFUSED exit=1). The POSITIVE CONTROL is the one that matters most: a clean three-row registry read 24// rows=3 ok=3 sum=3 verdict=OK exit=0, so this is not a guard that refuses everything and passes every absence test. 25// The budget was proven in both directions too: 3 sources x 10 requests = 30 planned, ceiling 100 -> OK exit 0, 26// ceiling 20 -> REFUSED-OVER-CEILING exit 1 with nothing fetched. 27// 28// exit: 0 OK | 1 REFUSED (a bad row, a partition that does not reconcile, or a budget over the ceiling) | 2 usage | 29// 3 NO-REGISTRY (unreadable or no src rows) 30// license_tier: ORIGINAL No hw writes (Rule 26). 31import "nx_syscalls.nx" 32import "nx_srcreg_lib.nx" 33import "nx_tool_run.nx" 34 35const SRC_STDOUT: i64 = 1 36const SRC_NUM_CAP: i64 = 24 37const SRC_ARG_VERB: i64 = 1 38const SRC_ARG_CONF: i64 = 2 39const SRC_ARG_REQS: i64 = 3 40const SRC_ARG_CEIL: i64 = 4 41const SRC_ARG_MAXPROBE: i64 = 5 42const SRC_DASH: i64 = 45 43// budget slots 44const SRC_B_REQUESTS: i64 = 0 45const SRC_B_LONGEST_MS: i64 = 1 46const SRC_B_TOTAL_MS: i64 = 2 47const SRC_B_N: i64 = 3 48 49func src_w(s: *u8) -> i64 { 50 return sys_write(SRC_STDOUT, s, sr_slen(s)) 51} 52func src_wn(v: i64) -> i64 { 53 let b: *u8 = sys_mmap(SRC_NUM_CAP) 54 var x: i64 = v 55 var neg: i64 = 0 56 if x < 0 { neg = 1; x = 0 - x } 57 var o: i64 = SRC_NUM_CAP - 1 58 b[o] = 0 as u8 59 o = o - 1 60 if x == 0 { b[o] = SR_DIGIT0 as u8; o = o - 1 } 61 while x > 0 { b[o] = (x - (x / 10) * 10 + SR_DIGIT0) as u8; o = o - 1; x = x / 10 } 62 if neg == 1 { b[o] = SRC_DASH as u8; o = o - 1 } 63 return src_w((b as i64 + o + 1) as *u8) 64} 65func src_wspan(buf: *u8, off: i64, len: i64) -> i64 { 66 if len <= 0 { return 0 } 67 return sys_write(SRC_STDOUT, (buf as i64 + off) as *u8, len) 68} 69 70// SR_PACE -- the outbound budget, computed with NO network touched. 71// requests = OK rows x reqs_per_source. This is the LOAD figure and the one the ceiling governs. 72// longest_ms = the slowest single source's wall floor, (reqs_per_source - 1) x its pace. Sources are paced PER HOST, 73// so they may interleave; the slowest host therefore sets the floor on wall time for the whole sweep. 74// total_ms = the sum of every source's wall floor, which is the wall time if the sweep runs strictly sequentially. 75// Both are printed because they bound the run from opposite ends and quoting one alone would misstate the cost. 76// A row that is not OK contributes NOTHING to the budget: an unparseable pace must never be spent against. 77func sr_pace(st: *i64, pace: *i64, nrows: i64, reqs_per_source: i64, out: *i64) -> i64 { 78 out[SRC_B_REQUESTS] = 0 79 out[SRC_B_LONGEST_MS] = 0 80 out[SRC_B_TOTAL_MS] = 0 81 if reqs_per_source <= 0 { return 0 } 82 var i: i64 = 0 83 while i < nrows { 84 if st[i] == SR_S_OK { 85 out[SRC_B_REQUESTS] = out[SRC_B_REQUESTS] + reqs_per_source 86 // reqs x pace, NOT (reqs-1) x pace. CORRECTED 2026-09-06 AFTER THE FIRST LIVE CENSUS EXPOSED THE 87 // DISAGREEMENT: the (reqs-1) form models waits BETWEEN a host's own requests, which is the textbook 88 // answer and is not what sr_probe does -- it sleeps pace_ms after EVERY request, including the last one 89 // to a host, because that sleep is also the spacer before the NEXT host. At one request per source the 90 // old model printed wall 0 while the run actually slept the sum of every pace. A PRODUCER AND A CONSUMER 91 // EACH DEFENSIBLE ALONE CAN STILL DESCRIBE DIFFERENT THINGS; the budget now models the behaviour that 92 // exists rather than the one a formula suggests, so the printed cost is the cost that will be paid. 93 let wall: i64 = reqs_per_source * pace[i] 94 out[SRC_B_TOTAL_MS] = out[SRC_B_TOTAL_MS] + wall 95 if wall > out[SRC_B_LONGEST_MS] { out[SRC_B_LONGEST_MS] = wall } 96 } 97 i = i + 1 98 } 99 return out[SRC_B_REQUESTS] 100} 101 102// ---- RR2: REACHABILITY, MEASURED, WITH THE BLOCKER NAMED ---- 103// Operator 2026-09-06: "we need to measure and make sure we never have any unreachable data whether by search or api". 104// It COMPOSES the estate's fetcher rather than re-implementing a TLS client, and it probes into ONE fixed scratch 105// name so forty probes leave one file behind instead of forty mirrors -- the same discipline nx_refdrift uses. 106// UNPROBED IS ITS OWN STATE AND IS COUNTED: a source nobody probed must never read as one that passed, and a row 107// refused by the registry reader is UNPROBED rather than silently dropped, so rows = sum(states) over EVERY row. 108// A TRANSPORT outcome is NOT folded into any status class: a host that never completed a handshake has told us 109// something different from a host that answered 404, and collapsing them would hide the codeberg TLS class entirely. 110const SRP_REACHABLE: i64 = 0 111const SRP_RATELIMITED: i64 = 1 112const SRP_AUTH: i64 = 2 113const SRP_NOTFOUND: i64 = 3 114const SRP_OTHER: i64 = 4 115const SRP_TRANSPORT: i64 = 5 116const SRP_UNPROBED: i64 = 6 117const SRP_N: i64 = 7 118const SRP_FETCHER: *u8 = "nx_research_fetch.elf" 119const SRP_PROBE_NAME: *u8 = "srcreach_probe.tmp" 120const SRP_CAP: i64 = 65536 121const SRP_URLCAP: i64 = 1024 122const SRP_TIMEOUT_MS: i64 = 45000 123const SRP_ARGV_SLOTS: i64 = 32 124 125func srp_name(s: i64) -> *u8 { 126 if s == SRP_REACHABLE { return "REACHABLE" as *u8 } 127 if s == SRP_RATELIMITED { return "RATE-LIMITED" as *u8 } 128 if s == SRP_AUTH { return "AUTH-REQUIRED" as *u8 } 129 if s == SRP_NOTFOUND { return "NOT-FOUND" as *u8 } 130 if s == SRP_OTHER { return "OTHER-STATUS" as *u8 } 131 if s == SRP_TRANSPORT { return "TRANSPORT-FAIL" as *u8 } 132 if s == SRP_UNPROBED { return "UNPROBED" as *u8 } 133 return "UNKNOWN" as *u8 134} 135 136// the remedy travels with the finding: a count without a worklist is not actionable, and neither is a blocker 137// without the action that would clear it 138func srp_remedy(s: i64) -> *u8 { 139 if s == SRP_RATELIMITED { return "raise this row's pace_ms and re-probe later; do not spend requests on a throttled host" as *u8 } 140 if s == SRP_AUTH { return "the endpoint needs a credential: set auth on the row and admit a key, or accept metadata-only reach" as *u8 } 141 if s == SRP_NOTFOUND { return "the endpoint moved: re-read the vendor's current API base and correct the row" as *u8 } 142 if s == SRP_OTHER { return "read the status the fetcher printed; a redirect needs the canonical url in the row" as *u8 } 143 if s == SRP_TRANSPORT { return "no handshake completed: our sovereign TLS chain may not accept this host (the measured codeberg class), or the host is down" as *u8 } 144 if s == SRP_UNPROBED { return "not examined this run -- raise the request ceiling or split the sweep; this is NOT a pass" as *u8 } 145 return "-" as *u8 146} 147 148// clamp FIRST, then copy: a loop that breaks by clobbering its own cursor cannot also report where it stopped 149func sr_copy_span(buf: *u8, off: i64, len: i64, dst: *u8, cap: i64) -> i64 { 150 var n: i64 = len 151 if n < 0 { n = 0 } 152 if n > cap - 1 { n = cap - 1 } 153 var i: i64 = 0 154 while i < n { dst[i] = buf[off + i]; i = i + 1 } 155 dst[n] = 0 as u8 156 return n 157} 158 159// classify one probe from the fetcher's own captured output. Ordered most-specific first; the bare status= test is 160// LAST so a named class is never swallowed by it, and absence of any status at all is TRANSPORT and not a status. 161func sr_probe_classify(out: *u8, n: i64) -> i64 { 162 if tr_contains(out, n, "status=200" as *u8) == 1 { return SRP_REACHABLE } 163 if tr_contains(out, n, "status=429" as *u8) == 1 { return SRP_RATELIMITED } 164 if tr_contains(out, n, "status=401" as *u8) == 1 { return SRP_AUTH } 165 if tr_contains(out, n, "status=403" as *u8) == 1 { return SRP_AUTH } 166 if tr_contains(out, n, "status=404" as *u8) == 1 { return SRP_NOTFOUND } 167 if tr_contains(out, n, "status=" as *u8) == 1 { return SRP_OTHER } 168 return SRP_TRANSPORT 169} 170 171// SR_PROBE -- probe every OK row up to the request ceiling, pacing per host. Every row gets a state, so the 172// partition covers the whole registry and must sum to rows. 173func sr_probe(buf: *u8, n: i64, off: *i64, st: *i64, pace: *i64, rows: i64, limit: i64, pst: *i64, pc: *i64) -> i64 { 174 var s: i64 = 0 175 while s < SRP_N { pc[s] = 0; s = s + 1 } 176 let url: *u8 = sys_mmap(SRP_URLCAP) 177 let out: *u8 = sys_mmap(SRP_CAP) 178 let olen: *i64 = sys_mmap(SR_I64) as *i64 179 let fo: *i64 = sys_mmap(SR_I64) as *i64 180 let av: *i64 = sys_mmap(SRP_ARGV_SLOTS) as *i64 181 var i: i64 = 0 182 var used: i64 = 0 183 while i < rows { 184 var state: i64 = SRP_UNPROBED 185 let p: i64 = off[i] 186 let e: i64 = sr_line_end(buf, n, p) 187 if st[i] == SR_S_OK { 188 if used < limit { 189 let ul: i64 = sr_field(buf, p, e, SR_F_URL, fo) 190 var uend: i64 = sr_copy_span(buf, fo[0], ul, url, SRP_URLCAP) 191 // OPTIONAL PROBE PATH appended to the base. A base-url is a BASE, and the first live census proved 192 // a probe that treats it as a target measures url shape as much as reach. 193 let ppl: i64 = sr_field(buf, p, e, SR_F_PROBE_PATH, fo) 194 if ppl > 0 { uend = uend + sr_copy_span(buf, fo[0], ppl, (url as i64 + uend) as *u8, SRP_URLCAP - uend) } 195 av[0] = SRP_FETCHER as i64 196 av[1] = url as i64 197 av[2] = SRP_PROBE_NAME as i64 198 av[3] = 0 199 olen[0] = 0 200 tr_run_capture_to(SRP_FETCHER, av, out, SRP_CAP, olen, SRP_TIMEOUT_MS) 201 state = sr_probe_classify(out, olen[0]) 202 used = used + 1 203 sys_sleep_ms(pace[i]) 204 } 205 } 206 pst[i] = state 207 pc[state] = pc[state] + 1 208 // PRINTED AS DECIDED, NOT AFTER THE SWEEP. A long run that buffers its findings has no partial results, so 209 // an interruption costs everything rather than the remainder -- the all-or-nothing shape this estate has 210 // already paid for. Each row is emitted the moment it is classified. 211 if state != SRP_REACHABLE { 212 src_w(" REACH state=" as *u8); src_w(srp_name(state)) 213 src_w(" key=" as *u8) 214 let kl: i64 = sr_field(buf, p, e, SR_F_KEY, fo) 215 if kl > 0 { src_wspan(buf, fo[0], kl) } else { src_w("-" as *u8) } 216 src_w(" remedy=" as *u8); src_w(srp_remedy(state)) 217 src_w("\n" as *u8) 218 } 219 i = i + 1 220 } 221 return used 222} 223 224func srp_partition_sum(pc: *i64) -> i64 { 225 var s: i64 = 0 226 var t: i64 = 0 227 while s < SRP_N { t = t + pc[s]; s = s + 1 } 228 return t 229} 230 231func main(argc: i64, argv: *i64) -> i64 { 232 if argc <= SRC_ARG_CONF { 233 src_w("usage: nx_srcreach load <conf> | nx_srcreach plan <conf> <reqs_per_source> <max_requests>\n" as *u8) 234 src_w(" exit 0 OK | 1 REFUSED | 2 usage | 3 NO-REGISTRY. The RR2 probe entry point is deliberately not defined yet.\n" as *u8) 235 return SR_EXIT_USAGE 236 } 237 let verb: *u8 = argv[SRC_ARG_VERB] as *u8 238 let conf: *u8 = argv[SRC_ARG_CONF] as *u8 239 var want_plan: i64 = 0 240 var want_probe: i64 = 0 241 if sr_span_eq(verb, 0, sr_slen(verb), "plan" as *u8) == 1 { want_plan = 1 } 242 if sr_span_eq(verb, 0, sr_slen(verb), "probe" as *u8) == 1 { want_probe = 1; want_plan = 1 } 243 if want_plan == 0 { 244 if sr_span_eq(verb, 0, sr_slen(verb), "load" as *u8) == 0 { 245 src_w("usage: the verb must be load, plan or probe\n" as *u8) 246 return SR_EXIT_USAGE 247 } 248 } 249 var reqs: i64 = 1 250 var ceil: i64 = 0 - 1 251 // THE PROBE LIMIT DEFAULTS TO THE CEILING AND MAY BE OVERRIDDEN. Two different questions: the ceiling asks what 252 // the budget PERMITS, the limit asks what this run ISSUES. Defaulting the second to the first keeps every 253 // existing call working while making probe five of forty expressible for the first time. 254 var maxprobe: i64 = 0 - 1 255 if want_plan == 1 { 256 if argc <= SRC_ARG_CEIL { 257 src_w("usage: plan needs <reqs_per_source> and <max_requests>\n" as *u8) 258 return SR_EXIT_USAGE 259 } 260 let ra: *u8 = argv[SRC_ARG_REQS] as *u8 261 let ca: *u8 = argv[SRC_ARG_CEIL] as *u8 262 reqs = sr_atoi_span(ra, 0, sr_slen(ra)) 263 ceil = sr_atoi_span(ca, 0, sr_slen(ca)) 264 if reqs == SR_NONE { src_w("usage: reqs_per_source must be a positive integer\n" as *u8); return SR_EXIT_USAGE } 265 if ceil == SR_NONE { src_w("usage: max_requests must be a positive integer\n" as *u8); return SR_EXIT_USAGE } 266 if reqs <= 0 { src_w("usage: reqs_per_source must be a positive integer\n" as *u8); return SR_EXIT_USAGE } 267 maxprobe = ceil 268 if argc > SRC_ARG_MAXPROBE { 269 let ma: *u8 = argv[SRC_ARG_MAXPROBE] as *u8 270 maxprobe = sr_atoi_span(ma, 0, sr_slen(ma)) 271 if maxprobe == SR_NONE { src_w("usage: max_probe must be a non-negative integer\n" as *u8); return SR_EXIT_USAGE } 272 } 273 } 274 275 let lenp: *i64 = sys_mmap(SR_I64) as *i64 276 let buf: *u8 = sr_load(conf, lenp) 277 src_w("SRCREACH conf=" as *u8); src_w(conf) 278 if lenp[0] <= 0 { 279 src_w(" registry=UNREADABLE rows=0 verdict=NO-REGISTRY\n" as *u8) 280 return SR_EXIT_NOREG 281 } 282 src_w(" bytes=" as *u8); src_wn(lenp[0]); src_w("\n" as *u8) 283 let n: i64 = lenp[0] 284 let nr: i64 = sr_count_rows(buf, n) 285 if nr <= 0 { 286 src_w("SRCREACH rows=0 verdict=NO-REGISTRY -- the file is readable and declares no src rows\n" as *u8) 287 return SR_EXIT_NOREG 288 } 289 let off: *i64 = sys_mmap((nr + 1) * SR_I64) as *i64 290 let st: *i64 = sys_mmap((nr + 1) * SR_I64) as *i64 291 let pace: *i64 = sys_mmap((nr + 1) * SR_I64) as *i64 292 let kind: *i64 = sys_mmap((nr + 1) * SR_I64) as *i64 293 let c: *i64 = sys_mmap(SR_C_N * SR_I64) as *i64 294 let rows: i64 = sr_classify(buf, n, off, st, pace, kind, c) 295 let fo: *i64 = sys_mmap(SR_I64) as *i64 296 297 // EVERY REFUSED ROW IS NAMED. A count without a worklist is not actionable. 298 var i: i64 = 0 299 while i < rows { 300 if st[i] != SR_S_OK { 301 let p: i64 = off[i] 302 let e: i64 = sr_line_end(buf, n, p) 303 src_w(" REFUSED-ROW state=" as *u8); src_w(sr_state_name(st[i])) 304 src_w(" key=" as *u8) 305 let kl: i64 = sr_field(buf, p, e, SR_F_KEY, fo) 306 if kl > 0 { src_wspan(buf, fo[0], kl) } else { src_w("-" as *u8) } 307 src_w("\n" as *u8) 308 } 309 i = i + 1 310 } 311 312 if want_plan == 1 { 313 let b: *i64 = sys_mmap(SRC_B_N * SR_I64) as *i64 314 sr_pace(st, pace, rows, reqs, b) 315 src_w("SRCREACH-BUDGET reqs_per_source=" as *u8); src_wn(reqs) 316 src_w(" sources_counted=" as *u8); src_wn(c[SR_C_OK]) 317 src_w(" planned_requests=" as *u8); src_wn(b[SRC_B_REQUESTS]) 318 src_w(" ceiling=" as *u8); src_wn(ceil) 319 src_w(" longest_source_wall_ms=" as *u8); src_wn(b[SRC_B_LONGEST_MS]) 320 src_w(" sequential_wall_ms=" as *u8); src_wn(b[SRC_B_TOTAL_MS]) 321 src_w("\n" as *u8) 322 if b[SRC_B_REQUESTS] > ceil { 323 src_w("SRCREACH REFUSED-OVER-CEILING planned_requests=" as *u8); src_wn(b[SRC_B_REQUESTS]) 324 src_w(" exceeds max_requests=" as *u8); src_wn(ceil) 325 src_w(" -- lower reqs_per_source, raise the ceiling deliberately, or split the sweep. Nothing was fetched.\n" as *u8) 326 src_w("SRCREACH rows=" as *u8); src_wn(c[SR_C_ROWS]) 327 src_w(" ok=" as *u8); src_wn(c[SR_C_OK]) 328 src_w(" bad_fields=" as *u8); src_wn(c[SR_C_BAD_FIELDS]) 329 src_w(" bad_kind=" as *u8); src_wn(c[SR_C_BAD_KIND]) 330 src_w(" bad_proto=" as *u8); src_wn(c[SR_C_BAD_PROTO]) 331 src_w(" bad_pace=" as *u8); src_wn(c[SR_C_BAD_PACE]) 332 src_w(" sum=" as *u8); src_wn(sr_partition_sum(c)) 333 src_w(" verdict=REFUSED-OVER-CEILING\n" as *u8) 334 return SR_EXIT_BAD 335 } 336 } 337 338 if want_probe == 1 { 339 let pst: *i64 = sys_mmap((rows + 1) * SR_I64) as *i64 340 let pc: *i64 = sys_mmap(SRP_N * SR_I64) as *i64 341 // maxprobe, NOT ceil. The ceiling governs how many requests the budget PERMITS; the probe limit governs how 342 // many this run ACTUALLY ISSUES. Conflating them meant there was no way to say probe five of forty -- the 343 // budget check refused first. Two questions, two arguments. 344 let probed: i64 = sr_probe(buf, n, off, st, pace, rows, maxprobe, pst, pc) 345 src_w("SRCREACH-REACH probed=" as *u8); src_wn(probed) 346 src_w(" max_probe=" as *u8); src_wn(maxprobe) 347 src_w(" reachable=" as *u8); src_wn(pc[SRP_REACHABLE]) 348 src_w(" rate_limited=" as *u8); src_wn(pc[SRP_RATELIMITED]) 349 src_w(" auth_required=" as *u8); src_wn(pc[SRP_AUTH]) 350 src_w(" not_found=" as *u8); src_wn(pc[SRP_NOTFOUND]) 351 src_w(" other_status=" as *u8); src_wn(pc[SRP_OTHER]) 352 src_w(" transport_fail=" as *u8); src_wn(pc[SRP_TRANSPORT]) 353 src_w(" unprobed=" as *u8); src_wn(pc[SRP_UNPROBED]) 354 src_w(" sum=" as *u8); src_wn(srp_partition_sum(pc)) 355 src_w(" rows=" as *u8); src_wn(rows) 356 src_w("\n" as *u8) 357 } 358 359 let v: i64 = sr_verdict(c) 360 src_w("SRCREACH rows=" as *u8); src_wn(c[SR_C_ROWS]) 361 src_w(" ok=" as *u8); src_wn(c[SR_C_OK]) 362 src_w(" bad_fields=" as *u8); src_wn(c[SR_C_BAD_FIELDS]) 363 src_w(" bad_kind=" as *u8); src_wn(c[SR_C_BAD_KIND]) 364 src_w(" bad_proto=" as *u8); src_wn(c[SR_C_BAD_PROTO]) 365 src_w(" bad_pace=" as *u8); src_wn(c[SR_C_BAD_PACE]) 366 src_w(" sum=" as *u8); src_wn(sr_partition_sum(c)) 367 src_w(" verdict=" as *u8); src_w(sr_verdict_name(v)); src_w("\n" as *u8) 368 return v 369}