Cleaned up downlink failures module.

Author: horned-sphere

runtime/swimos_runtime/src/agent/mod.rs

@@ -294,7 +294,6 @@ impl AgentContext for AgentRuntimeContext {
     }
 }
 
-
 /// Reasons that a remote connected to an agent runtime task could be disconnected.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum DisconnectionReason {

runtime/swimos_runtime/src/agent/task/external_links/mod.rs

@@ -45,7 +45,8 @@ use tracing::{debug, error, trace};
 use uuid::Uuid;
 
 use crate::{
-    agent::{CommanderKey, CommanderRequest, DownlinkRequest, LinkRequest}, Io,
+    agent::{CommanderKey, CommanderRequest, DownlinkRequest, LinkRequest},
+    Io,
 };
 
 use super::{AdHocChannelRequest, ExternalLinkRequest};

runtime/swimos_runtime/src/agent/task/external_links/tests.rs

@@ -19,7 +19,8 @@ use crate::{
         task::{external_links, AdHocChannelRequest, ExternalLinkRequest},
         CommanderKey, CommanderRequest, DownlinkRequest, LinkRequest,
     },
-    downlink::DownlinkOptions, Io,
+    downlink::DownlinkOptions,
+    Io,
 };
 use bytes::Bytes;
 use futures::{

runtime/swimos_runtime/src/downlink/failure.rs

@@ -12,12 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{convert::Infallible, fmt::Display, num::NonZeroUsize};
-
-use swimos_utilities::format::comma_sep;
-use thiserror::Error;
-use tracing::warn;
+use std::{convert::Infallible, fmt::Display};
 
+/// Configuration for the downlink runtime to instruct it how to respond to bad envelopes.
 #[derive(Debug, PartialEq, Eq)]
 pub enum BadFrameResponse<E> {
     /// Instruction to ignore the bad envelope.
@@ -63,6 +60,7 @@ impl<E: std::error::Error> BadFrameStrategy<E> for AlwaysAbortStrategy {
     }
 }
 
+/// Error type returned by [`ReportStrategy`].
 #[derive(Debug)]
 pub struct ErrReport {
     message: String,
@@ -84,6 +82,7 @@ impl ErrReport {
     }
 }
 
+/// [`BadFrameStrategy`] that creates an [`ErrReport`] from the error returned by the inner strategy.
 pub struct ReportStrategy<S> {
     inner: S,
 }
@@ -117,6 +116,7 @@ where
     }
 }
 
+#[doc(hidden)]
 pub type BoxedReportStrategy<'a, E> = Box<dyn BadFrameStrategy<E, Report = ErrReport> + Send + 'a>;
 
 impl<'a, E> BadFrameStrategy<E> for BoxedReportStrategy<'a, E> {
@@ -139,76 +139,10 @@ impl<E> BadFrameStrategy<E> for AlwaysIgnoreStrategy {
     }
 }
 
-/// A strategy that will ignore several bad envelopes, collecting the errors, and then
-/// abort.
-pub struct CountStrategy<E> {
-    max: usize,
-    count: usize,
-    errors: Vec<E>,
-}
-
-impl<E> CountStrategy<E> {
-    pub fn new(max: NonZeroUsize) -> Self {
-        CountStrategy {
-            max: max.get(),
-            count: 0,
-            errors: vec![],
-        }
-    }
-}
-
-/// A collection of errors, accumulated by a [`CountStrategy`].
-#[derive(Debug, Error, PartialEq, Eq)]
-#[error("Too many bad frames: {errors}")]
-pub struct ErrorLog<E> {
-    errors: Errors<E>,
-}
-
-impl<E> AsRef<[E]> for ErrorLog<E> {
-    fn as_ref(&self) -> &[E] {
-        &self.errors.0
-    }
-}
-
-#[derive(Debug, PartialEq, Eq)]
-pub struct Errors<E>(Vec<E>);
-
-impl<E: Display> Display for Errors<E> {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "[{}]", comma_sep(&self.0))
-    }
-}
-
-impl<E: std::error::Error> BadFrameStrategy<E> for CountStrategy<E> {
-    type Report = ErrorLog<E>;
-
-    fn failed_with(&mut self, error: E) -> BadFrameResponse<Self::Report> {
-        let CountStrategy { max, count, errors } = self;
-        *count += 1;
-        if *count == *max {
-            errors.push(error);
-            *count = 0;
-            BadFrameResponse::Abort(ErrorLog {
-                errors: Errors(std::mem::take(errors)),
-            })
-        } else {
-            warn!(
-                "Received {n} of a maximum of {m} invalid map messages: {e}",
-                n = *count,
-                m = *max,
-                e = error
-            );
-            errors.push(error);
-            BadFrameResponse::Ignore
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
 
     use super::*;
-    use swimos_utilities::non_zero_usize;
     use thiserror::Error;
 
     #[derive(Debug, Error, PartialEq, Eq)]
@@ -231,34 +165,4 @@ mod tests {
         let response = handler.failed_with(TestError("failed".to_string()));
         assert_eq!(response, BadFrameResponse::Ignore);
     }
-
-    const MAX: NonZeroUsize = non_zero_usize!(3);
-
-    #[test]
-    fn count_errors() {
-        let mut handler = CountStrategy::new(MAX);
-
-        let first = handler.failed_with(TestError("first".to_string()));
-        assert_eq!(BadFrameResponse::Ignore, first);
-
-        let second = handler.failed_with(TestError("second".to_string()));
-        assert_eq!(BadFrameResponse::Ignore, second);
-
-        let third = handler.failed_with(TestError("third".to_string()));
-        match third {
-            BadFrameResponse::Abort(report) => {
-                assert_eq!(
-                    report.as_ref(),
-                    &[
-                        TestError("first".to_string()),
-                        TestError("second".to_string()),
-                        TestError("third".to_string())
-                    ]
-                )
-            }
-            BadFrameResponse::Ignore => {
-                panic!("Error ignored.");
-            }
-        }
-    }
 }

runtime/swimos_runtime/src/downlink/interpretation/mod.rs

@@ -94,6 +94,8 @@ impl DownlinkInterpretation for MapInterpretation {
     }
 }
 
+/// Interpretation for map downlinks that does not interpret the events at all and passes through
+/// the frame unmodified.
 pub struct NoInterpretation;
 impl DownlinkInterpretation for NoInterpretation {
     type Error = Infallible;

runtime/swimos_runtime/src/downlink/mod.rs

@@ -25,8 +25,7 @@ use bytes::{Bytes, BytesMut};
 use futures::future::{join, select, Either};
 use futures::stream::SelectAll;
 use futures::{Future, FutureExt, Sink, SinkExt, Stream, StreamExt};
-pub use interpretation::MapInterpretation;
-pub use interpretation::NoInterpretation;
+use interpretation::MapInterpretation;
 use swimos_agent_protocol::{
     encoding::downlink::DownlinkNotificationEncoder, DownlinkNotification,
 };
@@ -47,6 +46,8 @@ use tokio_util::codec::{Decoder, Encoder, FramedRead, FramedWrite};
 use tracing::{error, info, info_span, trace, warn, Instrument};
 use uuid::Uuid;
 
+pub use interpretation::NoInterpretation;
+
 mod backpressure;
 pub mod failure;
 mod interpretation;

runtime/swimos_runtime/src/lib.rs

@@ -28,4 +28,4 @@ pub mod downlink;
 mod timeout_coord;
 
 /// Ends of two independent channels (for example the input and output channels of an agent).
-type Io = (ByteWriter, ByteReader);
+type Io = (ByteWriter, ByteReader);