From bfac0a8a11a27a566374fd26b0049a3f4ebddce5 Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Fri, 14 Nov 2025 14:56:19 -0500 Subject: [PATCH] fix: lints --- consumer/src/database_writer/locking.rs | 122 ------------------ .../operations/notifications.rs | 6 +- consumer/src/database_writer/timestamp.rs | 95 -------------- consumer/src/db/gates/queries.rs | 20 --- consumer/src/db/operations/feed.rs | 13 +- consumer/src/db/operations/labeler.rs | 4 +- consumer/src/events.rs | 6 +- consumer/tests/locking_test.rs | 80 ------------ parakeet-db/src/cid_util.rs | 14 -- 9 files changed, 15 insertions(+), 345 deletions(-) diff --git a/consumer/src/database_writer/locking.rs b/consumer/src/database_writer/locking.rs index bd982f5f..826630ff 100644 --- a/consumer/src/database_writer/locking.rs +++ b/consumer/src/database_writer/locking.rs @@ -30,43 +30,6 @@ fn hash_to_i32(s: &str) -> i32 { hasher.finish() as i32 } -/// Convert a DID to a PostgreSQL advisory lock ID (single-key form) -/// -/// **DEPRECATED**: Use `table_record_lock()` instead for table-scoped locks. -/// This function is kept for backward compatibility with existing code. -/// -/// Uses DefaultHasher for fast, stable hashing within a process. -/// Hash collisions are acceptable - they just cause false contention -/// (performance impact only, not a correctness issue). -#[deprecated( - note = "Use table_record_lock() for table-scoped locks instead of single-key locks" -)] -pub fn did_to_lock_id(did: &str) -> i64 { - let mut hasher = DefaultHasher::new(); - did.hash(&mut hasher); - - // Cast u64 hash to i64 (PostgreSQL bigint) - // This is safe - we just reinterpret the bits - hasher.finish() as i64 -} - -/// Convert an AT URI to a PostgreSQL advisory lock ID (single-key form) -/// -/// **DEPRECATED**: Use `table_record_lock()` instead for table-scoped locks. -/// This function is kept for backward compatibility with existing code. -/// -/// Uses DefaultHasher for fast, stable hashing within a process. -/// Hash collisions are acceptable - they just cause false contention -/// (performance impact only, not a correctness issue). -pub fn uri_to_lock_id(uri: &str) -> i64 { - let mut hasher = DefaultHasher::new(); - uri.hash(&mut hasher); - - // Cast u64 hash to i64 (PostgreSQL bigint) - // This is safe - we just reinterpret the bits - hasher.finish() as i64 -} - /// Generate a table-scoped advisory lock ID pair /// /// Returns `(table_id, key_id)` for use with PostgreSQL's two-key advisory lock form: @@ -178,95 +141,10 @@ pub async fn acquire_did_locks( Ok(()) } -/// Acquire transaction-level advisory locks for a set of AT URIs (records) -/// -/// Locks are acquired in sorted URI order to prevent deadlocks. -/// All locks are automatically released when the transaction commits/rolls back. -/// -/// # Deadlock Prevention -/// -/// By sorting URIs before acquiring locks, we ensure all workers acquire -/// locks in the same order, preventing circular wait conditions. -/// -/// # Performance -/// -/// All locks are acquired in a single PostgreSQL query, minimizing round-trips. -/// Empty URI lists short-circuit without a database call. -pub async fn acquire_record_locks( - conn: &(impl tokio_postgres::GenericClient + Sync), - uris: &[String], -) -> eyre::Result<()> { - if uris.is_empty() { - return Ok(()); - } - - // Sort URIs to ensure consistent lock ordering across all workers - let mut sorted_uris: Vec<&str> = uris.iter().map(|u| u.as_str()).collect(); - sorted_uris.sort_unstable(); - - // Deduplicate URIs (same URI shouldn't be locked twice) - sorted_uris.dedup(); - - // Convert to lock IDs - let lock_ids: Vec = sorted_uris.iter().map(|uri| uri_to_lock_id(uri)).collect(); - - // Acquire all locks in a single query (transaction-scoped) - // pg_advisory_xact_lock automatically releases on commit/rollback - conn.execute( - "SELECT pg_advisory_xact_lock(unnest($1::bigint[]))", - &[&lock_ids], - ) - .await?; - - tracing::trace!( - "Acquired {} advisory locks for {} record URIs", - lock_ids.len(), - sorted_uris.len() - ); - - Ok(()) -} - #[cfg(test)] mod tests { use super::*; - #[test] - #[allow(deprecated)] - fn test_did_hash_stability() { - let did = "did:plc:abcdef123456"; - let hash1 = did_to_lock_id(did); - let hash2 = did_to_lock_id(did); - assert_eq!(hash1, hash2, "Hash should be stable for same DID"); - } - - #[test] - #[allow(deprecated)] - fn test_did_hash_distribution() { - let dids = ["did:plc:abcdef123456", - "did:plc:abcdef123457", - "did:plc:xyz789000000", - "did:web:example.com"]; - - let hashes: Vec = dids.iter().map(|d| did_to_lock_id(d)).collect(); - - // Check for uniqueness (though collisions are acceptable) - let unique: std::collections::HashSet<_> = hashes.iter().collect(); - assert_eq!( - unique.len(), - hashes.len(), - "Hashes should generally be unique for different DIDs" - ); - } - - #[test] - #[allow(deprecated)] - fn test_different_dids_different_hashes() { - let hash1 = did_to_lock_id("did:plc:alice"); - let hash2 = did_to_lock_id("did:plc:bob"); - assert_ne!(hash1, hash2, "Different DIDs should have different hashes"); - } - #[test] fn test_table_record_lock_stability() { let did = "did:plc:abcdef123456"; diff --git a/consumer/src/database_writer/operations/notifications.rs b/consumer/src/database_writer/operations/notifications.rs index 072471a1..a5829fe0 100644 --- a/consumer/src/database_writer/operations/notifications.rs +++ b/consumer/src/database_writer/operations/notifications.rs @@ -248,9 +248,9 @@ mod tests { let liker_actor_id = 123; let post_author_actor_id = 456; let op = create_like_notification( - "at://did:plc:liker/app.bsky.feed.like/likerk3y", + "likerk3y", liker_actor_id, - "at://did:plc:author/app.bsky.feed.post/postk3y", + "postk3y", post_author_actor_id, "bafyreicid", created_at, @@ -285,7 +285,7 @@ mod tests { let author_actor_id = 789; let created_at = Utc::now(); let ops = create_mention_notifications( - "at://did:plc:author/app.bsky.feed.post/postk3y", + "postk3y", author_actor_id, &mentioned_actor_ids, "bafyreicid", diff --git a/consumer/src/database_writer/timestamp.rs b/consumer/src/database_writer/timestamp.rs index 8e7bffc7..393a25c7 100644 --- a/consumer/src/database_writer/timestamp.rs +++ b/consumer/src/database_writer/timestamp.rs @@ -158,87 +158,11 @@ pub fn validate_record_timestamp_with_tid( } } -/// Legacy validation function (deprecated, use validate_record_timestamp_with_tid) -/// -/// This function clamps timestamps to reasonable bounds but loses the actual -/// creation time for old records. Prefer using TID-based validation. -#[deprecated(note = "Use validate_record_timestamp_with_tid instead")] -pub fn validate_record_timestamp(created_at: DateTime) -> DateTime { - let now = Utc::now(); - - // 5 years in the past - let five_years_ago = now - chrono::Duration::days(5 * 365); - - // 1 day in the future - let one_day_future = now + chrono::Duration::days(1); - - // Check if timestamp is out of bounds - if created_at < five_years_ago { - tracing::warn!( - timestamp = %created_at, - "Record timestamp more than 5 years in the past, using current time" - ); - now - } else if created_at > one_day_future { - tracing::warn!( - timestamp = %created_at, - "Record timestamp more than 1 day in the future, using current time" - ); - now - } else { - created_at - } -} - -/// Validate an optional record timestamp and clamp to reasonable bounds -/// -/// Returns the validated timestamp, or None if the input is None. -/// If timestamp exists but is out of bounds, returns Some(current_time). -#[deprecated(note = "Use validate_record_timestamp_with_tid instead")] -#[allow(deprecated)] -pub fn validate_optional_timestamp(created_at: Option>) -> Option> { - created_at.map(validate_record_timestamp) -} - #[cfg(test)] mod tests { use super::*; use chrono::Duration; - #[test] - #[allow(deprecated)] - fn test_valid_timestamp() { - let now = Utc::now(); - let yesterday = now - Duration::days(1); - - // Timestamp within bounds should be returned unchanged - assert_eq!(validate_record_timestamp(yesterday), yesterday); - } - - #[test] - #[allow(deprecated)] - fn test_too_old_timestamp() { - let now = Utc::now(); - let six_years_ago = now - Duration::days(6 * 365); - - // Timestamp too old should be clamped to now - let result = validate_record_timestamp(six_years_ago); - // Allow 1 second difference due to execution time - assert!((result - now).num_seconds().abs() <= 1); - } - - #[test] - #[allow(deprecated)] - fn test_future_timestamp() { - let now = Utc::now(); - let two_days_future = now + Duration::days(2); - - // Future timestamp should be clamped to now - let result = validate_record_timestamp(two_days_future); - // Allow 1 second difference due to execution time - assert!((result - now).num_seconds().abs() <= 1); - } - #[test] fn test_edge_cases() { let now = Utc::now(); @@ -267,23 +191,4 @@ mod tests { let result = validate_record_timestamp_with_tid(two_days_future, &recent_tid); assert!((result - now).num_seconds().abs() <= 1); } - - #[test] - #[allow(deprecated)] - fn test_optional_timestamp() { - let now = Utc::now(); - let yesterday = now - Duration::days(1); - - // Valid optional timestamp - assert_eq!(validate_optional_timestamp(Some(yesterday)), Some(yesterday)); - - // None should return None - assert_eq!(validate_optional_timestamp(None), None); - - // Invalid optional timestamp should be clamped - let six_years_ago = now - Duration::days(6 * 365); - let result = validate_optional_timestamp(Some(six_years_ago)); - assert!(result.is_some()); - assert!((result.unwrap() - now).num_seconds().abs() <= 1); - } } diff --git a/consumer/src/db/gates/queries.rs b/consumer/src/db/gates/queries.rs index d09fe69d..cc821ddf 100644 --- a/consumer/src/db/gates/queries.rs +++ b/consumer/src/db/gates/queries.rs @@ -105,26 +105,6 @@ pub async fn check_list_membership( Ok(row.get(0)) } -/// Maintain postgate detachments using stored procedure (LEGACY - use maintain_postgates_cached) -/// -/// Calls the maintain_postgates stored procedure to update postgate state -#[deprecated( - note = "Use maintain_postgates_cached for better performance" -)] -pub async fn maintain_postgates( - conn: &C, - post: &str, - detached: &[String], - disable_effective: Option, -) -> QueryResult { - conn.execute( - "SELECT maintain_postgates($1, $2, $3)", - &[&post, &detached, &disable_effective], - ) - .await - .wrap_err_with(|| format!("Failed to maintain postgates for post {}", post)) -} - /// Maintain postgate detachments (OPTIMIZED) /// /// Performance improvements: diff --git a/consumer/src/db/operations/feed.rs b/consumer/src/db/operations/feed.rs index 5cb3f4a7..4f25b834 100644 --- a/consumer/src/db/operations/feed.rs +++ b/consumer/src/db/operations/feed.rs @@ -155,9 +155,8 @@ fn lexicon_to_embed_type(lexicon: &str) -> &str { /// Uses advisory locks to prevent concurrent transactions from racing on the same URI. async fn get_uri_id(conn: &C, uri: &str) -> Result { // Acquire advisory lock on URI to prevent concurrent access races - // Use a different namespace (0x55524900 = "URI\0") from other locks - let lock_id = crate::database_writer::locking::uri_to_lock_id(uri); - conn.execute("SELECT pg_advisory_xact_lock($1)", &[&lock_id]) + let (table_id, key_id) = crate::database_writer::locking::table_record_lock("uris", uri); + conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) .await?; let row = conn @@ -1379,8 +1378,8 @@ async fn get_feedgen_id( cid_str: &str, ) -> Result<(i64, bool)> { // Acquire advisory lock on feedgen URI to prevent concurrent access races - let lock_id = crate::database_writer::locking::uri_to_lock_id(at_uri); - conn.execute("SELECT pg_advisory_xact_lock($1)", &[&lock_id]) + let (table_id, key_id) = crate::database_writer::locking::table_record_lock("feedgens", at_uri); + conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) .await?; // Parse the CID string to get the digest @@ -1449,8 +1448,8 @@ pub(super) async fn ensure_list_id(conn: &C, at_uri: &str) -> /// The actual list data comes from the full list record when fetched. async fn get_list_id(conn: &C, at_uri: &str) -> Result<(i64, bool)> { // Acquire advisory lock on list URI to prevent concurrent access races - let lock_id = crate::database_writer::locking::uri_to_lock_id(at_uri); - conn.execute("SELECT pg_advisory_xact_lock($1)", &[&lock_id]) + let (table_id, key_id) = crate::database_writer::locking::table_record_lock("lists", at_uri); + conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) .await?; // Extract owner DID and rkey from AT URI diff --git a/consumer/src/db/operations/labeler.rs b/consumer/src/db/operations/labeler.rs index d2406c4f..f8e47ffb 100644 --- a/consumer/src/db/operations/labeler.rs +++ b/consumer/src/db/operations/labeler.rs @@ -23,8 +23,8 @@ pub async fn ensure_labeler_stub( // Acquire advisory lock on labeler URI to prevent concurrent access races // Labeler URI is always at://{did}/app.bsky.labeler.service/self let labeler_uri = format!("at://{}/app.bsky.labeler.service/self", did); - let lock_id = crate::database_writer::locking::uri_to_lock_id(&labeler_uri); - conn.execute("SELECT pg_advisory_xact_lock($1)", &[&lock_id]) + let (table_id, key_id) = crate::database_writer::locking::table_record_lock("labelers", &labeler_uri); + conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) .await?; // Parse the CID string to get the digest diff --git a/consumer/src/events.rs b/consumer/src/events.rs index 03d83872..13e8df63 100644 --- a/consumer/src/events.rs +++ b/consumer/src/events.rs @@ -101,14 +101,16 @@ pub struct AtpCommitEvent { pub since: Option, pub commit: Option, #[serde(rename = "tooBig")] - #[deprecated] + #[deprecated(note = "Legacy AT Protocol field, no longer used")] + #[allow(dead_code, reason = "Required for deserialization of legacy relay events")] pub too_big: bool, #[serde(default)] pub blocks: ByteBuf, #[serde(default)] pub ops: Vec, #[serde(default)] - #[deprecated] + #[deprecated(note = "Legacy AT Protocol field, no longer used")] + #[allow(dead_code, reason = "Required for deserialization of legacy relay events")] pub blobs: Vec, #[serde(rename = "prevData")] pub prev_data: Option, diff --git a/consumer/tests/locking_test.rs b/consumer/tests/locking_test.rs index 0deed804..1d919fac 100644 --- a/consumer/tests/locking_test.rs +++ b/consumer/tests/locking_test.rs @@ -89,86 +89,6 @@ async fn test_acquire_did_locks_duplicates() -> eyre::Result<()> { Ok(()) } -/// Test acquire_record_locks with a single URI -#[tokio::test] -async fn test_acquire_record_locks_single() -> eyre::Result<()> { - let pool = test_pool(); - let mut conn = pool.get().await.wrap_err("Failed to get connection")?; - let tx = conn.transaction().await.wrap_err("Failed to start transaction")?; - - let uris = vec!["at://did:plc:test/app.bsky.feed.post/123".to_string()]; - let result = locking::acquire_record_locks(&*tx, &uris).await; - - assert!( - result.is_ok(), - "acquire_record_locks should work: {:?}", - result.err() - ); - Ok(()) -} - -/// Test acquire_record_locks with multiple URIs -#[tokio::test] -async fn test_acquire_record_locks_multiple() -> eyre::Result<()> { - let pool = test_pool(); - let mut conn = pool.get().await.wrap_err("Failed to get connection")?; - let tx = conn.transaction().await.wrap_err("Failed to start transaction")?; - - let uris = vec![ - "at://did:plc:alice/app.bsky.feed.post/123".to_string(), - "at://did:plc:bob/app.bsky.feed.post/456".to_string(), - "at://did:plc:charlie/app.bsky.feed.post/789".to_string(), - ]; - let result = locking::acquire_record_locks(&*tx, &uris).await; - - assert!( - result.is_ok(), - "acquire_record_locks with multiple URIs should work: {:?}", - result.err() - ); - Ok(()) -} - -/// Test acquire_record_locks with empty URI list (should be no-op) -#[tokio::test] -async fn test_acquire_record_locks_empty() -> eyre::Result<()> { - let pool = test_pool(); - let mut conn = pool.get().await.wrap_err("Failed to get connection")?; - let tx = conn.transaction().await.wrap_err("Failed to start transaction")?; - - let uris: Vec = vec![]; - let result = locking::acquire_record_locks(&*tx, &uris).await; - - assert!( - result.is_ok(), - "acquire_record_locks with empty list should work: {:?}", - result.err() - ); - Ok(()) -} - -/// Test acquire_record_locks with duplicate URIs (should deduplicate) -#[tokio::test] -async fn test_acquire_record_locks_duplicates() -> eyre::Result<()> { - let pool = test_pool(); - let mut conn = pool.get().await.wrap_err("Failed to get connection")?; - let tx = conn.transaction().await.wrap_err("Failed to start transaction")?; - - let uris = vec![ - "at://did:plc:alice/app.bsky.feed.post/123".to_string(), - "at://did:plc:bob/app.bsky.feed.post/456".to_string(), - "at://did:plc:alice/app.bsky.feed.post/123".to_string(), // duplicate - ]; - let result = locking::acquire_record_locks(&*tx, &uris).await; - - assert!( - result.is_ok(), - "acquire_record_locks with duplicates should work: {:?}", - result.err() - ); - Ok(()) -} - /// Test that advisory locks are automatically released on transaction rollback #[tokio::test] async fn test_locks_released_on_rollback() -> eyre::Result<()> { diff --git a/parakeet-db/src/cid_util.rs b/parakeet-db/src/cid_util.rs index 513e0b54..ddbc6bf5 100644 --- a/parakeet-db/src/cid_util.rs +++ b/parakeet-db/src/cid_util.rs @@ -110,14 +110,6 @@ pub fn digest_to_record_cid(digest: &[u8]) -> Option> { None } } - -// Deprecated: Use digest_to_blob_cid or digest_to_record_cid instead -#[deprecated(since = "0.1.0", note = "Use digest_to_blob_cid or digest_to_record_cid")] -#[inline] -pub fn digest_to_cid(digest: &[u8]) -> Option> { - digest_to_record_cid(digest) -} - /// Convert a 32-byte digest to a base32 blob CID string /// /// Creates a CID string with raw codec (0x55) suitable for blobs. @@ -154,12 +146,6 @@ pub fn digest_to_record_cid_string(digest: &[u8]) -> Option { Some(multibase::encode(multibase::Base::Base32Lower, &full_cid)) } -// Deprecated: Use digest_to_blob_cid_string or digest_to_record_cid_string -#[deprecated(since = "0.1.0", note = "Use digest_to_blob_cid_string or digest_to_record_cid_string")] -pub fn digest_to_cid_string(digest: &[u8]) -> Option { - digest_to_record_cid_string(digest) -} - #[cfg(test)] mod tests { use super::*; -- 2.51.2