nx_atomic_publish.nx source

1// nx_atomic_publish.nx -- ALL-OR-NOTHING FILE PUBLISH WITH COMPARE-AND-SWAP, IN NISHILANG. 2// 3// WHY IT EXISTS. A multi-part source change applied as N sequential anchored edits is N chances to 4// half-land. Measured twice on 2026-08-07: the MCP daemon wedged mid-sequence, edit 1 landed and edits 5// 2-3 did not, leaving a live source with an UNCLOSED BRACE -- a broken file that blocks every build of 6// that target, found only because the artefact was verified rather than the response believed. 7// ★★★★★★A MULTI-PART EDIT OVER A FLAKY TRANSPORT IS AN ATOMICITY PROBLEM, NOT A RETRY PROBLEM: 8// retrying makes a half-applied file MORE likely, not less. Stage the whole file, publish once. 9// 10// WHY IT IS AN ORGAN AND NOT A SHELL HELPER. The first version of this was a .ps1, and it JUDGED: 11// it parsed a refusal to recover a token, classified transient-vs-deterministic, and decided 12// refuse-vs-retry. That is judgement, and judgement belongs in NishiLang -- shell is the wire and 13// nothing else. Here the only laptop-side act left is moving bytes to a scratch path; every decision 14// happens on this side. 15// 16// usage: nx_atomic_publish <staged-src> <target> <expect> 17// expect = "any" -- publish regardless (NAME IT OUT LOUD when you use it) 18// | "absent" -- publish only if the target does not exist yet 19// | <64-hex sha256> -- publish only if the target's CURRENT content hashes to exactly this 20// nx_atomic_publish landed <writelist> -- the READ half: adjudicate ambiguous writes (ap_landed) 21// 22// CAS SEMANTICS, and the direction of failure is the point: on mismatch it REFUSES and PRINTS the live 23// hash, so the caller can re-read, merge, and retry with the right token. It never re-resolves the token 24// itself -- that would turn a concurrent writer's edit into a silent clobber, which is the exact defect 25// compare-and-swap exists to prevent. 26// ★A CAS THAT RESOLVES ITS OWN EXPECTED VALUE IS NOT A CAS, IT IS AN OVERWRITE WITH EXTRA STEPS. 27// 28// ATOMIC BY renameat, NOT BY WRITE ORDER: content goes to <target>.aptmp then renames over the target, 29// so a reader never sees a half-written file and a crash mid-publish leaves the original intact. 30// VERIFY AFTER: the published bytes are re-read and re-hashed before GREEN is printed. 31// license_tier: ORIGINAL expect_exit: 0 32import "nx_syscalls.nx" 33import "nx_sha256.nx" 34 35const AP_STDOUT: i64 = 1 36const AP_PATHCAP: i64 = 1024 37const AP_MODE: i64 = 420 // 0644 38 39func ap_puts(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(AP_STDOUT, s, n); return 0 } 40 41func ap_len(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n } 42 43func ap_num(v: i64) -> i64 { 44 let b: *u8 = sys_mmap(32) 45 var m: i64 = v 46 if m < 0 { m = 0 - m; sys_write(AP_STDOUT, "-" as *u8, 1) } 47 let t: *u8 = sys_mmap(32) 48 var k: i64 = 0 49 if m == 0 { t[0] = 48 as u8; k = 1 } 50 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 } 51 var i: i64 = 0 52 while i < k { b[i] = t[k - 1 - i]; i = i + 1 } 53 sys_write(AP_STDOUT, b, k) 54 return 0 55} 56 57// 32 raw digest bytes -> 64 lowercase hex, NUL terminated. 58func ap_hex(dig: *u8, out: *u8) -> i64 { 59 let d: *u8 = "0123456789abcdef" as *u8 60 var i: i64 = 0 61 while i < 32 { 62 let v: i64 = dig[i] as i64 63 out[i * 2] = d[(v / 16) % 16] 64 out[i * 2 + 1] = d[v % 16] 65 i = i + 1 66 } 67 out[64] = 0 as u8 68 return 64 69} 70 71func ap_streq(a: *u8, b: *u8) -> i64 { 72 var i: i64 = 0 73 while a[i] != (0 as u8) { if a[i] != b[i] { return 0 } i = i + 1 } 74 if b[i] != (0 as u8) { return 0 } 75 return 1 76} 77 78func ap_cat2(out: *u8, a: *u8, b: *u8) -> i64 { 79 var o: i64 = 0 80 var i: i64 = 0 81 while a[i] != (0 as u8) { out[o] = a[i]; o = o + 1; i = i + 1 } 82 i = 0 83 while b[i] != (0 as u8) { out[o] = b[i]; o = o + 1; i = i + 1 } 84 out[o] = 0 as u8 85 return o 86} 87 88// Hash a file into `hex` and hand the caller the bytes it just read; returns the byte count, or -1 when 89// the file cannot be read. 90// ⚠SILENT CAP REMOVED 2026-08-20: this used to copy the file into a fixed 8 MiB scratch buffer before 91// hashing, so any file over that ceiling wrote past the allocation -- a LIMIT-class defect that fails as 92// corruption rather than as a refusal. sha256_digest does not mutate its input, so the copy bought 93// nothing; sys_read_file already sizes its own buffer from the file and cannot short-read. There is now 94// no ceiling to guess in either direction. 95func ap_hash_file(path: *u8, outp: *i64, hex: *u8) -> i64 { 96 let szp: *i64 = sys_mmap(16) as *i64 97 let b: *u8 = sys_read_file(path, szp) 98 if (b as i64) == 0 { return 0 - 1 } 99 let n: i64 = szp[0] 100 let dig: *u8 = sys_mmap(32) 101 sha256_digest(b, n, dig) 102 ap_hex(dig, hex) 103 outp[0] = b as i64 104 return n 105} 106 107// --------------------------------------------------------------------------------------------------- 108// ap_landed -- THE CRASH-WRITE ADJUDICATOR (DM4 on /compare/devmgmt). 109// 110// WHY IT LIVES HERE. This organ already owns the compare-and-swap publish: it knows what a settled 111// artifact is, it can hash one, and its refusal path is the estate's most authoritative "what is 112// actually there right now" answer. Adjudicating "did my write land?" is the READ half of the same 113// question and belongs beside the WRITE half, not in a second ruler. 114// 115// WHAT IT MECHANIZES. On 2026-08-19 two crashes killed thirteen seats mid-action and the in-flight 116// async writes split roughly half landed / half not; every lane adjudicated its own rows BY HAND from 117// artifacts. The same week produced four writes that returned 503-or-nothing and HAD landed, three the 118// same hour that had NOT, a bare empty-object reply that had landed, a debt filed twice because one 119// journal search returned zero, and a whole-function insert that only the CAS guard caught as a 120// duplicate. 121// 122// THE EVIDENCE HIERARCHY, AND THE DIRECTION OF ERROR IS THE POINT. 123// A false LANDED is a lost edit nobody ever re-applies. A false NOT-LANDED costs one idempotent 124// retry. So LANDED is emitted from EXACTLY TWO SITES in this code, and both require a SELF-COMPUTED 125// measurement over a SETTLED artifact matched against an expectation the caller recorded BEFORE the 126// write. Nothing else can reach them -- that is structural, not a promise: 127// tier 1 CONTENT HASH want=h<sha256> authoritative; we compute the digest ourselves 128// tier 2 MARKER COUNT mark= plus mark_pre= a witness; the only tier that can see DOUBLE-APPLIED 129// tier 3 PRE-IMAGE ONLY pre=h<sha256> can only ever prove NOT-LANDED, never LANDED 130// tier 4 NOTHING DECIDABLE -> UNKNOWN with a named reason 131// ⛔A BARE SIZE READ IS NOT EVIDENCE and neither is the transport's reply text. `size=` and `reply=` 132// are accepted, ECHOED so the reader sees what was believed, and NEVER consulted. Measured four times 133// on 2026-08-20: a size read taken right after a dropped write reports the PRE-write size, and an 134// empty / 503 / session-expired reply discriminated nothing in either direction. The record carries 135// the controlled pairs that settle it: the SAME 503 text over LANDED and NOT-LANDED writes in the same 136// hour, and the same bare empty-object reply over both outcomes in one lane. 137// 138// THE IMPRECISION CHOSEN, STATED SO THE NEXT READER DOES NOT TRUST THIS AS EXACT. A create whose target 139// reads absent in BOTH passes is called NOT-LANDED. That can be wrong: a read taken through a caching 140// layer reported ABSENT three times on 2026-08-20 for a file that had landed. It is wrong in the SAFE 141// direction (one idempotent create-retry, refused by the guard if it turns out to exist), this organ 142// reads the filesystem directly rather than through a caching tool, and a transient absence is caught by 143// the two-pass settle. It is still an imprecision and it is named here rather than left to be trusted. 144// 145// SETTLING IS A PRECONDITION, NOT A REFINEMENT. Every path is hashed TWICE -- once in pass A over the 146// whole list, once in pass B after a single settle interval -- because "the artifact says it did not 147// land" is not proof it did not land while a writer is still in flight. Two reads that disagree are 148// UNSETTLED and get UNKNOWN, never a verdict. The interval is paid ONCE for the entire worklist, not 149// per row, so the cost does not scale with the list. 150// settle_ms is READ FROM CONF and its source is announced. The default is not taste: it adopts the 151// sibling calibration in nx_memvel, whose sampling window exists for exactly this purpose (detect a 152// change over an interval longer than the event being watched) rather than inventing a fresh budget. 153// 154// ROW GRAMMAR -- one row per line, TAB-separated key=value tokens; blank lines and lines starting with 155// a double slash are skipped. 156// id=<slug> required 157// path=<file> required -- the artifact to read back 158// kind=publish|promote|create|replace|insert|append selects the RETRY ADVICE, never the verdict 159// want=h<64hex> the sha256 the file was meant to have AFTER the write 160// pre=h<64hex>|absent what the file was BEFORE it ("absent" for a create) 161// mark=<string> a distinctive string the write INTRODUCES (anchored/append kinds) 162// mark_pre=<n> how many times mark occurred BEFORE the write -- REQUIRED with mark, never assumed 163// mark_adds=<n> occurrences one correct apply adds (default APL_DEFAULT_MARK_ADDS, announced) 164// size=<n> reply=<t> RECORDED AND IGNORED, on purpose (see above) 165// An unrecognised token makes the row MALFORMED -> UNKNOWN. A row we cannot parse is a row we must 166// not judge. 167// 168// exit: 0 every row decided | 1 at least one UNKNOWN (a human must look) | 2 usage | 3 list unreadable 169// --------------------------------------------------------------------------------------------------- 170const APL_TAB: i64 = 9 171const APL_NL: i64 = 10 172const APL_CR: i64 = 13 173const APL_EQ: i64 = 61 174const APL_HCHAR: i64 = 104 // the 'h' prefix of an h<hex> compare-and-swap token 175const APL_SLASH: i64 = 47 // '/' -- a doubled slash starts a comment row in the grammar 176const APL_D0: i64 = 48 // ASCII '0' -- the base of decimal parsing, structural 177const APL_D9: i64 = 57 178const APL_DEC_BASE: i64 = 10 // every count in the row grammar is written in decimal 179const APL_NUL_SLACK: i64 = 2 // room to NUL-terminate the working copy AND its final line 180const APL_RC_LIST_UNREADABLE: i64 = 3 // the writelist itself could not be read: not a verdict about any row 181const APL_HEXSLOT: i64 = 80 // 64 hex + NUL, rounded up to a comfortable per-row slot 182const APL_PTRW: i64 = 8 // one i64 per parallel-array element 183const APL_DEFAULT_MARK_ADDS: i64 = 1 // one anchored apply introduces its marker once; stated, not assumed 184// DERIVATION, not a chosen number: nx_memvel samples change over 1200 ms windows precisely because a 185// window must outlast the event it watches. This is the same question (is this file being written right 186// now?) so it reuses that sibling calibration instead of inventing a second budget. 187const APL_SETTLE_MS_DEFAULT: i64 = 1200 188const APL_CONF: *u8 = "knowledge/status/atomic_publish.conf" 189 190const APL_V_LANDED: i64 = 0 191const APL_V_NOT: i64 = 1 192const APL_V_DOUBLE: i64 = 2 193const APL_V_UNKNOWN: i64 = 3 194 195// UNKNOWN reasons -- two causes with different remedies never share a counter 196const APL_R_NONE: i64 = 0 197const APL_R_UNSETTLED: i64 = 1 198const APL_R_UNREADABLE: i64 = 2 199const APL_R_WANT_EQ_PRE: i64 = 3 200const APL_R_THIRD_STATE: i64 = 4 201const APL_R_MARK_FELL: i64 = 5 202const APL_R_MARK_PARTIAL: i64 = 6 203const APL_R_CHANGED_UNDECLARED: i64 = 7 204const APL_R_SIZE_ONLY: i64 = 8 205const APL_R_REPLY_ONLY: i64 = 9 206const APL_R_NO_EXPECTATION: i64 = 10 207const APL_R_MALFORMED: i64 = 11 208const APL_R_MARK_NO_PRE: i64 = 12 209const APL_R_COUNT: i64 = 13 // one past the highest reason code; the counter table size 210 211func apl_reason_name(r: i64) -> *u8 { 212 if r == APL_R_UNSETTLED { return "unsettled-two-reads-differ" as *u8 } 213 if r == APL_R_UNREADABLE { return "artifact-unreadable" as *u8 } 214 if r == APL_R_WANT_EQ_PRE { return "want-equals-pre-indistinguishable" as *u8 } 215 if r == APL_R_THIRD_STATE { return "third-state-content-matches-neither" as *u8 } 216 if r == APL_R_MARK_FELL { return "marker-count-fell-below-pre" as *u8 } 217 if r == APL_R_MARK_PARTIAL { return "marker-count-between-pre-and-applied" as *u8 } 218 if r == APL_R_CHANGED_UNDECLARED { return "changed-but-target-content-undeclared" as *u8 } 219 if r == APL_R_SIZE_ONLY { return "size-only-not-evidence" as *u8 } 220 if r == APL_R_REPLY_ONLY { return "reply-text-not-evidence" as *u8 } 221 if r == APL_R_NO_EXPECTATION { return "no-expectation-recorded" as *u8 } 222 if r == APL_R_MALFORMED { return "malformed-row-unrecognised-or-missing-token" as *u8 } 223 if r == APL_R_MARK_NO_PRE { return "mark-without-mark-pre" as *u8 } 224 return "unnamed" as *u8 225} 226 227// RETRY ADVICE IS A PROPERTY OF THE EDIT KIND, NOT OF THE VERDICT. An insert is not idempotent: its 228// context anchor survives the apply, so a blind retry double-inserts -- and a stale multi-region diff 229// invalidates its own later anchors, measured on the 21-region ladder of 2026-08-20. 230// ⚠THE ANCHORED-REPLACE CAVEAT IS LOAD-BEARING AND WAS NEARLY LEFT OUT. The three-state retry table 231// (OK means it had not landed / NOMATCH means it had) is only sound while the replacement cannot 232// re-match its own anchor. Measured 2026-08-20: an edit whose `new` string CONTAINED its `old` anchor 233// would have inserted the same function twice on a blind retry. The advice carries the condition rather 234// than the conclusion, because the conclusion is false for a shape this estate actually writes. 235func apl_advice(kind: *u8) -> *u8 { 236 if (kind as i64) == 0 { return "KIND-UNDECLARED cannot advise on retry safety -- declare kind= before acting" as *u8 } 237 if ap_streq(kind, "publish" as *u8) == 1 { return "RETRY-SAFE re-issue the whole-file publish under expect=h<now_sha256> -- a CAS publish is idempotent" as *u8 } 238 if ap_streq(kind, "create" as *u8) == 1 { return "RETRY-SAFE re-issue with expect=absent -- the guard refuses if it turns out to exist" as *u8 } 239 if ap_streq(kind, "promote" as *u8) == 1 { return "RETRY-SELF-ADJUDICATING re-issue the promote -- NOTHING-STAGED with live_sha256 equal to your build sha means it IS live" as *u8 } 240 if ap_streq(kind, "register" as *u8) == 1 { return "RETRY-SAFE registration is idempotent -- re-issue and read callable= from the response" as *u8 } 241 if ap_streq(kind, "replace" as *u8) == 1 { return "RETRY-EXACT-CONDITIONAL re-issue the identical anchored edit ONLY IF the replacement cannot re-match its own anchor -- then OK means it had not landed and NOMATCH means it had; if the new text CONTAINS the old, treat it as an insert and RE-DIFF instead" as *u8 } 242 if ap_streq(kind, "insert" as *u8) == 1 { return "RE-DIFF do NOT re-issue -- an insert is not idempotent, its anchor survives the apply and a blind retry double-inserts" as *u8 } 243 if ap_streq(kind, "append" as *u8) == 1 { return "RE-DIFF do NOT re-issue -- an append is not idempotent; check the marker count against the journal first" as *u8 } 244 if ap_streq(kind, "put" as *u8) == 1 { return "CHECK-THE-PLANE-FIRST do NOT blind-retry -- a plane put appends a row, two puts leave two rows, and there is no row-delete primitive to undo the second" as *u8 } 245 return "KIND-UNDECLARED cannot advise on retry safety -- declare kind= before acting" as *u8 246} 247 248func apl_verdict_name(v: i64) -> *u8 { 249 if v == APL_V_LANDED { return "LANDED" as *u8 } 250 if v == APL_V_NOT { return "NOT-LANDED" as *u8 } 251 if v == APL_V_DOUBLE { return "DOUBLE-APPLIED" as *u8 } 252 return "UNKNOWN" as *u8 253} 254 255func apl_slot(base: *u8, i: i64, w: i64) -> *u8 { return ((base as i64) + i * w) as *u8 } 256 257// value side of `key=value` when the token's key is EXACTLY `key`, else 0. Exactness matters: it is what 258// keeps `mark=` from swallowing `mark_pre=` and `mark_adds=`. 259func apl_kv(tok: *u8, key: *u8) -> *u8 { 260 var i: i64 = 0 261 while key[i] != (0 as u8) { if tok[i] != key[i] { return 0 as *u8 } i = i + 1 } 262 if (tok[i] as i64) != APL_EQ { return 0 as *u8 } 263 return ((tok as i64) + i + 1) as *u8 264} 265 266// strip the 'h' of an h<hex> CAS token. Unambiguous: 'h' is not a hex digit, so a bare digest can never 267// be mistaken for a prefixed one. 268func apl_dehex(s: *u8) -> *u8 { 269 if (s[0] as i64) == APL_HCHAR { return ((s as i64) + 1) as *u8 } 270 return s 271} 272 273func apl_atoi(s: *u8) -> i64 { 274 var v: i64 = 0 275 var i: i64 = 0 276 while s[i] != (0 as u8) { 277 let c: i64 = s[i] as i64 278 if c < APL_D0 { return 0 - 1 } 279 if c > APL_D9 { return 0 - 1 } 280 v = v * APL_DEC_BASE + (c - APL_D0) 281 i = i + 1 282 } 283 if i == 0 { return 0 - 1 } 284 return v 285} 286 287// NON-OVERLAPPING occurrences of needle in the first n bytes of buf. The inner compare deliberately 288// runs to completion instead of breaking by clobbering its own cursor -- that idiom erases the answer. 289func apl_count_occ(buf: *u8, n: i64, needle: *u8) -> i64 { 290 let m: i64 = ap_len(needle) 291 if m == 0 { return 0 - 1 } 292 if m > n { return 0 } 293 var c: i64 = 0 294 var i: i64 = 0 295 while i <= n - m { 296 var j: i64 = 0 297 var ok: i64 = 1 298 while j < m { 299 if buf[i + j] != needle[j] { ok = 0 } 300 j = j + 1 301 } 302 if ok == 1 { c = c + 1; i = i + m } else { i = i + 1 } 303 } 304 return c 305} 306 307// Read <path> once: hex-digest it, and count `needle` over the same bytes. Returns the byte count, or 308// -1 when the file cannot be read (which is how ABSENT is detected). occ_out receives the occurrence 309// count, or -1 when no needle was asked for. 310func apl_scan(path: *u8, hexout: *u8, needle: *u8, occ_out: *i64) -> i64 { 311 let szp: *i64 = sys_mmap(16) as *i64 312 let b: *u8 = sys_read_file(path, szp) 313 if (b as i64) == 0 { occ_out[0] = 0 - 1; return 0 - 1 } 314 let n: i64 = szp[0] 315 let dig: *u8 = sys_mmap(32) 316 sha256_digest(b, n, dig) 317 ap_hex(dig, hexout) 318 if (needle as i64) == 0 { occ_out[0] = 0 - 1 } else { occ_out[0] = apl_count_occ(b, n, needle) } 319 return n 320} 321 322// settle_ms from conf, or -1 when the conf is absent or carries no row. The value used is ANNOUNCED by 323// the caller either way: an unannounced default is the defect, not the default itself. 324func apl_conf_settle_ms() -> i64 { 325 let szp: *i64 = sys_mmap(16) as *i64 326 let b: *u8 = sys_read_file(APL_CONF, szp) 327 if (b as i64) == 0 { return 0 - 1 } 328 let n: i64 = szp[0] 329 let key: *u8 = "settle_ms " as *u8 330 let m: i64 = ap_len(key) 331 var found: i64 = 0 - 1 332 var i: i64 = 0 333 while i <= n - m { 334 var atline: i64 = 0 335 if i == 0 { atline = 1 } 336 if i > 0 { if (b[i - 1] as i64) == APL_NL { atline = 1 } } 337 if atline == 1 { 338 var j: i64 = 0 339 var ok: i64 = 1 340 while j < m { 341 if b[i + j] != key[j] { ok = 0 } 342 j = j + 1 343 } 344 if ok == 1 { if found < 0 { 345 var v: i64 = 0 346 var got: i64 = 0 347 var k: i64 = i + m 348 var scanning: i64 = 1 349 while scanning == 1 { 350 if k >= n { scanning = 0 } else { 351 let c: i64 = b[k] as i64 352 if c < APL_D0 { scanning = 0 } else { 353 if c > APL_D9 { scanning = 0 } else { 354 v = v * APL_DEC_BASE + (c - APL_D0); got = 1; k = k + 1 355 } 356 } 357 } 358 } 359 if got == 1 { found = v } 360 } } 361 } 362 i = i + 1 363 } 364 return found 365} 366 367// THE CONTRACT SYMBOL. devmgmt.plan names this rung `ap_landed`, and the watch row measures that the 368// SYMBOL EXISTS -- so the entry point carries the contract name rather than leaving it to live only in 369// prose. ★A COMPLETION SIGNAL THAT KEYS ON A NAME REWARDS WRITING THE NAME: the flip is the receipt, 370// the gate is the proof. 371func ap_landed(listpath: *u8) -> i64 { 372 let szp: *i64 = sys_mmap(16) as *i64 373 let raw: *u8 = sys_read_file(listpath, szp) 374 if (raw as i64) == 0 { 375 ap_puts("AP-LANDED verdict=RED reason=writelist-unreadable path=" as *u8); ap_puts(listpath); ap_puts("\n" as *u8) 376 return APL_RC_LIST_UNREADABLE 377 } 378 let rn: i64 = szp[0] 379 if rn == 0 { 380 ap_puts("AP-LANDED verdict=RED reason=writelist-empty path=" as *u8); ap_puts(listpath); ap_puts("\n" as *u8) 381 return APL_RC_LIST_UNREADABLE 382 } 383 // Own copy so tokens can be NUL-terminated in place without mutating the read buffer. 384 let buf: *u8 = sys_mmap(rn + APL_NUL_SLACK) 385 var c0: i64 = 0 386 while c0 < rn { buf[c0] = raw[c0]; c0 = c0 + 1 } 387 buf[rn] = 0 as u8 388 389 // NO CAP: the row table is sized from the line count of the file itself, so nothing is silently 390 // dropped and there is no ceiling to guess. 391 var lines: i64 = 1 392 var li: i64 = 0 393 while li < rn { if (buf[li] as i64) == APL_NL { lines = lines + 1 } li = li + 1 } 394 395 let a_id: *i64 = sys_mmap(lines * APL_PTRW) as *i64 396 let a_path: *i64 = sys_mmap(lines * APL_PTRW) as *i64 397 let a_kind: *i64 = sys_mmap(lines * APL_PTRW) as *i64 398 let a_want: *i64 = sys_mmap(lines * APL_PTRW) as *i64 399 let a_pre: *i64 = sys_mmap(lines * APL_PTRW) as *i64 400 let a_mark: *i64 = sys_mmap(lines * APL_PTRW) as *i64 401 let a_size: *i64 = sys_mmap(lines * APL_PTRW) as *i64 402 let a_reply: *i64 = sys_mmap(lines * APL_PTRW) as *i64 403 let a_markpre: *i64 = sys_mmap(lines * APL_PTRW) as *i64 404 let a_hasmp: *i64 = sys_mmap(lines * APL_PTRW) as *i64 405 let a_adds: *i64 = sys_mmap(lines * APL_PTRW) as *i64 406 let a_hasadds: *i64 = sys_mmap(lines * APL_PTRW) as *i64 407 let a_bad: *i64 = sys_mmap(lines * APL_PTRW) as *i64 408 let a_nA: *i64 = sys_mmap(lines * APL_PTRW) as *i64 409 let a_nB: *i64 = sys_mmap(lines * APL_PTRW) as *i64 410 let a_occA: *i64 = sys_mmap(lines * APL_PTRW) as *i64 411 let a_occB: *i64 = sys_mmap(lines * APL_PTRW) as *i64 412 let hexA: *u8 = sys_mmap(lines * APL_HEXSLOT) 413 let hexB: *u8 = sys_mmap(lines * APL_HEXSLOT) 414 415 var nrows: i64 = 0 416 var p: i64 = 0 417 while p < rn { 418 var e: i64 = p 419 var sc: i64 = 1 420 while sc == 1 { 421 if e >= rn { sc = 0 } else { 422 if (buf[e] as i64) == APL_NL { sc = 0 } else { e = e + 1 } 423 } 424 } 425 var q: i64 = e 426 if q > p { if (buf[q - 1] as i64) == APL_CR { q = q - 1 } } 427 buf[q] = 0 as u8 428 let line: *u8 = ((buf as i64) + p) as *u8 429 var skip: i64 = 0 430 if q == p { skip = 1 } 431 if skip == 0 { if (line[0] as i64) == APL_TAB { skip = 1 } } 432 if skip == 0 { if ap_len(line) > 1 { if (line[0] as i64) == APL_SLASH { if (line[1] as i64) == APL_SLASH { skip = 1 } } } } 433 if skip == 0 { 434 let r: i64 = nrows 435 a_adds[r] = APL_DEFAULT_MARK_ADDS 436 var t: i64 = p 437 var tsc: i64 = 1 438 while tsc == 1 { 439 var te: i64 = t 440 var isc: i64 = 1 441 while isc == 1 { 442 if te >= q { isc = 0 } else { 443 if (buf[te] as i64) == APL_TAB { isc = 0 } else { te = te + 1 } 444 } 445 } 446 let hit_end: i64 = te 447 buf[te] = 0 as u8 448 let tok: *u8 = ((buf as i64) + t) as *u8 449 if (tok[0] as i64) != 0 { 450 var matched: i64 = 0 451 var v: *u8 = apl_kv(tok, "id" as *u8) 452 if (v as i64) != 0 { a_id[r] = v as i64; matched = 1 } 453 if matched == 0 { v = apl_kv(tok, "path" as *u8); if (v as i64) != 0 { a_path[r] = v as i64; matched = 1 } } 454 if matched == 0 { v = apl_kv(tok, "kind" as *u8); if (v as i64) != 0 { a_kind[r] = v as i64; matched = 1 } } 455 if matched == 0 { v = apl_kv(tok, "want" as *u8); if (v as i64) != 0 { a_want[r] = v as i64; matched = 1 } } 456 if matched == 0 { v = apl_kv(tok, "pre" as *u8); if (v as i64) != 0 { a_pre[r] = v as i64; matched = 1 } } 457 if matched == 0 { v = apl_kv(tok, "mark_pre" as *u8); if (v as i64) != 0 { a_markpre[r] = apl_atoi(v); a_hasmp[r] = 1; matched = 1 } } 458 if matched == 0 { v = apl_kv(tok, "mark_adds" as *u8); if (v as i64) != 0 { a_adds[r] = apl_atoi(v); a_hasadds[r] = 1; matched = 1 } } 459 if matched == 0 { v = apl_kv(tok, "mark" as *u8); if (v as i64) != 0 { a_mark[r] = v as i64; matched = 1 } } 460 if matched == 0 { v = apl_kv(tok, "size" as *u8); if (v as i64) != 0 { a_size[r] = v as i64; matched = 1 } } 461 if matched == 0 { v = apl_kv(tok, "reply" as *u8); if (v as i64) != 0 { a_reply[r] = v as i64; matched = 1 } } 462 if matched == 0 { a_bad[r] = 1 } 463 } 464 if hit_end >= q { tsc = 0 } else { t = hit_end + 1 } 465 } 466 if a_path[r] == 0 { a_bad[r] = 1 } 467 if a_id[r] == 0 { a_id[r] = "(unnamed)" as i64 } 468 if a_hasmp[r] == 1 { if a_markpre[r] < 0 { a_bad[r] = 1 } } 469 if a_hasadds[r] == 1 { if a_adds[r] < 1 { a_bad[r] = 1 } } 470 nrows = nrows + 1 471 } 472 p = e + 1 473 } 474 475 if nrows == 0 { 476 ap_puts("AP-LANDED verdict=RED reason=writelist-has-no-rows path=" as *u8); ap_puts(listpath); ap_puts("\n" as *u8) 477 return APL_RC_LIST_UNREADABLE 478 } 479 480 var settle: i64 = apl_conf_settle_ms() 481 var settle_src: *u8 = "conf" as *u8 482 if settle < 0 { settle = APL_SETTLE_MS_DEFAULT; settle_src = "default-nx_memvel-window" as *u8 } 483 484 // ---- PASS A ---- 485 let occp: *i64 = sys_mmap(16) as *i64 486 var r1: i64 = 0 487 while r1 < nrows { 488 if a_bad[r1] == 0 { 489 a_nA[r1] = apl_scan(a_path[r1] as *u8, apl_slot(hexA, r1, APL_HEXSLOT), a_mark[r1] as *u8, occp) 490 a_occA[r1] = occp[0] 491 } 492 r1 = r1 + 1 493 } 494 // ONE settle interval for the WHOLE worklist, not one per row. 495 sys_sleep_ms(settle) 496 // ---- PASS B ---- 497 var r2: i64 = 0 498 while r2 < nrows { 499 if a_bad[r2] == 0 { 500 a_nB[r2] = apl_scan(a_path[r2] as *u8, apl_slot(hexB, r2, APL_HEXSLOT), a_mark[r2] as *u8, occp) 501 a_occB[r2] = occp[0] 502 } 503 r2 = r2 + 1 504 } 505 506 let rcount: *i64 = sys_mmap(APL_R_COUNT * APL_PTRW) as *i64 507 var n_landed: i64 = 0 508 var n_not: i64 = 0 509 var n_double: i64 = 0 510 var n_unknown: i64 = 0 511 512 var r: i64 = 0 513 while r < nrows { 514 var v: i64 = APL_V_UNKNOWN 515 var rs: i64 = APL_R_NONE 516 var ev: *u8 = "none" as *u8 517 var done: i64 = 0 518 519 if a_bad[r] == 1 { v = APL_V_UNKNOWN; rs = APL_R_MALFORMED; ev = "row-not-parsed" as *u8; done = 1 } 520 521 let hA: *u8 = apl_slot(hexA, r, APL_HEXSLOT) 522 let hB: *u8 = apl_slot(hexB, r, APL_HEXSLOT) 523 var absA: i64 = 0 524 var absB: i64 = 0 525 if done == 0 { if a_nA[r] < 0 { absA = 1 } } 526 if done == 0 { if a_nB[r] < 0 { absB = 1 } } 527 528 // TIER 0 -- SETTLED, or nothing below it may be trusted. 529 if done == 0 { if absA != absB { v = APL_V_UNKNOWN; rs = APL_R_UNSETTLED; ev = "two-reads-differ-on-presence" as *u8; done = 1 } } 530 if done == 0 { if absA == 0 { if ap_streq(hA, hB) == 0 { v = APL_V_UNKNOWN; rs = APL_R_UNSETTLED; ev = "two-reads-differ-on-content" as *u8; done = 1 } } } 531 532 var pre_absent: i64 = 0 533 if a_pre[r] != 0 { if ap_streq(a_pre[r] as *u8, "absent" as *u8) == 1 { pre_absent = 1 } } 534 535 if done == 0 { if absA == 1 { 536 if pre_absent == 1 { v = APL_V_NOT; ev = "absent-now-and-pre-absent" as *u8; done = 1 } 537 if done == 0 { v = APL_V_UNKNOWN; rs = APL_R_UNREADABLE; ev = "artifact-unreadable" as *u8; done = 1 } 538 } } 539 540 // TIER 1 -- CONTENT HASH. One of the only two sites that can emit LANDED. 541 if done == 0 { if a_want[r] != 0 { 542 let wh: *u8 = apl_dehex(a_want[r] as *u8) 543 var prematch: i64 = 0 544 if pre_absent == 0 { if a_pre[r] != 0 { if ap_streq(apl_dehex(a_pre[r] as *u8), wh) == 1 { prematch = 1 } } } 545 if prematch == 1 { v = APL_V_UNKNOWN; rs = APL_R_WANT_EQ_PRE; ev = "content-hash" as *u8; done = 1 } 546 if done == 0 { if ap_streq(hA, wh) == 1 { v = APL_V_LANDED; ev = "content-hash" as *u8; done = 1 } } 547 if done == 0 { if pre_absent == 0 { if a_pre[r] != 0 { if ap_streq(hA, apl_dehex(a_pre[r] as *u8)) == 1 { v = APL_V_NOT; ev = "content-hash-equals-pre-image" as *u8; done = 1 } } } } 548 if done == 0 { v = APL_V_UNKNOWN; rs = APL_R_THIRD_STATE; ev = "content-hash" as *u8; done = 1 } 549 } } 550 551 // TIER 2 -- MARKER COUNT. The other site that can emit LANDED, and the only one that can see a 552 // double-apply: a blind retry of a non-idempotent edit shows up here and nowhere else. 553 if done == 0 { if a_mark[r] != 0 { 554 if a_hasmp[r] == 0 { v = APL_V_UNKNOWN; rs = APL_R_MARK_NO_PRE; ev = "marker-count" as *u8; done = 1 } 555 if done == 0 { if a_occA[r] != a_occB[r] { v = APL_V_UNKNOWN; rs = APL_R_UNSETTLED; ev = "marker-count-differs-across-reads" as *u8; done = 1 } } 556 if done == 0 { 557 let mp: i64 = a_markpre[r] 558 let ad: i64 = a_adds[r] 559 let oc: i64 = a_occA[r] 560 if oc < mp { v = APL_V_UNKNOWN; rs = APL_R_MARK_FELL; ev = "marker-count" as *u8; done = 1 } 561 if done == 0 { if oc == mp { v = APL_V_NOT; ev = "marker-count" as *u8; done = 1 } } 562 if done == 0 { if oc == mp + ad { v = APL_V_LANDED; ev = "marker-count" as *u8; done = 1 } } 563 if done == 0 { if oc > mp + ad { v = APL_V_DOUBLE; ev = "marker-count" as *u8; done = 1 } } 564 if done == 0 { v = APL_V_UNKNOWN; rs = APL_R_MARK_PARTIAL; ev = "marker-count" as *u8; done = 1 } 565 } 566 } } 567 568 // TIER 3 -- PRE-IMAGE ONLY. Structurally incapable of returning LANDED: a file that CHANGED is 569 // not evidence that YOUR change is the one that landed. 570 if done == 0 { if a_pre[r] != 0 { 571 if pre_absent == 1 { v = APL_V_UNKNOWN; rs = APL_R_CHANGED_UNDECLARED; ev = "pre-image-only" as *u8; done = 1 } 572 if done == 0 { if ap_streq(hA, apl_dehex(a_pre[r] as *u8)) == 1 { v = APL_V_NOT; ev = "pre-image-only" as *u8; done = 1 } } 573 if done == 0 { v = APL_V_UNKNOWN; rs = APL_R_CHANGED_UNDECLARED; ev = "pre-image-only" as *u8; done = 1 } 574 } } 575 576 // TIER 4 -- NOTHING DECIDABLE. Named separately because the remedies differ: a size-only row 577 // needs a hash recorded next time, a reply-only row needs any expectation at all. 578 if done == 0 { if a_size[r] != 0 { v = APL_V_UNKNOWN; rs = APL_R_SIZE_ONLY; ev = "none" as *u8; done = 1 } } 579 if done == 0 { if a_reply[r] != 0 { v = APL_V_UNKNOWN; rs = APL_R_REPLY_ONLY; ev = "none" as *u8; done = 1 } } 580 if done == 0 { v = APL_V_UNKNOWN; rs = APL_R_NO_EXPECTATION; ev = "none" as *u8; done = 1 } 581 582 if v == APL_V_LANDED { n_landed = n_landed + 1 } 583 if v == APL_V_NOT { n_not = n_not + 1 } 584 if v == APL_V_DOUBLE { n_double = n_double + 1 } 585 if v == APL_V_UNKNOWN { n_unknown = n_unknown + 1; rcount[rs] = rcount[rs] + 1 } 586 587 ap_puts("APL row=" as *u8); ap_puts(a_id[r] as *u8) 588 ap_puts(" verdict=" as *u8); ap_puts(apl_verdict_name(v)) 589 ap_puts(" evidence=" as *u8); ap_puts(ev) 590 if v == APL_V_UNKNOWN { ap_puts(" reason=" as *u8); ap_puts(apl_reason_name(rs)) } 591 ap_puts(" path=" as *u8) 592 if a_path[r] != 0 { ap_puts(a_path[r] as *u8) } else { ap_puts("(missing)" as *u8) } 593 ap_puts(" kind=" as *u8) 594 if a_kind[r] != 0 { ap_puts(a_kind[r] as *u8) } else { ap_puts("(undeclared)" as *u8) } 595 if a_bad[r] == 0 { 596 ap_puts(" now=" as *u8) 597 if absA == 1 { ap_puts("ABSENT" as *u8) } else { ap_puts(hA) } 598 ap_puts(" bytes=" as *u8); ap_num(a_nA[r]) 599 ap_puts(" settled=" as *u8) 600 if rs == APL_R_UNSETTLED { ap_puts("0" as *u8) } else { ap_puts("1" as *u8) } 601 if a_mark[r] != 0 { 602 ap_puts(" occ_now=" as *u8); ap_num(a_occA[r]) 603 ap_puts(" mark_pre=" as *u8) 604 if a_hasmp[r] == 1 { ap_num(a_markpre[r]) } else { ap_puts("UNDECLARED" as *u8) } 605 ap_puts(" mark_adds=" as *u8); ap_num(a_adds[r]) 606 if a_hasadds[r] == 0 { ap_puts("(default)" as *u8) } 607 } 608 } 609 ap_puts("\n" as *u8) 610 ap_puts(" advice=" as *u8); ap_puts(apl_advice(a_kind[r] as *u8)); ap_puts("\n" as *u8) 611 if a_size[r] != 0 { ap_puts(" NON-EVIDENCE-ECHO size=" as *u8); ap_puts(a_size[r] as *u8); ap_puts(" -- NOT used: a size read taken right after a dropped write can report the PRE-write size\n" as *u8) } 612 if a_reply[r] != 0 { ap_puts(" NON-EVIDENCE-ECHO reply=" as *u8); ap_puts(a_reply[r] as *u8); ap_puts(" -- NOT used: the transport reply text discriminated nothing in either direction\n" as *u8) } 613 r = r + 1 614 } 615 616 let total: i64 = n_landed + n_not + n_double + n_unknown 617 ap_puts("APL-SETTLE settle_ms=" as *u8); ap_num(settle) 618 ap_puts(" src=" as *u8); ap_puts(settle_src) 619 ap_puts(" passes=2 (one interval for the whole worklist)\n" as *u8) 620 ap_puts("APL-PARTITION rows=" as *u8); ap_num(nrows) 621 ap_puts(" landed=" as *u8); ap_num(n_landed) 622 ap_puts(" not_landed=" as *u8); ap_num(n_not) 623 ap_puts(" double_applied=" as *u8); ap_num(n_double) 624 ap_puts(" unknown=" as *u8); ap_num(n_unknown) 625 ap_puts(" sum=" as *u8); ap_num(total) 626 if total == nrows { ap_puts(" partition=RECONCILES\n" as *u8) } else { ap_puts(" partition=LEAK\n" as *u8) } 627 var ri: i64 = 0 628 while ri < APL_R_COUNT { 629 if rcount[ri] > 0 { 630 ap_puts("APL-UNKNOWN-REASON " as *u8); ap_puts(apl_reason_name(ri)) 631 ap_puts(" count=" as *u8); ap_num(rcount[ri]); ap_puts("\n" as *u8) 632 } 633 ri = ri + 1 634 } 635 if n_unknown > 0 { 636 ap_puts("AP-LANDED verdict=NEEDS-ADJUDICATION unknown=" as *u8); ap_num(n_unknown) 637 ap_puts(" -- abstained rather than acquitted; each UNKNOWN row names its reason above\n" as *u8) 638 return 1 639 } 640 ap_puts("AP-LANDED verdict=DECIDED-ALL rows=" as *u8); ap_num(nrows); ap_puts(" unknown=0\n" as *u8) 641 return 0 642} 643 644func main(argc: i64, argv: *i64) -> i64 { 645 // The `landed` verb is dispatched on argv[1] and the legacy three-positional publish form is 646 // untouched: a staged source literally named "landed" is the only collision, and it is not one. 647 if argc >= 2 { if ap_streq(argv[1] as *u8, "landed" as *u8) == 1 { 648 if argc < 3 { 649 ap_puts("usage: nx_atomic_publish landed <writelist>\n" as *u8) 650 ap_puts(" Adjudicates ambiguous writes. Rows are TAB-separated key=value:\n" as *u8) 651 ap_puts(" id= path= kind=publish|promote|create|replace|insert|append want=h<sha256> pre=h<sha256>|absent\n" as *u8) 652 ap_puts(" mark=<string> mark_pre=<n> mark_adds=<n> size=<n> reply=<text>\n" as *u8) 653 ap_puts(" size and reply are ECHOED AND IGNORED on purpose -- neither is evidence of landing.\n" as *u8) 654 ap_puts(" exit: 0 all decided | 1 some UNKNOWN | 2 usage | 3 list unreadable\n" as *u8) 655 sys_exit(2) 656 return 2 657 } 658 let rc: i64 = ap_landed(argv[2] as *u8) 659 sys_exit(rc) 660 return rc 661 } } 662 if argc < 4 { 663 ap_puts("usage: nx_atomic_publish <staged-src> <target> <expect>\n" as *u8) 664 ap_puts(" expect: any | absent | <64-hex sha256 of the target's CURRENT content>\n" as *u8) 665 ap_puts(" All-or-nothing: stages to <target>.aptmp then renameat over the target.\n" as *u8) 666 ap_puts(" On CAS mismatch it REFUSES and prints the live hash -- it never re-resolves its own\n" as *u8) 667 ap_puts(" expected value, because that would make a concurrent writer's edit a silent clobber.\n" as *u8) 668 ap_puts(" nx_atomic_publish landed <writelist> -- the READ half: adjudicate ambiguous writes.\n" as *u8) 669 sys_exit(2) 670 return 2 671 } 672 let src: *u8 = argv[1] as *u8 673 let dst: *u8 = argv[2] as *u8 674 let expect: *u8 = argv[3] as *u8 675 676 let sptr: *i64 = sys_mmap(16) as *i64 677 let shex: *u8 = sys_mmap(80) 678 let sn: i64 = ap_hash_file(src, sptr, shex) 679 if sn < 0 { 680 ap_puts("NX-ATOMIC-PUBLISH verdict=RED reason=source-unreadable path=" as *u8); ap_puts(src); ap_puts("\n" as *u8) 681 sys_exit(3) 682 return 3 683 } 684 // A ZERO-BYTE SOURCE IS ALMOST ALWAYS A FAILED UPLOAD, NOT AN INTENDED EMPTY FILE, and publishing it 685 // would truncate a real file to nothing. Refuse; `any` does not override this. 686 if sn == 0 { 687 ap_puts("NX-ATOMIC-PUBLISH verdict=RED reason=source-empty (refusing to publish 0 bytes over a live file)\n" as *u8) 688 sys_exit(3) 689 return 3 690 } 691 let sbuf: *u8 = sptr[0] as *u8 692 693 let dptr: *i64 = sys_mmap(16) as *i64 694 let dhex: *u8 = sys_mmap(80) 695 let dn: i64 = ap_hash_file(dst, dptr, dhex) 696 697 var ok: i64 = 0 698 if ap_streq(expect, "any" as *u8) == 1 { ok = 1 } 699 if ap_streq(expect, "absent" as *u8) == 1 { if dn < 0 { ok = 1 } } 700 if dn >= 0 { if ap_streq(expect, dhex) == 1 { ok = 1 } } 701 if dn < 0 { if ap_streq(expect, "absent" as *u8) == 0 { if ap_streq(expect, "any" as *u8) == 0 { 702 ap_puts("NX-ATOMIC-PUBLISH verdict=REFUSED reason=target-absent-but-expect-names-a-hash (use expect=absent for a create)\n" as *u8) 703 sys_exit(1) 704 return 1 705 } } } 706 707 if ok == 0 { 708 ap_puts("NX-ATOMIC-PUBLISH verdict=REFUSED reason=stale-expect -- the target changed under you; RE-READ AND MERGE, do not force.\n" as *u8) 709 ap_puts(" live_sha256=" as *u8); ap_puts(dhex); ap_puts("\n" as *u8) 710 ap_puts(" you_expected=" as *u8); ap_puts(expect); ap_puts("\n" as *u8) 711 ap_puts(" target UNCHANGED (" as *u8); ap_num(dn); ap_puts(" bytes)\n" as *u8) 712 sys_exit(1) 713 return 1 714 } 715 716 let tmp: *u8 = sys_mmap(AP_PATHCAP) 717 ap_cat2(tmp, dst, ".aptmp" as *u8) 718 let fd: i64 = sys_openat_wr(tmp, AP_MODE) 719 if fd < 0 { 720 ap_puts("NX-ATOMIC-PUBLISH verdict=RED reason=cannot-open-temp path=" as *u8); ap_puts(tmp); ap_puts("\n" as *u8) 721 sys_exit(3) 722 return 3 723 } 724 let w: i64 = sys_write(fd, sbuf, sn) 725 sys_close(fd) 726 if w != sn { 727 ap_puts("NX-ATOMIC-PUBLISH verdict=RED reason=short-write -- target UNTOUCHED, temp left for inspection at " as *u8) 728 ap_puts(tmp); ap_puts("\n" as *u8) 729 sys_exit(3) 730 return 3 731 } 732 if sys_renameat(tmp, dst) != 0 { 733 ap_puts("NX-ATOMIC-PUBLISH verdict=RED reason=rename-failed -- target UNTOUCHED, staged content at " as *u8) 734 ap_puts(tmp); ap_puts("\n" as *u8) 735 sys_exit(3) 736 return 3 737 } 738 739 // VERIFY BY ARTEFACT, NOT BY THE WRITE RETURNING. Re-read what is actually on disk and re-hash it. 740 let vptr: *i64 = sys_mmap(16) as *i64 741 let vhex: *u8 = sys_mmap(80) 742 let vn: i64 = ap_hash_file(dst, vptr, vhex) 743 if vn < 0 { 744 ap_puts("NX-ATOMIC-PUBLISH verdict=RED reason=published-but-unreadable\n" as *u8) 745 sys_exit(3) 746 return 3 747 } 748 if ap_streq(vhex, shex) == 0 { 749 ap_puts("NX-ATOMIC-PUBLISH verdict=RED reason=verify-mismatch published_sha256=" as *u8); ap_puts(vhex) 750 ap_puts(" expected=" as *u8); ap_puts(shex); ap_puts("\n" as *u8) 751 sys_exit(3) 752 return 3 753 } 754 ap_puts("NX-ATOMIC-PUBLISH verdict=GREEN published=" as *u8); ap_puts(dst) 755 ap_puts(" bytes=" as *u8); ap_num(vn) 756 ap_puts(" sha256=" as *u8); ap_puts(vhex) 757 if dn >= 0 { ap_puts(" replaced_sha256=" as *u8); ap_puts(dhex) } 758 if dn < 0 { ap_puts(" (created)" as *u8) } 759 ap_puts("\n" as *u8) 760 sys_exit(0) 761 return 0 762}