krabka_remote_storage_topic/manager/
readiness.rs

1//! Per-metadata-partition readiness of the topic-backed manager, for metrics.
2//!
3//! The broker exports one readiness sample per assigned `__remote_log_metadata`
4//! partition, so a partition that stays `NotReady` is visible and can alert.
5//! The sample reuses the read gate, so it cannot disagree with what a gated
6//! read answers.
7
8use {
9    super::{
10        TopicBasedRemoteLogMetadataManager,
11        assignment::{HWM_UNKNOWN, ReadGate, read_gate},
12    },
13    std::sync::PoisonError,
14};
15
16/// Readiness of one assigned `__remote_log_metadata` partition, as
17/// [`TopicBasedRemoteLogMetadataManager::metadata_partition_readiness`]
18/// reports it.
19#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20pub struct MetadataPartitionReadiness {
21    /// The metadata partition.
22    pub partition: i32,
23    /// The high-water mark observed when the partition was assigned: the
24    /// offset after the last event the consumer must apply before reads are
25    /// answered. `None` while that high-water mark is unknown, because the
26    /// assignment-time fetch failed and a later reconcile has not yet
27    /// refreshed it.
28    pub target_offset: Option<i64>,
29    /// The offset of the last event applied to the cache, or `-1` when none
30    /// has been applied.
31    pub applied_offset: i64,
32    /// `true` when gated reads for this partition are answered, and `false`
33    /// while they return the retriable `NotReady`.
34    pub ready: bool,
35}
36
37impl TopicBasedRemoteLogMetadataManager {
38    /// One [`MetadataPartitionReadiness`] for each metadata partition this
39    /// manager is assigned, sorted by partition. An unassigned partition has
40    /// no entry.
41    ///
42    /// `ready` comes from the same rule that every gated read applies, over
43    /// the target and applied offset the entry reports.
44    #[must_use]
45    pub fn metadata_partition_readiness(&self) -> Vec<MetadataPartitionReadiness> {
46        let mut targets: Vec<(i32, i64)> = self
47            .ready_targets
48            .lock()
49            .unwrap_or_else(PoisonError::into_inner)
50            .iter()
51            .map(|(&partition, &target)| (partition, target))
52            .collect();
53        targets.sort_unstable();
54        targets
55            .into_iter()
56            .map(|(partition, target)| {
57                // `ready` comes from the offsets this sample reports, so a
58                // pump that advances meanwhile cannot make it contradict them.
59                let applied = self.applied_offset(partition);
60                MetadataPartitionReadiness {
61                    partition,
62                    target_offset: (target != HWM_UNKNOWN).then_some(target),
63                    applied_offset: applied.unwrap_or(-1),
64                    ready: matches!(read_gate(target, applied), ReadGate::Ready),
65                }
66            })
67            .collect()
68    }
69}
70
71#[cfg(test)]
72mod tests {
73    use {
74        super::{HWM_UNKNOWN, MetadataPartitionReadiness},
75        crate::{
76            log::{InProcessMetadataEventLog, MetadataEventLog as _},
77            manager::test_support::{
78                HwmFlakyLog, SeededManager, seeded_manager, start_manager, tp, wait_ready,
79            },
80        },
81        assert2::{assert, check},
82        std::{collections::HashMap, sync::Arc},
83    };
84
85    /// One readiness case: the assignment targets and applied offsets the
86    /// reconciler and the pump would have recorded, and the report expected.
87    struct Case {
88        name: &'static str,
89        targets: &'static [(i32, i64)],
90        applied: [i64; 4],
91        expected: Vec<MetadataPartitionReadiness>,
92    }
93
94    const fn readiness(
95        partition: i32,
96        target_offset: Option<i64>,
97        applied_offset: i64,
98        ready: bool,
99    ) -> MetadataPartitionReadiness {
100        MetadataPartitionReadiness {
101            partition,
102            target_offset,
103            applied_offset,
104            ready,
105        }
106    }
107
108    #[tokio::test(flavor = "multi_thread")]
109    async fn readiness_reports_each_assigned_partition_against_its_gate() {
110        let cases = [
111            Case {
112                name: "unassigned",
113                targets: &[],
114                applied: [5, 5, 5, 5],
115                expected: vec![],
116            },
117            Case {
118                name: "assigned empty partition, target 0",
119                targets: &[(2, 0)],
120                applied: [-1, -1, -1, -1],
121                expected: vec![readiness(2, Some(0), -1, true)],
122            },
123            Case {
124                name: "assigned behind",
125                targets: &[(1, 10)],
126                applied: [-1, 7, -1, -1],
127                expected: vec![readiness(1, Some(10), 7, false)],
128            },
129            Case {
130                name: "assigned caught up",
131                targets: &[(1, 10)],
132                applied: [-1, 9, -1, -1],
133                expected: vec![readiness(1, Some(10), 9, true)],
134            },
135            Case {
136                name: "HWM unknown",
137                targets: &[(3, HWM_UNKNOWN)],
138                applied: [-1, -1, -1, 4],
139                expected: vec![readiness(3, None, 4, false)],
140            },
141            Case {
142                name: "several partitions, sorted",
143                targets: &[(3, 2), (0, HWM_UNKNOWN), (2, 0)],
144                applied: [-1, -1, -1, 0],
145                expected: vec![
146                    readiness(0, None, -1, false),
147                    readiness(2, Some(0), -1, true),
148                    readiness(3, Some(2), 0, false),
149                ],
150            },
151        ];
152        for Case {
153            name,
154            targets,
155            applied,
156            expected,
157        } in cases
158        {
159            // The manager consumes nothing, so the pump never touches the
160            // state the case installs.
161            let manager = start_manager(InProcessMetadataEventLog::new(4));
162            *manager.ready_targets.lock().unwrap() =
163                targets.iter().copied().collect::<HashMap<_, _>>();
164            *manager.applied.lock().unwrap() = applied.to_vec();
165            check!(manager.metadata_partition_readiness() == expected, "{name}");
166            manager.shutdown();
167        }
168    }
169
170    /// The reconciler's own sentinel reads as an unknown target, and the
171    /// refreshed real target as a known one once the pump catches up.
172    #[tokio::test(flavor = "multi_thread")]
173    async fn readiness_follows_a_failed_then_refreshed_high_water_mark() {
174        let log = HwmFlakyLog::new(4);
175        let SeededManager { manager, partition } =
176            seeded_manager(Arc::<HwmFlakyLog>::clone(&log)).await;
177        let hwm = log.high_water_marks().await.unwrap()[usize::try_from(partition).unwrap()];
178        assert!(hwm > 0);
179
180        log.set_fail_hwm(true);
181        manager.reconcile_assignment(&[partition]).await;
182        // The pump is already consuming, so the applied offset is whatever it
183        // has reached; the target and the gate are what the sentinel decides.
184        let unknown = manager.metadata_partition_readiness();
185        let applied = unknown.first().map_or(-1, |entry| entry.applied_offset);
186        check!(unknown == [readiness(partition, None, applied, false)]);
187
188        log.set_fail_hwm(false);
189        manager.reconcile_assignment(&[partition]).await;
190        wait_ready(&manager, &tp()).await;
191        check!(
192            manager.metadata_partition_readiness()
193                == [readiness(partition, Some(hwm), hwm - 1, true)]
194        );
195
196        manager.reconcile_assignment(&[]).await;
197        check!(manager.metadata_partition_readiness().is_empty());
198        manager.shutdown();
199    }
200}
krabka_remote_storage_topic/manager/readiness.rs

krabka_remote_storage_topic/manager/
readiness.rs
