set connection parameters, revert tx delay to 2s, update query-service

.env

@@ -124,7 +124,7 @@ ESPRESSO_BUILDER_BUFFER_VIEW_NUM_COUNT=50
 ESPRESSO_BUILDER_GENESIS_FILE=$ESPRESSO_SEQUENCER_GENESIS_FILE
 
 # Load generator
-ESPRESSO_SUBMIT_TRANSACTIONS_DELAY=0s
+ESPRESSO_SUBMIT_TRANSACTIONS_DELAY=2s
 ESPRESSO_SUBMIT_TRANSACTIONS_PUBLIC_PORT=24010
 ESPRESSO_SUBMIT_TRANSACTIONS_PRIVATE_PORT=24020
 ESPRESSO_SUBMIT_TRANSACTIONS_PRIVATE_FALLBACK_PORT=24030
@@ -156,3 +156,6 @@ INTEGRATION_TEST_PROTO=http
 # Version of sequencer protocol we want to test. If this is set to
 # `03`, marketplace upgrade will be tested.
 INTEGRATION_TEST_SEQUENCER_VERSION=02
+
+
+ESPRESSO_SEQUENCER_DATABASE_MAX_CONNECTIONS=30

process-compose.yaml

@@ -116,8 +116,8 @@ processes:
       - ESPRESSO_SEQUENCER_API_PORT=$ESPRESSO_SEQUENCER_API_PORT
       - ESPRESSO_SEQUENCER_LIBP2P_BIND_ADDRESS=0.0.0.0:$ESPRESSO_DEMO_SEQUENCER_LIBP2P_PORT_0
       - ESPRESSO_SEQUENCER_LIBP2P_ADVERTISE_ADDRESS=localhost:$ESPRESSO_DEMO_SEQUENCER_LIBP2P_PORT_0
-      - ESPRESSO_SEQUENCER_API_PEERS=http://localhost:$ESPRESSO_SEQUENCER6_API_PORT
-      - ESPRESSO_SEQUENCER_STATE_PEERS=http://localhost:$ESPRESSO_SEQUENCER6_API_PORT
+      - ESPRESSO_SEQUENCER_API_PEERS=http://localhost:$ESPRESSO_SEQUENCER1_API_PORT
+      - ESPRESSO_SEQUENCER_STATE_PEERS=http://localhost:$ESPRESSO_SEQUENCER1_API_PORT
       - ESPRESSO_SEQUENCER_POSTGRES_HOST=localhost
       - ESPRESSO_SEQUENCER_POSTGRES_PORT=$ESPRESSO_SEQUENCER0_DB_PORT
       - ESPRESSO_SEQUENCER_POSTGRES_USER=root
@@ -315,7 +315,7 @@ processes:
     command: sequencer -- http -- catchup -- status
     environment:
       - ESPRESSO_SEQUENCER_API_PORT=$ESPRESSO_SEQUENCER4_API_PORT
-      - ESPRESSO_SEQUENCER_STATE_PEERS=http://localhost:$ESPRESSO_SEQUENCER_API_PORT
+      - ESPRESSO_SEQUENCER_STATE_PEERS=http://localhost:$ESPRESSO_SEQUENCER5_API_PORT
       - ESPRESSO_SEQUENCER_STORAGE_PATH=$ESPRESSO_BASE_STORAGE_PATH/seq4
       - ESPRESSO_SEQUENCER_PRIVATE_STAKING_KEY=$ESPRESSO_DEMO_SEQUENCER_STAKING_PRIVATE_KEY_4
       - ESPRESSO_SEQUENCER_PRIVATE_STATE_KEY=$ESPRESSO_DEMO_SEQUENCER_STATE_PRIVATE_KEY_4
@@ -361,8 +361,8 @@ processes:
       - ESPRESSO_SEQUENCER_API_PORT=$ESPRESSO_SEQUENCER5_API_PORT
       - ESPRESSO_SEQUENCER_LIBP2P_BIND_ADDRESS=0.0.0.0:$ESPRESSO_DEMO_SEQUENCER_LIBP2P_PORT_5
       - ESPRESSO_SEQUENCER_LIBP2P_ADVERTISE_ADDRESS=localhost:$ESPRESSO_DEMO_SEQUENCER_LIBP2P_PORT_5
-      - ESPRESSO_SEQUENCER_API_PEERS=http://localhost:$ESPRESSO_SEQUENCER_API_PORT
-      - ESPRESSO_SEQUENCER_STATE_PEERS=http://localhost:$ESPRESSO_SEQUENCER_API_PORT
+      - ESPRESSO_SEQUENCER_API_PEERS=http://localhost:$ESPRESSO_SEQUENCER6_API_PORT
+      - ESPRESSO_SEQUENCER_STATE_PEERS=http://localhost:$ESPRESSO_SEQUENCER6_API_PORT
       - ESPRESSO_SEQUENCER_SQLITE_PATH=$ESPRESSO_BASE_STORAGE_PATH/seq5-sqlite.db
       - ESPRESSO_SEQUENCER_PRIVATE_STAKING_KEY=$ESPRESSO_DEMO_SEQUENCER_STAKING_PRIVATE_KEY_5
       - ESPRESSO_SEQUENCER_PRIVATE_STATE_KEY=$ESPRESSO_DEMO_SEQUENCER_STATE_PRIVATE_KEY_5
@@ -377,6 +377,7 @@ processes:
       - ESPRESSO_SEQUENCER_IDENTITY_LATITUDE=40.7128
       - ESPRESSO_SEQUENCER_IDENTITY_LONGITUDE=-74.0060
       - ESPRESSO_SEQUENCER_PUBLIC_API_URL=http://localhost:$ESPRESSO_SEQUENCER5_API_PORT/
+      - ESPRESSO_SEQUENCER_DATABASE_MAX_CONNECTIONS=10
     depends_on:
       orchestrator:
         condition: process_healthy
@@ -426,6 +427,7 @@ processes:
       - ESPRESSO_SEQUENCER_IDENTITY_LONGITUDE=21.8243
       - ESPRESSO_SEQUENCER_PUBLIC_API_URL=http://localhost:$ESPRESSO_SEQUENCER6_API_PORT/
       - ESPRESSO_SEQUENCER_L1_PROVIDER=ws://localhost:$ESPRESSO_SEQUENCER_L1_WS_PORT
+      - ESPRESSO_SEQUENCER_DATABASE_MAX_CONNECTIONS=10
     depends_on:
       orchestrator:
         condition: process_healthy

sequencer/src/persistence/sql.rs

@@ -152,7 +152,7 @@ pub struct Options {
     ///
     /// Any connection which has been open and unused longer than this duration will be
     /// automatically closed to reduce load on the server.
-    #[clap(long, env = "ESPRESSO_SEQUENCER_POSTGRES_IDLE_CONNECTION_TIMEOUT", value_parser = parse_duration, default_value = "10m")]
+    #[clap(long, env = "ESPRESSO_SEQUENCER_DATABASE_IDLE_CONNECTION_TIMEOUT", value_parser = parse_duration, default_value = "10m")]
     pub(crate) idle_connection_timeout: Duration,
 
     /// The maximum lifetime of a database connection.
@@ -161,7 +161,7 @@ pub struct Options {
     /// (and, if needed, replaced), even if it is otherwise healthy. It is good practice to refresh
     /// even healthy connections once in a while (e.g. daily) in case of resource leaks in the
     /// server implementation.
-    #[clap(long, env = "ESPRESSO_SEQUENCER_POSTGRES_CONNECTION_TIMEOUT", value_parser = parse_duration, default_value = "30m")]
+    #[clap(long, env = "ESPRESSO_SEQUENCER_DATABASE_CONNECTION_TIMEOUT", value_parser = parse_duration, default_value = "30m")]
     pub(crate) connection_timeout: Duration,
 
     /// The minimum number of database connections to maintain at any time.
@@ -171,7 +171,7 @@ pub struct Options {
     /// connections when at least this many simultaneous connections are frequently needed.
     #[clap(
         long,
-        env = "ESPRESSO_SEQUENCER_POSTGRES_MIN_CONNECTIONS",
+        env = "ESPRESSO_SEQUENCER_DATABASE_MIN_CONNECTIONS",
         default_value = "0"
     )]
     pub(crate) min_connections: u32,
@@ -182,7 +182,7 @@ pub struct Options {
     /// (or begin a transaction) will block until one of the existing connections is released.
     #[clap(
         long,
-        env = "ESPRESSO_SEQUENCER_POSTGRES_MAX_CONNECTIONS",
+        env = "ESPRESSO_SEQUENCER_DATABASE_MAX_CONNECTIONS",
         default_value = "25"
     )]
     pub(crate) max_connections: u32,
@@ -273,6 +273,11 @@ impl TryFrom<Options> for Config {
             cfg = cfg.pool(pool);
         }
 
+        cfg = cfg.max_connections(opt.max_connections);
+        cfg = cfg.idle_connection_timeout(opt.idle_connection_timeout);
+        cfg = cfg.min_connections(opt.min_connections);
+        cfg = cfg.connection_timeout(opt.connection_timeout);
+
         #[cfg(not(feature = "embedded-db"))]
         {
             cfg = cfg.migrations(include_migrations!(
@@ -396,6 +401,7 @@ impl From<PruningOptions> for PrunerCfg {
         if let Some(interval) = opt.interval {
             cfg = cfg.with_interval(interval);
         }
+
         cfg
     }
 }
@@ -564,10 +570,8 @@ impl SequencerPersistence for Persistence {
     async fn load_anchor_leaf(
         &self,
     ) -> anyhow::Result<Option<(Leaf, QuorumCertificate<SeqTypes>)>> {
-        let Some(row) = self
-            .db
-            .read()
-            .await?
+        let mut tx = self.db.read().await?;
+        let Some(row) = tx
             .fetch_optional("SELECT leaf, qc FROM anchor_leaf ORDER BY view DESC LIMIT 1")
             .await?
         else {
@@ -594,10 +598,9 @@ impl SequencerPersistence for Persistence {
     async fn load_undecided_state(
         &self,
     ) -> anyhow::Result<Option<(CommitmentMap<Leaf>, BTreeMap<ViewNumber, View<SeqTypes>>)>> {
-        let Some(row) = self
-            .db
-            .read()
-            .await?
+        let mut tx = self.db.read().await?;
+
+        let Some(row) = tx
             .fetch_optional("SELECT leaves, state FROM undecided_state WHERE id = 0")
             .await?
         else {
@@ -617,10 +620,9 @@ impl SequencerPersistence for Persistence {
         &self,
         view: ViewNumber,
     ) -> anyhow::Result<Option<Proposal<SeqTypes, DaProposal<SeqTypes>>>> {
-        let result = self
-            .db
-            .read()
-            .await?
+        let mut tx = self.db.read().await?;
+
+        let result = tx
             .fetch_optional(
                 query("SELECT data FROM da_proposal where view = $1").bind(view.u64() as i64),
             )
@@ -638,10 +640,8 @@ impl SequencerPersistence for Persistence {
         &self,
         view: ViewNumber,
     ) -> anyhow::Result<Option<Proposal<SeqTypes, VidDisperseShare<SeqTypes>>>> {
-        let result = self
-            .db
-            .read()
-            .await?
+        let mut tx = self.db.read().await?;
+        let result = tx
             .fetch_optional(
                 query("SELECT data FROM vid_share where view = $1").bind(view.u64() as i64),
             )
@@ -658,12 +658,9 @@ impl SequencerPersistence for Persistence {
     async fn load_quorum_proposals(
         &self,
     ) -> anyhow::Result<BTreeMap<ViewNumber, Proposal<SeqTypes, QuorumProposal<SeqTypes>>>> {
-        let rows = self
-            .db
-            .read()
-            .await?
-            .fetch_all("SELECT * FROM quorum_proposals")
-            .await?;
+        let mut tx = self.db.read().await?;
+
+        let rows = tx.fetch_all("SELECT * FROM quorum_proposals").await?;
 
         Ok(BTreeMap::from_iter(
             rows.into_iter()