Class ::nsmcp::AccessRecordStoreSQLite (public)
::nx::Class ::nsmcp::AccessRecordStoreSQLite![[i]](/resources/acs-subsite/ZoomIn16.gif)
Defined in /usr/local/ns/tcl/nsmcp/log-traffic-record-store.tcl
- Testcases:
- No testcase defined.
Source code: :property settings:required :property {mode writer} :property {observer {}} :variable db {} :variable handle {} :variable attached false :variable path :variable prior {} :variable querying false :variable fullScope false :variable peers {} :variable lastStamp {} :variable lastCompleted 0 :method init {} { try { set directory [dict get ${:settings} directory] if {${:mode} ni {reader writer}} {error "invalid access store mode"} if {${:mode} eq "writer"} {file mkdir $directory; file attributes $directory -permissions 0700} set key [ns_crypto::md string -digest sha256 -- [ns_info server]] set :path [file join $directory access-records-$key.sqlite3] # Create privately before SQLite opens the file. Never follow a symlink. if {${:mode} eq "reader" && ![file exists ${:path}]} {error "record-store-not-initialized"} if {![file exists ${:path}]} { if {[catch {open ${:path} {WRONLY CREAT EXCL} 0600} f createOptions]} { if {[lindex [dict get $createOptions -errorcode] 1] ne "EEXIST"} {return -options $createOptions $f} } else {close $f} } file lstat ${:path} st if {$st(type) ne "file"} {error "record-store-invalid-file"} if {${:mode} eq "writer"} {file attributes ${:path} -permissions 0600} accessIngestionObserve ${:observer} pool-checkout set :handle [ns_db gethandle -timeout 0:100000 [dict get ${:settings} pool]] if {${:handle} eq {}} {return} if {[ns_db dbtype ${:handle}] ne "sqlite"} {error "record-store-wrong-driver"} set :db [::nsmcp::DatabaseSession new -handle ${:handle}] set :prior [dict create busy [${:db} scalar {PRAGMA busy_timeout}] temp [${:db} scalar {PRAGMA temp_store}] queryOnly [${:db} scalar {PRAGMA query_only}]] foreach row [${:db} execute {PRAGMA database_list}] { if {[dict get $row name] eq "nsmcp_records"} {${:db} execute {DETACH DATABASE nsmcp_records}} } # Install the short native wait before ATTACH or schema inspection. ${:db} execute {PRAGMA busy_timeout=100} set path ${:path} accessIngestionObserve ${:observer} attach-store ${:db} execute {ATTACH DATABASE $path AS nsmcp_records} set :attached true ${:db} execute {PRAGMA nsmcp_records.cache_size=-8192; PRAGMA nsmcp_records.mmap_size=0; PRAGMA temp_store=FILE} if {${:mode} eq "writer"} { file stat ${:path} identity set key [ns_crypto::md string -digest sha256 -- [list ${:path} $identity(dev) $identity(ino) [dict get ${:settings} maxBytes] [expr {[dict exists ${:settings} unified] && [dict get ${:settings} unified]}] schema3]] if {[file size ${:path}]<512 || ![nsv_exists nsmcp:record-initialized $key]} { accessIngestionObserve ${:observer} initialize-schema :initialize nsv_set nsmcp:record-initialized $key 1 } # ATTACH creates a new connection-local pager. The page ceiling # must be applied here even when schema initialization was cached. accessIngestionObserve ${:observer} page-limit :pageLimit ${:db} execute {PRAGMA nsmcp_records.synchronous=FULL; PRAGMA nsmcp_records.journal_size_limit=0} } elseif {[${:db} scalar {SELECT version FROM nsmcp_records.meta}]!=2} { error "record-store-schema-not-initialized" } if {${:mode} eq "writer"} {${:db} execute {DROP TABLE IF EXISTS temp.nsmcp_retention_batch; DROP TABLE IF EXISTS temp.nsmcp_ingest_batch; CREATE TEMP TABLE nsmcp_ingest_batch AS SELECT * FROM nsmcp_records.rows WHERE 0}} else {${:db} execute {PRAGMA query_only=ON}} } on error {message opts} { if {$message eq "record-store-disk-limit"} { # The unified writer emits a rate-limited capacity warning. set failure {class disk-limit diagnosticId {}} } elseif {$message in {record-store-not-initialized record-store-schema-not-initialized}} { set failure [dict create class [string range $message 13 end] diagnosticId {}] } elseif {[regexp -nocase {database.*(locked|busy)} $message]} { set failure [dict create class database-locked diagnosticId {}] } else {set failure [:diagnostic $message $opts]} dict set opts -errorcode [list NSMCP RECORD_STORE [dict get $failure class] [dict get $failure diagnosticId]] :disconnect return -options $opts $message } } :method initialize {} { set journal [expr {[dict exists ${:settings} unified] && [dict get ${:settings} unified] ? "WAL" : "DELETE"}] ${:db} execute "PRAGMA nsmcp_records.journal_mode=$journal" ${:db} execute {PRAGMA busy_timeout=100; PRAGMA nsmcp_records.synchronous=FULL; PRAGMA nsmcp_records.cache_size=-8192; PRAGMA nsmcp_records.mmap_size=0; PRAGMA temp_store=FILE; PRAGMA nsmcp_records.journal_size_limit=0} # Reserve space for the rollback journal. The cap applies to all live # database/journal files, not SQL temporary sort files. No SQLite settings on main are changed. :pageLimit ${:db} execute {CREATE TABLE IF NOT EXISTS nsmcp_records.meta(version INTEGER NOT NULL); INSERT INTO nsmcp_records.meta SELECT 1 WHERE NOT EXISTS(SELECT 1 FROM nsmcp_records.meta); CREATE TABLE IF NOT EXISTS nsmcp_records.files(fid INTEGER PRIMARY KEY AUTOINCREMENT,k TEXT UNIQUE NOT NULL,identity TEXT NOT NULL,prefix TEXT NOT NULL,plen INTEGER NOT NULL,high INTEGER NOT NULL,stamp INTEGER NOT NULL,touched INTEGER NOT NULL); CREATE TABLE IF NOT EXISTS nsmcp_records.ranges(fid INTEGER NOT NULL,lo INTEGER NOT NULL,hi INTEGER NOT NULL,anchor TEXT NOT NULL,PRIMARY KEY(fid,lo)); CREATE TABLE IF NOT EXISTS nsmcp_records.rows(fid INTEGER NOT NULL,off INTEGER NOT NULL,stop INTEGER NOT NULL,kind INTEGER NOT NULL,t REAL,status INTEGER,path TEXT,method TEXT,pool TEXT,source TEXT,driver TEXT,auth TEXT,bytes INTEGER,phase INTEGER,fallback INTEGER,a REAL,q REAL,f REAL,r REAL,peer TEXT,agent TEXT,label TEXT,clipped INTEGER,redacted INTEGER,identifier TEXT,score TEXT,PRIMARY KEY(fid,off)); CREATE INDEX IF NOT EXISTS nsmcp_records.by_time ON rows(t); CREATE INDEX IF NOT EXISTS nsmcp_records.by_time_status ON rows(t,status,fid,off,kind)} set version [${:db} scalar {SELECT version FROM nsmcp_records.meta}] if {$version ni {1 2}} {error "record-store-schema-version"} set columns {}; foreach row [${:db} execute {SELECT name FROM pragma_table_info('rows','nsmcp_records')}] {lappend columns [dict get $row name]} if {"completed" ni $columns} {${:db} execute {ALTER TABLE nsmcp_records.rows ADD COLUMN completed REAL NOT NULL DEFAULT 0}} if {"rolled" ni $columns} {${:db} execute {ALTER TABLE nsmcp_records.rows ADD COLUMN rolled INTEGER NOT NULL DEFAULT 0}} set columns {}; foreach row [${:db} execute {SELECT name FROM pragma_table_info('files','nsmcp_records')}] {lappend columns [dict get $row name]} if {"active" ni $columns} {${:db} execute {ALTER TABLE nsmcp_records.files ADD COLUMN active INTEGER NOT NULL DEFAULT 1}} foreach column {dropping} { if {$column ni $columns} {${:db} execute "ALTER TABLE nsmcp_records.files ADD COLUMN $column INTEGER NOT NULL DEFAULT 0"} } ${:db} execute {CREATE TABLE IF NOT EXISTS nsmcp_records.state(key TEXT PRIMARY KEY,value TEXT NOT NULL); CREATE TABLE IF NOT EXISTS nsmcp_records.peer_addresses(k BLOB PRIMARY KEY,address TEXT NOT NULL,seen REAL NOT NULL); CREATE TABLE IF NOT EXISTS nsmcp_records.minutes(fid INTEGER NOT NULL,slot INTEGER NOT NULL,status INTEGER NOT NULL,pool TEXT NOT NULL,method TEXT NOT NULL,source TEXT NOT NULL,n INTEGER NOT NULL,bytes INTEGER NOT NULL,PRIMARY KEY(fid,slot,status,pool,method,source)); CREATE INDEX IF NOT EXISTS nsmcp_records.minute_time ON minutes(slot,status); CREATE INDEX IF NOT EXISTS nsmcp_records.peer_expiry ON peer_addresses(seen); CREATE INDEX IF NOT EXISTS nsmcp_records.diagnostic_expiry ON rows(completed) WHERE kind<>0; CREATE INDEX IF NOT EXISTS nsmcp_records.correlation ON rows(identifier,t); CREATE TABLE IF NOT EXISTS nsmcp_records.evictions(id INTEGER PRIMARY KEY,dropped_at INTEGER,first_time REAL,last_time REAL,rows INTEGER,reason TEXT); UPDATE nsmcp_records.meta SET version=2} } :method pageLimit {} { set pages [accessRecordMainPages ${:settings} [${:db} scalar {PRAGMA nsmcp_records.page_size}]] set capacity [:capacity] if {[dict get $capacity limitExceeded]} {error "record-store-disk-limit"} ${:db} execute "PRAGMA nsmcp_records.max_page_count=$pages" } :method retentionFloor {} { set floor [:stateGet retentionFloor 0] if {[dict exists ${:settings} unified] && [dict get ${:settings} unified]} { # A paused writer cannot prune or advance persisted metadata. # Readers still enforce current event-age retention immediately. set current [expr {wide(floor(([clock seconds]-[dict get ${:settings} retention])/60.0))*60}] set floor [expr {max($floor,$current)}] } return $floor } :public method capacity {} { set pageSize [${:db} scalar {PRAGMA nsmcp_records.page_size}] set allocated [expr {wide([${:db} scalar {PRAGMA nsmcp_records.page_count}])*$pageSize}] set mainLimit [expr {[accessRecordMainPages ${:settings} $pageSize]*$pageSize}] set disk [:diskBytes]; set budget [dict get ${:settings} maxBytes] set result [dict create maxBytes $budget mainMaxBytes $mainLimit mainAllocatedBytes $allocated diskBytes $disk limitExceeded [expr {$allocated>$mainLimit || $disk>$budget}] minimumBudgetBytes [expr {($allocated*5+1)/2}]] if {${:mode} eq "writer"} { nsv_set nsmcp:access-store:[ns_info server] capacitySnapshot [dict merge $result [dict create observedAt [clock seconds]]] } return $result } :public method diagnostic {message opts} { return [accessAggregationDiagnostic record-store $message $opts ${:handle}] } :public method available {} {return [expr {${:db} ne {}}]} :public method diskBytes {} { set total 0 foreach suffix {{} -journal -wal -shm} {if {[file exists ${:path}$suffix]} {incr total [file size ${:path}$suffix]}} return $total } :public method prepare {file} { 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 } :public method coveredEnd {fid offset end} { 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)}] } :public method dropState {fid offset} {return [${:db} scalar {SELECT CASE WHEN dropping=1 AND high<=$offset THEN 1 ELSE 0 END FROM nsmcp_records.files WHERE fid=$fid}]} :public method gapEnd {fid offset end} { 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)}] } :public method ingest {fid file begin end tuples} { accessIngestionObserve ${:observer} read-anchor 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}] set committed [dict create records 0 requestRecords 0 malformedRecords 0 oversizedRecords 0] accessIngestionObserve ${:observer} begin-transaction ${:db} transaction { if {![${:db} scalar {SELECT count(*) FROM nsmcp_records.files WHERE fid=$fid}]} {error "record-store-generation-changed"} if {[llength $tuples]} { accessIngestionObserve ${:observer} stage-cooked-records ${:db} execute {DELETE FROM temp.nsmcp_ingest_batch} ${:db} executeDml "INSERT INTO temp.nsmcp_ingest_batch VALUES [join $tuples ,]" accessIngestionObserve ${:observer} deduplicate-records ${: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)} accessIngestionObserve ${:observer} insert-records ${:db} execute {INSERT INTO nsmcp_records.rows SELECT * FROM temp.nsmcp_ingest_batch} accessIngestionObserve ${:observer} count-records set committed [lindex [${:db} execute {SELECT count(*) AS records, coalesce(sum(kind=0),0) AS requestRecords,coalesce(sum(kind=1),0) AS malformedRecords, coalesce(sum(kind=2),0) AS oversizedRecords FROM temp.nsmcp_ingest_batch}] 0] accessIngestionObserve ${:observer} rollup :rollBatch accessIngestionObserve ${:observer} observe-records :observeBatch } accessIngestionObserve ${:observer} store-peers 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} executeDml "INSERT INTO nsmcp_records.peer_addresses VALUES [join $values ,] ON CONFLICT(k) DO UPDATE SET address=excluded.address,seen=excluded.seen WHERE excluded.seen>=seen AND (excluded.seen<>seen OR excluded.address<>address)" } accessIngestionObserve ${:observer} save-ranges 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} accessIngestionObserve ${:observer} check-storage if {[:diskBytes]>[dict get ${:settings} maxBytes]} {error "record-store-disk-limit"} accessIngestionObserve ${:observer} commit } set :peers {} return $committed } # Native batches enter a private, connection-local staging table. The # canonical store still receives only policy-processed rows, committed with # their covered range by ingest(). No raw headers enter durable storage. :public method nativeRead {fid file offset stop budget limits redactor ingestOnly} { accessIngestionObserve ${:observer} batch-capabilities if {![llength [info commands ns_accesslog]]} {error "access batch ingestion requires nslog's scan API"} if {![dict get [${:db} batchCapabilities] supported]} { error "access batch ingestion requires a batch-capable SQLite driver" } set mapping {offset off endOffset stop timestamp stamp peerAddress peer threadName identifier user usr requestLine request status status responseBytes bytes userAgent agent startTime start acceptTime a queueTime q filterTime f runTime r parseStatus parsed diagnostic diagnostic} accessIngestionObserve ${:observer} stage-schema ${:db} execute {DROP TABLE IF EXISTS temp.nsmcp_scan_batch; CREATE TEMP TABLE nsmcp_scan_batch(off INTEGER,stop INTEGER,stamp TEXT,peer TEXT,identifier TEXT,usr TEXT,request TEXT,status INTEGER,bytes INTEGER,agent TEXT,start INTEGER,a INTEGER,q INTEGER,f INTEGER,r INTEGER,parsed TEXT,diagnostic TEXT)} set timing [::nsmcp::accessRecordParser cget -timingColumn] set timing [dict get {-2 none -1 auto 0 0 2 2} $timing] set scanOptions [list -filename [dict get $file path] -offset $offset -endoffset $stop -maxlinebytes [dict get $limits line]] if {$timing ne "auto"} {lappend scanOptions -timingcolumn $timing} accessIngestionObserve ${:observer} open-scanner set scanner [ns_accesslog scan open {*}$scanOptions] set begin $offset; set tuples {}; set pending false try { # The scanner enforces this invocation's planned end even if the # active log grew after discovery or this is a covered-range gap. # Keep framing state within this call, but serialize only completed # record offsets between worker slices. Never commit a snapshot tail. accessIngestionObserve ${:observer} scan-native set batch [ns_accesslog scan next $scanner -maxrecords 512 -maxscanbytes [expr {min(1073741824,max(1,$budget))}] -maxbytes [expr {max(8388608,4*[dict get $limits line]+2048)}] -timeout 100ms] set info [ns_db batchinfo $batch] accessIngestionObserve ${:observer} batchinsert ${:db} transaction {${:db} batchInsert {temp nsmcp_scan_batch} $mapping $batch} accessIngestionObserve ${:observer} cook-batch set escapes [list \\ \\\\ \" \\\" \n {\n} \r {\r} \t {\t}] foreach item [${:db} execute {SELECT *,agent IS NOT NULL AS combined FROM temp.nsmcp_scan_batch ORDER BY off}] { set next [dict get $item stop] if {$next>$stop} {set pending true; break} set kind 0; set row {} if {[dict get $item parsed] ne "ok"} { set kind [expr {[dict get $item diagnostic] eq "oversized" ? 2 : 1}] } else { # SQLite converts lexical seconds to integer microseconds. # Reuse the existing policy for start-time plausibility, # authentication, query stripping and unavailable phases. set fields {} if {[dict get $item combined]} { lappend fields {""} "\"[string map $escapes [dict get $item agent]]\"" } if {[dict get $item start] ne {}} { set phases {} foreach key {start a q f r} { set us [dict get $item $key] if {$us eq {}} {set us 0} lappend phases [format {%d.%06d} [expr {$us/1000000}] [expr {$us%1000000}]] } lappend fields "\"[join $phases { }]\"" } set bytes [dict get $item bytes] if {$bytes eq {}} {set bytes -} set row [::nsmcp::accessRecordParser fields [dict get $item peer] [dict get $item identifier] [string map $escapes [dict get $item usr]] [dict get $item stamp] [string map $escapes [dict get $item request]] [dict get $item status] $bytes {} $fields] if {$row ne {} && [dict exists $row timeDependent] && !$ingestOnly} { return [list $begin {} false true] } if {$row eq {} || [dict exists $row timeDependent]} {set row {}; set kind 1} } lappend tuples [:tuple $fid $offset $next $kind $row $redactor $file] set offset $next } if {[dict get $info state] eq "pending"} {set pending true} } finally { ns_accesslog scan close $scanner ${:db} execute {DROP TABLE IF EXISTS temp.nsmcp_scan_batch} } return [list $offset $tuples $pending] } :public method tuple {fid off stop kind row redactor file} { 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 ,])" } :public method scope {files} { set :fullScope false 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_scan_batch; 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}} } :method filters {options} { set from [dict get $options from]; set to [dict get $options to] if {[dict exists ${:settings} unified] && [dict get ${:settings} unified]} { # Minute-rollup queries bypass the retained-row view, so their # shared predicates must enforce the moving retention floor too. set from [expr {max($from,[:retentionFloor])}] } set context "kind=0 AND t>=[${:db} quote $from] AND t<[${:db} quote $to]" foreach {key column} {pool pool method method recordSource source} { set value [dict get $options $key] if {$value ne {} && !($key eq "recordSource" && $value eq "all")} {append context " AND $column=[${:db} quote $value]"} } set literal [dict get $options pathLiteral] if {$literal ne {}} {append context " AND instr(path,[${:db} quote $literal])>0"} set selected $context set status [dict get $options status]; set class [dict get $options statusClass] if {$status ne {}} {append selected " AND status=[${:db} quote $status]"} if {$class eq "4xx/5xx"} {append selected " AND status>=400"} elseif {$class ne {}} {set low [expr {[string index $class 0]*100}]; append selected " AND status>=$low AND status<$low+100"} return [list $context $selected] if {${:mode} eq "reader"} {${:db} execute {PRAGMA query_only=ON}} } :method boundaryPending {} { return [${:db} scalar {SELECT count(*) 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 WHERE r.stop>s.hi}] if {${:mode} eq "reader"} {${:db} execute {PRAGMA query_only=ON}} } :public method stats {options redactor limits} { lassign [:filters $options] context selected set from [dict get $options from]; set to [dict get $options to] set stats [accessScanEmpty] set rollup [:rollupSource $options] if {$rollup ne {}} { set row [lindex [${:db} execute "SELECT coalesce(sum(n),0) AS contextRecords,coalesce(sum(CASE WHEN status>=400 THEN n ELSE 0 END),0) AS contextFailedRecords FROM ($rollup) WHERE $context"] 0] } else { 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(redacted),0) AS labelRedactedRecords, coalesce(sum(source='driver' AND status=503),0) AS driver503Records,coalesce(sum(source='driver'),0) AS driverRecords 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 if {$rollup ne {} && $field in {methods statuses pools hours}} { set sql "SELECT $column AS value,sum(n) AS n FROM ($rollup) WHERE $where GROUP BY $column ORDER BY n DESC,value LIMIT $cap" } else {set sql "SELECT $column AS value,count(*) AS n FROM temp.nsmcp_record_window WHERE $where GROUP BY $column ORDER BY n DESC,value LIMIT $cap"} foreach row [${:db} execute $sql] { 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 [dict get $stats [expr {$field eq "drivers" ? "driverRecords" : "requestCount"}]] if {$known<$total} {dict set stats omittedRecords $field [expr {$total-$known}]} } foreach {field column} {authenticationStates auth recordSources source} { if {$rollup ne {} && $field eq "recordSources"} { set sql "SELECT $column AS value,sum(n) AS n FROM ($rollup) WHERE $selected GROUP BY $column" } else {set sql "SELECT $column AS value,count(*) AS n FROM temp.nsmcp_record_window WHERE $selected GROUP BY $column"} foreach row [${:db} execute $sql] {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 0 records 0 top1 0 top10 0 top100 0 rankedRecords 0 clippedRecords 0 redactedRecords 0] set entries {}; set i 0 foreach row [${:db} execute "SELECT *,count(*) OVER() AS distinctCount,sum(n) OVER() AS records FROM (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"] { dict set totals distinct [dict get $row distinctCount]; dict set totals records [dict get $row records] 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 } dict set aggregate labelTruncatedRecords [dict get $stats truncatedUserAgentRecords] dict set aggregate labelRedactedRecords [dict get $stats labelRedactedRecords] return [list $stats $aggregate $selected] } :public method coveredBytes {fid lo hi} { 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}] } :method rollupSource {options} { if {!${:fullScope} || ![dict exists ${:settings} unified] || ![dict get ${:settings} unified] || ![:stateGet rollupsReady 0] || [dict get $options pathLiteral] ne {}} {return {}} set from [dict get $options from]; set to [dict get $options to] # Full minutes use the ingestion rollup. Only the disjoint partial # minutes use the time index; never round the requested window. set from [expr {max($from,[:retentionFloor])}] if {$from<0} {return {}} set lo [expr {wide(ceil($from/60.0))*60}] set hi [expr {wide(floor($to/60.0))*60}] set parts {} if {$lo<$hi} { lappend parts "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 slot>=$lo AND slot<$hi" set edges [list [list $from $lo] [list $hi $to]] } else {set edges [list [list $from $to]]} foreach edge $edges { lassign $edge start end if {$start>=$end} continue lappend parts "SELECT kind,t,status,pool,method,source,1 AS n FROM nsmcp_records.rows r INDEXED BY by_time_status JOIN temp.nsmcp_record_scope s ON r.fid=s.fid AND r.off>=s.lo AND r.stop<=s.hi WHERE kind=0 AND t>=[${:db} quote $start] AND t<[${:db} quote $end]" } if {![llength $parts]} {return {SELECT 0 AS kind,0 AS t,0 AS status,'' AS pool,'' AS method,'' AS source,0 AS n WHERE 0}} return [join $parts { UNION ALL }] } :public method totals {options} { lassign [:filters $options] context selected set source [:rollupSource $options] if {$source ne {}} { set counts [dict create requestCount 0 failedRecords 0 selectedRecords 0]; set statuses {} # One grouped pass supplies both context and selected totals. foreach row [${:db} execute "SELECT status,sum(n) AS n,sum(CASE WHEN $selected THEN n ELSE 0 END) AS selected FROM ($source) WHERE $context GROUP BY status ORDER BY status"] { set status [dict get $row status]; set n [dict get $row n]; set selectedCount [dict get $row selected] dict incr counts requestCount $n if {$status>=400} {dict incr counts failedRecords $n} dict incr counts selectedRecords $selectedCount if {$selectedCount>0} {lappend statuses [dict create status $status count $selectedCount]} } set overview [:overview] return [list $counts $statuses [dict create queryStrategy minute-rollups-with-boundaries 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 queryStrategy detailed-rows dict set diagnostics pendingRecords [:boundaryPending] return [list $counts $statuses $diagnostics] } :public method samples {selected files redactor} { 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] } :public method finish {} {${:db} execute COMMIT; set :querying false} :public method descriptors {} { 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 } :public method legacyViews {} { if {${:mode} eq "reader"} {${:db} execute {PRAGMA query_only=OFF}} set floor [:retentionFloor] 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}} } :public method allScope {} { set :fullScope true 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_scan_batch; 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 [:retentionFloor] ${: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}} } :public method overview {} { set report [:stateGet continuous {}]; set observed [:stateGet observations {}] set floor [:retentionFloor]; 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 storage [:capacity] writer [accessIngestionStatus] retention [accessRetentionStatus]] } :public method observeBatch {} { 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 } :method rollBatch {} { ${:db} execute {INSERT INTO nsmcp_records.minutes SELECT fid,CAST(t/60 AS INTEGER)*60,status,pool,method,source,count(*),sum(bytes) FROM temp.nsmcp_ingest_batch WHERE kind=0 GROUP BY fid,CAST(t/60 AS INTEGER),status,pool,method,source ON CONFLICT(fid,slot,status,pool,method,source) DO UPDATE SET n=n+excluded.n,bytes=bytes+excluded.bytes} } :public method stateGet {key {default {}}} { set rows [${:db} execute {SELECT value FROM nsmcp_records.state WHERE key=$key}] return [expr {$rows eq {} ? $default : [dict get [lindex $rows 0] value]}] } :public method statePut {key value} {${:db} execute {INSERT OR REPLACE INTO nsmcp_records.state VALUES($key,$value)}} :public method session {} {return ${:db}} :public method rollupBackfill {{batches 1}} { 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 } :public method prune {{batchRows 10000}} { if {[dict exists ${:settings} unified] && [dict get ${:settings} unified]} {return [:pruneAge $batchRows]} # 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} } } :method pruneAge {batchRows} { if {![string is integer -strict $batchRows] || $batchRows<1 || $batchRows>10000} {error "retention batch rows must be 1..10000"} set now [clock seconds]; set cutoff [expr {wide(floor(($now-[dict get ${:settings} retention])/60.0))*60}] set cutoff [expr {max($cutoff,[:stateGet retentionFloor 0])}] set peerCutoff [expr {$now-[dict get ${:settings} peerRetention]}] set counts {requests 0 diagnostics 0 minutes 0 peers 0} # Select once for both eviction accounting and deletion. Explicit # indexes keep empty expiry probes from becoming full-table scans. ${:db} execute {DROP TABLE IF EXISTS temp.nsmcp_retention_batch; CREATE TEMP TABLE nsmcp_retention_batch(id INTEGER PRIMARY KEY)} try { ${:db} transaction { :statePut retentionFloor $cutoff ${:db} execute {INSERT INTO temp.nsmcp_retention_batch SELECT rowid FROM nsmcp_records.rows INDEXED BY by_time WHERE kind=0 AND t<$cutoff ORDER BY t LIMIT $batchRows} set deleted [lindex [${:db} execute {SELECT count(*) AS n,coalesce(min(t),0) AS first,coalesce(max(t),0) AS last FROM nsmcp_records.rows WHERE rowid IN(SELECT id FROM temp.nsmcp_retention_batch)}] 0] ${:db} execute {DELETE FROM nsmcp_records.rows WHERE rowid IN(SELECT id FROM temp.nsmcp_retention_batch)} dict set counts requests [dict get $deleted n] ${:db} execute {DELETE FROM nsmcp_records.rows WHERE rowid IN( SELECT rowid FROM nsmcp_records.rows INDEXED BY diagnostic_expiry WHERE kind<>0 AND completed<$cutoff ORDER BY completed LIMIT $batchRows)} dict set counts diagnostics [${:db} changes] ${:db} execute {DELETE FROM nsmcp_records.minutes WHERE rowid IN( SELECT rowid FROM nsmcp_records.minutes INDEXED BY minute_time WHERE slot<$cutoff ORDER BY slot LIMIT $batchRows)} dict set counts minutes [${:db} changes] ${:db} execute {DELETE FROM nsmcp_records.peer_addresses WHERE rowid IN( SELECT rowid FROM nsmcp_records.peer_addresses INDEXED BY peer_expiry WHERE seen<$peerCutoff ORDER BY seen LIMIT $batchRows)} dict set counts peers [${:db} changes] if {[dict get $deleted n]>0} { set first [dict get $deleted first]; set last [dict get $deleted last]; set count [dict get $deleted n] ${:db} execute {INSERT INTO nsmcp_records.evictions(dropped_at,first_time,last_time,rows,reason) VALUES($now,$first,$last,$count,'age'); DELETE FROM nsmcp_records.evictions WHERE id NOT IN(SELECT id FROM nsmcp_records.evictions ORDER BY id DESC LIMIT 64)} } :statePut droppedThrough [expr {$cutoff-1}] :statePut lastEvictionAt $now } } finally {${:db} execute {DROP TABLE IF EXISTS temp.nsmcp_retention_batch}} # Keep committed ranges as ingestion watermarks after expiry, preventing # the continuous backfill from resurrecting deliberately expired rows. return $counts } :public method close {} {:destroy} :method disconnect {} { if {${:db} ne {}} { if {${:querying} || [${:db} inTransaction]} {catch {${:db} execute ROLLBACK}; set :querying false} catch {${:db} execute {PRAGMA query_only=OFF}} catch {${:db} execute {DROP VIEW IF EXISTS temp.nsmcp_record_window; DROP TABLE IF EXISTS temp.nsmcp_scan_batch; DROP TABLE IF EXISTS temp.nsmcp_record_scope; DROP TABLE IF EXISTS temp.nsmcp_retention_batch; DROP TABLE IF EXISTS temp.nsmcp_ingest_batch; DROP VIEW IF EXISTS temp.nsmcp_access_requests; DROP VIEW IF EXISTS temp.nsmcp_access_metadata; DROP VIEW IF EXISTS temp.nsmcp_access_evictions}} if {${:attached}} {catch {${:db} execute {DETACH DATABASE nsmcp_records}}} if {${:prior} ne {}} { set busy [dict get ${:prior} busy]; set temp [dict get ${:prior} temp] set queryOnly [expr {[dict exists ${:prior} queryOnly] ? [dict get ${:prior} queryOnly] : 0}] catch {${:db} execute "PRAGMA busy_timeout=$busy; PRAGMA temp_store=$temp; PRAGMA query_only=$queryOnly"} } ${:db} close; set :db {}; set :handle {}; set :attached false; set :prior {} } elseif {${:handle} ne {}} {catch {ns_db releasehandle ${:handle}}} } :method destroy {} {:disconnect; next}XQL Not present: Generic, PostgreSQL, Oracle
