DEV Community: Autom Mate

Testing Ticket-Sync Boundaries: Duplicate Events, Stale Snapshots and Private Fields

Caglayan Aydin — Thu, 15 Oct 2026 17:00:00 +0000

AI disclosure: This article and its code example were generated by an AI agent for Autom Mate. The example was executed locally with the automated checks described below. It is an illustrative tutorial, not a report of a production deployment or a claim of human review.

A two-way ticket integration can fail even when every HTTP request succeeds. A delayed event can restore an old priority. A duplicated delivery can trigger repeated work. An internal note can appear in the wrong system.

Before building a complete connector, test a smaller boundary: which incoming information is allowed to update the integration's local view? A deterministic function makes that policy easier to inspect than a large workflow with several remote side effects.

This example extends Autom Mate's ticket integration test matrix with a runnable Python exercise. It filters snapshots into a local SQLite projection. It sends nothing to Jira or another application.

Write the event contract first

The example assumes that each event contains a full snapshot of the fields owned by its source, plus a strictly increasing version for that source's ticket. It uses separate namespaces for the customer tenant, source system and ticket.

Those assumptions are explicit because real webhook mechanisms differ. Atlassian documents a Jira Cloud webhook identifier that stays the same across retries and is unique within a tenant. That is useful for delivery deduplication, but it does not establish the snapshot version contract used here. See the Jira Cloud webhook documentation.

If your events contain partial patches, discarding an older patch may lose an independent field update. Use an appropriate ordering, per-field versioning or authoritative reconciliation design instead of copying this reducer unchanged.

Separate delivery, ordering and ownership

Three questions need separate answers:

Question	Local rule in this example
Have we processed this delivery before?	Check tenant, source and event identifier together
Is its snapshot newer?	Compare versions within that tenant, source and ticket
Which fields may travel?	Apply the source's explicit field allowlist
Does this prove remote delivery?	No; the example updates only a local projection

The desk owns priority and service_status. Engineering owns development_status. A field named internal_note is excluded from both projections.

“Development done” may still require deployment or customer confirmation.

Run the filter and its checks

Save this as ticket_snapshot_demo.py:

"""Local snapshot filter: no webhooks, credentials, or remote writes."""
import json
import sqlite3
import unittest

ALLOWED = {
    "desk": {"priority", "service_status"},
    "engineering": {"development_status"},
}


def connect():
    db = sqlite3.connect(":memory:")
    db.executescript("""
        CREATE TABLE seen (
            tenant TEXT, source TEXT, event_id TEXT,
            PRIMARY KEY (tenant, source, event_id));
        CREATE TABLE projection (
            tenant TEXT, source TEXT, ticket TEXT, version INTEGER, data TEXT,
            PRIMARY KEY (tenant, source, ticket));
    """)
    return db


def apply_snapshot(db, event):
    tenant, source = event["tenant"], event["source"]
    if source not in ALLOWED or type(event["version"]) is not int:
        raise ValueError("unsupported source or invalid version")
    key = (tenant, source, event["ticket"])
    with db:
        duplicate = db.execute(
            "SELECT 1 FROM seen WHERE tenant=? AND source=? AND event_id=?",
            (tenant, source, event["id"]),
        ).fetchone()
        if duplicate:
            return "duplicate"
        prior = db.execute(
            "SELECT version FROM projection WHERE tenant=? AND source=? AND ticket=?",
            key,
        ).fetchone()
        result = "stale"
        if prior is None or event["version"] > prior[0]:
            safe = {k: v for k, v in event["fields"].items() if k in ALLOWED[source]}
            db.execute("INSERT OR REPLACE INTO projection VALUES (?, ?, ?, ?, ?)",
                       (*key, event["version"], json.dumps(safe)))
            result = "applied"
        db.execute("INSERT INTO seen VALUES (?, ?, ?)",
                   (tenant, source, event["id"]))
    return result


def read(db, tenant, source, ticket):
    row = db.execute(
        "SELECT data FROM projection WHERE tenant=? AND source=? AND ticket=?",
        (tenant, source, ticket),
    ).fetchone()
    return json.loads(row[0]) if row else None


class SnapshotTests(unittest.TestCase):
    def test_duplicate_stale_and_private_fields(self):
        db = connect()
        e = dict(tenant="A", source="desk", ticket="7", id="event-2",
                 version=2, fields={"priority": "high", "internal_note": "test only"})
        self.assertEqual(apply_snapshot(db, e), "applied")
        self.assertEqual(apply_snapshot(db, e), "duplicate")
        old = dict(e, id="event-1", version=1, fields={"priority": "low"})
        self.assertEqual(apply_snapshot(db, old), "stale")
        self.assertEqual(read(db, "A", "desk", "7"), {"priority": "high"})
        db.close()

    def test_tenant_and_source_namespaces_are_separate(self):
        db = connect()
        e = dict(tenant="A", source="desk", ticket="7", id="same-id",
                 version=4, fields={"priority": "high"})
        apply_snapshot(db, e)
        other = dict(e, tenant="B", fields={"priority": "low"})
        self.assertEqual(apply_snapshot(db, other), "applied")
        eng = dict(e, source="engineering", version=1,
                   fields={"development_status": "done", "priority": "low"})
        self.assertEqual(apply_snapshot(db, eng), "applied")
        self.assertEqual(read(db, "A", "desk", "7"), {"priority": "high"})
        self.assertEqual(read(db, "B", "desk", "7"), {"priority": "low"})
        self.assertEqual(read(db, "A", "engineering", "7"),
                         {"development_status": "done"})
        db.close()


if __name__ == "__main__":
    unittest.main()

Run:

python ticket_snapshot_demo.py -v

The first test applies version two, repeats the delivery and then delivers version one. It verifies that the repeated event is recognised, the old snapshot cannot regress priority and the internal note is not stored in the projection.

The second test uses identical event and ticket identifiers across two tenants and two sources. Both namespaces must remain independent. Engineering's attempt to supply a desk-owned priority is filtered out.

Both tests passed with Python 3.12.14 during preparation. They check concrete state rather than merely asserting that a function returned without an exception.

Understand the transaction boundary

The local projection change and the “seen” record are committed together using the connection context manager. Python's sqlite3 documentation explains transaction handling and parameterised statements.

There is no remote call inside that boundary. If you later add an HTTP write after the commit, a process failure can occur between recording the event and delivering the change. If you write before the commit, the destination can accept the change while the local transaction later fails.

A production design needs a deliberate outbound delivery mechanism, such as durable queued work with destination-aware duplicate handling. This exercise does not claim exactly-once delivery across systems.

The code is also a single-consumer teaching example. It does not test concurrent transactions, database contention, persistent restarts or retention of delivery identifiers. Those require their own design and tests.

Test loops at the outbound boundary

Deduplicating an inbound event will not necessarily stop an echo loop. A remote update may create a new event identifier even though it reflects a change the bridge just sent.

Record enough outbound context to recognise that specific reflected change. Compare the relevant fields, origin and version before deciding to ignore it. Avoid dropping every event created by an integration account: that account may legitimately perform work for another service.

A useful integration test sends one allowed field change, captures the resulting destination event and verifies that processing it does not issue the same business change repeatedly. Also send a genuine new change from the same account; the loop control must not discard it blindly.

Extend the acceptance matrix beyond this sample

Before connecting real systems, add tests for:

Invalid or unauthenticated senders, including a forged tenant identifier.
Private attachments and comments, not only the sample internal-note field.
Missing events discovered through reconciliation rather than customer complaints.
A moved or deleted destination ticket.
Equal versions carrying different payloads, which this simplified example treats as stale.
A remote permission change after the local event has been accepted.

Each failed or unproven case needs a visible owner. Passing the local tests is useful evidence for the filter's stated rules, not acceptance of the entire service.

Use the wider ticket integration matrix to plan the remaining service checks.

Recovering an Ambiguous Workflow Timeout: A Python and SQLite Exercise

Caglayan Aydin — Mon, 12 Oct 2026 17:00:00 +0000

A workflow calls an application to grant access. The application commits the change, but the response never reaches the caller. The workflow reports a timeout. Running the entire sequence again is an attractive shortcut, but the error does not tell you whether the change happened.

This tutorial turns that ambiguity into a small recovery exercise. The destination is a local SQLite table, so there are no accounts, credentials or real access changes. The interesting boundary is between the destination committing a side effect and the caller receiving confirmation.

The operating context comes from Autom Mate's service desk recovery runbook. Here we focus on one operation that can be inspected independently.

Make the recovery decision before writing the retry loop

Microsoft's Retry pattern guidance distinguishes transient failures from persistent problems and warns about repeating operations with side effects. A timeout is a reason to investigate the result, not proof that the destination rejected the request.

A useful decision table is:

Destination evidence	Request still authorised?	Recovery decision
Matching action exists	Yes	Reconcile the result and continue
No matching action exists	Yes	Attempt the missing action under the original operation key
Matching action exists	No	Hold for an explicit access decision
No matching action exists	No	Hold without granting access
Key points to a different payload	Either	Stop and investigate the conflict

The last row matters. An idempotency key should not be reused to turn an approved read-only grant into administrator access. Matching only the operation identifier would hide that change.

Give the example a deliberately narrow contract

The fake destination stores one immutable grant per operation key. A repeated call with the same person and access returns the existing outcome. A call reusing that key with a different payload fails.

That is a capability of this example's destination. A local caller-side table cannot impose the same guarantee on an arbitrary remote API. Before applying the pattern elsewhere, verify whether the destination supports an appropriate idempotency key or an authoritative lookup.

The example also receives a simple authorisation boolean. A real service must derive that decision from its current policy and request state, with a design for changes that arrive during execution.

Run the exercise

Save the following as recovery_demo.py. It uses Python's standard library and creates only temporary test data.

"""Local recovery exercise. Uses only temporary SQLite data in tests."""
import sqlite3
import tempfile
import unittest
from pathlib import Path


def connect(path):
    db = sqlite3.connect(path)
    db.execute("CREATE TABLE IF NOT EXISTS grants "
               "(operation TEXT PRIMARY KEY, person TEXT, access TEXT)")
    return db


def lookup(db, operation):
    return db.execute(
        "SELECT person, access FROM grants WHERE operation = ?",
        (operation,),
    ).fetchone()


def grant(db, operation, person, access, lose_response=False):
    with db:
        db.execute("INSERT OR IGNORE INTO grants VALUES (?, ?, ?)",
                   (operation, person, access))
        if lookup(db, operation) != (person, access):
            raise ValueError("operation key reused for different payload")
    if lose_response:
        raise TimeoutError("committed, but acknowledgement was lost")


def recover(db, operation, person, access, authorised):
    actual = lookup(db, operation)
    if actual is not None and actual != (person, access):
        return "conflict"
    if not authorised:
        return "hold-existing" if actual else "hold-empty"
    if actual is not None:
        return "reconciled"
    grant(db, operation, person, access)
    return "completed"


class RecoveryTests(unittest.TestCase):
    def test_lost_response_and_restart(self):
        with tempfile.TemporaryDirectory() as folder:
            path = Path(folder) / "destination.db"
            db = connect(path)
            with self.assertRaises(TimeoutError):
                grant(db, "tenant-a:req-7:grant", "person-9", "reader", True)
            db.close()
            db = connect(path)
            self.assertEqual(
                recover(db, "tenant-a:req-7:grant", "person-9", "reader", True),
                "reconciled",
            )
            self.assertEqual(db.execute("SELECT COUNT(*) FROM grants").fetchone()[0], 1)
            db.close()

    def test_new_request_repeat_and_conflict(self):
        db = connect(":memory:")
        self.assertEqual(recover(db, "op-1", "p-1", "reader", True), "completed")
        self.assertEqual(recover(db, "op-1", "p-1", "reader", True), "reconciled")
        self.assertEqual(recover(db, "op-1", "p-1", "admin", True), "conflict")
        self.assertEqual(lookup(db, "op-1"), ("p-1", "reader"))
        db.close()

    def test_cancelled_request_does_not_grant_or_silently_revoke(self):
        db = connect(":memory:")
        self.assertEqual(recover(db, "op-1", "p-1", "reader", False), "hold-empty")
        self.assertIsNone(lookup(db, "op-1"))
        grant(db, "op-1", "p-1", "reader")
        self.assertEqual(recover(db, "op-1", "p-1", "reader", False), "hold-existing")
        self.assertEqual(lookup(db, "op-1"), ("p-1", "reader"))
        db.close()


if __name__ == "__main__":
    unittest.main()

Run:

python recovery_demo.py -v

The first test commits a grant and deliberately raises a timeout afterwards. It closes the database connection, opens a new connection to the same file and performs recovery. The assertion checks both the returned decision and the count of destination records. Recovery must recognise the completed effect without creating another grant.

The other tests exercise a new request, repeated recovery, conflicting payload reuse and a cancelled request. They passed with Python 3.12.14 during preparation. These checks demonstrate the listed local behaviours; they are not concurrency, availability or security tests.

Why cancellation produces a hold

Suppose a starter is cancelled after the destination granted access but before the workflow received confirmation. Treating that request as complete would ignore the cancellation. Deleting the grant automatically would invent a recovery policy.

The hold-existing result preserves both facts: something happened and the request is no longer authorised to continue. An accountable operator must decide the next step under the service's access policy. In a real runbook, include the evidence reference, recovery owner and next update time.

Likewise, a conflicting payload should not be “fixed” by overwriting the stored row. First determine whether the original operation key was reused incorrectly or the business request changed. Those cases need different decisions.

Keep these limits visible

This model deliberately leaves out several production concerns:

It assumes an immutable destination record; subsequent manual changes need additional reconciliation rules.
It does not coordinate multiple workers or lock the current authorisation decision across a remote action.
It does not model eventually consistent reads, rate limits, unavailable lookups or ambiguous search results.
It does not implement secret handling, customer access controls or a retained operational audit trail.

If the real destination cannot tell you whether an action occurred, return an unresolved state and give it an owner. Do not convert uncertainty into an automatic repeat merely to clear the queue.

For remote systems, also distinguish a stable business-operation key from an individual delivery attempt. A retry should retain the former while receiving its own attempt reference. That makes it possible to explain which business action several attempts were trying to complete.

Turn the result into an operator instruction

A practical instruction names the customer, original request, intended action, current authorisation and authoritative evidence source. It then identifies a permitted next step and a stop condition.

For example: inspect the grant associated with the original request; if it matches and the request remains valid, reconcile completion; if absent, execute only the missing operation through the destination's supported mechanism; if uncertain or conflicting, hold and escalate.

Keep the recovery outcome distinct from the whole service outcome. A confirmed access grant does not prove that equipment is ready or that the employee can complete their work.

The useful next step is to take one existing workflow failure and write this table for its actual destination. Use the full service desk recovery runbook as a worksheet.