Add comments around snapshot_latest, documenting the set invariant

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,15 @@ 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!
+    ///
+    /// # 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.
     async fn snapshot_latest(&self, id: GlobalId) -> Result<Vec<Row>, StorageError>;
 
     /// Returns a snapshot of the contents of collection `id` at `as_of`.
@@ -1496,6 +1503,8 @@ impl StorageCollections for StorageCollectionsImpl {
                 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
                     })