sui_rpc_store/schema/
package_versions.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! `(original_package_id, version)` → `PackageVersionInfo`.
5//!
6//! Lists every published version of a Move package: the storage id
7//! under which each version lives, and the checkpoint at which the
8//! version was published. The publish checkpoint lets a
9//! checkpoint-bounded read resolve the latest version of a package
10//! as of a given checkpoint — see
11//! [`get_package_at_checkpoint`](super::RpcStoreSchema::get_package_at_checkpoint).
12//!
13//! Rows written by the live-set restore at the anchor checkpoint
14//! leave the publish checkpoint unset (a restore floor): those
15//! versions predate the available window, so a checkpoint-bounded
16//! read treats them as having always existed.
17
18use bytes::Buf;
19use bytes::BufMut;
20use sui_consistent_store::Decode;
21use sui_consistent_store::Encode;
22use sui_consistent_store::Iter;
23use sui_consistent_store::Protobuf;
24use sui_consistent_store::error::DecodeError;
25use sui_consistent_store::error::EncodeError;
26use sui_consistent_store::error::Error;
27use sui_consistent_store::reader::Reader;
28use sui_types::base_types::ObjectID;
29
30use crate::proto::PackageVersionInfo;
31
32pub const NAME: &str = "package_versions";
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
35pub struct Key {
36    pub original_id: ObjectID,
37    pub version: u64,
38}
39
40pub type Value = Protobuf<PackageVersionInfo>;
41
42impl Encode for Key {
43    fn encode_into<B: BufMut>(&self, buf: &mut B) -> Result<(), EncodeError> {
44        buf.put_slice(self.original_id.as_ref());
45        buf.put_slice(&self.version.to_be_bytes());
46        Ok(())
47    }
48}
49
50impl Decode for Key {
51    fn decode<B: Buf>(buf: &mut B) -> Result<Self, DecodeError> {
52        if buf.remaining() != ObjectID::LENGTH + 8 {
53            return Err(DecodeError::msg(format!(
54                "expected {} bytes for {NAME} Key, got {}",
55                ObjectID::LENGTH + 8,
56                buf.remaining(),
57            )));
58        }
59        let mut id = [0u8; ObjectID::LENGTH];
60        buf.copy_to_slice(&mut id);
61        let version = buf.get_u64();
62        Ok(Key {
63            original_id: ObjectID::new(id),
64            version,
65        })
66    }
67}
68
69pub fn options(resolver: &sui_consistent_store::CfOptionsResolver) -> rocksdb::Options {
70    resolver.options(NAME)
71}
72
73/// Build the `(Key, Value)` pair recording that version `version`
74/// of the package originally published at `original_id` lives at
75/// the on-chain storage id `storage_id`, published in `checkpoint`.
76/// Written by tip indexing and the post-restore backfill.
77pub fn store(
78    original_id: ObjectID,
79    version: u64,
80    storage_id: ObjectID,
81    checkpoint: u64,
82) -> (Key, Value) {
83    (
84        Key {
85            original_id,
86            version,
87        },
88        Protobuf(PackageVersionInfo {
89            storage_id: storage_id.to_vec().into(),
90            checkpoint: Some(checkpoint),
91        }),
92    )
93}
94
95/// Like [`store`], but for rows written by the live-set restore at
96/// the anchor checkpoint: the publish checkpoint is left unset (a
97/// restore floor), marking a version that was published before the
98/// available window. A checkpoint-bounded read treats such a version
99/// as having always existed.
100pub fn store_restored(original_id: ObjectID, version: u64, storage_id: ObjectID) -> (Key, Value) {
101    (
102        Key {
103            original_id,
104            version,
105        },
106        Protobuf(PackageVersionInfo {
107            storage_id: storage_id.to_vec().into(),
108            checkpoint: None,
109        }),
110    )
111}
112
113/// Prefix encoder for "all versions of the package originally
114/// published at `original_id`". Encodes as the 32 raw id bytes —
115/// exactly the leading bytes of every `Key` whose `original_id`
116/// matches.
117pub struct OriginalIdPrefix(pub ObjectID);
118
119impl Encode for OriginalIdPrefix {
120    fn encode_into<B: BufMut>(&self, buf: &mut B) -> Result<(), EncodeError> {
121        buf.put_slice(self.0.as_ref());
122        Ok(())
123    }
124}
125
126/// Decode a `storage_id` field (the 32 raw `ObjectID` bytes) from a
127/// stored `PackageVersionInfo`.
128fn decode_storage_id(bytes: &[u8]) -> Result<ObjectID, DecodeError> {
129    let array: [u8; ObjectID::LENGTH] = bytes.try_into().map_err(|_| {
130        DecodeError::msg(format!(
131            "expected {} bytes for storage_id, got {}",
132            ObjectID::LENGTH,
133            bytes.len(),
134        ))
135    })?;
136    Ok(ObjectID::new(array))
137}
138
139impl<R: Reader> super::RpcStoreSchema<R> {
140    /// Look up the on-chain storage id that holds a specific
141    /// version of a package.
142    pub fn get_package_storage_id(
143        &self,
144        original_id: ObjectID,
145        version: u64,
146    ) -> Result<Option<ObjectID>, Error> {
147        let Some(stored) = self.package_versions.get(&Key {
148            original_id,
149            version,
150        })?
151        else {
152            return Ok(None);
153        };
154        Ok(Some(decode_storage_id(&stored.into_inner().storage_id)?))
155    }
156
157    /// Iterate every version of the package originally published
158    /// at `original_id`, in ascending version order.
159    pub fn iter_package_versions(
160        &self,
161        original_id: ObjectID,
162    ) -> Result<Iter<'_, Key, Value>, Error> {
163        self.package_versions
164            .iter_prefix(&OriginalIdPrefix(original_id))
165    }
166
167    /// Resolve the latest version of the package originally
168    /// published at `original_id` that existed as of `checkpoint`,
169    /// returning its `(version, storage_id)`.
170    ///
171    /// Versions are walked from highest to lowest, and the first
172    /// version whose publish checkpoint is at or before `checkpoint`
173    /// wins. Publish checkpoints are non-decreasing in version,
174    /// because every writer stamps the checkpoint being indexed, so
175    /// the scan's cost is bounded by the number of versions published
176    /// after `checkpoint`, not by the package's lineage length.
177    /// Restore-floor rows (no recorded publish checkpoint) count as
178    /// having always existed, since they predate the available window.
179    ///
180    /// Returns `Ok(None)` when no version of the package existed as
181    /// of `checkpoint` (it was first published later, or the package
182    /// is unknown to this store).
183    pub fn get_package_at_checkpoint(
184        &self,
185        original_id: ObjectID,
186        checkpoint: u64,
187    ) -> Result<Option<(u64, ObjectID)>, Error> {
188        for row in self
189            .package_versions
190            .iter_rev_prefix(&OriginalIdPrefix(original_id))?
191        {
192            let (key, value) = row?;
193            let info = value.into_inner();
194            if info
195                .checkpoint
196                .is_none_or(|published| published <= checkpoint)
197            {
198                return Ok(Some((key.version, decode_storage_id(&info.storage_id)?)));
199            }
200        }
201        Ok(None)
202    }
203}
204
205#[cfg(test)]
206mod tests {
207    use sui_consistent_store::Db;
208    use sui_consistent_store::DbOptions;
209
210    use super::*;
211    use crate::RpcStoreSchema;
212
213    fn fresh_db() -> (tempfile::TempDir, sui_consistent_store::Db, RpcStoreSchema) {
214        let dir = tempfile::tempdir().unwrap();
215        let (db, schema) = Db::open::<RpcStoreSchema>(dir.path(), DbOptions::default()).unwrap();
216        (dir, db, schema)
217    }
218
219    #[test]
220    fn get_returns_none_for_unknown_version() {
221        let (_dir, _db, schema) = fresh_db();
222        let original = ObjectID::random();
223        assert!(
224            schema
225                .get_package_storage_id(original, 1)
226                .unwrap()
227                .is_none()
228        );
229    }
230
231    #[test]
232    fn store_then_get_round_trips() {
233        let (_dir, db, schema) = fresh_db();
234        let original = ObjectID::random();
235        let storage = ObjectID::random();
236
237        let (k, v) = store(original, 3, storage, 100);
238        let mut batch = db.batch();
239        batch.put(&schema.package_versions, &k, &v).unwrap();
240        batch.commit().unwrap();
241
242        assert_eq!(
243            schema.get_package_storage_id(original, 3).unwrap(),
244            Some(storage),
245        );
246    }
247
248    #[test]
249    fn iter_walks_versions_for_one_package() {
250        let (_dir, db, schema) = fresh_db();
251        let original = ObjectID::random();
252        let other = ObjectID::random();
253
254        let mut batch = db.batch();
255        // Three versions of the target package.
256        for version in [1u64, 2, 5] {
257            let (k, v) = store(original, version, ObjectID::random(), version);
258            batch.put(&schema.package_versions, &k, &v).unwrap();
259        }
260        // An unrelated package — must not show up in the iter.
261        let (k, v) = store(other, 1, ObjectID::random(), 1);
262        batch.put(&schema.package_versions, &k, &v).unwrap();
263        batch.commit().unwrap();
264
265        let versions: Vec<u64> = schema
266            .iter_package_versions(original)
267            .unwrap()
268            .map(|res| res.unwrap().0.version)
269            .collect();
270        assert_eq!(versions, vec![1, 2, 5]);
271    }
272
273    #[test]
274    fn get_package_at_checkpoint_resolves_latest_in_window() {
275        let (_dir, db, schema) = fresh_db();
276        let original = ObjectID::random();
277        let (s1, s2, s3) = (ObjectID::random(), ObjectID::random(), ObjectID::random());
278
279        let mut batch = db.batch();
280        // v1 is a restore floor (published before the available
281        // window); v2 and v3 were published at checkpoints 10 and 20.
282        let (k, v) = store_restored(original, 1, s1);
283        batch.put(&schema.package_versions, &k, &v).unwrap();
284        let (k, v) = store(original, 2, s2, 10);
285        batch.put(&schema.package_versions, &k, &v).unwrap();
286        let (k, v) = store(original, 3, s3, 20);
287        batch.put(&schema.package_versions, &k, &v).unwrap();
288        batch.commit().unwrap();
289
290        // Before any real publish, only the restore-floor v1 exists.
291        assert_eq!(
292            schema.get_package_at_checkpoint(original, 5).unwrap(),
293            Some((1, s1)),
294        );
295        // v2's publish checkpoint, then the gap before v3.
296        assert_eq!(
297            schema.get_package_at_checkpoint(original, 10).unwrap(),
298            Some((2, s2)),
299        );
300        assert_eq!(
301            schema.get_package_at_checkpoint(original, 15).unwrap(),
302            Some((2, s2)),
303        );
304        // v3's publish checkpoint and beyond.
305        assert_eq!(
306            schema.get_package_at_checkpoint(original, 20).unwrap(),
307            Some((3, s3)),
308        );
309        assert_eq!(
310            schema.get_package_at_checkpoint(original, 9_999).unwrap(),
311            Some((3, s3)),
312        );
313    }
314
315    #[test]
316    fn get_package_at_checkpoint_returns_none_before_first_publish() {
317        let (_dir, db, schema) = fresh_db();
318        let original = ObjectID::random();
319
320        let mut batch = db.batch();
321        let (k, v) = store(original, 1, ObjectID::random(), 50);
322        batch.put(&schema.package_versions, &k, &v).unwrap();
323        batch.commit().unwrap();
324
325        // The package's first version was published at checkpoint 50,
326        // so it does not exist as of an earlier checkpoint.
327        assert_eq!(
328            schema.get_package_at_checkpoint(original, 49).unwrap(),
329            None,
330        );
331        assert!(
332            schema
333                .get_package_at_checkpoint(original, 50)
334                .unwrap()
335                .is_some()
336        );
337        // An unknown package resolves to nothing.
338        assert_eq!(
339            schema
340                .get_package_at_checkpoint(ObjectID::random(), 100)
341                .unwrap(),
342            None,
343        );
344    }
345}
sui_rpc_store/schema/package_versions.rs

sui_rpc_store/schema/
package_versions.rs
