persist: Validate full part tiling in unflatten, not just endpoints

Trace::unflatten's check that a spine batch's reconstructed parts cover its
id range only verified the endpoints: the first part starts at the batch
id's lower and the last ends at its upper. A crafted rollup could encode an
outer id [0, 3) tiled by non-contiguous parts [0, 2) and [1, 3): the endpoint
check passes, and Trace::validate also passes because it inspects the outer
SpineBatch ids and descriptions, never the adjacency of a SpineBatch's parts.

That malformed trace would be accepted at decode and panic later in normal
maintenance instead. fueled_merge_reqs_before_ms emits a FueledMergeReq from
the accepted parts, and Compactor::chunk_runs / compact_all (plus the id-range
merge path in apply_merge_res_checked) call id_range on those ids, whose
adjacency check is an assert_eq!. So the hardening still converted a corrupted
durable rollup into a process panic for this invariant.

Validate the full tiling in unflatten: require adjacent parts to be contiguous
in addition to the existing endpoint check, so non-adjacent parts fail with a
decode error. Adds a decode-rejection regression test for the overlapping-parts
vector alongside the existing partless-spine-batch one.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: def-

src/persist-client/src/internal/encoding.rs

@@ -2436,7 +2436,37 @@ mod tests {
             });
         });
         let err = rollup_decode_err(proto);
-        assert!(err.contains("do not cover"), "{err}");
+        assert!(err.contains("do not tile"), "{err}");
+    }
+
+    /// A spine batch whose parts hit the right endpoints but overlap in the
+    /// middle (id `[0, 3)` tiled by parts `[0, 2)` and `[1, 3)`) passes the
+    /// endpoint check but is not a valid tiling. Such parts reach compaction's
+    /// `id_range`, whose contiguity `assert_eq!` would panic later, so decoding
+    /// must reject them here instead.
+    #[mz_ore::test]
+    fn rollup_proto_with_noncontiguous_spine_parts_is_rejected() {
+        let outer = SpineId(0, 3);
+        let part_ids = [SpineId(0, 2), SpineId(1, 3)];
+        let proto = rollup_proto_with_trace(|trace| {
+            trace.spine_batches.push(ProtoIdSpineBatch {
+                id: Some(outer.into_proto()),
+                batch: Some(ProtoSpineBatch {
+                    level: 0,
+                    desc: Some(u64_desc_proto(0, 3, 0)),
+                    parts: part_ids.iter().map(|id| id.into_proto()).collect(),
+                    descs: vec![],
+                }),
+            });
+            for id in part_ids {
+                trace.hollow_batches.push(ProtoIdHollowBatch {
+                    id: Some(id.into_proto()),
+                    batch: Some(legacy_batch_proto(0, 1, 0)),
+                });
+            }
+        });
+        let err = rollup_decode_err(proto);
+        assert!(err.contains("do not tile"), "{err}");
     }
 
     /// Three spine batches at one level overflow the two-batch layer, which

src/persist-client/src/internal/trace.rs

@@ -451,13 +451,20 @@ impl<T: Timestamp + Lattice> Trace<T> {
                 .map(|(id, desc)| pop_batch(id, desc))
                 .collect::<Result<Vec<_>, _>>()?;
             // A spine batch's parts tile its id range (`SpineBatch::id`
-            // `debug_assert`s this). Real batches always have at least one
-            // part: an empty batch still carries an empty hollow batch.
+            // `debug_assert`s the endpoints). Real batches always have at least
+            // one part: an empty batch still carries an empty hollow batch.
+            // Validate the full tiling, not just the endpoints: downstream
+            // maintenance (`fueled_merge_reqs_before_ms` -> `id_range` in
+            // compaction, `apply_merge_res_checked`) `assert_eq!`s that the
+            // collected part ids are contiguous, so non-adjacent parts that
+            // happen to hit the right endpoints would panic later instead of
+            // here.
             if parts.first().map(|x| x.id.0) != Some(id.0)
                 || parts.last().map(|x| x.id.1) != Some(id.1)
+                || parts.windows(2).any(|w| w[0].id.1 != w[1].id.0)
             {
                 return Err(format!(
-                    "spine batch {id:?} parts do not cover the batch's id range"
+                    "spine batch {id:?} parts do not tile the batch's id range"
                 ));
             }
             let len = parts.iter().map(|p| (*p).batch.len).sum();