::nx::Class ::nsmcp::AccessRecordStoreSQLite

Methods: All Methods Documented Methods Hide Methods
Source: Display Source Hide Source
Variables: Show Variables Hide Variables

Class ::nsmcp::AccessRecordStoreSQLite

::nsmcp::AccessRecordStoreSQLite create ... \
           [ -attached (default "false") ] \
           [ -db (default "") ] \
           [ -handle (default "") ] \
           [ -lastCompleted (default "0") ] \
           [ -lastStamp (default "") ] \
           [ -mode (default "writer") ] \
           [ -path path ] \
           [ -peers (default "") ] \
           [ -prior (default "") ] \
           [ -querying (default "false") ] \
           [ -settings:required settings:required ]

Defined in /usr/local/ns/tcl/nsmcp/log-traffic-record-store.tcl

Class Relations

class: ::nx::Class
superclass: ::nx::Object

::nx::Class create ::nsmcp::AccessRecordStoreSQLite \
     -superclass ::nx::Object

Methods (to be applied on instances)

allScope (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> allScope

Testcases:
No testcase defined.

if {${:mode} eq "reader"} {${:db} execute {PRAGMA query_only=OFF}}
${:db} execute BEGIN; set :querying true
${:db} execute {DROP VIEW IF EXISTS temp.nsmcp_record_window; DROP TABLE IF EXISTS temp.nsmcp_record_scope;
    CREATE TEMP TABLE nsmcp_record_scope(fid INTEGER PRIMARY KEY,lo INTEGER,hi INTEGER);
    INSERT INTO temp.nsmcp_record_scope SELECT fid,0,high FROM nsmcp_records.files WHERE active=1}
set floor [:stateGet retentionFloor 0]
${:db} execute "CREATE TEMP VIEW nsmcp_record_window AS SELECT r.* FROM nsmcp_records.rows r JOIN temp.nsmcp_record_scope s ON r.fid=s.fid AND r.off>=s.lo AND r.stop<=s.hi WHERE r.kind<>0 OR r.t>=$floor"
if {${:mode} eq "reader"} {${:db} execute {PRAGMA query_only=ON}}

available (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> available

Testcases:
No testcase defined.

return [expr {${:db} ne {}}]

close (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> close

Testcases:
No testcase defined.

:destroy

coveredBytes (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> coveredBytes fid lo \
    hi

Parameters:

fid (required)

lo (required)

hi (required)

Testcases:
No testcase defined.

return [${:db} scalar {SELECT coalesce(sum(min(hi,$hi)-max(lo,$lo)),0) FROM nsmcp_records.ranges WHERE fid=$fid AND hi>$lo AND lo<$hi}]

coveredEnd (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> coveredEnd fid \
    offset end

Parameters:

fid (required)

offset (required)

end (required)

Testcases:
No testcase defined.

set hi [${:db} scalar {SELECT coalesce(max(hi),-1) FROM nsmcp_records.ranges WHERE fid=$fid AND lo<=$offset AND hi>$offset}]
return [expr {$hi<0 ? $offset : min($hi,$end)}]

descriptors (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> descriptors

Testcases:
No testcase defined.

set result {}; set i 0; set control [:stateGet continuous {}]
foreach row [${:db} execute {SELECT fid,identity,high FROM nsmcp_records.files WHERE active=1 ORDER BY fid}] {
    set fid [dict get $row fid]
    set live [expr {[dict exists $control live $fid]}]
    lappend result [dict create archiveFid $fid identity [dict get $row identity] fileIndex $i source 0 rotation [expr {!$live}] compressed false size [dict get $row high] offset [dict get $row high] status complete]
    incr i
}
return $result

diagnostic (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> diagnostic message \
    opts

Parameters:

message (required)

opts (required)

Testcases:
No testcase defined.

return [accessAggregationDiagnostic record-store $message $opts ${:handle}]

diskBytes (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> diskBytes

Testcases:
No testcase defined.

set total 0
foreach suffix {{} -journal -wal -shm} {if {[file exists ${:path}$suffix]} {incr total [file size ${:path}$suffix]}}
return $total

dropState (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> dropState fid \
    offset

Parameters:

fid (required)

offset (required)

Testcases:
No testcase defined.

return [${:db} scalar {SELECT CASE WHEN dropping=1 AND high<=$offset THEN 1 ELSE 0 END FROM nsmcp_records.files WHERE fid=$fid}]

finish (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> finish

Testcases:
No testcase defined.

${:db} execute COMMIT; set :querying false

gapEnd (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> gapEnd fid offset \
    end

Parameters:

fid (required)

offset (required)

end (required)

Testcases:
No testcase defined.

set lo [${:db} scalar {SELECT coalesce(min(lo),-1) FROM nsmcp_records.ranges WHERE fid=$fid AND lo>$offset}]
return [expr {$lo<0 ? $end : min($lo,$end)}]

ingest (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> ingest fid file \
    begin end tuples

Parameters:

fid (required)

file (required)

begin (required)

end (required)

tuples (required)

Testcases:
No testcase defined.

if {[${:db} scalar {SELECT count(*) FROM nsmcp_records.ranges WHERE fid=$fid}]>=4096} {error "record-store-range-limit"}
set f [open [dict get $file path] rb]
try {seek $f [expr {max(0,$end-64)}]; set anchor [ns_crypto::md string -digest sha256 -binary -- [read $f [expr {min(64,$end)}]]]} finally {close $f}
set stamp [dict get $file mtime]; set now [clock seconds]
set dropping [expr {[dict exists $file dropping] ? [dict get $file dropping] : 0}]
${:db} transaction {
    if {![${:db} scalar {SELECT count(*) FROM nsmcp_records.files WHERE fid=$fid}]} {error "record-store-generation-changed"}
    if {[llength $tuples]} {
        ${:db} execute {DELETE FROM temp.nsmcp_ingest_batch}
        ${:db} execute "INSERT INTO temp.nsmcp_ingest_batch VALUES [join $tuples ,]"
        ${:db} execute {DELETE FROM temp.nsmcp_ingest_batch WHERE EXISTS(SELECT 1 FROM nsmcp_records.rows r WHERE r.fid=nsmcp_ingest_batch.fid AND r.off=nsmcp_ingest_batch.off);
            INSERT INTO nsmcp_records.rows SELECT * FROM temp.nsmcp_ingest_batch}
        :rollBatch
        :observeBatch
    }
    if {[dict size ${:peers}]} {
        set values {}; dict for {key data} ${:peers} {
            lassign $data address seen
            lappend values "(X'$key',[${:db} quote $address],[${:db} quote $seen])"
        }
        ${:db} execute "INSERT INTO nsmcp_records.peer_addresses VALUES [join $values ,] ON CONFLICT(k) DO UPDATE SET address=CASE WHEN excluded.seen>=seen THEN excluded.address ELSE address END,seen=max(seen,excluded.seen)"
    }
    set lo $begin; set hi $end
    foreach span [${:db} execute {SELECT lo,hi,anchor FROM nsmcp_records.ranges WHERE fid=$fid AND lo<=$end AND hi>=$begin}] {
        set lo [expr {min($lo,[dict get $span lo])}]
        if {[dict get $span hi]>$hi} {set hi [dict get $span hi]; set anchor [dict get $span anchor]}
    }
    ${:db} execute {DELETE FROM nsmcp_records.ranges WHERE fid=$fid AND lo<=$end AND hi>=$begin;
        INSERT INTO nsmcp_records.ranges VALUES($fid,$lo,$hi,$anchor);
        UPDATE nsmcp_records.files SET dropping=CASE WHEN $end>=high THEN $dropping ELSE dropping END,stamp=CASE WHEN $end>=high THEN $stamp ELSE stamp END,high=max(high,$end),touched=$now WHERE fid=$fid}
    if {[:diskBytes]>[dict get ${:settings} maxBytes]} {error "record-store-disk-limit"}
}
set :peers {}

legacyViews (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> legacyViews

Testcases:
No testcase defined.

if {${:mode} eq "reader"} {${:db} execute {PRAGMA query_only=OFF}}
set floor [:stateGet retentionFloor 0]
set peerFloor [expr {[clock seconds]-[dict get ${:settings} peerRetention]}]
${:db} execute {DROP VIEW IF EXISTS temp.nsmcp_access_requests; DROP VIEW IF EXISTS temp.nsmcp_access_metadata; DROP VIEW IF EXISTS temp.nsmcp_access_evictions}
${:db} execute "CREATE TEMP VIEW nsmcp_access_requests AS SELECT r.fid AS generation,r.off AS offset,r.t AS start,r.completed AS completed,r.identifier AS request_id,
    CASE WHEN length(r.path)>512 THEN '' ELSE r.path END AS path,r.pool,r.method,r.status,r.bytes,r.fallback,(length(r.path)>512) AS excluded,
    CASE WHEN r.phase THEN r.a ELSE NULL END AS accept,CASE WHEN r.phase THEN r.q ELSE NULL END AS queue,
    CASE WHEN r.phase THEN r.f ELSE NULL END AS filter,CASE WHEN r.phase THEN r.r ELSE NULL END AS run,
    CASE WHEN r.t>=$peerFloor THEN coalesce(p.address,'') ELSE '' END AS peer,r.auth AS authenticationState,r.label AS userAgent
    FROM nsmcp_records.rows r JOIN nsmcp_records.files f ON r.fid=f.fid AND f.active=1
    LEFT JOIN nsmcp_records.peer_addresses p ON r.peer=p.k AND p.seen>=$peerFloor WHERE r.kind=0 AND r.t>=$floor"
${:db} execute {CREATE TEMP VIEW nsmcp_access_metadata AS SELECT key,value FROM nsmcp_records.state;
    CREATE TEMP VIEW nsmcp_access_evictions AS SELECT * FROM nsmcp_records.evictions}
if {${:mode} eq "reader"} {${:db} execute {PRAGMA query_only=ON}}

observeBatch (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> observeBatch

Testcases:
No testcase defined.

set row [lindex [${:db} execute {SELECT coalesce(min(CASE WHEN kind=0 THEN t END),0) AS first,coalesce(max(t),0) AS latest,coalesce(max(CASE WHEN kind=0 THEN completed END),0) AS completed,coalesce(sum(kind=1),0) AS malformed,coalesce(sum(kind=2),0) AS oversized FROM temp.nsmcp_ingest_batch}] 0]
set old [:stateGet observations {first 0 latest 0 completed 0 malformed 0 oversized 0}]
set first [dict get $row first]
if {$first>0 && ([dict get $old first]==0 || $first<[dict get $old first])} {dict set old first $first}
foreach key {latest completed} {dict set old $key [expr {max([dict get $old $key],[dict get $row $key])}]}
foreach key {malformed oversized} {dict incr old $key [dict get $row $key]}
:statePut observations $old

overview (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> overview

Testcases:
No testcase defined.

set report [:stateGet continuous {}]; set observed [:stateGet observations {}]
set floor [:stateGet retentionFloor 0]; set now [clock seconds]
set first [expr {[dict exists $observed first] && [dict get $observed first]>0 ? [dict get $observed first] : {}}]
set latest [expr {[dict exists $observed latest] && [dict get $observed latest]>0 ? [dict get $observed latest] : {}}]
set completed [expr {[dict exists $observed completed] && [dict get $observed completed]>0 ? [dict get $observed completed] : {}}]
set size [expr {[dict exists $report observedSize] ? [dict get $report observedSize] : 0}]
set cursor [expr {[dict exists $report cursor] ? [dict get $report cursor] : 0}]
set history [expr {[dict exists $report historyUnreadBytes] ? [dict get $report historyUnreadBytes] : 1}]
set unsupported [expr {[dict exists $report unsupportedCompressed] ? [dict get $report unsupportedCompressed] : 0}]
if {$unsupported} {incr history $unsupported}
set recent [expr {[dict exists $report recentFrom] ? [dict get $report recentFrom] : $now}]
set updated [expr {[dict exists $report updatedAt] ? [dict get $report updatedAt] : 0}]
return [dict create observedSize $size cursor $cursor historyEnd $history historyCursor 0 recentFrom $recent initialOffset 0 recentInitialized [expr {$updated>0}] seek {}  updatedAt $updated firstObserved $first latestObserved $latest latestCompletionObserved $completed pending [expr {[dict exists $report pending] ? [dict get $report pending] : 0}] generation 1 rotationGaps 0 lastRotationAt 0  malformed [expr {[dict exists $observed malformed] ? [dict get $observed malformed] : 0}] oversized [expr {[dict exists $observed oversized] ? [dict get $observed oversized] : 0}] lastMalformedAt {}  ingestionLagSeconds [expr {[dict exists $report caughtUp] && [dict get $report caughtUp] ? max(0,$now-$updated) : ($completed eq {} ? max(0,$now-$updated) : max(0,$now-$completed))}] retentionFloor $floor]

prepare (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> prepare file

Parameters:

file (required)

Testcases:
No testcase defined.

file stat [dict get $file path] actual
set identity [dict get $file identity]
set policy [expr {[dict exists ${:settings} policy] ? [dict get ${:settings} policy] : {}}]
set key [ns_crypto::md string -digest sha256 -- [list schema1 $identity $policy]]
set fid {}
set old [${:db} execute {SELECT * FROM nsmcp_records.files WHERE k=$key}]
set f [open [dict get $file path] rb]
try {
    set reset false
    if {$old ne {}} {
        set old [lindex $old 0]; set fid [dict get $old fid]
        set prefix [ns_crypto::md string -digest sha256 -binary -- [read $f [dict get $old plen]]]
        if {[dict get $old high]>$actual(size) || $prefix ne [dict get $old prefix]
            || ([dict get $old high]==$actual(size) && [dict get $old stamp]!=$actual(mtime))} {set reset true}
        if {!$reset} {
            # A small boundary anchor per contiguous imported range
            # detects rewrites near committed cursors before reuse.
            foreach span [${:db} execute {SELECT hi,anchor FROM nsmcp_records.ranges WHERE fid=$fid LIMIT 4097}] {
                set hi [dict get $span hi]; seek $f [expr {max(0,$hi-64)}]
                if {[ns_crypto::md string -digest sha256 -binary -- [read $f [expr {min(64,$hi)}]]] ne [dict get $span anchor]} {set reset true; break}
            }
        }
    }
    if {${:mode} eq "reader"} {return [expr {$old eq {} || $reset || ![dict get $old active] ? -1 : $fid}]}
    if {$old eq {} || $reset} {
        set plen [expr {min(64,[dict get $file size])}]; seek $f 0
        set prefix [ns_crypto::md string -digest sha256 -binary -- [read $f $plen]]
        set high 0; set stamp [dict get $file mtime]; set now [clock seconds]
        ${:db} transaction {
            if {[dict exists ${:settings} unified] && [dict get ${:settings} unified] && $fid ne {}} {
                set retired "${key}:retired:$fid"; set oldIdentity "${identity}:retired:$fid"
                ${:db} execute {UPDATE nsmcp_records.files SET k=$retired,identity=$oldIdentity WHERE fid=$fid}
            } else {
                ${:db} execute {DELETE FROM nsmcp_records.rows WHERE fid=$fid; DELETE FROM nsmcp_records.ranges WHERE fid=$fid; DELETE FROM nsmcp_records.files WHERE k=$key}
            }
            ${:db} execute {
                INSERT INTO nsmcp_records.files(k,identity,prefix,plen,high,stamp,touched) VALUES($key,$identity,$prefix,$plen,$high,$stamp,$now)}
        }
    }
} finally {close $f}
set fid [${:db} scalar {SELECT fid FROM nsmcp_records.files WHERE k=$key}]
set now [clock seconds]
set touchBefore [expr {$now-max(1,min(3600,int([dict get ${:settings} retention]/4)))}]
${:db} execute {UPDATE nsmcp_records.files SET touched=$now WHERE fid=$fid AND touched<$touchBefore}
${:db} execute {UPDATE nsmcp_records.files SET active=0 WHERE identity=$identity AND fid<>$fid AND active=1;
    UPDATE nsmcp_records.files SET active=1 WHERE fid=$fid}
return $fid

prune (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> prune

Testcases:
No testcase defined.

if {[dict exists ${:settings} unified] && [dict get ${:settings} unified]} {return [:pruneAge]}

# Expire whole source generations, not individual rows/ranges: keeping
# a range after deleting its rows would falsely certify cached coverage.
set cutoff [expr {[clock seconds]-[dict get ${:settings} retention]}]
${:db} transaction {
    ${:db} execute {DELETE FROM nsmcp_records.rows WHERE fid IN (SELECT fid FROM nsmcp_records.files WHERE touched<$cutoff);
        DELETE FROM nsmcp_records.ranges WHERE fid IN (SELECT fid FROM nsmcp_records.files WHERE touched<$cutoff);
        DELETE FROM nsmcp_records.files WHERE touched<$cutoff}
}

rollupBackfill (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> rollupBackfill \
    [ batches ]

Parameters:

batches (optional, defaults to "1")

Testcases:
No testcase defined.

lassign [:stateGet rollupCursor {0 -1}] fid off
for {set i 0} {$i<$batches} {incr i} {
    set rows [${:db} execute {SELECT fid,off FROM nsmcp_records.rows WHERE (fid,off)>($fid,$off) AND rolled=0 ORDER BY fid,off LIMIT 512}]
    if {$rows eq {}} {:statePut rollupsReady 1; return true}
    ${:db} transaction {
        ${:db} execute {DELETE FROM temp.nsmcp_ingest_batch;
            INSERT INTO temp.nsmcp_ingest_batch SELECT * FROM nsmcp_records.rows WHERE (fid,off)>($fid,$off) AND rolled=0 ORDER BY fid,off LIMIT 512}
        :rollBatch
        :observeBatch
        ${:db} execute {UPDATE nsmcp_records.rows SET rolled=1 WHERE (fid,off) IN(SELECT fid,off FROM temp.nsmcp_ingest_batch)}
        set last [lindex $rows end]; set fid [dict get $last fid]; set off [dict get $last off]
        :statePut rollupCursor [list $fid $off]
    }
}
return false

samples (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> samples selected \
    files redactor

Parameters:

selected (required)

files (required)

redactor (required)

Testcases:
No testcase defined.

set samples {}; set successes {}
foreach {field extra} {samples {} successSamples { AND status<400}} {
    set entries {}
    foreach row [${:db} execute "SELECT fid,off,t,status,path,method,pool,source,driver,auth,phase,label,clipped,hex(agent) AS agent,identifier,lower(hex(score)) AS score FROM temp.nsmcp_record_window WHERE $selected $extra ORDER BY score,fid,off LIMIT 5"] {
        set file {}; foreach entry $files {if {[dict exists $entry archiveFid] && [dict get $entry archiveFid] eq [dict get $row fid]} {set file $entry; break}}
        if {$file eq {}} continue
        set item [dict create path [dict get $row path] method [dict get $row method] status [dict get $row status] pool [dict get $row pool] authenticationState [dict get $row auth] startTime [dict get $row t] recordSource [dict get $row source] driver [dict get $row driver] requestIdentifier [dict get $row identifier] requestIdentifierUnique [expr {[dict get $row source] eq "connection"}] phaseTimingAvailable [dict get $row phase] phaseTimingReason [expr {[dict get $row phase] ? "logged-worker-phases" : ([dict get $row source] eq "driver" ? "driver-synthetic-phases" : "no-phase-counters")}]]
        if {[dict get $row agent] ne {}} {dict set item userAgent [dict get $row label]}
        set sample [accessScanSample $item $file [dict get $row off] $redactor]
        if {[dict get $row clipped]} {
            set index [lsearch -exact $sample userAgentTruncated]
            if {$index>=0} {lset sample [expr {$index+2}] true}
        }
        lappend entries [list [dict get $row score] $sample]
    }
    if {$field eq "samples"} {set samples $entries} else {set successes $entries}
}
return [list $samples $successes]

scope (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> scope files

Parameters:

files (required)

Testcases:
No testcase defined.

if {${:mode} eq "reader"} {${:db} execute {PRAGMA query_only=OFF}}
${:db} execute {DROP VIEW IF EXISTS temp.nsmcp_record_window; DROP TABLE IF EXISTS temp.nsmcp_record_scope;
    CREATE TEMP TABLE nsmcp_record_scope(fid INTEGER PRIMARY KEY,lo INTEGER,hi INTEGER)}
foreach file $files {
    if {![dict exists $file archiveFid] || [dict get $file status] eq "outside-window"} continue
    set fid [dict get $file archiveFid]; set lo [dict get $file archiveLo]; set hi [dict get $file offset]
    ${:db} execute {INSERT INTO temp.nsmcp_record_scope VALUES($fid,$lo,$hi)}
}
${:db} execute BEGIN; set :querying true
foreach file $files {
    if {[dict exists $file archiveFid]} {
        set fid [dict get $file archiveFid]
        if {$fid>=0 && ![${:db} scalar {SELECT count(*) FROM nsmcp_records.files WHERE fid=$fid}]} {error "record-store-generation-changed"}
    }
}
${:db} execute {CREATE TEMP VIEW nsmcp_record_window AS SELECT r.* FROM nsmcp_records.rows r
    JOIN temp.nsmcp_record_scope s ON r.fid=s.fid AND r.off>=s.lo AND r.off<s.hi AND r.stop<=s.hi}
if {${:mode} eq "reader"} {${:db} execute {PRAGMA query_only=ON}}

session (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> session

Testcases:
No testcase defined.

return ${:db}

stateGet (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> stateGet key \
    [ default ]

Parameters:

key (required)

default (optional)

Testcases:
No testcase defined.

set rows [${:db} execute {SELECT value FROM nsmcp_records.state WHERE key=$key}]
return [expr {$rows eq {} ? $default : [dict get [lindex $rows 0] value]}]

statePut (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> statePut key value

Parameters:

key (required)

value (required)

Testcases:
No testcase defined.

${:db} execute {INSERT OR REPLACE INTO nsmcp_records.state VALUES($key,$value)}

stats (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> stats options \
    redactor limits

Parameters:

options (required)

redactor (required)

limits (required)

Testcases:
No testcase defined.

lassign [:filters $options] context selected
set from [dict get $options from]; set to [dict get $options to]
set stats [accessScanEmpty]
set row [lindex [${:db} execute "SELECT count(*) AS contextRecords,coalesce(sum(status>=400),0) AS contextFailedRecords FROM temp.nsmcp_record_window WHERE $context"] 0]
set stats [dict merge $stats $row]
set row [lindex [${:db} execute "SELECT count(*) AS requestCount,coalesce(sum(status<400),0) AS statusBelow400Records,coalesce(sum(bytes),0) AS responseBytes,
    coalesce(sum(phase),0) AS timingCount,coalesce(sum(fallback),0) AS fallbackCount,coalesce(sum(a),0) AS acceptTotal,coalesce(sum(q),0) AS queueTotal,
    coalesce(sum(f),0) AS filterTotal,coalesce(sum(r),0) AS runTotal,coalesce(max(r),0) AS runMax,coalesce(min(t),'') AS matchedFirstAt,coalesce(max(t),'') AS matchedLastAt,
    coalesce(sum(peer=''),0) AS unknownPeerRecords,coalesce(sum(agent=''),0) AS unknownUserAgentRecords,coalesce(sum(clipped),0) AS truncatedUserAgentRecords,
    coalesce(sum(source='driver' AND status=503),0) AS driver503Records FROM temp.nsmcp_record_window WHERE $selected"] 0]
set stats [dict merge $stats $row]
set row [lindex [${:db} execute {SELECT count(*) AS examinedRecords,coalesce(sum(kind=1),0) AS parseErrors,coalesce(sum(kind=2),0) AS oversizedRecords,coalesce(min(t),'') AS earliestAt,coalesce(max(t),'') AS latestAt FROM temp.nsmcp_record_window}] 0]
set stats [dict merge $stats $row]
dict set stats pendingRecords [:boundaryPending]
foreach {field column cap} [list paths path [dict get $limits keys] methods method 32 statuses status 500 pools pool 64 hours {CAST(t/3600 AS INTEGER)*3600} 170 drivers driver 64] {
    set where $selected
    if {$field eq "drivers"} {append where " AND source='driver'"}
    set known 0
    foreach row [${:db} execute "SELECT $column AS value,count(*) AS n FROM temp.nsmcp_record_window WHERE $where GROUP BY $column ORDER BY n DESC,value LIMIT $cap"] {
        set value [dict get $row value]; set n [dict get $row n]
        if {$field ni {hours statuses}} {set value [$redactor message $value]}
        if {$field eq "paths" && [string length [encoding convertto utf-8 $value]]>1024} continue
        set counts [dict get $stats $field]; dict incr counts $value $n; dict set stats $field $counts; incr known $n
    }
    set total [${:db} scalar "SELECT count(*) FROM temp.nsmcp_record_window WHERE $where"]
    if {$known<$total} {dict set stats omittedRecords $field [expr {$total-$known}]}
}
foreach {field column} {authenticationStates auth recordSources source} {
    foreach row [${:db} execute "SELECT $column AS value,count(*) AS n FROM temp.nsmcp_record_window WHERE $selected GROUP BY $column"] {dict set stats $field [dict get $row value] [dict get $row n]}
}
foreach row [${:db} execute "SELECT status,count(*) AS requestCount,sum(phase) AS timingCount,sum(r) AS runTotal,max(r) AS runMax FROM temp.nsmcp_record_window WHERE $selected GROUP BY status"] {
    set status [dict get $row status]; dict unset row status; dict set stats statusTimings $status $row
}
set seconds [dict get $options bucketSeconds]; set index "CAST((t-[${:db} quote $from])/$seconds AS INTEGER)"
foreach row [${:db} execute "SELECT $index AS i,count(*) AS n,sum(status>=400) AS failed,sum(phase) AS timings,sum(r) AS runtime FROM temp.nsmcp_record_window WHERE $context GROUP BY i LIMIT 3600"] {
    dict set stats buckets [dict get $row i] [list contextRecords [dict get $row n] contextFailedRecords [dict get $row failed] selectedRecords 0 driver503Records 0 timingCount 0 runTotal 0.0 contextTimingCount [dict get $row timings] contextRunTotal [dict get $row runtime]]
}
foreach row [${:db} execute "SELECT $index AS i,count(*) AS n,sum(source='driver' AND status=503) AS driver503,sum(phase) AS timings,sum(r) AS runtime FROM temp.nsmcp_record_window WHERE $selected GROUP BY i LIMIT 3600"] {
    set i [dict get $row i]
    foreach {key col} {selectedRecords n driver503Records driver503 timingCount timings runTotal runtime} {dict set stats buckets $i $key [dict get $row $col]}
}
set aggregate [dict create backend sqlite cacheKiB 8192 status active reason {} maxBytes [dict get ${:settings} maxBytes] diskBytes [:diskBytes] peakBufferedKeys 512 peerOmitted 0 agentOmitted 0 labelTruncatedRecords 0 labelRedactedRecords 0]
foreach {field col} {peers peer agents agent} {
    set valid "$selected AND $col<>''"
    set totals [dict create distinct [${:db} scalar "SELECT count(DISTINCT $col) FROM temp.nsmcp_record_window WHERE $valid"] records [${:db} scalar "SELECT count(*) FROM temp.nsmcp_record_window WHERE $valid"] top1 0 top10 0 top100 0 rankedRecords 0 clippedRecords 0 redactedRecords 0]
    set entries {}; set i 0
    foreach row [${:db} execute "SELECT hex($col) AS k,count(*) AS n,min(label) AS label,max(clipped) AS clipped,max(redacted) AS redacted FROM temp.nsmcp_record_window WHERE $valid GROUP BY $col ORDER BY n DESC,k LIMIT 100"] {
        incr i; set n [dict get $row n]
        if {$i==1} {dict set totals top1 $n}; if {$i<=10} {dict incr totals top10 $n}; dict incr totals top100 $n
        if {$i<=[dict get $options maxGroups]} {
            dict incr totals rankedRecords $n
            lappend entries [dict create count $n value [dict get $row label] valueTruncated [dict get $row clipped] valueRedacted [dict get $row redacted]]
        }
    }
    dict set aggregate $field $entries; dict set aggregate ${field}Totals $totals
}
set last [${:db} scalar "SELECT coalesce(sum(clipped),0) FROM temp.nsmcp_record_window WHERE $selected"]
dict set aggregate labelTruncatedRecords $last
dict set aggregate labelRedactedRecords [${:db} scalar "SELECT coalesce(sum(redacted),0) FROM temp.nsmcp_record_window WHERE $selected"]
return [list $stats $aggregate $selected]

totals (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> totals options

Parameters:

options (required)

Testcases:
No testcase defined.

lassign [:filters $options] context selected
set from [dict get $options from]; set to [dict get $options to]
if {[dict exists ${:settings} unified] && [dict get ${:settings} unified] && [:stateGet rollupsReady 0] && $from%60==0 && $to%60==0 && [dict get $options pathLiteral] eq {}} {
    set counts [lindex [${:db} execute "SELECT coalesce(sum(n),0) AS requestCount,coalesce(sum(CASE WHEN status>=400 THEN n ELSE 0 END),0) AS failedRecords,coalesce(sum(CASE WHEN $selected THEN n ELSE 0 END),0) AS selectedRecords FROM (SELECT 0 AS kind,slot AS t,status,pool,method,source,n FROM nsmcp_records.minutes m JOIN temp.nsmcp_record_scope s ON m.fid=s.fid) WHERE $context"] 0]
    set statuses [${:db} execute "SELECT status,sum(n) AS count FROM (SELECT 0 AS kind,slot AS t,status,pool,method,source,n FROM nsmcp_records.minutes m JOIN temp.nsmcp_record_scope s ON m.fid=s.fid) WHERE $selected GROUP BY status ORDER BY status"]
    set overview [:overview]
    return [list $counts $statuses [dict create storedPhysicalRecords [dict get $counts requestCount] parseErrors [dict get $overview malformed] oversizedRecords [dict get $overview oversized] pendingRecords 0]]
}
set counts [lindex [${:db} execute "SELECT count(*) AS requestCount,coalesce(sum(status>=400),0) AS failedRecords,coalesce(sum(CASE WHEN $selected THEN 1 ELSE 0 END),0) AS selectedRecords FROM temp.nsmcp_record_window WHERE $context"] 0]
set statuses [${:db} execute "SELECT status,count(*) AS count FROM temp.nsmcp_record_window WHERE $selected GROUP BY status ORDER BY status"]
set diagnostics [lindex [${:db} execute {SELECT count(*) AS storedPhysicalRecords,coalesce(sum(kind=1),0) AS parseErrors,coalesce(sum(kind=2),0) AS oversizedRecords FROM temp.nsmcp_record_window}] 0]
dict set diagnostics pendingRecords [:boundaryPending]
return [list $counts $statuses $diagnostics]

tuple (scripted, public)

 <instance of nsmcp::AccessRecordStoreSQLite> tuple fid off stop \
    kind row redactor file

Parameters:

fid (required)

off (required)

stop (required)

kind (required)

row (required)

redactor (required)

file (required)

Testcases:
No testcase defined.

set values [list $fid $off $stop $kind]
if {$kind!=0} {
    for {set i 0} {$i<22} {incr i} {lappend values {}}
} else {
    set peer [dict get $row peerAddress]; set agent [expr {[dict exists $row userAgent] ? [dict get $row userAgent] : {}}]
    set peer [expr {$peer in {{} - {[REDACTED]}} ? {} : [ns_crypto::md string -digest sha256 -- $peer]}]
    set key [expr {$agent in {{} -} ? {} : [ns_crypto::md string -digest sha256 -- $agent]}]
    set label [$redactor message $agent]; set clipped [expr {[string length [encoding convertto utf-8 $label]]>1024 || [string length [encoding convertto utf-8 $agent]]>1024}]
    set redacted [expr {$label ne $agent}]
    if {$clipped} {
        set label [encoding convertfrom utf-8 [string range [encoding convertto utf-8 $label] 0 1023]]
        while {[string length [encoding convertto utf-8 $label]]>1024} {set label [string range $label 0 end-1]}
    }
    foreach k {startTime status path method pool recordSource driver authenticationState responseBytes phaseTimingAvailable} {lappend values [dict get $row $k]}
    lappend values [expr {[dict get $row timingSource] eq "completion"}]
    foreach k {acceptTime queueTime filterTime runTime} {lappend values [expr {[dict exists $row $k] ? [dict get $row $k] : 0.0}]}
    lappend values $peer $key $label $clipped $redacted [dict get $row requestIdentifier] [ns_crypto::md string -digest sha256 -- [list [dict get $file identity] $off]]
}
set completed 0
if {$kind==0} {
    set stamp [dict get $row timestamp]
    if {$stamp ne ${:lastStamp}} {
        set :lastStamp $stamp
        if {[string is entier -strict $stamp]} {set :lastCompleted $stamp} elseif {[catch {clock scan $stamp -format {%d/%b/%Y:%H:%M:%S %z} -locale C} value]} {set :lastCompleted 0} else {set :lastCompleted $value}
    }
    set completed ${:lastCompleted}
    if {[dict exists ${:settings} unified] && [dict get ${:settings} unified]} {
        set address [dict get $row peerAddress]; set time [dict get $row startTime]
        if {[ns_ip valid $address] && $time>=[clock seconds]-[dict get ${:settings} peerRetention] && $time<=[clock seconds]+60} {
            dict set :peers [lindex $values 19] [list $address $time]
        }
    }
}
lappend values $completed 1
set quoted {}; foreach value $values {lappend quoted [${:db} quote $value]}
if {$kind==0} {
    foreach index {19 20 25} {
        set value [lindex $values $index]
        if {$value ne {}} {lset quoted $index "X'$value'"}
    }
}
if {$kind!=0} {
    set at [expr {[dict exists ${:settings} unified] && [dict get ${:settings} unified] ? [clock seconds] : 0}]
    set quoted [concat [lrange $quoted 0 3] [lrepeat 22 NULL] $at 1]
}
return "([join $quoted ,])"

Methods: All Methods Documented Methods Hide Methods
Source: Display Source Hide Source
Variables: Show Variables Hide Variables