Pyranid Logo

Core Concepts

Notifications

Pyranid provides a small, database-neutral API for named notifications. PostgreSQL LISTEN/NOTIFY is the only currently supported implementation. The public API deliberately contains no PostgreSQL or JDBC-driver types; Pyranid might support other database notification systems in the future.

Notifications are just nice-to-haves

Database notifications are transient, lossy hints. Use them to wake code that reconciles authoritative database state. Do not treat them as a durable event log, exact event counter, change-data-capture stream, or work queue.

Sending Notifications

Use Database::sendNotification(...) to send a named notification with an optional string payload:

// Send a hint without a payload.
database.sendNotification("car_changed");

// Include the changed record's identifier when it helps reconciliation.
database.sendNotification("car_changed", carId.toString());

The one-argument overload is equivalent to passing null to the two-argument overload, which also accepts "" or a nonempty string. Pyranid does not normalize payloads generically: Notification::getPayload() is nullable, and each database determines how null and empty payloads are represented. PostgreSQL converts a null payload to "", so PostgreSQL receivers observe an empty string for null, omitted, and explicitly empty payloads. Portable code should not assume that null and empty payloads remain distinct.

Sending follows Pyranid's ordinary statement and transaction-selection path. When the durable change and notification should commit together, perform both through the same Database::transaction(...) call:

// Commit the durable change and its notification atomically.
database.transaction(() -> {
  database.query("""
    UPDATE car
    SET color = :color
    WHERE car_id = :carId
    """)
    .bind("color", Color.BLUE)
    .bind("carId", carId)
    .execute();

  // PostgreSQL delivers this only if the transaction commits.
  database.sendNotification("car_changed", carId.toString());
});

The backend determines the precise delivery point. PostgreSQL publishes the notification after commit and discards it on rollback.

Notification payloads should contain only enough information to locate or version durable state. Treat them as untrusted input and avoid secrets. Sends use normal Pyranid statement logging, parameter redaction, timeout, and metrics behavior.

Receiving Notifications

Database::withNotificationSession(...) opens one callback-scoped listener session for either one channel or a fixed set:

// Register both channels on one listener connection.
database.withNotificationSession(
  Set.of("car_changed", "owner_changed"),
  session -> {
    // Reconcile once after registration to close the setup gap.
    reconcileCars();

    for (;;) {
      List<Notification> notifications =
        session.awaitNotifications(Duration.ofSeconds(30));

      // Treat the batch as a hint to reread durable state.
      if (!notifications.isEmpty())
        reconcileCars();
    }
  }
);

The callback begins only after every requested channel has been registered. Performing the initial reconciliation inside the callback closes the setup gap: notifications committed while that reconciliation runs remain available to a later receive.

One withNotificationSession(...) invocation:

  • runs synchronously on the calling thread;
  • acquires at most one listener connection from that Database instance's configured DataSource;
  • invokes the callback at most once;
  • never replaces or reconnects the listener connection;
  • expires the supplied NotificationSession when the callback exits; and
  • completes listener cleanup before returning or throwing.

Pyranid does not create a listener thread, executor, timer, scheduler, or reconnect task. The application owns the calling thread and any supervision or retry policy.

Durable Progress Without A Notification

Callback-entry reconciliation closes the listener-registration gap, but it cannot recover every publication gap while one listener remains active. For example, an application that commits durable work and then sends a best-effort notification in a separate step can crash between those operations. The notification is also inherently lossy after publication.

When notifications must provide only lower latency - not the sole source of progress - schedule an authoritative poll independently of the listener thread. Do not run the only periodic poll after awaitNotifications(...) returns: a receive that is stuck inside driver protocol work cannot reach it.

// Poll durable state independently of notification delivery.
ScheduledExecutorService durablePoller =
  Executors.newSingleThreadScheduledExecutor();

durablePoller.scheduleWithFixedDelay(
  this::reconcilePendingJobs,
  0, // Start immediately.
  1, // Wait one minute after each completed poll.
  TimeUnit.MINUTES
);

An outbox with persistent relay retries can close the commit-to-publication crash gap, but consumers must still reconcile durable state because notification delivery itself is not durable.

Waiting And Draining

NotificationSession::awaitNotifications(...) waits for a nonempty batch until its best-effort elapsed-time budget expires:

// Wait up to 30 seconds for at least one notification.
List<Notification> notifications =
  session.awaitNotifications(Duration.ofSeconds(30));

An empty result means only that no notification was observed before that call's final check. It does not prove that a quiet connection is still healthy.

NotificationSession::drainNotifications() performs one driver-specific non-waiting poll:

// Poll once without waiting for a notification.
List<Notification> pending = session.drainNotifications();

if (!pending.isEmpty())
  reconcileCars();

Calling awaitNotifications(Duration.ZERO) has the same behavior as drainNotifications(). Returned batches are immutable and preserve the encounter order reported by the adapter. A conceptual burst can span multiple batches, so applications should normally reconcile once per nonempty batch rather than interpreting each Notification as one durable event.

Positive waits are best-effort rather than hard JDBC deadlines. Pyranid divides them into short driver calls so interruption can normally be observed between calls, but it cannot preempt a JDBC call already in progress.

For PostgreSQL, neither a positive wait nor drainNotifications() has a hard completion bound after pgjdbc begins parsing a partial protocol frame. Give the listener a bounded shutdown deadline and escalate to process-level termination if it does not join; no in-process timeout can universally preempt arbitrary JDBC I/O.

When a nonempty batch and a stop race, the batch wins and the interrupt flag remains set. Process the returned batch - or intentionally discard it according to application policy - before acting on that later stop.

Session Scope And Transaction Boundaries

NotificationSession is confined to the callback thread and may not be retained. Calls from another thread, calls after callback exit, concurrent/reentrant receives, and receives after a terminal transport failure throw IllegalStateException.

Opening a notification session or consuming notifications while any Pyranid transaction is active on the calling thread is also illegal. Notification consumption is not transactional: rolling back or retrying a transaction cannot put an already-consumed notification back.

This guard observes Pyranid's thread-local transaction stack only. It cannot detect an application-owned raw JDBC transaction or transaction-control SQL such as BEGIN issued through Database::useRawConnection(...). Ordinary lifecycle calls such as Connection::setAutoCommit(...), Connection::commit(), and Connection::rollback() remain blocked on Pyranid's guarded raw connection, but SQL text can still establish transaction state that the notification guard cannot see. Do not consume notifications while any such unmanaged transaction is active.

Complete each bounded reconciliation transaction before waiting again:

// Reconcile after registration, then after each observed batch.
listenerDatabase.withNotificationSession("job_ready", session -> {
  reconcilePendingJobs();

  for (;;) {
    if (!session.awaitNotifications(Duration.ofSeconds(30)).isEmpty())
      reconcilePendingJobs();
  }
});

Nested notification sessions are permitted, but strongly discouraged. The outer listener is not serviced while its callback runs the inner session, and the inner call needs another connection. Against a capacity-one source, that checkout can block indefinitely and deadlock application progress. Prefer separately supervised listener threads and provision at least one checkout for every simultaneously live listener. Concurrent top-level calls are thread-safe, but Pyranid promises no acquisition fairness or progress beyond the configured DataSource.

Interruption And Supervision

The methods withNotificationSession(...), awaitNotifications(...), and drainNotifications() declare InterruptedException. An interrupt observed before driver work or after an empty receive is normal cooperative cancellation, not a database failure.

If a supervisor catches the exception, it should normally restore the thread's interrupt flag before stopping:

try {
  listenerDatabase.withNotificationSession(
    "job_ready",
    this::consumeJobNotifications
  );
} catch (InterruptedException interruptedException) {
  // Preserve the cancellation signal for the caller.
  Thread.currentThread().interrupt();
}

A successfully decoded nonempty batch wins over an interrupt that races after the driver returns. Pyranid returns the batch and leaves the interrupt flag set so application code can perform its final bounded reconciliation before stopping.

On Java 21+, interrupting a virtual thread blocked in an applicable JDK system-default socket can close the socket and surface as a terminal DatabaseException rather than clean InterruptedException. If a returned batch carries a late interrupt and final database work is required, clear and remember the flag, perform only bounded work, and restore the flag in finally:

// Capture an interrupt that raced with a returned batch.
List<Notification> hints =
  session.awaitNotifications(Duration.ofSeconds(30));
boolean stopRequested = Thread.interrupted();

if (stopRequested && hints.isEmpty())
  throw new InterruptedException();

try {
  reconcilePendingJobs();
} finally {
  // Restore the signal after bounded reconciliation.
  if (stopRequested)
    Thread.currentThread().interrupt();
}

A later interrupt can still race with that bounded work; this pattern preserves a stop already observed but does not make socket I/O immune to interruption.

Connection loss is terminal for the current invocation and surfaces as DatabaseException. Pyranid does not reinvoke the callback. An application that wants recovery should observe the failure, apply its own backoff and retry policy, and call withNotificationSession(...) again. The new callback should reconcile authoritative state again after registration.

Catching a terminal receive failure inside the callback does not make the session healthy: later receives throw IllegalStateException. When the callback exits, withNotificationSession(...) completes cleanup and rethrows the retained failure. A retained transport Error propagates unwrapped; only a distinct callback Error takes precedence, with the retained failure suppressed beneath it.

There are intentionally no listener-health accessors. A Boolean snapshot cannot certify that a quiet physical connection remains healthy and registered.

A Java 17 Supervisor

The application owns restart and readiness policy. The following Java 17-compatible pattern fails startup until registration and the first authoritative reconciliation have succeeded. After readiness, it retries only failures selected by application policy, with anti-flap backoff. Here listenerDatabase owns the session-affine source and jobRunner uses the ordinary application Database for bounded reconciliation transactions.

private static final Set<String> JOB_CHANNELS = Set.of("job_ready");
private static final Duration RECEIVE_INTERVAL =
  Duration.ofSeconds(30);
private static final Duration LISTENER_STABILITY_WINDOW =
  Duration.ofMinutes(1);

private final RetryPolicy.Backoff listenerBackoff =
  RetryPolicy.Backoff.exponential(
    Duration.ofMillis(250),
    Duration.ofSeconds(30)
  );

private void runJobNotificationSupervisor(
  StartupSignal startup,
  Predicate<? super DatabaseException> retryableAfterReadiness
) {
  Objects.requireNonNull(startup);
  Objects.requireNonNull(retryableAfterReadiness);

  AtomicBoolean everReady = new AtomicBoolean();
  AtomicReference<Long> attemptReadySinceNanos =
    new AtomicReference<>();
  int consecutiveFailures = 0;

  for (;;) {
    if (Thread.currentThread().isInterrupted()) {
      startup.cancelled();
      return;
    }

    // Acquisition and registration are not healthy-session time.
    attemptReadySinceNanos.set(null);

    try {
      listenerDatabase.withNotificationSession(
        JOB_CHANNELS,
        session -> {
          throwIfInterrupted();

          // Registration is complete. Publish readiness only after
          // authoritative state is known to be drainable.
          jobRunner.drainClaimableBatch();
          throwIfInterrupted();

          attemptReadySinceNanos.set(System.nanoTime());
          everReady.set(true);
          startup.ready();

          for (;;) {
            List<Notification> hints =
              session.awaitNotifications(RECEIVE_INTERVAL);
            boolean stopRequested = Thread.interrupted();

            if (stopRequested && hints.isEmpty())
              throw new InterruptedException();

            try {
              // A reconciliation failure ends this physical session.
              jobRunner.drainClaimableBatch();
            } finally {
              // Preserve a stop that raced with a returned batch.
              if (stopRequested)
                Thread.currentThread().interrupt();
            }
          }
        }
      );

      // A normal return before this attempt reached readiness is not healthy.
      if (attemptReadySinceNanos.get() == null) {
        if (Thread.currentThread().isInterrupted()) {
          startup.cancelled();
          return;
        }

        throw new IllegalStateException(
          "Notification listener stopped before readiness"
        );
      }

      // A healthy callback return is an application-requested stop.
      return;
    } catch (InterruptedException interruptedException) {
      // Listener cleanup has completed. Restore normal cancellation state.
      Thread.currentThread().interrupt();
      startup.cancelled();
      return;
    } catch (DatabaseException listenerOrReconciliationFailure) {
      Long readySinceNanos =
        attemptReadySinceNanos.getAndSet(null);

      // A real failure that coincides with shutdown remains visible.
      if (Thread.currentThread().isInterrupted()) {
        startup.failed(listenerOrReconciliationFailure);
        throw listenerOrReconciliationFailure;
      }

      // This sample deliberately fails fast before first readiness.
      if (!everReady.get()) {
        startup.failed(listenerOrReconciliationFailure);
        throw listenerOrReconciliationFailure;
      }

      if (!retryableAfterReadiness.test(
        listenerOrReconciliationFailure
      ))
        throw listenerOrReconciliationFailure;

      if (readySinceNanos != null
        && System.nanoTime() - readySinceNanos
          >= LISTENER_STABILITY_WINDOW.toNanos())
        consecutiveFailures = 0;

      if (consecutiveFailures < Integer.MAX_VALUE)
        ++consecutiveFailures;

      Duration delay = Objects.requireNonNull(
        listenerBackoff.delayAfterFailedAttempt(
          consecutiveFailures,
          listenerOrReconciliationFailure
        )
      );

      if (delay.isNegative())
        throw new IllegalArgumentException(
          "Supervisor backoff must not be negative"
        );

      if (!sleepInterruptibly(delay))
        // Do not erase the failure that preceded the interrupted backoff.
        throw listenerOrReconciliationFailure;
    } catch (RuntimeException | Error terminalFailure) {
      // No-op if readiness already won the first-result race.
      startup.failed(terminalFailure);
      throw terminalFailure;
    }
  }
}

private static void throwIfInterrupted()
  throws InterruptedException {
  if (Thread.interrupted())
    throw new InterruptedException();
}

private static boolean sleepInterruptibly(Duration delay) {
  try {
    Thread.sleep(
      delay.toMillis(),
      delay.toNanosPart() % 1_000_000
    );
    return true;
  } catch (InterruptedException interruptedException) {
    Thread.currentThread().interrupt();
    return false;
  }
}

private static final class StartupSignal {
  private static final Object READY = new Object();

  private final AtomicReference<Object> outcome =
    new AtomicReference<>();
  private final CountDownLatch completed =
    new CountDownLatch(1);

  void ready() {
    complete(READY);
  }

  void failed(Throwable failure) {
    complete(Objects.requireNonNull(failure));
  }

  void cancelled() {
    failed(new CancellationException(
      "Notification listener stopped before readiness"
    ));
  }

  void await(long timeout, TimeUnit unit)
    throws InterruptedException, TimeoutException,
    ExecutionException {
    Objects.requireNonNull(unit);

    if (!completed.await(timeout, unit))
      throw new TimeoutException(
        "Timed out waiting for notification readiness"
      );

    Object result = outcome.get();
    if (result instanceof Throwable failure)
      throw new ExecutionException(failure);
  }

  private void complete(Object result) {
    if (outcome.compareAndSet(null, result))
      completed.countDown();
  }
}

Run the supervisor on a dedicated application-owned executor and retain its worker Future. Readiness is a first-result signal: registration alone is not ready, and a later failure cannot replace an already-published ready result. On shutdown, interrupt the dedicated worker without discarding its observable outcome, wait only for an application-chosen finite deadline, and inspect the Future. If the worker does not terminate, report the stuck listener and escalate at the process boundary.

The retry predicate is application policy; do not use DatabaseException::isTransient() as the sole classifier. Authentication, configuration, transport, and reconciliation failures can require different actions, and equivalent disconnects can surface through different SQLSTATEs.

Choosing The Listener Database

Listening requires a DataSource that preserves one backend session for the complete callback. The ordinary application Database is suitable when its source provides that affinity and has enough pool capacity:

// Reconcile after registration and after each observed batch.
database.withNotificationSession("car_changed", session -> {
  reconcileCars();

  for (;;) {
    if (!session.awaitNotifications(Duration.ofSeconds(30)).isEmpty())
      reconcileCars();
  }
});

If ordinary traffic uses a source that cannot preserve listener state, construct a second Database over a session-affine listener source:

// Dedicate this Database to a session-affine listener source.
Database listenerDatabase = Database.withDataSource(listenerDataSource)
  .databaseType(DatabaseType.POSTGRESQL)
  .build();

This is ordinary Database composition; Pyranid has no notification-specific DataSource setting. Reconciliation and transactional publication should continue through the application Database. Each Database retains its one caller-owned DataSource; Pyranid neither compares separately configured sources nor closes them.

Each live listener holds one connection for its full callback lifetime. A capacity-one listener source cannot simultaneously serve callback queries or sends, even if another Database object wraps that same source.

Configure the listener instance's DatabaseType explicitly when practical. Otherwise Database::isNotificationListeningSupported() or the first session open may resolve an uncached type through that instance's source, which performs a metadata checkout and can throw DatabaseException. Within one session invocation, detection finishes before listener acquisition; a concurrent first-time detection can still contend with a live listener for a capacity-one source.

Call Database::isNotificationListeningSupported() on the listener Database when application and listener instances differ. It reports whether that instance's configured dialect and currently loadable runtime adapter expose the APIs needed to attempt listening. It performs no notification-session checkout or lifecycle metric emission, but it does not inspect proxy mode, prove backend-session affinity, test a quiet connection, or guarantee that independently configured publisher, listener, and reconciliation instances reach the same database.

Pyranid performs no runtime topology validation or active liveness query. In particular, using PgBouncer transaction or statement pooling as the listener source is unsupported and can appear to register successfully before delivery silently stops. Use a direct or session-pooled listener source, and route every participating instance to the same active logical database and primary.

Portable Contract

Pyranid standardizes the callback scope and common value model while leaving database semantics with the backend.

Pyranid guaranteesBackend-specific behavior
Nonblank, NUL-free channel namesChannel case matching and additional limits
Nullable string payloads with no generic normalizationWhether null and empty remain distinct, whether payloads are supported, and their size limit
Immutable notification batchesDelivery timing, ordering, fan-out, and coalescing
One physical listener session per invocationProxy and pooling compatibility
No backend sender identifier or occurrence countAny metadata that Pyranid deliberately omits

Send and listen support are independent. A backend may support one without the other. Each backend defines whether it accepts, preserves, transforms, or rejects null and empty payloads; a backend without payload support may reject any supplied payload.

PostgreSQL

PostgreSQL notification receiving supports and is tested against pgjdbc 42.7.13 as its baseline. pgjdbc is a provided dependency, so applications receiving notifications must supply pgjdbc 42.7.13 or newer at runtime. Sending uses bound pg_notify(?, ?) SQL and remains available without the receive adapter.

PostgreSQL-specific behavior - including commit/rollback delivery, duplicate coalescing, channel and payload limits, PgBouncer pooling modes, TCP/socket considerations, and notification-queue operations - is covered under Database-Specific Recipes.

Metrics

Notification sessions emit low-cardinality lifecycle callbacks through the MetricsCollector configured on the listener Database. Sends and ordinary statements emit through the Database instance that executes them.

See Metrics Collection for lifecycle outcomes, in-memory counters, and current OpenTelemetry adapter coverage.

References:

Previous
ResultSet Mapping
Next
Error Handling