From dcf78e17f11174d406e1506dbd3c6f0a0fabc3fd Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Tue, 12 May 2026 17:43:25 +0400 Subject: [PATCH] feat: add nucleus engine crate --- Cargo.toml | 17 ++- nucleus/Cargo.toml | 85 +++++++++++++++ nucleus/README.md | 34 ++++++ nucleus/src/heed.rs | 52 ++++++++++ nucleus/src/ledger.rs | 26 +++++ nucleus/src/lib.rs | 32 ++++++ nucleus/src/metrics.rs | 138 +++++++++++++++++++++++++ nucleus/src/runtime.rs | 127 +++++++++++++++++++++++ nucleus/src/shutdown.rs | 223 ++++++++++++++++++++++++++++++++++++++++ nucleus/src/testkit.rs | 88 ++++++++++++++++ nucleus/src/tls.rs | 46 +++++++++ 11 files changed, 867 insertions(+), 1 deletion(-) create mode 100644 nucleus/Cargo.toml create mode 100644 nucleus/README.md create mode 100644 nucleus/src/heed.rs create mode 100644 nucleus/src/ledger.rs create mode 100644 nucleus/src/lib.rs create mode 100644 nucleus/src/metrics.rs create mode 100644 nucleus/src/runtime.rs create mode 100644 nucleus/src/shutdown.rs create mode 100644 nucleus/src/testkit.rs create mode 100644 nucleus/src/tls.rs diff --git a/Cargo.toml b/Cargo.toml index ad90967..a4b4c44 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,15 @@ [workspace] -members = ["solana/account", "solana/program-runtime", "solana/svm", "solana/transaction-context"] +members = [ + "nucleus", + "programs/magic-root-interface", + "programs/magic-root-program", + "programs/v42-calculator-interface", + "programs/v42-calculator-program", + "solana/account", + "solana/program-runtime", + "solana/svm", + "solana/transaction-context" +] resolver = "3" [workspace.package] @@ -14,6 +24,7 @@ version = "0.1.0" [workspace.dependencies] magic-root-interface = { path = "programs/magic-root-interface" } magic-root-program = { path = "programs/magic-root-program" } +nucleus = { path = "nucleus", package = "magicblock-engine-nucleus" } solana-account = { path = "solana/account" } solana-program-runtime = { path = "solana/program-runtime" } solana-svm = { path = "solana/svm" } @@ -30,6 +41,7 @@ cfg-if = "1.0.4" criterion = "0.7.0" derive_more = "2.1.1" env_logger = "0.11.8" +futures = { version = "0.3.32", default-features = false } itertools = "0.14.0" log = "0.4.29" qualifier_attr = "0.2.2" @@ -40,6 +52,9 @@ serde_bytes = "0.11.19" tar = "0.4.45" tempfile = "3" thiserror = "2.0.17" +tokio = "1.52.1" +tokio-util = "0.7.18" +tracing = "0.1.44" tracing-subscriber = { version = "0.3.23", features = ["env-filter", "fmt"] } wincode = "0.5.1" zstd = { version = "0.13.3", default-features = false } diff --git a/nucleus/Cargo.toml b/nucleus/Cargo.toml new file mode 100644 index 0000000..c197256 --- /dev/null +++ b/nucleus/Cargo.toml @@ -0,0 +1,85 @@ +[package] +name = "magicblock-engine-nucleus" + +authors.workspace = true +edition.workspace = true +homepage.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +version.workspace = true + +[lib] +name = "nucleus" + +[features] +default = [] +heed = ["dep:heed"] +ledger-schema = [ + "dep:solana-hash", + "dep:wincode", + "solana-hash/copy", + "solana-hash/wincode", + "wincode/derive" +] +metrics = ["dep:prometheus", "dep:tracing"] +notifier = ["dep:tokio"] +runtime = [ + "agave-transaction-view/agave-unstable-api", + "dep:agave-transaction-view", + "dep:derive_more", + "dep:oneshot", + "dep:solana-svm", + "dep:solana-transaction-error", + "dep:tokio", + "dep:wincode", + "ledger-schema", + "tokio/sync", + "wincode/derive" +] +shutdown = ["dep:futures", "dep:oneshot", "dep:tokio", "dep:tokio-util", "dep:tracing"] +testkit = [ + "dep:solana-instruction", + "dep:solana-keypair", + "dep:solana-message", + "dep:solana-pubkey", + "dep:solana-signature", + "dep:solana-signer", + "dep:solana-transaction", + "dep:tempfile", + "dep:tracing-subscriber", + "dep:v42-calculator-interface", + "ledger-schema", + "solana-transaction/wincode" +] +tls = ["dep:solana-instruction-error", "dep:solana-pubkey", "dep:wincode"] + +[dependencies] +derive_more = { workspace = true, optional = true, features = ["deref", "from"] } +futures = { workspace = true, features = ["alloc"], optional = true } +heed = { workspace = true, optional = true } +oneshot = { workspace = true, features = ["async"], optional = true } +prometheus = { workspace = true, optional = true } +solana-hash = { workspace = true, optional = true } +solana-instruction-error = { workspace = true, optional = true } +tempfile = { workspace = true, optional = true } +tokio = { workspace = true, features = ["macros", "signal", "time"], optional = true } +tokio-util = { workspace = true, optional = true } +tracing = { workspace = true, optional = true } +tracing-subscriber = { workspace = true, optional = true } +wincode = { workspace = true, optional = true } + +agave-transaction-view = { workspace = true, optional = true } +solana-instruction = { workspace = true, optional = true } +solana-keypair = { workspace = true, optional = true } +solana-message = { workspace = true, optional = true } +solana-pubkey = { workspace = true, optional = true } +solana-signature = { workspace = true, optional = true } +solana-signer = { workspace = true, optional = true } +solana-svm = { workspace = true, optional = true } +solana-transaction = { workspace = true, optional = true } +solana-transaction-error = { workspace = true, optional = true } +v42-calculator-interface = { workspace = true, optional = true } + +[lints] +workspace = true diff --git a/nucleus/README.md b/nucleus/README.md new file mode 100644 index 0000000..49a39b4 --- /dev/null +++ b/nucleus/README.md @@ -0,0 +1,34 @@ +# `magicblock-engine-nucleus` + +The bottom of the stack: the small set of types and helpers that every other +engine crate shares. Its whole job is to be *common vocabulary* — the things you'd +otherwise be tempted to redefine in three places — without smuggling in any +engine policy. +Because it sits under everything, it is built to stay light. Everything is behind +a feature flag, so a dependent pulls in exactly the pieces it uses and nothing +more. + +## Feature flags + +- **`heed`** — LMDB transaction aliases and the `DatabaseIndex` trait, the shared + shape for the on-disk indexes used by `accountsdb` and `ledger`. +- **`shutdown`** — ordered cooperative cancellation and termination reporting, + so background services can be stopped by priority tier and report why they did. +- **`notifier`** — `EventNotifier`, a one-shot async latch. Once fired it stays + fired: late waiters return immediately and it never resets. +- **`ledger-schema`** — the block-boundary types (`Block`, `SuperblockSeal`) shared + by the storage-adjacent crates that read and write them. +- **`metrics`** — small Prometheus helpers (metric specs, integer collector + constructors, default-registry registration, late-initialized access, operation + labels, and duration/count timers). The helpers are shared; metric names and + registration ownership stay in each crate. +- **`runtime`** — the runtime-facing shared types: sanitized transaction-view + aliases, execution I/O, the sequencer messages and the quiescence `barrier`. +- **`tls`** — thread-local execution state used by runtime-adjacent code: + MagicRoot authority publication and the encoded message queue. +- **`testkit`** — engine-agnostic test fixtures (fixture accounts, wincode + transactions, block boundaries, throwaway directories, and libtest-aware + tracing setup) shared across crate test suites. Compiled as normal code but + only under this feature, so it never reaches release builds; consumers enable + it as a dev-dependency. Keeper-level harness code lives in `keeper::testkit`, + not here. diff --git a/nucleus/src/heed.rs b/nucleus/src/heed.rs new file mode 100644 index 0000000..ae03f6f --- /dev/null +++ b/nucleus/src/heed.rs @@ -0,0 +1,52 @@ +//! Shared heed index plumbing. + +use std::mem; + +use ::heed::{Env, Result, RoTxn, RwTxn, WithTls}; + +/// Read-only transaction using heed thread-local storage. +pub type RoTxnTls<'e> = RoTxn<'e, WithTls>; +/// Optional write transaction used by batched updates. +pub type OptRwTxn<'t, 'e> = &'t mut Option>; +/// Optional read transaction used by batched reads. +pub type OptRoTxn<'t, 'e> = &'t mut Option>; + +/// Common transaction access for heed-backed indexes. +/// +/// # Safety +/// +/// Implementors must return an environment owned by `self`, and callers must +/// drop transactions opened through this trait before that environment is +/// dropped. This matches the batched transaction pattern where the transaction +/// is stored in the caller's `Option` and reused across index operations. +pub unsafe trait DatabaseIndex { + /// Returns the owning heed environment. + fn env(&self) -> &Env; + + /// Uses the supplied write transaction or opens one on demand. + fn write_txn<'t, 'e>(&self, txn: OptRwTxn<'t, 'e>) -> Result<&'t mut RwTxn<'e>> { + if let Some(txn) = txn { + return Ok(txn); + } + // SAFETY: guaranteed by the trait contract. The transaction is stored + // in the caller-owned option and must be dropped before `env`. + let write = unsafe { mem::transmute::, RwTxn<'e>>(self.env().write_txn()?) }; + Ok(txn.insert(write)) + } + + /// Uses the supplied read transaction or opens one on demand. + fn read_txn<'t, 'e>(&self, txn: OptRoTxn<'t, 'e>) -> Result<&'t RoTxnTls<'e>> { + if let Some(txn) = txn { + return Ok(txn); + } + // SAFETY: guaranteed by the trait contract. The transaction is stored + // in the caller-owned option and must be dropped before `env`. + let read = unsafe { mem::transmute::, RoTxnTls<'e>>(self.env().read_txn()?) }; + Ok(txn.insert(read)) + } + + /// Flushes the index databases to durable storage. + fn flush(&self) -> Result<()> { + self.env().force_sync() + } +} diff --git a/nucleus/src/ledger.rs b/nucleus/src/ledger.rs new file mode 100644 index 0000000..bf9a78f --- /dev/null +++ b/nucleus/src/ledger.rs @@ -0,0 +1,26 @@ +//! Ledger block-boundary schema shared by storage-adjacent crates. + +use solana_hash::Hash; +use wincode::{SchemaRead, SchemaWrite}; + +use crate::Slot; + +/// Block boundary entry stored after all transactions in the block. +#[derive(SchemaRead, SchemaWrite, Clone, Copy, Default)] +pub struct Block { + /// Slot that produced the block. + pub slot: Slot, + /// Block hash for `slot`. + pub hash: Hash, + /// Block timestamp in the producer's time base. + pub time: i64, +} + +/// Superblock boundary entry stored at the end of the blockstore stream. +#[derive(SchemaRead, SchemaWrite, Clone, Copy)] +pub struct SuperblockSeal { + /// Checksum of accountsdb at the moment the superblock was sealed. + pub checksum: u64, + /// Total transactions written before this seal. + pub transactions: u64, +} diff --git a/nucleus/src/lib.rs b/nucleus/src/lib.rs new file mode 100644 index 0000000..93a54f8 --- /dev/null +++ b/nucleus/src/lib.rs @@ -0,0 +1,32 @@ +#![doc = include_str!("../README.md")] + +#[cfg(feature = "heed")] +pub mod heed; + +#[cfg(feature = "shutdown")] +pub mod shutdown; + +#[cfg(feature = "notifier")] +pub mod notifier; + +#[cfg(feature = "ledger-schema")] +pub mod ledger; + +#[cfg(feature = "metrics")] +pub mod metrics; + +#[cfg(feature = "runtime")] +pub mod runtime; + +#[cfg(feature = "testkit")] +pub mod testkit; + +#[cfg(feature = "tls")] +pub mod tls; + +/// Ledger slot number. +pub type Slot = u64; +/// One mebibyte in bytes. +pub const MB: usize = 1024 * 1024; +/// One gibibyte in bytes. +pub const GB: usize = 1024 * MB; diff --git a/nucleus/src/metrics.rs b/nucleus/src/metrics.rs new file mode 100644 index 0000000..4865174 --- /dev/null +++ b/nucleus/src/metrics.rs @@ -0,0 +1,138 @@ +//! Prometheus metric helpers shared by engine crates. + +use std::{sync::OnceLock, time::Instant}; + +use prometheus::{HistogramOpts, HistogramVec, Opts, default_registry}; +pub use prometheus::{IntCounter, IntCounterVec, IntGauge, IntGaugeVec}; +use tracing::warn; + +/// Metric name and help text kept together so collector definitions stay grepable. +#[derive(Clone, Copy)] +pub struct MetricSpec { + /// Prometheus collector name. + pub name: &'static str, + /// Prometheus help text. + pub help: &'static str, +} + +/// Creates a counter and applies an initial value before registration. +pub fn counter(spec: MetricSpec, initial: u64) -> IntCounter { + let counter = validate(IntCounter::with_opts(Opts::new(spec.name, spec.help))); + counter.inc_by(initial); + counter +} + +/// Creates a labeled counter. +pub fn counter_vec(spec: MetricSpec, labels: &[&'static str]) -> IntCounterVec { + validate(IntCounterVec::new(Opts::new(spec.name, spec.help), labels)) +} + +/// Creates a gauge and applies an initial value before registration. +pub fn gauge(spec: MetricSpec, initial: i64) -> IntGauge { + let gauge = validate(IntGauge::with_opts(Opts::new(spec.name, spec.help))); + gauge.set(initial); + gauge +} + +/// Creates a labeled gauge. +pub fn gauge_vec(spec: MetricSpec, labels: &[&'static str]) -> IntGaugeVec { + validate(IntGaugeVec::new(Opts::new(spec.name, spec.help), labels)) +} + +/// Registers `collector`, logging duplicate-name or registry errors without aborting startup. +pub fn register(namespace: &'static str, collector: C) +where + C: prometheus::core::Collector + 'static, +{ + if let Err(error) = default_registry().register(Box::new(collector)) { + warn!(namespace, ?error, "failed to register metric"); + } +} + +/// Applies `f` when metrics have been initialized; early calls are intentionally no-ops. +pub fn with_metrics(metrics: &OnceLock, f: impl FnOnce(&T)) { + if let Some(metrics) = metrics.get() { + f(metrics); + } else { + warn!("metrics are not registered"); + } +} + +/// Low-cardinality operation label used by duration histograms. +pub trait MetricOperation: Copy { + /// Returns the Prometheus label value for this operation. + fn label(self) -> &'static str; + + /// Starts a timer against `counters`, or a no-op timer before metrics are initialized. + fn time(self, counters: Option<&OperationCounters>) -> OperationTimer<'_> { + counters.map(|c| c.time(self)).unwrap_or_else(|| OperationTimer::noop(self)) + } +} + +/// Runtime operation duration histogram sharing one `op` label. +pub struct OperationCounters(HistogramVec); + +/// Operation duration histogram buckets, in microseconds. +const OPERATION_BUCKETS_MICROS: [f64; 8] = + [50.0, 200.0, 800.0, 3_200.0, 12_800.0, 51_200.0, 204_800.0, 1_000_000.0]; + +impl OperationCounters { + /// Builds the duration histogram collector. + pub fn new(micros: MetricSpec) -> Self { + let opts = + HistogramOpts::new(micros.name, micros.help).buckets(OPERATION_BUCKETS_MICROS.to_vec()); + Self(validate(HistogramVec::new(opts, &["op"]))) + } + + /// Registers the operation histogram in the default Prometheus registry. + pub fn register(&self, namespace: &'static str) { + register(namespace, self.0.clone()); + } + + /// Starts an operation timer that records latency when the returned guard drops. + pub fn time(&self, op: impl MetricOperation) -> OperationTimer<'_> { + OperationTimer { + counters: Some(self), + op: op.label(), + started: Instant::now(), + } + } +} + +/// Drop guard that records elapsed operation time in the metrics registry. +pub struct OperationTimer<'a> { + /// Operation counters to update on drop. + counters: Option<&'a OperationCounters>, + /// Operation label recorded with the duration observation. + op: &'static str, + /// Monotonic start instant captured when the guard is created. + started: Instant, +} + +impl OperationTimer<'static> { + /// Returns a timer that intentionally records nothing. + pub fn noop(op: impl MetricOperation) -> Self { + Self { + counters: None, + op: op.label(), + started: Instant::now(), + } + } +} + +impl Drop for OperationTimer<'_> { + /// Records elapsed microseconds when the timer leaves scope. + fn drop(&mut self) { + let Some(counters) = self.counters else { + return; + }; + let elapsed = self.started.elapsed().as_micros() as f64; + counters.0.with_label_values(&[self.op]).observe(elapsed); + } +} + +/// Unwraps construction of static metric definitions. +#[allow(clippy::expect_used)] +fn validate(result: prometheus::Result) -> T { + result.expect("prometheus metric registration should succeed") +} diff --git a/nucleus/src/runtime.rs b/nucleus/src/runtime.rs new file mode 100644 index 0000000..d3e3a97 --- /dev/null +++ b/nucleus/src/runtime.rs @@ -0,0 +1,127 @@ +//! Runtime-facing shared types: transaction views, execution I/O, scheduling +//! messages, and the schema of the engine's built-in MagicRoot program (its +//! address, instruction set, and authority). The MagicRoot execution logic +//! lives in the `magic-root-program` crate, which depends on these shared +//! definitions. + +use std::sync::Arc; + +use agave_transaction_view::{ + resolved_transaction_view::ResolvedTransactionView, transaction_view::SanitizedTransactionView, +}; +use derive_more::{Deref, From}; +use solana_svm::{ + transaction_balances::BalanceCollector, + transaction_processing_result::TransactionProcessingResult, +}; +use solana_transaction_error::TransactionResult; +use tokio::sync::mpsc::Sender; + +use crate::{Slot, ledger::Block}; + +/// Sanitized transaction view backed by a shared, immutable payload buffer. +pub type TransactionView = SanitizedTransactionView>>; +/// Sanitized transaction view with its account keys already resolved. +pub type ResolvedTransaction = ResolvedTransactionView>>; + +/// Cloneable submission handle into the sequencer's execution and simulation +/// channels. +#[derive(Clone, Deref)] +pub struct SequencerHandle { + /// Channel for submitting transactions to be executed and committed. + #[deref] + pub execution: Sender, + /// Channel for submitting transactions to be simulated without committing. + pub simulation: Sender, +} + +/// Work item handed to the transaction sequencer. +#[derive(From)] +pub enum SequencerMessage { + /// A transaction to schedule and execute. + Transaction(TransactionView), + /// A block boundary to seal before scheduling further transactions. + Block(Block), + /// Quiesce the sequencer and all its executors until released — used to take + /// a consistent snapshot at superblock boundaries (see ledger replay and + /// `finalize_superblock`). + Barrier(BarrierGuard), +} + +/// Quiescence barrier handed to a running service. +/// +/// On receipt the service drains its in-flight work, signals `acknowledged`, +/// then blocks on `released` before resuming. Paired with a [`BarrierController`] +/// via [`barrier`]. +pub struct BarrierGuard { + /// Signals the controller that the service is now idle. + pub acknowledged: oneshot::Sender<()>, + /// Resolves when the controller permits the service to resume; a dropped + /// controller also releases it. + pub released: oneshot::Receiver<()>, +} + +/// Controller side of a quiescence barrier, held by the caller that raised it. +/// +/// Awaits `acknowledged` to learn the service is quiesced, then sends `released` +/// (or drops) to let it resume. +pub struct BarrierController { + /// Resolves once the service reports it has gone idle. + pub acknowledged: oneshot::Receiver<()>, + /// Releases the service to resume operation. + pub released: oneshot::Sender<()>, +} + +/// Constructs a paired [`BarrierController`] and [`BarrierGuard`] over two +/// oneshot channels. +pub fn barrier() -> (BarrierController, BarrierGuard) { + let (acknowledged_tx, acknowledged_rx) = oneshot::channel(); + let (released_tx, released_rx) = oneshot::channel(); + let controller = BarrierController { + acknowledged: acknowledged_rx, + released: released_tx, + }; + let guard = BarrierGuard { + acknowledged: acknowledged_tx, + released: released_rx, + }; + (controller, guard) +} + +/// Work item handed to the transaction simulator. +#[derive(From)] +pub enum SimulatorMessage { + /// A transaction to simulate against the current state. + Transaction(Simulation), + /// A block boundary advancing the simulator's environment. + Block(Block), + /// Quiesce the simulator until released — keeps it idle while a consistent + /// snapshot is taken at superblock boundaries. + Barrier(BarrierGuard), +} + +/// A single simulation request and the channel to deliver its outcome. +pub struct Simulation { + /// The transaction to simulate. + pub transaction: TransactionView, + /// Where the simulation result is returned to the caller. + pub response: oneshot::Sender>, +} + +/// Transaction payload paired with the SVM output needed to finalize state. +pub struct FullTransaction { + /// Sanitized transaction bytes and signatures accepted by the ledger. + pub transaction: TransactionView, + /// SVM execution output for the transaction. + pub execution: ExecutionRecord, +} + +/// SVM output produced by executing a transaction. +pub struct ExecutionRecord { + /// SVM processing result produced for this transaction. + pub result: TransactionProcessingResult, + /// Native pre/post balances collected during execution. + pub balances: Option, + /// Slot assigned to the execution result. + pub slot: Slot, +} diff --git a/nucleus/src/shutdown.rs b/nucleus/src/shutdown.rs new file mode 100644 index 0000000..dcd0867 --- /dev/null +++ b/nucleus/src/shutdown.rs @@ -0,0 +1,223 @@ +//! Cooperative shutdown for engine services. +//! +//! A [`ShutdownManager`] owns ordered cancellation tokens and a set of +//! registered service handles. Each service receives a [`ShutdownHandle`], +//! observes its tier token while running, and reports a [`ShutdownReason`] when +//! it exits. The manager waits for an OS shutdown signal, internal cancellation, +//! or service termination, then cancels each tier in order and gives it a +//! bounded window to stop before moving on. + +use std::{error::Error, time::Duration}; + +use futures::{ + StreamExt, + future::{BoxFuture, select}, + stream::FuturesUnordered, +}; +use oneshot::Sender; +use tokio::time::timeout; +use tokio_util::sync::CancellationToken; +use tracing::{error, info, warn}; + +const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(8); + +type HandleFuture = BoxFuture<'static, (Service, ShutdownTier, ShutdownReason)>; + +/// Background service tracked by the shutdown manager. +#[derive(Debug)] +pub enum Service { + /// Ledger append worker. + LedgerAppender, + /// Ledger read worker. + LedgerReader, + /// Ledger replay worker. + LedgerReplayer, + /// Transaction scheduler service. + Sequencer, + /// Transaction executor worker with its worker index. + TransactionExecutor(u32), + /// Transaction simulation worker. + TransactionSimulator, + /// Subscription map cleanup worker. + SubscriptionsCleanup, + /// Block pacing task. + PaceMaker, +} + +impl Service { + fn tier(&self) -> ShutdownTier { + match self { + Service::PaceMaker => ShutdownTier::One, + Service::Sequencer => ShutdownTier::Two, + _ => ShutdownTier::Three, + } + } +} + +#[derive(Clone, Copy, Debug)] +enum ShutdownTier { + One, + Two, + Three, +} + +impl ShutdownTier { + const COUNT: usize = 3; + const ORDER: [Self; Self::COUNT] = [Self::One, Self::Two, Self::Three]; +} + +/// Coordinates graceful shutdown across engine services. +#[derive(Default)] +pub struct ShutdownManager { + /// Internal token signalled when any shutdown path wins. + request: CancellationToken, + /// Service cancellation tokens, one per ordered shutdown tier. + tokens: [CancellationToken; ShutdownTier::COUNT], + /// Registered service termination reports. + handles: FuturesUnordered, + /// Number of services that have not reported termination, by tier. + pending: [isize; ShutdownTier::COUNT], +} + +impl ShutdownManager { + /// Wait for an OS shutdown signal, internal cancellation, or service failure. + pub async fn wait(&mut self) { + tokio::select! { + _ = graceful_shutdown() => { + self.request.cancel(); + info!("graceful shutdown has been requested"); + } + _ = self.request.cancelled() => { + error!("engine termination request was received internally"); + } + Some((service, tier, reason)) = self.handles.next(), if !self.handles.is_empty() => { + self.pending(tier, -1); + error!(?service, ?reason, "terminated prematurely"); + } + } + } + + /// Cancels services one tier at a time and drains their termination reports. + /// + /// Each tier gets `SHUTDOWN_TIMEOUT` to report before the next tier is + /// cancelled. Already terminated services are skipped by their tier. + pub async fn terminate(&mut self) { + info!("initiating graceful shutdown of the engine"); + for tier in ShutdownTier::ORDER { + self.tokens[tier as usize].cancel(); + if self.pending[tier as usize] == 0 { + continue; + } + if timeout(SHUTDOWN_TIMEOUT, self.drain(tier)).await.is_err() { + warn!( + ?tier, + remaining = self.pending[tier as usize], + "shutdown tier timed out" + ); + } + } + } + + /// Register a service and return its cancellation handle. + pub fn handle(&mut self, service: Service) -> ShutdownHandle { + let tier = service.tier(); + let (tx, rx) = oneshot::channel(); + let fut = async move { + let reason = rx.await.unwrap_or_default(); + (service, tier, reason) + }; + self.handles.push(Box::pin(fut)); + self.pending(tier, 1); + ShutdownHandle { + request: self.request.clone(), + cancel: self.tokens[tier as usize].child_token(), + reason: Some(tx), + } + } + + async fn drain(&mut self, tier: ShutdownTier) { + while self.pending[tier as usize] != 0 { + let Some((service, tier, reason)) = self.handles.next().await else { + return; + }; + self.pending(tier, -1); + Self::log(service, reason); + } + } + + fn pending(&mut self, tier: ShutdownTier, op: isize) { + self.pending[tier as usize] += op; + } + + fn log(service: Service, reason: ShutdownReason) { + match reason { + ShutdownReason::Unexpected => { + warn!(?service, "terminated unexpectedly") + } + ShutdownReason::Signalled => { + info!(?service, "terminated gracefully") + } + ShutdownReason::Error(error) => { + error!(?service, ?error, "terminated with error") + } + } + } +} + +/// Waits for SIGTERM or Ctrl-C. +async fn graceful_shutdown() { + use tokio::signal::unix::{SignalKind, signal}; + let term = Box::pin(async { + let Ok(mut term) = signal(SignalKind::terminate()) else { + return; + }; + term.recv().await; + }); + let ctrlc = Box::pin(tokio::signal::ctrl_c()); + select(term, ctrlc).await; +} + +/// Cancellation handle held by a running service. +pub struct ShutdownHandle { + /// Token used to wake the shutdown manager when this service exits. + request: CancellationToken, + /// Token observed by the running service. + cancel: CancellationToken, + /// One-shot report consumed by the manager when the service exits. + reason: Option>, +} + +/// Reason reported when a service terminates. +#[derive(Debug, Default)] +pub enum ShutdownReason { + /// Service handle was dropped without reporting a reason. + #[default] + Unexpected, + /// Service stopped after being signalled. + Signalled, + /// Service stopped because of an error. + Error(Box), +} + +impl ShutdownHandle { + /// Request engine shutdown and report this service's termination reason. + pub fn terminate(&mut self, reason: ShutdownReason) { + self.request.cancel(); + if let Some(tx) = self.reason.take() { + let _ = tx.send(reason); + }; + } + + /// Wait until this service's shutdown tier is cancelled. + pub async fn signalled(&self) { + self.cancel.cancelled().await + } +} + +impl Drop for ShutdownHandle { + fn drop(&mut self) { + if let Some(tx) = self.reason.take() { + let _ = tx.send(ShutdownReason::Unexpected); + }; + } +} diff --git a/nucleus/src/testkit.rs b/nucleus/src/testkit.rs new file mode 100644 index 0000000..455a043 --- /dev/null +++ b/nucleus/src/testkit.rs @@ -0,0 +1,88 @@ +//! Engine-agnostic test fixtures shared across crate test suites. +//! +//! Only the primitives that depend on nothing above nucleus live here — +//! wincode-serialized transactions, block boundaries, and throwaway directories. +//! Keeper-level harness code (building a `Keeper`, loading the v42 ELF) lives in +//! `keeper::testkit`. Compiled only under the `testkit` feature, so it never +//! reaches release builds. +#![allow(clippy::expect_used)] + +use std::sync::{Arc, Once}; + +use solana_hash::Hash; +use solana_instruction::{AccountMeta, Instruction}; +use solana_keypair::Keypair; +use solana_message::Message; +use solana_pubkey::Pubkey; +use solana_signature::Signature; +use solana_signer::Signer; +use solana_transaction::Transaction; +pub use tempfile::TempDir; +use tracing_subscriber::{EnvFilter, fmt}; + +pub use v42_calculator_interface::ID as V42_ID; + +use crate::{Slot, ledger::Block}; + +static TRACING: Once = Once::new(); + +/// Installs a libtest-aware tracing subscriber for test processes. +/// +/// The default filter is intentionally quiet. Set `RUST_LOG` and run tests with +/// `-- --nocapture` to see lower-level spans and events while debugging. +pub fn init_tracing() { + TRACING.call_once(|| { + let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("warn")); + let _ = fmt().with_env_filter(filter).with_test_writer().try_init(); + }); +} + +/// A throwaway on-disk directory; the returned guard must outlive the store +/// opened over it, since these stores keep their files open/mmapped. +pub fn tempdir() -> TempDir { + TempDir::new().expect("create tempdir") +} + +/// A block boundary with a distinct hash and time derived from `slot`. +pub fn block(slot: Slot) -> Block { + Block { + slot, + hash: Hash::new_from_array([slot as u8; 32]), + time: slot as i64, + } +} + +/// Signs `instruction` from `payer` against `blockhash`, returning the first +/// signature and the wincode-serialized transaction bytes. +/// +/// The single shared envelope builder for every test transaction: callers supply +/// a fully-built instruction (a v42 program call, or any other), so nothing has +/// to re-open-code the sign-and-serialize path. +pub fn sign_instruction( + payer: &Keypair, + instruction: Instruction, + blockhash: Hash, +) -> (Signature, Arc>) { + let message = Message::new(&[instruction], Some(&payer.pubkey())); + let mut tx = Transaction::new_unsigned(message); + tx.try_sign(&[payer], blockhash).expect("transaction signs"); + let signature = tx.signatures[0]; + ( + signature, + Arc::new(wincode::serialize(&tx).expect("transaction serializes")), + ) +} + +/// A signed, wincode-serialized v42 transaction plus its first signature. +/// +/// The instruction references `accounts` as read-only keys so they land in the +/// transaction's static account keys (and thus any account index), and targets +/// the v42 program so the transaction is executable, not just well-formed. A +/// fresh random payer per call keeps signatures unique without varying the +/// blockhash. +pub fn transaction(accounts: &[Pubkey]) -> (Signature, Arc>) { + let payer = Keypair::new(); + let metas = accounts.iter().map(|k| AccountMeta::new_readonly(*k, false)).collect(); + let ix = Instruction::new_with_bytes(V42_ID, &[], metas); + sign_instruction(&payer, ix, Hash::default()) +} diff --git a/nucleus/src/tls.rs b/nucleus/src/tls.rs new file mode 100644 index 0000000..3596832 --- /dev/null +++ b/nucleus/src/tls.rs @@ -0,0 +1,46 @@ +//! Thread-local execution state shared with runtime-adjacent code. + +use std::{ + cell::{Cell, RefCell}, + collections::VecDeque, +}; + +use solana_instruction_error::InstructionError; +use solana_pubkey::Pubkey; +use wincode::{SchemaWrite, config::Configuration}; + +/// Wincode-encoded message buffered for later handling on the same thread. +pub type EncodedMessage = Vec; + +thread_local! { + /// Per-thread queue for messages emitted while a transaction executes. + pub static TLS: RefCell = RefCell::new(Default::default()); + /// Signer authorized to invoke the MagicRoot program on the current thread. + pub static AUTHORITY: Cell = Cell::new(Default::default()); +} + +/// FIFO queue of encoded messages scoped to the current thread. +#[derive(Default)] +pub struct TlsManager(VecDeque); + +impl TlsManager { + /// Encodes `msg` and appends it to the current thread's queue. + pub fn enqueue(msg: &T) -> Result<(), InstructionError> + where + T: SchemaWrite, + { + let encoded = wincode::serialize(msg).map_err(|_| InstructionError::Custom(u32::MAX))?; + TLS.with_borrow_mut(|tls| tls.0.push_back(encoded)); + Ok(()) + } + + /// Removes the oldest encoded message from the current thread's queue. + pub fn dequeue() -> Option { + TLS.with_borrow_mut(|tls| tls.0.pop_front()) + } + + /// Drops every queued message for the current thread. + pub fn clear() { + TLS.with_borrow_mut(|tls| tls.0.clear()) + } +}