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..7811efcb --- /dev/null +++ b/di/heartbeat/deps.q @@ -0,0 +1,5 @@ +/ hard module dependencies and their minimum versions, validated by di.depcheck +/ di.heartbeat has no hard dependencies - all runtime dependencies (log, timer, +/ 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 new file mode 100644 index 00000000..2dd4954a --- /dev/null +++ b/di/heartbeat/heartbeat.md @@ -0,0 +1,566 @@ +# 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. `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. + +```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!(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.) + +> **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 | +| `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 | +| `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. + +### How `` `ALL `` resolves, and why the sweep still iterates + +`` `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`). + +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. + +> **`` `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 +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 + +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. +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 +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` 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. + +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 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 +``` + +### 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. + +**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: +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: + +- 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. + +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 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 +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 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 +> 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 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) | +| `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]; ... } + ``` + > **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 + `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. +- **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 + +```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 + +- **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 + **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..40715abf --- /dev/null +++ b/di/heartbeat/heartbeat.q @@ -0,0 +1,858 @@ +/ 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 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 + +/ ============================================================ +/ 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; + +/ 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. +/ 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`subscribeinterval, + `warningtolerance`errortolerance`subscribewarnsweeps`maxage, + `pid`host`port`connections`onwarning`onerror + )!( + 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 +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}; + +/ ============================================================ +/ 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 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; + }; + +/ ============================================================ +/ init - dependency validation and config +/ ============================================================ + +validatedeps:{[deps] + / 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"]; + 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 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)"]; + }; + +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`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`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 + / 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 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. + / 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 ", + "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 +/ ============================================================ + +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;();tosecs subscribeinterval;2;()!()]]; + }; + +/ ============================================================ +/ 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. + / 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) + @[{[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 ", + "actually connected to (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 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 + / 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. + / 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 + / 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. + / 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; + :()]; + 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; + 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. 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[]]; + }; + +/ ============================================================ +/ 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]; + / 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 + 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). 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; + 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: 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 + / publishroot off would otherwise leave teardown unable to clean up its own residue + if[rootinstalled;uninstallroot[]]; + .z.m.enabled:0b; + .z.m.subenabled:0b; + / 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 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; + .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]; + / 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.unseen:unseenschema; + resetdiscovery[]; + .z.m.rootowned:0b; + .z.m.rootinstalled: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.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; + .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[]; + / 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 the timer jobs 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..50b912b4 --- /dev/null +++ b/di/heartbeat/test.csv @@ -0,0 +1,594 @@ +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; 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) +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 +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,`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,delete from `.pt.cap,1,1,clear the log capture +run,0,0,q,hb.subscribe[-5i],1,1,subscribing a negative handle is accepted but skipped +true,0,0,q,0=count select from .pt.cap where lvl=`error,1,1,a negative handle is never attempted - neg of an already-negative handle is POSITIVE so it would become the blocking sync call the async switch removed + +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: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 + 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 +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"; + }; + +/ make the publisher unresponsive for a while WITHOUT killing it - the "alive but stalled" state a +/ heartbeat exists to detect. sent async so this call itself does not block +hangpeer:{[h;secs] + (neg h)"system\"sleep ",string[secs],"\""; + }; + +/ elapsed time for a unary call. used to prove the subscribe does not block against a hung peer: +/ the synchronous version waited out the ENTIRE hang, and because it runs inside a di.timer job on a +/ single thread it stalled publishheartbeat and checkheartbeat with it, so the monitor fell silent to +/ its own monitors at exactly the moment a peer was misbehaving +elapsed:{[f;arg] + t0:.z.p; + f arg; + :.z.p-t0; + }; + +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;::]; + / 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 new file mode 100644 index 00000000..f07cfc11 --- /dev/null +++ b/di/heartbeat/test_integration.csv @@ -0,0 +1,102 @@ +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,"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 +comment,,,,,,,"a suite written only in the module's own mock dialect can only prove self-consistency; this block caught a nested-list log message that every mock accepted" +run,0,0,q,logging:use`di.log,1,1,real di.log +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,"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 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