nx_capture_append_lib.nx source
↩ module page · 76 lines · 3534 B
1// Atomic immutable capture append: row, count and retry fingerprint in one segment commit.
2import "nx_store_seed_lib.nx"
3import "nx_sha256.nx"
4const CA_CONFLICT: i64 = -101
5const CA_CORRUPT: i64 = -102
6const CA_WRITE_FAILED: i64 = -103
7const CA_SCRATCH: i64 = 160
8const CA_HASH_BYTES: i64 = 64
9// Only ENOENT establishes an unseeded plane; unreadable or malformed existing history refuses.
10func ca_manifest_missing(prefix: *u8) -> i64 {
11 let suffix: *u8="manifest.txt" as *u8
12 let cap: i64=ss_len(prefix)+ss_len(suffix)+1
13 let path: *u8=sys_mmap(cap)
14 var n: i64=ss_cat(path,0,prefix); n=ss_cat(path,n,suffix); path[n]=0 as u8
15 let fd: i64=sys_openat_rd(path); sys_munmap(path,cap)
16 if fd>=0 { sys_close(fd); return 0 }
17 if fd == -2 { return 1 }
18 return 0
19}
20func ca_append_body(prefix: *u8,id: *u8,fingerprint: *u8,row: *u8,n: i64,tmp: *u8) -> i64 {
21 let key: *u8=tmp
22 key[0]=105 as u8; key[1]=100 as u8; key[2]=58 as u8
23 var i: i64=0; while i<CA_HASH_BYTES { key[3+i]=id[i]; i=i+1 }; key[67]=0 as u8
24 let box: *i64=(tmp as i64+72) as *i64
25 let rowkey: *u8=(tmp as i64+88) as *u8
26 let countbuf: *u8=(tmp as i64+120) as *u8
27 let gen: i64=ss_max_segid(prefix)
28 if gen<0 { if ca_manifest_missing(prefix)!=1 { return CA_CORRUPT } }
29 var h: *i64=0 as *i64
30 if gen>=0 { h=ss_open_cached(prefix) }
31 var count: i64=0
32 if (h as i64)!=0 {
33 if ss_hget(h,key,box,(box as i64+8) as *i64)==1 {
34 if box[1]!=CA_HASH_BYTES { return CA_CORRUPT }
35 let old: *u8=box[0] as *u8
36 i=0; while i<CA_HASH_BYTES { if old[i]!=fingerprint[i] { return CA_CONFLICT }; i=i+1 }
37 // A previous attempt may have published the manifest before its directory sync failed.
38 if ss_syncdir(prefix)!=0 { return CA_WRITE_FAILED }
39 return 0
40 }
41 if ss_hget(h,"q:n" as *u8,box,(box as i64+8) as *i64)!=1 { return CA_CORRUPT }
42 if box[1]<=0 || box[1]>19 { return CA_CORRUPT }
43 let cb: *u8=box[0] as *u8
44 i=0; while i<box[1] { let d: i64=cb[i] as i64-48; if d<0 || d>9 { return CA_CORRUPT }; if count>(9223372036854775806-d)/10 { return CA_CORRUPT }; count=count*10+d; i=i+1 }
45 } else { if gen>=0 { return CA_CORRUPT } }
46 sts_rowkey(count,rowkey)
47 let cn: i64=ss_catn(countbuf,0,count+1)
48 let w: *i64=ss_begin_cap(n+STS_WSLACK)
49 var rc: i64=CA_WRITE_FAILED
50 if ss_add(w,STS_KIND_LIVE,rowkey,row,n)>=0 {
51 if ss_add(w,STS_KIND_LIVE,"q:n" as *u8,countbuf,cn)>=0 {
52 if ss_add(w,STS_KIND_LIVE,key,fingerprint,CA_HASH_BYTES)>=0 {
53 let commit: i64=ss_commit_cas(prefix,w,ss_next_segid(prefix),gen)
54 if commit==0 { rc=count+1 } else { if commit==SS_ERR_STALE { rc=STS_ERR_STALE } }
55 }
56 }
57 }
58 sys_munmap(w[0] as *u8,w[2]); sys_munmap(w as *u8,32)
59 return rc
60}
61func ca_append_once(prefix: *u8,id: *u8,fingerprint: *u8,row: *u8,n: i64) -> i64 {
62 if n<=0 { return CA_CORRUPT }
63 let lock: i64=sts_lock(prefix); if lock<0 { return STS_ERR_LOCK }
64 let tmp: *u8=sys_mmap(CA_SCRATCH)
65 let rc: i64=ca_append_body(prefix,id,fingerprint,row,n,tmp)
66 sys_munmap(tmp,CA_SCRATCH); sts_unlock(lock); return rc
67}
68
69func ca_fingerprint(c: *Sha256,raw: *u8,out: *u8,domain: *u8,a: *u8,an: i64,b: *u8,bn: i64) {
70 sha256_init(c); sha256_update(c,domain,ss_len(domain)); sha256_update(c,a,an)
71 sha256_update(c,"\n" as *u8,1); sha256_update(c,b,bn); sha256_final(c,raw)
72 let hex: *u8="0123456789abcdef"
73 var i: i64=0; while i<32 { let v: i64=raw[i] as i64; out[i*2]=hex[v/16]; out[i*2+1]=hex[v%16]; i=i+1 }; out[64]=0 as u8
74 sys_munmap(c.bufptr as *u8,64); sys_munmap(c.kptr as *u8,512); sys_munmap(c.wptr as *u8,512)
75 sys_munmap(c.k32ptr as *u8,256); sys_munmap(c.st8ptr as *u8,32)
76}