Design: LISTEN/NOTIFY Support

[SeanTAllen]

Add support for PostgreSQL's LISTEN/NOTIFY asynchronous notification mechanism. Users subscribe to channels via regular queries (LISTEN channel) and receive notifications via a new callback on SessionStatusNotify.

Background

PostgreSQL allows connections to subscribe to named channels with LISTEN channel_name and send notifications with NOTIFY channel_name, 'payload'. Notifications are delivered asynchronously -- they arrive in the message stream between query cycles, specifically just before ReadyForQuery. A connection can receive notifications it sent to itself.

Wire Protocol

NotificationResponse ('A'): Byte1('A') Int32(length) Int32(pid) String(channel) String(payload)

• pid: Process ID of the notifying backend
• channel: Null-terminated channel name
• payload: Null-terminated payload string (can be empty)

NotificationResponse is an asynchronous message that can arrive between any other messages, though PostgreSQL delivers pending notifications just before ReadyForQuery.

Current Handling

The parser (_response_parser.pony line 214-217) matches NotificationResponse and returns _SkippedMessage -- the payload is discarded without parsing. The router (_response_message_parser.pony line 56-59) silently ignores _SkippedMessage. A test message builder (_IncomingNotificationResponseTestMessage) already exists in _test_response_parser.pony and correctly constructs the wire format.

Design

New Public Type

notification.pony:

class val Notification
  """
  A notification received via PostgreSQL's LISTEN/NOTIFY mechanism. Contains
  the channel name, payload string, and process ID of the notifying backend.

  Subscribe to notifications by executing `LISTEN channel_name` via
  `Session.execute()`. Unsubscribe with `UNLISTEN channel_name` or
  `UNLISTEN *` to remove all subscriptions.
  """
  let channel: String
  let payload: String
  let pid: I32

  new val create(channel': String, payload': String, pid': I32) =>
    channel = channel'
    payload = payload'
    pid = pid'

New Internal Message Type

Add to _backend_messages.pony:

class val _NotificationResponseMessage
  """
  Message from the backend delivering a LISTEN/NOTIFY notification.
  Contains the process ID of the notifying backend, the channel name,
  and the payload string.
  """
  let process_id: I32
  let channel: String
  let payload: String

  new val create(process_id': I32, channel': String, payload': String) =>
    process_id = process_id'
    channel = channel'
    payload = payload'

Add _NotificationResponseMessage to the _ResponseParserResult union type.

Parser Change

In _response_parser.pony, replace the skip-and-return-_SkippedMessage case with actual parsing:

| _MessageType.notification_response() =>
  // Slide past the header...
  buffer.skip(5)?
  let pid = buffer.i32_be()?
  let channel_bytes = buffer.read_until(0)?
  let channel = String.from_array(consume channel_bytes)
  let payload_bytes = buffer.read_until(0)?
  let payload = String.from_array(consume payload_bytes)
  return _NotificationResponseMessage(pid, channel, payload)

Update the _SkippedMessage docstring to remove NotificationResponse from the list -- it now only covers ParameterStatus and NoticeResponse.

Router Change

In _response_message_parser.pony, add a match clause before the _SkippedMessage case:

| let msg: _NotificationResponseMessage =>
  s.state.on_notification(s, msg)

Update the _SkippedMessage comment to remove NotificationResponse from the list.

New Callback

Add to SessionStatusNotify:

be pg_notification(session: Session, notification: Notification) =>
  """
  Called when the server delivers a LISTEN/NOTIFY notification. Subscribe
  to notifications by executing LISTEN via Session.execute(). Notifications
  arrive asynchronously between query cycles.
  """
  None

Non-breaking -- default body means existing implementations don't need to change.

State Machine Changes

Add on_notification to _SessionState interface:

fun ref on_notification(s: Session ref, msg: _NotificationResponseMessage)
  """
  Called when the server sends a LISTEN/NOTIFY notification.
  """

Default in _NotAuthenticated trait (notifications cannot arrive before auth):

fun ref on_notification(s: Session ref, msg: _NotificationResponseMessage) =>
  _IllegalState()

Implementation in _SessionLoggedIn:

fun ref on_notification(s: Session ref, msg: _NotificationResponseMessage) =>
  _notify.pg_notification(s,
    Notification(msg.channel, msg.payload, msg.process_id))

Callback Timing

Notifications are delivered inline during message parsing. Since callbacks are be (behavior sends), they are queued and don't block the parser loop. When SessionStatusNotify and ResultReceiver are the same actor, the delivery order matches the wire protocol order. For example, if a notification arrives between CommandComplete and ReadyForQuery, pg_query_result is delivered first, then pg_notification.

Files Changed

1. New: postgres/notification.pony -- Notification val class
2. Edit: postgres/_backend_messages.pony -- add _NotificationResponseMessage class
3. Edit: postgres/_response_parser.pony -- parse NotificationResponse instead of skipping; add to _ResponseParserResult union; update _SkippedMessage docstring
4. Edit: postgres/_response_message_parser.pony -- route _NotificationResponseMessage to session state; update _SkippedMessage comment
5. Edit: postgres/session_status_notify.pony -- add pg_notification callback
6. Edit: postgres/session.pony -- add on_notification to _SessionState interface; default _IllegalState() in _NotAuthenticated; implement in _SessionLoggedIn
7. Edit: postgres/_test_response_parser.pony -- update _TestResponseParserNotificationResponseSkipped to verify parsing into _NotificationResponseMessage with correct fields (rename to _TestResponseParserNotificationResponseMessage); update _TestResponseParserMultipleMessagesSkippedFirst to expect _NotificationResponseMessage instead of _SkippedMessage for the notification entry
8. New: postgres/_test_notification.pony -- mock server notification tests
9. Edit: postgres/_test.pony -- register new test classes
10. Edit: CLAUDE.md -- update architecture (async message handling), test organization (new file), public API types (Notification), file layout (new files), supported features

Tests

Parser Unit Tests

In _test_response_parser.pony:

• _TestResponseParserNotificationResponseMessage (renamed from _TestResponseParserNotificationResponseSkipped) -- parse NotificationResponse bytes and verify _NotificationResponseMessage has correct process_id, channel, and payload fields. Include a second case with empty payload to verify read_until(0) handles immediately null-terminated strings.

• _TestResponseParserMultipleMessagesSkippedFirst -- update to expect _NotificationResponseMessage for the notification entry instead of _SkippedMessage. Verify the other async messages (ParameterStatus, NoticeResponse) still return _SkippedMessage.

Unit Tests (Mock Servers)

New test file: postgres/_test_notification.pony

Mock port assignments: 7686, 7687

• _TestNotificationDelivery (port 7686) -- mock server authenticates, sends ReadyForQuery('I'). Client sends a query. Server responds with CommandComplete + NotificationResponse(pid=42, channel="test_ch", payload="hello") + ReadyForQuery('I'). Client verifies pg_notification fires with correct Notification fields (channel, payload, pid).

• _TestNotificationDuringDataRows (port 7687) -- mock server authenticates, sends ReadyForQuery('I'). Client sends a query. Server responds with RowDescription + DataRow + NotificationResponse(pid=1, channel="ch", payload="mid-query") + DataRow + CommandComplete + ReadyForQuery('I'). Client verifies both the query result (with both data rows) and the notification are delivered.

Integration Test

Added to postgres/_test_query.pony:

• _TestListenNotify -- execute LISTEN test_channel_XX (with a unique suffix to avoid interference from other tests), then execute NOTIFY test_channel_XX, 'hello'. Verify pg_notification fires with channel matching and payload "hello". Clean up with UNLISTEN test_channel_XX.

Existing Tests

No regressions expected. The parser change converts NotificationResponse from _SkippedMessage to _NotificationResponseMessage, but _ResponseMessageParser now routes it to on_notification instead of ignoring it. Existing tests don't involve NotificationResponse in their message streams, so behavior is unchanged. The two parser tests that reference NotificationResponse are updated explicitly.

Example

New example: examples/listen-notify/listen-notify-example.pony

Demonstrates:

• Subscribing with LISTEN channel_name
• Sending notifications with NOTIFY channel_name, 'payload'
• Receiving notifications via pg_notification callback
• Unsubscribing with UNLISTEN channel_name

Build/Run Commands

make ssl=3.0.x                      # full build and test
make unit-tests ssl=3.0.x           # unit tests only
make integration-tests ssl=3.0.x    # integration tests (needs postgres containers)

[SeanTAllen]

Two notes from review before implementation:

1. Parser should isolate the payload before parsing (will fix)

The proposed parser code reads variable-length strings directly from the shared buffer via read_until(0). Every other variable-length message parser in the codebase first extracts the payload into an isolated block with buffer.block(payload_size), then parses within that. This prevents accidental reads past the message boundary if the server sends a malformed message (missing null terminator). Examples: error_response, row_description, data_row, command_complete, SASL authentication — all follow this pattern. The fixed-size-only parsers (backend_key_data) read directly but use fixed-width reads (i32_be), not read_until.

Will follow the established pattern: extract with buffer.block(payload_size), wrap in a Reader, then parse with read_until(0) on that isolated reader.

2. _TestResponseParserMultipleMessagesSkippedFirst name (minor, will fix)

After the change, this test has ParameterStatus (skipped), NoticeResponse (skipped), NotificationResponse (now parsed as _NotificationResponseMessage), then AuthenticationOk. The name "SkippedFirst" becomes slightly misleading since one of the "first" messages is no longer skipped. Will rename to something like MultipleMessages/AsyncThenAuth.

[SeanTAllen]

Implemented in PR #108 (merged). The two findings from the earlier review comment were addressed in the implementation.