solana-autopilot: prune expired quotes and old order events

crates/autopilot-svm/example.toml

@@ -30,6 +30,10 @@ submission-deadline-slots = 25
 name = "baseline"
 url = "http://localhost:11088"
 
+[db-cleanup]
+cleanup-interval = "1d"
+cleanup-threshold = "30d"
+
 [logging]
 filter = "info,autopilot_svm=debug"
 

crates/autopilot-svm/src/infra/cleanup.rs

@@ -0,0 +1,39 @@
+//! Periodic deletion of database rows nothing reads anymore.
+
+use {crate::infra::db, sqlx::PgPool, std::time::Duration};
+
+/// Deletes expired quotes and old order events on a fixed interval, like the
+/// EVM autopilot's database cleanup.
+pub struct Cleanup {
+    pool: PgPool,
+    interval: Duration,
+    event_age: chrono::Duration,
+}
+
+impl Cleanup {
+    pub fn new(pool: PgPool, interval: Duration, event_age: Duration) -> Self {
+        Self {
+            pool,
+            interval,
+            event_age: chrono::Duration::from_std(event_age).expect("event age fits chrono"),
+        }
+    }
+
+    /// Run the cleanup forever. A failed run is logged and retried on the
+    /// next tick.
+    pub async fn run_forever(self) -> ! {
+        let mut interval = tokio::time::interval(self.interval);
+        loop {
+            interval.tick().await;
+            let now = chrono::Utc::now();
+            match db::remove_expired_quotes(&self.pool, now).await {
+                Ok(removed) => tracing::debug!(removed, "expired quotes cleanup"),
+                Err(err) => tracing::warn!(?err, "failed to delete expired quotes"),
+            }
+            match db::remove_order_events_before(&self.pool, now - self.event_age).await {
+                Ok(removed) => tracing::debug!(removed, "order events cleanup"),
+                Err(err) => tracing::warn!(?err, "failed to delete old order events"),
+            }
+        }
+    }
+}

crates/autopilot-svm/src/infra/config.rs

@@ -62,11 +62,44 @@ pub struct Config {
     /// sponsored orders: without it their winning solutions dispatch without
     /// creations and fail at the driver.
     pub sponsoring: Option<Sponsoring>,
+    /// Periodic deletion of expired quotes and old order events.
+    #[serde(default)]
+    pub db_cleanup: DbCleanup,
     /// Logging configuration.
     #[serde(default)]
     pub logging: LoggingConfig,
 }
 
+/// Database cleanup cadence. The age threshold applies to order events,
+/// expired quotes are removed regardless of age.
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "kebab-case", deny_unknown_fields)]
+pub struct DbCleanup {
+    /// Time between cleanup runs.
+    #[serde(with = "humantime_serde", default = "default_cleanup_interval")]
+    pub cleanup_interval: Duration,
+    /// Age at which an order event is deleted.
+    #[serde(with = "humantime_serde", default = "default_cleanup_threshold")]
+    pub cleanup_threshold: Duration,
+}
+
+impl Default for DbCleanup {
+    fn default() -> Self {
+        Self {
+            cleanup_interval: default_cleanup_interval(),
+            cleanup_threshold: default_cleanup_threshold(),
+        }
+    }
+}
+
+const fn default_cleanup_interval() -> Duration {
+    Duration::from_secs(24 * 60 * 60)
+}
+
+const fn default_cleanup_threshold() -> Duration {
+    Duration::from_secs(30 * 24 * 60 * 60)
+}
+
 impl Config {
     /// Build the `observe::Config` for the tracing framework from the logging
     /// configuration.
@@ -176,6 +209,14 @@ mod tests {
         assert_eq!(config.competition.submission_deadline_slots.get(), 25);
         assert_eq!(config.max_auction_age, Duration::from_secs(5 * 60));
         assert_eq!(config.min_auction_interval, Duration::from_secs(2));
+        assert_eq!(
+            config.db_cleanup.cleanup_interval,
+            Duration::from_secs(24 * 60 * 60)
+        );
+        assert_eq!(
+            config.db_cleanup.cleanup_threshold,
+            Duration::from_secs(30 * 24 * 60 * 60)
+        );
         assert_eq!(config.drivers.len(), 1);
         assert_eq!(config.drivers[0].name, "baseline");
         assert_eq!(config.logging.filter, "info,autopilot_svm=debug");

crates/autopilot-svm/src/infra/db.rs

@@ -5,6 +5,7 @@ use {
     anyhow::{Context, Result},
     bigdecimal::{BigDecimal, ToPrimitive},
     chain_types::solana::{AppData, IntentHash, Pubkey},
+    chrono::{DateTime, Utc},
     database::byte_array::ByteArray,
     sqlx::PgExecutor,
 };
@@ -246,10 +247,40 @@ fn to_amount(value: &BigDecimal) -> Result<u64> {
         .with_context(|| format!("amount {value} does not fit u64"))
 }
 
+/// Delete quotes whose expiration passed. Answers the number removed.
+pub async fn remove_expired_quotes(ex: impl PgExecutor<'_>, now: DateTime<Utc>) -> Result<u64> {
+    let result = sqlx::query("DELETE FROM solana.quotes WHERE expiration_timestamp < $1")
+        .bind(now)
+        .execute(ex)
+        .await
+        .context("delete expired quotes")?;
+    Ok(result.rows_affected())
+}
+
+/// Delete order events recorded before the timestamp. Answers the number
+/// removed.
+pub async fn remove_order_events_before(
+    ex: impl PgExecutor<'_>,
+    timestamp: DateTime<Utc>,
+) -> Result<u64> {
+    let result = sqlx::query("DELETE FROM solana.order_events WHERE timestamp < $1")
+        .bind(timestamp)
+        .execute(ex)
+        .await
+        .context("delete old order events")?;
+    Ok(result.rows_affected())
+}
+
 #[cfg(test)]
 mod tests {
     use {
-        super::{last_indexed_slot, open_orders},
+        super::{
+            Utc,
+            last_indexed_slot,
+            open_orders,
+            remove_expired_quotes,
+            remove_order_events_before,
+        },
         bigdecimal::BigDecimal,
         database::byte_array::ByteArray,
         sqlx::PgTransaction,
@@ -408,4 +439,54 @@ VALUES ($1, $2, CASE WHEN $3 THEN now() END, $4, $5)
             .unwrap();
         assert_eq!(last_indexed_slot(&mut *tx).await.unwrap(), Some(42));
     }
+
+    /// Expired quotes and old order events are deleted, fresh rows survive.
+    #[tokio::test]
+    #[ignore = "needs the solana.* schema applied to the local database"]
+    async fn solana_db_cleanup_removes_expired_rows() {
+        let pool = crate::test_db::pool().await;
+        sqlx::query("TRUNCATE solana.quotes, solana.order_events")
+            .execute(&pool)
+            .await
+            .unwrap();
+        sqlx::query(
+            "INSERT INTO solana.quotes (sell_token, buy_token, sell_amount, buy_amount, kind, \
+             solver, expiration_timestamp) VALUES ($1, $2, 1, 2, 'sell'::solana.OrderKind, $3, \
+             now() - interval '1 minute'), ($1, $2, 1, 2, 'sell'::solana.OrderKind, $3, now() + \
+             interval '1 hour')",
+        )
+        .bind(ByteArray([0x11; 32]))
+        .bind(ByteArray([0x22; 32]))
+        .bind(ByteArray([0x33; 32]))
+        .execute(&pool)
+        .await
+        .unwrap();
+        sqlx::query(
+            "INSERT INTO solana.order_events (order_uid, timestamp, label) VALUES ($1, now() - \
+             interval '40 days', 'created'::solana.OrderEventLabel), ($1, now(), \
+             'created'::solana.OrderEventLabel)",
+        )
+        .bind(ByteArray([0x44; 32]))
+        .execute(&pool)
+        .await
+        .unwrap();
+
+        let now = Utc::now();
+        assert_eq!(remove_expired_quotes(&pool, now).await.unwrap(), 1);
+        assert_eq!(
+            remove_order_events_before(&pool, now - chrono::Duration::days(30))
+                .await
+                .unwrap(),
+            1
+        );
+        let quotes: i64 = sqlx::query_scalar("SELECT count(*) FROM solana.quotes")
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        let events: i64 = sqlx::query_scalar("SELECT count(*) FROM solana.order_events")
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        assert_eq!((quotes, events), (1, 1));
+    }
 }

crates/autopilot-svm/src/infra/mod.rs

@@ -1,5 +1,6 @@
 //! Infrastructure: database access, driver clients, and the loop seams.
 
+pub mod cleanup;
 pub mod competition;
 pub mod config;
 pub mod db;

crates/autopilot-svm/src/run.rs

@@ -4,6 +4,7 @@ use {
     crate::{
         domain::arbitrator::SolanaArbitrator,
         infra::{
+            cleanup::Cleanup,
             competition::DriverCompetition,
             config::{self, Config},
             db,
@@ -87,6 +88,15 @@ async fn run(config: Config) {
         .await
         .expect("database connection");
 
+    tokio::spawn(
+        Cleanup::new(
+            pool.clone(),
+            config.db_cleanup.cleanup_interval,
+            config.db_cleanup.cleanup_threshold,
+        )
+        .run_forever(),
+    );
+
     let windows = SettlementWindows::new(pool.clone());
     let listen = ListenSession::spawn(
         pool.clone(),

crates/solana-orderbook/openapi.yml

@@ -522,7 +522,7 @@ components:
           type: string
           format: date-time
         id:
-          description: The quote's database id. Null until quotes are persisted.
+          description: The stored quote's id. Null when storing the quote failed.
           type: integer
           format: int64
           nullable: true

crates/solana-orderbook/src/infra/api/routes/create_order/mod.rs

@@ -20,7 +20,7 @@ use {
         instruction::{InstructionInputParsing, create_order::CreateOrderInput},
         pda::{order::find_order_pda, state::find_state_pda},
     },
-    database::solana::OrderKind,
+    database::{byte_array::ByteArray, solana::OrderKind},
     serde::Deserialize,
     serde_with::{base64::Base64, serde_as},
     solana_sdk::{
@@ -156,7 +156,7 @@ pub async fn create_order(
     // Short-circuit replays with a cheap read before the insert. A replayed
     // transaction usually dies at the blockhash check already, and the
     // insert's unique violation stays as the race-safe backstop.
-    let duplicate = db::order_exists(state.pool(), &order.uid)
+    let duplicate = db::order_exists(state.pool(), &order.uid.0)
         .await
         .map_err(|err| internal_error_reply(err, "order existence check failed"))?;
     if duplicate {
@@ -174,7 +174,7 @@ pub async fn create_order(
         }
         return Err(internal_error_reply(err, "sponsored order insert failed"));
     }
-    Ok((StatusCode::CREATED, Json(const_hex::encode_prefixed(uid))))
+    Ok((StatusCode::CREATED, Json(const_hex::encode_prefixed(uid.0))))
 }
 
 /// Check the transaction is exactly the sponsored-creation shape and derive
@@ -485,12 +485,12 @@ fn build_order(
     order_pda: Pubkey,
 ) -> db::SponsoredOrder {
     db::SponsoredOrder {
-        uid: uid.to_bytes(),
-        owner: intent.owner.to_bytes(),
-        sell_token: intent.sell_mint.to_bytes(),
-        buy_token: intent.buy_mint.to_bytes(),
-        sell_token_account: intent.sell_token_account.to_bytes(),
-        buy_token_account: intent.buy_token_account.to_bytes(),
+        uid: ByteArray(uid.to_bytes()),
+        owner: ByteArray(intent.owner.to_bytes()),
+        sell_token: ByteArray(intent.sell_mint.to_bytes()),
+        buy_token: ByteArray(intent.buy_mint.to_bytes()),
+        sell_token_account: ByteArray(intent.sell_token_account.to_bytes()),
+        buy_token_account: ByteArray(intent.buy_token_account.to_bytes()),
         sell_amount: intent.sell_amount,
         buy_amount: intent.buy_amount,
         valid_to: intent.valid_to,
@@ -499,8 +499,8 @@ fn build_order(
             IntentOrderKind::Buy => OrderKind::Buy,
         },
         partially_fillable: intent.flags.partially_fillable,
-        app_data: intent.app_data,
-        order_pda: order_pda.to_bytes(),
+        app_data: ByteArray(intent.app_data),
+        order_pda: ByteArray(order_pda.to_bytes()),
         presigned_transaction: Vec::new(),
         last_valid_block_height: 0,
     }

crates/solana-orderbook/src/infra/api/routes/quote/dto.rs

@@ -107,7 +107,7 @@ pub struct Response {
     pub from: Pubkey,
     /// When the quoted amounts stop being honored.
     pub expiration: DateTime<Utc>,
-    /// The quote's database id. Always absent: quotes are not persisted.
+    /// The stored quote's id, absent when the store was unavailable.
     pub id: Option<i64>,
     /// Whether the amounts were confirmed by simulating the settlement. No
     /// component simulates, so a quote is indicative.

crates/solana-orderbook/src/infra/api/routes/quote/mod.rs

@@ -5,16 +5,22 @@ pub mod dto;
 use {
     crate::infra::{
         api::{State, ValidationParameters, error, extract},
+        db,
         quoter,
     },
     axum::{Json, http::StatusCode},
     chrono::Utc,
+    database::{byte_array::ByteArray, solana::OrderKind},
     std::time::Duration,
 };
 
 /// How long a quoted order stays valid when the request names no validity.
 const DEFAULT_VALIDITY: Duration = Duration::from_secs(30 * 60);
 
+/// TODO: fail the quote on a store error instead, like the EVM orderbook,
+/// once fee policies consume the link and the id becomes load-bearing.
+const SAVE_TIMEOUT: Duration = Duration::from_secs(3);
+
 /// Handle `POST /api/v1/quote`.
 pub async fn quote(
     state: axum::extract::State<State>,
@@ -49,6 +55,35 @@ pub async fn quote(
             error::reply(StatusCode::NOT_FOUND, "NoLiquidity", "no route found")
         })?;
 
+    let expiration = now + state.quote_expiry();
+    // A failed insert answers without an id instead of failing the quote,
+    // since the fees are not yet implemented and stored quote are not mandatory
+    // at the moment.
+    let quote = db::Quote {
+        sell_token: ByteArray(request.sell_token.to_bytes()),
+        buy_token: ByteArray(request.buy_token.to_bytes()),
+        sell_amount: quoted.sell_amount,
+        buy_amount: quoted.buy_amount,
+        kind: match kind {
+            dto::Kind::Sell => OrderKind::Sell,
+            dto::Kind::Buy => OrderKind::Buy,
+        },
+        solver: ByteArray(quoted.solver.to_bytes()),
+        expiration,
+    };
+    let save = db::save_quote(state.pool(), &quote);
+    let id = match tokio::time::timeout(SAVE_TIMEOUT, save).await {
+        Ok(Ok(id)) => Some(id),
+        Ok(Err(err)) => {
+            tracing::error!(?err, "quote insert failed");
+            None
+        }
+        Err(_) => {
+            tracing::error!("quote insert timed out");
+            None
+        }
+    };
+
     Ok(Json(dto::Response {
         quote: dto::Quote {
             sell_token: request.sell_token,
@@ -63,8 +98,8 @@ pub async fn quote(
             partially_fillable: false,
         },
         from: request.from,
-        expiration: now + state.quote_expiry(),
-        id: None,
+        expiration,
+        id,
         verified: false,
     }))
 }