diff --git a/di/serverselect/VERSION b/di/serverselect/VERSION new file mode 100644 index 00000000..6c6aa7cb --- /dev/null +++ b/di/serverselect/VERSION @@ -0,0 +1 @@ +0.1.0 \ No newline at end of file diff --git a/di/serverselect/deps.q b/di/serverselect/deps.q new file mode 100644 index 00000000..5b7801b9 --- /dev/null +++ b/di/serverselect/deps.q @@ -0,0 +1,10 @@ +/ hard module dependencies and their minimum versions, validated by di.depcheck. +/ di.serverselect has NONE, and the empty manifest is deliberate rather than the file being absent: +/ di.depcheck's finddepsq returns (::) for a module that ships no deps.q, which is indistinguishable +/ from "nobody has decided yet". an explicit empty dict records that the STANDALONE classification in +/ the modularisation plan was checked against the source and holds. +/ verified: init.q and serverselect.q contain no `use` at all. logging is INJECTED via init, not a +/ hard dep - the plan's tier table excludes logging, timer and handler management from the dependency +/ tree by design. integration.q does load kx.log, but it is a harness run directly rather than loaded +/ by init.q, so it is not an edge of this module. +deps:(`$())!(); diff --git a/di/serverselect/init.q b/di/serverselect/init.q new file mode 100644 index 00000000..c0af8685 --- /dev/null +++ b/di/serverselect/init.q @@ -0,0 +1,27 @@ +\l ::serverselect.q + +/ logging is an injected dependency: the start-up script that wires the modules together - +/ or the user at run time - must call init with a required `log dependency before using the +/ module. kx.log is intentionally NOT loaded here. +/ note: the injected log dict must already be binary `info`warn`error!{[c;m]} - no adaptation +/ is done here; init fans it out into .z.m.loginfo/.z.m.logwarn/.z.m.logerr, called as +/ .z.m.loginfo[`ctx;"msg"] + +/ module version, read from the VERSION file rather than hardcoded, so a release bump touches one +/ plain-text file. read module-relative (`:::` resolves to di/serverselect) and BEFORE the export +/ line, since export:([...]) evaluates each name. +/ NB `version` must STAY in the export: di.depcheck resolves a dependency's version from the export +/ dict (checkdepversion) and classes a missing one as a FAILURE - which makes di.depcheck.init throw +/ for any process loading a module that declares this one as a hard dependency +/ trim, and fail LOUD on a missing/unreadable/empty VERSION, rather than a bare `first read0`: a raw +/ OS error names no module, and read0 strips the line terminator but NOT a trailing \r on a CRLF file +/ or trailing spaces - and di.depcheck compares versions as STRINGS, so a padded value silently fails +/ every dependent module's check. an empty value is worse still: it reads to depcheck as +/ "exports no version", i.e. the exact failure the VERSION file was added to prevent +version:@[{trim first read0 x};`:::VERSION;{'"di.serverselect: VERSION file missing or unreadable"}]; +if[0=count version;'"di.serverselect: VERSION file is empty"]; + +export:([init; + addserverfull;addserverattr;addserver;setserveractive;setserveridactive;getserverstable;addserversfromtable; + getservers;selector;setselector;getserverbytype;gethandlebytype;gethpbytype;getserverids; + removeinactive;getapimeta;version]) diff --git a/di/serverselect/integration.q b/di/serverselect/integration.q new file mode 100644 index 00000000..0eb4f2b5 --- /dev/null +++ b/di/serverselect/integration.q @@ -0,0 +1,360 @@ +/ =========================================================================== +/ di.serverselect integration test (real processes) +/ Spawns child q listeners, opens REAL IPC handles to them, and drives every +/ exported function through a realistic gateway lifecycle - register -> +/ query -> select -> route a live query to the chosen handle -> bulk-register +/ -> disconnect -> error handling -> init. Unlike a simulated test, the +/ selection functions hand back handles that are actually queried, so the +/ assertions prove queries reach the expected backend process. +/ . +/ Self-contained: no helper files. The child backends are bare `q` listeners +/ whose identity (`whoami`) and `ping` api are injected over IPC at setup; the +/ test launches and tears them down itself. +/ . +/ Requirements: +/ - run THIS script with KDB-X q (needs `use`) +/ - a q on PATH (override with $QBIN) to launch the child listeners +/ - some free TCP ports: the test scans for free ports from a pid-derived +/ base (so concurrent runs don't collide); set $SSPORT to pin the base +/ Run (QPATH must include this repo and the kx module dir): +/ QPATH=/path/to/kx/mod:/path/to/kdbx-modules /q integration.q +/ Exits with a non-zero code equal to the number of failed assertions. +/ =========================================================================== + +srvsel:use`di.serverselect; + +/ ---- tiny assertion harness ---- +/ note: avoid `desc`/`exp`/`log` as names - they are reserved words in KDB-X +PASS:0; FAIL:0; +lbl:{[ok] $[ok;" ok | ";" FAIL | "]}; +chk:{[nm;got;expd] + ok:got~expd; + $[ok;PASS+:1;FAIL+:1]; + m:lbl[ok],nm; + if[not ok; m:m," || got=",(.Q.s1 got)," exp=",.Q.s1 expd]; + -1 m; + }; +errtok:`$"__ERRORED__"; +chkerr:{[nm;f] + r:@[f;();{[e](errtok;e)}]; + ok:(0count free; + '"could not find ",string[n]," free ports scanning up from ",string start]; + :n#free; + }; +ports:findfree[baseport;count names]; + +launch:{[port] + / start a bare q listener detached and capture its shell background pid ($!) - so every + / child can be reaped by exact pid later, whether or not we ever connect to it, and with + / no dependence on the qbin path (identity/api are injected over IPC once connected). + / nohup is REQUIRED, not cosmetic: without it system returns the generic null rather than the + / echoed pid whenever the backgrounded child is itself a q process, so "I"$first threw a bare + / 'type here and the fleet never launched. a backgrounded sleep captures fine, which is what + / makes it look like a shell-quoting problem rather than a q-child one. measured, not assumed. + / nohup over setsid: it execs rather than forking, so $! is unambiguously the q child's own pid + / (verified against pgrep), and it is POSIX where setsid is util-linux. both redirects stay, so + / nohup writes no nohup.out + :"I"$first system "nohup ",qbin," -p ",string[port]," -q /dev/null 2>&1 & echo $!"; + }; + +connect:{[port] + / open a handle, retrying for up to ~10s while the child binds and accepts + hp:`$":localhost:",string port; + step:{[hp;h] if[not null h; :h]; system"sleep 0.2"; @[hopen;(hp;500);{0Ni}]}[hp]; + :50 step/ 0Ni; + }; + +teardown:{[] + / always-run cleanup. kill every launched child by its captured pid (exact). the pkill + / fallback is a last resort for the rare case a pid was never captured - keyed on our own + / unique port and -q flag (NOT the qbin path, which is configurable), so a custom qbin + / cannot break it. the [-]p bracket keeps the pattern from matching this very kill command. + live:pids where not null pids; + if[count live; @[system;"kill ",(" " sv string live)," 2>/dev/null; true";{}]]; + {@[system;"pkill -f \"[-]p ",string[x]," -q\" 2>/dev/null; true";{}]} each ports; + @[hclose;;{}] each handles where not null handles; + }; + +/ launch the fleet (capturing each child's shell pid for exact-pid teardown), then connect +-1 "selected free ports: ",.Q.s1 ports; +pids:launch each ports; +handles:connect each ports; +/ guarantee the fleet is reaped on ANY exit (normal, error-abort, or exit code) +.z.exit:{teardown[]}; +if[any null handles; + -2 "failed to connect to child processes on ports: ",(" " sv string ports where null handles); + exit 3]; +/ inject each backend's identity + ping api over IPC +{[h;nm] h "whoami:`",string nm; h "ping:{whoami}";}'[handles;names]; +-1 "spawned ",(string count handles)," backend processes; real handles: ",.Q.s1 handles; + +/ route: select a server of type t via strategy sel, then SEND ping[] over the chosen handle +route:{[t;sel] h:srvsel.gethandlebytype[t;sel]; $[()~h; `NONE; h "ping[]"]}; +/ map serverids -> their handles (to route attribute-path selections) +sidhandles:{[sids] exec handle from srvsel.getserverstable[] where serverid in sids}; +/ the gateway's host/port for the hpup column - its real (dynamically chosen) address +gwhp:`$":localhost:",string ports 4; + +/ test steps run at top level (a single enclosing function would exceed q's +/ per-function constant limit). chk/chkerr trap their own assertion failures; +/ .z.exit guarantees teardown regardless of how the script ends. + +/ init must be called before use (log is a required injected dependency, no default) +srvsel.init[enlist[`log]!enlist `info`warn`error!({[c;m]};{[c;m]};{[c;m]})]; + +/ ========================================================================= +hdr"STEP 1 initial empty state"; +chk["fresh module: server table empty";0=count srvsel.getserverstable[];1b]; +chk["schema columns correct";`serverid`handle`procname`servertype`hpup`active`lastp`hits`attributes`disconnecttime;cols srvsel.getserverstable[]]; + +/ ========================================================================= +hdr"STEP 2 registration of live handles (addserver / addserverattr / addserverfull)"; +srvsel.addserver[handles 0;`rdb]; +srvsel.addserver[handles 1;`rdb]; +srvsel.addserverattr[handles 2;`hdb;`date`sym!((d1;d2);`A`B)]; +srvsel.addserverattr[handles 3;`hdb;`date`sym!((d2;d3);`B`C)]; +srvsel.addserverfull[handles 4;`gw1;`gateway;gwhp;(enlist`region)!enlist`EU]; +chk["five servers registered";5=count srvsel.getserverstable[];1b]; +chk["serverids autoincrement 1..5";1 2 3 4 5i;exec serverid from srvsel.getserverstable[]]; +chk["handles stored are the real open handles";handles til 5;exec handle from srvsel.getserverstable[]]; +chk["servertypes as registered";`rdb`rdb`hdb`hdb`gateway;exec servertype from srvsel.getserverstable[]]; +chk["addserverfull populated procname";`gw1;first exec procname from srvsel.getserverstable[] where servertype=`gateway]; +chk["addserverfull populated hpup";gwhp;first exec hpup from srvsel.getserverstable[] where servertype=`gateway]; +chk["addserver -> empty attributes";(()!());first exec attributes from srvsel.getserverstable[] where handle=handles 0]; +chk["all active by default";1b;all exec active from srvsel.getserverstable[]]; + +/ ========================================================================= +hdr"STEP 3 getservers (lookup + attribute match scoring)"; +chk["lookups=` returns all active";5;count srvsel.getservers[`;`;()!()]]; +chk["servertype=hdb returns 2";2;count srvsel.getservers[`servertype;`hdb;()!()]]; +chk["servertype list rdb,hdb returns 4";4;count srvsel.getservers[`servertype;`rdb`hdb;()!()]]; +chk["procname=gw1 returns the gateway handle";enlist handles 4;exec handle from srvsel.getservers[`procname;`gw1;()!()]]; +chk["unknown servertype returns empty";0;count srvsel.getservers[`servertype;`nope;()!()]]; +chk["attribmatch column present";1b;`attribmatch in cols srvsel.getservers[`;`;()!()]]; +chk["complete date match flagged for some hdb on d1";1b; + any {x[`date]0} each (srvsel.getservers[`servertype;`hdb;(enlist`date)!enlist enlist d1])`attribmatch]; + +/ ========================================================================= +hdr"STEP 4 selector (strategies on a known table)"; +seltab:([]serverid:101 102 103i;handle:201 202 203i;lastp:(2024.01.03D0;2024.01.01D0;2024.01.02D0)); +chk["roundrobin picks oldest lastp";202i;(srvsel.selector[seltab;`roundrobin])`handle]; +chk["last picks newest lastp";201i;(srvsel.selector[seltab;`last])`handle]; +chk["any picks a row from the table";1b;((srvsel.selector[seltab;`any])`handle) in 201 202 203i]; +empt:([]serverid:`int$();handle:`int$();lastp:`timestamp$()); +chk["empty table -> null handle";0Ni;(srvsel.selector[empt;`roundrobin])`handle]; + +/ ========================================================================= +hdr"STEP 5 live routing (gethandlebytype / gethpbytype / getserverbytype)"; +/ round-robin across the two live rdbs: both start with null lastp, so LRU +/ visits rdb1, rdb2, rdb1, rdb2 - and each query is answered by that process +resp:{[i] route[`rdb;`roundrobin]} each til 4; +chk["round-robin routes to live rdbs in LRU order";`rdb1`rdb2`rdb1`rdb2;resp]; +chk["only the two live rdbs ever answer";`rdb1`rdb2;asc distinct resp]; +chk["gethpbytype gateway returns its hpup";gwhp;srvsel.gethpbytype[`gateway;`roundrobin]]; +chk["getserverbytype returns requested column";`gw1;srvsel.getserverbytype[`gateway;`procname;`roundrobin]]; +chk["gethandlebytype unknown type -> ()";();srvsel.gethandlebytype[`nope;`roundrobin]]; +chk["a single hdb query is answered by an hdb";1b;(route[`hdb;`roundrobin]) in `hdb1`hdb2]; +hitsbefore:exec sum hits from srvsel.getserverstable[] where servertype=`gateway; +srvsel.gethandlebytype[`gateway;`roundrobin]; +chk["selection increments hits";1b;(hitsbefore+1)=exec sum hits from srvsel.getserverstable[] where servertype=`gateway]; + +/ ========================================================================= +hdr"STEP 6 getserverids - servertype (symbol) path"; +chk["rdb -> the two rdb serverids";1 2i;asc raze srvsel.getserverids[`rdb]]; +chk["rdb,hdb -> all four serverids";1 2 3 4i;asc raze srvsel.getserverids[`rdb`hdb]]; +srvsel.setserveractive[handles 0;0b]; srvsel.setserveractive[handles 1;0b]; +chkerr["all requested servertype inactive -> error";{srvsel.getserverids[`rdb]}]; +srvsel.setserveractive[handles 0;1b]; srvsel.setserveractive[handles 1;1b]; + +/ ========================================================================= +hdr"STEP 7 getserverids - attribute path + routing to the matched backend"; +/ hdb1 covers (d1,d2)/(A,B); hdb2 covers (d2,d3)/(B,C) +chk["date=d1 -> hdb1 only";enlist`hdb1; + distinct {x"ping[]"} each sidhandles raze srvsel.getserverids[enlist[`date]!enlist enlist d1]]; +chk["cross (d1,d2)x(A,B) -> hdb1 alone covers it";enlist`hdb1; + distinct {x"ping[]"} each sidhandles raze srvsel.getserverids[`date`sym!((d1;d2);`A`B)]]; +chk["independent over d1,d2,d3 + A,B,C -> both hdbs contribute";`hdb1`hdb2; + asc distinct {x"ping[]"} each sidhandles raze srvsel.getserverids[`date`sym`attributetype!((d1;d2;d3);`A`B`C;`independent)]]; +chk["servertype key scopes attribute filter to hdbs";3 4i;asc raze srvsel.getserverids[`servertype`date!(`hdb;enlist d2)]]; +chk["empty requirement dict -> all active serverids";1 2 3 4 5i;asc raze srvsel.getserverids[()!()]]; +chkerr["besteffort=0b impossible -> error";{srvsel.getserverids[`date`besteffort!(enlist 2099.01.01;0b)]}]; +chkerr["no server has requested attribute value -> error";{srvsel.getserverids[enlist[`date]!enlist enlist 2099.01.01]}]; + +/ ========================================================================= +hdr"STEP 8 addserversfromtable (bulk registration of live handles)"; +conntab:([]w:handles 5 6;proctype:`rdb`hdb;attributes:(()!();`date`sym!((d1;d3);`A`C))); +srvsel.addserversfromtable[`rdb`hdb;conntab]; +chk["bulk-registered handles 5,6 now present";2=count select from srvsel.getserverstable[] where handle in handles 5 6;1b]; +chk["bulk-registered rdb3 is reachable via its handle";`rdb3;(handles 5)"ping[]"]; +chk["bulk-registered rdb3 now appears in rdb serverids";1b;6i in raze srvsel.getserverids[`rdb]]; +/ skip-already-active: re-offering a live handle must not duplicate it +n0:count srvsel.getserverstable[]; +srvsel.addserversfromtable[`rdb;([]w:enlist handles 0;proctype:enlist`rdb;attributes:enlist()!())]; +chk["already-active handle skipped (no new row)";n0;count srvsel.getserverstable[]]; +/ (proctype filter, `ALL, and optional procname/hpup columns are covered in test.csv) + +/ ========================================================================= +hdr"STEP 9 disconnect: kill a backend, deactivate it, prove routing adapts"; +/ a real gateway's .z.pc handler would call setserveractive[h;0b] on the dropped handle +/ guard: the kill is only a genuine test if rdb1's pid was captured. if pids 0 were null, +/ string 0Ni -> "0Ni" and the kill silently no-ops, degrading STEP 9 into a deactivation-only +/ test. fail loudly here rather than pass on a half-exercised path. +chk["rdb1 pid captured (so the kill below is real)";1b;not null pids 0]; +@[system;"kill ",string[pids 0]," 2>/dev/null; true";{}]; system"sleep 0.3"; +srvsel.setserveractive[handles 0;0b]; +chk["disconnect stamped disconnecttime on the dropped handle";1b; + not null first exec disconnecttime from srvsel.getserverstable[] where handle=handles 0]; +chk["a still-active server keeps a null disconnecttime";1b; + null first exec disconnecttime from srvsel.getserverstable[] where handle=handles 1]; +/ active rdbs are now rdb2 (live) and rdb3 (live); the dead rdb1 must never be picked +postresp:{[i] route[`rdb;`roundrobin]} each til 6; +chk["killed/deactivated rdb1 is never selected";1b;not `rdb1 in postresp]; +chk["no routed query hits a dead handle (all answered)";1b;not `NONE in postresp]; +chk["the two remaining live rdbs both serve";`rdb2`rdb3;asc distinct postresp]; + +/ ========================================================================= +hdr"STEP 10 error handling (every guarded path signals + logs)"; +chkerr["addserverfull rejects non-int handle";{srvsel.addserverfull[`bad;`;`rdb;`;()!()]}]; +chkerr["addserverfull rejects non-symbol servertype";{srvsel.addserverfull[71i;`;"rdb";`;()!()]}]; +chkerr["addserver rejects bad handle";{srvsel.addserver[`bad;`rdb]}]; +chkerr["addserverattr rejects bad servertype";{srvsel.addserverattr[73i;"hdb";()!()]}]; +chkerr["setserveractive rejects non-int handle";{srvsel.setserveractive["bad";0b]}]; +chkerr["setserveractive rejects non-boolean flag";{srvsel.setserveractive[1i;`yes]}]; +chkerr["getservers rejects bad nameortype";{srvsel.getservers[`badname;`x;()!()]}]; +chkerr["selector rejects unknown strategy";{srvsel.selector[seltab;`bogus]}]; +chkerr["getserverids rejects null servertype";{srvsel.getserverids[`]}]; +chkerr["getserverids rejects unregistered servertype";{srvsel.getserverids[`nope]}]; +chkerr["getserverids rejects non-symbol non-dict arg";{srvsel.getserverids[42]}]; +chkerr["getserverids rejects nested/malformed attribute value";{srvsel.getserverids[(enlist`date)!enlist enlist d1,d2]}]; +chkerr["addserversfromtable rejects missing columns";{srvsel.addserversfromtable[`rdb;([]x:enlist 1i)]}]; + +/ ========================================================================= +hdr"STEP 11 init (required log dependency - bespoke + kx.log)"; +caplog:([]lvl:`symbol$();msg:()); +caplogger:`info`warn`error!({[c;m]`caplog upsert(`info;m)};{[c;m]`caplog upsert(`warn;m)};{[c;m]`caplog upsert(`error;m)}); +srvsel.init[enlist[`log]!enlist caplogger]; +caplog:0#caplog; +srvsel.addserver[40i;`rdb]; +chk["injected logger captures an info message";1b;0type d;raiseerror[ctx;nm," must be a dictionary; got type ",string type d]]; + if[not $[0=count k:key d;not 98h=type k;11h=type k]; + raiseerror[ctx;nm," must be a symbol-keyed dictionary; got keys of type ",string type k]]; + }; + +requireinit:{[ctx] + / internal - every public entry point needs init to have run first. without this guard a bare read + / of an unwritten .z.m name surfaces as a raw '.m.di.0serverselect. error, leaking the mangled + / internal namespace to the caller instead of naming the actual problem. signals plainly rather than + / via raiseerror - there is no logger to log through yet. + / getapimeta is deliberately NOT guarded: it is pure data and di.torq collects it at startup, which + / may be before this module's init has run + if[not `logerr in key .z.m;'"di.serverselect: ",string[ctx],": init must be called first"]; + }; + +selectorarity:{[f] + / internal - arity of f WHERE Q CAN TELL, else 0N. + / a lambda reports its own parameter list. a projection's REMAINING arity is its underlying + / function's rank minus the arguments already supplied - counting elided (::) placeholders alone is + / wrong, because a trailing partial application such as f[x] carries none at all and would read as + / arity 0, rejecting a perfectly valid strategy. the underlying rank is resolved recursively so a + / projection OF a projection also works. + / primitives, compositions and adverb-derived functions report nothing usable, so they yield 0N and + / the caller accepts them unchecked - admitting the check does not apply beats rejecting valid input + :$[100h=type f; count value[f]1; + 104h=type f; $[null b:selectorarity first value f; + 0N; + b-count where not (::)~/:1_value f]; + 0N]; + }; + +getopt:{[deps;k;dflt] + / internal - read an optional config key from the deps dict, falling back to a default + $[k in key deps;deps k;dflt] + }; + +checkopt:{[deps;k;ok;what] + / internal - validate an optional config value's TYPE at init, so a misconfiguration fails loudly + / at startup instead of silently at first use + if[k in key deps; + if[not ok deps k;'"di.serverselect: ",string[k]," must be ",what]]; + }; + +init:{[deps] + / wire the injected logger - required, no silent fallback - and the optional config. + / deps keys: + / log (required) `info`warn`error!{[c;m]} dict - binary, already conforming. + / a raw monadic kx.log instance is NOT adapted here and will 'rank at first use + / cp (optional) current-time fn, default {.z.p}. stamps disconnecttime and drives + / removeinactive; override it to fast-forward in tests without sleeping + / clearinactivetime (optional) timespan, default 0D01:00. NOT read by any function here - + / removeinactive is caller-invoked, so this is the age di.torq passes it when + / it schedules the purge, mirroring di.dataaccess's requestkeeptime + / maxcrossproduct (optional) long, default 1000000. upper bound on the requirement cross + / product getserverids builds when cross matching, checked before it is built - + / cost is the PRODUCT of the requirement value counts, so a gateway forwarding + / client-supplied requirements can get very large very cheaply. 0W disables it + / e.g. di.serverselect.init[enlist[`log]!enlist logdep] + if[99h<>type deps; + '"di.serverselect: deps must be a dict with `log key"]; + if[not `log in key deps; + '"di.serverselect: log dependency is required; pass `info`warn`error functions keyed on `log"]; + if[99h<>type deps`log; + '"di.serverselect: log value must be a dict; pass `info`warn`error functions"]; + if[not all (`info`warn`error) in key deps`log; + '"di.serverselect: log dict must have `info`warn`error keys; got: ",(", " sv string key deps`log)]; + checkopt[deps;`cp;{type[x] within 100 112h};"a function"]; + checkopt[deps;`clearinactivetime;{-16h=type x};"a timespan"]; + checkopt[deps;`maxcrossproduct;{(-7h=type x) and 00Np is true for everything + if[0Wn=age;:()]; + .z.m.servers:delete from servers where not active,not null disconnecttime,.z.m.cp[]>disconnecttime+age; + }; + +getserverstable:{[] + / return the current registered server table + requireinit`getserverstable; + :servers; + }; + +addserversfromtable:{[proctypes;conntable] + / register active servers from a connection table filtered by proctype + / conntable must have columns: w (int handle), proctype (symbol), attributes (dict per row) + / optional columns: procname (symbol), hpup (symbol) - populated from conntable if present + / pass proctypes:`ALL to register all process types + requireinit`addserversfromtable; + / unkeyed table required: cols works on a keyed table but the select below does not, so without + / this the caller gets a raw 'type instead of a message naming the argument + if[not 98h=type conntable; + raiseerror[`addserversfromtable;"conntable must be an unkeyed table; got type ",string type conntable]]; + if[not all `w`proctype`attributes in cols conntable; + raiseerror[`addserversfromtable; + "conntable must have columns w, proctype and attributes; got: ",", " sv string cols conntable]]; + activehandles:(0i;0Ni),exec handle from servers where active; + rows:select from conntable where + ((proctype in proctypes) or proctypes~`ALL), + not w in activehandles; + .z.m.loginfo[`addserversfromtable;"registering ",(string count rows)," servers from connection table"]; + pnames:$[`procname in cols rows; rows`procname; count[rows]#`]; + hpups:$[`hpup in cols rows; rows`hpup; count[rows]#`]; + addserverfull'[rows`w;pnames;rows`proctype;hpups;rows`attributes]; + }; + +attributematch:{[req;avail] + / compute match result for each key in req against what avail advertises + / returns dict of attrname!(complete_match_bool;matched_values) for each required attribute key + / keys present in req but absent in avail return (0b;()) + vals:key[req] inter key avail; + notpresent:noval!(count noval:key[req] except key avail)#enlist(0b;()); + :notpresent,vals!{($[0>type y;x~y;all x in y];(x,()) inter y,())}'[req vals;avail vals]; + }; + +getservers:{[nameortype;lookups;req] + / look up active servers by servertype or procname with per-attribute match scoring + / nameortype: `servertype or `procname; pass ` as lookups to return all active servers + / req: attribute requirements dict - use ()!() for no attribute filtering + / returns table with attribmatch column showing (complete_bool;matched_values) per attribute key + requireinit`getservers; + if[(not `~lookups) and not nameortype in `servertype`procname; + raiseerror[`getservers;"nameortype must be `servertype or `procname; got: ",string nameortype]]; + requiredict[`getservers;"req";req]; + r:$[`~lookups; + select serverid,procname,servertype,hpup,handle,lastp,attributes from servers where active; + nameortype~`servertype; + select serverid,procname,servertype,hpup,handle,lastp,attributes from servers where active,servertype in lookups; + select serverid,procname,servertype,hpup,handle,lastp,attributes from servers where active,procname in lookups]; + if[0=count r;:update attribmatch:attributes from r]; + am:attributematch[req] each r`attributes; + :update attribmatch:am from r; + }; + +selector:{[servertable;selection] + / pick one row from servertable using the given strategy + / selection: `roundrobin (least recently used), `any (random), `last (most recently used) + requireinit`selector; + if[not selection in `roundrobin`any`last; + raiseerror[`selector;"unknown selection strategy: ",string selection]]; + :$[selection=`roundrobin; first `lastp xasc servertable; + selection=`any; rand servertable; + last `lastp xasc servertable]; + }; + +setselector:{[f] + / replace the strategy getserverbytype uses to pick one row from a candidate table. + / the built-in selector stays exported and unchanged - pass it back here to restore the default + requireinit`setselector; + if[not type[f] within 100 112h;raiseerror[`setselector;"selector must be a function; got type ",string type f]]; + / check arity HERE where the mistake is, not at the next getserverbytype. a monadic or niladic + / strategy used to be accepted happily and then throw a bare 'rank from inside the module, far from + / the call site. 0N means q cannot report an arity for this function type, so it is accepted + ar:selectorarity f; + if[not null ar; + if[2<>ar; + raiseerror[`setselector;"selector must take 2 arguments (servertable;selection); got ",string ar]]]; + .z.m.selector:f; + }; + +getserverbytype:{[ptype;serverval;selection] + / return a single server attribute value for a servertype using the given selection strategy + / ptype: servertype symbol; serverval: column to return e.g. `handle or `hpup; selection: `roundrobin`any`last + / dispatches through the live .z.m.selector so setselector can override the strategy + requireinit`getserverbytype; + r:getservers[`servertype;ptype;()!()]; + if[not count r;:()]; + r:.z.m.selector[r;selection]; + updatestats[r`serverid]; + :r serverval; + }; + +gethandlebytype:getserverbytype[;`handle;]; +gethpbytype:getserverbytype[;`hpup;]; + +normreq:{[req] + / internal - promote atom attribute-requirement values to one-element lists, so a caller may write + / (enlist`date)!enlist 2024.01.01 as well as the enlist form. getservers already tolerates an atom + / via attributematch; without this the cross matcher threw a raw, unlogged 'rank on the same input. + / req is PURE requirements by this point - control keys are split off at the boundary - so every + / value is promoted, including one whose key happens to be spelled like a control key + :(key req)!{(),x} each value req; + }; + +raisecaught:{[ctx;msg] + / internal - log an error the engine already prefixed, then re-signal it VERBATIM. the engine's own + / text ("no servers match hdb requested attributes") is more specific than anything composed here, + / and signalnomatch deliberately does not log - the boundary decides the level. used on the + / single-servertype path, where there is nothing to tolerate: one type missing IS the whole answer + .z.m.logerr[ctx;msg]; + 'msg; + }; + +fanoutids:{[req;besteffort;attype;typ] + / internal - resolve one servertype's ids, tolerating a miss. a miss is NOT a query failure: another + / requested type may match, and getserverids errors only when every type comes back empty. this + / revives getserverids' all-empty guard, which was unreachable while this call threw instead + :@[getserveridstype[req;besteffort;attype;];typ; + {[t;e] .z.m.logwarn[`getserverids;"no servers matched servertype ",string[t],": ",e];()}[typ]]; + }; + +getserverids:{[att] + / return server IDs matching a servertype list or attribute requirement dictionary + / att: symbol list of servertypes, or dict of attribute requirements (optionally keyed on `servertype) + / dispatch the symbol-list path to getserveridsbytype and the dict path to getserveridstype + requireinit`getserverids; + if[99h<>type att; :getserveridsbytype att]; + / TWO request shapes. flat (the original): control keys sit alongside the attribute requirements, so + / the three control names are reserved and an attribute cannot be called servertype/besteffort/ + / attributetype. nested: an `attrs key holds the requirements EXPLICITLY, leaving the rest of the + / dict to the controls - which lets an attribute carry any name at all, control names included. + / both are supported; nested is the way out of the namespace collision, flat stays for compatibility + nested:`attrs in key att; + if[nested; + requiredict[`getserverids;"attrs";att`attrs]; + / in the nested form EVERY top-level key that is not `attrs is a control, so an unrecognised one + / would be silently swallowed - dropping a requirement left outside attrs by a half-migrated + / caller, or ignoring a mistyped control. the flat form cannot have this problem (an unknown key + / there is simply a requirement), so the nested form has to be strict to stay as safe + if[count unknown:(key att) except `attrs,ctrlkeys; + raiseerror[`getserverids;"unknown keys beside attrs: ",(", " sv string unknown), + "; only attrs and the control keys belong at the top level of an attrs request"]]]; + ctl:$[nested;`attrs _ att;(key[att] inter ctrlkeys)#att]; + req:normreq $[nested;att`attrs;ctrlkeys _ att]; + / controls are validated HERE, at the boundary, and never inside getserveridstype: the fan-out below + / runs under a protected apply, so an error raised deeper down is caught and downgraded to a "this + / servertype did not match" warning. a misconfigured request is categorically not a per-type miss and + / must stay loud. before this, besteffort:0 (int) silently kept the 1b default and an unknown + / attributetype silently fell back to cross matching + if[`servertype in key ctl; + if[not 11h=abs type ctl`servertype; + raiseerror[`getserverids;"servertype must be a symbol or symbol list; got type ",string type ctl`servertype]]]; + if[`besteffort in key ctl; + if[not -1h=type ctl`besteffort; + raiseerror[`getserverids;"besteffort must be a boolean; got type ",string type ctl`besteffort]]]; + if[`attributetype in key ctl; + if[not -11h=type ctl`attributetype; + raiseerror[`getserverids;"attributetype must be a symbol; got type ",string type ctl`attributetype]]; + if[not (ctl`attributetype) in `cross`independent; + raiseerror[`getserverids;"attributetype must be `cross or `independent; got: ",string ctl`attributetype]]]; + / a nested/general-list requirement value (type 0h) would crash the cross matcher with a raw 'type + if[count badkeys:(key req) where 0h=type each value req; + raiseerror[`getserverids; + "attribute requirement values must be atoms or simple vectors; nested/mixed for: ",", " sv string badkeys]]; + besteffort:$[`besteffort in key ctl;ctl`besteffort;1b]; + attype:$[`attributetype in key ctl;ctl`attributetype;`cross]; + / bound the cross product before building it. cost is the PRODUCT of the requirement value counts, so + / a request a caller forwards straight from a client can get very large very cheaply. only cross + / matching builds it - independent matching is not combinatorial, so it is not bounded here. + / maxcrossproduct:0W disables the bound + / count req guard is NOT optional: prd of an empty list is () rather than 1, so an empty requirement + / dict would make the comparison return () and if[()] throw a raw 'type + if[attype=`cross; + if[.z.m.maxcrossproduct