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

↩ module page · 363 lines · 18606 B
1// nx_web_shard_compact.nx -- SCALE compaction for big multi-segment shards (the web shard's rung).
2// ss_compact/ss_compact_cap have EXACT semantics (last entry per key wins, tombstones kept, retired
3// segments archived then an atomic manifest swap) but their key dedup is a LINEAR table scan =
4// O(entries^2): fine at 4k keys, unusable at the web shard's ~1.4M entries (bulk docs + pr: priors).
5// This organ is the same merge with an open-addressing HASH dedup = O(entries), plus a post-swap
6// verification pass. Semantics kept byte-equal to ss_compact: chronological walk, last wins, ALL kinds
7// carried (incl. kind-2 tombstones), archive-before-swap, every failure path returns BEFORE the swap.
8//
9// BYTE-BUDGETED SUFFIX FOLD (2026-07-29): an unbounded whole-shard merge materializes the merged
10// segment in anonymous RAM (the writer buffer, plus aux blobs and key tables that all scale with
11// merged bytes) -- the OOM killer reaped it twice at bytes-in=1883885124 on 2026-07-25, mid-merge,
12// before the swap (never-brick held; shard untouched). The merge therefore now folds the LONGEST
13// NEWEST contiguous manifest suffix whose .docs bytes fit a budget derived from the box itself
14// (MemAvailable/WSC_BUDGET_SHARE; argv[2] in MB overrides). Replacing a contiguous suffix with its
15// fold preserves last-wins semantics EXACTLY: the run is merged in chronological order, kind-2
16// tombstones are carried, and older (kept) segments remain shadowed by the merged tail precisely as
17// they were by the unmerged tail. Repeated runs converge the shard LSM-style with bounded memory;
18// a streaming (file-backed) writer that lifts the budget entirely is the named next rung.
19//   usage: nx_web_shard_compact <domain> [budget-MB]
20// license_tier: ORIGINAL
21import "nx_docportal_search_seg.nx"    // dss_prefix + nx_seg_store (ss_* primitives)
22const WSC_MAGIC_1125899906842597: i64 = 1125899906842597
23const WSC_MAGIC_1048576: i64 = 1048576
24const WSC_MAGIC_65536: i64 = 65536
25
26const WSC_MAXSEGS: i64 = 8192        // manifest lines we can fold in one run
27const WSC_MAXENT:  i64 = 134217728  // hard sanity ceiling on entries (2^27; ~an order past any near plan)
28const WSC_HMUL:    i64 = 7046029254386353131  // odd 63-bit multiplicative-mix constant (same family as pagerank_build)
29const WSC_BUDGET_SHARE: i64 = 32        // /32 (was /16) 2026-07-30: the /16 reasoning below is RIGHT about the 9-10x amplification but it reserves a share of MemAvailable AT START on a box with OTHER GROWING CONSUMERS (torrent stack, jellyfin, syno services) -- the share is stale the moment it is computed, and it assumes this job is the only claimant. MEASURED TODAY: with ~25GB available at start the /16 budget projected a peak of ~15GB and this organ was caught at 21GB RSS with 536MB free, minutes from OOM, taking mgmt + the tools daemon down with it. /32 projects ~7-8GB, which also keeps it clear of the 16GiB RLIMIT_AS ceiling now imposed by hc_spawn_searchpipe_job -- TWO CONTROLS THAT CONTRADICT EACH OTHER ARE ONE CONTROL AND ONE OUTAGE, so the soft budget must project a peak BELOW the hard cap. Original note, still accurate on the amplification: fold budget = MemAvailable/16. MEASURED 2026-07-29 (death #3, OOM in ss_build_terms): a 1.88GB run took MemAvailable from 19.4GB to 1.2GB before the kill => PEAK TOUCH ~9-10x run bytes (occ arrays + pairs + writer + key tables + input page cache), not the ~6x the seq1048 memo estimated. /16 caps projected peak at ~60% of MemAvailable so the box keeps serving
30const WSC_BUDGET_FALLBACK: i64 = 268435456  // 256MB: used ONLY when /proc/meminfo is unreadable -- conservative enough to be safe on any box that boots this stack, keeps the pinned no-arg hostctl sub from wedging into a permanent refuse
31const WSC_BUDGET_FLOOR: i64 = 67108864  // 64MB: below this the box is too tight to merge anything safely -> refuse loudly rather than thrash
32const WSC_LOCK_EX: i64 = 2              // flock LOCK_EX on <prefix>plock -- the plane-lock discipline segguard + nx_seg_compact_cli already honor; a lockless manifest swap under a concurrent fold LOSES rows (the proven seq349 race)
33const WSC_MEMINFO_BUF: i64 = 8192       // scratch size for the bounded /proc/meminfo read (file is ~1.5KB)
34const WSC_MEMINFO_CAP: i64 = 7900       // read cap under the scratch -- headroom below WSC_MEMINFO_BUF
35const WSC_KB: i64 = 1024                // /proc/meminfo reports MemAvailable in kB
36
37func wc9_puts(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } sys_write(1, s, n); return 0 }
38func wc9_num(v: i64) -> i64 {
  let bb: *u8 = sys_mmap(28); var m: i64 = v
  if m < 0 { sys_write(1, "-" as *u8, 1); m = 0 - m }
  let t: *u8 = sys_mmap(28); var k: i64 = 0
  if m == 0 { t[0] = 48 as u8; k = 1 }
  while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
  var i: i64 = 0; while i < k { bb[i] = t[k - 1 - i]; i = i + 1 }
  sys_write(1, bb, k); return 0
46}
47// polynomial hash of key bytes -> mixed slot (same ci_hash polynomial so behavior is well-understood).
48// hshift/hmask parameterized: the table is sized DATA-DRIVEN from a first-pass entry count.
49func wsc_slot(k: *u8, kl: i64, hshift: i64, hmask: i64) -> i64 {
  var h: i64 = WSC_MAGIC_1125899906842597
  var i: i64 = 0
  while i < kl { h = (h * 131) + (k[i] as i64); i = i + 1 }
  var s: i64 = h * WSC_HMUL
  s = s & 0x7fffffffffffffff
  return (s >> hshift) & hmask
56}
57// MemAvailable in BYTES from /proc/meminfo -- the kernel's own "allocatable without swapping" figure.
58// procfs files report size 0 to lseek, so ss_readall (lseek-sized buffer) CANNOT read them (its 64KB
59// reads would overflow a 64-byte buffer); this is a bounded read loop into a fixed 8KB scratch instead.
60// Returns -1 if unreadable or unparseable -- callers must treat that as "cannot judge", never as 0.
61func wsc_memavail() -> i64 {
  let fd: i64 = sys_openat_rd("/proc/meminfo" as *u8)
  if fd < 0 { return 0 - 1 }
  let buf: *u8 = sys_mmap(WSC_MEMINFO_BUF)
  var got: i64 = 0
  var n: i64 = 1
  while n > 0 {
      if got >= WSC_MEMINFO_CAP { n = 0 } else {
          n = sys_read(fd, (buf as i64 + got) as *u8, WSC_MEMINFO_CAP - got)
          if n > 0 { got = got + n }
      }
  }
  sys_close(fd)
  if got <= 0 { return 0 - 1 }
  let pat: *u8 = "MemAvailable:" as *u8
  var i: i64 = 0
  while i + 13 < got {
      var hit: i64 = 1
      var j: i64 = 0
      while j < 13 { if buf[i + j] != pat[j] { hit = 0; j = 13 } else { j = j + 1 } }
      if hit == 1 {
          var v: i64 = 0
          var seen: i64 = 0
          var k2: i64 = i + 13
          while k2 < got {
              let c: i64 = buf[k2] as i64
              if c == 10 { k2 = got } else {
                  if c >= 48 { if c <= 57 { v = v * 10 + (c - 48); seen = 1 } }
                  k2 = k2 + 1
              }
          }
          if seen == 1 { return v * WSC_KB }
          return 0 - 1
      }
      i = i + 1
  }
  return 0 - 1
98}
99
100func main(argc: i64, argv: *i64) -> i64 {
  var domain: *u8 = "web" as *u8
  if argc >= 2 { domain = argv[1] as *u8 }
  let prefix: *u8 = sys_mmap(512)
  dss_prefix(domain, prefix)
  // PLANE LOCK FIRST: serialize vs segguard + nx_seg_compact_cli (both flock <prefix>plock).
  // Blocking LOCK_EX -- a short wait behind a sibling fold is correct; running beside one is not.
  let lkp: *u8 = sys_mmap(512)
  var lo: i64 = 0
  lo = ss_cat(lkp, lo, prefix)
  lo = ss_cat(lkp, lo, "plock" as *u8)
  lkp[lo] = 0 as u8
  let lfd: i64 = sys_openat_append(lkp, 0x1a4)
  if lfd < 0 { wc9_puts("cannot open plane lock\nverdict=FAIL\n" as *u8); return 13 }
  if sys_flock(lfd, WSC_LOCK_EX) != 0 { wc9_puts("cannot take plane lock\nverdict=FAIL\n" as *u8); return 13 }
  let segs: *i64 = sys_mmap(8 * WSC_MAXSEGS) as *i64
  let ns: i64 = ss_manifest_cap(prefix, segs, WSC_MAXSEGS)
  wc9_puts("=== nx_web_shard_compact " as *u8); wc9_puts(prefix); wc9_puts(" segments=" as *u8); wc9_num(ns); wc9_puts(" ===\n" as *u8)
  if ns <= 0 { wc9_puts("no live segments\nverdict=FAIL\n" as *u8); return 1 }
  if ns == 1 { wc9_puts("already a single segment -- nothing to fold\nverdict=PASS\n" as *u8); return 0 }
120
  // fold-run byte budget: argv[2] MB if given, else MemAvailable/WSC_BUDGET_SHARE, else fallback.
  var budget: i64 = 0
  if argc >= 3 {
      let a2c: *u8 = argv[2] as *u8
      var bi: i64 = 0
      while a2c[bi] != (0 as u8) { let c: i64 = a2c[bi] as i64; if c >= 48 { if c <= 57 { budget = budget * 10 + (c - 48) } } bi = bi + 1 }
      budget = budget * WSC_MAGIC_1048576
  }
  if budget <= 0 {
      let ma: i64 = wsc_memavail()
      if ma > 0 { budget = ma / WSC_BUDGET_SHARE }
      if ma <= 0 { budget = WSC_BUDGET_FALLBACK; wc9_puts("MemAvailable unreadable -> conservative fallback budget\n" as *u8) }
  }
  wc9_puts("fold-budget bytes=" as *u8); wc9_num(budget); wc9_puts("\n" as *u8)
  if budget < WSC_BUDGET_FLOOR { wc9_puts("budget below floor -- box too tight to merge safely, refusing\nverdict=FAIL\n" as *u8); return 11 }
136
  // fresh merged segid = manifest max + 1 (numeric part of "seg-<id>")
  var segid: i64 = 1
  var si: i64 = 0
  while si < ns {
      let nm: *u8 = segs[si] as *u8
      var v: i64 = 0; var ci: i64 = 0
      while nm[ci] != (0 as u8) { let c: i64 = nm[ci] as i64; if c >= 48 { if c <= 57 { v = v*10 + (c-48) } } ci = ci + 1 }
      if v >= segid { segid = v + 1 }
      si = si + 1
  }
147
  // read every live segment (chronological = manifest order)
  let bptrs: *i64 = sys_mmap(8 * WSC_MAXSEGS) as *i64
  let bszs: *i64 = sys_mmap(8 * WSC_MAXSEGS) as *i64
  var total: i64 = 0
  var s: i64 = 0
  while s < ns {
      let path: *u8 = sys_mmap(512)
      var o: i64 = 0
      o = ss_cat(path, o, prefix)
      o = ss_cat(path, o, segs[s] as *u8)
      o = ss_cat(path, o, ".docs" as *u8)
      path[o] = 0 as u8
      let szp: *i64 = sys_mmap(16) as *i64
      bptrs[s] = ss_loadfile(path, szp, 1) as i64   // mmap-open the input segments: sequential read, low RSS
      bszs[s] = szp[0]
      if bszs[s] < 0 { bszs[s] = 0 }
      total = total + bszs[s]
      s = s + 1
  }
167
  // SUFFIX-RUN SELECTION: longest newest contiguous run with run-bytes <= budget. kk = first
  // manifest index INSIDE the run; [0..kk-1] are kept verbatim, [kk..ns-1] fold into one segment.
  var kk: i64 = ns - 1
  var runbytes: i64 = bszs[ns - 1]
  var walking: i64 = 1
  while walking == 1 {
      if kk == 0 { walking = 0 } else {
          if runbytes + bszs[kk - 1] <= budget { runbytes = runbytes + bszs[kk - 1]; kk = kk - 1 } else { walking = 0 }
      }
  }
  if ns - kk < 2 {
      wc9_puts("newest two segments exceed the fold budget -- nothing foldable at this budget (streaming writer is the next rung)\nverdict=FAIL\n" as *u8)
      return 12
  }
  wc9_puts("fold-run: keep=" as *u8); wc9_num(kk); wc9_puts(" fold=" as *u8); wc9_num(ns - kk)
  wc9_puts(" run-bytes=" as *u8); wc9_num(runbytes); wc9_puts(" total-bytes=" as *u8); wc9_num(total); wc9_puts("\n" as *u8)
184
  // PASS 1 over the RUN ONLY: count real entries (record framing walk, no fixed guess)
  var nent: i64 = 0
  var s0: i64 = kk
  while s0 < ns {
      let b0: *u8 = bptrs[s0] as *u8
      let sz0: i64 = bszs[s0]
      var i0: i64 = 0
      while i0 + 9 <= sz0 {
          let kl0: i64 = ss_r32(b0, i0 + 1)
          let vl0: i64 = ss_r32(b0, i0 + 5 + kl0)
          i0 = i0 + 5 + kl0 + 4 + vl0
          nent = nent + 1
      }
      s0 = s0 + 1
  }
  if nent > WSC_MAXENT { wc9_puts("entry count exceeds WSC_MAXENT sanity ceiling\nverdict=FAIL\n" as *u8); return 2 }
  let maxk: i64 = nent + 16
  // hash table = smallest power of two >= 4x entries (load factor <= 0.25)
  var hsize: i64 = WSC_MAGIC_1048576
  var hbits: i64 = 20
  while hsize < nent * 4 { hsize = hsize * 2; hbits = hbits + 1 }
  let hshift: i64 = 63 - hbits
  let hmask: i64 = hsize - 1
  wc9_puts("pass1 entries=" as *u8); wc9_num(nent); wc9_puts(" hash-slots=" as *u8); wc9_num(hsize); wc9_puts("\n" as *u8)
209
  // entry tables (dense) + hash slot -> entry-index (+1 so 0 = empty)
  let tkp: *i64 = sys_mmap(8 * maxk) as *i64
  let tkl: *i64 = sys_mmap(8 * maxk) as *i64
  let tkind: *i64 = sys_mmap(8 * maxk) as *i64
  let tvp: *i64 = sys_mmap(8 * maxk) as *i64
  let tvl: *i64 = sys_mmap(8 * maxk) as *i64
  let hslot: *i64 = sys_mmap(8 * hsize) as *i64
  var nk: i64 = 0
  var entries: i64 = 0
  s = kk
  while s < ns {
      let b: *u8 = bptrs[s] as *u8
      let sz: i64 = bszs[s]
      var i: i64 = 0
      while i + 9 <= sz {
          let kind: i64 = b[i]
          let kl: i64 = ss_r32(b, i + 1)
          let koff: i64 = i + 5
          let vl: i64 = ss_r32(b, koff + kl)
          let voff: i64 = koff + kl + 4
          let kp: *u8 = (b as i64 + koff) as *u8
          // hash-probe for this key (byte-verify on hit; O(1) expected)
          var slot: i64 = wsc_slot(kp, kl, hshift, hmask)
          var hit: i64 = 0 - 1
          var probing: i64 = 1
          while probing == 1 {
              let e: i64 = hslot[slot]
              if e == 0 { probing = 0 } else {
                  if ss_kcmp(tkp[e - 1] as *u8, tkl[e - 1], kp, kl) == 0 { hit = e - 1; probing = 0 } else {
                      slot = (slot + 1) & hmask
                  }
              }
          }
          if hit < 0 {
              if nk >= maxk { wc9_puts("key table overflow (impossible: data-driven bound)\nverdict=FAIL\n" as *u8); return 3 }
              hit = nk
              hslot[slot] = nk + 1
              nk = nk + 1
          }
          tkp[hit] = kp as i64
          tkl[hit] = kl
          tkind[hit] = kind
          tvp[hit] = (b as i64) + voff
          tvl[hit] = vl
          entries = entries + 1
          i = voff + vl
      }
      s = s + 1
  }
  wc9_puts("entries=" as *u8); wc9_num(entries); wc9_puts(" live-keys=" as *u8); wc9_num(nk)
  wc9_puts(" run-bytes-in=" as *u8); wc9_num(runbytes); wc9_puts("\n" as *u8)
261
  // merged segment (latest per key, insertion order = first-seen key order, matching ss_compact)
  let w: *i64 = ss_begin_cap(runbytes + WSC_MAGIC_65536)
  var t2: i64 = 0
  while t2 < nk {
      if ss_add2(w, tkind[t2], tkp[t2] as *u8, tkl[t2], tvp[t2] as *u8, tvl[t2]) != 0 { wc9_puts("writer overflow\nverdict=FAIL\n" as *u8); return 4 }
      t2 = t2 + 1
  }
  wc9_puts("merge-write: emitting seg (docs+idx+pos+imp)...\n" as *u8)
  if ss_write_seg(prefix, w, segid) != 0 { wc9_puts("segment write failed\nverdict=FAIL\n" as *u8); return 5 }
  wc9_puts("merged segment written\n" as *u8)
272
  // archive retired names BEFORE the swap (crash-safe ordering, identical to ss_compact)
  let ap: *u8 = sys_mmap(512)
  var ao: i64 = 0
  ao = ss_cat(ap, ao, prefix)
  ao = ss_cat(ap, ao, "manifest-archive.txt" as *u8)
  ap[ao] = 0 as u8
  let afd: i64 = sys_openat_append(ap, 0x1a4)
  if afd < 0 { wc9_puts("archive open failed\nverdict=FAIL\n" as *u8); return 6 }
  s = kk
  while s < ns {
      let nm2: *u8 = segs[s] as *u8
      sys_write(afd, nm2, ss_len(nm2))
      sys_write(afd, "\n" as *u8, 1)
      s = s + 1
  }
  sys_fsync(afd)
  sys_close(afd)
  // atomic manifest swap -> kept prefix lines [0..kk-1] + the merged segment (order preserved)
  let mf: *u8 = sys_mmap(512)
  let mt: *u8 = sys_mmap(512)
  var o2: i64 = 0
  o2 = ss_cat(mf, o2, prefix)
  o2 = ss_cat(mf, o2, "manifest.txt" as *u8)
  mf[o2] = 0 as u8
  o2 = 0
  o2 = ss_cat(mt, o2, prefix)
  o2 = ss_cat(mt, o2, "manifest.tmp" as *u8)
  mt[o2] = 0 as u8
  let nb: *u8 = sys_mmap(40 * ns + 256)
  var no: i64 = 0
  var km: i64 = 0
  while km < kk {
      no = ss_cat(nb, no, segs[km] as *u8)
      nb[no] = 10 as u8
      no = no + 1
      km = km + 1
  }
  no = ss_cat(nb, no, "seg-" as *u8)
  no = ss_catn(nb, no, segid)
  nb[no] = 10 as u8
  no = no + 1
  if ss_writefile(mt, nb, no) != 0 { wc9_puts("manifest tmp write failed\nverdict=FAIL\n" as *u8); return 7 }
  if sys_renameat(mt, mf) != 0 { wc9_puts("manifest swap failed\nverdict=FAIL\n" as *u8); return 8 }
  ss_syncdir(prefix)
317
  // POST-SWAP VERIFY: reopen; every live key must resolve via ss_hget to the SAME bytes we merged
  // (sampled stride keeps this O(nk/stride); stride 97 ~= 1% coverage floor + first/last always).
  // ss_open2(...,1) = mmap open: the read-all ss_open pulls the WHOLE shard into anon RAM -- the
  // same OOM class this rewrite exists to kill; verify must not die of the disease it cures.
  let h2: *i64 = ss_open2(prefix, 1)
  if (h2 as i64) == 0 { wc9_puts("VERIFY reopen failed\nverdict=FAIL\n" as *u8); return 9 }
  let vp: *i64 = sys_mmap(16) as *i64
  let vl2: *i64 = sys_mmap(16) as *i64
  let kbuf: *u8 = sys_mmap(600)
  var bad: i64 = 0
  var checked: i64 = 0
  var t3: i64 = 0
  while t3 < nk {
      var pick: i64 = 0
      if t3 == 0 { pick = 1 }
      if t3 == nk - 1 { pick = 1 }
      if (t3 % 97) == 0 { pick = 1 }
      if pick == 1 { if tkind[t3] == 1 { if tkl[t3] < 590 {
          let sp2: *u8 = tkp[t3] as *u8
          var x: i64 = 0
          while x < tkl[t3] { kbuf[x] = sp2[x]; x = x + 1 }
          kbuf[tkl[t3]] = 0 as u8
          checked = checked + 1
          if ss_hget(h2, kbuf, vp, vl2) != 1 { bad = bad + 1 } else {
              if vl2[0] != tvl[t3] { bad = bad + 1 } else {
                  let a2: *u8 = vp[0] as *u8
                  let b2: *u8 = tvp[t3] as *u8
                  var y: i64 = 0
                  var mism: i64 = 0
                  while y < vl2[0] { if a2[y] != b2[y] { mism = 1; y = vl2[0] } else { y = y + 1 } }
                  if mism == 1 { bad = bad + 1 }
              }
          }
      } } }
      t3 = t3 + 1
  }
  wc9_puts("COMPACT done: segments " as *u8); wc9_num(ns); wc9_puts(" -> " as *u8); wc9_num(kk + 1); wc9_puts(" (folded tail -> seg-" as *u8); wc9_num(segid)
  wc9_puts(") live-keys=" as *u8); wc9_num(nk)
  wc9_puts(" verify-checked=" as *u8); wc9_num(checked); wc9_puts(" verify-bad=" as *u8); wc9_num(bad); wc9_puts("\n" as *u8)
  if bad > 0 {
      wc9_puts("VERIFY FAILED -- restore <prefix>manifest.txt from the retired list in manifest-archive.txt\nverdict=FAIL\n" as *u8)
      return 10
  }
  wc9_puts("verdict=PASS\n" as *u8)
  return 0
363}
nx_web_shard_compact.nx source

`nx_web_shard_compact.nx` source
