Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
add collection filtering for jetstream
Aly Raffauf
(Jul 2, 2026, 3:36 PM EDT)
5d2181c6
e54f6518
+22
-2
+8
constellation/src/bin/main.rs
+8
constellation/src/bin/main.rs
···
70
70
/// debugging: print the current jetstream cursor and exit
71
71
#[arg(long, action)]
72
72
print_cursor: bool,
73
+
/// Only request these Jetstream collections; supports 'foo.bar.*' wildcards
74
+
#[arg(long, value_delimiter = ',')]
75
+
collections: Vec<String>,
73
76
}
74
77
75
78
#[derive(Debug, Clone, ValueEnum)]
···
120
123
121
124
let collect_metrics = args.collect_metrics;
122
125
let stay_alive = CancellationToken::new();
126
+
let collections = args.collections;
123
127
124
128
match args.backend {
125
129
StorageBackend::Memory => run(
···
133
137
metrics_bind,
134
138
collect_metrics,
135
139
stay_alive,
140
+
collections,
136
141
),
137
142
#[cfg(feature = "rocks")]
138
143
StorageBackend::Rocks => {
···
165
170
metrics_bind,
166
171
collect_metrics,
167
172
stay_alive,
173
+
collections,
168
174
);
169
175
eprintln!("run finished: {r:?}");
170
176
r
···
188
194
metrics_bind: SocketAddr,
189
195
collect_metrics: bool,
190
196
stay_alive: CancellationToken,
197
+
collections: Vec<String>,
191
198
) -> Result<()> {
192
199
ctrlc::set_handler({
193
200
let mut desperation: u8 = 0;
···
224
231
fixture_preserve_cursor,
225
232
stream,
226
233
staying_alive,
234
+
collections,
227
235
) {
228
236
eprintln!("jetstream finished with error: {e}");
229
237
}
+6
-1
constellation/src/consumer/jetstream.rs
+6
-1
constellation/src/consumer/jetstream.rs
···
16
16
cursor: Option<u64>,
17
17
stream: String,
18
18
staying_alive: CancellationToken,
19
+
collections: Vec<String>,
19
20
) -> Result<()> {
20
21
let dict = DecoderDictionary::copy(JETSTREAM_ZSTD_DICTIONARY);
21
22
let mut connect_retries = 0;
22
23
let mut latest_cursor = cursor;
24
+
let collections_qs: String = collections
25
+
.iter()
26
+
.map(|c| format!("&wantedCollections={c}"))
27
+
.collect();
23
28
'outer: loop {
24
29
let stream_url = format!(
25
-
"{stream}?compress=true{}",
30
+
"{stream}?compress=true{collections_qs}{}",
26
31
latest_cursor
27
32
.map(|c| {
28
33
println!("starting with cursor from {:?} ago...", ts_age(c));
+8
-1
constellation/src/consumer/mod.rs
+8
-1
constellation/src/consumer/mod.rs
···
22
22
fixture_preserve_cursor: bool,
23
23
stream: String,
24
24
staying_alive: CancellationToken,
25
+
collections: Vec<String>,
25
26
) -> Result<()> {
26
27
let mut fixture_cursor = None;
27
28
let (receiver, consumer_handle) = if let Some(f) = fixture {
···
42
43
thread::spawn(move || consume_jsonl_file(f, sender)),
43
44
)
44
45
} else {
46
+
if !collections.is_empty() {
47
+
println!("requesting only these jetstream collections: {collections:?}");
48
+
}
49
+
45
50
let (sender, receiver) = flume::bounded(1024);
46
51
let cursor = store.get_cursor().unwrap();
47
52
(
48
53
receiver,
49
-
thread::spawn(move || consume_jetstream(sender, cursor, stream, staying_alive)),
54
+
thread::spawn(move || {
55
+
consume_jetstream(sender, cursor, stream, staying_alive, collections)
56
+
}),
50
57
)
51
58
};
52
59