PR feedback

Author: platypii

src/config.js

@@ -91,6 +91,10 @@ export function resolveOptions(argv, env) {
 }
 
 /**
+ * Parse a comma-separated signal list. Throws on any unrecognized
+ * token (including bad case like "Logs") or an empty result, so a
+ * typo fails loudly instead of silently disabling all uploads.
+ *
  * @param {string} value comma-separated signal list
  * @returns {ReadonlyArray<Signal>}
  */
@@ -99,9 +103,15 @@ function parseSignals(value) {
   const out = []
   for (const part of value.split(',')) {
     const trimmed = part.trim()
+    if (trimmed === '') continue
     if (trimmed === 'logs' || trimmed === 'traces' || trimmed === 'metrics') {
       out.push(trimmed)
+    } else {
+      throw new Error(`invalid upload signal "${trimmed}", expected one of: logs, traces, metrics`)
     }
   }
+  if (out.length === 0) {
+    throw new Error('upload signals list is empty, expected one or more of: logs, traces, metrics')
+  }
   return out
 }

src/upload/uploader.js

@@ -125,22 +125,30 @@ export async function uploadJob(job, options, connector, outputDir, committed) {
 
 /**
  * Upload every eligible job — used both by the daily timer and by
- * startup catch-up.
+ * startup catch-up. Per-job failures (transient 5xx, permanent 4xx,
+ * malformed JSONL, etc.) are logged and isolated so one bad file does
+ * not abort the whole run; the next tick will retry the failed jobs.
  *
  * @param {ResolvedUploadOptions} options
  * @param {StorageConnector} connector
  * @param {string} outputDir
  * @param {string} today YYYY-MM-DD UTC
- * @returns {Promise<Array<{ job: UploadJob, uploaded: boolean, key: string, rows: number, size: number }>>}
+ * @returns {Promise<Array<{ job: UploadJob, uploaded: boolean, key: string, rows: number, size: number, error?: Error }>>}
  */
 export async function uploadPending(options, connector, outputDir, today) {
   const committed = readLedger(outputDir)
   const jobs = discoverJobs(outputDir, today, options)
-  /** @type {Array<{ job: UploadJob, uploaded: boolean, key: string, rows: number, size: number }>} */
+  /** @type {Array<{ job: UploadJob, uploaded: boolean, key: string, rows: number, size: number, error?: Error }>} */
   const results = []
   for (const job of jobs) {
-    const result = await uploadJob(job, options, connector, outputDir, committed)
-    results.push({ job, ...result })
+    try {
+      const result = await uploadJob(job, options, connector, outputDir, committed)
+      results.push({ job, ...result })
+    } catch (err) {
+      const error = err instanceof Error ? err : new Error(String(err))
+      console.error(`[collectivus] upload failed for ${job.service}/${job.signal}/${job.date}: ${error.message}`)
+      results.push({ job, uploaded: false, key: '', rows: 0, size: 0, error })
+    }
   }
   return results
 }

test/config.test.js

@@ -69,4 +69,24 @@ describe('resolveOptions', () => {
       outputDir: '/tmp/x',
     })
   })
+
+  it('throws on unrecognized upload signals to fail loudly on typos', () => {
+    expect(() => resolveOptions([], {
+      COLLECTIVUS_UPLOAD_BUCKET: 'b',
+      COLLECTIVUS_UPLOAD_SIGNALS: 'Logs',
+    })).toThrow(/invalid upload signal "Logs"/)
+    expect(() => resolveOptions(['--upload-signals=loggs'], {})).toThrow(/invalid upload signal "loggs"/)
+  })
+
+  it('throws when upload signals list resolves to empty', () => {
+    expect(() => resolveOptions(['--upload-signals='], {})).toThrow(/upload signals list is empty/)
+    expect(() => resolveOptions(['--upload-signals', ', ,'], {})).toThrow(/upload signals list is empty/)
+  })
+
+  it('parses a valid comma-separated upload-signals list', () => {
+    const opts = resolveOptions(['--upload-signals=logs,traces'], {
+      COLLECTIVUS_UPLOAD_BUCKET: 'b',
+    })
+    expect(opts.upload?.signals).toEqual(['logs', 'traces'])
+  })
 })

test/upload/uploader.test.js

@@ -1,5 +1,5 @@
 import { parquetReadObjects } from 'hyparquet'
-import { afterEach, beforeEach, describe, expect, it } from 'vitest'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import fs from 'node:fs'
 import os from 'node:os'
 import path from 'node:path'
@@ -108,6 +108,47 @@ describe('uploadPending', () => {
     ])
   })
 
+  it('isolates per-job failures so one bad object does not abort the run', async () => {
+    writeJsonl('svc-bad', 'logs', yesterday, [
+      { serviceName: 'svc-bad', body: 'x', resource: {}, scope: { attributes: {} }, attributes: {} },
+    ])
+    writeJsonl('svc-good', 'logs', yesterday, [
+      { serviceName: 'svc-good', body: 'y', resource: {}, scope: { attributes: {} }, attributes: {} },
+    ])
+
+    const memory = memoryConnector()
+    /** @type {import('../../src/upload/upload.d.ts').StorageConnector} */
+    const connector = {
+      scheme: 'flaky',
+      async putObject(key, body, contentType) {
+        await memory.putObject(key, body, contentType)
+      },
+      headObject(key) {
+        if (key.includes('svc-bad')) return Promise.reject(new Error('s3 HEAD returned 503'))
+        return memory.headObject(key)
+      },
+    }
+
+    const errSpy = vi.spyOn(console, 'error').mockImplementation(() => {})
+    const results = await uploadPending(
+      { bucket: 'b', prefix: 'collectivus', time: '00:10', signals: ['logs', 'traces', 'metrics'], catchupDays: 7, region: 'us-east-1' },
+      connector,
+      outputDir,
+      today
+    )
+    errSpy.mockRestore()
+
+    expect(results).toHaveLength(2)
+    const bad = results.find((r) => r.job.service === 'svc-bad')
+    const good = results.find((r) => r.job.service === 'svc-good')
+    expect(bad?.uploaded).toBe(false)
+    expect(bad?.error?.message).toMatch(/503/)
+    expect(good?.uploaded).toBe(true)
+    expect([...memory.store.keys()]).toEqual([
+      `collectivus/svc-good/logs/date=${yesterday}/data.parquet`,
+    ])
+  })
+
   it('writes a ledger entry per uploaded file', async () => {
     writeJsonl('svc-a', 'logs', yesterday, [
       { serviceName: 'svc-a', body: 'a', resource: {}, scope: { attributes: {} }, attributes: {} },