cartel-pg 0.24.0

Async PostgreSQL driver for the dope runtime
Documentation 
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
139
140

use dope::{
    core::driver::route::KeyTag,
    manifold::{
        client::{close, conn, session::turns},
        service::bindings::endpoint,
    },
};

use crate::{
    port::{
        self, Disposition,
        pool::{backend, key},
        replies,
    },
    protocol::{self, logic},
    schema,
};

pub(crate) struct Control<'p, 'g, 'r, 's, 'b, 'c, 'd, I: schema::QuerySet, const ID: u8> {
    port: &'p port::Port<'p, 'g, 'r, 's, 'b, 'c, 'd, I, ID>,
}

impl<'p, 'g, 'r, 's, 'b, 'c, 'd, I: schema::QuerySet, const ID: u8>
    Control<'p, 'g, 'r, 's, 'b, 'c, 'd, I, ID>
{
    pub(in crate::port) const fn new(
        port: &'p port::Port<'p, 'g, 'r, 's, 'b, 'c, 'd, I, ID>,
    ) -> Self {
        Self { port }
    }

    pub(crate) fn activate(
        &self,
        entry: &conn::Activation<'_, 'd, KeyTag<ID>>,
    ) -> Result<(), close::Directive> {
        let token = entry.connection();
        let ready = entry.ready();
        if matches!(
            self.port.shared.ready.bind(token, ready),
            protocol::runtime::ready::Bind::Rejected
        ) {
            return Err(self.reject());
        }
        self.port.lane(token.lane()).bind_retire(ready);
        Ok(())
    }

    fn reject(&self) -> close::Directive {
        let reason =
            close::Directive::cut_final(close::Reason::Local(close::LocalOrigin::ResourceFailure));
        self.port.shared.reject(reason.reason());
        reason
    }

    pub(crate) fn try_install(
        &self,
        token: conn::Id<'d, KeyTag<ID>>,
        key: key::Key,
    ) -> Result<(), protocol::Reason> {
        let Some(slot) = self.port.conn(token) else {
            return Err(protocol::Reason::BackendKeyStale);
        };
        backend::Backend::new(slot).try_install(token, key)
    }

    pub(crate) fn cancel(
        &self,
        token: conn::Id<'d, KeyTag<ID>>,
    ) -> Option<(backend::Bound<'p, 'd, ID>, i32)> {
        backend::Backend::new(self.port.conn(token)?).bind(token)
    }

    pub(crate) fn try_ready(
        &self,
        token: conn::Id<'d, KeyTag<ID>>,
    ) -> Result<(), protocol::Reason> {
        self.port.shared.ready.try_ready(token)?;
        if self.port.shared.ready.count() == self.port.capacity() {
            self.port.shared.signals.clear_open();
        }
        Ok(())
    }

    pub(crate) fn deactivate(
        &self,
        drain: turns::Drained<
            'd,
            KeyTag<ID>,
            endpoint::State<
                'p,
                'd,
                port::Port<'p, 'g, 'r, 's, 'b, 'c, 'd, I, ID>,
                logic::Session,
                KeyTag<ID>,
                { port::CONNECTOR_CONNECTION_LIMIT },
            >,
        >,
    ) {
        let token = drain.connection();
        self.port.shared.waiters().wake_all();
        let Some(slot) = self.port.conn(token) else {
            return;
        };
        if let Err(error) = slot.try_retire(token) {
            self.port.shared.fail(error);
        }
    }

    pub(crate) fn wake(&self, token: conn::Id<'d, KeyTag<ID>>) {
        if let Some(slot) = self.port.conn(token) {
            slot.wake();
        }
    }

    pub(crate) fn fail(&self, token: conn::Id<'d, KeyTag<ID>>, cause: close::Reason) {
        self.port.shared.wake_close(
            token,
            replies::Failure::from_close(cause),
            Disposition::Reconnect,
        );
    }

    pub(crate) fn reconnect(&self, token: conn::Id<'d, KeyTag<ID>>, cause: replies::Failure) {
        self.port
            .shared
            .wake_close(token, cause, Disposition::Reconnect);
    }

    pub(crate) fn try_fail(
        &self,
        token: conn::Id<'d, KeyTag<ID>>,
        cause: close::Reason,
    ) -> Option<replies::Failure> {
        self.port.shared.try_close(
            token,
            replies::Failure::from_close(cause),
            Disposition::Reconnect,
        )
    }
}