nx_compare_receipt_lib_t138.nx source
↩ module page · 309 lines · 12712 B
1// Private receipt transaction for nx_compare_regen; not adopted.
2// Existing completion rows remain byte-compatible. Sidecar intents are append-only evidence.
3// A PREPARED/PUBLISHED intent without COMPLETED requires reconciliation; this is not automatic replay.
4import "nx_jrnl_archive_lib_t138.nx"
5const RGR_I64_MAX: i64 = 9223372036854775807
6const RGR_DECIMAL_BYTES: i64 = 20
7const RGR_ROW_PUNCTUATION_BYTES: i64 = 12
8const RGR_NO_EXTRA_RESERVE: i64 = 0
9const RGR_E_INPUT: i64 = 0-101
10const RGR_E_LOCK: i64 = 0-102
11const RGR_E_CAPACITY: i64 = 0-103
12const RGR_E_TAIL: i64 = 0-104
13const RGR_E_INTENT: i64 = 0-105
14const RGR_E_DURABLE: i64 = 0-106
15const RGR_E_WRITE: i64 = 0-107
16const RGR_E_RENAME: i64 = 0-108
17const RGR_E_COMPLETE: i64 = 0-109
18struct RgrTx {
19 lockfd: i64
20 intentfd: i64
21 row: *u8
22 rowcap: i64
23 rowlen: i64
24 intentpath: *u8
25 pathcap: i64
26 journal: *u8
27 published: i64
28 available: i64
29 required: i64
30 reserve: i64
31 detail: i64
32}
33// Receipt allocations bypass the small bump arena so release remains effective across rows.
34func rgr_alloc(n: i64) -> *u8 {
35 var actual:i64=n
36 if actual<=NXA_SMALL_MAX {actual=NXA_SMALL_MAX+1}
37 return sys_mmap_shared(actual)
38}
39func rgr_free(b: *u8, n: i64) -> i64 {
40 var actual:i64=n
41 if actual<=NXA_SMALL_MAX {actual=NXA_SMALL_MAX+1}
42 return sys_munmap(b,actual)
43}
44func rgr_len(s: *u8) -> i64 { return ja_slen(s) }
45func rgr_copy(d: *u8, off: i64, s: *u8) -> i64 { return ja_cat(d, off, s) }
46func rgr_decimal(d: *u8, off: i64, v: i64) -> i64 {
47 var x: i64 = v
48 var o: i64 = off
49 if x < 0 { d[o] = 45 as u8; o = o+1; x = 0-x }
50 let first: i64 = o
51 if x == 0 { d[o] = 48 as u8; o = o+1 }
52 while x > 0 { d[o] = (48+x%10) as u8; o=o+1; x=x/10 }
53 var a: i64 = first
54 var b: i64 = o-1
55 while a < b { let c: u8 = d[a]; d[a]=d[b]; d[b]=c; a=a+1; b=b-1 }
56 d[o] = 0 as u8
57 return o
58}
59func rgr_field(s: *u8) -> i64 {
60 if (s as i64) <= 0 { return 0 }
61 var i: i64 = 0
62 while s[i] != (0 as u8) { if s[i] < (32 as u8) { return 0 } i=i+1 }
63 if i == 0 { return 0 }
64 return 1
65}
66func rgr_id(s: *u8) -> i64 {
67 if rgr_field(s) == 0 { return 0 }
68 var i: i64=0
69 while s[i] != (0 as u8) {
70 let c: i64=s[i] as i64
71 var ok: i64=0
72 if c>=48 { if c<=57 {ok=1} }
73 if c>=65 { if c<=90 {ok=1} }
74 if c>=97 { if c<=122 {ok=1} }
75 if c==45 {ok=1}
76 if ok==0 {return 0}
77 i=i+1
78 }
79 return 1
80}
81// The write loop's result, not a capacity snapshot, is authoritative. EINTR is retryable.
82// EAGAIN/ENOSPC/zero writes are explicit failures; callers retain any partial evidence.
83func rgr_write_all(fd: i64, b: *u8, n: i64) -> i64 {
84 if n < 0 { return RGR_E_INPUT }
85 var off: i64=0
86 while off<n {
87 let rc: i64=sys_write(fd,(b as i64+off) as *u8,n-off)
88 if rc != (0-4) {
89 if rc<=0 { if rc==0 {return RGR_E_WRITE} return rc }
90 if rc>n-off {return RGR_E_WRITE}
91 off=off+rc
92 }
93 }
94 return 0
95}
96func rgr_sync_dir(path: *u8) -> i64 {
97 let cap: i64=rgr_len(path)+2
98 let d: *u8=rgr_alloc(cap)
99 if (d as i64)<=0 {return RGR_E_DURABLE}
100 ja_dirname(path,d)
101 let fd: i64=sys_openat_directory(d)
102 rgr_free(d,cap)
103 if fd<0 {return fd}
104 let rc: i64=sys_fsync(fd)
105 let cc: i64=sys_close(fd)
106 if rc!=0 {return rc}
107 return cc
108}
109// Validate arithmetic independently of the host statfs call so failure gates exercise it.
110func rgr_capacity_numbers(block: i64, blocks: i64, reserve: i64, pending: i64) -> i64 {
111 if block<=0 {return RGR_E_CAPACITY}
112 if blocks<0 {return RGR_E_CAPACITY}
113 if reserve<0 {return RGR_E_CAPACITY}
114 if pending<0 {return RGR_E_CAPACITY}
115 if blocks>RGR_I64_MAX/block {return RGR_E_CAPACITY}
116 let avail: i64=block*blocks
117 if reserve>avail {return RGR_E_CAPACITY}
118 if pending>avail-reserve {return RGR_E_CAPACITY}
119 return avail
120}
121func rgr_capacity(tx: *RgrTx, path: *u8, pending: i64) -> i64 {
122 tx.required=pending
123 let sc:i64=STATFS_BUF_BYTES
124 let st:*i64=rgr_alloc(sc) as *i64
125 if (st as i64)<=0 {return RGR_E_CAPACITY}
126 let rc:i64=sys_statfs(path,st)
127 var admitted:i64=RGR_E_CAPACITY
128 if rc==0 {
129 let measured:i64=rgr_capacity_numbers(st[STATFS_I_BSIZE],st[STATFS_I_BAVAIL],0,0)
130 if measured>=0 {tx.available=measured}
131 admitted=rgr_capacity_numbers(st[STATFS_I_BSIZE],st[STATFS_I_BAVAIL],tx.reserve,pending)
132 }
133 rgr_free(st as *u8,sc)
134 tx.detail=rc
135 if admitted<0 {return RGR_E_CAPACITY}
136 return 0
137}
138// Only inspect the final byte. Refuse to append across a torn row; never erase history.
139func rgr_tail(journal: *u8) -> i64 {
140 let fd: i64=sys_openat_rd(journal)
141 if fd==(0-2) {return 0}
142 if fd<0 {return RGR_E_TAIL}
143 let size: i64=sys_lseek(fd,0,2)
144 if size<0 {sys_close(fd);return RGR_E_TAIL}
145 var rc: i64=0
146 if size>0 {
147 let b: *u8=rgr_alloc(1)
148 if (b as i64)<=0 {sys_close(fd);return RGR_E_TAIL}
149 if sys_lseek(fd,size-1,0)!=(size-1) {rc=RGR_E_TAIL}
150 if rc==0 {if sys_read(fd,b,1)!=1 {rc=RGR_E_TAIL}}
151 if rc==0 {if b[0]!=(10 as u8) {rc=RGR_E_TAIL}}
152 rgr_free(b,1)
153 }
154 let cc: i64=sys_close(fd)
155 if cc!=0 {return RGR_E_TAIL}
156 return rc
157}
158func rgr_release(tx: *RgrTx) -> i64 {
159 var rc: i64=0
160 if tx.intentfd>=0 {rc=sys_close(tx.intentfd);tx.intentfd=0-1}
161 if tx.lockfd>=0 {let lc:i64=ja_unlock(tx.lockfd);if rc==0 {rc=lc} tx.lockfd=0-1}
162 if (tx.row as i64)>0 {rgr_free(tx.row,tx.rowcap);tx.row=0 as *u8}
163 if (tx.intentpath as i64)>0 {rgr_free(tx.intentpath,tx.pathcap);tx.intentpath=0 as *u8}
164 return rc
165}
166func rgr_prepare(tx: *RgrTx, journal: *u8, target: *u8, body: *u8, bytes: i64, prev: i64, epoch: i64, id: *u8, reserve: i64) -> i64 {
167 tx.lockfd=0-1;tx.intentfd=0-1;tx.row=0 as *u8;tx.intentpath=0 as *u8
168 tx.published=0;tx.available=0-1;tx.required=0;tx.reserve=reserve;tx.detail=0;tx.journal=journal
169 if rgr_field(journal)==0 {return RGR_E_INPUT}
170 if rgr_field(target)==0 {return RGR_E_INPUT}
171 if rgr_id(id)==0 {return RGR_E_INPUT}
172 if bytes<0 {return RGR_E_INPUT}
173 if prev<(0-1) {return RGR_E_INPUT}
174 if epoch<0 {return RGR_E_INPUT}
175 if reserve<0 {return RGR_E_INPUT}
176 if (body as i64)<=0 {return RGR_E_INPUT}
177 tx.rowcap=rgr_len(target)+3*RGR_DECIMAL_BYTES+RGR_ROW_PUNCTUATION_BYTES
178 tx.row=rgr_alloc(tx.rowcap)
179 if (tx.row as i64)<=0 {return RGR_E_INPUT}
180 var o:i64=rgr_decimal(tx.row,0,epoch)
181 o=rgr_copy(tx.row,o,"\t" as *u8);o=rgr_copy(tx.row,o,target)
182 o=rgr_copy(tx.row,o,"\t" as *u8);o=rgr_decimal(tx.row,o,bytes)
183 o=rgr_copy(tx.row,o,"\t" as *u8);o=rgr_decimal(tx.row,o,prev)
184 if prev>=0 {if bytes<prev {o=rgr_copy(tx.row,o,"\tSHRANK" as *u8)}}
185 tx.rowlen=rgr_copy(tx.row,o,"\n" as *u8)
186 tx.pathcap=rgr_len(journal)+rgr_len(id)+rgr_len(".intent.")+1
187 tx.intentpath=rgr_alloc(tx.pathcap)
188 if (tx.intentpath as i64)<=0 {return RGR_E_INPUT}
189 o=rgr_copy(tx.intentpath,0,journal);o=rgr_copy(tx.intentpath,o,".intent." as *u8);rgr_copy(tx.intentpath,o,id)
190 tx.lockfd=ja_lock(journal)
191 if tx.lockfd<0 {tx.detail=tx.lockfd;return RGR_E_LOCK}
192 let tail:i64=rgr_tail(journal)
193 if tail!=0 {return tail}
194 let dc:i64=rgr_len(journal)+2
195 let dir:*u8=rgr_alloc(dc)
196 if (dir as i64)<=0 {return RGR_E_CAPACITY}
197 ja_dirname(journal,dir)
198 // Three file data allocations: prepared intent, completion row, appended final intent state.
199 // Include filesystem allocation granularity; metadata/quota/concurrent writers remain syscall checks.
200 let st:*i64=rgr_alloc(STATFS_BUF_BYTES) as *i64
201 if (st as i64)<=0 {rgr_free(dir,dc);return RGR_E_CAPACITY}
202 let sr:i64=sys_statfs(dir,st)
203 let block:i64=st[STATFS_I_BSIZE]
204 rgr_free(st as *u8,STATFS_BUF_BYTES)
205 if sr!=0 {rgr_free(dir,dc);tx.detail=sr;return RGR_E_CAPACITY}
206 let labels:*u8="schema=1\nstate=PREPARED\nid=\nartifact_sha256=\nreserve_extra_bytes=\navailable_bavail_bytes=\nrequired_data_bytes=\ncompletion_row="
207 let mc:i64=rgr_len(labels)+rgr_len(id)+64+3*RGR_DECIMAL_BYTES+tx.rowlen+1
208 if block<=0 {rgr_free(dir,dc);return RGR_E_CAPACITY}
209 if block>RGR_I64_MAX/3 {rgr_free(dir,dc);return RGR_E_CAPACITY}
210 let base:i64=mc+tx.rowlen+rgr_len("state=PUBLISHED\nstate=COMPLETED\n")
211 if base>RGR_I64_MAX-block*3 {rgr_free(dir,dc);return RGR_E_CAPACITY}
212 let pending:i64=base+block*3
213 let cp:i64=rgr_capacity(tx,dir,pending)
214 rgr_free(dir,dc)
215 if cp!=0 {return cp}
216 let meta:*u8=rgr_alloc(mc)
217 let digest:*u8=rgr_alloc(32)
218 let hex:*u8=rgr_alloc(65)
219 if (meta as i64)<=0 || (digest as i64)<=0 || (hex as i64)<=0 {
220 if (meta as i64)>0 {rgr_free(meta,mc)}
221 if (digest as i64)>0 {rgr_free(digest,32)}
222 if (hex as i64)>0 {rgr_free(hex,65)}
223 return RGR_E_INTENT
224 }
225 let hr:i64=sha256_digest_checked_native(body,bytes,digest)
226 if hr!=0 {rgr_free(meta,mc);rgr_free(digest,32);rgr_free(hex,65);return RGR_E_INTENT}
227 ja_hex(digest,hex)
228 o=rgr_copy(meta,0,"schema=1\nstate=PREPARED\nid=" as *u8);o=rgr_copy(meta,o,id)
229 o=rgr_copy(meta,o,"\nartifact_sha256=" as *u8);o=rgr_copy(meta,o,hex)
230 o=rgr_copy(meta,o,"\nreserve_extra_bytes=" as *u8);o=rgr_decimal(meta,o,reserve)
231 o=rgr_copy(meta,o,"\navailable_bavail_bytes=" as *u8);o=rgr_decimal(meta,o,tx.available)
232 o=rgr_copy(meta,o,"\nrequired_data_bytes=" as *u8);o=rgr_decimal(meta,o,tx.required)
233 o=rgr_copy(meta,o,"\ncompletion_row=" as *u8);o=rgr_copy(meta,o,tx.row)
234 tx.intentfd=sys_openat_exclusive(tx.intentpath,420)
235 var wr:i64=tx.intentfd
236 if tx.intentfd>=0 {
237 wr=rgr_write_all(tx.intentfd,meta,o)
238 if wr==0 {wr=sys_fsync(tx.intentfd)}
239 if wr==0 {wr=rgr_sync_dir(tx.intentpath)}
240 }
241 rgr_free(meta,mc);rgr_free(digest,32);rgr_free(hex,65)
242 tx.detail=wr
243 if wr!=0 {return RGR_E_INTENT}
244 return 0
245}
246func rgr_state(tx: *RgrTx, state: *u8) -> i64 {
247 let rc:i64=rgr_write_all(tx.intentfd,state,rgr_len(state))
248 if rc!=0 {return rc}
249 return sys_fsync(tx.intentfd)
250}
251func rgr_complete(tx: *RgrTx) -> i64 {
252 let ps:i64=rgr_state(tx,"state=PUBLISHED\n" as *u8)
253 if ps!=0 {tx.detail=ps;return RGR_E_COMPLETE}
254 let fd:i64=sys_openat_append(tx.journal,420)
255 if fd<0 {tx.detail=fd;return RGR_E_COMPLETE}
256 var rc:i64=rgr_write_all(fd,tx.row,tx.rowlen)
257 if rc==0 {rc=sys_fsync(fd)}
258 let cc:i64=sys_close(fd)
259 if rc==0 {rc=cc}
260 if rc==0 {rc=rgr_sync_dir(tx.journal)}
261 if rc==0 {rc=rgr_state(tx,"state=COMPLETED\n" as *u8)}
262 tx.detail=rc
263 if rc!=0 {return RGR_E_COMPLETE}
264 return 0
265}
266// Atomic visibility and durability are separate observations. Never remove the prior live file.
267static g_rgr_atomic_sequence:i64
268func rgr_write_atomic(path: *u8, buf: *u8, n: i64, published: *i64, detail: *i64) -> i64 {
269 published[0]=0;detail[0]=0
270 if rgr_field(path)==0 {return RGR_E_INPUT}
271 if n<0 {return RGR_E_INPUT}
272 let cap:i64=rgr_len(path)+rgr_len(".new.")+2*RGR_DECIMAL_BYTES+2
273 let tmp:*u8=rgr_alloc(cap)
274 if (tmp as i64)<=0 {detail[0]=tmp as i64;return RGR_E_WRITE}
275 var o:i64=rgr_copy(tmp,0,path);o=rgr_copy(tmp,o,".new." as *u8)
276 o=rgr_decimal(tmp,o,sys_now_realtime_us());o=rgr_copy(tmp,o,"-" as *u8)
277 g_rgr_atomic_sequence=g_rgr_atomic_sequence+1
278 rgr_decimal(tmp,o,g_rgr_atomic_sequence)
279 let fd:i64=sys_openat_exclusive(tmp,420)
280 if fd<0 {detail[0]=fd;rgr_free(tmp,cap);return RGR_E_WRITE}
281 var rc:i64=rgr_write_all(fd,buf,n)
282 if rc==0 {rc=sys_fsync(fd)}
283 let cc:i64=sys_close(fd)
284 if rc==0 {rc=cc}
285 if rc!=0 {detail[0]=rc;rgr_free(tmp,cap);return RGR_E_WRITE}
286 rc=sys_renameat(tmp,path)
287 rgr_free(tmp,cap)
288 if rc!=0 {detail[0]=rc;return RGR_E_RENAME}
289 published[0]=1
290 rc=rgr_sync_dir(path)
291 if rc!=0 {detail[0]=rc;return RGR_E_DURABLE}
292 return 0
293}
294// Returns prepare/write/complete stage errors while keeping rename visibility in tx.published.
295func rgr_transact(tx: *RgrTx, journal: *u8, target: *u8, body: *u8, bytes: i64, prev: i64, epoch: i64, id: *u8, reserve: i64) -> i64 {
296 let prep:i64=rgr_prepare(tx,journal,target,body,bytes,prev,epoch,id,reserve)
297 if prep!=0 {return prep}
298 let visible:*i64=rgr_alloc(16) as *i64
299 if (visible as i64)<=0 {return RGR_E_WRITE}
300 let wr:i64=rgr_write_atomic(target,body,bytes,visible,(visible as i64+8) as *i64)
301 tx.published=visible[0];tx.detail=visible[1]
302 rgr_free(visible as *u8,16)
303 if wr!=0 {
304 if tx.published==0 {rgr_state(tx,"state=WRITE_FAILED_BEFORE_RENAME\n" as *u8)}
305 if tx.published==1 {rgr_state(tx,"state=PUBLISHED_DURABILITY_INCOMPLETE\n" as *u8)}
306 return wr
307 }
308 return rgr_complete(tx)
309}