Improve query cache refresh behavior (#117)

Author: philcunliffe

README.md

@@ -551,7 +551,7 @@ Freshness is treated asymmetrically (since v1.7.0):
 | Partition state | Behavior |
 | --- | --- |
 | `fresh` | Query proceeds silently. |
-| `stale` (cache exists, may be outdated) | Query proceeds; a `warning: querying stale data; …` line is written to stderr. Stdout is unchanged. |
+| `stale` (cache exists, source changed since refresh) | Query proceeds; a `warning: query cache last refreshed at …` line is written to stderr. Stdout is unchanged. |
 | `missing` (no cache table/cursor) | Query exits with the exact file-targeted `ctvs query refresh …` command to run when the source file is known. |
 
 Use `ctvs query refresh <file.jsonl>` to refresh selected source files, or

skills/collectivus-query/SKILL.md

@@ -12,7 +12,7 @@ Use `ctvs query` to inspect local Collectivus recordings. It reads local JSONL r
 1. Run `ctvs query doctor` or `ctvs query status` first to verify the recording root and cache state.
 2. If the command cannot find the intended config, discover the service config once with `ctvs status`, a LaunchAgent/systemd unit, or the user, then reuse `--config <path>` only for that setup.
 3. Cache freshness is handled asymmetrically:
-   - **Stale partitions are queried by default** and the CLI prints a `warning: querying stale data; …` line to stderr. Read stderr alongside stdout, and surface any stale warning to the user so they know the data may be outdated. Prefer the file-targeted `ctvs query refresh <file.jsonl>` command the CLI prints when updating cache data; use `--refresh always` only when the query should refresh before it runs.
+   - **Stale partitions are queried by default** and the CLI prints a `warning: query cache last refreshed at …` line to stderr. Read stderr alongside stdout, and surface the refresh timestamp to the user so they know the cache may not include newer source rows. Prefer the file-targeted `ctvs query refresh <file.jsonl>` command the CLI prints when updating cache data; use `--refresh always` only when the query should refresh before it runs.
    - **Missing partitions still error.** Run the exact `ctvs query refresh …` command the CLI prints, or rerun the target query with `--refresh always`.
    - Broad manual refreshes are explicit: `ctvs query refresh --all [dataset]`. Do not run a broad refresh when the printed file-targeted command is enough.
    - Pass `--strict-freshness` only when the user explicitly needs the pre-1.7 strict mode (e.g., scheduled checks that must never read stale data); it turns stale partitions back into a hard error.
@@ -107,7 +107,7 @@ Use `JSON_VALUE(<col>, '$.path')` to extract scalars from the `attributes` / `st
 ## Guardrails
 
 - Do not assume the cache auto-refreshes. Query commands default to `--refresh never`, and stale partitions return data with a stderr warning rather than refreshing themselves.
-- Always read stderr. A successful exit code does not mean the data is fresh — a `warning: querying stale data; …` line on stderr means stdout reflects outdated cache rows, and the user should be told before drawing conclusions.
+- Always read stderr. A successful exit code does not mean the cache is current — a `warning: query cache last refreshed at …` line on stderr means stdout reflects cache rows from that refresh, and the user should be told before drawing conclusions.
 - Do not paste `--config` into every command by habit. Use it when discovery shows the service is not using `~/.hyp/collectivus.json`.
 - Do not read arbitrary Parquet files directly for `ctvs query sql`; the CLI only allows logical tables.
 - Keep SQL read-only and use only logical datasets: `logs`, `traces`, `metrics`, `proxy_messages`, `gascity_messages`, and registered collection tables from `ctvs query catalog`.

skills/collectivus-query/references/query-cli.md

@@ -19,7 +19,7 @@ The cache is explicit. Query commands do not refresh it unless `--refresh always
 Freshness is asymmetric (since v1.7.0):
 
 - `fresh` — query proceeds silently.
-- `stale` (cache exists but may be outdated) — query proceeds and writes a `warning: querying stale data; N partition(s) outdated [...] — run '...' to update` line to stderr. Stdout is unchanged.
+- `stale` (cache exists, source changed since refresh) — query proceeds and writes a `warning: query cache last refreshed at ...; N partition(s) differ from source [...] — run '...' to refresh` line to stderr. Stdout is unchanged.
 - `missing` (no cache table/cursor) — query exits with the exact `ctvs query refresh ...` command to run.
 
 Pass `--strict-freshness` to restore the pre-1.7 behavior where stale partitions are a hard error.

src/cli/query.js

@@ -837,7 +837,7 @@ async function ensureCacheReady(paths, scope, parsed) {
     const more = stale.length > 3 ? `, +${stale.length - 3} more` : ''
     return {
       ok: true,
-      warnings: [`warning: querying stale data; ${stale.length} partition(s) outdated [${summary}${more}] — run '${refreshCommand(parsed, stale, scope)}' to update`],
+      warnings: [`warning: query cache ${refreshTimeSummary(stale)}; ${stale.length} partition(s) differ from source [${summary}${more}] — run '${refreshCommand(parsed, stale, scope)}' to refresh`],
     }
   }
 
@@ -1246,6 +1246,28 @@ function partitionLabel(state) {
   return `${String(partition.dataset)}/${String(partition.gatewayId)}/${String(partition.date)}`
 }
 
+/**
+ * @param {Array<{ meta?: unknown }>} states
+ * @returns {string}
+ */
+function refreshTimeSummary(states) {
+  const times = [...new Set(states.map(refreshedAtForState).filter((time) => time !== undefined))].sort()
+  if (times.length === 0) return 'refresh time unavailable'
+  if (times.length === 1) return `last refreshed at ${times[0]}`
+  return `last refreshed between ${times[0]} and ${times[times.length - 1]}`
+}
+
+/**
+ * @param {{ meta?: unknown }} state
+ * @returns {string | undefined}
+ */
+function refreshedAtForState(state) {
+  const meta = /** @type {{ refreshed_at?: unknown } | undefined} */ (state.meta)
+  return typeof meta?.refreshed_at === 'string' && meta.refreshed_at.length > 0
+    ? meta.refreshed_at
+    : undefined
+}
+
 /**
  * @param {number} limit
  * @returns {string}

src/query/iceberg/jsonl.js

@@ -27,13 +27,17 @@ export async function readJsonlEntryBatches(filePath, options = {}, onBatch = ()
   const {
     startByteOffset = 0,
     startLineNumber = 0,
+    endByteOffset,
     batchRows = DEFAULT_BATCH_ROWS,
     batchBytes = DEFAULT_BATCH_BYTES,
   } = options
   const stat = fs.statSync(filePath)
   if (startByteOffset > stat.size) {
     throw new Error(`source JSONL was truncated: ${filePath}`)
   }
+  const readEndOffset = endByteOffset === undefined
+    ? stat.size
+    : Math.max(startByteOffset, Math.min(endByteOffset, stat.size))
 
   /** @type {JsonlEntry[]} */
   let entries = []
@@ -59,10 +63,10 @@ export async function readJsonlEntryBatches(filePath, options = {}, onBatch = ()
     await onBatch(batch)
   }
 
-  if (startByteOffset < stat.size) {
+  if (startByteOffset < readEndOffset) {
     const stream = fs.createReadStream(filePath, {
       start: startByteOffset,
-      end: stat.size - 1,
+      end: readEndOffset - 1,
       highWaterMark: STREAM_HIGH_WATER_MARK,
     })
     for await (const chunk of stream) {

src/query/iceberg/store.js

@@ -1,6 +1,6 @@
 import fs from 'node:fs'
 import path from 'node:path'
-import { fileCatalog, icebergAppend, icebergCreateTable, icebergRead, loadLatestFileCatalogMetadata } from 'icebird'
+import { fileCatalog, icebergAppend, icebergCreateTable, icebergDataSource, icebergRead, loadLatestFileCatalogMetadata } from 'icebird'
 import { createLocalIcebergIO, tableUrlForDir } from './resolver.js'
 import { icebergSchemaForColumns, rowsToIcebergRecords } from './schema.js'
 
@@ -81,3 +81,36 @@ export async function readRowsFromCursor(cursor) {
   const rows = await icebergRead({ tableUrl, metadata, resolver })
   return /** @type {Record<string, unknown>[]} */ (rows)
 }
+
+/**
+ * @param {QueryCacheCursor} cursor
+ * @param {string[]} columns
+ * @returns {AsyncGenerator<Record<string, unknown>>}
+ */
+export async function* scanRowsFromCursor(cursor, columns) {
+  if (!queryCacheTableExists(cursor.table_path)) return
+  const { resolver, lister } = await createLocalIcebergIO()
+  const tableUrl = cursor.table_url || queryCacheTableUrl(cursor.table_path)
+  const { metadata } = await loadLatestFileCatalogMetadata({ tableUrl, resolver, lister })
+  if (metadata['current-snapshot-id'] === undefined || !metadata.snapshots?.length) return
+  const source = await icebergDataSource({ tableUrl, metadata, resolver, lister })
+  const scan = source.scan({ columns })
+  for await (const row of scan.rows()) {
+    yield await resolveAsyncRow(row, columns)
+  }
+}
+
+/**
+ * @param {import('squirreling').AsyncRow} row
+ * @param {string[]} columns
+ * @returns {Promise<Record<string, unknown>>}
+ */
+async function resolveAsyncRow(row, columns) {
+  /** @type {Record<string, unknown>} */
+  const out = row.resolved ? { ...row.resolved } : {}
+  for (const column of columns) {
+    if (Object.prototype.hasOwnProperty.call(out, column)) continue
+    out[column] = await row.cells[column]?.()
+  }
+  return out
+}

src/query/iceberg/types.d.ts

@@ -62,6 +62,7 @@ export interface JsonlReadResult {
 export interface JsonlReadOptions {
   startByteOffset?: number
   startLineNumber?: number
+  endByteOffset?: number
   batchRows?: number
   batchBytes?: number
 }