faucet_sink_sqs/
config.rs

1//! Configuration for the SQS sink. No I/O here.
2
3use faucet_common_sqs::SqsCredentials;
4use schemars::JsonSchema;
5use serde::{Deserialize, Serialize};
6
7/// SQS hard limit: `SendMessageBatch` accepts at most 10 entries per request.
8pub const MAX_ENTRIES_PER_REQUEST: usize = 10;
9/// SQS hard limit: 256 KiB per message body.
10pub const MAX_MESSAGE_BYTES: usize = 262_144;
11/// SQS hard limit: 256 KiB total payload per `SendMessageBatch` request.
12pub const MAX_BATCH_BYTES: usize = 262_144;
13
14/// Configuration for [`SqsSink`](crate::SqsSink).
15#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
16pub struct SqsSinkConfig {
17    /// SQS queue URL (e.g. `https://sqs.us-east-1.amazonaws.com/1234/my-q`).
18    pub queue_url: String,
19    /// AWS region. `None` uses the SDK default chain.
20    #[serde(default, skip_serializing_if = "Option::is_none")]
21    pub region: Option<String>,
22    /// Custom endpoint URL for LocalStack / VPC endpoints.
23    #[serde(default, skip_serializing_if = "Option::is_none")]
24    pub endpoint_url: Option<String>,
25    /// AWS credentials. Defaults to the SDK default provider chain.
26    #[serde(default)]
27    pub credentials: SqsCredentials,
28
29    /// FIFO message group id, applied to every message. Required by FIFO
30    /// queues; omit for standard queues. Default: none.
31    #[serde(default, skip_serializing_if = "Option::is_none")]
32    pub message_group_id: Option<String>,
33    /// Record field whose (stringified) value becomes each message's
34    /// `MessageDeduplicationId` on a FIFO queue. A record missing the field
35    /// (or with a non-scalar value) fails per-record (DLQ-routable). Only used
36    /// when set. Default: none.
37    #[serde(default, skip_serializing_if = "Option::is_none")]
38    pub message_deduplication_id_field: Option<String>,
39
40    /// Max entries per `SendMessageBatch` request (1–10). Default 10.
41    #[serde(default = "default_batch_size")]
42    pub batch_size: usize,
43
44    /// Bounded concurrent in-flight `SendMessageBatch` requests. Default 4.
45    #[serde(default = "default_concurrency")]
46    pub concurrency: usize,
47    /// Per-record retry budget for partial failures. Default 5.
48    #[serde(default = "default_retry_max_attempts")]
49    pub retry_max_attempts: usize,
50    /// Initial retry backoff (milliseconds). Default 100.
51    #[serde(default = "default_retry_initial_backoff_ms")]
52    pub retry_initial_backoff_ms: u64,
53    /// Backoff ceiling (milliseconds). Default 30000.
54    #[serde(default = "default_retry_max_backoff_ms")]
55    pub retry_max_backoff_ms: u64,
56}
57
58fn default_batch_size() -> usize {
59    MAX_ENTRIES_PER_REQUEST
60}
61fn default_concurrency() -> usize {
62    4
63}
64fn default_retry_max_attempts() -> usize {
65    5
66}
67fn default_retry_initial_backoff_ms() -> u64 {
68    100
69}
70fn default_retry_max_backoff_ms() -> u64 {
71    30_000
72}
73
74impl SqsSinkConfig {
75    /// Minimal config with defaults for everything but the queue URL.
76    pub fn new(queue_url: impl Into<String>) -> Self {
77        Self {
78            queue_url: queue_url.into(),
79            region: None,
80            endpoint_url: None,
81            credentials: SqsCredentials::default(),
82            message_group_id: None,
83            message_deduplication_id_field: None,
84            batch_size: default_batch_size(),
85            concurrency: default_concurrency(),
86            retry_max_attempts: default_retry_max_attempts(),
87            retry_initial_backoff_ms: default_retry_initial_backoff_ms(),
88            retry_max_backoff_ms: default_retry_max_backoff_ms(),
89        }
90    }
91
92    /// Fail-fast validation, called from `SqsSink::new`.
93    pub fn validate(&self) -> Result<(), faucet_core::FaucetError> {
94        use faucet_core::FaucetError;
95        if self.queue_url.trim().is_empty() {
96            return Err(FaucetError::Config(
97                "sqs sink: queue_url must not be empty".into(),
98            ));
99        }
100        if self.batch_size == 0 || self.batch_size > MAX_ENTRIES_PER_REQUEST {
101            return Err(FaucetError::Config(format!(
102                "sqs sink: batch_size must be 1..={MAX_ENTRIES_PER_REQUEST} (got {}) — the \
103                 SendMessageBatch API limit",
104                self.batch_size
105            )));
106        }
107        if self.concurrency == 0 {
108            return Err(FaucetError::Config(
109                "sqs sink: concurrency must be at least 1".into(),
110            ));
111        }
112        if self.retry_max_attempts == 0 {
113            return Err(FaucetError::Config(
114                "sqs sink: retry_max_attempts must be at least 1".into(),
115            ));
116        }
117        if let Some(field) = &self.message_deduplication_id_field
118            && field.is_empty()
119        {
120            return Err(FaucetError::Config(
121                "sqs sink: message_deduplication_id_field must not be empty".into(),
122            ));
123        }
124        Ok(())
125    }
126}
127
128#[cfg(test)]
129mod tests {
130    use super::*;
131
132    #[test]
133    fn defaults_are_the_api_limits() {
134        let c = SqsSinkConfig::new("https://q");
135        c.validate().unwrap();
136        assert_eq!(c.batch_size, 10);
137        assert_eq!(c.concurrency, 4);
138        assert!(c.message_group_id.is_none());
139        assert!(c.message_deduplication_id_field.is_none());
140    }
141
142    #[test]
143    fn validation_bounds() {
144        let mut c = SqsSinkConfig::new("https://q");
145        c.batch_size = 11;
146        assert!(c.validate().is_err(), "SendMessageBatch cap is 10");
147        c.batch_size = 0;
148        assert!(c.validate().is_err());
149
150        let mut c = SqsSinkConfig::new("https://q");
151        c.concurrency = 0;
152        assert!(c.validate().is_err());
153
154        let mut c = SqsSinkConfig::new("https://q");
155        c.retry_max_attempts = 0;
156        assert!(c.validate().is_err());
157
158        let mut c = SqsSinkConfig::new("https://q");
159        c.message_deduplication_id_field = Some(String::new());
160        assert!(c.validate().is_err());
161
162        let mut c = SqsSinkConfig::new("  ");
163        c.batch_size = 5;
164        assert!(c.validate().is_err());
165    }
166
167    #[test]
168    fn full_config_parses_from_yaml() {
169        let yaml = r#"
170queue_url: https://sqs.us-east-1.amazonaws.com/1/events.fifo
171region: us-east-1
172endpoint_url: http://127.0.0.1:4566
173credentials: { type: default }
174message_group_id: orders
175message_deduplication_id_field: order_id
176batch_size: 5
177concurrency: 2
178retry_max_attempts: 3
179"#;
180        let c: SqsSinkConfig = serde_yaml::from_str(yaml).unwrap();
181        c.validate().unwrap();
182        assert_eq!(c.message_group_id.as_deref(), Some("orders"));
183        assert_eq!(
184            c.message_deduplication_id_field.as_deref(),
185            Some("order_id")
186        );
187        assert_eq!(c.batch_size, 5);
188    }
189}
faucet_sink_sqs/config.rs

faucet_sink_sqs/
config.rs
