Class ::nsmcp::AccessRecordStoreSQLite (public)

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

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

Testcases:

No testcase defined.

Source code:

    :require trait ::nsmcp::AccessRecordSupport
    :require trait ::nsmcp::AccessRecordPartitions
    # nx::trait versions that export aliases publicly need explicit protection.
    foreach method [::nsmcp::AccessRecordSupport info methods -callprotection all] {
        ::nsf::method::property [self] $method call-protected true
    }

    foreach method [::nsmcp::AccessRecordPartitions info methods -callprotection all] {
        ::nsf::method::property [self] $method call-protected true
    }

    :property settings:required
    :property {mode writer}
    :property {observer {}}
    :variable db {}
    :variable handle {}
    :variable path
    :variable prior {}
    :variable batchSchema {}
    :variable partitionInterval 0
    :variable partitionDDL {}
    :variable recordTable main.rows
    :variable tupleTimes {}
    :variable sqliteSynchronous FULL
    :variable sqliteSynchronousValue 2
    :variable sqliteWalAutoCheckpointPages 1000
    :variable sqliteBaseline {}
    :variable sqlitePreCommit {}
    :variable sqliteCommitTicks 0
    :variable retentionCacheKiB 32768
    :variable querying false
    :variable fullScope false
    :variable peers {}
    :variable lastStamp {}
    :variable lastCompleted 0

    :method init {} {
        #
        # Ordered column declarations describe the prepared request rows.
        # Build the core batch schema once, before any ingestion can start.
        #
        set :partitionInterval [:dict getdef ${:settings} partitionInterval 0]
        if {![string is integer -strict ${:partitionInterval}] || ${:partitionInterval}<0} {
            error "invalid partition interval"
        }
        if {${:partitionInterval}>0 && (${:partitionInterval}<3600
            || ${:partitionInterval}%60!=0
            || int(ceil([dict get ${:settings} retention]/double(${:partitionInterval})))+2>256)} {
            error "partition interval must be minute-aligned, at least one hour, and retain at most 256 tables"
        }
        set columns {}
        foreach {name definition} {
            fid        {type number refinement int64 representation integer_decimal}
            off        {type number refinement int64 representation integer_decimal}
            stop       {type number refinement int64 representation integer_decimal}
            kind       {type number refinement int64 representation integer_decimal}
            t          {type number refinement float64 representation decimal}
            status     {type number refinement int64 representation integer_decimal}
            path       {type string}
            method     {type string}
            pool       {type string}
            source     {type string}
            driver     {type string}
            auth       {type string}
            bytes      {type string refinement int64_or_empty_text representation integer_decimal}
            phase      {type number refinement int64 representation integer_decimal}
            fallback   {type number refinement int64 representation integer_decimal}
            a          {type number refinement float64 representation decimal}
            q          {type number refinement float64 representation decimal}
            f          {type number refinement float64 representation decimal}
            r          {type number refinement float64 representation decimal}
            peer       {type string refinement hex_or_empty_text representation hex}
            agent      {type string refinement hex_or_empty_text representation hex}
            label      {type string}
            clipped    {type number refinement int64 representation integer_decimal}
            redacted   {type number refinement int64 representation integer_decimal}
            identifier {type string}
            score      {type string refinement encoded_binary representation hex}
            completed  {type number refinement float64 representation decimal}
            rolled     {type number refinement int64 representation integer_decimal}
        } {
            lappend columns [dict merge [dict create name $name] $definition]
        }
        set :batchSchema [dict create id nsmcp-cooked-access version 2 columns $columns]
        if {[dict exists ${:settings} retentionCacheBytes]} {
            set bytes [dict get ${:settings} retentionCacheBytes]
            if {![string is wideinteger -strict $bytes] || $bytes < 1048576 || $bytes > 67108864} {
                error "SQLite retentionCacheBytes must be 1MiB through 64MiB"
            }
            set :retentionCacheKiB [expr {($bytes+1023)/1024}]
        }
        try {
            set pool [dict get ${:settings} pool]
            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 :path [file normalize $datasource]
            set directory [file dirname ${:path}]
            if {${:mode} ni {reader writer}} {
                error "invalid access store mode"
            }
            if {${:mode} eq "writer" && ![file exists $directory]} {
                file mkdir $directory
                file attributes $directory -permissions 0700
            }
            # 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 pool-checkout
            set :handle [ns_db gethandle -timeout 100ms [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}]
            if {[file normalize [ns_db datasource ${:handle}]] ne ${:path}} {
                error "record-store-wrong-datasource"
            }
            set :prior [dict create queryOnly [${:db} scalar {PRAGMA query_only}]]
            foreach row [${:db} execute {
                SELECT name FROM main.sqlite_schema
                 WHERE type='table' AND name NOT LIKE 'sqlite_%'
                   AND NOT EXISTS(SELECT 1 FROM main.sqlite_schema WHERE type='table' AND name='meta')
            }] {
                set name [dict get $row name]
                if {$name ni {meta files scan_counts ranges rows state peer_addresses minutes evictions partitions row_template}
                    && ![regexp {^p_[0-9]+$} $name]} {
                    error "accessRecordPool must contain only access-record tables"
                }
            }
            set :sqliteSynchronousValue [${:db} scalar {PRAGMA main.synchronous}]
            set :sqliteSynchronous [lindex {OFF NORMAL FULL EXTRA} ${:sqliteSynchronousValue}]
            set :sqliteWalAutoCheckpointPages [${:db} scalar {PRAGMA wal_autocheckpoint}]

            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 getdef ${:settings} unified false]}]  schema4]]
                if {[file size ${:path}]<512 || ![nsv_exists nsmcp:record-initialized $key]} {
                    :accessIngestionObserve initialize-schema
                    :initialize
                    nsv_set nsmcp:record-initialized $key 1
                }
                # The page budget belongs to this store, not to the pool policy.
                :accessIngestionObserve page-limit
                :pageLimit
            } else {
                if {[${:db} scalar {SELECT version FROM main.meta}] ni {2 3 4}} {
                    error "record-store-schema-not-initialized"
                }
            }
            :partitionOpen
            if {${:mode} eq "writer"} {
                # Only migration backfill uses cooked staging. Both ingestion
                # backends insert directly into the canonical table.
                ${:db} execute {
                    DROP TABLE IF EXISTS temp.nsmcp_ingest_batch
                }
            } 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 record-store-partition-reset-required}} {
                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 {} {
        if {[${:db} scalar {SELECT EXISTS(SELECT 1 FROM main.sqlite_schema WHERE name='state' AND type='table')}] 
            && [:stateGet partitionInterval 0]>0} {
            if {[:stateGet partitionInterval 0]!=${:partitionInterval}} {
                error "record-store-partition-reset-required"
            }
            return
        }
        # Capacity is an application budget; connection policies belong to the pool.
        :pageLimit
        ${:db} execute {
            CREATE TABLE IF NOT EXISTS main.meta(version INTEGER NOT NULL);
            INSERT INTO main.meta SELECT 1 WHERE NOT EXISTS(SELECT 1 FROM main.meta);
            CREATE TABLE IF NOT EXISTS main.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 main.scan_counts(
                                                                 fid INTEGER NOT NULL,
                                                                 lo INTEGER NOT NULL,
                                                                 hi INTEGER NOT NULL,
                                                                 malformed INTEGER NOT NULL,
                                                                 oversized INTEGER NOT NULL,
                                                                 skipped INTEGER NOT NULL,
                                                                 completed INTEGER NOT NULL,
                                                                 PRIMARY KEY(fid,lo)
                                                                 );
            CREATE INDEX IF NOT EXISTS main.scan_counts_expiry ON scan_counts(completed);
            CREATE TABLE IF NOT EXISTS main.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 main.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 main.by_time ON rows(t);
            CREATE INDEX IF NOT EXISTS main.by_time_status ON rows(t,status,fid,off,kind)
        }
        set version [${:db} scalar {SELECT version FROM main.meta}]
        if {$version ni {1 2 3}} {
            error "record-store-schema-version"
        }
        set columns {}
        foreach row [${:db} execute {SELECT name FROM pragma_table_info('rows','main')}] {
            lappend columns [dict get $row name]
        }
        if {"completed" ni $columns} {
            ${:db} execute  {ALTER TABLE main.rows ADD COLUMN completed REAL NOT NULL DEFAULT 0}
        }
        if {"rolled" ni $columns} {
            ${:db} execute  {ALTER TABLE main.rows ADD COLUMN rolled INTEGER NOT NULL DEFAULT 0}
        }
        set columns {}
        foreach row [${:db} execute {SELECT name FROM pragma_table_info('files','main')}] {
            lappend columns [dict get $row name]
        }
        if {"active" ni $columns} {
            ${:db} execute  {ALTER TABLE main.files ADD COLUMN active INTEGER NOT NULL DEFAULT 1}
        }
        foreach column {dropping} {
            if {$column ni $columns} {
                ${:db} execute  "ALTER TABLE main.files ADD COLUMN $column INTEGER NOT NULL DEFAULT 0"
            }
        }
        ${:db} execute {
            CREATE TABLE IF NOT EXISTS main.state(key TEXT PRIMARY KEY,value TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS main.peer_addresses(
                                                                    k BLOB PRIMARY KEY,
                                                                    address TEXT NOT NULL,
                                                                    seen REAL NOT NULL
                                                                    );
            CREATE TABLE IF NOT EXISTS main.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 main.minute_time ON minutes(slot,status);
            CREATE INDEX IF NOT EXISTS main.peer_expiry ON peer_addresses(seen);
            CREATE INDEX IF NOT EXISTS main.diagnostic_expiry ON rows(completed) WHERE kind<>0;
            CREATE INDEX IF NOT EXISTS main.correlation ON rows(identifier,t);
            CREATE TABLE IF NOT EXISTS main.evictions(
                                                               id INTEGER PRIMARY KEY,
                                                               dropped_at INTEGER,
                                                               first_time REAL,
                                                               last_time REAL,
                                                               rows INTEGER,
                                                               reason TEXT
                                                               );
            UPDATE main.meta SET version=3
        }
    }

    :method pageLimit {} {
        set pages  [:accessRecordMainPages ${:settings} [${:db} scalar {PRAGMA main.page_size}]]
        set capacity [:capacity]
        if {[dict get $capacity limitExceeded]} {
            error "record-store-disk-limit"
        }
        ${:db} execute "PRAGMA main.max_page_count=$pages"
    }

    :method retentionFloor {} {
        set floor [:stateGet retentionFloor 0]
        if {[:dict getdef ${:settings} unified false]} {
            # 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 partitionInfo {} {
        #
        # Report partition layout and the bootstrap/import coverage boundary.
        #
        # Read bounded catalog metadata without scanning request tables. The
        # available-from timestamp advances only after a backfill pass completes.
        #
        # @return Partition width/count, bootstrap boundary and backfill policy.
        #

        return [:partitionStatus]
    }

    :public method capacity {} {
        #
        # Report the configured record-store budget and current allocation.
        #
        # Account for the main database and its journal files separately. Writer instances
        # also publish a timestamped capacity snapshot for the status interface.
        #
        # @return A dictionary containing byte budgets, allocation, disk usage and the
        #         limit-exceeded flag.
        #

        set pageSize [${:db} scalar {PRAGMA main.page_size}]
        set allocated [expr {wide([${:db} scalar {PRAGMA main.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} {
        #
        # Classify and log a record-store failure.
        #
        # Use the shared aggregation diagnostic path so callers receive a bounded
        # classification and diagnostic identifier rather than exposing the underlying error
        # details.
        #
        # @param message Original error message.
        # @param opts Tcl return-options dictionary for the failure.
        #
        # @return The shared diagnostic dictionary for the record-store error.
        #

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

    :public method available {} {
        #
        # Report whether this adapter owns a database session.
        #
        # This checks the current object state; it does not obtain another pool handle or
        # perform a database health query.
        #
        # @return True when a session is available, otherwise false.
        #

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

    :public method diskBytes {} {
        #
        # Measure the disk space occupied by the record-store files.
        #
        # Include the database, rollback journal, WAL and shared-memory file when present.
        # SQLite temporary files and unrelated files are excluded.
        #
        # @return The total size in bytes of the existing store files.
        #

        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} {
        #
        # Resolve or initialize the stored generation for a discovered log file.
        #
        # Validate the source prefix and committed-range anchors before reusing a generation.
        # Writers retire or replace changed generations; readers only accept an existing
        # active generation.
        #
        # @param file Discovered source descriptor, including path, identity, size and
        #        modification time.
        #
        # @return The stored file identifier, or -1 when a reader cannot use the generation.
        #

        file stat [dict get $file path] actual
        set identity [dict get $file identity]
        set policy [:dict getdef ${:settings} policy {}]
        set key [ns_crypto::md string -digest sha256 -- [list schema1 $identity $policy]]
        set fid {}
        set old [${:db} execute {SELECT * FROM main.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 main.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 getdef ${:settings} unified false] && $fid ne {}} {
                        set retired "${key}:retired:$fid"
                        set oldIdentity "${identity}:retired:$fid"
                        ${:db} execute {
                            UPDATE main.files SET k=:retired,identity=:oldIdentity WHERE fid=:fid
                        }
                    } else {
                        if {$fid ne {}} {
                            ${:db} execute {DELETE FROM main.rows WHERE fid=:fid}
                        }
                        ${:db} execute {
                            DELETE FROM main.ranges WHERE fid=:fid;
                            DELETE FROM main.files WHERE k=:key
                        }
                    }
                    ${:db} execute {
                        INSERT INTO main.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 main.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 main.files SET touched=:now WHERE fid=:fid AND touched<:touchBefore}
        ${:db} execute {
            UPDATE main.files SET active=0 WHERE identity=:identity AND fid<>:fid AND active=1;
            UPDATE main.files SET active=1 WHERE fid=:fid
        }
        return $fid
    }

    :public method coveredEnd {fid offset end} {
        #
        # Find the end of the committed range containing an offset.
        #
        # Use persisted byte coverage rather than reading the source file. Clamp the result
        # to the caller's planned scan boundary.
        #
        # @param fid Stored source-file identifier.
        # @param offset Starting byte offset to test.
        # @param end Upper byte boundary for this scan.
        #
        # @return The covered end offset, or offset itself when no range covers it.
        #

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

    :public method dropState {fid offset} {
        #
        # Check whether an offset follows a committed dropped section.
        #
        # Consult the source generation's persisted dropping flag and high watermark. This
        # allows the scanner to carry its framing state across ingestion slices.
        #
        # @param fid Stored source-file identifier.
        # @param offset Byte offset at which scanning will resume.
        #
        # @return True when the generation is dropping at or before this offset, otherwise
        #         false.
        #

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

    :public method gapEnd {fid offset end} {
        #
        # Find the next committed range after an uncovered offset.
        #
        # Bound scanning to the next stored range or the planned source end, whichever comes
        # first.
        #
        # @param fid Stored source-file identifier.
        # @param offset Uncovered byte offset.
        # @param end Upper byte boundary for this scan.
        #
        # @return The end offset of the uncovered interval.
        #

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

    :public method batchInsertionAvailable {} {
        #
        # Report use of the prepared-row batch insertion API.
        #
        # Current deployments require this API. Tests may override this method to select the
        # SQL reference implementation for comparison.
        #
        # @return True for the required production insertion path.
        #

        return true
    }

    :method sqliteSnapshot {} {
        # Read the connection owned by this writer, never a second pool handle.
        # Statistics are diagnostic: failure must not fail a committed import.
        set started [clock clicks -microseconds]
        if {[catch {
            set :sqliteWalAutoCheckpointPages [${:db} scalar {PRAGMA wal_autocheckpoint}]
            set stats [ns_db stats ${:handle} -database main]
            if {![dict get $stats supported]} {
                error "driver statistics unsupported"
            }
            set result [dict create available true]
            if {[dict exists $stats backend sqlite wal]} {
                dict set result wal [dict get $stats backend sqlite wal]
            }
            # Keep the ingestion delta labels stable while consuming shared
            # scalar metrics and explicitly scoped SQLite page/byte counters.
            foreach {key path} {
                cacheHits            {connection cacheHits}
                cacheMisses          {connection cacheMisses}
                cacheWrites          {backend sqlite connection cacheWrittenPages}
                cacheSpills          {backend sqlite connection cacheSpilledPages}
                tempBufferSpillBytes {backend sqlite connection temporarySpillBytes}
                cacheUsedBytes       {connection cacheUsedBytes}
            } {
                if {[dict exists $stats {*}$path]} {
                    dict set result $key [dict get $stats {*}$path]
                }
            }
            set result
        } result]} {
            set result {available false reason statistics-read-failed}
        }
        dict set result synchronous ${:sqliteSynchronous}
        dict set result synchronousValue ${:sqliteSynchronousValue}
        dict set result walAutoCheckpointPages ${:sqliteWalAutoCheckpointPages}
        dict set result samplingMicroseconds [expr {max(0,[clock clicks -microseconds]-$started)}]
        return $result
    }

    :method sqliteBeforeCommit {} {
        :accessIngestionObserve sqlite-statistics
        set :sqlitePreCommit [:sqliteSnapshot]
        :accessIngestionObserve -batch [dict create sqlite [dict create  beforeTransaction ${:sqliteBaseline} beforeCommit ${:sqlitePreCommit}]] commit
        set :sqliteCommitTicks [clock clicks -microseconds]
    }

    :method sqliteMeasurement {after finished} {
        # Counter deltas stay within one physical connection and transaction.
        # WAL snapshots are retained separately: frame counts can rewind on reuse.
        set before ${:sqliteBaseline}
        set pre ${:sqlitePreCommit}
        set result [dict create available false  synchronous ${:sqliteSynchronous} synchronousValue ${:sqliteSynchronousValue}  walAutoCheckpointPages ${:sqliteWalAutoCheckpointPages}  commitMicroseconds [expr {max(0,$finished-${:sqliteCommitTicks})}]  samplingMicroseconds [expr {[dict get $before samplingMicroseconds]  +[dict get $pre samplingMicroseconds]+[dict get $after samplingMicroseconds]}]  beforeTransaction $before beforeCommit $pre afterCommit $after]
        foreach snapshot [list $before $pre $after] {
            if {![dict get $snapshot available]} {
                dict set result reason statistics-read-failed
                return $result
            }
        }
        foreach key {cacheHits cacheMisses cacheWrites cacheSpills tempBufferSpillBytes} {
            foreach snapshot [list $before $pre $after] {
                if {![dict exists $snapshot $key]} {
                    dict set result reason counter-unavailable
                    return $result
                }
            }
            set delta [expr {[dict get $after $key]-[dict get $before $key]}]
            set commitDelta [expr {[dict get $after $key]-[dict get $pre $key]}]
            if {$delta<0 || $commitDelta<0} {
                dict set result reason counter-reset
                return $result
            }
            dict set result delta $key $delta
            dict set result commitDelta $key $commitDelta
        }
        dict set result available true
        return $result
    }

    :public method ingest {fid checkpoint tuples {batched false}} {
        #
        # Persist the scanner's supplied requests and checkpoint atomically.
        #
        # Insert every supplied request; duplicate source positions are errors. Commit
        # rollups, pending peer addresses, parsing-loss counters and byte coverage in the
        # same transaction. This method does not read the source file.
        #
        # @param fid Stored source-file identifier.
        # @param checkpoint Scanner-prepared coverage, framing, timestamp and parsing-loss
        #        metadata.
        # @param tuples Prepared request rows, or serialized SQL tuples for the reference
        #        path.
        # @param batched True for the batch API; false for the SQL reference implementation.
        #
        # @return Committed request and diagnostic counts, together with the
        #         insertion-backend label and writer-side SQLite measurements.
        #

        if {$batched && [llength $tuples]>512} {
            error "record-store-batch-limit"
        }
        :accessIngestionObserve sqlite-statistics
        set :sqliteBaseline [:sqliteSnapshot]
        if {${:partitionInterval}>0} {
            set committed [:partitionIngest $fid $checkpoint $tuples $batched]
        } else {
            set committed [:ingestCanonical $fid $checkpoint $tuples $batched]
        }
        set finished [clock clicks -microseconds]
        :accessIngestionObserve sqlite-statistics
        set after [:sqliteSnapshot]
        dict set committed sqlite [:sqliteMeasurement $after $finished]
        return $committed
    }

    :method ingestCanonical {fid checkpoint tuples batched {nested false}} {
        foreach key {begin end lo hi anchor stamp now dropping} {
            set $key [dict get $checkpoint $key]
        }
        set malformed [:dict getdef $checkpoint malformed 0]
        set oversized [:dict getdef $checkpoint oversized 0]
        set skipped [:dict getdef $checkpoint skipped 0]
        foreach count [list $malformed $oversized $skipped] {
            if {![string is entier -strict $count] || $count<0} {
                error "invalid checkpoint counter"
            }
        }
        set committed [dict create  records 0  requestRecords 0  malformedRecords $malformed  oversizedRecords $oversized]
        if {$batched && [llength $tuples] > 512} {
            error "record-store-batch-limit"
        }
        if {!$nested} {
            :accessIngestionObserve begin-transaction
        }
        set work {
            if {![${:db} scalar {SELECT count(*) FROM main.files WHERE fid=:fid}]} {
                error "record-store-generation-changed"
            }
            if {[llength $tuples]} {
                set previousRowid  [${:db} scalar "SELECT coalesce(max(rowid),0) FROM ${:recordTable}"]
                if {$previousRowid == 9223372036854775807} {
                    error "record-store-rowid-limit"
                }
                :accessIngestionObserve insert-records
                if {$batched} {
                    ${:db} batchInsert -schema ${:batchSchema} [split ${:recordTable} .] $tuples
                } else {
                    ${:db} executeDml "INSERT INTO ${:recordTable} VALUES [join $tuples ,]"
                }
                # The writer lock makes this watermark identify precisely the
                # inserted batch, without staging or filtering its records.
                set batchSource "(SELECT * FROM ${:recordTable} WHERE rowid>$previousRowid)"
                :accessIngestionObserve count-records
                set committed [dict merge $committed [lindex [${:db} execute "
                    SELECT count(*) AS records,
                    coalesce(sum(kind=0),0) AS requestRecords FROM $batchSource
                "] 0]]
                :accessIngestionObserve rollup
                :rollBatch $batchSource
                :accessIngestionObserve observe-records
                :observeBatch $batchSource
            }
            :accessIngestionObserve store-peers
            if {[dict size ${:peers}]} {
                set values {}
                dict for {key data} ${:peers} {
                    lassign $data address seen
                    lappend values [list $key $address $seen]
                }
                ${:db} executeDml  -bindtypes {values {rows {binary scalar scalar}}} {
                        INSERT INTO main.peer_addresses VALUES :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 checkpoint-counters
            # Parsing losses are committed with the same source interval. A
            # bad-only batch advances coverage without inserting a request row.
            if {$malformed+$oversized+$skipped>0} {
                set observed  [:stateGet observations {first 0 latest 0 completed 0 malformed 0 oversized 0}]
                dict incr observed malformed $malformed
                dict incr observed oversized $oversized
                :statePut observations $observed
                ${:db} execute {
                    INSERT INTO main.scan_counts VALUES(:fid,:begin,:end,:malformed,:oversized,:skipped,:now)
                }
            }
            dict incr committed records [expr {$malformed + $oversized + $skipped}]
            :accessIngestionObserve save-ranges
            ${:db} execute {
                DELETE FROM main.ranges WHERE fid=:fid AND lo<=:end AND hi>=:begin;
                INSERT INTO main.ranges VALUES(:fid,:lo,:hi,:anchor);
                UPDATE main.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 check-storage
            if {[:diskBytes] > [dict get ${:settings} maxBytes]} {
                error "record-store-disk-limit"
            }
            if {!$nested} {
                :sqliteBeforeCommit
            }
        }
        if {$nested} {
            eval $work
        } else {
            ${:db} transaction $work
        }
        set :peers {}
        dict set committed insertionBackend [expr {$batched ? "batch-api" : "sql-values"}]
        return $committed
    }

    # Raw scanner 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 cRead {
                          fid
                          file
                          offset
                          stop
                          budget
                          limits
                          redactor
                          ingestOnly
                          {batched false}
                      } {
        #
        # Read a bounded source slice through the C access-log scanner.
        #
        # Use a connection-local temporary table to materialize raw scanner fields, then cook
        # and redact valid requests. Raw fields are never written to the canonical store.
        # Leave an incomplete tail uncommitted.
        #
        # @param fid Stored source-file identifier.
        # @param file Discovered source descriptor.
        # @param offset Starting source byte offset.
        # @param stop Planned exclusive source boundary.
        # @param budget Maximum source bytes to scan in this invocation.
        # @param limits Scanner limits, including the maximum line size.
        # @param redactor Redaction object used when cooking requests.
        # @param ingestOnly True for ingestion; false allows time-dependent records to
        #        request the Tcl scanner fallback.
        # @param batched True to construct prepared rows; false to construct SQL reference
        #        tuples.
        #
        # @return A list of end offset, cooked tuples, pending flag, fallback flag and
        #         parsing-loss counts.
        #

        :accessIngestionObserve batch-capabilities
        if {![llength [info commands ns_accesslog]]} {
            error "C scanner ingestion requires nslog's C scan API"
        }
        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 prepare-raw-stage
        ${: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 open-c-scanner
        set scanner [ns_accesslog scan open {*}$scanOptions]
        set begin $offset
        set tuples {}
        set pending false
        set counts {malformed 0 oversized 0 skipped 0}
        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 scan-c
            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 stage-raw-batch
            ${:db} transaction {
                ${:db} batchInsert -mapping $mapping {temp nsmcp_scan_batch} $batch
            }
            :accessIngestionObserve cook-records
            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 {
                    # Consume decoded C fields and typed timing values directly.
                    set timing [::nsmcp::accessRecordParser timingMicroseconds [dict get $item stamp]  [dict get $item start] [dict get $item a] [dict get $item q]  [dict get $item f] [dict get $item r]]
                    set bytes [dict get $item bytes]
                    if {$bytes eq {}} {
                        set bytes 0
                    }
                    if {$timing eq {}} {
                        set row {}
                    } else {
                        set row [::nsmcp::accessRecordParser requestFields [dict get $item peer]  [dict get $item identifier] [dict get $item usr] [dict get $item stamp]  [dict get $item request] [dict get $item status] $bytes  [dict get $item combined] [dict get $item agent] $timing]
                    }
                    if {$row ne {} && [dict exists $row timeDependent] && !$ingestOnly} {
                        return [list $begin {} false true {malformed 0 oversized 0 skipped 0}]
                    }
                    if {$row eq {} || [dict exists $row timeDependent]} {
                        set row {}
                        set kind 1
                    }
                }
                if {$kind==0} {
                    lappend tuples [:tuple $fid $offset $next 0 $row $redactor $file $batched]
                } elseif {$kind==2} {
                    dict incr counts oversized
                } else {
                    dict incr counts malformed
                }
                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 false $counts]
    }

    :public method tuple {fid off stop kind row redactor file {batched false}} {
        #
        # Construct one policy-processed record from a valid parsed request.
        #
        # Redact display fields and encode the canonical row representation. Cache
        # completion-time parsing and retain eligible peer-address observations for the
        # enclosing ingestion transaction; do not insert a row here.
        #
        # @param fid Stored source-file identifier.
        # @param off Starting source byte offset.
        # @param stop Exclusive byte offset after the record.
        # @param kind Record kind; only valid requests (zero) are accepted.
        # @param row Parsed request dictionary.
        # @param redactor Redaction object for display fields.
        # @param file Source descriptor used to derive the stable sample identifier.
        # @param batched True to return a prepared row; false to return a quoted SQL tuple.
        #
        # @return The cooked request as a Tcl row list or serialized SQL tuple.
        #

        if {$kind!=0} {
            error "only parsed requests have tuples"
        }
        set values [list $fid $off $stop $kind]
        set peer [dict get $row peerAddress]
        set agent [:dict getdef $row userAgent {}]
        set peerBinary [expr  {$peer in {{} - {[REDACTED]}} ? {} : [ns_crypto::md string -digest sha256 -encoding binary -- $peer]}]
        set peer [binary encode hex $peerBinary]
        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 [:dict getdef $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 getdef ${:settings} unified false]} {
                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 $peerBinary [list $address $time]
                }
            }
        }
        lappend values $completed 1
        if {$batched} {
            return $values
        }
        # Hash columns retain BLOB storage; absent peer/agent hashes retain
        # empty TEXT. Quote each contiguous scalar group in one core call.
        set sql [ns_dbquotelist [lrange $values 0 18]]
        foreach index {19 20} {
            set hash [lindex $values $index]
            if {$hash eq {}} {
                append sql ",''"
            } else {
                append sql ",X[ns_dbquotevalue $hash]"
            }
        }
        append sql , [ns_dbquotelist [lrange $values 21 24]]
        append sql ,X [ns_dbquotevalue [lindex $values 25]]
        append sql , [ns_dbquotelist [lrange $values 26 27]]
        set result "($sql)"
        if {${:partitionInterval}>0} {
            dict set :tupleTimes $result [dict get $row startTime]
        }
        return $result
    }

    :public method scope {files} {
        #
        # Open a stored-record query scope for selected source intervals.
        #
        # Create connection-local range metadata and a bounded request view inside a read
        # transaction. Reject a generation that disappeared after discovery. Call finish to
        # end the scope.
        #
        # @param files Source descriptors containing archive file identifiers and byte
        #        bounds.
        #
        # @return An empty result after the scoped transaction and view are ready.
        #

        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 main.files WHERE fid=:fid}]} {
                    error "record-store-generation-changed"
                }
            }
        }
        ${:db} execute {
            CREATE TEMP VIEW nsmcp_record_window AS SELECT r.* FROM main.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 getdef ${:settings} unified false]} {
            # 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>=[ns_dbquotevalue $from] AND t<[ns_dbquotevalue $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=[ns_dbquotevalue $value]"
            }
        }
        set literal [dict get $options pathLiteral]
        if {$literal ne {}} {
            append context " AND instr(path,[ns_dbquotevalue $literal])>0"
        }
        set selected $context
        set status [dict get $options status]
        set class [dict get $options statusClass]
        if {$status ne {}} {
            append selected " AND status=[ns_dbquotevalue $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 main.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}
        }
    }

    :method checkpointCounts {} {
        set result  {parseErrors 0 oversizedRecords 0 skippedRecords 0 checkpointRecords 0 diagnosticCountsComplete true}
        if {[${:db} scalar  {SELECT count(*) FROM main.sqlite_master WHERE type='table' AND name='scan_counts'}]} {
            set row [lindex [${:db} execute {
                SELECT
                coalesce(sum(CASE WHEN c.lo>=s.lo AND c.hi<=s.hi THEN malformed ELSE 0 END),0) AS parseErrors,
                coalesce(sum(CASE WHEN c.lo>=s.lo AND c.hi<=s.hi THEN oversized ELSE 0 END),0) AS oversizedRecords,
                coalesce(sum(CASE WHEN c.lo>=s.lo AND c.hi<=s.hi THEN skipped ELSE 0 END),0) AS skippedRecords,
                coalesce(sum(CASE WHEN c.lo<s.lo OR c.hi>s.hi THEN 1 ELSE 0 END),0) AS partial
                FROM main.scan_counts c JOIN temp.nsmcp_record_scope s
                ON c.fid=s.fid AND c.lo<s.hi AND c.hi>s.lo
            }] 0]
            foreach key {parseErrors oversizedRecords skippedRecords} {
                dict set result $key [dict get $row $key]
            }
            dict set result checkpointRecords [expr  {[dict get $row parseErrors]+[dict get $row oversizedRecords]+[dict get $row skippedRecords]}]
            dict set result diagnosticCountsComplete [expr {[dict get $row partial]==0}]
        }
        return $result
    }

    :public method stats {options redactor limits} {
        #
        # Aggregate request statistics for the current stored-record scope.
        #
        # Combine detailed request rows with eligible minute rollups. Bound ranked
        # dimensions, redact returned labels and report omitted or unavailable observations
        # explicitly.
        #
        # @param options Normalized time window, filters, bucket size and result limits.
        # @param redactor Redaction object for returned dimension labels.
        # @param limits Aggregation limits, including the maximum number of retained keys.
        #
        # @return A list containing request statistics, aggregation metadata and the selected
        #         SQL predicate.
        #

        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]
        set counters [:checkpointCounts]
        foreach key {parseErrors oversizedRecords} {
            dict incr stats $key [dict get $counters $key]
        }
        dict incr stats examinedRecords [dict get $counters checkpointRecords]
        dict set stats diagnosticCountsComplete [dict get $counters diagnosticCountsComplete]
        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-[ns_dbquotevalue $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} {
        #
        # Count committed bytes overlapping a source interval.
        #
        # Intersect the interval with persisted coverage ranges. This is stored coverage, not
        # a measurement of current source-file contents.
        #
        # @param fid Stored source-file identifier.
        # @param lo Inclusive lower byte boundary.
        # @param hi Exclusive upper byte boundary.
        #
        # @return The number of covered bytes in the interval.
        #

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

    :method rollupSource {options} {
        if {
            !${:fullScope} || ![:dict getdef ${:settings} unified false] ||
            ![: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 {}
        set timeIndex [expr {${:partitionInterval}>0 ? {} : {INDEXED BY by_time_status}}]
        if {$lo<$hi} {
            lappend parts "
                SELECT 0 AS kind, slot AS t, status, pool, method, source, n FROM main.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 main.rows r $timeIndex
                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>=[ns_dbquotevalue $start] AND t<[ns_dbquotevalue $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} {
        #
        # Return stored request totals for the current query scope.
        #
        # Use minute rollups when the scope and filters permit, retaining detailed boundary
        # rows. Otherwise aggregate the scoped records and checkpoint diagnostic counters
        # directly.
        #
        # @param options Normalized time window and request filters.
        #
        # @return A list containing the counts dictionary, selected status counts and
        #         coverage diagnostics.
        #

        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]
        set counters [:checkpointCounts]
        foreach key {parseErrors oversizedRecords} {
            dict incr diagnostics $key [dict get $counters $key]
        }
        dict set diagnostics diagnosticCountsComplete [dict get $counters diagnosticCountsComplete]
        dict set diagnostics queryStrategy detailed-rows
        dict set diagnostics pendingRecords [:boundaryPending]
        return [list $counts $statuses $diagnostics]
    }

    :public method samples {selected files redactor} {
        #
        # Build deterministic samples from selected and successful requests.
        #
        # Restrict output to known source descriptors and apply the supplied redactor.
        # Preserve truncation metadata when a stored user-agent label was clipped.
        #
        # @param selected Trusted SQL predicate produced by the query-filter builder.
        # @param files Source descriptors used to attach sample locations.
        # @param redactor Redaction object used for sample output.
        #
        # @return A list of selected-request sample entries and successful-request sample
        #         entries.
        #

        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 {} {
        #
        # Finish the current scoped read transaction.
        #
        # Commit the transaction and clear the object's query-active flag. The adapter
        # continues to own its session until it is closed.
        #
        # @return False after the query-active flag has been cleared.
        #

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

    :public method descriptors {} {
        #
        # Describe the active generations present in the record store.
        #
        # Derive live and rotated-source descriptors from persisted state without discovering
        # or opening source files.
        #
        # @return A list of active stored-source descriptors and their committed byte
        #         boundaries.
        #

        set result {}
        set i 0
        set control [:stateGet continuous {}]
        foreach row [${:db} execute  {SELECT fid,identity,high FROM main.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 {} {
        #
        # Create connection-local views for the access-analytics interface.
        #
        # Expose canonical requests, control metadata and evictions through the legacy view
        # names. Enforce request and peer retention without duplicating stored records.
        #
        # @return An empty result after the temporary views have been created.
        #

        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 main.rows r JOIN main.files f ON r.fid=f.fid AND f.active=1
            LEFT JOIN main.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 main.state;
            CREATE TEMP VIEW nsmcp_access_evictions AS SELECT * FROM main.evictions
        }
        if {${:mode} eq "reader"} {
            ${:db} execute {PRAGMA query_only=ON}
        }
    }

    :public method allScope {} {
        #
        # Open a query scope covering every active stored generation.
        #
        # Create the source-scope table and retained-record view in a read transaction. This
        # enables complete-generation rollup queries; call finish after querying.
        #
        # @return An empty result after the all-generation scope is ready.
        #

        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 main.files WHERE active=1
        }
        set floor [:retentionFloor]
        ${:db} execute "
            CREATE TEMP VIEW nsmcp_record_window AS SELECT r.* FROM main.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 {} {
        #
        # Report persisted ingestion progress and current store status.
        #
        # Combine the continuous cursor, observation times, parsing-loss totals, retention
        # floor and capacity. Missing history metadata remains pending rather than implying
        # complete coverage.
        #
        # @return An overview dictionary containing coverage, lag, storage, writer and
        #         retention status.
        #

        set report [:stateGet continuous {}]
        set observed [:stateGet observations {}]
        set floor [:retentionFloor]
        set now [clock seconds]
        foreach key {first latest completed} {
            set $key [:dict getdef $observed $key 0]
            if {!([set $key] > 0)} {
                set $key {}
            }
        }
        set size [:dict getdef $report observedSize 0]
        set cursor [:dict getdef $report cursor 0]
        set history [:dict getdef $report historyUnreadBytes 1]
        set unsupported [:dict getdef $report unsupportedCompressed 0]
        if {$unsupported} {
            incr history $unsupported
        }
        set recent [:dict getdef $report recentFrom $now]
        set updated [:dict getdef $report updatedAt 0]
        return [dict merge [:partitionStatus] [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 [:dict getdef $report pending 0]  generation 1  rotationGaps 0  lastRotationAt 0  malformed  [:dict getdef $observed malformed 0]  oversized  [:dict getdef $observed oversized 0]  lastMalformedAt {}  ingestionLagSeconds [expr {
                                               [:dict getdef $report caughtUp false] ? 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 {{batchSource temp.nsmcp_ingest_batch}} {
        #
        # Update observation metadata from an inserted or migrated batch.
        #
        # Merge first and latest request times, completion time and parsing-loss counts into
        # persistent state. The caller supplies the surrounding transaction.
        #
        # @param batchSource Trusted SQL relation containing the batch; defaults to the
        #        migration staging table.
        #
        # @return An empty result after the observation state has been saved.
        #

        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 $batchSource
        "] 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 {{batchSource temp.nsmcp_ingest_batch}} {
        ${:db} execute "
            INSERT INTO main.minutes SELECT fid, CAST(t/60 AS INTEGER)*60, status, pool, method, source,
            count(*), sum(bytes) FROM $batchSource 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 {}}} {
        #
        # Read a named control-state value.
        #
        # Return the stored Tcl representation without interpreting its contents. Use the
        # supplied default when the key is absent.
        #
        # @param key Control-state key.
        # @param default Value to return when the key does not exist.
        #
        # @return The stored state value or the supplied default.
        #

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

    :public method statePut {key value} {
        #
        # Save a named control-state value.
        #
        # Replace the existing value using scalar binding. The caller controls transaction
        # boundaries when this state accompanies record or coverage changes.
        #
        # @param key Control-state key.
        # @param value Tcl value to serialize into the state table.
        #
        # @return An empty result after the value has been saved.
        #

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

    :public method session {} {
        #
        # Return the database session owned by this adapter.
        #
        # Use the session for operations within the adapter's lifetime. The adapter retains
        # responsibility for closing it and releasing the pool handle.
        #
        # @return The DatabaseSession object, or an empty value when no session is available.
        #

        return ${:db}
    }

    :public method rollupBackfill {{batches 1}} {
        #
        # Build missing minute rollups for previously stored requests.
        #
        # Process bounded batches in source order. Commit rollups, observations, rolled flags
        # and the migration cursor atomically for each batch.
        #
        # @param batches Maximum number of migration batches to process in this invocation.
        #
        # @return True when no unrolled rows remain; false when the batch budget was
        #         exhausted.
        #

        if {${:partitionInterval}>0} {
            :statePut rollupsReady 1
            return true
        }
        lassign [:stateGet rollupCursor {0 -1}] fid off
        for {set i 0} {$i<$batches} {incr i} {
            set rows [${:db} execute {
                SELECT fid, off FROM main.rows WHERE (fid, off)>(:fid, :off) AND rolled=0 ORDER BY fid, off
                LIMIT 512
            }]
            if {$rows eq {}} {
                :statePut rollupsReady 1
                return true
            }
            ${:db} execute {
                CREATE TEMP TABLE IF NOT EXISTS nsmcp_ingest_batch AS
                SELECT * FROM main.rows WHERE 0
            }
            ${:db} transaction {
                ${:db} execute {
                    DELETE FROM temp.nsmcp_ingest_batch;
                    INSERT INTO temp.nsmcp_ingest_batch SELECT * FROM main.rows WHERE (fid, off)>(:fid, :off)
                    AND rolled=0 ORDER BY fid, off LIMIT 512
                }
                :rollBatch
                :observeBatch
                ${:db} execute {
                    UPDATE main.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}} {
        #
        # Apply the configured retention policy to a writable store.
        #
        # Unified stores remove bounded batches of expired records and related metadata.
        # They require an idle session and own their transaction. A connection-local
        # candidate table is reused for selection/accounting and exact-ID deletion;
        # it is removed after success or rollback without changing other temporary objects.
        # Other stores expire whole source generations so retained coverage cannot certify
        # deleted requests.
        #
        # @param batchRows Maximum metadata rows per deletion, from 1 through 10000.
        #        Partitioned stores also drop at most one expired request table per
        #        transaction; that table can contain more than batchRows requests.
        #
        # @return Deletion counts for unified retention, or an empty result for generation
        #         retention.
        #

        if {[:dict getdef ${:settings} unified false]} {
            if {${:partitionInterval}>0} {
                return [:partitionPrune $batchRows]
            }
            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 main.rows WHERE fid IN (SELECT fid FROM main.files WHERE touched<:cutoff);
                DELETE FROM main.ranges WHERE fid IN (SELECT fid FROM main.files WHERE
                                                               touched<:cutoff);
                DELETE FROM main.scan_counts WHERE fid IN (SELECT fid FROM main.files WHERE
                                                                    touched<:cutoff);
                DELETE FROM main.files WHERE touched<:cutoff
            }
        }
    }

    :public method maintenancePressure {} {
        #
        # Inspect expiry backlog and reusable capacity without counting records.
        #
        # Indexed existence probes bound the work independently of store size.
        # Free pages remain reusable even when the database file is at its ceiling.
        #
        # @return Expiry and capacity-pressure flags used to defer backfill.
        #

        set cutoff [:retentionFloor]
        set peerCutoff [expr {[clock seconds]-[dict get ${:settings} peerRetention]}]
        if {${:partitionInterval}>0} {
            set expired [${:db} scalar {SELECT EXISTS(SELECT 1 FROM main.partitions WHERE stop<=:cutoff)
                OR EXISTS(SELECT 1 FROM main.peer_addresses WHERE seen<:peerCutoff LIMIT 1)}]
        } else {
            set expired [${:db} scalar {
                SELECT EXISTS(SELECT 1 FROM main.rows INDEXED BY by_time_status
                              WHERE t<:cutoff AND kind=0 LIMIT 1)
                    OR EXISTS(SELECT 1 FROM main.peer_addresses INDEXED BY peer_expiry
                              WHERE seen<:peerCutoff LIMIT 1)
            }]
        }
        set capacity [:capacity]
        set pageSize [${:db} scalar {PRAGMA main.page_size}]
        set freePages [${:db} scalar {PRAGMA main.freelist_count}]
        set used [expr {[dict get $capacity mainAllocatedBytes]-$freePages*$pageSize}]
        set pressure [expr {$used>=0.9*[dict get $capacity mainMaxBytes]
                            || [dict get $capacity diskBytes]>=0.9*[dict get $capacity maxBytes]}]
        return [dict create expired $expired capacityPressure $pressure]
    }

    :method pruneAge {batchRows} {
        if {![string is integer -strict $batchRows] || $batchRows<1 || $batchRows>10000} {
            error "retention batch rows must be 1..10000"
        }
        if {[${:db} inTransaction]} {
            error "retention requires an idle database session"
        }
        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}
        # Keep bounded candidates in a connection-local temporary table. The
        # pool policy determines whether temporary storage uses memory or files.
        # Reuse this one table for requests and loss counters inside the same
        # transaction as their deletions and checkpoint updates.
        :accessIngestionObserve prepare-retention
        set priorCache [${:db} scalar {PRAGMA main.cache_size}]
        set cacheKiB [expr {$priorCache<0 ? -$priorCache
                           : $priorCache*[${:db} scalar {PRAGMA main.page_size}]/1024}]
        set growCache [expr {$cacheKiB<${:retentionCacheKiB}}]
        set candidatesCreated false
        try {
            # A deletion touches table and index pages. Keep this bounded
            # working set cached through COMMIT rather than spilling dirty
            # pages early; do not shrink a larger caller-provided cache.
            if {$growCache} {
                set size [expr {-1*${:retentionCacheKiB}}]
                ${:db} execute "PRAGMA main.cache_size=$size"
            }
            ${:db} execute {
                CREATE TEMP TABLE nsmcp_retention_candidates(id INTEGER PRIMARY KEY,t REAL,loss_count INTEGER)
            }
            set candidatesCreated true
            :accessIngestionObserve begin-transaction
            ${:db} transaction {
                :statePut retentionFloor $cutoff
                :accessIngestionObserve candidate-selection
                ${:db} execute {
                    INSERT INTO temp.nsmcp_retention_candidates(id,t)
                    SELECT rowid,t FROM main.rows INDEXED BY by_time_status
                     WHERE kind=0 AND t<:cutoff ORDER BY t LIMIT :batchRows
                }
                set selected [lindex [${:db} execute {
                    SELECT count(*) AS n,coalesce(min(t),0) AS first,coalesce(max(t),0) AS last
                      FROM temp.nsmcp_retention_candidates
                }] 0]
                set count [dict get $selected n]
                set first [dict get $selected first]
                set last [dict get $selected last]
                :accessIngestionObserve request-deletion
                if {$count>0} {
                    ${:db} execute {
                        DELETE FROM main.rows
                         WHERE rowid IN(SELECT id FROM temp.nsmcp_retention_candidates)
                    }
                }
                dict set counts requests $count
                :accessIngestionObserve diagnostic-deletion
                ${:db} execute {
                    DELETE FROM main.rows WHERE rowid IN(
                                                                  SELECT rowid FROM main.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 temp.nsmcp_retention_candidates;
                    INSERT INTO temp.nsmcp_retention_candidates(id,loss_count)
                    SELECT rowid,malformed+oversized+skipped
                      FROM main.scan_counts INDEXED BY scan_counts_expiry
                     WHERE completed<:cutoff ORDER BY completed LIMIT :batchRows
                }
                set expiring [${:db} scalar {
                    SELECT coalesce(sum(loss_count),0) FROM temp.nsmcp_retention_candidates
                }]
                ${:db} execute {
                    DELETE FROM main.scan_counts
                     WHERE rowid IN(SELECT id FROM temp.nsmcp_retention_candidates)
                }
                dict incr counts diagnostics $expiring
                :accessIngestionObserve minute-deletion
                ${:db} execute {
                    DELETE FROM main.minutes WHERE rowid IN(
                                                                     SELECT rowid FROM main.minutes INDEXED BY minute_time
                                                                     WHERE slot<:cutoff ORDER BY slot LIMIT :batchRows)
                }
                dict set counts minutes [${:db} changes]
                :accessIngestionObserve peer-deletion
                ${:db} execute {
                    DELETE FROM main.peer_addresses WHERE rowid IN(
                                                                            SELECT rowid FROM main.peer_addresses INDEXED BY peer_expiry
                                                                            WHERE seen<:peerCutoff ORDER BY seen LIMIT :batchRows)
                }
                dict set counts peers [${:db} changes]
                :accessIngestionObserve checkpoint
                if {$count>0} {
                    ${:db} execute {
                        INSERT INTO main.evictions(dropped_at, first_time, last_time, rows, reason)
                        VALUES(:now, :first, :last, :count, 'age');
                        DELETE FROM main.evictions WHERE id NOT IN(SELECT id FROM main.evictions ORDER
                                                                            BY id DESC LIMIT 64)
                    }
                }
                :statePut droppedThrough [expr {$cutoff-1}]
                :statePut lastEvictionAt $now
                # DatabaseSession commits after this body returns. The next
                # transition therefore measures the actual COMMIT call.
                :accessIngestionObserve commit
            }
            :accessIngestionObserve committed
        } finally {
            try {
                if {$candidatesCreated} {
                    ${:db} execute {DROP TABLE temp.nsmcp_retention_candidates}
                }
            } finally {
                if {$growCache} {
                    ${:db} execute "PRAGMA main.cache_size=$priorCache"
                }
            }
        }
        # Keep committed ranges as ingestion watermarks after expiry, preventing
        # the continuous backfill from resurrecting deliberately expired rows.
        return $counts
    }

    :public method close {} {
        #
        # Destroy this adapter and release its database resources.
        #
        # Roll back unfinished work and remove temporary state before returning the
        # handle to its dedicated pool. The pooled SQLite connection stays open.
        #
        # @return No result; the adapter object is destroyed.
        #

        :destroy
    }

    :method disconnect {} {
        if {${:db} ne {}} {
            # Cleanup diagnostics must never prevent rollback or handle release.
            catch {
                :accessIngestionObserve store-cleanup
            }
            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_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;
                    DROP TABLE IF EXISTS temp.nsmcp_retention_candidates
                }
            }
            if {${:prior} ne {}} {
                catch {
                    :accessIngestionObserve restore-connection
                }
                set queryOnly [dict get ${:prior} queryOnly]
                catch {
                    ${:db} execute "PRAGMA query_only=$queryOnly"
                }
            }
            catch {
                :accessIngestionObserve release-handle
            }
            try {
                ${:db} close
            } finally {
                catch {
                    :accessIngestionObserve store-closed
                }
            }
            set :db {}
            set :handle {}
            set :prior {}
        } elseif {${:handle} ne {}} {
            catch {
                :accessIngestionObserve release-handle
            }
            try {
                catch {
                    ns_db releasehandle ${:handle}
                }
            } finally {
                catch {
                    :accessIngestionObserve store-closed
                }
            }
        }
    }

    :method destroy {} {
        :disconnect
        next
    }

XQL Not present:

Generic, PostgreSQL, Oracle