[sui-fork] Fix child object reads to support bounded versions (#26966)

## Description 

This PR adds support for reading child object bounded versions. Without
this, a bunch of transactions do not work correctly and the backend
complains with an invariant violation, because we're passing the latest
object rather than the correct bounded version of that child object.

## Test plan 

Several tests added that cover loading the object at the right version.

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] gRPC:
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] Indexing Framework:

Author: stefan-mysten

crates/sui-fork/src/filesystem.rs

@@ -141,6 +141,16 @@ struct ObjectLatestMetadata {
     state: ObjectLatestState,
 }
 
+/// Result of a bounded local object lookup.
+pub(crate) enum BoundedObjectLookup {
+    /// A live object version at or below the requested bound was found locally.
+    Hit(Object),
+    /// A local tombstone at or below the requested bound proves the object is absent.
+    NegativeHit,
+    /// Local storage has no authoritative answer for the requested bound.
+    Miss,
+}
+
 impl ObjectLatestMetadata {
     /// Construct numeric-only latest metadata for a live object version.
     fn live(version: u64) -> Self {
@@ -445,6 +455,40 @@ impl FilesystemStore {
         self.read_bcs_file(&version_file).map(Some)
     }
 
+    /// Get the highest locally persisted object version at or below `version_bound`.
+    pub(crate) fn get_object_lt_or_eq_version(
+        &self,
+        object_id: &ObjectID,
+        version_bound: u64,
+    ) -> anyhow::Result<BoundedObjectLookup> {
+        let object_dir = self.objects_dir().join(object_id.to_string());
+        if !object_dir.exists() {
+            return Ok(BoundedObjectLookup::Miss);
+        }
+
+        if let Some(latest) = self.read_object_latest_metadata_if_exists(&object_dir)?
+            && latest.version <= version_bound
+        {
+            if latest.state.is_removed() {
+                return Ok(BoundedObjectLookup::NegativeHit);
+            }
+
+            let version_file = object_dir.join(latest.version.to_string());
+            return self
+                .read_bcs_file(&version_file)
+                .map(BoundedObjectLookup::Hit);
+        }
+
+        let Some(version) =
+            self.find_highest_object_version_lt_or_eq(&object_dir, version_bound)?
+        else {
+            return Ok(BoundedObjectLookup::Miss);
+        };
+
+        self.read_bcs_file(&object_dir.join(version.to_string()))
+            .map(BoundedObjectLookup::Hit)
+    }
+
     /// Write an object fetched from remote/cache paths. Existing deleted or wrapped current-state
     /// metadata is preserved so remote reads cannot resurrect local removals.
     pub(crate) fn write_object(&self, object: &Object) -> anyhow::Result<()> {
@@ -603,6 +647,54 @@ impl FilesystemStore {
             .with_context(|| format!("Failed to write latest file: {}", latest_file.display()))
     }
 
+    /// Scan the object directory for the highest version at or below the bound, ignoring
+    /// non-version files.
+    fn find_highest_object_version_lt_or_eq(
+        &self,
+        object_dir: &Path,
+        version_bound: u64,
+    ) -> anyhow::Result<Option<u64>> {
+        let mut best = None;
+        for entry in fs::read_dir(object_dir)
+            .with_context(|| format!("Failed to read object directory: {}", object_dir.display()))?
+        {
+            let entry = entry.with_context(|| {
+                format!(
+                    "Failed to read object directory entry: {}",
+                    object_dir.display()
+                )
+            })?;
+
+            if !entry
+                .file_type()
+                .with_context(|| {
+                    format!(
+                        "Failed to read object file type: {}",
+                        entry.path().display()
+                    )
+                })?
+                .is_file()
+            {
+                continue;
+            }
+
+            let file_name = entry.file_name();
+            let Some(name) = file_name.to_str() else {
+                continue;
+            };
+            let Ok(version) = name.parse::<u64>() else {
+                continue;
+            };
+
+            // Compare the current best version with the candidate version, and update it if the
+            // candidate version is higher but still within bound.
+            if version <= version_bound && best.is_none_or(|best| version > best) {
+                best = Some(version);
+            }
+        }
+        Ok(best)
+    }
+
     /// Read the owned-object index. Missing index files represent an empty index.
     pub(crate) fn get_owned_object_entries(&self) -> anyhow::Result<Vec<OwnedObjectEntry>> {
         let path = self.owned_objects_index_path();

crates/sui-fork/src/store.rs

@@ -74,6 +74,7 @@ use crate::ObjectRead;
 use crate::TransactionInfo;
 use crate::TransactionRead;
 use crate::VersionQuery;
+use crate::filesystem::BoundedObjectLookup;
 use crate::filesystem::FilesystemStore;
 use crate::filesystem::ObjectLatestState;
 use crate::filesystem::OwnedObjectEntry;
@@ -342,6 +343,36 @@ impl DataStore {
         Ok(object)
     }
 
+    /// Get the latest object version at or below the given root version.
+    fn get_object_lt_or_eq_version(
+        &self,
+        object_id: &ObjectID,
+        version_bound: SequenceNumber,
+    ) -> anyhow::Result<Option<Object>> {
+        match self
+            .inner
+            .local
+            .get_object_lt_or_eq_version(object_id, version_bound.value())?
+        {
+            BoundedObjectLookup::Hit(object) => return Ok(Some(object)),
+            BoundedObjectLookup::NegativeHit => return Ok(None),
+            BoundedObjectLookup::Miss => {}
+        }
+
+        let mut objects = self.inner.gql.get_objects(&[ObjectKey {
+            object_id: *object_id,
+            version_query: VersionQuery::RootVersion(version_bound.value()),
+        }])?;
+        let object = objects.pop().flatten().map(|(object, _)| object);
+
+        if let Some(ref object) = object {
+            let _local_snapshot_guard = self.write_local_snapshot()?;
+            self.inner.local.write_object(object)?;
+        }
+
+        Ok(object)
+    }
+
     /// Local-first lookup for the latest known version of an object. Falls back to a remote
     /// `AtCheckpoint(forked_at_checkpoint)` query and caches the result on disk.
     fn get_latest_object(&self, object_id: &ObjectID) -> anyhow::Result<Option<Object>> {
@@ -847,7 +878,11 @@ impl ChildObjectResolver for DataStore {
         child: &ObjectID,
         child_version_upper_bound: SequenceNumber,
     ) -> SuiResult<Option<Object>> {
-        let child_object = match self.get_object(child).ok().flatten() {
+        let child_object = match self
+            .get_object_lt_or_eq_version(child, child_version_upper_bound)
+            .ok()
+            .flatten()
+        {
             None => return Ok(None),
             Some(obj) => obj,
         };
@@ -861,13 +896,6 @@ impl ChildObjectResolver for DataStore {
             .into());
         }
 
-        if child_object.version() > child_version_upper_bound {
-            return Err(sui_types::error::SuiErrorKind::UnsupportedFeatureError {
-                error: "DataStore::read_child_object does not yet support bounded reads".to_owned(),
-            }
-            .into());
-        }
-
         Ok(Some(child_object))
     }
 

crates/sui-fork/src/tests/filesystem.rs

@@ -222,6 +222,72 @@ fn test_get_object_at_version_returns_none_for_unknown_version() {
     assert!(result.is_none());
 }
 
+#[test]
+fn test_object_lt_or_eq_returns_highest_version_within_bound() {
+    let (_dir, store) = test_store();
+    let id = ObjectID::random();
+    let v1 = make_object(id, 1);
+    let v3 = make_object(id, 3);
+    let v5 = make_object(id, 5);
+
+    store.write_object(&v1).unwrap();
+    store.write_object(&v3).unwrap();
+    store.write_object(&v5).unwrap();
+
+    let bounded = store.get_object_lt_or_eq_version(&id, 4).unwrap();
+    let BoundedObjectLookup::Hit(object) = bounded else {
+        panic!("expected bounded object hit");
+    };
+    assert_eq!(object, v3);
+
+    let exact = store.get_object_lt_or_eq_version(&id, 5).unwrap();
+    let BoundedObjectLookup::Hit(object) = exact else {
+        panic!("expected exact-bound object hit");
+    };
+    assert_eq!(object, v5);
+}
+
+#[test]
+fn test_object_lt_or_eq_misses_when_no_local_version_within_bound() {
+    let (_dir, store) = test_store();
+    let id = ObjectID::random();
+    let object = make_object(id, 5);
+
+    store.write_object(&object).unwrap();
+
+    assert!(matches!(
+        store.get_object_lt_or_eq_version(&id, 4).unwrap(),
+        BoundedObjectLookup::Miss,
+    ));
+}
+
+#[test]
+fn test_removed_latest_at_or_below_bound_is_negative_hit() {
+    let (_dir, store) = test_store();
+    let id = ObjectID::random();
+    let object = make_object(id, 3);
+
+    store.write_object(&object).unwrap();
+    store
+        .mark_object_as_wrapped(id, SequenceNumber::from_u64(5))
+        .unwrap();
+
+    assert!(matches!(
+        store.get_object_lt_or_eq_version(&id, 5).unwrap(),
+        BoundedObjectLookup::NegativeHit,
+    ));
+    assert!(matches!(
+        store.get_object_lt_or_eq_version(&id, 6).unwrap(),
+        BoundedObjectLookup::NegativeHit,
+    ));
+
+    let before_wrap = store.get_object_lt_or_eq_version(&id, 4).unwrap();
+    let BoundedObjectLookup::Hit(before_wrap) = before_wrap else {
+        panic!("expected object version before wrap");
+    };
+    assert_eq!(before_wrap, object);
+}
+
 #[test]
 fn test_latest_tracks_highest_written_version() {
     let (_dir, store) = test_store();

crates/sui-fork/src/tests/store_execution.rs

@@ -136,6 +136,28 @@ fn object_at_checkpoint_response(object: &Object) -> serde_json::Value {
     })
 }
 
+fn objects_response(objects: &[Option<&Object>]) -> serde_json::Value {
+    serde_json::json!({
+        "data": {
+            "multiGetObjects": objects
+                .iter()
+                .map(|object| {
+                    object.map(|object| {
+                        serde_json::json!({
+                            "address": object.id().to_string(),
+                            "version": object.version().value(),
+                            "objectBcs": FastCryptoBase64::from_bytes(
+                                &bcs::to_bytes(object).expect("object should serialize"),
+                            )
+                            .encoded(),
+                        })
+                    })
+                })
+                .collect::<Vec<_>>(),
+        }
+    })
+}
+
 async fn mock_seed_object(server: &MockServer, checkpoint: u64, object: &Object) {
     Mock::given(method("POST"))
         .and(path("/"))
@@ -538,6 +560,100 @@ fn test_missing_owned_index_after_local_checkpoint_advancement_fails_closed() {
     );
 }
 
+#[test]
+fn test_read_child_object_uses_highest_local_version_within_bound() {
+    let (_temp, store) = test_data_store();
+    let parent = ObjectID::random();
+    let child_id = ObjectID::random();
+    let child_v5 = make_gas_object(child_id, 5, Owner::ObjectOwner(parent.into()));
+    let child_v7 = make_gas_object(child_id, 7, Owner::ObjectOwner(parent.into()));
+
+    store.local().write_object(&child_v5).unwrap();
+    store.local().write_object(&child_v7).unwrap();
+
+    let child = sui_types::storage::ChildObjectResolver::read_child_object(
+        &store,
+        &parent,
+        &child_id,
+        SequenceNumber::from_u64(6),
+    )
+    .expect("bounded child read should not error")
+    .expect("child object should be found");
+
+    assert_eq!(child, child_v5);
+}
+
+#[tokio::test]
+async fn test_read_child_object_falls_back_to_remote_root_version() {
+    let temp = tempfile::tempdir().expect("failed to create tempdir");
+    let checkpoint = 42;
+    let parent = ObjectID::random();
+    let child_id = ObjectID::random();
+    let child = make_gas_object(child_id, 5, Owner::ObjectOwner(parent.into()));
+
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path("/"))
+        .and(body_partial_json(serde_json::json!({
+            "variables": {
+                "keys": [
+                    {
+                        "address": child_id.to_string(),
+                        "rootVersion": 6,
+                    },
+                ],
+            }
+        })))
+        .respond_with(ResponseTemplate::new(200).set_body_json(objects_response(&[Some(&child)])))
+        .mount(&server)
+        .await;
+
+    let store =
+        DataStore::new_for_testing_with_remote(temp.path().to_path_buf(), server.uri(), checkpoint);
+    let read = sui_types::storage::ChildObjectResolver::read_child_object(
+        &store,
+        &parent,
+        &child_id,
+        SequenceNumber::from_u64(6),
+    )
+    .expect("remote bounded child read should not error")
+    .expect("child object should be found");
+
+    assert_eq!(read, child);
+    assert_eq!(
+        store.local().get_object_at_version(&child_id, 5).unwrap(),
+        Some(child),
+    );
+}
+
+#[test]
+fn test_read_child_object_rejects_wrong_owner_after_bounded_lookup() {
+    let (_temp, store) = test_data_store();
+    let parent = ObjectID::random();
+    let other_parent = ObjectID::random();
+    let child_id = ObjectID::random();
+    let child = make_gas_object(child_id, 5, Owner::ObjectOwner(other_parent.into()));
+
+    store.local().write_object(&child).unwrap();
+
+    let err = sui_types::storage::ChildObjectResolver::read_child_object(
+        &store,
+        &parent,
+        &child_id,
+        SequenceNumber::from_u64(6),
+    )
+    .expect_err("wrong child owner should error");
+
+    assert!(matches!(
+        err.as_inner(),
+        sui_types::error::SuiErrorKind::InvalidChildObjectAccess {
+            object,
+            given_parent,
+            actual_owner,
+        } if *object == child_id && *given_parent == parent && actual_owner == &child.owner
+    ));
+}
+
 #[test]
 fn test_local_deletion_removes_current_object_but_preserves_historical_lookup() {
     let (_temp, mut store) = test_data_store();