code wiki / _hdl_build / nx_apistack_idempotency.nx
`nx_apistack_idempotency.nx` source

↩ module page · 245 lines · 11245 B
1// Retained idempotency evidence. Sequential replay is not an exactly-once transaction:
2// mutation and journal commit remain separate. Missing or malformed evidence is not NEW.
3//
4// THE READ LOCK IS SHARED (2026-09-16, debt 1789503583). A reader used to take LOCK_EX|LOCK_NB, so a concurrent
5// reader, or the recorder inside its one append+fsync, made every keyed tools/call fail closed as EVIDENCE_UNAVAILABLE
6// and the tools API could not be re-shipped from this tree. Readers now take LOCK_SH and wait: a recorder holds LOCK_EX
7// for exactly one append and one fsync, and flock releases on death, so the wait is bounded by one write and is never
8// held by another reader. Blocking is chosen over a retry budget on purpose: a budget is a magic number that fails
9// closed under exactly the load that makes the wait long, and the old immediate refusal WAS the defect.
10// MALFORMED ROWS ARE QUARANTINED, NOT POISONOUS (same day). One row with a CR, a NUL, two TABs, no TAB, or a torn tail
11// used to refuse the whole ledger for every key. A malformed row is now skipped and COUNTED (r.quarantined, with the
12// first bad line in r.first_bad_line, both handed out by id_lookup_q) -- EXCEPT when its key field is the key being
13// looked up: evidence about the sought key that cannot be read is not NEW, so that one lookup still fails closed
14// (EBADMSG) exactly as before. A torn last row (no trailing newline) is a torn append and is judged by the same rule.
15import "nx_syscalls.nx"
16import "nx_fio.nx"
17const ID_NOT_FOUND:i64=0-1
18const ID_EVIDENCE_ERROR:i64=0-2
19const ID_MISSING:i64=0-3
20const ID_CAPACITY:i64=0-4
21const ID_RESOURCE:i64=0-5
22const ID_SIGNED_MAX:i64=9223372036854775807
23const ID_O_RDONLY:i64=0
24const ID_O_RDWR_APPEND_CREATE:i64=1090
25const ID_O_NONBLOCK:i64=2048
26const ID_O_NOFOLLOW:i64=131072
27const ID_O_CLOEXEC:i64=524288
28struct NxIdEvidence{
buffer:*u8,bytes:i64,allocated:i64,result:*u8,result_bytes:i64,state:i64,system_code:i64,quarantined:i64,first_bad_line:i64
30}
31func id_evidence_init(r:*NxIdEvidence)->i64{
r.buffer=0 as *u8;r.bytes=0;r.allocated=0;r.result=0 as *u8
r.result_bytes=0;r.state=ID_EVIDENCE_ERROR;r.system_code=0;r.quarantined=0;r.first_bad_line=0;return 0
34}
35func id_evidence_close(r:*NxIdEvidence)->i64{
if (r.buffer as i64)>0{sys_munmap_direct(r.buffer,r.allocated)}
r.buffer=0 as *u8;r.result=0 as *u8;r.allocated=0;return 0
38}
39func id_field_valid(p:*u8,n:i64,key:i64)->i64{
if n<0 || (p as i64)<=0{return 0}
if key==1&&n==0{return 0}
var i:i64=0;while i<n{
if p[i]==0 as u8 || p[i]==9 as u8 || p[i]==10 as u8 || p[i]==13 as u8{return 0};i=i+1
};return 1
45}
46func id_read_evidence(path:*u8,key:*u8,key_n:i64,r:*NxIdEvidence)->i64{
id_evidence_init(r)
if (path as i64)<=0 || id_field_valid(key,key_n,1)==0{return r.state}
let stat:*u8=sys_mmap_try(NX_STAT_X64_BYTES)
if (stat as i64)<=0{r.state=ID_RESOURCE;return r.state}
let fd:i64=__syscall(257,AT_FDCWD,path,ID_O_RDONLY|ID_O_NONBLOCK|ID_O_NOFOLLOW|ID_O_CLOEXEC,0,0,0)
if fd<0{
sys_munmap_direct(stat,NX_STAT_X64_BYTES);r.system_code=fd
if fd==0-2{r.state=ID_MISSING};return r.state
}
var rc:i64=sys_flock(fd,SYS_LOCK_SH)
if rc==0{rc=sys_fstat_fd(fd,stat)}
var n:i64=0
if rc==0{
let mode:i64=*((stat+NX_STAT_X64_MODE_OFFSET) as *i64)
n=*((stat+NX_STAT_X64_SIZE_OFFSET) as *i64)
if (mode&NX_STAT_TYPE_MASK)!=NX_STAT_REGULAR_FILE || n<0 || n==ID_SIGNED_MAX{rc=FIO_EINVAL}
}
if rc==0{
r.allocated=n+1;r.buffer=sys_mmap_try(r.allocated)
if (r.buffer as i64)<=0{r.buffer=0 as *u8;r.allocated=0;r.state=ID_RESOURCE;rc=0-12}
}
var got:i64=0
while rc==0&&got<n{
let z:i64=sys_read(fd,r.buffer+got,n-got)
if z==FIO_EINTR{continue}
if z<=0{rc=z;if z==0{rc=FIO_EIO}}else{got=got+z}
}
let unlock:i64=sys_flock(fd,SYS_LOCK_UN);let closed:i64=sys_close(fd)
sys_munmap_direct(stat,NX_STAT_X64_BYTES)
if rc==0&&unlock!=0{rc=unlock};if rc==0&&closed!=0{rc=closed}
if rc!=0{r.system_code=rc;id_evidence_close(r);return r.state}
r.bytes=n;r.buffer[n]=0 as u8
var tail_torn:i64=0
if n>0{if r.buffer[n-1]!=10 as u8{tail_torn=1}}
var start:i64=0;var found:i64=0;var chosen:i64=0;var chosen_n:i64=0
var line:i64=0;var bad:i64=0;var first_bad:i64=0;var bad_hit:i64=0
while start<n{
line=line+1
var end:i64=start;while end<n&&r.buffer[end]!=10 as u8{end=end+1}
var malformed:i64=0
if end>=n{if tail_torn==1{malformed=1}}
if end>start&&r.buffer[start]!=35 as u8{
 var tab:i64=0-1;var i:i64=start
 while i<end{
  let c:u8=r.buffer[i]
  if c==0 as u8||c==13 as u8{malformed=1}
  if c==9 as u8{if tab>=0{malformed=1}else{tab=i}}
  i=i+1
 }
 if tab<=start{malformed=1}
 if malformed==1{
  bad=bad+1;if first_bad==0{first_bad=line}
  var kend:i64=end;if tab>start{kend=tab}
  if kend-start==key_n{var same:i64=1;i=0;while i<key_n{if r.buffer[start+i]!=key[i]{same=0};i=i+1};if same==1{bad_hit=1}}
 }else{
  if found==0&&tab-start==key_n{
   var same:i64=1;i=0;while i<key_n{if r.buffer[start+i]!=key[i]{same=0};i=i+1}
   if same==1{found=1;chosen=tab+1;chosen_n=end-tab-1}
  }
 }
}else{if malformed==1{bad=bad+1;if first_bad==0{first_bad=line}}}
start=end+1
}
r.quarantined=bad;r.first_bad_line=first_bad
if bad_hit==1{r.system_code=FIO_EBADMSG;return r.state}
if found==0{r.state=ID_NOT_FOUND;return r.state}
r.result=r.buffer+chosen;r.result_bytes=chosen_n;r.result[chosen_n]=0 as u8
r.state=chosen_n;return r.state
115}
116// Legacy caller-buffer API: success returns complete result length; -1 alone means unseen.
117// Other negative values require reconciliation; no truncated success or out-of-bounds terminator.
118// id_lookup_q additionally hands out the quarantine count and the first quarantined line (q[0], q[1]) so a caller can
119// announce what it skipped; q may be 0 when the caller does not want them.
120func id_lookup_q(path:*u8,key:*u8,key_n:i64,out:*u8,cap:i64,q:*i64)->i64{
if cap<=0||(out as i64)<=0{return ID_CAPACITY};out[0]=0 as u8
if (q as i64)>0{q[0]=0;q[1]=0}
let r:*NxIdEvidence=sys_mmap_try(__size_of(NxIdEvidence)) as *NxIdEvidence
if (r as i64)<=0{return ID_RESOURCE}
var rc:i64=id_read_evidence(path,key,key_n,r)
if (q as i64)>0{q[0]=r.quarantined;q[1]=r.first_bad_line}
if rc>=0{
if rc>=cap{rc=ID_CAPACITY}else{
 var i:i64=0;while i<rc{out[i]=r.result[i];i=i+1};out[rc]=0 as u8
}
}
id_evidence_close(r);sys_munmap_direct(r as *u8,__size_of(NxIdEvidence));return rc
133}
134func id_lookup(path:*u8,key:*u8,key_n:i64,out:*u8,cap:i64)->i64{
return id_lookup_q(path,key,key_n,out,cap,0 as *i64)
136}
137func id_seen(path:*u8,key:*u8,key_n:i64)->i64{
let r:*NxIdEvidence=sys_mmap_try(__size_of(NxIdEvidence)) as *NxIdEvidence
if (r as i64)<=0{return ID_RESOURCE}
let rc:i64=id_read_evidence(path,key,key_n,r)
id_evidence_close(r);sys_munmap_direct(r as *u8,__size_of(NxIdEvidence))
if rc>=0{return 1};if rc==ID_NOT_FOUND{return 0};return rc
143}
144// A lock protects the complete append against cooperating recorders. It does not cover
145// a caller's prior lookup or mutation, nor make a crashed mutation exactly-once.
146func id_record(path:*u8,key:*u8,key_n:i64,result:*u8,result_n:i64)->i64{
if (path as i64)<=0||id_field_valid(key,key_n,1)==0||id_field_valid(result,result_n,0)==0{return FIO_EINVAL}
if key_n>ID_SIGNED_MAX-result_n-2{return ID_CAPACITY}
let bytes:i64=key_n+result_n+2
let row:*u8=sys_mmap_try(bytes)
let stat:*u8=sys_mmap_try(NX_STAT_X64_BYTES)
if (row as i64)<=0||(stat as i64)<=0{
if (row as i64)>0{sys_munmap_direct(row,bytes)}
if (stat as i64)>0{sys_munmap_direct(stat,NX_STAT_X64_BYTES)}
return ID_RESOURCE
}
var i:i64=0;while i<key_n{row[i]=key[i];i=i+1};row[key_n]=9 as u8
i=0;while i<result_n{row[key_n+1+i]=result[i];i=i+1};row[bytes-1]=10 as u8
let fd:i64=__syscall(257,AT_FDCWD,path,ID_O_RDWR_APPEND_CREATE|ID_O_NONBLOCK|ID_O_NOFOLLOW|ID_O_CLOEXEC,MODE_0644,0,0)
var rc:i64=fd
if fd>=0{
rc=sys_flock(fd,SYS_LOCK_EX)
if rc==0{rc=sys_fstat_fd(fd,stat)}
if rc==0{
 let mode:i64=*((stat+NX_STAT_X64_MODE_OFFSET) as *i64)
 let n:i64=*((stat+NX_STAT_X64_SIZE_OFFSET) as *i64)
 if (mode&NX_STAT_TYPE_MASK)!=NX_STAT_REGULAR_FILE||n<0{rc=FIO_EINVAL}
 if rc==0&&n>0{
  if sys_lseek(fd,n-1,0)!=n-1{rc=FIO_EIO}else{
   let got:i64=sys_read(fd,stat,1);if got!=1||stat[0]!=10 as u8{rc=FIO_EBADMSG}
  }
 }
}
var written:i64=0
while rc==0&&written<bytes{
 let z:i64=sys_write(fd,row+written,bytes-written)
 if z==FIO_EINTR{continue}
 if z<=0{rc=z;if z==0{rc=FIO_EIO}}else{written=written+z}
}
if rc==0{rc=sys_fsync(fd)}
let unlock:i64=sys_flock(fd,SYS_LOCK_UN);let closed:i64=sys_close(fd)
if rc==0&&unlock!=0{rc=unlock};if rc==0&&closed!=0{rc=closed}
}
sys_munmap_direct(row,bytes);sys_munmap_direct(stat,NX_STAT_X64_BYTES)
if rc==0{return 1};return rc
186}
187
188
189// Lossless legacy row inspection only. Shape cannot prove a request boundary:
190// four unlocked legacy writes and an embedded TAB can produce identical bytes.
191// Offsets refer to the unchanged source snapshot; no row is repaired or skipped.
192// Flags describe syntax, not permission to execute or replay an unknown request.
193const ID_ROW_EMPTY:i64=1
194const ID_ROW_COMMENT:i64=2
195const ID_ROW_NO_TAB:i64=4
196const ID_ROW_EXTRA_TAB:i64=8
197const ID_ROW_EMPTY_KEY:i64=16
198const ID_ROW_CONTROL:i64=32
199const ID_ROW_UNTERMINATED:i64=64
200struct NxIdLegacyRow{
start:i64,end:i64,next:i64,flags:i64,tabs:i64,first_tab:i64,second_tab:i64,
legacy_key_start:i64,legacy_key_bytes:i64,legacy_result_start:i64,legacy_result_bytes:i64
203}
204// Returns 1 for a row, 0 at exact EOF, negative for invalid memory/extent.
205// Caller owns disjoint row storage; source may have no sentinel and is never written.
206func id_legacy_row_next(source:*u8,bytes:i64,start:i64,row:*NxIdLegacyRow)->i64{
let base:i64=source as i64;let rp:i64=row as i64
let rn:i64=__size_of(NxIdLegacyRow)
if bytes<0||start<0||start>bytes||rp<=0{return FIO_EINVAL}
if rp>ID_SIGNED_MAX-rn{return FIO_EINVAL}
if bytes>0{
if base<=0{return FIO_EINVAL}
if base>ID_SIGNED_MAX-bytes{return FIO_EINVAL}
if rp<base+bytes&&base<rp+rn{return FIO_EINVAL}
}
row.start=start;row.end=start;row.next=start;row.flags=0;row.tabs=0
row.first_tab=0-1;row.second_tab=0-1
row.legacy_key_start=start;row.legacy_key_bytes=0
row.legacy_result_start=start;row.legacy_result_bytes=0
if start==bytes{return 0}
var end:i64=start
while end<bytes{
let ch:u8=source[end]
if ch==10 as u8{break}
if ch==9 as u8{
 row.tabs=row.tabs+1
 if row.first_tab<0{row.first_tab=end}else{if row.second_tab<0{row.second_tab=end}}
}
if ch==0 as u8||ch==13 as u8{row.flags=row.flags|ID_ROW_CONTROL}
end=end+1
}
row.end=end;row.next=end
if end<bytes{row.next=end+1}else{row.flags=row.flags|ID_ROW_UNTERMINATED}
if end==start{row.flags=row.flags|ID_ROW_EMPTY;return 1}
if source[start]==35 as u8{row.flags=row.flags|ID_ROW_COMMENT;return 1}
row.legacy_key_bytes=end-start;row.legacy_result_start=end
if row.first_tab<0{row.flags=row.flags|ID_ROW_NO_TAB;return 1}
row.legacy_key_bytes=row.first_tab-start
row.legacy_result_start=row.first_tab+1
var result_end:i64=end
if row.second_tab>=0{result_end=row.second_tab;row.flags=row.flags|ID_ROW_EXTRA_TAB}
row.legacy_result_bytes=result_end-row.legacy_result_start
if row.legacy_key_bytes==0{row.flags=row.flags|ID_ROW_EMPTY_KEY}
return 1
245}
nx_apistack_idempotency.nx source

`nx_apistack_idempotency.nx` source
