`nx_coedit.nx` source

↩ module page · 780 lines · 33873 B
1// nx_coedit.nx -- COLLABORATIVE-EDITING KERNEL (office census OF-C1's engine half): an RGA text CRDT.
2// Multiple replicas edit the SAME document concurrently; their op logs MERGE deterministically -- the same
3// final text on every replica REGARDLESS of merge order (convergence), no lost updates, tombstoned deletes,
4// deterministic (counter,replica) tie-break for concurrent inserts at the same point.
5//
6// OP LOG (append-only, LF lines; the natural fit for the additive-only store doctrine):
7//   I <replica> <counter> <afterid> <char>     insert ONE byte after node <afterid> ("R" = document root)
8//   D <replica> <counter> <targetid>           tombstone node <targetid>
9//   ids are "<replica>:<counter>"; an op's id must be UNIQUE (same id + different content = LOUD protocol
10//   violation; identical duplicate = idempotent skip, which is what makes re-merging safe).
11//   <char> is one byte, escaped: \n=newline \t=tab \s=space \\=backslash (any other 1 byte literal).
12//
13// COMMANDS
14//   apply <log> <out.txt>                      materialize (RGA linearization) -> file + verdict line
15//   ins <log> <rep> <ctr> <afterid> <char>     append a raw insert op
16//   del <log> <rep> <ctr> <targetid>           append a raw delete op
17//   merge <A> <B> <out>                        union by op id (idempotent dedup; conflicting id = LOUD)
18//   edit <log> <rep> <pos> ins <text>          position-based convenience: per-BYTE ops chained after the
19//   edit <log> <rep> <pos> del <count>         visible byte at pos-1 / tombstones for [pos,pos+count)
20//
21// HONEST SCOPE (v1, named): the KERNEL -- byte-granular ops (UTF-8 sequences survive as consecutive ops but
22// byte-position edits must not split a sequence), logs are plain files, ops cap 65536/doc. LIVE multi-client
23// sync (the R2 daemon + browser surface) is NOT built yet; the census OF-C1 axis stays ABSENT until R2 --
24// its needle requires the R2 sync function DEFINITION, which this file intentionally does not contain.
25// license_tier: ORIGINAL
26import "nx_syscalls.nx"
27const CD_MAGIC_1000003: i64 = 1000003
28const CD_MAGIC_10007: i64 = 10007
29const CD_MAGIC_12289: i64 = 12289
30const CD_MAGIC_4194320: i64 = 4194320
31const CD_MAGIC_4194304: i64 = 4194304
32const CD_MAGIC_4294967296: i64 = 4294967296
33const CD_MAGIC_8388640: i64 = 8388640
34const CD_MAGIC_8388608: i64 = 8388608
35const CD_MAGIC_8388600: i64 = 8388600
36
37const CD_MAXOPS: i64 = 65536
38const CD_HASHN:  i64 = 262144   // open-addressing id -> node-idx+1 (power of two)
39
40func cd_slen(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8) { n=n+1 } return n }
41func cd_puts(s: *u8) -> i64 { sys_write(1, s, cd_slen(s)); return 0 }
42func cd_num(v: i64) -> i64 {
  let b: *u8=sys_mmap(28); var m: i64=v; if m<0 { m=0-m; sys_write(1,"-" as *u8,1) }
  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 { b[i]=t[k-1-i]; i=i+1 } sys_write(1,b,k); return 0
47}
48func cd_streq(a: *u8, b: *u8) -> i64 { var i: i64=0; while a[i]!=(0 as u8) { if a[i]!=b[i] { return 0 } i=i+1 } if b[i]!=(0 as u8) { return 0 } return 1 }
49func cd_cat(dst: *u8, off: i64, s: *u8) -> i64 { var i: i64=0; while s[i]!=(0 as u8) { dst[off+i]=s[i]; i=i+1 } return off+i }
50func cd_uint(dst: *u8, off: i64, v: i64) -> i64 {
  var m: i64=v; 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 o: i64=off; var i: i64=0; while i<k { dst[o]=t[k-1-i]; o=o+1; i=i+1 } return o
54}
55func cd_readfile(path: *u8, buf: *u8, cap: i64) -> i64 {
  let fd: i64=sys_openat_rd(path); if fd<0 { return 0-1 }
  var tot: i64=0; while tot<cap { let r: i64=sys_read(fd,(buf as i64+tot) as *u8,cap-tot); if r<=0 { break } tot=tot+r }
  sys_close(fd); return tot
59}
60func cd_appendfile(path: *u8, line: *u8, n: i64) -> i64 {
  let fd: i64=sys_openat_append(path, 0x1a4); if fd<0 { return 0-1 }
  sys_write(fd, line, n); sys_close(fd); return 0
63}
64func cd_writefile(path: *u8, buf: *u8, n: i64) -> i64 {
  let fd: i64=sys_openat_wr(path, 0x1a4); if fd<0 { return 0-1 }
  sys_write(fd, buf, n); sys_close(fd); return 0
67}
68
69// parse an unsigned int at s[*ip..e); stops at non-digit; -1 if no digits
70func cd_int(s: *u8, ip: *i64, e: i64) -> i64 {
  var i: i64=ip[0]
  var v: i64=0
  var nd: i64=0
  while i<e { let c: i64=s[i] as i64; if c>=48 { if c<=57 { v=v*10+(c-48); nd=nd+1; i=i+1 } else { break } } else { break } }
  if nd==0 { return 0-1 }
  ip[0]=i
  return v
78}
79
80// ---- the replica state: parallel node arrays + an id hash + per-parent child lists (sorted DESC by id) ----
81struct CdDoc {
  nrep: *i64,      // node replica id
  nctr: *i64,      // node counter
  nch: *u8,        // node byte
  ntomb: *u8,      // tombstone flag
  cfirst: *i64,    // per node: first child idx (or 0 = none; node 0 = ROOT sentinel)
  cnext: *i64,     // per node: next sibling idx (or 0)
  hsh: *i64,       // id hash -> node idx+1
  n: i64,          // node count (incl ROOT at 0)
  err: i64         // 0 ok | 1 malformed | 2 bad-ref | 3 conflicting-duplicate | 4 cap
91}
92
93func cd_new() -> *CdDoc {
  let d: *CdDoc = sys_mmap(96) as *CdDoc
  d.nrep = sys_mmap(8*CD_MAXOPS) as *i64
  d.nctr = sys_mmap(8*CD_MAXOPS) as *i64
  d.nch  = sys_mmap(CD_MAXOPS)
  d.ntomb= sys_mmap(CD_MAXOPS)
  d.cfirst=sys_mmap(8*CD_MAXOPS) as *i64
  d.cnext =sys_mmap(8*CD_MAXOPS) as *i64
  d.hsh  = sys_mmap(8*CD_HASHN) as *i64
  d.n = 1                      // idx 0 = ROOT (rep 0, ctr 0)
  d.err = 0
  return d
105}
106
107func cd_hkey(rep: i64, ctr: i64) -> i64 {
  var h: i64 = ctr*CD_MAGIC_1000003 + rep*CD_MAGIC_10007 + CD_MAGIC_12289
  h = h ^ (h >> 15)
  if h < 0 { h = 0 - h }
  return h & (CD_HASHN - 1)
112}
113func cd_hfind(d: *CdDoc, rep: i64, ctr: i64) -> i64 {   // node idx or -1
  if rep==0 { if ctr==0 { return 0 } }
  var h: i64 = cd_hkey(rep, ctr)
  var g: i64 = 0
  while g < CD_HASHN {
      let v: i64 = d.hsh[h]
      if v == 0 { return 0-1 }
      let ix: i64 = v - 1
      if d.nrep[ix]==rep { if d.nctr[ix]==ctr { return ix } }
      h = (h + 1) & (CD_HASHN - 1)
      g = g + 1
  }
  return 0-1
126}
127func cd_hput(d: *CdDoc, rep: i64, ctr: i64, ix: i64) -> i64 {
  var h: i64 = cd_hkey(rep, ctr)
  var g: i64 = 0
  while g < CD_HASHN {
      if d.hsh[h] == 0 { d.hsh[h] = ix + 1; return 0 }
      h = (h + 1) & (CD_HASHN - 1)
      g = g + 1
  }
  return 0-1
136}
137
138// id A(rep,ctr) newer-than B? RGA order: counter first, replica tie-break (bigger = first among siblings)
139func cd_newer(ra: i64, ca: i64, rb: i64, cb: i64) -> i64 {
  if ca > cb { return 1 }
  if ca < cb { return 0 }
  if ra > rb { return 1 }
  return 0
144}
145
146// link node ix under parent pidx at its RGA position (siblings sorted DESC by id)
147func cd_link(d: *CdDoc, pidx: i64, ix: i64) -> i64 {
  let head: i64 = d.cfirst[pidx]
  if head == 0 { d.cfirst[pidx] = ix; d.cnext[ix] = 0; return 0 }
  if cd_newer(d.nrep[ix], d.nctr[ix], d.nrep[head], d.nctr[head]) == 1 {
      d.cnext[ix] = head
      d.cfirst[pidx] = ix
      return 0
  }
  var cur: i64 = head
  var run: i64 = 1
  while run == 1 {
      let nx: i64 = d.cnext[cur]
      if nx == 0 { d.cnext[cur] = ix; d.cnext[ix] = 0; run = 0 } else {
          if cd_newer(d.nrep[ix], d.nctr[ix], d.nrep[nx], d.nctr[nx]) == 1 {
              d.cnext[ix] = nx
              d.cnext[cur] = ix
              run = 0
          } else { cur = nx }
      }
  }
  return 0
168}
169
170// unescape a 1-byte payload at s[i..e): returns the byte in out[0]; -1 malformed
171func cd_unesc(s: *u8, i: i64, e: i64, out: *u8) -> i64 {
  if i >= e { return 0-1 }
  let c: i64 = s[i] as i64
  if c == 92 {                              // backslash escape: must be exactly two bytes
      if i+2 != e { return 0-1 }
      let c2: i64 = s[i+1] as i64
      if c2==110 { out[0]=10 as u8; return 0 }
      if c2==116 { out[0]=9 as u8; return 0 }
      if c2==115 { out[0]=32 as u8; return 0 }
      if c2==92 { out[0]=92 as u8; return 0 }
      return 0-1
  }
  if i+1 != e { return 0-1 }                // must be exactly one byte
  out[0]=c as u8
  return 0
186}
187
188// parse an id token "R" or "<rep>:<ctr>" at s[*ip..e) (stops at space/EOL); rc[0]=rep rc[1]=ctr; -1 malformed
189func cd_idtok(s: *u8, ip: *i64, e: i64, rc: *i64) -> i64 {
  var i: i64 = ip[0]
  if i < e { if s[i]==(82 as u8) {          // "R"
      var e2: i64 = i+1
      var isend: i64 = 0
      if e2 >= e { isend=1 } else { if s[e2]==(32 as u8) { isend=1 } }
      if isend==1 { rc[0]=0; rc[1]=0; ip[0]=i+1; return 0 }
  } }
  let rep: i64 = cd_int(s, ip, e)
  if rep < 0 { return 0-1 }
  if ip[0] >= e { return 0-1 }
  if s[ip[0]] != (58 as u8) { return 0-1 }
  ip[0] = ip[0] + 1
  let ctr: i64 = cd_int(s, ip, e)
  if ctr < 0 { return 0-1 }
  rc[0]=rep
  rc[1]=ctr
  return 0
207}
208
209// load an op log into the doc (two data passes packed into one: inserts create nodes with parent RESOLVED
210// on the spot when known, else deferred -- but merge always concatenates whole causally-ordered logs, so a
211// forward parent means the SIBLING log comes later; handle it with a deferred second pass).
212func cd_load(d: *CdDoc, log: *u8, ln: i64) -> i64 {
  // deferred parents: entries (nodeidx, prep, pctr)
  let dfi: *i64=sys_mmap(8*CD_MAXOPS) as *i64
  let dfr: *i64=sys_mmap(8*CD_MAXOPS) as *i64
  let dfc: *i64=sys_mmap(8*CD_MAXOPS) as *i64
  var ndf: i64=0
  // deferred deletes: (trep,tctr)
  let ddr: *i64=sys_mmap(8*CD_MAXOPS) as *i64
  let ddc: *i64=sys_mmap(8*CD_MAXOPS) as *i64
  var ndd: i64=0
  let rc: *i64=sys_mmap(32) as *i64
  let bch: *u8=sys_mmap(8)
  var i: i64=0
  while i<ln {
      var e: i64=i
      while e<ln { if log[e]==(10 as u8) { break } e=e+1 }
      var end: i64=e
      if end>i { if log[end-1]==(13 as u8) { end=end-1 } }
      if end>i+1 {
          let kind: i64=log[i] as i64
          if log[i+1] != (32 as u8) { d.err=1; return 0-1 }
          let ip: *i64=sys_mmap(16) as *i64
          ip[0]=i+2
          if kind==73 {                     // I rep ctr afterid char
              let rep: i64=cd_int(log, ip, end)
              if rep<1 { d.err=1; return 0-1 }
              if ip[0]>=end { d.err=1; return 0-1 }
              ip[0]=ip[0]+1
              let ctr: i64=cd_int(log, ip, end)
              if ctr<1 { d.err=1; return 0-1 }
              if ip[0]>=end { d.err=1; return 0-1 }
              ip[0]=ip[0]+1
              if cd_idtok(log, ip, end, rc)<0 { d.err=1; return 0-1 }
              if ip[0]>=end { d.err=1; return 0-1 }
              ip[0]=ip[0]+1
              if cd_unesc(log, ip[0], end, bch)<0 { d.err=1; return 0-1 }
              // dedup / conflict
              let ex: i64=cd_hfind(d, rep, ctr)
              if ex>=0 {
                  if d.nch[ex]==bch[0] { i=e+1 } else { d.err=3; return 0-1 }
              } else {
                  if d.n>=CD_MAXOPS { d.err=4; return 0-1 }
                  let ix: i64=d.n
                  d.n=d.n+1
                  d.nrep[ix]=rep
                  d.nctr[ix]=ctr
                  d.nch[ix]=bch[0]
                  d.ntomb[ix]=0 as u8
                  d.cfirst[ix]=0
                  d.cnext[ix]=0
                  cd_hput(d, rep, ctr, ix)
                  let pix: i64=cd_hfind(d, rc[0], rc[1])
                  if pix>=0 { cd_link(d, pix, ix) } else {
                      dfi[ndf]=ix
                      dfr[ndf]=rc[0]
                      dfc[ndf]=rc[1]
                      ndf=ndf+1
                  }
                  i=e+1
              }
          } else { if kind==68 {            // D rep ctr targetid
              let rep2: i64=cd_int(log, ip, end)
              if rep2<1 { d.err=1; return 0-1 }
              if ip[0]>=end { d.err=1; return 0-1 }
              ip[0]=ip[0]+1
              let ctr2: i64=cd_int(log, ip, end)
              if ctr2<1 { d.err=1; return 0-1 }
              if ip[0]>=end { d.err=1; return 0-1 }
              ip[0]=ip[0]+1
              if cd_idtok(log, ip, end, rc)<0 { d.err=1; return 0-1 }
              ddr[ndd]=rc[0]
              ddc[ndd]=rc[1]
              ndd=ndd+1
              i=e+1
          } else { d.err=1; return 0-1 } }
      } else { i=e+1 }
  }
  // resolve deferred parents (forward refs across concatenated logs)
  var k: i64=0
  while k<ndf {
      let pix2: i64=cd_hfind(d, dfr[k], dfc[k])
      if pix2<0 { d.err=2; return 0-1 }
      cd_link(d, pix2, dfi[k])
      k=k+1
  }
  // apply tombstones
  k=0
  while k<ndd {
      let tix: i64=cd_hfind(d, ddr[k], ddc[k])
      if tix<1 { d.err=2; return 0-1 }     // ROOT or missing = bad target
      d.ntomb[tix]=1 as u8
      k=k+1
  }
  return 0
306}
307
308// RGA linearization: preorder DFS (node byte, then children newest-first), tombstones skipped but traversed.
309// out gets the visible bytes; vis (optional, may be 0) gets the node idx per visible byte. Returns length.
310func cd_linear(d: *CdDoc, out: *u8, vis: *i64) -> i64 {
  let stk: *i64=sys_mmap(8*CD_MAXOPS) as *i64
  var sp: i64=0
  var o: i64=0
  // seed: ROOT's children pushed in REVERSE list order so the first child pops first
  // (list is desc; we want to VISIT desc order; push onto stack reversed = collect then push backward)
  let tmp: *i64=sys_mmap(8*CD_MAXOPS) as *i64
  var tn: i64=0
  var c: i64=d.cfirst[0]
  while c != 0 { tmp[tn]=c; tn=tn+1; c=d.cnext[c] }
  var q: i64=tn-1
  while q>=0 { stk[sp]=tmp[q]; sp=sp+1; q=q-1 }
  while sp>0 {
      sp=sp-1
      let ix: i64=stk[sp]
      if (d.ntomb[ix] as i64)==0 {
          out[o]=d.nch[ix]
          if (vis as i64)!=0 { vis[o]=ix }
          o=o+1
      }
      // push children reversed
      tn=0
      c=d.cfirst[ix]
      while c != 0 { tmp[tn]=c; tn=tn+1; c=d.cnext[c] }
      q=tn-1
      while q>=0 { stk[sp]=tmp[q]; sp=sp+1; q=q-1 }
  }
  return o
338}
339
340func cd_loadfile(d: *CdDoc, path: *u8) -> i64 {
  let buf: *u8=sys_mmap(CD_MAGIC_4194320)
  let n: i64=cd_readfile(path, buf, CD_MAGIC_4194304)
  if n<0 { return 0-9 }
  return cd_load(d, buf, n)
345}
346
347// emit one escaped byte into dst
348func cd_escput(dst: *u8, off: i64, b: i64) -> i64 {
  var o: i64=off
  if b==10 { o=cd_cat(dst,o,"\\n" as *u8); return o }
  if b==9 { o=cd_cat(dst,o,"\\t" as *u8); return o }
  if b==32 { o=cd_cat(dst,o,"\\s" as *u8); return o }
  if b==92 { o=cd_cat(dst,o,"\\\\" as *u8); return o }
  dst[o]=b as u8
  return o+1
356}
357
358// max counter used by replica `rep` in the log file (0 if none)
359func cd_maxctr(path: *u8, rep: i64) -> i64 {
  let buf: *u8=sys_mmap(CD_MAGIC_4194320)
  let n: i64=cd_readfile(path, buf, CD_MAGIC_4194304)
  if n<0 { return 0 }
  var best: i64=0
  var i: i64=0
  while i<n {
      var e: i64=i
      while e<n { if buf[e]==(10 as u8) { break } e=e+1 }
      if e>i+2 {
          let ip: *i64=sys_mmap(16) as *i64
          ip[0]=i+2
          let r: i64=cd_int(buf, ip, e)
          if r==rep { if ip[0]<e { if buf[ip[0]]==(32 as u8) {
              ip[0]=ip[0]+1
              let c: i64=cd_int(buf, ip, e)
              if c>best { best=c }
          } } }
      }
      i=e+1
  }
  return best
381}
382
383func cd_fail(msg: *u8, code: i64) -> i64 {
  cd_puts("COEDIT-FAIL " as *u8)
  cd_puts(msg)
  cd_puts(" err=" as *u8)
  cd_num(code)
  cd_puts(" (1=malformed 2=bad-ref 3=conflicting-duplicate-id 4=cap)\n" as *u8)
  sys_exit(1)
  return 1
391}
392
393// ---- R2 LIVE SYNC: multi-client convergence over HTTP. The canonical op log at <logpath> is the ONE union
394// point; because ops carry globally-unique ids and merge is idempotent + order-independent (proven by the
395// kernel gate), every client that pushes-then-pulls reaches the SAME text. This is what makes OF-C1 real
396// "real-time co-editing" rather than a lone merge engine.
397
398// dedup-append the op lines in body[0,blen) to the canonical log file: keep only ids not already present
399// (I-line "rep ctr", D-line "rep ctr" in a separate namespace). Returns #ops added (or <0 on I/O fail).
400func cd_seen_mark(seen: *i64, rep: i64, ctr: i64, isdel: i64) -> i64 {   // returns 1 if NEW, 0 if already present
  let key: i64=rep*CD_MAGIC_4294967296+ctr*2+isdel+1
  var h: i64=cd_hkey(rep, ctr*2+isdel)
  var g: i64=0
  while g<CD_HASHN {
      let v: i64=seen[h]
      if v==0 { seen[h]=key; return 1 }
      if v==key { return 0 }
      h=(h+1)&(CD_HASHN-1)
      g=g+1
  }
  return 0
412}
413// scan buf[0,blen) as op lines; for each unseen id, mark it and (if emit==1) copy the line into out. returns
414// bytes written to out.
415func cd_scan_dedup(buf: *u8, blen: i64, seen: *i64, out: *u8, emit: i64) -> i64 {
  let ip: *i64=sys_mmap(16) as *i64
  var oa: i64=0
  var i: i64=0
  while i<blen {
      var e: i64=i
      while e<blen { if buf[e]==(10 as u8) { break } e=e+1 }
      var end: i64=e
      if end>i { if buf[end-1]==(13 as u8) { end=end-1 } }
      if end>i+2 {
          var isdel: i64=0
          if buf[i]==(68 as u8) { isdel=1 }
          ip[0]=i+2
          let rep: i64=cd_int(buf, ip, end)
          var ctr: i64=0-1
          if rep>0 { if ip[0]<end { if buf[ip[0]]==(32 as u8) { ip[0]=ip[0]+1; ctr=cd_int(buf, ip, end) } } }
          if ctr>0 {
              if cd_seen_mark(seen, rep, ctr, isdel)==1 {
                  if emit==1 {
                      var q: i64=i
                      while q<end { out[oa]=buf[q]; oa=oa+1; q=q+1 }
                      out[oa]=10 as u8; oa=oa+1
                  }
              }
          }
      }
      i=e+1
  }
  return oa
444}
445func cd_sync_append(logpath: *u8, body: *u8, blen: i64) -> i64 {
  let cur: *u8=sys_mmap(CD_MAGIC_8388640)
  var curn: i64=cd_readfile(logpath, cur, CD_MAGIC_8388608)
  if curn<0 { curn=0 }
  let seen: *i64=sys_mmap(8*CD_HASHN) as *i64
  cd_scan_dedup(cur, curn, seen, 0 as *u8, 0)          // pass 1: mark ids already in the log
  let outapp: *u8=sys_mmap(CD_MAGIC_8388640)
  let oa: i64=cd_scan_dedup(body, blen, seen, outapp, 1) // pass 2: emit only NEW body ops
  if oa>0 {
      let fd: i64=sys_openat_append(logpath, 0x1a4)
      if fd<0 { return 0-1 }
      sys_write(fd, outapp, oa)
      sys_close(fd)
  }
  return oa
460}
461
462// THE live-sync request handler (the census OF-C1 needle). POST => dedup-append body ops to <logpath>; both
463// POST and GET => respond 200 with the FULL canonical merged log as the body (so the client materializes the
464// converged doc). Returns the response length written into out.
465func cd_live_sync(logpath: *u8, is_post: i64, body: *u8, blen: i64, out: *u8, outcap: i64) -> i64 {
  if is_post==1 { cd_sync_append(logpath, body, blen) }
  let logbuf: *u8=sys_mmap(CD_MAGIC_8388640)
  var ln: i64=cd_readfile(logpath, logbuf, CD_MAGIC_8388608)
  if ln<0 { ln=0 }
  var o: i64=0
  o=cd_cat(out, o, "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: " as *u8)
  o=cd_uint(out, o, ln)
  o=cd_cat(out, o, "\r\nConnection: close\r\n\r\n" as *u8)
  var i: i64=0
  while i<ln { out[o]=logbuf[i]; o=o+1; i=i+1 }
  return o
477}
478
479// header terminator offset (index just past \r\n\r\n), or -1 if headers not yet complete
480func cd_hdrend(req: *u8, rn: i64) -> i64 {
  var i: i64=0
  while i+3<rn { if req[i]==(13 as u8) { if req[i+1]==(10 as u8) { if req[i+2]==(13 as u8) { if req[i+3]==(10 as u8) { return i+4 } } } } i=i+1 }
  return 0-1
484}
485// parse the "Content-Length:" header value, searching only the header region; -1 if absent
486func cd_contentlen(req: *u8, he: i64) -> i64 {
  let pat: *u8="Content-Length:" as *u8
  let pl: i64=cd_slen(pat)
  var i: i64=0
  while i+pl<=he {
      var m: i64=1
      var j: i64=0
      while j<pl { if req[i+j]!=pat[j] { m=0; j=pl } else { j=j+1 } }
      if m==1 {
          var k: i64=i+pl
          while k<he { if req[k]==(32 as u8) { k=k+1 } else { break } }
          var v: i64=0
          var nd: i64=0
          while k<he { let c: i64=req[k] as i64; if c>=48 { if c<=57 { v=v*10+(c-48); nd=nd+1; k=k+1 } else { break } } else { break } }
          if nd>0 { return v }
          return 0
      }
      i=i+1
  }
  return 0-1
506}
507// minimal request parse: is it POST? and where is the body (after \r\n\r\n)? sets bp[0]=body offset, bp[1]=len
508func cd_parsereq(req: *u8, rn: i64, bp: *i64) -> i64 {
  var is_post: i64=0
  if rn>=4 { if req[0]==(80 as u8) { if req[1]==(79 as u8) { if req[2]==(83 as u8) { if req[3]==(84 as u8) { is_post=1 } } } } }
  let he: i64=cd_hdrend(req, rn)
  if he<0 { bp[0]=rn; bp[1]=0 } else { bp[0]=he; bp[1]=rn-he }
  return is_post
514}
515
516func cd_addr(out: *u8, port: i64) -> i64 {
  out[0]=2 as u8; out[1]=0 as u8
  out[2]=((port>>8)&0xff) as u8; out[3]=(port&0xff) as u8
  out[4]=127 as u8; out[5]=0 as u8; out[6]=0 as u8; out[7]=1 as u8
  var i: i64=8; while i<16 { out[i]=0 as u8; i=i+1 } return 0
521}
522
523// serve <logpath> <port> [maxconns]: accept-loop live-sync daemon. maxconns=0 => serve forever.
524func cd_serve(logpath: *u8, port: i64, maxconns: i64) -> i64 {
  let addr: *u8=sys_mmap(16)
  cd_addr(addr, port)
  let lfd: i64=sys_socket(AF_INET, SOCK_STREAM, 0)
  if lfd<0 { return 0-1 }
  let one: *i64=sys_mmap(8) as *i64; one[0]=1     // SO_REUSEADDR: rebind through a prior server's TIME_WAIT
  sys_setsockopt(lfd, SOL_SOCKET, SO_REUSEADDR, one as *u8, 4)
  if sys_bind(lfd, addr, 16)<0 { return 0-2 }
  if sys_listen(lfd, 16)<0 { return 0-3 }
  let reqb: *u8=sys_mmap(CD_MAGIC_8388640)
  let resb: *u8=sys_mmap(CD_MAGIC_8388640)
  let bp: *i64=sys_mmap(16) as *i64
  var served: i64=0
  var run: i64=1
  while run==1 {
      let cfd: i64=sys_accept(lfd)
      if cfd<0 { run=0 } else {
          // read the request: loop until headers complete AND Content-Length body bytes are in (or EOF).
          // For CL:0 (a pure pull) the request is complete at end-of-headers -- no blocking wait for a body.
          var rn: i64=0
          var g: i64=1
          while g==1 {
              let he: i64=cd_hdrend(reqb, rn)
              if he>=0 {
                  var cl: i64=cd_contentlen(reqb, he)
                  if cl<0 { cl=0 }
                  if rn>=he+cl { g=0 }
              }
              if g==1 {
                  if rn>=CD_MAGIC_8388600 { g=0 } else {
                      let r: i64=sys_read(cfd, (reqb as i64+rn) as *u8, CD_MAGIC_8388600-rn)
                      if r<=0 { g=0 } else { rn=rn+r }
                  }
              }
          }
          let is_post: i64=cd_parsereq(reqb, rn, bp)
          // clean shutdown: a GET/POST to /shutdown stops the accept loop (so a caller reaps us -- no kill race)
          var isdown: i64=0
          if rn>=13 { var s: i64=0; while s<rn-13 { if reqb[s]==(47 as u8) { if reqb[s+1]==(115 as u8) { if reqb[s+2]==(104 as u8) { if reqb[s+3]==(117 as u8) { if reqb[s+4]==(116 as u8) { isdown=1; s=rn } } } } } s=s+1 } }
          if isdown==1 {
              let ok: *u8="HTTP/1.1 200 OK\r\nContent-Length: 3\r\nConnection: close\r\n\r\nbye" as *u8
              rv_write_all_fd(cfd, ok, cd_slen(ok))
              sys_close(cfd)
              run=0
          } else {
              let on: i64=cd_live_sync(logpath, is_post, (reqb as i64+bp[0]) as *u8, bp[1], resb, CD_MAGIC_8388608)
              rv_write_all_fd(cfd, resb, on)
              sys_close(cfd)
              served=served+1
              if maxconns>0 { if served>=maxconns { run=0 } }
          }
      }
  }
  sys_close(lfd)
  return served
579}
580
581func rv_write_all_fd(fd: i64, buf: *u8, n: i64) -> i64 {
  var off: i64=0
  while off<n {
      let w: i64=sys_write(fd, (buf as i64+off) as *u8, n-off)
      if w<=0 { return 0-1 }
      off=off+w
  }
  return 0
589}
590
591func main(argc: i64, argv: *i64) -> i64 {
  if argc<3 { cd_puts("usage: nx_coedit apply <log> <out.txt> | ins <log> <rep> <ctr> <afterid> <char> | del <log> <rep> <ctr> <targetid> | merge <A> <B> <out> | edit <log> <rep> <pos> ins <text> | edit <log> <rep> <pos> del <count> | serve <log> <port> [maxconns]\n" as *u8); sys_exit(2); return 2 }
  let cmd: *u8=argv[1] as *u8
594
  if cd_streq(cmd,"apply" as *u8)==1 {
      if argc<4 { cd_puts("usage: nx_coedit apply <log> <out.txt>\n" as *u8); sys_exit(2); return 2 }
      let d: *CdDoc=cd_new()
      if cd_loadfile(d, argv[2] as *u8)<0 { return cd_fail("load" as *u8, d.err) }
      let out: *u8=sys_mmap(CD_MAXOPS+16)
      let n: i64=cd_linear(d, out, 0 as *i64)
      if cd_writefile(argv[3] as *u8, out, n)<0 { return cd_fail("write" as *u8, 0) }
      cd_puts("COEDIT-APPLY-OK bytes=" as *u8); cd_num(n); cd_puts(" ops=" as *u8); cd_num(d.n-1); cd_puts("\n" as *u8)
      sys_exit(0); return 0
  }
  if cd_streq(cmd,"ins" as *u8)==1 {
      if argc<7 { cd_puts("usage: nx_coedit ins <log> <rep> <ctr> <afterid> <char>\n" as *u8); sys_exit(2); return 2 }
      let line: *u8=sys_mmap(256)
      var o: i64=cd_cat(line,0,"I " as *u8)
      o=cd_cat(line,o,argv[3] as *u8); line[o]=32 as u8; o=o+1
      o=cd_cat(line,o,argv[4] as *u8); line[o]=32 as u8; o=o+1
      o=cd_cat(line,o,argv[5] as *u8); line[o]=32 as u8; o=o+1
      o=cd_cat(line,o,argv[6] as *u8); line[o]=10 as u8; o=o+1
      if cd_appendfile(argv[2] as *u8, line, o)<0 { return cd_fail("append" as *u8, 0) }
      cd_puts("COEDIT-INS-OK\n" as *u8)
      sys_exit(0); return 0
  }
  if cd_streq(cmd,"del" as *u8)==1 {
      if argc<6 { cd_puts("usage: nx_coedit del <log> <rep> <ctr> <targetid>\n" as *u8); sys_exit(2); return 2 }
      let line: *u8=sys_mmap(256)
      var o: i64=cd_cat(line,0,"D " as *u8)
      o=cd_cat(line,o,argv[3] as *u8); line[o]=32 as u8; o=o+1
      o=cd_cat(line,o,argv[4] as *u8); line[o]=32 as u8; o=o+1
      o=cd_cat(line,o,argv[5] as *u8); line[o]=10 as u8; o=o+1
      if cd_appendfile(argv[2] as *u8, line, o)<0 { return cd_fail("append" as *u8, 0) }
      cd_puts("COEDIT-DEL-OK\n" as *u8)
      sys_exit(0); return 0
  }
  if cd_streq(cmd,"merge" as *u8)==1 {
      if argc<5 { cd_puts("usage: nx_coedit merge <A> <B> <out>\n" as *u8); sys_exit(2); return 2 }
      // validate the union parses cleanly (incl conflict detection), then write A-lines + B-lines deduped by id
      let a: *u8=sys_mmap(CD_MAGIC_4194320)
      let an: i64=cd_readfile(argv[2] as *u8, a, CD_MAGIC_4194304)
      if an<0 { return cd_fail("read-A" as *u8, 0) }
      let b: *u8=sys_mmap(CD_MAGIC_4194320)
      let bn: i64=cd_readfile(argv[3] as *u8, b, CD_MAGIC_4194304)
      if bn<0 { return cd_fail("read-B" as *u8, 0) }
      let both: *u8=sys_mmap(CD_MAGIC_8388640)
      var bo: i64=0
      var i: i64=0
      while i<an { both[bo]=a[i]; bo=bo+1; i=i+1 }
      if bo>0 { if both[bo-1]!=(10 as u8) { both[bo]=10 as u8; bo=bo+1 } }
      i=0
      while i<bn { both[bo]=b[i]; bo=bo+1; i=i+1 }
      let d: *CdDoc=cd_new()
      if cd_load(d, both, bo)<0 { return cd_fail("merge-validate" as *u8, d.err) }
      // dedup pass: keep first occurrence of each op id (I ids from nodes; D lines dedup by (rep,ctr) too)
      let seen: *i64=sys_mmap(8*CD_HASHN) as *i64
      let out: *u8=sys_mmap(CD_MAGIC_8388640)
      var oo: i64=0
      i=0
      while i<bo {
          var e: i64=i
          while e<bo { if both[e]==(10 as u8) { break } e=e+1 }
          if e>i+2 {
              let ip: *i64=sys_mmap(16) as *i64
              ip[0]=i+2
              let rep: i64=cd_int(both, ip, e)
              var ctr: i64=0-1
              if rep>0 { if ip[0]<e { if both[ip[0]]==(32 as u8) { ip[0]=ip[0]+1; ctr=cd_int(both, ip, e) } } }
              var dup: i64=0
              if ctr>0 {
                  var isdel: i64=0
                  if both[i]==(68 as u8) { isdel=1 }
                  let key: i64=rep*CD_MAGIC_4294967296+ctr*2+isdel+1
                  var h: i64=cd_hkey(rep, ctr*2+isdel)
                  var g: i64=0
                  while g<CD_HASHN {
                      let v: i64=seen[h]
                      if v==0 { seen[h]=key; g=CD_HASHN } else {
                          if v==key { dup=1; g=CD_HASHN } else { h=(h+1)&(CD_HASHN-1); g=g+1 }
                      }
                  }
              }
              if dup==0 {
                  var q: i64=i
                  while q<e { out[oo]=both[q]; oo=oo+1; q=q+1 }
                  out[oo]=10 as u8
                  oo=oo+1
              }
          }
          i=e+1
      }
      if cd_writefile(argv[4] as *u8, out, oo)<0 { return cd_fail("write" as *u8, 0) }
      cd_puts("COEDIT-MERGE-OK ops=" as *u8); cd_num(d.n-1); cd_puts("\n" as *u8)
      sys_exit(0); return 0
  }
  if cd_streq(cmd,"edit" as *u8)==1 {
      if argc<6 { cd_puts("usage: nx_coedit edit <log> <rep> <pos> ins <text> | edit <log> <rep> <pos> del <count>\n" as *u8); sys_exit(2); return 2 }
      let ip0: *i64=sys_mmap(16) as *i64
      ip0[0]=0
      let reps: *u8=argv[3] as *u8
      let rep: i64=cd_int(reps, ip0, cd_slen(reps))
      ip0[0]=0
      let poss: *u8=argv[4] as *u8
      let pos: i64=cd_int(poss, ip0, cd_slen(poss))
      if rep<1 { return cd_fail("bad-replica" as *u8, 1) }
      if pos<0 { return cd_fail("bad-pos" as *u8, 1) }
      let mode: *u8=argv[5] as *u8
      let d: *CdDoc=cd_new()
      if cd_loadfile(d, argv[2] as *u8)<0 { return cd_fail("load" as *u8, d.err) }
      let cur: *u8=sys_mmap(CD_MAXOPS+16)
      let vis: *i64=sys_mmap(8*(CD_MAXOPS+16)) as *i64
      let vn: i64=cd_linear(d, cur, vis)
      var ctr: i64=cd_maxctr(argv[2] as *u8, rep)
      if cd_streq(mode,"ins" as *u8)==1 {
          if argc<7 { cd_puts("usage: nx_coedit edit <log> <rep> <pos> ins <text>\n" as *u8); sys_exit(2); return 2 }
          if pos>vn { return cd_fail("pos-out-of-range" as *u8, 2) }
          let txt: *u8=argv[6] as *u8
          let tl: i64=cd_slen(txt)
          if tl<1 { return cd_fail("empty-text" as *u8, 1) }
          // anchor: R (pos 0) or the id of the visible byte at pos-1; then chain each new byte after the last
          var isroot: i64=1
          var arep: i64=0
          var actr: i64=0
          if pos>0 { isroot=0; let aix: i64=vis[pos-1]; arep=d.nrep[aix]; actr=d.nctr[aix] }
          let line: *u8=sys_mmap(256)
          var t: i64=0
          while t<tl {
              ctr=ctr+1
              var o: i64=cd_cat(line,0,"I " as *u8)
              o=cd_uint(line,o,rep); line[o]=32 as u8; o=o+1
              o=cd_uint(line,o,ctr); line[o]=32 as u8; o=o+1
              if isroot==1 { o=cd_cat(line,o,"R" as *u8) } else { o=cd_uint(line,o,arep); line[o]=58 as u8; o=o+1; o=cd_uint(line,o,actr) }
              line[o]=32 as u8; o=o+1
              o=cd_escput(line,o,txt[t] as i64)
              line[o]=10 as u8; o=o+1
              if cd_appendfile(argv[2] as *u8, line, o)<0 { return cd_fail("append" as *u8, 0) }
              isroot=0
              arep=rep
              actr=ctr
              t=t+1
          }
          cd_puts("COEDIT-EDIT-OK ins ops=" as *u8); cd_num(tl); cd_puts("\n" as *u8)
          sys_exit(0); return 0
      }
      if cd_streq(mode,"del" as *u8)==1 {
          if argc<7 { cd_puts("usage: nx_coedit edit <log> <rep> <pos> del <count>\n" as *u8); sys_exit(2); return 2 }
          ip0[0]=0
          let cnts: *u8=argv[6] as *u8
          let cnt: i64=cd_int(cnts, ip0, cd_slen(cnts))
          if cnt<1 { return cd_fail("bad-count" as *u8, 1) }
          if pos+cnt>vn { return cd_fail("range-out-of-range" as *u8, 2) }
          let line: *u8=sys_mmap(256)
          var t: i64=0
          while t<cnt {
              ctr=ctr+1
              let tix: i64=vis[pos+t]
              var o: i64=cd_cat(line,0,"D " as *u8)
              o=cd_uint(line,o,rep); line[o]=32 as u8; o=o+1
              o=cd_uint(line,o,ctr); line[o]=32 as u8; o=o+1
              o=cd_uint(line,o,d.nrep[tix]); line[o]=58 as u8; o=o+1
              o=cd_uint(line,o,d.nctr[tix])
              line[o]=10 as u8; o=o+1
              if cd_appendfile(argv[2] as *u8, line, o)<0 { return cd_fail("append" as *u8, 0) }
              t=t+1
          }
          cd_puts("COEDIT-EDIT-OK del ops=" as *u8); cd_num(cnt); cd_puts("\n" as *u8)
          sys_exit(0); return 0
      }
      cd_puts("nx_coedit edit: mode must be ins|del\n" as *u8)
      sys_exit(2); return 2
  }
  if cd_streq(cmd,"serve" as *u8)==1 {
      if argc<4 { cd_puts("usage: nx_coedit serve <log> <port> [maxconns]\n" as *u8); sys_exit(2); return 2 }
      let ip0: *i64=sys_mmap(16) as *i64
      ip0[0]=0
      let ports: *u8=argv[3] as *u8
      let port: i64=cd_int(ports, ip0, cd_slen(ports))
      if port<1 { return cd_fail("bad-port" as *u8, 1) }
      var maxc: i64=0
      if argc>4 { ip0[0]=0; let ms: *u8=argv[4] as *u8; maxc=cd_int(ms, ip0, cd_slen(ms)) }
      cd_puts("COEDIT-SERVE listening port=" as *u8); cd_num(port); cd_puts(" log=" as *u8); cd_puts(argv[2] as *u8); cd_puts("\n" as *u8)
      let served: i64=cd_serve(argv[2] as *u8, port, maxc)
      if served<0 { return cd_fail("serve" as *u8, 0-served) }
      cd_puts("COEDIT-SERVE-DONE served=" as *u8); cd_num(served); cd_puts("\n" as *u8)
      sys_exit(0); return 0
  }
  cd_puts("nx_coedit: unknown command (apply|ins|del|merge|edit|serve)\n" as *u8)
  sys_exit(2); return 2
780}
nx_coedit.nx source

`nx_coedit.nx` source
