espresso_node/persistence/sql/
postgres.rs

1//! Persistence on a Postgres server.
2
3use hotshot_query_service::{
4    data_source::storage::sql::postgres::{self as qs, set_no_deferrable_on_read},
5    include_migrations,
6};
7use light_client::storage::{LightClientDbOptions, LightClientPostgresOptions};
8
9#[path = "../sql.rs"]
10mod imp;
11
12pub use imp::*;
13use qs::{Config, SqlStorage};
14
15use super::Options as SqlOptions;
16#[cfg(any(test, feature = "testing"))]
17use super::PostgresOptions;
18use crate::{api::options::Query, persistence::storage_probe::StorageProbe};
19
20impl BackendOptions for qs::Db {
21    fn apply(mut cfg: Config, opt: &SqlOptions) -> anyhow::Result<Config> {
22        cfg = cfg.query_max_connections(opt.query_max_connections.unwrap_or(opt.max_connections));
23        cfg = cfg.query_min_connections(opt.query_min_connections.unwrap_or(opt.min_connections));
24        set_no_deferrable_on_read(opt.postgres_options.no_deferrable);
25
26        cfg = cfg.migrations(include_migrations!(
27            "$CARGO_MANIFEST_DIR/api/migrations/postgres"
28        ));
29
30        let pg_options = &opt.postgres_options;
31
32        if let Some(host) = &pg_options.host {
33            cfg = cfg.host(host.clone());
34        }
35
36        if let Some(port) = pg_options.port {
37            cfg = cfg.port(port);
38        }
39
40        if let Some(database) = &pg_options.database {
41            cfg = cfg.database(database);
42        }
43
44        if let Some(user) = &pg_options.user {
45            cfg = cfg.user(user);
46        }
47
48        if let Some(password) = &pg_options.password {
49            cfg = cfg.password(password);
50        }
51
52        if pg_options.use_tls {
53            cfg = cfg.tls();
54        }
55
56        Ok(cfg)
57    }
58
59    /// A Postgres node keeps its light client in the same database, so it survives restarts and
60    /// container replacement without any extra volume.
61    fn light_client_db(cfg: &Config, query: &Query) -> LightClientDbOptions {
62        LightClientDbOptions::Postgres(Box::new(LightClientPostgresOptions {
63            connect: cfg.connect_options().clone(),
64            num_connections: query.light_client_db.num_connections,
65            num_leaves: query.light_client_db.num_leaves,
66        }))
67    }
68
69    async fn reset_light_client(cfg: &Config) -> anyhow::Result<()> {
70        LightClientPostgresOptions::reset(cfg.connect_options()).await
71    }
72
73    async fn storage_probe(
74        _db: &SqlStorage,
75        _opt: &SqlOptions,
76    ) -> anyhow::Result<Option<StorageProbe>> {
77        Ok(None)
78    }
79}
80
81#[cfg(any(test, feature = "testing"))]
82pub(crate) fn tmp_options(db: &qs::testing::TmpDb) -> SqlOptions {
83    PostgresOptions {
84        port: Some(db.port()),
85        host: Some(db.host()),
86        user: Some("postgres".into()),
87        password: Some("password".into()),
88        ..Default::default()
89    }
90    .into()
91}
92
93#[cfg(test)]
94mod test {
95    use std::sync::Arc;
96
97    use committable::Committable as _;
98    use espresso_types::{
99        FeeAccount, Header, Leaf, Leaf2, NodeState, Payload, Transaction as Tx,
100        traits::{PersistenceOptions as _, SequencerPersistence as _},
101    };
102    use hotshot_example_types::node_types::TEST_VERSIONS;
103    use hotshot_query_service::{
104        availability::{BlockQueryData, LeafQueryData},
105        data_source::{
106            Transaction as _, VersionedDataSource as _, storage::UpdateAvailabilityStorage,
107        },
108    };
109    use hotshot_types::{
110        data::{QuorumProposal, ViewNumber, vid_commitment},
111        event::HotShotAction,
112        simple_certificate::QuorumCertificate,
113        traits::{
114            EncodeBytes,
115            block_contents::{
116                BlockHeader, BlockPayload as _, BuilderFee, GENESIS_VID_NUM_STORAGE_NODES,
117            },
118            signature_key::BuilderSignatureKey,
119        },
120    };
121    use sqlx::{Row as _, query};
122
123    use super::*;
124    use crate::{SeqTypes, persistence::tests::TestablePersistence as _};
125
126    #[test]
127    fn sqlite_options_do_not_build_postgres_persistence() {
128        let opt = SqlOptions {
129            embedded_db: true,
130            ..SqlOptions::default()
131        };
132        assert!(Options::try_from(opt.clone()).is_err());
133        assert!(matches!(
134            opt.into_backend(),
135            super::super::SqlBackendOptions::Sqlite(_)
136        ));
137    }
138
139    #[tokio::test]
140    async fn a_sqlite_pool_is_not_reused_for_postgres() {
141        let pool = sqlx::sqlite::SqlitePoolOptions::new()
142            .connect(":memory:")
143            .await
144            .unwrap();
145        let opt = SqlOptions {
146            pool: Some(pool.into()),
147            ..SqlOptions::default()
148        };
149        let Err(err) = Config::try_from(&opt) else {
150            panic!("a SQLite pool was accepted for Postgres");
151        };
152        assert!(format!("{err:#}").contains("other SQL backend"), "{err:#}");
153    }
154
155    async fn test_postgres_read_ns_table(instance_state: NodeState) {
156        let tmp = Persistence::tmp_storage().await;
157        let mut opt = Persistence::options(&tmp);
158        let storage = opt.create().await.unwrap();
159
160        let txs = [
161            Tx::new(10001u32.into(), vec![1, 2, 3]),
162            Tx::new(10001u32.into(), vec![4, 5, 6]),
163            Tx::new(10009u32.into(), vec![7, 8, 9]),
164        ];
165
166        let validated_state = Default::default();
167        let justify_qc =
168            QuorumCertificate::genesis(&validated_state, &instance_state, TEST_VERSIONS.test).await;
169        let view_number: ViewNumber = justify_qc.view_number + 1;
170        let parent_leaf = Leaf::genesis(&validated_state, &instance_state, TEST_VERSIONS.test.base)
171            .await
172            .into();
173
174        let (payload, ns_table) =
175            Payload::from_transactions(&txs, &validated_state, &instance_state)
176                .await
177                .unwrap();
178        let payload_bytes = payload.encode();
179        let payload_commitment = vid_commitment(
180            &payload_bytes,
181            &ns_table.encode(),
182            GENESIS_VID_NUM_STORAGE_NODES,
183            instance_state.current_version,
184        );
185        let builder_commitment = payload.builder_commitment(&ns_table);
186        let (fee_account, fee_key) = FeeAccount::generated_from_seed_indexed([0; 32], 0);
187        let fee_amount = 0;
188        let fee_signature = FeeAccount::sign_fee(&fee_key, fee_amount, &ns_table).unwrap();
189        let block_header = Header::new(
190            &validated_state,
191            &instance_state,
192            &parent_leaf,
193            payload_commitment,
194            builder_commitment,
195            ns_table,
196            BuilderFee {
197                fee_amount,
198                fee_account,
199                fee_signature,
200            },
201            instance_state.current_version,
202            view_number.u64(),
203        )
204        .await
205        .unwrap();
206        let proposal = QuorumProposal {
207            block_header: block_header.clone(),
208            view_number,
209            justify_qc: justify_qc.clone(),
210            upgrade_certificate: None,
211            proposal_certificate: None,
212        };
213        let leaf: Leaf2 = Leaf::from_quorum_proposal(&proposal).into();
214        let mut qc = justify_qc.to_qc2();
215        qc.data.leaf_commit = leaf.commit();
216        qc.view_number = view_number;
217
218        let mut tx = storage.db.write().await.unwrap();
219        tx.insert_leaf(&LeafQueryData::new(leaf, qc).unwrap())
220            .await
221            .unwrap();
222        tx.insert_block(&BlockQueryData::<SeqTypes>::new(block_header, payload))
223            .await
224            .unwrap();
225        tx.commit().await.unwrap();
226
227        let mut tx = storage.db.read().await.unwrap();
228        let rows = query(
229            "
230            SELECT ns_id, read_ns_id(get_ns_table(h.data), t.ns_index) AS read_ns_id
231              FROM header AS h
232              JOIN transactions AS t ON t.block_height = h.height
233              ORDER BY t.ns_index, t.position
234        ",
235        )
236        .fetch_all(tx.as_mut())
237        .await
238        .unwrap();
239        assert_eq!(rows.len(), txs.len());
240        for (i, row) in rows.into_iter().enumerate() {
241            let ns = u64::from(txs[i].namespace()) as i64;
242            assert_eq!(row.get::<i64, _>("ns_id"), ns);
243            assert_eq!(row.get::<i64, _>("read_ns_id"), ns);
244        }
245    }
246
247    #[test_log::test(tokio::test(flavor = "multi_thread"))]
248    async fn test_postgres_read_ns_table_v0_1() {
249        let v0_1 = versions::version(0, 1);
250        test_postgres_read_ns_table(
251            NodeState::mock()
252                .with_current_version(v0_1)
253                .with_genesis_version(v0_1),
254        )
255        .await;
256    }
257
258    #[test_log::test(tokio::test(flavor = "multi_thread"))]
259    async fn test_postgres_read_ns_table_v0_2() {
260        test_postgres_read_ns_table(NodeState::mock_v2()).await;
261    }
262
263    #[test_log::test(tokio::test(flavor = "multi_thread"))]
264    async fn test_postgres_read_ns_table_v0_3() {
265        test_postgres_read_ns_table(NodeState::mock_v3().with_epoch_height(0)).await;
266    }
267
268    /// Verify that concurrent calls to `record_action` all succeed under
269    /// PostgreSQL SERIALIZABLE isolation. `self.serializable_backoff.retry_if` handles any
270    /// 40001 serialization failures that arise when many tasks race to update
271    /// the same row.
272    #[test_log::test(tokio::test(flavor = "multi_thread"))]
273    async fn test_record_action_concurrent() {
274        let tmp = Persistence::tmp_storage().await;
275        let storage = Arc::new(Persistence::connect(&tmp).await);
276
277        let handles: Vec<_> = (0u64..20)
278            .map(|i| {
279                let storage = Arc::clone(&storage);
280                tokio::spawn(async move {
281                    storage
282                        .record_action(ViewNumber::new(i), HotShotAction::Vote)
283                        .await
284                })
285            })
286            .collect();
287
288        for handle in handles {
289            handle.await.unwrap().unwrap();
290        }
291
292        let latest = storage.load_latest_acted_view().await.unwrap();
293        assert_eq!(latest, Some(ViewNumber::new(19)));
294    }
295
296    #[test_log::test(tokio::test(flavor = "multi_thread"))]
297    async fn test_light_client_shares_the_node_database() {
298        use hotshot_query_service::availability::LeafId;
299        use light_client::storage::Storage as _;
300
301        use crate::api::{data_source::DataSourceOptions as _, options::Query};
302
303        let tmp = Persistence::tmp_storage().await;
304        let mut opt = Persistence::options(&tmp);
305        let persistence = opt.create().await.unwrap();
306
307        let db = opt
308            .light_client_db(&Query::default())
309            .unwrap()
310            .connect()
311            .await
312            .unwrap();
313        let leaf = light_client::testing::leaf_chain(0..1, versions::EPOCH_VERSION)
314            .await
315            .remove(0);
316        db.insert_leaf(leaf.clone()).await.unwrap();
317        assert_eq!(
318            db.leaf_upper_bound(LeafId::Number(0)).await.unwrap(),
319            Some(leaf)
320        );
321
322        let (leaves,): (i64,) = sqlx::query_as("SELECT count(*) FROM light_client.leaf")
323            .fetch_one(&persistence.db.pool())
324            .await
325            .unwrap();
326        assert_eq!(leaves, 1);
327    }
328
329    #[test_log::test(tokio::test(flavor = "multi_thread"))]
330    async fn test_reset_drops_light_client_cache() {
331        use light_client::storage::Storage as _;
332
333        use crate::api::{data_source::DataSourceOptions as _, options::Query};
334
335        let tmp = Persistence::tmp_storage().await;
336        let opt = Persistence::options(&tmp);
337        let light_client = || opt.light_client_db(&Query::default()).unwrap().connect();
338        let leaf = light_client::testing::leaf_chain(0..1, versions::EPOCH_VERSION)
339            .await
340            .remove(0);
341        let db = light_client().await.unwrap();
342        db.insert_leaf(leaf).await.unwrap();
343        drop(db);
344
345        opt.clone().reset().await.unwrap();
346
347        let db = light_client().await.unwrap();
348        assert_eq!(db.block_height().await.unwrap(), 0);
349    }
350}
espresso_node/persistence/sql/postgres.rs

espresso_node/persistence/sql/
postgres.rs
