store: Replace std::sync::RwLock with parking_lot::RwLock in writable

Replace std::sync::RwLock with parking_lot::RwLock in the background
writer's Request::Write batch handling. This reduces lock contention
as parking_lot's RwLock is faster for short-held locks due to
efficient spinning before parking.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: lutter

store/postgres/src/writable.rs

@@ -1,7 +1,9 @@
 use std::collections::BTreeSet;
 use std::ops::{Deref, Range};
 use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::{Mutex, RwLock, TryLockError as RwLockError};
+use std::sync::Mutex;
+
+use graph::parking_lot::RwLock;
 use std::time::Instant;
 use std::{collections::BTreeMap, sync::Arc};
 
@@ -574,7 +576,7 @@ impl BlockTracker {
         // processed.
         let res = queue.find_map(|req| match req.as_ref() {
             Request::Write { batch, .. } => {
-                let batch = batch.read().unwrap();
+                let batch = batch.read();
                 tracker.write(&batch.block_ptr);
                 if batch.first_block <= tracker.revert {
                     let res = f(batch.deref(), tracker.revert);
@@ -613,7 +615,7 @@ impl BlockTracker {
         let accum = queue.fold(init, |accum, req| {
             match req.as_ref() {
                 Request::Write { batch, .. } => {
-                    let batch = batch.read().unwrap();
+                    let batch = batch.read();
                     let mut accum = accum;
                     tracker.write(&batch.block_ptr);
                     if batch.first_block <= tracker.revert {
@@ -740,7 +742,7 @@ impl std::fmt::Debug for Request {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::Write { batch, store, .. } => {
-                let batch = batch.read().unwrap();
+                let batch = batch.read();
                 write!(
                     f,
                     "write[{}, {:p}, {} entities]",
@@ -811,7 +813,7 @@ impl Request {
             } => {
                 let start = Instant::now();
 
-                let batch = batch.write().unwrap().close();
+                let batch = batch.write().close();
 
                 if let Some(err) = &batch.error {
                     // This can happen when appending to the batch failed
@@ -850,7 +852,7 @@ impl Request {
     fn should_process(&self) -> bool {
         match self {
             Request::Write { queued, batch, .. } => {
-                batch.read().unwrap().weight() >= ENV_VARS.store.write_batch_size
+                batch.read().weight() >= ENV_VARS.store.write_batch_size
                     || queued.elapsed() >= ENV_VARS.store.write_batch_duration
             }
             Request::RevertTo { .. } | Request::Stop => true,
@@ -1169,7 +1171,7 @@ impl Queue {
                             // duration of the write, and we do not want to
                             // slow down queueing requests unnecessarily
                             match existing.try_write() {
-                                Ok(mut existing) => {
+                                Some(mut existing) => {
                                     if existing.weight() < ENV_VARS.store.write_batch_size {
                                         let res = existing.append(batch).map(|()| None);
                                         if existing.weight() >= ENV_VARS.store.write_batch_size {
@@ -1180,16 +1182,13 @@ impl Queue {
                                         Ok(Some(batch))
                                     }
                                 }
-                                Err(RwLockError::WouldBlock) => {
+                                None => {
                                     // This branch can cause batches that
                                     // are not 'full' at the head of the
                                     // queue, something that start_writer
                                     // has to take into account
                                     Ok(Some(batch))
                                 }
-                                Err(RwLockError::Poisoned(e)) => {
-                                    panic!("rwlock on batch was poisoned {:?}", e);
-                                }
                             }
                         } else {
                             Ok(Some(batch))