urma_runtime/p2p/
headers.rs

1use super::bootstrap::{HeaderBootstrap, context_headers};
2use crate::config::{
3    Checkpoint, LITECOIN_MAINNET_CHECKPOINT, LITECOIN_TESTNET_CHECKPOINT, P2P_HEADER_CACHE_MAX,
4    P2P_HEADER_CACHE_REVALIDATE, P2P_LOCATOR_LINEAR, P2P_MAX_FUTURE_SECS, P2P_MEDIAN_TIME_SPAN,
5    P2P_POW_WORKERS,
6};
7use crate::error::{Context, Error, ensure};
8use bitcoin::BlockHash;
9use bitcoin::block::Header;
10use bitcoin::consensus::encode::{deserialize, serialize};
11use std::borrow::Cow;
12use std::collections::HashMap;
13use std::path::{Path, PathBuf};
14use std::time::Instant;
15use urma_chain::observation::Chain;
16use urma_chain::pow::{PowError, chain_work, check_header_pow, expected_bits};
17
18pub struct HeaderChain {
19    chain: Chain,
20    start_height: u64,
21    trusted_next: Option<BlockHash>,
22    ancestors: Option<Vec<Header>>,
23    headers: Vec<Header>,
24    index: HashMap<BlockHash, usize>,
25    path: PathBuf,
26}
27
28#[derive(Debug, PartialEq, Eq)]
29pub enum Extension {
30    Appended(usize),
31    Reorganized { fork_height: u64, appended: usize },
32    Ignored,
33}
34
35pub fn checkpoint(chain: Chain) -> Result<&'static Checkpoint, Error> {
36    match chain {
37        Chain::LitecoinMainnet => Ok(&LITECOIN_MAINNET_CHECKPOINT),
38        Chain::LitecoinTestnet => Ok(&LITECOIN_TESTNET_CHECKPOINT),
39        Chain::BitcoinMainnet | Chain::BitcoinRegtest | Chain::BitcoinTestnet4 => {
40            Err(Error::Unsupported(
41                "the peer-to-peer light client serves Litecoin networks only".into(),
42            ))
43        }
44    }
45}
46
47/// Legacy checkpoint bootstrap. Its first successor is pinned rather than
48/// contextually retargeted, and initial median-time checks use only the locally
49/// available suffix. Use `HeaderBootstrap` for full preceding context from the
50/// first successor; this compatibility API does not synthesize that history.
51#[derive(Clone, Copy, Debug, PartialEq, Eq)]
52pub struct Anchor {
53    pub height: u64,
54    pub header: Header,
55    pub next_hash: BlockHash,
56}
57
58impl Anchor {
59    pub fn embedded(chain: Chain) -> Result<Self, Error> {
60        let checkpoint = checkpoint(chain)?;
61        Ok(Self {
62            height: checkpoint.height,
63            header: deserialize(&hex::decode(checkpoint.header)?)?,
64            next_hash: checkpoint.next_hash.parse()?,
65        })
66    }
67}
68
69impl HeaderChain {
70    /// Open a strict checkpoint-relative view. Context was authenticated by
71    /// `HeaderBootstrap::new` before any cache access. Its cache namespace is
72    /// separate from legacy pinned-successor caches and binds height as well
73    /// as chain and anchor hash. Ancestors never enter the selected index.
74    pub fn open_bootstrap(cache_dir: &Path, bootstrap: HeaderBootstrap) -> Result<Self, Error> {
75        let (chain, height, header, ancestors) = bootstrap.into_parts();
76        let path = cache_dir.join(format!(
77            "strict-v1-{}-{height}-{}-headers.bin",
78            chain.label().replace(' ', "-"),
79            header.block_hash()
80        ));
81        Self {
82            chain,
83            start_height: height,
84            trusted_next: None,
85            ancestors: Some(ancestors),
86            headers: vec![header],
87            index: HashMap::from([(header.block_hash(), 0)]),
88            path,
89        }
90        .restore()
91    }
92
93    pub fn open(chain: Chain, cache_dir: &Path) -> Result<Self, Error> {
94        Self::open_at(chain, cache_dir, Anchor::embedded(chain)?)
95    }
96
97    pub fn open_at(chain: Chain, cache_dir: &Path, anchor: Anchor) -> Result<Self, Error> {
98        let path = cache_dir.join(format!("{}-headers.bin", chain.label().replace(' ', "-")));
99        Self::anchored(chain, path, anchor.height, anchor.header, anchor.next_hash)
100    }
101
102    pub fn anchored(
103        chain: Chain,
104        path: PathBuf,
105        start_height: u64,
106        anchor: Header,
107        trusted_next: BlockHash,
108    ) -> Result<Self, Error> {
109        check_header_pow(&anchor, chain).map_err(|cause| Error::Invalid(cause.to_string()))?;
110        Self {
111            chain,
112            start_height,
113            trusted_next: Some(trusted_next),
114            ancestors: None,
115            headers: vec![anchor],
116            index: HashMap::from([(anchor.block_hash(), 0)]),
117            path,
118        }
119        .restore()
120    }
121
122    fn restore(mut self) -> Result<Self, Error> {
123        if self.path.try_exists()? {
124            match self.load() {
125                Ok(()) => (),
126                Err(error) if error.is_header_context() => return Err(error),
127                Err(error) => {
128                    tracing::warn!(%error, path = %self.path.display(), "header cache rejected; restarting from the checkpoint");
129                    self.reset()?;
130                }
131            }
132        }
133        Ok(self)
134    }
135
136    fn reset(&mut self) -> Result<(), Error> {
137        self.headers.truncate(1);
138        self.index = HashMap::from([(self.headers[0].block_hash(), 0)]);
139        std::fs::remove_file(&self.path)?;
140        Ok(())
141    }
142
143    fn load(&mut self) -> Result<(), Error> {
144        let started = Instant::now();
145        let raw = urma_io::read_bounded(&self.path, P2P_HEADER_CACHE_MAX * Header::SIZE)?;
146        tracing::debug!(target: "urma_startup", chain = self.chain.label(), stage = "cache_read", elapsed_ms = started.elapsed().as_secs_f64() * 1000.0, headers = raw.len() / Header::SIZE, "light client stage completed");
147        let started = Instant::now();
148        ensure!(
149            raw.len() % Header::SIZE == 0 && raw.len() >= Header::SIZE,
150            "header cache {} is not a whole number of headers",
151            self.path.display()
152        );
153        let stored: Header = deserialize(&raw[..Header::SIZE])?;
154        ensure!(
155            stored == self.headers[0],
156            "header cache {} starts from a different checkpoint",
157            self.path.display()
158        );
159        let count = raw.len() / Header::SIZE;
160        let mut headers = Vec::new();
161        headers.try_reserve_exact(count)?;
162        for slice in raw.chunks_exact(Header::SIZE) {
163            headers.push(deserialize::<Header>(slice)?);
164        }
165        tracing::debug!(target: "urma_startup", chain = self.chain.label(), stage = "cache_decode", elapsed_ms = started.elapsed().as_secs_f64() * 1000.0, headers = count, "light client stage completed");
166        let revalidate_from = count.max(P2P_HEADER_CACHE_REVALIDATE) - P2P_HEADER_CACHE_REVALIDATE;
167        let started = Instant::now();
168        for (offset, header) in headers.iter().enumerate().skip(1) {
169            self.validate_context(&headers[..offset], header)
170                .with_context(|| {
171                    format!("header cache {} at offset {offset}", self.path.display())
172                })?;
173        }
174        tracing::debug!(target: "urma_startup", chain = self.chain.label(), stage = "cache_context", elapsed_ms = started.elapsed().as_secs_f64() * 1000.0, headers = count, "light client stage completed");
175        let tail = &headers[revalidate_from.max(1)..];
176        let started = Instant::now();
177        prove_batch(self.chain, tail)?;
178        tracing::debug!(target: "urma_startup", chain = self.chain.label(), stage = "cache_pow", elapsed_ms = started.elapsed().as_secs_f64() * 1000.0, headers = count, pow_headers = tail.len(), "light client stage completed");
179        let started = Instant::now();
180        self.index = headers
181            .iter()
182            .enumerate()
183            .map(|(offset, header)| (header.block_hash(), offset))
184            .collect();
185        self.headers = headers;
186        tracing::debug!(target: "urma_startup", chain = self.chain.label(), stage = "cache_index", elapsed_ms = started.elapsed().as_secs_f64() * 1000.0, headers = count, "light client stage completed");
187        Ok(())
188    }
189
190    pub fn persist(&self) -> Result<(), Error> {
191        let mut raw = Vec::new();
192        raw.try_reserve_exact(self.headers.len() * Header::SIZE)?;
193        for header in &self.headers {
194            raw.extend_from_slice(&serialize(header));
195        }
196        urma_io::write_replace(&self.path, &raw)?;
197        Ok(())
198    }
199
200    pub fn chain(&self) -> Chain {
201        self.chain
202    }
203
204    pub fn start_height(&self) -> u64 {
205        self.start_height
206    }
207
208    pub fn tip_height(&self) -> u64 {
209        self.start_height + self.offset_count()
210    }
211
212    pub fn synced_count(&self) -> u64 {
213        self.offset_count()
214    }
215
216    fn offset_count(&self) -> u64 {
217        match u64::try_from(self.headers.len() - 1) {
218            Ok(count) => count,
219            Err(error) => {
220                tracing::error!(%error, "header count exceeds u64");
221                u64::MAX
222            }
223        }
224    }
225
226    pub fn tip_hash(&self) -> BlockHash {
227        self.headers[self.headers.len() - 1].block_hash()
228    }
229
230    pub fn height_of(&self, hash: BlockHash) -> Result<u64, Error> {
231        let offset = *self
232            .index
233            .get(&hash)
234            .with_context(|| format!("block {hash} is not in the verified header chain"))?;
235        Ok(self.start_height + u64::try_from(offset)?)
236    }
237
238    pub fn header_at(&self, height: u64) -> Result<Header, Error> {
239        let offset = height
240            .checked_sub(self.start_height)
241            .with_context(|| format!("height {height} precedes the light client checkpoint"))?;
242        let offset = usize::try_from(offset)?;
243        self.headers
244            .get(offset)
245            .copied()
246            .with_context(|| format!("height {height} is above the verified header tip"))
247    }
248
249    pub fn hash_at(&self, height: u64) -> Result<BlockHash, Error> {
250        Ok(self.header_at(height)?.block_hash())
251    }
252
253    pub fn locator(&self) -> Vec<BlockHash> {
254        let mut hashes = Vec::new();
255        let mut back = self.headers.len() - 1;
256        let mut step = 1;
257        loop {
258            hashes.push(self.headers[back].block_hash());
259            if back == 0 {
260                break;
261            }
262            if hashes.len() > P2P_LOCATOR_LINEAR {
263                step *= 2;
264            }
265            back = back.max(step) - step;
266        }
267        hashes
268    }
269
270    pub fn extend(&mut self, batch: &[Header], now: u32) -> Result<Extension, Error> {
271        let Some(first) = batch.first() else {
272            return Ok(Extension::Ignored);
273        };
274        let fork = *self
275            .index
276            .get(&first.prev_blockhash)
277            .context("peer headers do not connect to the verified header chain")?;
278        let known = batch
279            .iter()
280            .zip(self.headers.iter().skip(fork + 1))
281            .take_while(|(incoming, held)| incoming.block_hash() == held.block_hash())
282            .count();
283        let fresh = &batch[known..];
284        if fresh.is_empty() {
285            return Ok(Extension::Ignored);
286        }
287        ensure!(
288            self.headers.len() + fresh.len() <= P2P_HEADER_CACHE_MAX,
289            "header chain would exceed the client capacity"
290        );
291        let base = fork + known;
292        prove_batch(self.chain, fresh)?;
293        if base == self.headers.len() - 1 {
294            self.append(fresh, now)?;
295            return Ok(Extension::Appended(fresh.len()));
296        }
297        let mut candidate = self.headers[..=base].to_vec();
298        for header in fresh {
299            self.validate_next(&candidate, header, now)?;
300            candidate.push(*header);
301        }
302        let challenger = chain_work(&candidate[base + 1..])
303            .map_err(|cause| Error::Invalid(cause.to_string()))?;
304        let incumbent = chain_work(&self.headers[base + 1..])
305            .map_err(|cause| Error::Invalid(cause.to_string()))?;
306        if challenger <= incumbent {
307            tracing::warn!(
308                fork_offset = base,
309                "peer branch carries no more work than the held chain"
310            );
311            return Ok(Extension::Ignored);
312        }
313        for dropped in &self.headers[base + 1..] {
314            self.index.remove(&dropped.block_hash());
315        }
316        self.headers = candidate;
317        for (offset, header) in self.headers.iter().enumerate().skip(base + 1) {
318            self.index.insert(header.block_hash(), offset);
319        }
320        Ok(Extension::Reorganized {
321            fork_height: self.start_height + u64::try_from(base)?,
322            appended: fresh.len(),
323        })
324    }
325
326    fn append(&mut self, fresh: &[Header], now: u32) -> Result<(), Error> {
327        let original = self.headers.len();
328        for header in fresh {
329            match self.validate_next(&self.headers, header, now) {
330                Ok(()) => {
331                    self.index.insert(header.block_hash(), self.headers.len());
332                    self.headers.push(*header);
333                }
334                Err(error) => {
335                    for rejected in &self.headers[original..] {
336                        self.index.remove(&rejected.block_hash());
337                    }
338                    self.headers.truncate(original);
339                    return Err(error);
340                }
341            }
342        }
343        Ok(())
344    }
345
346    fn validate_next(&self, window: &[Header], header: &Header, now: u32) -> Result<(), Error> {
347        self.validate_context(window, header)?;
348        let horizon = now
349            .checked_add(P2P_MAX_FUTURE_SECS)
350            .context("clock horizon overflow")?;
351        ensure!(
352            header.time <= horizon,
353            "peer header time is too far in the future"
354        );
355        Ok(())
356    }
357
358    fn validate_context(&self, window: &[Header], header: &Header) -> Result<(), Error> {
359        let last = window.last().context("header window is empty")?;
360        ensure!(
361            header.prev_blockhash == last.block_hash(),
362            "peer header does not extend its predecessor"
363        );
364        let (first_height, context) = self.preceding_context(window)?;
365        if window.len() == 1 && self.trusted_next.is_some() {
366            ensure!(
367                Some(header.block_hash()) == self.trusted_next,
368                "peer header after the checkpoint is not the checkpointed block"
369            );
370        } else {
371            let required = expected_bits(self.chain, &context, first_height, header.time).map_err(
372                |cause| match cause {
373                    cause @ (PowError::ShortWindow { .. }
374                    | PowError::MissingDifficultyContext { .. }) => Error::HeaderContext(cause),
375                    cause => Error::Invalid(cause.to_string()),
376                },
377            )?;
378            ensure!(
379                header.bits == required,
380                "peer header bits {:08x} differ from the required {:08x}",
381                header.bits.to_consensus(),
382                required.to_consensus()
383            );
384        }
385        ensure!(
386            header.time > median_time(&context),
387            "peer header time is not after the median of the previous blocks"
388        );
389        Ok(())
390    }
391
392    fn preceding_context<'a>(
393        &self,
394        window: &'a [Header],
395    ) -> Result<(u64, Cow<'a, [Header]>), Error> {
396        let Some(ancestors) = &self.ancestors else {
397            return Ok((self.start_height, Cow::Borrowed(window)));
398        };
399        let bound = context_headers(self.chain) + 1;
400        let next_height = self
401            .start_height
402            .checked_add(u64::try_from(window.len())?)
403            .context("header context height overflow")?;
404        let tail = if window.len() >= bound {
405            Cow::Borrowed(&window[window.len() - bound..])
406        } else {
407            let count = (bound - window.len()).min(ancestors.len());
408            let mut tail = Vec::with_capacity(count + window.len());
409            tail.extend_from_slice(&ancestors[ancestors.len() - count..]);
410            tail.extend_from_slice(window);
411            Cow::Owned(tail)
412        };
413        let first_height = next_height
414            .checked_sub(u64::try_from(tail.len())?)
415            .context("header context precedes genesis")?;
416        Ok((first_height, tail))
417    }
418}
419
420pub fn prove_batch(chain: Chain, headers: &[Header]) -> Result<(), Error> {
421    if headers.is_empty() {
422        return Ok(());
423    }
424    let workers = std::thread::available_parallelism()?
425        .get()
426        .min(P2P_POW_WORKERS)
427        .min(headers.len());
428    let chunk = headers.len().div_ceil(workers);
429    let verdicts: Vec<Result<(), (usize, String)>> = std::thread::scope(|scope| {
430        let mut handles = Vec::new();
431        for (slot, part) in headers.chunks(chunk).enumerate() {
432            handles.push(scope.spawn(move || {
433                for (offset, header) in part.iter().enumerate() {
434                    check_header_pow(header, chain)
435                        .map_err(|cause| (slot * chunk + offset, cause.to_string()))?;
436                }
437                Ok(())
438            }));
439        }
440        let mut verdicts = Vec::new();
441        for handle in handles {
442            match handle.join() {
443                Ok(verdict) => verdicts.push(verdict),
444                Err(panic) => {
445                    tracing::error!(?panic, "header proof worker panicked");
446                    verdicts.push(Err((usize::MAX, "header proof worker panicked".into())));
447                }
448            }
449        }
450        verdicts
451    });
452    for verdict in verdicts {
453        match verdict {
454            Ok(()) => (),
455            Err((offset, cause)) => {
456                return Err(Error::Invalid(format!(
457                    "header at batch offset {offset} fails proof of work: {cause}"
458                )));
459            }
460        }
461    }
462    Ok(())
463}
464
465fn median_time(window: &[Header]) -> u32 {
466    let start = window.len().max(P2P_MEDIAN_TIME_SPAN) - P2P_MEDIAN_TIME_SPAN;
467    let mut times: Vec<u32> = window[start..].iter().map(|header| header.time).collect();
468    times.sort_unstable();
469    times[times.len() / 2]
470}
urma_runtime/p2p/headers.rs

urma_runtime/p2p/
headers.rs
