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
121pub fn publish(
122 node: &Node,
123 plan: &PublicationPlan,
124 approved_id: &str,
125 journal: &Path,
126) -> Result<PublishReport, Error> {
127 let (mut report, anchor) = start(
128 node,
129 plan.id()?,
130 &plan.root_txid,
131 plan.chain,
132 approved_id,
133 journal,
134 )?;
135 let mut raw_transactions = Vec::new();
136 for pair in &plan.records {
137 raw_transactions.push(&pair.commit);
138 raw_transactions.push(&pair.reveal);
139 }
140 for raw in &raw_transactions {
141 if !advance(node, raw, &mut report)? {
142 store(journal, &report)?;
143 return Ok(report);
144 }
145 let presence = &report
146 .transactions
147 .last()
148 .context("transaction observation missing")?
149 .presence;
150 if !matches!(presence, Presence::Confirmed { .. }) {
151 report.blocked_reason =
152 "awaiting one confirmation before dependent publication or recovery".into();
153 store(journal, &report)?;
154 return Ok(report);
155 }
156 store(journal, &report)?;
157 }
158 ensure!(
159 node.block_hash(anchor.0)?.to_string() == anchor.1,
160 "chain changed during publication reconciliation; resume against the new chain"
161 );
162 report.complete = true;
163 report.confirmed = true;
164 store(journal, &report)?;
165 Ok(report)
166}
167
168pub(crate) fn advance(node: &Node, raw: &str, report: &mut PublishReport) -> Result<bool, Error> {
169 let transaction = decode(raw, usize::try_from(config::STANDARD_TX_WEIGHT)? * 2)?;
170 let txid = transaction.compute_txid();
171 let mut presence = node.presence(txid)?;
172 if matches!(presence, Presence::Missing) {
173 let reason = broadcast(node, raw)?;
174 if !reason.is_empty() {
175 report.blocked_reason = reason;
176 report.transactions.push(TransactionStatus {
177 txid: txid.to_string(),
178 presence,
179 });
180 return Ok(false);
181 }
182 presence = node.presence(txid)?;
183 }
184 report.transactions.push(TransactionStatus {
185 txid: txid.to_string(),
186 presence,
187 });
188 Ok(true)
189}
190
191pub fn ensure_journal_distinct(plan_path: &Path, journal_path: &Path) -> Result<(), Error> {
192 use std::os::unix::fs::MetadataExt;
193 let plan = std::fs::canonicalize(plan_path)?;
194 if journal_path.try_exists()? {
195 let journal = std::fs::canonicalize(journal_path)?;
196 ensure!(
197 plan != journal,
198 "journal must differ from immutable plan input"
199 );
200 let input = std::fs::metadata(&plan)?;
201 let output = std::fs::metadata(&journal)?;
202 ensure!(
203 input.dev() != output.dev() || input.ino() != output.ino(),
204 "journal must not be a hard link to immutable plan input"
205 );
206 } else {
207 let parent = std::fs::canonicalize(urma_io::output_parent(journal_path))?;
208 let name = journal_path
209 .file_name()
210 .context("journal filename missing")?;
211 ensure!(
212 plan != parent.join(name),
213 "journal must differ from immutable plan input"
214 );
215 }
216 Ok(())
217}
218
219pub(crate) fn start(
220 node: &Node,
221 id: String,
222 root_txid: &str,
223 chain: Chain,
224 approved_id: &str,
225 journal: &Path,
226) -> Result<(PublishReport, (u64, String)), Error> {
227 ensure!(
228 approved_id == id,
229 "approval does not match exact immutable signed plan"
230 );
231 node.verify_network()?;
232 node.require_txindex()?;
233 let anchor = node.tip()?;
234 ensure!(
235 node.chain().genesis()? == chain.genesis()?,
236 "publication network mismatch"
237 );
238 if journal.try_exists()? {
239 let old: PublishReport =
240 serde_json::from_slice(&storage::read_bounded(journal, 1024 * 1024)?)?;
241 ensure!(old.plan_id == id, "journal belongs to another plan");
242 }
243 let report = PublishReport {
244 plan_id: id,
245 root_txid: root_txid.to_owned(),
246 complete: false,
247 confirmed: false,
248 blocked_reason: String::new(),
249 transactions: Vec::new(),
250 };
251 store(journal, &report)?;
252 Ok((report, anchor))
253}