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 match test_accept(node, raw)? {
82 MempoolCheck::Allowed => (),
83 MempoolCheck::Rejected(reason) => return Ok(reason),
84 MempoolCheck::Unavailable(reason) => {
85 tracing::warn!(%reason, "broadcast has no node mempool preflight evidence");
86 tracing::debug!(target: "urma_progress", "Endpoint does not provide testmempoolaccept; acceptance is unverified until submission.");
87 }
88 }
89 let transaction = decode(raw, usize::try_from(config::STANDARD_TX_WEIGHT)? * 2)?;
90 let txid = transaction.compute_txid();
91 match node.call("sendrawtransaction", &[json!(raw)]) {
92 Ok(result) => ensure!(
93 result == json!(txid),
94 "broadcast RPC returned unexpected TXID"
95 ),
96 Err(error) => {
97 tracing::warn!(txid = %txid, "broadcast reply unavailable; reconciling exact signed transaction");
98 match node.presence(txid)? {
99 Presence::Missing => {
100 if temporary_rejection(&error) {
101 return Ok("mempool full".into());
102 }
103 return Err(error);
104 }
105 Presence::Mempool | Presence::Confirmed { .. } => {}
106 }
107 }
108 }
109 Ok(String::new())
110}
111
112fn temporary_rejection(error: &Error) -> bool {
113 match error {
114 Error::Rpc(bitcoincore_rpc::Error::JsonRpc(bitcoincore_rpc::jsonrpc::Error::Rpc(
115 rejection,
116 ))) => rejection.code == -26 && rejection.message == "mempool full",
117 _ => false,
118 }
119}
120
121#[derive(Clone, Copy, Debug, PartialEq, Eq)]
122pub enum RevealTiming {
123 AfterMempool,
124 AfterConfirmation,
125}
126
127pub fn publish(
128 node: &Node,
129 plan: &PublicationPlan,
130 approved_id: &str,
131 journal: &Path,
132) -> Result<PublishReport, Error> {
133 publish_with(node, plan, approved_id, journal, RevealTiming::AfterMempool)
134}
135
136pub fn publish_with(
137 node: &Node,
138 plan: &PublicationPlan,
139 approved_id: &str,
140 journal: &Path,
141 timing: RevealTiming,
142) -> Result<PublishReport, Error> {
143 let (mut report, anchor) = start(
144 node,
145 plan.id()?,
146 &plan.root_txid,
147 plan.chain,
148 approved_id,
149 journal,
150 )?;
151 let mut raw_transactions = Vec::new();
152 for pair in &plan.records {
153 raw_transactions.push(&pair.commit);
154 raw_transactions.push(&pair.reveal);
155 }
156 for raw in &raw_transactions {
157 if !advance(node, raw, &mut report)? {
158 store(journal, &report)?;
159 return Ok(report);
160 }
161 let presence = &report
162 .transactions
163 .last()
164 .context("transaction observation missing")?
165 .presence;
166 if matches!(presence, Presence::Missing) {
167 report.blocked_reason =
168 "transaction missing after submission; resume exact plan".into();
169 store(journal, &report)?;
170 return Ok(report);
171 }
172 let dependent = plan.records.len() != 1 || timing == RevealTiming::AfterConfirmation;
173 if dependent && !matches!(presence, Presence::Confirmed { .. }) {
174 report.blocked_reason =
175 "awaiting one confirmation before dependent publication or recovery".into();
176 store(journal, &report)?;
177 return Ok(report);
178 }
179 store(journal, &report)?;
180 }
181 ensure!(
182 node.block_hash(anchor.0)?.to_string() == anchor.1,
183 "chain changed during publication reconciliation; resume against the new chain"
184 );
185 report.confirmed = report
186 .transactions
187 .iter()
188 .all(|transaction| matches!(transaction.presence, Presence::Confirmed { .. }));
189 report.complete = report.confirmed;
190 if !report.confirmed {
191 report.blocked_reason = "awaiting confirmations for commit and reveal".into();
192 }
193 store(journal, &report)?;
194 Ok(report)
195}
196
197pub(crate) fn advance(node: &Node, raw: &str, report: &mut PublishReport) -> Result<bool, Error> {
198 let transaction = decode(raw, usize::try_from(config::STANDARD_TX_WEIGHT)? * 2)?;
199 let txid = transaction.compute_txid();
200 let mut presence = node.presence(txid)?;
201 if matches!(presence, Presence::Missing) {
202 let reason = broadcast(node, raw)?;
203 if !reason.is_empty() {
204 report.blocked_reason = reason;
205 report.transactions.push(TransactionStatus {
206 txid: txid.to_string(),
207 presence,
208 });
209 return Ok(false);
210 }
211 presence = node.presence(txid)?;
212 }
213 report.transactions.push(TransactionStatus {
214 txid: txid.to_string(),
215 presence,
216 });
217 Ok(true)
218}
219
220#[cfg(not(target_arch = "wasm32"))]
221pub fn ensure_journal_distinct(plan_path: &Path, journal_path: &Path) -> Result<(), Error> {
222 use std::os::unix::fs::MetadataExt;
223 let plan = std::fs::canonicalize(plan_path)?;
224 if journal_path.try_exists()? {
225 let journal = std::fs::canonicalize(journal_path)?;
226 ensure!(
227 plan != journal,
228 "journal must differ from immutable plan input"
229 );
230 let input = std::fs::metadata(&plan)?;
231 let output = std::fs::metadata(&journal)?;
232 ensure!(
233 input.dev() != output.dev() || input.ino() != output.ino(),
234 "journal must not be a hard link to immutable plan input"
235 );
236 } else {
237 let parent = std::fs::canonicalize(urma_io::output_parent(journal_path))?;
238 let name = journal_path
239 .file_name()
240 .context("journal filename missing")?;
241 ensure!(
242 plan != parent.join(name),
243 "journal must differ from immutable plan input"
244 );
245 }
246 Ok(())
247}
248
249#[cfg(target_arch = "wasm32")]
250pub fn ensure_journal_distinct(plan_path: &Path, journal_path: &Path) -> Result<(), Error> {
251 Err(Error::Unsupported(format!(
252 "native journal path validation is unavailable on WASM: {} and {}",
253 plan_path.display(),
254 journal_path.display()
255 )))
256}
257
258pub(crate) fn start(
259 node: &Node,
260 id: String,
261 root_txid: &str,
262 chain: Chain,
263 approved_id: &str,
264 journal: &Path,
265) -> Result<(PublishReport, (u64, String)), Error> {
266 ensure!(
267 approved_id == id,
268 "approval does not match exact immutable signed plan"
269 );
270 node.verify_network()?;
271 node.require_txindex()?;
272 let anchor = node.tip()?;
273 ensure!(
274 node.chain().genesis()? == chain.genesis()?,
275 "publication network mismatch"
276 );
277 if journal.try_exists()? {
278 let old: PublishReport =
279 serde_json::from_slice(&storage::read_bounded(journal, 1024 * 1024)?)?;
280 ensure!(old.plan_id == id, "journal belongs to another plan");
281 }
282 let report = PublishReport {
283 plan_id: id,
284 root_txid: root_txid.to_owned(),
285 complete: false,
286 confirmed: false,
287 blocked_reason: String::new(),
288 transactions: Vec::new(),
289 };
290 store(journal, &report)?;
291 Ok((report, anchor))
292}