nx_research_journal.nx source
↩ module page · 219 lines · 9610 B
1// nx_research_journal.nx -- RES-R6: a CRASH-RESUMABLE research session journal.
2//
3// module: nishi-core.research.journal
4// capability: CORE_COMPUTE (a durable, replay-able research progress log)
5//
6// WHAT this builds (one capability): a research session records each phase
7// (FETCH / EXTRACT / VERIFY / SYNTH) and per-source progress as ATOMIC FRAMED
8// LINES. On restart after a crash, the journal is REPLAYED: the highest phase
9// that has a DONE record becomes the resume high-water mark, and every phase at
10// or below it is SKIPPED -- the completed FETCH/EXTRACT work is NOT re-done.
11//
12// WHY it is correct under crash: every journal line is written by the WMS-R0
13// primitive fa_append (nx_framed_append.nx) -- the whole record is assembled in
14// ONE buffer and emitted as EXACTLY ONE sys_write() to an O_APPEND fd, so a
15// crash can never produce a half-written record that interleaves with another.
16// A crash CAN leave the very last record absent (it was never reached) or, on
17// other channels, torn; the replay therefore credits a phase DONE ONLY when it
18// reads a WELL-FORMED line ("RJ " head + " END" tail + matching sid + ev=DONE).
19// A torn / truncated last line is ignored -> replay can never SKIP work that did
20// not actually complete (proven by the gate's torn-line tamper row).
21//
22// REUSE / lineage (rule 15 DRY, check-registries): this organ adds NO new write
23// primitive. It COMPOSES fa_append / fa_cat / fa_catn from nx_framed_append.nx
24// (the WMS-R0 floor) for writing, and the count_torn-style line+token byte-walk
25// idiom from nx_framed_append_gate.nx for replay parsing. Sovereign: only
26// nx_syscalls + nx_framed_append.
27// license_tier: ORIGINAL
28import "nx_syscalls.nx"
29import "nx_framed_append.nx"
30
31// bounded framed record size (mirrors RECCAP idiom in the WMS-R0 gate)
32const RJ_REC_CAP: i64 = 256
33
34// phase codes (FETCH < EXTRACT < VERIFY < SYNTH -- ordered so "highest DONE
35// phase" is a valid resume high-water mark)
36const RJ_FETCH: i64 = 1
37const RJ_EXTRACT: i64 = 2
38const RJ_VERIFY: i64 = 3
39const RJ_SYNTH: i64 = 4
40
41// event codes
42const RJ_BEGIN: i64 = 1
43const RJ_DONE: i64 = 2
44
45// ---- WRITE side: one atomic framed line per event -----------------
46
47// Record ONE phase/source event as ONE atomic framed line:
48// "RJ sid=<sid> ph=<phase> src=<src_idx> ev=<event> ts=<epoch> END"
49// Returns fa_append's rc: > 0 ok (bytes written) ; -2 oversized ; -1/-3 io.
50// sid = the research session id (lets one journal hold many sessions).
51func rj_log(path: *u8, sid: i64, phase: i64, src_idx: i64, event: i64) -> i64 {
52 let buf: *u8 = sys_mmap(RJ_REC_CAP + 16)
53 var o: i64 = 0
54 o = fa_cat(buf, o, "RJ sid=\x00" as *u8)
55 o = fa_catn(buf, o, sid)
56 o = fa_cat(buf, o, " ph=\x00" as *u8)
57 o = fa_catn(buf, o, phase)
58 o = fa_cat(buf, o, " src=\x00" as *u8)
59 o = fa_catn(buf, o, src_idx)
60 o = fa_cat(buf, o, " ev=\x00" as *u8)
61 o = fa_catn(buf, o, event)
62 o = fa_cat(buf, o, " ts=\x00" as *u8)
63 o = fa_catn(buf, o, sys_now_realtime_sec())
64 o = fa_cat(buf, o, " END\x00" as *u8)
65 return fa_append(path, buf, o, RJ_REC_CAP)
66}
67
68// Convenience wrappers (single responsibility, rule 9).
69func rj_phase_begin(path: *u8, sid: i64, phase: i64) -> i64 {
70 return rj_log(path, sid, phase, 0 - 1, RJ_BEGIN) // src = -1 (whole phase)
71}
72func rj_phase_done(path: *u8, sid: i64, phase: i64) -> i64 {
73 return rj_log(path, sid, phase, 0 - 1, RJ_DONE)
74}
75func rj_source_done(path: *u8, sid: i64, phase: i64, src: i64) -> i64 {
76 return rj_log(path, sid, phase, src, RJ_DONE)
77}
78
79// ---- REPLAY side: parse the journal, derive the resume high-water ----
80
81// Within line [ls, le) of buffer b, find the integer that follows the 3-byte
82// token `tok` (e.g. "ph=", "ev=", a 3-char nul-terminated token whose 3rd char
83// is '='). Returns the parsed non-negative integer, or `dflt` if the token is
84// not present. Negative values in records (src=-1) are not parsed by callers
85// that need them (callers only read sid/ph/ev which are >= 0).
86// (Byte-walk idiom lifted from count_torn's " END" / "FA w=" token scans.)
87func rj_field(b: *u8, ls: i64, le: i64, tok: *u8, dflt: i64) -> i64 {
88 let t0: u8 = tok[0]
89 let t1: u8 = tok[1]
90 let t2: u8 = tok[2]
91 var i: i64 = ls
92 while i + 2 < le {
93 if b[i] == t0 {
94 if b[i + 1] == t1 {
95 if b[i + 2] == t2 {
96 // matched token; parse digits right after it
97 var j: i64 = i + 3
98 var have: i64 = 0
99 var v: i64 = 0
100 while j < le {
101 let c: u8 = b[j]
102 if c < (48 as u8) { j = le }
103 else {
104 if c > (57 as u8) { j = le }
105 else {
106 v = v * 10 + ((c as i64) - 48)
107 have = 1
108 j = j + 1
109 }
110 }
111 }
112 if have == 1 { return v }
113 return dflt
114 }
115 }
116 }
117 i = i + 1
118 }
119 return dflt
120}
121
122// A line [ls, le) (le = index of its '\n') is WELL-FORMED iff it begins "RJ "
123// and the 4 bytes immediately before '\n' are " END". (Mirrors the WMS-R0
124// gate's prefix+suffix sentinel discipline -- a torn/truncated line fails this
125// and is ignored, so it can never advance the resume point.)
126func rj_wellformed(b: *u8, ls: i64, le: i64) -> i64 {
127 if le - ls < 8 { return 0 }
128 if b[ls] != (82 as u8) { return 0 } // R
129 if b[ls + 1] != (74 as u8) { return 0 } // J
130 if b[ls + 2] != (32 as u8) { return 0 } // space
131 let e: i64 = le - 1
132 if b[e - 3] != (32 as u8) { return 0 } // space
133 if b[e - 2] != (69 as u8) { return 0 } // E
134 if b[e - 1] != (78 as u8) { return 0 } // N
135 if b[e] != (68 as u8) { return 0 } // D
136 return 1
137}
138
139// REPLAY the journal for session `sid`. Walks every '\n'-terminated line; for
140// each WELL-FORMED line whose sid matches and ev==DONE, tracks completion.
141// Torn / foreign-sid / non-DONE lines are ignored (defensive at the boundary,
142// rule 12). CRITICAL resume semantics: a phase counts as COMPLETE only when it
143// has a PHASE-LEVEL done record (src=-1). A crash mid-VERIFY may have logged
144// VERIFY BEGIN + some source_done records but NO phase_done -> VERIFY is NOT
145// complete -> it is NOT skipped on resume. This is what makes resume correct.
146// Token convention for rj_field: a 3-byte token ending in '=' so parsing starts
147// at the first digit. "id=" uniquely tags sid=, "rc=" uniquely tags src=.
148// Outputs:
149// out[0] = count of source_done records seen in the resume (high-water) phase
150// out[1] = highest phase with a phase-level DONE (== the return value)
151// out[2] = total well-formed lines parsed
152// Returns the RESUME HIGH-WATER = highest COMPLETED phase; 0 = nothing complete.
153func rj_replay(path: *u8, sid: i64, out: *i64) -> i64 {
154 let szp: *i64 = sys_mmap(16) as *i64
155 let b: *u8 = sys_read_file(path, szp)
156 let sz: i64 = szp[0]
157 var hi_done: i64 = 0 // highest phase with a PHASE-LEVEL DONE
158 var wf: i64 = 0
159 if b as i64 == 0 { out[0] = 0; out[1] = 0; out[2] = 0; return 0 }
160 // pass 1: find the high-water completed phase (phase-level DONE, src=-1).
161 var ls: i64 = 0
162 var i: i64 = 0
163 while i < sz {
164 if b[i] == (10 as u8) {
165 if rj_wellformed(b, ls, i) == 1 {
166 wf = wf + 1
167 let rsid: i64 = rj_field(b, ls, i, "id=\x00" as *u8, 0 - 9)
168 if rsid == sid {
169 let rev: i64 = rj_field(b, ls, i, "ev=\x00" as *u8, 0)
170 let rph: i64 = rj_field(b, ls, i, "ph=\x00" as *u8, 0)
171 // src=-1 (phase-level) -> rj_field finds no digit -> dflt -1.
172 let rsrc: i64 = rj_field(b, ls, i, "rc=\x00" as *u8, 0 - 1)
173 if rev == RJ_DONE {
174 if rsrc < 0 {
175 if rph > hi_done { hi_done = rph }
176 }
177 }
178 }
179 }
180 ls = i + 1
181 }
182 i = i + 1
183 }
184 // pass 2: count source-done records observed in the high-water phase (for
185 // diagnostics / progress within the resume phase). Small journals -> cheap.
186 var src_in_hi: i64 = 0
187 ls = 0; i = 0
188 while i < sz {
189 if b[i] == (10 as u8) {
190 if rj_wellformed(b, ls, i) == 1 {
191 let rsid2: i64 = rj_field(b, ls, i, "id=\x00" as *u8, 0 - 9)
192 if rsid2 == sid {
193 let rev2: i64 = rj_field(b, ls, i, "ev=\x00" as *u8, 0)
194 let rph2: i64 = rj_field(b, ls, i, "ph=\x00" as *u8, 0)
195 let rsrc2: i64 = rj_field(b, ls, i, "rc=\x00" as *u8, 0 - 1)
196 if rev2 == RJ_DONE {
197 if rph2 == hi_done {
198 if rsrc2 >= 0 { src_in_hi = src_in_hi + 1 }
199 }
200 }
201 }
202 }
203 ls = i + 1
204 }
205 i = i + 1
206 }
207 out[0] = src_in_hi
208 out[1] = hi_done
209 out[2] = wf
210 return hi_done
211}
212
213// RESUME decision: should this phase be SKIPPED given the replay high-water?
214// A phase whose code is <= the highest completed phase is already done -> skip.
215// Returns 1 = skip (already complete), 0 = execute it.
216func rj_should_skip(phase: i64, last_done_phase: i64) -> i64 {
217 if phase <= last_done_phase { return 1 }
218 return 0
219}