diff --git a/nexus/db-queries/src/db/datastore/inventory.rs b/nexus/db-queries/src/db/datastore/inventory.rs
index ccf453c5cd7..cd320de537d 100644
--- a/nexus/db-queries/src/db/datastore/inventory.rs
+++ b/nexus/db-queries/src/db/datastore/inventory.rs
@@ -28,6 +28,7 @@ use futures::FutureExt;
 use futures::future::BoxFuture;
 use iddqd::{IdOrdItem, IdOrdMap, id_upcast};
 use nexus_db_errors::ErrorHandler;
+use nexus_db_errors::OptionalError;
 use nexus_db_errors::public_error_from_diesel;
 use nexus_db_errors::public_error_from_diesel_lookup;
 use nexus_db_model::ArtifactHash;
@@ -95,7 +96,6 @@ use omicron_common::api::external::Error;
 use omicron_common::api::external::InternalContext;
 use omicron_common::api::external::LookupType;
 use omicron_common::api::external::ResourceType;
-use omicron_common::bail_unless;
 use omicron_common::disk::M2Slot;
 use omicron_uuid_kinds::CollectionUuid;
 use omicron_uuid_kinds::DatasetUuid;
@@ -2394,124 +2394,125 @@ impl DataStore {
         id: CollectionUuid,
         batch_size: NonZeroU32,
     ) -> Result<Collection, Error> {
+        let err = OptionalError::new();
         let conn = self.pool_connection_authorized(opctx).await?;
         let db_id = to_db_typed_uuid(id);
-        let (time_started, time_done, collector) = {
-            use nexus_db_schema::schema::inv_collection::dsl;
-
-            let collections = dsl::inv_collection
-                .filter(dsl::id.eq(db_id))
-                .limit(2)
-                .select(InvCollection::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
-            bail_unless!(collections.len() == 1);
-            let collection = collections.into_iter().next().unwrap();
-            (
-                collection.time_started,
-                collection.time_done,
-                collection.collector,
-            )
-        };
 
-        let errors: Vec<String> = {
-            use nexus_db_schema::schema::inv_collection_error::dsl;
-            let mut errors = Vec::new();
-            let mut paginator = Paginator::new(
-                batch_size,
-                dropshot::PaginationOrder::Ascending,
-            );
-            while let Some(p) = paginator.next() {
-                let batch = paginated(
-                    dsl::inv_collection_error,
-                    dsl::idx,
-                    &p.current_pagparams(),
-                )
-                .filter(dsl::inv_collection_id.eq(db_id))
-                .order_by(dsl::idx)
-                .select(InvCollectionError::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
-                paginator =
-                    p.found_batch(&batch, &|row: &InvCollectionError| row.idx);
-                errors.extend(batch.into_iter().map(|e| e.message));
-            }
-            errors
-        };
+        self.transaction_retry_wrapper("inventory_collection_read")
+            .transaction(&conn, |conn| {
+                let err = err.clone();
+                async move {
+                    let (time_started, time_done, collector) = {
+                        use nexus_db_schema::schema::inv_collection::dsl;
 
-        let sps: BTreeMap<_, _> = {
-            use nexus_db_schema::schema::inv_service_processor::dsl;
+                        let collections = dsl::inv_collection
+                            .filter(dsl::id.eq(db_id))
+                            .limit(2)
+                            .select(InvCollection::as_select())
+                            .load_async(&conn)
+                            .await?;
+                        if collections.len() != 1 {
+                            return Err(err.bail(Error::internal_error(
+                                &format!(
+                                    "failed runtime check: {:?}",
+                                    "collections.len() == 1"
+                                ),
+                            )));
+                        }
+                        let collection = collections.into_iter().next().unwrap();
+                        (
+                            collection.time_started,
+                            collection.time_done,
+                            collection.collector,
+                        )
+                    };
 
-            let mut sps = BTreeMap::new();
+                    let errors: Vec<String> = {
+                        use nexus_db_schema::schema::inv_collection_error::dsl;
+                        let mut errors = Vec::new();
+                        let mut paginator = Paginator::new(
+                            batch_size,
+                            dropshot::PaginationOrder::Ascending,
+                        );
+                        while let Some(p) = paginator.next() {
+                            let batch = paginated(
+                                dsl::inv_collection_error,
+                                dsl::idx,
+                                &p.current_pagparams(),
+                            )
+                            .filter(dsl::inv_collection_id.eq(db_id))
+                            .order_by(dsl::idx)
+                            .select(InvCollectionError::as_select())
+                            .load_async(&conn)
+                            .await?;
+                            paginator =
+                                p.found_batch(&batch, &|row: &InvCollectionError| row.idx);
+                            errors.extend(batch.into_iter().map(|e| e.message));
+                        }
+                        errors
+                    };
 
-            let mut paginator = Paginator::new(
-                batch_size,
-                dropshot::PaginationOrder::Ascending,
-            );
-            while let Some(p) = paginator.next() {
-                let batch = paginated(
-                    dsl::inv_service_processor,
-                    dsl::hw_baseboard_id,
-                    &p.current_pagparams(),
-                )
-                .filter(dsl::inv_collection_id.eq(db_id))
-                .select(InvServiceProcessor::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
-                paginator = p.found_batch(&batch, &|row| row.hw_baseboard_id);
-                sps.extend(batch.into_iter().map(|row| {
-                    let baseboard_id = row.hw_baseboard_id;
-                    (
-                        baseboard_id,
-                        nexus_types::inventory::ServiceProcessor::from(row),
-                    )
-                }));
-            }
-            sps
-        };
+                    let sps: BTreeMap<_, _> = {
+                        use nexus_db_schema::schema::inv_service_processor::dsl;
+
+                        let mut sps = BTreeMap::new();
+
+                        let mut paginator = Paginator::new(
+                            batch_size,
+                            dropshot::PaginationOrder::Ascending,
+                        );
+                        while let Some(p) = paginator.next() {
+                            let batch = paginated(
+                                dsl::inv_service_processor,
+                                dsl::hw_baseboard_id,
+                                &p.current_pagparams(),
+                            )
+                            .filter(dsl::inv_collection_id.eq(db_id))
+                            .select(InvServiceProcessor::as_select())
+                            .load_async(&conn)
+                            .await?;
+                            paginator = p.found_batch(&batch, &|row| row.hw_baseboard_id);
+                            sps.extend(batch.into_iter().map(|row| {
+                                let baseboard_id = row.hw_baseboard_id;
+                                (
+                                    baseboard_id,
+                                    nexus_types::inventory::ServiceProcessor::from(row),
+                                )
+                            }));
+                        }
+                        sps
+                    };
 
-        let rots: BTreeMap<_, _> = {
-            use nexus_db_schema::schema::inv_root_of_trust::dsl;
+                    let rots: BTreeMap<_, _> = {
+                        use nexus_db_schema::schema::inv_root_of_trust::dsl;
 
-            let mut rots = BTreeMap::new();
+                        let mut rots = BTreeMap::new();
 
-            let mut paginator = Paginator::new(
-                batch_size,
-                dropshot::PaginationOrder::Ascending,
-            );
-            while let Some(p) = paginator.next() {
-                let batch = paginated(
-                    dsl::inv_root_of_trust,
-                    dsl::hw_baseboard_id,
-                    &p.current_pagparams(),
-                )
-                .filter(dsl::inv_collection_id.eq(db_id))
-                .select(InvRootOfTrust::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
-                paginator = p.found_batch(&batch, &|row| row.hw_baseboard_id);
-                rots.extend(batch.into_iter().map(|rot_row| {
-                    let baseboard_id = rot_row.hw_baseboard_id;
-                    (
-                        baseboard_id,
-                        nexus_types::inventory::RotState::from(rot_row),
-                    )
-                }));
-            }
-            rots
-        };
+                        let mut paginator = Paginator::new(
+                            batch_size,
+                            dropshot::PaginationOrder::Ascending,
+                        );
+                        while let Some(p) = paginator.next() {
+                            let batch = paginated(
+                                dsl::inv_root_of_trust,
+                                dsl::hw_baseboard_id,
+                                &p.current_pagparams(),
+                            )
+                            .filter(dsl::inv_collection_id.eq(db_id))
+                            .select(InvRootOfTrust::as_select())
+                            .load_async(&conn)
+                            .await?;
+                            paginator = p.found_batch(&batch, &|row| row.hw_baseboard_id);
+                            rots.extend(batch.into_iter().map(|rot_row| {
+                                let baseboard_id = rot_row.hw_baseboard_id;
+                                (
+                                    baseboard_id,
+                                    nexus_types::inventory::RotState::from(rot_row),
+                                )
+                            }));
+                        }
+                        rots
+                    };
 
         let sled_agent_rows: Vec<_> = {
             use nexus_db_schema::schema::inv_sled_agent::dsl;
@@ -2530,11 +2531,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvSledAgent::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.sled_id);
                 rows.append(&mut batch);
             }
@@ -2565,11 +2563,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvNvmeDiskFirmware::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator =
                     p.found_batch(&batch, &|row| (row.sled_id(), row.slot()));
                 for firmware in batch {
@@ -2610,11 +2605,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvPhysicalDisk::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator =
                     p.found_batch(&batch, &|row| (row.sled_id, row.slot));
                 for disk in batch {
@@ -2659,11 +2651,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvZpool::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| (row.sled_id, row.id));
                 for zpool in batch {
                     zpools
@@ -2693,11 +2682,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvDataset::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| {
                     (row.sled_id, row.name.clone())
                 });
@@ -2737,11 +2723,8 @@ impl DataStore {
                 )
                 .filter(dsl::id.eq_any(baseboard_id_ids.clone()))
                 .select(HwBaseboardId::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.id);
                 bbs.extend(
                     batch
@@ -2766,7 +2749,8 @@ impl DataStore {
                     },
                 )
             })
-            .collect::<Result<BTreeMap<_, _>, _>>()?;
+            .collect::<Result<BTreeMap<_, _>, _>>()
+            .map_err(|e| err.bail(e))?;
         let rots = rots
             .into_iter()
             .map(|(id, rot)| {
@@ -2779,7 +2763,8 @@ impl DataStore {
                         )
                     })
             })
-            .collect::<Result<BTreeMap<_, _>, _>>()?;
+            .collect::<Result<BTreeMap<_, _>, _>>()
+            .map_err(|e| err.bail(e))?;
 
         // Fetch the host phase 1 active slots found.
         let host_phase_1_active_slots = {
@@ -2799,19 +2784,16 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvHostPhase1ActiveSlot::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.hw_baseboard_id);
                 for row in batch {
                     let bb = baseboards_by_id
                         .get(&row.hw_baseboard_id)
                         .ok_or_else(|| {
-                            Error::internal_error(
+                            err.bail(Error::internal_error(
                                 "missing baseboard that we should have fetched",
-                            )
+                            ))
                         })?;
                     slots.insert(Arc::clone(bb), row.into());
                 }
@@ -2838,11 +2820,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvHostPhase1FlashHash::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| {
                     (row.hw_baseboard_id, row.slot)
                 });
@@ -2864,7 +2843,7 @@ impl DataStore {
                      inv_host_phase_1_flash_hash: {}",
                     p.hw_baseboard_id
                 );
-                return Err(Error::internal_error(&msg));
+                return Err(err.bail(Error::internal_error(&msg)));
             };
 
             let previous = by_baseboard.insert(
@@ -2876,12 +2855,13 @@ impl DataStore {
                     hash: *p.hash,
                 },
             );
-            bail_unless!(
-                previous.is_none(),
-                "duplicate host phase 1 flash hash found: {:?} baseboard {:?}",
-                p.slot,
-                p.hw_baseboard_id
-            );
+            if previous.is_some() {
+                return Err(err.bail(Error::internal_error(&format!(
+                    "duplicate host phase 1 flash hash found: {:?} baseboard {:?}",
+                    p.slot,
+                    p.hw_baseboard_id
+                ))));
+            }
         }
 
         // Fetch records of cabooses found.
@@ -2902,11 +2882,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvCaboose::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| {
                     (row.hw_baseboard_id, row.which)
                 });
@@ -2936,11 +2913,8 @@ impl DataStore {
                     paginated(dsl::sw_caboose, dsl::id, &p.current_pagparams())
                         .filter(dsl::id.eq_any(sw_caboose_ids.clone()))
                         .select(SwCaboose::as_select())
-                        .load_async(&*conn)
-                        .await
-                        .map_err(|e| {
-                            public_error_from_diesel(e, ErrorHandler::Server)
-                        })?;
+                        .load_async(&conn)
+                        .await?;
                 paginator = p.found_batch(&batch, &|row| row.id);
                 cabooses.extend(batch.into_iter().map(|sw_caboose_row| {
                     (
@@ -2966,14 +2940,14 @@ impl DataStore {
                     "unknown baseboard found in inv_caboose: {}",
                     c.hw_baseboard_id
                 );
-                return Err(Error::internal_error(&msg));
+                return Err(err.bail(Error::internal_error(&msg)));
             };
             let Some(sw_caboose) = cabooses_by_id.get(&c.sw_caboose_id) else {
                 let msg = format!(
                     "unknown caboose found in inv_caboose: {}",
                     c.sw_caboose_id
                 );
-                return Err(Error::internal_error(&msg));
+                return Err(err.bail(Error::internal_error(&msg)));
             };
 
             let previous = by_baseboard.insert(
@@ -2984,12 +2958,13 @@ impl DataStore {
                     caboose: sw_caboose.clone(),
                 },
             );
-            bail_unless!(
-                previous.is_none(),
-                "duplicate caboose found: {:?} baseboard {:?}",
-                c.which,
-                c.hw_baseboard_id
-            );
+            if previous.is_some() {
+                return Err(err.bail(Error::internal_error(&format!(
+                    "duplicate caboose found: {:?} baseboard {:?}",
+                    c.which,
+                    c.hw_baseboard_id
+                ))));
+            }
         }
 
         // Fetch records of RoT pages found.
@@ -3010,11 +2985,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvRotPage::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| {
                     (row.hw_baseboard_id, row.which)
                 });
@@ -3047,11 +3019,8 @@ impl DataStore {
                 )
                 .filter(dsl::id.eq_any(sw_rot_page_ids.clone()))
                 .select(SwRotPage::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.id);
                 rot_pages.extend(batch.into_iter().map(|sw_rot_page_row| {
                     (
@@ -3077,7 +3046,7 @@ impl DataStore {
                     "unknown baseboard found in inv_root_of_trust_page: {}",
                     p.hw_baseboard_id
                 );
-                return Err(Error::internal_error(&msg));
+                return Err(err.bail(Error::internal_error(&msg)));
             };
             let Some(sw_rot_page) =
                 rot_pages_by_id.get(&p.sw_root_of_trust_page_id)
@@ -3086,7 +3055,7 @@ impl DataStore {
                     "unknown rot page found in inv_root_of_trust_page: {}",
                     p.sw_root_of_trust_page_id
                 );
-                return Err(Error::internal_error(&msg));
+                return Err(err.bail(Error::internal_error(&msg)));
             };
 
             let previous = by_baseboard.insert(
@@ -3097,12 +3066,13 @@ impl DataStore {
                     page: sw_rot_page.clone(),
                 },
             );
-            bail_unless!(
-                previous.is_none(),
-                "duplicate rot page found: {:?} baseboard {:?}",
-                p.which,
-                p.hw_baseboard_id
-            );
+            if previous.is_some() {
+                return Err(err.bail(Error::internal_error(&format!(
+                    "duplicate rot page found: {:?} baseboard {:?}",
+                    p.which,
+                    p.hw_baseboard_id
+                ))));
+            }
         }
 
         // Now read the `OmicronSledConfig`s.
@@ -3134,11 +3104,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvOmicronSledConfig::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.id);
                 for sled_config in batch {
                     configs
@@ -3156,11 +3123,11 @@ impl DataStore {
                             },
                         })
                         .map_err(|e| {
-                            Error::internal_error(&format!(
+                            err.bail(Error::internal_error(&format!(
                                 "duplicate omicron sled config ID found, but \
                                  database guarantees uniqueness: {}",
                                 InlineErrorChain::new(&e),
-                            ))
+                            )))
                         })?;
                 }
             }
@@ -3189,11 +3156,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvOmicronSledConfigZoneNic::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.id);
                 nics.extend(batch.into_iter().map(|found_zone_nic| {
                     (
@@ -3224,11 +3188,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvOmicronSledConfigZone::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.id);
                 zones.append(&mut batch);
             }
@@ -3252,7 +3213,8 @@ impl DataStore {
                         ))
                     })
                 })
-                .transpose()?;
+                .transpose()
+                .map_err(|e| err.bail(e))?;
             let mut config_with_id = omicron_sled_configs
                 .get_mut(&z.sled_config_id)
                 .ok_or_else(|| {
@@ -3260,10 +3222,10 @@ impl DataStore {
                     // inv_omicron_sled_config_zone with no associated record in
                     // inv_sled_omicron_config. This should be impossible and
                     // reflects either a bug or database corruption.
-                    Error::internal_error(&format!(
+                    err.bail(Error::internal_error(&format!(
                         "zone {:?}: unknown config ID: {:?}",
                         z.id, z.sled_config_id
-                    ))
+                    )))
                 })?;
             let zone_id = z.id;
             let zone = z
@@ -3272,16 +3234,17 @@ impl DataStore {
                     format!("zone {:?}: parse from database", zone_id)
                 })
                 .map_err(|e| {
-                    Error::internal_error(&format!("{:#}", e.to_string()))
+                    err.bail(Error::internal_error(&format!("{:#}", e.to_string())))
                 })?;
             config_with_id.config.zones.insert_overwrite(zone);
         }
 
-        bail_unless!(
-            omicron_zone_nics.is_empty(),
-            "found extra Omicron zone NICs: {:?}",
-            omicron_zone_nics.keys()
-        );
+        if !omicron_zone_nics.is_empty() {
+            return Err(err.bail(Error::internal_error(&format!(
+                "found extra Omicron zone NICs: {:?}",
+                omicron_zone_nics.keys()
+            ))));
+        }
 
         // Now load the datasets from all configs.
         {
@@ -3299,25 +3262,22 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvOmicronSledConfigDataset::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.id);
 
                 for row in batch {
                     let mut config_with_id = omicron_sled_configs
                         .get_mut(&row.sled_config_id)
                         .ok_or_else(|| {
-                            Error::internal_error(&format!(
+                            err.bail(Error::internal_error(&format!(
                                 "dataset config {:?}: unknown config ID: {:?}",
                                 row.id, row.sled_config_id
-                            ))
+                            )))
                         })?;
                     config_with_id.config.datasets.insert_overwrite(
                         row.try_into().map_err(|e| {
-                            Error::internal_error(&format!("{e:#}"))
+                            err.bail(Error::internal_error(&format!("{e:#}")))
                         })?,
                     );
                 }
@@ -3340,21 +3300,18 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvOmicronSledConfigDisk::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.id);
 
                 for row in batch {
                     let mut config_with_id = omicron_sled_configs
                         .get_mut(&row.sled_config_id)
                         .ok_or_else(|| {
-                            Error::internal_error(&format!(
+                            err.bail(Error::internal_error(&format!(
                                 "disk config {:?}: unknown config ID: {:?}",
                                 row.id, row.sled_config_id
-                            ))
+                            )))
                         })?;
                     config_with_id.config.disks.insert_overwrite(row.into());
                 }
@@ -3380,11 +3337,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvSledConfigReconciler::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.sled_id);
 
                 for row in batch {
@@ -3415,11 +3369,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvSledBootPartition::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| {
                     (row.sled_id, row.boot_disk_slot)
                 });
@@ -3427,8 +3378,8 @@ impl DataStore {
                 for row in batch {
                     let sled_map =
                         results.entry(row.sled_id.into()).or_default();
-                    let slot = row.slot().map_err(|err| {
-                        Error::internal_error(&format!("{err:#}"))
+                    let slot = row.slot().map_err(|e| {
+                        err.bail(Error::internal_error(&format!("{e:#}")))
                     })?;
                     sled_map.insert(slot, row);
                 }
@@ -3459,11 +3410,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvLastReconciliationDiskResult::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.disk_id);
 
                 for row in batch {
@@ -3498,11 +3446,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvLastReconciliationDatasetResult::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.dataset_id);
 
                 for row in batch {
@@ -3534,24 +3479,21 @@ impl DataStore {
             let rows = dsl::inv_last_reconciliation_orphaned_dataset
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvLastReconciliationOrphanedDataset::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
 
             for row in rows {
                 orphaned
                     .entry(row.sled_id.into())
                     .or_default()
-                    .insert_unique(row.try_into()?)
-                    .map_err(|err| {
+                    .insert_unique(row.try_into().map_err(|e| err.bail(e))?)
+                    .map_err(|e| {
                         // We should never get duplicates: the table's primary
                         // key is the dataset name (same as the IdOrdMap)
-                        Error::internal_error(&format!(
+                        err.bail(Error::internal_error(&format!(
                             "unexpected duplicate orphaned dataset: {}",
-                            InlineErrorChain::new(&err)
-                        ))
+                            InlineErrorChain::new(&e)
+                        )))
                     })?;
             }
 
@@ -3580,11 +3522,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvLastReconciliationZoneResult::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.zone_id);
 
                 for row in batch {
@@ -3618,11 +3557,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvZoneManifestZone::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| {
                     (row.sled_id, row.zone_file_name.clone())
                 });
@@ -3633,7 +3569,7 @@ impl DataStore {
                         .or_default()
                         .insert_unique(row.try_into().map_err(
                             |e: anyhow::Error| {
-                                Error::internal_error(&e.to_string())
+                                err.bail(Error::internal_error(&e.to_string()))
                             },
                         )?)
                         .expect("database ensures the row is unique");
@@ -3664,11 +3600,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvZoneManifestNonBoot::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| {
                     (row.sled_id, row.non_boot_zpool_id)
                 });
@@ -3706,11 +3639,8 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvMupdateOverrideNonBoot::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| {
                     (row.sled_id, row.non_boot_zpool_id)
                 });
@@ -3742,17 +3672,14 @@ impl DataStore {
                 )
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvClickhouseKeeperMembership::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
                 paginator = p.found_batch(&batch, &|row| row.queried_keeper_id);
                 for membership in batch.into_iter() {
                     memberships.insert(
                         ClickhouseKeeperClusterMembership::try_from(membership)
                             .map_err(|e| {
-                                Error::internal_error(&format!("{e:#}",))
+                                err.bail(Error::internal_error(&format!("{e:#}",)))
                             })?,
                     );
                 }
@@ -3768,11 +3695,8 @@ impl DataStore {
                 dsl::inv_cockroachdb_status
                     .filter(dsl::inv_collection_id.eq(db_id))
                     .select(InvCockroachStatus::as_select())
-                    .load_async(&*conn)
-                    .await
-                    .map_err(|e| {
-                        public_error_from_diesel(e, ErrorHandler::Server)
-                    })?;
+                    .load_async(&conn)
+                    .await?;
 
             status_records
                 .into_iter()
@@ -3784,7 +3708,8 @@ impl DataStore {
                         })?;
                     Ok((node_id, status))
                 })
-                .collect::<Result<BTreeMap<_, _>, Error>>()?
+                .collect::<Result<BTreeMap<_, _>, Error>>()
+                .map_err(|e| err.bail(e))?
         };
 
         // Load TimeSync statuses
@@ -3794,11 +3719,8 @@ impl DataStore {
             let records: Vec<InvNtpTimesync> = dsl::inv_ntp_timesync
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvNtpTimesync::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
 
             records
                 .into_iter()
@@ -3815,11 +3737,8 @@ impl DataStore {
             let records: Vec<InvInternalDns> = dsl::inv_internal_dns
                 .filter(dsl::inv_collection_id.eq(db_id))
                 .select(InvInternalDns::as_select())
-                .load_async(&*conn)
-                .await
-                .map_err(|e| {
-                    public_error_from_diesel(e, ErrorHandler::Server)
-                })?;
+                .load_async(&conn)
+                .await?;
 
             records
                 .into_iter()
@@ -3842,7 +3761,8 @@ impl DataStore {
                         )
                     })
                 })
-                .transpose()?;
+                .transpose()
+                .map_err(|e| err.bail(e))?;
 
             // Convert all the sled config foreign keys back into fully realized
             // `OmicronSledConfig`s. We have to clone these: the same
@@ -3859,7 +3779,8 @@ impl DataStore {
                         )
                     })
                 })
-                .transpose()?;
+                .transpose()
+                .map_err(|e| err.bail(e))?;
             let reconciler_status = s
                 .reconciler_status
                 .to_status(|config_id| {
@@ -3868,7 +3789,7 @@ impl DataStore {
                         .as_ref()
                         .map(|c| c.config.clone())
                 })
-                .map_err(|e| Error::internal_error(&format!("{e:#}")))?;
+                .map_err(|e| err.bail(Error::internal_error(&format!("{e:#}"))))?;
             let last_reconciliation = sled_config_reconcilers
                 .remove(&sled_id)
                 .map(|reconciler| {
@@ -3949,7 +3870,8 @@ impl DataStore {
                         remove_mupdate_override,
                     })
                 })
-                .transpose()?;
+                .transpose()
+                .map_err(|e| err.bail(e))?;
 
             let zone_image_resolver = s
                 .zone_image_resolver
@@ -3959,11 +3881,11 @@ impl DataStore {
                     mupdate_override_non_boot_by_sled_id.remove(&sled_id),
                 )
                 .map_err(|e| {
-                    Error::internal_error(&format!(
+                    err.bail(Error::internal_error(&format!(
                         "failed to create zone image resolver inventory \
                          for sled {sled_id}: {}",
                         InlineErrorChain::new(e.as_ref()),
-                    ))
+                    )))
                 })?;
 
             let sled_agent = nexus_types::inventory::SledAgent {
@@ -4013,11 +3935,12 @@ impl DataStore {
 
         // Check that we consumed all the reconciliation results we found in
         // this collection.
-        bail_unless!(
-            sled_config_reconcilers.is_empty(),
-            "found extra sled config reconcilers: {:?}",
-            sled_config_reconcilers.keys(),
-        );
+        if !sled_config_reconcilers.is_empty() {
+            return Err(err.bail(Error::internal_error(&format!(
+                "found extra sled config reconcilers: {:?}",
+                sled_config_reconcilers.keys()
+            ))));
+        }
         {
             // `sled_boot_partition_details` is a map of maps; we don't prune
             // the outermost map, but they should all be empty.
@@ -4032,64 +3955,80 @@ impl DataStore {
                         }
                     })
                     .collect::<Vec<_>>();
-            bail_unless!(
-                sleds_with_leftover_boot_partitions.is_empty(),
-                "found extra sled boot partition details: {:?}",
-                sleds_with_leftover_boot_partitions,
-            );
+            if !sleds_with_leftover_boot_partitions.is_empty() {
+                return Err(err.bail(Error::internal_error(&format!(
+                    "found extra sled boot partition details: {:?}",
+                    sleds_with_leftover_boot_partitions
+                ))));
+            }
+        }
+        if !last_reconciliation_disk_results.is_empty() {
+            return Err(err.bail(Error::internal_error(&format!(
+                "found extra config reconciliation disk results: {:?}",
+                last_reconciliation_disk_results.keys()
+            ))));
+        }
+        if !last_reconciliation_dataset_results.is_empty() {
+            return Err(err.bail(Error::internal_error(&format!(
+                "found extra config reconciliation dataset results: {:?}",
+                last_reconciliation_dataset_results.keys()
+            ))));
+        }
+        if !last_reconciliation_zone_results.is_empty() {
+            return Err(err.bail(Error::internal_error(&format!(
+                "found extra config reconciliation zone results: {:?}",
+                last_reconciliation_zone_results.keys()
+            ))));
+        }
+        if !zone_manifest_artifacts_by_sled_id.is_empty() {
+            return Err(err.bail(Error::internal_error(&format!(
+                "found extra zone manifest artifacts: {:?}",
+                zone_manifest_artifacts_by_sled_id.keys()
+            ))));
+        }
+        if !zone_manifest_non_boot_by_sled_id.is_empty() {
+            return Err(err.bail(Error::internal_error(&format!(
+                "found extra zone manifest non-boot entries: {:?}",
+                zone_manifest_non_boot_by_sled_id.keys()
+            ))));
+        }
+        if !mupdate_override_non_boot_by_sled_id.is_empty() {
+            return Err(err.bail(Error::internal_error(&format!(
+                "found extra mupdate override non-boot entries: {:?}",
+                mupdate_override_non_boot_by_sled_id.keys()
+            ))));
         }
-        bail_unless!(
-            last_reconciliation_disk_results.is_empty(),
-            "found extra config reconciliation disk results: {:?}",
-            last_reconciliation_disk_results.keys()
-        );
-        bail_unless!(
-            last_reconciliation_dataset_results.is_empty(),
-            "found extra config reconciliation dataset results: {:?}",
-            last_reconciliation_dataset_results.keys()
-        );
-        bail_unless!(
-            last_reconciliation_zone_results.is_empty(),
-            "found extra config reconciliation zone results: {:?}",
-            last_reconciliation_zone_results.keys()
-        );
-        bail_unless!(
-            zone_manifest_artifacts_by_sled_id.is_empty(),
-            "found extra zone manifest artifacts: {:?}",
-            zone_manifest_artifacts_by_sled_id.keys()
-        );
-        bail_unless!(
-            zone_manifest_non_boot_by_sled_id.is_empty(),
-            "found extra zone manifest non-boot entries: {:?}",
-            zone_manifest_non_boot_by_sled_id.keys()
-        );
-        bail_unless!(
-            mupdate_override_non_boot_by_sled_id.is_empty(),
-            "found extra mupdate override non-boot entries: {:?}",
-            mupdate_override_non_boot_by_sled_id.keys()
-        );
 
-        Ok(Collection {
-            id,
-            errors,
-            time_started,
-            time_done,
-            collector,
-            baseboards: baseboards_by_id.values().cloned().collect(),
-            cabooses: cabooses_by_id.values().cloned().collect(),
-            rot_pages: rot_pages_by_id.values().cloned().collect(),
-            sps,
-            host_phase_1_active_slots,
-            host_phase_1_flash_hashes,
-            rots,
-            cabooses_found,
-            rot_pages_found,
-            sled_agents,
-            clickhouse_keeper_cluster_membership,
-            cockroach_status,
-            ntp_timesync,
-            internal_dns_generation_status,
-        })
+                    Ok(Collection {
+                        id,
+                        errors,
+                        time_started,
+                        time_done,
+                        collector,
+                        baseboards: baseboards_by_id.values().cloned().collect(),
+                        cabooses: cabooses_by_id.values().cloned().collect(),
+                        rot_pages: rot_pages_by_id.values().cloned().collect(),
+                        sps,
+                        host_phase_1_active_slots,
+                        host_phase_1_flash_hashes,
+                        rots,
+                        cabooses_found,
+                        rot_pages_found,
+                        sled_agents,
+                        clickhouse_keeper_cluster_membership,
+                        cockroach_status,
+                        ntp_timesync,
+                        internal_dns_generation_status,
+                    })
+                }
+            })
+            .await
+            .map_err(|e| {
+                if let Some(err) = err.take() {
+                    return err;
+                }
+                public_error_from_diesel(e, ErrorHandler::Server)
+            })
     }
 
     pub async fn inventory_collections_latest(