Skip to content

feat(webhooks): let PgNotificationOutbox enqueue on a synchronous DB-API transaction #1607

Description

@bokelley

Summary

PgNotificationOutbox.enqueue_prepared only accepts a psycopg 3 AsyncConnection inside a transaction. A seller whose business writes run on synchronous SQLAlchemy (psycopg2 or psycopg 3 sync) cannot queue a notification in the same transaction as the change it announces, which is the outbox's whole point. Everything enqueue_prepared does is synchronous except two await conn.execute(...) calls, so a synchronous variant is small.

Details (adcp 9.0.0b4 and 9.1.0)

adcp/notification_outbox_pg.py, enqueue_prepared:

  • conn.info.transaction_status != TransactionStatus.INTRANS → ValueError. That is psycopg 3 only (psycopg2 has conn.get_transaction_status(), a SQLAlchemy Connection has in_transaction()).
  • Validation, _protected, _scope, _aad and the AES-GCM encryption are pure.
  • The I/O is one INSERT ... ON CONFLICT DO NOTHING RETURNING id and, on conflict, one SELECT ... FOR UPDATE plus _open to compare the stored binding. Both use %s placeholders, which psycopg2 also accepts.

Ask

One of:

  1. enqueue_prepared_sync(conn, prepared, *, notification_type, caller_scope_id, account_id=None, signing_scope_id=None) -> int, taking a synchronous DB-API connection (psycopg 3 Connection or psycopg2 connection) that is inside a transaction, sharing all the pure work with the async method.
  2. Or a public pure step, for example outbox.prepare_row(prepared, ...) -> OutboxInsert(sql: str, params: tuple, conflict_sql: str, conflict_params: tuple) plus outbox.check_existing(row) -> int, so a caller can run the two statements on its own session (session.execute(text(...))).

Option 1 keeps the binding check (identical re-enqueue returns the id, a different body raises) inside the SDK, so I'd prefer it.

Why an adopter hits this

The Embedded Sales Agent records each account change in a SQLAlchemy unit of work (sync, psycopg2) and queues account.change_recorded for each subscriber in that same transaction, so a change that rolls back sends nothing and a committed change cannot lose its notification (src/core/account_change_notifications.py). It cannot call enqueue_prepared from there. Until the SDK has a synchronous path, the ESA keeps a staging row in its own table in the business transaction and moves it into adcp_notification_outbox in a second psycopg 3 transaction that deletes the staging row and calls enqueue_prepared atomically (FIXME pointing here). That relay is the only ESA-local webhook machinery left for RFC 9421 notifications once this lands.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions