nx_msg_sync_gate.nx source
↩ module page · 264 lines · 12769 B
1// nx_msg_sync_gate.nx -- REFEREE for C2 (nx_msg_sync, contract msy_devsync). END-TO-END: builds a
2// server room and device replicas in /tmp/msgate (nothing production is touched), forks BOTH promoted
3// elfs -- nx_chat_store to author fixtures and READ BACK, nx_msg_sync as the subject -- and proves the
4// pre-declared done-rule: "the same history converges on a second device; a partition-printed
5// reconcile measures zero divergence."
6// THE ANTI-VACUITY DESIGN: convergence is NEVER taken from the syncer's own receipt. T3/T5 compare
7// nx_chat_store fetch output on the server root vs the device root BYTE FOR BYTE -- the consumer is
8// the referee, so a syncer that lied about pulling would fail the tooth it cannot see.
9// NEGATIVE CONTROLS: split-brain divergence refuses (seq named, nothing written) · device-ahead
10// refuses · absent source never conjures · verify itself goes RED on the diverged replica.
11// Fixture rooms are epoch-suffixed; every verb is its own fork, so persistence across process death
12// is structural. license_tier: ORIGINAL expect_exit: 0
13import "nx_syscalls.nx"
14import "nx_tool_run.nx"
15import "nx_gate_verdict.nx"
16const MSG_CAP: i64 = 65536
17func msg_len(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
18func msg_write(path: *u8, s: *u8) -> i64 {
19 let fd: i64 = sys_openat_wr(path, 420)
20 if fd < 0 { return 0 - 1 }
21 let n: i64 = msg_len(s)
22 var w2: i64 = 0
23 while w2 < n { let r: i64 = sys_write(fd, (s as i64 + w2) as *u8, n - w2); if r <= 0 { break } w2 = w2 + r }
24 sys_close(fd)
25 return 0
26}
27func msg_count(buf: *u8, n: i64, needle: *u8) -> i64 {
28 let m: i64 = msg_len(needle)
29 if m <= 0 { return 0 }
30 var c: i64 = 0
31 var i: i64 = 0
32 while i + m <= n {
33 var k: i64 = 0
34 var hit: i64 = 1
35 while k < m { if buf[i+k] != needle[k] { hit = 0; k = m } else { k = k + 1 } }
36 if hit == 1 { c = c + 1; i = i + m } else { i = i + 1 }
37 }
38 return c
39}
40func main(argc: i64, argv: *i64) -> i64 {
41 let ctr: *i64 = gv_ctr()
42 gv_head("nx_msg_sync -- the same history converges on a second device, proven through the consumer; split-brain refuses loudly" as *u8)
43 let CHAT: *u8 = "/volume1/homes/elderwesto/nishihost/nx_chat_store.elf" as *u8
44 let SYNC: *u8 = "/volume1/homes/elderwesto/nishihost/nx_msg_sync.elf" as *u8
45 let CSRV: *u8 = "/tmp/msgate/conf_srv" as *u8
46 let CDEV: *u8 = "/tmp/msgate/conf_dev" as *u8
47 let CDEV2: *u8 = "/tmp/msgate/conf_dev2" as *u8
48 sys_mkdir("/tmp/msgate" as *u8, 493)
49 sys_mkdir("/tmp/msgate/srv" as *u8, 493)
50 sys_mkdir("/tmp/msgate/dev" as *u8, 493)
51 sys_mkdir("/tmp/msgate/dev2" as *u8, 493)
52 msg_write(CSRV, "store_root|/tmp/msgate/srv/\nroom_bytes_cliff|100000\nroom_msgs_cliff|1000\nbody_bytes_max|500\n" as *u8)
53 msg_write(CDEV, "store_root|/tmp/msgate/dev/\nroom_bytes_cliff|100000\nroom_msgs_cliff|1000\nbody_bytes_max|500\n" as *u8)
54 msg_write(CDEV2, "store_root|/tmp/msgate/dev2/\nroom_bytes_cliff|100000\nroom_msgs_cliff|1000\nbody_bytes_max|500\n" as *u8)
55 let out: *u8 = sys_mmap(MSG_CAP)
56 let outb: *u8 = sys_mmap(MSG_CAP)
57 let ol: *i64 = sys_mmap(16) as *i64
58 let av: *i64 = sys_mmap(256) as *i64
59 let ms: i64 = sys_now_realtime_ms()
60 let rm: *u8 = sys_mmap(64)
61 var o: i64 = gv_cat(rm, 0, "r" as *u8)
62 o = gv_catn(rm, o, ms)
63 rm[o] = 0 as u8
64 var t0: i64 = 0
65 let cchk: *u8 = sys_mmap(4096)
66 let cfd: i64 = sys_openat_rd(CSRV)
67 if cfd >= 0 { if sys_read(cfd, cchk, 4096) > 0 { t0 = 1 } sys_close(cfd) }
68 gv_check("T0 fixture roots and confs on disk, all isolated under /tmp/msgate" as *u8, t0, ctr)
69 // ---- server room with 3 messages ----------------------------------------------------------
70 av[0] = CHAT as i64
71 av[1] = "open" as i64
72 av[2] = rm as i64
73 av[3] = "alice" as i64
74 av[4] = CSRV as i64
75 av[5] = 0
76 let r1: i64 = tr_run_capture(CHAT, av, out, MSG_CAP, ol)
77 av[1] = "append" as i64
78 av[3] = "alice" as i64
79 av[4] = "1" as i64
80 av[5] = "one" as i64
81 av[6] = CSRV as i64
82 av[7] = 0
83 let r2: i64 = tr_run_capture(CHAT, av, out, MSG_CAP, ol)
84 av[5] = "two" as i64
85 let r3: i64 = tr_run_capture(CHAT, av, out, MSG_CAP, ol)
86 av[5] = "three" as i64
87 let r4: i64 = tr_run_capture(CHAT, av, out, MSG_CAP, ol)
88 var t1: i64 = 0
89 if r1 == 0 { if r2 == 0 { if r3 == 0 { if r4 == 0 { t1 = 1 } } } }
90 gv_check("T1 server fixture: room opened and 3 messages appended through the C1 organ" as *u8, t1, ctr)
91 // ---- T2 BITE: first sync pulls everything into a fresh device -----------------------------
92 av[0] = SYNC as i64
93 av[1] = "sync" as i64
94 av[2] = rm as i64
95 av[3] = "/tmp/msgate/dev/" as i64
96 av[4] = CSRV as i64
97 av[5] = 0
98 let s1: i64 = tr_run_capture(SYNC, av, out, MSG_CAP, ol)
99 var t2: i64 = 0
100 if s1 == 0 { if msg_count(out, ol[0], "device_had=0 pulled=4" as *u8) == 1 { if msg_count(out, ol[0], "diverged=0" as *u8) == 1 { t2 = 1 } } }
101 gv_check("T2 BITE: a fresh device pulls the full history in one sync (had 0, pulled 4, diverged 0)" as *u8, t2, ctr)
102 // ---- T3 CONVERGENCE THROUGH THE CONSUMER: fetch both roots, byte-compare ------------------
103 av[0] = CHAT as i64
104 av[1] = "fetch" as i64
105 av[2] = rm as i64
106 av[3] = "0" as i64
107 av[4] = "100" as i64
108 av[5] = CSRV as i64
109 av[6] = 0
110 let f1: i64 = tr_run_capture(CHAT, av, out, MSG_CAP, ol)
111 let n1: i64 = ol[0]
112 av[5] = CDEV as i64
113 let f2: i64 = tr_run_capture(CHAT, av, outb, MSG_CAP, ol)
114 var t3: i64 = 0
115 if f1 == 0 { if f2 == 0 { if ol[0] == n1 {
116 var same: i64 = 1
117 var q: i64 = 0
118 while q < n1 { if out[q] != outb[q] { same = 0; q = n1 } else { q = q + 1 } }
119 if same == 1 { if n1 > 0 { t3 = 1 } }
120 } } }
121 gv_check("T3 ANTI-VACUITY: nx_chat_store itself reads server and device and the outputs are BYTE-IDENTICAL -- the consumer certifies convergence, not the syncer" as *u8, t3, ctr)
122 // ---- T4 verify verb agrees ----------------------------------------------------------------
123 av[0] = SYNC as i64
124 av[1] = "verify" as i64
125 av[3] = "/tmp/msgate/dev/" as i64
126 av[4] = CSRV as i64
127 av[5] = 0
128 let v1: i64 = tr_run_capture(SYNC, av, out, MSG_CAP, ol)
129 var t4: i64 = 0
130 if v1 == 0 { if msg_count(out, ol[0], "CONVERGED" as *u8) == 1 { if msg_count(out, ol[0], "rows=4" as *u8) == 1 { t4 = 1 } } }
131 gv_check("T4 the read-only verify reconciles the full population and reports CONVERGED rows=4" as *u8, t4, ctr)
132 // ---- T5 incremental sync ------------------------------------------------------------------
133 av[0] = CHAT as i64
134 av[1] = "append" as i64
135 av[3] = "alice" as i64
136 av[4] = "1" as i64
137 av[5] = "four" as i64
138 av[6] = CSRV as i64
139 av[7] = 0
140 tr_run_capture(CHAT, av, out, MSG_CAP, ol)
141 av[5] = "five" as i64
142 tr_run_capture(CHAT, av, out, MSG_CAP, ol)
143 av[0] = SYNC as i64
144 av[1] = "sync" as i64
145 av[3] = "/tmp/msgate/dev/" as i64
146 av[4] = CSRV as i64
147 av[5] = 0
148 let s2: i64 = tr_run_capture(SYNC, av, out, MSG_CAP, ol)
149 var t5: i64 = 0
150 if s2 == 0 { if msg_count(out, ol[0], "device_had=4 pulled=2" as *u8) == 1 { if msg_count(out, ol[0], "verified_after_write=6" as *u8) == 1 { t5 = 1 } } }
151 gv_check("T5 INCREMENTAL: only the missing rows move (had 4, pulled 2) and the post-write re-read confirms 6" as *u8, t5, ctr)
152 // ---- T6 tombstones travel -----------------------------------------------------------------
153 av[0] = CHAT as i64
154 av[1] = "del" as i64
155 av[3] = "2" as i64
156 av[4] = "alice" as i64
157 av[5] = CSRV as i64
158 av[6] = 0
159 tr_run_capture(CHAT, av, out, MSG_CAP, ol)
160 av[0] = SYNC as i64
161 av[1] = "sync" as i64
162 av[3] = "/tmp/msgate/dev/" as i64
163 av[4] = CSRV as i64
164 av[5] = 0
165 let s3: i64 = tr_run_capture(SYNC, av, out, MSG_CAP, ol)
166 var t6a: i64 = 0
167 if s3 == 0 { if msg_count(out, ol[0], "tombstones_copied=1" as *u8) == 1 { t6a = 1 } }
168 gv_check("T6a a server-side soft-delete syncs as a tombstone copy" as *u8, t6a, ctr)
169 av[0] = CHAT as i64
170 av[1] = "fetch" as i64
171 av[3] = "0" as i64
172 av[4] = "100" as i64
173 av[5] = CDEV as i64
174 av[6] = 0
175 let f3: i64 = tr_run_capture(CHAT, av, out, MSG_CAP, ol)
176 var t6: i64 = 0
177 if f3 == 0 { if msg_count(out, ol[0], "m|2|" as *u8) == 0 { if msg_count(out, ol[0], "deleted_hidden=1" as *u8) == 1 { t6 = 1 } } }
178 gv_check("T6 the device now hides the deleted row exactly as the server does (rule-13 semantics survive sync)" as *u8, t6, ctr)
179 // ---- T7 idempotent ------------------------------------------------------------------------
180 av[0] = SYNC as i64
181 av[1] = "sync" as i64
182 av[3] = "/tmp/msgate/dev/" as i64
183 av[4] = CSRV as i64
184 av[5] = 0
185 let s4: i64 = tr_run_capture(SYNC, av, out, MSG_CAP, ol)
186 var t7: i64 = 0
187 if s4 == 0 { if msg_count(out, ol[0], "pulled=0 tombstones_copied=0" as *u8) == 1 { t7 = 1 } }
188 gv_check("T7 neg-control-idempotent: a second sync writes NOTHING (pulled 0, copied 0) -- safe to run twice" as *u8, t7, ctr)
189 // ---- T8 determinism -----------------------------------------------------------------------
190 av[1] = "verify" as i64
191 tr_run_capture(SYNC, av, out, MSG_CAP, ol)
192 let vn1: i64 = ol[0]
193 tr_run_capture(SYNC, av, outb, MSG_CAP, ol)
194 var t8: i64 = 0
195 if ol[0] == vn1 {
196 var same2: i64 = 1
197 var q2: i64 = 0
198 while q2 < vn1 { if out[q2] != outb[q2] { same2 = 0; q2 = vn1 } else { q2 = q2 + 1 } }
199 if same2 == 1 { t8 = 1 }
200 }
201 gv_check("T8 DETERMINISM: verify twice is byte-identical" as *u8, t8, ctr)
202 // ---- T9 neg-control-diverged: an independently-opened replica refuses to merge ------------
203 av[0] = CHAT as i64
204 av[1] = "open" as i64
205 av[2] = rm as i64
206 av[3] = "mallory" as i64
207 av[4] = CDEV2 as i64
208 av[5] = 0
209 tr_run_capture(CHAT, av, out, MSG_CAP, ol)
210 av[0] = SYNC as i64
211 av[1] = "sync" as i64
212 av[3] = "/tmp/msgate/dev2/" as i64
213 av[4] = CSRV as i64
214 av[5] = 0
215 let s5: i64 = tr_run_capture(SYNC, av, out, MSG_CAP, ol)
216 var t9: i64 = 0
217 if s5 != 0 { if msg_count(out, ol[0], "diverged seq=1" as *u8) == 1 { t9 = 1 } }
218 gv_check("T9 neg-control-diverged: a replica that disagrees on row 1 REFUSES by seq -- split-brain never merges" as *u8, t9, ctr)
219 av[0] = CHAT as i64
220 av[1] = "fetch" as i64
221 av[3] = "0" as i64
222 av[4] = "100" as i64
223 av[5] = CDEV2 as i64
224 av[6] = 0
225 let f4: i64 = tr_run_capture(CHAT, av, out, MSG_CAP, ol)
226 var t10: i64 = 0
227 if f4 == 0 { if msg_count(out, ol[0], "total=1" as *u8) == 1 { t10 = 1 } }
228 gv_check("T10 the refused sync wrote NOTHING: the diverged replica still holds exactly its own 1 row" as *u8, t10, ctr)
229 // ---- T11 neg-control-device-ahead ---------------------------------------------------------
230 av[1] = "append" as i64
231 av[3] = "mallory" as i64
232 av[4] = "1" as i64
233 av[5] = "junk" as i64
234 av[6] = CDEV2 as i64
235 av[7] = 0
236 var ji: i64 = 0
237 while ji < 6 { tr_run_capture(CHAT, av, out, MSG_CAP, ol) ji = ji + 1 }
238 av[0] = SYNC as i64
239 av[1] = "sync" as i64
240 av[3] = "/tmp/msgate/dev2/" as i64
241 av[4] = CSRV as i64
242 av[5] = 0
243 let s6: i64 = tr_run_capture(SYNC, av, out, MSG_CAP, ol)
244 var t11: i64 = 0
245 if s6 != 0 { if msg_count(out, ol[0], "device-ahead" as *u8) == 1 { t11 = 1 } }
246 gv_check("T11 neg-control-device-ahead: a replica with MORE rows than the server refuses rather than truncating either side" as *u8, t11, ctr)
247 // ---- T12 verify goes RED on the diverged replica (verify has its own negative control) -----
248 av[1] = "verify" as i64
249 let v2: i64 = tr_run_capture(SYNC, av, out, MSG_CAP, ol)
250 var t12: i64 = 0
251 if v2 != 0 { if msg_count(out, ol[0], "DIVERGED" as *u8) == 1 { t12 = 1 } }
252 gv_check("T12 neg-control-verify: the convergence referee itself goes RED on the diverged replica, so T4 is not a tautology" as *u8, t12, ctr)
253 // ---- T13 neg-control-absent-source --------------------------------------------------------
254 av[1] = "sync" as i64
255 av[2] = "ghostroom" as i64
256 av[3] = "/tmp/msgate/dev/" as i64
257 av[4] = CSRV as i64
258 av[5] = 0
259 let s7: i64 = tr_run_capture(SYNC, av, out, MSG_CAP, ol)
260 var t13: i64 = 0
261 if s7 != 0 { if msg_count(out, ol[0], "room-absent" as *u8) == 1 { t13 = 1 } }
262 gv_check("T13 neg-control-absent-source: syncing a room the server does not have REFUSES -- a sync never conjures" as *u8, t13, ctr)
263 return gv_verdict("MSG-SYNC-GATE" as *u8, ctr, "convergence proven through the consumer byte-for-byte; split-brain, device-ahead and absent-source all refuse loudly with nothing written" as *u8)
264}