Add tx-granular ledger snapshot source using meta, RPC, and archives

Cargo.toml

@@ -4,19 +4,28 @@ resolver = "2"
 members = [
     "soroban-sdk",
     "soroban-sdk-macros",
+    "cache-to-file",
     "soroban-meta",
     "soroban-spec",
     "soroban-spec-rust",
     "soroban-ledger-snapshot",
     "soroban-token-sdk",
     "soroban-token-spec",
     "stellar-asset-spec",
+
+    "soroban-ledger-fetch-from/rpc",
+    "soroban-ledger-fetch-from/meta-storage",
+    "soroban-ledger-fetch-from/history-archive",
+    "soroban-ledger-fetch",
+    "soroban-ledger-snapshot-source-live",
+    "soroban-ledger-snapshot-source-tx",
+
     "tests/*",
 ]
 
 [workspace.package]
 version = "23.4.0"
-rust-version = "1.84.0"
+rust-version = "1.85.0"
 
 [workspace.dependencies]
 soroban-sdk = { version = "23.4.0", path = "soroban-sdk" }
@@ -29,6 +38,14 @@ soroban-token-sdk = { version = "23.4.0", path = "soroban-token-sdk" }
 soroban-token-spec = { version = "23.4.0", path = "soroban-token-spec" }
 stellar-asset-spec = { version = "23.4.0", path = "stellar-asset-spec" }
 
+cache-to-file = { version = "0.0.1", path = "cache-to-file" }
+soroban-ledger-fetch-from-rpc = { version = "23.4.0", path = "soroban-ledger-fetch-from/rpc" }
+soroban-ledger-fetch-from-meta-storage = { version = "23.4.0", path = "soroban-ledger-fetch-from/meta-storage" }
+soroban-ledger-fetch-from-history-archive = { version = "23.4.0", path = "soroban-ledger-fetch-from/history-archive" }
+soroban-ledger-fetch = { version = "23.4.0", path = "soroban-ledger-fetch" }
+soroban-ledger-snapshot-source-live = { version = "23.4.0", path = "soroban-ledger-snapshot-source-live" }
+soroban-ledger-snapshot-source-tx = { version = "23.4.0", path = "soroban-ledger-snapshot-source-tx" }
+
 [workspace.dependencies.soroban-env-common]
 version = "=23.0.1"
 #git = "https://github.com/stellar/rs-soroban-env"

cache-to-file/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+name = "cache-to-file"
+description = "File caching utilities with locking"
+homepage = "https://github.com/stellar/rs-soroban-sdk"
+repository = "https://github.com/stellar/rs-soroban-sdk"
+authors = ["Stellar Development Foundation <info@stellar.org>"]
+readme = "../README.md"
+license = "Apache-2.0"
+version = "0.0.1"
+edition = "2021"
+rust-version.workspace = true
+
+[dependencies]
+thiserror = "2.0"
+fs2 = "0.4"

cache-to-file/src/lib.rs

@@ -0,0 +1,49 @@
+use std::fs::{self, File};
+use std::io::{Read, Write};
+use std::path::Path;
+
+use fs2::FileExt;
+
+// TODO: Move this crate to its own repo.
+
+/// Cache a file, collecting bytes if not present, and return a reader
+pub fn cache<P, C, CE>(path: P, collect: C) -> Result<impl Read, CacheError<CE>>
+where
+    P: AsRef<Path>,
+    C: FnOnce(&mut dyn Write) -> Result<(), CE>,
+{
+    let path = path.as_ref();
+
+    // Fast path: if file already exists, just read it (no lock needed).
+    // Slow path: acquire lock, check again, collect and write if needed.
+    if !path.exists() {
+        let lock_path = path.with_extension("lock");
+        let lock_file = File::create(&lock_path).map_err(CacheError::Io)?;
+        lock_file.lock_exclusive().map_err(CacheError::Io)?;
+
+        if !path.exists() {
+            // Write atomically via temp file and rename.
+            let temp_path = path.with_extension("dl");
+            {
+                let mut temp = File::create(&temp_path).map_err(CacheError::Io)?;
+                collect(&mut temp).map_err(CacheError::Collector)?;
+                temp.sync_all().map_err(CacheError::Io)?;
+            }
+            fs::rename(&temp_path, path).map_err(CacheError::Io)?;
+        }
+    }
+
+    // Open data file for reading.
+    let file = File::open(path).map_err(CacheError::Io)?;
+    Ok(file)
+}
+
+/// Error type for cache operations
+#[derive(Debug, thiserror::Error)]
+pub enum CacheError<CE> {
+    #[error("collector error: {0}")]
+    Collector(CE),
+    #[error("io error: {0}")]
+    #[allow(dead_code)]
+    Io(#[from] std::io::Error),
+}

soroban-ledger-fetch-from/history-archive/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "soroban-ledger-fetch-from-history-archive"
+description = "Client for accessing Stellar History Archives to retrieve ledger snapshots."
+homepage = "https://github.com/stellar/rs-soroban-sdk"
+repository = "https://github.com/stellar/rs-soroban-sdk"
+authors = ["Stellar Development Foundation <info@stellar.org>"]
+readme = "../README.md"
+license = "Apache-2.0"
+version.workspace = true
+edition = "2021"
+rust-version.workspace = true
+
+[dependencies]
+stellar-xdr = { workspace = true, features = ["curr", "std"] }
+thiserror = "1.0"
+reqwest = { version = "0.12", default-features = false, features = ["json", "blocking", "rustls-tls"] }
+flate2 = { version = "1.0", features = ["rust_backend"] }
+serde = { version = "1.0.0", features = ["derive"] }
+serde_json = "1.0.0"

soroban-ledger-fetch-from/history-archive/src/lib.rs

@@ -0,0 +1,117 @@
+use flate2::read::GzDecoder;
+use std::io::{self, copy, Cursor, Write};
+use stellar_xdr::curr::{self as xdr, Frame, Limited, ReadXdr};
+
+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+    #[error("downloading history: {0}")]
+    DownloadingHistory(reqwest::Error),
+
+    #[error("downloading history: got status code {0}")]
+    DownloadingHistoryGotStatusCode(reqwest::StatusCode),
+
+    #[error("json decoding history: {0}")]
+    JsonDecodingHistory(serde_json::Error),
+
+    #[error("getting bucket: {0}")]
+    GettingBucket(reqwest::Error),
+
+    #[error("getting bucket: got status code {0}")]
+    GettingBucketGotStatusCode(reqwest::StatusCode),
+
+    #[error("streaming bucket: {0}")]
+    StreamingBucket(io::Error),
+
+    #[error("streaming history: {0}")]
+    StreamingHistory(io::Error),
+
+    #[error("xdr parsing error: {0}")]
+    Xdr(#[from] xdr::Error),
+}
+
+pub fn history(archive_url: &str, ledger: u32) -> Result<History, Error> {
+    let mut bytes = Vec::new();
+    get_history(archive_url, ledger, &mut bytes)?;
+    parse_history(Cursor::new(bytes))
+}
+
+pub fn get_history<W: Write + ?Sized>(
+    archive_url: &str,
+    ledger: u32,
+    writer: &mut W,
+) -> Result<(), Error> {
+    let history_url = {
+        let ledger_hex = format!("{ledger:08x}");
+        let ledger_hex_0 = ledger_hex[0..=1].to_string();
+        let ledger_hex_1 = ledger_hex[2..=3].to_string();
+        let ledger_hex_2 = ledger_hex[4..=5].to_string();
+        format!("{archive_url}/history/{ledger_hex_0}/{ledger_hex_1}/{ledger_hex_2}/history-{ledger_hex}.json")
+    };
+    //eprintln!("url: {history_url}");
+
+    let mut response = reqwest::blocking::Client::new()
+        .get(&history_url)
+        .send()
+        .map_err(Error::DownloadingHistory)?;
+
+    if !response.status().is_success() {
+        return Err(Error::DownloadingHistoryGotStatusCode(response.status()));
+    }
+
+    copy(&mut response, writer).map_err(Error::StreamingHistory)?;
+    Ok(())
+}
+
+pub fn parse_history<R: std::io::Read>(reader: R) -> Result<History, Error> {
+    let history: History = serde_json::from_reader(reader).map_err(Error::JsonDecodingHistory)?;
+    Ok(history)
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct History {
+    pub current_ledger: u32,
+    pub current_buckets: Vec<HistoryBucket>,
+    pub hot_archive_buckets: Option<Vec<HistoryBucket>>,
+    pub network_passphrase: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct HistoryBucket {
+    pub curr: String,
+    pub snap: String,
+}
+
+pub fn get_bucket<W: Write + ?Sized>(
+    archive_url: &str,
+    bucket: &str,
+    writer: &mut W,
+) -> Result<Option<u64>, Error> {
+    let bucket_0 = &bucket[0..=1];
+    let bucket_1 = &bucket[2..=3];
+    let bucket_2 = &bucket[4..=5];
+    let bucket_url =
+        format!("{archive_url}/bucket/{bucket_0}/{bucket_1}/{bucket_2}/bucket-{bucket}.xdr.gz");
+
+    let response = reqwest::blocking::Client::new()
+        .get(&bucket_url)
+        .send()
+        .map_err(Error::GettingBucket)?;
+
+    if !response.status().is_success() {
+        return Err(Error::GettingBucketGotStatusCode(response.status()));
+    }
+
+    let content_length = response.content_length();
+
+    let mut decoder = GzDecoder::new(response);
+    copy(&mut decoder, writer).map_err(Error::StreamingBucket)?;
+    Ok(content_length)
+}
+
+pub fn parse_bucket<'a, R: std::io::Read + 'a>(
+    reader: &'a mut Limited<R>,
+) -> impl Iterator<Item = Result<Frame<xdr::BucketEntry>, xdr::Error>> + 'a {
+    Frame::<xdr::BucketEntry>::read_xdr_iter(reader)
+}

soroban-ledger-fetch-from/meta-storage/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "soroban-ledger-fetch-from-meta-storage"
+description = "Download LedgerCloseMeta from SEP-54 compatible ledger storage."
+homepage = "https://github.com/stellar/rs-soroban-sdk"
+repository = "https://github.com/stellar/rs-soroban-sdk"
+authors = ["Stellar Development Foundation <info@stellar.org>"]
+readme = "../README.md"
+license = "Apache-2.0"
+version.workspace = true
+edition = "2021"
+rust-version.workspace = true
+
+[features]
+default = []
+cli = ["dep:clap"]
+
+[[bin]]
+name = "soroban-ledger-meta-storage"
+path = "src/bin/soroban-ledger-meta-storage/main.rs"
+required-features = ["cli"]
+
+[dependencies]
+stellar-xdr = { workspace = true, features = ["curr", "std", "serde"] }
+thiserror = "1.0"
+reqwest = { version = "0.12", default-features = false, features = ["blocking", "rustls-tls"] }
+zstd = "0.13"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+clap = { version = "4.0", features = ["derive"], optional = true }

soroban-ledger-fetch-from/meta-storage/src/bin/soroban-ledger-meta-storage/main.rs

@@ -0,0 +1,14 @@
+use soroban_ledger_fetch_from_meta_storage::cli;
+use std::env;
+
+fn main() {
+    if let Err(e) = cli::run(env::args_os()) {
+        match e {
+            cli::Error::Clap(e) => e.exit(),
+            _ => {
+                eprintln!("Error: {e}");
+                std::process::exit(1);
+            }
+        }
+    }
+}

soroban-ledger-fetch-from/meta-storage/src/cli/mod.rs

@@ -0,0 +1,96 @@
+//! CLI module for downloading LedgerCloseMeta from Stellar public blockchain data.
+//!
+//! This module is only available when the `cli` feature is enabled.
+
+use crate::ledger;
+use clap::{Parser, ValueEnum};
+use std::{ffi::OsString, fmt::Debug, io::Write};
+use stellar_xdr::curr::{Limited, Limits, WriteXdr};
+
+/// CLI tool to download LedgerCloseMeta from Stellar public blockchain data
+#[derive(Parser, Debug, Clone)]
+#[command(
+    author,
+    version,
+    about,
+    long_about = None,
+    disable_help_subcommand = true,
+)]
+pub struct Root {
+    /// Ledger sequence number to download
+    #[arg(short, long)]
+    sequence: u32,
+
+    /// Output format
+    #[arg(short, long, value_enum, default_value_t)]
+    format: Format,
+
+    /// Meta storage URL (SEP-54 compatible)
+    #[arg(
+        long,
+        default_value = "https://aws-public-blockchain.s3.us-east-2.amazonaws.com/v1.1/stellar/ledgers/pubnet"
+    )]
+    meta_url: String,
+}
+
+#[derive(Default, Clone, ValueEnum, Debug)]
+pub enum Format {
+    #[default]
+    Json,
+    Xdr,
+}
+
+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+    #[error(transparent)]
+    Clap(#[from] clap::Error),
+    #[error("failed to download ledger meta: {0}")]
+    Download(#[from] crate::Error),
+    #[error("failed to serialize to JSON: {0}")]
+    Json(#[from] serde_json::Error),
+    #[error("failed to write XDR: {0}")]
+    Xdr(#[from] stellar_xdr::curr::Error),
+    #[error("I/O error: {0}")]
+    Io(#[from] std::io::Error),
+}
+
+impl Root {
+    /// Run the CLI command.
+    ///
+    /// ## Errors
+    ///
+    /// If the command fails to execute.
+    pub fn run(&self) -> Result<(), Error> {
+        let meta = ledger(&self.meta_url, self.sequence)?;
+
+        match self.format {
+            Format::Json => {
+                let json = serde_json::to_string_pretty(&meta)?;
+                println!("{}", json);
+            }
+            Format::Xdr => {
+                let limits = Limits::none();
+                let mut stdout = std::io::stdout();
+                let mut limited_writer = Limited::new(&mut stdout, limits);
+                meta.write_xdr(&mut limited_writer)?;
+                stdout.flush()?;
+            }
+        }
+
+        Ok(())
+    }
+}
+
+/// Run the CLI with the given args.
+///
+/// ## Errors
+///
+/// If the input cannot be parsed or the command fails.
+pub fn run<I, T>(args: I) -> Result<(), Error>
+where
+    I: IntoIterator<Item = T>,
+    T: Into<OsString> + Clone,
+{
+    let root = Root::try_parse_from(args)?;
+    root.run()
+}