photon_backend_fluvio/
config.rs

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

photon_backend_fluvio/
config.rs
