nx_state_spool_core.nx source
↩ module page · 113 lines · 4756 B
1// nx_state_spool_core.nx -- importable CORE of the STORE-AND-FORWARD state spool (CR-R1a of the
2// crash-resume->SOTA roadmap, amended 07-15 by operator directive: "the hub is the NAS not the
3// laptop; the laptop when on network should sync to the NAS as much as possible; when the NAS
4// isn't available it should go local"). Semantics: capture writes are NAS-FIRST -- a line is
5// spooled locally, then DRAIN attempts delivery via a PLUGGABLE transport ELF (argv[1]=line,
6// exit 0 = delivered); delivered lines leave the spool, failed lines are KEPT (offline = nothing
7// lost) without blocking later lines (head-of-line non-blocking; state lines are latest-wins).
8// Durability: spool ops serialize under a sidecar flock; drain rewrites via tmp + renameat
9// (atomic -- a crash mid-drain leaves the old spool or the new one, never a torn file).
10// Transport is a parameter so the gate proves the semantics OFFLINE; CR-R1 plugs the real
11// NAS API poster in without touching these mechanics.
12// license_tier: ORIGINAL
13import "nx_syscalls.nx"
14import "nx_crashresume_census_core.nx"
15const K_MAGIC_262144: i64 = 262144
16const K_MAGIC_262143: i64 = 262143
17const K_MAGIC_8192: i64 = 8192
18const K_MAGIC_8191: i64 = 8191
19
20// take the sidecar lock: open-or-create lockpath, flock EX (blocking). returns fd, or -1.
21func ssp_lock(lockpath: *u8) -> i64 {
22 let fd: i64 = sys_openat_wr(lockpath, 420)
23 if fd < 0 { return 0 - 1 }
24 let r: i64 = sys_flock(fd, SYS_LOCK_EX)
25 if r < 0 { sys_close(fd); return 0 - 1 }
26 return fd
27}
28
29func ssp_unlock(fd: i64) -> i64 {
30 if fd >= 0 { sys_flock(fd, SYS_LOCK_UN); sys_close(fd) }
31 return 0
32}
33
34// append one line (caller passes WITHOUT trailing newline) to the spool under the lock.
35func ssp_spool(spoolpath: *u8, lockpath: *u8, line: *u8) -> i64 {
36 let lk: i64 = ssp_lock(lockpath)
37 if lk < 0 { return 0 - 1 }
38 let fd: i64 = sys_openat_append(spoolpath, 420)
39 if fd < 0 { ssp_unlock(lk); return 0 - 1 }
40 let n: i64 = ccz_slen(line)
41 sys_write(fd, line, n)
42 sys_write(fd, "\n" as *u8, 1)
43 sys_close(fd)
44 ssp_unlock(lk)
45 return 0
46}
47
48// fork+exec the transport with argv=[elf, line, NULL]; return the child's exit code (127 if the
49// exec itself failed -- e.g. missing transport binary -> treated as NOT delivered, fail-safe).
50func ssp_run_transport(elf: *u8, line: *u8) -> i64 {
51 let pid: i64 = sys_fork()
52 if pid < 0 { return 127 }
53 if pid == 0 {
54 let av: *i64 = sys_mmap(32) as *i64
55 av[0] = elf as i64
56 av[1] = line as i64
57 av[2] = 0
58 sys_execve(elf, av, 0 as *i64)
59 sys_exit(127)
60 }
61 let st: *i64 = sys_mmap(16) as *i64
62 let r: i64 = sys_wait4(pid, st, 0)
63 if r < 0 { return 127 }
64 return wait_exit_code(st[0])
65}
66
67// drain the spool: per line attempt the transport; delivered lines drop, failed lines are kept in
68// order; the spool is atomically rewritten (tmp + renameat). sentp[0]/keptp[0] carry the counts.
69// Missing/empty spool -> 0 with counts 0/0. Returns 0, or -1 on lock failure (spool untouched).
70func ssp_drain(spoolpath: *u8, lockpath: *u8, tmppath: *u8, elf: *u8, sentp: *i64, keptp: *i64) -> i64 {
71 sentp[0] = 0
72 keptp[0] = 0
73 let lk: i64 = ssp_lock(lockpath)
74 if lk < 0 { return 0 - 1 }
75 let buf: *u8 = sys_mmap(K_MAGIC_262144)
76 let n: i64 = ccz_read(spoolpath, buf, K_MAGIC_262143)
77 if n <= 0 { ssp_unlock(lk); return 0 }
78 let keep: *u8 = sys_mmap(K_MAGIC_262144)
79 var ko: i64 = 0
80 let scratch: *u8 = sys_mmap(K_MAGIC_8192)
81 var ls: i64 = 0
82 while ls < n {
83 var le: i64 = ls
84 var go: i64 = 1
85 while go == 1 { go = 0; if le < n { if buf[le] != (10 as u8) { le = le + 1; go = 1 } } }
86 if le > ls {
87 // copy line -> NUL-terminated scratch (transport argv needs a C string)
88 var i: i64 = 0
89 while i < le - ls { if i < K_MAGIC_8191 { scratch[i] = buf[ls+i] } i = i + 1 }
90 var sl: i64 = le - ls
91 if sl > K_MAGIC_8191 { sl = K_MAGIC_8191 }
92 scratch[sl] = 0 as u8
93 let ec: i64 = ssp_run_transport(elf, scratch)
94 if ec == 0 { sentp[0] = sentp[0] + 1 } else {
95 keptp[0] = keptp[0] + 1
96 var k: i64 = 0
97 while k < sl { keep[ko] = scratch[k]; ko = ko + 1; k = k + 1 }
98 keep[ko] = 10 as u8
99 ko = ko + 1
100 }
101 }
102 ls = le + 1
103 }
104 // atomic rewrite: kept lines (possibly zero) -> tmp -> rename over spool
105 let tfd: i64 = sys_openat_wr(tmppath, 420)
106 if tfd < 0 { ssp_unlock(lk); return 0 - 1 }
107 if ko > 0 { sys_write(tfd, keep, ko) }
108 sys_fsync(tfd)
109 sys_close(tfd)
110 sys_renameat(tmppath, spoolpath)
111 ssp_unlock(lk)
112 return 0
113}