以資料庫交易原子處理工作失敗與點數退款

An asynchronous generation job can have two observers: a provider callback and a status poll. If both observe failure, both may try to refund the credits charged for that job.

A process-local flag cannot coordinate different workers. A unique refund key helps, but it does not make a separately committed job status and balance change atomic. The invariant belongs inside the database transaction: claim one terminal transition, record its refund, and update its balance together.

Store the charged cost with the job

Refund the amount actually charged for that job, rather than calculating it again from a pricing table that may have changed. The example below assumes the job was already charged, and charged_cost holds that debit amount. Credit charging itself is outside this example.

These are fictional tables and a seven-credit test job, not a product's pricing or production schema:

CREATE TABLE balances (
  owner TEXT PRIMARY KEY,
  credits INTEGER NOT NULL CHECK (credits >= 0)
);
CREATE TABLE jobs (
  id TEXT PRIMARY KEY,
  owner TEXT NOT NULL,
  charged_cost INTEGER NOT NULL CHECK (charged_cost > 0),
  status TEXT NOT NULL CHECK (
    status IN ('pending', 'processing', 'completed', 'failed')
  )
);
CREATE TABLE refunds (
  event_key TEXT PRIMARY KEY,
  owner TEXT NOT NULL,
  amount INTEGER NOT NULL
);
INSERT INTO balances VALUES ('demo', 0);
INSERT INTO jobs VALUES ('job-a', 'demo', 7, 'processing');

Claim the transition conditionally

The following Python function uses SQLite so the example can run without a database server. It requires SQLite with RETURNING support. Each invocation opens its own connection; no Python lock is used to coordinate the workers.

import sqlite3

def finalize_job(db_path, job_id, outcome):
    if outcome not in ('completed', 'failed'):
        raise ValueError('Unsupported terminal outcome')
    conn = sqlite3.connect(db_path, timeout=10, isolation_level=None)
    try:
        conn.execute('BEGIN IMMEDIATE')
        row = conn.execute(
            "UPDATE jobs SET status = ? WHERE id = ? "
            "AND status IN ('pending', 'processing') "
            'RETURNING owner, charged_cost',
            (outcome, job_id),
        ).fetchone()
        if row is None:
            conn.commit()
            return False
        if outcome == 'failed':
            owner, cost = row
            conn.execute(
                'INSERT INTO refunds (event_key, owner, amount) VALUES (?, ?, ?)',
                (f'job-refund:{job_id}', owner, cost),
            )
            changed = conn.execute(
                'UPDATE balances SET credits = credits + ? WHERE owner = ?',
                (cost, owner),
            ).rowcount
            if changed != 1:
                raise RuntimeError('Missing balance row')
        conn.commit()
        return True
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

The status predicate is the gate: an invocation that cannot change a pending or processing job does not insert a refund. The winner commits all three changes together. A failure after the ledger insert rolls back the terminal transition as well as the insert, so a later attempt can still finish the job.

The unique key is retained as an invariant check. An ordinary duplicate request hits the terminal-state guard and returns False. A pre-existing refund key on a job that still claims to be processing is an inconsistency worth surfacing. Silently ignoring that conflict could hide incorrect balance or state assumptions.

This is an internal credit-ledger example. It does not implement refunds through a payment processor or promise exactly-once delivery of external callbacks.

Exercise actual connection races

After creating the schema in a fresh test database, run multiple failure notifications against the same job:

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=8) as pool:
    results = list(pool.map(
        lambda _: finalize_job('example.sqlite', 'job-a', 'failed'),
        range(16),
    ))
assert sum(results) == 1

with sqlite3.connect('example.sqlite') as conn:
    assert conn.execute('SELECT credits FROM balances').fetchone()[0] == 7
    assert conn.execute('SELECT COUNT(*) FROM refunds').fetchone()[0] == 1

This synthetic fixture uses independent connections and checks the final balance, ledger, and job status. Six other local scenarios checked competing completion/failure notifications, an already completed job, an injected balance-write failure, an unknown job, a missing balance row, and a conflicting ledger entry. All seven scenarios passed. They are correctness checks for this example, not throughput benchmarks or production traffic measurements.

Keep the guarantee within its scope

SQLite serializes writes, and BEGIN IMMEDIATE starts a write transaction. That makes this a useful local demonstration; it is not a PostgreSQL concurrency test. A production database needs equivalent transactional guards, constraints, error handling, and tests in its actual engine. Handle lock timeouts as failed attempts with a bounded retry policy, rather than treating them as successful refunds.

The example also deliberately accepts whichever valid terminal notification wins. A real integration must first authenticate the notification, bind it to the correct generation attempt, and decide which upstream state is authoritative. Otherwise a stale or untrusted failure could win a perfectly atomic transition.

Media storage and user notifications have their own failure boundaries. Do not hold a database write transaction open while downloading audio or calling a remote service. Persist a durable follow-up event in the same transaction when another system must be notified, and process that event separately with its own deduplication.

Disclosure: Published under the pen name Cynthia Hayes for HeartMuLa, the hosted music creation app. This article and illustrative code were prepared by an AI agent and labeled Fully Autonomous. Tests used synthetic temporary SQLite databases; no production database, payment service, customer record, or private credential was accessed. The example is not a claim that the deployed application uses this exact implementation.