diff --git a/engine/packages/depot/tests/compaction_byte_volume_txn_window.rs b/engine/packages/depot/tests/compaction_byte_volume_txn_window.rs index cac1fd2c55..8e7b4ec913 100644 --- a/engine/packages/depot/tests/compaction_byte_volume_txn_window.rs +++ b/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::(), - Some(DatabaseError::TransactionTooOld | DatabaseError::MaxRetriesReached) + Some(DatabaseError::TransactionTooOld | DatabaseError::MaxRetriesReached(_)) ) }); assert!( diff --git a/engine/packages/depot/tests/compaction_hot_read_byte_volume_txn_window.rs b/engine/packages/depot/tests/compaction_hot_read_byte_volume_txn_window.rs index 4f46704884..9a76db62aa 100644 --- a/engine/packages/depot/tests/compaction_hot_read_byte_volume_txn_window.rs +++ b/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::(), - Some(DatabaseError::TransactionTooOld | DatabaseError::MaxRetriesReached) + Some(DatabaseError::TransactionTooOld | DatabaseError::MaxRetriesReached(_)) ) }); assert!( diff --git a/engine/packages/depot/tests/sqlite_byte_volume_txn_window.rs b/engine/packages/depot/tests/sqlite_byte_volume_txn_window.rs index f948356e44..74670c70fa 100644 --- a/engine/packages/depot/tests/sqlite_byte_volume_txn_window.rs +++ b/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::(), - Some(DatabaseError::TransactionTooOld | DatabaseError::MaxRetriesReached) + Some(DatabaseError::TransactionTooOld | DatabaseError::MaxRetriesReached(_)) ) }); assert!( diff --git a/engine/packages/universaldb/src/driver/postgres/commit.rs b/engine/packages/universaldb/src/driver/postgres/commit.rs index 778c122a52..3310ca5b43 100644 --- a/engine/packages/universaldb/src/driver/postgres/commit.rs +++ b/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) -> Result { 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; } diff --git a/engine/packages/universaldb/src/driver/postgres/database.rs b/engine/packages/universaldb/src/driver/postgres/database.rs index d78d18a18b..25e31b47bf 100644 --- a/engine/packages/universaldb/src/driver/postgres/database.rs +++ b/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()) }) } diff --git a/engine/packages/universaldb/src/driver/rocksdb/database.rs b/engine/packages/universaldb/src/driver/rocksdb/database.rs index 6a68df8abc..9c4a6bf758 100644 --- a/engine/packages/universaldb/src/driver/rocksdb/database.rs +++ b/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()) }) } diff --git a/engine/packages/universaldb/src/error.rs b/engine/packages/universaldb/src/error.rs index ef98a05039..b6ddcbccf5 100644 --- a/engine/packages/universaldb/src/error.rs +++ b/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, } } diff --git a/engine/packages/universaldb/tests/conflict_parity.rs b/engine/packages/universaldb/tests/conflict_parity.rs index bde30d1733..f10403c62e 100644 --- a/engine/packages/universaldb/tests/conflict_parity.rs +++ b/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::(), - Some(universaldb::error::DatabaseError::MaxRetriesReached) + Some(universaldb::error::DatabaseError::MaxRetriesReached(_)) )), "expected the conflict to exhaust retries, got {err:?}" );