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