From 95fda144584006ca87992aeca10d5c9f854c4b8e Mon Sep 17 00:00:00 2001 From: alowrydi Date: Fri, 7 Aug 2026 16:48:17 +0100 Subject: [PATCH 1/5] di.heartbeat initial refactor extraction from TorQ heartbeat.q. Builds on Olly's PR initial implementation, integration layer rewritten for current di.* contracts and conventions. --- di/heartbeat/VERSION | 1 + di/heartbeat/deps.q | 4 + di/heartbeat/heartbeat.md | 369 ++++++++++++++++ di/heartbeat/heartbeat.q | 711 ++++++++++++++++++++++++++++++ di/heartbeat/init.q | 15 + di/heartbeat/test.csv | 461 +++++++++++++++++++ di/heartbeat/test.q | 78 ++++ di/heartbeat/test_integration.csv | 65 +++ 8 files changed, 1704 insertions(+) create mode 100644 di/heartbeat/VERSION create mode 100644 di/heartbeat/deps.q create mode 100644 di/heartbeat/heartbeat.md create mode 100644 di/heartbeat/heartbeat.q create mode 100644 di/heartbeat/init.q create mode 100644 di/heartbeat/test.csv create mode 100644 di/heartbeat/test.q create mode 100644 di/heartbeat/test_integration.csv diff --git a/di/heartbeat/VERSION b/di/heartbeat/VERSION new file mode 100644 index 00000000..6c6aa7cb --- /dev/null +++ b/di/heartbeat/VERSION @@ -0,0 +1 @@ +0.1.0 \ No newline at end of file diff --git a/di/heartbeat/deps.q b/di/heartbeat/deps.q new file mode 100644 index 00000000..42185407 --- /dev/null +++ b/di/heartbeat/deps.q @@ -0,0 +1,4 @@ +/ hard module dependencies and their minimum versions, validated by di.depcheck +/ di.heartbeat has no hard dependencies - all runtime dependencies (log, timer, +/ handlers, pubsub, servers) are injected via init as dictionaries of functions +deps:(`$())!(); diff --git a/di/heartbeat/heartbeat.md b/di/heartbeat/heartbeat.md new file mode 100644 index 00000000..3042f65b --- /dev/null +++ b/di/heartbeat/heartbeat.md @@ -0,0 +1,369 @@ +# di.heartbeat + +Periodic liveness signalling over pub/sub, and monitoring of other processes' beats. + +A process publishes a heartbeat row on a timer so downstream monitors can detect that it has stalled +or blocked, even when the underlying TCP connection is still perfectly valid. The same module carries +the monitor role: subscribing to other processes' heartbeats, tracking the last beat seen per process, +and raising warning and error transitions when they stop arriving. + +Consolidates TorQ's `code/common/heartbeat.q` (the `.hb` namespace), covering both roles. + +## Features + +- **Publisher** (`enabled`, default on) - publishes one row per `publishinterval` over `di.pubsub`. +- **Monitor** (`subenabled`, default off) - subscribes to configured process types, stores incoming + beats, and flags `warning` then `error` as processes go quiet. +- **Pluggable transitions** - `onwarning` and `onerror` callbacks receive the affected rows. No + dashboard dependency is baked in. +- **Local-only mode** (`publishroot:0b`) - track own heartbeats without touching root namespace at all. +- **Own clock** - `setcp` overrides the module's time source for testing and simulation, independently + of `di.timer`'s clock. + +## Dependencies + +All injected via `init`; there are no hard module dependencies (`deps.q` is empty). + +| Key | Required | Must expose | Notes | +|---|---|---|---| +| `log` | always | `info`, `warn`, `error` | binary `{[ctx;msg]}` - `di.log`'s `logdict` fits directly | +| `timer` | always | `addjob`, `deletejobs` | `addjob` must be the variant dict exposing `custom` | +| `pubsub` | always | `publish`, `subscribe` | `di.pubsub` | +| `servers` | when `subenabled` | `getservers` | `di.servers`; returns a **table**, handles are its `w` column | +| `handlers` | when `subenabled` | `register`, `remove` | `di.handlers`; 5-arg `register[event;phase;nm;pri;func]` | + +Nothing is silently defaulted. A missing or malformed dependency throws from `init` with a message +naming the module that supplies it. + +## Initialisation + +`init` takes exactly **one** dict, carrying dependencies and config side by side. + +```q +logging:use`di.log +timer:use`di.timer +ps:use`di.pubsub +hb:use`di.heartbeat + +timer.init[] +hb.init[logging.logdict,`timer`pubsub`proctype`procname!(timerdep;psdep;`rdb;`rdb1)] +ps.init[] / MUST run after hb.init - see Ordering below +``` + +`proctype` and `procname` are **required**. They are published as the heartbeat's `sym` and `procname` +and have no sensible default. (Legacy took them from `.proc.*`; a module cannot.) + +> **Dict-building gotcha.** `!` and `,` are both right-to-left with no precedence, so +> ``` `a`b!(x;y),moredeps ``` parses as ``` `a`b!((x;y),moredeps) ``` - the comma binds to the *value +> list*, not the dict, and you get a `'type`. Build the base dict first, or keep an inline `k!v` last. + +### Ordering: `di.pubsub.init[]` must run *after* `di.heartbeat.init[]` + +`di.pubsub` decides which tables it will serve by scanning **root** `tables[]` when its own `init` +runs, and resolves each name with `value`. `di.heartbeat.init` is what puts the `heartbeat` schema +table at root. Reverse the order and `di.pubsub` never sees the table: `subscribe` is refused and +`publish` silently discards every row. Nothing errors. The integration suite asserts this path +end-to-end for exactly that reason. + +> **Known gap - the wider discovery surface is LIVE, not fixed.** Because nothing currently calls +> `di.pubsub.setsubtables`, `di.pubsub` still auto-discovers *every* root table present when its +> `init` runs - not just `heartbeat`. `di.heartbeat` deliberately does **not** call `setsubtables` +> itself: that function replaces the list wholesale and `di.pubsub` exposes no getter, so a consumer +> cannot add itself without silently removing everyone else. Closing this belongs to **`di.torq`**, +> the only component that knows the full table set - the same reasoning that puts `getapimeta` +> collection there. Until `di.torq` does it, this remains a known, temporary gap. + +## Configuration + +Passed in the same dict as the dependencies. Every key is optional. + +| Key | Default | Source of default | Description | +|---|---|---|---| +| `enabled` | `1b` | both agree | publish and check heartbeats | +| `subenabled` | `0b` | both agree | monitor other processes | +| `debug` | `1b` | both agree | log warning/error transitions | +| `publishroot` | `1b` | new | publish at root; see below | +| `publishinterval` | `0D00:00:30` | both agree | how often to publish | +| `checkinterval` | `0D00:00:10` | both agree | how often to check | +| `warningtolerance` | `2f` | **shipped** | warning after `tolerance * publishinterval` | +| `errortolerance` | `3f` | **shipped** | error after `tolerance * publishinterval` | +| `maxage` | `0D24:00:00` | new | forget a process silent this long; `0Wn` to keep forever | +| `connections` | `` `ALL `` | **shipped** | process types to monitor; `` `ALL `` means every one | +| `onwarning` | no-op | new | unary callback given the rows entering warning | +| `onerror` | no-op | new | unary callback given the rows entering error | +| `pid` / `host` / `port` | `.z.i` / `.z.h` / `system"p"` | legacy | captured once at load, as legacy did | + +**On "shipped" vs in-file defaults.** Legacy carried two different sets: the `@[value;...]` fallbacks +inside `heartbeat.q` (`1.5f`, `2f`, `()`) and the values actually shipped in +`config/settings/default.q` (`2f`, `3f`, `` `ALL ``). This module takes the **shipped** values - they +reflect what really ran. The in-file `connections:()` could not be used at all: with it, `` `ALL in () `` +is false and every subscription lookup returns nothing, so the whole monitor path silently does nothing. + +### Config that `init` refuses + +Every one of these would otherwise produce a module that runs happily and does the wrong thing +silently, so each is rejected up front rather than left to be discovered in production: + +| Rejected | Why | +|---|---| +| `errortolerance <= warningtolerance` | a process reaches error before warning, so the warning transition can never fire | +| tolerance `<= 0` (including `0n`) | the grace period is zero or negative, so every process is flagged immediately and permanently | +| interval `< 1 second` | `di.timer` schedules in whole seconds; `tosecs` **rounds**, so anything under 500ms becomes a period of `0` and the job would run on every timer cycle | +| null `proctype` or `procname` | they key the monitor's store, so nulls collapse every publisher into a single row | + +Setting `subenabled:1b` with an **empty** `connections` list is legal but warns: it is the same shape +as legacy's in-file `connections:()` default, where the entire monitor path silently did nothing. Use +`` `ALL ``, or name the process types to watch. + +Setting `subenabled:1b` with an **empty** `connections` list is legal but warns: it is the same shape +as legacy's in-file `connections:()` default, where the entire monitor path silently did nothing. Use +`` `ALL ``, or name the process types to watch. + +## Robustness + +Three properties worth knowing about, because each one exists to stop a *local* failure becoming a +*permanent* one. + +**Two silent-projection traps are closed at `init`.** A callback of the wrong arity is *not* an error +in q — applying a two-argument function to one argument yields a projection, so it would never run and +never log. `init` therefore checks callback **arity**, not just type. Likewise `setcp` calls the clock +function once and checks it returns a timestamp: a clock returning anything else does not throw inside +`checkheartbeat`, it just makes every staleness comparison evaluate false, silently switching +monitoring off. (Consequence: the clock's state must exist when `setcp` is called.) + +**A broken callback cannot stop heartbeating.** `onwarning`, `onerror` and the `pubsub.publish` call +all run isolated: a throw is logged at error and execution continues. This matters more than it looks. +Both timer jobs are scheduled through `di.timer`, whose `addjob.opts` defaults `disableonfail:1b` — so +an unprotected throw would not merely skip one beat, it would **permanently disable** the job. A single +bug in a client's `onwarning`, or one transient pub/sub outage, would silently end the monitoring this +module exists to provide. State is always updated *before* a callback fires, so isolation never leaves +the store inconsistent. Same reasoning `di.handlers` applies to its `post` phase. + +**A misbehaving publisher cannot corrupt or blind the monitor.** `storeheartbeat` validates what +arrives over the wire: + +- a batch missing `sym`, `procname` or `time` is rejected with a named error (not a raw `'rank`); +- rows with a **null time** are dropped with a warning — such a row is permanently invisible to + `checkheartbeat`, since `now > time + period` is never true against a null, so the process could + never be flagged however long it stayed silent; +- **unknown columns are ignored, not rejected.** A publisher on a newer schema would otherwise have + every heartbeat refused and be reported *dead* by this monitor — a far worse failure than dropping a + column we have no use for. Version skew is warned about once per distinct column set, so it stays + visible without a log line per beat. + +**The store is bounded.** `hb` is keyed on `sym`+`procname`, so anything that churns identities — +containers with generated names, a process restarting under a new `procname` — would otherwise add a +row per incarnation that lives for the life of the monitor. `checkheartbeat` evicts processes silent +for longer than `maxage` (default one day), and logs what it dropped. + +Two properties of that policy are deliberate and worth stating, because the obvious implementations +get both wrong: + +- **Eviction is by age, never by row count.** A dead process has an *old* timestamp by definition, so + a "keep the newest N rows" cap would discard exactly the rows worth keeping. +- **A row that has never heartbeated is never evicted.** Those come from `addprocs` and represent + operator *intent* rather than an observation — silently forgetting a process you declared you were + expecting would stop it being reported missing, which is the whole point of seeding it. They are + cleared only by `removeprocs`. + +**Only a row already flagged `error` is ever evicted.** Validating `maxage > errorperiod` is *not* +sufficient on its own — it only guarantees the transition would have fired had a check run in the +window. A monitor that was paused or restarted, or one handed a heartbeat carrying an old timestamp, +gets its first check when the row is already past `maxage`; gating solely on age evicted it without +anyone ever being told the process had stopped. Gating on `error` makes "forgotten only after being +reported" structurally true rather than dependent on check cadence — the row simply survives one extra +check cycle. Set `maxage:0Wn` to disable eviction entirely. + +**Flipping a flag off on re-`init` cleans up after the previous one.** `registertimers`, +`registerhandlers` and the root-name sync all *reconcile* rather than only add. Re-initialising with +`subenabled:0b` deregisters the `.z.pc` observer; with `publishroot:0b` it removes the published root +names. Both leaks were real and both were permanent: `teardown` used to key off the *current* config, +so once a flag was off nothing could remove what the earlier init had installed. A lingering +`.heartbeat.subscribe` is the worse of the two — a remote monitor can still subscribe *successfully* +and then receive nothing for the life of the process, while believing it is watching you. + +The root-name sync runs **before** any timer is scheduled, since it is the only step that can fail on +external state — so a failure leaves the process untouched rather than half-configured. + +### `publishroot` - and what `publishroot:0b` actually means + +With `publishroot:1b` (the default) the module publishes two names at root, both removed by `teardown`: + +- `heartbeat` - the empty schema table `di.pubsub` discovers and resolves. +- `.heartbeat.subscribe` - the entry point a monitor calls to subscribe itself (see below). + +(A root table `heartbeat` and a root namespace `.heartbeat` do not collide in q — verified in both +creation orders.) + +**A pre-existing `heartbeat` table is never taken over — not even a column-identical one.** If the +name is already occupied at root by a table this module did not create, `init` **errors**. This is a +real case, not a hypothetical: `di.subscriptions` installs subscribed schemas at root, so a monitor +watching a tickerplant that carries `heartbeat` already has one. + +There is deliberately **no adoption path**, and column compatibility is deliberately *not* used as a +signal. "Identifier plus timestamp plus a few status fields" is a common table shape, so matching +columns are no evidence that a table means the same thing — adopting on that basis would silently +co-mingle liveness rows with whatever the real owner stores there, leaving only a log line to explain +it afterwards. Nothing is lost by refusing, either: the publish table is *only ever an empty schema +holder* (rows go out over pub/sub, never into it), so a stale one from an earlier load costs nothing +to remove. The error says exactly that, and names both remedies (remove it, or run `publishroot:0b`). + +Re-`init` over the module's **own** table is fine — ownership persists across re-init, and `teardown` +removes only a table this module created. + +**`publishroot:0b` means "do not publish at all, track locally" - not "publish, but privately".** +This is a real behavioural difference, not just a doc note. Without the root schema table `di.pubsub` +can never serve this topic, so a version that merely skipped the root call would build and discard a +row on every tick, forever, with no subscriber possible. Instead, with `publishroot:0b`: + +- no root names are created; +- `publishheartbeat` does **not** call `pubsub.publish` at all; +- the row is retained in module-private state, readable via `getownhb[]` (last row only, so memory + stays bounded). + +So `enabled:1b, publishroot:0b` is a valid, meaningful combination: this process keeps beating for +local introspection, and nothing it produces is visible to anyone else or written to root namespace. +Legacy had no equivalent - every heartbeat was always globally visible - so this is a deliberate +addition, not a port. + +## How a monitor subscribes + +`di.pubsub.subscribe` registers **the caller's own `.z.w`**. A monitor therefore cannot subscribe +itself to a remote publisher by calling it locally - it would only ever subscribe itself to itself. + +Legacy solved this with a synchronous IPC call (`heartbeat.q:92`), and so does this module: the monitor +calls the publisher's root ``.heartbeat.subscribe`` over the handle. During that inbound call the +publisher's `.z.w` *is* the connection back to the monitor, so the subscription lands correctly. + +```q +hb.subscribe[handle] / monitor side; sends `.heartbeat.subscribe to the publisher +``` + +A failed subscribe is logged and **not** recorded, so the next `hbsubscribe` tick retries it. + +**Handle 0 and null handles are refused.** A "remote" call on handle `0i` evaluates *locally*, so +subscribing it would register this process as a subscriber to its own heartbeats and then publish to +handle 0; a null handle is a disconnected server row. Both are dropped from `subscribe` and from the +`getservers` sweep, with a warning. Legacy guarded the same case by seeding +`subscribedhandles:0 0Ni` (`heartbeat.q:21`) — the guard is deliberate, not incidental. + +> **Security note.** `.heartbeat.subscribe` is callable by anyone holding a handle to this process, +> exactly as legacy's `.ps.subscribe` was. It takes no arguments and only subscribes the caller to the +> heartbeat table, so the exposure is small — but it is a root-published remote entry point, and a +> deployment running `di.permissions` will want it in scope of whatever gates `.z.pg`. + +## Exported functions + +| Function | Signature | Description | +|---|---|---| +| `init` | `[dict]` | wire dependencies and config; idempotent | +| `teardown` | `[]` | release timer jobs, the `.z.pc` registration and the root names | +| `version` | - | module version string, read from `VERSION` | +| `getapimeta` | `[]` | api metadata rows for `di.torq` to register with `di.api` | +| `publishheartbeat` | `[]` | publish one row and bump the counter (timer job) | +| `checkheartbeat` | `[]` | flag processes past their grace periods (timer job) | +| `storeheartbeat` | `[table]` | store incoming beats, latest per process, clearing state | +| `addprocs` | `[proctypes;procnames]` | seed expected processes so a silent one is still flagged | +| `removeprocs` | `[proctypes;procnames]` | forget the named processes entirely; the counterpart to `addprocs` | +| `subscribe` | `[handles]` | subscribe to heartbeats on the given remote handle(s) | +| `gethb` | `[]` | the store of heartbeats **received** from others | +| `getownhb` | `[]` | the last heartbeat **this** process produced (populated when `publishroot:0b`) | +| `setcp` | `[func]` | replace the module's clock, for tests and simulation | + +Every function except `init` requires `init` to have run first, and says so if it has not. + +**Input validation.** Every public function validates its arguments and routes failures through the +log before signalling, so a bad call names itself rather than surfacing as a raw `'type` / `'length` +from somewhere downstream: + +| Call | Behaviour | +|---|---| +| `storeheartbeat` on a non-table, or missing `sym`/`procname`/`time` | errors, naming the columns it got | +| `storeheartbeat` with a non-timestamp `time` column | errors, naming the type it got | +| `onwarning`/`onerror` of the wrong arity | rejected at `init` — see below | +| `setcp` with a function not returning a timestamp | rejected, the function is called once to check | +| `subscribe` with a negative handle | accepted, skipped, warned | +| `storeheartbeat` rows with a null `time` | dropped, warned (see Robustness) | +| `storeheartbeat` with unknown columns | ignored, warned once (see Robustness) | +| `addprocs` with mismatched lengths or non-symbols | errors, naming the two lengths | +| `subscribe` with a non-integer | errors | +| `subscribe` with `0i` or a null handle | accepted, skipped, warned | +| `setcp` with a non-function | errors | + +## Store schema + +`gethb[]` returns a table keyed on `sym` (the publishing process **type**) and `procname`: + +``` +sym procname | time counter pid host port warning error +``` + +`addprocs` seeds rows with null `counter`/`pid`/`port`, so a process that never beats at all still +appears and still transitions to warning and error. A real beat arriving later wins. + +**A *gradually* escalated process carries `warning:1b` and `error:1b` together.** Both statements are +literally true — it is past both thresholds — and it matches legacy, which sets each flag +independently and never clears either (`heartbeat.q:71,77`). Only a fresh heartbeat clears them, in +`storeheartbeat`. (Queried on PR #109; legacy semantics kept.) + +But a process that crosses **both** thresholds between two checks — a long `checkinterval`, or a +monitor that was paused — gets `error:1b` with `warning` still `0b`, because it was never observed in +the warning state and that transition never fired. So do not treat `warning` as implied by `error`. A +consumer rendering both columns should key off `error` first and treat `warning` as independent. + +The null `counter` is load-bearing beyond that: it is what marks a row as operator intent rather than +an observation, and so what exempts it from age-based eviction (see Robustness). Use `removeprocs` to +clear one. + +## Deliberate departures from legacy + +Three places where this module intentionally does not look like `heartbeat.q`. Each is a considered +choice, not a missed port. + +- **No `upd` hook.** Legacy composed `upd` at load time + (`upd:{[f;t;x] ...}@[value;`upd;...]`), which is load-order dependent and was silently overridden + by `monitor.q` in the one process that most needed it. Instead, `storeheartbeat` is exported and the + consuming process calls it from its own `upd`: + ```q + upd:{[t;x] if[t=`heartbeat;hb.storeheartbeat x]; ... } + ``` +- **No `.servers.connectcustom` wrapper.** Legacy wrapped it to filter auto-connections by + `.hb.CONNECTIONS`, mutating the table handed to whatever had registered before it - a cross-feature + side effect. `di.servers` owns its own connection strategy now; this module resolves handles through + `getservers` instead. +- **No `.html.pub`.** Legacy's `processwarning`/`processerror` published straight to a dashboard. + Those are now the `onwarning`/`onerror` callbacks. `di.html` is `di.monitor`'s dependency, not this + module's. + +## Running tests + +```q +k4unit:use`di.k4unit +k4unit.moduletest`di.heartbeat +``` + +The integration suite needs a real child process and is a separate file, so `moduletest` (which is +hardcoded to `test.csv`) does not pick it up: + +```q +.m.di.0k4unit.KUltf .Q.dd[hsym`$.Q.m.mp`di.heartbeat;`test_integration.csv] +.m.di.0k4unit.KUrt[] +``` + +It spawns a genuine second q process as a publisher and exercises the paths a mock cannot reach: the +root schema table, the remote-subscribe handshake, and a published row actually crossing the wire into +the monitor's store. + +## Notes + +- `warningperiod` and `errorperiod` take a process type so a deployment can vary grace periods per + type. The default implementations ignore it, as legacy's did. +- Both timer jobs use **mode 2** (period after the previous *actual* start). A heartbeat asserts + "alive now", so missed beats must not be replayed as a catch-up storm, which mode 1 would do. + Periods are converted to whole seconds, which is what `di.timer` expects. +- `pid`, `host` and `port` are captured once at load, matching legacy. A runtime port change is not + picked up. +- Re-running `init` is safe: it clears its own timer jobs before re-registering, and deliberately + **preserves** already-received heartbeats. +- `host` shares its name with a q error message, which `qlint` flags as `VAR_Q_ERROR`. The column name + is fixed by the published row shape and legacy wire compatibility, so it is kept. diff --git a/di/heartbeat/heartbeat.q b/di/heartbeat/heartbeat.q new file mode 100644 index 00000000..3b71d3f5 --- /dev/null +++ b/di/heartbeat/heartbeat.q @@ -0,0 +1,711 @@ +/ heartbeat module for kdb-x +/ every process can publish a periodic heartbeat over pub/sub so that downstream monitors can detect +/ when a process has stopped beating - i.e. it is stalled or blocked - even when the underlying +/ connection is still valid +/ the module handles both publishing heartbeats and, on the monitoring side, storing received +/ heartbeats and raising warnings / errors when they stop +/ config and dependencies arrive in a single dictionary passed to init: config keys are optional and +/ fall back to the defaults below; log/timer/pubsub are required (servers/handlers when subenabled) +/ and init errors immediately if a required dependency is missing +/ module-local state convention: constants are bare top-level names, mutable state lives in .z.m, and +/ injected dependencies are read through .z.m at every call site + +/ ============================================================ +/ constants +/ ============================================================ + +/ table used to publish heartbeats - sym holds the publishing process type +schema:( + [] time:`timestamp$(); + sym:`symbol$(); + procname:`symbol$(); + counter:`long$(); + pid:`int$(); + host:`symbol$(); + port:`int$() + ); + +/ keyed store of the latest received heartbeat per process, with warning / error state +storeschema:update warning:0b,error:0b from `sym`procname xkey schema; + +/ the table name published over pub/sub, and the root name di.pubsub discovers it under +tablename:`heartbeat; + +/ the dependency keys of the single init dict - everything else in that dict is config +depkeys:`log`timer`pubsub`servers`handlers; + +/ config defaults. proctype and procname are deliberately absent - they are self-identity and are +/ required, not defaulted, matching di.servers. legacy captured pid/host/port once at load time and +/ so do we, so a runtime port change is not picked up +configdefaults:( + `enabled`subenabled`debug`publishroot`publishinterval`checkinterval`warningtolerance`errortolerance, + `maxage`pid`host`port`connections`onwarning`onerror + )!( + 1b;0b;1b;1b;0D00:00:30;0D00:00:10;2f;3f; + 0D24:00:00;.z.i;.z.h;`int$system"p";`ALL;{[procs]};{[procs]} + ); + +/ timer job ids this module owns - deleted before re-registering so init is safe to call again +jobids:`hbpublish`hbcheck`hbsubscribe; + +/ ============================================================ +/ module state +/ ============================================================ + +/ current-time function - heartbeat owns its clock, separate from di.timer's; override via setcp +cp:{.z.p}; + +/ ============================================================ +/ internal helpers +/ ============================================================ + +initialised:{[] + / has init run? a direct (module-rewritten) reference detects prior setup without touching root + :@[{.z.m.enabled;1b};::;0b]; + }; + +raiseerror:{[ctx;msg] + / log an error under ctx then signal it, so a failure is observable in the log and not only as a + / throw. init's own dependency validation signals with a plain ' - the logger is not wired yet + .z.m.logerr[ctx;msg]; + '"di.heartbeat: ",string[ctx],": ",msg; + }; + +requireinit:{[ctx] + / every exported function except init depends on init having wired the logger and the config + if[not initialised[]; + '"di.heartbeat: ",string[ctx],": init must be called before any other function"]; + }; + +tosecs:{[span] + / di.timer mode-2 periods are in whole seconds; legacy expressed these intervals as timespans + :`int$span%0D00:00:01; + }; + +warningperiod:{[processtype] + / grace period before a process is flagged warning - takes the process type so a deployment can + / vary it per type, as legacy documented + :`timespan$warningtolerance*publishinterval; + }; + +errorperiod:{[processtype] + / grace period before a process is flagged error + :`timespan$errortolerance*publishinterval; + }; + +resolveconnections:{[] + / `ALL is the shipped default and means every connected process type. legacy converts the sentinel + / to a null symbol, which .servers.getservers treats as match-all + :$[`ALL in (),connections;`;connections]; + }; + +arity:{[f] + / parameter count of a lambda - value f returns (bytecode;params;...) so index 1 is the param list + :count value[f]1; + }; + +safecall:{[ctx;nm;f;arg] + / run an injected function in isolation and log any failure instead of letting it propagate. + / both call sites here are reached from a di.timer job, and di.timer disables a job that throws + / (addjob.opts sets disableonfail:1b), so an unprotected failure would PERMANENTLY stop heartbeating + / or heartbeat checking - the very thing this module exists to provide. one broken callback, or one + / transient publish failure, must not take the mechanism down with it. this is the same reasoning + / di.handlers applies to its post phase: a watcher must not be able to change the outcome + @[f;arg;{[ctx;nm;e] .z.m.logerr[ctx;(string nm)," failed: ",e]}[ctx;nm]]; + }; + +validhandles:{[handles] + / drop nulls (a dead server row) and handle 0. a "remote" call on handle 0 evaluates LOCALLY, so + / subscribing it registers this process as a subscriber to its own heartbeats and then publishes + / to handle 0. legacy guarded exactly this by seeding subscribedhandles with 0 0Ni (heartbeat.q:21) + / a negative handle is an ASYNC handle in q: the subscribe call would return immediately without + / confirming anything, so it would be recorded as subscribed on no evidence - and closeconnection + / matches against .z.w, which is always positive, so the row could never be cleaned up either + h:(),handles; + :h where not (null h) or 0i>=h; + }; + +/ ============================================================ +/ init - dependency validation and config +/ ============================================================ + +validatedeps:{[deps] + / log, timer and pubsub are always required; servers and handlers only when this process monitors + / others. nested if guards rather than and - and evaluates both sides eagerly, so key would be + / reached on a non-dict. no dependency is ever silently defaulted + if[99h<>type deps; + '"di.heartbeat: deps must be a dict of injectables + config - see di.log, di.timer, di.pubsub"]; + if[not all `log`timer`pubsub in key deps; + '"di.heartbeat: log, timer and pubsub dependencies are required (see di.log, di.timer, di.pubsub); got: ", + (", " sv string key deps)]; + if[99h<>type deps`log; + '"di.heartbeat: log value must be a dict; pass `info`warn`error functions - see di.log"]; + if[not all `info`warn`error in key deps`log; + '"di.heartbeat: log dict must have `info`warn`error keys; got: ",(", " sv string key deps`log)]; + if[99h<>type deps`timer; + '"di.heartbeat: timer value must be a dict (see di.timer)"]; + if[not all `addjob`deletejobs in key deps`timer; + '"di.heartbeat: timer dict must expose `addjob and `deletejobs (see di.timer)"]; + if[99h<>type deps[`timer]`addjob; + '"di.heartbeat: timer`addjob must be a variant dict (see di.timer addjob.custom/default/simple)"]; + if[not `custom in key deps[`timer]`addjob; + '"di.heartbeat: timer`addjob must expose the `custom variant [id;func;params;period;mode;opts]"]; + if[99h<>type deps`pubsub; + '"di.heartbeat: pubsub value must be a dict (see di.pubsub)"]; + if[not all `publish`subscribe in key deps`pubsub; + '"di.heartbeat: pubsub dict must expose `publish and `subscribe (see di.pubsub)"]; + if[not all `proctype`procname in key deps; + '"di.heartbeat: proctype and procname (self-identity) are required in deps - they are published ", + "as the heartbeat's sym and procname and have no sensible default"]; + if[not all -11h=type each deps`proctype`procname; + '"di.heartbeat: proctype and procname must be symbols"]; + if[any null deps`proctype`procname; + '"di.heartbeat: proctype and procname must not be null - they identify this process in every ", + "published heartbeat and key the monitor's store, so null ones collide every publisher ", + "into a single row"]; + }; + +validatemonitordeps:{[deps] + / servers and handlers are required only when subenabled - a publisher-only process needs neither. + / split out so the conditional in init stays a single statement, per the style guide + if[99h<>type deps`servers; + '"di.heartbeat: subenabled is set, so a servers dependency is required (see di.servers)"]; + if[not `getservers in key deps`servers; + '"di.heartbeat: servers dict must expose `getservers (see di.servers)"]; + if[99h<>type deps`handlers; + '"di.heartbeat: subenabled is set, so a handlers dependency is required (see di.handlers)"]; + if[not all `register`remove in key deps`handlers; + '"di.heartbeat: handlers dict must have `register`remove keys; got: ", + (", " sv string key deps`handlers)]; + }; + +resolveconfig:{[deps] + / merge the config half of the single init dict over the defaults, warning about anything + / unrecognised rather than dropping it silently. depkeys are dependencies, not config + config:(key[deps] except depkeys,`proctype`procname)#deps; + if[count unknown:(key config) except key configdefaults; + .z.m.logwarn[`init;"ignoring unrecognised config key(s): ",", " sv string unknown]]; + :configdefaults,(key[configdefaults] inter key config)#config; + }; + +validateconfig:{[cfg] + / catch config that would leave the module quietly doing nothing rather than failing loudly + if[not all -1h=type each cfg`enabled`subenabled`debug`publishroot; + raiseerror[`init;"enabled, subenabled, debug and publishroot must be booleans"]]; + if[not all -16h=type each cfg`publishinterval`checkinterval; + raiseerror[`init;"publishinterval and checkinterval must be timespans"]]; + / di.timer schedules in whole SECONDS and tosecs rounds, so anything under half a second becomes a + / period of 0 - a job that then runs on every timer cycle. reject the whole sub-second range rather + / than silently accept a schedule that cannot be represented + if[any 0D00:00:01>cfg`publishinterval`checkinterval; + raiseerror[`init;"publishinterval and checkinterval must be at least one second - di.timer ", + "schedules in whole seconds, so a shorter interval cannot be represented"]]; + if[not all -9h=type each cfg`warningtolerance`errortolerance; + raiseerror[`init;"warningtolerance and errortolerance must be floats"]]; + / a zero or negative tolerance makes the grace period zero or negative, so now>time+period is true + / the instant a process is seen and every process sits permanently in error. also catches 0n + if[any 0>=cfg`warningtolerance`errortolerance; + raiseerror[`init;"warningtolerance and errortolerance must be positive - a non-positive ", + "tolerance flags every process immediately and permanently"]]; + if[cfg[`errortolerance]<=cfg`warningtolerance; + raiseerror[`init;"errortolerance must exceed warningtolerance, else a process reaches error ", + "before warning and the warning transition never fires"]]; + if[not all 100h=type each cfg`onwarning`onerror; + raiseerror[`init;"onwarning and onerror must be unary functions taking the affected rows"]]; + / arity is checked, not just type: applying a two-argument callback to one argument yields a + / PROJECTION rather than throwing, so a wrong-arity callback is silently never called and never + / logged - the same silent-projection shape as a two-argument init + if[not all 1=arity each cfg`onwarning`onerror; + raiseerror[`init;"onwarning and onerror must take exactly one argument (the affected rows); a ", + "callback of any other arity is silently never called, it just yields a projection"]]; + if[-16h<>type cfg`maxage; + raiseerror[`init;"maxage must be a timespan (0Wn to keep every process forever)"]]; + / an evicted row must always have had its error transition first, otherwise a process could vanish + / from the store without anyone ever being told it had stopped + if[not cfg[`maxage]>`timespan$cfg[`errortolerance]*cfg`publishinterval; + raiseerror[`init;"maxage must exceed the error period (errortolerance*publishinterval), so a ", + "process always reaches its error transition before it can be evicted"]]; + / not an error - an explicit empty list is a legal way to say "monitor nothing" - but it is the + / same shape as legacy's in-file connections:() default, where the whole monitor path silently did + / nothing. say so rather than let it look configured + if[cfg[`subenabled] and 0=count (),cfg`connections; + .z.m.logwarn[`init;"subenabled is set but connections is empty - this process will monitor ", + "nothing. use `ALL, or list the process types to watch"]]; + }; + +validateroot:{[cfg] + / check the root name is available BEFORE init writes any state. claiming it is the only step that + / can fail on something outside this module, and letting installroot throw part-way through init + / would leave the module marked initialised, with config written and deps wired, but with no + / timers, no handlers and no root table - so publishheartbeat would happily run and publish into a + / topic di.pubsub cannot serve. exactly the silent failure this module exists to avoid + if[not cfg`publishroot;:()]; + if[not tablename in tables[];:()]; + / rootowned is unset until the first init has run, so read it defensively + if[@[{rootowned};::;0b];:()]; + raiseerror[`init;"a table named ",(string tablename)," already exists at root and was not ", + "created by this module - refusing to use it. if it is left over from an earlier load of ", + "di.heartbeat, remove it (the publish table is always an empty schema holder, so nothing is ", + "lost); if it belongs to another module, rename one of them or run with publishroot:0b"]; + }; + +/ ============================================================ +/ root namespace - the publish schema and the remote-subscribe entry point +/ ============================================================ + +ensureroottable:{[] + / put the publish schema at root, but NEVER take over a table this module did not create - not even + / a column-identical one. matching columns are not evidence that a table means the same thing: + / "identifier plus timestamp plus a few status fields" is a common shape, and adopting on that basis + / would silently co-mingle this module's liveness rows with whatever the real owner stores there. + / this is a reachable collision rather than a defensive hypothetical - di.subscriptions installs + / subscribed schemas at root, so a monitor watching a tickerplant that carries `heartbeat has one. + / there is deliberately no adoption path: the publish table is only ever an empty schema holder + / (rows go out over pub/sub, never into it), so a stale one from an earlier load costs nothing to + / remove, and "the columns matched" is far too weak a signal to hand over a name on + if[not tablename in tables[]; + set[tablename;schema]; + .z.m.rootowned:1b; + :()]; + / already there and we created it on an earlier init - keep ownership and leave the table alone + if[rootowned;:()]; + raiseerror[`installroot;"a table named ",(string tablename)," already exists at root and was not ", + "created by this module - refusing to use it. if it is left over from an earlier load of ", + "di.heartbeat, remove it (the publish table is always an empty schema holder, so nothing is ", + "lost); if it belongs to another module, rename one of them or run with publishroot:0b"]; + }; + +installroot:{[] + / di.pubsub discovers publishable tables by scanning ROOT tables[] when its own init runs, and + / resolves each name with value, so the schema must exist at root under the published name or + / publish is a silent no-op. legacy does exactly this at heartbeat.q:112 + ensureroottable[]; + / the remote-subscribe entry point. a monitor cannot subscribe itself by calling di.pubsub.subscribe + / locally - that function reads the CALLER's .z.w - so it makes a synchronous IPC call to this name + / instead. during that inbound call .z.w is the monitor's connection back to us, so the subscription + / lands against the right handle. this is the mechanism legacy uses at heartbeat.q:92 + / NB a root table `heartbeat and a root namespace .heartbeat do not collide - verified both orders + set[`.heartbeat.subscribe;{[] .z.m.pubsubsubscribe[tablename;`]}]; + .z.m.rootinstalled:1b; + .z.m.loginfo[`installroot;"published root names ",(string tablename)," and .heartbeat.subscribe"]; + }; + +droprootnames:{[] + / publishroot is off - remove anything an earlier init published. leaving the names behind would let + / di.pubsub keep serving the topic and let a remote monitor subscribe SUCCESSFULLY and then receive + / nothing for the life of the process, while believing it is watching us. a monitor that is silently + / watching a dead topic is worse than one that fails to subscribe at all + if[not rootinstalled;:()]; + uninstallroot[]; + .z.m.loginfo[`init;"publishroot is 0b - root names published by an earlier init have been removed"]; + }; + +uninstallroot:{[] + / remove the live names installroot created. the root table goes only if WE created it - an adopted + / one belongs to another part of the process. the empty .heartbeat namespace slot itself remains, + / since q has no way to remove one + if[rootowned;![`.;();0b;enlist tablename]]; + ![`.heartbeat;();0b;enlist `subscribe]; + / release the claim, so a later init re-evaluates whoever owns the name by then + .z.m.rootowned:0b; + .z.m.rootinstalled:0b; + }; + +/ ============================================================ +/ wiring - timers and handlers +/ ============================================================ + +registertimers:{[] + / mode 2 = period after the previous ACTUAL start. a heartbeat asserts "alive now", so missed beats + / must not be replayed as a catch-up storm, which mode 1 would do. periods are in whole seconds. + / di.timer's addjob throws on a duplicate id, so clear ours first and init stays safe to re-run + .z.m.timerdeletejobs jobids; + if[enabled; + .z.m.timeraddjob[`custom][`hbpublish;publishheartbeat;();tosecs publishinterval;2;()!()]; + .z.m.timeraddjob[`custom][`hbcheck;checkheartbeat;();tosecs checkinterval;2;()!()]]; + if[subenabled; + .z.m.timeraddjob[`custom][`hbsubscribe;hbsubscriptions;();60;2;()!()]]; + }; + +registerhandlers:{[] + / .z.pc is a SIMPLE event in di.handlers - side-effect only, return value discarded, so any number + / of registrants coexist and the phase must be ` (null). register is 5-arg [event;phase;nm;pri;func] + / drop any registration from an earlier init FIRST and unconditionally, mirroring registertimers. + / a re-init that turns subenabled off would otherwise orphan the observer permanently: teardown + / keys off whether we are registered, but before this it keyed off the CURRENT subenabled, which is + / now false - so nothing could ever remove it + if[handlerregistered; + .z.m.handlersremove[`.z.pc;`;`heartbeat]; + .z.m.handlerregistered:0b]; + if[subenabled; + .z.m.handlersregister[`.z.pc;`;`heartbeat;0j;closeconnection]; + .z.m.handlerregistered:1b]; + }; + +/ ============================================================ +/ monitor side - subscribing to other processes' heartbeats +/ ============================================================ + +subscribeone:{[h] + / ask the REMOTE publisher to subscribe us. calling di.pubsub.subscribe locally cannot work - it + / reads the caller's own .z.w and so can only ever subscribe the caller. a failed subscribe is not + / recorded, so the next tick retries it + / the error handler's first parameter is named hdl, not h: the projection [h] supplies the outer + / handle either way, but reusing the name would shadow it and a later refactor could silently bind + / the wrong value (raised on PR #109 and worth keeping fixed) + ok:@[{[h] h(`.heartbeat.subscribe;::);1b};h; + {[hdl;e] .z.m.logerr[`subscribeone;"failed to subscribe to heartbeats on handle ",(string hdl),": ",e];0b}[h]]; + if[ok;.z.m.subscribedhandles:distinct subscribedhandles,h]; + }; + +getheartbeats:{[proctypes] + / di.servers.getservers returns a TABLE of server rows - the handles are its w column, and a + / disconnected row carries a null handle + handles:validhandles exec w from .z.m.serversgetservers proctypes; + handles:handles except subscribedhandles; + if[count handles; + .z.m.loginfo[`getheartbeats;"subscribing to new heartbeat handle(s) ",", " sv string handles]; + subscribeone each handles]; + }; + +hbsubscriptions:{[] + / timer job - pick up any newly connected publisher of a configured process type + getheartbeats resolveconnections[]; + }; + +closeconnection:{[h] + / drop a closed handle from the tracked subscriptions - registered against .z.pc + .z.m.subscribedhandles:subscribedhandles except h; + }; + +/ ============================================================ +/ warning / error transitions +/ ============================================================ + +logwarnproc:{[r] + .z.m.logwarn[`checkheartbeat;"process ",(string r`procname)," (type ",(string r`sym), + ") has not heartbeated since ",string r`time]; + }; + +logerrproc:{[r] + .z.m.logerr[`checkheartbeat;"process ",(string r`procname)," (type ",(string r`sym), + ") has not heartbeated since ",string r`time]; + }; + +evictstale:{[now] + / forget processes that have been silent for longer than maxage, so a monitor does not grow without + / bound. the store is keyed on sym+procname, so anything that churns identities - containers with + / generated names, a process restarting under a new procname - adds a row per incarnation that + / would otherwise live for the life of the monitor. + / eviction is by AGE, never by row count: a dead process has an old timestamp by definition, so + / evicting "the oldest N" would discard exactly the rows worth keeping. + / a row that has never heartbeated (null counter) came from addprocs and is operator intent, not an + / observation - forgetting one would silently stop reporting a process you declared you expected. + / those are removed only by removeprocs. maxage is validated to exceed the error period, so an + / evicted row has always fired its error transition first + if[0Wn=maxage;:()]; + flat:0!hb; + / only an ALREADY-ERRORED row may be evicted. validating maxage > the error period is not enough on + / its own: it only guarantees the transition would have fired had a check run in between, and a + / monitor that was paused, restarted, or handed a beat carrying an old timestamp gets its first + / check when the row is already past maxage - which evicted it without anyone ever being told the + / process had stopped. gating on error makes "forgotten only after being reported" structurally + / true instead of dependent on check cadence. the row simply survives one extra check cycle + keep:(null flat`counter) or (not flat`error) or not now>flat[`time]+maxage; + if[all keep;:()]; + .z.m.loginfo[`checkheartbeat;"evicting ",(string sum not keep)," process(es) silent for longer ", + "than maxage: ",", " sv string flat[`procname] where not keep]; + .z.m.hb:2!flat where keep; + }; + +warn:{[procs] + / move processes into warning state, log the transition and fire the warning callback. state is + / updated BEFORE the callback runs, so an isolated callback failure cannot leave the store wrong + if[debug;logwarnproc each 0!procs]; + .z.m.hb:hb upsert select sym,procname,warning:1b from procs; + safecall[`checkheartbeat;`onwarning;onwarning;procs]; + }; + +err:{[procs] + / move processes into error state, log the transition and fire the error callback. + / NB warning is deliberately NOT cleared here, so an escalated process carries warning:1b AND + / error:1b. both flags are literally true - it is past both thresholds - and this matches legacy + / (heartbeat.q:71,77 set each flag independently and never clear). only a fresh heartbeat clears + / them, in storeheartbeat. queried on PR #109; keeping legacy semantics, and a consumer rendering + / both columns should treat error as taking precedence + if[debug;logerrproc each 0!procs]; + .z.m.hb:hb upsert select sym,procname,error:1b from procs; + safecall[`checkheartbeat;`onerror;onerror;procs]; + }; + +/ ============================================================ +/ public api +/ ============================================================ + +publishheartbeat:{[] + / publish one heartbeat row and bump the counter - timer job + requireinit[`publishheartbeat]; + if[not enabled;:()]; + row:enlist `time`sym`procname`counter`pid`host`port! + (cp[];proctype;procname;hbcounter;pid;host;port); + / publishroot 0b means do NOT publish at all - not "publish privately". without the root schema + / table di.pubsub can never serve this topic (its init scans root tables[] and resolves each name + / with value), so publishing would build and discard a row on every tick, forever. retain it + / locally instead, where getownhb can still read it and nothing touches root namespace + / the publish is isolated: a transient pub/sub failure must not permanently disable this timer job. + / the counter still advances, so it counts beats ATTEMPTED and a gap in what subscribers received + / stays visible to them + $[publishroot; + safecall[`publishheartbeat;`publish;.z.m.pubsubpublish[tablename;];row]; + .z.m.ownhb:row]; + .z.m.hbcounter:hbcounter+1; + }; + +checkheartbeat:{[] + / flag processes that have not heartbeated within the warning / error grace periods - timer job + / status: 0 healthy, 1 warning, 2+ error. grace periods are computed as locals first, since module + / functions do not resolve inside qsql + requireinit[`checkheartbeat]; + now:cp[]; + / evict first: maxage is validated to exceed the error period, so nothing can be evicted before it + / has already been through its error transition + evictstale[now]; + t:0!hb; + if[not count t;:()]; + wp:warningperiod each t`sym; + ep:errorperiod each t`sym; + stats:update status:(`short$now>time+wp)+`short$2*now>time+ep from t; + newwarn:select sym,procname,time from stats where status=1,not warning; + newerr:select sym,procname,time from stats where status>1,not error; + if[count newwarn;warn newwarn]; + if[count newerr;err newerr]; + }; + +storeheartbeat:{[batch] + / store incoming heartbeats, keeping the latest per process and clearing warning / error state. + / call this from the consuming process's own upd when a heartbeat arrives - deliberately NOT an + / automatic upd hook, which legacy composed at load time and which breaks on load order + requireinit[`storeheartbeat]; + / validated up front so a malformed payload names itself in the log, rather than surfacing as a + / raw 'sym from the select below and bypassing the logger entirely + if[98h<>type batch; + raiseerror[`storeheartbeat;"batch must be an unkeyed table of heartbeat rows; got type ", + string type batch]]; + if[not count batch;:()]; + / time is required, not just sym and procname: checkheartbeat compares now against time+period, so + / a row without one can never warn or error + if[not all `sym`procname`time in cols batch; + raiseerror[`storeheartbeat;"batch must carry sym, procname and time columns; got: ", + ", " sv string cols batch]]; + / the type is checked too, not just presence: a date column reaches the keyed upsert and throws a + / raw 'type that bypasses the logger entirely + if[12h<>type batch`time; + raiseerror[`storeheartbeat;"the time column must be a timestamp vector; got type ", + string type batch`time]]; + / tolerate a publisher on a newer schema by keeping the columns we know and ignoring the rest. + / rejecting the batch outright would make a version-skewed publisher look DEAD to this monitor, + / which is a far worse failure than dropping a column we have no use for. warn only when the + / unexpected set changes, so skew stays visible without a log line per beat + if[count extra:(cols batch) except cols schema; + if[not extra~warnedcols; + .z.m.warnedcols:extra; + .z.m.logwarn[`storeheartbeat;"ignoring unrecognised heartbeat column(s) - publisher may be ", + "on a newer schema: ",", " sv string extra]]]; + rows:((cols schema) inter cols batch)#batch; + / a null time is permanently invisible to checkheartbeat: now>time+period is never true against a + / null, so such a process could never be flagged however long it stays silent. drop it loudly + if[count bad:select from rows where null time; + .z.m.logwarn[`storeheartbeat;"dropping heartbeat row(s) with a null time - they could never be ", + "flagged as stale: ",", " sv string distinct exec procname from bad]]; + rows:select from rows where not null time; + if[not count rows;:()]; + .z.m.hb:hb upsert update warning:0b,error:0b from select by sym,procname from rows; + }; + +addprocs:{[proctypes;procnames] + / seed the store with expected processes so one that never heartbeats at all is still flagged. + / prepended, so a real heartbeat arriving later wins + requireinit[`addprocs]; + pt:(),proctypes; + pn:(),procnames; + if[not all 11h=type each (pt;pn); + raiseerror[`addprocs;"proctypes and procnames must be symbols or symbol lists"]]; + if[not (count pt)=count pn; + raiseerror[`addprocs;"proctypes and procnames must be the same length; got ", + (string count pt)," and ",string count pn]]; + / counter stays null, which is what marks the row as operator intent rather than an observation: + / evictstale keeps it forever, and only removeprocs will clear it + seed:2!([]sym:pt;procname:pn;time:cp[];counter:0N;pid:0Ni;host:`;port:0Ni;warning:0b;error:0b); + .z.m.hb:seed,hb; + }; + +removeprocs:{[proctypes;procnames] + / forget the named processes entirely - the counterpart to addprocs. needed because a seeded row is + / deliberately never evicted on age, so without this a decommissioned process declared via addprocs + / could never be removed from the store at all + requireinit[`removeprocs]; + pt:(),proctypes; + pn:(),procnames; + if[not all 11h=type each (pt;pn); + raiseerror[`removeprocs;"proctypes and procnames must be symbols or symbol lists"]]; + if[not (count pt)=count pn; + raiseerror[`removeprocs;"proctypes and procnames must be the same length; got ", + (string count pt)," and ",string count pn]]; + drop:([]sym:pt;procname:pn); + .z.m.hb:2!delete from 0!hb where ([]sym;procname) in drop; + }; + +subscribe:{[handles] + / subscribe to heartbeats on the given remote handle(s), tracking successful subscriptions + requireinit[`subscribe]; + h:(),handles; + if[not type[h] within 5 7h; + raiseerror[`subscribe;"handles must be integers; got type ",string type h]]; + valid:validhandles h; + if[count skipped:h except valid; + .z.m.logwarn[`subscribe;"skipping handle(s) that cannot be subscribed - 0i is this process ", + "itself and a null handle is a dead server row: ",", " sv string skipped]]; + subscribeone each valid; + }; + +gethb:{[] + / the store of heartbeats RECEIVED from other processes + requireinit[`gethb]; + :hb; + }; + +getownhb:{[] + / the last heartbeat this process produced. under publishroot 0b this is the only record of it, + / since nothing is published; under publishroot 1b it is empty and the row went out over pub/sub + requireinit[`getownhb]; + :ownhb; + }; + +setcp:{[f] + / replace the current-time function - used by tests and simulation + requireinit[`setcp]; + if[not type[f] within 100 112h;raiseerror[`setcp;"f must be a function returning a timestamp"]]; + / call it once and check the result type. a clock returning anything else does NOT throw inside + / checkheartbeat - the staleness comparison just silently evaluates false against every row, so + / monitoring would quietly stop flagging anything at all rather than failing loudly + probe:@[f;::;{[e] '"di.heartbeat: setcp: the clock function threw when called: ",e}]; + if[not -12h=type probe; + raiseerror[`setcp;"the clock function must return a timestamp; got type ",string type probe]]; + .z.m.cp:f; + }; + +teardown:{[] + / release everything init installed: timer jobs, the .z.pc registration and the root names + requireinit[`teardown]; + .z.m.timerdeletejobs jobids; + / keyed off what is actually installed, NOT off the current config - a re-init that flipped + / subenabled or publishroot off would otherwise leave teardown unable to clean up its own residue + if[handlerregistered; + .z.m.handlersremove[`.z.pc;`;`heartbeat]; + .z.m.handlerregistered:0b]; + if[rootinstalled;uninstallroot[]]; + .z.m.enabled:0b; + .z.m.subenabled:0b; + .z.m.loginfo[`teardown;"di.heartbeat torn down - timers, handlers and root names released"]; + }; + +init:{[deps] + / wire the injected deps and this process's config from ONE dict, then install the timer jobs, the + / .z.pc observer (when monitoring) and the root names (when publishroot). idempotent - a second + / call clears its own timer jobs first and re-registers, leaving the heartbeat store intact + / deps: `log`timer`pubsub (required), `servers`handlers (required when subenabled), + / `proctype`procname (required identity), plus any config key alongside them + / e.g. hb.init[(`log`timer`pubsub!(logdep;timerdep;psdep)),`proctype`procname!(`rdb;`rdb1)] + validatedeps[deps]; + .z.m.loginfo:(deps`log)`info; + .z.m.logwarn:(deps`log)`warn; + .z.m.logerr:(deps`log)`error; + cfg:resolveconfig[deps]; + if[cfg`subenabled;validatemonitordeps[deps]]; + validateconfig[cfg]; + / every validation that can fail runs here, before a single byte of module state is written + validateroot[cfg]; + .z.m.timeraddjob:(deps`timer)`addjob; + .z.m.timerdeletejobs:(deps`timer)`deletejobs; + .z.m.pubsubpublish:(deps`pubsub)`publish; + .z.m.pubsubsubscribe:(deps`pubsub)`subscribe; + if[cfg`subenabled; + .z.m.serversgetservers:(deps`servers)`getservers; + .z.m.handlersregister:(deps`handlers)`register; + .z.m.handlersremove:(deps`handlers)`remove]; + / first init only - a re-init must not discard heartbeats already received + if[not initialised[]; + .z.m.hb:storeschema; + .z.m.ownhb:0#schema; + .z.m.subscribedhandles:`int$(); + .z.m.rootowned:0b; + .z.m.rootinstalled:0b; + .z.m.handlerregistered:0b; + .z.m.warnedcols:`symbol$(); + .z.m.hbcounter:0]; + .z.m.config:cfg; + .z.m.proctype:deps`proctype; + .z.m.procname:deps`procname; + / written out one key at a time rather than looped over cfg: an explicit .z.m.: write is the + / documented form, it keeps every config key greppable, and it does not lean on dynamic .z.m indexing + .z.m.enabled:cfg`enabled; + .z.m.subenabled:cfg`subenabled; + .z.m.debug:cfg`debug; + .z.m.publishroot:cfg`publishroot; + .z.m.publishinterval:cfg`publishinterval; + .z.m.checkinterval:cfg`checkinterval; + .z.m.warningtolerance:cfg`warningtolerance; + .z.m.errortolerance:cfg`errortolerance; + .z.m.maxage:cfg`maxage; + .z.m.pid:cfg`pid; + .z.m.host:cfg`host; + .z.m.port:cfg`port; + .z.m.connections:cfg`connections; + .z.m.onwarning:cfg`onwarning; + .z.m.onerror:cfg`onerror; + / reconcile the root names FIRST. this is the one step that can fail on external state (a foreign + / table already at that name), so failing here leaves the process untouched rather than + / half-configured with timers already publishing into a topic di.pubsub cannot serve + $[publishroot;installroot[];droprootnames[]]; + if[not publishroot; + .z.m.loginfo[`init;"publishroot is 0b - nothing published at root and no pub/sub publishing; ", + "own heartbeats are tracked locally and readable via getownhb"]]; + registertimers[]; + registerhandlers[]; + / NB the index expressions are parenthesised deliberately. juxtaposition binds to the WHOLE + / right-hand expression, so ("disabled";"enabled")enabled,", monitoring ",... parses as + / ("disabled";"enabled")[enabled,", monitoring ",...] - indexing by the rest of the string, which + / yields a nested list rather than a flat one. di.log rejects that with 'type; a permissive mock + / logger does not, which is exactly why this survived a green suite + .z.m.loginfo[`init;"di.heartbeat initialised - proctype ",(string proctype),", procname ", + (string procname),", publishing ",(("disabled";"enabled")enabled),", monitoring ", + (("disabled";"enabled")subenabled)]; + }; + +getapimeta:{[] + / one row per CALLABLE api function, for di.torq to register with di.api. init and getapimeta are + / plumbing di.torq calls by convention and are deliberately omitted. names are bare + :flip `name`public`descrip`params`return!flip( + (`teardown; 1b; "release timer jobs, the .z.pc registration and the published root names"; + "[]"; "null"); + (`version; 1b; "module version string"; + "[]"; "string: version"); + (`publishheartbeat; 1b; "publish one heartbeat row over pub/sub and bump the counter"; + "[]"; "null"); + (`checkheartbeat; 1b; "flag processes that have not heartbeated within the grace periods"; + "[]"; "null"); + (`storeheartbeat; 1b; "store incoming heartbeats, latest per process, clearing warning/error"; + "[table: heartbeat rows]"; "null"); + (`addprocs; 1b; "seed the store with expected processes so a silent one is still flagged"; + "[symbol|list: proctypes; symbol|list: procnames]"; "null"); + (`removeprocs; 1b; "forget the named processes entirely - the counterpart to addprocs"; + "[symbol|list: proctypes; symbol|list: procnames]"; "null"); + (`subscribe; 1b; "subscribe to heartbeats on the given remote handle(s)"; + "[int|list: handles]"; "null"); + (`gethb; 1b; "the store of heartbeats received from other processes"; + "[]"; "table: keyed on sym,procname"); + (`getownhb; 1b; "the last heartbeat this process produced (populated when publishroot is 0b)"; + "[]"; "table: zero or one row"); + (`setcp; 1b; "replace the module's current-time function, for tests and simulation"; + "[function: niladic returning a timestamp]"; "null")); + }; diff --git a/di/heartbeat/init.q b/di/heartbeat/init.q new file mode 100644 index 00000000..aa9c3ee7 --- /dev/null +++ b/di/heartbeat/init.q @@ -0,0 +1,15 @@ +/ di.heartbeat - periodic liveness signalling over pub/sub, and monitoring of other processes' beats +/ consolidates TorQ's heartbeat.q (.hb), covering both the publisher and the monitor role + +\l ::heartbeat.q + +/ module version, read from the VERSION file rather than hardcoded, so a release bump touches one +/ plain-text file. NB `version` STAYS in the export: di.depcheck resolves a dependency's version from +/ the export dict and fails the check with "exports no version" for any module that drops it +version:first read0`:::VERSION + +/ public api - init and getapimeta are framework plumbing di.torq calls by convention; every other +/ name here carries a getapimeta row, which the test suite asserts +export:([init;teardown;version;getapimeta; + publishheartbeat;checkheartbeat;storeheartbeat; + addprocs;removeprocs;subscribe;gethb;getownhb;setcp]) diff --git a/di/heartbeat/test.csv b/di/heartbeat/test.csv new file mode 100644 index 00000000..70973f05 --- /dev/null +++ b/di/heartbeat/test.csv @@ -0,0 +1,461 @@ +action,ms,bytes,lang,code,repeat,minver,comment +comment,,,,,,,setup - load the module with capturing mocks for every injected dependency +before,0,0,q,hb:use`di.heartbeat,1,1,load di.heartbeat +before,0,0,q,.pt.cap:([]lvl:`symbol$();ctx:`symbol$();msg:()),1,1,log capture table +before,0,0,q,"caplog:`info`warn`error!({[c;m] if[not 10h=type m;'`msgnotastring]; `.pt.cap insert (`info;c;m)};{[c;m] if[not 10h=type m;'`msgnotastring]; `.pt.cap insert (`warn;c;m)};{[c;m] if[not 10h=type m;'`msgnotastring]; `.pt.cap insert (`error;c;m)})",1,1,capturing binary logger that REJECTS a non-flat message exactly as real di.log does - a permissive mock hid a nested-list message for a whole session +before,0,0,q,.pt.jobs:([]id:`symbol$();period:();mode:()),1,1,timer job capture +before,0,0,q,"mocktimer:`addjob`deletejobs!((enlist`custom)!enlist {[i;f;p;pe;m;o] if[i in exec id from .pt.jobs;'`duplicateid]; `.pt.jobs insert (i;pe;m)};{[ids] delete from `.pt.jobs where id in ids})",1,1,timer mock - variant dict exposing custom; THROWS on duplicate id as real di.timer addjob.custom does +before,0,0,q,.pt.pub:(),1,1,publish capture +before,0,0,q,"mockps:`publish`subscribe!({[t;x] .pt.pub,:enlist (t;x)};{[t;f] .pt.sub:(t;f)})",1,1,pubsub mock capturing publish and subscribe +before,0,0,q,.pt.reg:([]event:`symbol$();phase:`symbol$();nm:`symbol$();pri:`long$()),1,1,handler registration capture +before,0,0,q,"mockh:`register`remove!({[e;p;n;pr;f] `.pt.reg insert (e;p;n;pr)};{[e;p;n] delete from `.pt.reg where event=e,phase=p,nm=n})",1,1,handlers mock - register is 5-arg +before,0,0,q,"mocksrv:enlist[`getservers]!enlist {[p] ([]w:`int$())}",1,1,servers mock returning an empty server TABLE +before,0,0,q,.pt.asked:(),1,1,records which proctype getservers was asked for +before,0,0,q,"mocksrvrows:enlist[`getservers]!enlist {[p] .pt.asked,:enlist p; ([]proctype:`rdb`hdb`dead`self;w:4 5 0Ni,0i)}",1,1,servers mock returning a POPULATED table - live handles plus a dead (null) row and handle 0 +before,0,0,q,"deps:(`log`timer`pubsub!(caplog;mocktimer;mockps)),`proctype`procname!(`rdb;`rdb1)",1,1,the base deps dict - deps plus required identity + +comment,,,,,,,init must be called before any other function (asserted FIRST - the module is a singleton and cannot be un-initialised later) +true,0,0,q,"0time+0Wn is 1b not 0b +run,0,0,q,hb.init[deps],1,1,restore the default instance + +comment,,,,,,,escalation flag semantics - both flags set is deliberate and legacy-faithful (queried on PR #109) +run,0,0,q,hb.teardown[],1,1,tear down +run,0,0,q,hb.init[deps],1,1,clean init +run,0,0,q,hb.setcp[{[] .pt.now}],1,1,point the clock +run,0,0,q,.m.di.0heartbeat.hb:.m.di.0heartbeat.storeschema,1,1,clear the store +run,0,0,q,.pt.now:2025.01.01D00:00:00,1,1,reset time +run,0,0,q,hb.addprocs[`rdb;`esc],1,1,seed a process that will escalate +run,0,0,q,.pt.now:2025.01.01D00:01:05,1,1,past the warning threshold +run,0,0,q,hb.checkheartbeat[],1,1,check +true,0,0,q,first exec warning from hb.gethb[] where procname=`esc,1,1,warning set +true,0,0,q,not first exec error from hb.gethb[] where procname=`esc,1,1,error not yet set +run,0,0,q,.pt.now:2025.01.01D00:02:00,1,1,past the error threshold +run,0,0,q,hb.checkheartbeat[],1,1,check again +true,0,0,q,first exec error from hb.gethb[] where procname=`esc,1,1,error set +true,0,0,q,first exec warning from hb.gethb[] where procname=`esc,1,1,warning DELIBERATELY still set - the process is past both thresholds and legacy sets each flag independently +run,0,0,q,"hb.storeheartbeat[([]time:enlist .pt.now;sym:enlist`rdb;procname:enlist`esc;counter:enlist 1;pid:enlist 1i;host:enlist`h;port:enlist 1i)]",1,1,a fresh heartbeat arrives +true,0,0,q,not first exec warning from hb.gethb[] where procname=`esc,1,1,recovery clears warning +true,0,0,q,not first exec error from hb.gethb[] where procname=`esc,1,1,and clears error - storeheartbeat is the only thing that clears either + +comment,,,,,,,init is atomic - a root-name clash must not leave a half-configured module behind +run,0,0,q,hb.teardown[],1,1,tear down so the root name is free +run,0,0,q,"set[`heartbeat;([]foreign:`int$())]",1,1,another module takes the root name +run,0,0,q,delete from `.pt.jobs,1,1,clear the job capture +run,0,0,q,delete from `.pt.reg,1,1,clear the registration capture +fail,0,0,q,hb.init[deps],1,1,init refuses the clash +true,0,0,q,0=count .pt.jobs,1,1,no timer jobs were registered before the throw +true,0,0,q,0=count .pt.reg,1,1,no handlers were registered before the throw +true,0,0,q,`foreign in cols value`heartbeat,1,1,the foreign table is untouched +run,0,0,q,"![`.;();0b;enlist`heartbeat]",1,1,free the name +run,0,0,q,hb.init[deps],1,1,init now succeeds +true,0,0,q,`heartbeat in tables[],1,1,and claims the root name properly + +comment,,,,,,,eviction must never pre-empt the error report - a paused monitor's first check must still tell you +run,0,0,q,hb.teardown[],1,1,tear down +run,0,0,q,.pt.fired:0,1,1,reset the callback counter +run,0,0,q,"hb.init[deps,enlist[`onerror]!enlist {[p] .pt.fired+:1}]",1,1,init with a counting onerror +run,0,0,q,.pt.now:2025.01.01D00:00:00,1,1,set the clock variable BEFORE setcp - setcp now calls the function to check its return type +run,0,0,q,hb.setcp[{[] .pt.now}],1,1,point the clock +run,0,0,q,.m.di.0heartbeat.hb:.m.di.0heartbeat.storeschema,1,1,clear the store +run,0,0,q,"hb.storeheartbeat[([]time:enlist .pt.now;sym:enlist`rdb;procname:enlist`ghost;counter:enlist 1;pid:enlist 1i;host:enlist`h;port:enlist 1i)]",1,1,one heartbeat +run,0,0,q,.pt.now:2025.01.03D00:00:00,1,1,jump two days - past maxage - WITHOUT any check in between +run,0,0,q,hb.checkheartbeat[],1,1,the first check since the beat +true,0,0,q,1=count hb.gethb[],1,1,the row was NOT evicted on the first check - it had never been reported +true,0,0,q,1=.pt.fired,1,1,onerror fired instead - you are told the process stopped +true,0,0,q,first exec error from hb.gethb[] where procname=`ghost,1,1,and the row is flagged error +run,0,0,q,hb.checkheartbeat[],1,1,the next check +true,0,0,q,0=count hb.gethb[],1,1,NOW it is evicted - reported first, forgotten second +true,0,0,q,1=.pt.fired,1,1,and onerror did not fire twice + +comment,,,,,,,a wrong-arity callback is silently never called - reject it at init +fail,0,0,q,"hb.init[deps,enlist[`onwarning]!enlist {[a;b] }]",1,1,a two-argument onwarning is rejected - applying it to one argument yields a projection not a call +fail,0,0,q,"hb.init[deps,enlist[`onerror]!enlist {[a;b] }]",1,1,a two-argument onerror is rejected +run,0,0,q,"hb.init[deps,enlist[`onwarning]!enlist {[] }]",1,1,a niladic callback is ACCEPTED - q gives it arity 1 and it is genuinely invoked, it just ignores the rows +run,0,0,q,"hb.init[deps,enlist[`onwarning]!enlist {[p] }]",1,1,an explicit unary callback is accepted +run,0,0,q,"hb.init[deps,enlist[`onwarning]!enlist {x}]",1,1,and so is the implicit-argument form + +comment,,,,,,,the clock must actually return a timestamp - a bad one silently disables monitoring +run,0,0,q,hb.init[deps],1,1,clean init +fail,0,0,q,hb.setcp[{[] 42}],1,1,a clock returning a long is rejected - the staleness comparison would just evaluate false against every row +fail,0,0,q,hb.setcp[{[] `notatime}],1,1,a clock returning a symbol is rejected +fail,0,0,q,hb.setcp[{[] 2025.01.01}],1,1,a clock returning a date is rejected +run,0,0,q,.pt.now:2025.01.01D00:00:00,1,1,define the clock variable first +run,0,0,q,hb.setcp[{[] .pt.now}],1,1,a timestamp-returning clock is accepted + +comment,,,,,,,handles that cannot work are filtered rather than tracked on no evidence +true,0,0,q,"(enlist 4i)~.m.di.0heartbeat.validhandles -5 4 0Ni,0i",1,1,negative null and zero handles are all dropped - only the positive one survives +run,0,0,q,hb.subscribe[-5i],1,1,subscribing a negative handle is accepted but skipped +true,0,0,q,not -5i in .m.di.0heartbeat.subscribedhandles,1,1,a negative handle is async so it would be recorded as subscribed on no evidence and closeconnection could never clear it + +comment,,,,,,,storeheartbeat validates the time column TYPE not just its presence +fail,0,0,q,"hb.storeheartbeat[([]time:enlist 2025.01.01;sym:enlist`rdb;procname:enlist`d;counter:enlist 1;pid:enlist 1i;host:enlist`h;port:enlist 1i)]",1,1,a date time column is rejected by name rather than throwing a raw type from the keyed upsert +run,0,0,q,.pt.errstr2:@[{hb.storeheartbeat[([]time:enlist 2025.01.01;sym:enlist`rdb;procname:enlist`d)]};(::);{x}],1,1,capture the message +true,0,0,q,"0/tmp/dihbchild",string[port],".log 2>&1 &"; + h:0N; + i:0; + while[(null h) and i<100; + h:@[{hopen `$":localhost:",string x};port;0N]; + i+:1; + system"sleep 0.1"]; + if[null h;'"di.heartbeat test: publisher process failed to start - see /tmp/dihbchild",string[port],".log"]; + / wait for its init to complete + i:0; + while[(not @[h;"@[{ready};::;0b]";0b]) and i<100; + i+:1; + system"sleep 0.1"]; + .ht.port:port; + .ht.h:h; + :h; + }; + +/ the monitor's own upd - di.heartbeat deliberately does NOT install this itself. +/ hbm is indexed, not dotted: module dot-sugar only works on a plain top-level name, and fails +/ silently inside a lambda or on a dotted one +installupd:{[] + upd::{[t;x] if[t=`heartbeat;hbm[`storeheartbeat]x]}; + }; + +/ force the parent to drain any async messages the publisher has pushed +drain:{[h] + h"1"; + }; + +cleanup:{[] + / terminate the child FIRST - hclose only drops our end of the socket and would leave the process + / running for the lifetime of the test host. sent SYNC inside a protected apply: the call cannot + / return because the peer exits mid-request, and swallowing that error is the point. an async send + / is not reliable here - it can sit in the output buffer and be discarded by the hclose below + @[{x"exit 0"};.ht.h;::]; + @[hclose;.ht.h;::]; + @[{system"rm -f /tmp/dihbchild",string[x],".q /tmp/dihbchild",string[x],".log"};.ht.port;::]; + }; diff --git a/di/heartbeat/test_integration.csv b/di/heartbeat/test_integration.csv new file mode 100644 index 00000000..c16cf02c --- /dev/null +++ b/di/heartbeat/test_integration.csv @@ -0,0 +1,65 @@ +action,ms,bytes,lang,code,repeat,minver,comment +comment,,,,,,,integration - a real publisher process and a real di.pubsub. mocks cannot reach these paths +before,0,0,q,hbm:use`di.heartbeat,1,1,load di.heartbeat in the monitor +before,0,0,q,os:use`di.os,1,1,di.os for abspath (harness only) +before,0,0,q,"system ""l "", os.abspath[""di/heartbeat/test.q""]",1,1,load the spawn/fixture helpers +before,0,0,q,.ht.lg:`info`warn`error!({[c;m]};{[c;m]};{[c;m]}),1,1,silent logger for the monitor +before,0,0,q,".ht.tmr:`addjob`deletejobs!((enlist`custom)!enlist {[i;f;p;pe;m;o]};{[ids]})",1,1,stub timer - this suite drives the calls by hand +before,0,0,q,".ht.ps:`publish`subscribe!({[t;x]};{[t;f]})",1,1,monitor-side pubsub stub - the monitor never publishes here +before,0,0,q,".ht.hnd:`register`remove!({[e;p;n;pr;f]};{[e;p;n]})",1,1,handlers stub +before,0,0,q,".ht.srv:enlist[`getservers]!enlist {[p] ([]w:`int$())}",1,1,servers stub - handles are supplied directly to subscribe +before,0,0,q,"handle:spawnpublisher[]",1,1,launch a genuinely separate q process running di.heartbeat as a publisher +before,0,0,q,installupd[],1,1,install the monitor's own upd - the module never does this itself + +comment,,,,,,,the publisher published its root names into its OWN process +true,0,0,q,"handle""`heartbeat in tables[]""",1,1,the publisher created the root schema table di.pubsub needs to discover +true,0,0,q,"handle""@[{value`.heartbeat.subscribe;1b};::;0b]""",1,1,the publisher published the remote-subscribe entry point +true,0,0,q,"`time`sym`procname`counter`pid`host`port~handle""cols value`heartbeat""",1,1,the published schema matches the row shape + +comment,,,,,,,the remote-subscribe handshake - the mechanism a pubsub mock cannot exercise +run,0,0,q,"base:(`log`timer`pubsub!(.ht.lg;.ht.tmr;.ht.ps)),`proctype`procname!(`montype;`monproc)",1,1,build the base deps dict first - an inline k!(v) followed by a comma binds the comma to the VALUE list not the dict +run,0,0,q,"hbm[`init][base,`subenabled`servers`handlers!(1b;.ht.srv;.ht.hnd)]",1,1,init the monitor with the monitor-only deps joined last +run,0,0,q,hbm[`subscribe][handle],1,1,subscribe to the publisher over the real handle +true,0,0,q,"0 Date: Tue, 11 Aug 2026 14:01:26 +0100 Subject: [PATCH 2/5] async subscribe, drop handle cache and handlers dep, add discovery warnings --- di/heartbeat/deps.q | 3 +- di/heartbeat/heartbeat.md | 236 ++++++++++++++++++++++--- di/heartbeat/heartbeat.q | 285 ++++++++++++++++++++++-------- di/heartbeat/test.csv | 179 +++++++++++++++---- di/heartbeat/test.q | 16 ++ di/heartbeat/test_integration.csv | 19 +- 6 files changed, 605 insertions(+), 133 deletions(-) diff --git a/di/heartbeat/deps.q b/di/heartbeat/deps.q index 42185407..7811efcb 100644 --- a/di/heartbeat/deps.q +++ b/di/heartbeat/deps.q @@ -1,4 +1,5 @@ / hard module dependencies and their minimum versions, validated by di.depcheck / di.heartbeat has no hard dependencies - all runtime dependencies (log, timer, -/ handlers, pubsub, servers) are injected via init as dictionaries of functions +/ pubsub, servers) are injected via init as dictionaries of functions. a handlers +/ dict is accepted and ignored, for di.torq's uniform wiring - see depkeys deps:(`$())!(); diff --git a/di/heartbeat/heartbeat.md b/di/heartbeat/heartbeat.md index 3042f65b..f31a7c77 100644 --- a/di/heartbeat/heartbeat.md +++ b/di/heartbeat/heartbeat.md @@ -29,12 +29,18 @@ All injected via `init`; there are no hard module dependencies (`deps.q` is empt | `log` | always | `info`, `warn`, `error` | binary `{[ctx;msg]}` - `di.log`'s `logdict` fits directly | | `timer` | always | `addjob`, `deletejobs` | `addjob` must be the variant dict exposing `custom` | | `pubsub` | always | `publish`, `subscribe` | `di.pubsub` | -| `servers` | when `subenabled` | `getservers` | `di.servers`; returns a **table**, handles are its `w` column | -| `handlers` | when `subenabled` | `register`, `remove` | `di.handlers`; 5-arg `register[event;phase;nm;pri;func]` | +| `servers` | when `subenabled` | `getservers` | `di.servers`; returns a **table**, handles are its `w` column. `procname` and `proctype` are used for the never-beaten warning, and skipped if absent | Nothing is silently defaulted. A missing or malformed dependency throws from `init` with a message naming the module that supplies it. +> **No `handlers` dependency.** This module used to register a `.z.pc` observer, purely to prune a +> cache of already-subscribed handles. That cache is gone (see *How a monitor subscribes*), and the +> observer went with it. A `handlers` key passed anyway is **silently accepted and ignored**, so +> `di.torq` can keep wiring every module with one uniform dict — it is deliberately still listed in +> the module's `depkeys` so it is never mistaken for a stray config key and warned about on every +> publisher-only boot. + ## Initialisation `init` takes exactly **one** dict, carrying dependencies and config side by side. @@ -46,10 +52,13 @@ ps:use`di.pubsub hb:use`di.heartbeat timer.init[] -hb.init[logging.logdict,`timer`pubsub`proctype`procname!(timerdep;psdep;`rdb;`rdb1)] +hb.init[logging.logdict,`timer`pubsub`proctype`procname!(timer;ps;`rdb;`rdb1)] ps.init[] / MUST run after hb.init - see Ordering below ``` +The module export dicts (`timer`, `ps`) are passed straight through - `di.heartbeat` reads the keys it +needs off them. + `proctype` and `procname` are **required**. They are published as the heartbeat's `sym` and `procname` and have no sensible default. (Legacy took them from `.proc.*`; a module cannot.) @@ -85,10 +94,12 @@ Passed in the same dict as the dependencies. Every key is optional. | `publishroot` | `1b` | new | publish at root; see below | | `publishinterval` | `0D00:00:30` | both agree | how often to publish | | `checkinterval` | `0D00:00:10` | both agree | how often to check | +| `subscribeinterval` | `0D00:01:00` | legacy value, now configurable | how often to sweep for newly connected publishers | | `warningtolerance` | `2f` | **shipped** | warning after `tolerance * publishinterval` | | `errortolerance` | `3f` | **shipped** | error after `tolerance * publishinterval` | +| `subscribewarnsweeps` | `3` | new | consecutive-sweep threshold for both monitor warnings: a subscribed peer sending nothing, and discovering no peer at all | | `maxage` | `0D24:00:00` | new | forget a process silent this long; `0Wn` to keep forever | -| `connections` | `` `ALL `` | **shipped** | process types to monitor; `` `ALL `` means every one | +| `connections` | `` `ALL `` | **shipped** | process types to monitor; `` `ALL `` means every one. **See the caveat below - `` `ALL `` does not currently work** | | `onwarning` | no-op | new | unary callback given the rows entering warning | | `onerror` | no-op | new | unary callback given the rows entering error | | `pid` / `host` / `port` | `.z.i` / `.z.h` / `system"p"` | legacy | captured once at load, as legacy did | @@ -115,9 +126,42 @@ Setting `subenabled:1b` with an **empty** `connections` list is legal but warns: as legacy's in-file `connections:()` default, where the entire monitor path silently did nothing. Use `` `ALL ``, or name the process types to watch. -Setting `subenabled:1b` with an **empty** `connections` list is legal but warns: it is the same shape -as legacy's in-file `connections:()` default, where the entire monitor path silently did nothing. Use -`` `ALL ``, or name the process types to watch. +### ⚠️ `` `ALL `` does not currently work - a regression in `di.servers`, tracked + +The shipped default resolves to a null symbol, which every ancestor of `getservers` treats as +match-all. `di.servers`' version does not: + +| Implementation | Contract | +|---|---| +| legacy TorQ `.servers.getservers` (`trackservers.q:75`) | `` `~lookups `` → every server; otherwise `proctype in lookups` (so a **list** works) | +| `di.serverselect.getservers` | identical - `` ` `` as `lookups` returns all active servers | +| **`di.servers.getservers`** | requires a symbol **atom** and matches `proctype=pt` - a null matches **nothing**, a list **throws** | + +So this is a regression against both its ancestors, not a missing feature. Consequences today: + +- **`` connections:`ALL ``** (the default) discovers zero servers, silently. +- **`` connections:`rdb`hdb ``** works, because `di.heartbeat` calls `getservers` **once per + proctype** rather than passing the list. That iteration is this module's workaround and is + forward-compatible - a one-element iteration still works if `getservers` later accepts vectors. + +**Until `di.servers` is fixed, name the process types explicitly.** The `` `ALL `` path cannot be +worked around from this side: there is no proctype universe to iterate over without `di.servers` +providing one. + +It is, however, no longer **silent**. A monitor that is configured to watch something and discovers +no usable peer for `subscribewarnsweeps` consecutive sweeps warns once, naming `` `ALL `` as the +likeliest cause. See *The discovered-nothing warning* below — that check is deliberately +cause-agnostic, so it needs no maintenance when `di.servers` is fixed. + +**`connections` is not the same key as `di.servers.connections`.** They share a name and mean +different things: `di.servers.connections` decides which process types this process *connects to at +all*; `di.heartbeat.connections` filters which of those already-connected types get *heartbeat +subscribed*. A type named here but absent there is silently inert - `getservers` simply has no rows +for it. A config that looks complete can therefore monitor nothing. + +**Discovery is gradual on a cold start.** `hbsubscriptions` only finds what `di.servers.startup` / +`retry` has actually connected to by the time it sweeps, so a freshly started monitor picks peers up +over the first few sweeps rather than all at once. Not a bug, but worth expecting. ## Robustness @@ -133,7 +177,7 @@ monitoring off. (Consequence: the clock's state must exist when `setcp` is calle **A broken callback cannot stop heartbeating.** `onwarning`, `onerror` and the `pubsub.publish` call all run isolated: a throw is logged at error and execution continues. This matters more than it looks. -Both timer jobs are scheduled through `di.timer`, whose `addjob.opts` defaults `disableonfail:1b` — so +All three timer jobs are scheduled through `di.timer`, whose `addjob.opts` defaults `disableonfail:1b` — so an unprotected throw would not merely skip one beat, it would **permanently disable** the job. A single bug in a client's `onwarning`, or one transient pub/sub outage, would silently end the monitoring this module exists to provide. State is always updated *before* a callback fires, so isolation never leaves @@ -174,11 +218,11 @@ anyone ever being told the process had stopped. Gating on `error` makes "forgott reported" structurally true rather than dependent on check cadence — the row simply survives one extra check cycle. Set `maxage:0Wn` to disable eviction entirely. -**Flipping a flag off on re-`init` cleans up after the previous one.** `registertimers`, -`registerhandlers` and the root-name sync all *reconcile* rather than only add. Re-initialising with -`subenabled:0b` deregisters the `.z.pc` observer; with `publishroot:0b` it removes the published root -names. Both leaks were real and both were permanent: `teardown` used to key off the *current* config, -so once a flag was off nothing could remove what the earlier init had installed. A lingering +**Flipping a flag off on re-`init` cleans up after the previous one.** `registertimers` and the +root-name sync both *reconcile* rather than only add. Re-initialising with `subenabled:0b` removes the +subscription sweep job; with `publishroot:0b` it removes the published root names. Both leaks were +real and both were permanent: `teardown` used to key off the *current* config, so once a flag was off +nothing could remove what the earlier init had installed. A lingering `.heartbeat.subscribe` is the worse of the two — a remote monitor can still subscribe *successfully* and then receive nothing for the life of the process, while believing it is watching you. @@ -231,22 +275,151 @@ addition, not a port. `di.pubsub.subscribe` registers **the caller's own `.z.w`**. A monitor therefore cannot subscribe itself to a remote publisher by calling it locally - it would only ever subscribe itself to itself. -Legacy solved this with a synchronous IPC call (`heartbeat.q:92`), and so does this module: the monitor -calls the publisher's root ``.heartbeat.subscribe`` over the handle. During that inbound call the -publisher's `.z.w` *is* the connection back to the monitor, so the subscription lands correctly. +Legacy solved this with an IPC call (`heartbeat.q:92`), and so does this module: the monitor calls the +publisher's root ``.heartbeat.subscribe`` over the handle. During that inbound call the publisher's +`.z.w` *is* the connection back to the monitor, so the subscription lands correctly. ```q hb.subscribe[handle] / monitor side; sends `.heartbeat.subscribe to the publisher ``` -A failed subscribe is logged and **not** recorded, so the next `hbsubscribe` tick retries it. +### The send is asynchronous, and why that matters + +Legacy sent this **synchronously**. That is an availability hazard in a module whose entire job is +detecting unresponsive processes: a sync call waits as long as the peer takes to answer, and a peer +that is *alive but stalled* — a GC pause, a heavy query, exactly the condition heartbeats exist to +surface — never answers promptly. Because the sweep runs inside a `di.timer` job on a single thread, +that wait stalled `publishheartbeat` and `checkheartbeat` along with it, so **the monitor fell silent +to its own monitors at precisely the moment a peer was misbehaving.** A failed attempt was never +recorded either, so it was retried — and re-blocked — on every sweep. Measured: 10s of blocking +against a peer hung for 10s, versus 36µs for the async send. (`hopen`'s timeout does not help; it +bounds connection *establishment*, not later requests on an established handle.) + +The async send costs nothing in correctness — `.z.w` resolves to the monitor's connection during an +async inbound call just as it does for a sync one — but it does change what failures are visible: + +| Failure | Reported | +|---|---| +| dead handle, null handle | **yes**, immediately, at `error` — the send itself throws | +| the remote throws, or has no `.heartbeat.subscribe` | **no** — an async send gets no answer | + +That second row is covered indirectly instead. See *The never-beaten warning* below. + +### There is no cache of subscribed handles + +An earlier version tracked which handles it had already subscribed, to skip them on later sweeps. +That was removed rather than repaired, because **a handle number is not an identity**: kdb+ reissues +the lowest free descriptor immediately (measured: handle `4` → close → reopen → handle `4`), and +`hclose` does not fire `.z.pc` at all (also measured). A stale entry therefore made the monitor skip a +live peer *forever*, with nothing in any log to explain it — and no cheap liveness probe fixes it, +because after a close-and-reopen the recycled number is back in `.z.W` and looks perfectly valid. + +Nothing is lost by dropping the cache: `di.pubsub` already dedupes subscribers by handle +(`pubsub.q:11`), so a repeat subscribe is a no-op on the publisher, and an async send is far too cheap +to be worth caching around. Removing it also removed this module's only reason to register a `.z.pc` +handler, and with it the `handlers` dependency. + +### The never-beaten warning + +Since an async send cannot report a *remote* failure, the sweep watches for the consequence instead: +a peer it has subscribed to that never produces a heartbeat. After `subscribewarnsweeps` consecutive +sweeps (default 3) in which a discovered, attempted peer has no observed beat in the store, it warns +once — naming the peers. + +Three details are deliberate: + +- **Only peers actually attempted are counted.** A row with a null handle or handle `0` was skipped on + purpose, so warning that it never beat would be a false alarm about a peer nobody asked. +- **An `addprocs` seed does not count as an observation.** Those rows carry a null `counter`, the same + discriminator `evictstale` uses to tell operator *intent* from a real observation. +- **The count restarts if a peer leaves the discovered set and returns.** A connectivity blip is not + the continuously-stuck subscribe this warning exists to catch; only an uninterrupted run trips it. + +It warns at exactly the threshold sweep, not past it, so a permanently broken peer does not log +forever. + +### The discovered-nothing warning + +A companion to the above, one level further out. The never-beaten warning covers *"we found a peer and +subscribed to it, but nothing arrives"*. This one covers *"we never found a peer at all"* — a monitor +with `subenabled:1b` that discovers no usable handle for `subscribewarnsweeps` consecutive sweeps +warns once. + +It is deliberately **cause-agnostic**, watching the consequence rather than any single mechanism, so +one check covers all of: + +- `` connections:`ALL `` matching nothing against the current `di.servers` (see above); +- a `connections` list naming a process type `di.servers` is not configured to connect to — the + name-collision trap described earlier; +- an injected `servers` dependency that returns nothing for the configured types; +- every watched peer being genuinely down. + +Two deliberate exclusions. An **explicitly empty** `connections` list does not trip it — that is a +considered "monitor nothing" choice, already warned about at `init`, and repeating it every sweep +would be noise. And it fires **once**, at the threshold, not on every subsequent sweep. + +**Cold start and peer loss are reported differently**, because they have nothing in common but the +symptom: + +| Situation | Message | +|---|---| +| no peer has been discovered **yet** | *"no peer discovered yet, after N consecutive sweeps — on a cold start the peers may simply still be coming up, in which case this resolves itself and a recovery line follows…"* | +| peers were discovered, and are now **all gone** | *"every previously-discovered peer has gone — N consecutive sweeps found none. the peers may be down, or di.servers may have dropped their connections"* | + +**A cold-start warning is always closed out.** When peers become visible again after a warning, the +module logs `peer discovery recovered - now watching N connection(s)` — gated on whether a warning was +actually emitted, so it can never claim to resolve an outage nobody was told about. + +It is logged at **`warn`**, not `info`, even though recovery is good news. The alert fires exactly +once, so a resolution at a lower level would be invisible to anyone whose pipeline filters to warn and +above — they would see a dangling alert and nothing else. Matching the alert's level is what makes a +once-only warning safe to act on. This matters more than it +looks: a staggered rollout that starts the monitor before its peers will legitimately trip the +warning, and an alert that fires once and then goes quiet is indistinguishable from an alert that +fired and was ignored. The recovery line is what makes the warning safe to act on. It is logged once +per outage, not on every healthy sweep. + +**`teardown` forgets everything this check learned.** A monitor re-`init`ed after a teardown is +usually watching a different set of process types, and is cold starting in every sense that matters to +whoever reads the log — so it gets the cold-start wording rather than a claim about peers it never +watched. All three pieces of discovery state reset together; resetting only "have we ever seen a peer" +would be worse than resetting none, because a carried-over sweep count already past a new, lower +threshold never equals it again and the warning would then *silently never fire* for the new +configuration. + +This is the one place `teardown` discards state. The heartbeat store, its counter and the never-beaten +counters are all deliberately preserved — the store by design, and the never-beaten counters because +they are rebuilt from the current discovered set on every sweep and so self-heal anyway. + +If your deployments routinely start monitors well ahead of their peers and you would rather not see +the warning at all, raise `subscribewarnsweeps` — the threshold is in sweeps, so the wall-clock grace +is `subscribewarnsweeps × subscribeinterval` (3 minutes on the defaults). + +Because the check never names a mechanism in its logic, it stays correct once `di.servers` is fixed; +only the `` `ALL `` hint in the cold-start message would become stale. + +**Only the sweep feeds this check, not `subscribe[]`.** The manual entry point takes bare handles with +no identity attached, so there is nothing to key a pending row on — a handle number alone cannot be +matched against the store. An operator subscribing by hand should check `gethb[]` directly. This is +structural, not an oversight: giving `subscribe[]` the same coverage would mean asking the caller for +a `proctype`/`procname` it does not currently have to supply. + +> **Known coupling.** This matches `di.servers`' `procname`/`proctype` against the `sym`/`procname` a +> publisher puts in its own heartbeats. If `process.csv` disagrees with a process's own identity +> config, the warning fires spuriously — which is itself a deployment bug nothing else currently +> catches. If the injected `getservers` returns rows without those columns, the check is skipped +> entirely rather than guessing. **Handle 0 and null handles are refused.** A "remote" call on handle `0i` evaluates *locally*, so subscribing it would register this process as a subscriber to its own heartbeats and then publish to handle 0; a null handle is a disconnected server row. Both are dropped from `subscribe` and from the -`getservers` sweep, with a warning. Legacy guarded the same case by seeding +`getservers` sweep, with a warning. Legacy guarded the same case by seeding its own `subscribedhandles:0 0Ni` (`heartbeat.q:21`) — the guard is deliberate, not incidental. +A **negative** handle is dropped for a sharper reason now that the send is async: `neg` of an +already-negative handle is *positive*, so a negative handle reaching `subscribeone` would silently +become the blocking synchronous call the async switch exists to remove. + > **Security note.** `.heartbeat.subscribe` is callable by anyone holding a handle to this process, > exactly as legacy's `.ps.subscribe` was. It takes no arguments and only subscribes the caller to the > heartbeat table, so the exposure is small — but it is a root-published remote entry point, and a @@ -257,7 +430,7 @@ handle 0; a null handle is a disconnected server row. Both are dropped from `sub | Function | Signature | Description | |---|---|---| | `init` | `[dict]` | wire dependencies and config; idempotent | -| `teardown` | `[]` | release timer jobs, the `.z.pc` registration and the root names | +| `teardown` | `[]` | release the timer jobs and the published root names, and reset the discovery-warning state (the heartbeat store is preserved) | | `version` | - | module version string, read from `VERSION` | | `getapimeta` | `[]` | api metadata rows for `di.torq` to register with `di.api` | | `publishheartbeat` | `[]` | publish one row and bump the counter (timer job) | @@ -327,6 +500,10 @@ choice, not a missed port. ```q upd:{[t;x] if[t=`heartbeat;hb.storeheartbeat x]; ... } ``` + > **Use `:` here, not `::`.** At the top level of a script, `upd::{…}` defines a **view**, not a + > global assignment — `type upd` is then `101h`, and every published row throws `'rank` before your + > handler is reached. Nothing logs, and the publisher looks dead to this monitor. Inside a function + > `::` *is* the correct global assign, which is why `test.q` sets `upd` from within `installupd`. - **No `.servers.connectcustom` wrapper.** Legacy wrapped it to filter auto-connections by `.hb.CONNECTIONS`, mutating the table handed to whatever had registered before it - a cross-feature side effect. `di.servers` owns its own connection strategy now; this module resolves handles through @@ -334,6 +511,11 @@ choice, not a missed port. - **No `.html.pub`.** Legacy's `processwarning`/`processerror` published straight to a dashboard. Those are now the `onwarning`/`onerror` callbacks. `di.html` is `di.monitor`'s dependency, not this module's. +- **The subscribe is async, and there is no subscribed-handle cache.** Legacy did both synchronously + and cached `subscribedhandles`. Both were removed for measured reasons — a stalled peer could block + the monitor's whole timer thread, and a recycled handle number could make it skip a live peer + forever. See *How a monitor subscribes*. The confirmation the sync call used to provide is replaced + by the never-beaten warning. ## Running tests @@ -356,11 +538,21 @@ the monitor's store. ## Notes -- `warningperiod` and `errorperiod` take a process type so a deployment can vary grace periods per - type. The default implementations ignore it, as legacy's did. -- Both timer jobs use **mode 2** (period after the previous *actual* start). A heartbeat asserts +- **Known limitation.** `warningperiod` and `errorperiod` take a process type so a deployment can vary + grace periods per type, as legacy documented — but `init` exposes no way to *supply* a per-type + override, so the extensibility point is not reachable without editing the module. Designing that + config shape is deferred until there is a real requirement to shape it against. +- All three timer jobs use **mode 2** (period after the previous *actual* start). A heartbeat asserts "alive now", so missed beats must not be replayed as a catch-up storm, which mode 1 would do. Periods are converted to whole seconds, which is what `di.timer` expects. +- **Both timer entry points that can throw are isolated.** `hbsubscriptions` runs its whole sweep + through `safecall`, because `di.timer`'s `addjob.opts` defaults `disableonfail:1b` — an unprotected + throw would not skip one sweep, it would end monitor discovery for the life of the process. + `checkheartbeat`'s `cp[]` call is deliberately *not* wrapped: `setcp` probes the clock once at swap + time, so a clock with external dependencies could still throw there. That is an accepted, much + narrower risk than the sweep's — noted in the source so it reads as a decision, not an oversight. +- `test.q`'s `freeport[]` binds a port and immediately releases it, so it carries an inherent + bind-release-reuse race. Known and accepted as an occasional integration-suite flake source. - `pid`, `host` and `port` are captured once at load, matching legacy. A runtime port change is not picked up. - Re-running `init` is safe: it clears its own timer jobs before re-registering, and deliberately diff --git a/di/heartbeat/heartbeat.q b/di/heartbeat/heartbeat.q index 3b71d3f5..298c0614 100644 --- a/di/heartbeat/heartbeat.q +++ b/di/heartbeat/heartbeat.q @@ -5,7 +5,7 @@ / the module handles both publishing heartbeats and, on the monitoring side, storing received / heartbeats and raising warnings / errors when they stop / config and dependencies arrive in a single dictionary passed to init: config keys are optional and -/ fall back to the defaults below; log/timer/pubsub are required (servers/handlers when subenabled) +/ fall back to the defaults below; log/timer/pubsub are required (servers too when subenabled) / and init errors immediately if a required dependency is missing / module-local state convention: constants are bare top-level names, mutable state lives in .z.m, and / injected dependencies are read through .z.m at every call site @@ -28,21 +28,30 @@ schema:( / keyed store of the latest received heartbeat per process, with warning / error state storeschema:update warning:0b,error:0b from `sym`procname xkey schema; +/ keyed count of consecutive sweeps a discovered peer has gone without producing a heartbeat +unseenschema:2!([]sym:`symbol$();procname:`symbol$();sweeps:`long$()); + / the table name published over pub/sub, and the root name di.pubsub discovers it under tablename:`heartbeat; -/ the dependency keys of the single init dict - everything else in that dict is config +/ the dependency keys of the single init dict - everything else in that dict is config. +/ NB handlers stays in this list only to absorb di.torq's uniform wiring: nothing in this module uses +/ it any more (the .z.pc subscription cache it existed for was removed). do not "clean it up" - +/ dropping it makes resolveconfig warn about handlers on every publisher-only boot, which is the +/ common case since subenabled defaults off depkeys:`log`timer`pubsub`servers`handlers; / config defaults. proctype and procname are deliberately absent - they are self-identity and are / required, not defaulted, matching di.servers. legacy captured pid/host/port once at load time and / so do we, so a runtime port change is not picked up configdefaults:( - `enabled`subenabled`debug`publishroot`publishinterval`checkinterval`warningtolerance`errortolerance, - `maxage`pid`host`port`connections`onwarning`onerror + `enabled`subenabled`debug`publishroot`publishinterval`checkinterval`subscribeinterval, + `warningtolerance`errortolerance`subscribewarnsweeps`maxage, + `pid`host`port`connections`onwarning`onerror )!( - 1b;0b;1b;1b;0D00:00:30;0D00:00:10;2f;3f; - 0D24:00:00;.z.i;.z.h;`int$system"p";`ALL;{[procs]};{[procs]} + 1b;0b;1b;1b;0D00:00:30;0D00:00:10;0D00:01:00; + 2f;3f;3;0D24:00:00; + .z.i;.z.h;`int$system"p";`ALL;{[procs]};{[procs]} ); / timer job ids this module owns - deleted before re-registering so init is safe to call again @@ -117,10 +126,12 @@ safecall:{[ctx;nm;f;arg] validhandles:{[handles] / drop nulls (a dead server row) and handle 0. a "remote" call on handle 0 evaluates LOCALLY, so / subscribing it registers this process as a subscriber to its own heartbeats and then publishes - / to handle 0. legacy guarded exactly this by seeding subscribedhandles with 0 0Ni (heartbeat.q:21) - / a negative handle is an ASYNC handle in q: the subscribe call would return immediately without - / confirming anything, so it would be recorded as subscribed on no evidence - and closeconnection - / matches against .z.w, which is always positive, so the row could never be cleaned up either + / to handle 0. legacy guarded exactly this by seeding its subscribedhandles with 0 0Ni + / (heartbeat.q:21) + / a negative handle is an ASYNC handle in q, and this guard is LOAD-BEARING rather than defensive + / now that subscribeone sends with (neg h): neg of an already-negative handle is POSITIVE, so a + / negative handle slipping through here would silently become the blocking synchronous call the + / async switch exists to remove h:(),handles; :h where not (null h) or 0i>=h; }; @@ -130,8 +141,8 @@ validhandles:{[handles] / ============================================================ validatedeps:{[deps] - / log, timer and pubsub are always required; servers and handlers only when this process monitors - / others. nested if guards rather than and - and evaluates both sides eagerly, so key would be + / log, timer and pubsub are always required; servers only when this process monitors others. + / nested if guards rather than and - and evaluates both sides eagerly, so key would be / reached on a non-dict. no dependency is ever silently defaulted if[99h<>type deps; '"di.heartbeat: deps must be a dict of injectables + config - see di.log, di.timer, di.pubsub"]; @@ -166,17 +177,15 @@ validatedeps:{[deps] }; validatemonitordeps:{[deps] - / servers and handlers are required only when subenabled - a publisher-only process needs neither. - / split out so the conditional in init stays a single statement, per the style guide + / servers is required only when subenabled - a publisher-only process does not need it. split out + / so the conditional in init stays a single statement, per the style guide. + / NB no handlers requirement: this module registered a .z.pc observer solely to prune a cache of + / subscribed handles, and that cache is gone. a handlers dict passed anyway is silently accepted + / (see depkeys) so di.torq's uniform wiring keeps working unchanged if[99h<>type deps`servers; '"di.heartbeat: subenabled is set, so a servers dependency is required (see di.servers)"]; if[not `getservers in key deps`servers; '"di.heartbeat: servers dict must expose `getservers (see di.servers)"]; - if[99h<>type deps`handlers; - '"di.heartbeat: subenabled is set, so a handlers dependency is required (see di.handlers)"]; - if[not all `register`remove in key deps`handlers; - '"di.heartbeat: handlers dict must have `register`remove keys; got: ", - (", " sv string key deps`handlers)]; }; resolveconfig:{[deps] @@ -192,14 +201,21 @@ validateconfig:{[cfg] / catch config that would leave the module quietly doing nothing rather than failing loudly if[not all -1h=type each cfg`enabled`subenabled`debug`publishroot; raiseerror[`init;"enabled, subenabled, debug and publishroot must be booleans"]]; - if[not all -16h=type each cfg`publishinterval`checkinterval; - raiseerror[`init;"publishinterval and checkinterval must be timespans"]]; + if[not all -16h=type each cfg`publishinterval`checkinterval`subscribeinterval; + raiseerror[`init;"publishinterval, checkinterval and subscribeinterval must be timespans"]]; / di.timer schedules in whole SECONDS and tosecs rounds, so anything under half a second becomes a / period of 0 - a job that then runs on every timer cycle. reject the whole sub-second range rather / than silently accept a schedule that cannot be represented - if[any 0D00:00:01>cfg`publishinterval`checkinterval; - raiseerror[`init;"publishinterval and checkinterval must be at least one second - di.timer ", - "schedules in whole seconds, so a shorter interval cannot be represented"]]; + if[any 0D00:00:01>cfg`publishinterval`checkinterval`subscribeinterval; + raiseerror[`init;"publishinterval, checkinterval and subscribeinterval must be at least one ", + "second - di.timer schedules in whole seconds, so a shorter interval cannot be represented"]]; + / the never-beaten warning counts sweeps, so a non-positive threshold would either fire on a peer + / that has had no chance to answer yet, or (at 0) never match the post-increment count at all + if[not -7h=type cfg`subscribewarnsweeps; + raiseerror[`init;"subscribewarnsweeps must be a long"]]; + if[1>cfg`subscribewarnsweeps; + raiseerror[`init;"subscribewarnsweeps must be at least 1 - it counts consecutive sweeps a ", + "discovered peer has gone without heartbeating, and the count starts at 1"]]; if[not all -9h=type each cfg`warningtolerance`errortolerance; raiseerror[`init;"warningtolerance and errortolerance must be floats"]]; / a zero or negative tolerance makes the grace period zero or negative, so now>time+period is true @@ -237,7 +253,7 @@ validateroot:{[cfg] / check the root name is available BEFORE init writes any state. claiming it is the only step that / can fail on something outside this module, and letting installroot throw part-way through init / would leave the module marked initialised, with config written and deps wired, but with no - / timers, no handlers and no root table - so publishheartbeat would happily run and publish into a + / timers and no root table - so publishheartbeat would happily run and publish into a / topic di.pubsub cannot serve. exactly the silent failure this module exists to avoid if[not cfg`publishroot;:()]; if[not tablename in tables[];:()]; @@ -312,7 +328,7 @@ uninstallroot:{[] }; / ============================================================ -/ wiring - timers and handlers +/ wiring - timers / ============================================================ registertimers:{[] @@ -324,22 +340,7 @@ registertimers:{[] .z.m.timeraddjob[`custom][`hbpublish;publishheartbeat;();tosecs publishinterval;2;()!()]; .z.m.timeraddjob[`custom][`hbcheck;checkheartbeat;();tosecs checkinterval;2;()!()]]; if[subenabled; - .z.m.timeraddjob[`custom][`hbsubscribe;hbsubscriptions;();60;2;()!()]]; - }; - -registerhandlers:{[] - / .z.pc is a SIMPLE event in di.handlers - side-effect only, return value discarded, so any number - / of registrants coexist and the phase must be ` (null). register is 5-arg [event;phase;nm;pri;func] - / drop any registration from an earlier init FIRST and unconditionally, mirroring registertimers. - / a re-init that turns subenabled off would otherwise orphan the observer permanently: teardown - / keys off whether we are registered, but before this it keyed off the CURRENT subenabled, which is - / now false - so nothing could ever remove it - if[handlerregistered; - .z.m.handlersremove[`.z.pc;`;`heartbeat]; - .z.m.handlerregistered:0b]; - if[subenabled; - .z.m.handlersregister[`.z.pc;`;`heartbeat;0j;closeconnection]; - .z.m.handlerregistered:1b]; + .z.m.timeraddjob[`custom][`hbsubscribe;hbsubscriptions;();tosecs subscribeinterval;2;()!()]]; }; / ============================================================ @@ -348,34 +349,159 @@ registerhandlers:{[] subscribeone:{[h] / ask the REMOTE publisher to subscribe us. calling di.pubsub.subscribe locally cannot work - it - / reads the caller's own .z.w and so can only ever subscribe the caller. a failed subscribe is not - / recorded, so the next tick retries it + / reads the caller's own .z.w and so can only ever subscribe the caller. + / sent ASYNC: .z.w still resolves to the monitor's connection during an async inbound call, so the + / handshake is unaffected, and an async send to a hung peer returns in microseconds where the sync + / call blocked for the FULL duration of the hang. that mattered because this runs inside a di.timer + / job on a single thread - a peer that is merely slow (a gc pause, a heavy query) stalled + / publishheartbeat and checkheartbeat with it, so the monitor went quiet to ITS monitors. + / a LOCAL failure (a dead or null handle) still throws here and is still logged; a REMOTE failure is + / invisible to an async send, which is what trackunseen exists to catch / the error handler's first parameter is named hdl, not h: the projection [h] supplies the outer / handle either way, but reusing the name would shadow it and a later refactor could silently bind / the wrong value (raised on PR #109 and worth keeping fixed) - ok:@[{[h] h(`.heartbeat.subscribe;::);1b};h; - {[hdl;e] .z.m.logerr[`subscribeone;"failed to subscribe to heartbeats on handle ",(string hdl),": ",e];0b}[h]]; - if[ok;.z.m.subscribedhandles:distinct subscribedhandles,h]; + @[{[h] (neg h)(`.heartbeat.subscribe;::)};h; + {[hdl;e] .z.m.logerr[`subscribeone;"failed to send heartbeat subscribe on handle ", + (string hdl),": ",e]}[h]]; + }; + +trackunseen:{[rows] + / an async subscribe cannot report a REMOTE failure (a root-name collision on the publisher, a + / di.pubsub that has not initialised there, a refused call), so watch for the consequence instead: + / a peer we have subscribed to that never produces a beat. a null counter means an addprocs seed + / rather than an observation - the same discriminator evictstale uses. warned once, at exactly the + / threshold sweep, so a permanently broken peer does not log forever. + / NB the count is rebuilt from the CURRENT discovered set each sweep, so a peer that drops out of + / di.servers' view and comes back starts again from 1. that is deliberate: a connectivity blip is + / not the continuously-stuck subscribe this warning exists to catch, and only an uninterrupted run + / of pending sweeps should trip it + if[not all `proctype`procname in cols rows;:()]; + / the threshold is hoisted into a LOCAL first: a qsql where clause inside module code cannot resolve + / a module-level name and throws 'subscribewarnsweeps. same reason checkheartbeat hoists wp and ep + n:subscribewarnsweeps; + live:select sym:proctype,procname from rows; + observed:select sym,procname from 0!hb where not null counter; + pending:select sym,procname from live where not ([]sym;procname) in observed; + carried:select from 0!unseen where ([]sym;procname) in pending; + fresh:select sym,procname,sweeps:0 from pending where not ([]sym;procname) in `sym`procname#carried; + .z.m.unseen:2!update sweeps:sweeps+1 from carried,fresh; + due:select from 0!unseen where sweeps=n; + / NB "no heartbeat", not "never heartbeated" - a peer whose row was evicted by maxage re-enters this + / set, and it HAS beaten before, just not recently. the alert is still correct; the wording must be + if[count due; + .z.m.logwarn[`hbsubscriptions;"subscribed to ",(string count due)," peer(s) with no heartbeat in ", + (string n)," sweeps - the remote subscribe may have failed (an async subscribe cannot report a ", + "remote error): ",", " sv string exec procname from due]]; + }; + +resetdiscovery:{[] + / internal - forget what this monitoring session learned about peer discovery. all three vars move + / together because they are ONE mechanism, and resetting only everdiscovered would be worse than + / resetting none: a carried-over emptysweeps already past a new, lower threshold never equals it + / again, so the discovered-nothing warning would silently never fire for the new configuration - + / and a carried-over emptywarned would let the next recovery close out an outage from the old one. + / NB the heartbeat store, its counter and unseen are deliberately NOT reset here: the store persists + / across re-init by design, and unseen is rebuilt from the current discovered set on every sweep, so + / it self-heals where these three do not + .z.m.everdiscovered:0b; + .z.m.emptysweeps:0; + .z.m.emptywarned:0b; + }; + +emptysweepmsg:{[] + / internal - the two ways of discovering nothing have completely different diagnoses, so the message + / says which one this is. a monitor that has NEVER seen a peer is usually still coming up: di.servers + / connects on its own retry cycle, which in a staggered rollout can easily outlast a few sweeps. one + / that HAD peers and now has none has lost them - not a startup condition at all, and it reads very + / differently to whoever is on call + if[everdiscovered; + :"every previously-discovered peer has gone - ",(string subscribewarnsweeps)," consecutive ", + "sweeps found none. the peers may be down, or di.servers may have dropped their connections"]; + :"no peer discovered yet, after ",(string subscribewarnsweeps)," consecutive sweeps - on a cold ", + "start the peers may simply still be coming up, in which case this resolves itself and a ", + "recovery line follows. otherwise check that connections names process types di.servers is ", + "connected to. NB `ALL currently matches nothing against di.servers (see heartbeat.md)"; + }; + +warnemptysweeps:{[] + / internal - count consecutive sweeps that discovered no usable peer at all, and warn ONCE at the + / threshold. deliberately cause-AGNOSTIC: it watches the consequence, not any single cause, so it + / stays correct whatever the reason - a connections list naming a process type di.servers never + / connects to, a servers dependency returning nothing, or peers that are simply all down. that also + / means it needs no maintenance when the `ALL gap below is closed. reuses subscribewarnsweeps: it + / is the same "consecutive sweeps" unit as the never-beaten warning + .z.m.emptysweeps:emptysweeps+1; + if[emptysweeps<>subscribewarnsweeps;:()]; + / record that a warning was actually EMITTED, rather than leaving the recovery line to re-derive it + / from the counter. a re-init may change subscribewarnsweeps, and comparing a carried-over count + / against a new threshold claimed a recovery for an outage that had never been reported + .z.m.emptywarned:1b; + .z.m.logwarn[`hbsubscriptions;emptysweepmsg[]]; + }; + +notediscovery:{[n] + / internal - peers are visible again. if we had already warned, close the loop with an info line: + / a cold start that merely took longer than the threshold would otherwise leave a lone warning in + / the log with nothing to say it resolved, which is the failure mode of every "it warned once and + / then went quiet" alert. the recovery line is what makes the warning safe to act on + / gated on whether a warning was emitted, NOT on the sweep count against the current threshold - + / so a recovery line can never appear for an outage nobody was told about. + / NB "connection(s)": n is the count of distinct usable HANDLES, which is what was actually + / subscribed. it is not necessarily the number of server rows, since handles are deduped + / logged at WARN, not info, even though recovery is good news: the discovered-nothing warning fires + / exactly ONCE, so if its resolution went to a lower level then anyone whose pipeline filters to + / warn and above would see the alert and never the close-out - which is the dangling-alert problem + / this line exists to remove. matching the alert's level is what makes a once-only warning safe + if[emptywarned; + .z.m.logwarn[`hbsubscriptions;"peer discovery recovered - now watching ",(string n), + " connection(s)"]; + .z.m.emptywarned:0b]; + .z.m.everdiscovered:1b; + .z.m.emptysweeps:0; }; getheartbeats:{[proctypes] / di.servers.getservers returns a TABLE of server rows - the handles are its w column, and a - / disconnected row carries a null handle - handles:validhandles exec w from .z.m.serversgetservers proctypes; - handles:handles except subscribedhandles; + / disconnected row carries a null handle. + / ONE CALL PER PROCTYPE: di.servers.getservers takes a symbol ATOM and matches with =, so passing + / the whole list throws - and di.timer's disableonfail would then end monitor discovery for the + / process lifetime. forward-compatible either way: a one-element iteration still works if getservers + / later accepts vectors + pts:(),proctypes; + / an empty connections list is a LEGAL "monitor nothing" configuration - init warns about it rather + / than rejecting it - so it must be a clean no-op here. without this guard `each` over no proctypes + / razes to a general empty list, which exec cannot read: it threw 'type on every sweep. clear the + / never-beaten counters too, since nothing is being monitored to be pending about + if[0=count pts; + .z.m.unseen:unseenschema; + :()]; + rows:raze {[pt] .z.m.serversgetservers pt} each pts; + / distinct across the per-proctype results: a well-behaved servers impl puts each row under exactly + / one proctype, but a repeated entry in connections would otherwise send the same peer two subscribes + handles:distinct validhandles exec w from rows; + / no tracked handle set: di.pubsub dedupes subscribers by .z.w, so a repeat subscribe is a no-op on + / the publisher and an async send costs nothing worth caching around. tracking handles here was + / actively harmful - a handle number is not an identity, kdb+ reissues the lowest free descriptor, + / and hclose does not fire .z.pc, so a stale entry made the monitor skip a live peer forever if[count handles; - .z.m.loginfo[`getheartbeats;"subscribing to new heartbeat handle(s) ",", " sv string handles]; + notediscovery count handles; + .z.m.loginfo[`getheartbeats;"subscribing to heartbeat handle(s) ",", " sv string handles]; subscribeone each handles]; + / a monitor configured to watch something that nonetheless discovers NOTHING is the exact + / looks-configured-but-silent failure this module exists to eliminate - say so rather than sit quiet + if[0=count handles;warnemptysweeps[]]; + / only peers we ACTUALLY attempted are candidates for the never-beaten warning. a row carrying a + / null handle (a disconnected server) or handle 0 was deliberately skipped, so warning that it never + / heartbeated would be a false alarm about a peer we never asked + trackunseen[select from rows where w in handles]; }; hbsubscriptions:{[] - / timer job - pick up any newly connected publisher of a configured process type - getheartbeats resolveconnections[]; - }; - -closeconnection:{[h] - / drop a closed handle from the tracked subscriptions - registered against .z.pc - .z.m.subscribedhandles:subscribedhandles except h; + / timer job - pick up any newly connected publisher of a configured process type. isolated for the + / same reason the publish call and the callbacks are: di.timer's addjob.opts defaults + / disableonfail:1b, so an unprotected throw does not skip one sweep, it ends monitor discovery for + / the lifetime of the process + safecall[`hbsubscriptions;`sweep;getheartbeats;resolveconnections[]]; }; / ============================================================ @@ -466,6 +592,9 @@ checkheartbeat:{[] / status: 0 healthy, 1 warning, 2+ error. grace periods are computed as locals first, since module / functions do not resolve inside qsql requireinit[`checkheartbeat]; + / NB cp[] is deliberately outside any safecall. setcp probes the clock once at swap time, so a + / clock with external dependencies could still throw here and take this job out under di.timer's + / disableonfail. accepted as a much narrower risk than the sweep's (see hbsubscriptions), not missed now:cp[]; / evict first: maxage is validated to exceed the error period, so nothing can be evicted before it / has already been through its error transition @@ -556,7 +685,8 @@ removeprocs:{[proctypes;procnames] }; subscribe:{[handles] - / subscribe to heartbeats on the given remote handle(s), tracking successful subscriptions + / subscribe to heartbeats on the given remote handle(s). the send is async, so a failure on the + / REMOTE side is not reported here - the sweep's never-beaten warning is what surfaces that requireinit[`subscribe]; h:(),handles; if[not type[h] within 5 7h; @@ -595,26 +725,28 @@ setcp:{[f] }; teardown:{[] - / release everything init installed: timer jobs, the .z.pc registration and the root names + / release everything init installed: the timer jobs and the root names requireinit[`teardown]; .z.m.timerdeletejobs jobids; / keyed off what is actually installed, NOT off the current config - a re-init that flipped - / subenabled or publishroot off would otherwise leave teardown unable to clean up its own residue - if[handlerregistered; - .z.m.handlersremove[`.z.pc;`;`heartbeat]; - .z.m.handlerregistered:0b]; + / publishroot off would otherwise leave teardown unable to clean up its own residue if[rootinstalled;uninstallroot[]]; .z.m.enabled:0b; .z.m.subenabled:0b; - .z.m.loginfo[`teardown;"di.heartbeat torn down - timers, handlers and root names released"]; + / a monitor re-inited after teardown is typically watching a DIFFERENT set of process types, and is + / cold starting in every sense that matters to whoever reads the log - so it must get the cold-start + / wording, not "every previously-discovered peer has gone" + resetdiscovery[]; + .z.m.loginfo[`teardown;"di.heartbeat torn down - timer jobs and root names released"]; }; init:{[deps] - / wire the injected deps and this process's config from ONE dict, then install the timer jobs, the - / .z.pc observer (when monitoring) and the root names (when publishroot). idempotent - a second - / call clears its own timer jobs first and re-registers, leaving the heartbeat store intact - / deps: `log`timer`pubsub (required), `servers`handlers (required when subenabled), - / `proctype`procname (required identity), plus any config key alongside them + / wire the injected deps and this process's config from ONE dict, then install the timer jobs and + / the root names (when publishroot). idempotent - a second call clears its own timer jobs first and + / re-registers, leaving the heartbeat store intact + / deps: `log`timer`pubsub (required), `servers` (required when subenabled), + / `proctype`procname (required identity), plus any config key alongside them. + / a `handlers` key is accepted and ignored, so di.torq's uniform wiring needs no special case / e.g. hb.init[(`log`timer`pubsub!(logdep;timerdep;psdep)),`proctype`procname!(`rdb;`rdb1)] validatedeps[deps]; .z.m.loginfo:(deps`log)`info; @@ -630,17 +762,15 @@ init:{[deps] .z.m.pubsubpublish:(deps`pubsub)`publish; .z.m.pubsubsubscribe:(deps`pubsub)`subscribe; if[cfg`subenabled; - .z.m.serversgetservers:(deps`servers)`getservers; - .z.m.handlersregister:(deps`handlers)`register; - .z.m.handlersremove:(deps`handlers)`remove]; + .z.m.serversgetservers:(deps`servers)`getservers]; / first init only - a re-init must not discard heartbeats already received if[not initialised[]; .z.m.hb:storeschema; .z.m.ownhb:0#schema; - .z.m.subscribedhandles:`int$(); + .z.m.unseen:unseenschema; + resetdiscovery[]; .z.m.rootowned:0b; .z.m.rootinstalled:0b; - .z.m.handlerregistered:0b; .z.m.warnedcols:`symbol$(); .z.m.hbcounter:0]; .z.m.config:cfg; @@ -654,8 +784,10 @@ init:{[deps] .z.m.publishroot:cfg`publishroot; .z.m.publishinterval:cfg`publishinterval; .z.m.checkinterval:cfg`checkinterval; + .z.m.subscribeinterval:cfg`subscribeinterval; .z.m.warningtolerance:cfg`warningtolerance; .z.m.errortolerance:cfg`errortolerance; + .z.m.subscribewarnsweeps:cfg`subscribewarnsweeps; .z.m.maxage:cfg`maxage; .z.m.pid:cfg`pid; .z.m.host:cfg`host; @@ -671,7 +803,6 @@ init:{[deps] .z.m.loginfo[`init;"publishroot is 0b - nothing published at root and no pub/sub publishing; ", "own heartbeats are tracked locally and readable via getownhb"]]; registertimers[]; - registerhandlers[]; / NB the index expressions are parenthesised deliberately. juxtaposition binds to the WHOLE / right-hand expression, so ("disabled";"enabled")enabled,", monitoring ",... parses as / ("disabled";"enabled")[enabled,", monitoring ",...] - indexing by the rest of the string, which @@ -686,7 +817,7 @@ getapimeta:{[] / one row per CALLABLE api function, for di.torq to register with di.api. init and getapimeta are / plumbing di.torq calls by convention and are deliberately omitted. names are bare :flip `name`public`descrip`params`return!flip( - (`teardown; 1b; "release timer jobs, the .z.pc registration and the published root names"; + (`teardown; 1b; "release the timer jobs and the published root names"; "[]"; "null"); (`version; 1b; "module version string"; "[]"; "string: version"); diff --git a/di/heartbeat/test.csv b/di/heartbeat/test.csv index 70973f05..819483af 100644 --- a/di/heartbeat/test.csv +++ b/di/heartbeat/test.csv @@ -11,7 +11,7 @@ before,0,0,q,.pt.reg:([]event:`symbol$();phase:`symbol$();nm:`symbol$();pri:`lon before,0,0,q,"mockh:`register`remove!({[e;p;n;pr;f] `.pt.reg insert (e;p;n;pr)};{[e;p;n] delete from `.pt.reg where event=e,phase=p,nm=n})",1,1,handlers mock - register is 5-arg before,0,0,q,"mocksrv:enlist[`getservers]!enlist {[p] ([]w:`int$())}",1,1,servers mock returning an empty server TABLE before,0,0,q,.pt.asked:(),1,1,records which proctype getservers was asked for -before,0,0,q,"mocksrvrows:enlist[`getservers]!enlist {[p] .pt.asked,:enlist p; ([]proctype:`rdb`hdb`dead`self;w:4 5 0Ni,0i)}",1,1,servers mock returning a POPULATED table - live handles plus a dead (null) row and handle 0 +before,0,0,q,"mocksrvrows:enlist[`getservers]!enlist {[p] .pt.asked,:enlist p; t:([]proctype:`rdb`hdb`dead`self;procname:`rdb1`hdb1`dead1`self1;w:4 5 0Ni,0i); $[null p;t;select from t where proctype=p]}",1,1,servers mock returning a POPULATED table - live handles plus a dead (null) row and handle 0. FILTERS BY ITS ARGUMENT as real di.servers.getservers does - a mock that ignores its input cannot catch a caller passing the wrong shape before,0,0,q,"deps:(`log`timer`pubsub!(caplog;mocktimer;mockps)),`proctype`procname!(`rdb;`rdb1)",1,1,the base deps dict - deps plus required identity comment,,,,,,,init must be called before any other function (asserted FIRST - the module is a singleton and cannot be un-initialised later) @@ -36,8 +36,13 @@ true,0,0,q,"0elapsed[hbm[`subscribe];handle],1,1,subscribing to a stalled peer returns immediately instead of blocking for the full hang run,0,0,q,cleanup[],1,1,close the handle and remove the child script and log comment,,,,,,,di.torq-shaped wiring - REAL di.log/di.timer/di.handlers/di.pubsub instead of mocks @@ -48,18 +53,26 @@ run,0,0,q,realtimer:use`di.timer,1,1,real di.timer run,0,0,q,realhandlers:use`di.handlers,1,1,real di.handlers run,0,0,q,realps:use`di.pubsub,1,1,real di.pubsub run,0,0,q,realtimer[`init][],1,1,init the real timer +run,0,0,q,.ht.now:.z.p,1,1,a simulated clock that BOTH di.timer and di.heartbeat will share +run,0,0,q,realtimer[`setcp][{.ht.now}],1,1,point the REAL scheduler at it BEFORE any job is added so each nextstart is computed from it run,0,0,q,realhandlers[`init][logging[`logdict]],1,1,init real di.handlers off di.log's ready-made logdict run,0,0,q,"realdeps:logging[`logdict],`timer`pubsub`handlers`servers`proctype`procname`subenabled!(realtimer;realps;realhandlers;enlist[`getservers]!enlist {[p]([]w:`int$())};`rdb;`rdb1;1b)",1,1,wire with real module export dicts exactly as di.torq will run,0,0,q,hbm[`init][realdeps],1,1,init against the REAL logger - a non-flat message throws here where a permissive mock accepted it -true,0,0,q,"1=count select from realhandlers[`list][`.z.pc] where name=`heartbeat",1,1,the registration landed in the REAL di.handlers registry -true,0,0,q,"null first exec phase from realhandlers[`list][`.z.pc] where name=`heartbeat",1,1,and real di.handlers agrees .z.pc is a simple event with a null phase +true,0,0,q,"0=count select from realhandlers[`list][`.z.pc] where name=`heartbeat",1,1,nothing is registered on .z.pc even with a real di.handlers available - the cache that observer maintained was removed true,0,0,q,3=count select from realtimer[`getalljobs][] where id in `hbpublish`hbcheck`hbsubscribe,1,1,all three jobs landed in the REAL di.timer true,0,0,q,"all 2h=exec mode from realtimer[`getalljobs][] where id in `hbpublish`hbcheck",1,1,real di.timer recorded mode 2 as intended true,0,0,q,`heartbeat in tables[],1,1,root schema table published run,0,0,q,realps[`init][],1,1,pubsub init AFTER heartbeat init - the documented ordering true,0,0,q,`heartbeat in .m.di.0pubsub.t,1,1,di.pubsub now serves the heartbeat topic - the ordering constraint holds end to end run,0,0,q,hbm[`publishheartbeat][],1,1,publish through the real pubsub with no subscribers attached + +comment,,,,,,,"the REAL scheduler must actually FIRE the jobs - asserting they were registered is a different claim, and di.handlers has already shown a registry can report a handler that is no longer live" +run,0,0,q,hbm[`setcp][{.ht.now}],1,1,point di.heartbeat's own clock at the same simulated now +run,0,0,q,.ht.c0:.m.di.0heartbeat.hbcounter,1,1,record the beat counter before stepping the clock +run,0,0,q,.ht.now+:0D00:00:31,1,1,advance past the 30s publishinterval +run,0,0,q,.z.ts 0,1,1,drive ONE scheduler cycle by hand - a system sleep would block q's event loop so .z.ts would never fire at all +true,0,0,q,.m.di.0heartbeat.hbcounter>.ht.c0,1,1,the REAL di.timer invoked publishheartbeat unattended - the job is live not merely registered +true,0,0,q,all exec status from realtimer[`getalljobs][],1,1,and no job was disabled - di.timer's disableonfail did not trip on any of them run,0,0,q,hbm[`teardown][],1,1,teardown against the real registries -true,0,0,q,0=count realhandlers[`list][`.z.pc],1,1,the real handlers registry was cleaned true,0,0,q,0=count realtimer[`getalljobs][],1,1,the real timer jobs were cleaned true,0,0,q,not `heartbeat in tables[],1,1,the root names were cleaned From d2115a3fa9ea3a356bb7af8419c39f09ebce2e43 Mon Sep 17 00:00:00 2001 From: alowrydi Date: Tue, 11 Aug 2026 16:54:48 +0100 Subject: [PATCH 3/5] di.heartbeat: document the empty-connections warning bypass and publisher-side subscriber cleanup --- di/heartbeat/heartbeat.md | 10 +++++++++- di/heartbeat/heartbeat.q | 4 +++- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/di/heartbeat/heartbeat.md b/di/heartbeat/heartbeat.md index f31a7c77..3d2d0d7c 100644 --- a/di/heartbeat/heartbeat.md +++ b/di/heartbeat/heartbeat.md @@ -126,7 +126,7 @@ Setting `subenabled:1b` with an **empty** `connections` list is legal but warns: as legacy's in-file `connections:()` default, where the entire monitor path silently did nothing. Use `` `ALL ``, or name the process types to watch. -### ⚠️ `` `ALL `` does not currently work - a regression in `di.servers`, tracked +### ⚠️ `` `ALL `` does not currently work - a regression in `di.servers`, not yet raised The shipped default resolves to a null symbol, which every ancestor of `getservers` treats as match-all. `di.servers`' version does not: @@ -319,6 +319,14 @@ Nothing is lost by dropping the cache: `di.pubsub` already dedupes subscribers b to be worth caching around. Removing it also removed this module's only reason to register a `.z.pc` handler, and with it the `handlers` dependency. +**What cleans up, then?** The publisher does. `di.pubsub` registers its own `.z.pc` (`pubsub.q:75`) +which calls `closesub` to drop a dead subscriber's handle from `reqalldict` and `reqfilteredtbl`. +Subscriber lifetime is the publisher's concern, not the monitor's — which is why losing the +monitor-side cache costs nothing. A monitor that dies is forgotten by every publisher it was watching, +without this module tracking anything. (`di.pubsub` assigns `.z.pc` flatly rather than through +`di.handlers`, so on a shared process it can clobber other modules' cleanup — but not its own: it is +the last assignment to win, so subscriber cleanup stays reliable regardless.) + ### The never-beaten warning Since an async send cannot report a *remote* failure, the sweep watches for the consequence instead: diff --git a/di/heartbeat/heartbeat.q b/di/heartbeat/heartbeat.q index 298c0614..ad73f4ba 100644 --- a/di/heartbeat/heartbeat.q +++ b/di/heartbeat/heartbeat.q @@ -471,7 +471,9 @@ getheartbeats:{[proctypes] / an empty connections list is a LEGAL "monitor nothing" configuration - init warns about it rather / than rejecting it - so it must be a clean no-op here. without this guard `each` over no proctypes / razes to a general empty list, which exec cannot read: it threw 'type on every sweep. clear the - / never-beaten counters too, since nothing is being monitored to be pending about + / never-beaten counters too, since nothing is being monitored to be pending about. + / NB returning here also bypasses warnemptysweeps, deliberately: discovering nothing is exactly what + / was asked for, and it was already warned about once at init - repeating it every sweep is noise if[0=count pts; .z.m.unseen:unseenschema; :()]; From a2b07ed9d1c3351975c7168609e2b13870cdf7a4 Mon Sep 17 00:00:00 2001 From: alowrydi Date: Tue, 11 Aug 2026 20:06:33 +0100 Subject: [PATCH 4/5] remove obsolete ALL discovery caveat now fix it applied in di.servers --- di/heartbeat/heartbeat.md | 44 +++++++++++++++------------------------ di/heartbeat/heartbeat.q | 16 +++++++------- di/heartbeat/test.csv | 6 +++--- 3 files changed, 29 insertions(+), 37 deletions(-) diff --git a/di/heartbeat/heartbeat.md b/di/heartbeat/heartbeat.md index 3d2d0d7c..02f4b187 100644 --- a/di/heartbeat/heartbeat.md +++ b/di/heartbeat/heartbeat.md @@ -99,7 +99,7 @@ Passed in the same dict as the dependencies. Every key is optional. | `errortolerance` | `3f` | **shipped** | error after `tolerance * publishinterval` | | `subscribewarnsweeps` | `3` | new | consecutive-sweep threshold for both monitor warnings: a subscribed peer sending nothing, and discovering no peer at all | | `maxage` | `0D24:00:00` | new | forget a process silent this long; `0Wn` to keep forever | -| `connections` | `` `ALL `` | **shipped** | process types to monitor; `` `ALL `` means every one. **See the caveat below - `` `ALL `` does not currently work** | +| `connections` | `` `ALL `` | **shipped** | process types to monitor; `` `ALL `` means every one | | `onwarning` | no-op | new | unary callback given the rows entering warning | | `onerror` | no-op | new | unary callback given the rows entering error | | `pid` / `host` / `port` | `.z.i` / `.z.h` / `system"p"` | legacy | captured once at load, as legacy did | @@ -126,32 +126,22 @@ Setting `subenabled:1b` with an **empty** `connections` list is legal but warns: as legacy's in-file `connections:()` default, where the entire monitor path silently did nothing. Use `` `ALL ``, or name the process types to watch. -### ⚠️ `` `ALL `` does not currently work - a regression in `di.servers`, not yet raised +### How `` `ALL `` resolves, and why the sweep still iterates -The shipped default resolves to a null symbol, which every ancestor of `getservers` treats as -match-all. `di.servers`' version does not: +`` `ALL `` converts to a null symbol, which `di.servers.getservers` treats as match-all — the same +contract legacy TorQ's `.servers.getservers` (`trackservers.q:75`) and `di.serverselect.getservers` +both implement (`` ` `` → every server, otherwise `proctype in lookups`). -| Implementation | Contract | -|---|---| -| legacy TorQ `.servers.getservers` (`trackservers.q:75`) | `` `~lookups `` → every server; otherwise `proctype in lookups` (so a **list** works) | -| `di.serverselect.getservers` | identical - `` ` `` as `lookups` returns all active servers | -| **`di.servers.getservers`** | requires a symbol **atom** and matches `proctype=pt` - a null matches **nothing**, a list **throws** | - -So this is a regression against both its ancestors, not a missing feature. Consequences today: - -- **`` connections:`ALL ``** (the default) discovers zero servers, silently. -- **`` connections:`rdb`hdb ``** works, because `di.heartbeat` calls `getservers` **once per - proctype** rather than passing the list. That iteration is this module's workaround and is - forward-compatible - a one-element iteration still works if `getservers` later accepts vectors. - -**Until `di.servers` is fixed, name the process types explicitly.** The `` `ALL `` path cannot be -worked around from this side: there is no proctype universe to iterate over without `di.servers` -providing one. +The sweep nonetheless calls `getservers` **once per proctype** rather than passing the whole list. +That is deliberate: it is the one shape every version of the contract accepts, so this module works +against a `di.servers` that predates the list/null support as well as one that has it. The cost is +one call per configured proctype per sweep, which is not worth optimising away for the coupling it +would add. -It is, however, no longer **silent**. A monitor that is configured to watch something and discovers -no usable peer for `subscribewarnsweeps` consecutive sweeps warns once, naming `` `ALL `` as the -likeliest cause. See *The discovered-nothing warning* below — that check is deliberately -cause-agnostic, so it needs no maintenance when `di.servers` is fixed. +> **Requires `di.servers` with the null/list contract.** An older build accepting only a symbol atom +> returns nothing for `` `ALL `` and *throws* on a list — and because the sweep runs in a `di.timer` +> job with `disableonfail:1b`, that throw would permanently disable monitor discovery. If a monitor +> discovers nothing, the discovered-nothing warning below is what surfaces it. **`connections` is not the same key as `di.servers.connections`.** They share a name and mean different things: `di.servers.connections` decides which process types this process *connects to at @@ -356,9 +346,9 @@ warns once. It is deliberately **cause-agnostic**, watching the consequence rather than any single mechanism, so one check covers all of: -- `` connections:`ALL `` matching nothing against the current `di.servers` (see above); - a `connections` list naming a process type `di.servers` is not configured to connect to — the name-collision trap described earlier; +- a `di.servers` too old for the null/list `getservers` contract, so `` `ALL `` matches nothing; - an injected `servers` dependency that returns nothing for the configured types; - every watched peer being genuinely down. @@ -403,8 +393,8 @@ If your deployments routinely start monitors well ahead of their peers and you w the warning at all, raise `subscribewarnsweeps` — the threshold is in sweeps, so the wall-clock grace is `subscribewarnsweeps × subscribeinterval` (3 minutes on the defaults). -Because the check never names a mechanism in its logic, it stays correct once `di.servers` is fixed; -only the `` `ALL `` hint in the cold-start message would become stale. +Because the check never names a mechanism in its logic, it needs no maintenance as the causes it +catches come and go — which is why it survived the `di.servers` `getservers` fix unchanged. **Only the sweep feeds this check, not `subscribe[]`.** The manual entry point takes bare handles with no identity attached, so there is nothing to key a pending row on — a handle number alone cannot be diff --git a/di/heartbeat/heartbeat.q b/di/heartbeat/heartbeat.q index ad73f4ba..d633828e 100644 --- a/di/heartbeat/heartbeat.q +++ b/di/heartbeat/heartbeat.q @@ -420,7 +420,7 @@ emptysweepmsg:{[] :"no peer discovered yet, after ",(string subscribewarnsweeps)," consecutive sweeps - on a cold ", "start the peers may simply still be coming up, in which case this resolves itself and a ", "recovery line follows. otherwise check that connections names process types di.servers is ", - "connected to. NB `ALL currently matches nothing against di.servers (see heartbeat.md)"; + "actually connected to (see heartbeat.md)"; }; warnemptysweeps:{[] @@ -428,8 +428,9 @@ warnemptysweeps:{[] / threshold. deliberately cause-AGNOSTIC: it watches the consequence, not any single cause, so it / stays correct whatever the reason - a connections list naming a process type di.servers never / connects to, a servers dependency returning nothing, or peers that are simply all down. that also - / means it needs no maintenance when the `ALL gap below is closed. reuses subscribewarnsweeps: it - / is the same "consecutive sweeps" unit as the never-beaten warning + / means it needs no maintenance as those causes come and go - it survived the di.servers getservers + / fix unchanged. reuses subscribewarnsweeps: it is the same "consecutive sweeps" unit as the + / never-beaten warning .z.m.emptysweeps:emptysweeps+1; if[emptysweeps<>subscribewarnsweeps;:()]; / record that a warning was actually EMITTED, rather than leaving the recovery line to re-derive it @@ -463,10 +464,11 @@ notediscovery:{[n] getheartbeats:{[proctypes] / di.servers.getservers returns a TABLE of server rows - the handles are its w column, and a / disconnected row carries a null handle. - / ONE CALL PER PROCTYPE: di.servers.getservers takes a symbol ATOM and matches with =, so passing - / the whole list throws - and di.timer's disableonfail would then end monitor discovery for the - / process lifetime. forward-compatible either way: a one-element iteration still works if getservers - / later accepts vectors + / ONE CALL PER PROCTYPE, deliberately, even though di.servers.getservers now accepts a list: a bare + / symbol is the one shape EVERY version of that contract takes, so this works against a di.servers + / that predates the null/list support as well as one that has it. an older build throws on a list, + / and di.timer's disableonfail would then end monitor discovery for the process lifetime - a + / permanent failure to buy an optimisation worth one call per proctype per sweep pts:(),proctypes; / an empty connections list is a LEGAL "monitor nothing" configuration - init warns about it rather / than rejecting it - so it must be a clean no-op here. without this guard `each` over no proctypes diff --git a/di/heartbeat/test.csv b/di/heartbeat/test.csv index 819483af..d8de259f 100644 --- a/di/heartbeat/test.csv +++ b/di/heartbeat/test.csv @@ -193,7 +193,7 @@ true,0,0,q,-11h=type first .pt.asked,1,1,passed as an atom run,0,0,q,"hb.init[deps,`subenabled`servers`handlers`connections!(1b;mocksrvrows;mockh;`rdb`hdb)]",1,1,re-init with an explicit proctype list run,0,0,q,.pt.asked:(),1,1,clear the record run,0,0,q,.m.di.0heartbeat.hbsubscriptions[],1,1,sweep again -true,0,0,q,2=count .pt.asked,1,1,ONE CALL PER PROCTYPE - passing the whole list throws against the real di.servers and di.timer would then disable the job permanently +true,0,0,q,2=count .pt.asked,1,1,ONE CALL PER PROCTYPE - a bare symbol is the shape EVERY version of the getservers contract accepts so this works against an older di.servers that throws on a list true,0,0,q,all -11h=type each .pt.asked,1,1,each call got a symbol ATOM not the vector true,0,0,q,`rdb`hdb~raze .pt.asked,1,1,and between them they covered every configured proctype @@ -264,7 +264,7 @@ true,0,0,q,"0=count select from .pt.cap where lvl=`error,{0 Date: Fri, 14 Aug 2026 16:45:05 +0100 Subject: [PATCH 5/5] Addressing DIReviewBot comments --- di/heartbeat/heartbeat.md | 15 +++++--- di/heartbeat/heartbeat.q | 16 +++++++-- di/heartbeat/test.csv | 16 ++++++++- di/heartbeat/test.q | 59 ++++++++++++++++++++++--------- di/heartbeat/test_integration.csv | 24 +++++++++++++ 5 files changed, 107 insertions(+), 23 deletions(-) diff --git a/di/heartbeat/heartbeat.md b/di/heartbeat/heartbeat.md index 02f4b187..2dd4954a 100644 --- a/di/heartbeat/heartbeat.md +++ b/di/heartbeat/heartbeat.md @@ -138,10 +138,17 @@ against a `di.servers` that predates the list/null support as well as one that h one call per configured proctype per sweep, which is not worth optimising away for the coupling it would add. -> **Requires `di.servers` with the null/list contract.** An older build accepting only a symbol atom -> returns nothing for `` `ALL `` and *throws* on a list — and because the sweep runs in a `di.timer` -> job with `disableonfail:1b`, that throw would permanently disable monitor discovery. If a monitor -> discovers nothing, the discovered-nothing warning below is what surfaces it. +> **`` `ALL `` requires a `di.servers` with the null match-all contract.** An older build matching +> `proctype=pt` against an atom returns nothing for a null symbol, so the shipped default discovers +> zero peers — silently, apart from the discovered-nothing warning below. Naming process types +> explicitly still works against such a build, which is exactly what the per-proctype iteration buys: +> the sweep never passes a list, so the *other* half of an old contract's incompatibility (throwing on +> a vector) is unreachable from this module. +> +> This is asserted against the real module, not a mock: `test_integration.csv` stands up a real +> `di.servers`, connects it to a spawned publisher, and drives a `` connections:`ALL `` sweep through +> to the subscription landing on that peer. Every other `` `ALL `` test in `test.csv` runs against a +> mock this module wrote, and so can only show it is self-consistent. **`connections` is not the same key as `di.servers.connections`.** They share a name and mean different things: `di.servers.connections` decides which process types this process *connects to at diff --git a/di/heartbeat/heartbeat.q b/di/heartbeat/heartbeat.q index d633828e..40715abf 100644 --- a/di/heartbeat/heartbeat.q +++ b/di/heartbeat/heartbeat.q @@ -61,6 +61,13 @@ jobids:`hbpublish`hbcheck`hbsubscribe; / module state / ============================================================ +/ name resolution inside this module: a bare READ resolves to the current .z.m value - init writes +/ .z.m. and every later read follows it, including across a re-init that changes it. a WRITE +/ must always be explicit .z.m.:, since a bare assignment makes a function-local. the one +/ exception is a qsql select/where/by clause, which cannot resolve a module-level name at ALL and +/ throws on it - hence the local hoists in trackunseen (n) and checkheartbeat (wp/ep). those hoists +/ are about qsql scope, NOT about .z.m. raised three times on PR #122, so it is written down once here + / current-time function - heartbeat owns its clock, separate from di.timer's; override via setcp cp:{.z.p}; @@ -283,8 +290,13 @@ ensureroottable:{[] set[tablename;schema]; .z.m.rootowned:1b; :()]; - / already there and we created it on an earlier init - keep ownership and leave the table alone - if[rootowned;:()]; + / already there and we created it on an earlier init - keep ownership and leave the table alone. + / read rootowned defensively, matching validateroot: it is unset until the first init has run, and a + / bare read of an unset module name THROWS rather than returning a null. that path is unreachable + / through init (validateroot gates the foreign-table case before any state is written, and the first + / init writes rootowned before installroot runs), but a direct call then gets this module's own + / error instead of a bare 'rootowned. raised on PR #122 + if[@[{rootowned};::;0b];:()]; raiseerror[`installroot;"a table named ",(string tablename)," already exists at root and was not ", "created by this module - refusing to use it. if it is left over from an earlier load of ", "di.heartbeat, remove it (the publish table is always an empty schema holder, so nothing is ", diff --git a/di/heartbeat/test.csv b/di/heartbeat/test.csv index d8de259f..50b912b4 100644 --- a/di/heartbeat/test.csv +++ b/di/heartbeat/test.csv @@ -184,7 +184,7 @@ run,0,0,q,.m.di.0heartbeat.hbsubscriptions[],1,1,run the subscription sweep the true,0,0,q,1=count .pt.asked,1,1,getservers was consulted exactly once true,0,0,q,all null raze .pt.asked,1,1,the shipped `ALL default resolved to the null match-all symbol not a literal lookup for `ALL true,0,0,q,-11h=type first .pt.asked,1,1,and it was passed as a symbol ATOM - real di.servers.getservers rejects anything else -comment,,,,,,,"regression cover - a single bare symbol is the only connections shape that works against the REAL di.servers today" +comment,,,,,,,"the per-proctype iteration - a bare symbol is the one shape EVERY version of the getservers contract accepts, which is why the sweep still iterates now that di.servers takes lists too. the `ALL coupling itself is proved against the real di.servers in test_integration.csv" run,0,0,q,"hb.init[deps,`subenabled`servers`handlers`connections!(1b;mocksrvrows;mockh;`rdb)]",1,1,re-init with one bare proctype run,0,0,q,.pt.asked:(),1,1,clear the record run,0,0,q,.m.di.0heartbeat.hbsubscriptions[],1,1,sweep @@ -578,3 +578,17 @@ run,0,0,q,.pt.now:2025.01.01D00:02:00,1,1,jump past warning AND error in one ste run,0,0,q,hb.checkheartbeat[],1,1,single check true,0,0,q,first exec error from hb.gethb[] where procname=`jump,1,1,error is set true,0,0,q,not first exec warning from hb.gethb[] where procname=`jump,1,1,warning is NOT set - it was never in warning so the transition never fired; both flags only coexist on a gradual escalation +comment,,,,,,,"name resolution - a bare read follows .z.m across a re-init; the trackunseen/checkheartbeat hoists are a qsql-scope problem, not this one (raised three times on PR #122)" +run,0,0,q,hb.teardown[],1,1,tear down +run,0,0,q,"hb.init[deps,`subscribewarnsweeps`warningtolerance`errortolerance!(9;7f;9f)]",1,1,re-init with values that differ from the shipped defaults of 3 2f and 3f +true,0,0,q,"0/tmp/dihbchild",string[port],".log 2>&1 &"; - h:0N; - i:0; - while[(null h) and i<100; - h:@[{hopen `$":localhost:",string x};port;0N]; - i+:1; - system"sleep 0.1"]; + h:awaithandle port; if[null h;'"di.heartbeat test: publisher process failed to start - see /tmp/dihbchild",string[port],".log"]; / wait for its init to complete - i:0; - while[(not @[h;"@[{ready};::;0b]";0b]) and i<100; - i+:1; - system"sleep 0.1"]; + awaitready h; .ht.port:port; .ht.h:h; :h; }; +/ write the phone book the REAL di.servers reads, naming the spawned publisher as the one peer to +/ connect to. di.servers' header check is strict and positional (host,port,proctype,procname, v1 +/ 4-column), so the header line must be exactly this or readprocesscsv fails loud. proctype and +/ procname must match the identity writechild gives the child, or di.servers dials a row that is not +/ the process we spawned. no self row: di.servers' own suite covers self-exclusion, and leaving it out +/ keeps this fixture to the one thing it exists to prove +writeserverscsv:{[port] + path:"/tmp/dihbservers",string[port],".csv"; + (hsym `$path) 0: ( + "host,port,proctype,procname"; + "localhost,",string[port],",childtype,childproc"); + :path; + }; + / the monitor's own upd - di.heartbeat deliberately does NOT install this itself. / hbm is indexed, not dotted: module dot-sugar only works on a plain top-level name, and fails / silently inside a lambda or on a dotted one @@ -90,5 +114,8 @@ cleanup:{[] / is not reliable here - it can sit in the output buffer and be discarded by the hclose below @[{x"exit 0"};.ht.h;::]; @[hclose;.ht.h;::]; - @[{system"rm -f /tmp/dihbchild",string[x],".q /tmp/dihbchild",string[x],".log"};.ht.port;::]; + / the servers phone book is only written by the di.servers block, so rm -f covers the runs that + / never created one + @[{system"rm -f /tmp/dihbchild",string[x],".q /tmp/dihbchild",string[x],".log /tmp/dihbservers", + string[x],".csv"};.ht.port;::]; }; diff --git a/di/heartbeat/test_integration.csv b/di/heartbeat/test_integration.csv index 87a3375f..f07cfc11 100644 --- a/di/heartbeat/test_integration.csv +++ b/di/heartbeat/test_integration.csv @@ -76,3 +76,27 @@ true,0,0,q,all exec status from realtimer[`getalljobs][],1,1,and no job was disa run,0,0,q,hbm[`teardown][],1,1,teardown against the real registries true,0,0,q,0=count realtimer[`getalljobs][],1,1,the real timer jobs were cleaned true,0,0,q,not `heartbeat in tables[],1,1,the root names were cleaned + +comment,,,,,,,"the REAL di.servers - the last stub in the di.torq-shaped block above, and the one whose contract heartbeat.md makes a claim about" +comment,,,,,,,"`ALL resolves to a null symbol and every other test asserts that against a mock di.heartbeat itself wrote. only the real getservers can prove the coupling, and the caveat removed in a2b07ed rests on it" +run,0,0,q,realservers:use`di.servers,1,1,real di.servers +run,0,0,q,.ht.h2:spawnpublisher[],1,1,spawn a fresh publisher to be the discovered peer - the earlier child was cleaned up above +run,0,0,q,.ht.csv:writeserverscsv[.ht.port],1,1,write the 4-column phone book naming that publisher +run,0,0,q,"realservers[`init][logging[`logdict],`timer`handlers`proctype`procname`connections`processcsv!(realtimer;realhandlers;`montype;`monproc;`childtype;.ht.csv)]",1,1,init the real di.servers against the same real timer and handlers +run,0,0,q,realservers[`startup][],1,1,open the connection to the publisher +true,0,0,q,1=count realservers[`getservers][`childtype],1,1,di.servers connected to the publisher +true,0,0,q,0