Task patterns

These patterns show how Envoy primitives fit together. Adapt connector methods to the deployed type and keep source scope bounded during testing.

Stream a collection into an upsert

steps:
  - name: read
    type: call_method
    config:
      connector: source
      method: find
      args:
        table: contacts
 
  - name: normalize
    type: transform
    config: |
      {
        "id": $string(input.id),
        "email": $lowercase(input.email)
      }
 
  - name: write
    type: call_method
    config: |
      {
        "connector": "destination",
        "method": "upsert",
        "args": {
          "table": "contacts",
          "data": input,
          "conflict_columns": ["id"]
        }
      }

The destination must enforce uniqueness for the conflict key.

Stage before external writes

Use a database staging table or collection to separate extraction from a slow or failure-prone destination:

1. extract and normalize into staging;
2. mark each row pending;
3. process pending rows with bounded concurrency;
4. mark success only after confirmed writes;
5. retain an exception category for failures;
6. reconcile by source identity.

This supports controlled retries without rereading the entire source.

Parent and child API data

List parents as the source step, then query children from the current parent:

- name: list_children
  type: call_method
  config: |
    {
      "connector": "source",
      "method": "list_children",
      "args": {
        "parent_id": input.id
      }
    }

Carry parent identity into each child transform. Use a job DAG instead when child migration depends on completed destination mappings from a separate stage.

Enrich with a bounded lookup

Use $callMethod() for find_one, count, or another bounded result. For an unbounded related collection, use a streaming call_method step.

Index the lookup key; one full scan per record is not a production pattern.

Parallel writes

Add concurrency to the write step:

concurrency: 5

Start low. Verify provider rate limits, destination uniqueness, ordering requirements, and idempotency before increasing it.

Change detection

Normalize the source record, hash the canonical payload, and compare it with the stored mapping hash. Write only when new or changed.

Prefer sync_record for the standard form. See Build an idempotent migration.

Batch APIs

Use batch.size before a connector method that accepts arrays. Handle partial batch failures explicitly; do not mark the entire batch successful based on one response flag unless the provider guarantees atomicity.

File transfer

Generate a stable intermediate key from source identity. Transfer with streaming where supported, record the destination key, and let reruns reuse or replace it according to an explicit policy.

Anti-patterns

• Fetching all rows into memory before filtering.
• Using $callMethod() for an unbounded list.
• Marking processed after an ignored write without checking output.
• Retrying an ambiguous create without reconciliation.
• Hashing raw objects with unstable field or array order.
• Storing credentials in arguments or variables.
• Running dependent stages with skip_deps without verified upstream state.