code wiki / (root) / nx_jrnl_archive_gate_t138.nx
`nx_jrnl_archive_gate_t138.nx` source

↩ module page · 319 lines · 17299 B
1// nx_jrnl_archive_gate.nx -- GATE for nx_jrnl_archive (lib composed in-process, nothing forked).
2// The verb's whole contract is PRESERVE-BEFORE-REMOVE, so the load-bearing teeth are the ones where
3// the archive step FAILS and the live journal must come out byte-identical (T6), and the one where
4// the verify step can actually see a corrupted copy (T10 -- a hash compare that cannot fail is not a
5// verify). Fixtures live under /tmp/nx_jrnl_archive_gate_<pid>/ (data only; nothing is executed from
6// /tmp, which the NAS mounts noexec). license_tier: ORIGINAL   No hw writes (Rule 26).
7import "nx_syscalls.nx"
8import "nx_gate_verdict.nx"
9import "nx_jrnl_archive_lib_t138.nx"
10
11const JAG_FIX_LINES: i64 = 40           // enough rows that a short write is detectable, small enough to read by eye
12const JAG_EPOCH_A: i64 = 1700000001     // fixed epochs: archive names must be reproducible across runs
13const JAG_EPOCH_B: i64 = 1700000002
14const JAG_EPOCH_C: i64 = 1700000003
15
16func jg_pid() -> i64 { return __syscall(172, 0, 0, 0, 0, 0, 0) }
17func jg_path(base: *u8, leaf: *u8) -> *u8 {
  let p: *u8 = sys_mmap(JA_PATH_MAX)
  var o: i64 = ja_cat(p, 0, base)
  o = ja_cat(p, o, "/" as *u8)
  o = ja_cat(p, o, leaf)
  return p
23}
24func jg_size(path: *u8) -> i64 {
  let ln: *i64 = sys_mmap(16) as *i64
  let b: *u8 = sys_read_file(path, ln)
  if (b as i64) == 0 { return 0 - 1 }
  let n: i64 = ln[0]
  sys_free_file(b, n)
  return n
31}
32func jg_digest(path: *u8, out: *u8) -> i64 {
  let ln: *i64 = sys_mmap(16) as *i64
  let b: *u8 = sys_read_file(path, ln)
  if (b as i64) == 0 { return 0 - 1 }
  let n: i64 = ln[0]
  sha256_digest(b, n, out)
  sys_free_file(b, n)
  return n
40}
41func jg_same(a: *u8, b: *u8) -> i64 { var i: i64 = 0; while i < JA_DIGEST_BYTES { if a[i] != b[i] { return 0 } i = i + 1 } return 1 }
42// a journal fixture: JAG_FIX_LINES rows "epoch<TAB>target<TAB>bytes" -- the comparepub shape
43func jg_make_journal(path: *u8) -> i64 {
  let buf: *u8 = sys_mmap(JAG_FIX_LINES * 64)
  var o: i64 = 0
  var i: i64 = 0
  while i < JAG_FIX_LINES {
      o = ja_catn(buf, o, 1787500000 + i)
      buf[o] = 9 as u8; o = o + 1
      o = ja_cat(buf, o, "../sites/nishifamily/compare/fixture/api.json" as *u8)
      buf[o] = 9 as u8; o = o + 1
      o = ja_catn(buf, o, 1000 + i * 7)
      buf[o] = 10 as u8; o = o + 1
      i = i + 1
  }
  return ja_write_all(path, buf, o, JA_ARCH_MODE)
57}
58
59// One child writer, explicit pipe acknowledgements. Timeout bounds failures, never schedules success.
60const JAG_SYNC_TIMEOUT_MS: i64 = 10000
61func jg_receive(fd: i64, expected: i64) -> i64 {
  let pollfd: *i64 = sys_mmap(16) as *i64
  pollfd[0] = (fd & 0xFFFFFFFF) | (1 << 32)
  if sys_poll(pollfd as *u8, 1, JAG_SYNC_TIMEOUT_MS) <= 0 { return 0 }
  let byte: *u8 = sys_mmap(1)
  if sys_read(fd, byte, 1) != 1 { return 0 }
  return (byte[0] as i64) == expected
68}
69func jg_child_writer(journal: *u8, commands: i64, events: i64) -> i64 {
  let lock: i64 = ja_lock(journal)
  if lock < 0 { return 1 }
  if sys_write(events, "A" as *u8, 1) != 1 { ja_unlock(lock); return 2 }
  if jg_receive(commands, 82) != 1 { ja_unlock(lock); return 3 }
  if ja_unlock(lock) != 0 { return 4 }
  if sys_write(events, "B" as *u8, 1) != 1 { return 5 }
  if jg_receive(commands, 84) != 1 { return 6 }
  let busy: i64 = ja_lock(journal)
  if busy >= 0 { ja_unlock(busy); return 7 }
  if sys_write(events, "C" as *u8, 1) != 1 { return 8 }
  if jg_receive(commands, 87) != 1 { return 9 }
  let writer: i64 = ja_lock(journal)
  if writer < 0 { return 10 }
  let fd: i64 = sys_openat_append(journal, JA_ARCH_MODE)
  if fd < 0 { ja_unlock(writer); return 11 }
  let wrote: i64 = sys_write(fd, "CHILD-AFTER-ARCHIVE\n" as *u8, 20)
  let sync: i64 = sys_fsync(fd)
  let closed: i64 = sys_close(fd)
  let unlocked: i64 = ja_unlock(writer)
  if wrote != 20 || sync != 0 || closed != 0 || unlocked != 0 { return 12 }
  if sys_write(events, "D" as *u8, 1) != 1 { return 13 }
  return 0
92}
93func jg_process_gate(base: *u8, ctr: *i64) -> i64 {
  let journal: *u8 = jg_path(base, "process.jrnl" as *u8)
  jg_make_journal(journal)
  let initial: i64 = jg_size(journal)
  let commands: *i64 = sys_mmap(16) as *i64
  let events: *i64 = sys_mmap(16) as *i64
  if sys_pipe2(commands, 0) != 0 { return 0 }
  let cr: i64 = commands[0] & 0xFFFFFFFF
  let cw: i64 = (commands[0] >> 32) & 0xFFFFFFFF
  if sys_pipe2(events, 0) != 0 { sys_close(cr); sys_close(cw); return 0 }
  let er: i64 = events[0] & 0xFFFFFFFF
  let ew: i64 = (events[0] >> 32) & 0xFFFFFFFF
  let pid: i64 = sys_fork()
  if pid < 0 { sys_close(cr); sys_close(cw); sys_close(er); sys_close(ew); return 0 }
  if pid == 0 {
      sys_close(cw); sys_close(er)
      let result: i64 = jg_child_writer(journal, cr, ew)
      sys_close(cr); sys_close(ew)
      sys_exit(result); return result
  }
  sys_close(cr); sys_close(ew)
  let rep: *i64 = sys_mmap(32) as *i64
  var sequence: i64 = jg_receive(er, 65)
  var blocked_archive: i64 = 0
  var blocked_writer: i64 = 0
  var archived: i64 = 0
  var appended: i64 = 0
  if sequence == 1 {
      let busy: i64 = ja_archive_cooperative(journal, 0 as *u8, JAG_EPOCH_A, rep)
      blocked_archive = (busy == 7) & (jg_size(journal) == initial)
      if sys_write(cw, "R" as *u8, 1) != 1 { sequence = 0 }
      if sequence == 1 { sequence = jg_receive(er, 66) }
  }
  var held: i64 = 0 - 1
  if sequence == 1 {
      held = ja_lock(journal)
      if held < 0 { sequence = 0 } else {
          if sys_write(cw, "T" as *u8, 1) != 1 { sequence = 0 }
          if sequence == 1 { sequence = jg_receive(er, 67); blocked_writer = sequence }
      }
  }
  if sequence == 1 {
      let done: i64 = ja_archive_locked(journal, 0 as *u8, JAG_EPOCH_A, rep)
      archived = (done == 0) & (rep[0] == initial) & (jg_size(journal) == 0)
  }
  if held >= 0 { if ja_unlock(held) != 0 { sequence = 0 } }
  if sequence == 1 {
      if sys_write(cw, "W" as *u8, 1) != 1 { sequence = 0 }
      if sequence == 1 { sequence = jg_receive(er, 68) }
      if sequence == 1 {
          let actual: *u8 = sys_mmap(JA_DIGEST_BYTES)
          let wanted: *u8 = sys_mmap(JA_DIGEST_BYTES)
          let n: *i64 = sys_mmap(16) as *i64
          sha256_digest("CHILD-AFTER-ARCHIVE\n" as *u8, 20, wanted)
          if ja_transfer(journal, 0 - 1, actual, n) == 0 { appended = (n[0] == 20) & ja_digest_same(actual, wanted) }
      }
  }
  sys_close(cw); sys_close(er)
  if sequence == 0 { nx_kill(pid, 9) }
  let status: *i64 = sys_mmap(16) as *i64
  let reaped: i64 = sys_wait4(pid, status, 0)
  gv_check("T25-independent-writer-held-lock-refuses-archive-preserves-live" as *u8, blocked_archive, ctr)
  gv_check("T26-independent-writer-refused-while-parent-archive-holds-lock" as *u8, blocked_writer & archived, ctr)
  gv_check("T27-independent-writer-after-unlock-durable-row-retained-exactly" as *u8, appended & (reaped == pid) & (wait_status_rc(status[0]) == 0), ctr)
  return sequence
158}
159
160func main(argc: i64, argv: *i64) -> i64 {
  sys_ignore_sigpipe()
  gv_head("=== nx_jrnl_archive_gate -- the archive verb is cooperative-lock durable exclusive archive ===" as *u8)
  let ctr: *i64 = gv_ctr()
  let base: *u8 = sys_mmap(JA_PATH_MAX)
  var bo: i64 = ja_cat(base, 0, "/tmp/nx_jrnl_archive_gate_" as *u8)
  bo = ja_catn(base, bo, jg_pid())
  sys_mkdir(base, JA_DIR_MODE)
  let jr: *u8 = jg_path(base, "fixture.jrnl" as *u8)
  let mk: i64 = jg_make_journal(jr)
  let n0: i64 = jg_size(jr)
  ja_puts("  fixture_bytes=" as *u8); ja_pn(n0); ja_puts("\n" as *u8)
  var reached: i64 = 0
  if mk == 0 { if n0 > 0 { reached = 1 } }
  gv_check("T1 fixture-reached-the-condition (journal written, non-empty)" as *u8, reached, ctr)
  let d0: *u8 = sys_mmap(JA_DIGEST_BYTES + 8)
  jg_digest(jr, d0)
  let rep: *i64 = sys_mmap(32) as *i64
178
  // ---- the happy path: archive, verify, reset ----
  let rcA: i64 = ja_archive_cooperative(jr, 0 as *u8, JAG_EPOCH_A, rep)
  gv_check("T2 archive-returns-OK" as *u8, rcA == 0, ctr)
  let archA: *u8 = jg_path(base, "archive/fixture.jrnl.1700000001" as *u8)
  let dA: *u8 = sys_mmap(JA_DIGEST_BYTES + 8)
  let nA: i64 = jg_digest(archA, dA)
  ja_puts("  archived_bytes=" as *u8); ja_pn(nA); ja_puts(" original_bytes=" as *u8); ja_pn(n0); ja_puts("\n" as *u8)
  var sameA: i64 = 0
  if nA == n0 { if jg_same(d0, dA) == 1 { sameA = 1 } }
  gv_check("T3 archived-copy-byte-identical-to-original (re-read from disk, sha256 equal)" as *u8, sameA, ctr)
  let nLive: i64 = jg_size(jr)
  ja_puts("  live_bytes_after=" as *u8); ja_pn(nLive); ja_puts("\n" as *u8)
  gv_check("T4 live-journal-reset-to-zero-bytes" as *u8, nLive == 0, ctr)
  var repok: i64 = 0
  if rep[0] == n0 { if rep[1] == 1 { if rep[2] == 1 { repok = 1 } } }
  gv_check("T5 announce-carries-bytes-verified-reset (report slots match the measurement)" as *u8, repok, ctr)
195
  // ---- neg-control: an unwritable archive dir must leave the live journal byte-identical ----
  let jr2: *u8 = jg_path(base, "fixture2.jrnl" as *u8)
  jg_make_journal(jr2)
  let d2: *u8 = sys_mmap(JA_DIGEST_BYTES + 8)
  let n2: i64 = jg_digest(jr2, d2)
  // a directory path UNDER A FILE cannot be created or written into (ENOTDIR)
  let baddir: *u8 = jg_path(base, "fixture2.jrnl/notadir" as *u8)
  let rcB: i64 = ja_archive_cooperative(jr2, baddir, JAG_EPOCH_B, rep)
  let d2b: *u8 = sys_mmap(JA_DIGEST_BYTES + 8)
  let n2b: i64 = jg_digest(jr2, d2b)
  var untouched: i64 = 0
  if n2b == n2 { if jg_same(d2, d2b) == 1 { untouched = 1 } }
  ja_puts("  unwritable_dir_rc=" as *u8); ja_pn(rcB); ja_puts(" live_bytes_before=" as *u8); ja_pn(n2); ja_puts(" after=" as *u8); ja_pn(n2b); ja_puts("\n" as *u8)
  gv_check("T6 neg-control-archive-write-failure-leaves-live-journal-byte-identical (rc 4, preserve-before-remove)" as *u8, untouched, ctr)
  gv_check("T6b neg-control-archive-write-failure-returns-4" as *u8, rcB == 4, ctr)
211
  // ---- neg-control: a missing journal is rc 3 and creates nothing ----
  let jr3: *u8 = jg_path(base, "does_not_exist.jrnl" as *u8)
  let rcC: i64 = ja_archive_cooperative(jr3, 0 as *u8, JAG_EPOCH_C, rep)
  let arch3: *u8 = jg_path(base, "archive/does_not_exist.jrnl.1700000003" as *u8)
  gv_check("T7 neg-control-missing-journal-returns-3" as *u8, rcC == 3, ctr)
  gv_check("T7b neg-control-missing-journal-creates-no-archive" as *u8, jg_size(arch3) < 0, ctr)
218
  // ---- an empty journal: nothing to do, rc 1, no file made, live still empty ----
  let rcD: i64 = ja_archive_cooperative(jr, 0 as *u8, JAG_EPOCH_B, rep)
  let archD: *u8 = jg_path(base, "archive/fixture.jrnl.1700000002" as *u8)
  gv_check("T8 empty-journal-returns-1-nothing-to-archive" as *u8, rcD == 1, ctr)
  gv_check("T8b empty-journal-creates-no-archive-file" as *u8, jg_size(archD) < 0, ctr)
  gv_check("T8c empty-journal-stays-empty" as *u8, jg_size(jr) == 0, ctr)
225
  // ---- the verify tooth can FAIL: corrupt one byte of the archived copy and re-hash ----
  let ln: *i64 = sys_mmap(16) as *i64
  let ab: *u8 = sys_read_file(archA, ln)
  var corrupt_seen: i64 = 0
  if (ab as i64) != 0 { if ln[0] > 0 {
      if ab[0] == (48 as u8) { ab[0] = 49 as u8 } else { ab[0] = 48 as u8 }
      ja_write_all(archA, ab, ln[0], JA_ARCH_MODE)
      let dC: *u8 = sys_mmap(JA_DIGEST_BYTES + 8)
      jg_digest(archA, dC)
      if jg_same(d0, dC) == 0 { corrupt_seen = 1 }
  } }
  gv_check("T9 neg-control-one-byte-corruption-of-the-archive-changes-the-digest (the verify compare is not vacuous)" as *u8, corrupt_seen, ctr)
238
  // ---- second fixture, named dir: the archive_dir argument is honoured ----
  let jr4: *u8 = jg_path(base, "fixture4.jrnl" as *u8)
  jg_make_journal(jr4)
  let dir4: *u8 = jg_path(base, "elsewhere" as *u8)
  let rcE: i64 = ja_archive_cooperative(jr4, dir4, JAG_EPOCH_A, rep)
  let arch4: *u8 = jg_path(base, "elsewhere/fixture4.jrnl.1700000001" as *u8)
  var dir_ok: i64 = 0
  if rcE == 0 { if jg_size(arch4) > 0 { dir_ok = 1 } }
  gv_check("T10 named-archive-dir-is-honoured (file lands there, rc 0)" as *u8, dir_ok, ctr)
248
  // Stable sidecar contention must reject archive without changing the journal.
  let jr5: *u8 = jg_path(base, "locked.jrnl" as *u8)
  jg_make_journal(jr5)
  let before5: i64 = jg_size(jr5)
  let held: i64 = ja_lock(jr5)
  gv_check("T11 lock-acquired-on-real-filesystem" as *u8, held >= 0, ctr)
  let busy: i64 = ja_archive_cooperative(jr5, 0 as *u8, JAG_EPOCH_A, rep)
  gv_check("T12 held-writer-lock-refuses-archive-with-live-intact" as *u8, (busy == 7) & (jg_size(jr5) == before5), ctr)
  ja_unlock(held)
  let released: i64 = ja_archive_cooperative(jr5, 0 as *u8, JAG_EPOCH_A, rep)
  gv_check("T13 released-lock-allows-durable-archive" as *u8, released == 0, ctr)
  let again: i64 = ja_archive_cooperative(jr5, 0 as *u8, JAG_EPOCH_A, rep)
  gv_check("T14-repeat-empty-is-idempotent" as *u8, again == 1, ctr)
  // Same-second reuse must not truncate the old archive even when new journal bytes differ.
  let arch5: *u8 = jg_path(base, "archive/locked.jrnl.1700000001" as *u8)
  let saved5: *u8 = sys_mmap(JA_DIGEST_BYTES + 8)
  jg_digest(arch5, saved5)
  ja_write_all(jr5, "NEW-UNARCHIVED-HISTORY\n" as *u8, 23, JA_ARCH_MODE)
  let collision: i64 = ja_archive_cooperative(jr5, 0 as *u8, JAG_EPOCH_A, rep)
  let after5: *u8 = sys_mmap(JA_DIGEST_BYTES + 8)
  jg_digest(arch5, after5)
  gv_check("T15-epoch-collision-preserves-prior-archive-and-new-live-data" as *u8, (collision == 4) & (jg_size(jr5) == 23) & jg_same(saved5, after5), ctr)
  let strictlen: *i64 = sys_mmap(16) as *i64
  let strict_digest: *u8 = sys_mmap(JA_DIGEST_BYTES)
  let strict: i64 = ja_transfer(base, 0 - 1, strict_digest, strictlen)
  gv_check("T16-directory-is-not-successful-empty-journal-read" as *u8, strict != 0, ctr)
  gv_check("T17-directory-sync-errors-are-observable" as *u8, ja_sync_dir(jr5) < 0, ctr)
  let refused: i64 = ja_archive(jr5, 0 as *u8, JAG_EPOCH_B, rep)
  gv_check("T18-public-entry-refuses-without-writer-migration-proof" as *u8, (refused == 7) & (jg_size(jr5) == 23), ctr)
  let writer: i64 = ja_lock(jr5)
  let append: i64 = sys_openat_append(jr5, JA_ARCH_MODE)
  let appended: i64 = sys_write(append, "AFTER\n" as *u8, 6)
  let durable: i64 = sys_fsync(append)
  sys_close(append)
  ja_unlock(writer)
  gv_check("T19-cooperating-writer-after-archive-retains-new-row" as *u8, (writer >= 0) & (appended == 6) & (durable == 0) & (jg_size(jr5) == 29), ctr)
  let big: *u8 = jg_path(base, "stream.jrnl" as *u8)
  let fsobs: *i64 = sys_mmap(144) as *i64
  sys_statfs(base, fsobs)
  let stream_bytes: i64 = 1048576 + fsobs[1] + 63
  let payload: *u8 = sys_mmap(stream_bytes)
  var ix: i64 = 0
  while ix < stream_bytes { payload[ix] = (ix % 251) as u8; ix = ix + 1 }
  ja_write_all(big, payload, stream_bytes, JA_ARCH_MODE)
  let oracle: *u8 = sys_mmap(JA_DIGEST_BYTES)
  sha256_digest(payload, stream_bytes, oracle)
  let streaming: *u8 = sys_mmap(JA_DIGEST_BYTES)
  let total: *i64 = sys_mmap(16) as *i64
  let streamrc: i64 = ja_transfer(big, 0 - 1, streaming, total)
  ja_puts("  streaming_fixture_bytes=" as *u8); ja_pn(stream_bytes)
  ja_puts(" transfer_buffer_bytes=" as *u8); ja_pn(fsobs[1])
  ja_puts(" sha_workspace_bytes=" as *u8); ja_pn(sha256_workspace_bytes()); ja_puts("\n" as *u8)
  gv_check("T20-streaming-over-old-1MiB-budget-matches-one-shot-sha-across-boundaries" as *u8, (streamrc == 0) & (total[0] == stream_bytes) & ja_digest_same(oracle, streaming), ctr)
  let bigrc: i64 = ja_archive_cooperative(big, 0 as *u8, JAG_EPOCH_A, rep)
  gv_check("T21-streaming-archive-larger-than-old-budget-verifies-and-resets" as *u8, (bigrc == 0) & (rep[0] == stream_bytes) & (jg_size(big) == 0), ctr)
  let shortfd: i64 = sys_openat_rd(jr5)
  let short_read: i64 = ja_transfer_fd(shortfd, 0 - 1, 30, fsobs[1], streaming)
  sys_close(shortfd)
  gv_check("T22-early-EOF-is-refused-not-a-successful-prefix" as *u8, short_read != 0, ctr)
  let read_error: i64 = ja_transfer_fd(0 - 1, 0 - 1, 1, fsobs[1], streaming)
  gv_check("T23-real-EBADF-read-error-is-refused" as *u8, read_error != 0, ctr)
  let input: i64 = sys_openat_rd(jr5)
  let full: i64 = sys_openat_wr("/dev/full" as *u8, JA_ARCH_MODE)
  let write_error: i64 = ja_transfer_fd(input, full, 29, fsobs[1], streaming)
  sys_close(input); sys_close(full)
  gv_check("T24-real-ENOSPC-destination-error-is-refused-with-source-intact" as *u8, (full >= 0) & (write_error != 0) & (jg_size(jr5) == 29), ctr)
  gv_check("T28-independent-process-handshake-completed-without-time-based-ordering" as *u8, jg_process_gate(base, ctr), ctr)
  let nomem: i64 = ja_transfer_fd(0 - 1, 0 - 1, 1, SHA256_SIGNED_MAX - sha256_workspace_bytes(), streaming)
  gv_check("T29-real-kernel-mapping-refusal-returns-structured-error" as *u8, nomem == (0 - 2), ctr)
  return gv_verdict("JRNL-ARCHIVE" as *u8, ctr, "cooperative writers serialized; collisions and durability failure preserve history" as *u8)
319}
nx_jrnl_archive_gate_t138.nx source

`nx_jrnl_archive_gate_t138.nx` source
