Implement and test retry classifier sorting (#3392)

references = ["smithy-rs#3336", "smithy-rs#3391"]

meta = { "breaking" = false, "tada" = false, "bug" = false}

author = "iampkmone"

[[aws-sdk-rust]]

message = """

Retry classifiers will now be sorted by priority. This change only affects requests

that are retried. Some requests that were previously been classified as transient

errors may now be classified as throttling errors.

If you were

- configuring multiple custom retry classifiers

- that would disagree on how to classify a response

- that have differing priorities

you may see a behavior change in that classification for the same response is now

dependent on the classifier priority instead of the order in which the classifier

was added.

"""

references = ["smithy-rs#3322"]

meta = { "breaking" = false, "bug" = true, "tada" = false }

author = "Velfi"

[[smithy-rs]]

message = """

Retry classifiers will now be sorted by priority. This change only affects requests

that are retried. Some requests that were previously been classified as transient

errors may now be classified as throttling errors.

If you were

- configuring multiple custom retry classifiers

- that would disagree on how to classify a response

- that have differing priorities

you may see a behavior change in that classification for the same response is now

dependent on the classifier priority instead of the order in which the classifier

was added.

"""

references = ["smithy-rs#3322"]

meta = { "breaking" = false, "bug" = true, "tada" = false }

author = "Velfi"

    }

    fn priority(&self) -> RetryClassifierPriority {

        RetryClassifierPriority::with_lower_priority_than(

        RetryClassifierPriority::run_before(

            RetryClassifierPriority::modeled_as_retryable_classifier(),

        )

    }

# Note: This workspace exists so that these tests can be run without having to build the entire SDK. When you run

# `./gradlew -Paws.fullsdk=true :aws:sdk:assemble` these tests are copied into their respective Service crates.

[workspace]

resolver = "2"

members = [

    "dynamodb",

    "ec2",

use aws_sdk_s3::config::SharedAsyncSleep;

use aws_smithy_async::rt::sleep::TokioSleep;

use aws_smithy_runtime::client::http::test_util::{ReplayEvent, StaticReplayClient};

use aws_smithy_runtime_api::client::retries::classifiers::RetryClassifierPriority;

use aws_smithy_types::body::SdkBody;

use std::sync::{Arc, Mutex};

    // ensure our custom retry classifier was called at least once.

    assert_ne!(customization_test_classifier.counter(), 0);

}

#[derive(Debug, Clone)]

struct OrderingTestClassifier {

    counter: Arc<Mutex<u8>>,

    name: &'static str,

    priority: RetryClassifierPriority,

}

impl OrderingTestClassifier {

    pub fn new(name: &'static str, priority: RetryClassifierPriority) -> Self {

        Self {

            counter: Arc::new(Mutex::new(0u8)),

            name,

            priority,

        }

    }

    pub fn counter(&self) -> u8 {

        *self.counter.lock().unwrap()

    }

}

impl ClassifyRetry for OrderingTestClassifier {

    fn classify_retry(&self, _ctx: &InterceptorContext) -> RetryAction {

        tracing::debug!("Running classifier {}", self.name);

        *self.counter.lock().unwrap() += 1;

        RetryAction::NoActionIndicated

    }

    fn name(&self) -> &'static str {

        "Ordering Test Retry Classifier"

    }

    fn priority(&self) -> RetryClassifierPriority {

        self.priority.clone()

    }

}

#[tracing_test::traced_test]

#[tokio::test]

async fn test_retry_classifier_customization_ordering() {

    let http_client = StaticReplayClient::new(vec![

        ReplayEvent::new(req(), err()),

        ReplayEvent::new(req(), ok()),

    ]);

    let classifier_a = OrderingTestClassifier::new("6", RetryClassifierPriority::default());

    let classifier_b = OrderingTestClassifier::new(

        "5",

        RetryClassifierPriority::run_before(classifier_a.priority()),

    );

    let classifier_c = OrderingTestClassifier::new(

        "4",

        RetryClassifierPriority::run_before(classifier_b.priority()),

    );

    let classifier_d = OrderingTestClassifier::new(

        "3",

        RetryClassifierPriority::run_before(classifier_c.priority()),

    );

    let classifier_e = OrderingTestClassifier::new(

        "2",

        RetryClassifierPriority::run_before(classifier_d.priority()),

    );

    let classifier_f = OrderingTestClassifier::new(

        "1",

        RetryClassifierPriority::run_before(classifier_e.priority()),

    );

    let config = aws_sdk_s3::Config::builder()

        .with_test_defaults()

        .sleep_impl(SharedAsyncSleep::new(TokioSleep::new()))

        .http_client(http_client)

        .retry_config(RetryConfig::standard())

        .retry_classifier(classifier_d.clone())

        .retry_classifier(classifier_b.clone())

        .retry_classifier(classifier_f.clone())

        .build();

    let client = aws_sdk_s3::Client::from_conf(config);

    let _ = client

        .get_object()

        .bucket("bucket")

        .key("key")

        .customize()

        .config_override(

            aws_sdk_s3::config::Config::builder()

                .retry_classifier(classifier_c.clone())

                .retry_classifier(classifier_a.clone())

                .retry_classifier(classifier_e.clone()),

        )

        .send()

        .await

        .expect_err("fails without attempting a retry");

    // ensure our classifiers were each called at least once.

    assert_ne!(classifier_a.counter(), 0, "classifier_a was never called");

    assert_ne!(classifier_b.counter(), 0, "classifier_b was never called");

    assert_ne!(classifier_c.counter(), 0, "classifier_c was never called");

    assert_ne!(classifier_d.counter(), 0, "classifier_d was never called");

    assert_ne!(classifier_e.counter(), 0, "classifier_e was never called");

    assert_ne!(classifier_f.counter(), 0, "classifier_f was never called");

    // ensure the classifiers were called in the correct order.

    logs_assert(|lines: &[&str]| {

        let mut found_log_a = false;

        let mut line_iter = lines.iter();

        while found_log_a == false {

            match line_iter.next() {

                Some(&line) => {

                    if line.contains("Running classifier 1") {

                        found_log_a = true;

                    }

                }

                None => {

                    return Err("Couldn't find log line for classifier 1".to_owned());

                }

            }

        }

        for i in 2..=6 {

            match line_iter.next() {

                Some(&line) => {

                    if line.contains(&format!("Running classifier {i}")) {

                        // pass

                    } else {

                        return Err(format!("Expected to find log line for classifier {i} after {} but found '{line}'", i - 1));

                    }

                }

                None => {

                    return Err(format!("Logs ended earlier than expected ({i})"));

                }

            }

        }

        Ok(())

    });

}

 * SPDX-License-Identifier: Apache-2.0

 */

//! Classifier for determining if a retry is necessary and related code.

//! Classifiers for determining if a retry is necessary and related code.

//!

//! When a request fails, a retry strategy should inspect the result with retry

//! classifiers to understand if and how the request should be retried.

//!

//! Because multiple classifiers are often used, and because some are more

//! specific than others in what they identify as retryable, classifiers are

//! run in a sequence that is determined by their priority.

//!

//! Classifiers that are higher priority are run **after** classifiers

//! with a lower priority. The intention is that:

//!

//! 1. Generic classifiers that look at things like the HTTP error code run

//!     first.

//! 2. More specific classifiers such as ones that check for certain error

//!     messages are run **after** the generic classifiers. This gives them the

//!     ability to override the actions set by the generic retry classifiers.

//!

//! Put another way:

//!

//! | large nets target common failures with basic behavior | run before            | small nets target specific failures with special behavior|

//! |-------------------------------------------------------|-----------------------|----------------------------------------------------------|

//! | low priority classifiers                              | results overridden by | high priority classifiers                                |

use crate::box_error::BoxError;

use crate::client::interceptors::context::InterceptorContext;

use crate::client::runtime_components::sealed::ValidateConfig;

use crate::client::runtime_components::RuntimeComponents;

use crate::impl_shared_conversions;

use aws_smithy_types::config_bag::ConfigBag;

use aws_smithy_types::retry::ErrorKind;

use std::fmt;

use std::sync::Arc;

    RetryableError {

        /// The kind of error.

        kind: ErrorKind,

        /// A server may tells us to retry only after a specific time has elapsed.

        /// A server may tell us to retry only after a specific time has elapsed.

        retry_after: Option<Duration>,

    },

}

    }

}

/// The priority of a retry classifier. Classifiers with a higher priority will run before

/// classifiers with a lower priority. Classifiers with equal priorities make no guarantees

/// about which will run first.

/// The priority of a retry classifier. Classifiers with a higher priority will

/// run **after** classifiers with a lower priority and may override their

/// result. Classifiers with equal priorities make no guarantees about which

/// will run first.

#[derive(Debug, Clone, Copy, PartialEq, Eq)]

pub struct RetryClassifierPriority {

    inner: Inner,

#[derive(Debug, Clone, Copy, PartialEq, Eq)]

enum Inner {

    // The default priority for the `HttpStatusCodeClassifier`.

    /// The default priority for the `HttpStatusCodeClassifier`.

    HttpStatusCodeClassifier,

    // The default priority for the `ModeledAsRetryableClassifier`.

    /// The default priority for the `ModeledAsRetryableClassifier`.

    ModeledAsRetryableClassifier,

    // The default priority for the `TransientErrorClassifier`.

    /// The default priority for the `TransientErrorClassifier`.

    TransientErrorClassifier,

    // The priority of some other classifier.

    /// The priority of some other classifier.

    Other(i8),

}

impl PartialOrd for RetryClassifierPriority {

    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {

        Some(other.as_i8().cmp(&self.as_i8()))

        Some(self.as_i8().cmp(&other.as_i8()))

    }

}

impl Ord for RetryClassifierPriority {

    fn cmp(&self, other: &Self) -> std::cmp::Ordering {

        other.as_i8().cmp(&self.as_i8())

        self.as_i8().cmp(&other.as_i8())

    }

}

        }

    }

    #[deprecated = "use the less-confusingly-named `RetryClassifierPriority::run_before` instead"]

    /// Create a new `RetryClassifierPriority` with lower priority than the given priority.

    pub fn with_lower_priority_than(other: Self) -> Self {

        Self::run_before(other)

    }

    /// Create a new `RetryClassifierPriority` that can be overridden by the given priority.

    pub fn run_before(other: Self) -> Self {

        Self {

            inner: Inner::Other(other.as_i8() + 1),

            inner: Inner::Other(other.as_i8() - 1),

        }

    }

    #[deprecated = "use the less-confusingly-named `RetryClassifierPriority::run_after` instead"]

    /// Create a new `RetryClassifierPriority` with higher priority than the given priority.

    pub fn with_higher_priority_than(other: Self) -> Self {

        Self::run_after(other)

    }

    /// Create a new `RetryClassifierPriority` that can override the given priority.

    pub fn run_after(other: Self) -> Self {

        Self {

            inner: Inner::Other(other.as_i8() - 1),

            inner: Inner::Other(other.as_i8() + 1),

        }

    }

    }

}

impl ValidateConfig for SharedRetryClassifier {}

impl ValidateConfig for SharedRetryClassifier {

    fn validate_final_config(

        &self,

        runtime_components: &RuntimeComponents,

        _cfg: &ConfigBag,

    ) -> Result<(), BoxError> {

        #[cfg(debug_assertions)]

        {

            // Because this is validating that the implementation is correct rather

            // than validating user input, we only want to run this in debug builds.

            let retry_classifiers = runtime_components.retry_classifiers_slice();

            let out_of_order: Vec<_> = retry_classifiers

                .windows(2)

                .filter(|&w| w[0].value().priority() > w[1].value().priority())

                .collect();

            if !out_of_order.is_empty() {

                return Err("retry classifiers are mis-ordered; this is a bug".into());

            }

        }

        Ok(())

    }

}

#[cfg(test)]

mod tests {

    use super::RetryClassifierPriority;

    use super::{ClassifyRetry, RetryAction, RetryClassifierPriority, SharedRetryClassifier};

    use crate::client::interceptors::context::InterceptorContext;

    #[test]

    fn test_classifier_lower_priority_than() {

        let classifier_a = RetryClassifierPriority::default();

        let classifier_b = RetryClassifierPriority::with_lower_priority_than(classifier_a);

        let classifier_c = RetryClassifierPriority::with_lower_priority_than(classifier_b);

        let mut list = vec![classifier_b, classifier_a, classifier_c];

    fn test_preset_priorities() {

        let before_modeled_as_retryable = RetryClassifierPriority::run_before(

            RetryClassifierPriority::modeled_as_retryable_classifier(),

        );

        let mut list = vec![

            RetryClassifierPriority::modeled_as_retryable_classifier(),

            RetryClassifierPriority::http_status_code_classifier(),

            RetryClassifierPriority::transient_error_classifier(),

            before_modeled_as_retryable,

        ];

        list.sort();

        assert_eq!(vec![classifier_c, classifier_b, classifier_a], list);

        assert_eq!(

            vec![

                RetryClassifierPriority::http_status_code_classifier(),

                before_modeled_as_retryable,

                RetryClassifierPriority::modeled_as_retryable_classifier(),

                RetryClassifierPriority::transient_error_classifier(),

            ],

            list

        );

    }

    #[test]

    fn test_classifier_higher_priority_than() {

        let classifier_c = RetryClassifierPriority::default();

        let classifier_b = RetryClassifierPriority::with_higher_priority_than(classifier_c);

        let classifier_a = RetryClassifierPriority::with_higher_priority_than(classifier_b);

    fn test_classifier_run_before() {

        // Ensure low-priority classifiers run *before* high-priority classifiers.

        let high_priority_classifier = RetryClassifierPriority::default();

        let mid_priority_classifier = RetryClassifierPriority::run_before(high_priority_classifier);

        let low_priority_classifier = RetryClassifierPriority::run_before(mid_priority_classifier);

        let mut list = vec![

            mid_priority_classifier,

            high_priority_classifier,

            low_priority_classifier,

        ];

        list.sort();

        let mut list = vec![classifier_b, classifier_c, classifier_a];

        assert_eq!(

            vec![

                low_priority_classifier,

                mid_priority_classifier,

                high_priority_classifier

            ],

            list

        );

    }

    #[test]

    fn test_classifier_run_after() {

        // Ensure high-priority classifiers run *after* low-priority classifiers.

        let low_priority_classifier = RetryClassifierPriority::default();

        let mid_priority_classifier = RetryClassifierPriority::run_after(low_priority_classifier);

        let high_priority_classifier = RetryClassifierPriority::run_after(mid_priority_classifier);

        let mut list = vec![

            mid_priority_classifier,

            low_priority_classifier,

            high_priority_classifier,

        ];

        list.sort();

        assert_eq!(vec![classifier_c, classifier_b, classifier_a], list);

        assert_eq!(

            vec![

                low_priority_classifier,

                mid_priority_classifier,

                high_priority_classifier

            ],

            list

        );

    }

    #[derive(Debug)]

    struct ClassifierStub {

        name: &'static str,

        priority: RetryClassifierPriority,

    }

    impl ClassifyRetry for ClassifierStub {

        fn classify_retry(&self, _ctx: &InterceptorContext) -> RetryAction {

            todo!()

        }

        fn name(&self) -> &'static str {

            self.name

        }

        fn priority(&self) -> RetryClassifierPriority {

            self.priority

        }

    }

    fn wrap(name: &'static str, priority: RetryClassifierPriority) -> SharedRetryClassifier {

        SharedRetryClassifier::new(ClassifierStub { name, priority })

    }

    #[test]

    fn test_shared_classifier_run_before() {

        // Ensure low-priority classifiers run *before* high-priority classifiers,

        // even after wrapping.

        let high_priority_classifier = RetryClassifierPriority::default();

        let mid_priority_classifier = RetryClassifierPriority::run_before(high_priority_classifier);

        let low_priority_classifier = RetryClassifierPriority::run_before(mid_priority_classifier);

        let mut list = vec![

            wrap("mid", mid_priority_classifier),

            wrap("high", high_priority_classifier),

            wrap("low", low_priority_classifier),

        ];

        list.sort_by_key(|rc| rc.priority());

        let actual: Vec<_> = list.iter().map(|it| it.name()).collect();

        assert_eq!(vec!["low", "mid", "high"], actual);

    }

    #[test]

    fn test_shared_classifier_run_after() {

        // Ensure high-priority classifiers run *after* low-priority classifiers,

        // even after wrapping.

        let low_priority_classifier = RetryClassifierPriority::default();

        let mid_priority_classifier = RetryClassifierPriority::run_after(low_priority_classifier);

        let high_priority_classifier = RetryClassifierPriority::run_after(mid_priority_classifier);

        let mut list = vec![

            wrap("mid", mid_priority_classifier),

            wrap("high", high_priority_classifier),

            wrap("low", low_priority_classifier),

        ];

        list.sort_by_key(|rc| rc.priority());

        let actual: Vec<_> = list.iter().map(|it| it.name()).collect();

        assert_eq!(vec!["low", "mid", "high"], actual);

    }

    #[test]

    fn test_shared_preset_priorities() {

        let before_modeled_as_retryable = RetryClassifierPriority::run_before(

            RetryClassifierPriority::modeled_as_retryable_classifier(),

        );

        let mut list = vec![

            wrap(

                "modeled as retryable",

                RetryClassifierPriority::modeled_as_retryable_classifier(),

            ),

            wrap(

                "http status code",

                RetryClassifierPriority::http_status_code_classifier(),

            ),

            wrap(

                "transient error",

                RetryClassifierPriority::transient_error_classifier(),

            ),

            wrap("before 'modeled as retryable'", before_modeled_as_retryable),

        ];

        list.sort_by_key(|rc| rc.priority());

        let actual: Vec<_> = list.iter().map(|it| it.name()).collect();

        assert_eq!(

            vec![

                "http status code",

                "before 'modeled as retryable'",

                "modeled as retryable",

                "transient error"

            ],

            actual

        );

    }

}