photon_backend_fluvio/
config.rs

1//! Builder and resolved configuration for the Fluvio storage adapter.
2
3use std::time::Duration;
4
5use photon_backend::{BrokerTransportSecurity, PhotonError, Result, TransportCrypto};
6
7use crate::replicas::replicas_from_env;
8use crate::retention::retention_from_env;
9
10/// Environment variable for Fluvio SC endpoint.
11pub const ENDPOINT_ENV: &str = "PHOTON_FLUVIO_ENDPOINT";
12
13/// Environment variable for topic name prefix.
14pub const PREFIX_ENV: &str = "PHOTON_FLUVIO_TOPIC_PREFIX";
15
16/// How durable replay and checkpoints map to Fluvio offsets.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
18pub enum ReplayCursor {
19    /// Broker offset+1 on publish ack; seek replay by seq.
20    #[default]
21    StreamSeq,
22    /// Live tail only; checkpoints no-op; maximum publish throughput.
23    TailOnly,
24}
25
26/// Resolved Fluvio adapter settings (no env lookups at append time).
27#[derive(Clone)]
28pub struct FluvioConfig {
29    /// Fluvio Streaming Controller endpoint.
30    pub endpoint: String,
31    /// Topic prefix for Photon events.
32    pub topic_prefix: String,
33    /// Topic retention / replay window.
34    pub retention: Duration,
35    /// Topic replication factor.
36    pub replicas: i32,
37    /// Payload envelope crypto.
38    pub crypto: TransportCrypto,
39    /// Replay and checkpoint semantics.
40    pub replay_cursor: ReplayCursor,
41    /// Await produce ack per message.
42    pub sync_ack: bool,
43    /// Max concurrent in-flight publishes.
44    pub max_inflight: u32,
45    /// Independent Fluvio topics for ingress sharding (`1` = legacy single layout).
46    pub topic_shards: u32,
47    /// Plaintext vs TLS connect policy.
48    pub transport_security: BrokerTransportSecurity,
49}
50
51impl FluvioConfig {
52    /// Effective replication factor when creating shard topics.
53    #[must_use]
54    pub const fn effective_replicas(&self) -> i32 {
55        if self.topic_shards > 1 {
56            1
57        } else {
58            self.replicas
59        }
60    }
61
62    /// Whether publishes route across multiple topic shards.
63    #[must_use]
64    pub const fn is_sharded(&self) -> bool {
65        self.topic_shards > 1
66    }
67
68    /// Compact checkpoint topic name.
69    #[must_use]
70    pub fn checkpoint_topic(&self) -> String {
71        format!("{}-checkpoints", self.topic_prefix)
72    }
73}
74
75/// Environment variable for replay cursor mode.
76pub const REPLAY_CURSOR_ENV: &str = "PHOTON_FLUVIO_REPLAY_CURSOR";
77
78/// Environment variable for synchronous publish ack (`1` / `0`).
79pub const SYNC_ACK_ENV: &str = "PHOTON_FLUVIO_SYNC_ACK";
80
81/// Environment variable for max in-flight publishes per port.
82pub const MAX_INFLIGHT_ENV: &str = "PHOTON_FLUVIO_MAX_INFLIGHT";
83
84/// Builder for [`super::port::FluvioStoragePort`].
85///
86/// **Configuration lives here.** Set builder methods explicitly; unset fields fall back to
87/// `PHOTON_FLUVIO_*` environment variables via [`from_env_defaults`](Self::from_env_defaults).
88///
89/// Use the **same** builder settings on every Brokered publisher and worker binary that shares a
90/// cluster. Getting started:
91/// [Brokered](https://docs.rs/uf-photon/latest/photon/#brokered-publisher--worker-binaries).
92/// Teach the brokered path once with the NATS examples (`nats_worker` / `nats_publisher`); swap
93/// this builder for Fluvio in production.
94///
95/// # Options
96///
97/// | Method / env | Default | Purpose |
98/// |--------------|---------|---------|
99/// | [`.endpoint`](Self::endpoint) / [`ENDPOINT_ENV`] | **required** | Streaming Controller address (`host:9103`). |
100/// | [`.topic_prefix`](Self::topic_prefix) / [`PREFIX_ENV`] | `photon` | Topic name prefix. |
101/// | [`.retention`](Self::retention) / [`RETENTION_ENV`](crate::retention::RETENTION_ENV) | `15m` | Topic segment retention (applied at create). |
102/// | [`.replicas`](Self::replicas) / [`REPLICAS_ENV`](crate::replicas::REPLICAS_ENV) | `1` | Replication factor per topic. |
103/// | [`.replay_cursor`](Self::replay_cursor) / [`REPLAY_CURSOR_ENV`] | `stream_seq` | [`ReplayCursor::StreamSeq`] or [`ReplayCursor::TailOnly`]. |
104/// | [`.sync_ack`](Self::sync_ack) / [`SYNC_ACK_ENV`] | `1` | Await produce ack (`0` = firehose). |
105/// | [`.max_inflight`](Self::max_inflight) / [`MAX_INFLIGHT_ENV`] | `1` / `256` | Concurrent in-flight publishes (`256` for PFH). |
106/// | [`.topic_shards`](Self::topic_shards) / [`TOPIC_SHARDS_ENV`](crate::stream_shard::TOPIC_SHARDS_ENV) | `1` | Ingress shard count (`K>1` → `photon-s-{i}-{topic}`). |
107/// | [`.require_tls`](Self::require_tls) / [`PHOTON_ALLOW_INSECURE_BROKER`](photon_backend::ALLOW_INSECURE_BROKER_ENV) | require TLS | Plaintext endpoints need `.allow_insecure_plaintext()`. |
108///
109/// **Retention:** applied as Fluvio segment cleanup policy when Photon creates the topic.
110///
111/// # Examples
112///
113/// ## Publisher binary
114///
115/// Publish only — skip `start_executor` unless this process also runs handlers.
116///
117/// ```rust,ignore
118/// use std::sync::Arc;
119///
120/// use photon_backend_fluvio::{FluvioStoragePort, ReplayCursor};
121/// use photon_runtime::Photon;
122///
123/// # async fn boot_publisher() -> photon_backend::Result<()> {
124/// let port = Arc::new(
125///     FluvioStoragePort::builder()
126///         .from_env_defaults()
127///         .replay_cursor(ReplayCursor::StreamSeq)
128///         .sync_ack(true)
129///         .build()
130///         .await?,
131/// );
132/// let photon = Photon::builder()
133///     .storage_port(port)
134///     .auto_registry()
135///     .build()?;
136/// // EventType { … }.publish_on(&photon).await?;
137/// # let _ = photon;
138/// # Ok(())
139/// # }
140/// ```
141///
142/// ## Worker binary
143///
144/// Same port wiring as the publisher, plus `#[subscribe]` handlers and `start_executor`.
145/// Start workers before publishers.
146///
147/// ```rust,ignore
148/// use std::sync::Arc;
149///
150/// use photon_backend_fluvio::{FluvioStoragePort, ReplayCursor};
151/// use photon_core::JsonIdentityFactory;
152/// use photon_runtime::Photon;
153///
154/// # async fn boot_worker() -> photon_backend::Result<()> {
155/// let port = Arc::new(
156///     FluvioStoragePort::builder()
157///         .from_env_defaults()
158///         .replay_cursor(ReplayCursor::StreamSeq)
159///         .sync_ack(true)
160///         .build()
161///         .await?,
162/// );
163/// let photon = Photon::builder()
164///     .storage_port(port)
165///     .auto_registry()
166///     .build()?;
167/// photon.start_executor(Arc::new(JsonIdentityFactory))?;
168/// # let _ = photon;
169/// # Ok(())
170/// # }
171/// ```
172///
173/// See also: [`FluvioConfig`] (resolved snapshot), crate-level topic mapping in [`crate`](index.html).
174#[derive(Default)]
175pub struct FluvioStoragePortBuilder {
176    endpoint: Option<String>,
177    topic_prefix: Option<String>,
178    retention: Option<Duration>,
179    replicas: Option<i32>,
180    crypto: Option<TransportCrypto>,
181    replay_cursor: Option<ReplayCursor>,
182    sync_ack: Option<bool>,
183    max_inflight: Option<u32>,
184    topic_shards: Option<u32>,
185    transport_security: Option<BrokerTransportSecurity>,
186}
187
188impl FluvioStoragePortBuilder {
189    /// Empty builder; set fields or call [`Self::from_env_defaults`].
190    #[must_use]
191    pub fn new() -> Self {
192        Self::default()
193    }
194
195    /// Fill unset fields from `PHOTON_FLUVIO_*` environment variables.
196    #[must_use]
197    pub fn from_env_defaults(mut self) -> Self {
198        if self.endpoint.is_none() {
199            self.endpoint = std::env::var(ENDPOINT_ENV).ok();
200        }
201        if self.topic_prefix.is_none() {
202            self.topic_prefix = Some(std::env::var(PREFIX_ENV).unwrap_or_else(|_| "photon".into()));
203        }
204        if self.retention.is_none() {
205            self.retention = Some(retention_from_env());
206        }
207        if self.replicas.is_none() {
208            self.replicas = Some(replicas_from_env());
209        }
210        if self.replay_cursor.is_none() {
211            self.replay_cursor = Some(replay_cursor_from_env());
212        }
213        if self.sync_ack.is_none() {
214            self.sync_ack = Some(sync_ack_from_env());
215        }
216        if self.max_inflight.is_none() {
217            self.max_inflight = Some(max_inflight_from_env(self.sync_ack.unwrap_or(true)));
218        }
219        if self.topic_shards.is_none() {
220            self.topic_shards = Some(crate::stream_shard::topic_shards_from_env());
221        }
222        self
223    }
224
225    /// Fluvio SC endpoint.
226    #[must_use]
227    pub fn endpoint(mut self, endpoint: impl Into<String>) -> Self {
228        self.endpoint = Some(endpoint.into());
229        self
230    }
231
232    /// Topic prefix.
233    #[must_use]
234    pub fn topic_prefix(mut self, prefix: impl Into<String>) -> Self {
235        self.topic_prefix = Some(prefix.into());
236        self
237    }
238
239    /// Topic retention duration.
240    #[must_use]
241    pub const fn retention(mut self, retention: Duration) -> Self {
242        self.retention = Some(retention);
243        self
244    }
245
246    /// Topic replication factor.
247    #[must_use]
248    pub const fn replicas(mut self, replicas: i32) -> Self {
249        self.replicas = Some(replicas);
250        self
251    }
252
253    /// Transport crypto for publish path.
254    #[must_use]
255    pub fn crypto(mut self, crypto: TransportCrypto) -> Self {
256        self.crypto = Some(crypto);
257        self
258    }
259
260    /// Replay cursor mode.
261    #[must_use]
262    pub const fn replay_cursor(mut self, cursor: ReplayCursor) -> Self {
263        self.replay_cursor = Some(cursor);
264        self
265    }
266
267    /// Whether to await produce ack.
268    #[must_use]
269    pub const fn sync_ack(mut self, wait: bool) -> Self {
270        self.sync_ack = Some(wait);
271        self
272    }
273
274    /// Pipeline depth for concurrent publishes.
275    #[must_use]
276    pub fn max_inflight(mut self, n: u32) -> Self {
277        self.max_inflight = Some(n.max(1));
278        self
279    }
280
281    /// Topic shard count for ingress scaling (`1` = single shared layout).
282    #[must_use]
283    pub fn topic_shards(mut self, n: u32) -> Self {
284        self.topic_shards = Some(n.clamp(1, crate::stream_shard::MAX_TOPIC_SHARDS));
285        self
286    }
287
288    /// Allow plaintext broker endpoints (development/CI only).
289    #[must_use]
290    pub const fn allow_insecure_plaintext(mut self) -> Self {
291        self.transport_security = Some(BrokerTransportSecurity::AllowInsecurePlaintext);
292        self
293    }
294
295    /// Require TLS-oriented broker endpoints.
296    #[must_use]
297    pub const fn require_tls(mut self) -> Self {
298        self.transport_security = Some(BrokerTransportSecurity::RequireTls);
299        self
300    }
301
302    /// Resolve configuration.
303    ///
304    /// # Errors
305    ///
306    /// Returns an error when required fields (endpoint) are missing.
307    pub fn resolve(self) -> Result<FluvioConfig> {
308        let builder = self.from_env_defaults();
309        let endpoint = builder.endpoint.ok_or_else(|| {
310            PhotonError::Internal(format!("{ENDPOINT_ENV} not set for fluvio storage adapter"))
311        })?;
312        Ok(FluvioConfig {
313            endpoint,
314            topic_prefix: builder.topic_prefix.unwrap_or_else(|| "photon".into()),
315            retention: builder.retention.unwrap_or_else(retention_from_env),
316            replicas: builder.replicas.unwrap_or_else(replicas_from_env),
317            crypto: match builder.crypto {
318                Some(c) => c,
319                None => TransportCrypto::from_env()?,
320            },
321            replay_cursor: builder.replay_cursor.unwrap_or(ReplayCursor::StreamSeq),
322            sync_ack: builder.sync_ack.unwrap_or(true),
323            max_inflight: builder
324                .max_inflight
325                .unwrap_or_else(|| max_inflight_from_env(builder.sync_ack.unwrap_or(true))),
326            topic_shards: builder
327                .topic_shards
328                .unwrap_or_else(crate::stream_shard::topic_shards_from_env),
329            transport_security: builder
330                .transport_security
331                .unwrap_or_else(BrokerTransportSecurity::from_env),
332        })
333    }
334}
335
336fn replay_cursor_from_env() -> ReplayCursor {
337    match std::env::var(REPLAY_CURSOR_ENV)
338        .unwrap_or_else(|_| "stream_seq".into())
339        .to_ascii_lowercase()
340        .as_str()
341    {
342        "tail_only" | "tail" | "none" => ReplayCursor::TailOnly,
343        _ => ReplayCursor::StreamSeq,
344    }
345}
346
347fn sync_ack_from_env() -> bool {
348    !matches!(
349        std::env::var(SYNC_ACK_ENV)
350            .unwrap_or_else(|_| "1".into())
351            .as_str(),
352        "0" | "false" | "off" | "no"
353    )
354}
355
356fn max_inflight_from_env(sync_ack: bool) -> u32 {
357    if let Ok(raw) = std::env::var(MAX_INFLIGHT_ENV) {
358        if let Ok(n) = raw.parse::<u32>() {
359            return n.max(1);
360        }
361    }
362    if sync_ack {
363        1
364    } else {
365        256
366    }
367}
368
369#[cfg(test)]
370mod tests {
371    use super::*;
372
373    #[test]
374    fn replay_cursor_parses_tail_only() {
375        std::env::set_var(REPLAY_CURSOR_ENV, "tail_only");
376        assert_eq!(replay_cursor_from_env(), ReplayCursor::TailOnly);
377        std::env::remove_var(REPLAY_CURSOR_ENV);
378    }
379}
photon_backend_fluvio/config.rs

photon_backend_fluvio/
config.rs
