feat: add a transaction api

Signed-off-by: Eric Torreborre <etorreborre@yahoo.com>

Author: etorreborre

Cargo.lock

@@ -23,6 +23,7 @@ acto = { version = "0.8.0", features = ["tokio"] }
 anyhow = "1.0.100"
 async-compression = { version = "0.4.32", features = ["tokio", "gzip"] }
 async-trait = "0.1.83"
+axum = "0.8"
 assert-json-diff = "2.0.2"
 bech32 = "0.11.0"
 binrw = "0.15.0"

Cargo.toml

@@ -31,6 +31,7 @@ doc = false
 [dependencies]
 # External dependencies ───────────────────────────────────────────────────────┐
 anyhow.workspace = true
+axum.workspace = true
 async-compression.workspace = true
 async-trait.workspace = true
 clap.workspace = true

crates/amaru/Cargo.toml

@@ -22,7 +22,8 @@ use amaru::{
         config::{Config, MaxExtraLedgerSnapshots, StoreType},
     },
 };
-use amaru_kernel::NetworkName;
+use amaru_kernel::{NetworkName, Transaction};
+use amaru_ouroboros::ResourceMempool;
 use amaru_stores::rocksdb::RocksDbConfig;
 use clap::{ArgAction, Parser};
 use opentelemetry_sdk::metrics::SdkMeterProvider;
@@ -105,6 +106,16 @@ pub struct Args {
     )]
     network: NetworkName,
 
+    /// Address for the HTTP transaction submit API.
+    ///
+    /// When set, starts an HTTP server exposing POST /api/submit/tx (Cardano Submit API).
+    #[arg(
+        long,
+        value_name = amaru::value_names::ENDPOINT,
+        env = amaru::env_vars::SUBMIT_API_ADDRESS,
+    )]
+    submit_api_address: Option<String>,
+
     /// Upstream peer addresses to synchronize from.
     ///
     /// This option can be specified multiple times to connect to multiple peers.
@@ -139,17 +150,22 @@ impl Args {
 pub async fn run(args: Args, meter_provider: Option<SdkMeterProvider>) -> Result<(), Box<dyn std::error::Error>> {
     with_optional_pid_file(args.pid_file.clone(), async |_pid_file| {
         let config = parse_args(args)?;
+        let submit_api_address = config.submit_api_address()?;
         pre_flight_checks()?;
 
         let metrics = meter_provider.clone().map(track_system_metrics).transpose()?;
+        let running = build_and_run_node(config, meter_provider)?;
 
         let exit = amaru::exit::hook_exit_token();
-
-        let running = build_and_run_node(config, meter_provider)?;
+        let submit_api_handle = start_submit_api(submit_api_address, &running, &exit).await?;
 
         exit.cancelled().await;
         running.abort();
 
+        if let Some(handle) = submit_api_handle {
+            let _ = handle.await; // Let graceful shutdown complete
+        }
+
         if let Some(handle) = metrics {
             handle.abort();
         }
@@ -159,6 +175,21 @@ pub async fn run(args: Args, meter_provider: Option<SdkMeterProvider>) -> Result
     .await
 }
 
+/// Start an HTTP API endpoint to allow local users to post CBOR-serialized transactions.
+async fn start_submit_api(
+    address: Option<std::net::SocketAddr>,
+    running: &pure_stage::tokio::TokioRunning,
+    exit: &tokio_util::sync::CancellationToken,
+) -> Result<Option<tokio::task::JoinHandle<()>>, Box<dyn std::error::Error>> {
+    let Some(addr) = address else {
+        return Ok(None);
+    };
+    let mempool: ResourceMempool<Transaction> = running.resources().get::<ResourceMempool<Transaction>>()?.clone();
+    let shutdown = exit.child_token();
+    let (handle, _) = amaru::submit_api::start(addr, mempool, shutdown).await?;
+    Ok(Some(handle))
+}
+
 fn parse_args(args: Args) -> Result<Config, Box<dyn std::error::Error>> {
     let network = args.network;
 
@@ -177,6 +208,7 @@ fn parse_args(args: Args) -> Result<Config, Box<dyn std::error::Error>> {
         network = %args.network,
         peer_address = %args.peer_address.iter().map(|s| s.as_str()).collect::<Vec<_>>().join(", "),
         pid_file = %args.pid_file.unwrap_or_default().to_string_lossy(),
+        submit_api_address = %args.submit_api_address.as_deref().unwrap_or("disabled"),
         "running"
     );
 
@@ -190,6 +222,7 @@ fn parse_args(args: Args) -> Result<Config, Box<dyn std::error::Error>> {
         max_downstream_peers: args.max_downstream_peers,
         max_extra_ledger_snapshots: args.max_extra_ledger_snapshots,
         migrate_chain_db: args.migrate_chain_db,
+        submit_api_address: args.submit_api_address,
         ..Config::default()
     })
 }

crates/amaru/src/bin/amaru/cmd/run.rs

@@ -32,6 +32,7 @@ pub mod tests;
 /// The sync pipeline is responsible for fetching blocks from the upstream node and
 /// applying them to the local chain.
 pub mod stages;
+pub mod submit_api;
 
 pub const SNAPSHOTS_DIR: &str = "snapshots";
 
@@ -167,6 +168,9 @@ pub mod env_vars {
     /// --snapshots-dir
     pub const SNAPSHOTS_DIR: &str = "AMARU_SNAPSHOTS_DIR";
 
+    /// --submit-api-address
+    pub const SUBMIT_API_ADDRESS: &str = "AMARU_SUBMIT_API_ADDRESS";
+
     /// --target-dir
     pub const TARGET_DIR: &str = "AMARU_TARGET_DIR";
 }

crates/amaru/src/lib.rs

@@ -30,6 +30,7 @@ pub struct Config {
     pub max_downstream_peers: usize,
     pub max_extra_ledger_snapshots: MaxExtraLedgerSnapshots,
     pub migrate_chain_db: bool,
+    pub submit_api_address: Option<String>,
 
     // Number of allocation arenas to keep around for performing parallel evaluation of scripts in
     // the ledger.
@@ -46,6 +47,11 @@ impl Config {
     pub fn listen_address(&self) -> anyhow::Result<SocketAddr> {
         self.listen_address.parse().context("invalid listen address")
     }
+
+    /// Parse the optional submit API address into a `SocketAddr`.
+    pub fn submit_api_address(&self) -> anyhow::Result<Option<SocketAddr>> {
+        self.submit_api_address.as_deref().map(|addr| addr.parse().context("invalid submit API address")).transpose()
+    }
 }
 
 impl Default for Config {
@@ -60,6 +66,7 @@ impl Default for Config {
             max_downstream_peers: 10,
             max_extra_ledger_snapshots: MaxExtraLedgerSnapshots::default(),
             migrate_chain_db: false,
+            submit_api_address: None,
             ledger_vm_alloc_arena_count: 1,
             ledger_vm_alloc_arena_size: 1_024_000,
         }