From c4391a3c9f560028e077d155c1083670a8c8a87a Mon Sep 17 00:00:00 2001 From: Tuan Pham Anh Date: Sat, 18 Jul 2026 22:33:11 +0700 Subject: [PATCH 1/4] support DNS names in coins-chain peer addresses --- examples/coins/chain/src/testnet.rs | 81 +++- examples/coins/chain/src/tests/testnet.rs | 476 +++++++++++++++++++++- 2 files changed, 541 insertions(+), 16 deletions(-) diff --git a/examples/coins/chain/src/testnet.rs b/examples/coins/chain/src/testnet.rs index a1fc1ea..8fc8049 100644 --- a/examples/coins/chain/src/testnet.rs +++ b/examples/coins/chain/src/testnet.rs @@ -29,7 +29,7 @@ use commonware_p2p::{ use commonware_runtime::{ tokio, Clock as _, Handle, Runner as _, Spawner as _, Strategizer as _, Supervisor as _, }; -use commonware_utils::{ordered::Set, N3f1, NZUsize, NZU32}; +use commonware_utils::{ordered::Set, Hostname, N3f1, NZUsize, NZU32}; use governor::Quota; use nunchi_dkg::{ ContinueOnUpdate, PeerConfig, Storage as DkgStorage, StorageKey, StorageProtector, @@ -124,7 +124,12 @@ pub struct NodeConfig { pub share: String, pub peer_config: PeerConfig, pub listen_address: SocketAddr, - pub dialable_address: SocketAddr, + /// Address advertised to peers. IP literals use socket syntax (with brackets around IPv6), + /// while DNS names use `hostname:port` without a URL scheme or path. DNS is resolved again + /// whenever a disconnected peer retries this node, so operators should use a stable hostname + /// and an appropriate TTL. + #[serde(with = "ingress_serde")] + pub dialable_address: Ingress, pub rpc_address: SocketAddr, pub metrics_address: SocketAddr, pub bootstrappers: Vec, @@ -155,7 +160,64 @@ impl NodeConfig { #[derive(Clone, Debug, Serialize, Deserialize)] pub struct BootstrapperConfig { pub public_key: String, - pub address: SocketAddr, + /// Address used to dial the bootstrapper. IP literals use socket syntax (with brackets around + /// IPv6), while DNS names use `hostname:port` without a URL scheme or path. DNS is resolved on + /// each reconnect attempt rather than continuously while a connection is healthy. + #[serde(with = "ingress_serde")] + pub address: Ingress, +} + +mod ingress_serde { + use super::*; + use serde::{de::Error as _, Deserializer, Serializer}; + + pub fn deserialize<'de, D>(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = String::deserialize(deserializer)?; + if let Ok(address) = SocketAddr::from_str(&value) { + return Ok(Ingress::Socket(address)); + } + if value.contains("://") { + return Err(D::Error::custom( + "peer address must not contain a URL scheme or path", + )); + } + + let (host, port) = value + .rsplit_once(':') + .ok_or_else(|| D::Error::custom("peer address is missing a port"))?; + if host.is_empty() { + return Err(D::Error::custom("peer address is missing a host")); + } + if port.is_empty() { + return Err(D::Error::custom("peer address is missing a port")); + } + if host.contains(':') { + return Err(D::Error::custom( + "IPv6 peer addresses must use bracketed socket syntax", + )); + } + let port = port + .parse::() + .map_err(|error| D::Error::custom(format!("invalid peer address port: {error}")))?; + let host = Hostname::new(host) + .map_err(|error| D::Error::custom(format!("invalid peer address hostname: {error}")))?; + Ok(Ingress::Dns { host, port }) + } + + pub fn serialize(ingress: &Ingress, serializer: S) -> Result + where + S: Serializer, + { + match ingress { + Ingress::Socket(address) => serializer.serialize_str(&address.to_string()), + Ingress::Dns { host, port } => { + serializer.serialize_str(&format!("{host}:{port}")) + } + } + } } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -284,14 +346,14 @@ pub fn generate_local_testnet(config: LocalTestnetConfig) -> Result Result, Error>>()?; @@ -477,7 +539,7 @@ async fn start_node( .map(|bootstrapper| { Ok(( decode_unit::(&bootstrapper.public_key, "bootstrapper.public_key")?, - Ingress::from(bootstrapper.address), + bootstrapper.address.clone(), )) }) .collect::, Error>>()?; @@ -486,6 +548,7 @@ async fn start_node( node = %config.name, public_key = %public_key, listen = %config.listen_address, + dialable = ?config.dialable_address, rpc = %config.rpc_address, metrics = %config.metrics_address, "starting coins-chain validator" @@ -495,7 +558,7 @@ async fn start_node( private_key.clone(), NAMESPACE, config.listen_address, - config.dialable_address, + config.dialable_address.clone(), bootstrappers, config.networking.max_message_size, ); diff --git a/examples/coins/chain/src/tests/testnet.rs b/examples/coins/chain/src/tests/testnet.rs index ac240e4..cd701fb 100644 --- a/examples/coins/chain/src/tests/testnet.rs +++ b/examples/coins/chain/src/tests/testnet.rs @@ -5,16 +5,24 @@ //! as `narae` consume. [`run_node`] boots a single validator from one of those configs on the //! tokio runtime with authenticated peer discovery, and serves the aggregated JSON-RPC module. -use commonware_cryptography::{bls12381::primitives::group, ed25519}; +use commonware_cryptography::{bls12381::primitives::group, ed25519, Signer as _}; +use commonware_macros::select; +use commonware_p2p::{ + authenticated::discovery::{self, Network}, + Ingress, Manager, Receiver as _, Recipients, Sender as _, +}; +use commonware_runtime::{deterministic, Clock as _, Quota, Runner as _, Supervisor as _}; +use commonware_utils::{ordered::Set, Hostname, NZU32}; use std::collections::HashSet; use std::{ fs, - net::{IpAddr, Ipv4Addr}, + net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, num::NonZeroU32, path::PathBuf, + time::Duration, }; -use crate::testnet::*; +use crate::{testnet::*, NAMESPACE}; #[test] fn generated_testnet_has_unique_ports_dirs_and_complete_peer_sets() { @@ -124,17 +132,20 @@ fn generated_testnet_can_advertise_remote_hosts() { let first = NodeConfig::read(&manifest.nodes[0].config_path).expect("read first config"); let second = NodeConfig::read(&manifest.nodes[1].config_path).expect("read second config"); + let first_raw = fs::read_to_string(&manifest.nodes[0].config_path).expect("read first config"); assert_eq!(first.listen_address.ip(), IpAddr::V4(Ipv4Addr::UNSPECIFIED)); - assert_eq!(first.dialable_address.ip(), public_ips[0]); - assert_eq!(first.bootstrappers[0].address.ip(), public_ips[1]); + assert_eq!(first.dialable_address.ip(), Some(public_ips[0])); + assert_eq!(first.bootstrappers[0].address.ip(), Some(public_ips[1])); + assert!(first_raw.contains("dialable_address = \"192.0.2.10:30000\"")); + assert!(first_raw.contains("address = \"192.0.2.11:30001\"")); assert_eq!(first.storage_dir, storage_dir); assert_eq!( first.indexer_url.as_deref(), Some("https://indexer.example.com/coins-chain") ); - assert_eq!(second.dialable_address.ip(), public_ips[1]); - assert_eq!(second.bootstrappers[0].address.ip(), public_ips[0]); + assert_eq!(second.dialable_address.ip(), Some(public_ips[1])); + assert_eq!(second.bootstrappers[0].address.ip(), Some(public_ips[0])); assert_eq!( second.indexer_url.as_deref(), Some("https://indexer.example.com/coins-chain") @@ -142,3 +153,454 @@ fn generated_testnet_can_advertise_remote_hosts() { let _ = fs::remove_dir_all(dir); } + +#[test] +fn peer_addresses_round_trip_dns_and_ipv6_as_strings() { + let dir = std::env::temp_dir().join(format!( + "coins-chain-address-round-trip-{}", + std::process::id() + )); + let _ = fs::remove_dir_all(&dir); + let manifest = generate_local_testnet(LocalTestnetConfig { + validators: 2, + base_port: 32_000, + base_rpc_port: 33_000, + base_metrics_port: 34_000, + base_data_dir: dir.clone(), + bind_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), + public_ips: None, + storage_dir: None, + genesis_path: None, + indexer_url: None, + seed: 9, + }) + .expect("generate testnet"); + let path = &manifest.nodes[0].config_path; + let mut config = NodeConfig::read(path).expect("read generated config"); + config.dialable_address = Ingress::Dns { + host: Hostname::new("validator-0.example.com").unwrap(), + port: 30_000, + }; + config.bootstrappers[0].address = Ingress::Dns { + host: Hostname::new("validator-1.example.com").unwrap(), + port: 30_001, + }; + config.write(path).expect("write DNS config"); + + let raw = fs::read_to_string(path).expect("read DNS config"); + assert!(raw.contains("dialable_address = \"validator-0.example.com:30000\"")); + assert!(raw.contains("address = \"validator-1.example.com:30001\"")); + let decoded = NodeConfig::read(path).expect("read DNS config"); + assert_eq!(decoded.dialable_address, config.dialable_address); + assert_eq!( + decoded.bootstrappers[0].address, + config.bootstrappers[0].address + ); + + config.dialable_address = Ingress::Socket(SocketAddr::new( + IpAddr::V6("2001:db8::10".parse::().unwrap()), + 30_000, + )); + config.write(path).expect("write IPv6 config"); + let raw = fs::read_to_string(path).expect("read IPv6 config"); + assert!(raw.contains("dialable_address = \"[2001:db8::10]:30000\"")); + assert_eq!( + NodeConfig::read(path) + .expect("read IPv6 config") + .dialable_address, + config.dialable_address + ); + + let _ = fs::remove_dir_all(dir); +} + +#[test] +fn invalid_peer_addresses_are_rejected_through_node_config_read() { + let dir = std::env::temp_dir().join(format!( + "coins-chain-invalid-addresses-{}", + std::process::id() + )); + let _ = fs::remove_dir_all(&dir); + let manifest = generate_local_testnet(LocalTestnetConfig { + validators: 1, + base_port: 35_000, + base_rpc_port: 36_000, + base_metrics_port: 37_000, + base_data_dir: dir.clone(), + bind_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), + public_ips: None, + storage_dir: None, + genesis_path: None, + indexer_url: None, + seed: 10, + }) + .expect("generate testnet"); + let valid_path = &manifest.nodes[0].config_path; + let valid = fs::read_to_string(valid_path).expect("read generated config"); + let valid_address = "dialable_address = \"127.0.0.1:35000\""; + assert!(valid.contains(valid_address)); + + let cases = [ + ("validator.example.com", "missing a port"), + ("validator.example.com:not-a-port", "invalid peer address port"), + ("validator.example.com:65536", "invalid peer address port"), + ("-validator.example.com:30000", "invalid peer address hostname"), + ( + "https://validator.example.com:30000", + "must not contain a URL scheme or path", + ), + ("2001:db8::10:30000", "must use bracketed socket syntax"), + ]; + for (index, (address, expected)) in cases.into_iter().enumerate() { + let path = dir.join(format!("invalid-{index}.toml")); + let invalid = valid.replace( + valid_address, + &format!("dialable_address = \"{address}\""), + ); + fs::write(&path, invalid).expect("write invalid config"); + let error = NodeConfig::read(&path).expect_err("invalid address should fail"); + assert!( + error.to_string().contains(expected), + "unexpected error for {address}: {error}" + ); + } + + let _ = fs::remove_dir_all(dir); +} + +fn parse_dns_address_through_node_config(address: &str, seed: u64) -> Ingress { + let dir = std::env::temp_dir().join(format!( + "coins-chain-parse-dns-{seed}-{}", + std::process::id() + )); + let _ = fs::remove_dir_all(&dir); + let manifest = generate_local_testnet(LocalTestnetConfig { + validators: 1, + base_port: 38_000, + base_rpc_port: 38_100, + base_metrics_port: 38_200, + base_data_dir: dir.clone(), + bind_ip: IpAddr::V4(Ipv4Addr::LOCALHOST), + public_ips: None, + storage_dir: None, + genesis_path: None, + indexer_url: None, + seed, + }) + .expect("generate parser fixture"); + let path = &manifest.nodes[0].config_path; + let valid = fs::read_to_string(path).expect("read parser fixture"); + let dns = valid.replace( + "dialable_address = \"127.0.0.1:38000\"", + &format!("dialable_address = \"{address}\""), + ); + fs::write(path, dns).expect("write parser fixture"); + let ingress = NodeConfig::read(path) + .expect("parse DNS address") + .dialable_address; + let _ = fs::remove_dir_all(dir); + ingress +} + +fn reconnect_config( + key: ed25519::PrivateKey, + listen: SocketAddr, + dialable: Ingress, + bootstrappers: Vec<(ed25519::PublicKey, Ingress)>, +) -> discovery::Config { + let mut config = discovery::Config::local( + key, + NAMESPACE, + listen, + dialable, + bootstrappers, + 1024 * 1024, + ); + config.peer_connection_cooldown = Duration::from_millis(100); + config.dial_frequency = Duration::from_millis(50); + config.gossip_bit_vec_frequency = Duration::from_millis(100); + config +} + +fn run_dns_bootstrapper_redeployment(seed: u64, dns: Ingress) -> String { + let runner = deterministic::Runner::new( + deterministic::Config::new() + .with_seed(seed) + .with_timeout(Some(Duration::from_secs(20))), + ); + runner.start(|context| async move { + let key_a = ed25519::PrivateKey::from_seed(100); + let key_b = ed25519::PrivateKey::from_seed(101); + let public_a = key_a.public_key(); + let public_b = key_b.public_key(); + let peers = Set::try_from(vec![public_a.clone(), public_b.clone()]).unwrap(); + let socket_a = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 10)), 39_000); + let socket_b1 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 11)), 39_000); + let socket_b2 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 12)), 39_000); + let unreachable_a = Ingress::Socket(SocketAddr::new( + IpAddr::V4(Ipv4Addr::new(127, 0, 0, 250)), + 49_000, + )); + context.resolver_register("validator-b-bootstrap.test", Some(vec![socket_b1.ip()])); + + let config_b = reconnect_config(key_b.clone(), socket_b1, dns.clone(), vec![]); + let (mut network_b, mut oracle_b) = + Network::new(context.child("b_initial").child("network"), config_b); + oracle_b.track(0, peers.clone()); + let (mut sender_b, mut receiver_b) = + network_b.register(0, Quota::per_second(NZU32!(100)), 128); + let handle_b = network_b.start(); + + let config_a = reconnect_config( + key_a, + socket_a, + unreachable_a, + vec![(public_b.clone(), dns.clone())], + ); + assert!(config_a.allow_dns); + let (mut network_a, mut oracle_a) = + Network::new(context.child("a").child("network"), config_a); + oracle_a.track(0, peers.clone()); + let (mut sender_a, mut receiver_a) = + network_a.register(0, Quota::per_second(NZU32!(100)), 128); + let _handle_a = network_a.start(); + + let initial_exchange = async { + loop { + sender_a.send( + Recipients::One(public_b.clone()), + b"a-before".to_vec(), + true, + ); + select! { + result = receiver_b.recv() => { + let (from, message) = result.unwrap(); + assert_eq!(from, public_a); + assert_eq!(message.as_ref(), b"a-before"); + break; + }, + _ = context.sleep(Duration::from_millis(50)) => {}, + } + } + loop { + sender_b.send( + Recipients::One(public_a.clone()), + b"b-before".to_vec(), + true, + ); + select! { + result = receiver_a.recv() => { + let (from, message) = result.unwrap(); + assert_eq!(from, public_b); + assert_eq!(message.as_ref(), b"b-before"); + break; + }, + _ = context.sleep(Duration::from_millis(50)) => {}, + } + } + }; + select! { + _ = initial_exchange => {}, + _ = context.sleep(Duration::from_secs(5)) => panic!("initial DNS connection timed out"), + } + + handle_b.abort(); + drop(sender_b); + drop(receiver_b); + context.resolver_register("validator-b-bootstrap.test", Some(vec![socket_b2.ip()])); + + // The restarted peer has no bootstrappers and A advertises an unreachable address, so B + // cannot initiate the replacement connection. + let config_b = reconnect_config(key_b, socket_b2, dns, vec![]); + let (mut network_b, mut oracle_b) = + Network::new(context.child("b_restarted").child("network"), config_b); + oracle_b.track(0, peers); + let (_sender_b, mut receiver_b) = + network_b.register(0, Quota::per_second(NZU32!(100)), 128); + let _handle_b = network_b.start(); + + let reconnected = async { + loop { + sender_a.send( + Recipients::One(public_b.clone()), + b"a-after".to_vec(), + true, + ); + select! { + result = receiver_b.recv() => { + let (from, message) = result.unwrap(); + assert_eq!(from, public_a); + assert_eq!(message.as_ref(), b"a-after"); + break; + }, + _ = context.sleep(Duration::from_millis(50)) => {}, + } + } + }; + select! { + _ = reconnected => {}, + _ = context.sleep(Duration::from_secs(5)) => panic!("DNS bootstrapper did not reconnect"), + } + context.auditor().state() + }) +} + +#[test] +fn dns_bootstrapper_is_resolved_again_after_redeployment() { + let dns = parse_dns_address_through_node_config("validator-b-bootstrap.test:39000", 11); + let first = run_dns_bootstrapper_redeployment(42, dns.clone()); + let second = run_dns_bootstrapper_redeployment(42, dns); + assert_eq!(first, second); +} + +fn run_discovered_dns_redeployment(seed: u64, dns: Ingress) -> String { + let runner = deterministic::Runner::new( + deterministic::Config::new() + .with_seed(seed) + .with_timeout(Some(Duration::from_secs(20))), + ); + runner.start(|context| async move { + let key_a = ed25519::PrivateKey::from_seed(200); + let key_b = ed25519::PrivateKey::from_seed(201); + let key_c = ed25519::PrivateKey::from_seed(202); + let public_a = key_a.public_key(); + let public_b = key_b.public_key(); + let public_c = key_c.public_key(); + let peers = Set::try_from(vec![ + public_a.clone(), + public_b.clone(), + public_c.clone(), + ]) + .unwrap(); + let socket_a = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 1, 10)), 40_000); + let socket_b1 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 1, 11)), 40_000); + let socket_b2 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 1, 12)), 40_000); + let socket_c = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 1, 13)), 40_000); + let unreachable_a = Ingress::Socket(SocketAddr::new( + IpAddr::V4(Ipv4Addr::new(127, 0, 1, 250)), + 50_000, + )); + context.resolver_register("validator-b-discovered.test", Some(vec![socket_b1.ip()])); + + let config_c = reconnect_config(key_c, socket_c, socket_c.into(), vec![]); + let (mut network_c, mut oracle_c) = + Network::new(context.child("c").child("network"), config_c); + oracle_c.track(0, peers.clone()); + let (_sender_c, _receiver_c) = + network_c.register(0, Quota::per_second(NZU32!(100)), 128); + let _handle_c = network_c.start(); + + let config_b = reconnect_config( + key_b.clone(), + socket_b1, + dns.clone(), + vec![(public_c.clone(), socket_c.into())], + ); + let (mut network_b, mut oracle_b) = + Network::new(context.child("b_initial").child("network"), config_b); + oracle_b.track(0, peers.clone()); + let (mut sender_b, mut receiver_b) = + network_b.register(0, Quota::per_second(NZU32!(100)), 128); + let handle_b = network_b.start(); + + // A only knows C initially. It must learn B's DNS ingress through discovery. + let config_a = reconnect_config( + key_a, + socket_a, + unreachable_a, + vec![(public_c, socket_c.into())], + ); + let (mut network_a, mut oracle_a) = + Network::new(context.child("a").child("network"), config_a); + oracle_a.track(0, peers.clone()); + let (mut sender_a, mut receiver_a) = + network_a.register(0, Quota::per_second(NZU32!(100)), 128); + let _handle_a = network_a.start(); + + let learned_and_connected = async { + loop { + sender_a.send( + Recipients::One(public_b.clone()), + b"a-before".to_vec(), + true, + ); + select! { + result = receiver_b.recv() => { + let (from, message) = result.unwrap(); + assert_eq!(from, public_a); + assert_eq!(message.as_ref(), b"a-before"); + break; + }, + _ = context.sleep(Duration::from_millis(50)) => {}, + } + } + loop { + sender_b.send( + Recipients::One(public_a.clone()), + b"b-before".to_vec(), + true, + ); + select! { + result = receiver_a.recv() => { + let (from, message) = result.unwrap(); + assert_eq!(from, public_b); + assert_eq!(message.as_ref(), b"b-before"); + break; + }, + _ = context.sleep(Duration::from_millis(50)) => {}, + } + } + }; + select! { + _ = learned_and_connected => {}, + _ = context.sleep(Duration::from_secs(7)) => panic!("A did not discover B's DNS ingress"), + } + + handle_b.abort(); + drop(sender_b); + drop(receiver_b); + context.resolver_register("validator-b-discovered.test", Some(vec![socket_b2.ip()])); + + // B has no bootstrapper after restart and A's advertised address is unreachable. A must + // retain the discovered hostname, resolve it again, and dial B at its replacement IP. + let config_b = reconnect_config(key_b, socket_b2, dns, vec![]); + let (mut network_b, mut oracle_b) = + Network::new(context.child("b_restarted").child("network"), config_b); + oracle_b.track(0, peers); + let (_sender_b, mut receiver_b) = + network_b.register(0, Quota::per_second(NZU32!(100)), 128); + let _handle_b = network_b.start(); + + let reconnected = async { + loop { + sender_a.send( + Recipients::One(public_b.clone()), + b"a-after".to_vec(), + true, + ); + select! { + result = receiver_b.recv() => { + let (from, message) = result.unwrap(); + assert_eq!(from, public_a); + assert_eq!(message.as_ref(), b"a-after"); + break; + }, + _ = context.sleep(Duration::from_millis(50)) => {}, + } + } + }; + select! { + _ = reconnected => {}, + _ = context.sleep(Duration::from_secs(7)) => panic!("discovered DNS peer did not reconnect"), + } + context.auditor().state() + }) +} + +#[test] +fn discovered_dns_address_is_resolved_again_after_redeployment() { + let dns = parse_dns_address_through_node_config("validator-b-discovered.test:40000", 12); + let first = run_discovered_dns_redeployment(84, dns.clone()); + let second = run_discovered_dns_redeployment(84, dns); + assert_eq!(first, second); +} From 830c649f026330a872c293059f96534f05014876 Mon Sep 17 00:00:00 2001 From: Tuan Pham Anh Date: Wed, 22 Jul 2026 09:05:33 +0700 Subject: [PATCH 2/4] feat: Bound validator disk growth with prunable history and cleanup --- dkg/src/orchestrator/actor.rs | 152 +++++- examples/coins/chain/src/engine.rs | 303 ++++++----- examples/coins/chain/src/history.rs | 466 +++++++++++++++++ .../chain/src/indexer/backfiller/consumer.rs | 483 +++++++----------- .../chain/src/indexer/backfiller/producer.rs | 243 +++++---- .../chain/src/indexer/backfiller/state.rs | 182 +++++-- examples/coins/chain/src/indexer/metrics.rs | 140 ++--- examples/coins/chain/src/indexer/mod.rs | 109 +++- examples/coins/chain/src/lib.rs | 1 + examples/coins/chain/src/testnet.rs | 41 +- examples/coins/chain/src/tests/indexer.rs | 37 +- examples/coins/chain/tests/common/network.rs | 4 +- 12 files changed, 1446 insertions(+), 715 deletions(-) create mode 100644 examples/coins/chain/src/history.rs diff --git a/dkg/src/orchestrator/actor.rs b/dkg/src/orchestrator/actor.rs index f815252..5a250d0 100644 --- a/dkg/src/orchestrator/actor.rs +++ b/dkg/src/orchestrator/actor.rs @@ -26,9 +26,14 @@ use commonware_parallel::Strategy; use commonware_runtime::{ buffer::paged::CacheRef, spawn_cell, - telemetry::metrics::{Gauge, GaugeExt, MetricsExt as _}, + telemetry::metrics::{ + histogram::Buckets, CounterFamily, EncodeLabelSet, Gauge, GaugeExt, Histogram, + MetricsExt as _, + }, BufferPooler, Clock, ContextCell, Handle, Metrics, Network, Spawner, Storage, }; +use commonware_storage::metadata::{self, Metadata}; +use commonware_utils::sequence::U64; use commonware_utils::{vec::NonEmptyVec, NZUsize, NZU16}; use rand::CryptoRng; use std::{ @@ -39,6 +44,13 @@ use std::{ }; use tracing::{debug, info, warn}; +const CLEANUP_WATERMARK_KEY: U64 = U64::new(0); + +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, EncodeLabelSet)] +struct CleanupStatusLabel { + status: &'static str, +} + /// Reporter that discards consensus activity. pub struct NoopReporter { _phantom: PhantomData, @@ -146,6 +158,10 @@ where page_cache_ref: CacheRef, latest_epoch: Gauge, + partition_cleanup_total: CounterFamily, + partition_cleanup_watermark: Gauge, + partitions_active: Gauge, + partition_cleanup_duration: Histogram, _phantom: PhantomData, } @@ -180,6 +196,23 @@ where // Register latest_epoch gauge for Grafana integration let latest_epoch = context.gauge("latest_epoch", "current epoch"); + let partition_cleanup_total = context.family( + "consensus_partition_cleanup", + "Total number of consensus partition cleanup outcomes", + ); + let partition_cleanup_watermark = context.gauge( + "consensus_partition_cleanup_watermark", + "Next consensus epoch partition requiring cleanup", + ); + let partitions_active = context.gauge( + "consensus_partitions_active", + "Number of consensus epoch partitions with active engines", + ); + let partition_cleanup_duration = context.histogram( + "consensus_partition_cleanup_duration_seconds", + "Duration of consensus epoch partition cleanup operations", + Buckets::LOCAL, + ); ( Self { @@ -200,6 +233,10 @@ where recovered_floor: config.recovered_floor, page_cache_ref, latest_epoch, + partition_cleanup_total, + partition_cleanup_watermark, + partitions_active, + partition_cleanup_duration, _phantom: PhantomData, }, Mailbox::new(sender), @@ -268,6 +305,24 @@ where // Wait for instructions to transition epochs. let epocher = FixedEpocher::new(self.epoch_length); let mut engines: BTreeMap> = BTreeMap::new(); + let cleanup_partition = format!("{}_cleanup-v1", self.partition_prefix); + let mut cleanup_metadata = Metadata::::init( + self.context.child("partition_cleanup_metadata"), + metadata::Config { + partition: cleanup_partition, + codec_config: (), + }, + ) + .await + .expect("failed to initialize consensus partition cleanup metadata"); + let mut next_epoch_to_clean = cleanup_metadata + .get(&CLEANUP_WATERMARK_KEY) + .cloned() + .unwrap_or_else(|| U64::new(0)); + let _ = self + .partition_cleanup_watermark + .try_set(u64::from(&next_epoch_to_clean)); + let mut startup_cleanup_done = false; select_loop! { self.context, @@ -317,6 +372,28 @@ where continue; } + if !startup_cleanup_done { + assert!( + u64::from(&next_epoch_to_clean) <= transition.epoch.get(), + "consensus partition cleanup watermark {} is ahead of first entering epoch {}", + u64::from(&next_epoch_to_clean), + transition.epoch, + ); + if let Some(previous) = transition.epoch.previous() { + assert!( + !engines.keys().any(|epoch| *epoch <= previous), + "refusing to clean a consensus partition with an active engine" + ); + self.cleanup_through( + &mut cleanup_metadata, + &mut next_epoch_to_clean, + previous, + ) + .await; + } + startup_cleanup_done = true; + } + let floor = if self .recovered_floor .as_ref() @@ -364,6 +441,7 @@ where .await; engines.insert(transition.epoch, handle); let _ = self.latest_epoch.try_set(transition.epoch.get()); + let _ = self.partitions_active.try_set(engines.len()); info!(epoch = %transition.epoch, "entered epoch"); } @@ -374,16 +452,88 @@ where continue; }; handle.abort(); + match handle.await { + Ok(()) | Err(commonware_runtime::Error::Aborted) => {} + Err(error) => { + panic!("consensus engine for epoch {epoch} failed while stopping: {error}") + } + } + let _ = self.partitions_active.try_set(engines.len()); // Unregister the signing scheme for the epoch. assert!(self.provider.unregister(&epoch)); + if u64::from(&next_epoch_to_clean) <= epoch.get() { + assert!( + !engines.keys().any(|active| *active <= epoch), + "refusing to clean a consensus partition with an active engine" + ); + self.cleanup_through( + &mut cleanup_metadata, + &mut next_epoch_to_clean, + epoch, + ) + .await; + } + info!(%epoch, "exited epoch"); } }, } } + async fn cleanup_through( + &mut self, + metadata: &mut Metadata, + next_epoch_to_clean: &mut U64, + through: Epoch, + ) { + while u64::from(&*next_epoch_to_clean) <= through.get() { + let epoch = Epoch::new(u64::from(&*next_epoch_to_clean)); + let partition = format!("{}_consensus_{}", self.partition_prefix, epoch); + let started = std::time::Instant::now(); + let status = match self.context.remove(&partition, None).await { + Ok(()) => { + info!(%epoch, %partition, "removed retired consensus partition"); + "removed" + } + Err(commonware_runtime::Error::PartitionMissing(_)) => { + info!(%epoch, %partition, "retired consensus partition already missing"); + "missing" + } + Err(error) => { + self.partition_cleanup_total + .get_or_create(&CleanupStatusLabel { status: "failed" }) + .inc(); + self.partition_cleanup_duration + .observe(started.elapsed().as_secs_f64()); + warn!(%epoch, %partition, %error, "failed to remove retired consensus partition"); + panic!("failed to remove retired consensus partition {partition}: {error}"); + } + }; + self.partition_cleanup_total + .get_or_create(&CleanupStatusLabel { status }) + .inc(); + self.partition_cleanup_duration + .observe(started.elapsed().as_secs_f64()); + + let next = epoch + .get() + .checked_add(1) + .expect("consensus cleanup watermark overflow"); + metadata + .put_sync(CLEANUP_WATERMARK_KEY, U64::new(next)) + .await + .unwrap_or_else(|error| { + panic!( + "failed to persist consensus cleanup watermark after removing {partition}: {error}" + ) + }); + *next_epoch_to_clean = U64::new(next); + let _ = self.partition_cleanup_watermark.try_set(next); + } + } + // Epoch zero uses genesis as its floor; every later epoch is anchored by the // last finalized block from the previous epoch. fn floor_boundary(epocher: &FixedEpocher, epoch: Epoch) -> Option { diff --git a/examples/coins/chain/src/engine.rs b/examples/coins/chain/src/engine.rs index 00a47dd..3247e10 100644 --- a/examples/coins/chain/src/engine.rs +++ b/examples/coins/chain/src/engine.rs @@ -1,9 +1,10 @@ use crate::application::{self, Application}; use crate::execution::NodeHandle; use crate::genesis::{genesis_target, state_commitment, ChainGenesis}; +use crate::history::{self, BlocksArchive, FinalizationsArchive}; use crate::indexer; use crate::{ - Block, EpochProvider, Finalization, Provider, PublicKey, Scheme, Transaction, BLOCKS_PER_EPOCH, + Block, EpochProvider, Provider, PublicKey, Scheme, Transaction, BLOCKS_PER_EPOCH, NAMESPACE, }; use commonware_broadcast::buffered; @@ -13,7 +14,7 @@ use commonware_consensus::{ core::Actor as MarshalActor, resolver, standard::{Inline, Standard}, - store::{Blocks as _, Certificates}, + store::Certificates, }, simplex::elector::Random, types::{Epoch, FixedEpocher, Height, ViewDelta}, @@ -24,7 +25,6 @@ use commonware_cryptography::{ dkg::feldman_desmedt::Output, primitives::{group, variant::MinSig}, }, - certificate::Verifier as _, ed25519::{self, Batch}, sha256::Digest, BatchVerifier, Digestible, Signer, @@ -41,12 +41,8 @@ use commonware_runtime::{ buffer::paged::CacheRef, spawn_cell, BufferPooler, Clock, ContextCell, Handle, Metrics, Network, Spawner, Storage, Strategizer, }; -use commonware_storage::{ - archive::{immutable, Identifier as ArchiveIdentifier}, - metadata::{self, Metadata}, - queue, -}; -use commonware_utils::{sequence::U64, union, NZDuration, NZU64}; +use commonware_storage::{archive::{Archive as ArchiveStore, Identifier as ArchiveIdentifier}, metadata::{self, Metadata}, queue}; +use commonware_utils::{sequence::U64, union, NZDuration}; use futures::lock::Mutex as AsyncMutex; use governor::clock::Clock as GClock; use nunchi_chain::engine::*; @@ -84,8 +80,6 @@ pub struct Config, P: Manager, @@ -100,6 +94,7 @@ pub struct Config, P: Manager, pub indexer: Option, + pub indexer_spool_limits: indexer::SpoolLimits, } type DkgActor = nunchi_chain::DkgActor; @@ -110,8 +105,6 @@ type LimitedStatefulAppMailbox = VerifyLimiter>; type InlineApp = Inline, Block, FixedEpocher>; type Marshaled = BoxedAutomaton>; type SchemeProvider = Provider; -type FinalizationsArchive = immutable::Archive; -type BlocksArchive = immutable::Archive; type Marshal = MarshalActor< E, Standard, @@ -167,6 +160,7 @@ where stateful: StatefulApp, stateful_mailbox: StatefulAppMailbox, indexer_producer: Option, + indexer_producer_handle: Option>, indexer_consumer: Option>, } @@ -251,88 +245,36 @@ where ); let start = Instant::now(); - let finalizations_by_height = immutable::Archive::init( - context.child("finalizations_by_height"), - immutable::Config { - metadata_partition: format!( - "{}-finalizations-by-height-metadata", - config.partition_prefix - ), - freezer_table_partition: format!( - "{}-finalizations-by-height-freezer-table", - config.partition_prefix - ), - freezer_table_initial_size: config.finalized_freezer_table_initial_size, - freezer_table_resize_frequency: FREEZER_TABLE_RESIZE_FREQUENCY, - freezer_table_resize_chunk_size: FREEZER_TABLE_RESIZE_CHUNK_SIZE, - freezer_key_partition: format!( - "{}-finalizations-by-height-freezer-key", - config.partition_prefix - ), - freezer_key_page_cache: page_cache.clone(), - freezer_key_write_buffer: WRITE_BUFFER, - freezer_value_partition: format!( - "{}-finalizations-by-height-freezer-value", - config.partition_prefix - ), - freezer_value_write_buffer: WRITE_BUFFER, - freezer_value_target_size: FREEZER_VALUE_TARGET_SIZE, - freezer_value_compression: FREEZER_VALUE_COMPRESSION, - ordinal_partition: format!( - "{}-finalizations-by-height-ordinal", - config.partition_prefix - ), - ordinal_write_buffer: WRITE_BUFFER, - items_per_section: IMMUTABLE_ITEMS_PER_SECTION, - codec_config: Scheme::certificate_codec_config_unbounded(), - replay_buffer: REPLAY_BUFFER, - }, - ) - .await - .expect("failed to initialize finalizations by height archive"); - info!(elapsed = ?start.elapsed(), "restored finalizations by height archive"); - - let start = Instant::now(); - let finalized_blocks = immutable::Archive::init( - context.child("finalized_blocks"), - immutable::Config { - metadata_partition: format!( - "{}-finalized_blocks-metadata", - config.partition_prefix - ), - freezer_table_partition: format!( - "{}-finalized_blocks-freezer-table", - config.partition_prefix - ), - freezer_table_initial_size: config.blocks_freezer_table_initial_size, - freezer_table_resize_frequency: FREEZER_TABLE_RESIZE_FREQUENCY, - freezer_table_resize_chunk_size: FREEZER_TABLE_RESIZE_CHUNK_SIZE, - freezer_key_partition: format!( - "{}-finalized_blocks-freezer-key", - config.partition_prefix - ), - freezer_key_page_cache: page_cache.clone(), - freezer_key_write_buffer: WRITE_BUFFER, - freezer_value_partition: format!( - "{}-finalized_blocks-freezer-value", - config.partition_prefix - ), - freezer_value_write_buffer: WRITE_BUFFER, - freezer_value_target_size: FREEZER_VALUE_TARGET_SIZE, - freezer_value_compression: FREEZER_VALUE_COMPRESSION, - ordinal_partition: format!("{}-finalized_blocks-ordinal", config.partition_prefix), - ordinal_write_buffer: WRITE_BUFFER, - items_per_section: IMMUTABLE_ITEMS_PER_SECTION, - codec_config: block_codec_config, - replay_buffer: REPLAY_BUFFER, - }, + let opened_history = history::open( + &context, + &config.partition_prefix, + page_cache.clone(), + block_codec_config, ) .await - .expect("failed to initialize finalized blocks archive"); - info!(elapsed = ?start.elapsed(), "restored finalized blocks archive"); + .unwrap_or_else(|error| panic!("failed to initialize bounded finalized history: {error}")); + let history::Opened { + finalizations: mut finalizations_by_height, + blocks: mut finalized_blocks, + partitions: history_partitions, + marker_status, + } = opened_history; + info!( + elapsed = ?start.elapsed(), + format_version = history::FORMAT_VERSION, + finalizations_key = %history_partitions.finalizations_key, + finalizations_value = %history_partitions.finalizations_value, + blocks_key = %history_partitions.blocks_key, + blocks_value = %history_partitions.blocks_value, + logical_retention = history::LOGICAL_RETENTION, + items_per_section = history::PRUNABLE_ITEMS_PER_SECTION.get(), + maximum_section_and_cadence_slack = history::MAX_RETAINED_HEIGHTS - history::LOGICAL_RETENTION, + ?marker_status, + "restored prunable finalized history" + ); let recovered_height = Certificates::last_index(&finalizations_by_height); - let mut recovered_floor = if let Some(height) = recovered_height { + let recovered_floor = if let Some(height) = recovered_height { Certificates::get( &finalizations_by_height, ArchiveIdentifier::Index(height.get()), @@ -425,15 +367,35 @@ where .expect("failed to initialize state database for startup probe"); state.sync_target() }; - if repair_marshal_progress_if_state_is_behind( + if let Some(processed_height) = validate_marshal_progress_against_state( &context, &config.partition_prefix, &finalized_blocks, ¤t_state_target, ) - .await + .await { + validate_history_through_processed( + &finalizations_by_height, + &finalized_blocks, + processed_height, + ) + .await; + if let Some(tip) = ArchiveStore::last_index(&finalized_blocks) { + let policy_floor = history::policy_floor(tip, history::LOGICAL_RETENTION); + let safe_floor = policy_floor.min(processed_height.get()); + finalizations_by_height + .prune(safe_floor) + .await + .expect("failed to startup-prune finalized certificates"); + finalized_blocks + .prune(safe_floor) + .await + .expect("failed to startup-prune finalized blocks"); + } + } else if ArchiveStore::last_index(&finalized_blocks).is_some() + || ArchiveStore::last_index(&finalizations_by_height).is_some() { - recovered_floor = None; + panic!("finalized history exists without persisted marshal processed height; rebuild or perform verified peer state sync"); } let applied_height = Arc::new(AsyncMutex::new(Height::zero())); let app = Application::with_consensus( @@ -554,7 +516,7 @@ where FixedEpocher::new(BLOCKS_PER_EPOCH), )); - let (indexer_producer, indexer_pusher, indexer_consumer) = + let (indexer_producer, indexer_producer_handle, indexer_pusher, indexer_consumer) = if let Some(client) = config.indexer.clone() { let indexer_context = context.child("indexer"); let indexer_metrics = indexer::IndexerMetrics::register(&indexer_context); @@ -562,10 +524,13 @@ where let queue = queue::shared::init( context.child("indexer_queue"), queue::Config { - partition: format!("{}-indexer-finalized-queue", config.partition_prefix), - items_per_section: NZU64!(128), + partition: format!( + "{}-indexer-finalized-payload-queue-v1", + config.partition_prefix + ), + items_per_section: indexer::SPOOL_ITEMS_PER_SECTION, compression: None, - codec_config: (), + codec_config: block_codec_config, page_cache: page_cache.clone(), write_buffer: WRITE_BUFFER, }, @@ -581,14 +546,20 @@ where mailbox_size: MAILBOX_SIZE, backfiller_max_active: commonware_utils::NZUsize!(16), backfiller_retry: Duration::from_millis(500), + spool_limits: config.indexer_spool_limits, metrics: indexer_metrics, }, ) .await; - let (producer, pusher, consumer) = indexer.split(); - (Some(producer), Some(pusher), Some(consumer)) + let (producer, producer_handle, pusher, consumer) = indexer.split(); + ( + Some(producer), + Some(producer_handle), + Some(pusher), + Some(consumer), + ) } else { - (None, None, None) + (None, None, None, None) }; let (orchestrator, orchestrator_mailbox) = orchestrator::Actor::new( @@ -629,6 +600,7 @@ where stateful, stateful_mailbox, indexer_producer, + indexer_producer_handle, indexer_consumer, }; (engine, node_handle) @@ -747,10 +719,13 @@ where .mempool .start_p2p(self.context.child("mempool"), mempool); let indexer_consumer_handle = self.indexer_consumer.map(indexer::Consumer::start); + let indexer_producer_handle = self.indexer_producer_handle; let clob_handle = self.clob.start_p2p(self.context.child("clob"), clob); let mut shutdown = self.context.stopped(); - if let Some(indexer_consumer_handle) = indexer_consumer_handle { + if let (Some(indexer_producer_handle), Some(indexer_consumer_handle)) = + (indexer_producer_handle, indexer_consumer_handle) + { commonware_macros::select! { stopped = &mut shutdown => match stopped { Ok(0) => { @@ -769,6 +744,7 @@ where result = orchestrator_handle => unexpected_exit("orchestrator", result), result = mempool_handle => unexpected_exit("mempool", result), result = clob_handle => unexpected_exit("clob", result), + result = indexer_producer_handle => unexpected_exit("indexer_producer", result), result = indexer_consumer_handle => unexpected_exit("indexer_consumer", result), } } else { @@ -806,14 +782,13 @@ fn unexpected_exit( } const MARSHAL_PROCESSED_KEY: U64 = U64::new(0xFF); -const STATE_SYNC_METADATA_SUFFIX: &str = "state_sync_metadata"; -async fn repair_marshal_progress_if_state_is_behind( +async fn validate_marshal_progress_against_state( context: &E, partition_prefix: &str, finalized_blocks: &BlocksArchive, current_target: &commonware_storage::qmdb::sync::Target, -) -> bool +) -> Option where E: BufferPooler + Clock @@ -837,63 +812,87 @@ where ) .await .expect("failed to initialize marshal progress metadata probe"); - let Some(processed_height) = metadata.get(&MARSHAL_PROCESSED_KEY).copied() else { - return false; - }; + let processed_height = metadata.get(&MARSHAL_PROCESSED_KEY).copied()?; drop(metadata); - let Some(processed_block) = finalized_blocks - .get(ArchiveIdentifier::Index(processed_height.get())) - .await - .expect("failed to read processed block for state repair") - else { - warn!( - %processed_height, - "marshal progress references a missing finalized block; clearing startup metadata to replay" - ); - remove_partition_if_exists(context, &marshal_metadata_partition).await; - clear_disabled_state_sync_metadata(context, partition_prefix).await; - return true; + let Some(tip) = ArchiveStore::last_index(finalized_blocks) else { + panic!("marshal progress references height {processed_height}, but finalized block history is empty; rebuild or perform verified peer state sync"); }; - let processed_target = >::sync_targets(&processed_block); - let current_size = current_target.range.end().as_u64(); - let processed_size = processed_target.range.end().as_u64(); - if current_size >= processed_size { - return false; + let search_end = tip.min( + processed_height + .get() + .saturating_add(MAX_PENDING_ACKS.get() as u64), + ); + for height in processed_height.get()..=search_end { + let Some(block) = ArchiveStore::get( + finalized_blocks, + ArchiveIdentifier::Index(height), + ) + .await + .expect("failed to read finalized block while validating QMDB startup target") + else { + continue; + }; + let target = >::sync_targets(&block); + if &target == current_target { + return Some(processed_height); + } } - warn!( - %processed_height, - current_size, - processed_size, - "state database is behind marshal progress; clearing startup metadata to replay finalized blocks" + panic!( + "QMDB committed target does not exactly match canonical finalized history from processed height {} through {}; rebuild or perform verified peer state sync", + processed_height, + search_end, ); - remove_partition_if_exists(context, &marshal_metadata_partition).await; - remove_partition_if_exists( - context, - &format!("{partition_prefix}{STATE_SYNC_METADATA_SUFFIX}"), - ) - .await; - true } -async fn clear_disabled_state_sync_metadata(context: &E, partition_prefix: &str) +async fn validate_history_through_processed( + finalizations: &FinalizationsArchive, + blocks: &BlocksArchive, + processed_height: Height, +) where - E: Storage, + E: BufferPooler + Metrics + Storage, { - remove_partition_if_exists( - context, - &format!("{partition_prefix}{STATE_SYNC_METADATA_SUFFIX}"), - ) - .await; -} + let processed = processed_height.get(); + let first = match (finalizations.first_index(), blocks.first_index()) { + (Some(finalization), Some(block)) => finalization.max(block), + _ => panic!( + "marshal processed height {processed_height} has incomplete finalized history; rebuild or perform verified peer state sync" + ), + }; + if first > processed { + panic!( + "marshal processed height {processed_height} is below retained finalized history floor {first}; rebuild or perform verified peer state sync" + ); + } -async fn remove_partition_if_exists(context: &E, partition: &str) -where - E: Storage, -{ - match context.remove(partition, None).await { - Ok(()) | Err(commonware_runtime::Error::PartitionMissing(_)) => {} - Err(error) => panic!("failed to remove partition {partition}: {error}"), + for height in first..=processed { + let certificate = finalizations + .get(ArchiveIdentifier::Index(height)) + .await + .expect("failed to validate finalized certificate archive"); + let block = blocks + .get(ArchiveIdentifier::Index(height)) + .await + .expect("failed to validate finalized block archive"); + if height == processed { + assert!( + block.is_some(), + "marshal processed height {processed_height} references a missing finalized block; rebuild or perform verified peer state sync" + ); + } + if let (Some(certificate), Some(block)) = (certificate, block) { + assert_eq!( + certificate.proposal.payload, + block.digest(), + "finalized block and certificate conflict at height {height}" + ); + assert_eq!( + certificate.proposal.round.epoch(), + block.context.round.epoch(), + "finalized block and certificate epochs conflict at height {height}" + ); + } } } diff --git a/examples/coins/chain/src/history.rs b/examples/coins/chain/src/history.rs new file mode 100644 index 0000000..402af05 --- /dev/null +++ b/examples/coins/chain/src/history.rs @@ -0,0 +1,466 @@ +//! Bounded finalized-history storage for coins-chain validators. + +use crate::{Block, Finalization}; +use commonware_codec::Read; +use commonware_cryptography::sha256::Digest; +use commonware_runtime::{buffer::paged::CacheRef, BufferPooler, Clock, Metrics, Spawner, Storage}; +use commonware_storage::{ + archive::{prunable, Archive as _}, + metadata::{self, Metadata}, + translator::EightCap, +}; +use commonware_utils::{sequence::U64, NZU64}; +use nunchi_chain::engine::{ + MAX_PENDING_ACKS, PRUNE_MAINTENANCE_INTERVAL, PRUNE_RETAINED_MARSHAL_BLOCKS, REPLAY_BUFFER, + WRITE_BUFFER, +}; +use std::num::NonZeroU64; + +pub(crate) const FORMAT_VERSION: u64 = 1; +pub(crate) const PRUNABLE_ITEMS_PER_SECTION: NonZeroU64 = NZU64!(4_096); +pub(crate) const LOGICAL_RETENTION: u64 = + PRUNE_RETAINED_MARSHAL_BLOCKS as u64 + MAX_PENDING_ACKS.get() as u64 + 1; +pub(crate) const MAX_RETAINED_HEIGHTS: u64 = LOGICAL_RETENTION + + PRUNABLE_ITEMS_PER_SECTION.get() + - 1 + + PRUNE_MAINTENANCE_INTERVAL.get() as u64 + - 1; + +const MARKER_KEY: U64 = U64::new(0); + +pub(crate) type FinalizationsArchive = + prunable::Archive; +pub(crate) type BlocksArchive = prunable::Archive; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum MarkerStatus { + Created, + RecoveredPartialInitialization, + Present, +} + +#[derive(Debug, thiserror::Error)] +pub(crate) enum HistoryError { + #[error("legacy finalized-history partition exists: {partition}; stop the devnet and clear the complete validator storage directory before relaunch")] + LegacyPartition { partition: String }, + #[error("history format marker exists but required partition is missing: {partition}; clear the complete devnet validator storage directory before relaunch")] + MarkedPartitionMissing { partition: String }, + #[error("history format marker is missing but {partition} contains finalized data at height {height}; refusing to modify unmarked storage")] + NonEmptyUnmarked { partition: String, height: u64 }, + #[error("invalid history format marker in {partition}: expected version {expected}, found {found:?}")] + InvalidMarker { + partition: String, + expected: u64, + found: Option, + }, + #[error("failed to inspect history partition {partition}: {source}")] + Inspect { + partition: String, + #[source] + source: commonware_runtime::Error, + }, + #[error("failed to open history archive partition {partition}: {source}")] + Archive { + partition: String, + #[source] + source: commonware_storage::archive::Error, + }, + #[error("failed to access history marker partition {partition}: {source}")] + Marker { + partition: String, + #[source] + source: commonware_storage::metadata::Error, + }, + #[error("invalid history retention configuration: {0}")] + Retention(&'static str), +} + +#[derive(Clone, Debug)] +pub(crate) struct Partitions { + pub(crate) finalizations_key: String, + pub(crate) finalizations_value: String, + pub(crate) blocks_key: String, + pub(crate) blocks_value: String, + pub(crate) metadata: String, +} + +impl Partitions { + pub(crate) fn new(prefix: &str) -> Self { + Self { + finalizations_key: format!( + "{prefix}-finalizations-by-height-prunable-v1-key" + ), + finalizations_value: format!( + "{prefix}-finalizations-by-height-prunable-v1-value" + ), + blocks_key: format!("{prefix}-finalized-blocks-prunable-v1-key"), + blocks_value: format!("{prefix}-finalized-blocks-prunable-v1-value"), + metadata: format!("{prefix}-history-prunable-v1-metadata"), + } + } + + fn archives(&self) -> [&str; 4] { + [ + &self.finalizations_key, + &self.finalizations_value, + &self.blocks_key, + &self.blocks_value, + ] + } +} + +pub(crate) struct Opened +where + E: BufferPooler + Clock + Storage + Metrics, +{ + pub(crate) finalizations: FinalizationsArchive, + pub(crate) blocks: BlocksArchive, + pub(crate) partitions: Partitions, + pub(crate) marker_status: MarkerStatus, +} + +pub(crate) const fn policy_floor(tip: u64, retained: u64) -> u64 { + tip.saturating_add(1).saturating_sub(retained) +} + +pub(crate) fn validate_production_retention() -> Result<(), HistoryError> { + if MAX_PENDING_ACKS.get() != 16 { + return Err(HistoryError::Retention( + "MAX_PENDING_ACKS changed; update the mandatory rewind assertion", + )); + } + if LOGICAL_RETENTION != 217 { + return Err(HistoryError::Retention( + "logical finalized-history retention must be 217", + )); + } + if MAX_RETAINED_HEIGHTS != 4_343 { + return Err(HistoryError::Retention( + "section-granular finalized-history bound must be 4,343", + )); + } + Ok(()) +} + +pub(crate) async fn open( + context: &E, + prefix: &str, + page_cache: CacheRef, + block_codec_config: ::Cfg, +) -> Result, HistoryError> +where + E: BufferPooler + Clock + Spawner + Storage + Metrics, +{ + validate_production_retention()?; + reject_legacy(context, prefix).await?; + + let partitions = Partitions::new(prefix); + let marker_existed = partition_exists(context, &partitions.metadata).await?; + let mut archive_existed = [false; 4]; + for (exists, partition) in archive_existed + .iter_mut() + .zip(partitions.archives().into_iter()) + { + *exists = partition_exists(context, partition).await?; + } + let mut finalizations = + init_finalizations(context, &partitions, page_cache.clone()).await?; + let mut blocks = init_blocks( + context, + &partitions, + page_cache.clone(), + block_codec_config, + ) + .await?; + + let marker_status = if marker_existed { + validate_marker(context, &partitions.metadata).await?; + let archives_empty = finalizations.first_index().is_none() && blocks.first_index().is_none(); + let all_lazy = archive_existed.iter().all(|exists| !exists); + if !(archive_existed.iter().all(|exists| *exists) || archives_empty && all_lazy) { + let partition = partitions + .archives() + .into_iter() + .zip(archive_existed) + .find_map(|(partition, exists)| (!exists).then_some(partition)) + .expect("at least one marked archive partition is missing"); + return Err(HistoryError::MarkedPartitionMissing { + partition: partition.to_string(), + }); + } + MarkerStatus::Present + } else { + if let Some(height) = finalizations.first_index() { + return Err(HistoryError::NonEmptyUnmarked { + partition: partitions.finalizations_key.clone(), + height, + }); + } + if let Some(height) = blocks.first_index() { + return Err(HistoryError::NonEmptyUnmarked { + partition: partitions.blocks_key.clone(), + height, + }); + } + + // Ensure all newly-created archive blobs are durable and structurally reopenable before + // committing the format marker. + finalizations.sync().await.map_err(|source| HistoryError::Archive { + partition: partitions.finalizations_key.clone(), + source, + })?; + blocks.sync().await.map_err(|source| HistoryError::Archive { + partition: partitions.blocks_key.clone(), + source, + })?; + drop(finalizations); + drop(blocks); + finalizations = init_finalizations(context, &partitions, page_cache.clone()).await?; + blocks = init_blocks(context, &partitions, page_cache, block_codec_config).await?; + write_marker(context, &partitions.metadata).await?; + + if archive_existed.into_iter().any(|exists| exists) { + MarkerStatus::RecoveredPartialInitialization + } else { + MarkerStatus::Created + } + }; + + Ok(Opened { + finalizations, + blocks, + partitions, + marker_status, + }) +} + +async fn init_finalizations( + context: &E, + partitions: &Partitions, + page_cache: CacheRef, +) -> Result, HistoryError> +where + E: BufferPooler + Clock + Spawner + Storage + Metrics, +{ + prunable::Archive::init( + context.child("finalizations_by_height"), + prunable::Config { + translator: EightCap, + key_partition: partitions.finalizations_key.clone(), + key_page_cache: page_cache, + value_partition: partitions.finalizations_value.clone(), + compression: Some(3), + codec_config: (), + items_per_section: PRUNABLE_ITEMS_PER_SECTION, + key_write_buffer: WRITE_BUFFER, + value_write_buffer: WRITE_BUFFER, + replay_buffer: REPLAY_BUFFER, + }, + ) + .await + .map_err(|source| HistoryError::Archive { + partition: partitions.finalizations_key.clone(), + source, + }) +} + +async fn init_blocks( + context: &E, + partitions: &Partitions, + page_cache: CacheRef, + codec_config: ::Cfg, +) -> Result, HistoryError> +where + E: BufferPooler + Clock + Spawner + Storage + Metrics, +{ + prunable::Archive::init( + context.child("finalized_blocks"), + prunable::Config { + translator: EightCap, + key_partition: partitions.blocks_key.clone(), + key_page_cache: page_cache, + value_partition: partitions.blocks_value.clone(), + compression: Some(3), + codec_config, + items_per_section: PRUNABLE_ITEMS_PER_SECTION, + key_write_buffer: WRITE_BUFFER, + value_write_buffer: WRITE_BUFFER, + replay_buffer: REPLAY_BUFFER, + }, + ) + .await + .map_err(|source| HistoryError::Archive { + partition: partitions.blocks_key.clone(), + source, + }) +} + +async fn partition_exists(context: &E, partition: &str) -> Result { + match context.scan(partition).await { + Ok(_) => Ok(true), + Err(commonware_runtime::Error::PartitionMissing(_)) => Ok(false), + Err(source) => Err(HistoryError::Inspect { + partition: partition.to_string(), + source, + }), + } +} + +async fn validate_marker(context: &E, partition: &str) -> Result<(), HistoryError> +where + E: BufferPooler + Clock + Storage + Metrics + Spawner, +{ + let marker = Metadata::::init( + context.child("history_marker"), + metadata::Config { + partition: partition.to_string(), + codec_config: (), + }, + ) + .await + .map_err(|source| HistoryError::Marker { + partition: partition.to_string(), + source, + })?; + let found = marker.get(&MARKER_KEY).copied(); + if found != Some(FORMAT_VERSION) { + return Err(HistoryError::InvalidMarker { + partition: partition.to_string(), + expected: FORMAT_VERSION, + found, + }); + } + Ok(()) +} + +async fn write_marker(context: &E, partition: &str) -> Result<(), HistoryError> +where + E: BufferPooler + Clock + Storage + Metrics + Spawner, +{ + let mut marker = Metadata::::init( + context.child("history_marker"), + metadata::Config { + partition: partition.to_string(), + codec_config: (), + }, + ) + .await + .map_err(|source| HistoryError::Marker { + partition: partition.to_string(), + source, + })?; + marker + .put_sync(MARKER_KEY, FORMAT_VERSION) + .await + .map_err(|source| HistoryError::Marker { + partition: partition.to_string(), + source, + }) +} + +async fn reject_legacy(context: &E, prefix: &str) -> Result<(), HistoryError> { + const FINALIZATION_SUFFIXES: [&str; 5] = [ + "finalizations-by-height-metadata", + "finalizations-by-height-freezer-table", + "finalizations-by-height-freezer-key", + "finalizations-by-height-freezer-value", + "finalizations-by-height-ordinal", + ]; + const BLOCK_SUFFIXES: [&str; 5] = [ + "finalized_blocks-metadata", + "finalized_blocks-freezer-table", + "finalized_blocks-freezer-key", + "finalized_blocks-freezer-value", + "finalized_blocks-ordinal", + ]; + + for suffix in FINALIZATION_SUFFIXES.into_iter().chain(BLOCK_SUFFIXES) { + let partition = format!("{prefix}-{suffix}"); + if partition_exists(context, &partition).await? { + return Err(HistoryError::LegacyPartition { partition }); + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use commonware_runtime::{buffer::paged::CacheRef, deterministic, Runner as _}; + use commonware_utils::{NZU16, NZU32, NZUsize}; + + #[test] + fn production_retention_arithmetic() { + validate_production_retention().unwrap(); + assert_eq!(policy_floor(216, LOGICAL_RETENTION), 0); + assert_eq!(policy_floor(217, LOGICAL_RETENTION), 1); + assert_eq!(LOGICAL_RETENTION, 217); + assert_eq!(MAX_RETAINED_HEIGHTS, 4_343); + } + + #[test] + fn fresh_format_marker_is_durable_and_legacy_is_not_created() { + deterministic::Runner::default().start(|context| async move { + let page_cache = CacheRef::from_pooler(&context, NZU16!(4_096), NZUsize!(32)); + let first = open( + &context, + "validator", + page_cache.clone(), + (NZU32!(1), ()), + ) + .await + .expect("initialize fresh history"); + assert_eq!(first.marker_status, MarkerStatus::Created); + assert_eq!(first.finalizations.first_index(), None); + assert_eq!(first.blocks.first_index(), None); + drop(first); + + let reopened = open( + &context, + "validator", + page_cache, + (NZU32!(1), ()), + ) + .await + .expect("reopen marked history"); + assert_eq!(reopened.marker_status, MarkerStatus::Present); + + for suffix in [ + "finalizations-by-height-metadata", + "finalizations-by-height-freezer-table", + "finalizations-by-height-freezer-key", + "finalizations-by-height-freezer-value", + "finalizations-by-height-ordinal", + "finalized_blocks-metadata", + "finalized_blocks-freezer-table", + "finalized_blocks-freezer-key", + "finalized_blocks-freezer-value", + "finalized_blocks-ordinal", + ] { + assert!(matches!( + context.scan(&format!("validator-{suffix}")).await, + Err(commonware_runtime::Error::PartitionMissing(_)) + )); + } + }); + } + + #[test] + fn refuses_known_legacy_partition() { + deterministic::Runner::default().start(|context| async move { + context + .open("validator-finalized_blocks-ordinal", b"legacy") + .await + .expect("create legacy partition"); + let page_cache = CacheRef::from_pooler(&context, NZU16!(4_096), NZUsize!(32)); + let result = open( + &context, + "validator", + page_cache, + (NZU32!(1), ()), + ) + .await; + assert!(matches!(result, Err(HistoryError::LegacyPartition { .. }))); + }); + } +} diff --git a/examples/coins/chain/src/indexer/backfiller/consumer.rs b/examples/coins/chain/src/indexer/backfiller/consumer.rs index e714eb2..0bd66e8 100644 --- a/examples/coins/chain/src/indexer/backfiller/consumer.rs +++ b/examples/coins/chain/src/indexer/backfiller/consumer.rs @@ -1,17 +1,14 @@ -use super::{Decision, Entry, SharedState}; +use super::{ + producer::{Admission, AdmissionReceiver}, + Decision, Entry, SharedState, +}; use crate::indexer::{ metrics::{ - estimated_finalized_bytes, BackfillDecision, BackfillPhase, BackfillResetReason, - BackfillUploadGuard, BackfillWaitReason, BlockMetricSource, QueueReadSource, QueueStatus, - SharedCacheSource, + estimated_finalized_bytes, BackfillDecision, BackfillPhase, BackfillWaitReason, + BlockMetricSource, ProducerStatus, QueueReadSource, QueueStatus, }, - Client, IndexerMetrics, -}; -use crate::{Block, Finalized, Scheme}; -use commonware_consensus::marshal::{ - core::Mailbox as MarshalMailbox, standard::Standard, Identifier, + Client, IndexerMetrics, SpoolLimits, }; -use commonware_consensus::types::Height; use commonware_cryptography::sha256::Digest; use commonware_macros::select_loop; use commonware_runtime::{ @@ -22,7 +19,7 @@ use commonware_storage::queue; use commonware_utils::futures::{OptionFuture, Pool}; use std::{ num::NonZeroUsize, - time::{Duration, Instant}, + time::{Duration, Instant, UNIX_EPOCH}, }; use tracing::{debug, warn}; @@ -45,47 +42,39 @@ enum Completion { Skipped { position: u64, height: u64, - }, - Stale { - reason: BackfillResetReason, - first_height: u64, - first_digest: Digest, - attempts: u64, - elapsed: Duration, + digest: Digest, }, } pub(crate) struct Config { pub(crate) max_active: NonZeroUsize, pub(crate) retry: Duration, - pub(crate) missing_finalization_grace: Duration, - pub(crate) mismatched_finalization_grace: Duration, + pub(crate) spool_limits: SpoolLimits, } pub struct Consumer { context: ContextCell, client: C, - marshal: MarshalMailbox>, metrics: IndexerMetrics, upload_results: status::Counter, uploads: SharedState, writer: queue::Writer, reader: queue::Reader, + admission: AdmissionReceiver, active: Pool, max_active: NonZeroUsize, retry: Duration, - missing_finalization_grace: Duration, - mismatched_finalization_grace: Duration, + spool_limits: SpoolLimits, } impl Consumer { pub fn new( context: E, client: C, - marshal: MarshalMailbox>, metrics: IndexerMetrics, uploads: SharedState, backfiller: (queue::Writer, queue::Reader), + admission: AdmissionReceiver, config: Config, ) -> Self { let upload_results = context.register( @@ -96,25 +85,23 @@ impl Consumer< let Config { max_active, retry, - missing_finalization_grace, - mismatched_finalization_grace, + spool_limits, } = config; metrics.backfill_configured(max_active.get()); let (writer, reader) = backfiller; Self { context: ContextCell::new(context), client, - marshal, metrics, upload_results, uploads, writer, reader, + admission, active: Pool::default(), max_active, retry, - missing_finalization_grace, - mismatched_finalization_grace, + spool_limits, } } @@ -127,25 +114,24 @@ impl Consumer< self.context, on_start => { self.fill_slots().await; - if self.active.is_empty() { - let item = Self::recv_queue(&mut self.reader, self.metrics.clone()).await; - let Some((position, entry)) = item else { - warn!("consumer queue closed"); - break; - }; - self.start_upload(position, entry).await; - continue; - } let item = OptionFuture::from( (self.active.len() < self.max_active.get()).then(|| { Self::recv_queue(&mut self.reader, self.metrics.clone()) }), ); + let admission = self.admission.recv(); }, on_stopped => {}, completion = self.active.next_completed() => { self.complete(completion).await; }, + request = admission => { + let Some(request) = request else { + warn!("indexer spool admission coordinator closed"); + break; + }; + self.admit(request).await; + }, item = item => { match item { Some((position, entry)) => { @@ -160,6 +146,87 @@ impl Consumer< } } + async fn admit(&mut self, admission: Admission) { + let Admission { entry, response } = admission; + loop { + let now_millis: u64 = self + .context + .current() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .try_into() + .unwrap_or(u64::MAX); + let next = { + let uploads = self.uploads.lock(); + let (entries, bytes) = uploads.spool_usage(); + uploads.oldest_queued().and_then( + |(digest, position, height, old_bytes, enqueued_at)| { + let age = Duration::from_millis( + now_millis.saturating_sub(enqueued_at), + ); + let status = if age >= self.spool_limits.max_age { + Some(ProducerStatus::ExpiredAge) + } else if entries >= self.spool_limits.max_entries { + Some(ProducerStatus::ExpiredEntries) + } else if bytes.saturating_add(entry.encoded_len) + > self.spool_limits.max_bytes + { + Some(ProducerStatus::ExpiredBytes) + } else { + None + }; + status.map(|status| { + (digest, position, height, old_bytes, age, status) + }) + }, + ) + }; + let Some((digest, position, height, encoded_len, age, status)) = next else { + break; + }; + + // Advancing the floor invalidates positions held by upload tasks. Cancel all of them + // and replay the still-live suffix so a stale completion cannot be applied twice. + self.active.cancel_all(); + self.reader + .ack_up_to(position.saturating_add(1)) + .await + .unwrap_or_else(|error| panic!("failed to expire indexer spool floor: {error:?}")); + let sync_started = Instant::now(); + self.writer.sync().await.unwrap_or_else(|error| { + panic!("failed to durably sync expired indexer spool floor: {error:?}") + }); + self.metrics + .queue_synced(QueueStatus::Success, sync_started.elapsed()); + self.reader.reset().await; + let expired = self.uploads.lock().expire_through(position); + for (_, _, _) in &expired { + self.metrics.producer_recorded(status, Duration::ZERO); + } + warn!( + height, + ?digest, + encoded_len, + ?age, + ?status, + expired_entries = expired.len(), + "terminally expired oldest finalized indexer payload before admission" + ); + } + + self.metrics.queue_entry(entry.height()); + let position = self.writer.enqueue(entry.clone()).await.unwrap_or_else(|error| { + self.metrics.queue_enqueued(QueueStatus::Failure); + panic!("failed to enqueue finalized indexer payload: {error:?}") + }); + self.uploads.lock().mark_queued(position, &entry); + self.metrics.queue_enqueued(QueueStatus::Success); + response + .send(()) + .expect("indexer spool producer stopped before admission completed"); + } + async fn fill_slots(&mut self) { while self.active.len() < self.max_active.get() { let item = Self::try_recv_queue(&mut self.reader, self.metrics.clone()).await; @@ -177,7 +244,7 @@ impl Consumer< match reader.recv().await { Ok(Some((position, entry))) => { metrics.queue_read(QueueReadSource::Recv, QueueStatus::Success); - metrics.queue_entry(entry.height); + metrics.queue_entry(entry.height()); Some((position, entry)) } Ok(None) => { @@ -198,7 +265,7 @@ impl Consumer< match reader.try_recv().await { Ok(Some((position, entry))) => { metrics.queue_read(QueueReadSource::TryRecv, QueueStatus::Success); - metrics.queue_entry(entry.height); + metrics.queue_entry(entry.height()); Some((position, entry)) } Ok(None) => { @@ -213,7 +280,26 @@ impl Consumer< } async fn start_upload(&mut self, position: u64, entry: Entry) { - let Entry { height, digest } = entry; + let height = entry.height(); + let digest = entry.digest(); + let enqueued_at_millis = entry.enqueued_at_millis; + let finalized = entry.finalized; + if self.payload_age(enqueued_at_millis) >= self.spool_limits.max_age { + self.metrics + .producer_recorded(ProducerStatus::ExpiredAge, Duration::ZERO); + warn!( + height, + ?digest, + "terminally expiring finalized indexer payload at configured age bound" + ); + self.complete(Completion::Skipped { + position, + height, + digest, + }) + .await; + return; + } let decision = self.uploads.lock().should_upload(&digest); self.metrics .backfill_decision((&decision).into(), BackfillPhase::Start); @@ -222,7 +308,11 @@ impl Consumer< let mut upload = self.metrics.start_backfill_upload(); upload.skipped(); } - self.complete(Completion::Skipped { position, height }) + self.complete(Completion::Skipped { + position, + height, + digest, + }) .await; debug!(?digest, "consumer skipping already-uploaded block"); return; @@ -235,82 +325,35 @@ impl Consumer< .with_attribute("digest", digest) .with_attribute("height", height); let client = self.client.clone(); - let marshal = self.marshal.clone(); let metrics = self.metrics.clone(); let upload_results = self.upload_results.clone(); let uploads = self.uploads.clone(); let retry = self.retry; - let missing_finalization_grace = self.missing_finalization_grace; - let mismatched_finalization_grace = self.mismatched_finalization_grace; + let max_age = self.spool_limits.max_age; async move { let mut upload = metrics.start_backfill_upload(); - let Some(block) = - Self::wait_for_uploadable_block( - &context, &marshal, &metrics, &uploads, &mut upload, digest, retry, - ) - .await - else { - upload.skipped(); - debug!(?digest, "skipping previously uploaded block"); - return Completion::Skipped { position, height }; - }; + metrics.observe_block(BlockMetricSource::ConsumerCached, &finalized.block); + upload.hold_block(&finalized.block); - let mut proof_failure = ProofFailure::default(); - let finalized = loop { - let Some(proof) = marshal.get_finalization(Height::new(height)).await else { - if let Some(stale) = proof_failure.observe( - &context, - BackfillResetReason::MissingFinalization, - missing_finalization_grace, - height, - digest, - ) { - upload.abandoned(); - return stale; - } - warn!( - height, - ?digest, - "consumer could not find finalization, retrying" - ); - Self::wait( - &context, - &metrics, - BackfillWaitReason::MissingFinalization, - retry, - ) - .await; - continue; - }; - if proof.proposal.payload != digest { - if let Some(stale) = proof_failure.observe( - &context, - BackfillResetReason::MismatchedFinalization, - mismatched_finalization_grace, + loop { + let now_millis: u64 = context + .current() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .try_into() + .unwrap_or(u64::MAX); + // Backward wall-clock movement produces age zero until the clock catches up. + let age = Duration::from_millis(now_millis.saturating_sub(enqueued_at_millis)); + if age >= max_age { + metrics.producer_recorded(ProducerStatus::ExpiredAge, Duration::ZERO); + warn!(height, ?digest, ?age, "terminally expiring active finalized indexer upload at configured age bound"); + return Completion::Skipped { + position, height, digest, - ) { - upload.abandoned(); - return stale; - } - warn!( - height, - ?digest, - "consumer found mismatched finalization, retrying" - ); - Self::wait( - &context, - &metrics, - BackfillWaitReason::MismatchedFinalization, - retry, - ) - .await; - continue; + }; } - break Finalized::new(proof, block.clone()); - }; - - loop { let decision = { let uploads = uploads.lock(); uploads.should_upload(&digest) @@ -320,7 +363,11 @@ impl Consumer< Decision::Skip => { upload.skipped(); debug!(?digest, "skipping previously uploaded block"); - return Completion::Skipped { position, height }; + return Completion::Skipped { + position, + height, + digest, + }; } Decision::Wait => { Self::wait( @@ -363,86 +410,17 @@ impl Consumer< }); } - async fn wait_for_uploadable_block( - context: &E, - marshal: &MarshalMailbox>, - metrics: &IndexerMetrics, - uploads: &SharedState, - upload: &mut BackfillUploadGuard, - digest: Digest, - retry: Duration, - ) -> Option { - enum NextBlock { - AlreadyUploaded, - WaitForCertificate, - Ready(Box), - FetchFromMarshal, - } - - loop { - let next = { - let uploads = uploads.lock(); - match uploads.should_upload(&digest) { - Decision::Skip => { - metrics.backfill_decision( - BackfillDecision::Skip, - BackfillPhase::BeforeBlock, - ); - NextBlock::AlreadyUploaded - } - Decision::Wait => { - metrics.backfill_decision( - BackfillDecision::Wait, - BackfillPhase::BeforeBlock, - ); - NextBlock::WaitForCertificate - } - Decision::Proceed => { - metrics.backfill_decision( - BackfillDecision::Proceed, - BackfillPhase::BeforeBlock, - ); - uploads - .cached_block(&digest) - .map(|block| NextBlock::Ready(Box::new(block))) - .unwrap_or(NextBlock::FetchFromMarshal) - } - } - }; - - match next { - NextBlock::AlreadyUploaded => return None, - NextBlock::WaitForCertificate => { - Self::wait( - context, - metrics, - BackfillWaitReason::CertificateUpload, - retry, - ) - .await; - } - NextBlock::Ready(block) => { - metrics.observe_block(BlockMetricSource::ConsumerCached, &block); - upload.hold_block(&block); - return Some(*block); - } - NextBlock::FetchFromMarshal => { - if let Some(block) = marshal.get_block(Identifier::Digest(digest)).await { - metrics.observe_block(BlockMetricSource::ConsumerMarshal, &block); - uploads - .lock() - .cache_block(block.clone(), SharedCacheSource::ConsumerMarshal); - upload.hold_block(&block); - return Some(block); - } - warn!( - ?digest, - "consumer could not find block in marshal, retrying" - ); - Self::wait(context, metrics, BackfillWaitReason::MissingBlock, retry).await; - } - } - } + fn payload_age(&self, enqueued_at_millis: u64) -> Duration { + let now_millis: u64 = self + .context + .current() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .try_into() + .unwrap_or(u64::MAX); + // Backward wall-clock movement produces age zero until the clock catches up. + Duration::from_millis(now_millis.saturating_sub(enqueued_at_millis)) } async fn wait( @@ -462,28 +440,22 @@ impl Consumer< } async fn complete(&mut self, completion: Completion) { - let (position, height) = match completion { + let (position, height, digest) = match completion { Completion::Uploaded { position, height, digest, } => { self.uploads.lock().mark_uploaded(digest, height); - (position, height) - } - Completion::Skipped { position, height } => (position, height), - Completion::Stale { - reason, - first_height, - first_digest, - attempts, - elapsed, - } => { - self.reset_stale_queue(reason, first_height, first_digest, attempts, elapsed) - .await; - return; + (position, height, digest) } + Completion::Skipped { + position, + height, + digest, + } => (position, height, digest), }; + self.uploads.lock().finish_queued(&digest); let floor = self.reader.ack_floor().await; match self.reader.ack(position).await { @@ -511,109 +483,4 @@ impl Consumer< } } - async fn reset_stale_queue( - &mut self, - reason: BackfillResetReason, - first_height: u64, - first_digest: Digest, - attempts: u64, - elapsed: Duration, - ) { - let queue_size = self.writer.size().await; - let ack_floor = self.reader.ack_floor().await; - let abandoned_entries = queue_size.saturating_sub(ack_floor); - let tip = Self::latest_proof_bearing_tip(self.marshal.clone()).await; - let (tip_height, tip_digest) = tip.unwrap_or((0, first_digest)); - let abandoned_height_span = tip_height.saturating_sub(first_height); - - warn!( - ?reason, - first_height, - ?first_digest, - tip_height, - ?tip_digest, - attempts, - ?elapsed, - abandoned_entries, - abandoned_height_span, - "resetting stale durable backfill queue" - ); - - self.active.cancel_all(); - match self.reader.ack_up_to(queue_size).await { - Ok(()) => self.metrics.queue_acked(QueueStatus::Success), - Err(err) => { - self.metrics.queue_acked(QueueStatus::Failure); - panic!("failed to ack stale finalized queue: {err:?}"); - } - } - let sync_started = Instant::now(); - match self.writer.sync().await { - Ok(()) => self - .metrics - .queue_synced(QueueStatus::Success, sync_started.elapsed()), - Err(err) => { - self.metrics - .queue_synced(QueueStatus::Failure, sync_started.elapsed()); - panic!("failed to sync stale finalized queue reset: {err:?}"); - } - } - self.reader.reset().await; - self.metrics.queue_ack_floor(tip_height); - self.metrics.backfill_queue_reset( - reason, - abandoned_entries, - abandoned_height_span, - ); - self.uploads.lock().advance_queue_floor(tip_height); - self.uploads.lock().restart_above(tip_height); - } - - async fn latest_proof_bearing_tip( - marshal: MarshalMailbox>, - ) -> Option<(u64, Digest)> { - let (height, _) = marshal.get_info(Identifier::Latest).await?; - for height in (0..=height.get()).rev() { - if let Some(proof) = marshal.get_finalization(Height::new(height)).await { - return Some((height, proof.proposal.payload)); - } - } - None - } -} - -#[derive(Default)] -struct ProofFailure { - reason: Option, - first_seen: Option, - attempts: u64, -} - -impl ProofFailure { - fn observe( - &mut self, - context: &E, - reason: BackfillResetReason, - grace: Duration, - height: u64, - digest: Digest, - ) -> Option { - if self.reason != Some(reason) { - self.reason = Some(reason); - self.first_seen = Some(context.current()); - self.attempts = 0; - } - self.attempts = self.attempts.saturating_add(1); - let elapsed = context - .current() - .duration_since(self.first_seen.expect("proof failure start")) - .unwrap_or_default(); - (elapsed >= grace).then_some(Completion::Stale { - reason, - first_height: height, - first_digest: digest, - attempts: self.attempts, - elapsed, - }) - } } diff --git a/examples/coins/chain/src/indexer/backfiller/producer.rs b/examples/coins/chain/src/indexer/backfiller/producer.rs index 88b01e4..fccdb5d 100644 --- a/examples/coins/chain/src/indexer/backfiller/producer.rs +++ b/examples/coins/chain/src/indexer/backfiller/producer.rs @@ -3,23 +3,32 @@ use crate::{ indexer::{ metrics::{ estimated_block_bytes, BlockMetricSource, ProducerActivity, ProducerStatus, - QueueStatus, }, - IndexerMetrics, + IndexerMetrics, SpoolLimits, }, - Block, + Block, Finalized, Scheme, }; use commonware_actor::{ mailbox::{self, Overflow, Policy}, Feedback, }; -use commonware_consensus::{marshal::Update, Reporter}; +use commonware_consensus::{ + marshal::{core::Mailbox as MarshalMailbox, standard::Standard, Update}, + types::Height, + Reporter, +}; use commonware_runtime::{ spawn_cell, BufferPooler, Clock, ContextCell, Handle, Metrics, Spawner, Storage, }; -use commonware_storage::queue; use commonware_utils::{acknowledgement::Exact, Acknowledgement}; -use std::{collections::VecDeque, num::NonZeroUsize, sync::Arc, time::Instant}; +use commonware_utils::channel::oneshot; +use std::{ + collections::VecDeque, + num::NonZeroUsize, + sync::Arc, + time::{Duration, Instant, UNIX_EPOCH}, +}; +use tracing::warn; #[derive(Clone)] pub struct Producer { @@ -27,6 +36,22 @@ pub struct Producer { metrics: IndexerMetrics, } +pub(crate) struct Admission { + pub(crate) entry: Entry, + pub(crate) response: oneshot::Sender<()>, +} + +impl Policy for Admission { + type Overflow = VecDeque; + + fn handle(overflow: &mut Self::Overflow, message: Self) { + overflow.push_back(message); + } +} + +pub(crate) type AdmissionSender = mailbox::Sender; +pub(crate) type AdmissionReceiver = mailbox::Receiver; + struct Message { block: Arc, block_estimated_bytes: u64, @@ -87,8 +112,21 @@ struct Actor { context: ContextCell, uploads: SharedState, metrics: IndexerMetrics, - writer: queue::Writer, + marshal: MarshalMailbox>, + admission: AdmissionSender, receiver: mailbox::Receiver, + retry: Duration, + missing_finalization_grace: Duration, + mismatched_finalization_grace: Duration, + spool_limits: SpoolLimits, +} + +pub(crate) struct Config { + pub(crate) mailbox_size: NonZeroUsize, + pub(crate) retry: Duration, + pub(crate) missing_finalization_grace: Duration, + pub(crate) mismatched_finalization_grace: Duration, + pub(crate) spool_limits: SpoolLimits, } impl Reporter for Producer { @@ -123,16 +161,29 @@ impl Actor { context: E, uploads: SharedState, metrics: IndexerMetrics, - writer: queue::Writer, - mailbox_size: NonZeroUsize, + marshal: MarshalMailbox>, + admission: AdmissionSender, + config: Config, ) -> (Self, Producer) { + let Config { + mailbox_size, + retry, + missing_finalization_grace, + mismatched_finalization_grace, + spool_limits, + } = config; let (sender, receiver) = mailbox::new(context.child("mailbox"), mailbox_size); let actor = Self { context: ContextCell::new(context), uploads, metrics: metrics.clone(), - writer, + marshal, + admission, receiver, + retry, + missing_finalization_grace, + mismatched_finalization_grace, + spool_limits, }; ( actor, @@ -158,25 +209,91 @@ impl Actor { let started = Instant::now(); self.metrics .observe_block(BlockMetricSource::ProducerRecord, block); - let Some(entry) = self.uploads.lock().record(block) else { + let Some(candidate) = self.uploads.lock().record(block) else { self.metrics.producer_recorded( ProducerStatus::AlreadyUploaded, started.elapsed(), ); return; }; - self.metrics.queue_entry(entry.height); - match self.writer.enqueue(entry).await { - Ok(_) => { - self.metrics.queue_enqueued(QueueStatus::Success); - self.metrics - .producer_recorded(ProducerStatus::Recorded, started.elapsed()); - } - Err(err) => { - self.metrics.queue_enqueued(QueueStatus::Failure); - panic!("failed to enqueue finalized digest: {err:?}"); + let first_seen = self.context.current(); + self.metrics.certificate_request_started(); + let proof = loop { + let Some(proof) = self + .marshal + .get_finalization(Height::new(candidate.height)) + .await + else { + let elapsed = self + .context + .current() + .duration_since(first_seen) + .unwrap_or_default(); + assert!( + elapsed < self.missing_finalization_grace, + "marshal has no finalization certificate for durable indexer payload at height {} after {:?}", + candidate.height, + elapsed, + ); + warn!(height = candidate.height, ?elapsed, "waiting for finalized certificate before spooling indexer payload"); + self.context.sleep(self.retry).await; + continue; + }; + if proof.proposal.payload != candidate.digest + || proof.proposal.round.epoch() != block.context.round.epoch() + { + let elapsed = self + .context + .current() + .duration_since(first_seen) + .unwrap_or_default(); + assert!( + elapsed < self.mismatched_finalization_grace, + "marshal finalization certificate conflicts with block at height {} after {:?}", + candidate.height, + elapsed, + ); + warn!(height = candidate.height, ?elapsed, "waiting for matching finalized certificate before spooling indexer payload"); + self.context.sleep(self.retry).await; + continue; } + break proof; + }; + self.metrics.certificate_request_finished(); + let enqueued_at_millis = self + .context + .current() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .try_into() + .unwrap_or(u64::MAX); + let entry = Entry::new(enqueued_at_millis, Finalized::new(proof, block.clone())); + let expiry = if entry.encoded_len > self.spool_limits.max_payload_bytes + || entry.encoded_len > self.spool_limits.max_bytes + { + Some(ProducerStatus::ExpiredOversized) + } else { + None + }; + if let Some(status) = expiry { + self.metrics.producer_recorded(status, started.elapsed()); + warn!( + height = entry.height(), + digest = ?entry.digest(), + encoded_len = entry.encoded_len, + ?status, + "terminally expiring finalized indexer payload to preserve spool availability bound" + ); + return; } + let (response, completed) = oneshot::channel(); + let _ = self.admission.enqueue(Admission { entry, response }); + completed + .await + .expect("indexer spool admission coordinator stopped"); + self.metrics + .producer_recorded(ProducerStatus::Recorded, started.elapsed()); } } @@ -184,15 +301,23 @@ pub fn init( context: E, uploads: SharedState, metrics: IndexerMetrics, - writer: queue::Writer, - mailbox_size: NonZeroUsize, -) -> Producer + marshal: MarshalMailbox>, + admission: AdmissionSender, + config: Config, +) -> (Producer, Handle<()>) where E: BufferPooler + Clock + Storage + Metrics + Spawner, { - let (actor, producer) = Actor::new(context, uploads, metrics, writer, mailbox_size); - actor.start(); - producer + let (actor, producer) = Actor::new( + context, + uploads, + metrics, + marshal, + admission, + config, + ); + let handle = actor.start(); + (producer, handle) } #[cfg(test)] @@ -200,16 +325,12 @@ mod tests { use super::*; use crate::{StateCommitment, Transaction, EPOCH}; use commonware_consensus::types::{Height, Round, View}; - use commonware_cryptography::{ed25519, Digestible, Hasher, Sha256, Signer}; - use commonware_runtime::{ - buffer::paged::CacheRef, deterministic, Runner as _, Supervisor as _, - }; + use commonware_cryptography::{ed25519, Hasher, Sha256, Signer}; + use commonware_runtime::{deterministic, Runner as _, Supervisor as _}; use commonware_storage::mmr::Location; use commonware_utils::{ - acknowledgement::Exact, range::NonEmptyRange, sync::Mutex, NZUsize, NZU16, NZU64, + acknowledgement::Exact, range::NonEmptyRange, NZUsize, }; - use futures::FutureExt; - use std::{sync::Arc, time::Duration}; fn state(height: u64) -> StateCommitment { StateCommitment { @@ -239,60 +360,6 @@ mod tests { ) } - #[test] - fn queues_finalized_block_before_acknowledging() { - deterministic::Runner::default().start(|context| async move { - let page_cache = CacheRef::from_pooler(&context, NZU16!(4_096), NZUsize!(128)); - let (writer, mut reader) = queue::shared::init( - context.child("queue"), - queue::Config { - partition: "indexer-producer-test".to_string(), - items_per_section: NZU64!(16), - compression: None, - codec_config: (), - page_cache, - write_buffer: NZUsize!(1024), - }, - ) - .await - .expect("init queue"); - let uploads = Arc::new(Mutex::new(crate::indexer::backfiller::State::new())); - let metrics = IndexerMetrics::register(&context.child("indexer")); - let mut producer = init(context.child("producer"), uploads, metrics, writer, NZUsize!(4)); - let block = block(2, 2, b"block"); - let digest = block.digest(); - let (ack, waiter) = Exact::handle(); - - assert!(producer - .report(commonware_consensus::marshal::Update::Block( - block.into(), - ack, - )) - .accepted()); - commonware_macros::select! { - result = waiter.fuse() => result.expect("acknowledged"), - _ = context.sleep(Duration::from_secs(1)) => panic!("ack timed out"), - } - - let (_, entry) = reader - .recv() - .await - .expect("read queue") - .expect("queued entry"); - assert_eq!(entry.height, 2); - assert_eq!(entry.digest, digest); - - let encoded = context.encode(); - assert!(encoded.contains( - "indexer_producer_report_total{activity=\"block\",status=\"enqueued\"} 1", - )); - assert!(encoded.contains( - "indexer_producer_report_total{activity=\"block\",status=\"recorded\"} 1", - )); - assert!(encoded.contains("indexer_producer_record_duration_seconds_bucket")); - }); - } - #[test] fn mailbox_overflow_tracks_retained_block_pressure() { deterministic::Runner::default().start(|context| async move { diff --git a/examples/coins/chain/src/indexer/backfiller/state.rs b/examples/coins/chain/src/indexer/backfiller/state.rs index 217b70f..e52c24c 100644 --- a/examples/coins/chain/src/indexer/backfiller/state.rs +++ b/examples/coins/chain/src/indexer/backfiller/state.rs @@ -5,10 +5,10 @@ use crate::{ }, IndexerMetrics, }, - Block, + Block, Finalized, }; use bytes::{Buf, BufMut}; -use commonware_codec::{self, FixedSize, Read, Write}; +use commonware_codec::{self, EncodeSize, FixedSize, Read, Write}; use commonware_cryptography::{sha256::Digest, Digestible}; use commonware_utils::{sync::Mutex, PrioritySet}; use std::{collections::BTreeMap, sync::Arc}; @@ -19,30 +19,84 @@ pub enum Decision { Proceed, } -#[derive(Clone, Copy, PartialEq, Eq)] -pub struct Entry { +pub struct Candidate { pub height: u64, pub digest: Digest, } -impl FixedSize for Entry { - const SIZE: usize = u64::SIZE + Digest::SIZE; +const ENTRY_VERSION: u8 = 1; + +#[derive(Clone)] +pub struct Entry { + pub version: u8, + pub enqueued_at_millis: u64, + pub encoded_len: u64, + pub finalized: Finalized, +} + +impl Entry { + pub fn new(enqueued_at_millis: u64, finalized: Finalized) -> Self { + let mut entry = Self { + version: ENTRY_VERSION, + enqueued_at_millis, + encoded_len: 0, + finalized, + }; + entry.encoded_len = entry.encode_size() as u64; + entry + } + + pub fn height(&self) -> u64 { + self.finalized.block.height.get() + } + + pub fn digest(&self) -> Digest { + self.finalized.block.digest() + } } impl Write for Entry { fn write(&self, buf: &mut impl BufMut) { - self.height.write(buf); - self.digest.write(buf); + self.version.write(buf); + self.enqueued_at_millis.write(buf); + self.encoded_len.write(buf); + self.finalized.write(buf); } } impl Read for Entry { - type Cfg = (); + type Cfg = ::Cfg; + + fn read_cfg(buf: &mut impl Buf, cfg: &Self::Cfg) -> Result { + let version = u8::read_cfg(buf, &())?; + if version != ENTRY_VERSION { + return Err(commonware_codec::Error::Invalid( + "indexer spool entry", + "unsupported version", + )); + } + let enqueued_at_millis = u64::read_cfg(buf, &())?; + let encoded_len = u64::read_cfg(buf, &())?; + let finalized = Finalized::read_cfg(buf, cfg)?; + let entry = Self { + version, + enqueued_at_millis, + encoded_len, + finalized, + }; + if entry.encode_size() as u64 != encoded_len { + return Err(commonware_codec::Error::Invalid( + "indexer spool entry", + "encoded length mismatch", + )); + } + Ok(entry) + } +} - fn read_cfg(buf: &mut impl Buf, _: &()) -> Result { - let height = u64::read_cfg(buf, &())?; - let digest = Digest::read_cfg(buf, &())?; - Ok(Self { height, digest }) +impl EncodeSize for Entry { + fn encode_size(&self) -> usize { + u8::SIZE + u64::SIZE + u64::SIZE + self.finalized.encode_size() } } @@ -55,6 +109,8 @@ pub struct State { cached_block_estimated_bytes: u64, certificate_uploads: BTreeMap, certificate_upload_refs: usize, + queued: BTreeMap, + queued_bytes: u64, metrics: Option, } @@ -74,6 +130,8 @@ impl State { cached_block_estimated_bytes: 0, certificate_uploads: BTreeMap::new(), certificate_upload_refs: 0, + queued: BTreeMap::new(), + queued_bytes: 0, metrics: None, } } @@ -89,8 +147,8 @@ impl State { self.uploaded.contains(digest) } - pub fn record(&mut self, block: &Block) -> Option { - let entry = Entry { + pub fn record(&mut self, block: &Block) -> Option { + let entry = Candidate { height: block.height.get(), digest: block.digest(), }; @@ -99,7 +157,7 @@ impl State { self.sync_metrics(); return None; } - if self.is_uploaded(&entry.digest) { + if self.is_uploaded(&entry.digest) || self.queued(&entry.digest) { self.sync_metrics(); return None; } @@ -108,12 +166,74 @@ impl State { Some(entry) } + pub fn queued(&self, digest: &Digest) -> bool { + self.queued.contains_key(digest) + } + + pub fn mark_queued(&mut self, position: u64, entry: &Entry) { + if let Some((_, _, old_bytes, _)) = + self.queued + .insert( + entry.digest(), + ( + position, + entry.height(), + entry.encoded_len, + entry.enqueued_at_millis, + ), + ) + { + self.queued_bytes = self.queued_bytes.saturating_sub(old_bytes); + } + self.queued_bytes = self.queued_bytes.saturating_add(entry.encoded_len); + self.sync_metrics(); + } + + pub fn recover_queued(&mut self, position: u64, entry: &Entry) { + self.observe_finalization(entry.height()); + self.mark_queued(position, entry); + } + + pub fn finish_queued(&mut self, digest: &Digest) { + if let Some((_, _, bytes, _)) = self.queued.remove(digest) { + self.queued_bytes = self.queued_bytes.saturating_sub(bytes); + } + self.sync_metrics(); + } + + pub fn spool_usage(&self) -> (u64, u64) { + (self.queued.len() as u64, self.queued_bytes) + } + + pub fn oldest_queued(&self) -> Option<(Digest, u64, u64, u64, u64)> { + self.queued + .iter() + .min_by_key(|(_, (position, _, _, _))| *position) + .map(|(digest, (position, height, bytes, enqueued_at))| { + (*digest, *position, *height, *bytes, *enqueued_at) + }) + } + + pub fn expire_through(&mut self, position: u64) -> Vec<(Digest, u64, u64)> { + let expired: Vec<_> = self + .queued + .iter() + .filter(|(_, (queued_position, _, _, _))| *queued_position <= position) + .map(|(digest, (_, height, bytes, _))| (*digest, *height, *bytes)) + .collect(); + for (digest, _, _) in &expired { + self.finish_queued(digest); + } + expired + } + pub fn advance_queue_floor(&mut self, height: u64) { self.acked_through = self.acked_through.max(height); self.prune(); self.sync_metrics(); } + #[cfg(test)] pub fn restart_above(&mut self, height: u64) { self.restart_watermark = self.restart_watermark.max(height); let pruned = self @@ -145,6 +265,7 @@ impl State { self.sync_metrics(); } + #[cfg(test)] pub fn cached_block(&self, digest: &Digest) -> Option { self.cached_blocks .get(digest) @@ -279,6 +400,20 @@ impl State { uploaded_digests: self.uploaded.len(), latest_finalized_height: self.latest_finalized, acked_through_height: self.acked_through, + spool_entries: self.queued.len(), + spool_logical_bytes: self.queued_bytes, + spool_oldest_height: self + .queued + .values() + .map(|(_, height, _, _)| *height) + .min() + .unwrap_or(0), + spool_oldest_enqueued_at_millis: self + .queued + .values() + .map(|(_, _, _, enqueued_at)| *enqueued_at) + .min() + .unwrap_or(0), }); } } @@ -290,7 +425,6 @@ pub type SharedState = Arc>; mod tests { use super::*; use crate::{StateCommitment, Transaction, EPOCH}; - use commonware_codec::{DecodeExt, Encode}; use commonware_consensus::types::{Height, Round, View}; use commonware_cryptography::{ed25519, Hasher, Sha256, Signer}; use commonware_runtime::{deterministic, Metrics as _, Runner as _, Supervisor as _}; @@ -325,20 +459,6 @@ mod tests { ) } - #[test] - fn entry_codec_roundtrips() { - let entry = Entry { - height: 7, - digest: Sha256::hash(b"block"), - }; - - let encoded = entry.encode(); - let decoded = Entry::decode(encoded).expect("decode entry"); - - assert_eq!(decoded.height, entry.height); - assert_eq!(decoded.digest, entry.digest); - } - #[test] fn record_caches_block_until_uploaded() { let mut state = State::new(); diff --git a/examples/coins/chain/src/indexer/metrics.rs b/examples/coins/chain/src/indexer/metrics.rs index 18d2f08..ca211ea 100644 --- a/examples/coins/chain/src/indexer/metrics.rs +++ b/examples/coins/chain/src/indexer/metrics.rs @@ -163,7 +163,6 @@ pub(crate) enum BlockMetricSource { ProducerRecord, LiveCertificate, ConsumerCached, - ConsumerMarshal, } impl BlockMetricSource { @@ -172,7 +171,6 @@ impl BlockMetricSource { Self::ProducerRecord => "producer_record", Self::LiveCertificate => "live_certificate", Self::ConsumerCached => "consumer_cached", - Self::ConsumerMarshal => "consumer_marshal", } } } @@ -192,7 +190,6 @@ impl EncodeLabelValueTrait for BlockMetricSource { pub(crate) enum SharedCacheSource { ProducerRecord, LiveCertificate, - ConsumerMarshal, } impl SharedCacheSource { @@ -200,7 +197,6 @@ impl SharedCacheSource { match self { Self::ProducerRecord => "producer_record", Self::LiveCertificate => "live_certificate", - Self::ConsumerMarshal => "consumer_marshal", } } } @@ -248,7 +244,6 @@ impl EncodeLabelValueTrait for SharedRetentionReason { pub(crate) enum BackfillStatus { Uploaded, Skipped, - Abandoned, } impl BackfillStatus { @@ -256,7 +251,6 @@ impl BackfillStatus { match self { Self::Uploaded => "uploaded", Self::Skipped => "skipped", - Self::Abandoned => "abandoned", } } } @@ -303,7 +297,6 @@ impl EncodeLabelValueTrait for BackfillDecision { #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub(crate) enum BackfillPhase { Start, - BeforeBlock, BeforeAttempt, } @@ -311,7 +304,6 @@ impl BackfillPhase { const fn as_str(self) -> &'static str { match self { Self::Start => "start", - Self::BeforeBlock => "before_block", Self::BeforeAttempt => "before_attempt", } } @@ -331,9 +323,6 @@ impl EncodeLabelValueTrait for BackfillPhase { #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub(crate) enum BackfillWaitReason { CertificateUpload, - MissingBlock, - MissingFinalization, - MismatchedFinalization, HttpError, } @@ -341,40 +330,11 @@ impl BackfillWaitReason { const fn as_str(self) -> &'static str { match self { Self::CertificateUpload => "certificate_upload", - Self::MissingBlock => "missing_block", - Self::MissingFinalization => "missing_finalization", - Self::MismatchedFinalization => "mismatched_finalization", Self::HttpError => "http_error", } } } -#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] -pub(crate) enum BackfillResetReason { - MissingFinalization, - MismatchedFinalization, -} - -impl BackfillResetReason { - const fn as_str(self) -> &'static str { - match self { - Self::MissingFinalization => "missing_finalization", - Self::MismatchedFinalization => "mismatched_finalization", - } - } -} - -impl EncodeLabelValueTrait for BackfillResetReason { - fn encode( - &self, - encoder: &mut commonware_runtime::telemetry::metrics::LabelValueEncoder, - ) -> Result<(), fmt::Error> { - use fmt::Write as _; - - encoder.write_str(self.as_str()) - } -} - impl EncodeLabelValueTrait for BackfillWaitReason { fn encode( &self, @@ -475,6 +435,10 @@ pub(crate) enum ProducerStatus { Ignored, Recorded, AlreadyUploaded, + ExpiredOversized, + ExpiredEntries, + ExpiredBytes, + ExpiredAge, } impl ProducerStatus { @@ -485,6 +449,10 @@ impl ProducerStatus { Self::Ignored => "ignored", Self::Recorded => "recorded", Self::AlreadyUploaded => "already_uploaded", + Self::ExpiredOversized => "expired_oversized", + Self::ExpiredEntries => "expired_entries", + Self::ExpiredBytes => "expired_bytes", + Self::ExpiredAge => "expired_age", } } } @@ -538,6 +506,10 @@ pub(crate) struct SharedStateSnapshot { pub(crate) uploaded_digests: usize, pub(crate) latest_finalized_height: u64, pub(crate) acked_through_height: u64, + pub(crate) spool_entries: usize, + pub(crate) spool_logical_bytes: u64, + pub(crate) spool_oldest_height: u64, + pub(crate) spool_oldest_enqueued_at_millis: u64, } #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, EncodeLabelSet)] @@ -599,11 +571,6 @@ struct BackfillWaitReasonLabel { reason: BackfillWaitReason, } -#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, EncodeLabelSet)] -struct BackfillResetReasonLabel { - reason: BackfillResetReason, -} - #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, EncodeLabelSet)] struct QueueStatusLabel { status: QueueStatus, @@ -666,6 +633,11 @@ pub(crate) struct Inner { shared_uploaded_digests: Gauge, shared_latest_finalized_height: Gauge, shared_acked_through_height: Gauge, + spool_entries: Gauge, + spool_logical_bytes: Gauge, + spool_oldest_height: Gauge, + spool_oldest_enqueued_at_millis: Gauge, + producer_certificate_requests: Gauge, shared_pruned: CounterFamily, shared_cache_inserted: CounterFamily, shared_cache_removed: CounterFamily, @@ -677,9 +649,6 @@ pub(crate) struct Inner { backfill_decision: CounterFamily, backfill_wait_duration: HistogramFamily, backfill_retry: CounterFamily, - backfill_queue_reset: CounterFamily, - backfill_queue_reset_abandoned_entries: HistogramFamily, - backfill_queue_reset_abandoned_height_span: HistogramFamily, backfill_active_block_estimated_bytes: Gauge, backfill_active_body_estimated_bytes: Gauge, queue_enqueue: CounterFamily, @@ -817,6 +786,26 @@ impl IndexerMetrics { "shared_acked_through_height", "Queue acknowledgement floor tracked by shared indexer state", ), + spool_entries: context.gauge( + "spool_entries", + "Current number of self-contained finalized payloads in the durable spool", + ), + spool_logical_bytes: context.gauge( + "spool_logical_bytes", + "Current encoded bytes in self-contained finalized spool payloads", + ), + spool_oldest_height: context.gauge( + "spool_oldest_height", + "Oldest finalized height retained in the durable spool", + ), + spool_oldest_enqueued_at_millis: context.gauge( + "spool_oldest_enqueued_at_millis", + "Wall-clock Unix millisecond timestamp of the oldest durable spool payload", + ), + producer_certificate_requests: context.gauge( + "spool_certificate_requests", + "Current finalized blocks awaiting a marshal certificate before spool admission", + ), shared_pruned: context.family( "shared_prune", "Total shared indexer state retention removals by reason", @@ -867,24 +856,6 @@ impl IndexerMetrics { "backfill_retry", "Total durable backfill retries by reason", ), - backfill_queue_reset: context.family( - "backfill_queue_reset", - "Total durable backfill queue resets by reason", - ), - backfill_queue_reset_abandoned_entries: context.register( - "backfill_queue_reset_abandoned_entries", - "Durable backfill entries abandoned by queue reset", - raw::Family:: raw::Histogram>::new_with_constructor( - local_histogram, - ), - ), - backfill_queue_reset_abandoned_height_span: context.register( - "backfill_queue_reset_abandoned_height_span", - "Durable backfill height span abandoned by queue reset", - raw::Family:: raw::Histogram>::new_with_constructor( - local_histogram, - ), - ), backfill_active_block_estimated_bytes: context.gauge( "backfill_active_block_estimated_bytes", "Current estimated encoded block bytes held by durable backfill upload tasks", @@ -1074,6 +1045,16 @@ impl IndexerMetrics { let _ = self .shared_acked_through_height .try_set(snapshot.acked_through_height); + let _ = self.spool_entries.try_set(snapshot.spool_entries); + let _ = self + .spool_logical_bytes + .try_set(snapshot.spool_logical_bytes); + let _ = self + .spool_oldest_height + .try_set(snapshot.spool_oldest_height); + let _ = self + .spool_oldest_enqueued_at_millis + .try_set(snapshot.spool_oldest_enqueued_at_millis); self.queue_ack_floor(snapshot.acked_through_height); let lag = snapshot .latest_finalized_height @@ -1081,6 +1062,14 @@ impl IndexerMetrics { let _ = self.queue_lag_height.try_set(lag); } + pub(crate) fn certificate_request_started(&self) { + self.producer_certificate_requests.inc(); + } + + pub(crate) fn certificate_request_finished(&self) { + self.producer_certificate_requests.dec(); + } + pub(crate) fn shared_cache_inserted(&self, source: SharedCacheSource) { self.shared_cache_inserted .get_or_create(&SharedCacheSourceLabel { source }) @@ -1132,22 +1121,6 @@ impl IndexerMetrics { .observe(duration.as_secs_f64()); } - pub(crate) fn backfill_queue_reset( - &self, - reason: BackfillResetReason, - abandoned_entries: u64, - abandoned_height_span: u64, - ) { - let label = BackfillResetReasonLabel { reason }; - self.backfill_queue_reset.get_or_create(&label).inc(); - self.backfill_queue_reset_abandoned_entries - .get_or_create(&label) - .observe(abandoned_entries as f64); - self.backfill_queue_reset_abandoned_height_span - .get_or_create(&label) - .observe(abandoned_height_span as f64); - } - pub(crate) fn start_backfill_body(&self, bytes: u64) -> BackfillBodyGuard { let bytes = gauge_u64(bytes); self.backfill_active_body_estimated_bytes.inc_by(bytes); @@ -1403,9 +1376,6 @@ impl BackfillUploadGuard { self.status = Some(BackfillStatus::Skipped); } - pub(crate) const fn abandoned(&mut self) { - self.status = Some(BackfillStatus::Abandoned); - } } impl Drop for BackfillUploadGuard { diff --git a/examples/coins/chain/src/indexer/mod.rs b/examples/coins/chain/src/indexer/mod.rs index 2604221..fac82a2 100644 --- a/examples/coins/chain/src/indexer/mod.rs +++ b/examples/coins/chain/src/indexer/mod.rs @@ -6,9 +6,9 @@ use commonware_consensus::types::Epoch; use commonware_cryptography::bls12381::{ dkg::feldman_desmedt::Output as DkgOutput, primitives::variant::MinSig, }; -use commonware_runtime::{BufferPooler, Clock, Metrics, Spawner, Storage}; +use commonware_runtime::{BufferPooler, Clock, Handle, Metrics, Spawner, Storage}; use commonware_storage::queue; -use commonware_utils::sync::Mutex; +use commonware_utils::{sync::Mutex, NZU64}; use std::{future::Future, num::NonZeroUsize, sync::Arc, time::Duration}; use thiserror::Error; @@ -21,7 +21,7 @@ use backfiller::{SharedState, State}; pub(crate) use metrics::{DkgUploadStatus, IndexerMetrics}; #[cfg(test)] pub(crate) use metrics::{ - BackfillDecision, BackfillPhase, BackfillResetReason, BackfillWaitReason, BlockMetricSource, + BackfillDecision, BackfillPhase, BackfillWaitReason, BlockMetricSource, HttpArtifact, LiveUploadArtifact, ProducerActivity, ProducerStatus, QueueReadSource, QueueStatus, SharedCacheSource, SharedRetentionReason, SharedStateSnapshot, }; @@ -32,6 +32,64 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); const MISSING_FINALIZATION_GRACE: Duration = Duration::from_secs(120); const MISMATCHED_FINALIZATION_GRACE: Duration = Duration::from_secs(15); +pub(crate) const SPOOL_ITEMS_PER_SECTION: std::num::NonZeroU64 = NZU64!(128); + +/// Availability window for durable finalized uploads while the external indexer is unavailable. +#[derive(Clone, Copy, Debug)] +pub struct SpoolLimits { + pub max_entries: u64, + pub max_bytes: u64, + pub max_payload_bytes: u64, + pub max_age: Duration, +} + +impl Default for SpoolLimits { + fn default() -> Self { + Self { + max_entries: crate::BLOCKS_PER_EPOCH.get(), + max_bytes: 2 * 1024 * 1024 * 1024, + max_payload_bytes: 16 * 1024 * 1024, + max_age: Duration::from_secs(24 * 60 * 60), + } + } +} + +impl SpoolLimits { + fn validate(self) { + assert!(self.max_entries > 0, "indexer spool max_entries must be non-zero"); + assert!(self.max_bytes > 0, "indexer spool max_bytes must be non-zero"); + assert!( + self.max_payload_bytes > 0 && self.max_payload_bytes <= self.max_bytes, + "indexer spool max_payload_bytes must be in 1..=max_bytes" + ); + assert!(!self.max_age.is_zero(), "indexer spool max_age must be non-zero"); + self.max_encoded_payload_bytes() + .expect("indexer spool encoded payload bound overflows u64"); + } + + pub fn max_encoded_payload_bytes(self) -> Option { + (SPOOL_ITEMS_PER_SECTION.get() - 1) + .checked_mul(self.max_payload_bytes)? + .checked_add(self.max_bytes) + } +} + +#[cfg(test)] +mod spool_tests { + use super::*; + + #[test] + fn default_spool_bound_includes_section_slack() { + let limits = SpoolLimits::default(); + assert_eq!(limits.max_entries, crate::BLOCKS_PER_EPOCH.get()); + assert_eq!(limits.max_bytes, 2 * 1024 * 1024 * 1024); + assert_eq!(limits.max_age, Duration::from_secs(24 * 60 * 60)); + assert_eq!( + limits.max_encoded_payload_bytes(), + Some(limits.max_bytes + 127 * limits.max_payload_bytes) + ); + } +} /// Errors returned by the HTTP indexer client. #[derive(Debug, Error)] @@ -184,11 +242,13 @@ pub(crate) struct Config { pub(crate) mailbox_size: NonZeroUsize, pub(crate) backfiller_max_active: NonZeroUsize, pub(crate) backfiller_retry: Duration, + pub(crate) spool_limits: SpoolLimits, pub(crate) metrics: IndexerMetrics, } pub(crate) struct Indexer { producer: Producer, + producer_handle: Handle<()>, pusher: Pusher, consumer: Consumer, } @@ -205,8 +265,10 @@ impl Indexer Indexer (Producer, Pusher, Consumer) { + pub(crate) fn split(self) -> (Producer, Handle<()>, Pusher, Consumer) { let Self { producer, + producer_handle, pusher, consumer, } = self; - (producer, pusher, consumer) + (producer, producer_handle, pusher, consumer) } } diff --git a/examples/coins/chain/src/lib.rs b/examples/coins/chain/src/lib.rs index b034460..359f7f8 100644 --- a/examples/coins/chain/src/lib.rs +++ b/examples/coins/chain/src/lib.rs @@ -19,6 +19,7 @@ pub mod application; pub mod engine; pub mod execution; pub mod genesis; +pub(crate) mod history; pub mod indexer; pub mod rpc; pub mod runtime; diff --git a/examples/coins/chain/src/testnet.rs b/examples/coins/chain/src/testnet.rs index 8fc8049..9d51d91 100644 --- a/examples/coins/chain/src/testnet.rs +++ b/examples/coins/chain/src/testnet.rs @@ -48,7 +48,6 @@ use std::{ }; use tracing::{info, warn, Level}; -const FREEZER_TABLE_INITIAL_SIZE: u32 = 2u32.pow(14); const DEFAULT_MAX_BLOCK_TRANSACTIONS: usize = 4_096; const DEFAULT_MAX_MESSAGE_SIZE: u32 = 1024 * 1024; const DEFAULT_CHANNEL_BACKLOG: usize = 1024; @@ -137,6 +136,18 @@ pub struct NodeConfig { pub genesis_path: Option, #[serde(default)] pub indexer_url: Option, + /// Maximum self-contained finalized payloads retained while the indexer is unavailable. + #[serde(default = "default_indexer_spool_max_entries")] + pub indexer_spool_max_entries: u64, + /// Maximum logical encoded payload bytes retained by the indexer spool. + #[serde(default = "default_indexer_spool_max_bytes")] + pub indexer_spool_max_bytes: u64, + /// Maximum encoded size accepted for one finalized payload. + #[serde(default = "default_indexer_spool_max_payload_bytes")] + pub indexer_spool_max_payload_bytes: u64, + /// Maximum payload age before visible terminal expiry. + #[serde(default = "default_indexer_spool_max_age_seconds")] + pub indexer_spool_max_age_seconds: u64, pub consensus: ConsensusConfig, pub networking: NetworkConfig, /// Enable one-time peer QMDB state sync for a fresh joining node. @@ -157,6 +168,22 @@ impl NodeConfig { } } +fn default_indexer_spool_max_entries() -> u64 { + indexer::SpoolLimits::default().max_entries +} + +fn default_indexer_spool_max_bytes() -> u64 { + indexer::SpoolLimits::default().max_bytes +} + +fn default_indexer_spool_max_payload_bytes() -> u64 { + indexer::SpoolLimits::default().max_payload_bytes +} + +fn default_indexer_spool_max_age_seconds() -> u64 { + indexer::SpoolLimits::default().max_age.as_secs() +} + #[derive(Clone, Debug, Serialize, Deserialize)] pub struct BootstrapperConfig { pub public_key: String, @@ -390,6 +417,10 @@ pub fn generate_local_testnet(config: LocalTestnetConfig) -> Result> = ContinueOnUpdate::boxed(); diff --git a/examples/coins/chain/src/tests/indexer.rs b/examples/coins/chain/src/tests/indexer.rs index b26e49e..d26399b 100644 --- a/examples/coins/chain/src/tests/indexer.rs +++ b/examples/coins/chain/src/tests/indexer.rs @@ -1,6 +1,6 @@ use crate::{ indexer::{ - BackfillDecision, BackfillPhase, BackfillResetReason, BackfillWaitReason, + BackfillDecision, BackfillPhase, BackfillWaitReason, BlockMetricSource, DkgUploadStatus, HttpArtifact, IndexerMetrics, LiveUploadArtifact, ProducerActivity, ProducerStatus, QueueReadSource, QueueStatus, SharedCacheSource, SharedRetentionReason, @@ -346,10 +346,13 @@ fn shared_state_metrics_use_expected_labels() { uploaded_digests: 4, latest_finalized_height: 8, acked_through_height: 6, + spool_entries: 5, + spool_logical_bytes: 512, + spool_oldest_height: 2, + spool_oldest_enqueued_at_millis: 1_000, }); metrics.shared_cache_inserted(SharedCacheSource::ProducerRecord); metrics.shared_cache_inserted(SharedCacheSource::LiveCertificate); - metrics.shared_cache_inserted(SharedCacheSource::ConsumerMarshal); metrics.shared_cache_removed(SharedRetentionReason::Uploaded); metrics.shared_cache_removed(SharedRetentionReason::CertificateFinished); metrics.shared_cache_removed(SharedRetentionReason::Pruned); @@ -373,9 +376,6 @@ fn shared_state_metrics_use_expected_labels() { assert!(encoded.contains( "indexer_shared_cache_insert_total{source=\"live_certificate\"} 1", )); - assert!(encoded.contains( - "indexer_shared_cache_insert_total{source=\"consumer_marshal\"} 1", - )); assert!(encoded.contains( "indexer_shared_cache_remove_total{reason=\"uploaded\"} 1", )); @@ -402,7 +402,6 @@ fn block_metrics_use_expected_sources() { metrics.observe_block(BlockMetricSource::ProducerRecord, &block); metrics.observe_block(BlockMetricSource::LiveCertificate, &block); metrics.observe_block(BlockMetricSource::ConsumerCached, &block); - metrics.observe_block(BlockMetricSource::ConsumerMarshal, &block); let encoded = context.encode(); assert!(encoded.contains("indexer_block_estimated_bytes_bucket")); @@ -410,7 +409,6 @@ fn block_metrics_use_expected_sources() { assert!(encoded.contains("source=\"producer_record\"")); assert!(encoded.contains("source=\"live_certificate\"")); assert!(encoded.contains("source=\"consumer_cached\"")); - assert!(encoded.contains("source=\"consumer_marshal\"")); }); } @@ -433,20 +431,11 @@ fn backfill_metrics_use_expected_labels() { upload.skipped(); } metrics.backfill_decision(BackfillDecision::Skip, BackfillPhase::Start); - metrics.backfill_decision(BackfillDecision::Wait, BackfillPhase::BeforeBlock); metrics.backfill_decision(BackfillDecision::Proceed, BackfillPhase::BeforeAttempt); metrics.backfill_retry(BackfillWaitReason::CertificateUpload); - metrics.backfill_retry(BackfillWaitReason::MissingBlock); - metrics.backfill_retry(BackfillWaitReason::MissingFinalization); - metrics.backfill_retry(BackfillWaitReason::MismatchedFinalization); metrics.backfill_retry(BackfillWaitReason::HttpError); metrics.backfill_waited(BackfillWaitReason::CertificateUpload, Duration::from_millis(1)); - metrics.backfill_waited(BackfillWaitReason::MissingBlock, Duration::from_millis(1)); - metrics.backfill_waited(BackfillWaitReason::MissingFinalization, Duration::from_millis(1)); - metrics.backfill_waited(BackfillWaitReason::MismatchedFinalization, Duration::from_millis(1)); metrics.backfill_waited(BackfillWaitReason::HttpError, Duration::from_millis(1)); - metrics.backfill_queue_reset(BackfillResetReason::MissingFinalization, 32, 31); - metrics.backfill_queue_reset(BackfillResetReason::MismatchedFinalization, 2, 1); metrics.queue_enqueued(QueueStatus::Success); metrics.queue_enqueued(QueueStatus::Failure); metrics.queue_acked(QueueStatus::Success); @@ -486,9 +475,6 @@ fn backfill_metrics_use_expected_labels() { assert!(encoded.contains( "indexer_backfill_decision_total{decision=\"skip\",phase=\"start\"} 1", )); - assert!(encoded.contains( - "indexer_backfill_decision_total{decision=\"wait\",phase=\"before_block\"} 1", - )); assert!(encoded.contains( "indexer_backfill_decision_total{decision=\"proceed\",phase=\"before_attempt\"} 1", )); @@ -496,20 +482,7 @@ fn backfill_metrics_use_expected_labels() { assert!(encoded.contains( "indexer_backfill_retry_total{reason=\"certificate_upload\"} 1", )); - assert!(encoded.contains("indexer_backfill_retry_total{reason=\"missing_block\"} 1")); - assert!(encoded - .contains("indexer_backfill_retry_total{reason=\"missing_finalization\"} 1")); - assert!(encoded - .contains("indexer_backfill_retry_total{reason=\"mismatched_finalization\"} 1")); assert!(encoded.contains("indexer_backfill_retry_total{reason=\"http_error\"} 1")); - assert!(encoded.contains( - "indexer_backfill_queue_reset_total{reason=\"missing_finalization\"} 1", - )); - assert!(encoded.contains( - "indexer_backfill_queue_reset_total{reason=\"mismatched_finalization\"} 1", - )); - assert!(encoded.contains("indexer_backfill_queue_reset_abandoned_entries_bucket")); - assert!(encoded.contains("indexer_backfill_queue_reset_abandoned_height_span_bucket")); assert!(encoded.contains("indexer_backfill_active_block_estimated_bytes 0")); assert!(encoded.contains("indexer_backfill_active_body_estimated_bytes 0")); assert!(encoded.contains("indexer_queue_enqueue_total{status=\"success\"} 1")); diff --git a/examples/coins/chain/tests/common/network.rs b/examples/coins/chain/tests/common/network.rs index 04495cb..843caa3 100644 --- a/examples/coins/chain/tests/common/network.rs +++ b/examples/coins/chain/tests/common/network.rs @@ -38,7 +38,6 @@ use std::{ time::Duration, }; -const FREEZER_TABLE_INITIAL_SIZE: u32 = 2u32.pow(14); // 1MB const TEST_QUOTA: Quota = Quota::per_second(NZU32!(u32::MAX)); const MAX_BLOCK_TRANSACTIONS: usize = 256; @@ -512,8 +511,6 @@ async fn start_validator( blocker: oracle.control(public_key.clone()), manager: oracle.manager(), partition_prefix: uid.clone(), - blocks_freezer_table_initial_size: FREEZER_TABLE_INITIAL_SIZE, - finalized_freezer_table_initial_size: FREEZER_TABLE_INITIAL_SIZE, signer: signer.clone(), dkg_storage_key: [9u8; 32], output, @@ -527,6 +524,7 @@ async fn start_validator( pool_config: PoolConfig::default(), genesis: None, indexer: None, + indexer_spool_limits: Default::default(), }; let validator_context = context.child("validator").with_attribute("id", &uid); From fe267437f8c2c59b6baf9a1556191f7cfcc32a0d Mon Sep 17 00:00:00 2001 From: Tuan Pham Anh Date: Wed, 22 Jul 2026 19:48:16 +0700 Subject: [PATCH 3/4] fix(dkg): handle closed consensus tasks at epoch rollover Treat Closed as an expected result when retiring an aborted consensus engine while preserving fatal handling for genuine task failures. Make the coins engine epoch length internally configurable and add a deterministic regression test covering epoch transition, partition cleanup, continued finalization, and clean shutdown. --- dkg/src/orchestrator/actor.rs | 5 +- examples/coins/chain/src/engine.rs | 15 ++-- examples/coins/chain/src/testnet.rs | 3 +- examples/coins/chain/tests/coins.rs | 54 ++++++++++++- examples/coins/chain/tests/common/network.rs | 85 +++++++++++++++++++- 5 files changed, 149 insertions(+), 13 deletions(-) diff --git a/dkg/src/orchestrator/actor.rs b/dkg/src/orchestrator/actor.rs index 5a250d0..060b701 100644 --- a/dkg/src/orchestrator/actor.rs +++ b/dkg/src/orchestrator/actor.rs @@ -452,8 +452,11 @@ where continue; }; handle.abort(); + // Spawned task handles close their completion channel when aborted. match handle.await { - Ok(()) | Err(commonware_runtime::Error::Aborted) => {} + Ok(()) + | Err(commonware_runtime::Error::Aborted) + | Err(commonware_runtime::Error::Closed) => {} Err(error) => { panic!("consensus engine for epoch {epoch} failed while stopping: {error}") } diff --git a/examples/coins/chain/src/engine.rs b/examples/coins/chain/src/engine.rs index 3247e10..2930cbf 100644 --- a/examples/coins/chain/src/engine.rs +++ b/examples/coins/chain/src/engine.rs @@ -3,10 +3,7 @@ use crate::execution::NodeHandle; use crate::genesis::{genesis_target, state_commitment, ChainGenesis}; use crate::history::{self, BlocksArchive, FinalizationsArchive}; use crate::indexer; -use crate::{ - Block, EpochProvider, Provider, PublicKey, Scheme, Transaction, BLOCKS_PER_EPOCH, - NAMESPACE, -}; +use crate::{Block, EpochProvider, Provider, PublicKey, Scheme, Transaction, NAMESPACE}; use commonware_broadcast::buffered; use commonware_consensus::{ marshal::{ @@ -57,6 +54,7 @@ use nunchi_mempool::{Mempool, PoolConfig}; use rand::{CryptoRng, Rng}; use std::{ marker::PhantomData, + num::NonZeroU64, sync::Arc, time::{Duration, Instant}, }; @@ -85,6 +83,7 @@ pub struct Config, P: Manager, pub share: Option, pub peer_config: PeerConfig, + pub epoch_length: NonZeroU64, pub leader_timeout: Duration, pub certification_timeout: Duration, pub strategy: S, @@ -228,7 +227,7 @@ where max_supported_mode: MAX_SUPPORTED_MODE, namespace: NAMESPACE.to_vec(), storage_protector: dkg::StorageProtector::new(config.dkg_storage_key), - epoch_length: BLOCKS_PER_EPOCH, + epoch_length: config.epoch_length, }, ); @@ -459,7 +458,7 @@ where finalized_blocks, marshal::Config { provider: provider.clone(), - epocher: FixedEpocher::new(BLOCKS_PER_EPOCH), + epocher: FixedEpocher::new(config.epoch_length), start: marshal_start, partition_prefix: format!("{}_marshal", config.partition_prefix), mailbox_size: MAILBOX_SIZE, @@ -513,7 +512,7 @@ where APPLICATION_VERIFY_CONCURRENCY, ), marshal_mailbox.clone(), - FixedEpocher::new(BLOCKS_PER_EPOCH), + FixedEpocher::new(config.epoch_length), )); let (indexer_producer, indexer_producer_handle, indexer_pusher, indexer_consumer) = @@ -576,7 +575,7 @@ where muxer_size: MAILBOX_SIZE.get(), mailbox_size: MAILBOX_SIZE, partition_prefix: format!("{}_consensus", config.partition_prefix), - epoch_length: BLOCKS_PER_EPOCH, + epoch_length: config.epoch_length, genesis_digest, recovered_floor, _phantom: PhantomData, diff --git a/examples/coins/chain/src/testnet.rs b/examples/coins/chain/src/testnet.rs index 9d51d91..feaf62a 100644 --- a/examples/coins/chain/src/testnet.rs +++ b/examples/coins/chain/src/testnet.rs @@ -10,7 +10,7 @@ use crate::{ channels, engine::{Config as EngineConfig, Engine}, genesis::{ChainGenesis, GenesisError}, - indexer, rpc, PublicKey, NAMESPACE, + indexer, rpc, PublicKey, BLOCKS_PER_EPOCH, NAMESPACE, }; use commonware_codec::{Decode, DecodeExt, Encode, EncodeSize}; use commonware_consensus::marshal; @@ -640,6 +640,7 @@ async fn start_node( output, share: Some(share), peer_config: config.peer_config.clone(), + epoch_length: BLOCKS_PER_EPOCH, leader_timeout: Duration::from_millis(config.consensus.leader_timeout_ms), certification_timeout: Duration::from_millis(config.consensus.certification_timeout_ms), strategy: context diff --git a/examples/coins/chain/tests/coins.rs b/examples/coins/chain/tests/coins.rs index 9051289..da9ec09 100644 --- a/examples/coins/chain/tests/coins.rs +++ b/examples/coins/chain/tests/coins.rs @@ -8,7 +8,8 @@ use commonware_cryptography::Signer as _; use commonware_cryptography::{Hasher, Sha256}; use commonware_macros::{select, test_traced}; use commonware_p2p::simulated::Link; -use commonware_runtime::{deterministic, Clock, Runner as _}; +use commonware_runtime::{deterministic, Clock, Runner as _, Spawner as _, Supervisor as _}; +use commonware_utils::NZU64; use nunchi_authority::{ proposal_id, AuthorityOperation, MultisigPolicy, RegistryChange, Transaction as AuthorityTransaction, @@ -104,6 +105,56 @@ fn reaches_height_100() { }); } +#[test_traced] +fn crosses_epoch_boundary_and_reclaims_retired_partition() { + with_large_stack(|| { + let executor = deterministic::Runner::timed(Duration::from_secs(60)); + executor.start(|mut context| async move { + let cfg = ValidatorConfig { + epoch_length: NZU64!(20), + ..ValidatorConfig::default() + }; + let mut network = TestNetworkBuilder::new(VALIDATORS) + .with_initial_link(reliable_link()) + .with_validator_config(cfg) + .build(&mut context) + .await; + network.start_all().await; + + let validator_task_prefix = "validator"; + assert!(network.running_tasks(validator_task_prefix) > 0); + + network.run_until_height(25).await; + + network.assert_validator_metric("engine_orchestrator_latest_epoch", &[], 1); + network.assert_validator_metric( + "engine_orchestrator_consensus_partitions_active", + &[], + 1, + ); + network.assert_validator_metric( + "engine_orchestrator_consensus_partition_cleanup_watermark", + &[], + 1, + ); + network.assert_validator_metric( + "engine_orchestrator_consensus_partition_cleanup_total", + &[("status", "removed")], + 1, + ); + network.assert_consensus_partition_missing(0).await; + assert!(network.running_tasks(validator_task_prefix) > 0); + + let shutdown = network.context().child("shutdown"); + shutdown + .stop(0, Some(Duration::from_secs(10))) + .await + .expect("validator tasks should stop cleanly"); + assert_eq!(network.running_tasks(validator_task_prefix), 0); + }); + }); +} + #[test_traced] fn state_syncs_late_validator() { with_large_stack(|| { @@ -154,6 +205,7 @@ fn recovers_unclean_shutdown() { let cfg = ValidatorConfig { leader_timeout: Duration::from_millis(250), certification_timeout: Duration::from_millis(500), + ..ValidatorConfig::default() }; let wait = diff --git a/examples/coins/chain/tests/common/network.rs b/examples/coins/chain/tests/common/network.rs index 843caa3..21d473f 100644 --- a/examples/coins/chain/tests/common/network.rs +++ b/examples/coins/chain/tests/common/network.rs @@ -15,7 +15,7 @@ use commonware_p2p::{ use commonware_parallel::Sequential; use commonware_runtime::{ deterministic::{self, Runner}, - Clock, Metrics, Runner as _, Supervisor, + Clock, Error as RuntimeError, Metrics, Runner as _, Storage, Supervisor, }; use commonware_utils::{ ordered::{Map, Set}, @@ -27,7 +27,7 @@ use nunchi_coins::{Address, Ledger}; use nunchi_coins_chain::{ engine::{Config, Engine}, execution::NodeHandle, - PublicKey, Transaction, + PublicKey, Transaction, BLOCKS_PER_EPOCH, }; use nunchi_common::QmdbReader; use nunchi_dkg::{ContinueOnUpdate, PeerConfig}; @@ -35,6 +35,7 @@ use nunchi_mempool::{MempoolHandle, PoolConfig}; use nunchi_oracle::OracleLedger; use std::{ collections::{HashMap, HashSet}, + num::NonZeroU64, time::Duration, }; @@ -108,6 +109,7 @@ pub(crate) fn lossy_link() -> Link { #[derive(Clone)] pub(crate) struct ValidatorConfig { + pub(crate) epoch_length: NonZeroU64, pub(crate) leader_timeout: Duration, pub(crate) certification_timeout: Duration, } @@ -115,6 +117,7 @@ pub(crate) struct ValidatorConfig { impl Default for ValidatorConfig { fn default() -> Self { Self { + epoch_length: BLOCKS_PER_EPOCH, leader_timeout: Duration::from_secs(1), certification_timeout: Duration::from_secs(2), } @@ -355,6 +358,83 @@ impl TestNetwork<'_> { } } + pub(crate) fn assert_validator_metric( + &self, + suffix: &str, + required_labels: &[(&str, &str)], + expected_value: u64, + ) { + let expected = self.started_validator_ids(); + let mut observed = HashSet::new(); + let metrics = self.context.encode(); + for line in metrics.lines() { + let Some((metric, labels, value)) = validator_metric_sample(line) else { + continue; + }; + if !metric.ends_with(suffix) + || !required_labels + .iter() + .all(|(name, expected)| metric_label(labels, name) == Some(*expected)) + { + continue; + } + let Some(id) = metric_label(labels, "id") else { + continue; + }; + if !expected.contains(id) { + continue; + } + assert_eq!( + value.parse::().unwrap(), + expected_value, + "unexpected {suffix} value for {id}: {line}", + ); + assert!( + observed.insert(id.to_string()), + "duplicate {suffix} for {id}" + ); + } + assert_eq!( + observed, expected, + "missing {suffix} samples in metrics: {metrics}", + ); + } + + pub(crate) async fn assert_consensus_partition_missing(&self, epoch: u64) { + for signer in &self.private_keys { + let public_key = signer.public_key(); + if self.registrations.contains_key(&public_key) { + continue; + } + let partition = format!("validator_{public_key}_consensus_consensus_{epoch}"); + let result = self.context.scan(&partition).await; + assert!( + matches!( + result, + Err(RuntimeError::PartitionMissing(ref missing)) if missing == &partition + ), + "retired consensus partition {partition} still exists: {result:?}", + ); + } + } + + pub(crate) fn running_tasks(&self, prefix: &str) -> usize { + self.context + .encode() + .lines() + .filter_map(|line| { + if !line.starts_with("runtime_tasks_running{") { + return None; + } + let name = line.split("name=\"").nth(1)?.split('"').next()?; + if !name.starts_with(prefix) { + return None; + } + line.rsplit(' ').next()?.parse::().ok() + }) + .sum() + } + fn started_validator_ids(&self) -> HashSet { self.private_keys .iter() @@ -516,6 +596,7 @@ async fn start_validator( output, share: Some(share), peer_config, + epoch_length: cfg.epoch_length, leader_timeout: cfg.leader_timeout, certification_timeout: cfg.certification_timeout, strategy: Sequential, From 0e95fef07895ecdc7bd2ab7b8d59dee97458e5e8 Mon Sep 17 00:00:00 2001 From: Tuan Pham Anh Date: Thu, 23 Jul 2026 03:03:30 +0700 Subject: [PATCH 4/4] fix clippy --- examples/coins/chain/src/history.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/coins/chain/src/history.rs b/examples/coins/chain/src/history.rs index 402af05..6d39607 100644 --- a/examples/coins/chain/src/history.rs +++ b/examples/coins/chain/src/history.rs @@ -159,7 +159,7 @@ where let mut archive_existed = [false; 4]; for (exists, partition) in archive_existed .iter_mut() - .zip(partitions.archives().into_iter()) + .zip(partitions.archives()) { *exists = partition_exists(context, partition).await?; }