Class ::nsmcp::AccessRecordSupport (public)

 ::nx::Trait ::nsmcp::AccessRecordSupport

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

Testcases:

No testcase defined.

Source code:

    :method accessIngestionObserve {
        -observer
        {-bytes 0}
        {-batch {}}
        phase
    } {
        #
        # Publish progress through the instance's observer callback.
        #
        # Orchestration may override the callback for one invocation without
        # retaining its token on the object. An explicit empty override disables
        # dispatch even when the instance has an observer.
        #
        # @param observer Optional invocation-local callback override.
        # @param bytes Committed source bytes for this event.
        # @param batch Optional committed-batch statistics.
        # @param phase Progress phase name.
        #
        # @return The callback result, or an empty result when disabled.
        #

        if {![info exists observer]} {
            if {[info exists :observer]} {
                set observer ${:observer}
            } else {
                set observer {}
            }
        }
        if {$observer ne {}} {
            set call [list {*}$observer $phase $bytes]
            if {$batch ne {}} {
                lappend call $batch
            }
            return [uplevel #0 $call]
        }
    }

    # Prepare file evidence and merged coverage before persistence opens a transaction.
    :method accessRecordCheckpoint {store fid file begin end} {
        set db [$store session]
        # Bound fragmented coverage metadata, not the number of batches.
        # A batch joining existing coverage is permitted at the limit.
        set maxRanges 4096
        if {[$db scalar {SELECT count(*) FROM main.ranges WHERE fid=:fid}] >= $maxRanges
            && ![$db scalar {SELECT count(*) FROM main.ranges
                WHERE fid=:fid AND lo<=:end AND hi>=:begin}]} {
            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 [:dict getdef $file dropping 0]
        set lo $begin
        set hi $end
        foreach span [$db execute {SELECT lo,hi,anchor FROM main.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]
            }
        }
        return [dict create begin $begin end $end lo $lo hi $hi anchor $anchor  stamp $stamp now $now dropping $dropping]
    }

    # Bound the whole-operation contention retry allowance.
    :method accessRecordLockWait {section} {
        set ms [expr {int(1000*[ns_baseunit -time [ns_config $section accessRecordLockWait 3s]])}]
        if {$ms<100 || $ms>30000} {
            error "accessRecordLockWait must be 100ms through 30s"
        }
        return $ms
    }

    # Reserve journal space with integer, page-aligned accounting.
    :method accessRecordMainPages {settings pageSize} {
        return [expr {wide([dict get $settings maxBytes])*2/5/$pageSize}]
    }

    # Open a read-only adapter and verify its configured source binding.
    :method accessRecordReaderOpen {settings} {
        set store [:accessRecordStoreOpen $settings reader]
        if {![$store available]} {
            $store close
            return -code error -errorcode {NSMCP RECORD_STORE pool-busy} "Access store pool is temporarily busy"
        }
        if {[dict exists $settings sourcePath]} {
            try {
                set source [$store stateGet sourcePath {}]
                if {$source eq {} || $source ne [dict get $settings sourcePath]} {
                    error "record-store-source-mismatch" {} {NSMCP RECORD_STORE source-mismatch}
                }
            } on error {message options} {
                $store close
                return -options $options $message
            }
        }
        return $store
    }

    # Retry the whole operation with a fresh snapshot in the original caller scope.
    :method accessRecordRetry {settings script {publish {}} {observer {}}} {
        set budget [:dict getdef $settings lockWaitMs 3000]
        set began [clock milliseconds]
        set delay 25
        while {1} {
            try {
                return [uplevel 1 $script]
            } on error {message options} {
                set code [dict get $options -errorcode]
                set transient [expr {[lrange $code 0 1] eq {NSMCP RECORD_STORE} && [lindex $code 2] in {database-locked database-busy pool-busy}}]
                if {!$transient && [regexp -nocase {database (is |table is |schema is )?(locked|busy)} $message]} {
                    set transient true
                }
                if {!$transient || [clock milliseconds]-$began >= $budget} {
                    return -options $options $message
                }
                :accessIngestionObserve -observer $observer wait-lock
                accessScanProgress $publish waiting-record-store {files {} index 0 examinedBytes 0} {}
                set left [expr {$budget-([clock milliseconds]-$began)}]
                if {$left>0} {
                    ns_sleep [expr {min($left,$delay)/1000.0}]
                }
                set delay [expr {min(250,$delay*2)}]
            }
        }
    }

    # Run one bounded slice; resumable state contains cursors, never live handles.
    :method accessRecordScan {paths directory options redactor limits publish settings {observer {}}} {
        if {![dict exists $settings ingestion]} {
            dict set settings ingestion [ns_config ns/server/[ns_info server]/module/nsmcp accessRecordIngestion tcl]
        }
        if {[dict get $settings ingestion] ni {tcl c batch}} {
            error "accessRecordIngestion must be tcl or c (batch is a legacy alias for c)"
        }
        set options [accessFailureOptions $options]
        set selection [accessFailureSelection $options]
        set policy [list schema1 [::nsmcp::accessRecordParser cget -timingColumn]]
        foreach key {queryParameters headerNames maskPeerAddresses maskUserIdentifiers replacement} {
            lappend policy [$redactor cget -$key]
        }
        if {[dict exists $settings policy] && [dict get $settings policy] ne $policy} {
            error "Record-store parsing/redaction policy changed; start a new analysis"
        }
        dict set settings policy $policy
        set configured [:dict getdef $options scanLimits {}]
        set limits [dict merge [accessScanDefaultLimits] [accessAggregateSettings] $configured $limits]
        if {[dict exists $options scanState]} {
            set state [dict get $options scanState]
            if {![dict exists $state recordStore] || [dict get $state failedTrafficSelection] ne $selection} {
                error "Record-store selection changed; start a new analysis"
            }
            set settings [dict merge [dict create ingestion [dict get $settings ingestion]] [dict get $state recordStore]]
            dict set state recordStore $settings
            set limits [dict get $state limits]
        } else {
            set state [accessScanDiscover $paths [dict get $options includeRotations] [dict get $limits files]]
            dict set state index 0
            dict set state stats [accessScanEmpty]
            dict set state examinedBytes 0
            dict set state parsedRecordBytes 0
            dict set state reusedBytes 0
            dict set state nativeSearch false
            dict set state recordStore $settings
            dict set state limits $limits
            dict set state failedTrafficSelection $selection
            # Compressed generations retain the established bounded scanner. No
            # partial persistent result is silently substituted for these sources.
            foreach file [dict get $state files] {
                if {[dict get $file compressed] && ![dict exists $options ingestOnly]} {
                    return {}
                }
            }
            set files {}
            foreach file [dict get $state files] {
                if {[dict get $file compressed]} {
                    dict set file status unsupported-compressed
                } elseif {[accessOrderedEnabled $options]} {
                    accessOrderedPlan file state $options $limits $publish
                }
                dict set file archiveLo [dict get $file offset]
                lappend files $file
            }
            dict set state files $files
        }
        set store {}
        set files [dict get $state files]
        set index [dict get $state index]
        try {
            set waitStart [clock milliseconds]
            :accessIngestionObserve -observer $observer open-store
            set store [:accessRecordStoreOpen $settings writer $observer]
            if {![$store available]} {
                dict incr state recordPoolWaitMilliseconds [expr {max(1,[clock milliseconds]-$waitStart)}]
                set waitMs [:dict getdef $settings waitMs 30000]
                if {[dict get $state recordPoolWaitMilliseconds]>=$waitMs} {
                    error "record-store-pool-wait-limit"
                }
                accessScanProgress $publish waiting-record-store $state {}
                return [list {} object {} $state]
            }
            set batchedInsertion [$store batchInsertionAvailable]
            if {![:dict getdef $settings unified false]} {
                $store prune
            }
            set budget [expr {min([dict get $limits slice],[dict get $limits bytes]-[dict get $state examinedBytes])}]
            set started [clock milliseconds]
            set maxMs [:dict getdef $limits milliseconds 1000]
            while {$index<[llength $files] && $budget>0 && [clock milliseconds]-$started<$maxMs} {
                :accessIngestionObserve -observer $observer prepare-source
                set file [lindex $files $index]
                if {[dict get $file status] in {outside-window unsupported-compressed}} {
                    incr index
                    continue
                }
                accessScanUnchanged $file
                set fid [$store prepare $file]
                if {[dict exists $file archiveFid] && $fid ne [dict get $file archiveFid]} {
                    error "record-store-generation-changed"
                }
                dict set file archiveFid $fid
                set offset [dict get $file offset]
                dict set file dropping [$store dropState $fid $offset]
                set end [expr {[dict exists $file scanEnd] ? [dict get $file scanEnd] : [dict get $file size]}]
                set covered [$store coveredEnd $fid $offset $end]
                if {$covered>$offset} {
                    dict incr state reusedBytes [expr {$covered-$offset}]
                    set offset $covered
                } else {
                    set batchStarted [clock clicks -milliseconds]
                    set stop [$store gapEnd $fid $offset $end]
                    set begin $offset
                    set tuples {}
                    set pending false
                    set counters {malformed 0 oversized 0 skipped 0}
                    set useCScanner [expr {[dict exists $settings ingestion]
                                           && [dict get $settings ingestion] in {c batch} && ![dict get $file dropping]
                                           && [dict get $limits line]>=128
                                           && [dict get $limits line]<=1048576}]
                    if {$useCScanner} {
                        lassign [$store cRead $fid $file $offset $stop $budget $limits $redactor [dict exists $options ingestOnly] $batchedInsertion] next tuples pending direct counters
                        if {$direct eq "true"} {
                            return {}
                        }
                        # A framing budget may end before emitting a row. Retain the
                        # established bounded Tcl reader for such exceptional spans.
                        set useCScanner [expr {$next>$offset || $pending}]
                        set offset $next
                    }
                    # The Tcl reader follows when C framing made no progress.
                    if {!$useCScanner} {
                        :accessIngestionObserve -observer $observer scan-tcl
                        set f [open [dict get $file path] rb]
                        seek $f $offset
                        set readLimit [expr {min($stop,$offset+max([dict get $limits slice],[dict get $limits line]+1),$offset+[dict get $limits bytes]-[dict get $state examinedBytes])}]
                        set reader [::nsmcp::LogSummaryReader new -channel $f -limit $readLimit -offset $offset -maxBytes [dict get $limits line]]
                        set begin $offset
                        set tuples {}
                        set pending false
                        set counters {malformed 0 oversized 0 skipped 0}
                        set processed 0
                        try {
                            while {$offset<$stop && $processed<512 && $offset-$begin<$budget} {
                                set line [$reader nextLine]
                                if {$line eq {}} {
                                    break
                                }
                                if {[dict exists $options ingestOnly] && [dict get $file dropping]} {
                                    set kind 3
                                    set row {}
                                    if {[dict get $line terminated]} {
                                        dict set file dropping 0
                                    }
                                    dict incr counters skipped
                                    incr processed
                                    set offset [dict get $line end]
                                    continue
                                }
                                if {![dict get $line terminated]} {
                                    if {$readLimit<$stop} {
                                        if {[dict exists $options ingestOnly] && [dict get $line truncated]} {
                                            dict set file dropping 1
                                            dict incr counters oversized
                                            incr processed
                                            set offset [dict get $line end]
                                            break
                                        }
                                        return {}
                                    }
                                    # A slice boundary can cut a line. Retry it from its
                                    # original byte offset in the next slice; a true
                                    # snapshot tail stays pending until a future query.
                                    if {$stop==$end} {
                                        set pending true
                                    }
                                    break
                                }
                                set next [dict get $line end]
                                set kind 0
                                set row {}
                                if {[dict get $line truncated]} {
                                    set kind 2
                                } else {
                                    set row [::nsmcp::accessRecordParser parse [dict get $line text]]
                                    if {$row ne {} && [dict exists $row timeDependent]} {
                                        if {[dict exists $options ingestOnly]} {
                                            set row {}
                                        } else {
                                            return {}
                                        }
                                    }
                                    if {$row eq {}} {
                                        set kind 1
                                    }
                                }
                                incr processed
                                if {$kind==0} {
                                    lappend tuples [$store tuple $fid $offset $next 0 $row $redactor $file $batchedInsertion]
                                } elseif {$kind==2} {
                                    dict incr counters oversized
                                } else {
                                    dict incr counters malformed
                                }
                                set offset $next
                            }
                        } finally {
                            $reader destroy
                            close $f
                        }
                    }
                    accessScanUnchanged $file
                    if {$offset>$begin} {
                        :accessIngestionObserve -observer $observer prepare-checkpoint
                        set checkpoint [dict merge [:accessRecordCheckpoint $store $fid $file $begin $offset] $counters]
                        set batch [$store ingest $fid $checkpoint $tuples $batchedInsertion]
                        dict set batch scannerBackend [expr {$useCScanner ? "c" : "tcl"}]
                        dict set batch durationMs [expr {max(0,[clock clicks -milliseconds]-$batchStarted)}]
                        dict set batch recordsPerSecond [expr {[dict get $batch durationMs]>0 ? 1000.0*[dict get $batch records]/[dict get $batch durationMs] : {}}]
                        set consumed [expr {$offset-$begin}]
                        dict incr state examinedBytes $consumed
                        dict incr state parsedRecordBytes $consumed
                        incr budget -$consumed
                        dict set batch parsedBytes $consumed
                        :accessIngestionObserve -observer $observer -bytes $consumed -batch $batch committed
                        dict incr file scanReadBytes $consumed
                    } elseif {!$pending && $stop<$end} {
                        # Ensure a configured slice smaller than one complete line
                        # cannot cause an endless zero-progress retry.
                        error "record-store-slice-smaller-than-record"
                    }
                    if {$pending} {
                        dict set file archivePending true
                        set end $offset
                    }
                }
                dict set file offset $offset
                if {$offset>=$end} {
                    dict set file status complete
                    incr index
                } else {
                    dict set file status running
                }
                lset files [dict get $file fileIndex] $file
                dict set state files $files
                dict set state index $index
                accessScanProgress $publish ingesting-records $state $file $offset [dict get $file size]
                if {[dict get $state examinedBytes]>=[dict get $limits bytes]} {
                    break
                }
            }
            if {$index<[llength $files]} {
                if {[dict get $state examinedBytes]<[dict get $limits bytes]} {
                    return [list {} object {} $state]
                }
                for {set i $index} {$i<[llength $files]} {incr i} {
                    set file [lindex $files $i]
                    dict set file status scan-byte-limit
                    lset files $i $file
                }
            }
            foreach file $files {
                if {[dict get $file status] ne "outside-window"} {
                    accessScanUnchanged $file
                }
            }
            dict set state files $files
            if {[dict exists $options ingestOnly]} {
                dict set state ingestComplete true
                return [list {} object {} $state]
            }
            accessScanProgress $publish querying-records $state {}
            $store scope $files
            lassign [$store stats $options $redactor $limits] stats aggregate selected
            lassign [$store samples $selected $files $redactor] samples successes
            dict set stats samples $samples
            dict set stats successSamples $successes
            foreach file $files {
                if {[dict exists $file archivePending]} {
                    dict incr stats pendingRecords
                }
            }
            dict set state stats $stats
            dict set state aggregateResult $aggregate
            set result [accessScanResult $state $options $limits $redactor]
            set fields [lindex $result 2]
            lappend fields recordStorage object [list backend string sqlite persistent boolean true parsedBytesThisJob number [dict get $state parsedRecordBytes] locatorProbeBytes number [expr {[dict get $state examinedBytes]-[dict get $state parsedRecordBytes]}]  reusedBytes number [dict get $state reusedBytes] diskBytes number [$store diskBytes] maxBytes number [dict get $settings maxBytes]  retentionSeconds number [dict get $settings retention] retentionScope string unused-source-generations temporarySortDiskIncluded boolean false reuseValidation string {File identity, original prefix and committed range boundary anchors; assumes immutable interiors of append-only log generations.}]
            lset result 2 $fields
            # Run retention only after querying this job's fixed scopes. Expiration
            # never removes current active-query rows before they are summarized.
            $store finish
            if {![:dict getdef $settings unified false]} {
                $store prune
            }
            return $result
        } on error {message opts} {
            if {[lrange [dict get $opts -errorcode] 0 1] eq {NSMCP RECORD_STORE}} {
                set failure [dict create class [lindex [dict get $opts -errorcode] 2] diagnosticId [lindex [dict get $opts -errorcode] 3]]
            } elseif {$store ne {}} {
                set failure [$store diagnostic $message $opts]
            } else {
                set failure [accessAggregationDiagnostic record-store $message $opts]
            }
            if {[dict get $failure class] in {database-locked database-busy}} {
                dict incr state recordPoolWaitMilliseconds 100
                set waitMs [:dict getdef $settings waitMs 30000]
                if {[dict get $state recordPoolWaitMilliseconds]<$waitMs} {
                    accessScanProgress $publish waiting-record-store $state {}
                    return [list {} object {} $state]
                }
            }
            return -code error -errorcode [list NSMCP RECORD_STORE [dict get $failure class]] "Persistent access-record analysis failed ([dict get $failure class], diagnostic [dict get $failure diagnosticId]); committed ranges remain available."
        } finally {
            if {$store ne {}} {
                $store close
            }
        }
    }

    # Keep concrete engine selection and failed-constructor cleanup in one factory.
    :method accessRecordStoreOpen {settings {mode writer} {observer {}}} {
        # This factory is the only scanner entry point tied to a concrete engine.
        # Adapter batches are opaque to the scanner; no handle crosses a slice.
        # Keep the allocated name so failed NX initialization cannot leave an
        # orphaned adapter behind on each scheduled retry.
        set object ::nsmcp::recordStore_[ns_crypto::randombytes -encoding hex 12]
        try {
            return [::nsmcp::AccessRecordStoreSQLite create $object -settings $settings -mode $mode -observer $observer]
        } on error {message options} {
            if {[nsf::object::exists $object]} {
                catch {
                    $object close
                }
            }
            return -options $options $message
        }
    }

    # Read configured storage, retention and contention limits.
    :method accessRecordStoreSettings {} {
        set section ns/server/[ns_info server]/module/nsmcp
        if {[nsv_exists nsmcp:access-store:[ns_info server] settings]} {
            return [nsv_get nsmcp:access-store:[ns_info server] settings]
        }
        set pool [ns_config $section accessRecordPool {}]
        if {$pool eq {}} {
            return {}
        }
        set datasource [ns_config ns/db/pool/$pool datasource {}]
        if {$datasource eq {} || [file pathtype $datasource] ne "absolute"} {
            error "accessRecordPool requires a persistent absolute datasource"
        }
        set directory [file dirname [file normalize $datasource]]
        set backend [ns_config $section accessRecordBackend sqlite]
        if {$backend ne "sqlite"} {
            error "unsupported accessRecordBackend"
        }
        set ingestion [ns_config $section accessRecordIngestion tcl]
        if {$ingestion ni {tcl c batch}} {
            error "accessRecordIngestion must be tcl or c (batch is a legacy alias for c)"
        }
        set settings [accessAggregateSettings]
        set bytes [accessScanBudget [ns_config $section accessRecordMaxBytes 16GiB]]
        if {$bytes < 4194304} {
            error "accessRecordMaxBytes must be at least 4MiB"
        }
        set retention [ns_baseunit -time [ns_config $section accessRecordRetention 7d]]
        if {$retention<60} {
            error "accessRecordRetention must be at least 60s"
        }
        return [dict create directory [file normalize $directory]  pool $pool  maxBytes $bytes  retention $retention  ingestion $ingestion  waitMs [dict get $settings aggregateWaitMs]  lockWaitMs [:accessRecordLockWait $section]]
    }

    # A totals lookup reads stored snapshots and never imports missing log ranges.
    :method accessRecordTotals {paths options redactor limits publish settings} {
        set began [clock milliseconds]
        try {
            return [:accessRecordRetry $settings [list :accessRecordTotalsOnce $paths $options $redactor $limits $publish $settings] $publish]
        } on error {message opts} {
            return [:accessRecordUnavailable $options $settings $publish $message $opts $began]
        }
    }

    # Validate source coverage and read one consistent totals snapshot.
    :method accessRecordTotalsOnce {paths options redactor limits publish settings} {
        set began [clock milliseconds]
        set fields [list server string [ns_info server] mode string totals from number [dict get $options from] to number [dict get $options to] timezone string [dict get $options timezone] ingestionPerformed boolean false storageConfigured boolean [expr {$settings ne {}}]]
        set baseFields $fields
        if {$settings eq {}} {
            accessScanProgress $publish complete {files {} index 0 examinedBytes 0} {} {} {} [dict create scanStatus partial completionSummary {Stored totals unavailable: persistent record storage is not configured. No log import was started.} truncatedByLimits false]
            lappend fields available boolean false reason string record-store-not-configured counts null {} coverage object [list countsCompleteForSnapshots boolean false windowComplete boolean false] lookupTime object [list wallMs number [expr {[clock milliseconds]-$began}] scope string stored-totals-lookup]
            return [list {} object $fields]
        }
        set policy [list schema1 [::nsmcp::accessRecordParser cget -timingColumn]]
        foreach key {queryParameters headerNames maskPeerAddresses maskUserIdentifiers replacement} {
            lappend policy [$redactor cget -$key]
        }
        dict set settings policy $policy
        set configured [:dict getdef $options scanLimits {}]
        set limits [dict merge [accessScanDefaultLimits] [accessAggregateSettings] $configured $limits]
        set state [accessScanDiscover $paths [dict get $options includeRotations] [dict get $limits files]]
        dict set state stats [accessScanEmpty]
        dict set state index 0
        dict set state examinedBytes 0
        accessScanProgress $publish checking-stored-totals $state {}
        set store {}
        set scopes {}
        set sourceFiles {}
        set complete true
        set compressed 0
        set pending 0
        set totalBytes 0
        set storedBytes 0
        set skipped 0
        try {
            set store [:accessRecordReaderOpen $settings]
            if {![$store available]} {
                accessScanProgress $publish complete $state {} {} {} [dict create scanStatus partial completionSummary {Stored totals unavailable: record pool is busy. No log import was started.} truncatedByLimits false]
                lappend fields available boolean false reason string record-store-pool-busy counts null {} coverage object [list countsCompleteForSnapshots boolean false windowComplete boolean false]
                return [list {} object $fields]
            }
            foreach file [dict get $state files] {
                accessScanUnchanged $file
                if {[dict get $file compressed]} {
                    dict set file status unsupported-compressed-source
                    dict set state files [lreplace [dict get $state files] [dict get $file fileIndex] [dict get $file fileIndex] $file]
                    incr compressed
                    set complete false
                    lappend sourceFiles [list fileIndex [dict get $file fileIndex] status unsupported-compressed-source snapshotBytes [dict get $file size]]
                    continue
                }
                if {[accessOrderedEnabled $options]} {
                    accessOrderedPlan file state $options $limits $publish
                }
                if {[dict get $file status] eq "outside-window"} {
                    dict set state files [lreplace [dict get $state files] [dict get $file fileIndex] [dict get $file fileIndex] $file]
                    incr skipped
                    continue
                }
                set fid [$store prepare $file]
                if {$fid<0} {
                    set fid [expr {-[dict get $file fileIndex]-1}]
                }
                set lo [dict get $file offset]
                set hi [expr {[dict exists $file scanEnd] ? [dict get $file scanEnd] : [dict get $file size]}]
                dict set file archiveFid $fid
                dict set file archiveLo $lo
                dict set file offset $hi
                lappend scopes $file
                incr totalBytes [expr {$hi-$lo}]
            }
            # Counts and committed-range coverage share a single read transaction.
            $store scope $scopes
            foreach file $scopes {
                accessScanUnchanged $file
                set lo [dict get $file archiveLo]
                set hi [dict get $file offset]
                set covered [$store coveredBytes [dict get $file archiveFid] $lo $hi]
                incr storedBytes $covered
                dict set file status [expr {$covered==$hi-$lo ? "complete" : "not-ingested"}]
                dict set state files [lreplace [dict get $state files] [dict get $file fileIndex] [dict get $file fileIndex] $file]
                if {$covered!=$hi-$lo} {
                    set complete false
                    incr pending
                }
                lappend sourceFiles [list fileIndex [dict get $file fileIndex] status [expr {$covered==$hi-$lo ? "stored" : "not-fully-ingested"}] snapshotBytes [dict get $file size] selectedBytes [expr {$hi-$lo}] storedBytes $covered missingBytes [expr {$hi-$lo-$covered}]]
            }
            lassign [$store totals $options] counts statuses diagnostics
            foreach file $scopes {
                accessScanUnchanged $file
            }
            if {[dict get $state omittedFiles] || [dict get $state unavailableFiles]} {
                set complete false
            }
            set rangesComplete $complete
            if {[dict get $diagnostics parseErrors] || [dict get $diagnostics oversizedRecords] || [dict get $diagnostics pendingRecords] || ![:dict getdef $diagnostics diagnosticCountsComplete true]} {
                set complete false
            }
            set countFields {}
            dict for {key value} $counts {
                lappend countFields $key number $value
            }
            set statusFields {}
            set i 0
            foreach row $statuses {
                lappend statusFields $i object [list status number [dict get $row status] count number [dict get $row count]]
                incr i
            }
            set files {}
            set i 0
            foreach row $sourceFiles {
                set rowFields {}
                foreach {key value} $row {
                    lappend rowFields $key [expr {$key eq "status" ? "string" : "number"}] $value
                }
                lappend files $i object $rowFields
                incr i
            }
            set coverage [list countsCompleteForSnapshots boolean $complete sourceRangesFullyIngested boolean $rangesComplete countsAreLowerBounds boolean [expr {!$complete}]  diagnosticCountsComplete boolean [:dict getdef $diagnostics diagnosticCountsComplete true] windowComplete boolean false orderingAssumed boolean [accessOrderedEnabled $options] files array $files  filesDiscovered number [llength [dict get $state files]] filesSkippedOutsideWindow number $skipped filesMissingIngestion number $pending  compressedFilesUnsupported number $compressed omittedFiles number [dict get $state omittedFiles] unavailableFiles number [dict get $state unavailableFiles]  selectedSourceBytes number $totalBytes storedSourceBytes number $storedBytes missingSourceBytes number [expr {$totalBytes-$storedBytes}]  retentionVerified boolean false]
            dict for {key value} $diagnostics {
                lappend coverage $key [expr {$key eq "queryStrategy" ? "string" : "number"}] $value
            }
            lappend fields available boolean true reason string [expr {$complete ? "validated-stored-snapshot-totals" : "stored-prefix-or-partial-ranges-only"}]  counts object $countFields statuses array $statusFields coverage object $coverage  originDetailsIncluded boolean false countsScope string {Stored matching access records only. Missing ingestion, parse diagnostics, unsupported compressed files, ordered selection and unverified retention qualify these counts. No daily extrapolation.}  lookupTime object [list wallMs number [expr {[clock milliseconds]-$began}] scope string stored-totals-lookup]
            $store finish
            accessScanProgress $publish complete $state {} {} {} [dict create scanStatus [expr {$complete ? "complete" : "partial"}] completionSummary [expr {$complete ? "Stored totals cover the selected source snapshots; retention remains unverified." : "Stored totals are incomplete: missing ingestion or source/parse limitations; no log import was started."}] truncatedByLimits false]
            return [list {} object $fields]
        } on error {message opts} {
            if {[regexp -nocase {database.*(locked|busy)} $message] || [lindex [dict get $opts -errorcode] 2] in {database-locked database-busy pool-busy}} {
                return -options $opts $message
            }
            if {[lrange [dict get $opts -errorcode] 0 1] eq {NSMCP RECORD_STORE}} {
                set kind [lindex [dict get $opts -errorcode] 2]
                set id [lindex [dict get $opts -errorcode] 3]
            } else {
                if {$store ne {}} {
                    set failure [$store diagnostic $message $opts]
                } else {
                    set failure [accessAggregationDiagnostic record-store $message $opts]
                }
                set kind [dict get $failure class]
                set id [dict get $failure diagnosticId]
            }
            set fields $baseFields
            accessScanProgress $publish complete  $state {} {} {}  [dict create scanStatus partial completionSummary "Stored totals unavailable ($kind). No log import was started." truncatedByLimits false]
            lappend fields available boolean false reason string record-store-$kind counts null {}  diagnostic object [list class string $kind id string $id]  coverage object [list countsCompleteForSnapshots boolean false windowComplete boolean false]  lookupTime object [list wallMs number [expr {[clock milliseconds]-$began}] scope string stored-totals-lookup]
            return [list {} object $fields]
        } finally {
            if {$store ne {}} {
                $store close
            }
        }
    }

    # Report unavailable evidence explicitly; never substitute zero counts.
    :method accessRecordUnavailable {options settings publish message opts began} {
        set code [dict get $opts -errorcode]
        if {[lrange $code 0 1] eq {NSMCP RECORD_STORE}} {
            set kind [lindex $code 2]
            set id [lindex $code 3]
            if {$kind in {database-locked database-busy pool-busy} && $id eq {}} {
                set failure [accessAggregationDiagnostic record-store $message $opts]
                set id [dict get $failure diagnosticId]
            }
        } else {
            set failure [accessAggregationDiagnostic record-store $message $opts]
            set kind [dict get $failure class]
            set id [dict get $failure diagnosticId]
        }
        accessScanProgress $publish complete  {files {} index 0 examinedBytes 0} {} {} {}  [dict create scanStatus partial completionSummary "Stored totals unavailable ($kind) after bounded contention retries; no scan started." truncatedByLimits false]
        return [list {} object [list  server string [ns_info server]  mode string totals  from number [dict get $options from]  to number [dict get $options to]  timezone string [dict get $options timezone]  storageConfigured boolean [expr {$settings ne {}}]  ingestionPerformed boolean false  available boolean false  reason string record-store-$kind  counts null {}  diagnostic object [list class string $kind id string $id]  coverage object {
                                        countsCompleteForSnapshots boolean false 
                                        windowComplete boolean false
                                    } lookupTime object [list wallMs number [expr {[clock milliseconds]-$began}]]]]
    }

XQL Not present:

Generic, PostgreSQL, Oracle