nsmcp::AccessRecordStoreSQLite method pruneAge (protected)

 <instance of nsmcp::AccessRecordStoreSQLite> pruneAge batchRows

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

Parameters:

batchRows (required)

Testcases:

No testcase defined.

Source code:

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

XQL Not present:

Generic, PostgreSQL, Oracle