Pipeline Alerting Reference Guide: AWS Lambda, DynamoDB, Teams and Jira Without Duplicate Incidents

TL;DR / Key Takeaways

• Keep one DynamoDB item per alert and change it only with conditional writes. A repeated, retried or concurrent failure event then updates the open incident instead of opening another.
• Write the notifications an incident owes into the same item, in the same write that changes its status. A crash or a Teams or Jira outage after that write delays a notification; it cannot forget one.
• Delivery across DynamoDB, Teams and Jira is at least once, not exactly once. Design for a repeated Teams message, look for a Jira issue by label before creating one, and reconcile.
• Put a TTL only on incidents that are recovered and fully delivered. An open incident that expires gets announced again as new.
• Office 365 connectors in Teams were retired in May 2026. A new integration posts to a Workflows webhook, and the old connector webhook URLs no longer work.

A pipeline alerting system that does not flood operators needs memory and an outbox: a DynamoDB item per alert that records whether an incident is open, which runs it has already seen, and which Teams and Jira actions are still owed. This guide builds that as two Lambda handlers, tests it against synthetic services that fail and crash on purpose, and marks which behavior comes from vendor documentation, checked October 12, 2026, and which is this example's own policy.

In the demo below, 9 webhook deliveries across 2 days produce 2 Jira issues and 7 Teams messages, one of them a repeat caused by a simulated crash.

If you would rather have alerting like this built and supported around your own pipelines, that is part of our Airflow and data pipeline service and our Fivetran pipeline service.

What is documented and what is not

Nothing in this guide was run against AWS, a Teams tenant, a Jira site, dbt or Fivetran. Request and payload shapes come from the vendors' public documentation. The table, the Teams webhook, the Jira API and every event in the example are synthetic.

Run the example

The example is standard-library Python 3.10 or newer. It needs no AWS account, Teams tenant, Jira site or credentials, makes no network calls, and contains no customer data: every event, ID and key is generated or a labeled placeholder.

mkdir pipeline-alerting && cd pipeline-alerting
base=https://itsmorethansoftware.com/examples/pipeline-alerting
for f in README.md events.py incident.py delivery.py adapters.py handler.py \
    reconcile.py synthetic.py demo.py test_alerting.py test_negative_controls.py; do
  curl -fsSO "$base/$f"
done
python3 -m unittest -v
python3 demo.py

$ python3 -m unittest
............................................
----------------------------------------------------------------------
Ran 44 tests in 0.559s

OK

Architecture and data flow

dbt and Fivetran both push webhooks, so the system is event-driven: nothing polls. One Lambda function behind a function URL receives the events. A second handler runs on a schedule and retries whatever the first could not deliver.

dbt job webhook ──┐
                  ├─► Lambda function URL: webhook_handler
Fivetran webhook ─┘     1 verify the HMAC signature
                        2 normalize to one signal
                        3 conditional write ──────────► DynamoDB: one item per alert
                        4 deliver what is pending          status, version, seen,
                        5 answer 200                       pending, lease, expires_at
                              │                                  ▲
                              ├─► Teams Workflows webhook        │
                              └─► Jira Cloud REST API v3         │
                                        ▲                        │
schedule ─► Lambda: sweep_handler ──────┴────────────────────────┘
              retry pending actions, report parked ones

The order of steps 3 to 5 is the design. The state write comes first, and it is the only step that can fail the webhook. If it raises, the sender gets an error and redelivers: Fivetran retries for up to 24 hours and dbt five times, per the pages above. Once the write succeeds, the handler answers 200 whatever Teams and Jira do, because the work they are owed is already in the item.

That matters for the time budget too. Both senders give the endpoint 10 seconds. The adapters use timeouts of 5 and 8 seconds as illustrative values; in production, size them so that one pass fits inside the sender's timeout, or return after step 3 and leave delivery to the sweep.

The events

This is the failure event the demo sends first. It is synthetic: built by synthetic.py with the field names of dbt's documented job.run.errored payload and generated values.

{
  "accountId": 1,
  "data": {
    "environmentId": "40",
    "environmentName": "Production",
    "jobId": "301",
    "jobName": "Daily build",
    "projectId": "50",
    "projectName": "Synthetic Analytics",
    "runErroredAt": "2026-10-12T14:05:20Z",
    "runId": "7001",
    "runReason": "scheduled",
    "runStatus": "Errored",
    "runStatusCode": 20,
    "runStatusMessage": "None"
  },
  "eventId": "wev_synthetic_7001_errored",
  "eventType": "job.run.errored",
  "timestamp": "2026-10-12T14:05:20Z",
  "webhookId": "wsu_synthetic",
  "webhookName": "pipeline-alerting"
}

And the event that ends the incident, a job.run.completed with a status of Success:

{
  "accountId": 1,
  "data": {
    "environmentId": "40",
    "environmentName": "Production",
    "jobId": "301",
    "jobName": "Daily build",
    "projectId": "50",
    "projectName": "Synthetic Analytics",
    "runFinishedAt": "2026-10-12T15:05:18Z",
    "runId": "7002",
    "runReason": "scheduled",
    "runStatus": "Success",
    "runStatusCode": 10,
    "runStatusMessage": "None"
  },
  "eventId": "wev_synthetic_7002_completed",
  "eventType": "job.run.completed",
  "timestamp": "2026-10-12T15:05:18Z",
  "webhookId": "wsu_synthetic",
  "webhookName": "pipeline-alerting"
}

Both vendors sign the raw body with HMAC-SHA256, so the handler verifies the bytes it received before it parses them, and compares in constant time:

def verify_signature(secret, body, received):
    """Both vendors sign the raw body with HMAC-SHA256. Compare in constant time."""
    if not secret or not received:
        return False
    expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected.lower(), received.strip().lower())

A function URL that third parties call has to use the NONE auth type, because the alternative requires every request to be signed with AWS credentials, which dbt and Fivetran cannot do. The signature check is the only gate, so an event that fails it gets a 401 and touches nothing.

The normalizer then reduces either vendor's payload to one signal: source, environment, resource, a kind of failure or healthy, the run ID, and the time the run finished. Three decisions are worth copying:

• One run and outcome is one signal. dbt documents that a failed run fires both job.run.errored and job.run.completed. The signal's dedupe ID is dbt:7001:failure for both, so the second is recognized as a duplicate whichever arrives first.
• Unknown is not healthy. Fivetran documents SUCCESSFUL for a sync_end. The example treats any other status as a failure and keeps the value for the operator, instead of guessing at failure names Fivetran's page does not list.
• Use the event's time, not the arrival time. Retries arrive late and out of order. Every ordering decision below compares the times in the payloads.

State model and idempotency

There is one item per alert identity. The identity is configurable: by default it is source, environment and resource, which gives a key such as dbt/Production/301. Dropping the environment from the identity merges a job's environments into one alert; the suite tests both.

This is the item after the first failure, at a moment when Teams has been notified and Jira is unavailable:

{
  "alert_key": "dbt/Production/301",
  "attempts": 1,
  "delivered": {
    "teams_opened": 1
  },
  "detail": "Run 7001 finished with status Errored",
  "environment": "Production",
  "escalated": false,
  "failure_count": 1,
  "incident_id": "dbt/Production/301@2026-10-12T14:05:20Z",
  "last_failure_at": "2026-10-12T14:05:20Z",
  "opened_at": "2026-10-12T14:05:20Z",
  "pending": [
    "jira_create"
  ],
  "reopen_count": 0,
  "resource": "301",
  "resource_name": "Daily build",
  "retry_at": 1791813939,
  "seen": [
    "dbt:7001:failure"
  ],
  "source": "dbt",
  "status": "open",
  "version": 4
}

Each event is applied by reading the item, deciding in a pure function, and writing on the condition that nothing changed in between:

The 30-minute reopen window and the escalation at three failed runs are configuration, not vendor behavior. The branch for a failure that arrives after a recovery shows how the rules read in code:

    if signal.kind == FAILURE:
        if signal.occurred_at <= item["recovered_at"]:
            return "stale", None
        quiet = epoch(signal.occurred_at) - epoch(item["recovered_at"])
        if is_closed(item) and quiet > config.reopen_window_s:
            return "opened", new_incident(key, signal)
        return "reopened", {
            "status": OPEN,
            "reopen_count": item["reopen_count"] + 1,
            "failure_count": item["failure_count"] + 1,
            "last_failure_at": signal.occurred_at,
            "detail": signal.detail,
            "seen": seen,
            "pending": (pending - {TEAMS_RECOVERED}) | {TEAMS_OPENED, JIRA_COMMENT},
            "recovered_at": REMOVE,
            "expires_at": REMOVE,
        }

    if item["status"] == RECOVERED:
        return "healthy", None
    if signal.occurred_at <= item["last_failure_at"]:
        return "stale", None
    return "recovered", {
        "status": RECOVERED,
        "recovered_at": signal.occurred_at,
        "seen": seen,
        "pending": (pending - {TEAMS_ESCALATED}) | {TEAMS_RECOVERED, JIRA_COMMENT},
    }

A new incident replaces the old one only when the old one is closed. If a recovery notification is still pending, a new failure continues the same incident, so nothing owed is overwritten.

The conditional writes

The first failure creates the item with a put that succeeds only if no item has that key. This is the request the example sent, in the form a boto3 Table resource takes:

{
  "ConditionExpression": "attribute_not_exists(#alert_key)",
  "ExpressionAttributeNames": {
    "#alert_key": "alert_key"
  },
  "Item": {
    "alert_key": "dbt/Production/301",
    "detail": "Run 7001 finished with status Errored",
    "environment": "Production",
    "escalated": false,
    "failure_count": 1,
    "incident_id": "dbt/Production/301@2026-10-12T14:05:20Z",
    "last_failure_at": "2026-10-12T14:05:20Z",
    "opened_at": "2026-10-12T14:05:20Z",
    "pending": [
      "jira_create",
      "teams_opened"
    ],
    "reopen_count": 0,
    "resource": "301",
    "resource_name": "Daily build",
    "seen": [
      "dbt:7001:failure"
    ],
    "source": "dbt",
    "status": "open",
    "version": 1
  }
}

If two invocations process the first failure at once, DynamoDB accepts one put and rejects the other, per its condition expression documentation. The loser reads again, finds the incident, and takes the repeat or duplicate path.

Every later write is an update guarded by the version. This is the recovery:

{
  "ConditionExpression": "#version = :expected",
  "ExpressionAttributeNames": {
    "#pending": "pending",
    "#recovered_at": "recovered_at",
    "#seen": "seen",
    "#status": "status",
    "#version": "version"
  },
  "ExpressionAttributeValues": {
    ":expected": 7,
    ":pending": [
      "jira_comment",
      "teams_recovered"
    ],
    ":recovered_at": "2026-10-12T15:05:18Z",
    ":seen": [
      "dbt:7001:failure",
      "dbt:7002:healthy"
    ],
    ":status": "recovered",
    ":version": 8
  },
  "Key": {
    "alert_key": "dbt/Production/301"
  },
  "ReturnValues": "ALL_NEW",
  "UpdateExpression": "SET #pending = :pending, #recovered_at = :recovered_at, #seen = :seen, #status = :status, #version = :version"
}

The builder generates a placeholder for every attribute name, because status and source are on DynamoDB's reserved word list:

def update_request(key, expected_version, changes):
    """Apply changes only if the item is still at the version that was read."""
    changes = {**changes, "version": expected_version + 1}
    sets = sorted(name for name, value in changes.items() if value is not REMOVE and value != set())
    removes = sorted(name for name in changes if name not in sets)
    expression = "SET " + ", ".join(f"#{name} = :{name}" for name in sets)
    if removes:
        expression += " REMOVE " + ", ".join(f"#{name}" for name in removes)
    return {
        "Key": {KEY: key},
        "UpdateExpression": expression,
        "ConditionExpression": "#version = :expected",
        # Placeholders for every name: status, source and others are reserved words.
        "ExpressionAttributeNames": {f"#{name}": name for name in sorted(changes)},
        "ExpressionAttributeValues": {":expected": expected_version, **{f":{name}": changes[name] for name in sets}},
        "ReturnValues": "ALL_NEW",
    }

def change(table, key, decide, attempts=5):
    """Read, decide, write on the condition that nothing changed in between."""
    for _ in range(attempts):
        item = table.get_item(Key={KEY: key}, ConsistentRead=True).get("Item")
        outcome, changes = decide(item)
        if changes is None:
            return outcome, item
        try:
            if item is None:
                created = {KEY: key, "version": 1, **{k: v for k, v in changes.items() if v is not REMOVE and v != set()}}
                table.put_item(**put_request(created))
                return outcome, created
            return outcome, table.update_item(**update_request(key, item["version"], changes))["Attributes"]
        except Exception as exc:
            if not is_conditional_failure(exc):
                raise
    raise Conflict(f"{key}: no stable read in {attempts} attempts")

Three details in those two functions carry weight:

• Read with ConsistentRead. DynamoDB reads are eventually consistent by default. A stale read would not corrupt anything, because the version condition rejects the write, but it costs a retry every time.
• The condition also protects against TTL. UpdateItem creates an item that does not exist. Without a condition, an update that lands just after TTL deleted the item would create a fragment with a status and no history. With #version = :expected it is rejected, and the suite tests that.
• An empty set is removed, not stored. pending and failed are string sets, and the builder turns an empty one into a REMOVE. "Nothing owed" is then simply the absence of the attribute.

The Lambda handlers

The webhook handler's last steps are the ones that define the failure behavior:

        # If this raises, Lambda returns an error and the sender retries.
        outcome, item, key = record(deps.table, signal, deps.config)
        deps.log("signal_recorded", alert_key=key, outcome=outcome, kind=signal.kind, run_id=signal.run_id, incident_id=(item or {}).get("incident_id"))
        try:
            deliver(deps, key, worker)
        except Exception as exc:  # the state is saved; the sweep will retry
            deps.log("delivery_error", level="ERROR", alert_key=key, error=f"{type(exc).__name__}: {exc}")
        return respond(200, result=outcome, alert_key=key)

It answers 401 for a bad signature, 404 for an unknown source, 400 for an authentic payload it cannot use, and 200 with ignored for events it does not act on, such as a run that started. None of those four touch the table.

The sweep handler scans for items with pending and calls the same deliver function. Schedule it every minute or so; that interval is the longest a retry waits beyond its retry_at. The example's sweep uses scan, which is fine for a table of a few hundred alerts. For a larger table, add an attribute that exists only while work is pending and query a sparse global secondary index on it. Index reads are eventually consistent, which is safe here: the index only finds keys, and deliver reads each item consistently before it acts.

A scheduled invocation is asynchronous, and AWS documents that an asynchronous event can reach a function more than once. Two sweeps running at once is therefore a normal case, and the lease in the next section is what keeps them from both posting.

Teams and Jira adapter contracts

The adapters build requests and classify answers. They do not decide what to send or when.

def call(transport, request, timeout):
    """Send once. A timeout is retryable, and the request may still have landed."""
    try:
        status, headers, body = transport.send(request, timeout=timeout)
    except (TimeoutError, ConnectionError) as exc:
        raise Retryable(f"{type(exc).__name__}: {exc}") from exc
    if 200 <= status < 300:
        return json.loads(body) if body else None
    if status in RETRYABLE_STATUS:
        raise Retryable(f"HTTP {status}", retry_after_seconds(headers))
    raise Permanent(f"HTTP {status}: {body[:200]}")

Teams

Microsoft retired Office 365 connectors in Teams in 2026. Its developer blog gives the schedule: the rollout began May 18, 2026 and completed May 22, 2026, and "After these dates, Office 365 Connectors will no longer function." The replacement is a workflow in the Workflows app, built on Power Automate, created from a template such as "Send webhook alerts to a channel" (Microsoft Support, Microsoft Learn).

The webhook trigger's documented body is a message with an array of Adaptive Card attachments. This is the request the example sent for the first failure:

{
  "attachments": [
    {
      "content": {
        "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
        "body": [
          {
            "text": "Pipeline failure: dbt Daily build (Production)",
            "type": "TextBlock",
            "wrap": true
          },
          {
            "text": "Incident: dbt/Production/301@2026-10-12T14:05:20Z",
            "type": "TextBlock",
            "wrap": true
          },
          {
            "text": "Failed runs: 1, first at 2026-10-12T14:05:20Z, latest at 2026-10-12T14:05:20Z",
            "type": "TextBlock",
            "wrap": true
          },
          {
            "text": "Run 7001 finished with status Errored",
            "type": "TextBlock",
            "wrap": true
          }
        ],
        "type": "AdaptiveCard",
        "version": "1.2"
      },
      "contentType": "application/vnd.microsoft.card.adaptive",
      "contentUrl": null
    }
  ],
  "type": "message"
}

What to know before pointing this at a tenant:

• The URL is the credential when the workflow allows anyone to trigger it. The connector reference documents three choices: anyone, any user in the tenant, or specific users. The last two require a token on the request, and the reference says not to send one when the choice is anyone.
• A workflow belongs to a person. Microsoft Learn notes that workflows are linked to their owners, not to the team or channel, and can be orphaned when the owner leaves. Add a co-owner.
• A 2xx is not proof of a post. The pages read for this guide do not state what the trigger returns or tie it to the card appearing. The adapter treats 2xx as accepted by the workflow. Check the flow's run history when you integrate, and note that Microsoft Learn puts the message size limit at 28 KB.
• There is no way to ask whether a card was posted. A repeated card cannot be prevented from this side. Every card carries the incident ID on its second line so a repeat is recognizable.

Jira

The example uses three Jira Cloud REST API v3 operations: create issue, JQL search and add comment. The description and comments are Atlassian Document Format, which v3 requires for rich text. This is the create request for the first incident, with placeholder project and issue type IDs:

{
  "fields": {
    "description": {
      "content": [
        {
          "content": [
            {
              "text": "dbt Daily build (Production) failed at 2026-10-12T14:05:20Z.",
              "type": "text"
            }
          ],
          "type": "paragraph"
        },
        {
          "content": [
            {
              "text": "Run 7001 finished with status Errored",
              "type": "text"
            }
          ],
          "type": "paragraph"
        },
        {
          "content": [
            {
              "text": "Incident: dbt/Production/301@2026-10-12T14:05:20Z",
              "type": "text"
            }
          ],
          "type": "paragraph"
        },
        {
          "content": [
            {
              "text": "Opened automatically. Recovery and repeat failures are added as comments.",
              "type": "text"
            }
          ],
          "type": "paragraph"
        }
      ],
      "type": "doc",
      "version": 1
    },
    "issuetype": {
      "id": "10001"
    },
    "labels": [
      "pipeline-alert",
      "alert-f985b7f10670518e"
    ],
    "project": {
      "id": "10000"
    },
    "summary": "Pipeline failure: dbt Daily build (Production)"
  }
}

Atlassian documents no idempotency key for creating an issue, so the second label is the example's substitute: a hash of the incident ID. Before creating, the adapter searches for that label and adopts the issue if it exists.

    if action == JIRA_CREATE:
        label = incident_label(item)
        found = deps.jira.find_issues(label)
        if len(found) > 1:
            deps.log("jira_duplicate_issues", level="ERROR", alert_key=item[KEY], issues=found)
        if found:
            deps.log("jira_issue_adopted", alert_key=item[KEY], jira_key=found[0])
            return {"jira_key": found[0]}
        return {"jira_key": deps.jira.create_issue(f"Pipeline failure: {subject(item)}", issue_paragraphs(item), [BASE_LABEL, label])}

That closes the common cases: a timeout after Jira created the issue, and a crash before the key was saved. It does not close all of them, because Atlassian's search documentation says recent updates might not be immediately visible. A retry that runs before the new issue is searchable creates a second one. The suite reproduces that with a synthetic search lag, and the reconciliation reports it.

Recovery adds a comment and leaves the issue's status alone. Transition IDs depend on each project's workflow, so closing the issue automatically is a decision for the team that owns the board.

Retry and recovery logic

deliver is the only code that calls Teams or Jira. It claims the incident, attempts each pending action once, and records each result with its own conditional write.

    def claim(item):
        if item is None or not item.get("pending"):
            return "idle", None
        if item.get("retry_at", 0) > now:
            return "waiting", None
        if item.get("lease_until", 0) > now and item.get("lease_owner") != worker:
            return "busy", None
        return "claimed", {"lease_owner": worker, "lease_until": now + config.lease_s}

    for action in ACTION_ORDER:
        if action not in item.get("pending", ()):
            continue
        if action == JIRA_COMMENT and JIRA_CREATE in item["pending"]:
            continue  # the comment needs the issue; it stays pending
        done = {"pending": set(item["pending"]) - {action}}
        try:
            result = perform(deps, action, item)
            if result is not None:
                # Count what was actually sent, so reconciliation can compare.
                done.update(result, delivered={**item.get("delivered", {}), action: item.get("delivered", {}).get(action, 0) + 1})
        except Retryable as exc:
            retried.add(action)
            wait_at_least = max(wait_at_least, exc.retry_after or 0)
            deps.log("action_failed", level="WARNING", alert_key=key, action=action, error=str(exc))
            continue
        except Permanent as exc:
            done["failed"] = set(item.get("failed", ())) | {action}
            deps.log("action_parked", level="ERROR", alert_key=key, action=action, error=str(exc))
        else:
            if result is not None:
                deps.log("action_delivered", alert_key=key, incident_id=item["incident_id"], action=action, **result)
        try:
            item = deps.table.update_item(**update_request(key, item["version"], done))["Attributes"]
        except Exception as exc:
            if not is_conditional_failure(exc):
                raise
            # The incident changed while the action was in flight. The action
            # stays pending and will be repeated, which is the safe direction.
            deps.log("delivery_interrupted", level="WARNING", alert_key=key, action=action)
            break

The rules that follow from that loop:

• Perform, then mark. A crash between the two repeats the action. Marking first would lose it, which the fourth negative control demonstrates.
• One failing service does not block the other. A retryable Teams failure leaves teams_opened pending and the loop moves on to Jira.
• No sleeping inside Lambda. A failed pass sets retry_at using exponential backoff from 30 seconds to a 15-minute cap, with the upper half randomized, and never less than a Retry-After header asks for. Atlassian's rate limiting page says to treat Retry-After as the minimum delay.
• Retries are bounded. After 6 failed passes, or one permanent failure such as a 403, the action moves from pending to failed and an ERROR line is logged. A parked action is never retried on its own. After fixing the cause, requeue puts it back.
• A lease, not a lock. The lease lasts 60 seconds. If the invocation holding it dies, the next sweep after expiry takes over.
• A changed incident wins. If an event changes the item while a request is in flight, the mark is rejected, the action stays pending, and it is sent again from current state.
• The comment waits for the issue. If Jira is down from failure to recovery, the recovery comment stays pending until the issue exists.

No part of this is exactly once. The state write is atomic and conditional. The calls to Teams and Jira are separate systems with no shared transaction, and the design chooses a repeat over a loss every time it has to choose.

Failure and recovery timeline

python3 demo.py runs one dbt job through two days. Jira's search returns HTTP 503 for the first failure:

2026-10-12
14:05:20  dbt run 7001 job.run.errored (Errored)  ->  opened, HTTP 200
            teams_opened: delivered
            jira_create: failed, HTTP 503
            delivery: retry 1 in 19s, pending jira_create
14:05:26  dbt run 7001 job.run.completed (Errored)  ->  duplicate, HTTP 200
            delivery skipped: waiting
14:06:00  scheduled sweep
            jira_create: delivered (DATA-1)

The webhook answered 200 while Jira was down. dbt's second notification for the same run was a duplicate, and delivery was still waiting out its backoff. The sweep created the issue. Two more failed runs follow:

15:05:18  dbt run 7002 job.run.completed (Errored)  ->  repeat, HTTP 200
16:05:22  dbt run 7003 job.run.completed (Errored)  ->  escalated, HTTP 200
            teams_escalated: delivered

The second failure wrote to the item and notified nobody. The third posted one escalation. Then the job succeeds, and the process is killed after Teams accepted the recovery card and before the result was saved:

17:05:09  dbt run 7004 job.run.completed (Success)  ->  recorded, then process died after post was accepted
17:05:39  dbt run 7004 job.run.completed (Success)  ->  duplicate, HTTP 200 (the sender's retry)
            delivery skipped: busy
17:07:00  scheduled sweep
            teams_recovered: delivered
            jira_comment: delivered

dbt saw no response and sent the event again. It was a duplicate, and the dead invocation still held the lease. Once the lease expired, the sweep sent the recovery card a second time and added the Jira comment. The job then fails again within 30 minutes and recovers:

17:20:31  dbt run 7005 job.run.completed (Errored)  ->  reopened, HTTP 200
            teams_opened: delivered
            jira_comment: delivered
17:50:12  dbt run 7006 job.run.completed (Success)  ->  recovered, HTTP 200
            teams_recovered: delivered
            jira_comment: delivered

reconciliation after the first incident
  state: dbt/Production/301@2026-10-12T14:05:20Z is recovered, 4 failed run(s), Jira DATA-1
  received: 6 Teams message(s); Jira DATA-1 with 3 comment(s)
  table writes accepted: 28, rejected by a condition: 0
  incidents checked: 1
  WARNING dbt/Production/301: repeated notification: teams_recovered marked delivered 2 time(s), Teams received 3
  result: OK (0 errors, 1 warnings)

The reopened failure reused incident dbt/Production/301@2026-10-12T14:05:20Z and issue DATA-1. The reconciliation found the repeat from the crash and reported it as a warning. The next morning's failure is outside the window:

2026-10-13
09:05:27  dbt run 7007 job.run.completed (Errored)  ->  opened, HTTP 200
            teams_opened: delivered
            jira_create: delivered (DATA-2)

reconciliation after the second incident opens
  state: dbt/Production/301@2026-10-13T09:05:27Z is open, 1 failed run(s), Jira DATA-2
  received: 7 Teams message(s); Jira DATA-1 with 3 comment(s), DATA-2 with 0 comment(s)
  table writes accepted: 33, rejected by a condition: 0
  incidents checked: 1
  result: OK (0 errors, 0 warnings)

Those are counts from synthetic services on a fake clock. They are not timings.

TTL and incident lifecycle

An incident's item gets expires_at in exactly one place: when it is recovered and both pending and failed are empty. The default retention is 30 days.

• An open incident never expires. If it did, the next failure of the same outage would find no item and open a second incident with a second Jira issue. The sixth negative control does this.
• Expiry is not recovery. Nothing is sent when an item is deleted. A recovery is recorded only when a healthy event arrives.
• Reopening removes the TTL in the same write that sets the status back to open.
• Deletion is late and that is fine. AWS documents that TTL deletes an expired item within a few days and that it stays readable until then. A recovered item that lingers changes nothing: a new failure is judged against its recovered_at either way.
• Keep retention longer than the senders' retries. A redelivered failure from a closed incident is ignored because its run is in seen. Once the item is gone, so is that memory. Fivetran retries for up to 24 hours, so 30 days is far more than needed.

The item holds the current incident only. When a new incident opens for the same alert, the previous one's history lives in its Jira issue and in the logs.

Tests and negative controls

The suite has 44 tests and runs in under a second, all against the synthetic services. What it covers:

The reconciliation is itself a check you can run in production. Each item counts the actions it marked delivered; reconcile.py compares those counts with what the services received. A lost notification or a second Jira issue for one incident is an error. A repeated card or comment is a warning. Against real services, Jira can be searched by the incident label, and the Teams side needs the flow's run history, because a webhook cannot be read back.

A test that has never failed proves little, so test_negative_controls.py swaps in six plausible bugs, reruns the named tests above, and asserts that they fail:

I also removed 18 safeguards from a copy of the example, one at a time, including the seen check, both stale-event checks, the version condition, the lease, the retry_at wait, the retry bound, both halves of the TTL condition and the signature check. The suite failed every time. The first pass missed one: nothing failed when the recovery comment no longer waited for the issue, so that test was added.

Structured logging and observability

Every log line is one JSON object with time, level and event, plus the alert key and whatever the event needs. These are the first lines of the demo:

{"alert_key": "dbt/Production/301", "event": "signal_recorded", "incident_id": "dbt/Production/301@2026-10-12T14:05:20Z", "kind": "failure", "level": "INFO", "outcome": "opened", "run_id": "7001", "time": "2026-10-12T14:05:20Z"}
{"action": "teams_opened", "alert_key": "dbt/Production/301", "event": "action_delivered", "incident_id": "dbt/Production/301@2026-10-12T14:05:20Z", "level": "INFO", "time": "2026-10-12T14:05:20Z"}
{"action": "jira_create", "alert_key": "dbt/Production/301", "error": "HTTP 503", "event": "action_failed", "level": "WARNING", "time": "2026-10-12T14:05:20Z"}
{"alert_key": "dbt/Production/301", "event": "delivery_settled", "level": "INFO", "pending": ["jira_create"], "result": "retry 1 in 19s", "time": "2026-10-12T14:05:20Z"}
{"action": "jira_create", "alert_key": "dbt/Production/301", "event": "action_delivered", "incident_id": "dbt/Production/301@2026-10-12T14:05:20Z", "jira_key": "DATA-1", "level": "INFO", "time": "2026-10-12T14:06:00Z"}

The lines never contain a payload, a header, a signing key, the Teams URL or the Jira credential, and a test asserts that. The failure text copied from the vendor is cut to 300 characters before it is stored or sent.

What to alarm on:

Operational runbook

If a production pipeline is already alerting badly, or not at all, and the cause is not obvious, that is what Pipeline Rescue is for: a fixed-scope diagnosis of the pipeline you already have.

Security and permissions

• Verify before parsing. The function URL is public. Check the signature over the raw body, and reject anything unsigned.
• Keep four secrets in a secret store: each sender's signing key, the Teams workflow URL, and the Jira credential. The adapters take them as callables so they are read at send time and never held in configuration or logs.
• Give the function one table. It needs GetItem, PutItem and UpdateItem on the alert table, and Scan or a Query on the sweep index. Nothing else in DynamoDB.
• Use a dedicated Jira account. Atlassian documents the permissions per operation: Browse projects and Create issues to create, Add comments to comment. Its basic auth page builds the header from an email and an API token and calls basic auth less secure than OAuth for integrations, so treat the token accordingly and scope the account to the one project.
• Restrict who can trigger the workflow if your sender can attach a token. Otherwise treat the URL like a password and rotate it by creating a new workflow.
• Do not forward raw payloads. Vendor error text can contain hostnames, table names or query fragments. Send the summary and link to the vendor's own run page from the ticket.

Known limitations

• Synthetic services. The table evaluates only the expressions the example writes. Run the suite against a test table before relying on it, and send one real card and one real issue before trusting the adapters.
• At least once. A repeated Teams card or Jira comment is possible after a crash or a concurrent change. A second Jira issue is possible when search lags a retry.
• Fivetran failures are assumed. The failure status, reason field and signature header name are not in Fivetran's webhook page. Confirm all three with a test delivery.
• Silence is invisible. This design reacts to events. A job that stops running, or a webhook that was deactivated, sends nothing, so nothing alerts. Pair it with a freshness check on the data.
• A healthy event is the only way an incident closes. A deleted job leaves its incident open until someone closes it.
• Second-level ordering. Timestamps are compared in whole seconds. A healthy run that finishes in the same second as a failure is treated as stale.
• One incident per alert. Two different failures of the same job are one incident. Add the failing step to the identity if you need them apart.
• Bounded duplicate memory. An incident remembers 100 run IDs. Beyond that, a redelivered event is counted as another failure.
• No performance numbers. The clock is fake. The timeouts, backoff, lease, window and retention are starting points to tune.

Where this fits

Alerting tells people a pipeline failed. It does not make the pipeline safe to rerun. For that side, the Fivetran Connector SDK Production Guide: State, Retries and Restart Safety covers checkpoints and restart tests for custom connectors, and two extraction guides apply the same method to specific APIs: the Workday SOAP API Extraction Guide: Pagination, Throttling, Retries and Restart Safety and the CxAlloy API Extraction Guide: Request Signing, Nested Pagination, Rate Limits and Reconciliation. A connector built that way fails less often, and when it does fail, the alert from this guide and the checkpoint from those are what make the recovery short.

Related reading

• Fivetran Connector SDK Production Guide: State, Retries and Restart Safety
• Workday SOAP API Extraction Guide: Pagination, Throttling, Retries and Restart Safety
• CxAlloy API Extraction Guide: Request Signing, Nested Pagination, Rate Limits and Reconciliation
• Why your reports keep breaking

Sources

Vendor documentation, checked October 12, 2026:

• dbt: Webhooks for your jobs
• Fivetran: Webhooks
• AWS Lambda: Invoking function URLs
• AWS Lambda: Errors and retries with asynchronous invocation
• Amazon DynamoDB: Condition expressions
• Amazon DynamoDB: UpdateItem
• Amazon DynamoDB: Time to live
• Amazon DynamoDB: Read consistency
• Amazon DynamoDB: Sparse indexes
• Amazon DynamoDB: Reserved words
• Microsoft 365 Developer Blog: Retirement of Office 365 connectors within Microsoft Teams
• Microsoft Learn: Create an Incoming Webhook
• Microsoft Learn: Microsoft Teams connector reference
• Microsoft Support: Send messages in Teams using incoming webhooks
• Atlassian: Jira Cloud REST API v3, Issues
• Atlassian: Jira Cloud REST API v3, Issue search
• Atlassian: Jira Cloud REST API v3, Issue comments
• Atlassian: Rate limiting
• Atlassian: Basic auth for REST APIs