Partial sync, propagate tx

docs/openapi.json

@@ -473,6 +473,59 @@
         }
       }
     },
+    "/transactions/submit": {
+      "post": {
+        "tags": [
+          "Transactions"
+        ],
+        "summary": "Propagate Transaction",
+        "description": "Propagate a transaction to the Bitcoin network. If the transaction is not valid it will fail\nsilently as it is directly passed to a peer via P2P.",
+        "operationId": "submit_transaction",
+        "requestBody": {
+          "description": "Transaction to submit",
+          "content": {
+            "application/json": {
+              "schema": {
+                "$ref": "#/components/schemas/SubmitTransactionRequest"
+              }
+            }
+          },
+          "required": true
+        },
+        "responses": {
+          "200": {
+            "description": "Transaction submitted successfully",
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/ServeResponse_SubmitTransactionResponse"
+                },
+                "example": {
+                  "data": {
+                    "status": "submitted",
+                    "txid": "c0345bb5906257a05cdc2d11b6580ce75fdfe8b7ac09b7b2711d435e2ba0a9b3"
+                  },
+                  "indexer_info": {
+                    "chain_tip": {
+                      "block_hash": "00000000000000108a4cd9755381003a01bea7998ca2d770fe09b576753ac7ef",
+                      "block_height": 31633
+                    },
+                    "estimated_blocks": [],
+                    "mempool_timestamp": null
+                  }
+                }
+              }
+            }
+          },
+          "400": {
+            "description": "Invalid transaction data"
+          },
+          "500": {
+            "description": "Internal server error"
+          }
+        }
+      }
+    },
     "/{rune}/utxos/{utxo}/balance": {
       "get": {
         "tags": [
@@ -933,6 +986,35 @@
           }
         }
       },
+      "ServeResponse_SubmitTransactionResponse": {
+        "type": "object",
+        "required": [
+          "data",
+          "indexer_info"
+        ],
+        "properties": {
+          "data": {
+            "type": "object",
+            "required": [
+              "txid",
+              "status"
+            ],
+            "properties": {
+              "status": {
+                "type": "string",
+                "description": "Status of the submission"
+              },
+              "txid": {
+                "type": "string",
+                "description": "Transaction ID of the submitted transaction"
+              }
+            }
+          },
+          "indexer_info": {
+            "$ref": "#/components/schemas/IndexerInfo"
+          }
+        }
+      },
       "ServeResponse_Vec_AddressUtxo": {
         "type": "object",
         "required": [
@@ -1114,6 +1196,35 @@
             "$ref": "#/components/schemas/IndexerInfo"
           }
         }
+      },
+      "SubmitTransactionRequest": {
+        "type": "object",
+        "required": [
+          "raw_tx"
+        ],
+        "properties": {
+          "raw_tx": {
+            "type": "string",
+            "description": "Raw transaction bytes encoded as hex string"
+          }
+        }
+      },
+      "SubmitTransactionResponse": {
+        "type": "object",
+        "required": [
+          "txid",
+          "status"
+        ],
+        "properties": {
+          "status": {
+            "type": "string",
+            "description": "Status of the submission"
+          },
+          "txid": {
+            "type": "string",
+            "description": "Transaction ID of the submitted transaction"
+          }
+        }
       }
     }
   }

examples/testnet3_partial.toml

@@ -0,0 +1,23 @@
+db_path = "tmp/symphony"
+
+[sync.node]
+p2p_address = "137.184.2.124:18333"
+# RPC user
+rpc_user = "..."
+# RPC password
+rpc_pass = "..."
+
+[sync]
+intersect = [4605959, "0000000010943e6e27a31f59d9e7595dc88e81a51547fd1e2d076b90c6613082"]
+network = "testnet3"
+max_rollback = 32
+mempool = true
+
+[sync.indexers]
+transaction_indexers = [
+    { type = "UtxosByAddress" },
+    { type = "Runes", start_height = 30562, index_activity = true }
+]
+
+[server]
+address = "0.0.0.0:8080"

src/error.rs

@@ -25,7 +25,7 @@ pub enum Error {
     #[error("{0}")]
     P2P(#[from] P2PError),
 
-    #[error("{0:?}")]
+    #[error("missing utxo {0:?}")]
     MissingUtxo(TxoRef),
 
     #[error("invalid merge operator combination")]

src/serve/mod.rs

@@ -153,6 +153,7 @@ pub async fn run(db: StorageHandler, address: &str) -> Result<(), Error> {
         .route("/tip", get(tip))
         .nest("/addresses", routes::addresses::router())
         .nest("/runes", routes::runes::router())
+        .nest("/transactions", routes::transactions::router())
         .layer(middleware::from_fn_with_state(
             app_state.clone(),
             auto_refresh,
@@ -231,3 +232,6 @@ async fn tip(State(state): State<AppState>) -> impl IntoResponse {
 
     json.into_response()
 }
+
+// Re-export the transaction submission function
+pub use routes::transactions::set_tx_submission_sender;

src/serve/openapi.rs

@@ -27,6 +27,7 @@ use utoipa::OpenApi;
         addresses::tx_count_by_address::addresses_tx_count_by_address,
         runes::rune_info_batch::runes_rune_info_batch,
         runes::rune_balance_at_utxo::rune_balance_at_utxo,
+        transactions::submit_transaction,
     ),
     components(schemas(
         IndexerInfo,
@@ -40,12 +41,15 @@ use utoipa::OpenApi;
         ServeResponse<String>,
         ServeResponse<AddressUtxo>,
         ServeResponse<HashMap<String, Option<RuneInfo>>>,
+        ServeResponse<super::routes::transactions::SubmitTransactionResponse>,
         // ---
         RuneAndAmount,
         RuneUtxo,
         RuneEdict,
         AddressUtxo,
         RuneInfo,
+        super::routes::transactions::SubmitTransactionRequest,
+        super::routes::transactions::SubmitTransactionResponse,
     )),
 )]
 pub struct APIDoc;

src/serve/routes/mod.rs

@@ -1,2 +1,3 @@
 pub mod addresses;
 pub mod runes;
+pub mod transactions;

src/serve/routes/transactions.rs

@@ -0,0 +1,116 @@
+use crate::serve::AppState;
+use crate::serve::error::ServeError;
+use crate::serve::types::ServeResponse;
+use axum::{Json, Router, extract::State, http::StatusCode, response::IntoResponse, routing::post};
+use bitcoin::{Transaction, consensus::Decodable};
+use serde::{Deserialize, Serialize};
+use std::sync::OnceLock;
+use tokio::sync::mpsc;
+use utoipa::ToSchema;
+
+#[derive(Deserialize, ToSchema)]
+pub struct SubmitTransactionRequest {
+    /// Raw transaction bytes encoded as hex string
+    pub raw_tx: String,
+}
+
+#[derive(Serialize, ToSchema)]
+pub struct SubmitTransactionResponse {
+    /// Transaction ID of the submitted transaction
+    pub txid: String,
+    /// Status of the submission
+    pub status: String,
+}
+
+// Channel for sending transactions to the P2P layer
+static TX_SUBMISSION_CHANNEL: OnceLock<mpsc::UnboundedSender<Transaction>> = OnceLock::new();
+
+pub fn set_tx_submission_sender(sender: mpsc::UnboundedSender<Transaction>) {
+    TX_SUBMISSION_CHANNEL
+        .set(sender)
+        .expect("Transaction submission channel already set");
+}
+
+pub fn get_tx_submission_sender() -> Option<&'static mpsc::UnboundedSender<Transaction>> {
+    TX_SUBMISSION_CHANNEL.get()
+}
+
+#[utoipa::path(
+    tag = "Transactions",
+    post,
+    path = "/transactions/submit",
+    request_body(
+        content = SubmitTransactionRequest,
+        description = "Transaction to submit",
+        content_type = "application/json"
+    ),
+    responses(
+        (
+            status = 200,
+            description = "Transaction submitted successfully",
+            body = ServeResponse<SubmitTransactionResponse>,
+            example = json!({
+                "data": {
+                    "txid": "c0345bb5906257a05cdc2d11b6580ce75fdfe8b7ac09b7b2711d435e2ba0a9b3",
+                    "status": "submitted"
+                },
+                "indexer_info": {
+                    "chain_tip": {
+                        "block_hash": "00000000000000108a4cd9755381003a01bea7998ca2d770fe09b576753ac7ef",
+                        "block_height": 31633
+                    },
+                    "mempool_timestamp": null,
+                    "estimated_blocks": []
+                }
+            })
+        ),
+        (status = 400, description = "Invalid transaction data"),
+        (status = 500, description = "Internal server error"),
+    )
+)]
+/// Propagate Transaction
+///
+/// Propagate a transaction to the Bitcoin network. If the transaction is not valid it will fail
+/// silently as it is directly passed to a peer via P2P.
+pub async fn submit_transaction(
+    State(state): State<AppState>,
+    Json(request): Json<SubmitTransactionRequest>,
+) -> Result<impl IntoResponse, ServeError> {
+    // Parse the raw transaction hex
+    let tx_bytes = hex::decode(&request.raw_tx)
+        .map_err(|_| ServeError::malformed_request("invalid hex encoding"))?;
+
+    // Decode the transaction
+    let transaction = Transaction::consensus_decode_from_finite_reader(&mut &tx_bytes[..])
+        .map_err(|_| ServeError::malformed_request("invalid transaction data"))?;
+
+    let txid = transaction.compute_txid();
+
+    // Get the transaction submission channel
+    let sender = get_tx_submission_sender()
+        .ok_or_else(|| ServeError::internal("transaction submission not available"))?;
+
+    // Send the transaction to the P2P layer for broadcasting
+    sender
+        .send(transaction)
+        .map_err(|_| ServeError::internal("failed to queue transaction for submission"))?;
+
+    // Get indexer info for response
+    let (_, indexer_info) = state.start_reader(false).await?;
+
+    let response_data = SubmitTransactionResponse {
+        txid: txid.to_string(),
+        status: "submitted".to_string(),
+    };
+
+    let response = ServeResponse {
+        data: response_data,
+        indexer_info,
+    };
+
+    Ok((StatusCode::OK, Json(response)))
+}
+
+pub fn router() -> Router<AppState> {
+    Router::new().route("/submit", post(submit_transaction))
+}

src/sync/mod.rs

@@ -2,6 +2,8 @@ use bitcoin::{Block, consensus::Decodable, p2p::Magic};
 use serde::Deserialize;
 use stages::index::indexers::custom::TransactionIndexerFactory;
 
+use crate::sync::stages::Point;
+
 pub mod pipeline;
 pub mod stages;
 
@@ -21,6 +23,8 @@ pub struct Config {
     pub stage_queue_size: Option<usize>,
     pub stage_timeout_secs: Option<u64>,
 
+    pub intersect: Option<Point>,
+
     pub indexers: IndexersConfig,
 
     pub max_rollback: Option<usize>,
@@ -30,7 +34,7 @@ pub struct Config {
 #[derive(Deserialize, Debug, Clone)]
 pub struct NodeConfig {
     pub p2p_address: String,
-    pub rpc_address: String,
+    pub rpc_address: Option<String>,
     pub rpc_user: String,
     pub rpc_pass: String,
 }
@@ -45,6 +49,7 @@ pub struct IndexersConfig {
 #[serde(rename_all = "lowercase")]
 pub enum Network {
     Mainnet,
+    Testnet3,
     Testnet4,
     Regtest,
 }
@@ -53,6 +58,7 @@ impl Network {
     pub fn genesis_block(&self) -> Block {
         match self {
             Self::Mainnet => bitcoin::constants::genesis_block(bitcoin::Network::Bitcoin),
+            Self::Testnet3 => bitcoin::constants::genesis_block(bitcoin::Network::Testnet),
             Self::Testnet4 => {
                 let raw_block = hex::decode("0100000000000000000000000000000000000000000000000000000000000000000000004e7b2b9128fe0291db0693af2ae418b767e657cd407e80cb1434221eaea7a07a046f3566ffff001dbb0c78170101000000010000000000000000000000000000000000000000000000000000000000000000ffffffff5504ffff001d01044c4c30332f4d61792f323032342030303030303030303030303030303030303030303165626435386332343439373062336161396437383362623030313031316662653865613865393865303065ffffffff0100f2052a010000002321000000000000000000000000000000000000000000000000000000000000000000ac00000000").unwrap();
                 Block::consensus_decode_from_finite_reader(&mut &raw_block[..]).unwrap()
@@ -64,6 +70,7 @@ impl Network {
     pub fn magic(&self) -> Magic {
         match self {
             Self::Mainnet => bitcoin::Network::Bitcoin.magic(),
+            Self::Testnet3 => bitcoin::Network::Testnet.magic(),
             Self::Testnet4 => Magic::from_bytes([0x1c, 0x16, 0x3f, 0x28]),
             Self::Regtest => bitcoin::Network::Regtest.magic(),
         }

src/sync/pipeline.rs

@@ -40,15 +40,23 @@ pub fn pipeline(
 
     let rpc_auth = Auth::UserPass(config.node.rpc_user, config.node.rpc_pass);
 
+    // Initialize transaction submission channel
+    let (tx_sender, tx_receiver) = mpsc::unbounded_channel();
+
+    // Store the sender in the global static for the HTTP endpoint to use
+    crate::serve::set_tx_submission_sender(tx_sender);
+
     // create Pull stage for pulling blocks/mempool from node
     let mut pull = pull::Stage::new(
         config.node.p2p_address,
         config.node.rpc_address,
         rpc_auth,
         config.network,
         config.mempool,
+        config.intersect,
         db,
         shutdown_signals,
+        Some(tx_receiver),
     );
 
     // // create Health stage for exposing health info