Add comments around snapshot_latest, documenting its preconditions

Author: ggevay

src/adapter/src/coord/caught_up.rs

@@ -63,6 +63,9 @@ impl Coordinator {
             .get_entry(&replica_frontier_item_id)
             .latest_global_id();
 
+        // `snapshot_latest` requires that the collection consolidates to a
+        // set. `mz_cluster_replica_frontiers` is a controller-managed builtin
+        // written with ±1 diffs, so it satisfies that invariant.
         let live_frontiers = self
             .controller
             .storage_collections

src/storage-client/src/storage_collections.rs

@@ -163,8 +163,22 @@ pub trait StorageCollections: Debug + Sync {
         as_of: Timestamp,
     ) -> BoxFuture<'static, Result<Vec<(Row, StorageDiff)>, StorageError>>;
 
-    /// Returns a snapshot of the contents of collection `id` at the largest
-    /// readable `as_of`.
+    /// Returns a snapshot of the contents of collection `id` at the largest readable `as_of`.
+    /// The collection must consolidate to a set, i.e., the multiplicity of every row must be 1!
+    ///
+    /// # Errors
+    ///
+    /// - Returns `StorageError::InvalidUsage` if the collection is closed.
+    /// - Propagates the error if the underlying `snapshot` call errors.
+    ///
+    /// # Panics
+    ///
+    /// - Panics if the collection does not consolidate to a set at that `as_of`
+    ///   (i.e., if any row survives with a multiplicity other than `+1`). Only
+    ///   safe to call on collections whose producer guarantees set semantics;
+    ///   not safe on arbitrary user collections.
+    /// - Panics if the collection's upper is `0`, i.e., if it doesn't have data at
+    ///   any timestamp.
     async fn snapshot_latest(&self, id: GlobalId) -> Result<Vec<Row>, StorageError>;
 
     /// Returns a snapshot of the contents of collection `id` at `as_of`.
@@ -1492,10 +1506,12 @@ impl StorageCollections for StorageCollectionsImpl {
             Some(f) if f > &Timestamp::MIN => {
                 let as_of = f.step_back().unwrap();
 
-                let snapshot = self.snapshot(id, as_of, &self.txns_read).await.unwrap();
+                let snapshot = self.snapshot(id, as_of, &self.txns_read).await?;
                 snapshot
                     .into_iter()
                     .map(|(row, diff)| {
+                        // See the trait doc: `snapshot_latest` is only meant for collections that
+                        // consolidate to a set.
                         assert_eq!(diff, 1, "snapshot doesn't accumulate to set");
                         row
                     })