Hightouch Sync Run Health Gate with Kestra

id: hightouch-sync-run-health-gate
namespace: company.team

description: |
  Trigger a Hightouch reverse ETL sync, classify the run as HEALTHY, DEGRADED or PROBE_FAILED
  from the real status and row counts Hightouch reports, alert Slack when rows fail to land in
  the destination, optionally retry once, and log a final audit summary either way.

inputs:
  - id: sync_id
    type: INT
    displayName: Hightouch sync ID
    description: Numeric ID of the Hightouch sync to trigger and monitor. Maps to
      the Sync task's syncId field.
    defaults: 1127166

  - id: full_resync
    type: BOOL
    displayName: Force full resynchronization
    description: When true, Hightouch reloads every row for this run instead of only
      the incremental changes. Maps to the Sync task's fullResynchronization
      field.
    defaults: false

  - id: max_wait
    type: STRING
    displayName: Maximum wait duration
    description: ISO-8601 duration the Sync task polls the run status for before
      timing out, for example PT5M for 5 minutes. Maps to the Sync task's
      maxDuration field.
    defaults: "PT5M"

  - id: failed_rows_warning_threshold
    type: INT
    displayName: Failed rows warning threshold
    description: Total failed rows, added plus changed plus removed, above which a
      run is classified DEGRADED even when Hightouch reports status SUCCESS.
    defaults: 0

  - id: retry_mode
    type: SELECT
    displayName: Retry resync mode
    description: Resync mode used for the retry when auto_retry is true. incremental
      retries only changed rows, full forces a complete reload on the retry.
    values:
      - "incremental"
      - "full"
    defaults: "incremental"

  - id: auto_retry
    type: BOOL
    displayName: Auto retry
    description: When true, together with dry_run false, retrigger the sync once
      after a DEGRADED or PROBE_FAILED classification.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_retry true, for the retry to
      actually run.
    defaults: true

tasks:
  - id: probe_sync
    type: io.kestra.plugin.hightouch.Sync
    description: Triggers the sync and waits for a terminal status. Hightouch throws
      rather than returning output for a FAILED, CANCELLED or INTERRUPTED run,
      allowFailure keeps the flow going so that case can still be classified
      below.
    allowFailure: true
    token: "{{ secret('HIGHTOUCH_API_TOKEN') }}"
    syncId: "{{ inputs.sync_id }}"
    fullResynchronization: "{{ inputs.full_resync }}"
    maxDuration: "{{ inputs.max_wait }}"

  - id: classify_sync_status
    type: io.kestra.plugin.core.debug.Return
    description: PROBE_FAILED when the probe produced no output at all, DEGRADED
      when it succeeded but with failed rows over threshold or a non-SUCCESS
      terminal status, HEALTHY otherwise.
    format: "{{ outputs.probe_sync is not defined or outputs.probe_sync.metadata is
      not defined ? 'PROBE_FAILED' : ((outputs.probe_sync.metadata.status !=
      'SUCCESS' or ((outputs.probe_sync.metadata.failedRows is defined and
      outputs.probe_sync.metadata.failedRows is not null ?
      (outputs.probe_sync.metadata.failedRows.addedCount ?? 0) +
      (outputs.probe_sync.metadata.failedRows.changedCount ?? 0) +
      (outputs.probe_sync.metadata.failedRows.removedCount ?? 0) : 0) >
      inputs.failed_rows_warning_threshold)) ? 'DEGRADED' : 'HEALTHY') }}"

  - id: route_by_sync_status
    type: io.kestra.plugin.core.flow.Switch
    description: Branch A logs a healthy sync. Branch B alerts Slack when the sync
      produced no result or landed fewer rows than it should have.
    value: "{{ outputs.classify_sync_status.value }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "Hightouch sync {{ inputs.sync_id }} healthy. status={{
            outputs.probe_sync.metadata.status }} completionRatio={{
            outputs.probe_sync.metadata.completionRatio }}."
      "DEGRADED":
        - id: alert_degraded
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {{ {"text": "Hightouch sync " ~ inputs.sync_id ~ " DEGRADED. status=" ~ outputs.probe_sync.metadata.status ~ ", failed rows added=" ~ outputs.probe_sync.metadata.failedRows.addedCount ~ " changed=" ~ outputs.probe_sync.metadata.failedRows.changedCount ~ " removed=" ~ outputs.probe_sync.metadata.failedRows.removedCount ~ ". completionRatio=" ~ outputs.probe_sync.metadata.completionRatio ~ "."} | toJson }}
      "PROBE_FAILED":
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {{ {"text": "Hightouch sync " ~ inputs.sync_id ~ " produced no result. Hightouch reports FAILED, CANCELLED and INTERRUPTED runs by throwing rather than returning a status, so this likely means the sync failed outright, was cancelled, or exceeded the " ~ inputs.max_wait ~ " wait window."} | toJson }}

  - id: remediate_gate
    type: io.kestra.plugin.core.flow.If
    description: Runs only when the sync was not healthy and auto_retry is true and
      dry_run is false.
    condition: "{{ outputs.classify_sync_status.value != 'HEALTHY' and
      inputs.auto_retry and not inputs.dry_run }}"
    then:
      - id: retry_sync
        type: io.kestra.plugin.hightouch.Sync
        allowFailure: true
        token: "{{ secret('HIGHTOUCH_API_TOKEN') }}"
        syncId: "{{ inputs.sync_id }}"
        fullResynchronization: "{{ inputs.retry_mode == 'full' }}"
        maxDuration: "{{ inputs.max_wait }}"

      - id: reclassify_sync_status
        type: io.kestra.plugin.core.debug.Return
        format: "{{ outputs.retry_sync is not defined or outputs.retry_sync.metadata is
          not defined ? 'PROBE_FAILED' : ((outputs.retry_sync.metadata.status !=
          'SUCCESS' or ((outputs.retry_sync.metadata.failedRows is defined and
          outputs.retry_sync.metadata.failedRows is not null ?
          (outputs.retry_sync.metadata.failedRows.addedCount ?? 0) +
          (outputs.retry_sync.metadata.failedRows.changedCount ?? 0) +
          (outputs.retry_sync.metadata.failedRows.removedCount ?? 0) : 0) >
          inputs.failed_rows_warning_threshold)) ? 'DEGRADED' : 'HEALTHY') }}"

      - id: verify_retry
        type: io.kestra.plugin.core.flow.If
        description: Fails the flow when the retry itself did not come back healthy, so
          a retry is never silently reported as a fix.
        condition: "{{ outputs.reclassify_sync_status.value != 'HEALTHY' }}"
        then:
          - id: retry_failed
            type: io.kestra.plugin.core.execution.Fail
            errorMessage: "Retry of Hightouch sync {{ inputs.sync_id }} is still not HEALTHY
              (status {{ outputs.reclassify_sync_status.value }})."
        else:
          - id: retried
            type: io.kestra.plugin.core.log.Log
            message: "Retry of Hightouch sync {{ inputs.sync_id }} is HEALTHY. status={{
              outputs.retry_sync.metadata.status }}."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: Final audit log confirming the gate ran, regardless of the branch
      taken above.
    message: "Hightouch sync run health gate completed. sync_id={{ inputs.sync_id }}
      status={{ outputs.classify_sync_status.value }} auto_retry={{
      inputs.auto_retry }} dry_run={{ inputs.dry_run }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {"text": "Hightouch sync run health gate flow failed (execution " ~ execution.id ~ "). Check the execution logs."} | toJson }}

triggers:
  - id: schedule_every_6_hours
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, and always in safe mode by default.
    cron: "0 */6 * * *"
    disabled: true
    inputs:
      sync_id: 1127166
      full_resync: false
      max_wait: "PT5M"
      failed_rows_warning_threshold: 0
      retry_mode: "incremental"
      auto_retry: false
      dry_run: true

A reverse ETL sync that silently drops rows is worse than one that fails outright, the destination (a CRM, an ad platform, a support tool) looks current, but the sales rep is calling a lead with last week's lifecycle stage. Hightouch reports exactly how many rows succeeded and failed on every run, status, failedRows, completionRatio, but nothing reads that report unless you build it. This blueprint turns a Hightouch sync into a Day 2 governance gate: it triggers the sync, classifies the real outcome, alerts Slack with the exact row counts, and can retry once before anyone has to go looking.

How it works

probe_sync (io.kestra.plugin.hightouch.Sync) triggers the sync by sync_id and waits for a terminal status, polling every second up to max_wait. allowFailure: true matters here: Hightouch's own task throws an exception rather than returning output for a FAILED, CANCELLED or INTERRUPTED run, or if max_wait is exceeded, so without allowFailure the flow would stop before it could classify or alert on exactly the case that matters most.
classify_sync_status (io.kestra.plugin.core.debug.Return) reads outputs.probe_sync.metadata.status and the three failedRows counts (addedCount, changedCount, removedCount). PROBE_FAILED when probe_sync produced no output at all (guarded with is not defined), DEGRADED when it succeeded but the total failed rows exceed failed_rows_warning_threshold or the status is not exactly SUCCESS (for example WARNING or COMPLETED_WITH_ERRORS), HEALTHY otherwise.
route_by_sync_status (io.kestra.plugin.core.flow.Switch) branches on that status:
- HEALTHY: log_healthy logs the status and completionRatio.
- DEGRADED: a SlackIncomingWebhook task alerts with the exact status and the three failed row counts.
- PROBE_FAILED: a SlackIncomingWebhook task alerts that the sync produced no result, with the plain-language explanation of why Hightouch's own task behaves that way for a failed or cancelled run.
remediate_gate (io.kestra.plugin.core.flow.If) runs only when the classification was not HEALTHY and auto_retry is true and dry_run is false:
- retry_sync triggers the same sync again, with fullResynchronization set from retry_mode.
- reclassify_sync_status runs the identical classification against the retry's output.
- verify_retry (io.kestra.plugin.core.flow.If) fails the flow with io.kestra.plugin.core.execution.Fail if the retry is still not HEALTHY, so a failed retry is never silently reported as a fix.
log_audit_summary always runs last and logs the final classification.
The errors block posts to Slack if the flow itself fails outside the gate logic above.

What you get

A real answer to "did this sync actually land the data", not just "did the task finish".
A Slack alert carrying Hightouch's own row counts and status, never an invented number.
A safe-by-default retry path: dry_run: true ships enabled, nothing retries until you explicitly turn both auto_retry on and dry_run off.
A verified retry: the flow reclassifies the retry's own result and fails loudly if it is still not healthy, instead of assuming a second attempt fixed it.

Who it's for

Data and RevOps teams who depend on Hightouch to keep a CRM, ad platform or support tool in sync with the warehouse, and currently find out about a bad sync from the business side instead of an alert.
Teams evaluating the Hightouch plugin who want a working example of io.kestra.plugin.hightouch.Sync, its metadata output, and the exception-on-failure behavior that is easy to miss on a first read of the plugin.

Why orchestrate this with Kestra

Hightouch's own scheduler can retrigger a sync, but it has no concept of an allow-listed failure threshold, a classification, or an alert that explains what actually went wrong in plain numbers. Kestra adds the schedule, the DEGRADED versus PROBE_FAILED distinction, the Slack alert built from Hightouch's own reported counts, the gated retry, and the verification that the retry worked, all as one declarative YAML file you can version and review like code.

Prerequisites

A Hightouch workspace with at least one sync configured. Hightouch is a managed reverse ETL service with no self-hostable open-source Docker image, there is no local container to start for Hightouch itself.
An API bearer token with permission to trigger the sync (Hightouch Settings, API tokens).
A Slack incoming webhook for alerts.

Secrets

HIGHTOUCH_API_TOKEN: API bearer token used by probe_sync and retry_sync.
SLACK_WEBHOOK_URL: Slack incoming webhook URL used by the alert tasks and the errors block.

In Kestra OSS, set both as base64-encoded environment variables: SECRET_HIGHTOUCH_API_TOKEN, SECRET_SLACK_WEBHOOK_URL.

Inputs

Input	Type	Default	Purpose
`sync_id`	INT	`1127166`	Numeric ID of the Hightouch sync to trigger and monitor
`full_resync`	BOOL	`false`	When true, forces a full resynchronization of all records
`max_wait`	STRING	`PT5M`	Maximum wait duration before timeout
`failed_rows_warning_threshold`	INT	`0`	Failed row count threshold before marking run DEGRADED
`retry_mode`	SELECT	`incremental`	Resync mode for retry (`incremental` or `full`)
`auto_retry`	BOOL	`false`	When true with `dry_run: false`, triggers a single retry
`dry_run`	BOOL	`true`	Must be false for the retry execution to run

Quick start

Create a Hightouch API bearer token and note the numeric ID of the sync to monitor.
Add the HIGHTOUCH_API_TOKEN and SLACK_WEBHOOK_URL secrets.
Run this flow manually with the defaults and read log_audit_summary for the result.
If it reports DEGRADED or PROBE_FAILED and a retry is the right call, set auto_retry: true and dry_run: false and run it again.
Enable the schedule_every_6_hours trigger, which always ships in safe mode.

How to extend

Lower failed_rows_warning_threshold to 0 on a sync where every row must land, or raise it on a sync where a small drop rate is expected and tolerated.
Add a second probe_sync for a different sync_id and reuse the same classification logic to monitor several syncs from one flow.
Read outputs.probe_sync.metadata.querySize and alert separately when it grows far past its usual size, a sign the source query itself has drifted.
Chain a downstream task after HEALTHY that writes completionRatio to a Kestra KV entry, to track sync health over time instead of only per run.

Pitfalls

Hightouch's own Sync task does not return a FAILED, CANCELLED or INTERRUPTED run as output with that status, it throws an exception instead. Without allowFailure: true on probe_sync and retry_sync, the flow stops before classify_sync_status ever runs.
A timeout past max_wait also throws, with the same effect, a sync that is simply slow looks identical to one that failed outright. Raise max_wait before assuming every PROBE_FAILED is a real failure.
completionRatio can be high while failedRows is still non-zero, most of the batch landed but not all of it, which is exactly the case failed_rows_warning_threshold exists to catch even when Hightouch itself reports status: SUCCESS.
The API token only needs permission to trigger and read this sync, it does not need workspace-admin access.

Hightouch Sync Run Health Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Hightouch Sync Run Health Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

More Business Blueprints

New to Kestra?