nxtask_store.nx source
↩ module page · 360 lines · 11915 B
1// nxtask_store.nx -- persistence layer for nxtask (beads-equivalent).
2//
3// Provides buffer round-trip of a TaskGraph + convenience wrappers
4// that go through fs.nx for disk persistence. Single-file binary
5// format; zero runtime deps beyond syscalls + fs.
6//
7// Format v1 -- magic "NXT0":
8//
9// 4 magic ("NXT0", little-endian bytes)
10// 8 n_tasks (i64 LE)
11// repeat n_tasks:
12// 8 state
13// 8 created_ns
14// 8 updated_ns
15// 8 desc_len
16// desc_len desc bytes (no trailing NUL)
17// 8 claimed_by_len
18// claimed_by_len claimed_by bytes (omitted when 0)
19// 8 summary_len
20// summary_len summary bytes (omitted when 0)
21// 8 n_blocked
22// 8*n_blocked blocked_by task indices (i64 LE each)
23//
24// Design notes:
25//
26// * We store strings inline rather than via a separate string pool
27// because each task's text is small (<2 KB) and the whole file
28// typically fits in one page. A pool only pays off when many
29// tasks share identical text, which isn't the nxtask workload.
30//
31// * id_hash is NOT persisted. v0.0.1 uses counter-hash which is a
32// function of the task's position + creation order, both of
33// which are recovered at load time. v0.1.0 will persist a real
34// sha256(desc + created_ns) once we switch to content-addressed
35// ids.
36//
37// * Little-endian is the RISC-V default so this is the native byte
38// order. No byte-swap needed.
39//
40// This commit ships the buffer API only. An fs_save/fs_load wrapper
41// lands in a follow-up (depends on fs_mkdir parent-dir creation).
42
43import "syscalls.nx"
44import "nxtask.nx"
45
46const NXT_MAGIC_0: i64 = 0x4E // 'N'
47const NXT_MAGIC_1: i64 = 0x58 // 'X'
48const NXT_MAGIC_2: i64 = 0x54 // 'T'
49const NXT_MAGIC_3: i64 = 0x30 // '0'
50
51// === helpers: i64 <-> bytes (little-endian) ========================
52
53func write_i64_le(buf: *u8, off: i64, val: i64) -> i64 {
54 var i: i64 = 0
55 while i < 8 {
56 buf[off + i] = (val >> (i * 8)) & 0xFF
57 i = i + 1
58 }
59 return off + 8
60}
61
62func read_i64_le(buf: *u8, off: i64) -> i64 {
63 var v: i64 = 0
64 var i: i64 = 0
65 while i < 8 {
66 let b: i64 = buf[off + i] & 0xFF
67 v = v | (b << (i * 8))
68 i = i + 1
69 }
70 return v
71}
72
73// Copy `n` bytes from `src` to `dst`. Returns new dst offset.
74func copy_bytes(dst: *u8, off: i64, src: *u8, n: i64) -> i64 {
75 var i: i64 = 0
76 while i < n {
77 dst[off + i] = src[i]
78 i = i + 1
79 }
80 return off + n
81}
82
83// === serialize =====================================================
84//
85// Write a graph to `buf` (must be >= nxtask_serialized_size).
86// Returns bytes written, or -1 on overflow.
87
88func nxtask_serialized_size(g: *TaskGraph) -> i64 {
89 var total: i64 = 4 + 8 // magic + n_tasks
90 var i: i64 = 0
91 while i < g.n_tasks {
92 let t: *Task = task_at(g, i)
93 total = total + 8 + 8 + 8 // state + created + updated
94 total = total + 8 + t.desc_len
95 total = total + 8 + t.claimed_by_len
96 total = total + 8 + t.summary_len
97 total = total + 8 + t.n_blocked * 8
98 i = i + 1
99 }
100 return total
101}
102
103func nxtask_serialize(g: *TaskGraph, buf: *u8, cap: i64) -> i64 {
104 let need: i64 = nxtask_serialized_size(g)
105 if need > cap { return -1 }
106
107 // Magic
108 buf[0] = NXT_MAGIC_0 & 0xFF
109 buf[1] = NXT_MAGIC_1 & 0xFF
110 buf[2] = NXT_MAGIC_2 & 0xFF
111 buf[3] = NXT_MAGIC_3 & 0xFF
112 var off: i64 = 4
113 off = write_i64_le(buf, off, g.n_tasks)
114
115 var i: i64 = 0
116 while i < g.n_tasks {
117 let t: *Task = task_at(g, i)
118 off = write_i64_le(buf, off, t.state)
119 off = write_i64_le(buf, off, t.created_ns)
120 off = write_i64_le(buf, off, t.updated_ns)
121 off = write_i64_le(buf, off, t.desc_len)
122 if t.desc_len > 0 {
123 off = copy_bytes(buf, off, t.desc, t.desc_len)
124 }
125 off = write_i64_le(buf, off, t.claimed_by_len)
126 if t.claimed_by_len > 0 {
127 off = copy_bytes(buf, off, t.claimed_by, t.claimed_by_len)
128 }
129 off = write_i64_le(buf, off, t.summary_len)
130 if t.summary_len > 0 {
131 off = copy_bytes(buf, off, t.summary, t.summary_len)
132 }
133 off = write_i64_le(buf, off, t.n_blocked)
134 var j: i64 = 0
135 while j < t.n_blocked {
136 off = write_i64_le(buf, off, t.blocked_by[j])
137 j = j + 1
138 }
139 i = i + 1
140 }
141 return off
142}
143
144// === deserialize ===================================================
145//
146// Read from a buf of size `n`. Returns 0 (null) on malformed input.
147
148func nxtask_deserialize(buf: *u8, n: i64) -> *TaskGraph {
149 if n < 12 { return 0 as *TaskGraph } // magic + header
150 if buf[0] != NXT_MAGIC_0 { return 0 as *TaskGraph }
151 if buf[1] != NXT_MAGIC_1 { return 0 as *TaskGraph }
152 if buf[2] != NXT_MAGIC_2 { return 0 as *TaskGraph }
153 if buf[3] != NXT_MAGIC_3 { return 0 as *TaskGraph }
154
155 var off: i64 = 4
156 let n_tasks: i64 = read_i64_le(buf, off)
157 off = off + 8
158 if n_tasks < 0 { return 0 as *TaskGraph }
159
160 let cap: i64 = n_tasks + 32 // grow headroom
161 let g: *TaskGraph = nxtask_graph_new(cap)
162 g.n_tasks = n_tasks
163
164 var i: i64 = 0
165 while i < n_tasks {
166 if off + 32 > n { return 0 as *TaskGraph } // truncated
167 let t: *Task = task_at(g, i)
168 t.state = read_i64_le(buf, off)
169 off = off + 8
170 t.created_ns = read_i64_le(buf, off)
171 off = off + 8
172 t.updated_ns = read_i64_le(buf, off)
173 off = off + 8
174
175 // desc
176 let d_len: i64 = read_i64_le(buf, off)
177 off = off + 8
178 if d_len < 0 { return 0 as *TaskGraph }
179 if off + d_len > n { return 0 as *TaskGraph }
180 t.desc_len = d_len
181 if d_len > 0 {
182 let d: *u8 = sys_mmap(d_len + 16)
183 var k: i64 = 0
184 while k < d_len {
185 d[k] = buf[off + k]
186 k = k + 1
187 }
188 d[d_len] = 0
189 t.desc = d
190 off = off + d_len
191 } else {
192 t.desc = 0 as *u8
193 }
194
195 // claimed_by
196 let c_len: i64 = read_i64_le(buf, off)
197 off = off + 8
198 if c_len < 0 { return 0 as *TaskGraph }
199 if off + c_len > n { return 0 as *TaskGraph }
200 t.claimed_by_len = c_len
201 if c_len > 0 {
202 let c: *u8 = sys_mmap(c_len + 16)
203 var k: i64 = 0
204 while k < c_len {
205 c[k] = buf[off + k]
206 k = k + 1
207 }
208 c[c_len] = 0
209 t.claimed_by = c
210 off = off + c_len
211 } else {
212 t.claimed_by = 0 as *u8
213 }
214
215 // summary
216 let s_len: i64 = read_i64_le(buf, off)
217 off = off + 8
218 if s_len < 0 { return 0 as *TaskGraph }
219 if off + s_len > n { return 0 as *TaskGraph }
220 t.summary_len = s_len
221 if s_len > 0 {
222 let s: *u8 = sys_mmap(s_len + 16)
223 var k: i64 = 0
224 while k < s_len {
225 s[k] = buf[off + k]
226 k = k + 1
227 }
228 s[s_len] = 0
229 t.summary = s
230 off = off + s_len
231 } else {
232 t.summary = 0 as *u8
233 }
234
235 // n_blocked + blocked_by
236 let nb: i64 = read_i64_le(buf, off)
237 off = off + 8
238 if nb < 0 { return 0 as *TaskGraph }
239 if off + nb * 8 > n { return 0 as *TaskGraph }
240 t.n_blocked = nb
241 if nb > 0 {
242 let deps: *i64 = sys_mmap(nb * 8 + 16) as *i64
243 var k: i64 = 0
244 while k < nb {
245 deps[k] = read_i64_le(buf, off)
246 off = off + 8
247 k = k + 1
248 }
249 t.blocked_by = deps
250 } else {
251 t.blocked_by = 0 as *i64
252 }
253
254 i = i + 1
255 }
256
257 return g
258}
259
260// === self-test =====================================================
261//
262// Round-trip 3 tasks with mixed state + deps. Assert every field
263// survives. Exit 42 on success, other codes on specific mismatches.
264
265func assert_bytes_equal(a: *u8, b: *u8, n: i64) -> i64 {
266 var i: i64 = 0
267 while i < n {
268 if a[i] != b[i] { return 0 }
269 i = i + 1
270 }
271 return 1
272}
273
274func main() -> i64 {
275 // Build source graph: 3 tasks, mid-workflow state.
276 let g: *TaskGraph = nxtask_graph_new(8)
277 nxtask_add(g, "Port 12-bit imm expansion" as *u8, 25, 1)
278 nxtask_add(g, "Add persistence to nxtask" as *u8, 25, 2)
279 nxtask_add(g, "Build Mormon v0" as *u8, 15, 3)
280
281 nxtask_add_dep(g, 1, 0) // persistence blocked by imm port
282 nxtask_add_dep(g, 2, 1) // mormon blocked by persistence
283
284 nxtask_claim(g, 0, "agent-foo" as *u8, 9, 10)
285 nxtask_close(g, 0, "Shipped with four helpers" as *u8, 25, 11)
286
287 // Serialize.
288 let buf: *u8 = sys_mmap(16384)
289 let n: i64 = nxtask_serialize(g, buf, 16384)
290 if n < 0 { return __syscall(93, 80, 0, 0, 0, 0, 0) }
291 if n < 12 { return __syscall(93, 81, 0, 0, 0, 0, 0) }
292
293 // Magic bytes at offset 0-3.
294 if buf[0] != 0x4E { return __syscall(93, 82, 0, 0, 0, 0, 0) }
295 if buf[1] != 0x58 { return __syscall(93, 83, 0, 0, 0, 0, 0) }
296 if buf[2] != 0x54 { return __syscall(93, 84, 0, 0, 0, 0, 0) }
297 if buf[3] != 0x30 { return __syscall(93, 85, 0, 0, 0, 0, 0) }
298
299 // Deserialize.
300 let g2: *TaskGraph = nxtask_deserialize(buf, n)
301 if g2 == (0 as *TaskGraph) { return __syscall(93, 86, 0, 0, 0, 0, 0) }
302 if g2.n_tasks != 3 { return __syscall(93, 87, 0, 0, 0, 0, 0) }
303
304 // Per-task field round-trip.
305 let t0: *Task = task_at(g, 0)
306 let u0: *Task = task_at(g2, 0)
307 if u0.state != t0.state { return __syscall(93, 90, 0, 0, 0, 0, 0) }
308 if u0.created_ns != t0.created_ns { return __syscall(93, 91, 0, 0, 0, 0, 0) }
309 if u0.updated_ns != t0.updated_ns { return __syscall(93, 92, 0, 0, 0, 0, 0) }
310 if u0.desc_len != t0.desc_len { return __syscall(93, 93, 0, 0, 0, 0, 0) }
311 if assert_bytes_equal(u0.desc, t0.desc, t0.desc_len) != 1 {
312 return __syscall(93, 94, 0, 0, 0, 0, 0)
313 }
314 if u0.claimed_by_len != t0.claimed_by_len {
315 return __syscall(93, 95, 0, 0, 0, 0, 0)
316 }
317 if assert_bytes_equal(u0.claimed_by, t0.claimed_by, t0.claimed_by_len) != 1 {
318 return __syscall(93, 96, 0, 0, 0, 0, 0)
319 }
320 if u0.summary_len != t0.summary_len {
321 return __syscall(93, 97, 0, 0, 0, 0, 0)
322 }
323 if assert_bytes_equal(u0.summary, t0.summary, t0.summary_len) != 1 {
324 return __syscall(93, 98, 0, 0, 0, 0, 0)
325 }
326
327 // Dep round-trip on task 1.
328 let t1: *Task = task_at(g, 1)
329 let u1: *Task = task_at(g2, 1)
330 if u1.n_blocked != t1.n_blocked {
331 return __syscall(93, 100, 0, 0, 0, 0, 0)
332 }
333 if u1.blocked_by[0] != 0 {
334 return __syscall(93, 101, 0, 0, 0, 0, 0)
335 }
336
337 // Task 2 (never claimed) -- claimed_by + summary should be null/zero.
338 let u2: *Task = task_at(g2, 2)
339 if u2.claimed_by_len != 0 {
340 return __syscall(93, 110, 0, 0, 0, 0, 0)
341 }
342 if u2.summary_len != 0 {
343 return __syscall(93, 111, 0, 0, 0, 0, 0)
344 }
345 if u2.n_blocked != 1 {
346 return __syscall(93, 112, 0, 0, 0, 0, 0)
347 }
348 if u2.blocked_by[0] != 1 {
349 return __syscall(93, 113, 0, 0, 0, 0, 0)
350 }
351
352 // Malformed-magic deserialise path returns null.
353 buf[0] = 0x00
354 let g_bad: *TaskGraph = nxtask_deserialize(buf, n)
355 if g_bad != (0 as *TaskGraph) {
356 return __syscall(93, 120, 0, 0, 0, 0, 0)
357 }
358
359 return __syscall(93, 42, 0, 0, 0, 0, 0)
360}