code wiki / _hdl_build / nx_coedit_live_gate.nx
nx_coedit_live_gate.nx source
↩ module page · 267 lines · 13562 B
1// nx_coedit_live_gate.nx -- LIVE real-time co-editing, END-TO-END over real loopback HTTP (the fork idiom of
2// nx_relate_live_gate). This is the R2 proof that flips census OF-C1: the REAL blessed daemon
3// (/tmp/nx_coedit.sov.elf serve) is one process; TWO independent client PROCESSES concurrently edit the SAME
4// document through it and both CONVERGE to identical text -- which is exactly "real-time co-editing".
5//
6// server child = `nx_coedit serve <canon.log> <port>` (accept-loop, serves until killed)
7// client A (rep 1) = POST its ops "AC-after-root" then POLL-pull until the log carries BOTH replicas
8// client B (rep 2) = POST its ops "BD-after-root" then POLL-pull likewise
9// convergence (pinned): root children order desc by (counter,replica) => rep2 first => "BDAC" on BOTH.
10// Then the parent fork-execs `nx_coedit apply` on each client's pulled log and asserts BDAC == BDAC, and that
11// the canonical server log UNIONED all 4 ops with NO duplicates (idempotent under repeated polling).
12// expect_exit: 0 license_tier: ORIGINAL
13import "nx_syscalls.nx"
14
15import "nx_connect.nx" // bounded connect: a raw sys_connect hangs ~127s on a black-holed host
16func lg_slen(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8) { n=n+1 } return n }
17func lg_p(s: *u8) -> i64 { sys_write(1, s, lg_slen(s)); return 0 }
18func lg_cat(dst: *u8, off: i64, s: *u8) -> i64 { var i: i64=0; while s[i]!=(0 as u8) { dst[off+i]=s[i]; i=i+1 } return off+i }
19func lg_catn(dst: *u8, off: i64, v: i64) -> i64 {
20 var o: i64=off; var m: i64=v; if m<0 { dst[o]=45 as u8; o=o+1; m=0-m }
21 let t: *u8=sys_mmap(28); var k: i64=0; if m==0 { t[0]=48 as u8; k=1 }
22 while m>0 { t[k]=(48+(m%10)) as u8; m=m/10; k=k+1 }
23 var i: i64=0; while i<k { dst[o+i]=t[k-1-i]; i=i+1 } return o+k
24}
25func lg_uint(dst: *u8, off: i64, v: i64) -> i64 { return lg_catn(dst, off, v) }
26func lg_write(path: *u8, buf: *u8, n: i64) -> i64 {
27 let fd: i64=sys_openat_wr(path, 0x1a4); if fd<0 { return 0-1 }
28 sys_write(fd, buf, n); sys_close(fd); return 0
29}
30func lg_mkempty(path: *u8) -> i64 { let fd: i64=sys_openat_wr(path, 0x1a4); if fd<0 { return 0-1 } sys_close(fd); return 0 }
31func lg_readall(path: *u8, szout: *i64) -> *u8 {
32 let fd: i64=sys_openat_rd(path); if fd<0 { szout[0]=0-1; return 0 as *u8 }
33 let buf: *u8=sys_mmap(1048576); var got: i64=0; var n: i64=1
34 while n>0 { n=sys_read(fd,(buf as i64+got) as *u8,65536); if n>0 { got=got+n } }
35 sys_close(fd); szout[0]=got; return buf
36}
37func lg_has(buf: *u8, n: i64, needle: *u8) -> i64 {
38 let nl: i64=lg_slen(needle); if n<nl { return 0 }
39 var i: i64=0
40 while i+nl<=n { var ok: i64=1; var j: i64=0; while j<nl { if buf[i+j]!=needle[j] { ok=0; j=nl } else { j=j+1 } } if ok==1 { return 1 } i=i+1 }
41 return 0
42}
43func lg_count(buf: *u8, n: i64, needle: *u8) -> i64 {
44 let nl: i64=lg_slen(needle); if n<nl { return 0 }
45 var c: i64=0; var i: i64=0
46 while i+nl<=n { var ok: i64=1; var j: i64=0; while j<nl { if buf[i+j]!=needle[j] { ok=0; j=nl } else { j=j+1 } } if ok==1 { c=c+1; i=i+nl } else { i=i+1 } }
47 return c
48}
49func lg_filehas(path: *u8, needle: *u8) -> i64 { let sp: *i64=sys_mmap(16) as *i64; let b: *u8=lg_readall(path, sp); if sp[0]<0 { return 0 } return lg_has(b, sp[0], needle) }
50func lg_fileis(path: *u8, exact: *u8) -> i64 {
51 let sp: *i64=sys_mmap(16) as *i64; let b: *u8=lg_readall(path, sp)
52 let n: i64=lg_slen(exact); if sp[0]!=n { return 0 }
53 var i: i64=0; while i<n { if b[i]!=exact[i] { return 0 } i=i+1 } return 1
54}
55func lg_fileeq(p1: *u8, p2: *u8) -> i64 {
56 let s1: *i64=sys_mmap(16) as *i64; let s2: *i64=sys_mmap(16) as *i64
57 let b1: *u8=lg_readall(p1, s1); let b2: *u8=lg_readall(p2, s2)
58 if s1[0]!=s2[0] { return 0 } if s1[0]<0 { return 0 }
59 var i: i64=0; while i<s1[0] { if b1[i]!=b2[i] { return 0 } i=i+1 } return 1
60}
61
62func lg_addr(out: *u8, port: i64) -> i64 {
63 out[0]=2 as u8; out[1]=0 as u8
64 out[2]=((port>>8)&0xff) as u8; out[3]=(port&0xff) as u8
65 out[4]=127 as u8; out[5]=0 as u8; out[6]=0 as u8; out[7]=1 as u8
66 var i: i64=8; while i<16 { out[i]=0 as u8; i=i+1 } return 0
67}
68func lg_sleep(spins: i64) -> i64 { var s: i64=0; while s<spins { s=s+1 } return 0 }
69
70// one HTTP exchange to 127.0.0.1:port; returns response length in out (headers+body), or <0
71func lg_http(port: i64, req: *u8, reqn: i64, out: *u8, cap: i64) -> i64 {
72 let addr: *u8=sys_mmap(16)
73 lg_addr(addr, port)
74 let cfd: i64=sys_socket(AF_INET, SOCK_STREAM, 0)
75 if cfd<0 { return 0-1 }
76 sys_set_socket_timeout(cfd, 4) // no client socket can block indefinitely
77 if nx_connect_bounded(cfd, addr, 16, NX_CONN_DEFAULT_MS)<0 { sys_close(cfd); return 0-2 }
78 var off: i64=0
79 while off<reqn { let w: i64=sys_write(cfd, (req as i64+off) as *u8, reqn-off); if w<=0 { sys_close(cfd); return 0-3 } off=off+w }
80 var n: i64=0
81 var go: i64=1
82 while go==1 {
83 if n>=cap-1 { go=0 } else { let r: i64=sys_read(cfd, (out as i64+n) as *u8, cap-1-n); if r<=0 { go=0 } else { n=n+r } }
84 }
85 sys_close(cfd)
86 return n
87}
88
89// POST /sync with body[0,blen); response into out. blen=0 => a pure pull.
90func lg_post(port: i64, body: *u8, blen: i64, out: *u8, cap: i64) -> i64 {
91 let req: *u8=sys_mmap(65536)
92 var o: i64=0
93 o=lg_cat(req, o, "POST /sync HTTP/1.1\r\nHost: c\r\nContent-Length: " as *u8)
94 o=lg_uint(req, o, blen)
95 o=lg_cat(req, o, "\r\nConnection: close\r\n\r\n" as *u8)
96 var i: i64=0
97 while i<blen { req[o]=body[i]; o=o+1; i=i+1 }
98 return lg_http(port, req, o, out, cap)
99}
100
101// body offset in an HTTP response (after \r\n\r\n), or the whole thing if no header terminator
102func lg_bodyoff(resp: *u8, rn: i64) -> i64 {
103 var i: i64=0
104 while i+3<rn { if resp[i]==(13 as u8) { if resp[i+1]==(10 as u8) { if resp[i+2]==(13 as u8) { if resp[i+3]==(10 as u8) { return i+4 } } } } i=i+1 }
105 return 0
106}
107
108// a client process: push its `ops` to the shared server (one POST), then one pull to confirm it gets a
109// well-formed response body. BOUNDED (no poll loop => no orchestration hang). The CONVERGENCE assertion is
110// the PARENT's final pull after BOTH clients have pushed; order-independence itself is proven by the kernel
111// gate. Returns 0 if push + pull both succeeded, else a diagnostic code.
112func lg_push(port: i64, ops: *u8, outlog: *u8) -> i64 {
113 lg_sleep(700000) // stagger past the server bind+listen
114 let resp: *u8=sys_mmap(1048576)
115 let pn: i64=lg_post(port, ops, lg_slen(ops), resp, 1048576) // push (length from the string, no truncation)
116 if pn<=0 { return 21 }
117 let rn: i64=lg_post(port, ops, 0, resp, 1048576) // one pull (empty body)
118 if rn<=0 { return 22 }
119 let bo: i64=lg_bodyoff(resp, rn)
120 lg_write(outlog, ((resp as i64)+bo) as *u8, rn-bo)
121 return 0
122}
123
124// fork+exec `nx_coedit apply <inlog> <outtxt>`; wait; return child exit code
125func lg_apply(inlog: *u8, outtxt: *u8) -> i64 {
126 let pid: i64=sys_fork()
127 if pid==0 {
128 let lfd: i64=sys_openat_wr("/tmp/clv_apply.log\x00" as *u8, 0x1a4)
129 if lfd>=0 { sys_dup3(lfd,1,0); sys_dup3(lfd,2,0) }
130 let argv: *i64=sys_mmap(64) as *i64
131 argv[0]="/tmp/nx_coedit.sov.elf\x00" as *u8 as i64
132 argv[1]="apply\x00" as *u8 as i64
133 argv[2]=inlog as i64
134 argv[3]=outtxt as i64
135 argv[4]=0
136 let envp: *i64=sys_mmap(16) as *i64; envp[0]=0
137 sys_execve("/tmp/nx_coedit.sov.elf\x00" as *u8, argv, envp)
138 sys_exit(127)
139 }
140 let st: *i64=sys_mmap(16) as *i64; sys_wait4(pid, st, 0)
141 return (st[0]>>8)&0xff
142}
143func lg_kill(pid: i64) -> i64 { return __syscall(129, pid, 9, 0, 0, 0, 0) } // rv64 kill=129 -- raw x86 62 is an RV64 KEY translated to lseek(8), the kill never happened (debt idx 2277)
144
145func lg_row(name: *u8, pass: i64) -> i64 {
146 lg_p("ROW " as *u8); lg_p(name); if pass==1 { lg_p(" PASS\n" as *u8) } else { lg_p(" FAIL\n" as *u8) } return pass
147}
148
149func main(argc: i64, argv: *i64) -> i64 {
150 lg_p("=== COEDIT LIVE GATE: 2 client PROCESSES concurrently edit ONE doc over real HTTP -> converge (BDAC) ===\n" as *u8)
151 if sys_openat_rd("/tmp/nx_coedit.sov.elf\x00" as *u8)<0 {
152 lg_p(" instrument missing -> rebuilding\n" as *u8)
153 let pid0: i64=sys_fork()
154 if pid0==0 {
155 let argv0: *i64=sys_mmap(64) as *i64
156 argv0[0]="_offc/nx_sov_build_run.elf\x00" as *u8 as i64
157 argv0[1]="nx_coedit\x00" as *u8 as i64
158 argv0[2]="--build-only\x00" as *u8 as i64
159 argv0[3]=0
160 let envp0: *i64=sys_mmap(16) as *i64; envp0[0]=0
161 sys_execve("_offc/nx_sov_build_run.elf\x00" as *u8, argv0, envp0)
162 sys_exit(127)
163 }
164 let s0: *i64=sys_mmap(16) as *i64; sys_wait4(pid0, s0, 0)
165 }
166
167 // UNIQUE per-run port + canon path (collision-proof: repeated runs / a stray prior server never clash)
168 let now: i64 = sys_now_realtime_sec()
169 let PORT: i64 = 9100 + (now % 700)
170 let portstr: *u8 = sys_mmap(16)
171 var pso: i64 = lg_catn(portstr, 0, PORT); portstr[pso]=0 as u8
172 let canon: *u8 = sys_mmap(64)
173 var cxo: i64 = lg_cat(canon, 0, "/tmp/clv_canon_" as *u8); cxo=lg_catn(canon, cxo, now); cxo=lg_cat(canon, cxo, ".log" as *u8); canon[cxo]=0 as u8
174 lg_mkempty(canon)
175
176 // 1) server child: exec the REAL daemon, serve until a clean /shutdown request
177 let spid: i64=sys_fork()
178 if spid==0 {
179 let lfd: i64=sys_openat_wr("/tmp/clv_serve.log\x00" as *u8, 0x1a4)
180 if lfd>=0 { sys_dup3(lfd,1,0); sys_dup3(lfd,2,0) }
181 let argv: *i64=sys_mmap(64) as *i64
182 argv[0]="/tmp/nx_coedit.sov.elf\x00" as *u8 as i64
183 argv[1]="serve\x00" as *u8 as i64
184 argv[2]=canon as i64
185 argv[3]=portstr as i64
186 argv[4]=0
187 let envp: *i64=sys_mmap(16) as *i64; envp[0]=0
188 sys_execve("/tmp/nx_coedit.sov.elf\x00" as *u8, argv, envp)
189 sys_exit(127)
190 }
191
192 // 2) two concurrent client PROCESSES push their edits to the shared server (bounded => no hang)
193 let apid: i64=sys_fork()
194 if apid==0 {
195 let rc: i64=lg_push(PORT, "I 1 1 R A\nI 1 2 1:1 C\n" as *u8, "/tmp/clv_A.log\x00" as *u8)
196 sys_exit(rc); return rc
197 }
198 let bpid: i64=sys_fork()
199 if bpid==0 {
200 let rc: i64=lg_push(PORT, "I 2 1 R B\nI 2 2 2:1 D\n" as *u8, "/tmp/clv_B.log\x00" as *u8)
201 sys_exit(rc); return rc
202 }
203
204 let sa: *i64=sys_mmap(16) as *i64; sys_wait4(apid, sa, 0)
205 let acode: i64=(sa[0]>>8)&0xff
206 let sb: *i64=sys_mmap(16) as *i64; sys_wait4(bpid, sb, 0)
207 let bcode: i64=(sb[0]>>8)&0xff
208
209 // 3) BOTH clients have pushed -> the PARENT pulls the merged doc and materializes it: convergence proof.
210 let fresp: *u8=sys_mmap(1048576)
211 let fn: i64=lg_post(PORT, "\x00" as *u8, 0, fresp, 1048576) // empty-body pull = GET the union
212 if fn>0 { let bo: i64=lg_bodyoff(fresp, fn); lg_write("/tmp/clv_final.log\x00" as *u8, ((fresp as i64)+bo) as *u8, fn-bo) }
213
214 // clean shutdown: stop the accept loop, then reap (no SIGKILL race)
215 let sd: *u8=sys_mmap(4096)
216 lg_http(PORT, "GET /shutdown HTTP/1.1\r\nHost: c\r\nConnection: close\r\n\r\n" as *u8, 52, sd, 4096)
217 let ss: *i64=sys_mmap(16) as *i64; sys_wait4(spid, ss, 0)
218
219 lg_apply("/tmp/clv_final.log\x00" as *u8, "/tmp/clv_final.txt\x00" as *u8)
220
221 var pass: i64=0
222 var r: i64=0
223
224 // both independent client processes pushed successfully over HTTP
225 r=0; if acode==0 { if bcode==0 { r=1 } }
226 pass=pass+lg_row("two-processes-pushed-over-http" as *u8, r)
227
228 // the merged doc, pulled AFTER both concurrent pushes, materializes to the pinned converged text
229 r=lg_fileis("/tmp/clv_final.txt\x00" as *u8, "BDAC" as *u8)
230 pass=pass+lg_row("CONVERGED-live-to-BDAC" as *u8, r)
231
232 // canonical server log UNIONED all 4 ops from the two clients
233 r=0
234 if lg_filehas(canon, "I 1 1 R A" as *u8)==1 { if lg_filehas(canon, "I 1 2 1:1 C" as *u8)==1 { if lg_filehas(canon, "I 2 1 R B" as *u8)==1 { if lg_filehas(canon, "I 2 2 2:1 D" as *u8)==1 { r=1 } } } }
235 pass=pass+lg_row("server-log-unioned-4ops" as *u8, r)
236
237 // idempotent: each op appears EXACTLY once (dedup under the client's push + pull re-sends)
238 r=0
239 let sp: *i64=sys_mmap(16) as *i64
240 let cb: *u8=lg_readall(canon, sp)
241 if sp[0]>0 { if lg_count(cb, sp[0], "I 1 1 R A" as *u8)==1 { if lg_count(cb, sp[0], "I 2 1 R B" as *u8)==1 { r=1 } } }
242 pass=pass+lg_row("server-log-idempotent-no-dup" as *u8, r)
243
244 // the client's own pull also carried the response body (liveness of the sync endpoint)
245 r=0
246 if lg_filehas("/tmp/clv_A.log\x00" as *u8, "I 1 1 R A" as *u8)==1 { if lg_filehas("/tmp/clv_B.log\x00" as *u8, "I 2 1 R B" as *u8)==1 { r=1 } }
247 pass=pass+lg_row("client-pulls-carry-body" as *u8, r)
248
249 let permil: i64=(pass*1000)/5
250 let logfd: i64=sys_openat_append("knowledge/status/vizsla_gate.log\x00" as *u8, 0x1a4)
251 var fdi: i64=0
252 while fdi<2 {
253 var fd: i64=1; if fdi==1 { fd=logfd }
254 if fd>0 {
255 let line: *u8=sys_mmap(256); var o: i64=0
256 o=lg_cat(line,o,"COEDIT-LIVE-GATE epoch=" as *u8); o=lg_catn(line,o,sys_now_realtime_sec())
257 o=lg_cat(line,o," rows=5 pass=" as *u8); o=lg_catn(line,o,pass)
258 o=lg_cat(line,o," permil=" as *u8); o=lg_catn(line,o,permil)
259 if pass==5 { o=lg_cat(line,o," verdict=GREEN\n" as *u8) } else { o=lg_cat(line,o," verdict=RED\n" as *u8) }
260 sys_write(fd, line, o)
261 }
262 fdi=fdi+1
263 }
264 if logfd>0 { sys_close(logfd) }
265 if pass==5 { return 0 }
266 return 1
267}