feat: persistent job progress + logs

.github/cross-shim-tests/package.json

@@ -7,7 +7,8 @@
   "scripts": {
     "producer": "tsx producer.ts",
     "worker": "tsx worker.ts",
-    "verify": "tsx verify_results.ts"
+    "verify": "tsx verify_results.ts",
+    "verify-progress": "tsx verify_progress.ts"
   },
   "devDependencies": {
     "tsx": "^4.19.0",

.github/cross-shim-tests/verify_progress.py

@@ -0,0 +1,132 @@
+"""Cross-shim progress + log verifier (Python side).
+
+Reads job IDs one-per-line from JOB_IDS_FILE, then for each id asserts:
+
+  - ``Queue.get_job(id).progress == EXPECT_PROGRESS``
+  - ``Queue.get_job_logs(id) == (EXPECT_LOGS, len(EXPECT_LOGS))``
+
+so a regression that breaks the per-job progress STRING or log Stream
+wire format on either shim surfaces here. Exits 0 on full match, else 1.
+
+Pair with a worker run that set ``EMIT_PROGRESS=1`` and
+``STORE_RESULT=1`` (the latter keeps the completed job discoverable by
+the introspector's result-key probe).
+
+Env vars:
+  QUEUE           — required, queue name.
+  JOB_IDS_FILE    — required, path written by the producer.
+  EXPECT_PROGRESS — optional, default ``75``.
+  EXPECT_LOGS     — optional, JSON-encoded list[str], default
+                     ``["step 1","step 2"]``.
+  TIMEOUT_SECS    — optional, polling deadline per id (default 10).
+  REDIS_URL       — optional.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import sys
+
+from chasquimq import Queue
+
+
+async def _wait_for_progress(
+    queue: Queue, job_id: str, expect: int, timeout_s: float
+) -> int | None:
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout_s
+    while loop.time() < deadline:
+        job = await queue.get_job(job_id)
+        if job is not None and job.progress == expect:
+            return job.progress
+        await asyncio.sleep(0.05)
+    job = await queue.get_job(job_id)
+    return None if job is None else job.progress
+
+
+async def _wait_for_logs(
+    queue: Queue, job_id: str, expect_count: int, timeout_s: float
+) -> tuple[list[str], int]:
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout_s
+    while loop.time() < deadline:
+        logs, count = await queue.get_job_logs(job_id)
+        if count >= expect_count:
+            return logs, count
+        await asyncio.sleep(0.05)
+    return await queue.get_job_logs(job_id)
+
+
+async def main() -> int:
+    queue_name = os.environ["QUEUE"]
+    ids_file = os.environ["JOB_IDS_FILE"]
+    expect_progress = int(os.environ.get("EXPECT_PROGRESS", "75"))
+    expect_logs_raw = os.environ.get(
+        "EXPECT_LOGS", '["step 1","step 2"]'
+    )
+    expect_logs = json.loads(expect_logs_raw)
+    timeout_secs = float(os.environ.get("TIMEOUT_SECS", "10"))
+    redis_url = os.environ.get("REDIS_URL", "redis://127.0.0.1:6379")
+
+    if not isinstance(expect_logs, list) or not all(
+        isinstance(s, str) for s in expect_logs
+    ):
+        print(
+            "[py-verify-progress] ERROR: EXPECT_LOGS must be a JSON list of strings",
+            file=sys.stderr,
+        )
+        return 1
+
+    with open(ids_file, "r", encoding="utf-8") as fh:
+        ids = [line.strip() for line in fh if line.strip()]
+    if not ids:
+        print(
+            f"[py-verify-progress] ERROR: {ids_file!r} contains no ids",
+            file=sys.stderr,
+        )
+        return 1
+
+    queue = Queue(queue_name, redis_url=redis_url)
+    progress_errors: list[str] = []
+    log_errors: list[str] = []
+    try:
+        for jid in ids:
+            got_progress = await _wait_for_progress(
+                queue, jid, expect_progress, timeout_secs
+            )
+            if got_progress != expect_progress:
+                progress_errors.append(
+                    f"{jid}: progress got {got_progress!r} want {expect_progress!r}"
+                )
+
+            got_logs, got_count = await _wait_for_logs(
+                queue, jid, len(expect_logs), timeout_secs
+            )
+            if got_logs != expect_logs or got_count != len(expect_logs):
+                log_errors.append(
+                    f"{jid}: logs got ({got_logs!r}, {got_count!r}) "
+                    f"want ({expect_logs!r}, {len(expect_logs)!r})"
+                )
+    finally:
+        await queue.close()
+
+    if progress_errors:
+        for e in progress_errors[:5]:
+            print(f"[py-verify-progress] ERROR: {e}", file=sys.stderr)
+        return 1
+    if log_errors:
+        for e in log_errors[:5]:
+            print(f"[py-verify-progress] ERROR: {e}", file=sys.stderr)
+        return 1
+
+    print(
+        f"[py-verify-progress] OK — {len(ids)} jobs round-tripped "
+        f"progress={expect_progress} logs={expect_logs!r}"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))

.github/cross-shim-tests/verify_progress.ts

@@ -0,0 +1,174 @@
+// Cross-shim progress + log verifier (Node side).
+//
+// Reads job IDs one-per-line from JOB_IDS_FILE, then for each id asserts:
+//
+//   - `Queue.getJob(id).progress === EXPECT_PROGRESS`
+//   - `Queue.getJobLogs(id)` deep-equals `{ logs: EXPECT_LOGS, count: EXPECT_LOGS.length }`
+//
+// so a regression that breaks the per-job progress STRING or log Stream
+// wire format on either shim surfaces here. Exits 0 on full match, else 1.
+//
+// Pair with a worker run that set `EMIT_PROGRESS=1` and `STORE_RESULT=1`
+// (the latter keeps the completed job discoverable by the introspector's
+// result-key probe).
+//
+// Env vars:
+//   QUEUE           — required, queue name.
+//   JOB_IDS_FILE    — required, path written by the producer.
+//   EXPECT_PROGRESS — optional, default '75'.
+//   EXPECT_LOGS     — optional, JSON-encoded string[], default
+//                      '["step 1","step 2"]'.
+//   TIMEOUT_SECS    — optional, polling deadline per id (default 10).
+//   REDIS_URL       — optional.
+
+import { readFileSync } from 'node:fs'
+import { Queue } from '../../chasquimq-node/dist/index.js'
+
+async function main(): Promise<number> {
+  const queueName = requireEnv('QUEUE')
+  const idsFile = requireEnv('JOB_IDS_FILE')
+  const expectProgress = Number(process.env.EXPECT_PROGRESS ?? '75')
+  const expectLogsRaw = process.env.EXPECT_LOGS ?? '["step 1","step 2"]'
+  const expectLogs = JSON.parse(expectLogsRaw) as unknown
+  const timeoutSecs = Number(process.env.TIMEOUT_SECS ?? '10')
+  const redisUrl = process.env.REDIS_URL ?? 'redis://127.0.0.1:6379'
+
+  if (
+    !Array.isArray(expectLogs) ||
+    !expectLogs.every((s) => typeof s === 'string')
+  ) {
+    console.error(
+      '[node-verify-progress] ERROR: EXPECT_LOGS must be a JSON array of strings',
+    )
+    return 1
+  }
+
+  const ids = readFileSync(idsFile, 'utf8')
+    .split('\n')
+    .map((s) => s.trim())
+    .filter((s) => s.length > 0)
+  if (ids.length === 0) {
+    console.error(`[node-verify-progress] ERROR: '${idsFile}' contains no ids`)
+    return 1
+  }
+
+  const queue = new Queue(queueName, { connection: parseConn(redisUrl) })
+  const progressErrors: string[] = []
+  const logErrors: string[] = []
+  try {
+    for (const jid of ids) {
+      const gotProgress = await waitForProgress(
+        queue,
+        jid,
+        expectProgress,
+        timeoutSecs * 1000,
+      )
+      if (gotProgress !== expectProgress) {
+        progressErrors.push(
+          `${jid}: progress got ${JSON.stringify(gotProgress)} want ${expectProgress}`,
+        )
+      }
+
+      const { logs, count } = await waitForLogs(
+        queue,
+        jid,
+        (expectLogs as string[]).length,
+        timeoutSecs * 1000,
+      )
+      if (
+        !arraysEqual(logs, expectLogs as string[]) ||
+        count !== (expectLogs as string[]).length
+      ) {
+        logErrors.push(
+          `${jid}: logs got (${JSON.stringify(logs)}, ${count}) ` +
+            `want (${JSON.stringify(expectLogs)}, ${(expectLogs as string[]).length})`,
+        )
+      }
+    }
+  } finally {
+    await queue.close()
+  }
+
+  if (progressErrors.length > 0) {
+    for (const e of progressErrors.slice(0, 5)) {
+      console.error(`[node-verify-progress] ERROR: ${e}`)
+    }
+    return 1
+  }
+  if (logErrors.length > 0) {
+    for (const e of logErrors.slice(0, 5)) {
+      console.error(`[node-verify-progress] ERROR: ${e}`)
+    }
+    return 1
+  }
+
+  console.log(
+    `[node-verify-progress] OK — ${ids.length} jobs round-tripped ` +
+      `progress=${expectProgress} logs=${JSON.stringify(expectLogs)}`,
+  )
+  return 0
+}
+
+async function waitForProgress(
+  queue: Queue,
+  jobId: string,
+  expect: number,
+  timeoutMs: number,
+): Promise<number | null | undefined> {
+  const deadline = Date.now() + timeoutMs
+  while (Date.now() < deadline) {
+    const job = await queue.getJob(jobId)
+    if (job != null && job.progress === expect) return job.progress
+    await new Promise((r) => setTimeout(r, 50))
+  }
+  const job = await queue.getJob(jobId)
+  return job?.progress
+}
+
+async function waitForLogs(
+  queue: Queue,
+  jobId: string,
+  expectCount: number,
+  timeoutMs: number,
+): Promise<{ logs: string[]; count: number }> {
+  const deadline = Date.now() + timeoutMs
+  // eslint-disable-next-line no-constant-condition
+  while (true) {
+    const res = await queue.getJobLogs(jobId)
+    if (res.count >= expectCount || Date.now() >= deadline) return res
+    await new Promise((r) => setTimeout(r, 50))
+  }
+}
+
+function arraysEqual(a: string[], b: string[]): boolean {
+  if (a.length !== b.length) return false
+  for (let i = 0; i < a.length; i++) if (a[i] !== b[i]) return false
+  return true
+}
+
+function requireEnv(name: string): string {
+  const v = process.env[name]
+  if (v == null || v === '') {
+    throw new Error(`missing required env var ${name}`)
+  }
+  return v
+}
+
+function parseConn(url: string) {
+  const u = new URL(url)
+  return {
+    host: u.hostname || '127.0.0.1',
+    port: u.port ? Number(u.port) : 6379,
+    password: u.password || undefined,
+    username: u.username || undefined,
+    db: u.pathname && u.pathname !== '/' ? Number(u.pathname.slice(1)) : undefined,
+  }
+}
+
+main().then(
+  (code) => process.exit(code),
+  (err) => {
+    console.error(err)
+    process.exit(1)
+  },
+)

.github/cross-shim-tests/worker.py

@@ -25,6 +25,13 @@
                    ``json.loads(RESULT_VALUE)`` instead of ``None``. The
                    verifier asserts this value round-trips through the
                    shim's msgpack wire format.
+  EMIT_PROGRESS — optional. When ``"1"`` the handler calls
+                   ``job.update_progress(75)`` and appends two log lines
+                   (``"step 1"``, ``"step 2"``) before returning so a
+                   ``verify_progress`` step in the opposite shim can read
+                   them back via ``Queue.get_job`` / ``Queue.get_job_logs``.
+                   Pair with ``STORE_RESULT=1`` to keep completed jobs
+                   discoverable by the introspector.
   EXPECT_TAG, TIMEOUT_SECS, REDIS_URL — optional.
 """
 
@@ -48,6 +55,7 @@ async def main() -> int:
     store_results = os.environ.get("STORE_RESULT", "") == "1"
     result_value_raw = os.environ.get("RESULT_VALUE", "")
     result_value = json.loads(result_value_raw) if result_value_raw else None
+    emit_progress = os.environ.get("EMIT_PROGRESS", "") == "1"
 
     seen: set[int] = set()
     done = asyncio.Event()
@@ -75,6 +83,15 @@ async def handler(job: Job):
             )
             done.set()
             return None
+        if emit_progress:
+            try:
+                await job.update_progress(75)
+                await job.log("step 1")
+                await job.log("step 2")
+            except Exception as exc:  # noqa: BLE001
+                errors.append(f"emit_progress failed: {exc!r}")
+                done.set()
+                return None
         seen.add(i)
         if len(seen) >= count:
             done.set()

.github/cross-shim-tests/worker.ts

@@ -14,6 +14,13 @@
 //                 read back. Default off.
 // RESULT_VALUE  — optional, JSON-encoded. When set, the handler returns
 //                 `JSON.parse(RESULT_VALUE)` instead of `undefined`.
+// EMIT_PROGRESS — optional. When '1' the handler calls
+//                 `job.updateProgress(75)` and appends two log lines
+//                 ('step 1', 'step 2') before returning so a
+//                 `verify_progress` step in the opposite shim can read
+//                 them back via `Queue.getJob` / `Queue.getJobLogs`.
+//                 Pair with `STORE_RESULT=1` to keep completed jobs
+//                 discoverable by the introspector.
 
 import { Worker } from '../../chasquimq-node/dist/index.js'
 
@@ -29,6 +36,7 @@ async function main(): Promise<number> {
   const resultValue: unknown = resultValueRaw
     ? JSON.parse(resultValueRaw)
     : undefined
+  const emitProgress = process.env.EMIT_PROGRESS === '1'
 
   const seen = new Set<number>()
   const errors: string[] = []
@@ -62,6 +70,17 @@ async function main(): Promise<number> {
         resolveDone()
         return resultValue
       }
+      if (emitProgress) {
+        try {
+          await job.updateProgress(75)
+          await job.log('step 1')
+          await job.log('step 2')
+        } catch (err) {
+          errors.push(`emitProgress failed: ${(err as Error).message}`)
+          resolveDone()
+          return resultValue
+        }
+      }
       seen.add(i)
       if (seen.size >= count) {
         resolveDone()