Energy Settlement Reconciliation Automation Hub

Designing idempotent ETRM sync jobs

A sync job times out after it has already written 4,000 of 5,000 trades, the scheduler retries the whole batch, and now half the book is posted twice — the same execution counted as two positions, the same MWh billed on two settlement lines. That is the failure mode an idempotent sync job removes: a re-run, a redelivered file, or an at-least-once queue replay must converge to exactly one row per trade version, never two. This guide sits under ETRM System Architecture, and it focuses narrowly on the write path — the keying, the upsert, and the retry envelope that let a job crash at any point and be re-run to a clean, identical result. The transport that pulls those batches is covered in ETRM API Integration Patterns; here we assume the bytes arrived and ask only how to persist them safely.

The diagram traces the write path each incoming trade record follows: derive a deterministic key, check whether that exact version already landed, and upsert so a duplicate is absorbed instead of appended.

Idempotent write path for an ETRM sync job A pulled trade batch flows into key derivation, which computes a deterministic idempotency key per trade version. A dedup check compares the key against the ledger: already-seen versions branch down to a no-op sink, while new or changed versions flow to an upsert-on-key step that writes to the append-only settlement ledger. A retry arrow loops the upsert back to itself to show that re-running is safe. Pull batchraw trade rows Derive keySHA-256 per version Dedup checkseen this key? Upsert on keyinsert or replace Ledgerappend-only No-opalready applied duplicate retry-safe

An idempotent operation is one where applying it once and applying it many times leave the store in the same state. For \(n \geq 1\) redeliveries of the same trade version, the invariant is \(f(f(x)) = f(x)\): the second and later writes must be observationally inert. Everything below is in service of that one property.

Prerequisites

  • Python packages: the standard library hashlib, json, and decimal cover key derivation and money math; sqlalchemy (2.x) drives the upsert against Postgres or SQLite. No third-party hashing library is needed.
  • A store with an atomic upsert: Postgres INSERT ... ON CONFLICT, SQLite INSERT ... ON CONFLICT, or an equivalent MERGE. The idempotency key must back a UNIQUE constraint or primary key, otherwise concurrent workers can both insert the same version.
  • Data dependencies: each incoming record must expose the natural business identifiers — trade_id, a source version or amendment sequence, and the economic fields (node_id, settlement_interval, mwh, price). If the source cannot supply a monotonic version, capture its last-modified timestamp instead.
  • Permissions: write scope to the ledger schema and read scope on the source API. Run the job under a service principal governed by the platform’s Security & Access Boundaries so the write credential cannot also mutate reference data.

Implementation

The whole design rests on one decision: what goes into the idempotency key. It must be deterministic (the same logical trade version always hashes to the same key) and content-sensitive at the version level (an amendment that changes economics must produce a new key, so the upsert replaces the prior version rather than colliding with it). The key is derived from the stable business identity plus the source version — not from ingest time, not from a row counter, both of which change on replay and would defeat deduplication.

import hashlib
import json
from decimal import Decimal
from datetime import datetime

def idempotency_key(trade: dict) -> str:
    """Deterministic key for one trade *version*.

    Only the business identity and version go in. Ingest-time fields
    (run_id, received_at, batch offset) are deliberately excluded so a
    replayed batch hashes identically to the original.
    """
    identity = {
        "trade_id": str(trade["trade_id"]),
        "version": int(trade["version"]),        # amendment sequence from source
        "node_id": str(trade["node_id"]),
        "settlement_interval": trade["settlement_interval"],  # ISO-8601 UTC string
    }
    # sort_keys makes the serialization order-stable across dict insertion order.
    canonical = json.dumps(identity, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

With the key defined, the write becomes an upsert keyed on it. The ledger table carries the key as a unique column; the database resolves the race between a retry and the original write, so no application-level lock is required.

from decimal import Decimal, ROUND_HALF_EVEN
from sqlalchemy import (
    create_engine, MetaData, Table, Column, String, DateTime, Numeric, text,
)
from sqlalchemy.dialects.postgresql import insert as pg_insert

CENT = Decimal("0.01")
engine = create_engine("postgresql+psycopg://svc_settle@db/etrm")
metadata = MetaData()

trade_ledger = Table(
    "trade_ledger", metadata,
    Column("idem_key", String(64), primary_key=True),   # SHA-256 hex, UNIQUE
    Column("trade_id", String, nullable=False),
    Column("version", String, nullable=False),
    Column("node_id", String, nullable=False),
    Column("settlement_interval", DateTime(timezone=True), nullable=False),
    Column("charge_usd", Numeric(18, 2), nullable=False),  # Decimal-exact money
    Column("run_id", String, nullable=False),
    Column("posted_at", DateTime(timezone=True), nullable=False),
)

def upsert_trade(conn, trade: dict, run_id: str) -> bool:
    """Insert a trade version once. A redelivery is absorbed, not duplicated.

    Returns True when this call actually wrote (or advanced) a row.
    """
    key = idempotency_key(trade)
    mwh = Decimal(str(trade["mwh"]))
    price = Decimal(str(trade["price"]))          # $/MWh, never float
    charge = (mwh * price).quantize(CENT, rounding=ROUND_HALF_EVEN)

    stmt = pg_insert(trade_ledger).values(
        idem_key=key,
        trade_id=str(trade["trade_id"]),
        version=str(trade["version"]),
        node_id=str(trade["node_id"]),
        settlement_interval=trade["settlement_interval"],
        charge_usd=charge,
        run_id=run_id,
        posted_at=datetime.now().astimezone(),
    )
    # ON CONFLICT DO NOTHING: the second write for an identical key is inert.
    stmt = stmt.on_conflict_do_nothing(index_elements=["idem_key"])
    result = conn.execute(stmt)
    return result.rowcount == 1

ON CONFLICT DO NOTHING gives strict once-only semantics: a replayed version is silently discarded because its economics are already posted under that key. When the source issues a true amendment, version increments, the key changes, and the new version inserts as its own row — the ledger keeps both, which is what a settlement audit trail requires. If instead you want the latest version to overwrite in place, switch to on_conflict_do_update keyed on (trade_id, version) so a corrected price replaces the stale one deterministically.

Deduplicating a whole replayed batch then costs nothing extra: run each record through the same upsert inside one transaction. Because every write is individually idempotent, re-running the batch after a mid-flight crash re-applies only the rows that never committed and no-ops the rest.

def sync_batch(records: list[dict], run_id: str) -> dict:
    """Apply a batch idempotently; safe to re-run after any partial failure."""
    applied, skipped = 0, 0
    with engine.begin() as conn:                 # one transaction per batch
        for trade in records:
            if upsert_trade(conn, trade, run_id):
                applied += 1
            else:
                skipped += 1                     # duplicate / already-applied
    return {"run_id": run_id, "applied": applied, "skipped": skipped,
            "total": len(records)}

Retries need a bound. Wrap the pull-and-apply in bounded exponential backoff so a transient database or network fault is retried, but a poison batch does not spin forever. Because sync_batch is idempotent, retrying it after a partial write is always safe — the already-committed rows are skipped on the next pass.

import time

def run_with_retry(records, run_id, max_attempts: int = 4) -> dict:
    delay = 1.0
    for attempt in range(1, max_attempts + 1):
        try:
            return sync_batch(records, run_id)   # idempotent: re-run cannot double-write
        except Exception as exc:                 # narrow to DB/transport errors in prod
            if attempt == max_attempts:
                route_to_dead_letter(run_id, records, reason=str(exc))
                raise
            time.sleep(delay)
            delay *= 2                            # 1s, 2s, 4s ...

Reusing the same run_id across every retry of one logical run keeps lineage honest: the ledger records which run first posted each version, and a later replay under a new run_id still no-ops on the key. This is the same versioned-state discipline formalized in Trade Lifecycle State Management, applied here to the persistence boundary.

Verification

Confirm idempotency empirically before trusting the job in production:

  1. Double-apply invariant. Run sync_batch on the same records twice. The first pass reports applied == total; the second reports applied == 0 and skipped == total. The row count in trade_ledger must be identical after both passes.
  2. Key stability. Assert idempotency_key(trade) returns the same digest across two calls and across a serialized round-trip, and that changing only version changes the key while changing run_id or received_at does not.
  3. Amendment path. Post version 1, then version 2 of the same trade_id; confirm two rows exist with distinct keys and the later charge_usd reflects the amended economics.
  4. Reconciliation diff. Group the ledger by (trade_id, settlement_interval) and assert the summed charge_usd equals the source statement to the cent — a duplicate would inflate the sum and surface here immediately.
def test_replay_is_idempotent():
    records = [_sample_trade(trade_id="T1", version=1)]
    first = sync_batch(records, run_id="R1")
    second = sync_batch(records, run_id="R1")     # exact replay
    assert first["applied"] == 1
    assert second["applied"] == 0 and second["skipped"] == 1

def test_amendment_writes_new_version():
    v1 = _sample_trade(trade_id="T1", version=1, price="42.10")
    v2 = _sample_trade(trade_id="T1", version=2, price="43.55")
    assert idempotency_key(v1) != idempotency_key(v2)

Compliance Note

An idempotent sync job is a precondition for the traceability FERC and the ISO tariffs require of settlement records. Under FERC’s Open Access framework and the RTO settlement manuals (PJM Manual 28, ERCOT Nodal Protocols Section 9, the CAISO Business Practice Manual for Settlements), every posted charge must trace to exactly one metered obligation and one market run — a double-posted trade breaks that one-to-one mapping and understates or overstates a participant’s invoice. Persisting on a deterministic idempotency key, keeping each version as its own append-only row, and stamping every write with a run_id and UTC posted_at gives auditors the immutable lineage they demand and lets a disputed interval be replayed to a bit-identical result. Validate the job against your market’s dispute-window rules so that an amendment arriving during resettlement inserts a new version rather than silently overwriting a value already cited on a filed statement.

Frequently Asked Questions

What should and should not go into an ETRM idempotency key?

Include only the stable business identity and the source version — trade_id, amendment version, node_id, and settlement_interval. Exclude anything that changes on replay: ingest timestamps, run_id, batch offsets, or a database sequence. If those leak in, the same logical trade hashes differently on the second delivery and the dedup check fails to recognize it, reintroducing the double-write you were trying to prevent.

Should a duplicate be discarded or should it overwrite the existing row?

It depends on whether the record is a true replay or a genuine amendment. For an exact replay of an already-posted version, discard it with ON CONFLICT DO NOTHING — the economics are identical, so writing again is pointless. For an amendment that changes price or volume, let the source version increment so it inserts as a new keyed row and the prior version is retained for audit. Use an upsert-that-updates only when your model deliberately keeps one current row per trade rather than a full version history.

How do idempotency keys interact with retries and at-least-once queues?

They make retries free. A queue that guarantees at-least-once delivery will occasionally hand you the same message twice, and a bounded-backoff retry will re-run a batch that partially committed. Because each write keys on the deterministic idempotency key, the redelivered or re-run records collide with rows already present and no-op, so the final state is identical regardless of how many times the message was delivered.