spin_factor_outbound_mysql/
host.rs

1use std::sync::Arc;
2
3use anyhow::Result;
4use opentelemetry_semantic_conventions::attribute as otel_attribute;
5use spin_core::wasmtime::component::{Accessor, FutureReader, Resource, StreamReader};
6use spin_factor_outbound_networking::ConnectionPermit;
7use spin_telemetry::traces::{self, Blame};
8use spin_world::MAX_HOST_BUFFERED_BYTES;
9use spin_world::spin::mysql::mysql as v3;
10use spin_world::v1::mysql as v1;
11use spin_world::v2::mysql as v2;
12use spin_world::v2::rdbms_types as v2_types;
13use tokio::sync::Mutex;
14use tracing::field::Empty;
15use tracing::{Level, instrument};
16
17use crate::client::Client;
18use crate::{InstanceState, InstanceStateInner, MysqlFactorData};
19
20impl<C: Client> InstanceStateInner<C> {
21    async fn open_connection(
22        &mut self,
23        address: &str,
24        permit: ConnectionPermit,
25    ) -> Result<u32, v2::Error> {
26        spin_factor_outbound_networking::record_address_fields(address);
27
28        if !self.is_address_allowed(address).await.map_err(|e| {
29            // The allow-list check infrastructure itself failed; that's a
30            // host problem, not anything the guest did wrong.
31            let err = v2::Error::Other(e.to_string());
32            traces::mark_as_error(&err, Some(Blame::Host));
33            err
34        })? {
35            // The check succeeded but returned false: the guest supplied an
36            // address that isn't on the allow list.
37            let err = v2::Error::ConnectionFailed(format!("address {address} is not permitted"));
38            traces::mark_as_error(&err, Some(Blame::Guest));
39            return Err(err);
40        }
41        let client = C::build_client(address).await.map_err(|e| {
42            // The guest supplies the address and credentials; connection
43            // failures (wrong password, TLS error, unreachable host, etc.)
44            // are the guest's problem.
45            let err = v2::Error::ConnectionFailed(format!("{e:?}"));
46            traces::mark_as_error(&err, Some(Blame::Guest));
47            err
48        })?;
49        self.connections
50            .push((Arc::new(Mutex::new(client)), permit))
51            .map_err(|_| {
52                // The guest exceeded the host-imposed connection limit.
53                let err = v2::Error::ConnectionFailed("too many connections".into());
54                traces::mark_as_error(&err, Some(Blame::Guest));
55                err
56            })
57    }
58
59    fn get_client(&mut self, connection: u32) -> Result<Arc<Mutex<C>>, v2::Error> {
60        self.connections
61            .get(connection)
62            .map(|(conn, _permit)| conn.clone())
63            .ok_or_else(|| {
64                // The connection table is managed entirely by the host, so a
65                // missing handle indicates a host-side bug, not a guest mistake.
66                let err = v2::Error::ConnectionFailed("no connection found".into());
67                traces::mark_as_error(&err, Some(Blame::Host));
68                err
69            })
70    }
71
72    async fn is_address_allowed(&self, address: &str) -> Result<bool> {
73        self.allowed_hosts.check_url(address, "mysql").await
74    }
75}
76
77impl<C: Client> v3::Host for InstanceState<C> {
78    fn convert_error(&mut self, error: v3::Error) -> Result<v3::Error> {
79        Ok(error)
80    }
81}
82
83impl<C: Client> v3::HostConnection for InstanceState<C> {
84    async fn drop(&mut self, connection: Resource<v3::Connection>) -> Result<()> {
85        let mut state = self.inner.lock().await;
86        state.connections.remove(connection.rep());
87        Ok(())
88    }
89}
90
91type QueryTuple = (
92    Vec<v3::Column>,
93    StreamReader<v3::Row>,
94    FutureReader<Result<(), v3::Error>>,
95);
96
97impl<C: Client, T> v3::HostConnectionWithStore<T> for MysqlFactorData<C> {
98    #[instrument(name = "spin_outbound_mysql.open", skip(accessor, address), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))]
99    async fn open(
100        accessor: &Accessor<T, Self>,
101        address: String,
102    ) -> Result<Resource<v3::Connection>, v3::Error> {
103        let (state_arc, semaphore) = accessor.with(|mut access| {
104            let host = access.get();
105            (host.inner.clone(), host.semaphore.clone())
106        });
107        let permit = semaphore
108            .acquire()
109            .await
110            .map_err(|_| v3::Error::ConnectionFailed("too many connections".into()))?;
111        let mut state = state_arc.lock().await;
112        state.otel.reparent_tracing_span();
113        Ok(Resource::new_own(
114            state.open_connection(&address, permit).await?,
115        ))
116    }
117
118    #[instrument(name = "spin_outbound_mysql.execute", skip(accessor, connection, params), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql", otel.name = statement))]
119    async fn execute(
120        accessor: &Accessor<T, Self>,
121        connection: Resource<v3::Connection>,
122        statement: String,
123        params: Vec<v3::ParameterValue>,
124    ) -> Result<(), v3::Error> {
125        let state = accessor.with(|mut access| access.get().inner.clone());
126        let client = {
127            let mut state = state.lock().await;
128            state.otel.reparent_tracing_span();
129            state.get_client(connection.rep())?
130        };
131        client
132            .lock()
133            .await
134            .execute(statement, params.into_iter().map(Into::into).collect())
135            .await
136            .map_err(track_db_error_on_span)?;
137        Ok(())
138    }
139
140    #[instrument(name = "spin_outbound_mysql.query", skip(accessor, connection, params), err(level = Level::INFO), fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql", otel.name = statement))]
141    async fn query(
142        accessor: &Accessor<T, Self>,
143        connection: Resource<v3::Connection>,
144        statement: String,
145        params: Vec<v3::ParameterValue>,
146    ) -> Result<QueryTuple, v3::Error> {
147        let state = accessor.with(|mut access| access.get().inner.clone());
148        let client = {
149            let mut state = state.lock().await;
150            state.otel.reparent_tracing_span();
151            state.get_client(connection.rep())?
152        };
153
154        let (columns, stream, future) =
155            C::query_async(client, statement, params, MAX_HOST_BUFFERED_BYTES)
156                .await
157                .map_err(|v| v3::Error::from(track_db_error_on_span(v2::Error::from(v))))?;
158
159        let (stream, future) = accessor
160            .with(|mut access| {
161                anyhow::Ok((
162                    StreamReader::new(&mut access, spin_wasi_async::stream::producer(stream))?,
163                    FutureReader::new(&mut access, future)?,
164                ))
165            })
166            .map_err(|e| {
167                // Setting up the async stream/future channels is a host
168                // implementation detail; if it fails, that's a host bug.
169                let err = v3::Error::Other(e.to_string());
170                traces::mark_as_error(&err, Some(Blame::Host));
171                err
172            })?;
173
174        Ok((columns, stream, future))
175    }
176}
177
178impl<C: Client> v2::Host for InstanceState<C> {}
179
180impl<C: Client> v2::HostConnection for InstanceState<C> {
181    #[instrument(name = "spin_outbound_mysql.open", skip(self, address), err(level = Level::INFO),
182        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))]
183    async fn open(&mut self, address: String) -> Result<Resource<v2::Connection>, v2::Error> {
184        let permit = self
185            .semaphore
186            .acquire()
187            .await
188            .map_err(|_| v2::Error::ConnectionFailed("too many connections".into()))?;
189        let mut state = self.inner.lock().await;
190        state.otel.reparent_tracing_span();
191        state
192            .open_connection(&address, permit)
193            .await
194            .map(Resource::new_own)
195    }
196
197    #[instrument(name = "spin_outbound_mysql.execute", skip(self, connection, params), err(level = Level::INFO),
198        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql"))]
199    async fn execute(
200        &mut self,
201        connection: Resource<v2::Connection>,
202        statement: String,
203        params: Vec<v2_types::ParameterValue>,
204    ) -> Result<(), v2::Error> {
205        let mut state = self.inner.lock().await;
206        state.otel.reparent_tracing_span();
207        state
208            .get_client(connection.rep())?
209            .lock()
210            .await
211            .execute(statement, params)
212            .await
213            .map_err(track_db_error_on_span)
214    }
215
216    #[instrument(name = "spin_outbound_mysql.query", skip(self, connection, params), err(level = Level::INFO),
217        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "mysql"))]
218    async fn query(
219        &mut self,
220        connection: Resource<v2::Connection>,
221        statement: String,
222        params: Vec<v2_types::ParameterValue>,
223    ) -> Result<v2_types::RowSet, v2::Error> {
224        let mut state = self.inner.lock().await;
225        state.otel.reparent_tracing_span();
226        state
227            .get_client(connection.rep())?
228            .lock()
229            .await
230            .query(statement, params, MAX_HOST_BUFFERED_BYTES)
231            .await
232            .map_err(track_db_error_on_span)
233    }
234
235    async fn drop(&mut self, connection: Resource<v2::Connection>) -> Result<()> {
236        let mut state = self.inner.lock().await;
237        state.connections.remove(connection.rep());
238        Ok(())
239    }
240}
241
242impl<C: Send> v2_types::Host for InstanceState<C> {
243    fn convert_error(&mut self, error: v2::Error) -> Result<v2::Error> {
244        Ok(error)
245    }
246}
247
248/// Delegate a function call to the v2::HostConnection implementation
249macro_rules! delegate {
250    ($self:ident.$name:ident($address:expr, $($arg:expr),*)) => {{
251        let permit = $self
252            .semaphore
253            .acquire()
254            .await
255            .map_err(|_| v2::Error::ConnectionFailed("too many connections".into()))?;
256        let connection = {
257            let mut state = $self.inner.lock().await;
258            Resource::new_own(state.open_connection(&$address, permit).await?)
259        };
260        // v1 has no persistent connections, so remove the table entry immediately
261        // after the call to release the semaphore permit.
262        let rep = connection.rep();
263        let result = <Self as v2::HostConnection>::$name($self, connection, $($arg),*)
264            .await
265            .map_err(Into::into);
266        $self.inner.lock().await.connections.remove(rep);
267        result
268    }};
269}
270
271impl<C: Client> v1::Host for InstanceState<C> {
272    async fn execute(
273        &mut self,
274        address: String,
275        statement: String,
276        params: Vec<v1::ParameterValue>,
277    ) -> Result<(), v1::MysqlError> {
278        delegate!(self.execute(
279            address,
280            statement,
281            params.into_iter().map(Into::into).collect()
282        ))
283    }
284
285    async fn query(
286        &mut self,
287        address: String,
288        statement: String,
289        params: Vec<v1::ParameterValue>,
290    ) -> Result<v1::RowSet, v1::MysqlError> {
291        delegate!(self.query(
292            address,
293            statement,
294            params.into_iter().map(Into::into).collect()
295        ))
296        .map(Into::into)
297    }
298
299    fn convert_mysql_error(&mut self, error: v1::MysqlError) -> Result<v1::MysqlError> {
300        Ok(error)
301    }
302}
303
304/// Only for actual DB client calls (execute/query).
305/// Blame is inferred from the error variant returned by the DB driver.
306fn track_db_error_on_span(err: v2::Error) -> v2::Error {
307    let blame = match &err {
308        // The guest brings their own database, so connection failures during
309        // execution (dropped connection, auth rejected mid-session, etc.) are
310        // the guest's problem, not the host's.
311        v2::Error::ConnectionFailed(_) => Blame::Guest,
312        v2::Error::BadParameter(_) => Blame::Guest,
313        v2::Error::QueryFailed(_) => Blame::Guest,
314        // The host is responsible for mapping DB wire types to WIT types;
315        // a conversion failure is a host-side limitation or bug.
316        v2::Error::ValueConversionFailed(_) => Blame::Host,
317        v2::Error::Other(_) => Blame::Host,
318    };
319    traces::mark_as_error(&err, Some(blame));
320    err
321}
spin_factor_outbound_mysql/host.rs

spin_factor_outbound_mysql/
host.rs
