Job Queues on Tawa

The Short Version

Declare queues in catalog-info.yaml. iec-queue POSTs jobs to your endpoint. Return 2xx = complete. Non-2xx = retry.

spec:
  queues:
    - name: process-claim
      endpoint: /internal/jobs/process-claim
      concurrency: 5
      retries: 3
      retryDelayMs: 2000
      timeoutMs: 30000

  internalDependencies:
    - service: iec-queue    # → IEC_QUEUE_URL

Worker Endpoint

iec-queue POSTs the request body as your enqueued payload (the data you passed to /jobs). Job metadata is delivered in headers, not the body:

Header	Value
`X-Job-Id`	Unique job ID — use for idempotency
`X-Job-Attempt`	Attempt number (starts at `1`)
`X-Queue-Name`	The queue this job came from

app.post('/internal/jobs/process-claim', async (req, res) => {
  const jobId = req.headers['x-job-id']
  const attempt = Number(req.headers['x-job-attempt'])
  const { claimId } = req.body  // the payload you enqueued

  // Idempotency check — retries mean this may run multiple times
  if (await isAlreadyProcessed(claimId, jobId)) {
    return res.json({ success: true, skipped: true })
  }

  try {
    const result = await processClaim(claimId)

    septor.emit('claim.processed', {
      entityId: claimId,
      data: { jobId, attempt, result },
      metadata: { who: 'process-claim-worker' },
    }).catch((err) => logger.error({ err }, 'Septor emit failed'))

    res.json({ success: true })
  } catch (err) {
    logger.error({ jobId, claimId, attempt, err }, 'Claim processing failed')
    res.status(500).json({ success: false })  // triggers retry
  }
})

Enqueuing a Job

const response = await fetch(`${process.env.IEC_QUEUE_URL}/jobs`, {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    namespace: process.env.SERVICE_NAMESPACE || 'my-service',
    queue: 'process-claim',
    data: { claimId },
  }),
})

if (!response.ok) throw new Error(`Failed to enqueue: ${response.status}`)
const { id: jobId } = await response.json()

Fan-Out Pattern (cron → many jobs)

app.post('/internal/cron/nightly-sync', async (req, res) => {
  res.json({ success: true })  // Return 200 FIRST — don't make cron wait

  try {
    const records = await getPendingRecords()
    await Promise.allSettled(
      records.map((r) =>
        fetch(`${process.env.IEC_QUEUE_URL}/jobs`, {
          method: 'POST',
          headers: { 'Content-Type': 'application/json' },
          body: JSON.stringify({
            namespace: 'my-service',
            queue: 'sync-record',
            data: { recordId: r.id },
          }),
        }).catch((err) => logger.warn({ err, recordId: r.id }, 'Failed to enqueue'))
      )
    )
  } catch (err) {
    logger.error({ err }, 'Fan-out failed')
  }
})

When to Use Queues vs Direct HTTP

Use iec-queue when...	Use direct HTTP when...
Work takes > 1 second	Work is fast (< 100ms)
You need retries	You handle errors yourself
Processing many items	You need synchronous result
Work must survive restarts	Simple request-response

Job Retention

iec-queue keeps finished jobs for a limited window, then removes them:

Outcome	Retention
Completed	24 hours
Failed	7 days

Query a job's status before it ages out — don't rely on completed jobs persisting past a day.

Queue Fields

Field	Required	Default	Description
`name`	Yes	—	Queue name, unique within namespace
`endpoint`	Yes	—	Your service endpoint path
`concurrency`	No	5	Max simultaneous jobs
`retries`	No	3	Retry attempts on failure
`retryDelayMs`	No	2000	Delay between retries
`timeoutMs`	No	30000	HTTP timeout for your handler

What NOT to Do

// ❌ WRONG: queue endpoints in routes (they must NOT be public)
spec:
  routes:
    - path: /internal/jobs/process-claim  // PUBLIC — exposed through Janus

// ✅ CORRECT: queue endpoints only in queues spec
spec:
  queues:
    - name: process-claim
      endpoint: /internal/jobs/process-claim  // INTERNAL — called only by iec-queue

// ❌ WRONG: not returning 500 on error (iec-queue won't retry)
} catch (err) {
  res.json({ success: false })  // 200 = complete, no retry!

// ✅ CORRECT
} catch (err) {
  res.status(500).json({ success: false })  // 500 = retry

// ❌ WRONG: no idempotency check (retries will re-process)
// ✅ CORRECT: check jobId or data.recordId before processing

Last updated: July 15, 2026