code wiki / _hdl_build / nx_coedit_live_gate.nx
`nx_coedit_live_gate.nx` source

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

`nx_coedit_live_gate.nx` source
