Address PR #2746 review comments for AWS SQS support

Changes made:
- Fix typo: `chaning` → `changing` in sqs.rs comment
- Replace concrete AWS account ID/queue URL with placeholders in example
- Weaken `source_sqs_fifo` guarantees from `TotalOrder`/`ExactlyOnce` to
  `NoOrder`/`AtLeastOnce` since SQS FIFO only guarantees ordering within a
  single message group; removed `NonDet` parameter; added detailed doc comment
- Add doc comment to `dest_sqs` noting it's for standard queues only (no
  `message_group_id` set)
- Add `features = ["alloc"]` to `futures-util` dependency for stream module
- Remove `aws_sqs` from `.vscode/settings.json` cargo features
- Improve `nondet!` justification comments in example
- Remove unused `TotalOrder` import from sqs.rs

Comments dismissed as incorrect/not actionable:
- EC2 hardcoded region/AMI: acceptable for an example
- Delete-before-return: already has TODO, AtLeastOnce marker is accurate
- `?` terminates stream on empty poll: intentional design (same as resolved)
- flat_map yields Vec not Message: wrong (Option iterates over inner Vec)

Co-authored-by: Infinity 🤖 <infinity@hydro.run>

Author: MingweiSamuel

.vscode/settings.json

@@ -11,9 +11,7 @@
     "rust-analyzer.rustfmt.extraArgs": [
         "+nightly"
     ],
-    "rust-analyzer.cargo.features": [
-        "aws_sqs"
-    ],
+    "rust-analyzer.cargo.features": [],
     "editor.semanticTokenColorCustomizations": {
         "enabled": true,
         "rules": {

hydro_test/Cargo.toml

@@ -45,7 +45,7 @@ aws-config = { version = "1", features = [
     "credentials-login",
 ], optional = true }
 aws-sdk-sqs = { version = "1", optional = true }
-futures-util = { version = "0.3.0", default-features = false, optional = true }
+futures-util = { version = "0.3.0", default-features = false, features = ["alloc"], optional = true }
 
 [build-dependencies]
 stageleft_tool.workspace = true

hydro_test/examples/aws_sqs.rs

@@ -13,8 +13,8 @@ use stageleft::q;
 
 type HostCreator = Box<dyn Fn(&mut Deployment) -> Arc<dyn Host>>;
 
-// aws sqs send-message --queue-url 'https://sqs.us-west-1.amazonaws.com/557701608578/hydro_test_queue_standard' --message-body 'foobar'
-// AWS_PROFILE='test-profile' cargo run -p hydro_test --example aws_sqs --all-features -- --queue-url 'https://sqs.us-west-1.amazonaws.com/557701608578/hydro_test_queue_standard'
+// aws sqs send-message --queue-url 'https://sqs.<REGION>.amazonaws.com/<ACCOUNT_ID>/<QUEUE_NAME>' --message-body 'foobar'
+// AWS_PROFILE='<AWS_PROFILE>' cargo run -p hydro_test --example aws_sqs --all-features -- --queue-url 'https://sqs.<REGION>.amazonaws.com/<ACCOUNT_ID>/<QUEUE_NAME>'
 #[derive(Parser, Debug)]
 struct Args {
     #[clap(flatten)]
@@ -62,10 +62,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         let sqs_client = sqs_client(sdk_config);
         source_sqs_standard(sqs_client, &args.queue_url)
             .assume_ordering::<TotalOrder>(
-                nondet!(/** Print all messages at least once in arbitrary order */),
+                nondet!(/** Safe to assume total order because the side effect is only printing messages for observation/debugging. */),
             )
             .assume_retries::<ExactlyOnce>(
-                nondet!(/** Print all messages at least once in arbitrary order */),
+                nondet!(/** Safe to assume exactly-once because the side effect is only printing messages for observation/debugging. */),
             )
             .for_each(q!(|msg| {
                 println!("MSG: {:?} {:?}", msg.message_id(), msg.body());

hydro_test/src/aws/sqs.rs

@@ -6,7 +6,7 @@ use aws_sdk_sqs::types::{DeleteMessageBatchRequestEntry, Message};
 use futures_util::stream::StreamExt as _;
 use hydro_lang::live_collections::boundedness::Boundedness;
 use hydro_lang::live_collections::stream::{
-    AtLeastOnce, ExactlyOnce, NoOrder, Ordering, TotalOrder,
+    AtLeastOnce, ExactlyOnce, NoOrder, Ordering,
 };
 use hydro_lang::location::Location;
 use hydro_lang::prelude::*;
@@ -23,7 +23,7 @@ fn init_rewrites() {
 pub fn sqs_client<'a, Loc: Location<'a>>(
     sdk_config: Singleton<SdkConfig, Loc, Bounded>,
 ) -> Singleton<Client, Loc, Bounded> {
-    // TODO(mingwei): Should this be unbounded since it is a ball of chaning state?
+    // TODO(mingwei): Should this be unbounded since it is a ball of changing state?
     sdk_config.map(q!(|c| Client::new(&c)))
 }
 
@@ -48,15 +48,21 @@ pub fn source_sqs_standard<'a, Loc: Location<'a>>(
         .weaken_ordering()
 }
 
-/// Exactly-once in-order delivery from a FIFO SQS queue.
+/// Delivery from a FIFO SQS queue.
+///
+/// SQS FIFO only guarantees ordering within a single message group, and
+/// duplicates can still occur around visibility-timeout boundaries. Therefore
+/// this source conservatively returns `NoOrder` / `AtLeastOnce`. Callers who
+/// know their FIFO configuration enforces a single message group and
+/// deduplication can `assume_ordering` / `assume_retries` with a [`NonDet`]
+/// justification.
 ///
 /// # Non-Determinism
 /// The user must ensure `queue_url` is a FIFO queue. If it is not, the output order will be non-deterministic.
 pub fn source_sqs_fifo<'a, Loc: Location<'a>>(
     client: Singleton<Client, Loc, Bounded>,
     queue_url: &'a str,
-    _nondet: NonDet,
-) -> Stream<Message, Loc, Bounded, TotalOrder, ExactlyOnce> {
+) -> Stream<Message, Loc, Bounded, NoOrder, AtLeastOnce> {
     client
         .into_stream()
         .flat_map_stream_blocking(q!(move |client| {
@@ -69,9 +75,14 @@ pub fn source_sqs_fifo<'a, Loc: Location<'a>>(
             })
             .flat_map(futures_util::stream::iter)
         }))
+        .weaken_retries()
+        .weaken_ordering()
 }
 
-/// Writes messages to SQS.
+/// Writes messages to a _standard_ SQS queue.
+///
+/// Does not set `message_group_id`, so sending to a FIFO queue will fail at
+/// runtime with `MissingParameter`.
 pub fn dest_sqs<'a, Loc: Location<'a>, Bound: Boundedness, Order: Ordering>(
     client: Singleton<Client, Loc, Bounded>,
     input: Stream<String, Loc, Bound, Order, ExactlyOnce>,