fix(universaldb): report the last error when transaction retries are exhausted

engine/packages/depot/tests/compaction_byte_volume_txn_window.rs

@@ -127,7 +127,7 @@ async fn unbounded_compaction_read_ages_out_bounded_survives() -> Result<()> {
 	let aged_out = err.chain().any(|cause| {
 		matches!(
 			cause.downcast_ref::<DatabaseError>(),
-			Some(DatabaseError::TransactionTooOld | DatabaseError::MaxRetriesReached)
+			Some(DatabaseError::TransactionTooOld | DatabaseError::MaxRetriesReached(_))
 		)
 	});
 	assert!(

engine/packages/depot/tests/compaction_hot_read_byte_volume_txn_window.rs

@@ -196,7 +196,7 @@ async fn hot_input_read_stays_bounded_at_byte_scale() -> Result<()> {
 		let aged_out = err.chain().any(|cause| {
 			matches!(
 				cause.downcast_ref::<DatabaseError>(),
-				Some(DatabaseError::TransactionTooOld | DatabaseError::MaxRetriesReached)
+				Some(DatabaseError::TransactionTooOld | DatabaseError::MaxRetriesReached(_))
 			)
 		});
 		assert!(

engine/packages/depot/tests/sqlite_byte_volume_txn_window.rs

@@ -131,7 +131,7 @@ async fn get_pages_ages_out_unbounded_bounded_survives() -> Result<()> {
 	let aged_out = err.chain().any(|cause| {
 		matches!(
 			cause.downcast_ref::<DatabaseError>(),
-			Some(DatabaseError::TransactionTooOld | DatabaseError::MaxRetriesReached)
+			Some(DatabaseError::TransactionTooOld | DatabaseError::MaxRetriesReached(_))
 		)
 	});
 	assert!(

engine/packages/universaldb/src/driver/postgres/commit.rs

@@ -74,14 +74,20 @@ async fn submit_local(
 
 	if commit_tx.send(job).await.is_err() {
 		// The leader drain loop is gone (driver shutting down). Retryable.
-		return Err(DatabaseError::NotCommitted.into());
+		return Err(
+			anyhow::Error::from(DatabaseError::NotCommitted).context("leader drain loop is gone")
+		);
 	}
 
 	match response_rx.await {
 		Ok(CommitOutcome::Committed { .. }) => Ok(()),
-		Ok(CommitOutcome::Conflict) => Err(DatabaseError::NotCommitted.into()),
+		// The leader resolved this commit as a loser. A cold-window rejection during leader recovery
+		// arrives as the same outcome, so the leader's batch log is what separates the two.
+		Ok(CommitOutcome::Conflict) => Err(anyhow::Error::from(DatabaseError::NotCommitted)
+			.context("leader resolved the commit as a conflict")),
 		// The leader dropped the job without responding; it was not applied.
-		Err(_) => Err(DatabaseError::NotCommitted.into()),
+		Err(_) => Err(anyhow::Error::from(DatabaseError::NotCommitted)
+			.context("leader dropped the commit without responding")),
 	}
 }
 
@@ -128,7 +134,9 @@ async fn submit_nats(
 					return Ok(());
 				}
 				Ok(CommitOutcome::Conflict) => {
-					return Err(DatabaseError::NotCommitted.into());
+					// As in the single-node path, a cold-window rejection is reported as a conflict.
+					return Err(anyhow::Error::from(DatabaseError::NotCommitted)
+						.context("leader resolved the commit as a conflict"));
 				}
 				Err(err) => {
 					tracing::warn!(?err, client_seq, "malformed udb commit reply; resending");
@@ -161,7 +169,11 @@ async fn submit_nats(
 		wait_ms = submit_start.elapsed().as_millis() as u64,
 		"udb commit exhausted resend attempts; treating as not committed"
 	);
-	Err(DatabaseError::NotCommitted.into())
+	Err(
+		anyhow::Error::from(DatabaseError::NotCommitted).context(format!(
+			"exhausted {MAX_SUBMIT_ATTEMPTS} commit resend attempts without a determinate reply"
+		)),
+	)
 }
 
 /// Wait for a known leader, returning a retryable error if none is elected in time.
@@ -172,7 +184,12 @@ async fn wait_for_leader(shared: &Arc<PostgresShared>) -> Result<LeaseInfo> {
 			return Ok(lease);
 		}
 		if Instant::now() >= deadline {
-			return Err(DatabaseError::NotCommitted.into());
+			return Err(
+				anyhow::Error::from(DatabaseError::NotCommitted).context(format!(
+					"no leader elected within {}s",
+					LEADER_WAIT_TIMEOUT.as_secs()
+				)),
+			);
 		}
 		tokio::time::sleep(LEADER_POLL_INTERVAL).await;
 	}

engine/packages/universaldb/src/driver/postgres/database.rs

@@ -339,18 +339,6 @@ impl DatabaseDriver for PostgresDatabaseDriver {
 
 			let mut attempt = 0;
 			loop {
-				// Re-read every iteration. The first attempt always runs, because nothing has called
-				// `retry_limit` yet; from then on the closure's limit wins over the database-wide one.
-				let limit = retry_limit.load(Ordering::SeqCst);
-				let max_attempts = if limit == RETRY_LIMIT_UNSET {
-					max_retries
-				} else {
-					limit.saturating_add(1)
-				};
-				if attempt >= max_attempts {
-					break;
-				}
-
 				let tx = Transaction::new(Arc::new(PostgresTransactionDriver::with_retry_limit(
 					self.shared.clone(),
 					retry_limit.clone(),
@@ -380,17 +368,31 @@ impl DatabaseDriver for PostgresDatabaseDriver {
 							maybe_committed = MaybeCommitted(true);
 						}
 
+						// Re-read every iteration. Nothing has called `retry_limit` before the first
+						// attempt; from then on the closure's limit wins over the database-wide one.
+						// The check runs after an attempt failed, so both values bound retries rather
+						// than total attempts.
+						let limit = retry_limit.load(Ordering::SeqCst);
+						let retry_budget = if limit == RETRY_LIMIT_UNSET {
+							max_retries
+						} else {
+							limit
+						};
+						if attempt >= retry_budget {
+							return Err(DatabaseError::MaxRetriesReached(error).into());
+						}
+
+						attempt += 1;
+
 						let backoff_ms = calculate_tx_retry_backoff(attempt as usize);
 						tokio::time::sleep(tokio::time::Duration::from_millis(backoff_ms)).await;
-						attempt += 1;
-						continue;
+					} else {
+						return Err(error);
 					}
+				} else {
+					return Err(error);
 				}
-
-				return Err(error);
 			}
-
-			Err(DatabaseError::MaxRetriesReached.into())
 		})
 	}
 

engine/packages/universaldb/src/driver/rocksdb/database.rs

@@ -79,18 +79,6 @@ impl DatabaseDriver for RocksDbDatabaseDriver {
 
 			let mut attempt = 0;
 			loop {
-				// Re-read every iteration. The first attempt always runs, because nothing has called
-				// `retry_limit` yet; from then on the closure's limit wins over the database-wide one.
-				let limit = retry_limit.load(Ordering::SeqCst);
-				let max_attempts = if limit == RETRY_LIMIT_UNSET {
-					max_retries
-				} else {
-					limit.saturating_add(1)
-				};
-				if attempt >= max_attempts {
-					break;
-				}
-
 				let tx = Transaction::new(Arc::new(RocksDbTransactionDriver::with_retry_limit(
 					self.db.clone(),
 					self.txn_conflict_tracker.clone(),
@@ -121,17 +109,31 @@ impl DatabaseDriver for RocksDbDatabaseDriver {
 							maybe_committed = MaybeCommitted(true);
 						}
 
+						// Re-read every iteration. Nothing has called `retry_limit` before the first
+						// attempt; from then on the closure's limit wins over the database-wide one.
+						// The check runs after an attempt failed, so both values bound retries rather
+						// than total attempts.
+						let limit = retry_limit.load(Ordering::SeqCst);
+						let retry_budget = if limit == RETRY_LIMIT_UNSET {
+							max_retries
+						} else {
+							limit
+						};
+						if attempt >= retry_budget {
+							return Err(DatabaseError::MaxRetriesReached(error).into());
+						}
+
+						attempt += 1;
+
 						let backoff_ms = calculate_tx_retry_backoff(attempt as usize);
 						tokio::time::sleep(tokio::time::Duration::from_millis(backoff_ms)).await;
-						attempt += 1;
-						continue;
+					} else {
+						return Err(error);
 					}
+				} else {
+					return Err(error);
 				}
-
-				return Err(error);
 			}
-
-			Err(DatabaseError::MaxRetriesReached.into())
 		})
 	}
 

engine/packages/universaldb/src/error.rs

@@ -7,8 +7,10 @@ pub enum DatabaseError {
 	#[error("transaction is too old to perform reads or be committed")]
 	TransactionTooOld,
 
-	#[error("max number of transaction retries reached")]
-	MaxRetriesReached,
+	// Stores the last error. The alternate format prints the whole context chain, so the cause the
+	// context names is reported alongside the underlying variant.
+	#[error("max number of transaction retries reached, last error: {0:#}")]
+	MaxRetriesReached(anyhow::Error),
 
 	#[error("operation issued while a commit was outstanding")]
 	UsedDuringCommit,
@@ -22,7 +24,7 @@ impl DatabaseError {
 		use DatabaseError::*;
 
 		match self {
-			NotCommitted | TransactionTooOld | MaxRetriesReached => true,
+			NotCommitted | TransactionTooOld | MaxRetriesReached(_) => true,
 			_ => false,
 		}
 	}

engine/packages/universaldb/tests/conflict_parity.rs

@@ -403,7 +403,7 @@ async fn writer_conflicts_when_a_key_it_read_was_written(db: Database) {
 	assert!(
 		err.chain().any(|x| matches!(
 			x.downcast_ref::<universaldb::error::DatabaseError>(),
-			Some(universaldb::error::DatabaseError::MaxRetriesReached)
+			Some(universaldb::error::DatabaseError::MaxRetriesReached(_))
 		)),
 		"expected the conflict to exhaust retries, got {err:?}"
 	);