nx_contentput_reap.nx source
↩ module page · 314 lines · 15428 B
1// nx_contentput_reap.nx -- CENSUS of content-upload staging. IT DOES NOT DELETE, AND THAT IS A
2// CORRECTION MADE ON THE DAY IT SHIPPED.
3//
4// I BUILT THIS AS A REAPER AND IT WAS A DUPLICATE. nx_content_put ALREADY reaps -- v_reap/cp_reap, in
5// the producer itself -- with a BETTER bound than the flat TTL this organ carried: a per-transfer
6// deadline derived from nchunks x worst_call_ms x resume_allowance, so a large transfer earns more
7// time, and it already abstains on an untrustworthy age. The coverage argument for a second reaper does
8// not survive either: index.txt carries 278 rows against 50 directory metas, so the index the incumbent
9// walks is a SUPERSET of the directory this one walks.
10//
11// HOW THE DUPLICATE GOT BUILT, recorded because the search WAS run and still missed it: nx_spendgate
12// and nx_capsearch rank REGISTERED TOOLS, and this capability lives as a VERB INSIDE AN ORGAN, where no
13// tool-name search can see it. The words `abort` and `reap` sat in nx_content_put's own usage string
14// the whole time. BEFORE BUILDING ANYTHING THAT ACTS ON ANOTHER ORGAN'S DATA, READ THAT ORGAN'S USAGE
15// LINE -- a capability search over names is not a capability search over behaviour.
16//
17// SO THE DELETE PATH IS GONE AND ONE RULER OWNS DELETION. What is kept is the half the incumbent does
18// not do: a DIRECTORY-walk census that prints a partition and reconciles it, which is how the stale
19// rows were found in the first place. Observing and deleting are different jobs; only the second has to
20// be unique.
21//
22// nx_contentput_reap [--dir <path>] [ttl_seconds] (READ-ONLY -- --apply is REFUSED)
23//
24// WHY IT EXISTS, MEASURED NOT ASSUMED. nx_content_put allocates a transfer on `begin` and writes its
25// staging under knowledge/contentput/<id>.meta, then one <id>.c<n> per chunk. A transfer that never
26// reaches `commit` leaves that state behind and ANNOUNCES NOTHING -- the leak is invisible until
27// somebody lists the directory. Measured 2026-09-04: 82 entries, roughly 40 of them BARE .meta files
28// from begins that never received a single chunk, the oldest about eleven days back, with no reaper
29// anywhere in the estate.
30//
31// THE INFLOW CAUSE IS ALREADY FIXED, AND SAYING SO CHANGES WHAT THIS ORGAN IS FOR. Those begins failed
32// because the shared TLS transport wrote every request as ONE record, capping a body at 8 KB against
33// the 48,402-byte chunks the server offers -- so every large put died at its first chunk. That is fixed
34// (nx_tls13_frag_len, gate nx_tls13_frag_gate). This organ therefore drains HISTORICAL debt plus
35// whatever future transients leave; it is not a workaround for a live defect, and it must not be read
36// as one.
37//
38// THE INDUSTRY BAR IS AN EXPIRY, NOT A SWEEP. S3 lifecycle AbortIncompleteMultipartUpload expires
39// unfinished multipart uploads after a declared number of days, and the OCI Distribution Spec gives an
40// upload session a timeout. Both are DECLARED durations, which is why the TTL here is read from conf
41// and its provenance printed rather than picked in code.
42//
43// WRONG IN THE DIRECTION OF DOING NOTHING, BY CONSTRUCTION:
44// - DRY-RUN IS THE DEFAULT. Deleting requires --apply, spelled out, every time.
45// - A TRANSFER HOLDING ANY CHUNK IS NEVER REAPED, whatever its age. A resumable upload that is paused
46// is indistinguishable from an abandoned one by age alone, and the resume rung (DL10) will make
47// paused transfers NORMAL -- so the chunk test is what keeps this organ correct as that lands.
48// - A TRANSFER YOUNGER THAN THE TTL IS NEVER REAPED, so a put in flight right now cannot be hit.
49// - An unreadable stat ABSTAINS (counted UNKNOWN, never reaped): a stat that failed is not evidence
50// of age, and reaping on a failed read is the acquit-on-no-evidence defect with a delete attached.
51//
52// THE PARTITION SUMS AND IS PRINTED, because a partition you cannot reconcile is a leak -- which is
53// exactly what this organ exists to drain.
54//
55// exit: 0 ok | 2 usage | 3 staging dir unreadable
56// license_tier: ORIGINAL No hw writes (Rule 26). expect_exit: 0
57import "nx_syscalls.nx"
58
59const CR_DIR: *u8 = "knowledge/contentput"
60const CR_CONF: *u8 = "knowledge/status/contentput_reap.conf"
61// Default TTL when no conf row exists: 7 days. DERIVED, and the derivation is stated so the next reader
62// can argue with it -- the oldest observed abandoned meta was about eleven days old, and the longest a
63// legitimate transfer has ever taken on this estate is minutes, so a week is three orders of magnitude
64// of headroom over the real work and still drains the fossils. Provenance is PRINTED on every run.
65const CR_TTL_DEFAULT: i64 = 604800
66const CR_DIRBUF: i64 = 65536
67const CR_PATH: i64 = 2048
68const CR_NAMECAP: i64 = 256
69const CR_CONFCAP: i64 = 8192
70const CR_LIST_CAP: i64 = 20
71
72func cr_slen(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
73func cr_out(s: *u8) -> i64 { sys_write(1, s, cr_slen(s)); return 0 }
74func cr_num(v: i64) -> i64 {
75 let b: *u8 = sys_mmap(32)
76 var m: i64 = v
77 var neg: i64 = 0
78 if m < 0 { neg = 1; m = 0 - m }
79 var i: i64 = 31
80 if m == 0 { i = i - 1; b[i] = 48 as u8 }
81 while m > 0 { i = i - 1; b[i] = (48 + (m % 10)) as u8; m = m / 10 }
82 if neg == 1 { i = i - 1; b[i] = 45 as u8 }
83 sys_write(1, (b as i64 + i) as *u8, 31 - i)
84 return 0
85}
86func cr_streq(a: *u8, b: *u8) -> i64 {
87 var i: i64 = 0
88 while a[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 }
89 if b[i] != (0 as u8) { return 0 }
90 return 1
91}
92func cr_ends_with(name: *u8, ext: *u8) -> i64 {
93 let nl: i64 = cr_slen(name)
94 let el: i64 = cr_slen(ext)
95 if el > nl { return 0 }
96 var i: i64 = 0
97 while i < el { if name[nl - el + i] != ext[i] { return 0 } i = i + 1 }
98 return 1
99}
100// dir + "/" + name, NUL-terminated
101func cr_join(dir: *u8, name: *u8, out: *u8) -> i64 {
102 var o: i64 = 0
103 var a: i64 = 0
104 while dir[a] != (0 as u8) { out[o] = dir[a]; o = o + 1; a = a + 1 }
105 out[o] = 47 as u8
106 o = o + 1
107 a = 0
108 while name[a] != (0 as u8) { out[o] = name[a]; o = o + 1; a = a + 1 }
109 out[o] = 0 as u8
110 return o
111}
112// name minus its trailing ext, into out
113func cr_strip_ext(name: *u8, ext: *u8, out: *u8) -> i64 {
114 let keep: i64 = cr_slen(name) - cr_slen(ext)
115 var i: i64 = 0
116 while i < keep { out[i] = name[i]; i = i + 1 }
117 out[keep] = 0 as u8
118 return keep
119}
120func cr_exists(path: *u8) -> i64 {
121 let sb: *u8 = sys_mmap(256)
122 if sys_fstatat(path, sb) < 0 { return 0 }
123 return 1
124}
125// mtime seconds, or -1 when the stat fails. -1 must ABSTAIN at every call site.
126func cr_mtime(path: *u8) -> i64 {
127 let sb: *u8 = sys_mmap(256)
128 if sys_fstatat(path, sb) < 0 { return 0 - 1 }
129 let sw: *i64 = sb as *i64
130 return sw[11]
131}
132func cr_now() -> i64 {
133 let ts: *i64 = sys_mmap(32) as *i64
134 sys_clock_gettime_real(ts)
135 return ts[0]
136}
137// first integer after `key` in the conf, or -1. Anchored on the key text; a conf with no such row
138// leaves the caller on its declared default rather than on a silent zero.
139func cr_conf_int(buf: *u8, n: i64, key: *u8) -> i64 {
140 let m: i64 = cr_slen(key)
141 var i: i64 = 0
142 while i + m <= n {
143 var j: i64 = 0
144 var same: i64 = 1
145 while j < m { if buf[i + j] != key[j] { same = 0; j = m } else { j = j + 1 } }
146 if same == 1 {
147 var p: i64 = i + m
148 var v: i64 = 0
149 var got: i64 = 0
150 var go: i64 = 1
151 while go == 1 {
152 if p >= n { go = 0 } else {
153 let c: i64 = buf[p] as i64
154 if c >= 48 { if c <= 57 { v = v * 10 + (c - 48); got = 1; p = p + 1 } else { go = 0 } } else { go = 0 }
155 }
156 }
157 if got == 1 { return v }
158 }
159 i = i + 1
160 }
161 return 0 - 1
162}
163
164func main(argc: i64, argv: *i64) -> i64 {
165 var apply: i64 = 0
166 var ttl_arg: i64 = 0 - 1
167 // AN ORGAN THAT CAN ONLY EVER LOOK AT ONE HARDCODED DIRECTORY CANNOT BE GATED, and a reaper whose
168 // only test subject is the LIVE staging tree is one nobody dares bite-prove. --dir lets a gate aim
169 // it at a per-run fixture; it defaults to the production path so no caller changes.
170 var dir: *u8 = CR_DIR
171 var ai: i64 = 1
172 while ai < argc {
173 let a: *u8 = argv[ai] as *u8
174 if cr_streq(a, "--apply" as *u8) == 1 { apply = 1 }
175 else { if cr_streq(a, "--dir" as *u8) == 1 { ai = ai + 1; if ai < argc { dir = argv[ai] as *u8 } }
176 else {
177 var v: i64 = 0
178 var k: i64 = 0
179 var ok: i64 = 1
180 while a[k] != (0 as u8) {
181 let c: i64 = a[k] as i64
182 if c < 48 { ok = 0 } else { if c > 57 { ok = 0 } else { v = v * 10 + (c - 48) } }
183 k = k + 1
184 }
185 if ok == 1 { if k > 0 { ttl_arg = v } }
186 } }
187 ai = ai + 1
188 }
189
190 // TTL provenance is PRINTED, never assumed: a bound whose origin the reader cannot see is a bound
191 // nobody can argue with, and this one gates a delete.
192 var ttl: i64 = CR_TTL_DEFAULT
193 var prov: *u8 = "DEFAULT-derived" as *u8
194 let cbox: *i64 = sys_mmap(16) as *i64
195 let cb: *u8 = sys_read_file(CR_CONF, cbox)
196 if (cb as i64) != 0 {
197 let cv: i64 = cr_conf_int(cb, cbox[0], "ttl_s" as *u8)
198 if cv > 0 { ttl = cv; prov = "CONF" as *u8 }
199 }
200 if ttl_arg > 0 { ttl = ttl_arg; prov = "ARGV" as *u8 }
201
202 cr_out("NX-CONTENTPUT-REAP dir=" as *u8); cr_out(dir)
203 cr_out(" ttl_s=" as *u8); cr_num(ttl)
204 cr_out(" ttl_provenance=" as *u8); cr_out(prov)
205 // REFUSE RATHER THAN BECOME A SECOND DELETER. A caller asking to delete is asking for the
206 // incumbent; censusing silently instead would let them believe a reap had happened.
207 if apply == 1 {
208 cr_out(" mode=REFUSED\n" as *u8)
209 cr_out("REFUSED: this organ does not delete. Deletion belongs to `nx_content_put reap`, which owns\n" as *u8)
210 cr_out(" the transfer index and derives a PER-TRANSFER deadline from nchunks rather than applying a\n" as *u8)
211 cr_out(" flat TTL, and which already abstains on an untrustworthy age. Two reapers over one store is\n" as *u8)
212 cr_out(" the duplicate-ruler defect and they WILL drift apart.\n" as *u8)
213 cr_out(" Run: nx_content_put reap -- then re-run this organ to confirm the partition moved.\n" as *u8)
214 return 2
215 }
216 cr_out(" mode=CENSUS (read-only by construction; deletion belongs to nx_content_put reap)" as *u8)
217 cr_out("\n" as *u8)
218
219 let fd: i64 = sys_openat_rd(dir)
220 if fd < 0 {
221 cr_out("REFUSED: staging dir unreadable -- nothing was examined, so this run proves NOTHING about the leak\n" as *u8)
222 return 3
223 }
224
225 let dbuf: *u8 = sys_mmap(CR_DIRBUF)
226 let path: *u8 = sys_mmap(CR_PATH)
227 let path2: *u8 = sys_mmap(CR_PATH)
228 let idbuf: *u8 = sys_mmap(CR_NAMECAP)
229 let now: i64 = cr_now()
230
231 var n_meta: i64 = 0
232 var n_chunks: i64 = 0
233 var n_young: i64 = 0
234 var n_reapable: i64 = 0
235 var n_unknown: i64 = 0
236 var n_reaped: i64 = 0
237 var shown: i64 = 0
238
239 // ONE getdents64 CALL IS NOT A DIRECTORY LISTING -- loop until it returns 0, or a big directory is
240 // silently read as a prefix and its total published as a fact.
241 var run: i64 = 1
242 while run == 1 {
243 let got: i64 = sys_getdents64(fd, dbuf, CR_DIRBUF)
244 if got <= 0 { run = 0 } else {
245 var off: i64 = 0
246 while off < got {
247 let rec: *u8 = ((dbuf as i64 + off) as *u8)
248 let reclen: i64 = dirent_reclen(rec)
249 if reclen <= 0 { off = got } else {
250 let name: *u8 = dirent_name(rec)
251 if cr_ends_with(name, ".meta" as *u8) == 1 {
252 n_meta = n_meta + 1
253 cr_strip_ext(name, ".meta" as *u8, idbuf)
254 cr_join(dir, name, path)
255 let mt: i64 = cr_mtime(path)
256 // chunk 0 is the discriminator: a transfer that received ANY chunk is work in
257 // progress, and age cannot tell a paused upload from an abandoned one.
258 var c0: *u8 = sys_mmap(CR_NAMECAP)
259 var q: i64 = 0
260 while idbuf[q] != (0 as u8) { c0[q] = idbuf[q]; q = q + 1 }
261 c0[q] = 46 as u8; c0[q+1] = 99 as u8; c0[q+2] = 48 as u8; c0[q+3] = 0 as u8
262 cr_join(dir, c0, path2)
263 if cr_exists(path2) == 1 { n_chunks = n_chunks + 1 }
264 else {
265 if mt < 0 { n_unknown = n_unknown + 1 }
266 else {
267 let age: i64 = now - mt
268 if age < ttl { n_young = n_young + 1 }
269 else {
270 n_reapable = n_reapable + 1
271 if shown < CR_LIST_CAP {
272 shown = shown + 1
273 cr_out(" REAPABLE id=" as *u8); cr_out(idbuf)
274 cr_out(" age_s=" as *u8); cr_num(age)
275 cr_out(" chunks=0\n" as *u8)
276 }
277 if apply == 1 {
278 // .shas first, then .meta: the meta is the row's identity, so
279 // removing it last means an interrupted reap leaves a transfer
280 // that is still CLASSIFIED, never a chunk index with no owner.
281 var sh: *u8 = sys_mmap(CR_NAMECAP)
282 var z: i64 = 0
283 while idbuf[z] != (0 as u8) { sh[z] = idbuf[z]; z = z + 1 }
284 sh[z] = 46 as u8; sh[z+1] = 115 as u8; sh[z+2] = 104 as u8
285 sh[z+3] = 97 as u8; sh[z+4] = 115 as u8; sh[z+5] = 0 as u8
286 cr_join(dir, sh, path2)
287 if cr_exists(path2) == 1 { sys_unlinkat(path2) }
288 if sys_unlinkat(path) == 0 { n_reaped = n_reaped + 1 }
289 }
290 }
291 }
292 }
293 }
294 off = off + reclen
295 }
296 }
297 }
298 }
299 sys_close(fd)
300
301 if shown >= CR_LIST_CAP {
302 cr_out(" <== THE LIST ABOVE IS A PREFIX OF ITS OWN COUNT: it stopped at the cap so this organ stays inside a bounded capture. reapable= below is the POPULATION.\n" as *u8)
303 }
304 cr_out("metas=" as *u8); cr_num(n_meta)
305 cr_out(" has_chunks=" as *u8); cr_num(n_chunks)
306 cr_out(" too_young=" as *u8); cr_num(n_young)
307 cr_out(" reapable=" as *u8); cr_num(n_reapable)
308 cr_out(" unknown_stat=" as *u8); cr_num(n_unknown)
309 cr_out(" reaped=" as *u8); cr_num(n_reaped)
310 let sum: i64 = n_chunks + n_young + n_reapable + n_unknown
311 cr_out(" partition_sum=" as *u8); cr_num(sum)
312 if sum == n_meta { cr_out(" RECONCILES=yes\n" as *u8) } else { cr_out(" RECONCILES=NO -- a class is missing, do not act on these numbers\n" as *u8) }
313 return 0
314}