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