chore: test

.github/workflows/main.yml

@@ -2,9 +2,7 @@ name: CI
 
 on:
  push:
- branches: [main]
- pull_request_review:
- types: [submitted]
+ pull_request:
 
 jobs:
  fmt:
@@ -20,7 +18,6 @@ jobs:
 
  check:
  name: Check
- if: github.event.review.state == 'approved'
  runs-on: ubuntu-latest
  steps:
  - uses: actions/checkout@v3
@@ -30,7 +27,7 @@ jobs:
  run: git submodule update --init --recursive
  - uses: actions/setup-go@v4
  with:
- go-version: '1.19'
+ go-version: '1.20'
  - uses: actions-rs/toolchain@v1
  with:
  profile: minimal
@@ -47,7 +44,6 @@ jobs:
 
  test:
  name: Test Suite
- if: github.event.review.state == 'approved'
  strategy:
  matrix:
  os: [ubuntu-latest]
@@ -77,7 +73,7 @@ jobs:
  run: git submodule update --init --recursive
  - uses: actions/setup-go@v4
  with:
- go-version: '1.19'
+ go-version: '1.20'
  - uses: actions-rs/toolchain@v1
  with:
  profile: minimal

src/db/resolver.rs

@@ -1,7 +1,7 @@
 use async_graphql::{OutputType, SimpleObject};
 use chrono::Utc;
 use serde::{de::DeserializeOwned, Serialize};
-use sqlx::{types::Json, FromRow, PgPool, Row as SqliteRow};
+use sqlx::{postgres::PgQueryResult, types::Json, FromRow, PgPool, Row as SqliteRow};
 use std::ops::Deref;
 use tracing::trace;
 
@@ -83,6 +83,23 @@ ORDER BY id
  Ok(rows)
 }
 
+pub async fn count_messages(pool: &PgPool) -> anyhow::Result<i64> {
+ let result = sqlx::query!(
+ r#"
+ SELECT COUNT(*) as "count!: i64"
+ FROM messages
+ "#
+ )
+ .fetch_one(pool)
+ .await
+ .map_err(|e| {
+ trace!("Database query error: {:#?}", e);
+ anyhow::Error::new(e)
+ })?;
+
+ Ok(result.count)
+}
+
 pub async fn list_rows<T>(pool: &PgPool) -> Result<Vec<GraphQLRow<T>>, anyhow::Error>
 where
  T: Clone + Serialize + DeserializeOwned + OutputType + std::marker::Unpin,
@@ -201,24 +218,51 @@ RETURNING id
  Ok(deleted_ids.try_into().unwrap())
 }
 
-/// Function to delete messages older than `retention` minutes
-/// Returns the number of messages deleted
-pub async fn prune_old_messages(pool: &PgPool, retention: i32) -> Result<i64, anyhow::Error> {
+/// Function to delete messages older than `retention` minutes in batches
+/// Returns the total number of messages deleted
+/// Arguments:
+/// - `pool`: &PgPool - A reference to the PostgreSQL connection pool
+/// - `retention`: i32 - The retention time in minutes
+/// - `batch_size`: i64 - The number of messages to delete in each batch
+pub async fn prune_old_messages(
+ pool: &PgPool,
+ retention: i32,
+ batch_size: i64,
+) -> Result<i64, anyhow::Error> {
  let cutoff_nonce = Utc::now().timestamp() - (retention as i64 * 60);
+ let mut total_deleted = 0i64;
+
+ loop {
+ let delete_query = sqlx::query(
+ r#"
+ WITH deleted AS (
+ SELECT id
+ FROM messages
+ WHERE (message->>'nonce')::bigint < $1
+ ORDER BY id ASC
+ LIMIT $2
+ FOR UPDATE SKIP LOCKED
+ )
+ DELETE FROM messages
+ WHERE id IN (SELECT id FROM deleted)
+ RETURNING id
+ "#,
+ )
+ .bind(cutoff_nonce)
+ .bind(batch_size);
 
- let deleted_count = sqlx::query!(
- r#"
- DELETE FROM messages
- WHERE (message->>'nonce')::bigint < $1
- RETURNING id
- "#,
- cutoff_nonce
- )
- .fetch_all(pool)
- .await?
- .len() as i64;
+ let result: PgQueryResult = delete_query.execute(pool).await?;
+ let deleted_count = result.rows_affected() as i64;
+
+ total_deleted += deleted_count;
+
+ // Break the loop if we deleted fewer rows than the batch size, indicating we've processed all eligible messages.
+ if deleted_count < batch_size {
+ break;
+ }
+ }
 
- Ok(deleted_count)
+ Ok(total_deleted)
 }
 
 pub async fn list_active_indexers(

src/operator/mod.rs

@@ -13,11 +13,11 @@ use tracing::{debug, info, trace, warn};
 
 use graphcast_sdk::graphcast_agent::{message_typing::GraphcastMessage, GraphcastAgent};
 
-use crate::db::resolver::{prune_old_messages, retain_max_storage};
+use crate::db::resolver::{count_messages, prune_old_messages, retain_max_storage};
 use crate::metrics::{CONNECTED_PEERS, GOSSIP_PEERS, PRUNED_MESSAGES, RECEIVED_MESSAGES};
 use crate::{
  config::Config,
- db::resolver::{add_message, list_messages},
+ db::resolver::add_message,
  message_types::{PublicPoiMessage, SimpleMessage, UpgradeIntentMessage},
  metrics::{handle_serve_metrics, ACTIVE_PEERS, CACHED_MESSAGES},
  server::run_server,
@@ -109,7 +109,7 @@ impl RadioOperator {
  let skip_iteration_clone = skip_iteration.clone();
 
  let mut network_update_interval = interval(Duration::from_secs(600));
- let mut comparison_interval = interval(Duration::from_secs(30));
+ let mut summary_interval = interval(Duration::from_secs(180));
 
  let iteration_timeout = Duration::from_secs(180);
  let update_timeout = Duration::from_secs(5);
@@ -160,7 +160,7 @@ impl RadioOperator {
  .set(self.graphcast_agent.number_of_peers().try_into().unwrap());
  }
  },
- _ = comparison_interval.tick() => {
+ _ = summary_interval.tick() => {
  trace!("Local summary update");
  if skip_iteration.load(Ordering::SeqCst) {
  skip_iteration.store(false, Ordering::SeqCst);
@@ -185,10 +185,12 @@ impl RadioOperator {
  };
  }
 
+ let batch_size = 1000;
+
  // Always prune old messages based on RETENTION
  match timeout(
  update_timeout,
- prune_old_messages(&self.db, self.config.retention)
+ prune_old_messages(&self.db, self.config.retention, batch_size)
  ).await {
  Err(e) => debug!(err = tracing::field::debug(e), "Pruning by retention timed out"),
  Ok(Ok(num_pruned)) => {
@@ -199,14 +201,13 @@ impl RadioOperator {
  };
 
  // List the remaining messages
- let result = timeout(update_timeout, list_messages::<GraphcastMessage<PublicPoiMessage>>(&self.db)).await;
+ let result = timeout(update_timeout, count_messages(&self.db)).await.expect("could not count messages");
 
  match result {
- Err(e) => warn!(err = tracing::field::debug(e), "Public PoI messages summary timed out"),
- Ok(msgs) => {
- let msg_num = msgs.map_or(0, |m| m.len());
- CACHED_MESSAGES.set(msg_num as i64);
- info!(total_messages = msg_num,
+ Err(e) => warn!(err = tracing::field::debug(e), "Database query for message count timed out"),
+ Ok(count) => {
+ CACHED_MESSAGES.set(count);
+ info!(total_messages = count,
  total_num_pruned,
  "Monitoring summary"
  )

src/server/model/mod.rs

@@ -1,9 +1,9 @@
 use async_graphql::{Context, EmptySubscription, Object, OutputType, Schema, SimpleObject};
 
-use chrono::{Duration, Utc};
+use chrono::Utc;
 use serde::{de::DeserializeOwned, Serialize};
 use sqlx::{Pool, Postgres};
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 use thiserror::Error;
 
 use crate::{
@@ -61,13 +61,13 @@ impl QueryRoot {
  &self,
  ctx: &Context<'_>,
  indexers: Option<Vec<String>>,
- minutes_ago: Option<i64>,
+ minutes_ago: Option<u64>,
  ) -> Result<Vec<String>, HttpServiceError> {
  let pool = ctx.data_unchecked::<Pool<Postgres>>();
  // Use a default time window if not specified
  // Default to 1440 minutes (24 hours) if not provided
  let minutes_ago = minutes_ago.unwrap_or(1440);
- let from_timestamp = (Utc::now() - Duration::minutes(minutes_ago)).timestamp();
+ let from_timestamp = (Utc::now() - Duration::from_secs(minutes_ago * 60)).timestamp();
 
  let active_indexers = list_active_indexers(pool, indexers, from_timestamp).await?;
  Ok(active_indexers)
@@ -77,11 +77,11 @@ impl QueryRoot {
  &self,
  ctx: &Context<'_>,
  indexers: Option<Vec<String>>,
- minutes_ago: Option<i64>,
+ minutes_ago: Option<u64>,
  ) -> Result<Vec<IndexerStats>, HttpServiceError> {
  let pool = ctx.data_unchecked::<Pool<Postgres>>();
  let minutes_ago = minutes_ago.unwrap_or(1440);
- let from_timestamp = (Utc::now() - Duration::minutes(minutes_ago)).timestamp();
+ let from_timestamp = (Utc::now() - Duration::from_secs(minutes_ago * 60)).timestamp();
 
  let stats = get_indexer_stats(pool, indexers, from_timestamp).await?;
  Ok(stats)