chore: consolidate queries; advisory lock · quilling.dev/parakeet@205a575

+27

consumer/src/database_writer/locking.rs

··· 96 96 (table_id, key_id) 97 97 } 98 98 99 + /// Acquire a single advisory lock 100 + /// 101 + /// This is a simple wrapper around the common pattern of: 102 + /// ```ignore 103 + /// conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]).await?; 104 + /// ``` 105 + /// 106 + /// The lock is automatically released when the transaction commits/rolls back. 107 + /// 108 + /// # Example 109 + /// ```ignore 110 + /// let (table_id, key_id) = table_record_lock("actors", did); 111 + /// acquire_lock(conn, table_id, key_id).await?; 112 + /// ``` 113 + pub async fn acquire_lock<C: deadpool_postgres::GenericClient>( 114 + conn: &C, 115 + table_id: i32, 116 + key_id: i32, 117 + ) -> eyre::Result<()> { 118 + conn.execute( 119 + "SELECT pg_advisory_xact_lock($1, $2)", 120 + &[&table_id, &key_id], 121 + ) 122 + .await?; 123 + Ok(()) 124 + } 125 + 99 126 /// Acquire transaction-level advisory locks for a set of DIDs 100 127 /// 101 128 /// Locks are acquired in sorted DID order to prevent deadlocks.

+3 -6

consumer/src/db/actor.rs

··· 16 16 ) -> Result<u64> { 17 17 // Acquire advisory lock on DID to prevent races 18 18 let (table_id, key_id) = crate::database_writer::locking::table_record_lock("actors", did); 19 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 20 - .await?; 19 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 21 20 22 21 // Allow allowlist states (synced, dirty, processing) to flow freely 23 22 // Allow upgrading from partial to allowlist states ··· 327 326 // This ensures only one transaction at a time can create/access this specific actor 328 327 // Prevents sequence consumption from concurrent inserts or rollback scenarios 329 328 let (table_id, key_id) = crate::database_writer::locking::table_record_lock("actors", did); 330 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 331 - .await?; 329 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 332 330 333 331 // Use CTE to SELECT first, then conditionally INSERT only if not found 334 332 // This prevents unnecessary sequence consumption when actor already exists ··· 449 447 // Slow path: Not in cache, do database operation 450 448 // Acquire per-DID advisory lock to prevent race conditions 451 449 let (table_id, key_id) = crate::database_writer::locking::table_record_lock("actors", did); 452 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 453 - .await?; 450 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 454 451 455 452 // Modified CTE that also returns sync_state for allowlist check and was_created flag 456 453 let row = match (status, handle) {

+2 -4

consumer/src/db/operations/feed/helpers.rs

··· 93 93 ) -> Result<(i64, bool)> { 94 94 // Acquire advisory lock on feedgen URI to prevent concurrent access races 95 95 let (table_id, key_id) = crate::database_writer::locking::table_record_lock("feedgens", at_uri); 96 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 97 - .await?; 96 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 98 97 99 98 // Parse the CID string to get the digest 100 99 let cid = ipld_core::cid::Cid::try_from(cid_str) ··· 163 162 pub(super) async fn get_list_id<C: GenericClient>(conn: &C, at_uri: &str) -> Result<(i64, bool)> { 164 163 // Acquire advisory lock on list URI to prevent concurrent access races 165 164 let (table_id, key_id) = crate::database_writer::locking::table_record_lock("lists", at_uri); 166 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 167 - .await?; 165 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 168 166 169 167 // Extract owner DID and rkey from AT URI 170 168 let did = parakeet_db::at_uri_util::extract_did(at_uri)

+1 -2

consumer/src/db/operations/feed/like.rs

··· 51 51 // Acquire advisory lock on record to prevent deadlocks 52 52 let lock_start = std::time::Instant::now(); 53 53 let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("likes", actor_id, rkey); 54 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 55 - .await?; 54 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 56 55 let lock_ms = lock_start.elapsed().as_millis(); 57 56 58 57 // Note: via_repost_id is already resolved in the reference extraction phase (workers.rs)

+1 -2

consumer/src/db/operations/feed/post.rs

··· 76 76 77 77 // Acquire advisory lock using actor_id and rkey to prevent deadlocks 78 78 let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("posts", actor_id, rkey); 79 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 80 - .await?; 79 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 81 80 82 81 // Get parent and root post natural keys (create stubs if necessary) 83 82 // First get the actor IDs, then get the post natural keys

+1 -2

consumer/src/db/operations/feed/repost.rs

··· 44 44 45 45 // Acquire advisory lock on record to prevent deadlocks 46 46 let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("reposts", actor_id, rkey); 47 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 48 - .await?; 47 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 49 48 50 49 // Note: via_repost_id is already resolved in the reference extraction phase (workers.rs) 51 50 // This ensures the FK constraint is satisfied before we reach this point

+6 -12

consumer/src/db/operations/graph.rs

··· 20 20 ) -> Result<u64> { 21 21 // Acquire advisory lock on record to prevent deadlocks 22 22 let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("follows", actor_id, rkey); 23 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 24 - .await?; 23 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 25 24 26 25 // Insert follow - simple insert on primary key (actor_id, rkey) 27 26 // Note: CID is synthetic, generated from actor_id + rkey ··· 69 68 ) -> Result<u64> { 70 69 // Acquire advisory lock on record to prevent deadlocks 71 70 let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("blocks", actor_id, rkey); 72 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 73 - .await?; 71 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 74 72 75 73 // Insert block - simple insert on primary key (actor_id, rkey) 76 74 // Note: CID is synthetic, generated from actor_id + rkey ··· 114 112 115 113 // Acquire table-scoped advisory lock on lists table for this record 116 114 let (table_id, key_id) = crate::database_writer::locking::actor_record_lock_str("lists", actor_id, rkey); 117 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 118 - .await?; 115 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 119 116 120 117 conn.query_one( 121 118 include_str!("../sql/list_upsert.sql"), ··· 162 159 163 160 // Acquire advisory lock on record to prevent deadlocks 164 161 let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("list_blocks", actor_id, rkey); 165 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 166 - .await?; 162 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 167 163 168 164 conn.execute( 169 165 include_str!("../sql/list_block_upsert.sql"), ··· 210 206 211 207 // Acquire advisory lock on record to prevent deadlocks 212 208 let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("list_items", actor_id, rkey); 213 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 214 - .await?; 209 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 215 210 216 211 // Resolve list_id by creating stub list if needed 217 212 let list_id = super::feed::ensure_list_id(conn, &rec.list).await?; ··· 262 257 263 258 // Acquire advisory lock on record to prevent deadlocks 264 259 let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("verification", actor_id, rkey); 265 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 266 - .await?; 260 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 267 261 268 262 // Insert verification - actors already resolved (no SELECT subquery needed) 269 263 // Note: created_at is derived from TID rkey

+2 -4

consumer/src/db/operations/labeler.rs

··· 24 24 // Labeler URI is always at://{did}/app.bsky.labeler.service/self 25 25 let labeler_uri = format!("at://{}/app.bsky.labeler.service/self", did); 26 26 let (table_id, key_id) = crate::database_writer::locking::table_record_lock("labelers", &labeler_uri); 27 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 28 - .await?; 27 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 29 28 30 29 // Parse the CID string to get the digest 31 30 let cid = ipld_core::cid::Cid::try_from(cid_str) ··· 75 74 // Acquire table-scoped advisory lock on labelers table for this actor 76 75 // This prevents concurrent labeler updates for the same actor 77 76 let (table_id, key_id) = crate::database_writer::locking::table_record_lock("labelers", did); 78 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 79 - .await?; 77 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 80 78 81 79 let cid_bytes = cid.to_bytes(); 82 80 let cid_digest = parakeet_db::cid_util::cid_to_digest(&cid_bytes)

+1 -2

consumer/src/db/operations/starter_pack.rs

··· 24 24 25 25 // Acquire advisory lock on record to prevent deadlocks 26 26 let (table_id, key_id) = crate::database_writer::locking::actor_record_lock("starterpacks", actor_id, rkey); 27 - conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) 28 - .await?; 27 + crate::database_writer::locking::acquire_lock(conn, table_id, key_id).await?; 29 28 30 29 // Resolve list_id by creating stub list if needed 31 30 let list_id = super::feed::ensure_list_id(conn, &rec.list).await?;

Configure Feed

Configure Feed