code wiki / _hdl_build / nx_pub_ship.nx
nx_pub_ship.nx source
↩ module page · 159 lines · 8015 B
1// nx_pub_ship.nx -- THE NISHI PUBLISHER R3: STAGE + VERIFY transport (composes nx_publisher's primitives,
2// so nx_publisher.nx itself is untouched -- additive, no collision with the publisher workstream).
3//
4// For each PENDING request in the queue: ship `src` -> `dest`+".staged" on the NAS via the sovereign sender
5// (_offc/nx_aw_send.elf -> nx_hostctl recv), read the LANDED bytes back, and VERIFY landed-sha == request-sha.
6// FAIL-CLOSED: a mismatch / unreadable-landed is NEVER ledgered (retryable; live binary never touched).
7// On success record STAGED to the ledger (idempotent, rule #10 -- re-run stages 0). STOPS at "staged on NAS":
8// the atomic promote (rename .staged -> live) + daemon restart stay with the OWNING workstream's supervisor
9// (coordinate, never brick -- #26). NEVER touches the live binary by construction (only the .staged sibling).
10//
11// argv (all optional; defaults = the real publish): [1]=queue [2]=ledger [3]=sender-elf [4]=dest-root
12// [5]=verify-root. dest-root->verify-root maps the NAS path to where the publisher can read it back (the
13// homes drvfs mount on a WSL-hosted publisher; "/volume1/homes" directly when run on the NAS). license_tier: ORIGINAL
14import "nx_publisher.nx"
15import "nx_syscalls.nx"
16import "nx_framed_append.nx"
17import "nx_arbiter.nx"
18const SHIP_MAGIC_4096: i64 = 4096
19const SHIP_MAGIC_1024: i64 = 1024
20const SHIP_MAGIC_1100: i64 = 1100
21const SHIP_MAGIC_1200: i64 = 1200
22const SHIP_MAGIC_100000: i64 = 100000
23
24const SHIP_SENDER: *u8 = "_offc/nx_aw_send.elf"
25const SHIP_DEST_ROOT: *u8 = "/volume1/homes"
26const SHIP_VERIFY_ROOT: *u8 = "/mnt/nas_homes"
27const SHIP_STAGED_SFX: *u8 = ".staged"
28const SHIP_LOCKRES: *u8 = "publish:nishifamily"
29
30// --- small string helpers (consts are passed in as PARAMS so we never index a const-global inline) ---
31func ship_puts(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return n }
32func ship_puti(v: i64) -> i64 {
33 let t: *u8 = sys_mmap(28); var m: i64 = v; if m < 0 { m = 0 - m }
34 var k: i64 = 0; if m == 0 { t[0] = 48 as u8; k = 1 }
35 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
36 let o: *u8 = sys_mmap(30); var w: i64 = 0; var q: i64 = k - 1
37 while q >= 0 { o[w] = t[q]; w = w + 1; q = q - 1 }
38 sys_write(1, o, w); return 0
39}
40func ship_slen(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
41func ship_cat(dst: *u8, off: i64, s: *u8) -> i64 { var o: i64 = off; var i: i64 = 0; while s[i] != (0 as u8) { dst[o] = s[i]; o = o + 1; i = i + 1 } return o }
42func ship_startswith(buf: *u8, pfx: *u8) -> i64 { var i: i64 = 0; while pfx[i] != (0 as u8) { if buf[i] != pfx[i] { return 0 } i = i + 1 } return 1 }
43
44// out = dest + SHIP_STAGED_SFX (NUL-terminated). returns length.
45func ship_staged_name(dest: *u8, out: *u8) -> i64 {
46 var o: i64 = ship_cat(out, 0, dest)
47 o = ship_cat(out, o, SHIP_STAGED_SFX)
48 out[o] = 0 as u8
49 return o
50}
51// translate NAS path under dest_root -> the local verify path under verify_root. 1 ok / 0 no-match.
52func ship_verify_path(nas: *u8, dest_root: *u8, verify_root: *u8, out: *u8) -> i64 {
53 if ship_startswith(nas, dest_root) == 0 { return 0 }
54 let plen: i64 = ship_slen(dest_root)
55 var o: i64 = ship_cat(out, 0, verify_root)
56 o = ship_cat(out, o, ((nas as i64) + plen) as *u8)
57 out[o] = 0 as u8
58 return 1
59}
60// fork/exec `sender` <src> <dest>; return child exit code (0=ok).
61func ship_send(sender: *u8, src: *u8, dest: *u8) -> i64 {
62 let pid: i64 = sys_fork()
63 if pid == 0 {
64 let argv: *i64 = sys_mmap(64) as *i64
65 argv[0] = sender as i64; argv[1] = src as i64; argv[2] = dest as i64; argv[3] = 0
66 let envp: *i64 = sys_mmap(16) as *i64
67 envp[0] = "PATH=/usr/bin:/bin" as *u8 as i64; envp[1] = 0
68 sys_execve(sender, argv, envp)
69 sys_exit(127)
70 }
71 let st: *i64 = sys_mmap(16) as *i64
72 sys_wait4(pid, st, 0)
73 return (st[0] >> 8) & 0xff
74}
75// append a STAGED ledger record (idempotency key = sha + staged-dest): sha<TAB>dest<TAB>site<TAB>STAGED.
76func ship_record_staged(ledpath: *u8, sha: *u8, dest: *u8, site: *u8) -> i64 {
77 let rec: *u8 = sys_mmap(SHIP_MAGIC_4096)
78 var o: i64 = 0
79 o = fa_cat(rec, o, sha); rec[o] = 9 as u8; o = o + 1
80 o = fa_cat(rec, o, dest); rec[o] = 9 as u8; o = o + 1
81 o = fa_cat(rec, o, site); rec[o] = 9 as u8; o = o + 1
82 o = fa_cat(rec, o, "STAGED" as *u8)
83 rec[o] = 0 as u8
84 return fa_appendz(ledpath, rec, SHIP_MAGIC_4096)
85}
86
87// THE R3 RUNG. Returns the number newly staged+verified this pass.
88func pub_ship(qpath: *u8, ledpath: *u8, lockres: *u8, sender: *u8, dest_root: *u8, verify_root: *u8) -> i64 {
89 let lenp: *i64 = sys_mmap(8) as *i64
90 let data: *u8 = sys_read_file(qpath, lenp)
91 if (data as i64) == 0 { ship_puts("queue empty/unreadable\n" as *u8); return 0 }
92 let n: i64 = lenp[0]
93 let st: *u8 = sys_mmap(64)
94 let sha: *u8 = sys_mmap(256)
95 let src: *u8 = sys_mmap(SHIP_MAGIC_1024)
96 let dest: *u8 = sys_mmap(SHIP_MAGIC_1024)
97 let site: *u8 = sys_mmap(256)
98 let staged: *u8 = sys_mmap(SHIP_MAGIC_1100)
99 let vpath: *u8 = sys_mmap(SHIP_MAGIC_1200)
100 let gotsha: *u8 = sys_mmap(72)
101 var i: i64 = 0
102 var ls: i64 = 0
103 var nstaged: i64 = 0
104 while i < n {
105 if data[i] == (10 as u8) {
106 let line: *u8 = ((data as i64) + ls) as *u8
107 let ll: i64 = i - ls
108 pub_field(line, ll, 0, st)
109 if pub_streq(st, "PENDING" as *u8) == 1 {
110 pub_field(line, ll, 1, site)
111 pub_field(line, ll, 3, sha)
112 pub_field(line, ll, 4, src)
113 pub_field(line, ll, 5, dest)
114 ship_staged_name(dest, staged)
115 let lfd: i64 = fl_acquire(lockres, SHIP_MAGIC_100000, 1)
116 if pub_led_has(ledpath, sha, staged) == 0 {
117 ship_puts("SHIP " as *u8); ship_puts(src); ship_puts(" -> " as *u8); ship_puts(staged); ship_puts("\n" as *u8)
118 let src_rc: i64 = ship_send(sender, src, staged)
119 if src_rc == 0 {
120 if ship_verify_path(staged, dest_root, verify_root, vpath) == 1 {
121 if pub_sha_file(vpath, gotsha) == 1 {
122 if pub_streq(gotsha, sha) == 1 {
123 ship_record_staged(ledpath, sha, staged, site)
124 ship_puts(" VERIFIED + STAGED (landed sha == request sha)\n" as *u8)
125 nstaged = nstaged + 1
126 } else {
127 ship_puts(" MISMATCH: landed sha != request sha -> NOT ledgered (fail-closed)\n" as *u8)
128 }
129 } else { ship_puts(" VERIFY-READ-FAIL: cannot read landed bytes back\n" as *u8) }
130 } else { ship_puts(" VERIFY-PATH-FAIL: dest not under dest-root\n" as *u8) }
131 } else { ship_puts(" SEND-FAIL: sender exit != 0\n" as *u8) }
132 } else {
133 ship_puts("SKIP already-staged " as *u8); ship_puts(staged); ship_puts("\n" as *u8)
134 }
135 fl_release(lfd)
136 }
137 ls = i + 1
138 }
139 i = i + 1
140 }
141 return nstaged
142}
143
144func main(argc: i64, argv: *i64) -> i64 {
145 pub_init()
146 var q: *u8 = pub_queue_default()
147 var led: *u8 = pub_ledger_default()
148 var sender: *u8 = SHIP_SENDER
149 var droot: *u8 = SHIP_DEST_ROOT
150 var vroot: *u8 = SHIP_VERIFY_ROOT
151 if argc > 1 { q = argv[1] as *u8 }
152 if argc > 2 { led = argv[2] as *u8 }
153 if argc > 3 { sender = argv[3] as *u8 }
154 if argc > 4 { droot = argv[4] as *u8 }
155 if argc > 5 { vroot = argv[5] as *u8 }
156 let nstaged: i64 = pub_ship(q, led, SHIP_LOCKRES, sender, droot, vroot)
157 ship_puts("R3 SHIP pass done, newly staged+verified = " as *u8); ship_puti(nstaged); ship_puts("\n" as *u8)
158 sys_exit(0); return 0
159}