Publish v3.2.1

Author: philcunliffe

CHANGELOG.md

@@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [3.2.1] — 2026-05-20
+
+### Fixed
+
+- `ctvs gascity backfill --all` now gives large city session discovery and
+  transcript fetches up to two minutes before timing out, and reports timeout
+  failures clearly instead of the raw aborted fetch message.
+
 ## [3.2.0] — 2026-05-20
 
 ### Added

package.json

@@ -1,6 +1,6 @@
 {
   "name": "collectivus",
-  "version": "3.2.0",
+  "version": "3.2.1",
   "description": "OTLP collector and pass-through LLM proxy",
   "author": "Hyperparam",
   "homepage": "https://hyperparam.app",

src/cli/gascity.js

@@ -761,7 +761,7 @@ Options:
   --config <path>      collectivus config (default: ~/.hyp/collectivus.json)
   --help, -h           Show this help
 `
-const BACKFILL_DISCOVERY_TIMEOUT_MS = 5000
+const BACKFILL_DISCOVERY_TIMEOUT_MS = 120_000
 
 /**
  * @typedef {object} BackfillParseResult
@@ -1013,6 +1013,11 @@ async function fetchSupervisorSessions(args) {
   let response
   try {
     response = await args.fetchFn(url, { signal: ac.signal })
+  } catch (err) {
+    if (ac.signal.aborted || isAbortError(err)) {
+      throw new Error(`session discovery timed out after ${formatTimeout(BACKFILL_DISCOVERY_TIMEOUT_MS)}`)
+    }
+    throw err
   } finally {
     clearTimeout(timer)
   }
@@ -1100,6 +1105,25 @@ function pickString(obj, key) {
   return typeof value === 'string' ? value : undefined
 }
 
+/**
+ * @param {unknown} err
+ * @returns {boolean}
+ */
+function isAbortError(err) {
+  if (err === null || typeof err !== 'object') return false
+  return /** @type {{ name?: unknown }} */ (err).name === 'AbortError'
+}
+
+/**
+ * @param {number} ms
+ * @returns {string}
+ */
+function formatTimeout(ms) {
+  if (ms % 60_000 === 0) return `${ms / 60_000}m`
+  if (ms % 1000 === 0) return `${ms / 1000}s`
+  return `${ms}ms`
+}
+
 /**
  * Parse a duration like `7d`, `12h`, `30m`, `45s`. Returns the equivalent
  * milliseconds, or undefined for malformed input.

src/gascity/backfill.js

@@ -7,7 +7,7 @@ import { cursorsDir } from './paths.js'
  * @import { SessionCursor, SessionContext } from './types.d.ts'
  */
 
-const POLL_TIMEOUT_MS = 5000
+const POLL_TIMEOUT_MS = 120_000
 
 /**
  * Run a one-shot backfill against the supervisor's `/transcript` endpoint
@@ -128,6 +128,11 @@ export async function backfillSession(args) {
   let response
   try {
     response = await args.fetchFn(url, { signal: ac.signal })
+  } catch (err) {
+    if (ac.signal.aborted || isAbortError(err)) {
+      throw new Error(`transcript fetch timed out after ${formatTimeout(POLL_TIMEOUT_MS)}`)
+    }
+    throw err
   } finally {
     clearTimeout(timer)
   }
@@ -210,3 +215,22 @@ function wrapProviderFrames(frames, provider) {
 function formatError(err) {
   return err instanceof Error ? err.message : String(err)
 }
+
+/**
+ * @param {unknown} err
+ * @returns {boolean}
+ */
+function isAbortError(err) {
+  if (err === null || typeof err !== 'object') return false
+  return /** @type {{ name?: unknown }} */ (err).name === 'AbortError'
+}
+
+/**
+ * @param {number} ms
+ * @returns {string}
+ */
+function formatTimeout(ms) {
+  if (ms % 60_000 === 0) return `${ms / 60_000}m`
+  if (ms % 1000 === 0) return `${ms / 1000}s`
+  return `${ms}ms`
+}

test/cli/gascity.test.js

@@ -508,6 +508,25 @@ describe('runBackfill', () => {
     expect(stdout.value()).toMatch(/Discovered 2 sessions/)
   })
 
+  it('reports --all session discovery timeouts clearly', async () => {
+    const paths = buildPaths()
+    await writeConfig(paths.configPath, {
+      gascity: [{ name: 'hyptown', api_url: 'http://127.0.0.1:8372' }],
+    })
+    const abort = new Error('This operation was aborted')
+    abort.name = 'AbortError'
+    const fetchFn = vi.fn(async () => { throw abort })
+    const stderr = memo()
+    const code = await runBackfill(['hyptown', '--all', '--config', paths.configPath], {
+      stdout: memo(), stderr,
+      sinkRoot: paths.sinkRoot,
+      fetchFn: /** @type {typeof fetch} */ (/** @type {unknown} */ (fetchFn)),
+    })
+    expect(code).toBe(1)
+    expect(stderr.value()).toMatch(/session discovery timed out after 2m/)
+    expect(stderr.value()).not.toMatch(/This operation was aborted/)
+  })
+
   it('skips sessions whose cursor is older than --since', async () => {
     const paths = buildPaths()
     await writeConfig(paths.configPath, {

test/gascity/backfill.test.js

@@ -160,6 +160,24 @@ describe('backfillCity', () => {
     expect(stderr.value()).toMatch(/backfill_session_failed.*hy-1.*HTTP 500/)
   })
 
+  it('logs transcript timeouts as explicit timeout failures', async () => {
+    await writeCursorFile('hyptown', 'hy-1', { last_uuid: 'u-0', retired: false })
+    const stderr = memoStream()
+    const abort = new Error('This operation was aborted')
+    abort.name = 'AbortError'
+    const fetchFn = vi.fn(async () => { throw abort })
+    const result = await backfillCity({
+      city: { name: 'hyptown', api_url: 'http://h:8372' },
+      sinkRoot,
+      dispatcher: new NormalizerDispatcher({ stderr: memoStream() }),
+      stderr,
+      fetchFn,
+    })
+    expect(result).toEqual({ sessionsAttempted: 1, framesDispatched: 0, sessionsFailed: 1 })
+    expect(stderr.value()).toMatch(/backfill_session_failed.*hy-1.*transcript fetch timed out after 2m/)
+    expect(stderr.value()).not.toMatch(/This operation was aborted/)
+  })
+
   it('treats HTTP 404 as a quiet skip (session retired upstream)', async () => {
     await writeCursorFile('hyptown', 'hy-1', { last_uuid: 'u-0', retired: false })
     const stderr = memoStream()