parakeet.at
Parakeet is a Rust-based Bluesky AppServer aiming to implement most of the functionality required to support the Bluesky client
appview
atproto
bluesky
rust
appserver
fix(consumer): tweaks to consumer metrics and make repo update warning debug
mia.omg.lol
+9
-1
consumer/src/backfill/mod.rs
+9
-1
consumer/src/backfill/mod.rs
···
10
10
use reqwest::{Client, StatusCode};
11
11
use std::str::FromStr;
12
12
use std::sync::Arc;
13
+
use metrics::counter;
13
14
use tracing::{instrument, Instrument};
14
15
15
16
mod db;
···
68
69
tracing::trace!("backfilling {did}");
69
70
if let Err(e) = backfill_actor(&inner, &did).await {
70
71
tracing::error!(did, "backfill failed: {e}");
72
+
counter!("backfill_failure").increment(1);
73
+
} else {
74
+
counter!("backfill_success").increment(1);
71
75
}
76
+
77
+
counter!("backfill_pending").absolute(rx.len() as u64);
72
78
}
73
79
tracing::error!("backfill rx error?");
74
80
}
···
164
170
let mut follow_stats = vec![did.to_string()];
165
171
166
172
for (path, (cid, record)) in records {
167
-
let Some((_, rkey)) = path.split_once("/") else {
173
+
let Some((collection, rkey)) = path.split_once("/") else {
168
174
tracing::warn!("record contained invalid path {}", path);
169
175
return Err(diesel::result::Error::RollbackTransaction);
170
176
};
177
+
178
+
counter!("backfilled_commits", "collection" => collection.to_string()).increment(1);
171
179
172
180
let full_path = format!("at://{did}/{path}");
173
181
+3
-3
consumer/src/indexer/mod.rs
+3
-3
consumer/src/indexer/mod.rs
···
6
6
use diesel_async::{AsyncConnection, AsyncPgConnection};
7
7
use futures::StreamExt;
8
8
use ipld_core::cid::Cid;
9
+
use metrics::counter;
9
10
use parakeet_db::types::{ActorStatus, ActorSyncState};
10
11
use std::collections::HashMap;
11
12
use std::sync::Arc;
12
-
use metrics::counter;
13
13
use tokio::sync::mpsc::{channel, Receiver, Sender};
14
14
use tracing::instrument;
15
15
···
208
208
.await?;
209
209
210
210
if !updated {
211
-
tracing::warn!(
211
+
tracing::debug!(
212
212
"Got a repo update older than the current rev. not processing ops."
213
213
);
214
214
return Ok(false);
···
249
249
return None;
250
250
};
251
251
252
-
counter!("backfilled_ops", "collection" => collection_raw.to_string(), "action" => op.action.clone()).increment(1);
252
+
counter!("backfill_pending_ops", "collection" => collection_raw.to_string(), "action" => op.action.clone()).increment(1);
253
253
254
254
let collection = CollectionType::from_str(collection_raw);
255
255
if collection == CollectionType::Unsupported {