1use crate::error::{Error, ensure};
2use crate::publication::PublicPlan;
3use crate::transaction::decode;
4use crate::{config, disk_journal, publication_progress};
5use crate::{
6 disk_plan::DiskPlan,
7 node::{Node, Presence},
8 publication_progress::{Observation, Progress, State, Target},
9 publish::{PublishReport, TransactionStatus, broadcast, store},
10};
11use std::{collections::HashMap, path::Path};
12use urma_core::multipart::{MultipartRecord, VerifiedRecord};
13
14pub fn publish(
15 node: &Node,
16 plan: &DiskPlan,
17 approved_id: &str,
18 journal: &Path,
19) -> Result<PublishReport, Error> {
20 Ok(
21 publish_progress(node, plan, approved_id, journal, &mut |progress| {
22 tracing::debug!(target: "urma_progress", "{}", progress.summary());
23 Ok(())
24 })?
25 .report,
26 )
27}
28
29pub fn publish_progress(
30 node: &Node,
31 plan: &DiskPlan,
32 approved_id: &str,
33 journal: &Path,
34 progress: &mut impl FnMut(&Progress) -> Result<(), Error>,
35) -> Result<Progress, Error> {
36 ensure!(
37 approved_id == plan.id()?,
38 "approval does not match exact immutable signed plan"
39 );
40 disk_journal::guard(plan, journal)?;
41 reconcile(node, plan, journal, true, progress)
42}
43
44pub fn watch(
45 node: &Node,
46 plan: &DiskPlan,
47 journal: &Path,
48 progress: &mut impl FnMut(&Progress) -> Result<(), Error>,
49) -> Result<Progress, Error> {
50 reconcile(node, plan, journal, false, progress)
51}
52
53struct Session<'a> {
54 node: &'a Node,
55 journal: &'a Path,
56 submit: bool,
57 history: HashMap<String, Presence>,
58 blocks: HashMap<String, Presence>,
59 progress: Progress,
60}
61
62struct BufferedPair {
63 role: &'static str,
64 commit_weight: u64,
65 reveal_weight: u64,
66}
67
68struct Buffer {
69 weight: u64,
70 broadcast_weight: u64,
71 commits: usize,
72 commit_vbytes: u64,
73}
74
75impl Buffer {
76 fn new(pairs: &[BufferedPair], observations: &[Observation]) -> Self {
77 let mut buffer = Self {
78 weight: 0,
79 broadcast_weight: 0,
80 commits: 0,
81 commit_vbytes: 0,
82 };
83 for (pair, rows) in pairs.iter().zip(observations.as_chunks::<2>().0) {
84 if rows[0].state == State::Mempool {
85 buffer.weight += pair.commit_weight;
86 buffer.broadcast_weight += pair.commit_weight;
87 buffer.commits += 1;
88 buffer.commit_vbytes += pair.commit_weight.div_ceil(4);
89 }
90 if rows[1].state == State::Mempool {
91 buffer.broadcast_weight += pair.reveal_weight;
92 }
93 if rows[1].state != State::Confirmed
94 && (matches!(rows[0].state, State::Confirmed | State::Mempool)
95 || rows[1].state == State::Mempool)
96 {
97 buffer.weight += pair.reveal_weight;
98 }
99 }
100 buffer
101 }
102
103 fn admits(&self, pair: &BufferedPair) -> bool {
104 self.commits < config::PUBLICATION_PENDING_COMMITS
105 && self.commit_vbytes + pair.commit_weight.div_ceil(4)
106 <= config::PUBLICATION_COMMIT_VBYTES
107 && self.weight + pair.commit_weight + pair.reveal_weight
108 <= config::PUBLICATION_BUFFER_WEIGHT
109 }
110
111 fn add(&mut self, pair: &BufferedPair) {
112 self.commits += 1;
113 self.commit_vbytes += pair.commit_weight.div_ceil(4);
114 self.weight += pair.commit_weight + pair.reveal_weight;
115 self.broadcast_weight += pair.commit_weight;
116 }
117}
118
119fn reconcile(
120 node: &Node,
121 plan: &DiskPlan,
122 journal: &Path,
123 submit: bool,
124 notify: &mut impl FnMut(&Progress) -> Result<(), Error>,
125) -> Result<Progress, Error> {
126 plan.validate()?;
127 node.verify_network()?;
128 node.require_txindex()?;
129 ensure!(
130 node.chain().genesis()? == plan.chain.genesis()?,
131 "publication network mismatch"
132 );
133 let anchor = node.tip()?;
134 let id = plan.id()?;
135 let mut session = Session {
136 node,
137 journal,
138 submit,
139 history: disk_journal::history(journal, &id)?,
140 blocks: HashMap::new(),
141 progress: Progress {
142 report: PublishReport {
143 plan_id: id,
144 root_txid: plan.root_txid.clone(),
145 complete: false,
146 confirmed: false,
147 blocked_reason: String::new(),
148 transactions: Vec::new(),
149 },
150 total: usize::try_from(plan.record_count)? * 2,
151 observations: Vec::new(),
152 retryable: true,
153 },
154 };
155 session.walk(plan, &anchor, notify)?;
156 if node.block_hash(anchor.0)?.to_string() != anchor.1 {
157 session.progress.observations.clear();
158 session.progress.report.transactions.clear();
159 session.progress.report.blocked_reason =
160 "chain changed during reconciliation; waiting for fresh observations".into();
161 }
162 session.progress.report.complete = session.progress.reached(Target::Confirmed);
163 session.progress.report.confirmed = session.progress.report.complete;
164 if !session.progress.report.complete && session.progress.report.blocked_reason.is_empty() {
165 session.progress.report.blocked_reason = if submit {
166 "awaiting confirmations; eligible transactions will be reconciled on the next pass"
167 } else {
168 "read-only watch; prepared or missing transactions require publish/resume in another process"
169 }.into();
170 }
171 if submit {
172 store(journal, &session.progress.report)?;
173 }
174 notify(&session.progress)?;
175 Ok(session.progress)
176}
177
178impl Session<'_> {
179 fn walk(
180 &mut self,
181 plan: &DiskPlan,
182 anchor: &(u64, String),
183 notify: &mut impl FnMut(&Progress) -> Result<(), Error>,
184 ) -> Result<(), Error> {
185 let pairs = self.inventory(plan, notify)?;
186 if !self.submit || !self.progress.retryable {
187 return Ok(());
188 }
189 if self.node.block_hash(anchor.0)?.to_string() != anchor.1 {
190 self.progress.report.blocked_reason =
191 "chain changed during reconciliation; waiting for fresh observations".into();
192 return Ok(());
193 }
194 let mut buffer = Buffer::new(&pairs, &self.progress.observations);
195 let mut data_available = true;
196 let mut leaves_available = true;
197 let mut parent_available = true;
198 let mut limited = false;
199 for (index, pair) in pairs.iter().enumerate() {
200 notify(&self.progress)?;
201 let offset = index * 2;
202 let mut commit = self.progress.observations[offset].state.clone();
203 if matches!(commit, State::Prepared | State::Missing) && parent_available {
204 if buffer.admits(pair) && self.progress.report.blocked_reason.is_empty() {
205 commit = self.submit_record(plan, index, offset, true)?;
206 if commit == State::Mempool {
207 buffer.add(pair);
208 }
209 } else {
210 limited = true;
211 }
212 }
213 parent_available = matches!(commit, State::Confirmed | State::Mempool);
214 let dependencies = match pair.role {
215 "leaf" => data_available,
216 "root" => data_available && leaves_available,
217 _ => true,
218 };
219 let mut reveal = self.progress.observations[offset + 1].state.clone();
220 if commit == State::Confirmed
221 && dependencies
222 && self.progress.retryable
223 && self.progress.report.blocked_reason.is_empty()
224 && matches!(reveal, State::Prepared | State::Missing)
225 {
226 if buffer.broadcast_weight + pair.reveal_weight <= config::PUBLICATION_BUFFER_WEIGHT
227 {
228 reveal = self.submit_record(plan, index, offset + 1, false)?;
229 if reveal == State::Mempool {
230 buffer.broadcast_weight += pair.reveal_weight;
231 }
232 } else {
233 limited = true;
234 }
235 }
236 match pair.role {
237 "data" => data_available &= matches!(reveal, State::Confirmed | State::Mempool),
238 "leaf" => leaves_available &= matches!(reveal, State::Confirmed | State::Mempool),
239 _ => (),
240 }
241 notify(&self.progress)?;
242 if !self.progress.retryable || !self.progress.report.blocked_reason.is_empty() {
243 break;
244 }
245 }
246 if limited && self.progress.report.blocked_reason.is_empty() {
247 self.progress.report.blocked_reason =
248 "Publication buffer full; waiting for blockchain confirmation. Publication continues automatically while watching; no user input needed.".into();
249 }
250 Ok(())
251 }
252
253 fn inventory(
254 &mut self,
255 plan: &DiskPlan,
256 notify: &mut impl FnMut(&Progress) -> Result<(), Error>,
257 ) -> Result<Vec<BufferedPair>, Error> {
258 let mut pairs = Vec::new();
259 self.progress.report.transactions.clear();
260 for index in 0..plan.record_count {
261 notify(&self.progress)?;
262 let pair = plan.record(index)?;
263 let role = match record(&pair)? {
264 MultipartRecord::Data(_) => "data",
265 MultipartRecord::Leaf(_) => "leaf",
266 MultipartRecord::Root(_) => "root",
267 };
268 let limit = usize::try_from(config::STANDARD_TX_WEIGHT)? * 2;
269 let commit = decode(&pair.commit, limit)?;
270 let reveal = decode(&pair.reveal, limit)?;
271 pairs.push(BufferedPair {
272 role,
273 commit_weight: commit.weight().to_wu(),
274 reveal_weight: reveal.weight().to_wu(),
275 });
276 self.transaction(&pair.commit, "commit", false)?;
277 if !self.progress.retryable {
278 break;
279 }
280 self.transaction(&pair.reveal, role, false)?;
281 if !self.progress.retryable {
282 break;
283 }
284 }
285 notify(&self.progress)?;
286 Ok(pairs)
287 }
288
289 fn submit_record(
290 &mut self,
291 plan: &DiskPlan,
292 index: usize,
293 offset: usize,
294 commit: bool,
295 ) -> Result<State, Error> {
296 let pair = plan.record(u32::try_from(index)?)?;
297 let role = self.progress.observations[offset].role.clone();
298 let raw = if commit { &pair.commit } else { &pair.reveal };
299 let state = self.transaction(raw, &role, true)?;
300 self.progress
301 .observations
302 .truncate(self.progress.observations.len() - 1);
303 self.progress.observations[offset].state = state.clone();
304 Ok(state)
305 }
306
307 fn transaction(&mut self, raw: &str, role: &str, eligible: bool) -> Result<State, Error> {
308 let txid = decode(raw, usize::try_from(config::STANDARD_TX_WEIGHT)? * 2)?.compute_txid();
309 let mut presence = match self.node.presence_cached(txid, &mut self.blocks) {
310 Ok(presence) => presence,
311 Err(error) => {
312 self.progress.observations.push(Observation {
313 txid: txid.to_string(),
314 role: role.into(),
315 state: State::SourceUnavailable {
316 reason: error.to_string(),
317 },
318 });
319 self.progress.report.blocked_reason = format!("source unavailable: {error}");
320 self.progress.retryable = false;
321 tracing::warn!(%error, "publication source unavailable; not treating as missing");
322 return Ok(State::SourceUnavailable {
323 reason: error.to_string(),
324 });
325 }
326 };
327 if eligible && matches!(presence, Presence::Missing) {
328 self.remember(txid.to_string(), presence.clone())?;
329 let reason = broadcast(self.node, raw)?;
330 if reason.is_empty() {
331 presence = self.node.presence_cached(txid, &mut self.blocks)?;
332 if matches!(presence, Presence::Missing) {
333 self.progress.report.blocked_reason =
334 "submission not yet observable; waiting before spending its change".into();
335 }
336 } else {
337 self.progress.retryable = reason == "mempool full";
338 self.progress.report.blocked_reason = format!("{reason}; approved fees unchanged");
339 }
340 }
341 let attempted = self.history.contains_key(&txid.to_string());
342 let state = publication_progress::state(&presence, attempted);
343 if self.submit && (attempted || !matches!(presence, Presence::Missing)) {
344 self.remember(txid.to_string(), presence.clone())?;
345 }
346 self.progress.report.transactions.push(TransactionStatus {
347 txid: txid.to_string(),
348 presence,
349 });
350 self.progress.observations.push(Observation {
351 txid: txid.to_string(),
352 role: role.into(),
353 state: state.clone(),
354 });
355 Ok(state)
356 }
357
358 fn remember(&mut self, txid: String, presence: Presence) -> Result<(), Error> {
359 let previous = self.history.get(&txid);
360 let unchanged = previous.iter().any(|previous| **previous == presence);
361 if unchanged {
362 return Ok(());
363 }
364 let mut event = self.progress.report.clone();
365 event.transactions = vec![TransactionStatus {
366 txid: txid.clone(),
367 presence: presence.clone(),
368 }];
369 disk_journal::observe(self.node, self.journal, &event)?;
370 self.history.insert(txid, presence);
371 Ok(())
372 }
373}
374
375fn record(pair: &PublicPlan) -> Result<MultipartRecord, Error> {
376 let limit = usize::try_from(config::STANDARD_TX_WEIGHT)? * 2;
377 let commit = decode(&pair.commit, limit)?;
378 let reveal = decode(&pair.reveal, limit)?;
379 Ok(VerifiedRecord::verify(reveal.compute_txid(), &reveal, &commit)?.decode()?)
380}
381
382pub fn may_resume_automatically(report: &PublishReport) -> bool {
383 !report.complete
384 && (report.blocked_reason.starts_with("awaiting confirmations;")
385 || report.blocked_reason == "eight funding commits pending; waiting for confirmation"
386 || report
387 .blocked_reason
388 .starts_with("Publication buffer full;")
389 || report.blocked_reason.starts_with("Commit window full (8);")
390 || report.blocked_reason
391 == "submission not yet observable; waiting before spending its change"
392 || report.blocked_reason == "mempool full; approved fees unchanged")
393 || (!report.complete
394 && report.blocked_reason
395 == "chain changed during reconciliation; waiting for fresh observations")
396}