Skip to main content

sidestr_round/
node.rs

1//! One signer of a federated chain, running (feature `bin`): the chain on
2//! disk, the block round and the peg-out round driven every second, relays
3//! followed and published to, the parent polled when there is one, and
4//! what a mirror needs served over HTTP. `bin/siding.mjs produce` on a
5//! level-2 document, in Rust, minus what is not level 2's (the desk, the
6//! EVM rule, checkpoints).
7//!
8//! Everything that owns state runs on one thread — `sidestr-core`'s state
9//! is not `Send` — so the HTTP server asks the loop over a channel, and
10//! only the relay sockets are separate tasks.
11
12use std::path::{Path, PathBuf};
13use std::time::Duration;
14
15use bitcoin::{OutPoint, Txid};
16use serde::{Deserialize, Serialize};
17use sidestr_core::block::{HeaderFamily, SidestrBlock};
18use sidestr_core::blockfile::HEADER;
19use sidestr_core::chain::ChainOf;
20use sidestr_core::document::ChainDocument;
21use sidestr_core::marker::{parse_claims, parse_peg_marker};
22use sidestr_core::parent::rpc::CoreRpc;
23use sidestr_core::parent::{
24    claimable, outpoints_to_lock, owned_by_peg_wallet, paid_pegouts_in, parent_network,
25    scan_pegins, FoundPegin, ParentRpc, PegOwner, PegWallet,
26};
27use sidestr_nostr::event::Event;
28use sidestr_nostr::kinds::{
29    KIND_BLOCK_PROPOSAL, KIND_PARTIAL_SIGNATURE, KIND_PEGOUT_PSBT, KIND_PEGOUT_SIGNED,
30    KIND_SEALED_BLOCK, KIND_TRANSACTION,
31};
32use sidestr_nostr::relay::Follower;
33use sidestr_nostr::tip::{sign_tip, TipTemplate, TIP_HEADERS};
34use tokio::sync::{mpsc, oneshot};
35
36use crate::error::{Error, Result};
37use crate::journal::{FileJournal, VoteJournal};
38use crate::pegout::{
39    burn_key, PaidPegout, PegCoin, PegoutAction, PegoutConfig, PegoutLedger, PegoutRound,
40};
41use crate::relay::{follow, ok_count, publish_all, unix_now, unix_now_ms};
42use crate::round::{Action, ClaimChecker, Round, RoundConfig};
43use crate::signer::LocalKey;
44
45/// How a signer is run.
46#[derive(Debug, Clone)]
47pub struct Settings {
48    /// The chain document.
49    pub chain: PathBuf,
50    /// The block file directory; `pegins.json`, `pegouts.json` and the
51    /// journal live beside it.
52    pub dir: PathBuf,
53    /// The key file (32-byte hex). Never a command-line value.
54    pub key_file: PathBuf,
55    /// The HTTP port on 127.0.0.1. `/blocks.dat` serves only what the
56    /// accepted index covers; `POST /tx` over [`MAX_TX_BODY`] bytes is 413.
57    pub port: u16,
58    /// Seconds between blocks when the mempool is empty.
59    pub interval: u64,
60    /// Seconds between blocks when it is not.
61    pub tx_interval: u64,
62    /// The round's options.
63    pub round: RoundConfig,
64    /// The peg-out round's options (network filled from the document).
65    pub pegout: PegoutConfig,
66    /// Relays to follow and publish to.
67    pub relays: Vec<String>,
68    /// Mirrors to name in the tip announcement; none means no announcement.
69    pub mirrors: Vec<String>,
70    /// The parent node's JSON-RPC, cookie file and peg wallet.
71    pub parent: Option<ParentSettings>,
72    /// The block round's vote journal; default `<dir>/votes.jsonl`. The
73    /// peg-out round journals beside it in `<stem>-pegout.jsonl` (default
74    /// `<dir>/votes-pegout.jsonl`): one file per round, one writer per file.
75    pub journal: Option<PathBuf>,
76}
77
78/// The peg-out round's journal beside the block round's: `votes.jsonl` →
79/// `votes-pegout.jsonl`.
80pub fn pegout_journal_path_for(journal: &Path) -> PathBuf {
81    let stem = journal
82        .file_stem()
83        .map(|s| s.to_string_lossy().into_owned())
84        .unwrap_or_else(|| "votes".into());
85    let ext = journal
86        .extension()
87        .map(|e| format!(".{}", e.to_string_lossy()))
88        .unwrap_or_default();
89    journal.with_file_name(format!("{stem}-pegout{ext}"))
90}
91
92/// Before 0.1.0 shipped, both rounds wrote one file. If the block journal
93/// holds burn-scoped entries and the peg-out journal does not exist yet,
94/// copy those entries across once, so a signer restarting on an old
95/// directory keeps every burn authorisation it made (never re-signs a burn
96/// it already authorised). The block journal is left as it was: the block
97/// round ignores burn scopes when it reloads.
98fn split_combined_journal(journal: &Path, pegout_journal: &Path) -> Result<()> {
99    if pegout_journal.exists() || !journal.exists() {
100        return Ok(());
101    }
102    let combined = FileJournal::open(journal)?;
103    let burns: Vec<_> = combined
104        .entries()?
105        .into_iter()
106        .filter(|e| matches!(e.scope, crate::journal::VoteScope::Burn(_)))
107        .collect();
108    drop(combined);
109    if burns.is_empty() {
110        return Ok(());
111    }
112    let mut split = FileJournal::open(pegout_journal)?;
113    for e in &burns {
114        split.record(e)?;
115    }
116    Ok(())
117}
118
119/// A parent view.
120#[derive(Debug, Clone)]
121pub struct ParentSettings {
122    /// `http://127.0.0.1:48332/`.
123    pub url: String,
124    /// The node's cookie file.
125    pub cookie: PathBuf,
126    /// The peg wallet's name, if peg-outs are to be paid from here.
127    pub wallet: Option<String>,
128    /// Seconds between polls.
129    pub poll: u64,
130    /// The first parent height to scan for peg-ins.
131    pub from: u32,
132}
133
134/// `pegins.json`: the scan position and what was found (`bin/siding.mjs pegState`).
135#[derive(Debug, Clone, Default, Serialize, Deserialize)]
136struct PeginState {
137    scanned: i64,
138    pegins: Vec<PeginRecord>,
139}
140
141#[derive(Debug, Clone, Serialize, Deserialize)]
142#[serde(rename_all = "camelCase")]
143struct PeginRecord {
144    txid: String,
145    vout: u32,
146    amount: u64,
147    script: String,
148    height: u32,
149    #[serde(default, skip_serializing_if = "Option::is_none")]
150    parent_address: Option<String>,
151}
152
153impl From<&FoundPegin> for PeginRecord {
154    fn from(p: &FoundPegin) -> Self {
155        Self {
156            txid: p.txid.clone(),
157            vout: p.vout,
158            amount: p.amount,
159            script: p.script.to_hex_string(),
160            height: p.height,
161            parent_address: p.parent_address.clone(),
162        }
163    }
164}
165
166impl PeginRecord {
167    fn found(&self) -> FoundPegin {
168        FoundPegin {
169            txid: self.txid.clone(),
170            vout: self.vout,
171            amount: self.amount,
172            script: bitcoin::ScriptBuf::from_bytes(hex::decode(&self.script).unwrap_or_default()),
173            height: self.height,
174            parent_address: self.parent_address.clone(),
175        }
176    }
177}
178
179/// `bin/siding.mjs checkClaims`: every claim in a proposal is a confirmed,
180/// unspent peg-in on the parent paying the peg, whose marker names the
181/// script the claim pays.
182struct ParentClaims {
183    rpc: std::rc::Rc<CoreRpc>,
184    chain_id: String,
185    need: u32,
186    challenge: bitcoin::ScriptBuf,
187}
188
189impl<F: HeaderFamily> ClaimChecker<F> for ParentClaims {
190    fn check(&self, block: &F::Block) -> Option<String> {
191        let coinbase = block.txdata().first()?;
192        let (claims, errors) = parse_claims(coinbase);
193        if let Some(e) = errors.first() {
194            return Some(e.clone());
195        }
196        for c in claims {
197            let short = &c.txid[..c.txid.len().min(12)];
198            let txid: Txid = c.txid.parse().ok()?;
199            let o = match self.rpc.tx_out(&txid, c.vout) {
200                Ok(Some(o)) => o,
201                Ok(None) => {
202                    return Some(format!("{short}…:{} is not unspent on the parent", c.vout))
203                }
204                Err(e) => return Some(e.to_string()),
205            };
206            if o.confirmations < self.need {
207                return Some(format!(
208                    "{short}… has {} of {} confirmations",
209                    o.confirmations, self.need
210                ));
211            }
212            if o.value != c.payout.value || o.script_pubkey != self.challenge {
213                return Some(format!(
214                    "{short}… does not pay the peg {} sats",
215                    c.payout.value
216                ));
217            }
218            let raw = match self
219                .rpc
220                .call("getrawtransaction", serde_json::json!([c.txid, true]))
221            {
222                Ok(v) => v,
223                Err(e) => return Some(e.to_string()),
224            };
225            let marker = raw["vout"]
226                .as_array()
227                .into_iter()
228                .flatten()
229                .filter_map(|v| v["scriptPubKey"]["hex"].as_str())
230                .filter_map(|h| hex::decode(h).ok())
231                .find_map(|b| parse_peg_marker(bitcoin::Script::from_bytes(&b), &self.chain_id));
232            if marker.as_deref() != Some(c.payout.script_pubkey.as_script()) {
233                return Some(format!(
234                    "{short}…'s marker names {}…, the claim pays {}…",
235                    marker
236                        .map(|m| m.to_hex_string())
237                        .unwrap_or_else(|| "nothing".into())
238                        .chars()
239                        .take(12)
240                        .collect::<String>(),
241                    &c.payout.script_pubkey.to_hex_string()[..12]
242                ));
243            }
244        }
245        None
246    }
247}
248
249/// The most a `POST /tx` body may be, bytes; more is 413.
250pub const MAX_TX_BODY: usize = 262_144;
251
252/// `/blocks.dat` as the loop answers it: only the bytes the accepted index
253/// covers, and a `Range` bounded to them.
254struct DatReply {
255    code: u16,
256    body: Vec<u8>,
257    /// `Content-Range`, when there is one.
258    content_range: Option<String>,
259}
260
261/// What the HTTP thread asks the loop.
262enum Query {
263    Status(oneshot::Sender<serde_json::Value>),
264    /// The block file, bounded to the accepted index; the `Range` header
265    /// if there was one.
266    Dat(Option<String>, oneshot::Sender<DatReply>),
267    Tip(oneshot::Sender<serde_json::Value>),
268    Chain(oneshot::Sender<serde_json::Value>),
269    Blocks(oneshot::Sender<serde_json::Value>),
270    Pegouts(oneshot::Sender<serde_json::Value>),
271    Coins(String, oneshot::Sender<serde_json::Value>),
272    Tx(
273        String,
274        oneshot::Sender<core::result::Result<serde_json::Value, String>>,
275    ),
276}
277
278fn log(s: impl AsRef<str>) {
279    let now = unix_now();
280    let (h, m, sec) = ((now / 3600) % 24, (now / 60) % 60, now % 60);
281    println!("{h:02}:{m:02}:{sec:02} {}", s.as_ref());
282}
283
284fn read_json<T: for<'a> Deserialize<'a> + Default>(path: &Path) -> T {
285    std::fs::read_to_string(path)
286        .ok()
287        .and_then(|t| serde_json::from_str(&t).ok())
288        .unwrap_or_default()
289}
290
291fn write_json<T: Serialize>(path: &Path, v: &T) {
292    if let Ok(text) = serde_json::to_string_pretty(v) {
293        if let Err(e) = std::fs::write(path, text) {
294            log(format!("{}: {e}", path.display()));
295        }
296    }
297}
298
299/// The signer's state, on the loop's thread.
300struct Node<F: HeaderFamily> {
301    doc: ChainDocument,
302    chain: ChainOf<F>,
303    round: Round<F>,
304    pegout: Option<PegoutRound>,
305    parent: Option<std::rc::Rc<CoreRpc>>,
306    pegins: PeginState,
307    settings: Settings,
308    txs: Follower,
309    key: LocalKey,
310    last_block: u64,
311    announced: Option<u32>,
312    announce_retry_at: u64,
313    started: u64,
314}
315
316impl<F: HeaderFamily> Node<F> {
317    fn status(&self) -> serde_json::Value {
318        let s = self.chain.state();
319        let tip = s.tip();
320        let fed = self.round.federation();
321        serde_json::json!({
322            "chain": self.doc.id, "parent": self.doc.parent,
323            "height": tip.height, "hash": tip.hash.to_string(), "time": tip.time,
324            "coins": s.utxo().len(), "mempool": s.mempool().count(), "minFeeRate": s.min_fee_rate(),
325            "relays": self.settings.relays,
326            "announce": if self.settings.mirrors.is_empty() { serde_json::Value::Null } else { serde_json::json!({"mirrors": self.settings.mirrors, "announced": self.announced.map(i64::from).unwrap_or(-1)}) },
327            "pegouts": {"burned": s.pegouts().len(), "paid": self.pegout.as_ref().map(|p| p.ledger().paid.len()).unwrap_or(0), "min": s.pegout_min(), "payer": self.settings.parent.as_ref().and_then(|p| p.wallet.clone())},
328            "pegins": self.parent.as_ref().map(|_| serde_json::json!({"scanned": self.pegins.scanned, "known": self.pegins.pegins.len(), "claimed": self.pegins.pegins.iter().filter(|p| s.claimed(&p.txid, p.vout)).count()})),
329            "signer": self.key.pubkey_hex(), "genesis": s.genesis_hash().to_string(), "interval": self.settings.interval,
330            "level2": {"signers": fed.signers.len(), "threshold": fed.threshold, "slot": self.round.slot() + 1,
331                        "proposeAfter": self.settings.round.propose_after, "resignAfter": self.settings.round.resign_after,
332                        "pending": self.round.pending().map(|p| serde_json::json!({"height": p.height, "signatures": p.sigs.len(), "at": p.at})),
333                        "journal": self.settings.journal.as_ref().map(|j| j.display().to_string())},
334            "engine": "sidestr-round", "started": self.started,
335        })
336    }
337
338    fn tip_json(&self) -> serde_json::Value {
339        let t = self.chain.state().tip();
340        serde_json::json!({"height": t.height, "hash": t.hash.to_string(), "time": t.time})
341    }
342
343    /// The bytes of the block file the accepted index covers: from its
344    /// first entry to the end of its last. A tail the index does not name
345    /// — a block appended but not (yet) accepted — is not served, so what a
346    /// mirror reads and what `/tip` says are the same chain. Read on the
347    /// loop's thread, where nothing can accept a block meanwhile.
348    fn committed_dat(&self, range: Option<&str>) -> DatReply {
349        let end = self
350            .chain
351            .index()
352            .blocks
353            .last()
354            .map(|e| e.offset + HEADER + u64::from(e.size))
355            .unwrap_or(0);
356        let bytes = match std::fs::read(self.chain.dat_path()) {
357            Ok(b) => b,
358            Err(e) => {
359                return DatReply {
360                    code: 500,
361                    body: serde_json::json!({"error": e.to_string()})
362                        .to_string()
363                        .into_bytes(),
364                    content_range: None,
365                }
366            }
367        };
368        if (bytes.len() as u64) < end {
369            return DatReply {
370                code: 500,
371                body: serde_json::json!({"error": "the block file is shorter than its index"})
372                    .to_string()
373                    .into_bytes(),
374                content_range: None,
375            };
376        }
377        let committed = &bytes[..end as usize];
378        match range {
379            None => DatReply {
380                code: 200,
381                body: committed.to_vec(),
382                content_range: None,
383            },
384            Some(h) => match parse_range(h, end) {
385                Some((s, e)) => DatReply {
386                    code: 206,
387                    body: committed[s as usize..=e as usize].to_vec(),
388                    content_range: Some(format!("bytes {s}-{e}/{end}")),
389                },
390                None => DatReply {
391                    code: 416,
392                    body: Vec::new(),
393                    content_range: Some(format!("bytes */{end}")),
394                },
395            },
396        }
397    }
398
399    fn answer(&mut self, q: Query) {
400        match q {
401            Query::Status(r) => {
402                let _ = r.send(self.status());
403            }
404            Query::Tip(r) => {
405                let _ = r.send(self.tip_json());
406            }
407            Query::Dat(range, r) => {
408                let _ = r.send(self.committed_dat(range.as_deref()));
409            }
410            Query::Chain(r) => {
411                let mut d = self.doc.clone();
412                d.genesis_hash = Some(self.chain.state().genesis_hash().to_string());
413                let _ = r.send(serde_json::to_value(&d).unwrap_or_default());
414            }
415            Query::Blocks(r) => {
416                let _ = r.send(serde_json::to_value(self.chain.index()).unwrap_or_default());
417            }
418            Query::Pegouts(r) => {
419                let _ = r.send(
420                    self.pegout
421                        .as_ref()
422                        .map(|p| serde_json::to_value(p.ledger()).unwrap_or_default())
423                        .unwrap_or_else(|| serde_json::json!({"paid": {}})),
424                );
425            }
426            Query::Coins(hex_spk, r) => {
427                let list: Vec<serde_json::Value> = hex::decode(&hex_spk)
428                    .map(|b| self.chain.state().coins(bitcoin::Script::from_bytes(&b)))
429                    .unwrap_or_default()
430                    .into_iter()
431                    .map(|c| serde_json::json!({"outpoint": format!("{}:{}", c.outpoint.txid, c.outpoint.vout), "value": c.value, "height": c.height, "coinbase": c.coinbase}))
432                    .collect();
433                let _ = r.send(serde_json::Value::Array(list));
434            }
435            Query::Tx(hex_tx, r) => {
436                let res = hex::decode(hex_tx.trim())
437                    .map_err(|e| e.to_string())
438                    .and_then(|b| self.chain.submit(&b).map_err(|e| e.to_string()))
439                    .map(|s| {
440                        log(format!("tx {}… accepted, fee {}", &s.txid.to_string()[..16], s.fee));
441                        serde_json::json!({"txid": s.txid.to_string(), "fee": s.fee, "vsize": s.vsize, "dup": s.dup})
442                    });
443                let _ = r.send(res);
444            }
445        }
446    }
447
448    fn headers_hex(&self) -> Vec<String> {
449        let s = self.chain.state();
450        let tip = s.height();
451        let from = tip.saturating_sub(TIP_HEADERS as u32 - 1);
452        (from..=tip)
453            .filter_map(|h| s.header_at(h))
454            .map(|h| hex::encode(s.family().encode_header(h)))
455            .collect()
456    }
457
458    fn peg_coins(&self) -> Vec<PegCoin> {
459        let (Some(rpc), Some(p)) = (&self.parent, &self.settings.parent) else {
460            return vec![];
461        };
462        if p.wallet.is_none() {
463            return vec![];
464        }
465        let challenge = self.chain.state().challenge().to_hex_string();
466        match rpc.wallet_call("listunspent", serde_json::json!([1, 9_999_999, [], true])) {
467            Ok(v) => v
468                .as_array()
469                .into_iter()
470                .flatten()
471                .filter(|u| u["scriptPubKey"].as_str() == Some(challenge.as_str()))
472                .filter_map(|u| {
473                    Some(PegCoin {
474                        outpoint: OutPoint {
475                            txid: u["txid"].as_str()?.parse().ok()?,
476                            vout: u32::try_from(u["vout"].as_u64()?).ok()?,
477                        },
478                        value: (u["amount"].as_f64()? * 1e8).round() as u64,
479                    })
480                })
481                .collect(),
482            Err(e) => {
483                log(format!("peg-out round: listunspent: {e}"));
484                vec![]
485            }
486        }
487    }
488
489    /// `bin/siding.mjs pegTick`: scan new parent blocks for peg-ins, hand
490    /// the claimable ones to the round, keep them out of the wallet's own
491    /// payments until claimed.
492    fn peg_tick(&mut self) {
493        let Some(rpc) = self.parent.clone() else {
494            return;
495        };
496        let network = self.doc.parent().ok().and_then(parent_network);
497        let tip = match rpc.block_count() {
498            Ok(t) => t,
499            Err(e) => {
500                log(format!("parent: {e}"));
501                return;
502            }
503        };
504        if i64::from(tip) > self.pegins.scanned {
505            let from = u32::try_from(self.pegins.scanned + 1).unwrap_or(0);
506            // SPEC 6 (0.0.3): with a peg wallet, the peg is the output it owns
507            // (the k-of-n descriptor it imported); without one, the first taproot
508            // output (`parent.mjs scanPegins`, `parent.walletRpc`)
509            let wallet_owner = owned_by_peg_wallet(rpc.as_ref());
510            let owner: Option<PegOwner<'_>> = rpc.wallet().is_some().then_some(&wallet_owner);
511            match scan_pegins(
512                rpc.as_ref(),
513                &self.doc.id,
514                from,
515                tip,
516                network,
517                owner,
518                |_| {},
519            ) {
520                Ok(found) => {
521                    for p in &found {
522                        if !self
523                            .pegins
524                            .pegins
525                            .iter()
526                            .any(|q| q.txid == p.txid && q.vout == p.vout)
527                        {
528                            log(format!(
529                                "peg-in {}…:{}: {} sats to {}…, parent h{}",
530                                &p.txid[..16],
531                                p.vout,
532                                p.amount,
533                                &p.script.to_hex_string()[..12],
534                                p.height
535                            ));
536                            self.pegins.pegins.push(p.into());
537                        }
538                    }
539                    self.pegins.scanned = i64::from(tip);
540                    write_json(&self.settings.dir.join("pegins.json"), &self.pegins);
541                }
542                Err(e) => log(format!("peg-in scan: {e}")),
543            }
544        }
545        let found: Vec<FoundPegin> = self.pegins.pegins.iter().map(PeginRecord::found).collect();
546        let s = self.chain.state();
547        let claims = claimable(&found, tip, self.doc.peg_confirmations, |t, v| {
548            s.claimed(t, v)
549        });
550        if self
551            .settings
552            .parent
553            .as_ref()
554            .is_some_and(|p| p.wallet.is_some())
555        {
556            let lock = outpoints_to_lock(&found, |t, v| s.claimed(t, v));
557            let unlock: Vec<OutPoint> = found
558                .iter()
559                .filter(|p| s.claimed(&p.txid, p.vout))
560                .filter_map(|p| {
561                    Some(OutPoint {
562                        txid: p.txid.parse().ok()?,
563                        vout: p.vout,
564                    })
565                })
566                .collect();
567            if !lock.is_empty() {
568                if let Err(e) = rpc.lock_outputs(&lock, true) {
569                    log(format!("lockunspent: {e}"));
570                }
571            }
572            if !unlock.is_empty() {
573                if let Err(e) = rpc.lock_outputs(&unlock, false) {
574                    log(format!("lockunspent: {e}"));
575                }
576            }
577        }
578        // empty included: a claim sealed by another signer must leave the list
579        self.round.want_claims(claims);
580    }
581
582    /// `bin/siding.mjs reconcile`: burns the wallet's history shows paid
583    /// (by another signer, or before a crash) are recorded, not paid twice.
584    fn reconcile(&mut self) {
585        let (Some(rpc), Some(round)) = (&self.parent, &mut self.pegout) else {
586            return;
587        };
588        if self
589            .settings
590            .parent
591            .as_ref()
592            .is_none_or(|p| p.wallet.is_none())
593        {
594            return;
595        }
596        let sent = match rpc.sent_transactions() {
597            Ok(s) => s,
598            Err(e) => {
599                log(format!("peg-out reconcile: {e}"));
600                return;
601            }
602        };
603        let paid = paid_pegouts_in(&sent, &self.doc.id);
604        let mut changed = false;
605        for b in self.chain.state().pegouts() {
606            let key = burn_key(&b);
607            if round.ledger().paid.contains_key(&key) {
608                continue;
609            }
610            if let Some(txid) = paid.get(&b.txid) {
611                round.mark_paid(
612                    &key,
613                    PaidPegout {
614                        parent_txid: txid.to_string(),
615                        address: None,
616                        value: b.value,
617                        script: b.script.clone(),
618                        height: b.height,
619                        at: unix_now(),
620                        signers: vec![],
621                        reconciled: Some(true),
622                    },
623                );
624                log(format!(
625                    "peg-out {}… was paid by the federation in {}…",
626                    &key[..16],
627                    &txid.to_string()[..16]
628                ));
629                changed = true;
630            }
631        }
632        if changed {
633            write_json(&self.settings.dir.join("pegouts.json"), round.ledger());
634        }
635    }
636}
637
638fn parse_range(h: &str, size: u64) -> Option<(u64, u64)> {
639    let r = h.strip_prefix("bytes=")?;
640    let (a, b) = r.split_once('-')?;
641    let start: u64 = a.parse().ok()?;
642    let end: u64 = if b.is_empty() {
643        size.saturating_sub(1)
644    } else {
645        b.parse().ok()?
646    };
647    (start <= end && end < size).then_some((start, end))
648}
649
650fn serve_http(port: u16, to_loop: mpsc::UnboundedSender<Query>) -> Result<()> {
651    let server = tiny_http::Server::http(("127.0.0.1", port))
652        .map_err(|e| Error::Io(std::io::Error::other(e.to_string())))?;
653    std::thread::spawn(move || {
654        for mut req in server.incoming_requests() {
655            let path = req.url().split('?').next().unwrap_or("/").to_string();
656            let cors = [
657                ("access-control-allow-origin", "*"),
658                ("access-control-allow-headers", "range, content-type"),
659                ("access-control-allow-methods", "GET, POST, OPTIONS"),
660            ];
661            let with_cors = |mut r: tiny_http::Response<std::io::Cursor<Vec<u8>>>| {
662                for (k, v) in cors {
663                    r = r.with_header(tiny_http::Header::from_bytes(k, v).expect("static header"));
664                }
665                r
666            };
667            let json = |code: u16, v: &serde_json::Value| {
668                with_cors(
669                    tiny_http::Response::from_string(v.to_string())
670                        .with_status_code(code)
671                        .with_header(
672                            tiny_http::Header::from_bytes("content-type", "application/json")
673                                .expect("static header"),
674                        ),
675                )
676            };
677            let ask = |q: Query, rx: oneshot::Receiver<serde_json::Value>| {
678                let _ = to_loop.send(q);
679                rx.blocking_recv().unwrap_or(serde_json::Value::Null)
680            };
681            if req.method() == &tiny_http::Method::Options {
682                let _ = req.respond(with_cors(
683                    tiny_http::Response::from_data(Vec::new()).with_status_code(204),
684                ));
685                continue;
686            }
687            let response = match (req.method().as_str(), path.as_str()) {
688                ("GET", "/") | ("GET", "/status.json") => {
689                    let (tx, rx) = oneshot::channel();
690                    json(200, &ask(Query::Status(tx), rx))
691                }
692                ("GET", "/tip") => {
693                    let (tx, rx) = oneshot::channel();
694                    json(200, &ask(Query::Tip(tx), rx))
695                }
696                ("GET", "/chain.json") => {
697                    let (tx, rx) = oneshot::channel();
698                    json(200, &ask(Query::Chain(tx), rx))
699                }
700                ("GET", "/blocks.json") => {
701                    let (tx, rx) = oneshot::channel();
702                    json(200, &ask(Query::Blocks(tx), rx))
703                }
704                ("GET", "/pegouts.json") => {
705                    let (tx, rx) = oneshot::channel();
706                    json(200, &ask(Query::Pegouts(tx), rx))
707                }
708                ("GET", "/blocks.dat") => {
709                    let range = req
710                        .headers()
711                        .iter()
712                        .find(|h| h.field.equiv("range"))
713                        .map(|h| h.value.as_str().to_string());
714                    let (tx, rx) = oneshot::channel();
715                    let _ = to_loop.send(Query::Dat(range, tx));
716                    match rx.blocking_recv() {
717                        Ok(d) if d.code == 500 => {
718                            let v: serde_json::Value =
719                                serde_json::from_slice(&d.body).unwrap_or_default();
720                            json(500, &v)
721                        }
722                        Ok(d) => {
723                            let mut r = with_cors(
724                                tiny_http::Response::from_data(d.body).with_status_code(d.code),
725                            )
726                            .with_header(
727                                tiny_http::Header::from_bytes(
728                                    "content-type",
729                                    "application/octet-stream",
730                                )
731                                .expect("static header"),
732                            )
733                            .with_header(
734                                tiny_http::Header::from_bytes("accept-ranges", "bytes")
735                                    .expect("static header"),
736                            );
737                            if let Some(cr) = d.content_range {
738                                r = r.with_header(
739                                    tiny_http::Header::from_bytes("content-range", cr.as_str())
740                                        .expect("static header"),
741                                );
742                            }
743                            r
744                        }
745                        Err(_) => json(500, &serde_json::json!({"error": "the signer is gone"})),
746                    }
747                }
748                ("GET", p) if p.starts_with("/coins/") => {
749                    let (tx, rx) = oneshot::channel();
750                    json(200, &ask(Query::Coins(p[7..].to_ascii_lowercase(), tx), rx))
751                }
752                ("POST", "/tx") => {
753                    let too_large = json(
754                        413,
755                        &serde_json::json!({"error": format!("the body is over {MAX_TX_BODY} bytes")}),
756                    );
757                    if req.body_length().is_some_and(|n| n > MAX_TX_BODY) {
758                        let _ = req.respond(too_large);
759                        continue;
760                    }
761                    let mut body = String::new();
762                    let read = std::io::Read::read_to_string(
763                        &mut std::io::Read::take(req.as_reader(), MAX_TX_BODY as u64 + 1),
764                        &mut body,
765                    );
766                    if read.is_err() || body.len() > MAX_TX_BODY {
767                        let _ = req.respond(too_large);
768                        continue;
769                    }
770                    let (tx, rx) = oneshot::channel();
771                    let _ = to_loop.send(Query::Tx(body, tx));
772                    match rx.blocking_recv() {
773                        Ok(Ok(v)) => json(200, &v),
774                        Ok(Err(e)) => json(400, &serde_json::json!({"error": e})),
775                        Err(_) => json(500, &serde_json::json!({"error": "the signer is gone"})),
776                    }
777                }
778                _ => json(404, &serde_json::json!({"error": "not found"})),
779            };
780            let _ = req.respond(response);
781        }
782    });
783    Ok(())
784}
785
786/// Run one signer until the process ends. Generic over the header family;
787/// [`run`] picks it from the document.
788pub async fn run_as<F: HeaderFamily>(doc: ChainDocument, settings: Settings) -> Result<()> {
789    let key_text = std::fs::read_to_string(&settings.key_file)
790        .map_err(|e| Error::Key(format!("{}: {e}", settings.key_file.display())))?;
791    let key = LocalKey::from_hex(&key_text)?;
792    let dir = settings.dir.clone();
793    std::fs::create_dir_all(&dir)?;
794    let chain = ChainOf::<F>::open_sealed(doc.clone(), &dir, |_| {
795        Err(sidestr_core::Error::Chain(
796            "no block file: copy blocks.dat and blocks.json from a mirror of this chain first"
797                .into(),
798        ))
799    })?;
800    let journal_path = settings
801        .journal
802        .clone()
803        .unwrap_or_else(|| dir.join("votes.jsonl"));
804    // one journal file per round: the block round and the peg-out round each
805    // hold their own exclusive handle (journal.rs "One writer per file")
806    let pegout_journal_path = pegout_journal_path_for(&journal_path);
807    split_combined_journal(&journal_path, &pegout_journal_path)?;
808    let journal = FileJournal::open(&journal_path)?;
809    let loaded = journal.entries()?.len();
810    let mut round = Round::new(
811        chain.state(),
812        Box::new(LocalKey::from_hex(&key_text)?),
813        Box::new(journal),
814        settings.round.clone(),
815    )?;
816    let fed = round.federation().clone();
817    let network = doc.parent().ok().and_then(parent_network);
818    let parent = settings
819        .parent
820        .as_ref()
821        .map(|p| std::rc::Rc::new(CoreRpc::new(&p.url, &p.cookie, p.wallet.as_deref())));
822    if let Some(rpc) = &parent {
823        round = round.with_claim_checker(Box::new(ParentClaims {
824            rpc: rpc.clone(),
825            chain_id: doc.id.clone(),
826            need: doc.peg_confirmations,
827            challenge: chain.state().challenge().to_owned(),
828        }));
829    }
830    let pegout = match (&parent, &settings.parent) {
831        (Some(_), Some(p)) if p.wallet.is_some() => Some(PegoutRound::new(
832            fed.clone(),
833            &doc.id,
834            Box::new(LocalKey::from_hex(&key_text)?),
835            Box::new(FileJournal::open(&pegout_journal_path)?),
836            PegoutConfig {
837                network,
838                ..settings.pegout.clone()
839            },
840            read_json::<PegoutLedger>(&dir.join("pegouts.json")),
841        )?),
842        _ => None,
843    };
844    let mut pegins: PeginState = read_json(&dir.join("pegins.json"));
845    if pegins.pegins.is_empty() && pegins.scanned == 0 {
846        pegins.scanned = i64::from(settings.parent.as_ref().map(|p| p.from).unwrap_or(0)) - 1;
847    }
848    let now = unix_now();
849    let mut node = Node {
850        txs: Follower::new(KIND_TRANSACTION, &doc.id),
851        doc,
852        chain,
853        round,
854        pegout,
855        parent,
856        pegins,
857        settings: settings.clone(),
858        key,
859        last_block: now,
860        announced: None,
861        announce_retry_at: 0,
862        started: now,
863    };
864    log(format!(
865        "level 2: signer {} of {}, threshold {}, proposing after {} s when it is another signer's turn; journal {} ({loaded} entries)",
866        node.round.slot() + 1,
867        fed.signers.len(),
868        fed.threshold,
869        settings.round.propose_after,
870        journal_path.display()
871    ));
872    log(format!(
873        "chain {} at {} ({} coins), {} relay(s), {} mirror(s), parent {}",
874        node.doc.id,
875        node.chain.state().height(),
876        node.chain.state().utxo().len(),
877        settings.relays.len(),
878        settings.mirrors.len(),
879        settings
880            .parent
881            .as_ref()
882            .map(|p| p.url.as_str())
883            .unwrap_or("none")
884    ));
885    if let Some(p) = node.pegout.as_ref() {
886        log(format!(
887            "parent wallet {}: peg-outs for {} are paid by the federation's PSBT round ({} of {}), {} paid so far",
888            settings.parent.as_ref().and_then(|p| p.wallet.clone()).unwrap_or_default(),
889            node.doc.id,
890            fed.threshold,
891            fed.signers.len(),
892            p.ledger().paid.len()
893        ));
894    }
895
896    let (to_loop, mut queries) = mpsc::unbounded_channel::<Query>();
897    serve_http(settings.port, to_loop)?;
898    log(format!(
899        "producer on http://127.0.0.1:{}/ every {} s ({} s with transactions)",
900        settings.port, settings.interval, settings.tx_interval
901    ));
902    let mut events = follow(
903        settings.relays.clone(),
904        vec![
905            KIND_BLOCK_PROPOSAL,
906            KIND_PARTIAL_SIGNATURE,
907            KIND_SEALED_BLOCK,
908            KIND_PEGOUT_PSBT,
909            KIND_PEGOUT_SIGNED,
910            KIND_TRANSACTION,
911        ],
912        600,
913        log,
914    );
915    let mut second = tokio::time::interval(Duration::from_secs(1));
916    let poll = settings
917        .parent
918        .as_ref()
919        .map(|p| p.poll.max(1))
920        .unwrap_or(60);
921    let mut parent_tick = tokio::time::interval(Duration::from_secs(poll));
922    let mut announce_tick = tokio::time::interval(Duration::from_secs(3));
923    let relays = settings.relays.clone();
924
925    let publish = |ev: Event, what: String| {
926        let relays = relays.clone();
927        tokio::spawn(async move {
928            let r = publish_all(&relays, &ev, Duration::from_secs(8)).await;
929            log(format!("{what} reached {} relay(s)", ok_count(&r)));
930        });
931    };
932    let handle = |node: &mut Node<F>, actions: Vec<Action>| {
933        for a in actions {
934            match a {
935                Action::Log(s) => log(s),
936                Action::Publish(ev) => {
937                    let what = match ev.kind {
938                        KIND_BLOCK_PROPOSAL => format!(
939                            "round: proposal h{} {}…",
940                            sidestr_nostr::tags::first(&ev.tags, "h").unwrap_or("?"),
941                            &ev.id[..12]
942                        ),
943                        KIND_PARTIAL_SIGNATURE => format!(
944                            "round: partial h{}",
945                            sidestr_nostr::tags::first(&ev.tags, "h").unwrap_or("?")
946                        ),
947                        _ => format!(
948                            "round: sealed h{}",
949                            sidestr_nostr::tags::first(&ev.tags, "h").unwrap_or("?")
950                        ),
951                    };
952                    publish(ev, what);
953                }
954                Action::Sealed(_) => {
955                    node.last_block = unix_now();
956                }
957            }
958        }
959    };
960    let handle_pegout = |node: &mut Node<F>, actions: Vec<PegoutAction>| {
961        for a in actions {
962            match a {
963                PegoutAction::Log(s) => log(s),
964                PegoutAction::Publish(ev) => {
965                    let what = format!(
966                        "peg-out round: kind {} for {}…",
967                        ev.kind,
968                        sidestr_nostr::tags::first(&ev.tags, "d")
969                            .map(|d| &d[..d.len().min(16)])
970                            .unwrap_or("?")
971                    );
972                    publish(ev, what);
973                }
974                PegoutAction::Broadcast(f) => {
975                    let Some(rpc) = node.parent.clone() else {
976                        continue;
977                    };
978                    let hex = hex::encode(bitcoin::consensus::encode::serialize(&f.tx));
979                    match rpc.call("sendrawtransaction", serde_json::json!([hex])) {
980                        Ok(_) => {
981                            if let Some(p) = node.pegout.as_mut() {
982                                p.mark_paid(&f.burn, f.record);
983                                write_json(&node.settings.dir.join("pegouts.json"), p.ledger());
984                            }
985                        }
986                        Err(e) => log(format!("peg-out round: sendrawtransaction: {e}")),
987                    }
988                }
989            }
990        }
991    };
992
993    loop {
994        tokio::select! {
995            _ = second.tick() => {
996                let now = unix_now();
997                let wait = if node.chain.state().mempool().count() > 0 { settings.tx_interval } else { settings.interval };
998                let due = now.saturating_sub(node.last_block) >= wait;
999                let actions = node.round.tick(unix_now_ms(), &mut node.chain, due);
1000                handle(&mut node, actions);
1001                while let Ok(q) = queries.try_recv() { node.answer(q); }
1002            }
1003            Some(q) = queries.recv() => { node.answer(q); }
1004            Some((_, ev)) = events.recv() => {
1005                let now = unix_now_ms();
1006                match ev.kind {
1007                    KIND_TRANSACTION => {
1008                        if node.txs.accept(&ev).is_some() {
1009                            if let Ok(t) = sidestr_nostr::tx::parse_transaction(&ev, Some(&node.doc.id)) {
1010                                match hex::decode(&t.tx_hex).map_err(|e| e.to_string()).and_then(|b| node.chain.submit(&b).map_err(|e| e.to_string())) {
1011                                    Ok(s) => if !s.dup { log(format!("tx {}… accepted from a relay, fee {}", &s.txid.to_string()[..16], s.fee)) },
1012                                    Err(e) => log(format!("tx from a relay refused: {e}")),
1013                                }
1014                            }
1015                        }
1016                    }
1017                    KIND_PEGOUT_PSBT | KIND_PEGOUT_SIGNED => {
1018                        let burns = node.chain.state().pegouts();
1019                        if let Some(p) = node.pegout.as_mut() {
1020                            let actions = p.on_event(now, &ev, &burns);
1021                            handle_pegout(&mut node, actions);
1022                        }
1023                    }
1024                    _ => {
1025                        let actions = node.round.on_event(now, &mut node.chain, &ev);
1026                        handle(&mut node, actions);
1027                    }
1028                }
1029            }
1030            _ = parent_tick.tick() => {
1031                if node.parent.is_some() {
1032                    node.peg_tick();
1033                    node.reconcile();
1034                    let coins = node.peg_coins();
1035                    let burns = node.chain.state().pegouts();
1036                    if let Some(p) = node.pegout.as_mut() {
1037                        let actions = p.tick(unix_now_ms(), &burns, &coins);
1038                        handle_pegout(&mut node, actions);
1039                    }
1040                }
1041            }
1042            _ = announce_tick.tick() => {
1043                if !relays.is_empty() && !settings.mirrors.is_empty() {
1044                    let tip = node.chain.state().tip();
1045                    let now = unix_now();
1046                    if node.announced != Some(tip.height) && now >= node.announce_retry_at {
1047                        let headers = node.headers_hex();
1048                        match TipTemplate::new(node.doc.id.clone(), tip.height, headers.clone(), settings.mirrors.clone()).and_then(|t| sign_tip(&node.key, &t, now)) {
1049                            Ok(ev) => {
1050                                let r = publish_all(&relays, &ev, Duration::from_secs(8)).await;
1051                                let ok = ok_count(&r);
1052                                if ok > 0 { node.announced = Some(tip.height); } else { node.announce_retry_at = now + 60; }
1053                                log(format!("announced tip {} {}… (kind 33333, {} headers, {} mirror(s)) to {ok}/{} relay(s)", tip.height, &tip.hash.to_string()[..16], headers.len(), settings.mirrors.len(), relays.len()));
1054                            }
1055                            Err(e) => log(format!("announce: {e}")),
1056                        }
1057                    }
1058                }
1059            }
1060        }
1061    }
1062}
1063
1064/// Run one signer for the document at `settings.chain`, picking the header
1065/// family from its parent.
1066pub async fn run(settings: Settings) -> Result<()> {
1067    let doc = ChainDocument::from_json(
1068        &std::fs::read_to_string(&settings.chain)
1069            .map_err(|e| Error::Federation(format!("{}: {e}", settings.chain.display())))?,
1070    )?;
1071    doc.validate()?;
1072    if doc.signers.is_none() {
1073        return Err(Error::Federation(
1074            "not a federated chain: the document has no signers; a level-1 chain is produced by `siding produce`".into(),
1075        ));
1076    }
1077    match doc.family()? {
1078        sidestr_core::parents::Family::Stock => {
1079            run_as::<sidestr_core::block::Stock>(doc, settings).await
1080        }
1081        sidestr_core::parents::Family::Blake2b => {
1082            run_as::<sidestr_header::Blake2bV2>(doc, settings).await
1083        }
1084    }
1085}
1086
1087#[cfg(test)]
1088mod tests {
1089    use super::*;
1090
1091    #[test]
1092    fn ranges_parse_as_a_mirror_sends_them() {
1093        assert_eq!(parse_range("bytes=8-15", 100), Some((8, 15)));
1094        assert_eq!(parse_range("bytes=90-", 100), Some((90, 99)));
1095        assert_eq!(parse_range("bytes=90-100", 100), None);
1096        assert_eq!(parse_range("items=1-2", 100), None);
1097    }
1098
1099    #[test]
1100    fn pegin_records_round_trip() {
1101        let f = FoundPegin {
1102            txid: "ab".repeat(32),
1103            vout: 1,
1104            amount: 5,
1105            script: bitcoin::ScriptBuf::from_bytes(vec![0x51, 0x20]),
1106            height: 9,
1107            parent_address: None,
1108        };
1109        let r = PeginRecord::from(&f);
1110        assert_eq!(r.found(), f);
1111    }
1112}