id: dolt-reference-data-promotion-with-revert
namespace: company.team
description: |
Promote reviewed changes from a Dolt staging branch to main without trusting the branch name. The run diffs the
merge, checks main for commits made outside the flow since the last promotion, runs your SQL invariants against the
reviewed commit and waits for an approval. It then merges that exact commit, proves the merge changed what was
reviewed, and reverts it if anything differs.
concurrency:
limit: 1
behavior: CANCEL
inputs:
- id: dry_run
type: BOOL
defaults: true
description: Plan and report only. Nothing is merged and no approval is requested.
- id: auto_promote_clean
type: BOOL
defaults: false
description: Skip the approval for a plan that trips no review rule. Off by
default, so every promotion is approved by a person.
- id: ignore_handled
type: BOOL
defaults: false
description: Plan again for a staging commit that was already planned, declined
or blocked against the current main.
- id: max_delete_pct
type: INT
defaults: 10
description: Review when any table loses more than this percent of its rows.
- id: max_change_pct
type: INT
defaults: 50
description: Review when any table has more than this percent of its rows added,
deleted or modified.
- id: invariants
type: JSON
defaults: |
[
{"name": "positive_prices", "sql": "SELECT sku FROM price_list AS OF '__REV__' WHERE price <= 0"},
{"name": "known_currencies", "sql": "SELECT sku FROM price_list AS OF '__REV__' WHERE currency NOT IN ('EUR', 'USD', 'GBP')"}
]
description: Each entry is a name and a SELECT that returns the violating rows.
__REV__ is replaced with the commit under test (the reviewed staging
commit before the merge, the merge commit after it). Any returned row
blocks the promotion.
- id: approval_timeout
type: DURATION
defaults: PT12H
description: How long a review waits. Without an answer nothing is merged, and
the next run plans again.
variables:
staging_branch: staging
target_branch: main
state_scope: pricing
last_key: "dolt_promo_{{ vars.state_scope }}_last"
marker_key: "dolt_promo_{{ vars.state_scope }}_h_{{ outputs.inspect.row.staging_head }}"
triggers:
- id: staging_ahead
type: io.kestra.plugin.jdbc.mysql.Trigger
description: Fires when the staging branch has commits that main does not. The
branch names are literals here because a trigger has no inputs; keep them
equal to the staging_branch and target_branch variables.
url: "{{ secret('DOLT_JDBC_URL') }}"
username: "{{ secret('DOLT_USER') }}"
password: "{{ secret('DOLT_PASSWORD') }}"
interval: PT5M
fetchType: FETCH_ONE
sql: "SELECT DOLT_HASHOF('staging') AS staging_head FROM dual WHERE (SELECT
COUNT(*) FROM dolt_log('main..staging')) > 0"
disabled: true
tasks:
- id: load_last_promotion
type: io.kestra.plugin.core.kv.Get
description: The main head this flow left behind at its last promotion. Missing
before the first promotion.
key: "{{ render(vars.last_key) }}"
errorOnMissing: false
- id: inspect
type: io.kestra.plugin.jdbc.mysql.Query
description: One read of the whole situation. The three-dot diff
(target...staging) is what a merge applies, so commits that landed on main
after staging branched are not counted as changes.
url: "{{ secret('DOLT_JDBC_URL') }}"
username: "{{ secret('DOLT_USER') }}"
password: "{{ secret('DOLT_PASSWORD') }}"
fetchType: FETCH_ONE
sql: |
SELECT
active_branch() AS active_branch,
DOLT_HASHOF('{{ vars.target_branch }}') AS target_head,
DOLT_HASHOF('{{ vars.staging_branch }}') AS staging_head,
(SELECT COUNT(*) FROM dolt_log('{{ vars.target_branch }}..{{ vars.staging_branch }}')) AS ahead,
(SELECT COUNT(*) FROM dolt_log('{{ vars.staging_branch }}..{{ vars.target_branch }}')) AS behind,
(SELECT COUNT(*) FROM dolt_status) AS dirty_tables,
(SELECT CAST(COALESCE(SUM(num_data_conflicts + num_schema_conflicts), 0) AS SIGNED) FROM dolt_preview_merge_conflicts_summary('{{ vars.target_branch }}', '{{ vars.staging_branch }}')) AS conflicts,
(SELECT COUNT(*) FROM dolt_diff_summary('{{ vars.target_branch }}...{{ vars.staging_branch }}')) AS tables_changed,
(SELECT CAST(COALESCE(SUM(schema_change), 0) AS SIGNED) FROM dolt_diff_summary('{{ vars.target_branch }}...{{ vars.staging_branch }}')) AS schema_changes,
(SELECT CAST(COALESCE(SUM(rows_added), 0) AS SIGNED) FROM dolt_diff_stat('{{ vars.target_branch }}...{{ vars.staging_branch }}')) AS rows_added,
(SELECT CAST(COALESCE(SUM(rows_deleted), 0) AS SIGNED) FROM dolt_diff_stat('{{ vars.target_branch }}...{{ vars.staging_branch }}')) AS rows_deleted,
(SELECT CAST(COALESCE(SUM(rows_modified), 0) AS SIGNED) FROM dolt_diff_stat('{{ vars.target_branch }}...{{ vars.staging_branch }}')) AS rows_modified,
(SELECT ROUND(COALESCE(MAX(rows_deleted * 100.0 / NULLIF(old_row_count, 0)), 0), 1) FROM dolt_diff_stat('{{ vars.target_branch }}...{{ vars.staging_branch }}')) AS worst_delete_pct,
(SELECT ROUND(COALESCE(MAX((rows_added + rows_deleted + rows_modified) * 100.0 / NULLIF(old_row_count, 0)), 0), 1) FROM dolt_diff_stat('{{ vars.target_branch }}...{{ vars.staging_branch }}')) AS worst_change_pct
- id: load_marker
type: io.kestra.plugin.core.kv.Get
description: What this flow already did with this exact staging commit. Written
as pending while a review waits, so the 5 minute trigger does not post the
same plan again.
key: "{{ render(vars.marker_key) }}"
errorOnMissing: false
- id: diff_by_table
type: io.kestra.plugin.jdbc.mysql.Query
description: Per table counts for the review message.
url: "{{ secret('DOLT_JDBC_URL') }}"
username: "{{ secret('DOLT_USER') }}"
password: "{{ secret('DOLT_PASSWORD') }}"
runIf: "{{ outputs.inspect.row.ahead > 0 }}"
fetchType: FETCH
sql: "SELECT table_name, rows_added, rows_deleted, rows_modified, old_row_count,
new_row_count FROM dolt_diff_stat('{{ vars.target_branch }}...{{
vars.staging_branch }}')"
- id: check_drift
type: io.kestra.plugin.jdbc.mysql.Query
description: Commits on main since the head this flow left at its last
promotion. Someone changed main outside the flow if the count is above
zero. allowFailure because a rewritten history makes the old head unknown,
and that must not stop the run.
url: "{{ secret('DOLT_JDBC_URL') }}"
username: "{{ secret('DOLT_USER') }}"
password: "{{ secret('DOLT_PASSWORD') }}"
runIf: "{{ outputs.load_last_promotion.value.main_head is defined and
outputs.load_last_promotion.value.main_head != '' and
outputs.inspect.row.ahead > 0 }}"
allowFailure: true
fetchType: FETCH_ONE
sql: "SELECT COUNT(*) AS outside_commits FROM dolt_log('{{
outputs.load_last_promotion.value.main_head }}..{{ vars.target_branch
}}')"
- id: evaluate_invariants
type: io.kestra.plugin.jdbc.mysql.Query
description: Runs every invariant against the staging commit that was inspected,
not against the moving branch, so a commit pushed during the review is
never tested or merged by accident.
url: "{{ secret('DOLT_JDBC_URL') }}"
username: "{{ secret('DOLT_USER') }}"
password: "{{ secret('DOLT_PASSWORD') }}"
runIf: "{{ outputs.inspect.row.ahead > 0 }}"
fetchType: FETCH
sql: |
{% for inv in inputs.invariants %}SELECT '{{ inv.name }}' AS invariant, COUNT(*) AS violations FROM ({{ inv.sql | replace({'__REV__': outputs.inspect.row.staging_head}) }}) AS v{% if not loop.last %}
UNION ALL
{% endif %}{% endfor %}
- id: findings
type: io.kestra.plugin.core.output.OutputValues
description: Splits what was found into blockers, which stop the promotion, and
review reasons, which require a person.
values:
seen: "{{ not inputs.ignore_handled and (outputs.load_marker.value.main_head is
defined) and (outputs.load_marker.value.main_head ==
outputs.inspect.row.target_head) }}"
blockers: "{% if outputs.inspect.row.active_branch != vars.target_branch
%}connected to branch {{ outputs.inspect.row.active_branch }}, not {{
vars.target_branch }}; {% endif %}{% if outputs.inspect.row.dirty_tables
> 0 %}{{ vars.target_branch }} has uncommitted changes in {{
outputs.inspect.row.dirty_tables }} table(s); {% endif %}{% if
outputs.inspect.row.conflicts > 0 %}the merge would conflict ({{
outputs.inspect.row.conflicts }}); {% endif %}{% if
outputs.evaluate_invariants.rows is defined %}{% for r in
outputs.evaluate_invariants.rows %}{% if r.violations > 0 %}invariant {{
r.invariant }} has {{ r.violations }} violating row(s); {% endif %}{%
endfor %}{% endif %}"
reasons: "{% if outputs.inspect.row.schema_changes > 0 %}schema changes in {{
outputs.inspect.row.schema_changes }} table(s); {% endif %}{% if
outputs.inspect.row.worst_delete_pct > inputs.max_delete_pct %}a table
loses {{ outputs.inspect.row.worst_delete_pct }}% of its rows (limit {{
inputs.max_delete_pct }}%); {% endif %}{% if
outputs.inspect.row.worst_change_pct > inputs.max_change_pct %}a table
has {{ outputs.inspect.row.worst_change_pct }}% of its rows changed
(limit {{ inputs.max_change_pct }}%); {% endif %}{% if
outputs.check_drift.row.outside_commits is defined and
outputs.check_drift.row.outside_commits > 0 %}{{
outputs.check_drift.row.outside_commits }} commit(s) reached {{
vars.target_branch }} outside this flow since the last promotion; {%
endif %}"
- id: verdict
type: io.kestra.plugin.core.output.OutputValues
description: NOOP when staging has nothing new, SEEN when this exact plan was
already handled, BLOCKED when it cannot be promoted, REVIEW when it needs
a person, CLEAN otherwise.
values:
verdict: "{{ outputs.inspect.row.ahead == 0 ? 'NOOP' :
(outputs.findings.values.seen == 'true' ? 'SEEN' :
(outputs.findings.values.blockers != '' ? 'BLOCKED' :
(outputs.findings.values.reasons != '' ? 'REVIEW' : 'CLEAN'))) }}"
- id: blocked_gate
type: io.kestra.plugin.core.flow.If
description: A blocked plan is reported once and remembered, so the trigger does
not repeat it until something changes.
condition: "{{ outputs.verdict.values.verdict == 'BLOCKED' }}"
then:
- id: mark_blocked
type: io.kestra.plugin.core.kv.Set
key: "{{ render(vars.marker_key) }}"
kvType: JSON
ttl: P30D
value: "{{ {'status': 'blocked', 'main_head': outputs.inspect.row.target_head,
'at': execution.startDate, 'execution': execution.id, 'why':
outputs.findings.values.blockers} | toJson }}"
- id: report_blocked
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Dolt promotion of `{{ vars.staging_branch }}` into `{{
vars.target_branch }}` is blocked: {{ outputs.findings.values.blockers
}}Nothing was merged. Staging is at `{{
outputs.inspect.row.staging_head }}`; fix it and the next run plans
again."
- id: plan_gate
type: io.kestra.plugin.core.flow.If
description: Only a plan that can be promoted gets a review request and a merge.
condition: "{{ outputs.verdict.values.verdict == 'REVIEW' or
outputs.verdict.values.verdict == 'CLEAN' }}"
then:
- id: mark_pending
type: io.kestra.plugin.core.kv.Set
description: Expires with the approval window, so an execution that is killed
while waiting never blocks the next run for longer than the window.
key: "{{ render(vars.marker_key) }}"
kvType: JSON
ttl: "{{ inputs.approval_timeout }}"
value: "{{ {'status': 'pending', 'main_head': outputs.inspect.row.target_head,
'at': execution.startDate, 'execution': execution.id} | toJson }}"
- id: announce_plan
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: |
Dolt promotion plan, `{{ vars.staging_branch }}` -> `{{ vars.target_branch }}` ({{ outputs.verdict.values.verdict }}{% if inputs.dry_run %}, dry run{% endif %}).
Staging commit `{{ outputs.inspect.row.staging_head }}` on top of {{ vars.target_branch }} `{{ outputs.inspect.row.target_head }}`: +{{ outputs.inspect.row.rows_added }} rows, -{{ outputs.inspect.row.rows_deleted }}, ~{{ outputs.inspect.row.rows_modified }} in {{ outputs.inspect.row.tables_changed }} table(s).{% if outputs.inspect.row.behind > 0 %} Staging is {{ outputs.inspect.row.behind }} commit(s) behind {{ vars.target_branch }}; the diff counts only what staging adds.{% endif %}
{% if outputs.diff_by_table.rows is defined %}{% for t in outputs.diff_by_table.rows %}- `{{ t.table_name }}`: +{{ t.rows_added }} / -{{ t.rows_deleted }} / ~{{ t.rows_modified }} (rows {{ t.old_row_count }} -> {{ t.new_row_count }})
{% endfor %}{% endif %}{% if outputs.findings.values.reasons != '' %}Needs a person because: {{ outputs.findings.values.reasons }}
{% endif %}{% if not inputs.dry_run %}Only this commit will be merged. Resume execution {{ execution.id }} in Kestra and tick Approve; without an answer within {{ inputs.approval_timeout }} nothing is merged.{% endif %}
- id: approval_gate
type: io.kestra.plugin.core.flow.If
description: A person approves unless the plan is clean and auto_promote_clean
is on.
condition: "{{ not inputs.dry_run and (outputs.verdict.values.verdict ==
'REVIEW' or not inputs.auto_promote_clean) }}"
then:
- id: approve_promotion
type: io.kestra.plugin.core.flow.Pause
description: Human gate. The default is false and a timeout resumes with the
default, so silence never merges.
pauseDuration: "{{ inputs.approval_timeout }}"
behavior: RESUME
onResume:
- id: approve
type: BOOL
displayName: Merge this exact commit into main?
defaults: false
- id: note
type: STRING
displayName: Note for the audit record
required: false
- id: promotion_gate
type: io.kestra.plugin.core.flow.If
description: Merges for a clean plan with auto_promote_clean on, or for an
approved plan. A dry run never gets here.
condition: "{{ not inputs.dry_run and ((outputs.verdict.values.verdict ==
'CLEAN' and inputs.auto_promote_clean) or
(outputs.approve_promotion.onResume.approve is defined and
outputs.approve_promotion.onResume.approve == true)) }}"
then:
- id: recheck_heads
type: io.kestra.plugin.jdbc.mysql.Query
description: Approval can take hours. The merge result depends only on the two
heads, so if both are what was reviewed the conflict and diff
checks still hold.
url: "{{ secret('DOLT_JDBC_URL') }}"
username: "{{ secret('DOLT_USER') }}"
password: "{{ secret('DOLT_PASSWORD') }}"
fetchType: FETCH_ONE
sql: "SELECT DOLT_HASHOF('{{ vars.target_branch }}') AS target_head,
DOLT_HASHOF('{{ vars.staging_branch }}') AS staging_head, (SELECT
COUNT(*) FROM dolt_status) AS dirty_tables"
- id: heads_moved
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.recheck_heads.row.target_head !=
outputs.inspect.row.target_head or
outputs.recheck_heads.row.staging_head !=
outputs.inspect.row.staging_head or
outputs.recheck_heads.row.dirty_tables > 0 }}"
then:
- id: mark_stale
type: io.kestra.plugin.core.kv.Set
key: "{{ render(vars.marker_key) }}"
kvType: JSON
ttl: P30D
value: "{{ {'status': 'stale', 'main_head': outputs.inspect.row.target_head,
'at': execution.startDate, 'execution': execution.id} | toJson
}}"
- id: fail_heads_moved
type: io.kestra.plugin.core.execution.Fail
errorMessage: "A branch moved while the review waited: {{ vars.target_branch }}
was {{ outputs.inspect.row.target_head }} and is {{
outputs.recheck_heads.row.target_head }}, {{
vars.staging_branch }} was {{ outputs.inspect.row.staging_head
}} and is {{ outputs.recheck_heads.row.staging_head }},
uncommitted tables on {{ vars.target_branch }}: {{
outputs.recheck_heads.row.dirty_tables }}. Nothing was merged.
The next run plans again."
- id: promote
type: io.kestra.plugin.jdbc.mysql.Query
description: Merges the reviewed commit by hash, with --no-ff so the promotion
is one merge commit that can be reverted as a unit.
url: "{{ secret('DOLT_JDBC_URL') }}"
username: "{{ secret('DOLT_USER') }}"
password: "{{ secret('DOLT_PASSWORD') }}"
fetchType: FETCH_ONE
sql: "CALL DOLT_MERGE('--no-ff', '-m', 'Promote {{ vars.staging_branch }} {{
outputs.inspect.row.staging_head }} via Kestra execution {{
execution.id }}', '{{ outputs.inspect.row.staging_head }}')"
- id: verify_promotion
type: io.kestra.plugin.jdbc.mysql.Query
description: Checks the merge commit is a merge of exactly the two reviewed
heads, that main points at it, that the working set is clean, and
what the merge changed compared with the head it was merged onto.
url: "{{ secret('DOLT_JDBC_URL') }}"
username: "{{ secret('DOLT_USER') }}"
password: "{{ secret('DOLT_PASSWORD') }}"
fetchType: FETCH_ONE
sql: |
SELECT
CASE WHEN (SELECT parents FROM dolt_log('--parents', '{{ vars.target_branch }}') LIMIT 1) = CONCAT('{{ outputs.inspect.row.target_head }}', ', ', '{{ outputs.inspect.row.staging_head }}') THEN 1 ELSE 0 END AS lineage_ok,
CASE WHEN DOLT_HASHOF('{{ vars.target_branch }}') = '{{ outputs.promote.row.hash }}' THEN 1 ELSE 0 END AS head_ok,
CASE WHEN (SELECT COUNT(*) FROM dolt_status) = 0 THEN 1 ELSE 0 END AS clean_ok,
(SELECT CAST(COALESCE(SUM(rows_added), 0) AS SIGNED) FROM dolt_diff_stat('{{ outputs.inspect.row.target_head }}', '{{ outputs.promote.row.hash }}')) AS rows_added,
(SELECT CAST(COALESCE(SUM(rows_deleted), 0) AS SIGNED) FROM dolt_diff_stat('{{ outputs.inspect.row.target_head }}', '{{ outputs.promote.row.hash }}')) AS rows_deleted,
(SELECT CAST(COALESCE(SUM(rows_modified), 0) AS SIGNED) FROM dolt_diff_stat('{{ outputs.inspect.row.target_head }}', '{{ outputs.promote.row.hash }}')) AS rows_modified
- id: verify_invariants_after
type: io.kestra.plugin.jdbc.mysql.Query
description: The same invariants again, now on the merge commit.
url: "{{ secret('DOLT_JDBC_URL') }}"
username: "{{ secret('DOLT_USER') }}"
password: "{{ secret('DOLT_PASSWORD') }}"
fetchType: FETCH
sql: |
{% for inv in inputs.invariants %}SELECT '{{ inv.name }}' AS invariant, COUNT(*) AS violations FROM ({{ inv.sql | replace({'__REV__': outputs.promote.row.hash}) }}) AS v{% if not loop.last %}
UNION ALL
{% endif %}{% endfor %}
- id: verification
type: io.kestra.plugin.core.output.OutputValues
description: Everything that must hold after the merge. Any problem sends the
run to the revert branch.
values:
problems: "{% if outputs.promote.row.conflicts != 0 %}merge reported {{
outputs.promote.row.conflicts }} conflict(s); {% endif %}{% if
outputs.verify_promotion.row.lineage_ok != 1 %}the new head is
not a merge of the reviewed heads; {% endif %}{% if
outputs.verify_promotion.row.head_ok != 1 %}{{
vars.target_branch }} does not point at the merge commit; {%
endif %}{% if outputs.verify_promotion.row.clean_ok != 1 %}the
working set is not clean; {% endif %}{% if
outputs.verify_promotion.row.rows_added !=
outputs.inspect.row.rows_added or
outputs.verify_promotion.row.rows_deleted !=
outputs.inspect.row.rows_deleted or
outputs.verify_promotion.row.rows_modified !=
outputs.inspect.row.rows_modified %}the merge changed +{{
outputs.verify_promotion.row.rows_added }} -{{
outputs.verify_promotion.row.rows_deleted }} ~{{
outputs.verify_promotion.row.rows_modified }} rows but +{{
outputs.inspect.row.rows_added }} -{{
outputs.inspect.row.rows_deleted }} ~{{
outputs.inspect.row.rows_modified }} were reviewed; {% endif
%}{% if outputs.verify_invariants_after.rows is defined %}{% for
r in outputs.verify_invariants_after.rows %}{% if r.violations >
0 %}invariant {{ r.invariant }} has {{ r.violations }} violating
row(s) after the merge; {% endif %}{% endfor %}{% endif %}"
- id: promotion_outcome
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.verification.values.problems == '' }}"
then:
- id: save_last_promotion
type: io.kestra.plugin.core.kv.Set
description: The head this flow leaves behind. The next run counts commits on
main after it as changes made outside the flow.
key: "{{ render(vars.last_key) }}"
kvType: JSON
value: "{{ {'main_head': outputs.promote.row.hash, 'promoted_staging_head':
outputs.inspect.row.staging_head, 'promoted_at':
execution.startDate, 'execution': execution.id,
'approval_note': (outputs.approve_promotion.onResume.note is
defined ? outputs.approve_promotion.onResume.note : '')} |
toJson }}"
- id: mark_promoted
type: io.kestra.plugin.core.kv.Set
key: "{{ render(vars.marker_key) }}"
kvType: JSON
ttl: P30D
value: "{{ {'status': 'promoted', 'main_head': outputs.promote.row.hash, 'at':
execution.startDate, 'execution': execution.id} | toJson }}"
- id: report_promoted
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Dolt promotion done and verified: `{{ vars.staging_branch }}` @
`{{ outputs.inspect.row.staging_head }}` is merged into `{{
vars.target_branch }}` as `{{ outputs.promote.row.hash }}`
(+{{ outputs.verify_promotion.row.rows_added }} -{{
outputs.verify_promotion.row.rows_deleted }} ~{{
outputs.verify_promotion.row.rows_modified }} rows). The merge
is one commit: revert it with `CALL DOLT_REVERT('{{
outputs.promote.row.hash }}')`."
else:
- id: revert_promotion
type: io.kestra.plugin.jdbc.mysql.Query
description: Reverts the merge commit. A revert adds a commit instead of
rewriting history, so replicas that already pulled the merge
follow it back.
url: "{{ secret('DOLT_JDBC_URL') }}"
username: "{{ secret('DOLT_USER') }}"
password: "{{ secret('DOLT_PASSWORD') }}"
fetchType: FETCH_ONE
sql: "CALL DOLT_REVERT('{{ outputs.promote.row.hash }}')"
- id: verify_revert
type: io.kestra.plugin.jdbc.mysql.Query
description: After a revert main must hold exactly what it held right before the
merge. That is the merge commit's first parent (<hash>~1),
which equals the reviewed head unless main moved between the
head recheck and the merge.
url: "{{ secret('DOLT_JDBC_URL') }}"
username: "{{ secret('DOLT_USER') }}"
password: "{{ secret('DOLT_PASSWORD') }}"
fetchType: FETCH_ONE
sql: "SELECT DOLT_HASHOF('{{ outputs.promote.row.hash }}~1') AS pre_merge_head,
(SELECT COUNT(*) FROM dolt_diff_summary('{{
outputs.promote.row.hash }}~1', '{{ vars.target_branch }}'))
AS tables_differing, (SELECT COUNT(*) FROM dolt_status) AS
dirty_tables"
- id: mark_rolled_back
type: io.kestra.plugin.core.kv.Set
key: "{{ render(vars.marker_key) }}"
kvType: JSON
ttl: P30D
value: "{{ {'status': 'rolled_back', 'main_head':
outputs.inspect.row.target_head, 'at': execution.startDate,
'execution': execution.id, 'why':
outputs.verification.values.problems} | toJson }}"
- id: report_rolled_back
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Dolt promotion FAILED verification and was reverted: {{
outputs.verification.values.problems }}`{{ vars.target_branch
}}` {% if outputs.verify_revert.row.tables_differing == 0 and
outputs.verify_revert.row.dirty_tables == 0 %}holds exactly
the data it had before the merge (revert commit `{{
outputs.revert_promotion.row.hash }}`){% else %}STILL DIFFERS
from `{{ outputs.verify_revert.row.pre_merge_head }}` in {{
outputs.verify_revert.row.tables_differing }} table(s), check
it by hand{% endif %}.{% if
outputs.verify_revert.row.pre_merge_head !=
outputs.inspect.row.target_head %} Note: {{ vars.target_branch
}} moved from `{{ outputs.inspect.row.target_head }}` to `{{
outputs.verify_revert.row.pre_merge_head }}` before the merge,
so someone committed to it during the run.{% endif %}"
- id: fail_after_revert
type: io.kestra.plugin.core.execution.Fail
errorMessage: "The promotion of {{ outputs.inspect.row.staging_head }} failed
verification ({{ outputs.verification.values.problems }}) and
{{ vars.target_branch }} was reverted; tables still differing
from the head before the merge ({{
outputs.verify_revert.row.pre_merge_head }}): {{
outputs.verify_revert.row.tables_differing }}."
else:
- id: mark_not_promoted
type: io.kestra.plugin.core.kv.Set
description: Remembers a declined or planned-only commit so the trigger does not
ask again for the same heads.
key: "{{ render(vars.marker_key) }}"
kvType: JSON
ttl: P30D
value: "{{ {'status': (inputs.dry_run ? 'planned' : 'declined'), 'main_head':
outputs.inspect.row.target_head, 'at': execution.startDate,
'execution': execution.id} | toJson }}"
- id: report_not_promoted
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Only for a declined or timed-out review. A dry run already said so
in the plan message.
runIf: "{{ not inputs.dry_run }}"
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Dolt promotion of `{{ vars.staging_branch }}` @ `{{
outputs.inspect.row.staging_head }}` was not approved and is not
merged. `{{ vars.target_branch }}` is unchanged at `{{
outputs.inspect.row.target_head }}`."
errors:
- id: alert_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Dolt promotion flow failed in execution {{ execution.id }}{% if
errorLogs() | length > 0 %} at `{{ errorLogs()[0]['taskId'] }}`: {{
errorLogs()[0]['message'] | abbreviate(300) }}{% endif %}. Check `{{
vars.target_branch }}` with `SELECT DOLT_HASHOF('{{ vars.target_branch
}}')` against the head in the plan message before running again."
