Async Workflows

About

The core pattern for building async workflows on Tawa: iec-cron fires on a schedule, your service fans out jobs to iec-queue, workers process items in parallel with retries. Covers catalog-info.yaml queue and schedule declarations, the fan-out trigger pattern (return 200 first), idempotent workers, sequential pipelines, and all anti-patterns.

Skill Content

This is the raw markdown that gets installed as a Claude Code rule.

# Async Workflows — iec-cron + iec-queue

## What this skill covers
Building multi-step async workflows by composing iec-cron (scheduling) and iec-queue (parallel job processing). The pattern for fan-out, retries, and idempotency.

## The Core Pattern

```
iec-cron (fires on schedule)
  ↓ POST /internal/cron/{endpoint}
Your service (fetches pending items, fans out jobs)
  ↓ POST {IEC_QUEUE_URL}/jobs  (one per item)
iec-queue (processes jobs in parallel with retries)
  ↓ POST /internal/jobs/{endpoint}
Your worker (processes one item, writes Septor audit)
```

## catalog-info.yaml

```yaml
spec:
  schedules:
    - name: nightly-sync
      cron: "0 2 * * *"
      endpoint: /internal/cron/nightly-sync
      timezone: America/Denver
      timeoutMs: 60000

  queues:
    - name: sync-record
      endpoint: /internal/jobs/sync-record
      concurrency: 10
      retries: 3
      retryDelayMs: 5000
      timeoutMs: 30000

  internalDependencies:
    - service: iec-queue     # creates IEC_QUEUE_URL
```

## The Trigger (fan-out)

```typescript
// POST /internal/cron/nightly-sync
app.post('/internal/cron/nightly-sync', async (req, res) => {
  // Return 200 FIRST — iec-cron must not wait for fan-out
  res.json({ success: true })

  try {
    const records = await getPendingRecords()

    await Promise.allSettled(
      records.map((record) =>
        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: record.id },
          }),
        }).catch((err) => logger.warn({ err, recordId: record.id }, 'Failed to enqueue'))
      )
    )
  } catch (err) {
    logger.error({ err }, 'Fan-out failed')
  }
})
```

## The Worker (process one item)

```typescript
// POST /internal/jobs/sync-record
app.post('/internal/jobs/sync-record', async (req, res) => {
  const { jobId, data, attempt } = req.body
  const { recordId } = data

  // Idempotency check — retries may re-deliver the same job
  if (await isAlreadySynced(recordId)) {
    return res.json({ success: true, skipped: true })
  }

  try {
    await syncRecord(recordId)

    septor.emit('record.synced', {
      entityId: recordId,
      data: { jobId, attempt },
      metadata: { who: 'sync-worker' },
    }).catch((err) => logger.error({ err }, 'Septor emit failed'))

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

## Schedule Fields

| Field | Required | Description | Default |
|-------|----------|-------------|---------|
| `name` | Yes | Unique name within namespace | — |
| `cron` | Yes | 5-part cron expression | — |
| `endpoint` | Yes | Path to POST on schedule | — |
| `timezone` | No | IANA timezone string | `UTC` |
| `timeoutMs` | No | HTTP callback timeout | `30000` |

## Queue Fields

| Field | Required | Description | Default |
|-------|----------|-------------|---------|
| `name` | Yes | Queue name | — |
| `endpoint` | Yes | Path in your service to POST jobs to | — |
| `concurrency` | No | Max simultaneous jobs | `5` |
| `retries` | No | Retry attempts on failure | `3` |
| `retryDelayMs` | No | Delay between retries | `5000` |
| `timeoutMs` | No | HTTP timeout for handler | `30000` |

## Cron Syntax Reference

```
"*/15 * * * *"     Every 15 minutes
"0 * * * *"        Top of every hour
"0 0 * * *"        Daily at midnight UTC
"0 9 * * 1-5"      Weekdays at 9am
"0 2 1 * *"        First of each month at 2am
```

## Gas Implications

| Operation | Gas charged? |
|-----------|-------------|
| iec-cron schedule fires | 2 tokens (production) / 0 (sandbox) |
| iec-queue job processed | 0 (platform cost) |
| Public API call through Janus | Yes — per route `gas` setting |
| Internal worker HTTP | 0 |

## Pattern Variants

**User-triggered enqueue** (no cron, immediate):
```typescript
app.post('/api/reports/generate', auth, async (req, res) => {
  const { reportId } = await createReport(req.body)
  await fetch(`${process.env.IEC_QUEUE_URL}/jobs`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ namespace: 'my-service', queue: 'generate-report', data: { reportId } }),
  })
  res.json({ success: true, reportId, status: 'queued' })
})
```

**Sequential pipeline** (job triggers next job):
```typescript
app.post('/internal/jobs/step-one', async (req, res) => {
  await processStepOne(req.body.data)
  await fetch(`${process.env.IEC_QUEUE_URL}/jobs`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ namespace: 'my-service', queue: 'step-two', data: req.body.data }),
  })
  res.json({ success: true })
})
```

## Anti-Patterns

| Wrong | Right |
|-------|-------|
| Awaiting all fan-out enqueues before responding 200 | Return 200 first, then fan out |
| `setTimeout` for scheduling | Use iec-cron |
| Rolling your own BullMQ inside your service | Use iec-queue |
| Registering queue/cron endpoints in `routes:` | Use `/internal/` paths — they are NOT public routes |
| Not checking idempotency on retried jobs | Use `jobId` or business key to detect retries |
| Blocking the cron handler with slow work | Return 200 immediately, do work async |

Install

Copy the skill content and save it to:

~/.claude/rules/workflow-pattern.md

Details

Format

Rule

Category

configure

Version

1.0.0

Tokens

~1,100

Updated

2026-03-01