nx_jobclaim_lib.nx source
↩ module page · 472 lines · 21779 B
1// nx_jobclaim_lib.nx -- ONE CONTRACT shared by the _jobs claim WRITER and the _jobs claim READER.
2//
3// WHY THIS IS A LIB AND NOT TWO ORGANS THAT EACH KNOW THE FORMAT.
4// nx_jobclaim_reap marks a dead job claim by APPENDING a tombstone. nx_joblost_gate decides what a
5// claim IS by looking for markers inside it. Those two are a PRODUCER and a CONSUMER of one wire,
6// and this estate has already measured what that costs when each is written separately: a producer
7// and a consumer each correct in isolation can still disagree on the wire.
8//
9// THE SPECIFIC TRAP THIS FILE MAKES IMPOSSIBLE.
10// History here is additive (rule 13): the tombstone is APPENDED, so a reaped claim STILL CONTAINS
11// its original `state=CLAIMED` line. The census classified by substring over the whole file and
12// tested CLAIMED before anything else -- so a reader written on its own sees every reaped claim as
13// still pending. The tombstone lands, the count does not move, the partition tooth breaks, and the
14// reaper reads as a fix while being a no-op.
15// A REAPER SHIPPED ALONE IS A NO-OP THAT LOOKS LIKE A FIX.
16// The ORDER is therefore not each organ's private business. There is ONE classifier -- jr_state_of --
17// and both organs import it, so the ordering cannot be right in one and wrong in the other.
18//
19// WHY A TOMBSTONE AND NOT AN UNLINK. The claims store is a DIRECTORY OF FILES, not a last-wins
20// seg-store, so the clock-plane's tombstone-row pattern does not transfer unchanged. `sys_unlinkat`
21// exists and would work -- and it would also destroy the only record that the job was ever claimed,
22// which is the evidence any later investigation of the loss needs. Rule 13: soft-delete, never
23// DELETE. The precedent is nx_claims' own `reap` verb: release ONLY provably-expired claims, leave
24// an audit note, additive-only, idempotent. This is that shape applied to a different store.
25//
26// license_tier: ORIGINAL
27import "nx_syscalls.nx"
28
29// ---------------------------------------------------------------------------------------------
30// THE WIRE. Every marker literal lives here exactly once. The tombstone LINE is COMPOSED from
31// JR_MARK_REAPED rather than retyped, so the string the reader searches for is the string the
32// writer emitted BY CONSTRUCTION -- a hand-typed second copy is a second copy of the same shape and
33// the two drift silently.
34const JR_MARK_REAPED: *u8 = "state=REAPED" as *u8
35const JR_MARK_DONE: *u8 = "state=DONE" as *u8
36const JR_MARK_CLAIMED: *u8 = "state=CLAIMED" as *u8
37const JR_SELF: *u8 = "nx_jobclaim_reap" as *u8
38const JR_EXT_CLAIM: *u8 = ".claim" as *u8
39const JR_EXT_OUT: *u8 = ".out" as *u8
40const JR_PREFIX: *u8 = "job_" as *u8
41const JR_JOBS: *u8 = "_jobs" as *u8
42const JR_TMP: *u8 = "/tmp/" as *u8
43const JR_DOTDOT: *u8 = ".." as *u8
44
45// ---------------------------------------------------------------------------------------------
46// THE STATES a claim file can be in. UNKNOWN IS ITS OWN BUCKET: a file whose contents we cannot
47// classify must never fall into a bucket we can, because the bucket it lands in becomes the number
48// somebody plans against.
49const JR_ST_EMPTY: i64 = 0
50const JR_ST_REAPED: i64 = 1
51const JR_ST_DONE: i64 = 2
52const JR_ST_CLAIMED: i64 = 3
53const JR_ST_UNKNOWN: i64 = 4
54
55// THE DECISIONS. Exactly one per examined claim; every refusal names its own rule rather than
56// collapsing into a generic "no".
57const JR_D_REAP: i64 = 0
58const JR_D_ALREADY: i64 = 1
59const JR_D_ABSENT: i64 = 2
60const JR_D_DONE: i64 = 3
61const JR_D_HAS_OUT: i64 = 4
62const JR_D_TOO_YOUNG: i64 = 5
63const JR_D_UNPARSED: i64 = 6
64const JR_D_WRITEFAIL: i64 = 7
65const JR_D_PATHLONG: i64 = 8
66const JR_D_BADID: i64 = 9
67
68// info[] slots filled by jr_examine, so a caller can PRINT THE VALUES and not merely a verdict.
69const JR_I_STATE: i64 = 0
70const JR_I_AGE: i64 = 1
71const JR_I_HASOUT: i64 = 2
72const JR_I_BYTES: i64 = 3
73const JR_I_NEEDNL: i64 = 4
74const JR_I_SLOTS: i64 = 5
75// sizeof(i64). The info[] table holds i64 HANDLES, not bytes, so every caller sizing its allocation
76// writes JR_I_SLOTS * 8 -- and nx_magic found that same bare 8 in BOTH consumers. Two copies of one
77// constant is two constants to every scanner, so it is bound ONCE here and derived at both sites.
78const JR_I64_BYTES: i64 = 8
79
80// THE THRESHOLD IS DATA, NOT A LITERAL IN TWO ORGANS. The reaper's age guard is only sound while it
81// is at least as generous as the census's -- if the reaper reaped younger than the census calls LOST
82// it would tombstone jobs that are still running. Sharing ONE conf makes that relation hold by
83// construction instead of by two people remembering the same number. ABSENT IS NOT ZERO: an absent
84// conf yields the declared default and the caller is TOLD which it got.
85const JR_MAXAGE_CONF: *u8 = "knowledge/status/jobclaim_maxage.conf" as *u8
86const JR_DEFAULT_MAX_AGE_SEC: i64 = 3600
87
88const JR_CH_NL: i64 = 10
89const JR_CH_MINUS: i64 = 45
90const JR_CH_DOT: i64 = 46
91const JR_CH_SLASH: i64 = 47
92const JR_CH_ZERO: i64 = 48
93const JR_CH_NINE: i64 = 57
94const JR_B10: i64 = 10
95const JR_SCRATCH: i64 = 16
96
97// The tombstone line is BOUNDED and the bound is DERIVED, not guessed: one optional newline, the
98// marker, three fixed joiners, two decimal integers (an i64 is at most 20 characters), the writer's
99// own name, and a terminating newline. JR_TOMB_MAX is that sum rounded up; JR_LINE is what callers
100// are asked to hand in. jr_tombstone REFUSES a buffer smaller than JR_TOMB_MAX rather than
101// truncating, because a truncated tombstone is a marker the reader cannot find.
102const JR_I64_DIGITS: i64 = 20
103const JR_TOMB_JOIN: i64 = 24
104const JR_TOMB_MAX: i64 = 128
105const JR_LINE: i64 = 256
106const JR_PATH: i64 = 1024
107
108func jr_slen(p: *u8) -> i64 {
109 var n: i64 = 0
110 while p[n] != (0 as u8) { n = n + 1 }
111 return n
112}
113
114// Literal substring search over a COUNTED buffer (not a C string): claim files are read whole and
115// may legitimately contain a NUL, so a strstr over a terminator would stop early and silently.
116func jr_has(buf: *u8, n: i64, pat: *u8) -> i64 {
117 let m: i64 = jr_slen(pat)
118 if m == 0 { return 1 }
119 var i: i64 = 0
120 while i + m <= n {
121 var j: i64 = 0
122 var ok: i64 = 1
123 while j < m { if buf[i+j] != pat[j] { ok = 0; j = m } else { j = j + 1 } }
124 if ok == 1 { return 1 }
125 i = i + 1
126 }
127 return 0
128}
129
130func jr_prefix(s: *u8, pre: *u8) -> i64 {
131 let m: i64 = jr_slen(pre)
132 let n: i64 = jr_slen(s)
133 if n < m { return 0 }
134 var i: i64 = 0
135 while i < m { if s[i] != pre[i] { return 0 } i = i + 1 }
136 return 1
137}
138
139func jr_streq(a: *u8, b: *u8) -> i64 {
140 let n: i64 = jr_slen(a)
141 if n != jr_slen(b) { return 0 }
142 var i: i64 = 0
143 while i < n { if a[i] != b[i] { return 0 } i = i + 1 }
144 return 1
145}
146
147// ---------------------------------------------------------------------------------------------
148// THE CLASSIFIER. THE ORDER IS THE CONTRACT.
149// REAPED is tested FIRST, before DONE and before CLAIMED, because the tombstone is APPENDED to a
150// file that still carries its original marker. Every other order reports a reaped claim as whatever
151// it used to be, which is exactly the silent no-op this lib exists to prevent.
152// A claim is EMPTY when the reservation was created and never filled -- that is the ORIGINAL
153// representation of "claimed, not finished" and it is still on disk for every pre-2026-08-07 job.
154func jr_state_of(buf: *u8, n: i64) -> i64 {
155 if n == 0 { return JR_ST_EMPTY }
156 if jr_has(buf, n, JR_MARK_REAPED) == 1 { return JR_ST_REAPED }
157 if jr_has(buf, n, JR_MARK_DONE) == 1 { return JR_ST_DONE }
158 if jr_has(buf, n, JR_MARK_CLAIMED) == 1 { return JR_ST_CLAIMED }
159 return JR_ST_UNKNOWN
160}
161
162// The state of the claim AT `path`, or -1 when it cannot be opened. Still ONE classifier -- this
163// only fetches the bytes and hands them to jr_state_of. -1 rather than a state, because "I could not
164// look" must never be spelled the same way as an answer.
165func jr_state_of_path(path: *u8, sc: *i64) -> i64 {
166 let b: *u8 = sys_read_file(path, sc)
167 if (b as i64) == 0 { return 0 - 1 }
168 let n: i64 = sc[0]
169 let st: i64 = jr_state_of(b, n)
170 sys_free_file(b, n)
171 return st
172}
173
174func jr_state_name(st: i64) -> *u8 {
175 if st == JR_ST_EMPTY { return "EMPTY" as *u8 }
176 if st == JR_ST_REAPED { return "REAPED" as *u8 }
177 if st == JR_ST_DONE { return "DONE" as *u8 }
178 if st == JR_ST_CLAIMED { return "CLAIMED" as *u8 }
179 return "UNKNOWN" as *u8
180}
181
182// Each refusal NAMES ITS RULE. A generic "refused" sends the reader to re-derive the scan by hand.
183func jr_decision_name(d: i64) -> *u8 {
184 if d == JR_D_REAP { return "REAP" as *u8 }
185 if d == JR_D_ALREADY { return "ALREADY-REAPED" as *u8 }
186 if d == JR_D_ABSENT { return "REFUSED-ABSENT" as *u8 }
187 if d == JR_D_DONE { return "REFUSED-DONE" as *u8 }
188 if d == JR_D_HAS_OUT { return "REFUSED-HAS-OUT" as *u8 }
189 if d == JR_D_TOO_YOUNG { return "REFUSED-TOO-YOUNG" as *u8 }
190 if d == JR_D_UNPARSED { return "REFUSED-UNPARSED" as *u8 }
191 if d == JR_D_WRITEFAIL { return "WRITE-FAILED" as *u8 }
192 if d == JR_D_PATHLONG { return "REFUSED-PATH-TOO-LONG" as *u8 }
193 if d == JR_D_BADID { return "REFUSED-BAD-ID" as *u8 }
194 return "REFUSED-UNCLASSIFIED" as *u8
195}
196
197func jr_decision_why(d: i64) -> *u8 {
198 if d == JR_D_REAP { return "old, outputless, unfinished -- the id was reserved and nothing came back" as *u8 }
199 if d == JR_D_ALREADY { return "already carries a tombstone -- reaping is idempotent, nothing written" as *u8 }
200 if d == JR_D_ABSENT { return "no such claim file -- a reap that invented its victim would report success for work it did not do" as *u8 }
201 if d == JR_D_DONE { return "state=DONE: the job finished, this is healthy history" as *u8 }
202 if d == JR_D_HAS_OUT { return "a sibling .out EXISTS: this is ORPHAN, whose remedy is to publish the result, not to bury it" as *u8 }
203 if d == JR_D_TOO_YOUNG { return "younger than the shared age threshold -- it may still be running" as *u8 }
204 if d == JR_D_UNPARSED { return "contents match no known marker -- UNKNOWN is its own bucket and is never reaped on a guess" as *u8 }
205 if d == JR_D_WRITEFAIL { return "the tombstone could not be appended -- the claim is UNCHANGED" as *u8 }
206 if d == JR_D_PATHLONG { return "the composed path exceeds the buffer -- refused rather than truncated to a different file" as *u8 }
207 if d == JR_D_BADID { return "the id is not a positive integer -- a job id is an epoch second" as *u8 }
208 return "unclassified" as *u8
209}
210
211// ---------------------------------------------------------------------------------------------
212// FIRST decimal run on the FIRST line of a conf. `seen` distinguishes "no digits present" from a
213// genuine 0 -- ABSENT IS NOT ZERO. Semantics are deliberately identical to the hand-rolled parse
214// nx_joblost_gate used for its ratchet floor, which now calls this instead of keeping a second copy.
215func jr_num_from(buf: *u8, n: i64, seen: *i64) -> i64 {
216 seen[0] = 0
217 var v: i64 = 0
218 var i: i64 = 0
219 while i < n {
220 let c: i64 = buf[i] as i64
221 if c >= JR_CH_ZERO { if c <= JR_CH_NINE { v = v * JR_B10 + (c - JR_CH_ZERO); seen[0] = 1 } }
222 if c == JR_CH_NL { i = n } else { i = i + 1 }
223 }
224 return v
225}
226
227// "12345" -> 12345, or -1 when the text is not a positive decimal. STRICT: any non-digit anywhere
228// rejects, because a lenient parse turns a typo'd id into a real file's id.
229func jr_atoi_strict(s: *u8) -> i64 {
230 let n: i64 = jr_slen(s)
231 if n == 0 { return 0 - 1 }
232 if n > JR_I64_DIGITS { return 0 - 1 }
233 var v: i64 = 0
234 var i: i64 = 0
235 while i < n {
236 let c: i64 = s[i] as i64
237 if c < JR_CH_ZERO { return 0 - 1 }
238 if c > JR_CH_NINE { return 0 - 1 }
239 v = v * JR_B10 + (c - JR_CH_ZERO)
240 i = i + 1
241 }
242 if v <= 0 { return 0 - 1 }
243 return v
244}
245
246// THE SHARED AGE THRESHOLD. Reader and writer both call this, so they cannot hold different numbers.
247// Reports through `defaulted` whether the conf was used, so a caller can STATE which bar it applied
248// instead of publishing a threshold nobody can see.
249func jr_maxage(defaulted: *i64, scratch: *i64) -> i64 {
250 defaulted[0] = 1
251 let buf: *u8 = sys_read_file(JR_MAXAGE_CONF, scratch)
252 if (buf as i64) == 0 { return JR_DEFAULT_MAX_AGE_SEC }
253 let n: i64 = scratch[0]
254 if n <= 0 { sys_free_file(buf, n); return JR_DEFAULT_MAX_AGE_SEC }
255 let seen: *i64 = sys_mmap(JR_SCRATCH) as *i64
256 let v: i64 = jr_num_from(buf, n, seen)
257 let got: i64 = seen[0]
258 sys_free_file(buf, n)
259 if got == 0 { return JR_DEFAULT_MAX_AGE_SEC }
260 if v <= 0 { return JR_DEFAULT_MAX_AGE_SEC }
261 defaulted[0] = 0
262 return v
263}
264
265// ---------------------------------------------------------------------------------------------
266// BLAST-RADIUS GUARD. This organ appends to a file named by an id inside a directory named by argv.
267// It is allowed exactly two neighbourhoods: the production _jobs plane, and /tmp (where gates build
268// their fixtures). A reaper that will write anywhere is a much larger tool than this one is, and the
269// guard is checked against a POSITIVE CONTROL in the gate -- a guard that refuses everything passes
270// every negative test.
271func jr_dir_allowed(dir: *u8) -> i64 {
272 let n: i64 = jr_slen(dir)
273 if n == 0 { return 0 }
274 if jr_has(dir, n, JR_DOTDOT) == 1 { return 0 }
275 if jr_streq(dir, JR_JOBS) == 1 { return 1 }
276 if jr_prefix(dir, JR_TMP) == 1 { return 1 }
277 return 0
278}
279
280func jr_cat(d: *u8, o: i64, s: *u8) -> i64 {
281 var i: i64 = 0
282 var p: i64 = o
283 while s[i] != (0 as u8) { d[p] = s[i]; p = p + 1; i = i + 1 }
284 return p
285}
286
287func jr_catnum(d: *u8, o: i64, v: i64) -> i64 {
288 var p: i64 = o
289 var x: i64 = v
290 if x < 0 { d[p] = JR_CH_MINUS as u8; p = p + 1; x = 0 - x }
291 var div: i64 = 1
292 var t: i64 = x
293 while t >= JR_B10 { div = div * JR_B10; t = t / JR_B10 }
294 while div > 0 { d[p] = ((x / div) % JR_B10 + JR_CH_ZERO) as u8; p = p + 1; div = div / JR_B10 }
295 return p
296}
297
298// "<dir>/job_<id><ext>" into buf. CAP-CHECKED: returns -1 rather than truncating, because a
299// truncated path names a DIFFERENT file and the caller would then act on it.
300func jr_mkpath(buf: *u8, cap: i64, dir: *u8, id: i64, ext: *u8) -> i64 {
301 if id <= 0 { return 0 - 1 }
302 let dl: i64 = jr_slen(dir)
303 let el: i64 = jr_slen(ext)
304 let pl: i64 = jr_slen(JR_PREFIX)
305 var digits: i64 = 1
306 var t: i64 = id
307 while t >= JR_B10 { digits = digits + 1; t = t / JR_B10 }
308 if dl + 1 + pl + digits + el + 1 > cap { return 0 - 1 }
309 var p: i64 = 0
310 var i: i64 = 0
311 while i < dl { buf[p] = dir[i]; p = p + 1; i = i + 1 }
312 buf[p] = JR_CH_SLASH as u8; p = p + 1
313 p = jr_cat(buf, p, JR_PREFIX)
314 p = jr_catnum(buf, p, id)
315 p = jr_cat(buf, p, ext)
316 buf[p] = 0 as u8
317 return p
318}
319
320// ---------------------------------------------------------------------------------------------
321// APPEND THE TOMBSTONE. Never overwrite: rule 13, history is sacred, and the original marker is the
322// evidence any later investigation of the loss needs. Returns the bytes written, or -1 with the
323// claim UNCHANGED.
324// need_nl exists because a claim that does not end in a newline would otherwise have the tombstone
325// glued onto its last line -- still findable by substring, but unreadable to a human and a
326// different shape from every other row in the plane.
327func jr_tombstone(path: *u8, now: i64, age: i64, need_nl: i64, line: *u8, lcap: i64) -> i64 {
328 if lcap < JR_TOMB_MAX { return 0 - 1 }
329 var p: i64 = 0
330 if need_nl == 1 { line[p] = JR_CH_NL as u8; p = p + 1 }
331 p = jr_cat(line, p, JR_MARK_REAPED)
332 p = jr_cat(line, p, " ts=" as *u8)
333 p = jr_catnum(line, p, now)
334 p = jr_cat(line, p, " age=" as *u8)
335 p = jr_catnum(line, p, age)
336 p = jr_cat(line, p, "s by=" as *u8)
337 p = jr_cat(line, p, JR_SELF)
338 line[p] = JR_CH_NL as u8; p = p + 1
339 let fd: i64 = sys_openat_append(path, MODE_0644)
340 if fd < 0 { return 0 - 1 }
341 let w: i64 = sys_write(fd, line, p)
342 sys_fsync(fd)
343 sys_close(fd)
344 if w != p { return 0 - 1 }
345 return p
346}
347
348// ---------------------------------------------------------------------------------------------
349// THE DECISION, pure and total: every input maps to exactly one code.
350// ORDER: idempotence first (a tombstoned claim is finished business), then healthy history, then
351// "cannot judge", then the different-remedy case, then age. The age test is LAST so that every
352// earlier refusal is reported by its own name instead of being masked by "too young".
353func jr_decide(state: i64, age: i64, maxage: i64, has_out: i64) -> i64 {
354 if state == JR_ST_REAPED { return JR_D_ALREADY }
355 if state == JR_ST_DONE { return JR_D_DONE }
356 if state == JR_ST_UNKNOWN { return JR_D_UNPARSED }
357 if has_out == 1 { return JR_D_HAS_OUT }
358 if age <= maxage { return JR_D_TOO_YOUNG }
359 return JR_D_REAP
360}
361
362// Read one claim, fill info[], and decide. Writes nothing. On return pbuf holds the CLAIM path, so
363// the caller can act on exactly the file that was examined rather than recomposing it.
364func jr_examine(dir: *u8, id: i64, now: i64, maxage: i64, info: *i64, pbuf: *u8, pcap: i64, sc: *i64) -> i64 {
365 info[JR_I_STATE] = JR_ST_UNKNOWN
366 info[JR_I_AGE] = now - id
367 info[JR_I_HASOUT] = 0
368 info[JR_I_BYTES] = 0 - 1
369 info[JR_I_NEEDNL] = 0
370 if id <= 0 { return JR_D_BADID }
371 if jr_mkpath(pbuf, pcap, dir, id, JR_EXT_CLAIM) < 0 { return JR_D_PATHLONG }
372
373 // ABSENT and EMPTY are different answers and sys_read_file distinguishes them: a NULL return is
374 // "could not open", a non-NULL with len 0 is "opened, and it is empty". Folding them would make
375 // reaping a claim that never existed look exactly like reaping a real reservation.
376 let buf: *u8 = sys_read_file(pbuf, sc)
377 if (buf as i64) == 0 { return JR_D_ABSENT }
378 let n: i64 = sc[0]
379 info[JR_I_BYTES] = n
380 info[JR_I_STATE] = jr_state_of(buf, n)
381 if n > 0 { if buf[n-1] != (JR_CH_NL as u8) { info[JR_I_NEEDNL] = 1 } }
382 sys_free_file(buf, n)
383
384 if jr_mkpath(pbuf, pcap, dir, id, JR_EXT_OUT) < 0 { return JR_D_PATHLONG }
385 let ofd: i64 = sys_openat_rd(pbuf)
386 if ofd >= 0 { info[JR_I_HASOUT] = 1; sys_close(ofd) }
387
388 if jr_mkpath(pbuf, pcap, dir, id, JR_EXT_CLAIM) < 0 { return JR_D_PATHLONG }
389 return jr_decide(info[JR_I_STATE], info[JR_I_AGE], maxage, info[JR_I_HASOUT])
390}
391
392// Examine, and when eligible and not a dry run, tombstone. dry=1 decides and writes nothing, so the
393// same code path answers "what would you do" and "do it" -- a dry run that exercises different code
394// from the real one proves nothing about the real one.
395func jr_reap_one(dir: *u8, id: i64, now: i64, maxage: i64, dry: i64,
396 info: *i64, pbuf: *u8, pcap: i64, sc: *i64, line: *u8, lcap: i64) -> i64 {
397 let d: i64 = jr_examine(dir, id, now, maxage, info, pbuf, pcap, sc)
398 if d != JR_D_REAP { return d }
399 if dry == 1 { return JR_D_REAP }
400 if jr_tombstone(pbuf, now, info[JR_I_AGE], info[JR_I_NEEDNL], line, lcap) < 0 { return JR_D_WRITEFAIL }
401 return JR_D_REAP
402}
403
404// Exclusive claim reservation shared by dispatch paths. A positive return is
405// ownership of exactly path/id; failures never authorize launching a worker.
406// out[0] is a reserved id even if marker persistence failed (retained, never reused).
407// out[1] counts open attempts and out[2] counts EEXIST conflicts; out[3] counts orphan output slots.
408const JR_ID_MAX:i64=9223372036854775807
409const JR_ERR_EXIST:i64=0-17
410const JR_ERR_INTR:i64=0-4
411const JR_ERR_INVALID:i64=0-22
412const JR_ERR_OVERFLOW:i64=0-75
413const JR_ERR_DEADLINE:i64=0-110
414const JR_ERR_PATH:i64=0-36
415const JR_ERR_IO:i64=0-5
416func jr_reserve(dir:*u8,first:i64,deadline_ms:i64,path:*u8,pathcap:i64,line:*u8,linecap:i64,out:*i64)->i64{
417 out[0]=0;out[1]=0;out[2]=0;out[3]=0
418 if first<=0{return JR_ERR_INVALID}
419 if deadline_ms<=0{return JR_ERR_INVALID}
420 if jr_dir_allowed(dir)!=1{return JR_ERR_INVALID}
421 if linecap<jr_slen(JR_MARK_CLAIMED)+4+JR_I64_DIGITS+1{return JR_ERR_PATH}
422 var id:i64=first
423 while 1==1{
424 if sys_now_ms()>=deadline_ms{return JR_ERR_DEADLINE}
425 let output_slot:i64=jr_slot_occupied(dir,id,path,pathcap,JR_EXT_OUT)
426 if output_slot<0{return output_slot}
427 var occupied:i64=output_slot
428 if occupied==0{occupied=jr_slot_occupied(dir,id,path,pathcap,".tmp")}
429 if occupied<0{return occupied}
430 if occupied==1{
431 out[3]=out[3]+1
432 if id==JR_ID_MAX{return JR_ERR_OVERFLOW}
433 id=id+1
434 }else{
435 if jr_mkpath(path,pathcap,dir,id,JR_EXT_CLAIM)<0{return JR_ERR_PATH}
436 let fd:i64=sys_openat_exclusive(path,MODE_0644)
437 out[1]=out[1]+1
438 if fd>=0{
439 out[0]=id
440 var n:i64=jr_cat(line,0,JR_MARK_CLAIMED)
441 n=jr_cat(line,n," ts=");n=jr_catnum(line,n,sys_now_realtime_sec());line[n]=10;n=n+1
442 var off:i64=0
443 while off<n{
444 if sys_now_ms()>=deadline_ms{sys_close(fd);return JR_ERR_DEADLINE}
445 let w:i64=sys_write(fd,line+off,n-off)
446 if w>0{off=off+w}else{
447 if w!=JR_ERR_INTR{sys_close(fd);if w==0{return JR_ERR_IO};return w}
448 }
449 }
450 let closed:i64=sys_close(fd)
451 if closed<0{return closed}
452 return id
453 }
454 if fd==JR_ERR_EXIST{
455 out[2]=out[2]+1
456 if id==JR_ID_MAX{return JR_ERR_OVERFLOW}
457 id=id+1
458 }else{if fd!=JR_ERR_INTR{return fd}}
459 }
460 }
461 return JR_ERR_IO
462}
463
464func jr_slot_occupied(dir:*u8,id:i64,path:*u8,cap:i64,ext:*u8)->i64{
465 if jr_mkpath(path,cap,dir,id,ext)<0{return JR_ERR_PATH}
466 // NOFOLLOW prevents an old dangling symlink being mistaken for an unused slot.
467 let fd:i64=__syscall(SYS_OPENAT,AT_FDCWD,path,O_RDONLY|O_NOFOLLOW|O_NONBLOCK,0,0,0)
468 if fd>=0{let rc:i64=sys_close(fd);if rc<0{return rc};return 1}
469 if fd==(0-2){return 0}
470 if fd==(0-40){return 1}
471 return fd
472}