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

↩ module page · 393 lines · 19566 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(p: *NxRequestPool, rows: *u8, accept_input: i64, protocol_fd: i64, timeout_secs: i64) -> i64 {
  let r: i64 = rp_pump(p, accept_input, protocol_fd)
  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}
nx_mcp_pending_http_t174.nx source

`nx_mcp_pending_http_t174.nx` source
