nx_swarm_heal.nx source
↩ module page · 264 lines · 14525 B
1// nx_swarm_heal.nx -- HUB-SIDE swarm endpoint self-heal (2026-08-12, residue debt 1786545237 item 4).
2// THE GAP: knowledge/swarm_nodes.conf is the resolver SSOT (nx_swarm_endpoint_lib), re-read per call,
3// but its ROW goes stale when the GPU node roams (DHCP/adapter flap: .192 -> .146 -> off-LAN). The
4// gen orchestrator then dispatches to a dead addr. Beat telemetry can't fix it: the NODE beacon row is
5// positional CPU/mem and carries no address, and a spoke reporting over the public edge cannot convey
6// its LAN address anyway. So the hub REPAIRS BY VERIFIED REACHABILITY, never by a claim:
7// check <role> [conf] -- probe the CURRENT conf addr for <role>; report FRESH/DOWN. No write.
8// heal <role> <candidates.conf> [conf] [apply]
9// -- if the current addr is DOWN, probe each candidate host:port;
10// rewrite the row's addr field ONLY to the UNIQUE reachable one.
11// 0 reachable -> LEFT (fail-closed); >1 -> AMBIGUOUS (never guess).
12// Writes only with the literal `apply`; else DRY.
13// selftest -- positive+negative probe controls + a conf-rewrite round-trip.
14// SAFETY: a prober wrong in the ADOPT direction rewrites the SSOT to a dead endpoint -- the exact outage
15// the conf's own law forbids ("a wrong endpoint is worse than none"). So adoption requires the STRONGEST
16// evidence (getsockopt SO_ERROR==0, not merely poll-writable) AND uniqueness; anything short -> no write.
17// PROBE PROVENANCE: go_probe_up is copied from nx_gen_worker.nx go_worker_up (the live multi-worker
18// liveness selector). DUPLICATE-BY-NECESSITY: that func sits inside a daemon with a main() so it is not
19// importable; both should move to a shared nx_netprobe_lib on the next nx_gen_worker touch (debt filed).
20// license_tier: ORIGINAL No hw writes (Rule 26).
21import "nx_syscalls.nx"
22import "nx_swarm_endpoint_lib.nx"
23import "nx_gate_verdict.nx"
24import "nx_netprobe_lib.nx"
25const SH_MAGIC_2048: i64 = 2048
26const SH_MAGIC_2000: i64 = 2000
27const SH_MAGIC_39271: i64 = 39271
28const SH_MAGIC_39272: i64 = 39272
29
30const SH_CONF: *u8 = "/volume1/homes/elderwesto/nishihost/knowledge/swarm_nodes.conf"
31const SH_CAP: i64 = 65536
32const SH_MODE_644: i64 = 420
33
34func sh_p(s: *u8) -> i64 { sys_write(1, s, se_len(s)); return 0 }
35func sh_pn(v: i64) -> i64 { let b: *u8=sys_mmap(28); let t: *u8=sys_mmap(28); var m: i64=v; if m<0{sys_write(1,"-" as *u8,1);m=0-m} var k: i64=0; if m==0{t[0]=48 as u8;k=1} while m>0{t[k]=(48+(m%10)) as u8;m=m/10;k=k+1} var i: i64=0; while i<k{b[i]=t[k-1-i];i=i+1} sys_write(1,b,k); return 0 }
36
37// the reachability probe now lives in nx_netprobe_lib (np_probe_up) -- the inline copy was extracted
38// 2026-08-12 (DRY, rule 15). go_probe_up is kept as a thin alias so the call sites below are untouched
39// and the change is provably behavior-preserving.
40func go_probe_up(ha: i64, hb: i64, hc: i64, hd: i64, port: i64) -> i64 { return np_probe_up(ha, hb, hc, hd, port) }
41
42// parse "a.b.c.d:port" (len-bounded) into oct[0..3] + oct[4]=port. 1 ok / 0 malformed.
43func sh_parse_addr(s: *u8, n: i64, oct: *i64) -> i64 {
44 var i: i64 = 0
45 var f: i64 = 0
46 while f < 4 {
47 var v: i64 = 0; var d: i64 = 0
48 var scan: i64 = 1
49 while scan == 1 { if i >= n { scan = 0 } else { let c: i64 = s[i] as i64; if c < 48 { scan = 0 } else { if c > 57 { scan = 0 } else { v = v*10 + (c-48); d = d+1; i = i+1 } } } }
50 if d == 0 { return 0 }
51 if v > 255 { return 0 }
52 oct[f] = v
53 if f < 3 { if i >= n { return 0 } if s[i] as i64 != 46 { return 0 } i = i + 1 }
54 f = f + 1
55 }
56 if i >= n { return 0 }
57 if s[i] as i64 != 58 { return 0 }
58 i = i + 1
59 var port: i64 = 0; var pd: i64 = 0
60 var scan2: i64 = 1
61 while scan2 == 1 { if i >= n { scan2 = 0 } else { let c: i64 = s[i] as i64; if c < 48 { scan2 = 0 } else { if c > 57 { scan2 = 0 } else { port = port*10 + (c-48); pd = pd+1; i = i+1 } } } }
62 if pd == 0 { return 0 }
63 oct[4] = port
64 return 1
65}
66
67// find role row -> box[0]=addr_off box[1]=addr_len (into buf). 1 found / 0 not.
68func sh_addr_span(buf: *u8, n: i64, role: *u8, box: *i64) -> i64 {
69 var p: i64 = 0
70 while p < n {
71 var q: i64 = p
72 var eol: i64 = 0
73 while eol == 0 { if q >= n { eol = 1 } else { if buf[q] == (10 as u8) { eol = 1 } else { q = q + 1 } } }
74 if q > p + 1 { if buf[p] == (82 as u8) { if buf[p+1] == (9 as u8) {
75 let fb: *i64 = sys_mmap(32) as *i64
76 se_field(buf, p, q, 1, fb)
77 if se_seq(buf, fb[0], fb[1], role) == 1 {
78 se_field(buf, p, q, 3, fb)
79 box[0] = fb[0]; box[1] = fb[1]
80 return 1
81 }
82 } } }
83 p = q + 1
84 }
85 return 0
86}
87
88// atomic replace of the addr span with new_addr. 1 ok / 0 fail. tmp+rename.
89func sh_rewrite(conf: *u8, buf: *u8, n: i64, off: i64, len: i64, new_addr: *u8) -> i64 {
90 let out: *u8 = sys_mmap(SH_CAP)
91 var o: i64 = 0
92 var i: i64 = 0
93 while i < off { out[o] = buf[i]; o = o+1; i = i+1 }
94 var j: i64 = 0
95 while new_addr[j] != (0 as u8) { out[o] = new_addr[j]; o = o+1; j = j+1 }
96 i = off + len
97 while i < n { out[o] = buf[i]; o = o+1; i = i+1 }
98 let tmp: *u8 = sys_mmap(512)
99 var to: i64 = 0
100 while conf[to] != (0 as u8) { tmp[to] = conf[to]; to = to+1 }
101 let suf: *u8 = ".healtmp" as *u8
102 var sj: i64 = 0
103 while suf[sj] != (0 as u8) { tmp[to] = suf[sj]; to = to+1; sj = sj+1 }
104 tmp[to] = 0 as u8
105 let fd: i64 = sys_openat_wr(tmp, SH_MODE_644)
106 if fd < 0 { return 0 }
107 sys_write(fd, out, o)
108 sys_close(fd)
109 if sys_renameat(tmp, conf) < 0 { return 0 }
110 return 1
111}
112
113// probe a role's current conf addr. returns 2 up / 1 down / 0 unresolved.
114func sh_probe_role(role: *u8) -> i64 {
115 let a: *u8 = se_addr(role)
116 if (a as i64) == 0 { return 0 }
117 let oct: *i64 = sys_mmap(64) as *i64
118 if sh_parse_addr(a, se_len(a), oct) == 0 { return 0 }
119 if go_probe_up(oct[0], oct[1], oct[2], oct[3], oct[4]) == 1 { return 2 }
120 return 1
121}
122
123// count reachable candidates; last reachable -> winner[] (unique iff count==1). returns reachable count.
124func sh_scan_candidates(path: *u8, winner: *u8) -> i64 {
125 let buf: *u8 = sys_mmap(SH_CAP)
126 let n: i64 = se_slurp(path, buf, SH_CAP - 1)
127 if n <= 0 { return 0 }
128 var reach: i64 = 0
129 var p: i64 = 0
130 while p < n {
131 var q: i64 = p
132 var eol: i64 = 0
133 while eol == 0 { if q >= n { eol = 1 } else { if buf[q] == (10 as u8) { eol = 1 } else { q = q + 1 } } }
134 if q > p { if buf[p] != (35 as u8) { if buf[p] != (10 as u8) {
135 var le: i64 = q
136 if le > p { if buf[le-1] == (13 as u8) { le = le - 1 } }
137 let ln: i64 = le - p
138 if ln > 6 { if ln < 40 {
139 let cand: *u8 = sys_mmap(64)
140 var ci: i64 = 0
141 while ci < ln { cand[ci] = buf[p+ci]; ci = ci+1 }
142 cand[ln] = 0 as u8
143 let oct: *i64 = sys_mmap(64) as *i64
144 if sh_parse_addr(cand, ln, oct) == 1 {
145 if go_probe_up(oct[0], oct[1], oct[2], oct[3], oct[4]) == 1 {
146 reach = reach + 1
147 var wi: i64 = 0
148 while wi <= ln { winner[wi] = cand[wi]; wi = wi+1 }
149 }
150 }
151 } }
152 } } }
153 p = q + 1
154 }
155 return reach
156}
157
158func sh_heal(role: *u8, candpath: *u8, conf: *u8, apply: i64) -> i64 {
159 // Probe the addr FROM THE conf being healed (NOT se_addr, which always reads the canonical SSOT):
160 // the probe and the rewrite must agree on one file, or heal <conf> checks the wrong endpoint. The
161 // e2e test caught exactly this -- a dead-port fixture read FRESH because se_addr probed the live SSOT.
162 let buf: *u8 = sys_mmap(SH_CAP)
163 let n: i64 = se_slurp(conf, buf, SH_CAP - 1)
164 if n <= 0 { sh_p("NX-SWARM-HEAL verdict=CONF-UNREADABLE (no write)\n" as *u8); return 1 }
165 let box: *i64 = sys_mmap(32) as *i64
166 if sh_addr_span(buf, n, role, box) == 0 { sh_p("NX-SWARM-HEAL role=" as *u8); sh_p(role); sh_p(" verdict=UNRESOLVED (role absent from conf; fail-closed, no write)\n" as *u8); return 0 }
167 let cur: *u8 = sys_mmap(64)
168 var ci: i64 = 0
169 while ci < box[1] { cur[ci] = buf[box[0]+ci]; ci = ci+1 }
170 cur[box[1]] = 0 as u8
171 let coct: *i64 = sys_mmap(64) as *i64
172 var up: i64 = 0
173 if sh_parse_addr(cur, box[1], coct) == 1 { if go_probe_up(coct[0], coct[1], coct[2], coct[3], coct[4]) == 1 { up = 1 } }
174 if up == 1 { sh_p("NX-SWARM-HEAL role=" as *u8); sh_p(role); sh_p(" verdict=FRESH (current addr reachable; nothing to heal)\n" as *u8); return 0 }
175 let winner: *u8 = sys_mmap(64)
176 let reach: i64 = sh_scan_candidates(candpath, winner)
177 if reach == 0 { sh_p("NX-SWARM-HEAL role=" as *u8); sh_p(role); sh_p(" verdict=NO-CANDIDATE (current addr DOWN, 0 candidates reachable; row LEFT unchanged -- fail-closed)\n" as *u8); return 0 }
178 if reach > 1 { sh_p("NX-SWARM-HEAL role=" as *u8); sh_p(role); sh_p(" verdict=AMBIGUOUS (" as *u8); sh_pn(reach); sh_p(" candidates reachable; never guess; row LEFT unchanged)\n" as *u8); return 0 }
179 if apply == 0 { sh_p("NX-SWARM-HEAL role=" as *u8); sh_p(role); sh_p(" verdict=WOULD-HEAL -> " as *u8); sh_p(winner); sh_p(" (DRY; pass apply to write)\n" as *u8); return 0 }
180 if sh_rewrite(conf, buf, n, box[0], box[1], winner) == 1 { sh_p("NX-SWARM-HEAL role=" as *u8); sh_p(role); sh_p(" verdict=HEALED -> " as *u8); sh_p(winner); sh_p("\n" as *u8); return 0 }
181 sh_p("NX-SWARM-HEAL verdict=REWRITE-FAILED (conf UNCHANGED)\n" as *u8); return 1
182}
183
184func sh_mkfile(path: *u8, content: *u8) -> i64 { let fd: i64 = sys_openat_wr(path, SH_MODE_644); if fd < 0 { return 0-1 } sys_write(fd, content, se_len(content)); sys_close(fd); return 0 }
185
186func sh_listen(port: i64) -> i64 {
187 let fd: i64 = sys_socket(2, 1, 0)
188 if fd < 0 { return 0 - 1 }
189 let optv: *u8 = sys_mmap(4); optv[0]=1 as u8; optv[1]=0 as u8; optv[2]=0 as u8; optv[3]=0 as u8
190 __syscall(54, fd, 1, 2, optv as i64, 4, 0) // setsockopt SO_REUSEADDR
191 let a: *u8 = sys_mmap(16)
192 a[0]=2 as u8; a[1]=0 as u8; a[2]=((port>>8)&0xff) as u8; a[3]=(port&0xff) as u8
193 a[4]=127 as u8; a[5]=0 as u8; a[6]=0 as u8; a[7]=1 as u8
194 var zi: i64=8; while zi<16 { a[zi]=0 as u8; zi=zi+1 }
195 if sys_bind(fd, a, 16) < 0 { sys_close(fd); return 0 - 1 }
196 if sys_listen(fd, 8) < 0 { sys_close(fd); return 0 - 1 }
197 return fd
198}
199
200func sh_selftest() -> i64 {
201 let ctr: *i64 = gv_ctr()
202 gv_head("=== nx_swarm_heal selftest -- probe controls + conf-rewrite round-trip (fixtures at runtime) ===" as *u8)
203 let lport: i64 = SH_MAGIC_39271
204 let lfd: i64 = sh_listen(lport)
205 gv_check("listener bound (test precondition)" as *u8, ( (lfd >= 0) as i64 ), ctr)
206 gv_bite("probe: UP on live listener, DOWN on dead port" as *u8, go_probe_up(127,0,0,1,lport), go_probe_up(127,0,0,1,SH_MAGIC_39272), ctr)
207 let cf: *u8 = "/tmp/nxsh_conf.conf" as *u8
208 sh_mkfile(cf, "# fixture\nR\tgpu-image\tlaptop\t192.168.8.192:7861\tlaptop 5080\tnote\nR\tgpu-video\twest\t10.0.4.13:7861\tdesc\tn2\n" as *u8)
209 let buf: *u8 = sys_mmap(SH_CAP)
210 let n: i64 = se_slurp(cf, buf, SH_CAP-1)
211 let box: *i64 = sys_mmap(32) as *i64
212 let found: i64 = sh_addr_span(buf, n, "gpu-image" as *u8, box)
213 gv_check("addr span located for gpu-image" as *u8, found, ctr)
214 var span_ok: i64 = 0
215 if found == 1 { if se_seq(buf, box[0], box[1], "192.168.8.192:7861" as *u8) == 1 { span_ok = 1 } }
216 gv_check("addr span is exactly the addr field (not the whole line)" as *u8, span_ok, ctr)
217 sh_rewrite(cf, buf, n, box[0], box[1], "192.168.10.146:7999" as *u8)
218 let b2: *u8 = sys_mmap(SH_CAP)
219 let n2: i64 = se_slurp(cf, b2, SH_CAP-1)
220 var t_new: i64 = 0; if se_has(b2, 0, n2, "192.168.10.146:7999" as *u8) == 1 { t_new = 1 }
221 var t_old: i64 = 1; if se_has(b2, 0, n2, "192.168.8.192:7861" as *u8) == 1 { t_old = 0 }
222 var t_other: i64 = 0; if se_has(b2, 0, n2, "10.0.4.13:7861" as *u8) == 1 { t_other = 1 }
223 var t_cmt: i64 = 0; if se_has(b2, 0, n2, "# fixture" as *u8) == 1 { t_cmt = 1 }
224 var t_desc: i64 = 0; if se_has(b2, 0, n2, "laptop 5080" as *u8) == 1 { t_desc = 1 }
225 gv_check("rewrite: new addr written" as *u8, t_new, ctr)
226 gv_check("rewrite: old addr gone (surgical, not appended)" as *u8, t_old, ctr)
227 gv_check("rewrite: sibling gpu-video row intact" as *u8, t_other, ctr)
228 gv_check("rewrite: comment line intact" as *u8, t_cmt, ctr)
229 gv_check("rewrite: gpu-image desc field intact" as *u8, t_desc, ctr)
230 let oc: *i64 = sys_mmap(64) as *i64
231 gv_bite("addr parse: accepts a.b.c.d:port, rejects garbage" as *u8, sh_parse_addr("1.2.3.4:56" as *u8, 10, oc), sh_parse_addr("not-an-addr" as *u8, 11, oc), ctr)
232 if lfd >= 0 { sys_close(lfd) }
233 sys_unlinkat(cf)
234 let rc: i64 = gv_verdict("SWARM-HEAL-SELFTEST" as *u8, ctr, "probe up/down controls + surgical fail-closed conf rewrite" as *u8)
235 return rc
236}
237
238func sh_streq(a: *u8, b: *u8, cap: i64) -> i64 { var i: i64=0; while i < cap { if a[i]!=b[i] { return 0 } if a[i]==(0 as u8) { return 1 } i=i+1 } return 1 }
239
240func main(argc: i64, argv: *i64) -> i64 {
241 if argc >= 2 {
242 let verb: *u8 = argv[1] as *u8
243 if sh_streq(verb, "selftest" as *u8, 9) == 1 { let rc: i64 = sh_selftest(); sys_exit(rc); return rc }
244 if sh_streq(verb, "check" as *u8, 6) == 1 { if argc >= 3 {
245 let st: i64 = sh_probe_role(argv[2] as *u8)
246 sh_p("NX-SWARM-HEAL check role=" as *u8); sh_p(argv[2] as *u8); sh_p(" " as *u8)
247 if st == 2 { sh_p("FRESH (reachable)\n" as *u8) }
248 if st == 1 { sh_p("DOWN (unreachable)\n" as *u8) }
249 if st == 0 { sh_p("UNRESOLVED (role absent from conf)\n" as *u8) }
250 sys_exit(0); return 0
251 } }
252 if sh_streq(verb, "heal" as *u8, 5) == 1 { if argc >= 4 {
253 var conf: *u8 = SH_CONF
254 if argc >= 5 { conf = argv[4] as *u8 }
255 var apply: i64 = 0
256 if argc >= 6 { if sh_streq(argv[5] as *u8, "apply" as *u8, 6) == 1 { apply = 1 } }
257 let rc: i64 = sh_heal(argv[2] as *u8, argv[3] as *u8, conf, apply)
258 sys_exit(rc); return rc
259 } }
260 }
261 sh_p("usage: nx_swarm_heal check <role> | heal <role> <candidates.conf> [conf] [apply] | selftest\n" as *u8)
262 sys_exit(2)
263 return 2
264}