krabka_barrier/cli.rs
1//! The tool's command line: the `clap` types and the argument conversions.
2//!
3//! `Cli` and `Command` spell every flag each barrier subcommand takes, and
4//! `parse_time` and `Time::millis_i64` are the two halves of the one conversion
5//! the wire forces: an operator writes a duration in whatever unit reads best,
6//! and the barrier apis carry milliseconds. Parsing is separate from sending
7//! so a test can prove a command line reaches the right shape without a
8//! broker.
9
10use {
11 clap::{Parser, Subcommand},
12 krabka_units::{Time, parse},
13};
14
15/// The tool's command line.
16///
17/// Shared by the binary and by [`crate::run_from_args`], so both accept
18/// exactly the same flags.
19#[krabka_macros::bootstrap_cli_fields]
20#[derive(Parser)]
21#[command(
22 name = "krabka-barrier",
23 version,
24 about = "Define barrier groups, trigger cuts, and verify a cut against the log"
25)]
26pub struct Cli {}
27
28/// The subcommands, one per barrier api.
29#[derive(Subcommand)]
30pub enum Command {
31 /// Create a barrier group, or update one that exists.
32 Define {
33 /// The group's name.
34 #[arg(long)]
35 group: String,
36 /// A topic the group cuts across. Repeat for each one.
37 #[arg(long = "topic", required = true)]
38 topics: Vec<String>,
39 /// How often the coordinator injects a cut on its own. Takes any
40 /// time unit, so `500ms`, `30s`, `5m` and `1h` all work. A number
41 /// with no unit is refused. Omit for on-demand only.
42 #[arg(long, value_parser = parse_time)]
43 interval: Option<Time>,
44 /// How many cuts the group keeps before the oldest is tombstoned.
45 #[arg(long, default_value_t = 100)]
46 retained_cuts: i32,
47 },
48 /// Delete a barrier group and its cuts.
49 Delete {
50 /// The group's name.
51 #[arg(long)]
52 group: String,
53 },
54 /// Print a group's definition and its latest epoch.
55 Describe {
56 /// A group to describe. Repeat for each one, or omit for every group.
57 #[arg(long = "group")]
58 groups: Vec<String>,
59 },
60 /// Inject a cut now and print the offsets it took.
61 Trigger {
62 /// The group's name.
63 #[arg(long)]
64 group: String,
65 /// How long the coordinator retries a partition that has no marker
66 /// yet. Takes any time unit, the same as `--interval`. The broker
67 /// clamps this to its own configured ceiling. Omit to take the
68 /// broker's default.
69 #[arg(long, value_parser = parse_time)]
70 timeout: Option<Time>,
71 },
72 /// Print a group's retained cuts.
73 List {
74 /// The group's name.
75 #[arg(long)]
76 group: String,
77 /// Skip every cut below this epoch.
78 #[arg(long, default_value_t = 0)]
79 from_epoch: i64,
80 /// How many cuts to print. `-1` prints every retained cut.
81 #[arg(long, default_value_t = -1)]
82 max_results: i32,
83 },
84 /// Read the log at a cut's offsets and prove each one holds that cut's
85 /// marker.
86 ///
87 /// A cut is only worth as much as the markers behind it. This fetches the
88 /// batch at every offset the cut names and checks it is a barrier control
89 /// batch carrying this group and this epoch.
90 Verify {
91 /// The group's name.
92 #[arg(long)]
93 group: String,
94 /// The epoch to verify.
95 #[arg(long)]
96 epoch: i64,
97 },
98}
99
100/// Parse a time argument.
101///
102/// Delegates to `krabka_units`, so every unit the broker's own configuration
103/// accepts works here too: `ns`, `us`, `ms`, `s`, `m`, `h`, `d` and their long
104/// forms.
105///
106/// A number with no unit is refused, and only `0` is exempt. That is the units
107/// crate's rule and it is the right one here: `--timeout 30` from someone who
108/// meant milliseconds would otherwise wait thirty seconds without complaining.
109fn parse_time(raw: &str) -> Result<Time, String> { parse::time(raw).map_err(|e| e.to_string()) }
110
111#[cfg(test)]
112mod tests {
113 use {super::*, assert2::check, krabka_units::convert::TimeExt as _};
114
115 krabka_macros::duration_parser_fixture!(
116 {
117 /// A time argument takes any unit the broker's own configuration takes,
118 /// so an operator never has to convert to milliseconds by hand, and a
119 /// number with no unit is refused rather than guessed at.
120
121 },
122 [
123 five_minutes = { "5m", Some(300_000) },
124 one_hour = { "1h", Some(3_600_000) },
125 two_days = { "2d", Some(172_800_000) },
126 submillisecond = { "250us", Some(0) },
127 one_minute_words = { "1 minute", Some(60_000) },
128 // A unit is required, so a number alone cannot be read as the
129 // wrong scale. Zero is the one exemption, having no scale.
130 missing_unit = { "90", None },
131 zero = { "0", Some(0) },
132 invalid_word = { "banana", None },
133 empty = { "", None },
134 ]
135 );
136
137 krabka_macros::bootstrap_parser_fixture!(["krabka-barrier", "describe"], [
138 "krabka-barrier",
139 "-b",
140 "localhost:9092",
141 "describe"
142 ]);
143
144 /// A group with no periodic injection is the default, and it reaches the
145 /// wire as the -1 the broker reads as "on demand only".
146 #[test]
147 fn define_without_an_interval_asks_for_on_demand_only() {
148 let cli = Cli::try_parse_from([
149 "krabka-barrier",
150 "-b",
151 "localhost:9092",
152 "define",
153 "--group",
154 "g",
155 "--topic",
156 "orders",
157 ])
158 .expect("parses");
159 let Command::Define {
160 interval,
161 retained_cuts,
162 topics,
163 ..
164 } = cli.command
165 else {
166 panic!("expected define");
167 };
168 check!(interval == None);
169 check!(retained_cuts == 100);
170 check!(topics == vec!["orders".to_owned()]);
171 }
172
173 /// `list` defaults to every retained cut, matching the request default the
174 /// wire carries.
175 #[test]
176 fn list_defaults_to_every_retained_cut() {
177 let cli = Cli::try_parse_from([
178 "krabka-barrier",
179 "-b",
180 "localhost:9092",
181 "list",
182 "--group",
183 "g",
184 ])
185 .expect("parses");
186 let Command::List {
187 max_results,
188 from_epoch,
189 ..
190 } = cli.command
191 else {
192 panic!("expected list");
193 };
194 check!(max_results == -1);
195 check!(from_epoch == 0);
196 }
197}