cartel-pg 0.29.0

Async PostgreSQL driver for the dope runtime
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138

use cartel_core::reply::{order, usage};
use dope::{
    core::driver::route::KeyTag,
    manifold::client::session::{response, response::work, turns},
};

use crate::{
    port::{self, CONNECTOR_CONNECTION_LIMIT, replies, requests},
    protocol::{
        Reason,
        logic::{Owner, Session, State, drain, turn::Turn},
        runtime::tx::poll,
    },
    schema,
};

/// Linear response continuation moved into the connector connection slot.
#[doc(hidden)]
pub struct Pending<'d>(Kind<'d>);

enum Kind<'d> {
    Row(replies::Pending<'d>),
    Ready(poll::Status),
}

impl<'d> Pending<'d> {
    pub(super) const fn row(pending: replies::Pending<'d>) -> Self {
        Self(Kind::Row(pending))
    }

    pub(super) const fn ready(pending: poll::Status) -> Self {
        Self(Kind::Ready(pending))
    }

    fn settle_row<'life, I: schema::QuerySet, const ID: u8>(
        port: &'life port::Port<'life, '_, '_, '_, '_, '_, 'd, I, ID>,
        pending: replies::Pending<'d>,
        ctx: &mut turns::Context<
            '_,
            '_,
            'd,
            State<'life, 'd, I, ID>,
            requests::Frame<'d>,
            KeyTag<ID>,
        >,
    ) -> response::Settlement<Self> {
        let Some(queue) = port.responses(ctx.conn_id) else {
            Turn::new(port, ctx).fail(crate::Error::Protocol(Reason::RowUnknown));
            return response::Settlement::Ready;
        };
        match pending {
            replies::Pending::Row(row) => match queue.try_push(ctx.region(), row) {
                Ok(()) => response::Settlement::Ready,
                Err(usage::InsertError::Backpressure { value, .. }) => {
                    response::Settlement::Blocked(Self::row(replies::Pending::Row(value)))
                }
                Err(usage::InsertError::Capacity { cause, .. }) => {
                    Turn::new(port, ctx).fail(crate::Error::ReplyCapacity(cause));
                    response::Settlement::Ready
                }
                Err(usage::InsertError::Front {
                    front: order::Front::Detached,
                    ..
                }) => response::Settlement::Ready,
                Err(usage::InsertError::Front { front, .. }) => {
                    Turn::new(port, ctx).fail(crate::Error::ReplyFront(front));
                    response::Settlement::Ready
                }
            },
            replies::Pending::Failure(failed) => match queue.try_fail(ctx.region(), failed) {
                Ok(()) => response::Settlement::Ready,
                Err(usage::InsertError::Backpressure { value, .. }) => {
                    response::Settlement::Blocked(Self::row(replies::Pending::Failure(value)))
                }
                Err(usage::InsertError::Capacity { cause, .. }) => {
                    Turn::new(port, ctx).fail(crate::Error::ReplyCapacity(cause));
                    response::Settlement::Ready
                }
                Err(usage::InsertError::Front {
                    front: order::Front::Detached,
                    ..
                }) => {
                    queue.complete(ctx.region());
                    response::Settlement::Ready
                }
                Err(usage::InsertError::Front { front, .. }) => {
                    Turn::new(port, ctx).fail(crate::Error::ReplyFront(front));
                    response::Settlement::Ready
                }
            },
        }
    }
}

impl<'life, 'g, 'r, 's, 'b, 'c, 'd: 'life, I: schema::QuerySet, const ID: u8>
    response::Action<
        'life,
        'd,
        port::Port<'life, 'g, 'r, 's, 'b, 'c, 'd, I, ID>,
        Session,
        KeyTag<ID>,
        CONNECTOR_CONNECTION_LIMIT,
    > for Pending<'d>
{
    fn settle<'e, A: work::Admission>(
        item: response::Item<Self>,
        owner: &mut Owner<'life, 'g, 'r, 's, 'b, 'c, 'd, I, ID>,
        work: &mut A,
        ctx: &mut turns::Context<
            '_,
            'e,
            'd,
            State<'life, 'd, I, ID>,
            requests::Frame<'d>,
            KeyTag<ID>,
        >,
    ) -> response::Settlement<Self> {
        let port = owner.root();
        match item.into_inner().0 {
            Kind::Row(pending) => {
                match work.admit_with(pending, |pending| Self::settle_row(port, pending, ctx)) {
                    work::Capacity::Available(settled) => settled,
                    work::Capacity::Exhausted(pending) => {
                        response::Settlement::Yield(Self::row(pending))
                    }
                }
            }
            Kind::Ready(pending) => match Turn::new(port, ctx).try_settle(pending, work) {
                Ok(drain::Progress::Complete) => response::Settlement::Ready,
                Ok(drain::Progress::Blocked) => response::Settlement::Yield(Self::ready(pending)),
                Err(error) => {
                    Turn::new(port, ctx).fail(error);
                    response::Settlement::Ready
                }
            },
        }
    }
}