Merge pull request #34304 from ggevay/frontend-peek-rtr-RtrDropFailure

Solve some frontend peek sequencing Nightly fails: RTR `catalog out of sync` and COPY TO `cannot fail to drop collections`

Author: ggevay

src/adapter/src/coord/peek.rs

@@ -1339,6 +1339,10 @@ impl crate::coord::Coordinator {
             .await
             .map_err(AdapterError::concurrent_dependency_drop_from_dataflow_creation_error)
         {
+            // Clean up the active compute sink that was added above, since the dataflow was never
+            // created. If we don't do this, the sink_id remains in drop_sinks but no collection
+            // exists in the compute controller, causing a panic when the connection terminates.
+            self.remove_active_compute_sink(sink_id).await;
             let _ = tx.send(Err(e));
             return;
         }

src/adapter/src/coord/sequencer/inner.rs

@@ -2512,14 +2512,20 @@ impl Coordinator {
         real_time_recency_timeout: Duration,
     ) -> Result<Option<BoxFuture<'static, Result<Timestamp, StorageError<Timestamp>>>>, AdapterError>
     {
-        let item_ids = source_ids.map(|gid| self.catalog.resolve_item_id(&gid));
+        let item_ids = source_ids
+            .map(|gid| {
+                self.catalog
+                    .try_resolve_item_id(&gid)
+                    .ok_or_else(|| AdapterError::RtrDropFailure(gid.to_string()))
+            })
+            .collect::<Result<Vec<_>, _>>()?;
 
         // Find all dependencies transitively because we need to ensure that
         // RTR queries determine the timestamp from the sources' (i.e.
         // storage objects that ingest data from external systems) remap
         // data. We "cheat" a little bit and filter out any IDs that aren't
         // user objects because we know they are not a RTR source.
-        let mut to_visit = VecDeque::from_iter(item_ids.filter(CatalogItemId::is_user));
+        let mut to_visit = VecDeque::from_iter(item_ids.into_iter().filter(CatalogItemId::is_user));
         // If none of the sources are user objects, we don't need to provide
         // a RTR timestamp.
         if to_visit.is_empty() {