Don't error on duplicate pagination token

CHANGELOG.next.toml

@@ -21,3 +21,15 @@ message = "Implement support for pure Python request middleware. Improve idiomat
 references = ["smithy-rs#1734"]
 meta = { "breaking" = false, "tada" = false, "bug" = false, "target" = "server" }
 author = "crisidev"
+
+[[aws-sdk-rust]]
+message = "Paginators now stop on encountering a duplicate token by default rather than panic. This behavior can be customized by toggling the `stop_on_duplicate_token` property on the paginator before calling `send`."
+references = ["aws-sdk-rust#620", "smithy-rs#1748"]
+meta = { "breaking" = false, "tada" = false, "bug" = true }
+author = "jdisanti"
+
+[[smithy-rs]]
+message = "Paginators now stop on encountering a duplicate token by default rather than panic. This behavior can be customized by toggling the `stop_on_duplicate_token` property on the paginator before calling `send`."
+references = ["aws-sdk-rust#620", "smithy-rs#1748"]
+meta = { "breaking" = false, "tada" = false, "bug" = true, "target" = "client"}
+author = "jdisanti"

aws/sdk/integration-tests/dynamodb/tests/paginators.rs

@@ -171,7 +171,7 @@ async fn paginators_handle_errors() {
 }
 
 #[tokio::test]
-async fn paginators_error_on_repeated_token() {
+async fn paginators_stop_on_duplicate_token_by_default() {
     let response = r#"{
         "Count": 1,
         "Items": [{
@@ -212,11 +212,80 @@ async fn paginators_error_on_repeated_token() {
             .get("PostedBy"),
         Some(&AttributeValue::S("[email protected]".to_string()))
     );
-    let err = rows.try_next().await.expect_err("failure");
-    assert!(
-        format!("{}", err).contains("next token did not change"),
-        "{}",
-        err
+    assert_eq!(
+        rows.try_next()
+            .await
+            .expect("no error")
+            .expect("not EOS")
+            .get("PostedBy"),
+        Some(&AttributeValue::S("[email protected]".to_string()))
+    );
+    assert_eq!(None, rows.try_next().await.expect("success"));
+}
+
+#[tokio::test]
+async fn paginators_can_continue_on_duplicate_token() {
+    let response = r#"{
+        "Count": 1,
+        "Items": [{
+            "PostedBy": {
+                "S": "[email protected]"
+            }
+        }],
+        "LastEvaluatedKey": {
+            "PostedBy": { "S": "[email protected]" }
+        }
+    }"#;
+    // send the same response twice with the same pagination token
+    let conn = TestConnection::new(vec![
+        (
+            mk_request(r#"{"TableName":"test-table","Limit":32}"#),
+            mk_response(response),
+        ),
+        (
+            mk_request(
+                r#"{"TableName":"test-table","Limit":32,"ExclusiveStartKey":{"PostedBy":{"S":"[email protected]"}}}"#,
+            ),
+            mk_response(response),
+        ),
+        (
+            mk_request(
+                r#"{"TableName":"test-table","Limit":32,"ExclusiveStartKey":{"PostedBy":{"S":"[email protected]"}}}"#,
+            ),
+            mk_response(response),
+        ),
+    ]);
+    let client = Client::from_conf_conn(stub_config(), conn.clone());
+    let mut rows = client
+        .scan()
+        .table_name("test-table")
+        .into_paginator()
+        .stop_on_duplicate_token(false)
+        .page_size(32)
+        .items()
+        .send();
+    assert_eq!(
+        rows.try_next()
+            .await
+            .expect("no error")
+            .expect("not EOS")
+            .get("PostedBy"),
+        Some(&AttributeValue::S("[email protected]".to_string()))
+    );
+    assert_eq!(
+        rows.try_next()
+            .await
+            .expect("no error")
+            .expect("not EOS")
+            .get("PostedBy"),
+        Some(&AttributeValue::S("[email protected]".to_string()))
+    );
+    assert_eq!(
+        rows.try_next()
+            .await
+            .expect("no error")
+            .expect("not EOS")
+            .get("PostedBy"),
+        Some(&AttributeValue::S("[email protected]".to_string()))
     );
-    assert_eq!(rows.try_next().await.expect("ok"), None);
 }

codegen-client/src/main/kotlin/software/amazon/smithy/rust/codegen/client/smithy/generators/PaginatorGenerator.kt

@@ -130,7 +130,8 @@ class PaginatorGenerator private constructor(
             /// Paginator for #{operation:D}
             pub struct $paginatorName#{generics:W} {
                 handle: std::sync::Arc<crate::client::Handle${generics.inst}>,
-                builder: #{Builder}
+                builder: #{Builder},
+                stop_on_duplicate_token: bool,
             }
 
             impl${generics.inst} ${paginatorName}${generics.inst} #{bounds:W} {
@@ -139,13 +140,25 @@ class PaginatorGenerator private constructor(
                     Self {
                         handle,
                         builder,
+                        stop_on_duplicate_token: true,
                     }
                 }
 
                 #{page_size_setter:W}
 
                 #{items_fn:W}
 
+                /// Stop paginating when the service returns the same pagination token twice in a row.
+                ///
+                /// Defaults to true.
+                ///
+                /// For certain operations, it may be useful to continue on duplicate token. For example,
+                /// if an operation is for tailing a log file in real-time, then continuing may be desired.
+                /// This option can be set to `false` to accommodate these use cases.
+                pub fn stop_on_duplicate_token(mut self, stop_on_duplicate_token: bool) -> Self {
+                    self.stop_on_duplicate_token = stop_on_duplicate_token;
+                    self
+                }
 
                 /// Create the pagination stream
                 ///
@@ -177,12 +190,12 @@ class PaginatorGenerator private constructor(
                                 Ok(ref resp) => {
                                     let new_token = #{output_token}(resp);
                                     let is_empty = new_token.map(|token| token.is_empty()).unwrap_or(true);
-                                    if !is_empty && new_token == input.$inputTokenMember.as_ref() {
-                                        let _ = tx.send(Err(#{SdkError}::ConstructionFailure("next token did not change, aborting paginator. This indicates an SDK or AWS service bug.".into()))).await;
-                                        return;
+                                    if !is_empty && new_token == input.$inputTokenMember.as_ref() && self.stop_on_duplicate_token {
+                                        true
+                                    } else {
+                                        input.$inputTokenMember = new_token.cloned();
+                                        is_empty
                                     }
-                                    input.$inputTokenMember = new_token.cloned();
-                                    is_empty
                                 },
                                 Err(_) => true,
                             };