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

↩ module page · 567 lines · 27908 B
1// MCP-specific identity ownership over the generic process pool.
2import "nx_mcp_control.nx"
3import "nx_request_pool.nx"
4
5struct NxMcpPending {
  pid: i64,
  kind: i64,
  method: i64,
  data: *u8,
  length: i64,
  allocated: i64,
12}
13// mp_spawn returns this (not a bare -1) when the id is ALREADY in flight, so the
14// caller can answer that ONE request with a correlated error and keep serving the
15// rest instead of tearing the session down. Still < 0, so mp_spawn's existing
16// "< 0 means not spawned" contract (and its gate) are unchanged.
17const MP_SPAWN_DUP: i64 = 0 - 2
18// Parent-side recovery error codes (JSON-RPC server range / internal). A worker
19// that died WITHOUT self-reporting (a crash, or the SIGALRM deadline killing it)
20// is turned into one of these correlated errors by mp_recover_pump.
21const MP_CODE_DEADLINE: i64 = 0 - 32012
22const MP_CODE_WORKER:   i64 = 0 - 32603
23const MP_SIG_ALARM:     i64 = 14
24// Capacity is an admitted execution budget, not an instruction to block the
25// control channel. Only requests can receive a correlated refusal response.
26func mp_admit(p: *NxRequestPool, body: *u8, control: *NxMcpControl, output: i64) -> i64 {
  if p.active < p.capacity { return 1 }
  if control.id_kind == 0 { return -1 }
  let prefix: *u8="{\"jsonrpc\":\"2.0\",\"id\":" as *u8
  let suffix: *u8=",\"error\":{\"code\":-32000,\"message\":\"Connector execution capacity is occupied\",\"data\":{\"owner\":\"nx_mcp_stdio\",\"stage\":\"admission\",\"cause\":\"admitted_workers_in_use\",\"dispatched\":false,\"next\":\"Wait for an outstanding request to finish or cancel it before retrying\"}}}\n" as *u8
  var off: i64=control.id_off
  var count: i64=control.id_len
  if control.id_kind == NX_JSON_STRING { off=off-1; count=count+2 }
  if rp_write(output,prefix,mr_len(prefix)) != 0 { return -1 }
  if rp_write(output,body+off,count) != 0 { return -1 }
  if rp_write(output,suffix,mr_len(suffix)) != 0 { return -1 }
  return 0
38}
39func mp_at(rows: *u8, slot: i64) -> *NxMcpPending {
  return (rows+slot*__size_of(NxMcpPending)) as *NxMcpPending
41}
42func mp_clear(row: *NxMcpPending) -> i64 {
  if row.allocated > 0 { sys_munmap(row.data,row.allocated) }
  row.pid=0; row.kind=0; row.method=0; row.data=0 as *u8; row.length=0; row.allocated=0
  return 0
46}
47func mp_sweep(p: *NxRequestPool, rows: *u8) -> i64 {
  var i: i64=0
  while i<p.capacity {
      let row: *NxMcpPending=mp_at(rows,i)
      if row.pid != 0 { if row.pid != rp_worker(p,i).pid { mp_clear(row) } }
      i=i+1
  }
  return 0
55}
56func mp_normalize(kind: i64, src: *u8, n: i64, dst: *u8) -> i64 {
  if kind == NX_JSON_STRING { return jx_decode_span(src,n,dst,n+1) }
  if kind != NX_JSON_NUMBER { return -1 }
  var i: i64=0
  // JSON -0 and 0 denote the same integer identity.
  if n == 2 { if src[0] == (45 as u8) { if src[1] == (48 as u8) { dst[0]=48 as u8; return 1 } } }
  while i<n { dst[i]=src[i]; i=i+1 }
  return n
64}
65func mp_find(p: *NxRequestPool, rows: *u8, kind: i64, data: *u8, n: i64) -> i64 {
  mp_sweep(p,rows)
  var slot: i64=0
  while slot<p.capacity {
      let row: *NxMcpPending=mp_at(rows,slot)
      if row.pid != 0 { if row.kind == kind { if row.length == n {
          var equal: i64=1
          var i: i64=0
          while i<n { if row.data[i] != data[i] { equal=0 }; i=i+1 }
          if equal != 0 { return slot }
      } } }
      slot=slot+1
  }
  return -1
79}
80func mp_spawn(p: *NxRequestPool, rows: *u8, body: *u8, control: *NxMcpControl) -> i64 {
  if control.id_kind == 0 { return rp_fork(p) }
  let allocated: i64=control.id_len+1
  let data: *u8=sys_mmap(allocated)
  let n: i64=mp_normalize(control.id_kind,body+control.id_off,control.id_len,data)
  if n<0 { sys_munmap(data,allocated); return -1 }
  if mp_find(p,rows,control.id_kind,data,n)>=0 { sys_munmap(data,allocated); return MP_SPAWN_DUP }
  let pid: i64=rp_fork(p)
  if pid==0 { return 0 }
  if pid<0 { sys_munmap(data,allocated); return pid }
  var slot: i64=0
  while slot<p.capacity {
      if rp_worker(p,slot).pid==pid {
          let row: *NxMcpPending=mp_at(rows,slot)
          mp_clear(row)
          row.pid=pid; row.kind=control.id_kind; row.method=control.method
          row.data=data; row.length=n; row.allocated=allocated
          return pid
      }
      slot=slot+1
  }
  sys_munmap(data,allocated)
  return -1
103}
104// Returns 1 for confirmed local worker cancellation, 0 for no active target,
105// -1 for failed/prohibited cancellation. This does not cancel a remote job.
106func mp_cancel(p: *NxRequestPool, rows: *u8, body: *u8, control: *NxMcpControl) -> i64 {
  let allocated: i64=control.cancel_len+1
  let data: *u8=sys_mmap(allocated)
  let n: i64=mp_normalize(control.cancel_kind,body+control.cancel_off,control.cancel_len,data)
  if n<0 { sys_munmap(data,allocated); return -1 }
  let slot: i64=mp_find(p,rows,control.cancel_kind,data,n)
  sys_munmap(data,allocated)
  if slot<0 { return 0 }
  let row: *NxMcpPending=mp_at(rows,slot)
  if row.method==MCT_INITIALIZE { return -1 }
  if rp_cancel_owned(p,slot,row.pid)!=0 { return -1 }
  mp_clear(row)
  return 1
119}
120
121// Extract a top-level "id" and its COMPLETE value from a (possibly truncated) request prefix -- the
122// bytes we DID retain before an oversized or malformed line. Returns 1 and fills kind/off/len (string
123// content EXCLUDES the quotes; number is its digit span with an optional leading '-'); returns 0 when
124// no whole top-level id is present (nested only, absent, or the value is cut short by the truncation).
125// Lives here (not in the connector's main-bearing file) so the gate can test it directly.
126func mo_prefix_id(src: *u8, n: i64, kind: *i64, off: *i64, len: *i64) -> i64 {
  var depth: i64 = 0
  var quoted: i64 = 0
  var escaped: i64 = 0
  var start: i64 = 0
  var i: i64 = 0
  while i < n {
      let c: i64 = src[i] as i64
      if quoted == 1 {
          if escaped == 1 { escaped = 0 }
          else { if c == 92 { escaped = 1 }
          else { if c == 34 {
              quoted = 0
              if depth == 1 { if i - start == 2 {
                  if src[start] == (105 as u8) { if src[start + 1] == (100 as u8) {
                      var j: i64 = i + 1
                      while j < n { if src[j] == (32 as u8) { j = j + 1 } else { break } }
                      if j < n { if src[j] == (58 as u8) {
                          j = j + 1
                          while j < n { if src[j] == (32 as u8) { j = j + 1 } else { break } }
                          if j >= n { return 0 }
                          let vc: i64 = src[j] as i64
                          if vc == 34 {
                              let vstart: i64 = j + 1
                              var k: i64 = vstart
                              var esc: i64 = 0
                              var closed: i64 = 0
                              while k < n {
                                  let cc: i64 = src[k] as i64
                                  if esc == 1 { esc = 0 }
                                  else { if cc == 92 { esc = 1 } else { if cc == 34 { closed = 1; break } } }
                                  k = k + 1
                              }
                              if closed == 0 { return 0 }
                              *kind = NX_JSON_STRING
                              *off = vstart
                              *len = k - vstart
                              return 1
                          }
                          var k2: i64 = j
                          if src[k2] == (45 as u8) { k2 = k2 + 1 }
                          let dstart: i64 = k2
                          while k2 < n { if nx_ascii_is_digit(src[k2]) == 1 { k2 = k2 + 1 } else { break } }
                          if k2 == dstart { return 0 }
                          if k2 >= n { return 0 }
                          *kind = NX_JSON_NUMBER
                          *off = j
                          *len = k2 - j
                          return 1
                      } }
                  } }
              } }
          } } }
      } else {
          if c == 34 { quoted = 1; start = i + 1 }
          else { if c == 123 { depth = depth + 1 }
          else { if c == 125 { depth = depth - 1 } } }
      }
      i = i + 1
  }
  return 0
187}
188
189// ---- CORRELATED JSON-RPC ERROR EMISSION -------------------------------------
190// One place builds the JSON-RPC 2.0 error envelope so the request CHILD (rich
191// soft-failure errors, original id span) and the PARENT recovery (a crashed
192// worker, decoded id from its pending row) produce the SAME shape. Self-contained
193// cat helpers: nx_mcp_pending sees neither the transport's mc_* nor the tools API.
194func mpe_slen(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
195func mpe_cat(d: *u8, o: i64, s: *u8) -> i64 { var i: i64 = 0; var p: i64 = o; while s[i] != (0 as u8) { d[p] = s[i]; p = p + 1; i = i + 1 } return p }
196func mpe_catb(d: *u8, o: i64, s: *u8, n: i64) -> i64 { var i: i64 = 0; while i < n { d[o+i] = s[i]; i = i + 1 } return o + i }
197func mpe_catn(d: *u8, o: i64, v: i64) -> i64 {
  var w: i64 = o
  var x: i64 = v
  if x < 0 { d[w] = 45 as u8; w = w + 1; x = 0 - x }
  let t: *u8 = sys_mmap(24)
  var k: i64 = 0
  if x == 0 { t[0] = 48 as u8; k = 1 }
  while x > 0 { t[k] = (48 + (x % 10)) as u8; x = x / 10; k = k + 1 }
  var i: i64 = 0
  while i < k { d[w] = t[k-1-i]; w = w + 1; i = i + 1 }
  sys_munmap(t, 24)
  return w
209}
210// JSON-escape a DECODED id string span (parent path): a raw quote, backslash, or
211// control byte in a decoded id would break the envelope, so escape them. The
212// child path passes the ORIGINAL span verbatim (already valid JSON) and skips this.
213func mpe_cat_esc(d: *u8, o: i64, s: *u8, n: i64) -> i64 {
  var w: i64 = o
  var i: i64 = 0
  while i < n {
      let c: i64 = s[i] as i64
      if c == 34 { d[w] = 92 as u8; d[w+1] = 34 as u8; w = w + 2 }
      else { if c == 92 { d[w] = 92 as u8; d[w+1] = 92 as u8; w = w + 2 }
      else { if c < 32 {
          d[w] = 92 as u8; d[w+1] = 117 as u8; d[w+2] = 48 as u8; d[w+3] = 48 as u8
          let hi: i64 = (c >> 4) & 15
          let lo: i64 = c & 15
          var hc: i64 = 48 + hi
          if hi > 9 { hc = 87 + hi }
          var lc: i64 = 48 + lo
          if lo > 9 { lc = 87 + lo }
          d[w+4] = hc as u8; d[w+5] = lc as u8; w = w + 6
      } else { d[w] = c as u8; w = w + 1 } } }
      i = i + 1
  }
  return w
233}
234// Write a correlated JSON-RPC 2.0 error to fd for request id (id_kind/id_ptr/id_len).
235// escape=0: id_ptr is the ORIGINAL request id span (string content or number
236// digits) -- emitted verbatim. escape=1: id_ptr is a DECODED id (pending row) --
237// string content is re-escaped. id_kind==0 (a notification) writes NOTHING and
238// returns 0: a notification never gets a response. Returns 0 on success (or on
239// nothing-to-write), non-zero when the protocol fd write failed (broken output).
240func mpe_write_error_detail(fd: i64, id_kind: i64, id_ptr: *u8, id_len: i64, escape: i64, code: i64, stage: *u8, cause: *u8, measured: i64, next: *u8, detail:*u8, detail_n:i64) -> i64 {
  if id_kind == 0 { return 0 }
  let cap: i64 = id_len*6 + mpe_slen(stage) + mpe_slen(cause)*2 + mpe_slen(next) + 512 + detail_n
  let d: *u8 = sys_mmap(cap)
  var o: i64 = mpe_cat(d, 0, "{\"jsonrpc\":\"2.0\",\"id\":" as *u8)
  if id_kind == NX_JSON_STRING {
      d[o] = 34 as u8; o = o + 1
      if escape == 0 { o = mpe_catb(d, o, id_ptr, id_len) } else { o = mpe_cat_esc(d, o, id_ptr, id_len) }
      d[o] = 34 as u8; o = o + 1
  } else {
      o = mpe_catb(d, o, id_ptr, id_len)
  }
  o = mpe_cat(d, o, ",\"error\":{\"code\":" as *u8)
  o = mpe_catn(d, o, code)
  o = mpe_cat(d, o, ",\"message\":\"" as *u8)
  o = mpe_cat(d, o, cause)
  o = mpe_cat(d, o, "\",\"data\":{\"owner\":\"nx_mcp_stdio\",\"stage\":\"" as *u8)
  o = mpe_cat(d, o, stage)
  o = mpe_cat(d, o, "\",\"cause\":\"" as *u8)
  o = mpe_cat(d, o, cause)
  o = mpe_cat(d, o, "\",\"measured\":" as *u8)
  o = mpe_catn(d, o, measured)
  o = mpe_cat(d, o, ",\"next\":\"" as *u8)
  o = mpe_cat(d, o, next)
  o = mpe_cat(d, o, "\"" as *u8)
  if (detail as i64)>0&&detail_n>0{o=mpe_cat(d,o,",\"http\":");o=mpe_catb(d,o,detail,detail_n)}
  o = mpe_cat(d, o, "}}}\n" as *u8)
  let rc: i64 = rp_write(fd, d, o)
  sys_munmap(d, cap)
  return rc
270}
271
272// ---- mp_recover_pump: A PER-REQUEST WORKER FAILURE IS ONE ERROR, NOT AN OUTAGE
273// Wrap rp_pump so a child that failed WITHOUT self-reporting (a crash, or the
274// SIGALRM deadline that killed it) becomes a CORRELATED error for that one request
275// id and the connector keeps serving every other request. Only a genuine fatal --
276// poll failure, a broken protocol fd (rp_pump last_error -3), or a reap fault --
277// returns -1. Success/progress (1 stdin-ready, 0 progress) is passed through;
278// a recovered per-request failure returns 0 (progress, keep going).
279func mp_recover_pump_timeout(p: *NxRequestPool, rows: *u8, accept_input: i64, protocol_fd: i64, timeout_secs: i64, wait_ms:i64) -> i64 {
  let r: i64 = rp_pump_timeout(p, accept_input, protocol_fd, wait_ms)
  if r >= 0 { mp_sweep(p, rows); return r }
  // r == -1. A known failed slot AND not a broken-output-fd fatal => recoverable.
  let slot: i64 = p.failed_slot
  if slot < 0 { return 0 - 1 }
  if p.last_error == 0 - 3 { return 0 - 1 }
  // Over-budget leaves the child alive; reap it before reusing the slot. A
  // non-zero exit already released it (rp_worker(p,slot).pid == 0), so skip.
  if rp_worker(p, slot).pid != 0 {
      if rp_cancel_owned(p, slot, rp_worker(p, slot).pid) != 0 { return 0 - 1 }
  }
  let row: *NxMcpPending = mp_at(rows, slot)
  var wrc: i64 = 0
  if row.pid != 0 {
      var code:  i64 = MP_CODE_WORKER
      var stage: *u8 = "worker" as *u8
      var cause: *u8 = "the request worker terminated before delivering a response" as *u8
      var measured: i64 = p.last_error
      if wait_term_signal(p.last_error) == MP_SIG_ALARM {
          code = MP_CODE_DEADLINE
          stage = "deadline" as *u8
          cause = "the request exceeded its per-request time budget" as *u8
          measured = timeout_secs
      }
      wrc = mpe_write_error(protocol_fd, row.kind, row.data, row.length, 1, code, stage, cause, measured, "check service health and byte budgets; verify any write outcome before retrying" as *u8)
      mp_clear(row)
  }
  p.failed_slot = 0 - 1
  mp_sweep(p, rows)
  if wrc != 0 { return 0 - 1 }
  return 0
311}
312
313// Selective HTTP evidence for per-request failures. Raw response bodies, cookies,
314// authorization headers and free-form upstream detail never enter protocol errors.
315import "nx_http_response_parse.nx"
316import "nx_mcp_route.nx"
317import "nx_sha256.nx"
318func mpe_http_header(src:*u8,n:i64,p:*i64,key:*u8,offset:*i64,length:*i64)->i64 {
var pos:i64=p[4];var count:i64=0
var no:i64=0;var nl:i64=0;var vo:i64=0;var vl:i64=0;var next:i64=0
while pos<p[6]-2{
if nx_http_resp_parse_header_line(src,n,pos,&no,&nl,&vo,&vl,&next)!=NX_HTTP_RESP_OK{return -1}
if next<=pos{return -1}
if nx_http_resp_cieq(src+no,nl,key,mpe_slen(key))==1{
 var i:i64=0;while i<vl{let c:i64=src[vo+i] as i64;if c<32||c>126{return -1};i=i+1}
 count=count+1;*offset=vo;*length=vl
};pos=next
}
return count
330}
331func mpe_http_field(d:*u8,o0:i64,src:*u8,n:i64,p:*i64,name:*u8,label:*u8)->i64 {
var off:i64=0;var len:i64=0;let count:i64=mpe_http_header(src,n,p,name,&off,&len)
var o:i64=mpe_cat(d,o0,",\"");o=mpe_cat(d,o,label);o=mpe_cat(d,o,"\":")
if count==1{o=mpe_cat(d,o,"\"");o=mpe_cat_esc(d,o,src+off,len);o=mpe_cat(d,o,"\"")}else{o=mpe_cat(d,o,"null")}
o=mpe_cat(d,o,",\"");o=mpe_cat(d,o,label);o=mpe_cat(d,o,"_observation\":\"")
var state:*u8="absent";if count==1{state="present"};if count>1{state="ambiguous-duplicate"};if count<0{state="malformed"}
o=mpe_cat(d,o,state);return mpe_cat(d,o,"\"")
338}
339// Returned object is validated/escaped here and borrowed by the error emitter.
340// Allocations live only in the existing one-request worker; no persistent collector.
341func mpe_http_evidence(src:*u8,n:i64,p:*i64,out_len:*i64)->*u8 {
*out_len=0
if n<=0||n>(NX_RA_SIZE_MAX-4096)/6||p[6]<0||p[6]>n{return 0 as *u8}
let cap:i64=n*6+4096;let d:*u8=sys_mmap(cap);if (d as i64)<=0{return 0 as *u8}
let bn:i64=n-p[6];var o:i64=mpe_cat(d,0,"{\"schema\":\"nishi.mcp-http-observation.v1\",\"status\":")
o=mpe_catn(d,o,p[1]);o=mpe_cat(d,o,",\"observed_at_unix\":");o=mpe_catn(d,o,sys_now_realtime_sec())
o=mpe_http_field(d,o,src,n,p,"Retry-After","retry_after")
o=mpe_http_field(d,o,src,n,p,"X-Request-Id","request_id")
o=mpe_http_field(d,o,src,n,p,"X-Nishi-Request-Id","nishi_request_id")
o=mpe_http_field(d,o,src,n,p,"Traceparent","traceparent")
o=mpe_http_field(d,o,src,n,p,"Content-Type","content_type")
o=mpe_cat(d,o,",\"observed_body_bytes\":");o=mpe_catn(d,o,bn)
o=mpe_cat(d,o,",\"declared_body_bytes\":")
if p[8]==NX_HTTP_BODY_CONTENT_LENGTH{o=mpe_catn(d,o,p[7])}else{o=mpe_cat(d,o,"null")}
var complete:i64=0
if p[8]==NX_HTTP_BODY_CONTENT_LENGTH&&p[7]==bn{complete=1}
o=mpe_cat(d,o,",\"body_framing\":\"")
var framing:*u8="unverified-transfer";if p[8]==NX_HTTP_BODY_CONTENT_LENGTH{framing="length-mismatch";if complete==1{framing="content-length-complete"}}
o=mpe_cat(d,o,framing);o=mpe_cat(d,o,"\",\"observed_body_sha256\":")
let hash:*u8=sys_mmap(SHA256_DIGEST_BYTES)
if (hash as i64)>0&&sha256_digest_checked_native(src+p[6],bn,hash)==0{
let hex:*u8="0123456789abcdef";o=mpe_cat(d,o,"\"");var i:i64=0
while i<SHA256_DIGEST_BYTES{let b:i64=hash[i] as i64;d[o]=hex[b>>4];d[o+1]=hex[b&15];o=o+2;i=i+1}
o=mpe_cat(d,o,"\"")
}else{o=mpe_cat(d,o,"null")}
if (hash as i64)>0{sys_munmap(hash,SHA256_DIGEST_BYTES)}
var stage:*u8=0 as *u8;var retry:*u8=0 as *u8
if complete==1{
var parsed:i64=0;let root:*NxValue=nx_value_parse_json(src+p[6],bn,&parsed)
if parsed==NX_VAL_PARSE_OK{
 var bad:i64=0;let sv:*NxValue=mr_member(root,"stage",&bad);let rv:*NxValue=mr_member(root,"retry",&bad)
 if bad==0{
  if sv!=(0 as *NxValue){if sv.kind==NX_VAL_STRING{
   if mr_equal(sv.str_ptr,sv.str_len,"connect")==1{stage="connect"}
   if mr_equal(sv.str_ptr,sv.str_len,"read")==1{stage="read"}
   if mr_equal(sv.str_ptr,sv.str_len,"deadline")==1{stage="deadline"}
   if mr_equal(sv.str_ptr,sv.str_len,"read-after-accept")==1{stage="read-after-accept"}
  }}
  if rv!=(0 as *NxValue){if rv.kind==NX_VAL_STRING{if mr_equal(rv.str_ptr,rv.str_len,"unsafe")==1{retry="unsafe"}}}
 }
}
}
o=mpe_cat(d,o,",\"upstream_stage\":")
if (stage as i64)>0{o=mpe_cat(d,o,"\"");o=mpe_cat(d,o,stage);o=mpe_cat(d,o,"\"")}else{o=mpe_cat(d,o,"null")}
o=mpe_cat(d,o,",\"upstream_retry\":")
if (retry as i64)>0{o=mpe_cat(d,o,"\"");o=mpe_cat(d,o,retry);o=mpe_cat(d,o,"\"")}else{o=mpe_cat(d,o,"null")}
o=mpe_cat(d,o,",\"request_outcome\":\"unknown\",\"automatic_retry\":false,\"body_retention\":\"digest-only; original body not retained\",\"correlation\":\"outer JSON-RPC id preserved; optional upstream headers are declarations\"}")
*out_len=o;return d
389}
390
391func mpe_write_error(fd:i64,id_kind:i64,id_ptr:*u8,id_len:i64,escape:i64,code:i64,stage:*u8,cause:*u8,measured:i64,next:*u8)->i64 {
return mpe_write_error_detail(fd,id_kind,id_ptr,id_len,escape,code,stage,cause,measured,next,0 as *u8,0)
393}
394
395func mp_recover_pump(p:*NxRequestPool,rows:*u8,accept_input:i64,protocol_fd:i64,timeout_secs:i64)->i64 {return mp_recover_pump_timeout(p,rows,accept_input,protocol_fd,timeout_secs,-1)}
396
397// Queue ownership stays in MCP pending: copied request bytes live until dispatch,
398// cancellation or expiry. A removed request is never automatically replayed.
399// Budget covers node, copied body and normalized identity; it is a configured
400// virtual-memory admission envelope, not a promise of resident physical RAM.
401struct NxMcpQueued {
next:*NxMcpQueued, bytes:i64, body:*u8, body_n:i64,
identity:*u8, identity_n:i64, control:*NxMcpControl,
deadline_ms:i64, barrier:i64
405}
406struct NxMcpQueue {
head:*NxMcpQueued, tail:*NxMcpQueued, used:i64, budget:i64,
count:i64, wait_ms:i64, barrier_pid:i64, first:i64
409}
410const MQ_CODE_CANCEL:i64=0-32800
411const MQ_CODE_CAPACITY:i64=0-32000
412const MQ_CODE_DUP:i64=0-32013
413const MQ_DECIMAL_I64_BYTES:i64=20
414func mq_new(budget:i64,wait_ms:i64)->*NxMcpQueue {
if budget<=__size_of(NxMcpQueued)||wait_ms<=0{return 0 as *NxMcpQueue}
let q:*NxMcpQueue=sys_mmap_try(__size_of(NxMcpQueue)) as *NxMcpQueue
if q==(0 as *NxMcpQueue){return q}
q.budget=budget;q.wait_ms=wait_ms;q.first=1
return q
420}
421func mq_event(q:*NxMcpQueue,node:*NxMcpQueued,state:*u8,dispatched:i64)->i64 {
let a:*u8="{\"owner\":\"nx_mcp_stdio\",\"stage\":\"admission-queue\",\"state\":\""
let b:*u8="\",\"request_id\":"
let c:*u8=",\"queued\":"
let d:*u8=",\"held_bytes\":"
let e:*u8=",\"budget_bytes\":"
let f:*u8=",\"deadline_monotonic_ms\":"
let g:*u8=",\"dispatched\":"
let cap:i64=mpe_slen(a)+mpe_slen(b)+mpe_slen(c)+mpe_slen(d)+mpe_slen(e)+mpe_slen(f)+mpe_slen(g)+mpe_slen(state)+node.control.id_len*6+MQ_DECIMAL_I64_BYTES*5+mpe_slen("nullfalse}\n")+2
let out:*u8=sys_mmap_try(cap);if out==(0 as *u8){return -1}
var o:i64=mpe_cat(out,0,a);o=mpe_cat(out,o,state);o=mpe_cat(out,o,b)
if node.control.id_kind==0{o=mpe_cat(out,o,"null")}else{
if node.control.id_kind==NX_JSON_STRING{o=mpe_cat(out,o,"\"")}
o=mpe_catb(out,o,node.body+node.control.id_off,node.control.id_len)
if node.control.id_kind==NX_JSON_STRING{o=mpe_cat(out,o,"\"")}
}
o=mpe_cat(out,o,c);o=mpe_catn(out,o,q.count)
o=mpe_cat(out,o,d);o=mpe_catn(out,o,q.used)
o=mpe_cat(out,o,e);o=mpe_catn(out,o,q.budget)
o=mpe_cat(out,o,f);o=mpe_catn(out,o,node.deadline_ms)
o=mpe_cat(out,o,g)
if dispatched!=0{o=mpe_cat(out,o,"true")}else{o=mpe_cat(out,o,"false")}
o=mpe_cat(out,o,"}\n")
let rc:i64=rp_write(2,out,o);sys_munmap_direct(out,cap);return rc
445}
446func mq_same(node:*NxMcpQueued,kind:i64,id:*u8,n:i64)->i64 {
if kind==0||node.control.id_kind!=kind||node.identity_n!=n{return 0}
var i:i64=0;while i<n{if node.identity[i]!=id[i]{return 0};i=i+1};return 1
449}
450func mq_error(node:*NxMcpQueued,fd:i64,code:i64,cause:*u8,measured:i64)->i64 {
return mpe_write_error(fd,node.control.id_kind,node.body+node.control.id_off,node.control.id_len,0,code,"admission-queue",cause,measured,"request was not dispatched; inspect queue evidence before resubmitting")
452}
453func mq_unlink(q:*NxMcpQueue,prior:*NxMcpQueued,node:*NxMcpQueued)->i64 {
if prior==(0 as *NxMcpQueued){q.head=node.next}else{prior.next=node.next}
if q.tail==node{q.tail=prior}
q.used=q.used-node.bytes;q.count=q.count-1
return 0
458}
459func mq_free(node:*NxMcpQueued)->i64 {return sys_munmap_direct(node as *u8,node.bytes)}
460func mq_enqueue(q:*NxMcpQueue,p:*NxRequestPool,rows:*u8,body:*u8,n:i64,control:*NxMcpControl,now:i64,fd:i64)->i64 {
if n<=0||now<0||q.wait_ms>NX_RA_SIZE_MAX-now{return -1}
let overhead:i64=__size_of(NxMcpQueued)+__size_of(NxMcpControl)+control.id_len+1
if overhead<0||n>NX_RA_SIZE_MAX-overhead{return -1}
let bytes:i64=n+overhead
// Stack metadata permits a correlated refusal without reserving queue memory.
var refusal:NxMcpQueued
refusal.body=body;refusal.body_n=n;refusal.control=control;refusal.deadline_ms=now+q.wait_ms
if bytes>q.budget-q.used{
mq_event(q,&refusal,"refused-memory-budget",0)
if mq_error(&refusal,fd,MQ_CODE_CAPACITY,"configured queue memory budget exhausted; not dispatched",q.budget)!=0{return -1};return 0
}
let node:*NxMcpQueued=sys_mmap_try(bytes) as *NxMcpQueued
if node==(0 as *NxMcpQueued){
mq_event(q,&refusal,"refused-allocation",0)
if mq_error(&refusal,fd,MQ_CODE_CAPACITY,"queue allocation unavailable; not dispatched",bytes)!=0{return -1};return 0
}
node.bytes=bytes;node.control=((node as *u8)+__size_of(NxMcpQueued)) as *NxMcpControl;node.body=(node as *u8)+__size_of(NxMcpQueued)+__size_of(NxMcpControl);node.body_n=n
var i:i64=0;while i<n{node.body[i]=body[i];i=i+1}
node.deadline_ms=now+q.wait_ms
node.control.method=control.method;node.control.id_kind=control.id_kind
node.control.id_off=control.id_off;node.control.id_len=control.id_len
node.control.cancel_kind=control.cancel_kind;node.control.cancel_off=control.cancel_off;node.control.cancel_len=control.cancel_len
node.identity=node.body+n
node.identity_n=0
if control.id_kind!=0{
node.identity_n=mp_normalize(control.id_kind,body+control.id_off,control.id_len,node.identity)
var duplicate:i64=mp_find(p,rows,control.id_kind,node.identity,node.identity_n)>=0
var cur:*NxMcpQueued=q.head
while cur!=(0 as *NxMcpQueued){
 if mq_same(cur,control.id_kind,node.identity,node.identity_n)!=0{duplicate=1};cur=cur.next
}
if duplicate!=0{
 let rc:i64=mq_error(node,fd,MQ_CODE_DUP,"request id already queued or running; duplicate not dispatched",q.count)
 mq_event(q,node,"refused-duplicate-id",0);mq_free(node);if rc!=0{return -1};return 0
}
}
node.deadline_ms=now+q.wait_ms
node.barrier=q.first!=0||control.method==MCT_INITIALIZE||control.method==MCT_INITIALIZED
q.first=0
if q.tail==(0 as *NxMcpQueued){q.head=node}else{q.tail.next=node}
q.tail=node;q.used=q.used+bytes;q.count=q.count+1
mq_event(q,node,"queued",0)
return 1
504}
505func mq_cancel(q:*NxMcpQueue,body:*u8,control:*NxMcpControl,fd:i64)->i64 {
let size:i64=control.cancel_len+1;let id:*u8=sys_mmap_try(size)
if id==(0 as *u8){return -1}
let n:i64=mp_normalize(control.cancel_kind,body+control.cancel_off,control.cancel_len,id)
var prior:*NxMcpQueued=0 as *NxMcpQueued;var node:*NxMcpQueued=q.head
while node!=(0 as *NxMcpQueued){
if mq_same(node,control.cancel_kind,id,n)!=0{
 sys_munmap_direct(id,size)
 if node.control.method==MCT_INITIALIZE{return -1}
 mq_unlink(q,prior,node)
 let rc:i64=mq_error(node,fd,MQ_CODE_CANCEL,"queued request cancelled before dispatch",0)
 mq_event(q,node,"cancelled-before-dispatch",0);mq_free(node)
 if rc!=0{return -2};return 1
}
prior=node;node=node.next
}
sys_munmap_direct(id,size);return 0
522}
523func mq_expire(q:*NxMcpQueue,now:i64,fd:i64)->i64 {
var prior:*NxMcpQueued=0 as *NxMcpQueued;var node:*NxMcpQueued=q.head
while node!=(0 as *NxMcpQueued){
let next:*NxMcpQueued=node.next
if node.deadline_ms<=now{
 mq_unlink(q,prior,node)
 let rc:i64=mq_error(node,fd,MP_CODE_DEADLINE,"queue deadline elapsed before dispatch",q.wait_ms)
 mq_event(q,node,"expired-before-dispatch",0);mq_free(node);if rc!=0{return -1}
}else{prior=node}
node=next
}
return 0
535}
536func mq_wait(q:*NxMcpQueue,now:i64)->i64 {
var wait:i64=0-1;var node:*NxMcpQueued=q.head
while node!=(0 as *NxMcpQueued){
let remaining:i64=node.deadline_ms-now
if remaining<=0{return 0}
if wait<0||remaining<wait{wait=remaining};node=node.next
}
return wait
544}
545func mq_ready(q:*NxMcpQueue,p:*NxRequestPool)->i64 {
if q.barrier_pid!=0{
var i:i64=0;var live:i64=0
while i<p.capacity{if rp_worker(p,i).pid==q.barrier_pid{live=1};i=i+1}
if live!=0{return 0};q.barrier_pid=0
}
if q.head==(0 as *NxMcpQueued)||p.active>=p.capacity{return 0}
if q.head.barrier!=0&&p.active!=0{return 0}
return 1
554}
555
556// Each admitted maximum line may also retain an identity as long as that line.
557// Include both variable spans and their actual metadata, then checked multiply.
558func mq_default_budget(workers:i64,request_cap:i64)->i64 {
if workers<=0||request_cap<=0{return -1}
let metadata:i64=__size_of(NxMcpQueued)+__size_of(NxMcpControl)+1
if request_cap>NX_RA_SIZE_MAX-metadata{return -1}
let body_envelope:i64=request_cap+metadata
if request_cap>NX_RA_SIZE_MAX-body_envelope{return -1}
let per_worker:i64=body_envelope+request_cap
if workers>NX_RA_SIZE_MAX/per_worker{return -1}
return workers*per_worker
567}
nx_mcp_pending.nx source

`nx_mcp_pending.nx` source
