Neo4j Long-Running Transaction Killer Guard with Kestra

id: neo4j-long-running-transaction-killer-guard
namespace: company.team

description: |
  Find Neo4j transactions that have been running longer than a threshold, report them
  to Slack and after approval terminate the offenders, then re-list and fail the run if
  any transaction it targeted is still open. A single stuck or runaway transaction holds
  locks and memory and blocks writers, so this guard surfaces it and can clear it on a
  schedule, never touching its own session or any user you exclude.

inputs:
  - id: url
    type: STRING
    displayName: Neo4j Bolt URL
    defaults: "bolt://localhost:7687"

  - id: user
    type: STRING
    displayName: Neo4j user
    defaults: "neo4j"

  - id: password
    type: STRING
    displayName: Neo4j password
    description: Defaults to the NEO4J_PASSWORD secret. Override only for local testing.
    defaults: "{{ secret('NEO4J_PASSWORD') }}"

  - id: threshold_seconds
    type: INT
    displayName: Elapsed-time threshold in seconds
    description: A transaction whose elapsed time is at or above this is treated as
      long running.
    defaults: 120

  - id: exclude_users
    type: STRING
    displayName: Usernames never terminated (comma separated)
    description: Transactions owned by these users are reported but never
      terminated. Leave empty to protect none.
    defaults: ""

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, for any TERMINATE to run.
    defaults: true

tasks:
  - id: audit
    type: io.kestra.plugin.neo4j.Query
    description: Read-only. SHOW TRANSACTIONS is Neo4j own live transaction list,
      filtered to transactions at or over the threshold. The guard own SHOW and
      TERMINATE statements are excluded by query text and any excluded user is
      left out, so neither is ever a kill target.
    url: "{{ inputs.url }}"
    username: "{{ inputs.user }}"
    password: "{{ inputs.password }}"
    storeType: FETCH
    query: |
      SHOW TRANSACTIONS
        YIELD transactionId, elapsedTime, currentQuery, username, status
      WHERE elapsedTime.milliseconds >= ({{ inputs.threshold_seconds }} * 1000)
        AND NOT (currentQuery CONTAINS 'SHOW TRANSACTIONS' OR currentQuery CONTAINS 'TERMINATE TRANSACTIONS')
        AND NOT (username IN [u IN split('{{ inputs.exclude_users | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ',') | trim(u)])
      RETURN transactionId AS transaction_id,
             elapsedTime.milliseconds AS elapsed_ms,
             status,
             username,
             left(coalesce(currentQuery, ''), 300) AS query_text
      ORDER BY elapsed_ms DESC

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Neo4j transaction audit on {{ inputs.url }}: {{ outputs.audit.size }}
      transaction(s) at or over {{ inputs.threshold_seconds }}s. {% for r in
      outputs.audit.rows %}{{ r.transaction_id }} ({{ r.elapsed_ms }}ms, {{
      r.username }}, {{ r.status }}): {{ r.query_text }}. {% endfor %}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.audit.size > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Neo4j ' ~ inputs.url ~ ': ' ~ outputs.audit.size ~ ' transaction(s) running longer than ' ~ inputs.threshold_seconds ~ 's. Worst: ' ~ (outputs.audit.rows | jq('.[0] | .transaction_id + " at " + (.elapsed_ms|tostring) + "ms (user " + .username + ")"') | first) ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Terminating after approval.' : ' No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: approve
            type: io.kestra.plugin.core.flow.Pause
            description: Terminating a live transaction rolls back its work and frees its
              locks, so a human approves before any kill. Resumed with approved
              true to proceed, false to leave every transaction alone.
            onResume:
              - id: approved
                type: BOOL
                defaults: false

          - id: apply_gate
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approve.onResume.approved ?? false }}"
            then:
              - id: kill
                type: io.kestra.plugin.neo4j.Query
                allowFailure: true
                description: Re-list and terminate in one statement, so the filter is evaluated
                  fresh at kill time. TERMINATE returns one row per transaction,
                  including a "Transaction not found." row for any that ended on
                  its own, so this task is allowed to fail and the re-list below
                  is the authoritative gate.
                url: "{{ inputs.url }}"
                username: "{{ inputs.user }}"
                password: "{{ inputs.password }}"
                storeType: FETCH
                query: |
                  SHOW TRANSACTIONS
                    YIELD transactionId AS txid, elapsedTime AS et, currentQuery AS cq, username AS un
                  WHERE et.milliseconds >= ({{ inputs.threshold_seconds }} * 1000)
                    AND NOT (cq CONTAINS 'SHOW TRANSACTIONS' OR cq CONTAINS 'TERMINATE TRANSACTIONS')
                    AND NOT (un IN [u IN split('{{ inputs.exclude_users | replace({'\\': '\\\\'}, regexp=true) | replace({"'": "\'"}) }}', ',') | trim(u)])
                  TERMINATE TRANSACTIONS txid
                    YIELD transactionId, message
                  RETURN transactionId AS killed_id, message

              - id: settle
                type: io.kestra.plugin.core.flow.Pause
                description: A terminated transaction is marked for rollback and clears a moment
                  later, so wait briefly before the re-list.
                pauseDuration: PT5S

              - id: reaudit
                type: io.kestra.plugin.neo4j.Query
                description: Read-only. List every live transaction again so the verify step can
                  check whether any id the audit targeted is still open.
                url: "{{ inputs.url }}"
                username: "{{ inputs.user }}"
                password: "{{ inputs.password }}"
                storeType: FETCH
                query: |
                  SHOW TRANSACTIONS
                    YIELD transactionId, elapsedTime, currentQuery, status
                  RETURN transactionId AS transaction_id,
                         elapsedTime.milliseconds AS elapsed_ms,
                         status,
                         left(coalesce(currentQuery, ''), 120) AS query_text
                  ORDER BY elapsed_ms DESC

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                description: Fail if any transaction the audit targeted is still present after
                  the terminate and settle.
                condition: >-
                  {{ {'targeted': outputs.audit.rows, 'after':
                  (outputs.reaudit.rows ?? [])} |
                  jq('([.targeted[].transaction_id]) as $t |
                  ([.after[].transaction_id]) as $a | [$t[] | select(. as $x |
                  ($a | index($x)) != null)] | length') | first > 0 }}
                then:
                  - id: remediation_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: "Transactions still open after TERMINATE and settle: {{
                      {'targeted': outputs.audit.rows, 'after':
                      outputs.reaudit.rows} | jq('([.targeted[].transaction_id])
                      as $t | ([.after[].transaction_id]) as $a | [$t[] |
                      select(. as $x | ($a | index($x)) != null)] | join(\",
                      \")') | first }}. A transaction can refuse to roll back
                      while a long call runs, so check SHOW TRANSACTIONS and the
                      client that opened it."
                else:
                  - id: remediated
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Terminated the long-running transactions. {{
                      outputs.kill.size ?? 0 }} TERMINATE row(s) returned.
                      Targeted transactions still open: 0.
            else:
              - id: approval_denied
                type: io.kestra.plugin.core.log.Log
                message: "Approval not granted, no transaction was terminated. The audit list
                  stands."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "No Neo4j transaction is running longer than {{
          inputs.threshold_seconds }}s."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Neo4j long-running transaction guard failed on ' ~ inputs.url ~ ' (execution ' ~ execution.id ~ '). Transaction state unknown, check the logs.'} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, always in safe mode.
    cron: "0 3 * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true

A single transaction that runs for too long is a real operational problem in Neo4j. It holds locks that block other writers. It pins memory and page cache. A forgotten BEGIN from a client that never committed can sit open for hours. Neo4j will not end it for you. Someone has to see it and terminate it.

This blueprint lists the live transactions, keeps the ones whose elapsed time is at or over a threshold you set and reports them to Slack. When you turn remediation on and a human approves, it terminates those transactions, waits a moment, lists the transactions again and fails the run if any transaction it targeted is still open. It never terminates its own session, because the audit and kill statements are excluded by their query text and it never touches a username you choose to protect.

This blueprint was created by zkasuran.

Safety

Terminating a transaction rolls back its work and frees its locks, which is disruptive on purpose. Two independent inputs (auto_remediate and dry_run) and a manual Pause sit between a finding and any kill, so the default run only reports. The exclude_users input keeps named service or admin accounts off the kill list. The guard excludes its own SHOW TRANSACTIONS and TERMINATE TRANSACTIONS statements by query text, so it can never terminate itself.

How it works

audit (io.kestra.plugin.neo4j.Query, read-only, storeType: FETCH) runs SHOW TRANSACTIONS ... WHERE elapsedTime.milliseconds >= threshold. It drops its own statements by query text and drops any excluded username, then returns the transaction id, elapsed milliseconds, status, username and a trimmed current query, worst first.
summary logs every transaction it found.
route (io.kestra.plugin.core.flow.If on outputs.audit.size > 0):
- clean: clean logs that nothing is over the threshold.
- findings: alert posts the count and the worst offender to Slack, saying whether it is about to terminate.
  - remediate_gate runs only when auto_remediate is true and dry_run is false.
  - approve is a Pause. A human resumes it with approved true to go ahead or false to stop.
  - apply_gate runs the kill only when approved is true.
  - kill (io.kestra.plugin.neo4j.Query, allowFailure: true) re-lists and terminates in one statement, so the filter is fresh at kill time. TERMINATE TRANSACTIONS returns one row per id, so the task is allowed to fail and the re-list is the real gate.
  - settle waits five seconds for the rollbacks to clear.
  - reaudit lists the transactions again. verify_result fails the run if any id the audit targeted is still open.
errors alerts Slack when the flow itself fails, for example an unreachable instance or bad credentials, so a failed audit is never read as "nothing wrong".

What you get

A direct list of the transactions Neo4j is actually running past your threshold, from its own SHOW TRANSACTIONS, with the owning user and the current query.
A terminate step that re-checks the filter at kill time rather than trusting a stale snapshot.
A guard that cannot terminate its own session and leaves any excluded user alone.
Two inputs plus a manual approval between a finding and any kill.
Proof the fix landed: the run fails loudly if a targeted transaction is still open after the terminate, so a transaction that refuses to roll back is surfaced rather than hidden.

Who it's for

Teams running Neo4j that have had a writer blocked by one forgotten open transaction.
Platform and data engineers who want a scheduled, approved way to clear runaway or leaked transactions instead of doing it by hand at 2am.
Anyone evaluating the Neo4j plugin who wants a worked example of Query driving SHOW TRANSACTIONS and TERMINATE TRANSACTIONS with a verify step.

Why orchestrate this with Kestra

Spotting a long transaction is one query. Clearing it safely is a sequence: capture the offenders, get a human to approve, terminate, wait for the rollback, then check the result and fail if it did not take. Kestra runs that sequence on a schedule behind two gates and an approval. It keeps protected users and the guard own session out of it, alerts Slack on both the finding and a failure and keeps a history of what was terminated and where.

Tested end to end

On Kestra 2.0.5 OSS against Neo4j 5.26 Community, with a local HTTP sink standing in for Slack. A long-running transaction was produced for real by running UNWIND range(1,1000000000) AS x RETURN count(x) from a side cypher-shell session, which holds one transaction open for over two minutes. cypher-shell confirmed the transaction id and its elapsed time before and after every run.

Run	Inputs	Result
1	defaults, no long transaction	0 found, `clean` logged. No alert
2	`threshold_seconds: 3`, one long transaction	1 reported, alert sent. No change. `cypher-shell` shows it still running
3	`auto_remediate: true`, `dry_run: true`	Same report, kill held. The transaction is untouched
4	`auto_remediate: true`, `dry_run: false`, approved false	Report, `approve` resumed with false, no kill. The transaction survives
5	`auto_remediate: true`, `dry_run: false`, approved true	Terminated. `settle` waited, `reaudit` clean, `verify_result` passed. `cypher-shell` shows the transaction gone
6	unreachable URL	`audit` failed, `errors` alert sent

Exact execution ids and the before and after transaction counts are in the pull request notes.

Prerequisites

A reachable Neo4j 5.x instance. SHOW TRANSACTIONS and TERMINATE TRANSACTIONS are Cypher administration commands available on Community and Enterprise.
For the audit, a user that can run SHOW TRANSACTIONS. For remediation, a user allowed to TERMINATE TRANSACTIONS.
A Slack incoming webhook for alerts.
Local testing:
- docker run -d --name neo4j -p 7687:7687 -e NEO4J_AUTH=neo4j/testpassword neo4j:5;
- open a long transaction from another shell: cypher-shell -u neo4j -p testpassword "UNWIND range(1,1000000000) AS x RETURN count(x);" and run the flow with threshold_seconds set low.

Secrets

NEO4J_PASSWORD: password for the Neo4j user, the default for the password input.
SLACK_WEBHOOK_URL: Slack incoming webhook used by alert and the errors block.
In Kestra OSS, provide each one as a base64-encoded environment variable prefixed SECRET_, for example SECRET_NEO4J_PASSWORD=$(echo -n 'testpassword' | base64).

Inputs

Input	Default	Purpose
`url`	`bolt://localhost:7687`	Bolt endpoint
`user`	`neo4j`	Neo4j user
`password`	`{{ secret('NEO4J_PASSWORD') }}`	Password, from the secret by default
`threshold_seconds`	`120`	Elapsed time at or over which a transaction is long running
`exclude_users`	empty	Comma-separated usernames never terminated
`auto_remediate`	`false`	Allow terminate
`dry_run`	`true`	Must be `false` for any terminate to run

Quick start

Add the NEO4J_PASSWORD and SLACK_WEBHOOK_URL secrets.
Import the flow and set url and user.
Run it once with the defaults and read the transactions over the threshold.
Decide a sensible threshold_seconds for your workload, since legitimate bulk writes can be long.
Add any service accounts to exclude_users so they are never terminated.
Run it with auto_remediate: true and dry_run: false, approve the Pause, then the offenders are terminated and verified.
Enable the daily trigger, which always runs in safe mode.

How to extend

Add a max_terminations cap so a single run never kills more than N transactions.
Alert only, by routing the finding to a ticket instead of terminating, for an advisory deployment.
Narrow the audit to one database in a multi-database setup or to transactions whose current query matches a pattern.
Treat a known long batch job as healthy by adding its username to exclude_users or filtering its query text.
Replace the elapsed-time threshold with a lock-wait or idle-time signal for a tighter definition of stuck.

Neo4j Long-Running Transaction Killer Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Neo4j Long-Running Transaction Killer Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

More Data Blueprints

New to Kestra?