photon_backend/retention/
partition.rs

1//! Topic partition and subscription partition keys for reclaim.
2
3use std::collections::HashSet;
4
5use crate::handler_registry::HandlerRegistry;
6use crate::retention::config::RetentionPolicy;
7use crate::retention::hook::RetentionHook;
8
9/// Storage partition `(topic, optional key)`.
10#[derive(Debug, Clone, PartialEq, Eq, Hash)]
11pub struct TopicPartition {
12    /// Topic name.
13    pub topic_name: String,
14    /// Optional partition / shard key.
15    pub topic_key: Option<String>,
16}
17
18impl TopicPartition {
19    /// Create a partition key.
20    pub fn new(topic_name: impl Into<String>, topic_key: Option<String>) -> Self {
21        Self {
22            topic_name: topic_name.into(),
23            topic_key,
24        }
25    }
26}
27
28/// Durable subscription scoped to one transport partition.
29#[derive(Debug, Clone, PartialEq, Eq, Hash)]
30pub struct SubscriptionPartition {
31    /// Durable subscription name.
32    pub subscription_name: String,
33    /// Topic name.
34    pub topic_name: String,
35    /// Optional partition / shard key.
36    pub topic_key: Option<String>,
37}
38
39/// Collect known durable subscriptions for watermark checkpoint loads.
40pub fn known_subscriptions(
41    policy: &RetentionPolicy,
42    hook: Option<&dyn RetentionHook>,
43) -> Vec<SubscriptionPartition> {
44    let mut seen = HashSet::new();
45    let mut out = Vec::new();
46
47    let registry = HandlerRegistry::auto_discover();
48    for handler in registry.iter() {
49        let entry = SubscriptionPartition {
50            subscription_name: handler.subscription_name.to_string(),
51            topic_name: handler.topic_name.to_string(),
52            topic_key: None,
53        };
54        if seen.insert(entry.clone()) {
55            out.push(entry);
56        }
57    }
58
59    for entry in &policy.extra_subscriptions {
60        if seen.insert(entry.clone()) {
61            out.push(entry.clone());
62        }
63    }
64
65    if let Some(h) = hook {
66        for entry in h.extra_subscriptions() {
67            if seen.insert(entry.clone()) {
68                out.push(entry);
69            }
70        }
71    }
72
73    out
74}
75
76/// Subscriptions whose checkpoint applies to the given transport partition.
77pub fn subscriptions_for_partition<'a>(
78    subs: &'a [SubscriptionPartition],
79    topic: &str,
80    topic_key: Option<&str>,
81) -> Vec<&'a SubscriptionPartition> {
82    subs.iter()
83        .filter(|s| {
84            s.topic_name == topic
85                && match (&s.topic_key, topic_key) {
86                    (None, _) => true,
87                    (Some(sk), Some(tk)) => sk == tk,
88                    (Some(_), None) => false,
89                }
90        })
91        .collect()
92}
93
94/// Merge partition sets from multiple sources.
95pub fn merge_partitions(sources: impl IntoIterator<Item = TopicPartition>) -> Vec<TopicPartition> {
96    let mut seen = HashSet::new();
97    let mut out = Vec::new();
98    for part in sources {
99        if seen.insert(part.clone()) {
100            out.push(part);
101        }
102    }
103    out
104}
photon_backend/retention/partition.rs

photon_backend/retention/
partition.rs
