diff --git a/di/servers/VERSION b/di/servers/VERSION new file mode 100644 index 00000000..6e8bf73a --- /dev/null +++ b/di/servers/VERSION @@ -0,0 +1 @@ +0.1.0 diff --git a/di/servers/init.q b/di/servers/init.q new file mode 100644 index 00000000..ad8a274c --- /dev/null +++ b/di/servers/init.q @@ -0,0 +1,12 @@ +/ connection management and handle-by-type lookup for the modular torq world. +\l ::servers.q +/ module version, read from the VERSION file rather than hardcoded, so a release bump touches one +/ plain-text file. read module-relative at load (`:::` resolves to di/servers), and BEFORE the export +/ line since export:([...]) evaluates each name. NB `version` stays in the export: di.depcheck +/ resolves a dependency's version from the export dict. +/ trim so a trailing newline/CRLF cannot pad the semver; fail loud with a clear message if VERSION is +/ missing/unreadable/empty (it is a required module file - better than a raw OS error, a silent empty +/ value, or a misleading 0.0.0 that would corrupt depcheck's version comparison). +version:@[{trim first read0 x};`:::VERSION;{'"di.servers: VERSION file missing or unreadable"}]; +if[0=count version;'"di.servers: VERSION file is empty"]; +export:([init;teardown;startup;getservers;gethandlebytype;waitfortype;getapimeta;version]) diff --git a/di/servers/servers.md b/di/servers/servers.md new file mode 100644 index 00000000..dda421ea --- /dev/null +++ b/di/servers/servers.md @@ -0,0 +1,279 @@ +# di.servers + +Connection management and handle-by-type lookup for the modular TorQ world — the `di.*` +analogue of TorQ's `.servers` (`code/handlers/trackservers.q` + `servers.q`), scoped down +for v1: no discovery service, no password/access-list files, no non-TorQ process tracking, +no FinSpace. `process.csv` here is a static **phone book** (who to *dial*), **not** an +identity source — self-identity comes from config, injected by `di.torq`. + +FRAMEWORK-tier module: no hard `di.*` dependencies; `log`, `timer` and `handlers` are all +**injected** (all required, no fallback). + +## init and config + +Standard **one-arg `init[deps]`**: `di.torq` merges this process's resolved config slice into +the same `deps` dict it passes the injectables in, so `deps` carries both the injectable +dependencies **and** the config keys. `init` wires the deps, records self-identity, and +installs two one-time process-global side effects — a `.z.pc` cleanup handler and a 10s retry +timer job. It is **idempotent** (guarded by an internal `registered` flag): `di.torq` calls +it once per process, but a second call refreshes the dep refs without re-registering (a +duplicate `di.timer.addjob` id would throw). `teardown` resets that flag, so an +`init`→`teardown`→`init` cycle re-installs both side effects rather than being skipped by the +guard. `init` does **not** open connections. + +`deps` keys: + +| key | kind | meaning | +|---|---|---| +| `log` | injectable | binary `` `info`warn`error `` `{[c;m]}` logger dict — di.log's `logdict``log` satisfies this directly (it carries all six levels; the extra `trace`/`debug`/`fatal` are ignored) | +| `timer` | injectable | the di.timer export dict. Must expose `` `addjob `` (a variant dict of `custom`/`default`/`simple`; di.servers calls the 6-arg `` timer[`addjob][`custom] ``) **and** `` `deletejobs ``, which must be **callable** — `teardown` needs it, and a timer dep missing it fails *silently* there rather than loudly at `init` (see [Lifecycle](#lifecycle-requireinit-and-teardown)) | +| `handlers` | injectable | di.handlers contract. Must expose callable `` `register `` (`register[event;phase;nm;pri;func]` — `init` uses it) **and** `` `remove `` (`remove[event;phase;nm]` — `teardown` uses it) | +| `proctype`/`procname` | config | this process's own identity (required); used to exclude self from `process.csv` | +| `connections` | config | proctypes this process should dial (symbols, or strings from a `.toml` cascade — normalised). Optional; default = none | +| `processcsv` | config | **path** to `process.csv`; supplied by di.torq. Optional; required only once `connections` is non-empty | + +```q +svc:use`di.servers +svc.init[deps] / deps = injectables + config, assembled by di.torq +svc.startup[] / open the configured connections (reads init config) +h:svc.gethandlebytype[`hdb;`any] +h "1+1" +``` + +## Exported functions + +| Function | Signature | Description | +|---|---|---| +| `init` | `init[deps]` | Wire deps + config, record identity, install the `.z.pc` handler + retry job. Idempotent. **The only function that may be called before `init`** — every other one below (except `getapimeta`) refuses to run until it has completed. | +| `teardown` | `teardown[]` | Release both process-global registrations `init` installed — the `.z.pc` handler and the `serversretry` timer job. Module state, `SERVERS` included, is **deliberately left intact**. Idempotent. See [Lifecycle](#lifecycle-requireinit-and-teardown). | +| `startup` | `startup[]` | Read `process.csv` (`processcsv`), drop self, connect to each row whose proctype is in `connections`. A failed connection is logged (not raised) and left as `w:0Ni` for `retry`. No-op if no connections configured. **Idempotent**: skips procs already tracked in `SERVERS`, so a repeat call (or a grown `process.csv`) adds only new rows — never a duplicate or a leaked second handle. `process.csv` must be the strict v1 4-column `host,port,proctype,procname` layout — a reordered or wider header is **rejected loudly** (the reader is positional, so it would otherwise misparse silently). **Self-exclusion is an exact `(proctype;procname)` match against the identity from config**, so a `process.csv` that disagrees would leave this process dialling itself; any row carrying this process's `procname` **that survives the connections filter (i.e. one this process would otherwise dial)** is therefore skipped with a warning naming both proctypes. A drifted row whose proctype isn't a connection type is already filtered out and poses no self-connection risk, so it is not warned about. | +| `getservers` | ``getservers[proctype]`` | Live (`w` non-null) `SERVERS` rows. Accepts a **symbol**, a **symbol list** (rows for any of them), or `` ` `` for **every** proctype — the contract legacy TorQ's `.servers.getservers` (`trackservers.q:75`) and sibling `di.serverselect.getservers` both implement, so a consumer written against either works here unchanged. | +| `gethandlebytype` | `gethandlebytype[proctype;selection]` | One live handle via `` `any``/`roundrobin`/`last``; `0Ni` if none. Bumps usage stats. | +| `waitfortype` | `waitfortype[proctype;timeoutms;pollms]` | Block until a live connection exists or timeout; `1b`/`0b`. Caller decides if a timeout is fatal. `startup` must have run first. A **zero or already-expired `timeoutms`** performs no active retry — see [waitfortype with a zero timeout](#waitfortype-with-a-zero-timeout). | +| `getapimeta` | `getapimeta[]` | This module's api metadata, one row per **callable** API function (`init`/`getapimeta`/`version` plumbing omitted), for `di.torq` to register with `di.api`. | +| `version` | `version` | The module's semver string (`"0.1.0"`) — metadata, not a function. Read at load from the plain-text `VERSION` file in the module folder; `di.depcheck` resolves it from the export dict to satisfy other modules' declared minimum-version requirements. | + +Export is deliberately conservative — only functions `di.torq` or a consumer actually calls +(so `di.api` lists exactly these), plus the `version` metadata string. The rest are **internal**: `retry` (the scheduled +`serversretry` job — passed to the timer *by value* at init, so it needs no export; it first +runs `cleanup` to sweep ungracefully-vanished handles, then reopens every dead handle), +`cleanup`, `formathp`, `opencon`, `readprocesscsv`, `retryrows`, `selector`, `updatestats`, +`signalfound`, `raiseerror`, `initialised`, `requireinit`, `iscallable`; plus state (`SERVERS`, `self`, +`registered`, `HOPENTIMEOUT`, `connections`, `processcsv`). + +## Lifecycle: requireinit and teardown + +### `requireinit` — nothing runs before `init` + +`startup`, `getservers`, `gethandlebytype`, `waitfortype` and `teardown` each call +`requireinit[ctx]` as their first statement and signal +`di.servers: : init must be called before any other function` if `init` has not completed. +`getapimeta` deliberately does **not** — it returns static metadata derived from no init'd state, +matching every other module in this project. + +This is not a nicety. `SERVERS`, `self`, `registered` and `HOPENTIMEOUT` are **load-time** +constants, so before the guard existed a pre-init call was *silently* wrong rather than loud: + +| call | before `init`, without the guard | indistinguishable from | +|---|---|---| +| `getservers[pt]` | an empty table | no process of that type is connected | +| `gethandlebytype[pt;sel]` | `0Ni` | nothing of that type is connected right now | +| `waitfortype[pt;t;p]` | spins the **full** timeout, then `0b` | a peer that never came up | + +Only `startup` threw at all, and then with a bare unnamed error off the unset `.z.m.connections`. +With no `di.torq` yet, every process here is wired by a hand-written script calling `init` on +several modules in order — exactly the situation where an ordering mistake is plausible — and a +failure that looks precisely like "the dependency is still starting up" is the worst shape this bug +could take. `di.rdb` already wraps its `gethandlebytype` calls protectively *on the assumption this +guard exists* (`rdb.q:656`). + +**Why the probe reads `.z.m.loginfo`.** `initialised[]` must read something whose value can only +come from `init`. `self` and `registered` cannot serve — both are declared at **load** time, so +neither distinguishes "init ran" from "the module was merely loaded". `.z.m.loginfo` has no +load-time default and is written only by `init`; it is also the truest probe of what the guard +protects, since without a wired logger the module cannot even report its own failures. + +It is deliberately **not** `registered`. `teardown` resets that flag, and conflating the two would +make every function report *"init must be called before any other function"* after a teardown — +untrue, and it would put `SERVERS` out of reach of exactly the shutdown path `teardown` leaves it +intact for. The two flags stay separate, as in `di.rdb` (probes `hdbdir`, resets `started`) and +`di.subscriptions` (probes `subscriptions`, resets `observing`). + +### `teardown[]` — what it releases, and what it does not + +`init` installs **two** process-global things, so there is something to give back: + +| released by `teardown` | how | +|---|---| +| the `.z.pc` cleanup handler | `` handlers[`remove][`.z.pc;`;`servers] `` — same event, phase and registrant name it was registered under | +| the `serversretry` timer job | `` timer[`deletejobs][`serversretry] `` | + +**Not** released — module state is deliberately left intact, the same convention every other +teardown in this project follows: + +- `SERVERS` keeps every row, so a shutdown path can still see what was connected and to whom. +- The logger, timer and handler dep refs, self-identity, `connections` and `processcsv` all stay. +- `getservers`, `gethandlebytype`, `waitfortype` and `startup` all remain callable. What is + withdrawn is only the *automatic* behaviour: with the `serversretry` job gone nothing reattempts + a dropped connection on a cycle, and with the `.z.pc` handler gone a clean disconnect is no longer + swept. `waitfortype` still drives `retry` itself, so it keeps working. There is no + `requireobserver`-style refusal on `startup` after a teardown. + +`teardown` resets `registered` to `0b`, so a later `init` installs both side effects again rather +than being skipped by the idempotency guard — a `teardown`→`init` cycle fully works. + +**Idempotent — a second `teardown[]` does not throw.** Both release calls are no-ops on an +already-removed registration, verified in the dependencies rather than assumed: +`di.timer.deletejobs` is a delete-where over its jobs table (`di/timer/init.q:95`), so an id that +is not there matches nothing; `di.handlers`' `removesimple` early-returns with an info log when the +event or the name is not registered (`di/handlers/handlers.q:71-72`). + +**Why `init` validates the timer's `deletejobs`.** A timer dep's value side is dict-typed, so a +*missing* key returns a null-shaped **dict** rather than erroring. `@[x;y;z]` is "try `x[y]`, catch +with `z`" only when `x` is a **function** — with a dict, `teardown`'s +`` @[.z.m.timer[`deletejobs];`serversretry;handler] `` is read as three-argument **amend** instead: +it upserts the id into that throwaway dict using the error handler as the value, discards the result +and carries on. Nothing throws, nothing warns, the job is never deleted, and `teardown` still logs +success — a false positive. Presence alone is not enough either: a non-callable `deletejobs` lands +in the identical amend. `init` therefore checks both presence and callable type, and the same pair +is applied to `handlers`' `register` and `remove`. The checks must be at `init` time because +`teardown` cannot detect the problem at all. + +The callable test is the internal `iscallable`, **not** a bare `` within 100 112h ``. That range +spans every genuinely callable form — lambda, primitive, operator, iterator, projection, +composition — but it also admits **`101h`, the generic null `::`**, which is callable in no useful +sense and is precisely what a dict hands back for a *missing* key when its value side is plain +functions (`` `deletejobs _ di.timer `` yields `101h`; only a *table*-valued dep yields the `99h` +null-shaped dict). Left in, a dep that was absent or explicitly null passed `init`, and `teardown` +then ran `` @[::;`serversretry;handler] `` — which simply returns the id, deleting nothing, while +`teardown` logged success. Measured end to end, which is why `iscallable` excludes `101h`: + +```q +iscallable:{[x] t:type x; :(t within 100 112h) and 101h<>t; }; +``` + +> **Known divergence.** `di.rdb` performs the equivalent check with a bare `` within 100 112h `` and +> therefore still carries this hole. It was left untouched deliberately — it is a separate module and +> its own unit of work — but the same tightening applies there. + +## `waitfortype` with a zero timeout + +**Decision: `waitfortype[pt;0;pollms]` performs no active retry, and that is intended.** With a +zero (or already-expired) `timeoutms` the deadline is in the past by the time the loop condition is +first evaluated, so `retry[]` is never called; the function reports current state, logs the usual +timeout warning, and returns `0b`. + +Reasoning, recorded rather than left implicit: + +- `retry[]` calls `opencon` for every dead row at `HOPENTIMEOUT` = 2000 ms. Granting a "0 ms" call + one active attempt could therefore block for **seconds** per unreachable peer — violating the + caller's explicit budget in the most surprising possible direction. A caller who asks for no wait + should get no wait. +- Nothing is lost. The scheduled `serversretry` job reattempts every 10 s regardless, so a reconnect + is *deferred*, not skipped. +- A caller who does want one active attempt asks for one by passing a `timeoutms` of at least + `pollms`; there is no need for a zero timeout to mean something different from what it says. +- The `0b` return still carries the existing timeout warning, so the outcome stays diagnosable. + +## The `SERVERS` table + +```q +SERVERS:([]procname:`symbol$();proctype:`symbol$();hpup:`symbol$();w:`int$();hits:`int$();startp:`timestamp$();lastp:`timestamp$();endp:`timestamp$()) +``` + +A direct analogue of legacy TorQ's `.servers.SERVERS`: `w` is the live handle (`0Ni` when +disconnected), `hits`/`lastp` drive handle selection, `startp`/`endp` track lifecycle. + +## `.z.pc` registration via di.handlers + +`.z.pc` (connection closed) is a **simple/observer** event in di.handlers — side-effect only, +fan-out — so di.servers registers its cleanup callback through the injected `handlers` +dependency rather than assigning `.z.pc` directly: + +```q +(handlers[`register])[`.z.pc;`;`servers;0j;pcfunc] +``` + +`register`'s signature is `register[event;phase;nm;pri;func]`; for a simple event the `phase` +must be `` ` `` (null) — di.handlers rejects a non-null phase on an observer event. This lets +di.servers' disconnect hook coexist with every other `.z.pc` registrant in the same +priority-ordered fan-out. + +`teardown` gives the registration back through the same dependency, under the same event, phase +and registrant name: + +```q +(handlers[`remove])[`.z.pc;`;`servers] +``` + +## Conventions (learnings from di.config) + +- **One-arg `init[deps]`** with config folded into `deps` (the project convention; matches + `di.eodtime`'s optional-config-in-deps pattern), not a two-arg `init[config;deps]`. +- **Three-flat-var logging** — `.z.m.loginfo`/`.z.m.logwarn`/`.z.m.logerr`, matching + `consistency.md`, `di.compression` and `di.config`. (The project hasn't globally frozen this + vs. the single-dict form — flag before changing.) +- **`raiseerror` (log-then-signal)** for all post-init domain errors (`selector` unknown + selection, missing or malformed `process.csv`). `init`'s own dependency validation is the one + exception (plain `'` — no logger yet). +- **`getapimeta`** exported; a test asserts it documents exactly the module's *callable* + exports — `init`/`getapimeta`/`version` are plumbing/metadata (di.torq calls or reads them by + convention) and are deliberately omitted from the registry rows, matching di.toml and the skill + convention. +- **`version` export** — a bare exported semver string (`"0.1.0"`, numeric `major.minor.patch`), + read by di.depcheck to satisfy other modules' declared minimum-version requirements. The single + source of truth is the **`VERSION`** file in the module folder, read module-relative at load by + `init.q` and `trim`med so a trailing newline cannot pad the semver. There is no compiled-in + fallback: a missing, unreadable or empty `VERSION` signals a named error at load rather than + yielding a silently stale or `0.0.0` version that would corrupt di.depcheck's comparison. Bump the + release by editing `VERSION` alone. +- **Env-free** — di.servers reads no environment variable; the `process.csv` path arrives via + `config`processcsv` (di.torq resolves it), holding di.config's env-free boundary. + +## Open items / not yet done + +- **Live-peer integration tests are in place** (`test.q` + `test.csv`, 84 checks), and now wire the + **real merged `di.timer` and `di.log`** — only `di.handlers` (not yet merged) is mocked. They spawn + a genuinely separate `q` peer (a self-connect returns pseudo-handle `0`, not a real socket) and + cover: `startup` connecting to a live peer and logging a failed dial while excluding self, + `gethandlebytype` returning a live remote handle (`2=h"1+1"`), the retry cycle recovering an + ungraceful kill (`cleanup`+reopen), and `waitfortype` connected-vs-timeout — plus init validation, + dep-wiring, idempotency, input validation, and `getapimeta`. `init` schedules `serversretry` in the + real `di.timer` (asserted via `` timer.getalljobs[] ``); because `retry`/`cleanup` are internal, the + retry cycle is driven by invoking the exact func di.servers handed the timer (`` firejob `` reads it + back from `` getalljobs[] `` — the actually-wired path), not a direct export. Idempotent re-init is + a genuine test here: the real timer's `` addjob[`custom] `` throws on a duplicate id, so a + non-idempotent `init` would fail outright. A final check re-inits against di.log's real `logdict` to + prove the injected-log contract holds end-to-end. + The lifecycle rows added alongside `requireinit`/`teardown` are worth knowing about when editing + the suite: the pre-init guard rows must stay **above every `init` attempt**, failing ones included + — an `init` validation gap would let one of those succeed, wire the logger, and make the guard rows + fail for that reason instead of a missing guard (measured). They assert the *named* message via + `ss`, not merely that something threw, since a bare `fail` row would pass on any unrelated throw + and a "returned empty" row would pass under the old silent behaviour. Each guard also asserts the + **function name** in the message: `gethandlebytype` calls `getservers` internally, so dropping its + own `requireinit` still produces an "init must be called" error — from the nested guard — and only + the function-name assertion catches it. `teardown` is covered for both its releases, a second call, + and an `init` after it re-installing both side effects (which is what proves the `registered` + reset). The handlers mock's `remove` deletes its recorded row rather than being a no-op, so a + broken `teardown` cannot pass. +- **`di.handlers` not in kdbx-modules yet**, so only that injected contract is mocked. The handlers + mock uses the real `register[event;phase;nm;pri;func]` shape from `handlers.q`; the `.z.pc` observer + path is therefore exercised only via the explicit `retry`→`cleanup` sweep, not a live auto-fired + `.z.pc` (which real di.handlers would install). +- **`config`processcsv` and the assembled `connections` list** depend on di.torq's config + wiring — coordinate when di.torq's servers dep is built. +- Scoped-out (v1): discovery service, password/access-list files, non-TorQ tracking, and + `tcps`/`unix` socket types. Only `tcp` is supported; `formathp` builds a `tcp` handle with no + socket-type arg — a future `SOCKETTYPE` config reintroduces that (with a test) when needed. + +## Tests + +Run in a fresh q session (spawns and kills a real peer process; don't interleave with other +modules' tests). Needs `QHOME` set (the peer is launched via `$QHOME/bin/q`) and `di.os` on +`QPATH` (the harness uses `os.abspath` to load `test.q`): + +```q +k4unit:use`di.k4unit +k4unit.moduletest`di.servers +``` diff --git a/di/servers/servers.q b/di/servers/servers.q new file mode 100644 index 00000000..c3e52659 --- /dev/null +++ b/di/servers/servers.q @@ -0,0 +1,371 @@ +/ connection management and handle-by-type lookup for the modular torq world - the di.* analogue +/ of TorQ's .servers (code/handlers/trackservers.q + servers.q), scoped down for v1: no discovery +/ service, no password/access-list files, no non-torq process tracking, no FinSpace. process.csv +/ is a static phone book (who to dial), NOT an identity source - self-identity comes from config. +/ FRAMEWORK-tier module: no hard di.* deps; log, timer and handlers are injected (all required). +/ standard one-arg init[deps]: di.torq merges this process's config slice (proctype/procname, +/ connections, processcsv) into the same deps dict it passes the injectables in. conventions match +/ di.config: strict init validation (no fallback), three-flat-var logging, log-then-signal via +/ raiseerror, getapimeta for di.api, and the env-free boundary (the process.csv path arrives via +/ config; di.servers reads no env itself). + +/ --- module-local state (initial values at load; read/written via .z.m at runtime) --- + +SERVERS:([] + procname:`symbol$(); + proctype:`symbol$(); + hpup:`symbol$(); + w:`int$(); + hits:`int$(); + startp:`timestamp$(); + lastp:`timestamp$(); + endp:`timestamp$()); + +HOPENTIMEOUT:2000; + +self:`proctype`procname!``; + +/ guards init's one-time process-global side effects (the .z.pc observer + the retry timer job) so +/ init is IDEMPOTENT - di.torq calls it once per process, but a second call (a test re-run, a +/ future re-init) must not re-register: di.timer.addjob throws on a duplicate id. the dep refs are +/ always refreshed; only the one-time registrations are guarded. teardown resets it to 0b so a +/ later init registers again - it is the SIDE-EFFECT guard only, deliberately not the initialised[] +/ probe below (see there for why the two are separate flags). +registered:0b; + +initialised:{[] + / has init run? .z.m.loginfo has no load-time default - its only value comes from init, so this + / probe cannot be fooled by a constant of the same name. that rules out the two obvious + / alternatives: `self` and `registered` are both declared at LOAD time above, so neither can tell + / "init ran" from "the module was merely loaded". it is also the truest probe of what requireinit + / actually protects - without a wired logger the module cannot even report its own failures. + / deliberately NOT `registered`: teardown resets that, and conflating the two would make every + / function report "init must be called" after a teardown, which is both untrue and would put + / SERVERS out of reach of the shutdown path teardown exists to serve. di.rdb (probes hdbdir, + / resets started) and di.subscriptions (probes subscriptions, resets observing) split them the + / same way. + :@[{.z.m.loginfo;1b};::;{[e] :0b}]; + }; + +requireinit:{[ctx] + / every exported function except init/getapimeta refuses to run before init has wired the deps. + / without this a pre-init call is SILENTLY wrong rather than loud, which is the worst shape a bug + / here could take: getservers returns an empty table, gethandlebytype 0Ni, and waitfortype spins + / its full timeout before returning 0b - all three indistinguishable from "genuinely nothing of + / that type is connected right now". signals with a plain ' and NOT raiseerror: the logger is the + / very thing that may not be wired yet. + if[not initialised[]; + '"di.servers: ",string[ctx],": init must be called before any other function"]; + }; + +iscallable:{[x] + / internal - is x a genuinely callable value? 100 112h spans every callable form (lambda, + / primitive, operator, iterator, projection, composition), but 101h - the generic null :: - sits + / INSIDE that range while being callable in no useful sense, so the bare range check is not enough. + / that matters because :: is exactly what a dict hands back for a MISSING key when its value side + / is plain functions, which is what a real dep dict looks like (`deletejobs _ di.timer gives 101h, + / not the 99h a table-valued dep gives). admitting it lets an absent or explicitly-null dep reach + / teardown and no-op in silence: @[::;id;handler] simply returns id, deleting nothing, while + / teardown logs success. measured end to end, hence the extra exclusion + t:type x; + :(t within 100 112h) and 101h<>t; + }; + +raiseerror:{[ctx;msg] + / internal - log an error under ctx via the injected logger, then signal it, so a failure is + / observable in the log as well as thrown. used for all post-init domain errors (init's own + / dependency validation signals with a plain ' - the logger is not wired yet). + .z.m.logerr[ctx;msg]; + '"di.servers: ",string[ctx],": ",msg; + }; + +init:{[deps] + / wire the injected deps (log/timer/handlers - all required, no fallback) and this process's + / config (proctype/procname identity, connections, processcsv), and install the one-time side + / effects (a .z.pc cleanup observer via handlers + a 10s serversretry job via timer). config + / arrives in the SAME deps dict (the one-arg init convention - di.torq merges the config slice + / into it). idempotent (see `registered). does NOT open connections - that is startup's job. + if[99h<>type deps; + '"di.servers: deps must be a dict of injectables + config"]; + if[not all `log`timer`handlers in key deps; + '"di.servers: log, timer and handlers dependencies are required (see di.log, di.timer, di.handlers)"]; + if[99h<>type deps`log; + '"di.servers: log value must be a dict; pass `info`warn`error functions"]; + if[not all (`info`warn`error) in key deps`log; + '"di.servers: log dict must have `info`warn`error keys; got: ",(", " sv string key deps`log)]; + if[99h<>type deps`timer; + '"di.servers: timer value must be a dict (see di.timer)"]; + if[not `addjob in key deps`timer; + '"di.servers: timer dict must expose `addjob (see di.timer)"]; + if[99h<>type deps[`timer]`addjob; + '"di.servers: timer`addjob must be a variant dict (see di.timer addjob.custom/default/simple)"]; + if[not `custom in key deps[`timer]`addjob; + '"di.servers: timer`addjob must expose the `custom variant [id;func;params;period;mode;opts]"]; + / deletejobs is checked here for the same reason addjob is, and the consequence of NOT checking it + / is worse than a late error - it is a SILENT one. the timer dep's value side is dict-typed, so a + / MISSING key returns a null-shaped DICT rather than erroring. teardown's + / @[.z.m.timer[`deletejobs];ids;handler] then stops being protected-apply at all: @[x;y;z] is + / "try x[y], catch with z" only when x is a FUNCTION, and here x is a dict, so q reads the whole + / expression as three-argument AMEND. it upserts the job id into that throwaway dict using the + / error handler as the new value, discards the result (nothing in teardown captures it) and + / carries on. nothing throws, nothing warns, serversretry is never deleted, and teardown still + / logs success - a false positive. see di.rdb, which closes the identical trap + if[not `deletejobs in key deps`timer; + '"di.servers: timer dict must expose `deletejobs - teardown needs it, and a timer dep without ", + "it fails SILENTLY at teardown rather than loudly here; see di.timer"]; + / presence is NOT enough: a non-callable deletejobs reaches teardown's @[...] and lands in exactly + / the same amend interpretation as a missing key, so it too returns quietly having deleted + / nothing. iscallable, not a bare `within 100 112h`: the range admits 101h (::), which is both a + / non-callable and the exact value a function-valued dep dict returns for a missing key - see there + if[not iscallable deps[`timer]`deletejobs; + '"di.servers: timer`deletejobs must be a function [ids]; a non-callable or null value fails ", + "silently at teardown - see di.timer"]; + if[99h<>type deps`handlers; + '"di.servers: handlers value must be a dict (see di.handlers)"]; + / register is what init calls and remove is what teardown calls; neither was validated before, so + / a handlers dep missing either failed late and obscurely at the call site instead of here, naming + / neither the module nor the missing key. same presence-then-callable pair as the timer checks + if[not all `register`remove in key deps`handlers; + '"di.servers: handlers dict must have `register`remove keys (init registers, teardown removes); ", + "got: ",(", " sv string key deps`handlers)]; + if[not all iscallable each deps[`handlers]`register`remove; + '"di.servers: handlers`register and handlers`remove must both be functions, neither null ", + "- see di.handlers"]; + if[not all `proctype`procname in key deps; + '"di.servers: proctype and procname (self-identity) are required in deps"]; + if[not all -11h=type each deps`proctype`procname; + '"di.servers: proctype and procname must be symbols"]; + .z.m.loginfo:deps[`log]`info; + .z.m.logwarn:deps[`log]`warn; + .z.m.logerr:deps[`log]`error; + .z.m.timer:deps`timer; + .z.m.handlers:deps`handlers; + .z.m.self:`proctype`procname!deps`proctype`procname; + .z.m.connections:$[`connections in key deps;deps`connections;`symbol$()]; + .z.m.processcsv:$[`processcsv in key deps;deps`processcsv;""]; + if[not .z.m.registered; + / .z.pc is a SIMPLE (observer) event in di.handlers - side-effect only, fan-out. registered via + / the injected handlers dep with di.handlers' register[event;phase;nm;pri;func] contract; phase + / is ` (null) for a simple event, pri 0. the callback marks a closed handle's row disconnected. + / (param `wh`, not `w`, so it does not shadow the SERVERS column w.) + pcfunc:{[wh] .z.m.SERVERS:update endp:.z.p,w:0Ni from .z.m.SERVERS where w=wh; }; + (.z.m.handlers[`register])[`.z.pc;`;`servers;0j;pcfunc]; + / di.timer's addjob is a VARIANT DICT; take `custom - the fully-configurable 6-arg form + / [id;func;params;period;mode;opts]. mode-1h period is in SECONDS, so 10 = a 10s retry (a bare + / 10000 would be ~2.8h - the latent typo that made dead-handle recovery never fire in early POCs). + / retry is passed BY VALUE (a lambda, not a symbol) so di.timer stores and runs it directly; its + / compile-time .z.m rewrite means it still updates di.servers' SERVERS when the timer fires it. + (.z.m.timer[`addjob][`custom])[`serversretry;retry;();10;1;()!()]; + .z.m.registered:1b; + ]; + .z.m.loginfo[`init;"di.servers initialised"]; + }; + +teardown:{[] + / release both process-global registrations init installed - the .z.pc observer and the + / serversretry timer job - so nothing of di.servers is left bound process-wide. paired with init's + / side effects, exactly as di.subscriptions.teardown is paired with its .z.pc registration and + / di.rdb.teardown with its root entry points and timer jobs. + / module state is deliberately LEFT INTACT - SERVERS above all - so a shutdown path can still + / inspect what was connected and to whom; only the process-global bindings are withdrawn. the same + / convention every other teardown in this project follows. + / IDEMPOTENT: a second call must not die on what the first already removed. both release calls are + / no-ops on an already-removed registration - di.timer.deletejobs is a delete-where over its jobs + / table (an id that is not there matches nothing) and di.handlers' removesimple early-returns with + / an info log when the event or the name is not registered. verified in both modules, not assumed. + requireinit[`teardown]; + / called directly rather than through @[...]: init guarantees `remove is present and callable, so + / a throw here is a genuine handlers failure and should surface rather than be swallowed + .z.m.handlers[`remove][`.z.pc;`;`servers]; + / NB this @[...] IS protected-apply only because init guarantees `deletejobs is present and is a + / function. were the key missing, x would be the null-shaped dict a dict-valued dep returns for an + / absent key and q would read the line as three-argument amend instead, deleting nothing and + / logging nothing while teardown reported success. the init check is what keeps this line honest + @[.z.m.timer[`deletejobs];`serversretry; + {[e] .z.m.logwarn[`teardown;"could not delete the serversretry timer job: ",e]}]; + / reset the SIDE-EFFECT guard only, so a later init registers both again. initialised[] probes + / .z.m.loginfo, not this, so everything else stays callable after a teardown - see initialised[] + .z.m.registered:0b; + .z.m.loginfo[`teardown;"di.servers .z.pc registration and serversretry job removed"]; + }; + +formathp:{[host;port] + / internal - build the tcp connection-handle symbol from a process.csv row. v1 is tcp only; a + / future SOCKETTYPE config would reintroduce tcps/unix handling (and a type arg) when there is a + / real requirement and a test - we do not ship unexercised branches. + lower `$":",(string host),":",string port + }; + +opencon:{[hpup] + / open a connection, logging (not erroring) on failure - a downed peer isn't necessarily an + / error at connect time; retry keeps trying. NOTE the timeout form is hopen[(handle;timeoutms)] + / (a single 2-item list), not the dyadic hopen[handle;timeoutms], which throws 'rank. + r:@[{(hopen (x;.z.m.HOPENTIMEOUT);"")};hpup;{(0Ni;x)}]; + if[null first r;.z.m.logwarn[`servers;"failed to open connection to ",(string hpup),": ",last r]]; + first r + }; + +readprocesscsv:{[path] + / internal - read the static process.csv phone book. the PATH comes from config`processcsv (di.torq + / resolves it; di.servers reads no env). v1 is a STRICT 4-column host,port,proctype,procname layout: + / validate the header up front and FAIL LOUD, because ("SISS";",") is positional and would otherwise + / silently misread a reordered or wider file (e.g. a real 13-column TorQ process.csv) into garbage. + fsym:`$":",path; + if[0=count key fsym;raiseerror[`readprocesscsv;"process.csv not found at ",path]]; + lines:read0 fsym; + if[0=count lines;raiseerror[`readprocesscsv;"process.csv is empty at ",path]]; + if[not `host`port`proctype`procname~`$trim each "," vs first lines; + raiseerror[`readprocesscsv;"process.csv header must be exactly host,port,proctype,procname (v1 4-column phone book); got: ",first lines]]; + ("SISS";enlist",") 0: lines + }; + +startup:{[] + / open connections to every process.csv row whose proctype is in the configured connections list, + / excluding this process's own row. reads the config stored at init. a failed connection is logged + / (not raised) and left as w:0Ni for retry. a no-op if no connections are configured. + / normalise connections to symbols to match process.csv's `proctype column (always a symbol via + / the "S" spec): a .q settings file gives symbols already (`$ throws 'type on a symbol - it is + / NOT idempotent, hence the type check); a .toml one gives plain strings (TOML has no symbol). + requireinit[`startup]; + conns:.z.m.connections; + conns:$[11h=abs type conns;conns;`$conns]; + if[0=count conns;.z.m.loginfo[`servers;"no configured connections to make"];:()]; + if[0=count .z.m.processcsv;raiseerror[`startup;"processcsv (path to process.csv) is required in config to open connections"]]; + procs:readprocesscsv[.z.m.processcsv]; + pt:.z.m.self`proctype; + pn:.z.m.self`procname; + procs:update isme:(proctype=pt)&procname=pn from procs; + procs:select from procs where not isme; + procs:select from procs where proctype in conns; + / SELF-CONNECTION GUARD. the exclusion above is an exact (proctype;procname) match against the + / identity from config. if process.csv disagrees - a drifted proctype for this procname - the self + / row is not recognised and this process dials ITSELF. procname is process.csv's unique key (the + / idempotency filter below relies on that too), so a surviving row carrying our procname IS us: + / drop it and say why, rather than opening a self-connection nobody would think to look for. + / checked AFTER the connections filter: a row that could never be connected to was never at risk + if[count mismatched:select from procs where procname=pn; + .z.m.logwarn[`startup;"process.csv lists procname ",(string pn)," as proctype ", + (string first mismatched`proctype),", but this process is configured as proctype ",(string pt), + " - identity drift, skipping that row rather than connecting to myself"]; + procs:select from procs where not procname=pn]; + / idempotent: skip any proc already tracked in SERVERS. a repeat startup (or a process.csv that has + / grown since) then adds only NEW rows - never a duplicate row or a leaked second handle to a proc + / already connected. reconnecting a dropped peer is retry's job, not startup's. + procs:select from procs where not procname in exec procname from .z.m.SERVERS; + if[0=count procs;.z.m.loginfo[`servers;"no new process.csv rows to connect"];:()]; + {[row] + hpup:formathp[row`host;row`port]; + w:opencon[hpup]; + if[not null w;.z.m.loginfo[`servers;"connected to ",(string row`proctype),"/",(string row`procname)," at ",string hpup]]; + / catenate+reassign, NOT `tablename insert - a symbol-based insert into `.z.m.SERVERS` misses + / the compile-time module-local rewrite a source-level .z.m.SERVERS gets, silently targeting the + / wrong (literal) table. + newrow:([]procname:enlist row`procname;proctype:enlist row`proctype;hpup:enlist hpup;w:enlist w;hits:enlist 0i;startp:enlist $[null w;0Np;.z.p];lastp:enlist .z.p;endp:enlist 0Np); + .z.m.SERVERS:.z.m.SERVERS,newrow; + } each 0!procs; + }; + +retryrows:{[rows] + / internal - reattempt opencon for the given SERVERS row indices, updating w/lastp (and startp on + / a successful reconnect). + hs:opencon each exec hpup from .z.m.SERVERS where i in rows; + .z.m.SERVERS:update w:hs,lastp:.z.p from .z.m.SERVERS where i in rows; + .z.m.SERVERS:update startp:.z.p from .z.m.SERVERS where i in rows, not null w; + }; + +cleanup:{[] + / internal - sweep any row whose handle has vanished from key .z.W (a peer that died WITHOUT a + / clean .z.pc on this side) and mark it disconnected, so retry will reopen it. the .z.pc observer + / already catches clean closes; this catches the ungraceful ones. + dead:exec w from .z.m.SERVERS where not null w, not w in key .z.W; + if[count dead;.z.m.SERVERS:update endp:.z.p,w:0Ni from .z.m.SERVERS where w in dead]; + }; + +retry:{[] + / internal - the scheduled `serversretry job (driven by the injected timer; passed by value at + / init, so it needs no export). first sweep ungracefully-vanished handles (cleanup), then reopen + / every dead (null) handle - so both clean and unclean drops are recovered on the retry cycle. + cleanup[]; + rows:exec i from .z.m.SERVERS where null w; + if[count rows;retryrows[rows]]; + }; + +getservers:{[pt] + / every live (non-null handle) SERVERS row for a proctype. ` matches EVERY proctype and a list + / matches any of them - the contract legacy TorQ's .servers.getservers (trackservers.q:75) and + / di.serverselect.getservers both implement, and which any consumer written against either expects. + / matching with `in` rather than `=` is what makes both shapes work; a bare symbol still behaves + / exactly as before, so every existing caller is unaffected + requireinit[`getservers]; + if[not 11h=abs type pt;raiseerror[`getservers;"proctype must be a symbol or symbol list"]]; + $[`~pt; + select from .z.m.SERVERS where not null w; + select from .z.m.SERVERS where proctype in pt, not null w] + }; + +selector:{[tab;selection] + / internal - pick one row from a live-server table by algorithm. + $[selection=`roundrobin;first `lastp xasc tab; + selection=`any; rand tab; + selection=`last; last `lastp xasc tab; + raiseerror[`selector;"unknown selection type ",string selection]] + }; + +updatestats:{[wh] + / internal - bump hits/lastp on the row whose handle was just handed out. + .z.m.SERVERS:update lastp:.z.p,hits:1+hits from .z.m.SERVERS where w=wh + }; + +gethandlebytype:{[pt;selection] + / get a single live handle for a proctype via a selection algorithm (`any`roundrobin`last), or + / 0Ni if none is connected. bumps usage stats on the chosen row. + requireinit[`gethandlebytype]; + if[not -11h=type pt;raiseerror[`gethandlebytype;"proctype must be a symbol"]]; + if[not -11h=type selection;raiseerror[`gethandlebytype;"selection must be a symbol (`any`roundrobin`last)"]]; + r:getservers[pt]; + if[0=count r;:0Ni]; + wh:(selector[r;selection])`w; + updatestats[wh]; + wh + }; + +signalfound:{[pt] + / internal - log and return 1b once a connection to pt exists. + .z.m.loginfo[`servers;"connected to ",string pt]; + 1b + }; + +waitfortype:{[pt;timeoutms;pollms] + / block until at least one LIVE connection to pt exists, or timeoutms elapses. the DI-scoped + / analogue of legacy TorQ's startupdepcycles - "fail fast, but wait for a hard dependency to come + / up". startup must have run first (so a pt row exists to reattempt). polls retry between tries, + / sleeping pollms. returns 1b once connected, 0b on timeout - the CALLER decides if that is fatal. + / NOTE the blocking system"sleep" is fine at startup (single-threaded; the injected timer's .z.ts + / just doesn't fire during the sleep). + / a zero or already-expired timeout performs NO active retry - see servers.md, waitfortype[pt;0;p] + requireinit[`waitfortype]; + if[not -11h=type pt;raiseerror[`waitfortype;"proctype must be a symbol"]]; + if[not (abs type timeoutms) within 5 7h;raiseerror[`waitfortype;"timeoutms must be an integer (ms)"]]; + if[not (abs type pollms) within 5 7h;raiseerror[`waitfortype;"pollms must be an integer (ms)"]]; + deadline:.z.p+`timespan$1000000*`long$timeoutms; + .z.m.loginfo[`servers;"waiting up to ",(string timeoutms),"ms for a ",(string pt)," connection"]; + while[(0=count getservers pt) and .z.p/di/servers/test.q: No such file, which" +comment,,,,,,,"reads like a missing fixture rather than a path bug. .Q.m.mp is what k4unit itself uses to find" +comment,,,,,,,this very test.csv +before,0,0,q,".t.MODDIR:.Q.m.mp`di.servers",1,1,this module's own directory - independent of where q was started +before,0,0,q,"system ""l "", .t.MODDIR,""/test.q""",1,1,load recording mocks + real-peer fixture helpers +before,0,0,q,setupfixture[],1,1,write a temp process.csv fixture (self/otherproc/deadproc) and pick free ports +before,0,0,q,spawnpeer[],1,1,launch a genuinely separate q peer to act as otherproc +comment,,,,,,,"requireinit - every exported function except getapimeta refuses to run before init. these rows" +comment,,,,,,,"MUST stay above EVERY init attempt below, including the failing ones: an init validation gap" +comment,,,,,,,"would let one of those rows succeed and wire the logger, and these rows would then fail for that" +comment,,,,,,,"reason instead of a missing guard (measured). they assert the NAMED message rather than just" +comment,,,,,,,"that something threw: a bare fail row would pass on any unrelated throw, and a row checking only" +comment,,,,,,,"""returned 0/empty"" would still pass under the old silent behaviour and prove nothing" +run,0,0,q,"estartup:@[{svc[`startup][]};::;{[e] e}]",1,1,call startup before init and capture the error text +true,0,0,q,"0 +/ cleanup, tested in isolation from the auto .z.pc hook (which di.handlers would install for real). +handlercalls:([]event:`symbol$();name:`symbol$()); +/ remove DELETES the recorded row (it used to be a no-op), so teardown's release is observable the +/ same way register's is - a no-op remove would let a broken teardown pass every assertion. +mockhandlers:`register`remove`list!( + {[ev;ph;nm;pri;fn]`handlercalls upsert(ev;nm)}; + {[ev;ph;nm] handlercalls::delete from handlercalls where event=ev,name=nm;}; + {[ev]}); + +warnlogged:{[s] any (exec msg from logrows where lvl=`warn) like "*",s,"*"}; +/ fire a scheduled job's stored func exactly as the real di.timer's cycle would (retry/cleanup are +/ INTERNAL - not exported - so this is the only handle on them: the func di.servers gave the timer). +firejob:{[jid] (first exec func from rtmr.getalljobs[] where id=jid)[]}; + +/ --- real peer process fixture --- +FIXDIR:"/tmp/diserverstest"; +isfree:{[p] not @[{hclose hopen x;1b};(`$":localhost:",string p;100);0b]}; +pickport:{[start] first (start+til 500) where isfree each start+til 500}; +PEERPORT:0N; DEADPORT:0N; PEERPID:0N; + +waitlisten:{[port;timeoutms] + deadline:.z.p+`timespan$1000000*timeoutms; + while[(.z.p/dev/null 2>&1 &"; + if[not waitlisten[PEERPORT;3000];'"test: peer failed to listen on ",string PEERPORT]; + h:hopen (`$":localhost:",string PEERPORT;2000); + PEERPID::h ".z.i"; + hclose h;}; + +killpeer:{[] if[not null PEERPID;@[system;"kill ",string PEERPID;{}]]; PEERPID::0N; system "sleep 0.3";}; + +setupfixture:{[] + / pick two free ports (peer + a never-listening dead one), then write a header'd process.csv + / phone book with self, the peer (otherproc), and a dead proctype. + PEERPORT::pickport 20000+`int$.z.i mod 20000; + DEADPORT::pickport PEERPORT+1; + system "mkdir -p ",FIXDIR; + (`$":",FIXDIR,"/process.csv") 0: ( + "host,port,proctype,procname"; + "localhost,",string[PEERPORT-2],",selfproc,selfinst"; + "localhost,",string[PEERPORT],",otherproc,otherinst"; + "localhost,",string[DEADPORT],",deadproc,deadinst"); + }; + +teardownfixture:{[] killpeer[]; system "rm -rf ",FIXDIR;}; + +/ write a process.csv whose header is REORDERED vs the assumed host,port,proctype,procname (the exact +/ shape that used to silently misparse) and return its path - used to prove readprocesscsv fails loud. +writebadcsv:{[] (`$":",p:FIXDIR,"/bad.csv") 0: ("port,host,proctype,procname"; "5010,localhost,rdb,rdb1"); p}; + +/ write a process.csv listing THIS process's procname (selfinst) under a DIFFERENT proctype than its +/ configured identity (selfproc), and return its path. that is the identity-drift shape the exact +/ (proctype;procname) self-exclusion cannot recognise - without the guard, startup dials its own row. +writedriftcsv:{[] + (`$":",p:FIXDIR,"/drift.csv") 0: ( + "host,port,proctype,procname"; + "localhost,",string[PEERPORT-2],",rdb,selfinst"; + "localhost,",string[PEERPORT],",otherproc,otherinst"); + p}; + +/ build the deps dict di.torq would assemble: injectables + this process's config slice. +svrdeps:{[conns] `log`timer`handlers`proctype`procname`connections`processcsv!(mocklog;rtmr;mockhandlers;`selfproc;`selfinst;conns;FIXDIR,"/process.csv")};