Feature heartbeat

di/heartbeat/VERSION

@@ -0,0 +1 @@
+0.1.0

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:(`$())!();

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])

di/heartbeat/test.q

@@ -0,0 +1,121 @@
+/ integration fixtures for di.heartbeat - spawn a real publisher process and subscribe to it
+/ this is the only way to exercise the two failures a pubsub mock cannot catch:
+/   1. the remote-subscribe handshake, which depends on .z.w resolving to the MONITOR's connection
+/      inside an inbound call on the PUBLISHER
+/   2. the root schema table, without which di.pubsub silently refuses the subscription and
+/      discards every published row
+
+/ pick a free port by binding one and immediately releasing it. the while-iterator {cond}{body}/ is
+/ used rather than a while loop: the style guide bans do/while outright, and a bounded retry is still
+/ an iterator - 0 is the "not found yet" state, and a failed bind leaves it at 0 for another go
+freeport:{[]
+  p:{0=x}{@[{[c] system"p ",string c;c};10000+rand 40000;0]}/0;
+  system"p 0";
+  :p;
+  };
+
+/ write the publisher script the child process will run
+writechild:{[path;port]
+  src:(
+    "hb:use`di.heartbeat;";
+    "ps:use`di.pubsub;";
+    "/ stub timer - this test drives publishheartbeat by hand, so nothing needs scheduling";
+    "tmr:`addjob`deletejobs!((enlist`custom)!enlist {[i;f;p;pe;m;o]};{[ids]});";
+    "lg:`info`warn`error!({[c;m]};{[c;m]};{[c;m]});";
+    "/ heartbeat init FIRST - it publishes the root schema table that di.pubsub then discovers.";
+    "/ reversing these two lines is the silent failure this whole test exists to catch.";
+    "hb.init[(`log`timer`pubsub!(lg;tmr;`publish`subscribe!(ps.publish;ps.subscribe))),";
+    "  `proctype`procname!(`childtype;`childproc)];";
+    "ps.init[];";
+    "ready:1b;");
+  path 0: src;
+  };
+
+/ poll for the child's port for up to 10s, then give up. state is (attempts;handle), threaded through
+/ the while-iterator so the retry cap is part of the condition rather than a counter in a loop body.
+/ the null handle is seeded 0Ni, not 0N - hopen returns an int, and a long seed would make the state
+/ change type on the first iteration
+awaithandle:{[port]
+  s:{(null x 1) and x[0]<100}{[port;x]
+    system"sleep 0.1";
+    (x[0]+1;@[{hopen `$":localhost:",string x};port;0Ni])}[port]/(0;0Ni);
+  :s 1;
+  };
+
+/ poll until the child's own init has completed. the probe is protected on the CHILD side, so a call
+/ that lands before ready exists comes back 0b rather than throwing back across the connection
+awaitready:{[h]
+  {(not x 1) and x[0]<100}{[h;x]
+    system"sleep 0.1";
+    (x[0]+1;@[h;"@[{ready};::;0b]";0b])}[h]/(0;0b);
+  };
+
+/ launch the child and wait for it to answer
+spawnpublisher:{[]
+  port:freeport[];
+  path:"/tmp/dihbchild",string[port],".q";
+  writechild[hsym `$path;port];
+  system"q ",path," -p ",string[port]," -q </dev/null >/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;::];
+  };

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,"0<handle""count raze value .m.di.0pubsub.reqalldict""",1,1,the publisher registered a subscriber - .z.w resolved to the monitor's connection inside the inbound call
+
+comment,,,,,,,a published row actually crosses the wire and lands in the monitor's store
+true,0,0,q,0=count hbm[`gethb][],1,1,monitor store starts empty
+run,0,0,q,"handle""hb.publishheartbeat[]""",1,1,make the publisher publish one heartbeat
+run,0,0,q,drain[handle],1,1,force the monitor to process the async upd the publisher pushed
+true,0,0,q,1=count hbm[`gethb][],1,1,the published row reached the monitor and was stored
+true,0,0,q,`childtype in exec sym from hbm[`gethb][],1,1,the row carries the publisher's process type
+true,0,0,q,`childproc in exec procname from hbm[`gethb][],1,1,the row carries the publisher's process name
+true,0,0,q,not first exec warning from hbm[`gethb][],1,1,a freshly received heartbeat is not in warning
+run,0,0,q,"handle""hb.publishheartbeat[]""",1,1,publish a second heartbeat
+run,0,0,q,drain[handle],1,1,drain again
+true,0,0,q,1=count hbm[`gethb][],1,1,the second beat updated the existing row rather than adding one
+true,0,0,q,1=first exec counter from hbm[`gethb][],1,1,and the counter advanced - proving the second row was genuinely received
+
+comment,,,,,,,teardown on the publisher removes its root names
+run,0,0,q,"handle""hb.teardown[]""",1,1,tear the publisher down
+true,0,0,q,"handle""not `heartbeat in tables[]""",1,1,the publisher's root schema table was removed
+true,0,0,q,"handle""not @[{value`.heartbeat.subscribe;1b};::;0b]""",1,1,the publisher's root subscribe entry point was removed
+
+comment,,,,,,,"a stalled publisher must not block the monitor - the availability bug the async switch exists to fix"
+comment,,,,,,,"the synchronous subscribe waited out the entire hang; inside a di.timer job on a single thread that stalled publishheartbeat and checkheartbeat too"
+run,0,0,q,hangpeer[handle;3],1,1,make the publisher alive but unresponsive for three seconds
+true,0,0,q,0D00:00:01>elapsed[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<count realservers[`getservers][`],1,1,a NULL symbol matches every live server - the getservers contract heartbeat.md relies on, asserted against the real module not a mock
+comment,,,,,,,"and the coupling itself - the shipped `ALL default must reach a real peer through the real getservers"
+run,0,0,q,"hbm[`init][logging[`logdict],`timer`pubsub`handlers`servers`proctype`procname`subenabled`connections!(realtimer;realps;realhandlers;realservers;`montype;`monproc;1b;`ALL)]",1,1,init the monitor with the REAL di.servers export dict and the shipped `ALL default
+true,0,0,q,"0=.ht.h2""count raze value .m.di.0pubsub.reqalldict""",1,1,the fresh publisher has no subscribers before the sweep
+run,0,0,q,.ht.sh:first exec w from realservers[`getservers][`childtype],1,1,the handle DI.SERVERS opened - not the monitor's own hopen
+run,0,0,q,.m.di.0heartbeat.hbsubscriptions[],1,1,run the discovery sweep the timer would run
+run,0,0,q,drain[.ht.sh],1,1,"the subscribe is async and went over di.servers' handle, so it must be flushed on THAT handle; a sync call there also waits for the child to process it in order. draining the monitor's own handle proves nothing and asserted 0 subscribers against a sweep that had in fact worked"
+true,0,0,q,"0<.ht.h2""count raze value .m.di.0pubsub.reqalldict""",1,1,`ALL discovery found the peer through the real getservers and the subscribe reached it
+run,0,0,q,.m.di.0heartbeat.hb:.m.di.0heartbeat.storeschema,1,1,"clear the store - it still holds the FIRST child's beats under the same childtype/childproc identity, and this block must not pass on those"
+run,0,0,q,".ht.h2""hb.publishheartbeat[]""",1,1,make the discovered peer publish one heartbeat
+run,0,0,q,drain[.ht.sh],1,1,drain the handle the subscription was made on - that is the connection the published row arrives over
+true,0,0,q,1=count hbm[`gethb][],1,1,a beat from the `ALL-discovered peer reached the monitor - discovery subscribe and delivery all the way through
+true,0,0,q,`childproc in exec procname from hbm[`gethb][],1,1,and it is the peer di.servers discovered not a row left over from earlier
+run,0,0,q,hbm[`teardown][],1,1,tear the monitor down
+run,0,0,q,cleanup[],1,1,close the handle and remove the child script log and phone book