nsmcp::LogCorrelation method accessFileScan (public)

 <instance of nsmcp::LogCorrelation> accessFileScan paths \
    directory options redactor [ limits ] [ publish ]

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

Scan selected access-log sources for a bounded file summary. Use the configured limits and optional progress callback while retaining source coverage information.

Parameters:

paths (required)

Selected access-log source paths.

directory (required)

Private cache or source directory used by the operation.

options (required)

Tcl dictionary containing the operation options.

redactor (required)

Redactor object applied before returning diagnostic evidence.

limits (optional)

Scan resource limits; empty selects the configured defaults.

publish (optional)

Optional progress-publication callback.

Returns:

The file-summary scan result and continuation information when work remains.

Testcases:

No testcase defined.

Source code:

set options [::nsmcp::accessFailureOptions $options]
set unified [:accessRecordStoreSettings]
if {$unified ne {} && [:dict getdef $unified unified false]} {
    return [: -local accessUnifiedRead $options $redactor $publish $unified]
}
if {[:dict getdef $options mode {}] eq "totals"} {
    set recordSettings [expr {[dict exists $limits recordStore] ? [dict get $limits recordStore] : [:accessRecordStoreSettings]}]
    return [:accessRecordTotals $paths $options $redactor $limits $publish $recordSettings]
}
# Evidence remains bounded and independent of the durable store. Existing
# pre-upgrade jobs retain their original scanner and snapshot semantics.
set recordSettings {}
if {[dict exists $options scanState recordStore]} {
    set recordSettings [dict get $options scanState recordStore]
} elseif {![dict exists $options scanState] && ![accessEvidenceLimit $options]} {
    if {[dict exists $limits recordStore]} {
        set recordSettings [dict get $limits recordStore]
    } else {
        set recordSettings [:accessRecordStoreSettings]
    }
}
if {$recordSettings ne {}} {
    set cached [:accessRecordScan $paths $directory $options $redactor $limits $publish $recordSettings]
    if {$cached ne {}} {
        return $cached
    }
    if {[dict exists $options scanState recordStore]} {
        dict unset options scanState
    }
}
set selectionKey [::nsmcp::accessFailureSelection $options]
if {[dict exists $options scanState] && (![dict exists $options scanState failedTrafficSelection]
    || [dict get $options scanState failedTrafficSelection] ne $selectionKey)} {
    error "Access summary selection changed or predates failed-traffic accounting; start a new scan"
}
set configured [:dict getdef $options scanLimits {}]
if {[dict exists $options scanState limits aggregatePool] && [dict exists $options scanState limits aggregateBytes]} {
    set aggregateDefaults [dict get $options scanState limits]
} elseif {[dict exists $configured aggregatePool] && [dict exists $configured aggregateBytes]} {
    set aggregateDefaults $configured
} else {
    set aggregateDefaults [accessAggregateSettings]
}
set limits [dict merge [accessScanDefaultLimits] $aggregateDefaults $configured $limits]
if {[accessEvidenceLimit $options]>0} {
    dict set limits bytes [expr {min([dict get $limits bytes],33554432)}]
    dict set limits decoded [expr {min([dict get $limits decoded],67108864)}]
}
if {![string is wideinteger -strict [dict get $limits bytes]] || [dict get $limits bytes]<1} {
    error "invalid access scan byte budget"
}
if {[dict exists $options scanState]} {
    set state [dict get $options scanState]
    if {[dict exists $state limits]} {
        set limits [dict get $state limits]
    } elseif {![dict exists $options scanLimits]} {
        # Preserve the original bound for pre-upgrade continuations.
        dict set limits bytes 2147483648
    }
} else {
    accessScanProgress $publish discovering {files {} index 0 examinedBytes 0} {}
    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
    set planned {}
    foreach file [dict get $state files] {
        if {![catch {
            accessScanUnchanged $file
        }]} {
            if {[accessOrderedEnabled $options]} {
                accessOrderedPlan file state $options $limits $publish
                lappend planned $file
                continue
            }
            accessScanRangeInit file
            set range [accessScanRangeLoad $file $options $limits]
            if {$range ne "" && ([dict get $range minimum] eq "" || [dict get $range maximum]<[dict get $options from] || [dict get $range minimum]>=[dict get $options to])} {
                dict set file rangeMin [dict get $range minimum]
                dict set file rangeMax [dict get $range maximum]
                dict set file timeRangeCacheHit true
                dict set file status outside-window
            } else {
                accessTimeIndexExcludeFile file $options $limits
            }
        }
        lappend planned $file
    }
    dict set state files $planned
}
dict set state failedTrafficSelection $selectionKey
dict set state limits $limits
dict set state nativeSearch [accessScanNativeEligible $options]
# Persist only the private spool descriptor across worker slices.
if {![dict exists $limits aggregatePool] || ![dict exists $limits aggregateBytes]} {
    foreach {key value} [accessAggregateSettings] {
        if {![dict exists $limits $key]} {
            dict set limits $key $value
        }
    }
}
dict set state limits $limits
set writer {}
try {
    set writer [accessAggregateOpen state $directory $limits]
    if {$writer ne "" && [dict get $state aggregate status] eq "waiting"} {
        set file [expr {[dict get $state index]<[llength [dict get $state files]] ? [lindex [dict get $state files] [dict get $state index]] : {}}]
        accessScanProgress $publish waiting-aggregation $state $file
        return [list {} object {} $state]
    }
    if {$writer ne ""} {
        dict set options aggregateWriter $writer
    }
    set files [dict get $state files]
    set index [dict get $state index]
    set budget [dict get $limits slice]
    while {$index<[llength $files] && $budget>0} {
        set fileIndex $index
        set file [lindex $files $index]
        dict set state files $files
        dict set state index $index
        accessScanProgress $publish checking $state $file
        if {[dict get $file status] eq "outside-window"} {
            if {[catch {
                accessScanUnchanged $file
            }]} {
                dict set file status changed-or-unreadable
            }
            lset files $index $file
            incr index
            continue
        }
        if {[catch {
            accessScanUnchanged $file
        }]} {
            dict set file status changed-or-unreadable
            lset files $index $file
            incr index
            continue
        }
        accessScanRangeInit file
        set range {}
        if {![accessOrderedEnabled $options]} {
            set range [accessScanRangeLoad $file $options $limits]
        }
        if {$range ne ""} {
            dict set file rangeMin [dict get $range minimum]
            dict set file rangeMax [dict get $range maximum]
            dict set file timeRangeCacheHit true
            if {[dict get $range minimum] eq "" || [dict get $range maximum]<[dict get $options from]
                || [dict get $range minimum]>=[dict get $options to]} {
                dict set file status outside-window
                lset files $index $file
                incr index
                continue
            }
        }
        if {[dict get $state examinedBytes]>=[dict get $limits bytes]} {
            dict set file status scan-byte-limit
            if {[dict exists $file spool]} {
                file delete -- [dict get $file spool]
                dict unset file spool
            }
            lset files $index $file
            incr index
            continue
        }
        if {[dict get $file compressed] && ![dict exists $file spool]} {
            if {[catch {
                accessScanInflate $file $directory [dict get $limits decoded] [list ::nsmcp::accessScanProgress $publish decompressing $state $file] [dict get $limits heartbeat]
            } inflated inflateOptions]} {
                dict set file status gzip-invalid
                switch -- [dict get $inflateOptions -errorcode] {
                    {NSMCP ACCESS LIMIT compressed-bytes} {
                        dict set file status gzip-compressed-byte-limit
                    }
                    {NSMCP ACCESS LIMIT decoded-bytes} {
                        dict set file status gzip-decoded-byte-limit
                    }
                    {NSMCP ACCESS LIMIT gzip-time} {
                        dict set file status gzip-time-limit
                    }
                }
                lset files $index $file
                incr index
                continue
            }
            set file $inflated
            if {[catch {
                accessScanUnchanged $file
            }]} {
                file delete -- [dict get $file spool]
                dict unset file spool
                dict set file status changed-during-inflate
                lset files $index $file
                incr index
                continue
            }
            lset files $index $file
            dict set state files $files
        }
        if {[accessOrderedEnabled $options]} {
            accessOrderedPlan file state $options $limits $publish
            if {[dict get $file status] eq "outside-window"} {
                if {[dict exists $file spool]} {
                    file delete -- [dict get $file spool]
                    dict unset file spool
                }
                lset files $index $file
                incr index
                continue
            }
        }
        set size [dict get $file size]
        set path [dict get $file path]
        if {[dict exists $file spool]} {
            set path [dict get $file spool]
            set size [dict get $file decodedSize]
        }
        if {[dict exists $file scanEnd]} {
            set size [dict get $file scanEnd]
        }
        set channel [open $path rb]
        set offset [dict get $file offset]
        seek $channel $offset
        try {
            while {![accessOrderedEnabled $options] && $offset<$size && [accessTimeIndexAdvance $channel file state budget $options $limits $publish $size]} {
                set offset [dict get $file offset]
            }
        } on error {message opts} {
            close $channel
            return -options $opts $message
        }
        set offset [dict get $file offset]
        seek $channel $offset
        if {[dict exists $file timeIndexYield]} {
            dict unset file timeIndexYield
            set budget 0
        }
        if {$offset<$size && $budget>0} {
            dict set file scanStarted true
        }
        lset files $fileIndex $file
        dict set state files $files
        set readLimit [expr {min($size,$offset+[dict get $limits bytes]-[dict get $state examinedBytes])}]
        set cached {}
        if {![accessOrderedEnabled $options]} {
            set cached [accessTimeIndexLoad $file]
        }
        if {$cached ne "" && [dict exists $cached segments $offset]} {
            dict set file timeIndexReadThrough [dict get $cached segments $offset end]
        }
        if {[dict exists $file timeIndexReadThrough]} {
            set readLimit [expr {min($readLimit,[dict get $file timeIndexReadThrough])}]
        }
        if {$offset>=$size || $budget<=0} {
            set stats [dict get $state stats]
            close $channel
        } elseif {[dict get $state nativeSearch]} {
            set stats [dict get $state stats]
            accessScanProgress $publish searching $state $file $offset $size
            try {
                accessScanNativeSlice $channel $directory file state stats budget $options $redactor $limits $publish $readLimit $size
                set offset [dict get $file offset]
            } finally {
                close $channel
            }
        } else {
            set reader [::nsmcp::LogSummaryReader new -channel $channel -limit $readLimit -offset $offset -maxBytes [dict get $limits line]]
            set stats [dict get $state stats]
            set published [clock milliseconds]
            accessScanProgress $publish scanning $state $file $offset $size
            try {
                while {$offset<$readLimit && $budget>0 && [dict get $state examinedBytes]<[dict get $limits bytes]
                    && ![accessEvidenceDone $stats $options]} {
                    set line [$reader nextLine]
                    if {$line eq ""} {
                        error "access log ended before snapshot boundary"
                    }
                    set consumed [expr {[dict get $line end]-$offset}]
                    if {$consumed<=0} {
                        error "access log ended before snapshot boundary"
                    }
                    incr budget -$consumed
                    dict incr state examinedBytes $consumed
                    dict incr file scanReadBytes $consumed
                    set offset [dict get $line end]
                    dict incr stats examinedRecords
                    # Clock/NSV work is amortized; never publish per request.
                    if {([dict get $stats examinedRecords] & 255)==0} {
                        set now [clock milliseconds]
                        if {$publish ne "" && $now-$published >= [dict get $limits heartbeat]} {
                            accessScanProgress $publish scanning $state $file $offset $size
                            set published $now
                        }
                    }
                    if {[dict get $line truncated]} {
                        dict incr stats oversizedRecords
                        dict set file rangeUncertain true
                        continue
                    }
                    if {![dict get $line terminated]} {
                        dict incr stats pendingRecords
                        dict set file rangeUncertain true
                        continue
                    }
                    set row [::nsmcp::accessRecordParser parse [dict get $line text]]
                    if {$row eq ""} {
                        dict incr stats parseErrors
                        dict set file rangeUncertain true
                        continue
                    }
                    accessScanRangeAdd file $row $options
                    set time [dict get $row startTime]
                    foreach key {earliestAt latestAt} {
                        if {[dict get $stats $key] eq "" || ($key eq "earliestAt" && $time<[dict get $stats $key])
                            || ($key eq "latestAt" && $time>[dict get $stats $key])} {
                            dict set stats $key $time
                        }
                    }
                    accessScanAdd stats $row $file [dict get $line start] $options $redactor $limits
                }
            } finally {
                $reader destroy
                close $channel
            }
        }
        dict set state stats $stats
        if {[accessEvidenceDone $stats $options]} {
            dict set state evidenceStopped true
        }
        dict set file offset $offset
        if {[dict exists $file timeIndexReadThrough] && $offset>=[dict get $file timeIndexReadThrough]} {
            dict unset file timeIndexReadThrough
        }
        if {$offset >= $size} {
            dict set file status complete
            if {[dict exists $file spool]} {
                file delete -- [dict get $file spool]
                dict unset file spool
            }
            incr index
        } else {
            dict set file status running
        }
        # Verify again after reading; changed sources invalidate coverage.
        if {[catch {
            accessScanUnchanged $file
        }]} {
            dict set file status changed-during-read
            if {$index==$fileIndex} {
                incr index
            }
        }
        if {[dict exists $state evidenceStopped]} {
            dict set file status evidence-record-limit
            set index [expr {$fileIndex+1}]
        }
        if {[dict get $file status] eq "complete" && ![accessOrderedEnabled $options]} {
            accessScanRangeSave $file $options $limits
        }
        lset files $fileIndex $file
        # Advance the stored index together with the processed file.
        dict set state index $index
        dict set state files $files
        if {[dict exists $state evidenceStopped] || ($offset<$size && [dict get $file status] eq "running")} {
            break
        }
    }
    dict set state files $files
    dict set state index $index
    if {$writer ne ""} {
        dict set state aggregate [$writer snapshot]
    }
    if {$index<[llength $files] && ![dict exists $state evidenceStopped]} {
        accessScanProgress $publish waiting $state [lindex $files $index]
        return [list {} object {} $state]
    }
    accessScanProgress $publish finalizing $state {}
    if {$writer ne ""} {
        # Reuse the healthy slice connection. A failed write can require a
        # read session to recover the committed prefix; contention here
        # also yields at the saved EOF checkpoint instead of rescanning.
        if {![$writer readable]} {
            $writer close
            set writer {}
            set writer [::nsmcp::AccessCardinalityStore new -directory $directory -limits $limits -checkpoint [dict get $state aggregate] -readResults true]
            dict set state aggregate [$writer snapshot]
            if {[dict get $state aggregate status] eq "waiting"} {
                accessScanProgress $publish waiting-aggregation $state {}
                return [list {} object {} $state]
            }
        }
        dict set state aggregateResult [$writer results [dict get $options maxGroups]  [expr {[dict get $state stats requestCount]-[dict get $state stats unknownPeerRecords]}]  [expr {[dict get $state stats requestCount]-[dict get $state stats unknownUserAgentRecords]}]]
        foreach key {status reason diskBytes failure poolWaitCount poolWaitMilliseconds poolWaitLimitMilliseconds} {
            dict set state aggregate $key [dict get $state aggregateResult $key]
        }
        $writer close
        set writer {}
    }
    set result [accessScanResult $state $options $limits $redactor]
    accessScanCleanup $state
    set completion {}
    foreach {key type value} [lindex $result 2] {
        if {$key in {scanStatus truncatedByLimits completionSummary}} {
            dict set completion $key $value
        }
    }
    accessScanProgress $publish complete $state {} {} {} $completion
    return $result
} on error {message opts} {
    if {$writer ne ""} {
        catch {
            $writer close
        }
        set writer {}
    }
    accessScanProgress $publish failed $state {}
    accessScanCleanup $state
    return -options $opts $message
} finally {
    if {$writer ne ""} {
        $writer close
    }
}

XQL Not present:

Generic, PostgreSQL, Oracle