This procedure is defined in the server but not documented via ad_proc or proc_doc and may be intended as a private interface.
The procedure is defined as:
proc nsmcp::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 batch}} {error "accessRecordIngestion must be tcl or batch"}
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 [expr {[dict exists $options scanLimits] ? [dict get $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 open-store
set store [accessRecordStoreOpen $settings writer $observer]
if {![$store available]} {
dict incr state recordPoolWaitMilliseconds [expr {max(1,[clock milliseconds]-$waitStart)}]
set waitMs [expr {[dict exists $settings waitMs] ? [dict get $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]
}
if {![dict exists $settings unified] || ![dict get $settings unified]} {$store prune}
set budget [expr {min([dict get $limits slice],[dict get $limits bytes]-[dict get $state examinedBytes])}]; set started [clock milliseconds]
set maxMs [expr {[dict exists $limits milliseconds] ? [dict get $limits milliseconds] : 1000}]
while {$index<[llength $files] && $budget>0 && [clock milliseconds]-$started<$maxMs} {
accessIngestionObserve $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 native [expr {[dict exists $settings ingestion] && [dict get $settings ingestion] eq "batch" && ![dict get $file dropping] && [dict get $limits line]>=128 && [dict get $limits line]<=1048576}]
if {$native} {
lassign [$store nativeRead $fid $file $offset $stop $budget $limits $redactor [dict exists $options ingestOnly]] next tuples pending direct
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 native [expr {$next>$offset || $pending}]
set offset $next
}
# Legacy reader follows when native framing made no progress.
if {!$native} {
accessIngestionObserve $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
try {
while {$offset<$stop && [llength $tuples]<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}
lappend tuples [$store tuple $fid $offset [dict get $line end] $kind $row $redactor $file]
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
lappend tuples [$store tuple $fid $offset [dict get $line end] 2 {} $redactor $file]
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}
}
lappend tuples [$store tuple $fid $offset $next $kind $row $redactor $file]
set offset $next
}
} finally {$reader destroy; close $f}
}
accessScanUnchanged $file
if {$offset>$begin} {
set batch [$store ingest $fid $file $begin $offset $tuples]
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 committed $consumed $batch
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 exists $settings unified] || ![dict get $settings unified]} {$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 [expr {[dict exists $settings waitMs] ? [dict get $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}}
}
