Idempotent sync and the mapping DB

This guide shows how to build a migration task that can run any number of times without creating duplicate destination records. You want this for every migration write: reruns after partial failures, delta passes, and launch-day retries all depend on it.

The mechanism is a mapping DB: a data storage connector (sqlite, postgres, mysql, or mssql) that persists source-to-destination ID mappings. Each record you migrate gets a row that says "source record X became destination record Y", plus an optional content hash for change detection.

Prerequisites

• A source connector with records that carry a stable, immutable identifier.
• A destination connector with create (and optionally update) methods.
• A data storage connector to act as the mapping DB — sqlite is the simplest choice.
• Familiarity with task anatomy and JSONata step configs.

How the mapping DB works

Mapping-capable connectors store mappings in a resource_mapping table that the connector creates automatically on first use. Each row holds:

Rows are unique per (resource, source_id), so writing a mapping for the same source record again updates the existing row instead of adding a duplicate.

Steps that use the mapping DB name the connector directly in their config. Separately, the top-level lookup_connector task key designates a connector that backs lookup steps and the $callMethod and $getDestinationId JSONata helpers; the validator counts it as connector usage even when no step references it.

Prefer sync_record

The sync_record step is the composite, preferred way to write idempotently. In one step it:

1. Looks up the mapping for mapping.resource + mapping.source_id.
2. Computes a content hash of the input record when a hash block is configured.
3. Calls destination.create when the record is unmapped, or destination.update when it is mapped, an update call is configured, hash tracking is on, and the stored hash differs from the new one. An already-mapped record whose content is unchanged — or with no update or no hash configured — is left untouched.
4. Upserts the mapping (destination ID plus hash) in the mapping DB.
5. Optionally writes the destination ID and hash back onto a staging row via writeback.

Key config blocks, as defined in the step implementation:

• mapping — connector, resource, source_id (required; the step fails if source_id is empty).
• hash — optional; fields (which fields to hash), algorithm (sha256 or md5), from (hash a nested object field instead of the whole record).
• destination — connector, create, and optional update. Each call takes method, optional resource, args, and id (the response field to read the new destination ID from; by default the step tries id, Id, contactid, accountid).
• writeback — optional; connector, table or collection, key_column, key, fields.

See sync_record for the full configuration table.

Steps

1. Declare the connector slots, including the mapping DB:

   connectors:
     source:
       type: csv
     destination:
       type: dynamics365
     mapping_db:
       type: sqlite

2. Stream the source records with a call_method step.

3. Add a sync_record step. The config references input, so it must be a JSONata string. The complete task:

   name: sync-contacts
   description: Idempotently sync contacts from CSV into the destination CRM
   output: none
    
   connectors:
     source:
       type: csv
     destination:
       type: dynamics365
     mapping_db:
       type: sqlite
    
   steps:
     - name: read_contacts
       type: call_method
       config:
         connector: source
         method: read
         args:
           resource: contacts
    
     - name: sync_contact
       type: sync_record
       config: |
         {
           "mapping": {
             "connector": "mapping_db",
             "resource": "contact",
             "source_id": $string(input.id)
           },
           "hash": {
             "fields": ["first_name", "last_name", "email"]
           },
           "destination": {
             "connector": "destination",
             "create": {
               "method": "create",
               "resource": "contacts",
               "args": {
                 "firstname": input.first_name,
                 "lastname": input.last_name,
                 "emailaddress1": input.email
               }
             },
             "update": {
               "method": "update",
               "resource": "contacts",
               "args": {
                 "firstname": input.first_name,
                 "lastname": input.last_name,
                 "emailaddress1": input.email
               }
             }
           }
         }

4. Run the task with your connector configs:

   envoy sync-contacts --config connectors.yaml

The manual triad

Before sync_record, the same result took three steps plus a hash step. The manual steps still exist and are the right tool when the composite step cannot express your workflow — for example when the "write" is not a single create or update call, or when you need the hash comparison to feed custom skip logic:

• get_destination_id — reads the mapping and merges destination_id and hash into the item (the original input fields are preserved).
• compute_hash — adds a content hash to the record. fields selects what to hash (default: all fields not starting with _), algorithm is sha256 (default) or md5, and field_name sets the output field (default _hash).
• set_destination_id — writes the mapping after a successful create.

The dedup pattern gates the write and the mapping write with skip_if:

- name: check_existing
  type: get_destination_id
  config: |
    {
      "connector": "mapping_db",
      "resource": "account",
      "source_id": steps.read_accounts.output.account_id
    }
 
- name: create_account
  type: call_method
  skip_if: "steps.check_existing.output.destination_id != null"
  config: |
    {
      "connector": "destination",
      "method": "create",
      "args": { "resource": "accounts", "data": input }
    }
 
- name: store_mapping
  type: set_destination_id
  skip_if: "steps.check_existing.output.destination_id != null"
  config: |
    {
      "connector": "mapping_db",
      "resource": "account",
      "source_id": steps.read_accounts.output.account_id,
      "destination_id": input.accountid
    }

For change detection, add compute_hash before the mapping lookup and compare steps.check_existing.output.hash against the computed hash in skip_if, passing the new hash to set_destination_id.

Keep hashes canonical

Both sync_record and compute_hash hash a stable serialization of the selected fields, so the same field values always produce the same hash. What you must keep canonical is the payload: an initial-load task and a delta task must hash the same fields with the same normalization, defaults, and array order. A single difference makes every record look changed on every delta run. See Migration playbook for delta parity in practice.

Verify it worked

Run the task twice against the same source data. On the second run:

• No new destination records are created — every record is already mapped.
• With hash and update configured, records whose content changed between runs are updated in place; unchanged records are skipped entirely.
• The resource_mapping table in the mapping DB has exactly one row per (resource, source_id).

Related

• sync_record — full configuration reference for the composite step
• Migration playbook — end-to-end migration structure using these patterns
• Error handling, retries, and loops — what happens when a destination write fails
• get_destination_id — mapping lookup step for the manual triad