fix(batch): refuse an empty path on every reference hop instead of panicking

CHANGELOG.md

@@ -90,6 +90,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   an absent path. `AxisKeys::empty_for_axis` is new. (#965)
 
 ### Fixed
+- A batch whose reference chain reaches another pending reference that
+  resolves to an empty path no longer panics. `apply_batch` refused an empty
+  path only for the reference op it was processing. Keys in a path run in
+  byte order, so an earlier op could follow into a later pending reference
+  (an `AbsolutePathReference(vec![])`, say, or a trusted `RefreshReference`
+  carrying one) before that op's own check ran. The next hop then reached
+  `process_reference(&[])`, where `split_last().expect(..)` panicked. Every
+  hop now refuses an empty path with `InvalidBatchOperation`, so the batch
+  fails and writes nothing. The fix is not version-gated. A batch that took
+  this path never committed at any grove version, so no stored state or
+  replayed block can change, and a batch that succeeded before resolves the
+  same way now.
 - A V1 proof whose limit runs out before its query does verifies at its own
   limit. The walk hides what it has not reached, but still passes back up
   through the ancestors of its last result, and one of them can be a range

grovedb/src/batch/mod.rs

@@ -2291,7 +2291,7 @@ where
         let mut cost = OperationCost::default();
         let (key, reference_path) = qualified_path
             .split_last()
-            .expect("path validated non-empty above");
+            .expect("follow_reference_get_value_hash refuses an empty path");
 
         // Fast path: `recursions_allowed == 1` means the user-declared
         // `max_reference_hop` budget allows exactly one more hop. Under
@@ -2724,6 +2724,18 @@ where
         if recursions_allowed == 0 {
             return Err(Error::ReferenceLimit).wrap_with_cost(cost);
         }
+        // Every hop enters here, so this is the one place an empty path is
+        // refused. The apply arms check only the reference they are
+        // processing; a hop can still land on another pending reference
+        // whose path resolves to `[]` before that op's own check runs.
+        // Everything below (`process_reference` and the flagged-item arm)
+        // splits the path and relies on this check.
+        if qualified_path.is_empty() {
+            return Err(Error::InvalidBatchOperation(
+                "reference chain resolves to an empty path",
+            ))
+            .wrap_with_cost(cost);
+        }
         let path_vec = qualified_path.to_vec();
         if !visited.insert(path_vec) {
             return Err(Error::CyclicReference).wrap_with_cost(cost);
@@ -2811,7 +2823,7 @@ where
                                 // it can be unmerged, let's get the value on disk
                                 let (key, reference_path) = qualified_path
                                     .split_last()
-                                    .expect("path validated non-empty above");
+                                    .expect("empty path refused at the top of this function");
                                 let serialized_element_result = cost_return_on_error!(
                                     &mut cost,
                                     self.get_and_deserialize_referenced_element(

grovedb/src/tests/batch_unit_tests.rs

@@ -1668,4 +1668,125 @@ mod tests {
             "ref chain through refresh to tree should fail"
         );
     }
+
+    // ===================================================================
+    // Group 16: follow_reference_get_value_hash — a later hop resolves to
+    //           an empty path
+    // ===================================================================
+    //
+    // The apply arms refuse an empty path only for the reference they are
+    // processing. Keys in a path run in byte order, so `a` follows its
+    // sibling `b` before `b`'s own check runs. `b`'s pending element
+    // resolves to `[]`, and the next hop used to reach
+    // `process_reference(&[])`, whose `split_last().expect(..)` panicked.
+
+    /// Every reference path type that resolves to `[]` from a reference
+    /// stored directly under `[TEST_LEAF]`.
+    fn empty_resolving_reference_paths() -> Vec<ReferencePathType> {
+        vec![
+            ReferencePathType::AbsolutePathReference(vec![]),
+            ReferencePathType::UpstreamRootHeightReference(0, vec![]),
+            ReferencePathType::UpstreamFromElementHeightReference(1, vec![]),
+        ]
+    }
+
+    /// Applies `ops` and asserts the batch is refused with the empty-hop
+    /// error and leaves the root hash untouched.
+    fn assert_empty_hop_refused(
+        db: &crate::tests::TempGroveDb,
+        ops: Vec<QualifiedGroveDbOp>,
+        grove_version: &GroveVersion,
+        case: &str,
+    ) {
+        let root_before = db
+            .root_hash(None, grove_version)
+            .unwrap()
+            .expect("root hash before");
+        let result = db.apply_batch(ops, None, None, grove_version).unwrap();
+        assert!(
+            matches!(
+                result,
+                Err(Error::InvalidBatchOperation(msg)) if msg.contains("empty path")
+            ),
+            "{case}: expected the empty-hop refusal, got {result:?}"
+        );
+        let root_after = db
+            .root_hash(None, grove_version)
+            .unwrap()
+            .expect("root hash after");
+        assert_eq!(root_before, root_after, "{case}: refused batch wrote");
+    }
+
+    #[test]
+    fn test_batch_ref_hop_to_pending_empty_reference_errors() {
+        for grove_version in grovedb_version::version::GROVE_VERSIONS {
+            for empty_path in empty_resolving_reference_paths() {
+                // InsertOrReplace and InsertWithKnownToNotAlreadyExist reach
+                // the empty hop through different match arms.
+                for insert_only in [false, true] {
+                    let case = format!(
+                        "v{} {empty_path:?} insert_only={insert_only}",
+                        grove_version.protocol_version
+                    );
+                    let db = make_test_grovedb(grove_version);
+                    let target = Element::new_reference(empty_path.clone());
+                    let target_op = if insert_only {
+                        QualifiedGroveDbOp::insert_only_known_to_not_already_exist_op(
+                            vec![TEST_LEAF.to_vec()],
+                            b"b".to_vec(),
+                            target,
+                        )
+                    } else {
+                        QualifiedGroveDbOp::insert_or_replace_op(
+                            vec![TEST_LEAF.to_vec()],
+                            b"b".to_vec(),
+                            target,
+                        )
+                    };
+                    let ops = vec![
+                        QualifiedGroveDbOp::insert_or_replace_op(
+                            vec![TEST_LEAF.to_vec()],
+                            b"a".to_vec(),
+                            Element::new_reference(ReferencePathType::SiblingReference(
+                                b"b".to_vec(),
+                            )),
+                        ),
+                        target_op,
+                    ];
+                    assert_empty_hop_refused(&db, ops, grove_version, &case);
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn test_batch_ref_hop_to_trusted_refresh_with_empty_path_errors() {
+        for grove_version in grovedb_version::version::GROVE_VERSIONS {
+            for empty_path in empty_resolving_reference_paths() {
+                let case = format!("v{} {empty_path:?}", grove_version.protocol_version);
+                let db = make_test_grovedb(grove_version);
+                // A trusted refresh carries its new path without a disk
+                // read, so `b` need not exist. `a` follows into the refresh
+                // and resolves the next hop through `process_reference`'s
+                // in-batch branch.
+                let ops = vec![
+                    QualifiedGroveDbOp::insert_or_replace_op(
+                        vec![TEST_LEAF.to_vec()],
+                        b"a".to_vec(),
+                        Element::new_reference(ReferencePathType::SiblingReference(b"b".to_vec())),
+                    ),
+                    QualifiedGroveDbOp::refresh_reference_op(
+                        vec![TEST_LEAF.to_vec()],
+                        b"b".to_vec(),
+                        empty_path,
+                        None,
+                        None,
+                        /* non_counted = */ false,
+                        true, // trust_refresh_reference
+                    ),
+                ];
+                assert_empty_hop_refused(&db, ops, grove_version, &case);
+            }
+        }
+    }
 }