sui_rpc_store/schema/
pruning_watermark.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! `()` → `PruningWatermarks`.
5//!
6//! Singleton row that holds the lowest still-available `tx_seq`
7//! and `checkpoint_seq`. It is the durable authority for the
8//! bitmap CFs' compaction filters and feeds `available_range`
9//! requests.
10//!
11//! Each open RPC-store schema owns an `Arc<AtomicU64>` whose clones
12//! are captured by that database's bitmap compaction filters. Schema
13//! construction loads the persisted `tx_seq` floor into the atomic.
14//! Pruning and restore paths commit the singleton row before
15//! publishing its value to the matching database-local atomic.
16
17use std::sync::atomic::Ordering;
18
19use sui_consistent_store::Protobuf;
20use sui_consistent_store::error::Error;
21use sui_consistent_store::reader::Reader;
22
23use crate::proto::PruningWatermarks;
24use crate::schema::primitives::UnitKey;
25
26pub const NAME: &str = "pruning_watermark";
27
28pub type Key = UnitKey;
29pub type Value = Protobuf<PruningWatermarks>;
30
31pub fn options(resolver: &sui_consistent_store::CfOptionsResolver) -> rocksdb::Options {
32    resolver.options(NAME)
33}
34
35/// Caller-facing view of the pruning watermarks.
36///
37/// `tx_seq_lo` is the lowest `tx_seq` whose downstream rows
38/// (`tx_metadata_by_seq`, `transactions`, `effects`, `events`,
39/// and the bitmap CFs) are still present. Everything strictly
40/// below it has been pruned.
41///
42/// `checkpoint_lo` is the analogous floor for the
43/// `checkpoint_summary` / `checkpoint_contents` CFs.
44#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
45pub struct Watermarks {
46    pub tx_seq_lo: u64,
47    pub checkpoint_lo: u64,
48}
49
50/// Build the singleton `(Key, Value)` pair recording the current
51/// pruning floor.
52pub fn store(watermarks: &Watermarks) -> (Key, Value) {
53    (
54        UnitKey,
55        Protobuf(PruningWatermarks {
56            tx_seq_lo: watermarks.tx_seq_lo,
57            checkpoint_lo: watermarks.checkpoint_lo,
58        }),
59    )
60}
61
62impl<R: Reader> super::RpcStoreSchema<R> {
63    /// Read the persisted pruning watermarks from disk.
64    pub fn get_pruning_watermarks(&self) -> Result<Option<Watermarks>, Error> {
65        let Some(stored) = self.pruning_watermark.get(&UnitKey)? else {
66            return Ok(None);
67        };
68        let stored = stored.into_inner();
69        Ok(Some(Watermarks {
70            tx_seq_lo: stored.tx_seq_lo,
71            checkpoint_lo: stored.checkpoint_lo,
72        }))
73    }
74}
75
76impl super::RpcStoreSchema {
77    /// Publish the `tx_seq` floor used by this database's bitmap
78    /// compaction filters.
79    ///
80    /// Callers publish only a committed `tx_seq_lo`, or zero
81    /// immediately after durably clearing the persisted watermark.
82    /// Restores may intentionally install a lower committed floor
83    /// before history backfill writes begin.
84    pub fn set_pruning_floor(&self, tx_seq_lo: u64) {
85        self.tx_seq_pruning_floor
86            .store(tx_seq_lo, Ordering::Relaxed);
87    }
88
89    #[cfg(test)]
90    pub(crate) fn current_pruning_floor(&self) -> u64 {
91        self.tx_seq_pruning_floor.load(Ordering::Relaxed)
92    }
93}
94
95#[cfg(test)]
96mod tests {
97    use sui_consistent_store::Db;
98    use sui_consistent_store::DbOptions;
99
100    use super::*;
101    use crate::RpcStoreSchema;
102    use crate::schema::event_bitmap;
103    use crate::schema::transaction_bitmap;
104
105    fn fresh_db() -> (tempfile::TempDir, Db, RpcStoreSchema) {
106        let dir = tempfile::tempdir().unwrap();
107        let (db, schema) = Db::open::<RpcStoreSchema>(dir.path(), DbOptions::default()).unwrap();
108        (dir, db, schema)
109    }
110
111    fn put_materialized_bucket_zero(db: &Db, schema: &RpcStoreSchema, dimension: &[u8]) {
112        let (tx_key, tx_value) = transaction_bitmap::store_match(dimension.to_vec(), 5);
113        let (event_key, event_value) = event_bitmap::store_match(dimension.to_vec(), 5, 0);
114        let mut batch = db.batch();
115        batch
116            .put(&schema.transaction_bitmap, &tx_key, &tx_value)
117            .unwrap();
118        batch
119            .put(&schema.event_bitmap, &event_key, &event_value)
120            .unwrap();
121        batch.commit().unwrap();
122        db.flush().unwrap();
123    }
124
125    #[test]
126    fn fresh_empty_db_starts_with_zero_pruning_floor() {
127        let (_dir, _db, schema) = fresh_db();
128        assert!(schema.get_pruning_watermarks().unwrap().is_none());
129        assert_eq!(schema.current_pruning_floor(), 0);
130    }
131
132    #[test]
133    fn reopen_loads_persisted_pruning_floor() {
134        let dir = tempfile::tempdir().unwrap();
135        let (db, schema) = Db::open::<RpcStoreSchema>(dir.path(), DbOptions::default()).unwrap();
136        let floor = transaction_bitmap::TX_BUCKET_SIZE;
137        let (watermark_key, watermark_value) = store(&Watermarks {
138            tx_seq_lo: floor,
139            checkpoint_lo: 1,
140        });
141        let mut batch = db.batch();
142        batch
143            .put(&schema.pruning_watermark, &watermark_key, &watermark_value)
144            .unwrap();
145        batch.commit().unwrap();
146        put_materialized_bucket_zero(&db, &schema, b"reopen");
147        assert_eq!(schema.current_pruning_floor(), 0);
148
149        drop(schema);
150        drop(db);
151
152        let (db, schema) = Db::open::<RpcStoreSchema>(dir.path(), DbOptions::default()).unwrap();
153        assert_eq!(schema.current_pruning_floor(), floor);
154        db.compact_range_cf(transaction_bitmap::NAME, None, None)
155            .unwrap();
156        db.compact_range_cf(event_bitmap::NAME, None, None).unwrap();
157        assert!(
158            schema
159                .get_transaction_bitmap(b"reopen".to_vec(), 0)
160                .unwrap()
161                .is_none()
162        );
163        assert!(
164            schema
165                .get_event_bitmap(b"reopen".to_vec(), 0)
166                .unwrap()
167                .is_none()
168        );
169    }
170
171    #[test]
172    fn bitmap_pruning_floors_are_isolated_per_database() {
173        let (_dir_a, db_a, schema_a) = fresh_db();
174        let (_dir_b, db_b, schema_b) = fresh_db();
175        let floor = transaction_bitmap::TX_BUCKET_SIZE;
176
177        let (watermark_key, watermark_value) = store(&Watermarks {
178            tx_seq_lo: floor,
179            checkpoint_lo: 1,
180        });
181        let mut batch = db_a.batch();
182        batch
183            .put(
184                &schema_a.pruning_watermark,
185                &watermark_key,
186                &watermark_value,
187            )
188            .unwrap();
189        batch.commit().unwrap();
190        schema_a.set_pruning_floor(floor);
191
192        put_materialized_bucket_zero(&db_a, &schema_a, b"isolated");
193        put_materialized_bucket_zero(&db_b, &schema_b, b"isolated");
194
195        for db in [&db_a, &db_b] {
196            db.compact_range_cf(transaction_bitmap::NAME, None, None)
197                .unwrap();
198            db.compact_range_cf(event_bitmap::NAME, None, None).unwrap();
199        }
200
201        assert!(
202            schema_a
203                .get_transaction_bitmap(b"isolated".to_vec(), 0)
204                .unwrap()
205                .is_none()
206        );
207        assert!(
208            schema_a
209                .get_event_bitmap(b"isolated".to_vec(), 0)
210                .unwrap()
211                .is_none()
212        );
213        assert!(
214            schema_b
215                .get_transaction_bitmap(b"isolated".to_vec(), 0)
216                .unwrap()
217                .is_some()
218        );
219        assert!(
220            schema_b
221                .get_event_bitmap(b"isolated".to_vec(), 0)
222                .unwrap()
223                .is_some()
224        );
225        assert!(schema_b.get_pruning_watermarks().unwrap().is_none());
226        assert_eq!(schema_b.current_pruning_floor(), 0);
227    }
228}
sui_rpc_store/schema/pruning_watermark.rs

sui_rpc_store/schema/
pruning_watermark.rs
