spin_factor_outbound_redis/
host.rs

1use std::net::SocketAddr;
2
3use anyhow::Result;
4use opentelemetry_semantic_conventions::attribute as otel_attribute;
5use redis::AsyncConnectionConfig;
6use redis::io::AsyncDNSResolver;
7use redis::{AsyncCommands, FromRedisValue, Value, aio::MultiplexedConnection};
8use spin_core::wasmtime::component::{Accessor, Resource};
9use spin_factor_otel::OtelFactorState;
10use spin_factor_outbound_networking::ConnectionSemaphore;
11use spin_factor_outbound_networking::config::blocked_networks::BlockedNetworks;
12use spin_world::MAX_HOST_BUFFERED_BYTES;
13use spin_world::spin::redis::redis as v3;
14use spin_world::v1::{redis as v1, redis_types};
15use spin_world::v2::redis as v2;
16use tracing::field::Empty;
17use tracing::{Level, instrument};
18
19use crate::allowed_hosts::AllowedHostChecker;
20
21pub struct InstanceState {
22    pub(crate) allowed_host_checker: AllowedHostChecker,
23    pub blocked_networks: BlockedNetworks,
24    pub connections: spin_resource_table::Table<(
25        MultiplexedConnection,
26        spin_factor_outbound_networking::ConnectionPermit,
27    )>,
28    pub semaphore: ConnectionSemaphore,
29    pub otel: OtelFactorState,
30}
31
32impl InstanceState {
33    async fn is_address_allowed(&self, address: &str) -> Result<bool> {
34        self.allowed_host_checker.is_address_allowed(address).await
35    }
36
37    async fn establish_connection(
38        &mut self,
39        address: String,
40    ) -> Result<Resource<v2::Connection>, v2::Error> {
41        let permit = self
42            .semaphore
43            .acquire()
44            .await
45            .map_err(|_| v2::Error::TooManyConnections)?;
46        let config = AsyncConnectionConfig::new()
47            .set_dns_resolver(SpinDnsResolver(self.blocked_networks.clone()));
48        let conn = redis::Client::open(address.as_str())
49            .map_err(|_| v2::Error::InvalidAddress)?
50            .get_multiplexed_async_connection_with_config(&config)
51            .await
52            .map_err(other_error_v2)?;
53        self.connections
54            .push((conn, permit))
55            .map(Resource::new_own)
56            .map_err(|_| v2::Error::TooManyConnections)
57    }
58
59    async fn get_conn(
60        &mut self,
61        connection: Resource<v2::Connection>,
62    ) -> Result<&mut MultiplexedConnection, v2::Error> {
63        self.connections
64            .get_mut(connection.rep())
65            .map(|(conn, _permit)| conn)
66            .ok_or(v2::Error::Other(
67                "could not find connection for resource".into(),
68            ))
69    }
70
71    fn get_conn_v3(
72        &mut self,
73        connection: Resource<v3::Connection>,
74    ) -> Result<MultiplexedConnection, v3::Error> {
75        self.connections
76            .get(connection.rep())
77            .map(|(conn, _permit)| conn.clone())
78            .ok_or(v3::Error::Other(
79                "could not find connection for resource".into(),
80            ))
81    }
82}
83
84mod operations {
85    use super::*;
86
87    pub async fn publish(
88        conn: &mut MultiplexedConnection,
89        channel: String,
90        payload: v3::Payload,
91    ) -> Result<(), v3::Error> {
92        // The `let () =` syntax is needed to suppress a warning when the result type is inferred.
93        // You can read more about the issue here: <https://github.com/redis-rs/redis-rs/issues/1228>
94        let () = conn
95            .publish(&channel, &payload)
96            .await
97            .map_err(other_error_v3)?;
98        Ok(())
99    }
100
101    pub async fn get(
102        conn: &mut MultiplexedConnection,
103        key: String,
104    ) -> Result<Option<Vec<u8>>, v3::Error> {
105        let value = conn
106            .get::<_, Option<Vec<u8>>>(&key)
107            .await
108            .map_err(other_error_v3)?;
109
110        // Currently there's no way to stream a `GET` result using the `redis`
111        // crate without buffering, so the damage (in terms of host memory
112        // usage) is already done, but we can still enforce the limit:
113        if std::mem::size_of::<Option<Vec<u8>>>() + value.as_ref().map(|v| v.len()).unwrap_or(0)
114            > MAX_HOST_BUFFERED_BYTES
115        {
116            Err(v3::Error::Other(format!(
117                "query result exceeds limit of {MAX_HOST_BUFFERED_BYTES} bytes"
118            )))
119        } else {
120            Ok(value)
121        }
122    }
123
124    pub async fn set(
125        conn: &mut MultiplexedConnection,
126        key: String,
127        value: Vec<u8>,
128    ) -> Result<(), v3::Error> {
129        // The `let () =` syntax is needed to suppress a warning when the result type is inferred.
130        // You can read more about the issue here: <https://github.com/redis-rs/redis-rs/issues/1228>
131        let () = conn.set(&key, &value).await.map_err(other_error_v3)?;
132        Ok(())
133    }
134
135    pub async fn incr(conn: &mut MultiplexedConnection, key: String) -> Result<i64, v3::Error> {
136        conn.incr(&key, 1).await.map_err(other_error_v3)
137    }
138
139    pub async fn del(
140        conn: &mut MultiplexedConnection,
141        keys: Vec<String>,
142    ) -> Result<u32, v3::Error> {
143        conn.del(&keys).await.map_err(other_error_v3)
144    }
145
146    pub async fn sadd(
147        conn: &mut MultiplexedConnection,
148        key: String,
149        values: Vec<String>,
150    ) -> Result<u32, v3::Error> {
151        let value = conn.sadd(&key, &values).await.map_err(|e| {
152            if e.kind() == redis::ErrorKind::TypeError {
153                v3::Error::TypeError
154            } else {
155                v3::Error::Other(e.to_string())
156            }
157        })?;
158        Ok(value)
159    }
160
161    pub async fn smembers(
162        conn: &mut MultiplexedConnection,
163        key: String,
164    ) -> Result<Vec<String>, v3::Error> {
165        conn.smembers(&key).await.map_err(other_error_v3)
166    }
167
168    pub async fn srem(
169        conn: &mut MultiplexedConnection,
170        key: String,
171        values: Vec<String>,
172    ) -> Result<u32, v3::Error> {
173        conn.srem(&key, &values).await.map_err(other_error_v3)
174    }
175
176    pub async fn execute(
177        conn: &mut MultiplexedConnection,
178        command: String,
179        arguments: impl IntoIterator<Item = v3::RedisParameter>,
180    ) -> Result<RedisResults, v3::Error> {
181        let mut cmd = redis::cmd(&command);
182        arguments.into_iter().for_each(|value| match value {
183            v3::RedisParameter::Int64(v) => {
184                cmd.arg(v);
185            }
186            v3::RedisParameter::Binary(v) => {
187                cmd.arg(v);
188            }
189        });
190
191        let results = cmd
192            .query_async::<RedisResults>(conn)
193            .await
194            .map_err(other_error_v3)?;
195
196        // Currently there's no way to stream results using the `redis`
197        // crate without buffering, so the damage (in terms of host memory
198        // usage) is already done, but we can still enforce the limit:
199        if std::mem::size_of::<Vec<v3::RedisResult>>()
200            + results.0.iter().map(memory_size).sum::<usize>()
201            > MAX_HOST_BUFFERED_BYTES
202        {
203            Err(v3::Error::Other(format!(
204                "query result exceeds limit of {MAX_HOST_BUFFERED_BYTES} bytes"
205            )))
206        } else {
207            Ok(results)
208        }
209    }
210}
211
212impl v3::Host for crate::InstanceState {
213    fn convert_error(&mut self, error: v3::Error) -> anyhow::Result<v3::Error> {
214        Ok(error)
215    }
216}
217
218impl v3::HostConnection for crate::InstanceState {
219    async fn drop(&mut self, connection: Resource<v3::Connection>) -> anyhow::Result<()> {
220        self.connections.remove(connection.rep());
221        Ok(())
222    }
223}
224
225impl crate::RedisFactorData {
226    fn get_conn<T: Send>(
227        accessor: &Accessor<T, Self>,
228        connection: Resource<v3::Connection>,
229    ) -> Result<MultiplexedConnection, v3::Error> {
230        accessor.with(|mut access| {
231            let host = access.get();
232            host.otel.reparent_tracing_span();
233            host.get_conn_v3(connection)
234        })
235    }
236}
237
238impl<T: Send> v3::HostConnectionWithStore<T> for crate::RedisFactorData {
239    #[instrument(name = "spin_outbound_redis.open_connection", skip(accessor, address), err(level = Level::INFO),
240        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))]
241    async fn open(
242        accessor: &Accessor<T, Self>,
243        address: String,
244    ) -> Result<Resource<v3::Connection>, v3::Error> {
245        let (allowed_host_checker, blocked_networks, semaphore) = accessor.with(|mut access| {
246            let host = access.get();
247            host.otel.reparent_tracing_span();
248            (
249                host.allowed_host_checker.clone(),
250                host.blocked_networks.clone(),
251                host.semaphore.clone(),
252            )
253        });
254
255        if !allowed_host_checker
256            .is_address_allowed(&address)
257            .await
258            .map_err(|e| v3::Error::Other(e.to_string()))?
259        {
260            return Err(v3::Error::InvalidAddress);
261        }
262
263        let permit = semaphore
264            .acquire()
265            .await
266            .map_err(|_| v3::Error::TooManyConnections)?;
267
268        let config =
269            AsyncConnectionConfig::new().set_dns_resolver(SpinDnsResolver(blocked_networks));
270        let conn = redis::Client::open(address.as_str())
271            .map_err(|_| v3::Error::InvalidAddress)?
272            .get_multiplexed_async_connection_with_config(&config)
273            .await
274            .map_err(other_error_v3)?;
275
276        accessor.with(|mut access| {
277            let host = access.get();
278            host.connections
279                .push((conn, permit))
280                .map(Resource::new_own)
281                .map_err(|_| v3::Error::TooManyConnections)
282        })
283    }
284
285    #[instrument(name = "spin_outbound_redis.publish", skip(accessor, connection, payload), err(level = Level::INFO),
286        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "PUBLISH"))]
287    async fn publish(
288        accessor: &Accessor<T, Self>,
289        connection: Resource<v3::Connection>,
290        channel: String,
291        payload: v3::Payload,
292    ) -> Result<(), v3::Error> {
293        let mut conn = Self::get_conn(accessor, connection)?;
294        operations::publish(&mut conn, channel, payload).await
295    }
296
297    #[instrument(name = "spin_outbound_redis.get", skip(accessor, connection), err(level = Level::INFO),
298        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "GET"))]
299    async fn get(
300        accessor: &Accessor<T, Self>,
301        connection: Resource<v3::Connection>,
302        key: String,
303    ) -> Result<Option<v3::Payload>, v3::Error> {
304        let mut conn = Self::get_conn(accessor, connection)?;
305        operations::get(&mut conn, key).await
306    }
307
308    #[instrument(name = "spin_outbound_redis.set", skip(accessor, connection, value), err(level = Level::INFO),
309        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SET"))]
310    async fn set(
311        accessor: &Accessor<T, Self>,
312        connection: Resource<v3::Connection>,
313        key: String,
314        value: v3::Payload,
315    ) -> Result<(), v3::Error> {
316        let mut conn = Self::get_conn(accessor, connection)?;
317        operations::set(&mut conn, key, value).await
318    }
319
320    #[instrument(name = "spin_outbound_redis.incr", skip(accessor, connection), err(level = Level::INFO),
321        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "INCRBY"))]
322    async fn incr(
323        accessor: &Accessor<T, Self>,
324        connection: Resource<v3::Connection>,
325        key: String,
326    ) -> Result<i64, v3::Error> {
327        let mut conn = Self::get_conn(accessor, connection)?;
328        operations::incr(&mut conn, key).await
329    }
330
331    #[instrument(name = "spin_outbound_redis.del", skip(accessor, connection), err(level = Level::INFO),
332        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "DEL"))]
333    async fn del(
334        accessor: &Accessor<T, Self>,
335        connection: Resource<v3::Connection>,
336        keys: Vec<String>,
337    ) -> Result<u32, v3::Error> {
338        let mut conn = Self::get_conn(accessor, connection)?;
339        operations::del(&mut conn, keys).await
340    }
341
342    #[instrument(name = "spin_outbound_redis.sadd", skip(accessor, connection, values), err(level = Level::INFO),
343        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SADD"))]
344    async fn sadd(
345        accessor: &Accessor<T, Self>,
346        connection: Resource<v3::Connection>,
347        key: String,
348        values: Vec<String>,
349    ) -> Result<u32, v3::Error> {
350        let mut conn = Self::get_conn(accessor, connection)?;
351        operations::sadd(&mut conn, key, values).await
352    }
353
354    #[instrument(name = "spin_outbound_redis.smembers", skip(accessor, connection), err(level = Level::INFO),
355        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SMEMBERS"))]
356    async fn smembers(
357        accessor: &Accessor<T, Self>,
358        connection: Resource<v3::Connection>,
359        key: String,
360    ) -> Result<Vec<String>, v3::Error> {
361        let mut conn = Self::get_conn(accessor, connection)?;
362        operations::smembers(&mut conn, key).await
363    }
364
365    #[instrument(name = "spin_outbound_redis.srem", skip(accessor, connection, values), err(level = Level::INFO),
366        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SREM"))]
367    async fn srem(
368        accessor: &Accessor<T, Self>,
369        connection: Resource<v3::Connection>,
370        key: String,
371        values: Vec<String>,
372    ) -> Result<u32, v3::Error> {
373        let mut conn = Self::get_conn(accessor, connection)?;
374        operations::srem(&mut conn, key, values).await
375    }
376
377    #[instrument(name = "spin_outbound_redis.execute", skip(accessor, connection), err(level = Level::INFO),
378        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = format!("{}", command)))]
379    async fn execute(
380        accessor: &Accessor<T, Self>,
381        connection: Resource<v3::Connection>,
382        command: String,
383        arguments: Vec<v3::RedisParameter>,
384    ) -> Result<Vec<v3::RedisResult>, v3::Error> {
385        let mut conn = Self::get_conn(accessor, connection)?;
386        Ok(operations::execute(&mut conn, command, arguments)
387            .await?
388            .into_v3())
389    }
390}
391
392impl v2::Host for crate::InstanceState {
393    fn convert_error(&mut self, error: v2::Error) -> Result<v2::Error> {
394        Ok(error)
395    }
396}
397
398impl v2::HostConnection for crate::InstanceState {
399    #[instrument(name = "spin_outbound_redis.open_connection", skip(self, address), err(level = Level::INFO),
400        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", {otel_attribute::SERVER_ADDRESS} = Empty, {otel_attribute::SERVER_PORT} = Empty, {otel_attribute::DB_NAMESPACE} = Empty))]
401    async fn open(&mut self, address: String) -> Result<Resource<v2::Connection>, v2::Error> {
402        self.otel.reparent_tracing_span();
403        if !self
404            .is_address_allowed(&address)
405            .await
406            .map_err(|e| v2::Error::Other(e.to_string()))?
407        {
408            return Err(v2::Error::InvalidAddress);
409        }
410
411        self.establish_connection(address).await
412    }
413
414    #[instrument(name = "spin_outbound_redis.publish", skip(self, connection, payload), err(level = Level::INFO),
415        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "PUBLISH"))]
416    async fn publish(
417        &mut self,
418        connection: Resource<v2::Connection>,
419        channel: String,
420        payload: Vec<u8>,
421    ) -> Result<(), v2::Error> {
422        self.otel.reparent_tracing_span();
423
424        let conn = self.get_conn(connection).await.map_err(other_error_v2)?;
425
426        Ok(operations::publish(conn, channel, payload).await?)
427    }
428
429    #[instrument(name = "spin_outbound_redis.get", skip(self, connection), err(level = Level::INFO),
430        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "GET"))]
431    async fn get(
432        &mut self,
433        connection: Resource<v2::Connection>,
434        key: String,
435    ) -> Result<Option<Vec<u8>>, v2::Error> {
436        self.otel.reparent_tracing_span();
437
438        let conn = self.get_conn(connection).await.map_err(other_error_v2)?;
439
440        Ok(operations::get(conn, key).await?)
441    }
442
443    #[instrument(name = "spin_outbound_redis.set", skip(self, connection, value), err(level = Level::INFO),
444        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SET"))]
445    async fn set(
446        &mut self,
447        connection: Resource<v2::Connection>,
448        key: String,
449        value: Vec<u8>,
450    ) -> Result<(), v2::Error> {
451        self.otel.reparent_tracing_span();
452
453        let conn = self.get_conn(connection).await.map_err(other_error_v2)?;
454        Ok(operations::set(conn, key, value).await?)
455    }
456
457    #[instrument(name = "spin_outbound_redis.incr", skip(self, connection), err(level = Level::INFO),
458        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "INCRBY"))]
459    async fn incr(
460        &mut self,
461        connection: Resource<v2::Connection>,
462        key: String,
463    ) -> Result<i64, v2::Error> {
464        self.otel.reparent_tracing_span();
465
466        let conn = self.get_conn(connection).await.map_err(other_error_v2)?;
467        Ok(operations::incr(conn, key).await?)
468    }
469
470    #[instrument(name = "spin_outbound_redis.del", skip(self, connection), err(level = Level::INFO),
471        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "DEL"))]
472    async fn del(
473        &mut self,
474        connection: Resource<v2::Connection>,
475        keys: Vec<String>,
476    ) -> Result<u32, v2::Error> {
477        self.otel.reparent_tracing_span();
478
479        let conn = self.get_conn(connection).await.map_err(other_error_v2)?;
480        Ok(operations::del(conn, keys).await?)
481    }
482
483    #[instrument(name = "spin_outbound_redis.sadd", skip(self, connection, values), err(level = Level::INFO),
484        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SADD"))]
485    async fn sadd(
486        &mut self,
487        connection: Resource<v2::Connection>,
488        key: String,
489        values: Vec<String>,
490    ) -> Result<u32, v2::Error> {
491        self.otel.reparent_tracing_span();
492
493        let conn = self.get_conn(connection).await.map_err(other_error_v2)?;
494        Ok(operations::sadd(conn, key, values).await?)
495    }
496
497    #[instrument(name = "spin_outbound_redis.smembers", skip(self, connection), err(level = Level::INFO),
498        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SMEMBERS"))]
499    async fn smembers(
500        &mut self,
501        connection: Resource<v2::Connection>,
502        key: String,
503    ) -> Result<Vec<String>, v2::Error> {
504        self.otel.reparent_tracing_span();
505
506        let conn = self.get_conn(connection).await.map_err(other_error_v2)?;
507        Ok(operations::smembers(conn, key).await?)
508    }
509
510    #[instrument(name = "spin_outbound_redis.srem", skip(self, connection, values), err(level = Level::INFO),
511        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = "SREM"))]
512    async fn srem(
513        &mut self,
514        connection: Resource<v2::Connection>,
515        key: String,
516        values: Vec<String>,
517    ) -> Result<u32, v2::Error> {
518        self.otel.reparent_tracing_span();
519
520        let conn = self.get_conn(connection).await.map_err(other_error_v2)?;
521        Ok(operations::srem(conn, key, values).await?)
522    }
523
524    #[instrument(name = "spin_outbound_redis.execute", skip(self, connection, arguments), err(level = Level::INFO),
525        fields(otel.kind = "client", {otel_attribute::DB_SYSTEM_NAME} = "redis", otel.name = command))]
526    async fn execute(
527        &mut self,
528        connection: Resource<v2::Connection>,
529        command: String,
530        arguments: Vec<v2::RedisParameter>,
531    ) -> Result<Vec<v2::RedisResult>, v2::Error> {
532        fn to_v3_param(value: v2::RedisParameter) -> v3::RedisParameter {
533            match value {
534                v2::RedisParameter::Int64(v) => v3::RedisParameter::Int64(v),
535                v2::RedisParameter::Binary(v) => v3::RedisParameter::Binary(v),
536            }
537        }
538
539        self.otel.reparent_tracing_span();
540
541        let conn = self.get_conn(connection).await?;
542
543        let arguments = arguments.into_iter().map(to_v3_param);
544        Ok(operations::execute(conn, command, arguments)
545            .await?
546            .into_v2())
547    }
548
549    async fn drop(&mut self, connection: Resource<v2::Connection>) -> anyhow::Result<()> {
550        self.connections.remove(connection.rep());
551        Ok(())
552    }
553}
554
555fn other_error_v2(e: impl std::fmt::Display) -> v2::Error {
556    v2::Error::Other(e.to_string())
557}
558
559fn other_error_v3(e: impl std::fmt::Display) -> v3::Error {
560    v3::Error::Other(e.to_string())
561}
562
563/// Delegate a function call to the v2::HostConnection implementation
564macro_rules! delegate {
565    ($self:ident.$name:ident($address:expr, $($arg:expr),*)) => {{
566        if !$self.is_address_allowed(&$address).await.map_err(|_| v1::Error::Error)?  {
567            return Err(v1::Error::Error);
568        }
569        let connection = match $self.establish_connection($address).await {
570            Ok(c) => c,
571            Err(_) => return Err(v1::Error::Error),
572        };
573        // v1 has no persistent connections, so remove the table entry immediately
574        // after the call to release the semaphore permit.
575        let rep = connection.rep();
576        let result = <Self as v2::HostConnection>::$name($self, connection, $($arg),*)
577            .await
578            .map_err(|_| v1::Error::Error);
579        $self.connections.remove(rep);
580        result
581    }};
582}
583
584impl v1::Host for crate::InstanceState {
585    async fn publish(
586        &mut self,
587        address: String,
588        channel: String,
589        payload: Vec<u8>,
590    ) -> Result<(), v1::Error> {
591        delegate!(self.publish(address, channel, payload))
592    }
593
594    async fn get(&mut self, address: String, key: String) -> Result<Vec<u8>, v1::Error> {
595        delegate!(self.get(address, key)).map(|v| v.unwrap_or_default())
596    }
597
598    async fn set(&mut self, address: String, key: String, value: Vec<u8>) -> Result<(), v1::Error> {
599        delegate!(self.set(address, key, value))
600    }
601
602    async fn incr(&mut self, address: String, key: String) -> Result<i64, v1::Error> {
603        delegate!(self.incr(address, key))
604    }
605
606    async fn del(&mut self, address: String, keys: Vec<String>) -> Result<i64, v1::Error> {
607        delegate!(self.del(address, keys)).map(|v| v as i64)
608    }
609
610    async fn sadd(
611        &mut self,
612        address: String,
613        key: String,
614        values: Vec<String>,
615    ) -> Result<i64, v1::Error> {
616        delegate!(self.sadd(address, key, values)).map(|v| v as i64)
617    }
618
619    async fn smembers(&mut self, address: String, key: String) -> Result<Vec<String>, v1::Error> {
620        delegate!(self.smembers(address, key))
621    }
622
623    async fn srem(
624        &mut self,
625        address: String,
626        key: String,
627        values: Vec<String>,
628    ) -> Result<i64, v1::Error> {
629        delegate!(self.srem(address, key, values)).map(|v| v as i64)
630    }
631
632    async fn execute(
633        &mut self,
634        address: String,
635        command: String,
636        arguments: Vec<v1::RedisParameter>,
637    ) -> Result<Vec<v1::RedisResult>, v1::Error> {
638        delegate!(self.execute(
639            address,
640            command,
641            arguments.into_iter().map(Into::into).collect()
642        ))
643        .map(|v| v.into_iter().map(Into::into).collect())
644    }
645}
646
647impl redis_types::Host for crate::InstanceState {
648    fn convert_error(&mut self, error: redis_types::Error) -> Result<redis_types::Error> {
649        Ok(error)
650    }
651}
652
653struct RedisResults(Vec<v3::RedisResult>);
654
655impl RedisResults {
656    fn into_v2(self) -> Vec<v2::RedisResult> {
657        fn into_v2(value: v3::RedisResult) -> v2::RedisResult {
658            match value {
659                v3::RedisResult::Nil => v2::RedisResult::Nil,
660                v3::RedisResult::Status(v) => v2::RedisResult::Status(v),
661                v3::RedisResult::Int64(v) => v2::RedisResult::Int64(v),
662                v3::RedisResult::Binary(v) => v2::RedisResult::Binary(v),
663            }
664        }
665
666        self.0.into_iter().map(into_v2).collect()
667    }
668
669    fn into_v3(self) -> Vec<v3::RedisResult> {
670        self.0
671    }
672}
673
674impl FromRedisValue for RedisResults {
675    fn from_redis_value(value: &Value) -> redis::RedisResult<Self> {
676        fn append(values: &mut Vec<v3::RedisResult>, value: &Value) -> redis::RedisResult<()> {
677            match value {
678                Value::Nil => {
679                    values.push(v3::RedisResult::Nil);
680                    Ok(())
681                }
682                Value::Int(v) => {
683                    values.push(v3::RedisResult::Int64(*v));
684                    Ok(())
685                }
686                Value::BulkString(bytes) => {
687                    values.push(v3::RedisResult::Binary(bytes.to_owned()));
688                    Ok(())
689                }
690                Value::SimpleString(s) => {
691                    values.push(v3::RedisResult::Status(s.to_owned()));
692                    Ok(())
693                }
694                Value::Okay => {
695                    values.push(v3::RedisResult::Status("OK".to_string()));
696                    Ok(())
697                }
698                Value::Map(_) => Err(redis::RedisError::from((
699                    redis::ErrorKind::TypeError,
700                    "Could not convert Redis response",
701                    "Redis Map type is not supported".to_string(),
702                ))),
703                Value::Attribute { .. } => Err(redis::RedisError::from((
704                    redis::ErrorKind::TypeError,
705                    "Could not convert Redis response",
706                    "Redis Attribute type is not supported".to_string(),
707                ))),
708                Value::Array(arr) | Value::Set(arr) => {
709                    arr.iter().try_for_each(|value| append(values, value))
710                }
711                Value::Double(v) => {
712                    values.push(v3::RedisResult::Binary(v.to_string().into_bytes()));
713                    Ok(())
714                }
715                Value::VerbatimString { .. } => Err(redis::RedisError::from((
716                    redis::ErrorKind::TypeError,
717                    "Could not convert Redis response",
718                    "Redis string with format attribute is not supported".to_string(),
719                ))),
720                Value::Boolean(v) => {
721                    values.push(v3::RedisResult::Int64(if *v { 1 } else { 0 }));
722                    Ok(())
723                }
724                Value::BigNumber(v) => {
725                    values.push(v3::RedisResult::Binary(v.to_string().as_bytes().to_owned()));
726                    Ok(())
727                }
728                Value::Push { .. } => Err(redis::RedisError::from((
729                    redis::ErrorKind::TypeError,
730                    "Could not convert Redis response",
731                    "Redis Pub/Sub types are not supported".to_string(),
732                ))),
733                Value::ServerError(err) => Err(redis::RedisError::from((
734                    redis::ErrorKind::ResponseError,
735                    "Server error",
736                    format!("{err:?}"),
737                ))),
738            }
739        }
740        let mut values = Vec::new();
741        append(&mut values, value)?;
742        Ok(RedisResults(values))
743    }
744}
745
746fn memory_size(value: &v3::RedisResult) -> usize {
747    match value {
748        v3::RedisResult::Nil | v3::RedisResult::Int64(_) => std::mem::size_of::<v3::RedisResult>(),
749        v3::RedisResult::Binary(b) => std::mem::size_of::<v3::RedisResult>() + b.len(),
750        v3::RedisResult::Status(s) => std::mem::size_of::<v3::RedisResult>() + s.len(),
751    }
752}
753
754/// Resolves DNS using Tokio's resolver, filtering out blocked IPs.
755struct SpinDnsResolver(BlockedNetworks);
756
757impl AsyncDNSResolver for SpinDnsResolver {
758    fn resolve<'a, 'b: 'a>(
759        &'a self,
760        host: &'b str,
761        port: u16,
762    ) -> redis::RedisFuture<'a, Box<dyn Iterator<Item = std::net::SocketAddr> + Send + 'a>> {
763        Box::pin(async move {
764            let mut addrs = tokio::net::lookup_host((host, port))
765                .await?
766                .collect::<Vec<_>>();
767            // Remove blocked IPs
768            let blocked_addrs = self.0.remove_blocked(&mut addrs);
769            if addrs.is_empty() && !blocked_addrs.is_empty() {
770                tracing::error!(
771                    "error.type" = "destination_ip_prohibited",
772                    ?blocked_addrs,
773                    "all destination IP(s) prohibited by runtime config"
774                );
775            }
776            Ok(Box::new(addrs.into_iter()) as Box<dyn Iterator<Item = SocketAddr> + Send>)
777        })
778    }
779}
spin_factor_outbound_redis/host.rs

spin_factor_outbound_redis/
host.rs
