// Shared helpers used by the higher-level client APIs.
///|
/// Queue a request for the connection loop and return its response stream.
///
/// The caller chooses the response queue kind. Most requests use the bounded
/// per-request queue so unread streams eventually exert backpressure.
fn Client::send_request(
self : Client,
kind : RequestKind,
bytes : Bytes,
response_kind? : @aqueue.Kind = request_response_queue_kind,
) -> @async.Queue[@backend.Message] raise {
let responses = @async.Queue(kind=response_kind)
// `shared.requests` is created as `Unbounded`, so `try_put` cannot return
// `false` for "queue full". The only failure mode here is a closed queue,
// which still raises and propagates out of this helper.
ignore(
self.shared.requests.try_put({ kind, bytes, responses, copy_input: None, }),
)
responses
}
///|
/// Queue a COPY IN request that also carries a client-input queue.
fn Client::send_request_with_copy(
self : Client,
bytes : Bytes,
input : @async.Queue[CopyInAction],
) -> @async.Queue[@backend.Message] raise {
let responses = @async.Queue(kind=request_response_queue_kind)
// Same reasoning as `send_request`: the shared request queue is unbounded,
// so ignoring the `Bool` only discards an impossible "queue full" result.
ignore(
self.shared.requests.try_put({
kind: CopyIn,
bytes,
responses,
copy_input: Some(input),
}),
)
responses
}
///|
/// Fail if the shared runtime is already closed or shutting down.
fn ensure_open(shared : Shared) -> Unit raise {
if shared.closed.val || shared.closing.val {
// `closing` flips before the socket is fully torn down so new work stops
// racing against a connection that is already converging toward shutdown.
raise ClientError::Closed("connection is closed")
}
}
///|
/// Spawn one detached stream drain on the connection task group.
///
/// Terminal database errors are intentionally swallowed for detached streams,
/// but any other failure still tears down the task group so protocol bugs or
/// runtime failures are not silently ignored.
fn spawn_detached_drain(
background_group : @ref.Ref[@async.TaskGroup[Unit]?],
drain : async () -> Unit,
) -> Unit {
match background_group.val {
Some(group) =>
group.spawn_bg(no_wait=true, () => {
try drain() catch {
// Detached callers gave up synchronous handling of database failures,
// but protocol bugs and runtime errors should still surface.
err => if err is ClientError::Database(_) { () } else { raise err }
} noraise {
_ => ()
}
})
None => ()
}
}
///|
/// Fail if the transaction or savepoint handle has already been finished.
fn Transaction::assert_open(self : Transaction) -> Unit raise {
if self.finished.val {
raise ClientError::Closed("transaction is already finished")
}
}
///|
/// Mark the shared runtime as closed and close its public queues.
///
/// This is the central place where shutdown semantics are normalized for
/// graceful termination, error-driven teardown, and cancellation.
fn close_runtime(shared : Shared, error? : Error) -> Unit {
// Close the queues even if Client::close already marked the runtime closed
// after failing to enqueue Terminate.
// Flip the visible shutdown flags first so concurrent submitters observe a
// consistent "no new work" state even if queue closure races with them.
shared.closed.val = true
shared.closing.val = true
match error {
None => {
shared.requests.close()
shared.async_messages.close()
}
Some(err) => {
shared.requests.close(error=err, clear=true)
shared.async_messages.close()
}
}
}
///|
/// Return the next monotonically increasing runtime-local identifier.
fn next_id(shared : Shared) -> Int {
shared.next_id.val += 1
shared.next_id.val
}
///|
/// Build a protocol-safe identifier name for statements and portals.
fn next_name(shared : Shared, prefix : String) -> Bytes {
@proto.utf8_encode("\{prefix}\{next_id(shared).to_string()}")
}
///|
/// Seed the shared type cache with built-in descriptors known without catalog I/O.
fn seed_builtin_types(types : Map[@proto.Oid, Type]) -> Unit {
for
type_ in [
Type::bool(),
Type::bytea(),
Type::char(),
Type::name_type(),
Type::int8(),
Type::int2(),
Type::int4(),
Type::text(),
Type::oid_type(),
Type::json(),
Type::json_array(),
Type::float4(),
Type::float8(),
Type::varchar(),
Type::date(),
Type::time(),
Type::timestamp(),
Type::timestamptz(),
Type::uuid(),
Type::jsonb(),
Type::jsonb_array(),
Type::bool_array(),
Type::bytea_array(),
Type::int2_array(),
Type::int4_array(),
Type::text_array(),
Type::varchar_array(),
Type::int8_array(),
Type::float4_array(),
Type::float8_array(),
Type::timestamp_array(),
Type::date_array(),
Type::uuid_array(),
] {
types[type_.oid] = type_
}
}