From b79854032d952570214b0e5808b77544d63ed8fc Mon Sep 17 00:00:00 2001 From: mfw78 Date: Thu, 16 Jul 2026 22:42:22 +0000 Subject: [PATCH] feat: make the log pipeline pluggable through the components seam ComponentsBuilder grows a fourth logs slot, defaulting to the new LogPipelineBuilder (the in-memory pipeline sized from [limits.logs]); with_logs substitutes a custom builder. The Runtime preset names the slot as LogsBuilder and the launcher cars carry the parameter through. --- crates/nexum-runtime/src/builder.rs | 24 +++-- .../src/host/component/builder.rs | 94 ++++++++++++++++--- .../nexum-runtime/src/host/component/mod.rs | 2 +- crates/nexum-runtime/src/preset.rs | 14 ++- 4 files changed, 106 insertions(+), 28 deletions(-) diff --git a/crates/nexum-runtime/src/builder.rs b/crates/nexum-runtime/src/builder.rs index 412c4d8f..f5559506 100644 --- a/crates/nexum-runtime/src/builder.rs +++ b/crates/nexum-runtime/src/builder.rs @@ -494,10 +494,10 @@ impl<'a, T: RuntimeTypes> TypedBuilder<'a, T> { } /// Bind the component builders that open the backends at launch. - pub fn with_components( + pub fn with_components( self, - components: ComponentsBuilder, - ) -> ComponentsStage<'a, T, C, S, E> { + components: ComponentsBuilder, + ) -> ComponentsStage<'a, T, C, S, E, L> { ComponentsStage { config: self.config, extensions: self.extensions, @@ -511,19 +511,22 @@ impl<'a, T: RuntimeTypes> TypedBuilder<'a, T> { } /// The component builders are bound; the add-on set remains. -pub struct ComponentsStage<'a, T: RuntimeTypes, C, S, E> { +pub struct ComponentsStage<'a, T: RuntimeTypes, C, S, E, L> { config: &'a EngineConfig, extensions: Vec>>, wasm: Option, manifest: Option, clocks: Option, - components: ComponentsBuilder, + components: ComponentsBuilder, _t: PhantomData T>, } -impl<'a, T: RuntimeTypes, C, S, E> ComponentsStage<'a, T, C, S, E> { +impl<'a, T: RuntimeTypes, C, S, E, L> ComponentsStage<'a, T, C, S, E, L> { /// Bind the cross-cutting add-on set installed before the engine boots. - pub fn with_add_ons(self, add_ons: &'a [&'a dyn RuntimeAddOn]) -> ReadyBuilder<'a, T, C, S, E> { + pub fn with_add_ons( + self, + add_ons: &'a [&'a dyn RuntimeAddOn], + ) -> ReadyBuilder<'a, T, C, S, E, L> { ReadyBuilder { config: self.config, extensions: self.extensions, @@ -538,22 +541,23 @@ impl<'a, T: RuntimeTypes, C, S, E> ComponentsStage<'a, T, C, S, E> { /// The assembly is complete; [`launch`](Self::launch) opens the backends and /// runs. -pub struct ReadyBuilder<'a, T: RuntimeTypes, C, S, E> { +pub struct ReadyBuilder<'a, T: RuntimeTypes, C, S, E, L> { config: &'a EngineConfig, extensions: Vec>>, wasm: Option, manifest: Option, clocks: Option, - components: ComponentsBuilder, + components: ComponentsBuilder, add_ons: &'a [&'a dyn RuntimeAddOn], } -impl ReadyBuilder<'_, T, C, S, E> +impl ReadyBuilder<'_, T, C, S, E, L> where T: RuntimeTypes, C: ComponentBuilder, S: ComponentBuilder, E: ComponentBuilder, + L: ComponentBuilder, { /// Open the backends and launch. Builds the [`Components`] bundle from the /// bound builders, then drives [`LaunchRuntime::launch`] with a fresh diff --git a/crates/nexum-runtime/src/host/component/builder.rs b/crates/nexum-runtime/src/host/component/builder.rs index 705896bf..60b734cf 100644 --- a/crates/nexum-runtime/src/host/component/builder.rs +++ b/crates/nexum-runtime/src/host/component/builder.rs @@ -3,8 +3,9 @@ //! //! Each core backend is wrapped as a [`ComponentBuilder`], and //! [`ComponentsBuilder`] assembles the core seams (plus the lattice `Ext` -//! payload) into a [`Components`] bundle. The composition root names the -//! concrete builders once; boot drives them through this trait. +//! payload and the log pipeline) into a [`Components`] bundle. The +//! composition root names the concrete builders once; boot drives them +//! through this trait. use std::future::Future; use std::path::Path; @@ -81,6 +82,18 @@ impl ComponentBuilder for LocalStoreBuilder { } } +/// Builds the default [`LogPipeline`]: the byte-bounded in-memory backend +/// sized from `[limits.logs]`. +pub struct LogPipelineBuilder; + +impl ComponentBuilder for LogPipelineBuilder { + type Output = LogPipeline; + + async fn build(self, ctx: &BuilderContext<'_>) -> anyhow::Result { + Ok(LogPipeline::in_memory(ctx.config.limits.logs())) + } +} + /// Names the component slot whose build failed. The leaf cause stays an /// `anyhow::Error` because the backends fail for heterogeneous reasons /// (I/O for the store, network for the chain). @@ -95,6 +108,9 @@ pub enum BuildError { /// The extension payload builder failed. #[error("build the extension payload: {0}")] Ext(anyhow::Error), + /// The log pipeline builder failed. + #[error("build the log pipeline: {0}")] + Logs(anyhow::Error), } /// The empty extension payload: a no-op builder for a core-only lattice @@ -107,41 +123,62 @@ impl ComponentBuilder for () { } } -/// Assembles the core backend builders and the lattice `Ext` builder into -/// a [`Components`] bundle. The log pipeline is sized from `[limits.logs]` -/// and built here; the embedder retains its read handle by cloning -/// [`Components::logs`] after the build. -pub struct ComponentsBuilder { +/// Assembles the core backend builders, the lattice `Ext` builder, and the +/// log pipeline builder into a [`Components`] bundle. The logs slot defaults +/// to [`LogPipelineBuilder`]; the embedder retains the read handle by +/// cloning [`Components::logs`] after the build. +pub struct ComponentsBuilder { /// Builds the chain backend ([`RuntimeTypes::Chain`]). pub chain: C, /// Builds the store backend ([`RuntimeTypes::Store`]). pub store: S, /// Builds the extension payload ([`RuntimeTypes::Ext`]). pub ext: E, + /// Builds the shared [`LogPipeline`]. + pub logs: L, } impl ComponentsBuilder { - /// Create a new [`ComponentsBuilder`]. + /// Create a new [`ComponentsBuilder`] with the default log pipeline. pub fn new(chain: C, store: S, ext: E) -> Self { - Self { chain, store, ext } + Self { + chain, + store, + ext, + logs: LogPipelineBuilder, + } + } +} + +impl ComponentsBuilder { + /// Replace the log pipeline builder. + pub fn with_logs(self, logs: L2) -> ComponentsBuilder { + ComponentsBuilder { + chain: self.chain, + store: self.store, + ext: self.ext, + logs, + } } - /// Drive each builder against `ctx`, then bundle the backends with a - /// fresh log pipeline. The builder outputs must match the lattice - /// seams: chain to [`RuntimeTypes::Chain`], store to - /// [`RuntimeTypes::Store`], ext to [`RuntimeTypes::Ext`]. A failing - /// sub-build returns the [`BuildError`] variant naming that slot. + /// Drive each builder against `ctx` and bundle the backends. The + /// builder outputs must match the lattice seams: chain to + /// [`RuntimeTypes::Chain`], store to [`RuntimeTypes::Store`], ext to + /// [`RuntimeTypes::Ext`]; logs always yields a [`LogPipeline`]. A + /// failing sub-build returns the [`BuildError`] variant naming that + /// slot. pub async fn build(self, ctx: &BuilderContext<'_>) -> Result, BuildError> where T: RuntimeTypes, C: ComponentBuilder, S: ComponentBuilder, E: ComponentBuilder, + L: ComponentBuilder, { let chain = self.chain.build(ctx).await.map_err(BuildError::Chain)?; let store = self.store.build(ctx).await.map_err(BuildError::Store)?; let ext = self.ext.build(ctx).await.map_err(BuildError::Ext)?; - let logs = LogPipeline::in_memory(ctx.config.limits.logs()); + let logs = self.logs.build(ctx).await.map_err(BuildError::Logs)?; Ok(Components { chain, store, @@ -189,4 +226,31 @@ mod tests { // The bundle carries a live in-memory log pipeline. let _ = &components.logs; } + + /// `with_logs` substitutes the log pipeline builder: the bundle carries + /// the exact pipeline the custom builder yields. + #[tokio::test] + async fn with_logs_substitutes_the_pipeline() { + let dir = tempfile::tempdir().expect("tempdir"); + let config = EngineConfig::default(); + let tasks = nexum_tasks::TaskManager::new(); + let executor = tasks.executor(); + let ctx = BuilderContext { + config: &config, + data_dir: dir.path(), + executor: &executor, + }; + + let custom = LogPipeline::in_memory(config.limits.logs()); + let components = ComponentsBuilder::new(ProviderPoolBuilder, LocalStoreBuilder, ()) + .with_logs(crate::test_utils::Prebuilt(custom.clone())) + .build::(&ctx) + .await + .expect("build with a custom log pipeline"); + + assert!( + std::sync::Arc::ptr_eq(&components.logs.router(), &custom.router()), + "bundle carries the substituted pipeline", + ); + } } diff --git a/crates/nexum-runtime/src/host/component/mod.rs b/crates/nexum-runtime/src/host/component/mod.rs index 0adb15fa..eaa0e70a 100644 --- a/crates/nexum-runtime/src/host/component/mod.rs +++ b/crates/nexum-runtime/src/host/component/mod.rs @@ -11,7 +11,7 @@ mod state; pub use builder::{ BuildError, BuilderContext, ComponentBuilder, ComponentsBuilder, LocalStoreBuilder, - ProviderPoolBuilder, + LogPipelineBuilder, ProviderPoolBuilder, }; pub use chain::{ChainMethod, ChainProvider}; pub use runtime_types::{Handle, RuntimeTypes}; diff --git a/crates/nexum-runtime/src/preset.rs b/crates/nexum-runtime/src/preset.rs index 05556ee3..19a948a9 100644 --- a/crates/nexum-runtime/src/preset.rs +++ b/crates/nexum-runtime/src/preset.rs @@ -8,9 +8,11 @@ use crate::addons::{AddOns, PrometheusAddOn}; use crate::host::component::{ - ComponentBuilder, ComponentsBuilder, LocalStoreBuilder, ProviderPoolBuilder, RuntimeTypes, + ComponentBuilder, ComponentsBuilder, LocalStoreBuilder, LogPipelineBuilder, + ProviderPoolBuilder, RuntimeTypes, }; use crate::host::local_store_redb::LocalStore; +use crate::host::logs::LogPipeline; use crate::host::provider_pool::ProviderPool; /// A bundled runtime assembly: the [`RuntimeTypes`] lattice plus the component @@ -27,9 +29,16 @@ pub trait Runtime { type StoreBuilder: ComponentBuilder::Store>; /// Builds the extension payload ([`RuntimeTypes::Ext`]). type ExtBuilder: ComponentBuilder::Ext>; + /// Builds the shared [`LogPipeline`]. + type LogsBuilder: ComponentBuilder; /// The component builders that open the backends at launch. - fn components() -> ComponentsBuilder; + fn components() -> ComponentsBuilder< + Self::ChainBuilder, + Self::StoreBuilder, + Self::ExtBuilder, + Self::LogsBuilder, + >; /// The cross-cutting add-ons installed before the engine boots. fn add_ons() -> AddOns; @@ -52,6 +61,7 @@ impl Runtime for CoreRuntime { type ChainBuilder = ProviderPoolBuilder; type StoreBuilder = LocalStoreBuilder; type ExtBuilder = (); + type LogsBuilder = LogPipelineBuilder; fn components() -> ComponentsBuilder { ComponentsBuilder::new(ProviderPoolBuilder, LocalStoreBuilder, ())