1use crate::config;
2use crate::error::{Context, Error, ensure};
3#[cfg(not(target_arch = "wasm32"))]
4use crate::native_fs;
5use crate::storage;
6use crate::transaction::decode;
7use crate::{
8 node::{Node, Presence},
9 plan::PublicationPlan,
10};
11use serde::{Deserialize, Serialize};
12use serde_json::json;
13use std::path::Path;
14use urma_chain::observation::Chain;
15
16#[derive(Clone, Serialize, Deserialize)]
17#[serde(deny_unknown_fields)]
18pub struct TransactionStatus {
19 pub txid: String,
20 pub presence: Presence,
21}
22
23#[derive(Clone, Serialize, Deserialize)]
24#[serde(deny_unknown_fields)]
25pub struct PublishReport {
26 pub plan_id: String,
27 pub root_txid: String,
28 pub complete: bool,
29 pub confirmed: bool,
30 pub blocked_reason: String,
31 pub transactions: Vec<TransactionStatus>,
32}
33
34pub(crate) fn store(path: &Path, report: &PublishReport) -> Result<(), Error> {
35 urma_io::write_replace(path, &serde_json::to_vec_pretty(report)?).map_err(Error::from)
36}
37
38#[derive(Clone, Debug, Serialize, Deserialize)]
39pub enum MempoolCheck {
40 Allowed,
41 Rejected(String),
42 Unavailable(String),
43}
44
45pub fn test_accept(node: &Node, raw: &str) -> Result<MempoolCheck, Error> {
46 let transaction = decode(raw, usize::try_from(config::STANDARD_TX_WEIGHT)? * 2)?;
47 ensure!(
48 transaction.weight().to_wu() <= config::STANDARD_TX_WEIGHT,
49 "transaction exceeds standard weight"
50 );
51 let acceptance = match node.call("testmempoolaccept", &[json!([raw])]) {
52 Ok(value) => value,
53 Err(cause) if node.is_public() => {
54 tracing::warn!(%cause, "public mempool preflight unavailable");
55 return Ok(MempoolCheck::Unavailable(cause.to_string()));
56 }
57 Err(cause) => return Err(cause),
58 };
59 let rows = acceptance
60 .as_array()
61 .context("missing mempool acceptance")?;
62 ensure!(rows.len() == 1, "incorrect mempool acceptance count");
63 let result = &rows[0];
64 ensure!(
65 result["txid"] == json!(transaction.compute_txid()),
66 "mempool preflight TXID mismatch"
67 );
68 match result["allowed"]
69 .as_bool()
70 .context("missing mempool decision")?
71 {
72 true => Ok(MempoolCheck::Allowed),
73 false => Ok(MempoolCheck::Rejected(
74 result["reject-reason"]
75 .as_str()
76 .context("mempool policy rejected transaction without reason")?
77 .to_owned(),
78 )),
79 }
80}
81
82pub(crate) fn broadcast(node: &Node, raw: &str) -> Result<String, Error> {
83 let transaction = decode(raw, usize::try_from(config::STANDARD_TX_WEIGHT)? * 2)?;
84 let txid = transaction.compute_txid();
85 let parent_pending = parent_in_mempool(node, &transaction)?;
86 match test_accept(node, raw)? {
87 MempoolCheck::Allowed => (),
88 MempoolCheck::Rejected(reason) if parent_pending && missing_inputs(&reason) => {
89 tracing::warn!(%reason, %txid, "preflight node has not seen the unconfirmed parent yet; submitting where the parent is known");
90 }
91 MempoolCheck::Rejected(reason) => return Ok(reason),
92 MempoolCheck::Unavailable(reason) => {
93 tracing::warn!(%reason, "broadcast has no node mempool preflight evidence");
94 tracing::debug!(target: "urma_progress", "Endpoint does not provide testmempoolaccept; acceptance is unverified until submission.");
95 }
96 }
97 match node.call("sendrawtransaction", &[json!(raw)]) {
98 Ok(result) => ensure!(
99 result == json!(txid),
100 "broadcast RPC returned unexpected TXID"
101 ),
102 Err(error) => {
103 tracing::warn!(txid = %txid, "broadcast reply unavailable; reconciling exact signed transaction");
104 match node.presence(txid)? {
105 Presence::Missing => {
106 if temporary_rejection(&error) {
107 return Ok("mempool full".into());
108 }
109 if parent_pending {
110 tracing::warn!(%error, %txid, "child refused while its parent is still propagating; retrying on the next pass");
111 return Ok(config::AWAITING_PARENT.into());
112 }
113 return Err(error);
114 }
115 Presence::Mempool | Presence::Confirmed { .. } => {}
116 }
117 }
118 }
119 Ok(String::new())
120}
121
122fn parent_in_mempool(node: &Node, transaction: &bitcoin::Transaction) -> Result<bool, Error> {
123 let mut pending = false;
124 for input in &transaction.input {
125 match node.presence(input.previous_output.txid)? {
126 Presence::Mempool => pending = true,
127 Presence::Confirmed { .. } | Presence::Missing => (),
128 }
129 }
130 Ok(pending)
131}
132
133fn missing_inputs(reason: &str) -> bool {
134 config::MISSING_INPUT_REJECTIONS.contains(&reason)
135}
136
137fn temporary_rejection(error: &Error) -> bool {
138 match error {
139 Error::Rpc(bitcoincore_rpc::Error::JsonRpc(bitcoincore_rpc::jsonrpc::Error::Rpc(
140 rejection,
141 ))) => rejection.code == -26 && rejection.message == "mempool full",
142 _ => false,
143 }
144}
145
146#[derive(Clone, Copy, Debug, PartialEq, Eq)]
147pub enum RevealTiming {
148 AfterMempool,
149 AfterConfirmation,
150}
151
152pub fn publish(
153 node: &Node,
154 plan: &PublicationPlan,
155 approved_id: &str,
156 journal: &Path,
157) -> Result<PublishReport, Error> {
158 publish_with(node, plan, approved_id, journal, RevealTiming::AfterMempool)
159}
160
161pub fn publish_with(
162 node: &Node,
163 plan: &PublicationPlan,
164 approved_id: &str,
165 journal: &Path,
166 timing: RevealTiming,
167) -> Result<PublishReport, Error> {
168 let (mut report, anchor) = start(
169 node,
170 plan.id()?,
171 &plan.root_txid,
172 plan.chain,
173 approved_id,
174 journal,
175 )?;
176 let mut raw_transactions = Vec::new();
177 for pair in &plan.records {
178 raw_transactions.push(&pair.commit);
179 raw_transactions.push(&pair.reveal);
180 }
181 for raw in &raw_transactions {
182 if !advance(node, raw, &mut report)? {
183 store(journal, &report)?;
184 return Ok(report);
185 }
186 let presence = &report
187 .transactions
188 .last()
189 .context("transaction observation missing")?
190 .presence;
191 if matches!(presence, Presence::Missing) {
192 report.blocked_reason =
193 "transaction missing after submission; resume exact plan".into();
194 store(journal, &report)?;
195 return Ok(report);
196 }
197 let dependent = plan.records.len() != 1 || timing == RevealTiming::AfterConfirmation;
198 if dependent && !matches!(presence, Presence::Confirmed { .. }) {
199 report.blocked_reason =
200 "awaiting one confirmation before dependent publication or recovery".into();
201 store(journal, &report)?;
202 return Ok(report);
203 }
204 store(journal, &report)?;
205 }
206 ensure!(
207 node.block_hash(anchor.0)?.to_string() == anchor.1,
208 "chain changed during publication reconciliation; resume against the new chain"
209 );
210 report.confirmed = report
211 .transactions
212 .iter()
213 .all(|transaction| matches!(transaction.presence, Presence::Confirmed { .. }));
214 report.complete = report.confirmed;
215 if !report.confirmed {
216 report.blocked_reason = "awaiting confirmations for commit and reveal".into();
217 }
218 store(journal, &report)?;
219 Ok(report)
220}
221
222pub(crate) fn advance(node: &Node, raw: &str, report: &mut PublishReport) -> Result<bool, Error> {
223 let transaction = decode(raw, usize::try_from(config::STANDARD_TX_WEIGHT)? * 2)?;
224 let txid = transaction.compute_txid();
225 let mut presence = node.presence(txid)?;
226 if matches!(presence, Presence::Missing) {
227 let reason = broadcast(node, raw)?;
228 if !reason.is_empty() {
229 report.blocked_reason = reason;
230 report.transactions.push(TransactionStatus {
231 txid: txid.to_string(),
232 presence,
233 });
234 return Ok(false);
235 }
236 presence = node.presence_after_submission(txid)?;
237 }
238 report.transactions.push(TransactionStatus {
239 txid: txid.to_string(),
240 presence,
241 });
242 Ok(true)
243}
244
245#[cfg(not(target_arch = "wasm32"))]
246pub fn ensure_journal_distinct(plan_path: &Path, journal_path: &Path) -> Result<(), Error> {
247 let plan = std::fs::canonicalize(plan_path)?;
248 if journal_path.try_exists()? {
249 let journal = std::fs::canonicalize(journal_path)?;
250 ensure!(
251 plan != journal,
252 "journal must differ from immutable plan input"
253 );
254 ensure!(
255 native_fs::file_identity(&plan)? != native_fs::file_identity(&journal)?,
256 "journal must not be a hard link to immutable plan input"
257 );
258 } else {
259 let parent = std::fs::canonicalize(urma_io::output_parent(journal_path))?;
260 let name = journal_path
261 .file_name()
262 .context("journal filename missing")?;
263 ensure!(
264 plan != parent.join(name),
265 "journal must differ from immutable plan input"
266 );
267 }
268 Ok(())
269}
270
271#[cfg(target_arch = "wasm32")]
272pub fn ensure_journal_distinct(plan_path: &Path, journal_path: &Path) -> Result<(), Error> {
273 Err(Error::Unsupported(format!(
274 "native journal path validation is unavailable on WASM: {} and {}",
275 plan_path.display(),
276 journal_path.display()
277 )))
278}
279
280pub(crate) fn start(
281 node: &Node,
282 id: String,
283 root_txid: &str,
284 chain: Chain,
285 approved_id: &str,
286 journal: &Path,
287) -> Result<(PublishReport, (u64, String)), Error> {
288 ensure!(
289 approved_id == id,
290 "approval does not match exact immutable signed plan"
291 );
292 node.verify_network()?;
293 node.require_txindex()?;
294 let anchor = node.tip()?;
295 ensure!(
296 node.chain().genesis()? == chain.genesis()?,
297 "publication network mismatch"
298 );
299 if journal.try_exists()? {
300 let old: PublishReport =
301 serde_json::from_slice(&storage::read_bounded(journal, 1024 * 1024)?)?;
302 ensure!(old.plan_id == id, "journal belongs to another plan");
303 }
304 let report = PublishReport {
305 plan_id: id,
306 root_txid: root_txid.to_owned(),
307 complete: false,
308 confirmed: false,
309 blocked_reason: String::new(),
310 transactions: Vec::new(),
311 };
312 store(journal, &report)?;
313 Ok((report, anchor))
314}