PostgreSQL and transactions · WebSocket

Share the pool through DI, connect services in the same message to Scoped PgDbContext, and explicitly establish transaction boundaries.

In default single mode, lily_postgresql provides a DI pool service and Scoped PgDbContext. Do not make a WebSocket connection's entire lifetime one transaction; run a message's unit of work inside an explicit transaction.

Dependency and configuration#

Update the first application's lilyrs dependency to use the following feature list. Using driver types from lilyrs::postgresql::{diesel, diesel_async} re-exports keeps versions compatible.

lilyrs = { version = "0.1.0", features = ["websocket", "postgresql"] }

[postgresql]
mode = "single"
connection_string = "postgresql://lily:lily@127.0.0.1:5432/lily"
[postgresql.tls]
mode = "disable"
[postgresql.pool]
max_size = 10
acquire_timeout_secs = 5
transaction_cleanup_timeout_secs = 5

This is local configuration. Production requires verify-full, hostname/CA verification, and a real secret policy. Schema and migrations belong to your application; repository derive does not create them. The HTTP PostgreSQL guide details the shared setup.

A message-scoped unit of work#

The example assumes a users(id, name) table. The complete controller, service, and repository appear below; add real user/object authorization in a guard or domain service.

use std::sync::Arc;
use lilyrs::websocket::Injectable;
use lilyrs::websocket::ServiceTrait;
use lilyrs::postgresql::{diesel, diesel_async::RunQueryDsl, PgDbContext, PgResult};
use lilyrs::websocket::{Ack, Extensions, Payload, Service, WebSocketActionError,
    WebSocketController, WebSocketControllerInitError, WebSocketControllerTrait,
    async_trait, websocket_controller};
use serde::Deserialize;

#[derive(Injectable)]
#[service(lifetime = "Scoped")]
struct UserRepository {
    #[inject]
    context: Arc<PgDbContext>,
}
impl ServiceTrait for UserRepository {}
impl UserRepository {
    async fn rename(&self, id: i64, name: String) -> PgResult<()> {
        self.context.with_connection(move |connection, _| Box::pin(async move {
            diesel::sql_query("UPDATE users SET name = $1 WHERE id = $2")
                .bind::<diesel::sql_types::Text, _>(name)
                .bind::<diesel::sql_types::BigInt, _>(id)
                .execute(connection).await?;
            Ok(())
        }), None).await
    }
}

#[derive(Injectable)]
#[service(lifetime = "Scoped")]
struct UserService {
    #[inject]
    context: Arc<PgDbContext>,
    #[inject]
    users: Arc<UserRepository>,
}
impl ServiceTrait for UserService {}
impl UserService {
    async fn rename_pair(&self, first: i64, second: i64) -> PgResult<()> {
        let users = Arc::clone(&self.users);
        self.context.transaction(move |_| async move {
            users.rename(first, "Ada".into()).await?;
            users.rename(second, "Deniz".into()).await?;
            Ok(())
        }, None).await // WS cancellation cannot be converted to Pg ExecutionCancellation.
    }
}

#[derive(Deserialize)]
struct RenamePair { first: i64, second: i64 }
#[derive(WebSocketController)]
#[namespace("users")]
struct UserController;
#[async_trait]
impl WebSocketControllerTrait for UserController {
    async fn new(_: Arc<Extensions>) -> Result<Self, WebSocketControllerInitError> { Ok(Self) }
}
#[websocket_controller]
impl UserController {
    #[message("renamePair")]
    async fn rename_pair(&self, Payload(input): Payload<RenamePair>,
        Service(service): Service<UserService>) -> Result<Ack<bool>, WebSocketActionError>
    {
        // In a real application, a guard or domain service checks user/object permissions.
        service.rename_pair(input.first, input.second).await
            .map_err(WebSocketActionError::internal)?;
        Ok(Ack::new(true))
    }
}

Download the complete example.

UserService and UserRepository receive the same PgDbContext instance within a message. Both repository calls in the transaction callback join the same transaction. A new scope opens for the next WebSocket message. Even if payload appears first in the DTO signature, the Service parts extractor resolves first.

Cancellation type boundary#

lilyrs::postgresql::ExecutionCancellation re-exports the shared lily_cancellation type. lilyrs::websocket::ExecutionCancellation is a different type; there is no public conversion between them. The HTTP example's Some(cancellation) does not compile here.

The example therefore deliberately uses None. It does not promise that the WebSocket signal simultaneously cancels SQL. Interrupting message execution does not prove all SQL work has ended or a commit has been undone. Preserve PostgreSQL's bounded operation, cleanup, and scope contracts; do not reach for private cancellation APIs.

Repositories and outcomes#

Use direct PgDatabaseService::with_connection/transaction, a Scoped context, or optional PgRepository. The postgresql feature selects single mode; for named pools/contexts use postgresql-factory instead and configure the factory model. Do not enable both in the same feature graph. The shared transaction guide explains the SQL unit of work; the WebSocket cancellation distinction is as described on this page.

ACK delivery can fail after a successful commit. Tie client retries to an application idempotency key. The example was compiled; this is not evidence of a live database schema or transaction integration test.