nx_msg_sync.nx source
↩ module page · 400 lines · 16129 B
1// nx_msg_sync.nx -- C2 of the comms lane: MULTI-DEVICE SYNC for nx_chat_store rooms. Contract symbol
2// msy_devsync == the /compare/comms C2 watch contract.
3// DONE-RULE (pre-declared in comms.plan before this organ existed): "The same history converges on a
4// second device; a partition-printed reconcile measures zero divergence." REFEREE: nx_msg_sync_gate.
5//
6// MODEL: the SERVER plane (store_root from the shared chatstore.conf) is authoritative; a DEVICE holds
7// a replica plane under its own root with the SAME layout, readable by nx_chat_store itself via a conf
8// whose store_root names the device root -- one format, one ruler, no translation layer. The gate
9// exploits exactly that: convergence is verified through the CONSUMER (nx_chat_store fetch on both
10// roots, byte-compared), never self-certified by this syncer.
11//
12// SYNC IS A PULL (Telegram-cloud-sync class): devices SEND by appending at the server (C1); devices
13// RECEIVE by pulling. Overlap rows are BYTE-COMPARED before anything is written: a replica that
14// disagrees with the server on a row it already holds is SPLIT-BRAIN and the sync REFUSES LOUDLY --
15// it never merges, never overwrites, wrong in the direction of doing nothing. A replica holding MORE
16// rows than the server likewise REFUSES (device-ahead). Missing rows land in ONE commit --
17// O(missing), the sts_append_fast shape batched -- never a whole-plane rewrite (the seq724
18// amplification law), and reads go through ss_open_cached + ss_hget (the seq356 law).
19//
20// SCOPE, stated so the next reader trusts nothing by accident: messages and del:<seq> tombstones
21// sync; cur:<member> cursors are DEVICE-LOCAL BY DESIGN (each device owns its read state -- syncing
22// them would make two phones share one unread badge); meta:bytes copies from the server. The full
23// population is always reconciled: NO batch cap, NO sampling -- sync has no tunable numbers at all,
24// and the only conf it reads is C1's store_root.
25// license_tier: ORIGINAL No hw writes (Rule 26).
26import "nx_syscalls.nx"
27import "nx_itoa_lib.nx"
28import "nx_store_seed_lib.nx"
29import "nx_seg_store.nx"
30
31const MSY_EXIT_OK: i64 = 0
32const MSY_EXIT_USAGE: i64 = 2
33const MSY_EXIT_REFUSED: i64 = 3
34const MSY_EXIT_RETRY: i64 = 4
35const MSY_EXIT_DIVERGED: i64 = 5
36// structural, not policy:
37const MSY_NAME_MAX: i64 = 64
38const MSY_PFX_CAP: i64 = 256
39const MSY_CONF_CAP: i64 = 192
40const MSY_WSLACK: i64 = 65536 // writer slack, the seed lib's own margin
41const MSY_NL: i64 = 10
42const MSY_PIPE: i64 = 124
43
44func msy_w(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
45func msy_n(v: i64) -> i64 { nxi_out(v); return 0 }
46func msy_len(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
47func msy_eq(a: *u8, b: *u8) -> i64 {
48 var i: i64 = 0
49 while a[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 }
50 if b[i] != (0 as u8) { return 0 }
51 return 1
52}
53func msy_tok_ok(s: *u8) -> i64 {
54 var i: i64 = 0
55 while s[i] != (0 as u8) {
56 let c: i64 = s[i] as i64
57 var ok: i64 = 0
58 if c >= 48 { if c <= 57 { ok = 1 } }
59 if c >= 65 { if c <= 90 { ok = 1 } }
60 if c >= 97 { if c <= 122 { ok = 1 } }
61 if c == 95 { ok = 1 }
62 if ok == 0 { return 0 }
63 i = i + 1
64 }
65 if i < 1 { return 0 }
66 if i > MSY_NAME_MAX { return 0 }
67 return 1
68}
69func msy_confs(confpath: *u8, key: *u8, out: *u8, cap: i64) -> i64 {
70 let ol: *i64 = sts_mm(16) as *i64
71 let b: *u8 = sys_read_file(confpath, ol)
72 let n: i64 = ol[0]
73 if b as i64 == 0 { return 0 - 1 }
74 if n <= 0 { return 0 - 1 }
75 let kl: i64 = msy_len(key)
76 var i: i64 = 0
77 while i < n {
78 var j: i64 = 0
79 var hit: i64 = 1
80 while j < kl {
81 if i + j >= n { hit = 0; break }
82 if b[i+j] != key[j] { hit = 0; break }
83 j = j + 1
84 }
85 if hit == 1 {
86 if i + kl >= n { hit = 0 } else { if b[i+kl] != (MSY_PIPE as u8) { hit = 0 } }
87 }
88 if hit == 1 {
89 var v: i64 = i + kl + 1
90 var o: i64 = 0
91 while v < n {
92 if b[v] == (MSY_NL as u8) { break }
93 if o < cap - 1 { out[o] = b[v]; o = o + 1 }
94 v = v + 1
95 }
96 out[o] = 0 as u8
97 return o
98 }
99 while i < n { if b[i] == (MSY_NL as u8) { break } i = i + 1 }
100 i = i + 1
101 }
102 return 0 - 1
103}
104// plane prefix for a room under an explicit root: <root>chat_<room>-
105func msy_prefix_at(root: *u8, room: *u8, out: *u8) -> i64 {
106 if msy_tok_ok(room) == 0 { return 0 - 1 }
107 var o: i64 = ss_cat(out, 0, root)
108 o = ss_cat(out, o, "chat_" as *u8)
109 o = ss_cat(out, o, room)
110 o = ss_cat(out, o, "-" as *u8)
111 out[o] = 0 as u8
112 return o
113}
114func msy_delkey(seq: i64, out: *u8) -> i64 {
115 var o: i64 = ss_cat(out, 0, "del:" as *u8)
116 o = ss_catn(out, o, seq)
117 out[o] = 0 as u8
118 return o
119}
120// byte-compare the overlap rows [0, upto); returns 0 if identical, else the 1-based seq of the FIRST
121// divergent row, or -(idx+1000000) on a row the handle cannot produce (corrupt count).
122func msy_cmp_overlap(sh: *i64, dh: *i64, upto: i64) -> i64 {
123 let key: *u8 = sts_mm(64)
124 let sp: *i64 = sts_mm(16) as *i64
125 let sl: *i64 = sts_mm(16) as *i64
126 let dp: *i64 = sts_mm(16) as *i64
127 let dl: *i64 = sts_mm(16) as *i64
128 var i: i64 = 0
129 while i < upto {
130 sts_rowkey(i, key)
131 if ss_hget(sh, key, sp, sl) != 1 { return 0 - (i + 1000000) }
132 if ss_hget(dh, key, dp, dl) != 1 { return 0 - (i + 1000000) }
133 if sl[0] != dl[0] { return i + 1 }
134 let a: *u8 = sp[0] as *u8
135 let b: *u8 = dp[0] as *u8
136 let m: i64 = sl[0]
137 var k: i64 = 0
138 var same: i64 = 1
139 while k < m { if a[k] != b[k] { same = 0; break } k = k + 1 }
140 if same == 0 { return i + 1 }
141 i = i + 1
142 }
143 return 0
144}
145
146// THE C2 CONTRACT SYMBOL. Pull-reconcile one room from the server plane into a device root.
147// Returns MSY_EXIT_* (announced). Wrong in the direction of doing nothing: any disagreement on rows
148// the device already holds refuses BEFORE anything is written.
149func msy_devsync(room: *u8, devroot: *u8, confpath: *u8) -> i64 {
150 let sroot: *u8 = sts_mm(MSY_CONF_CAP)
151 if msy_confs(confpath, "store_root" as *u8, sroot, MSY_CONF_CAP) <= 0 {
152 msy_w("MSG-SYNC-REFUSED conf-missing row=store_root conf=" as *u8)
153 msy_w(confpath)
154 msy_w("\n" as *u8)
155 return MSY_EXIT_REFUSED
156 }
157 let spfx: *u8 = sts_mm(MSY_PFX_CAP)
158 if msy_prefix_at(sroot, room, spfx) < 0 { msy_w("MSG-SYNC-REFUSED bad-token room\n" as *u8) return MSY_EXIT_REFUSED }
159 let dpfx: *u8 = sts_mm(MSY_PFX_CAP)
160 msy_prefix_at(devroot, room, dpfx)
161 let pq: *i64 = sts_mm(16) as *i64
162 let lq: *i64 = sts_mm(16) as *i64
163 if ss_get(spfx, "q:n" as *u8, pq, lq) != 1 {
164 msy_w("MSG-SYNC-REFUSED room-absent room=" as *u8)
165 msy_w(room)
166 msy_w(" -- the server room does not exist; a sync never conjures\n" as *u8)
167 return MSY_EXIT_REFUSED
168 }
169 let sn: i64 = sts_atoi(pq[0] as *u8, lq[0])
170 let lk: i64 = sts_lock(dpfx)
171 if lk < 0 { msy_w("MSG-SYNC-RETRY cannot-lock device plane\n" as *u8) return MSY_EXIT_RETRY }
172 var dn: i64 = 0
173 if ss_get(dpfx, "q:n" as *u8, pq, lq) == 1 { dn = sts_atoi(pq[0] as *u8, lq[0]) }
174 if dn > sn {
175 sts_unlock(lk)
176 msy_w("MSG-SYNC-REFUSED device-ahead device_total=" as *u8)
177 msy_n(dn)
178 msy_w(" server_total=" as *u8)
179 msy_n(sn)
180 msy_w(" -- the replica holds rows the server does not (split-brain); nothing written\n" as *u8)
181 return MSY_EXIT_REFUSED
182 }
183 let sh: *i64 = ss_open_cached(spfx)
184 if sh as i64 == 0 { sts_unlock(lk) msy_w("MSG-SYNC-RETRY server-open-failed\n" as *u8) return MSY_EXIT_RETRY }
185 var dh: *i64 = 0 as *i64
186 if dn > 0 {
187 dh = ss_open_cached(dpfx)
188 if dh as i64 == 0 { sts_unlock(lk) msy_w("MSG-SYNC-RETRY device-open-failed\n" as *u8) return MSY_EXIT_RETRY }
189 let dv: i64 = msy_cmp_overlap(sh, dh, dn)
190 if dv > 0 {
191 sts_unlock(lk)
192 msy_w("MSG-SYNC-REFUSED diverged seq=" as *u8)
193 msy_n(dv)
194 msy_w(" -- replica disagrees with the server on a row it already holds (split-brain); REFUSING to merge or overwrite, nothing written\n" as *u8)
195 return MSY_EXIT_DIVERGED
196 }
197 if dv < 0 {
198 sts_unlock(lk)
199 msy_w("MSG-SYNC-REFUSED corrupt-count -- a declared row could not be produced\n" as *u8)
200 return MSY_EXIT_DIVERGED
201 }
202 }
203 // measure what is missing: rows dn..sn-1 (sizes summed for the writer cap -- data-driven, no ceiling)
204 let key: *u8 = sts_mm(64)
205 var need: i64 = 0
206 var i: i64 = dn
207 while i < sn {
208 sts_rowkey(i, key)
209 if ss_hget(sh, key, pq, lq) != 1 {
210 sts_unlock(lk)
211 msy_w("MSG-SYNC-REFUSED corrupt-count server row missing idx=" as *u8)
212 msy_n(i)
213 msy_w("\n" as *u8)
214 return MSY_EXIT_DIVERGED
215 }
216 need = need + lq[0] + 96
217 i = i + 1
218 }
219 // tombstones to copy (present on server, absent on device)
220 let dk: *u8 = sts_mm(64)
221 let dp2: *i64 = sts_mm(16) as *i64
222 let dl2: *i64 = sts_mm(16) as *i64
223 var tcopy: i64 = 0
224 var s2: i64 = 1
225 while s2 <= sn {
226 msy_delkey(s2, dk)
227 if ss_hget(sh, dk, pq, lq) == 1 {
228 var have: i64 = 0
229 if dh as i64 != 0 { if ss_hget(dh, dk, dp2, dl2) == 1 { have = 1 } }
230 if have == 0 { tcopy = tcopy + 1; need = need + 96 }
231 }
232 s2 = s2 + 1
233 }
234 let pulled: i64 = sn - dn
235 if pulled == 0 { if tcopy == 0 {
236 sts_unlock(lk)
237 msy_w("MSG-SYNC-OK room=" as *u8)
238 msy_w(room)
239 msy_w(" server_total=" as *u8)
240 msy_n(sn)
241 msy_w(" device_had=" as *u8)
242 msy_n(dn)
243 msy_w(" pulled=0 tombstones_copied=0 overlap_verified=" as *u8)
244 msy_n(dn)
245 msy_w(" diverged=0 -- already converged, nothing to write (idempotent)\n" as *u8)
246 return MSY_EXIT_OK
247 } }
248 let w: *i64 = ss_begin_cap(need + MSY_WSLACK)
249 i = dn
250 while i < sn {
251 sts_rowkey(i, key)
252 if ss_hget(sh, key, pq, lq) != 1 { sts_unlock(lk) msy_w("MSG-SYNC-RED reread-failed\n" as *u8) return MSY_EXIT_DIVERGED }
253 if ss_add(w, 1, key, pq[0] as *u8, lq[0]) < 0 { sts_unlock(lk) msy_w("MSG-SYNC-RED writer-add-failed\n" as *u8) return MSY_EXIT_DIVERGED }
254 i = i + 1
255 }
256 let cb: *u8 = sts_mm(32)
257 let cl: i64 = ss_catn(cb, 0, sn)
258 if ss_add(w, 1, "q:n" as *u8, cb, cl) < 0 { sts_unlock(lk) msy_w("MSG-SYNC-RED writer-add-count-failed\n" as *u8) return MSY_EXIT_DIVERGED }
259 if ss_get(spfx, "meta:bytes" as *u8, pq, lq) == 1 {
260 if ss_add(w, 1, "meta:bytes" as *u8, pq[0] as *u8, lq[0]) < 0 { sts_unlock(lk) msy_w("MSG-SYNC-RED writer-add-meta-failed\n" as *u8) return MSY_EXIT_DIVERGED }
261 }
262 s2 = 1
263 while s2 <= sn {
264 msy_delkey(s2, dk)
265 if ss_hget(sh, dk, pq, lq) == 1 {
266 var have2: i64 = 0
267 if dh as i64 != 0 { if ss_hget(dh, dk, dp2, dl2) == 1 { have2 = 1 } }
268 if have2 == 0 {
269 if ss_add(w, 1, dk, "1" as *u8, 1) < 0 { sts_unlock(lk) msy_w("MSG-SYNC-RED writer-add-del-failed\n" as *u8) return MSY_EXIT_DIVERGED }
270 }
271 }
272 s2 = s2 + 1
273 }
274 if ss_commit_cas(dpfx, w, ss_next_segid(dpfx), SS_CAS_ANY) != 0 {
275 sts_unlock(lk)
276 msy_w("MSG-SYNC-RED commit-failed -- device plane unchanged\n" as *u8)
277 return MSY_EXIT_DIVERGED
278 }
279 // the write-landed check: never trust the receipt
280 var after: i64 = 0 - 1
281 if ss_get(dpfx, "q:n" as *u8, pq, lq) == 1 { after = sts_atoi(pq[0] as *u8, lq[0]) }
282 sts_unlock(lk)
283 if after != sn {
284 msy_w("MSG-SYNC-RED post-commit count wrong expected=" as *u8)
285 msy_n(sn)
286 msy_w(" got=" as *u8)
287 msy_n(after)
288 msy_w("\n" as *u8)
289 return MSY_EXIT_DIVERGED
290 }
291 msy_w("MSG-SYNC-OK room=" as *u8)
292 msy_w(room)
293 msy_w(" server_total=" as *u8)
294 msy_n(sn)
295 msy_w(" device_had=" as *u8)
296 msy_n(dn)
297 msy_w(" pulled=" as *u8)
298 msy_n(pulled)
299 msy_w(" tombstones_copied=" as *u8)
300 msy_n(tcopy)
301 msy_w(" overlap_verified=" as *u8)
302 msy_n(dn)
303 msy_w(" diverged=0 verified_after_write=" as *u8)
304 msy_n(after)
305 msy_w(" device_root=" as *u8)
306 msy_w(devroot)
307 msy_w("\n" as *u8)
308 return MSY_EXIT_OK
309}
310
311// read-only convergence referee: full population, both directions, rows AND tombstones.
312func msy_verify(room: *u8, devroot: *u8, confpath: *u8) -> i64 {
313 let sroot: *u8 = sts_mm(MSY_CONF_CAP)
314 if msy_confs(confpath, "store_root" as *u8, sroot, MSY_CONF_CAP) <= 0 { msy_w("MSG-VERIFY REFUSED conf-missing row=store_root\n" as *u8) return MSY_EXIT_REFUSED }
315 let spfx: *u8 = sts_mm(MSY_PFX_CAP)
316 if msy_prefix_at(sroot, room, spfx) < 0 { msy_w("MSG-VERIFY REFUSED bad-token room\n" as *u8) return MSY_EXIT_REFUSED }
317 let dpfx: *u8 = sts_mm(MSY_PFX_CAP)
318 msy_prefix_at(devroot, room, dpfx)
319 let pq: *i64 = sts_mm(16) as *i64
320 let lq: *i64 = sts_mm(16) as *i64
321 if ss_get(spfx, "q:n" as *u8, pq, lq) != 1 { msy_w("MSG-VERIFY REFUSED room-absent on server\n" as *u8) return MSY_EXIT_REFUSED }
322 let sn: i64 = sts_atoi(pq[0] as *u8, lq[0])
323 if ss_get(dpfx, "q:n" as *u8, pq, lq) != 1 { msy_w("MSG-VERIFY DIVERGED device plane absent -- absent is not converged\n" as *u8) return MSY_EXIT_DIVERGED }
324 let dn: i64 = sts_atoi(pq[0] as *u8, lq[0])
325 if dn != sn {
326 msy_w("MSG-VERIFY DIVERGED counts server=" as *u8)
327 msy_n(sn)
328 msy_w(" device=" as *u8)
329 msy_n(dn)
330 msy_w("\n" as *u8)
331 return MSY_EXIT_DIVERGED
332 }
333 let sh: *i64 = ss_open_cached(spfx)
334 let dh: *i64 = ss_open_cached(dpfx)
335 if sh as i64 == 0 { msy_w("MSG-VERIFY RETRY server-open-failed\n" as *u8) return MSY_EXIT_RETRY }
336 if dh as i64 == 0 { msy_w("MSG-VERIFY RETRY device-open-failed\n" as *u8) return MSY_EXIT_RETRY }
337 let dv: i64 = msy_cmp_overlap(sh, dh, sn)
338 if dv != 0 {
339 msy_w("MSG-VERIFY DIVERGED first-mismatch seq=" as *u8)
340 msy_n(dv)
341 msy_w("\n" as *u8)
342 return MSY_EXIT_DIVERGED
343 }
344 let dk: *u8 = sts_mm(64)
345 let dp2: *i64 = sts_mm(16) as *i64
346 let dl2: *i64 = sts_mm(16) as *i64
347 var tb: i64 = 0
348 var s2: i64 = 1
349 while s2 <= sn {
350 msy_delkey(s2, dk)
351 var sv: i64 = 0
352 var dvv: i64 = 0
353 if ss_hget(sh, dk, pq, lq) == 1 { sv = 1 }
354 if ss_hget(dh, dk, dp2, dl2) == 1 { dvv = 1 }
355 if sv != dvv {
356 msy_w("MSG-VERIFY DIVERGED tombstone-set seq=" as *u8)
357 msy_n(s2)
358 msy_w("\n" as *u8)
359 return MSY_EXIT_DIVERGED
360 }
361 if sv == 1 { tb = tb + 1 }
362 s2 = s2 + 1
363 }
364 msy_w("MSG-VERIFY CONVERGED room=" as *u8)
365 msy_w(room)
366 msy_w(" rows=" as *u8)
367 msy_n(sn)
368 msy_w(" tombstones=" as *u8)
369 msy_n(tb)
370 msy_w(" byte_identical=" as *u8)
371 msy_n(sn)
372 msy_w(" diverged=0\n" as *u8)
373 return MSY_EXIT_OK
374}
375
376func msy_usage() -> i64 {
377 msy_w("usage: nx_msg_sync sync <room> <device_root> [conf] | verify <room> <device_root> [conf]\n" as *u8)
378 return MSY_EXIT_USAGE
379}
380func msy_confpick(argc: i64, argv: *i64, idx: i64) -> *u8 {
381 if argc > idx { return argv[idx] as *u8 }
382 return "knowledge/comms/chatstore.conf" as *u8
383}
384func main(argc: i64, argv: *i64) -> i64 {
385 if argc < 4 { let u: i64 = msy_usage() sys_exit(u) return u }
386 let verb: *u8 = argv[1] as *u8
387 if msy_eq(verb, "sync" as *u8) == 1 {
388 let rc: i64 = msy_devsync(argv[2] as *u8, argv[3] as *u8, msy_confpick(argc, argv, 4))
389 sys_exit(rc)
390 return rc
391 }
392 if msy_eq(verb, "verify" as *u8) == 1 {
393 let rc: i64 = msy_verify(argv[2] as *u8, argv[3] as *u8, msy_confpick(argc, argv, 4))
394 sys_exit(rc)
395 return rc
396 }
397 let u2: i64 = msy_usage()
398 sys_exit(u2)
399 return u2
400}