Feature subscriptions - #123
Open
alowrydi wants to merge 11 commits into
Open
Conversation
…ions Releases a handle's subscriptions before close (the local-close case .z.pc misses); validates schema type, duplicate table names, logfilelist shape and sign, empty selectors, and setschema:0b with replay. Tests: 318, was 209.
…some general housekeeping too: chain .z.pc instead of replacing it, add VERSION/version and getsubtables, signal from the string subscribe entry points, remove hardcoded ports and paths from the suite
DIReview Summary0 critical | 4 warning(s) | 0 suggestion(s)
|
DIReview Summary0 critical | 1 warning(s) | 0 suggestion(s)
|
TorQ's code/common/pubsub.q:19 broadcasts (`endofperiod;x;y;z) and both of
its subscribers - code/rdb/endofperiod.q and code/wdb/writedown.q:52 - are
{[currp;nextp;data]}. Ours was {(neg getallhandles[])@\:(`endofperiod;x)},
which failed two ways at once, both measured:
callendofperiod[c;n;d] -> 'rank, so a caller following that contract
could not call it at all
callendofperiod[c] -> a ternary subscriber is left PARTIALLY
APPLIED. q returns a projection, the body
never runs, and nothing throws, logs or
comes back to say so.
Same defect class as the callendofday bug fixed in #118.
callendofday deliberately stays UNARY. TorQ sends (`endofday;x;y), but the
second argument is processdata, which legacy's own rdb never reads and the
shipped .u.end alias passes ()!() for - di.rdb's endofday is unary to
match. Fixing it "for symmetry" would projection-ise every unary
subscriber, so the asymmetry is documented in pubsub.md rather than left
looking like an oversight.
15 regression rows added. Negative control: against the unary version they
fail exactly 3 - the ternary call on 'rank, plus both subscriber
assertions - and nothing else.
DIReview Summary0 critical | 1 warning(s) | 0 suggestion(s)
|
di.servers now refuses every accessor until its own init has run (requireinit, servers.q:50). That is deliberate on its side - a pre-init getservers used to return an empty table, indistinguishable from "nothing is connected" - but the signal it raises names only di.servers and reaches only di.servers' logger, so it travelled straight through this module: a di.subscriptions call handed back a di.servers error and nothing landed in this module's log. getsubscriptionhandles now catches it and re-raises through raiseerror, so the caller learns which of the two modules is unwired and the failure is observable here like every other domain error in the module. Returning an empty table instead was considered and rejected: "I cannot tell you" is not the same answer as "no handles", and a caller that treats them alike waits for a tickerplant that was never going to appear. Only the getservers[`] call is wrapped. The proctype call below it is reachable only once the first has succeeded, so init is known to have run by then. 8 unit rows added, placed ABOVE the di.servers init row - once that module is initialised the pre-init path cannot be reached again in the same process. They assert all three things that matter separately: the error carries this module's context, it names di.servers as the module needing wiring, and it was logged here rather than bypassing this log. Driving the real getservers also means the suite must init di.servers itself, then tear it down immediately: init registers a .z.pc observer and a serversretry timer job, and this suite fires .z.pc by hand (markdeadfor) against handles that overlap the seeded SERVERS rows. teardown releases both while leaving the accessors callable, which is what di.servers' separate registered/initialised flags are for. Integration suite: di.servers also now requires `deletejobs on the timer dep - teardown deletes its retry job through it, and it is validated as callable at init rather than left to fail inside a protected apply - so the mocktimer stub gained one. Without it svc.init threw and took the whole 12-row di.servers end-to-end block with it. Suites: 570 unit + 131 integration = 701, all green. Was 562 + 131 with the integration block failing 12.
DIReview Summary0 critical | 1 warning(s) | 0 suggestion(s)
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Subscription management module, di.subscriptions
Extracts
code/common/subscriptions.qfrom TorQ and packages it as a standalone kdb-x module. Manages tickerplant subscriptions for RDB/WDB/chained-TP consumers - subscribe with optional exactly-once log replay, resubscribe after a dropped connection, unsubscribe, and a live registry of what's currently held.The modularisation plan places
di.subscriptionsin the FRAMEWORK tier with hard dependencies ondi.serversanddi.pubsub.getsubscriptionhandles(the legacy handle-resolution function TorQ consumers call before subscribing) is ported here rather than duplicated across every futuredi.proc.*module, and a new opt-inrepublishcapability lets a chained/segmented tickerplant consumer hand its subscribed tables todi.pubsubfor downstream fan-out - the actual job a chained TP exists to do. Closing thedi.pubsubedge safely required three small, necessary changes to that module, included here rather than as a follow-up: its.z.pcwas overwriting rather than chaining, meaning depending on it as shipped would silently destroy any other module's observer; it shipped noVERSION, so a version pin against it would have been unenforceable; and it had no read counterpart tosetsubtables, which the additive handoff needs. A fourth, unrelateddi.pubsubfix (a broken result-shape guard insubscribestr/subscribestrfilter) is bundled in as well - found during this due-diligence read, not required by anything here, worth knowing that distinction if this gets split later.Trello ticket - https://trello.com/c/aUtlpkfA/133-kdb-x-subscriptions-module
Files created / modified
di/subscriptions/init.qdi/subscriptions/subscriptions.qdi/subscriptions/subscriptions.mddi/subscriptions/deps.qdi/subscriptions/test.csvdi/subscriptions/test_integration.csvdi/pubsub/pubsub.q.z.pcnow chains via di.handlers-style capture instead of overwriting; getsubtables added; subscribestr/subscribestrfilter's result-shape guard fixeddi/pubsub/init.qdi/pubsub/VERSIONdi/pubsub/test.csvHow to test
784 suite assertions, 0 failures. Plus 317 smoke/adversarial-harness assertions across 12 scratchpad suites (
smoke.q/smoke2.q/smoke3.q,adversarial.q,iter3.q,probe7/probe8/probe10/probe11.q,gsh.q,repub.q,pssmoke.q) - 1101 assertions total, 0 failures, 0 flags.The integration suite runs three real processes for the republish path specifically - an upstream tickerplant, a middle process using this module with
republishon, and a downstream subscriber - since a subscriber that serves its own subscribers can't be tested in-process. Its tickerplant peer runs the realdi.pubsub, and assertions read that module's own registry directly, so the coupling can't regress into a stand-in.The handoff-failure warn-not-throw path was proven non-vacuous, not just exercised: the same suite run against a copy with the fix reverted to
raiseerrorproduces 7 failures (the module throws, the completed subscription is lost, nothing is logged atwarn); with the fix in place, 0.di.depcheck:Resolved against
deps.q =di.serversdi.pubsub!("0.1.0";"0.2.0"), withdi.serversat 0.1.0 anddi.pubsubat 0.2.0 on disk.Design decisions
Protocol correctness (subscribe/replay against a real segmented tickerplant)
1. Guards that don't need the tickerplant's reply run before it -
fetchdetails's underlyingsubdetailscall is.u.sub, which registers the handle for live delivery as a side effect of just fetching schemas. Every check not needing that reply (root-updpresence, the already-held-table guard, the setschema-implies-exists guard) now runs first; only the log-integrity preflight is irreducible, since log filenames only exist in the reply itself.2.
`(all tables) is resolved via atablelistround trip beforesubdetails- a segmented tickerplant'ssubdetailscan't accept the bare sentinel directly (throws'rank); legacy always resolved a concrete list first. Falls back to sending`directly if no usabletablelistis offered.3. A shared log (segmented
singular/periodicmultilog mode) collapses to one whole-file replay - the same physical file is reported once per table with different counts; replaying each entry separately re-applies the file's head. Collapsing to one whole-file replay trades a bounded, diagnosable trailing duplicate for never silently under-replaying, which a naive per-table sum could do.4. A whole-file replay is narrowed to the offered tables; a genuine per-table log is not - a segmented tickerplant logs more tables than it publishes, so a whole-file read (from the shared-log collapse or the
0Wjsentinel) can carry a table the caller never subscribed to. Narrowed via the existing table filter; the common per-table-log fast path is unaffected.5.
0Wjmeans "replay everything", not "more than expected" - resolved to the log's own readable total rather than forwarded to-11!, which would otherwise replay a corrupt log's good prefix and only then throw - the exact half-populated state the preflight exists to prevent.Framework-tier dependencies and republish
6.
getsubscriptionhandlesported rather than left to each caller - real TorQ callers acrossrdb.q,wdb.q,chainedtp.q,sctp.qall resolve a handle the same way before subscribing; centralising it here avoids four futuredi.proc.*modules each reimplementing legacy's`/()/intersection-vs-union selection semantics independently.7.
republishis opt-in, default off - realises the chained/segmented-tickerplant role (subscribe upstream, serve the same tables downstream) thatchainedtp.q/sctp.qsplit across themselves in TorQ. Off by default: a plain RDB must not silently become a publisher.8. The handoff to
di.pubsubis additive - unions with whatever the publisher already serves rather than replacing it, becausesetsubtablesreplaces its list andunsubscribedeletes registry rows; a registry-only union would drop an unsubscribed table at the next unrelated subscribe. Never shrinks toward empty either, since an empty list makesdi.pubsubpublish every root table.9. A handoff failure warns, doesn't throw - it's the last step of
subscribe, by which point the subscription has already fully succeeded (schemas defined, log replayed, registry committed). Throwing would discard completed work over an optional secondary step.10.
di.pubsub's.z.pcnow chains - a bare.z.pc:{closesub[x]}at load silently destroyed any observer another module had already registered, withdi.handlerscontinuing to report it as live. Depending ondi.pubsubas a hard dependency without fixing this would import that risk.Checklist
Documentation
See
subscriptions.mdfor full reference, including thesubdetails/tablelistprotocol notes, the shared-log and whole-file replay behaviour,republishconfiguration, and known non-blocking limitations (live-feed filtering delegation is unverified end-to-end pendingdi.tickerplant; a shared log with exactly-equal per-table counts is indistinguishable from an ordinary reply and silently under-replays - both carried forward as requirements fordi.tickerplantto address upstream).