Store auction with liquidity (#2663)

# Description
Store auction instances from one of the solvers which receives
liquidity, similarly to how it was done for the legacy format. Auction
instances from the autopilot are not enough since they do not contain
liquidity.

## How to test
1. Regression tests
2. It can be tested in staging by configuring quasidomo and checking the
S3 bucket

Fixes #2633

crates/autopilot/src/infra/persistence/dto/order.rs

@@ -258,7 +258,6 @@ impl From<boundary::EcdsaSignature> for domain::auction::order::EcdsaSignature {
     }
 }
 
-#[serde_as]
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub enum FeePolicy {

crates/driver/Cargo.toml

@@ -17,6 +17,7 @@ path = "src/main.rs"
 [dependencies]
 app-data = { path = "../app-data" }
 bytes-hex = { path = "../bytes-hex" }
+s3 = { path = "../s3" }
 async-trait = { workspace = true }
 axum = { workspace = true }
 bigdecimal = { workspace = true }

crates/driver/src/domain/competition/auction.rs

@@ -4,7 +4,7 @@ use {
         domain::{
             competition::{self, auction},
             eth,
-            liquidity,
+            liquidity::{self},
             time,
         },
         infra::{self, blockchain, observe, Ethereum},

crates/driver/src/infra/config/file/load.rs

@@ -89,6 +89,7 @@ pub async fn load(chain: eth::ChainId, path: &Path) -> infra::Config {
                     true => SolutionMerging::Allowed,
                     false => SolutionMerging::Forbidden,
                 },
+                s3: config.s3.map(Into::into),
             }
         }))
         .await,

crates/driver/src/infra/config/file/mod.rs

@@ -225,6 +225,11 @@ struct SolverConfig {
     /// auction together.
     #[serde(default)]
     merge_solutions: bool,
+
+    /// S3 configuration for storing the auctions in the form they are sent to
+    /// the solver engine
+    #[serde(default)]
+    s3: Option<S3>,
 }
 
 #[derive(Clone, Copy, Debug, Default, Deserialize, PartialEq, Serialize)]
@@ -235,6 +240,17 @@ pub enum FeeHandler {
     Solver,
 }
 
+#[derive(Clone, Debug, Default, Deserialize)]
+#[serde(rename_all = "kebab-case", deny_unknown_fields)]
+pub struct S3 {
+    /// Name of the AWS S3 bucket in which the auctions will be stored
+    pub bucket: String,
+
+    /// Prepended to the auction id to form the final instance filename on AWS
+    /// S3 bucket. Something like "staging/mainnet/"
+    pub prefix: String,
+}
+
 #[serde_as]
 #[derive(Debug, Deserialize)]
 #[serde(untagged)]

crates/driver/src/infra/mod.rs

@@ -6,6 +6,7 @@ pub mod liquidity;
 pub mod mempool;
 pub mod notify;
 pub mod observe;
+pub mod persistence;
 pub mod simulator;
 pub mod solver;
 pub mod time;

crates/driver/src/infra/persistence/mod.rs

@@ -0,0 +1,75 @@
+use {
+    crate::{
+        domain::competition::auction::Id,
+        infra::{config::file, solver::Config},
+    },
+    std::sync::Arc,
+    tracing::Instrument,
+};
+
+#[derive(Clone, Debug, Default)]
+pub struct S3 {
+    /// Name of the AWS S3 bucket in which the auctions will be stored
+    pub bucket: String,
+
+    /// Prepended to the auction id to form the final instance filename on AWS
+    /// S3 bucket. Something like "staging/mainnet/"
+    pub prefix: String,
+}
+
+impl From<file::S3> for S3 {
+    fn from(value: file::S3) -> Self {
+        Self {
+            bucket: value.bucket,
+            prefix: value.prefix,
+        }
+    }
+}
+
+impl From<S3> for s3::Config {
+    fn from(value: S3) -> Self {
+        Self {
+            bucket: value.bucket,
+            filename_prefix: value.prefix,
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub struct Persistence {
+    s3: Option<Arc<s3::Uploader>>,
+}
+
+impl Persistence {
+    pub async fn build(config: &Config) -> Self {
+        if let Some(s3) = &config.s3 {
+            Self {
+                s3: Some(Arc::new(s3::Uploader::new(s3.clone().into()).await)),
+            }
+        } else {
+            Self { s3: None }
+        }
+    }
+
+    /// Saves the given auction with liquidity with fire and forget mentality
+    /// (non-blocking operation)
+    pub fn archive_auction(&self, auction_id: Id, body: &str) {
+        let Some(uploader) = self.s3.clone() else {
+            return;
+        };
+        let body = body.to_string();
+        tokio::spawn(
+            async move {
+                match uploader.upload(auction_id.to_string(), body).await {
+                    Ok(key) => {
+                        tracing::debug!(?key, "uploaded auction with liquidity to s3");
+                    }
+                    Err(err) => {
+                        tracing::warn!(?err, "failed to upload auction to s3");
+                    }
+                }
+            }
+            .instrument(tracing::Span::current()),
+        );
+    }
+}

crates/driver/src/infra/solver/mod.rs

@@ -10,7 +10,11 @@ use {
             liquidity,
             time::Remaining,
         },
-        infra::{blockchain::Ethereum, config::file::FeeHandler},
+        infra::{
+            blockchain::Ethereum,
+            config::file::FeeHandler,
+            persistence::{Persistence, S3},
+        },
         util,
     },
     anyhow::Result,
@@ -85,6 +89,7 @@ pub struct Solver {
     client: reqwest::Client,
     config: Config,
     eth: Ethereum,
+    persistence: Persistence,
 }
 
 #[derive(Debug, Clone)]
@@ -108,10 +113,13 @@ pub struct Config {
     /// TODO: Remove once all solvers are moved to use limit orders for quoting
     pub quote_using_limit_orders: bool,
     pub merge_solutions: SolutionMerging,
+    /// S3 configuration for storing the auctions in the form they are sent to
+    /// the solver engine
+    pub s3: Option<S3>,
 }
 
 impl Solver {
-    pub fn new(config: Config, eth: Ethereum) -> Result<Self> {
+    pub async fn new(config: Config, eth: Ethereum) -> Result<Self> {
         let mut headers = reqwest::header::HeaderMap::new();
         headers.insert(
             reqwest::header::CONTENT_TYPE,
@@ -124,15 +132,22 @@ impl Solver {
             headers.insert(header_name, val.parse()?);
         }
 
+        let persistence = Persistence::build(&config).await;
+
         Ok(Self {
             client: reqwest::ClientBuilder::new()
                 .default_headers(headers)
                 .build()?,
             config,
             eth,
+            persistence,
         })
     }
 
+    pub fn persistence(&self) -> Persistence {
+        self.persistence.clone()
+    }
+
     pub fn name(&self) -> &Name {
         &self.config.name
     }
@@ -187,6 +202,11 @@ impl Solver {
             self.config.fee_handler,
         ))
         .unwrap();
+        // Only auctions with IDs are real auctions (/quote requests don't have an ID,
+        // and it makes no sense to store them)
+        if let Some(id) = auction.id() {
+            self.persistence.archive_auction(id, &body);
+        };
         let url = shared::url::join(&self.config.endpoint, "solve");
         super::observe::solver_request(&url, &body);
         let mut req = self

crates/driver/src/run.rs

@@ -13,6 +13,7 @@ use {
         },
     },
     clap::Parser,
+    futures::future::join_all,
     std::{net::SocketAddr, sync::Arc, time::Duration},
     tokio::sync::oneshot,
 };
@@ -51,7 +52,7 @@ async fn run_with(args: cli::Args, addr_sender: Option<oneshot::Sender<SocketAdd
     let (shutdown_sender, shutdown_receiver) = tokio::sync::oneshot::channel();
     let eth = ethereum(&config, ethrpc).await;
     let serve = Api {
-        solvers: solvers(&config, &eth),
+        solvers: solvers(&config, &eth).await,
         liquidity: liquidity(&config, &eth).await,
         simulator: simulator(&config, &eth),
         mempools: Mempools::new(
@@ -134,12 +135,15 @@ async fn ethereum(config: &infra::Config, ethrpc: blockchain::Rpc) -> Ethereum {
     Ethereum::new(ethrpc, config.contracts, gas).await
 }
 
-fn solvers(config: &config::Config, eth: &Ethereum) -> Vec<Solver> {
-    config
-        .solvers
-        .iter()
-        .map(|config| Solver::new(config.clone(), eth.clone()).unwrap())
-        .collect()
+async fn solvers(config: &config::Config, eth: &Ethereum) -> Vec<Solver> {
+    join_all(
+        config
+            .solvers
+            .iter()
+            .map(|config| async move { Solver::new(config.clone(), eth.clone()).await.unwrap() })
+            .collect::<Vec<_>>(),
+    )
+    .await
 }
 
 async fn liquidity(config: &config::Config, eth: &Ethereum) -> liquidity::Fetcher {

crates/s3/src/lib.rs

@@ -15,7 +15,7 @@ pub struct Config {
     pub filename_prefix: String,
 }
 
-#[derive(Clone)]
+#[derive(Debug, Clone)]
 pub struct Uploader {
     bucket: String,
     filename_prefix: String,