Highly ambitious ATProtocol AppView service and sdks
untangled jetstream connected state from events processed
+29
-63
+29
-63
api/src/main.rs
+29
-63
api/src/main.rs
···
203
203
// Reset retry delay on successful connection
204
204
retry_delay = tokio::time::Duration::from_secs(5);
205
205
206
-
// Start health checker to monitor event processing
207
-
let health_check_connected = jetstream_connected_clone.clone();
208
-
tokio::spawn({
209
-
let consumer_for_health = consumer_arc.clone();
210
-
async move {
211
-
let mut last_count = 0u64;
212
-
let mut no_events_duration = 0u64;
213
-
214
-
loop {
215
-
tokio::time::sleep(tokio::time::Duration::from_secs(30)).await;
216
-
217
-
// Get current event count from consumer
218
-
let current_count = consumer_for_health
219
-
.event_count
220
-
.load(std::sync::atomic::Ordering::Relaxed);
221
-
222
-
if current_count == last_count {
223
-
no_events_duration += 30;
224
-
if no_events_duration >= 60 {
225
-
// No events for 60+ seconds - mark as disconnected
226
-
health_check_connected
227
-
.store(false, std::sync::atomic::Ordering::Relaxed);
228
-
let message = format!(
229
-
"Jetstream marked as disconnected: no events processed in {} seconds",
230
-
no_events_duration
231
-
);
232
-
tracing::warn!("{}", message);
233
-
Logger::global().log_jetstream(
234
-
LogLevel::Warn,
235
-
&message,
236
-
Some(serde_json::json!({
237
-
"no_events_duration_secs": no_events_duration,
238
-
"action": "health_check_disconnected"
239
-
})),
240
-
);
241
-
}
242
-
} else {
243
-
// Events are flowing - mark as connected
244
-
no_events_duration = 0;
245
-
health_check_connected
246
-
.store(true, std::sync::atomic::Ordering::Relaxed);
247
-
if last_count == 0 && current_count > 0 {
248
-
let message =
249
-
"Jetstream health check: events flowing, marked as connected";
250
-
tracing::info!("{}", message);
251
-
Logger::global().log_jetstream(
252
-
LogLevel::Info,
253
-
message,
254
-
Some(serde_json::json!({
255
-
"event_count": current_count,
256
-
"action": "health_check_connected"
257
-
})),
258
-
);
259
-
}
260
-
}
261
-
262
-
last_count = current_count;
263
-
}
264
-
}
265
-
});
206
+
// Mark as connected when consumer starts successfully
207
+
jetstream_connected_clone.store(true, std::sync::atomic::Ordering::Relaxed);
208
+
tracing::info!("Jetstream consumer connected successfully");
209
+
Logger::global().log_jetstream(
210
+
LogLevel::Info,
211
+
"Jetstream consumer connected",
212
+
Some(serde_json::json!({
213
+
"action": "consumer_connected"
214
+
})),
215
+
);
266
216
267
-
// Start consuming events (don't mark as connected immediately)
217
+
// Start consuming events
268
218
let cancellation_token = atproto_jetstream::CancellationToken::new();
269
219
match consumer_arc.start_consuming(cancellation_token).await {
270
220
Err(e) => {
271
221
tracing::error!(
272
-
"Jetstream consumer failed: {} - will retry in {:?}",
222
+
"Jetstream consumer disconnected: {} - will retry in {:?}",
273
223
e,
274
224
retry_delay
275
225
);
226
+
Logger::global().log_jetstream(
227
+
LogLevel::Error,
228
+
&format!("Jetstream consumer disconnected: {}", e),
229
+
Some(serde_json::json!({
230
+
"error": e.to_string(),
231
+
"retry_delay_secs": retry_delay.as_secs(),
232
+
"action": "consumer_disconnected"
233
+
})),
234
+
);
276
235
// Mark as disconnected on failure
277
236
jetstream_connected_clone.store(false, std::sync::atomic::Ordering::Relaxed);
278
237
}
279
238
Ok(_) => {
280
-
tracing::info!("Jetstream consumer completed normally");
239
+
tracing::info!("Jetstream consumer closed normally");
240
+
Logger::global().log_jetstream(
241
+
LogLevel::Info,
242
+
"Jetstream consumer closed normally",
243
+
Some(serde_json::json!({
244
+
"action": "consumer_closed"
245
+
})),
246
+
);
281
247
// This shouldn't happen in normal operation since start_consuming should run indefinitely
282
248
jetstream_connected_clone.store(false, std::sync::atomic::Ordering::Relaxed);
283
249
}