From 5b84d4b53c9b3f33f342b6d29d8c1924c34f7b62 Mon Sep 17 00:00:00 2001 From: ascottDI Date: Thu, 13 Aug 2026 16:56:48 +0100 Subject: [PATCH 1/3] initial implimentation and pass following a couple pointers --- di/tplog/VERSION | 1 + di/tplog/deps.q | 5 ++ di/tplog/init.q | 77 ++------------------- di/tplog/tplog.q | 177 +++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 190 insertions(+), 70 deletions(-) create mode 100644 di/tplog/VERSION create mode 100644 di/tplog/deps.q create mode 100644 di/tplog/tplog.q diff --git a/di/tplog/VERSION b/di/tplog/VERSION new file mode 100644 index 00000000..6e8bf73a --- /dev/null +++ b/di/tplog/VERSION @@ -0,0 +1 @@ +0.1.0 diff --git a/di/tplog/deps.q b/di/tplog/deps.q new file mode 100644 index 00000000..ca3444e4 --- /dev/null +++ b/di/tplog/deps.q @@ -0,0 +1,5 @@ +/ hard module dependencies and their minimum versions, validated by di.depcheck. +/ di.tplog is self-contained - it uses no other di.* module via `use` (lifecycle and byte-scan +/ recovery are built on base q only). its one runtime dependency, log, is injected via init as a +/ dictionary of functions and is validated by di.depcheck's core-contract check, not declared here. +deps:(`$())!(); diff --git a/di/tplog/init.q b/di/tplog/init.q index 3220f689..194e21ed 100644 --- a/di/tplog/init.q +++ b/di/tplog/init.q @@ -1,72 +1,9 @@ -/ header to build deserialisable msg -header:8#-8!(`upd;`trade;()); -/ first part of tp update msg -updmsg:`char$10#8_-8!(`upd;`trade;()); -/ size of default chunk to read (10MB) -chunk:10*1024*1024; -/ don't let single read exceed this -maxchunk:8*chunk; +/ di.tplog - tickerplant log lifecycle (open/write/roll/replay/replayupto/logname) plus corruption +/ check/repair. self-contained (no hard `use` deps); log is injected via init. see tplog.md +\l ::tplog.q -check:{[logfile;lastmsgtoreplay] - / logfile (symbol) is the handle to the logsfile - / lastmsgtoreplay (long) is index position of the last message to be replayed from the log - / check if the logfile is corrupt - loginfo:-11!(-2;logfile); - :$[1 = count loginfo; - / - the log file is good so return the good log file handle - :logfile; - loginfo[0] <= lastmsgtoreplay + 1; - :logfile; - repair[logfile] - ] - }; - -repair:{[logfile] - / - append ".good" to the "good" log file - goodlog: `$ string[logfile],".good"; - / - create file and open handle to it - goodlogh: hopen goodlog set (); - / - loop through the file in chunks - repairover[logfile;goodlogh] over `start`size!(0j;chunk); - / - return goodlog - goodlog - }; - -repairover:{[logfile;goodlogh;d] - / logfile (symbol) is the handle to the logsfile - / goodlogh (int) is the handle to the "good" log file - / d (dictionary) has two keys start and size, the point to start reading from and size of chunk to read - / read bytes from - x:read1 logfile,d`start`size; - / find the start points of upd messages - u: ss[`char$x;updmsg]; - / nothing in this block - if[not count u; - / EOF - we're done - if[hcount[logfile] <= sum d`start`size;:d]; - / move on bytes - :@[d;`start;+;d`size]]; - / split bytes into msgs - m: u _ x; - / message sizes as bytes - mz: 0x0 vs' `int$ 8 + ms: count each m; - / set msg size at correct part of hdr - hd: @[header;7 6 5 4;:;] each mz; - / try and deserialize each msg - g: @[(1b;)@-9!;;(0b;)@] each hd,'m; - / write good msgs to the "good" log - goodlogh g[;1] where k:g[;0]; - / saw msg(s) but couldn't read - if[not any k; - / read as much as we dare, give up - if[maxchunk <= d`size; - :@[d;`start`size;:;(sum d`start`size;chunk)]]; - / read a bigger chunk - :@[d;`size;*;2]]; - / move to the end of the last good msg - ns: d[`start] + sums[ms] last where k; - :@[d;`start`size;:;(ns;chunk)]; - }; - -export:([check;repair]) +/ module version, read from the VERSION file before the export line evaluates each name. +/ trim so a trailing newline / CRLF cannot pad the semver di.depcheck compares +version:trim first read0`:::VERSION +export:([init;logname;open;write;roll;replay;replayupto;check;repair;getapimeta;version]) diff --git a/di/tplog/tplog.q b/di/tplog/tplog.q new file mode 100644 index 00000000..f02d953d --- /dev/null +++ b/di/tplog/tplog.q @@ -0,0 +1,177 @@ +/ di.tplog - tickerplant log lifecycle plus corruption check/repair, in one module. +/ lifecycle (open/write/roll/replay/replayupto/logname) is ported from the inline log handling +/ in TorQ/code/processes/tickerplant.q (.u.ld / .u.endofday); the byte-scanning check/repair +/ recovery is ported from TorQ/code/common/tplogutils.q. self-contained: no hard `use` deps. +/ log is an injected, required dependency (see init) - best-effort recovery is narrated so silent +/ message drops are observable. version is sourced from the VERSION file in init.q. + +/ --- message-signature constants used by the byte-scan recovery (repairover) --- +/ these are geared to the (`upd;`trade;...) message shape, inherited from TorQ tplogutils; logs of +/ other table shapes are recovered only if their messages share this prefix (see known gaps in .md) +/ header template to rebuild a deserialisable message header +header:8#-8!(`upd;`trade;()); +/ first bytes of a tp update message, the signature searched for in the raw log +updmsg:`char$10#8_-8!(`upd;`trade;()); +/ default chunk to read (10mb) +chunk:10*1024*1024; +/ never let a single read exceed this +maxchunk:8*chunk; + +init:{[deps] + / wire the injected log dependency - required, no fallback. deps is a dict with a `log key holding + / `info`warn`error!({[ctx;msg]};...) (extra levels like di.log's six are accepted and ignored). + / examples: + / tp.init[(use`di.log)`logdict] / di.log.logdict is pre-shaped as `log!(...) + / tp.init[enlist[`log]!enlist `info`warn`error!(f;f;f)] + / signalled with a plain ' (not raiseerror) - the logger is not yet wired while init runs. + if[99h<>type deps; + '"di.tplog: deps must be a dict with a `log key"]; + if[not `log in key deps; + '"di.tplog: log dependency is required; pass `info`warn`error functions keyed on `log"]; + if[99h<>type deps`log; + '"di.tplog: log value must be a dict of `info`warn`error functions"]; + if[not all `info`warn`error in key deps`log; + '"di.tplog: log dict must have `info`warn`error keys; got: ",", " sv string key deps`log]; + .z.m.loginfo:deps[`log]`info; + .z.m.logwarn:deps[`log]`warn; + .z.m.logerr:deps[`log]`error; + }; + +raiseerror:{[ctx;msg] + / internal - log an error under ctx via the injected logger, then signal it, so a failure lands in + / the log as well as being thrown. every post-init domain error routes through here. + .z.m.logerr[ctx;msg]; + '"di.tplog: ",string[ctx],": ",msg; + }; + +corruptp:{[logfile] + / internal - true if the log is unreadable/corrupt. -11!(-2;...) is the NON-EXECUTING mode: on a + / clean log it returns the message count without running upd, and on this kdb-x build it THROWS on + / any corruption (classic kdb+ instead returns a (goodcount;bytes) pair). corruption is therefore + / detected by trapping that throw. NB -11!(-1;...) also counts but EXECUTES upd, so is not used here. + / does not execute upd and never signals. + `corrupt~@[{-11!(-2;x);`ok};logfile;{`corrupt}] + }; + +logname:{[dir;date] + / build the log file handle for an absolute-path dir (string) and a date; one file per date, + / /tp, e.g. `:/var/tplog/tp2026.08.13 + :`$":",dir,"/tp",string date; + }; + +open:{[dir;date] + / open (creating if absent) the log for date under dir. absent: create empty, return (handle;0). + / present: count with the non-executing -11!(-2;...) FIRST, so a corrupt log fails fast BEFORE any + / partial replay mutates state (a tickerplant must not continue on a bad log - use replay to recover + / instead). clean: replay through the root-level upd exactly once, return (handle;count). + l:logname[dir;date]; + if[not type key l; + .z.m.loginfo[`open;"creating new log ",1_string l]; + .[l;();:;()]; + :(hopen l;0)]; + cnt:@[{-11!(-2;x)};l;{[lf;e] raiseerror[`open;"corrupt log ",(1_string lf),": ",e," - use replay to recover"]}[l;]]; + .z.m.loginfo[`open;"replaying ",(string cnt)," message(s) from ",1_string l]; + -11! l; + :(hopen l;cnt); + }; + +write:{[h;msg] + / append one message (typically (`upd;t;x)) to an open log handle + h enlist msg; + }; + +roll:{[h;dir;olddate] + / roll to the next day's log: close the current handle, open (create) the olddate+1 log + .z.m.loginfo[`roll;"rolling log for ",string olddate]; + hclose h; + :open[dir;olddate+1]; + }; + +replay:{[logfile] + / replay a log through the root-level upd, repairing first if corrupt (recovers rather than failing - + / for consumers like an rdb on startup). corruption is checked with the non-executing corruptp FIRST, + / so good messages before the corruption point are never replayed twice (a naive trap-and-retry would + / partially replay before throwing, then replay again). returns the replayed message count. + good:$[corruptp logfile;repair logfile;logfile]; + :-11! good; + }; + +replayupto:{[logfile;n] + / replay only the first n messages of a log through the root upd (repair-aware). for a subscriber on + / startup replaying exactly its pre-subscription rowcount, so live messages that arrive after + / subscription are not double-processed. n>=good-count replays the whole (repaired) log. + good:$[corruptp logfile;repair logfile;logfile]; + :-11!(n;good); + }; + +check:{[logfile;lastmsgtoreplay] + / return logfile if it is usable as-is, else a repaired .good. lastmsgtoreplay is the index + / of the last message the caller intends to replay; it is retained for signature compatibility with + / TorQ's .tplog.check, but on kdb-x the "corrupt yet enough good messages, skip repair" optimisation + / is unavailable (-11! throws rather than returning a partial good-count), so any corruption repairs. + .z.m.loginfo[`check;"checking ",(1_string logfile)," (caller replays up to index ",(string lastmsgtoreplay),")"]; + if[not corruptp logfile; + .z.m.loginfo[`check;"log is clean - using as-is"]; + :logfile]; + .z.m.logwarn[`check;"log is corrupt - writing a repaired good log"]; + :repair logfile; + }; + +repair:{[logfile] + / scan a corrupt log in chunks and write every recoverable message to .good, returning that + / handle. best-effort: only messages that deserialise are kept, so unrecoverable messages are dropped. + goodlog:`$string[logfile],".good"; + .z.m.loginfo[`repair;"writing recovered messages to ",1_string goodlog]; + goodlogh:hopen goodlog set (); + repairover[logfile;goodlogh] over `start`size!(0j;chunk); + hclose goodlogh; + .z.m.loginfo[`repair;"finished repairing ",1_string logfile]; + :goodlog; + }; + +repairover:{[logfile;goodlogh;d] + / internal - one pass of the chunked byte-scan recovery, driven by `over` on a `start`size dict. + / d has keys start (offset to read from) and size (bytes to read); returns the next d, or d itself + / at eof to terminate the scan. + / read bytes from + x:read1 logfile,d`start`size; + / find the start points of upd messages + u:ss[`char$x;updmsg]; + if[not count u; + / nothing in this block - stop at eof, else move on one chunk + if[hcount[logfile]<=sum d`start`size;:d]; + :@[d;`start;+;d`size]]; + / split bytes into candidate messages + m:u _ x; + / message sizes as bytes + mz:0x0 vs' `int$ 8+ms:count each m; + / set each message size into the correct header bytes + hd:@[header;7 6 5 4;:;] each mz; + / try to deserialise each candidate; g is a list of (ok;value) pairs + g:@[(1b;)@-9!;;(0b;)@] each hd,'m; + / write the good messages to the good log + goodlogh g[;1] where k:g[;0]; + if[not any k; + / saw candidate(s) but none deserialised - give up past maxchunk, else read a bigger chunk + if[maxchunk<=d`size;:@[d;`start`size;:;(sum d`start`size;chunk)]]; + :@[d;`size;*;2]]; + / advance to the end of the last good message + ns:d[`start]+sums[ms] last where k; + :@[d;`start`size;:;(ns;chunk)]; + }; + +getapimeta:{[] + / this module's api metadata, one row per CALLABLE api function (NOT init/getapimeta/version - those + / are plumbing di.torq reads by convention, never registered), for di.torq to collect and register + / with di.api. names are bare; di.torq applies the process-wide qualification. one self-contained + / (name;public;descrip;params;return) row per line - flip cols!flip rows. + :flip `name`public`descrip`params`return!flip( + (`logname;1b;"build the log file handle for a dir and date (/tp, one file per date)";"[string dir; date date]";"symbol: log file handle"); + (`open;1b;"open a log (create if absent); replay a clean log through root upd, fail fast if corrupt";"[string dir; date date]";"(int handle; long count)"); + (`write;1b;"append one message (typically (`upd;t;x)) to an open log handle";"[int handle; any msg]";"null"); + (`roll;1b;"close the current handle and open (create) the next day's log";"[int handle; string dir; date olddate]";"(int handle; long count)"); + (`replay;1b;"replay a log through root upd, repairing first if corrupt (recover rather than fail)";"[symbol logfile]";"long: replayed message count"); + (`replayupto;1b;"replay only the first n messages of a log (repair-aware), for subscriber startup";"[symbol logfile; long n]";"long: replayed count"); + (`check;1b;"return the logfile if clean, else a repaired .good";"[symbol logfile; long lastmsgtoreplay]";"symbol: usable log handle"); + (`repair;1b;"scan a corrupt log and write recoverable messages to .good";"[symbol logfile]";"symbol: .good handle")); + }; From 0728f6ec2422cb0b3e9efafb6f33464846d0ba0c Mon Sep 17 00:00:00 2001 From: ascottDI Date: Mon, 17 Aug 2026 10:17:35 +0100 Subject: [PATCH 2/3] fixing testing --- di/tplog/test.csv | 52 +++-- di/tplog/test.q | 485 +++++++++++++--------------------------------- di/tplog/tplog.md | 264 ++++++++----------------- di/tplog/tplog.q | 8 +- 4 files changed, 249 insertions(+), 560 deletions(-) diff --git a/di/tplog/test.csv b/di/tplog/test.csv index be2c81a5..5f91ae59 100644 --- a/di/tplog/test.csv +++ b/di/tplog/test.csv @@ -1,21 +1,33 @@ action,ms,bytes,lang,code,repeat,minver,comment -before,0,0,q,tplog:use`di.tplog,1,,Initialize module -before,0,0,q,os:use`di.os,1,,Initialize module -/ note location of the test.q file might be different, built under assumption using di module file system -before,0,0,q,"system ""l "", os.abspath[""di/tplog/test.q""]",1,1,load additional testing functions / dependencies -run,0,0,q,testrepairandreplay[],1,1, -run,0,0,q,testrepairrecoversmessages[],1,1, -run,0,0,q,testrepaircreatesgoodfile[],1,1, -run,0,0,q,testcheckvalidlog[],1,1, -run,0,0,q,testcheckcorruptsufficientmessages[],1,1, -run,0,0,q,testrepaircreatesgoodfile[],1,1, -run,0,0,q,testrepairrecoversmessages[],1,1, -run,0,0,q,testchecktriggersrepair[],1,1, -run,0,0,q,testrepairgarbageatend[],1,1, -run,0,0,q,testmultiplecorruptsections[],1,1, -run,0,0,q,testcompletelycorruptlog[],1,1, -run,0,0,q,testemptylog[],1,1, -run,0,0,q,testrepairandreplay[],1,1, -run,0,0,q,testlargefilehandling[],1,1, -run,0,0,q,testrepaircreatesgoodfile[],1,1, -run,0,0,q,testsequentialoperations[],1,1, +before,0,0,q,tp:use`di.tplog,1,,load the module under test +before,0,0,q,os:use`di.os,1,,os module for portable path resolution +before,0,0,q,"system ""l "",os.abspath[""di/tplog/test.q""]",1,,load fixture helpers (defines root upd + trade + helpers) +before,0,0,q,tp.init[enlist[`log]!enlist capturelog[]],1,,init with a capturing logger (the required log dependency) +before,0,0,q,setupfixture[],1,,create the temp fixture root +comment,,,,,,,module metadata - version and getapimeta +true,0,0,q,10h=type tp`version,1,1,version is a string +true,0,0,q,0/tp handle +comment,,,,,,,open / write / roll lifecycle +true,0,0,q,testopenfresh[],1,1,a freshly created log opens with count 0 +true,0,0,q,testwritereopen[],1,1,write two then reopen replays both through the root upd +true,0,0,q,testroll[],1,1,roll closes the handle and creates the next day's log +fail,0,0,q,testopenfailsfast[],1,1,open fails fast on a corrupt log +comment,,,,,,,replay / replayupto (repair-aware) +true,0,0,q,testreplayrepairs[],1,1,replay repairs a corrupt log and recovers messages +true,0,0,q,testreplaynodouble[],1,1,replay processes each recovered message exactly once +true,0,0,q,testreplayupto[],1,1,replayupto replays only the first n messages +comment,,,,,,,check / repair corruption utilities +true,0,0,q,testcheckclean[],1,1,check returns a clean log unchanged +true,0,0,q,testcheckcorruptwarns[],1,1,check repairs a corrupt log and logs a warning +true,0,0,q,testrepaircreatesgood[],1,1,repair writes a .good file +after,0,0,q,teardownfixture[],1,,remove the temp fixture root diff --git a/di/tplog/test.q b/di/tplog/test.q index 953f3fba..afb51591 100644 --- a/di/tplog/test.q +++ b/di/tplog/test.q @@ -1,360 +1,137 @@ -/ ============================================================================= -/ TEST HELPERS -/ ============================================================================= - -upd:{[t;x] t upsert x}; -trade:([] time:`timestamp$(); sym:`symbol$(); price:`float$(); size:`long$()); - -/ @function createvalidlog -/ @description Create a valid tickerplant log file for testing -/ @param filepath {symbol} Path where to create the log file -/ @param msgcount {long} Number of messages to write -createvalidlog:{[filepath;msgcount] - / create test table - trade:([] time:.z.p + til msgcount; sym:msgcount?`AAPL`GOOGL`MSFT`AMZN`TSLA; price:100+msgcount?100.0; size:100+msgcount?1000); - / create log file and write messages - h:hopen filepath set (); - {[h;i;t] h enlist (`upd;`trade;value t[i])} [h;;trade] each til msgcount; +/ fixture helpers for di.tplog's k4unit tests. +/ di.tplog replays through the ROOT-level upd, so a recorder upd + a trade schema are defined here at +/ root. message tuples carry commas/backticks, so they are built in these helpers and never written +/ inline in test.csv (raw commas would break the csv fields). the module handle `tp` and the module's +/ init are set up by test.csv's `before` rows before any of these run. + +base:"/tmp/di_tplog_k4unit" +d:2026.08.13 + +/ root-level replay target + a recorder that also captures what was replayed +trade:([] time:`timestamp$(); sym:`symbol$(); price:`float$()) +replayed:() +resetstate:{[] `replayed set (); `trade set 0#trade;} +upd:{[t;x] `replayed set replayed,enlist(t;x); t insert x;} + +/ capturing logger - one row per call, so tests can assert on narration and the log contract +logcap:([] level:`symbol$(); ctx:`symbol$(); msg:()) +capturelog:{[] + `logcap set 0#logcap; + `info`warn`error!( + {[c;m] `logcap insert (`info;c;m);}; + {[c;m] `logcap insert (`warn;c;m);}; + {[c;m] `logcap insert (`error;c;m);}) + } + +/ a single trade message matching the trade schema above +trademsg:{[ts;s] (`upd;`trade;(enlist ts;enlist s;enlist 1.0))} + +setupfixture:{[] system "rm -rf ",base; system "mkdir -p ",base;} +teardownfixture:{[] system "rm -rf ",base;} + +/ a fresh, empty per-test directory; returns the dir string +freshdir:{[sub] dd:base,"/",sub; system "rm -rf ",dd; system "mkdir -p ",dd; dd} + +/ write n trade messages into a fresh clean log under sub; close; return the log filename handle +writelog:{[sub;n] + dd:freshdir sub; + r:tp[`open][dd;d]; h:r 0; + tp[`write][h;] each trademsg[;`AAPL] each d+0D00:01*til n; hclose h; - }; + tp[`logname][dd;d] + } -/ @function createcorruptlog -/ @description Create a log file with valid messages followed by corruption -/ @param filepath {symbol} Path where to create the log file -/ @param msgcount {long} Number of messages in log file -/ @param corruptpos {long} Message position where to insert corruption -createcorruptlog:{[filepath;msgcount;corruptpos] - / create test table - trade:([] time:.z.p + til msgcount; sym:msgcount?`AAPL`GOOGL`MSFT`AMZN`TSLA; price:100+msgcount?100.0; size:100+msgcount?1000); - / create log file and write messages - h:hopen filepath set (); - {[h;i;t;corruptpos] - if[=[i;corruptpos]; - data:enlist (`upd;`trade;value t[i]); - databytes:-18!data; - data_bytes[10+til 20]:`byte$(20?50); - :h data_bytes; - ] - h enlist (`upd;`trade;value t[i]) - } [h;;trade;corruptpos] each til msgcount; - hclose h; - }; - -/ @function countLogMessages -/ @description Count number of messages in a log file -/ @param filepath {symbol} Path to log file -/ @returns {long} Number of messages in the log -countlogmessages:{[filepath] - count -11!(1;filepath) - }; - -/ @function cleanup -/ @description Delete test files -/ @param filepaths {symbol[]} List of file paths to delete -cleanup:{[filepaths] - {[fp] @[hdel;fp;{}]} each filepaths; - }; +/ smash k bytes near the middle of a log file, in place; returns the filename +corrupt:{[fn;k] + b:read1 fn; + p:count[b] div 2; + fn set @[b;p+til k&count[b]-p;:;k#0xff]; + fn + } / ============================================================================= -/ BASIC FUNCTIONALITY TESTS +/ tests (each returns 1b on success) / ============================================================================= -/ @test Valid log file tplog.check returns original filepath -testcheckvalidlog: { - testfile:`:test_valid.log; - msgcount:10; - - / setup - createvalidlog[testfile;msgcount]; - - / test - result:tplog.check[testfile;msgcount-1]; - - / assert - passes:result~testfile; - - / cleanup - cleanup enlist testfile; - - / Return - passes - }; - -/ @test tplog.check returns original when enough good messages exist -testcheckcorruptsufficientmessages:{ - testfile:`:test_corrupt_sufficient.log; - validmsgcount:20; - lastmsgtoreplay:10j; - - / setup: corrupt after position where we have enough good messages - createcorruptlog[testfile;validmsgcount;500]; - - / test - result:tplog.check[testfile;lastmsgtoreplay]; - - / assert - should return original since we have enough good messages - goodmsgcount:first -11!(-2;testfile); - passes:(result~testfile) and (goodmsgcount > lastmsgtoreplay); - - / cleanup - cleanup enlist testfile; - - passes - }; - -/ @test tplog.repair creates .good file with correct name -testrepaircreatesgoodfile: { - testfile:`:test_tplog.repair.log; - expectedgoodfile:`$string[testfile],".good"; - - / setup - createcorruptlog[testfile;15;150]; - - / test - result:tplog.repair[testfile]; - - / assert - namecorrect:result~expectedgoodfile; - fileexists:not ()~key expectedgoodfile; - passes:namecorrect and fileexists; - - / cleanup - cleanup (testfile;expectedgoodfile); - - passes - }; - -/ @test tplog.repair recovers valid messages from corrupt log -testrepairrecoversmessages: { - testfile:`:test_recover.log; - goodfile:`$string[testfile],".good"; - validmsgcount:20; - - / setup - createcorruptlog[testfile;validmsgcount;250]; - - / test - tplog.repair[testfile]; - - / Count messages in good file - recoveredcount:countlogmessages[goodfile]; - - / assert - should recover at least some messages - passes:(recoveredcount>0) and (recoveredcount<=validmsgcount); - - / cleanup - cleanup (testfile;goodfile); - - passes - }; - -/ @test tplog.check triggers tplog.repair when insufficient good messages -testchecktriggersrepair: { - testfile:`:test_tplog.check_tplog.repair.log; - goodfile:`$string[testfile],".good"; - validmsgcount:10; - lastmsgtoreplay:15j; / Need more messages than available good ones - - / setup - corrupt early so not enough good messages - createcorruptlog[testfile;validmsgcount;100]; - - / test - result:tplog.check[testfile;lastmsgtoreplay]; - - / assert - triggerstplog.repair:result~goodfile; - filecreated:not ()~key goodfile; - passes:triggerstplog.repair and filecreated; - - / cleanup - cleanup (testfile;goodfile); - - passes - }; - -/ ============================================================================= -/ EDGE CASE TESTS -/ ============================================================================= - -/ @test tplog.repair handles garbage at end of file -testrepairgarbageatend: { - testfile:`:test_garbage_end.log; - goodfile:`$string[testfile],".good"; - - / setup - create log and append garbage at end - createvalidlog[testfile;10]; - bytes:read1 testfile; - testfile set bytes,100#0x00; - - / test - result:tplog.repair[testfile]; - - / assert - namecorrect:result~goodfile; - hasMessages:countlogmessages[goodfile]>0; - passes:namecorrect and hasMessages; - - / cleanup - cleanup (testfile;goodfile); - - passes - }; - -/ @test Handles multiple corruption points -testmultiplecorruptsections: { - testfile:`:test_multi_corrupt.log; - goodfile:`$string[testfile],".good"; - - / setup - create log with corruption in middle - createvalidlog[testfile;30]; - bytes:read1 testfile; - - / insert corruption at position (should have valid messages before and after) - if[200 < count bytes; - corrupted:bytes[til 200],10#0xFF,bytes[210+til count[bytes]-210]; - testfile set corrupted; - ]; - - / test - result:tplog.repair[testfile]; - - / assert - should create file and recover something - fileCorrect:result~goodfile; - fileExists:not ()~key goodfile; - passes:fileCorrect and fileExists; - - / cleanup - cleanup (testfile;goodfile); - - passes - }; - -/ @test Completely corrupt log creates empty .good file -testcompletelycorruptlog: { - testfile:`:test_all_corrupt.log; - goodfile:`$string[testfile],".good"; - - / setup - create completely corrupt file - testfile set 1000#0x00; - - / test - result:tplog.repair[testfile]; - - / assert - should create .good file even if empty/minimal - namecorrect:result~goodfile; - fileExists:not ()~key goodfile; - passes:namecorrect and fileExists; - - / cleanup - cleanup (testfile;goodfile); - - passes - }; - -/ @test Empty log file handling -testemptylog: { - testfile:`:test_empty.log; - - / setup - create empty log - testfile set (); - - / test - should not crash - result:tplog.check[testfile;0j]; - - / If we got here without error, test passes - passes:1b; - - / cleanup - cleanup enlist testfile; - - passes - }; - -/ ============================================================================= -/ CONFIGURATION TESTS -/ ============================================================================= - -/ @test Module metadata is present -testmoduleinfo: { - hasname:`name in key info; - hasversion:`version in key info; - hasdesc:`description in key info; - - hasname and hasversion and hasdesc - }; - -/ ============================================================================= -/ INTEGRATION TESTS -/ ============================================================================= - -/ @test tplog.repair then replay workflow -testrepairandreplay: { - testfile:`:test_replay.log; - goodfile:`$string[testfile],".good"; - - / setup - createcorruptlog[testfile;20;200]; - - / test - tplog.repair and try to replay - tplog.repair[testfile]; - - / This should not throw an error if the .good file is valid - replayOk:@[{-11!(1;x);1b};goodfile;{0b}]; - - / cleanup - cleanup (testfile;goodfile); - - replayOk - }; - -/ @test Large file handling (performance test) -testlargefilehandling: { - testfile:`:test_large.log; - goodfile:`$string[testfile],".good"; - msgcount:500; / Reasonable size for testing - - / setup - createcorruptlog[testfile;msgcount;5000]; - - / test - measure time - start:.z.p; - result:tplog.repair[testfile]; - elapsed:`second$.z.p-start; - - / assert - should complete and create file - completed:result~goodfile; - reasonable:elapsed<30; / Should complete in under 30 seconds - passes:completed and reasonable; - - / cleanup - cleanup (testfile;goodfile); - - passes - }; - -/ @test Sequential tplog.check and tplog.repair calls -testsequentialoperations: { - testfile:`:test_sequential.log; - goodfile:`$string[testfile],".good"; - - / setup - createcorruptlog[testfile;15;150]; - - / test - tplog.check then tplog.repair - tplog.checkresult:tplog.check[testfile;20j]; - - / if tplog.check triggered tplog.repair, goodfile should exist - / if not, manually tplog.repair - if[not tplog.checkresult~goodfile; - tplog.repair[testfile]; - ]; - - / assert - .good file should exist in either case - passes:not ()~key goodfile; - - / cleanup - cleanup (testfile;goodfile); - - passes - }; - +/ logname builds /tp +testlogname:{[] (`$":/x/tp2026.08.13")~tp[`logname]["/x";2026.08.13]} + +/ a freshly created log opens with count 0 +testopenfresh:{[] + dd:freshdir"fresh"; + r:tp[`open][dd;d]; hclose r 0; + 0=r 1 + } + +/ write two, reopen replays both through the root upd +testwritereopen:{[] + resetstate[]; + dd:freshdir"wr"; + r:tp[`open][dd;d]; h:r 0; + tp[`write][h;trademsg[d+0D10:00;`AAPL]]; + tp[`write][h;trademsg[d+0D10:01;`MSFT]]; + hclose h; + r2:tp[`open][dd;d]; c:r2 1; hclose r2 0; + (2=c) and (2=count replayed) and (2=count trade) + } + +/ roll closes the current handle and creates the next day's log +testroll:{[] + dd:freshdir"roll"; + r:tp[`open][dd;d]; h:r 0; + tp[`write][h;trademsg[d+0D10:00;`AAPL]]; + r2:tp[`roll][h;dd;d]; hclose r2 0; + not ()~key tp[`logname][dd;d+1] + } + +/ open fails fast (signals) on a corrupt log - used by a `fail` row +testopenfailsfast:{[] + writelog["off";5]; + corrupt[tp[`logname][base,"/off";d];12]; + tp[`open][base,"/off";d] + } + +/ replay repairs a corrupt log and recovers a sensible number of messages +testreplayrepairs:{[] + resetstate[]; + fn:corrupt[writelog["rrp";10];12]; + n:tp[`replay] fn; + (n>0) and n<=10 + } + +/ replay processes each recovered message EXACTLY once (double-processing regression guard) +testreplaynodouble:{[] + resetstate[]; + fn:corrupt[writelog["nd";8];12]; + n:tp[`replay] fn; + n=count replayed + } + +/ replayupto replays ONLY the first n messages +testreplayupto:{[] + resetstate[]; + fn:writelog["upto";4]; + n:tp[`replayupto][fn;2]; + (2=n) and (2=count replayed) + } + +/ check returns a clean log unchanged +testcheckclean:{[] + fn:writelog["ckc";3]; + fn~tp[`check][fn;100] + } + +/ check repairs a corrupt log (returns .good) AND logs a warning under ctx `check +testcheckcorruptwarns:{[] + `logcap set 0#logcap; + fn:corrupt[writelog["ckx";6];12]; + res:tp[`check][fn;100]; + (res~`$string[fn],".good") and `warn in exec level from logcap where ctx=`check + } + +/ repair writes a .good file +testrepaircreatesgood:{[] + fn:corrupt[writelog["rep";5];12]; + g:tp[`repair] fn; + (g~`$string[fn],".good") and not ()~key g + } diff --git a/di/tplog/tplog.md b/di/tplog/tplog.md index 25542e07..3cf0a30d 100644 --- a/di/tplog/tplog.md +++ b/di/tplog/tplog.md @@ -1,210 +1,110 @@ -# `tplog` – Tickerplant Log Check & Repair Utilities for kdb+/q +# di.tplog -A small utility module for **checking** and **best‑effort repairing** tickerplant-style log files by scanning raw bytes for update-message boundaries, attempting to deserialize candidate messages, and writing any recoverable messages into a new `*.good` logfile. +Tickerplant **log lifecycle** — create/open, append, roll, and replay-on-startup — together with +best-effort **corruption check/repair**. It is the modular replacement for TorQ's inline log handling +in `TorQ/code/processes/tickerplant.q` (`.u.ld` / `.u.endofday`) plus the recovery utilities in +`TorQ/code/common/tplogutils.q`, folded into a single import surface. -> **Note:** As currently implemented, recovery is keyed off the signature of `(`upd;`trade;...)` (see **Configuration**). If your logs contain other tables or message shapes, you may need to adapt the signature constants. +The module is **self-contained**: it has no hard `use` dependencies and is built on base q only. Its +one runtime dependency, a logger, is **injected** via `init`. ---- - -## :sparkles: Features - -- Check whether a logfile should be used as-is or repaired (based on the logic in `check`). -- Repair a corrupt logfile by extracting messages that can be successfully deserialized. -- Chunked scanning to avoid loading large files into memory. -- Adaptive read sizing when no valid messages are found in a chunk. -- Produces a new `.good` output file (append-only write during recovery). -- Includes a test suite (`test.q`, `test.csv`) that generates valid/corrupt logs and validates recovery outcomes. - ---- - -## :file_folder: Directory contents - -- `init.q` – module implementation (constants + `check`, `repair`) -- `tplog.md` – documentation (you can replace/rename to `README.md` if desired) -- `test.q` – tests + helpers for creating valid/corrupted logs -- `test.csv` – test manifest for your project’s test harness - ---- - -## :inbox_tray: Loading - -### KDB-X (supports `use`) -If you are using KDB-X (where `use` exists), load the module using the symbol that matches your `QPATH` layout. - -If your `QPATH` includes the `di` directory (e.g. `~/kdbx-modules/di`), a common pattern is: - -```q -tplog:use`tplog -``` - ---- - -## :gear: Configuration - -These constants are defined at the top of `init.q`: - -| Name | Type | Description | -|------------|-------------|-------------| -| `HEADER` | byte list | Template bytes used to build a deserialisable message header. | -| `UPDMSG` | char list | Prefix used to detect candidate update messages within raw bytes. | -| `CHUNK` | long | Default chunk size (bytes) to read (10MB). | -| `MAXCHUNK` | long | Maximum chunk size for a single read attempt (`8 * CHUNK`). | - -### Current default signature - -The module sets `UPDMSG` based on the serialized form of: - -```q -(`upd;`trade;()) -``` - -This means: -- it is geared toward logs containing `upd` messages for the `trade` table -- logs containing other table names or different update call shapes may not be recovered unless you adjust the signature logic - ---- - -## :wrench: Functions - -### Summary - -| Function | Description | -|----------|-------------| -| `check[logfile;lastmsgtoreplay]` | Returns `logfile` if it should be used as-is per `check` logic, otherwise triggers `repair` and returns `.good`. | -| `repair[logfile]` | Creates `.good` and writes any recoverable messages into it. Returns the new filename. | ---- - -### `check` - -```q -tplog.check[logfile; lastmsgtoreplay] -``` - -**Parameters** - -| Parameter | Type | Description | -|----------:|------|-------------| -| `logfile` | symbol | Path to logfile as a symbol (e.g. ```:tp.log```), as used by `-11!`, `hcount`, `read1`, etc. | -| `lastmsgtoreplay` | long | Index position of the last message the caller intends to replay. | - -**Behavior (as implemented)** -- inspects logfile info via `-11!(-2; logfile)` -- returns either: - - the original `logfile`, or - - a repaired logfile produced by `repair[logfile]` - -**Returns** -- `logfile` **or** `.good` - ---- - -### `repair` - -```q -tplog.repair[logfile] -``` - -**Purpose** -Create a “good” logfile containing only recoverable messages. - -**Behavior (as implemented)** -- writes output to `.good` -- processes the input logfile in chunks -- for each chunk: - - searches for occurrences of the configured `UPDMSG` signature - - splits the chunk into candidate messages - - constructs a header for each candidate - - attempts to deserialize each candidate - - writes successfully decoded messages into the output logfile - -**Returns** -- symbol path of the repaired logfile (e.g. ```:tp.log.good```) - ---- - -## :rocket: Typical usage - -### Repair-if-needed flow +## Import and init ```q -/ Load module -tplog:use`tplog - -/ Decide whether to repair -log:`:tp.log -safe:tplog.check[log; 0j] - -/ safe is either `:tp.log or `:tp.log.good -safe +tp:use`di.tplog + +/ using di.log (its logdict is pre-shaped as `log!(`info`warn`error!...)) +logging:use`di.log +tp.init[logging`logdict] + +/ or a hand-rolled binary logger +mylog:`info`warn`error!( + {[c;m] -1 string[c],": ",m;}; + {[c;m] -1 string[c],": ",m;}; + {[c;m] -2 string[c],": ",m;}) +tp.init[enlist[`log]!enlist mylog] ``` -### Always repair +`init` **must** be called before any other function — there is no default logger. It validates the +`log` dependency strictly and errors immediately if it is missing or malformed (no silent fallback). -```q -tplog:use`tplog +## The `upd` replay contract -log:`:tp.log -good:tplog.repair log -good -``` +`open`, `replay`, and `replayupto` restore state by running `-11!` over the log, which executes the +**root-level `upd`** for each stored `(`upd;t;x)` message. **A caller must define a root `upd` before +calling any of them.** A tickerplant publishes its `upd` at root during init, ahead of opening the +log; an RDB/subscriber does the same before replaying. ---- +## Exported functions -## :test_tube: Tests +| Function | Signature | Description | +|---|---|---| +| `logname` | `[dir;date]` → `` `:/tp `` | Build the log-file handle for an absolute-path `dir` string and a `date` (one file per date). | +| `open` | `[dir;date]` → `(handle;count)` | Open (creating if absent) the log. Absent → empty log, count `0`. Present → **fail fast** if corrupt, else replay through the root `upd` once and return the message count. | +| `write` | `[handle;msg]` → `` (::) `` | Append one message (typically `` (`upd;t;x) ``) to an open handle. | +| `roll` | `[handle;dir;olddate]` → `(handle;count)` | Close `handle` and open (create) the `olddate+1` log. | +| `replay` | `[logfile]` → `count` | Replay through the root `upd`, **repairing first if corrupt** (recovers rather than failing). Returns the replayed count. | +| `replayupto` | `[logfile;n]` → `count` | Replay only the **first `n`** messages (repair-aware). For a subscriber replaying exactly its pre-subscription rowcount so later live-and-logged messages are not double-processed. | +| `check` | `[logfile;lastmsgtoreplay]` → `logfile` \| `` `.good `` | Return `logfile` if usable as-is, else a repaired `` `.good ``. | +| `repair` | `[logfile]` → `` `.good `` | Scan a corrupt log in chunks and write every recoverable message to `` `.good ``. | -The module includes `test.q` and `test.csv`. +`getapimeta[]` and `version` are also exported, as module metadata for `di.torq` / `di.depcheck` — not +callable API. -### What the tests do (high level) +## Injectable dependencies -`test.q` provides helpers to: -- create a valid log by writing records shaped like `enlist (`upd;`trade; rowData)` -- create a corrupt log by introducing byte-level corruption into one record -- verify that `check` and `repair` behave as expected across scenarios: - - valid logs - - corruption with enough valid messages - - corruption requiring repair - - garbage at end-of-file - - multiple corrupt sections - - completely corrupt logs - - empty logs - - sequential operations +| Injectable | Required keys | Signature | +|---|---|---| +| `log` (required) | `` `info`warn`error `` | `{[ctx;msg]}` — context symbol, message string. Extra levels (e.g. di.log's six) are accepted and ignored. | -### Running tests manually +No config keys beyond `log` are accepted. There are **no hard `use` dependencies** (`deps.q` is empty). -```q -/ Load module -tplog:use`tplog +## Design notes -/ Load tests -\l /path/to/kdbx-modules/di/tplog/test.q +### Observer/decider classification -/ Run a few key tests -test_check_valid_log[] -test_repair_creates_good_file[] -test_repair_recovers_messages[] -test_repair_garbage_at_end[] -``` +`di.tplog` registers **no `.z.*` handlers** — it is a pure log-file utility invoked directly by the +tickerplant (`open`/`write`/`roll`) and by subscribers on startup (`replay`/`replayupto`). It therefore +sits outside the observer/decider handler model entirely and never touches `di.handlers`. -> **Note:** If the tests reference the module under a different name than the one used when loading it, either: -> - load the module under the expected name as well, or -> - update the test references to use the loaded module name. +### KDB-X `-11!` behaviour (verified on the installed build) ---- +This module's corruption handling was rebuilt against **measured** KDB-X behaviour, which diverges from +classic kdb+ (and from assumptions in the TorqX POC): -## :bulb: Notes & limitations +- **`-11!(-2;logfile)` is the non-executing primitive.** On a clean log it returns the message count + **without running `upd`**; on **any** corruption it **throws** (it does *not* return a + `(goodcount;bytes)` 2-list as classic kdb+ does). Corruption is therefore detected by trapping that + throw (`corruptp`), and `open` counts with `-11!(-2)` *before* replaying, so a corrupt log fails fast + before any partial replay mutates state. +- **`-11!(-1;logfile)` also counts but *executes* `upd`** — so it is unsafe for detection and is not + used here (using it caused an early double-replay bug). +- **No double-processing:** `replay`/`replayupto` detect corruption with the non-executing `corruptp`, + repair to a `.good` file (a byte-scan that never calls `upd`), then replay the good log exactly once. + A naive trap-and-retry would partially replay before throwing and then replay again. -- **Best-effort recovery only:** The repair process only keeps messages that can be successfully deserialized by the module’s decode attempt. -- **Signature-specific:** The scan is currently tuned to the prefix of `(`upd;`trade;...)`. -- **Chunk-boundary sensitivity:** Recovery depends on being able to locate the message signature within the bytes read for a given chunk. -- **Validate output:** Always validate that `.good` replays correctly in your environment before using it as a production recovery artifact. +### Known gaps / limitations ---- +- **`repair`'s message signature is hardcoded to the `` (`upd;`trade;…) `` shape** (inherited from + TorQ `tplogutils`). Logs of other tables are recovered only if their messages share that prefix. The + corruption *detection* (`corruptp`, and `open`'s fail-fast) is schema-agnostic; only the *repair* + byte-scan is `trade`-specific. Generalising the signature is future work. +- **`check`'s `lastmsgtoreplay` optimisation is not available on KDB-X.** In classic TorQ, `check` could + skip repair when a corrupt log still held enough good messages for the caller's needs. That relied on + `-11!(-2)` returning a partial good-count, which this build does not do (it throws). The parameter is + retained for signature compatibility, but `check` now conservatively repairs on any corruption. +- **Filename convention is fixed** (`/tp`, one log per date). Sharing a directory between + multiple logical logs would need a prefix parameter on `logname`. -## :package: Exported symbols +## Testing -The module exports: +`test.csv` / `test.q` (k4unit) cover: version/`getapimeta` metadata, strict `init` dependency +validation (a `fail` row per guard), `logname`, the open/write/roll lifecycle, fail-fast `open` on a +corrupt log, `replay` repairing a corrupt log while processing each recovered message exactly once (the +double-processing regression), `replayupto` replaying only the first `n`, and `check`/`repair` on clean +and corrupt logs (asserting the warning is logged via a capturing logger). Run with: ```q -export:([check;repair]) +q)k4unit:use`di.k4unit +q)k4unit.moduletest`di.tplog ``` - diff --git a/di/tplog/tplog.q b/di/tplog/tplog.q index f02d953d..105e3b51 100644 --- a/di/tplog/tplog.q +++ b/di/tplog/tplog.q @@ -166,12 +166,12 @@ getapimeta:{[] / with di.api. names are bare; di.torq applies the process-wide qualification. one self-contained / (name;public;descrip;params;return) row per line - flip cols!flip rows. :flip `name`public`descrip`params`return!flip( - (`logname;1b;"build the log file handle for a dir and date (/tp, one file per date)";"[string dir; date date]";"symbol: log file handle"); - (`open;1b;"open a log (create if absent); replay a clean log through root upd, fail fast if corrupt";"[string dir; date date]";"(int handle; long count)"); + (`logname;1b;"build the log file handle for a dir and date (/tp)";"[string dir; date date]";"symbol: log file handle"); + (`open;1b;"open/create a log; replay a clean one via root upd, fail fast if corrupt";"[string dir; date date]";"(int handle; long count)"); (`write;1b;"append one message (typically (`upd;t;x)) to an open log handle";"[int handle; any msg]";"null"); (`roll;1b;"close the current handle and open (create) the next day's log";"[int handle; string dir; date olddate]";"(int handle; long count)"); - (`replay;1b;"replay a log through root upd, repairing first if corrupt (recover rather than fail)";"[symbol logfile]";"long: replayed message count"); - (`replayupto;1b;"replay only the first n messages of a log (repair-aware), for subscriber startup";"[symbol logfile; long n]";"long: replayed count"); + (`replay;1b;"replay via root upd, repairing first if corrupt (recover not fail)";"[symbol logfile]";"long: replayed message count"); + (`replayupto;1b;"replay only the first n messages (repair-aware), for subscriber startup";"[symbol logfile; long n]";"long: replayed count"); (`check;1b;"return the logfile if clean, else a repaired .good";"[symbol logfile; long lastmsgtoreplay]";"symbol: usable log handle"); (`repair;1b;"scan a corrupt log and write recoverable messages to .good";"[symbol logfile]";"symbol: .good handle")); }; From 6c51adb844e2cc91165d79f7042cb590440c726e Mon Sep 17 00:00:00 2001 From: ascottDI Date: Mon, 17 Aug 2026 16:52:44 +0100 Subject: [PATCH 3/3] final iteration and passover ready for PR --- di/tplog/test.csv | 6 ++ di/tplog/test.q | 4 +- di/tplog/tplog.md | 120 +++++++++++++++------------------ di/tplog/tplog.q | 164 ++++++++++++++++------------------------------ 4 files changed, 120 insertions(+), 174 deletions(-) diff --git a/di/tplog/test.csv b/di/tplog/test.csv index 5f91ae59..e8c72c97 100644 --- a/di/tplog/test.csv +++ b/di/tplog/test.csv @@ -30,4 +30,10 @@ comment,,,,,,,check / repair corruption utilities true,0,0,q,testcheckclean[],1,1,check returns a clean log unchanged true,0,0,q,testcheckcorruptwarns[],1,1,check repairs a corrupt log and logs a warning true,0,0,q,testrepaircreatesgood[],1,1,repair writes a .good file +comment,,,,,,,public-function input validation +fail,0,0,q,tp[`logname][`notastring;2026.08.13],1,1,logname rejects a non-string dir +fail,0,0,q,tp[`replay]["notasymbol"],1,1,replay rejects a non-symbol logfile +fail,0,0,q,tp[`check]["notasymbol"],1,1,check rejects a non-symbol logfile +fail,0,0,q,tp[`repair]["notasymbol"],1,1,repair rejects a non-symbol logfile +fail,0,0,q,tp[`replayupto][tp[`logname]["/x";2026.08.13];`notanint],1,1,replayupto rejects a non-integral n after,0,0,q,teardownfixture[],1,,remove the temp fixture root diff --git a/di/tplog/test.q b/di/tplog/test.q index afb51591..5da0a377 100644 --- a/di/tplog/test.q +++ b/di/tplog/test.q @@ -118,14 +118,14 @@ testreplayupto:{[] / check returns a clean log unchanged testcheckclean:{[] fn:writelog["ckc";3]; - fn~tp[`check][fn;100] + fn~tp[`check] fn } / check repairs a corrupt log (returns .good) AND logs a warning under ctx `check testcheckcorruptwarns:{[] `logcap set 0#logcap; fn:corrupt[writelog["ckx";6];12]; - res:tp[`check][fn;100]; + res:tp[`check] fn; (res~`$string[fn],".good") and `warn in exec level from logcap where ctx=`check } diff --git a/di/tplog/tplog.md b/di/tplog/tplog.md index 3cf0a30d..72811109 100644 --- a/di/tplog/tplog.md +++ b/di/tplog/tplog.md @@ -1,23 +1,23 @@ # di.tplog -Tickerplant **log lifecycle** — create/open, append, roll, and replay-on-startup — together with -best-effort **corruption check/repair**. It is the modular replacement for TorQ's inline log handling -in `TorQ/code/processes/tickerplant.q` (`.u.ld` / `.u.endofday`) plus the recovery utilities in -`TorQ/code/common/tplogutils.q`, folded into a single import surface. +Tickerplant log utilities: create/open a log, append to it, roll to the next day, replay it on +startup, and repair a corrupt one. It is the modular replacement for the inline log handling in +TorQ's `code/processes/tickerplant.q` (`.u.ld` / `.u.endofday`) and the recovery code in +`code/common/tplogutils.q`, folded into one module. -The module is **self-contained**: it has no hard `use` dependencies and is built on base q only. Its -one runtime dependency, a logger, is **injected** via `init`. +Self-contained: no hard `use` dependencies (built on base q). Its one runtime dependency, a logger, +is injected via `init`. ## Import and init ```q tp:use`di.tplog -/ using di.log (its logdict is pre-shaped as `log!(`info`warn`error!...)) +/ with di.log (its logdict is already shaped as `log!(`info`warn`error!...)) logging:use`di.log tp.init[logging`logdict] -/ or a hand-rolled binary logger +/ or a hand-rolled logger mylog:`info`warn`error!( {[c;m] -1 string[c],": ",m;}; {[c;m] -1 string[c],": ",m;}; @@ -25,84 +25,72 @@ mylog:`info`warn`error!( tp.init[enlist[`log]!enlist mylog] ``` -`init` **must** be called before any other function — there is no default logger. It validates the -`log` dependency strictly and errors immediately if it is missing or malformed (no silent fallback). +`init` must be called before anything else — there is no default logger. It validates the `log` +dependency and signals immediately if it is missing or malformed. -## The `upd` replay contract +## The upd replay contract -`open`, `replay`, and `replayupto` restore state by running `-11!` over the log, which executes the -**root-level `upd`** for each stored `(`upd;t;x)` message. **A caller must define a root `upd` before -calling any of them.** A tickerplant publishes its `upd` at root during init, ahead of opening the -log; an RDB/subscriber does the same before replaying. +`open`, `replay`, and `replayupto` restore state with `-11!`, which runs the root-level `upd` for +each stored `(`upd;t;x)` message. A caller must define a root `upd` before calling them: a +tickerplant publishes its `upd` at root during init, before opening the log; a subscriber does the +same before replaying. ## Exported functions | Function | Signature | Description | |---|---|---| -| `logname` | `[dir;date]` → `` `:/tp `` | Build the log-file handle for an absolute-path `dir` string and a `date` (one file per date). | -| `open` | `[dir;date]` → `(handle;count)` | Open (creating if absent) the log. Absent → empty log, count `0`. Present → **fail fast** if corrupt, else replay through the root `upd` once and return the message count. | +| `logname` | `[dir;date]` → `` `:/tp `` | Log-file handle for an absolute-path `dir` string and a `date`, one file per date. | +| `open` | `[dir;date]` → `(handle;count)` | Open the log, creating it if absent (count `0`). If present, fail fast when corrupt, else replay through the root `upd` once and return the message count. | | `write` | `[handle;msg]` → `` (::) `` | Append one message (typically `` (`upd;t;x) ``) to an open handle. | -| `roll` | `[handle;dir;olddate]` → `(handle;count)` | Close `handle` and open (create) the `olddate+1` log. | -| `replay` | `[logfile]` → `count` | Replay through the root `upd`, **repairing first if corrupt** (recovers rather than failing). Returns the replayed count. | -| `replayupto` | `[logfile;n]` → `count` | Replay only the **first `n`** messages (repair-aware). For a subscriber replaying exactly its pre-subscription rowcount so later live-and-logged messages are not double-processed. | -| `check` | `[logfile;lastmsgtoreplay]` → `logfile` \| `` `.good `` | Return `logfile` if usable as-is, else a repaired `` `.good ``. | -| `repair` | `[logfile]` → `` `.good `` | Scan a corrupt log in chunks and write every recoverable message to `` `.good ``. | +| `roll` | `[handle;dir;olddate]` → `(handle;count)` | Close `handle` and open the `olddate+1` log. | +| `replay` | `[logfile]` → `count` | Replay through the root `upd`, repairing first if corrupt. | +| `replayupto` | `[logfile;n]` → `count` | Replay only the first `n` messages (repair-aware) — a subscriber replays up to the point it subscribed, so live messages logged after that are not processed twice. | +| `check` | `[logfile]` → `logfile` \| `` `.good `` | Return `logfile` if usable, else a repaired copy. | +| `repair` | `[logfile]` → `` `.good `` | Scan the log and write every message that still deserialises to `` `.good ``. | -`getapimeta[]` and `version` are also exported, as module metadata for `di.torq` / `di.depcheck` — not -callable API. +`getapimeta[]` and `version` are also exported, as metadata for `di.torq` / `di.depcheck`. ## Injectable dependencies -| Injectable | Required keys | Signature | +| Injectable | Keys | Signature | |---|---|---| -| `log` (required) | `` `info`warn`error `` | `{[ctx;msg]}` — context symbol, message string. Extra levels (e.g. di.log's six) are accepted and ignored. | +| `log` | `` `info`warn`error `` | `{[ctx;msg]}` — context symbol, message string. Extra levels (e.g. di.log's six) are accepted and ignored. | -No config keys beyond `log` are accepted. There are **no hard `use` dependencies** (`deps.q` is empty). +No config keys beyond `log`, and no hard `use` dependencies (`deps.q` is empty). ## Design notes -### Observer/decider classification - -`di.tplog` registers **no `.z.*` handlers** — it is a pure log-file utility invoked directly by the -tickerplant (`open`/`write`/`roll`) and by subscribers on startup (`replay`/`replayupto`). It therefore -sits outside the observer/decider handler model entirely and never touches `di.handlers`. - -### KDB-X `-11!` behaviour (verified on the installed build) - -This module's corruption handling was rebuilt against **measured** KDB-X behaviour, which diverges from -classic kdb+ (and from assumptions in the TorqX POC): - -- **`-11!(-2;logfile)` is the non-executing primitive.** On a clean log it returns the message count - **without running `upd`**; on **any** corruption it **throws** (it does *not* return a - `(goodcount;bytes)` 2-list as classic kdb+ does). Corruption is therefore detected by trapping that - throw (`corruptp`), and `open` counts with `-11!(-2)` *before* replaying, so a corrupt log fails fast - before any partial replay mutates state. -- **`-11!(-1;logfile)` also counts but *executes* `upd`** — so it is unsafe for detection and is not - used here (using it caused an early double-replay bug). -- **No double-processing:** `replay`/`replayupto` detect corruption with the non-executing `corruptp`, - repair to a `.good` file (a byte-scan that never calls `upd`), then replay the good log exactly once. - A naive trap-and-retry would partially replay before throwing and then replay again. - -### Known gaps / limitations - -- **`repair`'s message signature is hardcoded to the `` (`upd;`trade;…) `` shape** (inherited from - TorQ `tplogutils`). Logs of other tables are recovered only if their messages share that prefix. The - corruption *detection* (`corruptp`, and `open`'s fail-fast) is schema-agnostic; only the *repair* - byte-scan is `trade`-specific. Generalising the signature is future work. -- **`check`'s `lastmsgtoreplay` optimisation is not available on KDB-X.** In classic TorQ, `check` could - skip repair when a corrupt log still held enough good messages for the caller's needs. That relied on - `-11!(-2)` returning a partial good-count, which this build does not do (it throws). The parameter is - retained for signature compatibility, but `check` now conservatively repairs on any corruption. -- **Filename convention is fixed** (`/tp`, one log per date). Sharing a directory between - multiple logical logs would need a prefix parameter on `logname`. +The module registers no `.z.*` handlers — it is called directly by the tickerplant +(`open`/`write`/`roll`) and by subscribers on startup (`replay`/`replayupto`), so it sits outside the +`di.handlers` observer/decider model. + +Corruption handling was written against measured KDB-X `-11!` behaviour, which differs from classic +kdb+: + +- `-11!(-2;logfile)` counts a clean log without running `upd`, and throws on any corruption — it does + not return the classic `(goodcount;bytes)` pair. Corruption is detected by trapping that throw + (`corruptp`); `open` counts with it before replaying, so a corrupt log fails fast before any partial + replay. +- `-11!(-1;logfile)` counts but runs `upd`, so it is not used for detection. + +`replay`/`replayupto` check with the non-executing `corruptp`, repair to a `.good` file if needed (a +byte-scan that never calls `upd`), then replay once, so no message is processed twice. + +## Known limitations + +- `repair` is tuned to the `(`upd;`trade;…)` message shape (inherited from `tplogutils`); other tables + are recovered only if their messages share that prefix. Detection (`corruptp`, `open`'s fail-fast) is + schema-agnostic — only the repair byte-scan is trade-specific. +- `check` takes only the logfile. TorQ's `lastmsgtoreplay` argument is dropped: its skip-repair + optimisation needed `-11!(-2)`'s partial good-count, which this build does not provide. +- The filename convention is fixed (`/tp`, one log per date). ## Testing -`test.csv` / `test.q` (k4unit) cover: version/`getapimeta` metadata, strict `init` dependency -validation (a `fail` row per guard), `logname`, the open/write/roll lifecycle, fail-fast `open` on a -corrupt log, `replay` repairing a corrupt log while processing each recovered message exactly once (the -double-processing regression), `replayupto` replaying only the first `n`, and `check`/`repair` on clean -and corrupt logs (asserting the warning is logged via a capturing logger). Run with: +`test.csv` / `test.q` (k4unit, 25 checks) cover the metadata/version contract, strict `init` +validation, public-input validation, the open/write/roll lifecycle, fail-fast `open` on a corrupt log, +`replay` recovering a corrupt log while processing each message exactly once, `replayupto`, and +`check`/`repair` on clean and corrupt logs (asserting the warning via a capturing logger). ```q q)k4unit:use`di.k4unit diff --git a/di/tplog/tplog.q b/di/tplog/tplog.q index 105e3b51..9fbcba70 100644 --- a/di/tplog/tplog.q +++ b/di/tplog/tplog.q @@ -1,177 +1,129 @@ -/ di.tplog - tickerplant log lifecycle plus corruption check/repair, in one module. -/ lifecycle (open/write/roll/replay/replayupto/logname) is ported from the inline log handling -/ in TorQ/code/processes/tickerplant.q (.u.ld / .u.endofday); the byte-scanning check/repair -/ recovery is ported from TorQ/code/common/tplogutils.q. self-contained: no hard `use` deps. -/ log is an injected, required dependency (see init) - best-effort recovery is narrated so silent -/ message drops are observable. version is sourced from the VERSION file in init.q. +/ tickerplant log utilities - open/append/roll/replay and corruption repair. +/ lifecycle ported from TorQ tickerplant.q (.u.ld/.u.endofday), repair from tplogutils.q. -/ --- message-signature constants used by the byte-scan recovery (repairover) --- -/ these are geared to the (`upd;`trade;...) message shape, inherited from TorQ tplogutils; logs of -/ other table shapes are recovered only if their messages share this prefix (see known gaps in .md) -/ header template to rebuild a deserialisable message header +/ (`upd;`trade;...) signature the byte-scan repair looks for - see the repair notes in the .md header:8#-8!(`upd;`trade;()); -/ first bytes of a tp update message, the signature searched for in the raw log updmsg:`char$10#8_-8!(`upd;`trade;()); -/ default chunk to read (10mb) chunk:10*1024*1024; -/ never let a single read exceed this maxchunk:8*chunk; init:{[deps] - / wire the injected log dependency - required, no fallback. deps is a dict with a `log key holding - / `info`warn`error!({[ctx;msg]};...) (extra levels like di.log's six are accepted and ignored). - / examples: - / tp.init[(use`di.log)`logdict] / di.log.logdict is pre-shaped as `log!(...) - / tp.init[enlist[`log]!enlist `info`warn`error!(f;f;f)] - / signalled with a plain ' (not raiseerror) - the logger is not yet wired while init runs. - if[99h<>type deps; - '"di.tplog: deps must be a dict with a `log key"]; - if[not `log in key deps; - '"di.tplog: log dependency is required; pass `info`warn`error functions keyed on `log"]; - if[99h<>type deps`log; - '"di.tplog: log value must be a dict of `info`warn`error functions"]; + / deps`log: an `info`warn`error dict of {[ctx;msg]}, required. signalled plainly - no logger yet. + if[99h<>type deps;'"di.tplog: deps must be a dict with a `log key"]; + if[not `log in key deps;'"di.tplog: log dependency is required"]; + if[99h<>type deps`log;'"di.tplog: log must be a dict of `info`warn`error functions"]; if[not all `info`warn`error in key deps`log; - '"di.tplog: log dict must have `info`warn`error keys; got: ",", " sv string key deps`log]; + '"di.tplog: log needs `info`warn`error, got ",", " sv string key deps`log]; .z.m.loginfo:deps[`log]`info; .z.m.logwarn:deps[`log]`warn; .z.m.logerr:deps[`log]`error; }; raiseerror:{[ctx;msg] - / internal - log an error under ctx via the injected logger, then signal it, so a failure lands in - / the log as well as being thrown. every post-init domain error routes through here. + / log then signal, so the failure lands in the log too .z.m.logerr[ctx;msg]; '"di.tplog: ",string[ctx],": ",msg; }; -corruptp:{[logfile] - / internal - true if the log is unreadable/corrupt. -11!(-2;...) is the NON-EXECUTING mode: on a - / clean log it returns the message count without running upd, and on this kdb-x build it THROWS on - / any corruption (classic kdb+ instead returns a (goodcount;bytes) pair). corruption is therefore - / detected by trapping that throw. NB -11!(-1;...) also counts but EXECUTES upd, so is not used here. - / does not execute upd and never signals. - `corrupt~@[{-11!(-2;x);`ok};logfile;{`corrupt}] - }; +/ true if the log won't cleanly replay. -11!(-2) counts a clean log without running upd and throws +/ on corruption, so trap the throw. (-11!(-1) counts too but runs upd - don't use it here.) +corruptp:{[logfile] `corrupt~@[{-11!(-2;x);`ok};logfile;{`corrupt}]}; logname:{[dir;date] - / build the log file handle for an absolute-path dir (string) and a date; one file per date, - / /tp, e.g. `:/var/tplog/tp2026.08.13 - :`$":",dir,"/tp",string date; + / /tp, one log file per date + if[not 10h=type dir;raiseerror[`logname;"dir must be a string"]]; + if[not -14h=type date;raiseerror[`logname;"date must be a date"]]; + `$":",dir,"/tp",string date }; open:{[dir;date] - / open (creating if absent) the log for date under dir. absent: create empty, return (handle;0). - / present: count with the non-executing -11!(-2;...) FIRST, so a corrupt log fails fast BEFORE any - / partial replay mutates state (a tickerplant must not continue on a bad log - use replay to recover - / instead). clean: replay through the root-level upd exactly once, return (handle;count). + / new log -> (handle;0); existing -> replay once through the root upd, returning (handle;count). + / a corrupt log throws here rather than half-replaying - use replay to recover from one. l:logname[dir;date]; if[not type key l; .z.m.loginfo[`open;"creating new log ",1_string l]; .[l;();:;()]; :(hopen l;0)]; - cnt:@[{-11!(-2;x)};l;{[lf;e] raiseerror[`open;"corrupt log ",(1_string lf),": ",e," - use replay to recover"]}[l;]]; - .z.m.loginfo[`open;"replaying ",(string cnt)," message(s) from ",1_string l]; + cnt:@[{-11!(-2;x)};l;{[lf;e] raiseerror[`open;"corrupt log ",(1_string lf),": ",e]}[l;]]; + .z.m.loginfo[`open;"replaying ",(string cnt)," messages from ",1_string l]; -11! l; - :(hopen l;cnt); + (hopen l;cnt) }; -write:{[h;msg] - / append one message (typically (`upd;t;x)) to an open log handle - h enlist msg; - }; +write:{[h;msg] h enlist msg;}; roll:{[h;dir;olddate] - / roll to the next day's log: close the current handle, open (create) the olddate+1 log + / close the current handle and open the next day's log + if[not -6h=type h;raiseerror[`roll;"handle must be an int"]]; + if[not -14h=type olddate;raiseerror[`roll;"olddate must be a date"]]; .z.m.loginfo[`roll;"rolling log for ",string olddate]; hclose h; - :open[dir;olddate+1]; + open[dir;olddate+1] }; replay:{[logfile] - / replay a log through the root-level upd, repairing first if corrupt (recovers rather than failing - - / for consumers like an rdb on startup). corruption is checked with the non-executing corruptp FIRST, - / so good messages before the corruption point are never replayed twice (a naive trap-and-retry would - / partially replay before throwing, then replay again). returns the replayed message count. - good:$[corruptp logfile;repair logfile;logfile]; - :-11! good; + / repair if corrupt, then replay through the root upd. the non-executing corruptp check up front + / means good messages are not replayed twice. + if[not -11h=type logfile;raiseerror[`replay;"logfile must be a symbol"]]; + -11! $[corruptp logfile;repair logfile;logfile] }; replayupto:{[logfile;n] - / replay only the first n messages of a log through the root upd (repair-aware). for a subscriber on - / startup replaying exactly its pre-subscription rowcount, so live messages that arrive after - / subscription are not double-processed. n>=good-count replays the whole (repaired) log. - good:$[corruptp logfile;repair logfile;logfile]; - :-11!(n;good); + / replay only the first n messages - a subscriber replays up to the point it subscribed + if[not -11h=type logfile;raiseerror[`replayupto;"logfile must be a symbol"]]; + if[not (type n) in -7 -6h;raiseerror[`replayupto;"n must be an int or long"]]; + -11!(n;$[corruptp logfile;repair logfile;logfile]) }; -check:{[logfile;lastmsgtoreplay] - / return logfile if it is usable as-is, else a repaired .good. lastmsgtoreplay is the index - / of the last message the caller intends to replay; it is retained for signature compatibility with - / TorQ's .tplog.check, but on kdb-x the "corrupt yet enough good messages, skip repair" optimisation - / is unavailable (-11! throws rather than returning a partial good-count), so any corruption repairs. - .z.m.loginfo[`check;"checking ",(1_string logfile)," (caller replays up to index ",(string lastmsgtoreplay),")"]; - if[not corruptp logfile; - .z.m.loginfo[`check;"log is clean - using as-is"]; - :logfile]; - .z.m.logwarn[`check;"log is corrupt - writing a repaired good log"]; - :repair logfile; +check:{[logfile] + / logfile if it is usable, else a repaired copy + if[not -11h=type logfile;raiseerror[`check;"logfile must be a symbol"]]; + if[not corruptp logfile;:logfile]; + .z.m.logwarn[`check;"corrupt log, repairing ",1_string logfile]; + repair logfile }; repair:{[logfile] - / scan a corrupt log in chunks and write every recoverable message to .good, returning that - / handle. best-effort: only messages that deserialise are kept, so unrecoverable messages are dropped. + / write every message that still deserialises to .good + if[not -11h=type logfile;raiseerror[`repair;"logfile must be a symbol"]]; goodlog:`$string[logfile],".good"; - .z.m.loginfo[`repair;"writing recovered messages to ",1_string goodlog]; goodlogh:hopen goodlog set (); repairover[logfile;goodlogh] over `start`size!(0j;chunk); hclose goodlogh; - .z.m.loginfo[`repair;"finished repairing ",1_string logfile]; - :goodlog; + .z.m.loginfo[`repair;"repaired ",(1_string logfile)," -> ",1_string goodlog]; + goodlog }; repairover:{[logfile;goodlogh;d] - / internal - one pass of the chunked byte-scan recovery, driven by `over` on a `start`size dict. - / d has keys start (offset to read from) and size (bytes to read); returns the next d, or d itself - / at eof to terminate the scan. - / read bytes from + / one scan pass over a `start`size window; returns the next window, or d unchanged at eof x:read1 logfile,d`start`size; - / find the start points of upd messages u:ss[`char$x;updmsg]; if[not count u; - / nothing in this block - stop at eof, else move on one chunk if[hcount[logfile]<=sum d`start`size;:d]; :@[d;`start;+;d`size]]; - / split bytes into candidate messages m:u _ x; - / message sizes as bytes + / rebuild each candidate's header with its true length, then try to deserialise it mz:0x0 vs' `int$ 8+ms:count each m; - / set each message size into the correct header bytes hd:@[header;7 6 5 4;:;] each mz; - / try to deserialise each candidate; g is a list of (ok;value) pairs g:@[(1b;)@-9!;;(0b;)@] each hd,'m; - / write the good messages to the good log goodlogh g[;1] where k:g[;0]; if[not any k; - / saw candidate(s) but none deserialised - give up past maxchunk, else read a bigger chunk + / nothing readable in the window - grow it, or skip past it once we hit maxchunk if[maxchunk<=d`size;:@[d;`start`size;:;(sum d`start`size;chunk)]]; :@[d;`size;*;2]]; - / advance to the end of the last good message ns:d[`start]+sums[ms] last where k; - :@[d;`start`size;:;(ns;chunk)]; + @[d;`start`size;:;(ns;chunk)] }; getapimeta:{[] - / this module's api metadata, one row per CALLABLE api function (NOT init/getapimeta/version - those - / are plumbing di.torq reads by convention, never registered), for di.torq to collect and register - / with di.api. names are bare; di.torq applies the process-wide qualification. one self-contained - / (name;public;descrip;params;return) row per line - flip cols!flip rows. - :flip `name`public`descrip`params`return!flip( - (`logname;1b;"build the log file handle for a dir and date (/tp)";"[string dir; date date]";"symbol: log file handle"); - (`open;1b;"open/create a log; replay a clean one via root upd, fail fast if corrupt";"[string dir; date date]";"(int handle; long count)"); - (`write;1b;"append one message (typically (`upd;t;x)) to an open log handle";"[int handle; any msg]";"null"); - (`roll;1b;"close the current handle and open (create) the next day's log";"[int handle; string dir; date olddate]";"(int handle; long count)"); - (`replay;1b;"replay via root upd, repairing first if corrupt (recover not fail)";"[symbol logfile]";"long: replayed message count"); - (`replayupto;1b;"replay only the first n messages (repair-aware), for subscriber startup";"[symbol logfile; long n]";"long: replayed count"); - (`check;1b;"return the logfile if clean, else a repaired .good";"[symbol logfile; long lastmsgtoreplay]";"symbol: usable log handle"); - (`repair;1b;"scan a corrupt log and write recoverable messages to .good";"[symbol logfile]";"symbol: .good handle")); + / callable api for di.torq to register with di.api (init/getapimeta/version are plumbing, omitted) + flip `name`public`descrip`params`return!flip( + (`logname;1b;"log file handle for a dir and date (/tp)";"[string dir; date date]";"symbol"); + (`open;1b;"open/create a log, replaying an existing one; fail fast if corrupt";"[string dir; date date]";"(handle;count)"); + (`write;1b;"append a message to an open log handle";"[int handle; any msg]";"null"); + (`roll;1b;"close the handle and open the next day's log";"[int handle; string dir; date olddate]";"(handle;count)"); + (`replay;1b;"replay through the root upd, repairing first if corrupt";"[symbol logfile]";"long count"); + (`replayupto;1b;"replay the first n messages only";"[symbol logfile; long n]";"long count"); + (`check;1b;"logfile if usable, else a repaired copy";"[symbol logfile]";"symbol"); + (`repair;1b;"recover readable messages into .good";"[symbol logfile]";"symbol")) };