fix: clarify queued repo lock lifecycle

docs/postgres-event-store.md

@@ -39,6 +39,12 @@ Event payload bytes are currently bitcode-encoded. Postgres rows must carry code
 
 The repository should reject unknown codec labels or versions unless an explicit decoder/upcaster path exists. Payload schema changes still use `event_version` and aggregate upcasters; codec metadata describes the byte encoding, not the domain event version.
 
+## Locking Model
+
+`QueuedRepository` remains a process-local coordination wrapper for examples, tests, and single-process adapters. It complements a Postgres repository but must not be the durable cross-process locking mechanism.
+
+The Postgres repository should enforce optimistic concurrency with the `(aggregate_type, aggregate_id, sequence)` uniqueness constraint. If a queued read/modify/write API is exposed for Postgres, it should use database-backed row locks or advisory locks inside the repository/transaction boundary. Those database locks replace process-local queue locks for cross-process writer coordination; `QueuedRepository` can still wrap a Postgres repository only as an additional in-process convenience layer.
+
 ## Backward Compatibility
 
 Rows or imported JSON records without event metadata deserialize with empty metadata. Postgres migrations should still write `metadata jsonb NOT NULL DEFAULT '{}'` so newly stored rows are explicit.

src/queued_repo/repository.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use crate::entity::{Committable, Entity};
-use crate::lock::{InMemoryLockManager, Lock, LockManager};
+use crate::lock::{InMemoryLockManager, Lock, LockError, LockManager};
 use crate::repository::{
     Commit, CommitBatch, Count, Exists, Find, FindOne, Get, GetMany, GetOne, RepositoryError,
     TransactionalCommit,
@@ -28,6 +28,17 @@ impl ReadOpts {
     }
 }
 
+/// Repository wrapper that serializes access with process-local per-stream locks.
+///
+/// Locking reads (`get`, `get_many`, `find`, and `find_one`) intentionally keep
+/// matching locks held after returning. Call `commit` to release those locks
+/// after a successful write, or call `abort`/`unlock` when the loaded entity is
+/// no longer being written. Dropping a loaded entity without `commit` or `abort`
+/// leaves its in-memory lock held until an explicit unlock.
+///
+/// Commit releases held locks only after the inner repository succeeds. On
+/// commit errors, locks remain held so callers can inspect state, retry, or
+/// explicitly abort.
 pub struct QueuedRepository<R, L: LockManager = InMemoryLockManager> {
     inner: R,
     lock_manager: Arc<L>,
@@ -71,8 +82,11 @@ impl<R, L: LockManager> QueuedRepository<R, L> {
     }
 
     pub fn lock(&self, id: impl AsRef<str>) -> Result<(), RepositoryError> {
-        let lock = self.ensure_lock(id.as_ref())?;
-        let _ = lock.try_lock()?;
+        let id = id.as_ref();
+        let lock = self.ensure_lock(id)?;
+        if !lock.try_lock()? {
+            return Err(LockError::AcquireFailed(format!("lock for {id} is already held")).into());
+        }
         Ok(())
     }
 
@@ -200,7 +214,9 @@ impl<R: Commit, L: LockManager> Commit for QueuedRepository<R, L> {
     fn commit<C: Committable + ?Sized>(&self, committable: &mut C) -> Result<(), RepositoryError> {
         let entities = committable.entities_mut();
 
-        // Acquire locks for all entities
+        // Commit releases locks that were acquired by a prior locking read or
+        // manual lock call. It does not acquire ownership itself because this
+        // lock implementation has no guard token or owner tracking.
         let mut locks = Vec::with_capacity(entities.len());
         for entity in &entities {
             locks.push(self.ensure_lock(entity.id())?);
@@ -209,7 +225,7 @@ impl<R: Commit, L: LockManager> Commit for QueuedRepository<R, L> {
         // Delegate to inner repository
         let result = self.inner.commit(committable);
 
-        // Unlock on success
+        // Keep locks held on errors so callers can retry or explicitly abort.
         if result.is_ok() {
             for lock in locks {
                 lock.unlock()?;
@@ -223,6 +239,8 @@ impl<R: Commit, L: LockManager> Commit for QueuedRepository<R, L> {
 impl<R: TransactionalCommit, L: LockManager> TransactionalCommit for QueuedRepository<R, L> {
     fn commit_batch(&self, batch: CommitBatch<'_>) -> Result<(), RepositoryError> {
         let ids: Vec<&str> = batch.entities.iter().map(|entity| entity.id()).collect();
+        // See `Commit::commit`: these handles are released after successful
+        // inner commit and intentionally kept held on errors.
         let mut locks = Vec::with_capacity(ids.len());
         for id in ids {
             locks.push(self.ensure_lock(id)?);

tests/todos/main.rs

@@ -4,7 +4,8 @@ use aggregate::{Todo, TodoSnapshot};
 use bitcode;
 use sourced_rust::{
     AggregateBuilder, Commit, EventEmitter, GetAggregate, HashMapRepository, LocalEmitterPublisher,
-    LogPublisher, OutboxCommitExt, OutboxMessage, OutboxRepositoryExt, OutboxWorker, Queueable,
+    LockError, LogPublisher, OutboxCommitExt, OutboxMessage, OutboxRepositoryExt, OutboxWorker,
+    Queueable, RepositoryError,
 };
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{mpsc, Arc, Mutex};
@@ -455,6 +456,72 @@ fn queued_repo_blocks_get_until_commit() {
     assert!(rx_done.recv_timeout(Duration::from_millis(500)).is_ok());
 }
 
+#[test]
+fn manual_lock_reports_failure_when_already_held() {
+    let repo = HashMapRepository::new().queued();
+    let id = next_id();
+
+    repo.lock(&id).unwrap();
+    let err = repo.lock(&id).expect_err("second manual lock should fail");
+    let is_lock_failure = matches!(
+        &err,
+        RepositoryError::Lock(LockError::AcquireFailed(message)) if message.contains(&id)
+    );
+
+    assert!(is_lock_failure, "unexpected error: {err}");
+    repo.unlock(&id).unwrap();
+}
+
+#[test]
+fn commit_failure_keeps_lock_until_abort() {
+    let repo = Arc::new(HashMapRepository::new().queued().aggregate::<Todo>());
+    let mut todo = Todo::new();
+    let id = next_id();
+    todo.initialize(
+        id.clone(),
+        "user1".to_string(),
+        "Commit failure lock".to_string(),
+    )
+    .unwrap();
+    repo.commit(&mut todo).unwrap();
+
+    let mut locked = repo.get(&id).unwrap().unwrap();
+    let mut concurrent = repo
+        .repo()
+        .inner()
+        .get_aggregate::<Todo>(&id)
+        .unwrap()
+        .unwrap();
+    concurrent.complete().unwrap();
+    repo.repo().inner().commit(&mut concurrent.entity).unwrap();
+
+    locked.complete().unwrap();
+    let err = repo
+        .commit(&mut locked)
+        .expect_err("stale locked aggregate should fail optimistic commit");
+    assert!(
+        matches!(err, RepositoryError::ConcurrentWrite { .. }),
+        "unexpected error: {err}"
+    );
+
+    let (tx_started, rx_started) = mpsc::channel();
+    let (tx_got, rx_got) = mpsc::channel();
+    let repo_other = Arc::clone(&repo);
+    let id_other = id.clone();
+    thread::spawn(move || {
+        tx_started.send(()).unwrap();
+        let todo = repo_other.get(&id_other).unwrap().unwrap();
+        repo_other.abort(&todo).unwrap();
+        tx_got.send(()).unwrap();
+    });
+
+    rx_started.recv().unwrap();
+    assert!(rx_got.recv_timeout(Duration::from_millis(200)).is_err());
+
+    repo.abort(&locked).unwrap();
+    assert!(rx_got.recv_timeout(Duration::from_millis(500)).is_ok());
+}
+
 #[test]
 fn outbox_worker_process_next_with_commit() {
     let repo = HashMapRepository::new();