save place to check on first iteration of diffs

Author: dgrkotsonis

Cargo.toml

@@ -87,7 +87,9 @@ members = [
   "synthesizer/snark",
   "utilities",
   "utilities/derives",
-  "wasm"
+  "wasm",
+  "plugins/slipstream_plugin_interface",
+  "plugins/slipstream_plugin_manager"
 ]
 
 [lib]
@@ -389,6 +391,14 @@ default-features = false
 path = "ledger/store"
 version = "=4.5.0"
 
+[workspace.dependencies.snarkvm-slipstream-plugin-interface]
+path = "plugins/slipstream_plugin_interface"
+version = "=4.5.0"
+
+[workspace.dependencies.snarkvm-slipstream-plugin-manager]
+path = "plugins/slipstream_plugin_manager"
+version = "=4.5.0"
+
 [workspace.dependencies.snarkvm-ledger-test-helpers]
 path = "ledger/test-helpers"
 version = "=4.5.0"

ledger/store/Cargo.toml

@@ -18,8 +18,8 @@ edition = "2024"
 
 [features]
 default = [ "indexmap/rayon" ]
-history = [ ]
-history-staking-rewards = [ ]
+history = [ "dep:snarkvm-slipstream-plugin-manager" ]
+history-staking-rewards = [ "dep:snarkvm-slipstream-plugin-manager" ]
 locktick = [ "dep:locktick", "snarkvm-ledger-puzzle/locktick" ]
 rocks = [ "rocksdb", "smallvec" ]
 serial = [
@@ -42,6 +42,10 @@ wasm = [
 ]
 test = [ ]
 
+[dependencies.snarkvm-slipstream-plugin-manager]
+workspace = true
+optional = true
+
 [dependencies.snarkvm-console]
 workspace = true
 

ledger/store/src/program/finalize.rs

@@ -36,6 +36,10 @@ use std::{
     borrow::Cow,
     sync::atomic::{AtomicU32, Ordering},
 };
+#[cfg(any(feature = "history", feature = "history-staking-rewards"))]
+use std::sync::{Arc, OnceLock, RwLock, atomic::AtomicBool};
+#[cfg(any(feature = "history", feature = "history-staking-rewards"))]
+use snarkvm_slipstream_plugin_manager::SlipstreamPluginManager;
 
 /// TODO (howardwu): Remove this.
 /// Returns the mapping ID for the given `program ID` and `mapping name`.
@@ -249,6 +253,7 @@ pub trait FinalizeStorage<N: Network>: 'static + Clone + Send + Sync {
         Ok(FinalizeOperation::InitializeMapping(to_mapping_id(&program_id, &mapping_name)?))
     }
 
+    // NOTE: THIS IS NEVER USED IN PROD
     /// Stores the given `(key, value)` pair at the given `program ID` and `mapping name` in storage.
     /// If the `mapping name` is not initialized, an error is returned.
     /// If the `key` already exists, the method returns an error.
@@ -279,6 +284,9 @@ pub trait FinalizeStorage<N: Network>: 'static + Clone + Send + Sync {
             // Update the historical maps.
             #[cfg(feature = "history")]
             {
+                // TODO: TODO: Is here where we would want to stream the data?
+                // NOTE: MAYBE WE WOULD WANT TO PUSH TO A BUFFER HERE AND THEN STREAM IT
+                // IN ATOMIC_POST_RATIFY()
                 let current_height = self.current_block_height().load(Ordering::SeqCst);
 
                 // Insert the initial value as the first historical update.
@@ -329,9 +337,10 @@ pub trait FinalizeStorage<N: Network>: 'static + Clone + Send + Sync {
         let value_id = N::hash_bhp1024(&(key_id, N::hash_bhp1024(&value.to_bits_le())?).to_bits_le())?;
 
         atomic_batch_scope!(self, {
-            // Update the historical maps.
+            // Update the historical maps. // NOTE: THIS GETS CALLED IN vm/finalize.rs, TODO: STREAM HERE
             #[cfg(feature = "history")]
             {
+                // TODO: Add to buffer here for optional streaming?
                 let current_height = self.current_block_height().load(Ordering::SeqCst);
 
                 // Register the updated value at the current height.
@@ -654,6 +663,14 @@ pub struct FinalizeStore<N: Network, P: FinalizeStorage<N>> {
     storage: P,
     /// PhantomData.
     _phantom: PhantomData<N>,
+    /// Indicates that canonical finalize is currently in progress.
+    /// When `true`, storage writes notify registered Slipstream plugins.
+    #[cfg(any(feature = "history", feature = "history-staking-rewards"))]
+    is_finalize_mode: Arc<AtomicBool>,
+    /// Optional plugin manager for streaming canonical mapping and staking updates.
+    /// Uses `OnceLock` so it can be installed from a shared reference after construction.
+    #[cfg(any(feature = "history", feature = "history-staking-rewards"))]
+    slipstream_plugin_manager: OnceLock<Arc<RwLock<SlipstreamPluginManager>>>,
 }
 
 impl<N: Network, P: FinalizeStorage<N>> FinalizeStore<N, P> {
@@ -665,7 +682,14 @@ impl<N: Network, P: FinalizeStorage<N>> FinalizeStore<N, P> {
     /// Initializes a finalize store from storage.
     pub fn from(storage: P) -> Result<Self> {
         // Return the finalize store.
-        Ok(Self { storage, _phantom: PhantomData })
+        Ok(Self {
+            storage,
+            _phantom: PhantomData,
+            #[cfg(any(feature = "history", feature = "history-staking-rewards"))]
+            is_finalize_mode: Arc::new(AtomicBool::new(false)),
+            #[cfg(any(feature = "history", feature = "history-staking-rewards"))]
+            slipstream_plugin_manager: OnceLock::new(),
+        })
     }
 
     /// Starts an atomic batch write operation.
@@ -714,6 +738,61 @@ impl<N: Network, P: FinalizeStorage<N>> FinalizeStore<N, P> {
         self.storage.current_block_height()
     }
 
+    /// Returns a reference to the canonical finalize mode flag.
+    ///
+    /// When `true`, storage writes notify registered Slipstream plugins.
+    /// Set to `true` by the VM before canonical finalize runs and reset to `false` afterwards.
+    #[cfg(any(feature = "history", feature = "history-staking-rewards"))]
+    pub fn is_finalize_mode(&self) -> &Arc<AtomicBool> {
+        &self.is_finalize_mode
+    }
+
+    /// Installs a Slipstream plugin manager to receive canonical mapping and staking updates.
+    ///
+    /// May be called from a shared reference. Logs a warning if called more than once.
+    #[cfg(any(feature = "history", feature = "history-staking-rewards"))]
+    pub fn set_slipstream_plugin_manager(&self, manager: Arc<RwLock<SlipstreamPluginManager>>) {
+        if self.slipstream_plugin_manager.set(manager).is_err() {
+            tracing::warn!("Slipstream plugin manager is already set; ignoring subsequent call.");
+        }
+    }
+
+    /// Notifies all interested plugins of a staking reward, if canonical finalize is active.
+    ///
+    /// Errors from plugin calls are logged but never propagated.
+    #[cfg(feature = "history-staking-rewards")]
+    pub fn notify_staking_reward(
+        &self,
+        staker: &Address<N>,
+        validator: &Address<N>,
+        reward: u64,
+        new_stake: u64,
+        block_height: u32,
+    ) {
+        if !self.is_finalize_mode.load(std::sync::atomic::Ordering::SeqCst) {
+            return;
+        }
+        if let Some(mgr) = self.slipstream_plugin_manager.get() {
+            let staker_bytes = match staker.to_bytes_le() {
+                Ok(b) => b,
+                Err(e) => {
+                    tracing::warn!("Slipstream: failed to serialize staker address: {e}");
+                    return;
+                }
+            };
+            let validator_bytes = match validator.to_bytes_le() {
+                Ok(b) => b,
+                Err(e) => {
+                    tracing::warn!("Slipstream: failed to serialize validator address: {e}");
+                    return;
+                }
+            };
+            mgr.read()
+                .unwrap()
+                .notify_staking_reward(&staker_bytes, &validator_bytes, reward, new_stake, block_height);
+        }
+    }
+
     /// Returns the historical value of a mapping.
     #[cfg(feature = "history")]
     pub fn get_historical_mapping_value(
@@ -827,7 +906,32 @@ impl<N: Network, P: FinalizeStorage<N>> FinalizeStoreTrait<N> for FinalizeStore<
         key: Plaintext<N>,
         value: Value<N>,
     ) -> Result<FinalizeOperation<N>> {
-        self.storage.update_key_value(program_id, mapping_name, key, value)
+        // Serialize before moving, if a plugin notification may be needed.
+        #[cfg(feature = "history")]
+        let plugin_data =
+            if self.is_finalize_mode.load(Ordering::SeqCst) && self.slipstream_plugin_manager.get().is_some() {
+                Some((
+                    program_id.to_bytes_le()?,
+                    mapping_name.to_bytes_le()?,
+                    key.to_bytes_le()?,
+                    value.to_bytes_le()?,
+                ))
+            } else {
+                None
+            };
+
+        let result = self.storage.update_key_value(program_id, mapping_name, key, value)?;
+
+        // Notify plugins of the update if in canonical finalize mode.
+        #[cfg(feature = "history")]
+        if let Some((pid, mname, k, v)) = plugin_data {
+            let height = self.storage.current_block_height().load(Ordering::SeqCst);
+            if let Some(mgr) = self.slipstream_plugin_manager.get() {
+                mgr.read().unwrap().notify_mapping_update(&pid, &mname, &k, &v, height);
+            }
+        }
+
+        Ok(result)
     }
 
     /// Removes the key-value pair for the given `program ID`, `mapping name`, and `key` from storage.
@@ -860,7 +964,35 @@ impl<N: Network, P: FinalizeStorage<N>> FinalizeStore<N, P> {
         mapping_name: Identifier<N>,
         entries: Vec<(Plaintext<N>, Value<N>)>,
     ) -> Result<FinalizeOperation<N>> {
-        self.storage.replace_mapping(program_id, mapping_name, entries)
+        // Serialize mapping identity and all entries before moving them into storage,
+        // so they are available for plugin notification after the storage call.
+        #[cfg(feature = "history")]
+        let plugin_data: Option<(Vec<u8>, Vec<u8>, Vec<(Vec<u8>, Vec<u8>)>)> =
+            if self.is_finalize_mode.load(Ordering::SeqCst) && self.slipstream_plugin_manager.get().is_some() {
+                let mut serialized_entries = Vec::with_capacity(entries.len());
+                for (key, value) in &entries {
+                    serialized_entries.push((key.to_bytes_le()?, value.to_bytes_le()?));
+                }
+                Some((program_id.to_bytes_le()?, mapping_name.to_bytes_le()?, serialized_entries))
+            } else {
+                None
+            };
+
+        let result = self.storage.replace_mapping(program_id, mapping_name, entries)?;
+
+        // Notify plugins of each updated key-value pair if in canonical finalize mode.
+        #[cfg(feature = "history")]
+        if let Some((pid, mname, serialized_entries)) = plugin_data {
+            let height = self.storage.current_block_height().load(Ordering::SeqCst);
+            if let Some(mgr) = self.slipstream_plugin_manager.get() {
+                let mgr_guard = mgr.read().unwrap();
+                for (k, v) in &serialized_entries {
+                    mgr_guard.notify_mapping_update(&pid, &mname, k, v, height);
+                }
+            }
+        }
+
+        Ok(result)
     }
 
     /// Removes the mapping for the given `program ID` and `mapping name` from storage,

plugins/slipstream_plugin_interface/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "snarkvm-slipstream-plugin-interface"
+version = "4.5.0"
+authors = [ "The Aleo Team <hello@aleo.org>" ]
+description = "The SnarkVM Slipstream plugin interface."
+homepage = "https://aleo.org"
+repository = "https://github.com/ProvableHQ/snarkVM"
+license = "Apache-2.0"
+edition = "2024"
+
+[dependencies.anyhow]
+workspace = true

plugins/slipstream_plugin_interface/README.md

@@ -0,0 +1,7 @@
+# Aleo Slipstream Plugin Interface
+
+This crate enables a plugin to be added into a SnarkVM runtime to
+take actions at the time of mapping updates at block finalization;
+for example, saving historical mappings state and staking data to an external database. The plugin must
+implement the `SlipstreamPlugin` trait. Please see the details of the
+`slipstream_plugin_interface.rs` for the interface definition.

plugins/slipstream_plugin_interface/src/lib.rs

@@ -0,0 +1 @@
+pub mod slipstream_plugin_interface;

plugins/slipstream_plugin_interface/src/slipstream_plugin_interface.rs

@@ -0,0 +1,73 @@
+// Copyright (c) 2019-2026 Provable Inc.
+// This file is part of the snarkVM library.
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at:
+
+// http://www.apache.org/licenses/LICENSE-2.0
+
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+/// The interface for Aleo Slipstream plugins. A plugin must implement
+/// the `SlipstreamPlugin` trait to work with the runtime. In addition,
+/// the dynamic library must export a `C` function `_create_plugin` that
+/// creates the implementation of the plugin.
+use anyhow::Result;
+use std::any::Any;
+
+pub trait SlipstreamPlugin: Any + Send + Sync + std::fmt::Debug {
+    /// Returns the name of the plugin.
+    fn name(&self) -> &'static str;
+
+    /// The callback called when a plugin is loaded by the system, used for
+    /// doing whatever initialization is required by the plugin. The
+    /// `_config_file` contains the name of the config file (JSON format) with
+    /// a `libpath` field indicating the full path of the shared library.
+    fn on_load(&mut self, _config_file: &str, _is_reload: bool) -> Result<()> {
+        Ok(())
+    }
+
+    /// The callback called right before a plugin is unloaded by the system.
+    /// Used for doing cleanup before unload.
+    fn on_unload(&mut self) {}
+
+    /// Called when a mapping key-value pair is inserted or updated during canonical finalize.
+    /// All arguments are serialized to bytes to keep the trait object-safe.
+    fn notify_mapping_update(
+        &self,
+        _program_id: &[u8],
+        _mapping_name: &[u8],
+        _key: &[u8],
+        _value: &[u8],
+        _block_height: u32,
+    ) -> Result<()> {
+        Ok(())
+    }
+
+    /// Called once per staker per block during staking reward distribution.
+    fn notify_staking_reward(
+        &self,
+        _staker: &[u8],
+        _validator: &[u8],
+        _reward: u64,
+        _new_stake: u64,
+        _block_height: u32,
+    ) -> Result<()> {
+        Ok(())
+    }
+
+    /// Returns `true` if the plugin is interested in general mapping update data.
+    fn history_enabled(&self) -> bool {
+        false
+    }
+
+    /// Returns `true` if the plugin is interested in staking reward data.
+    fn history_staking_rewards_enabled(&self) -> bool {
+        false
+    }
+}

plugins/slipstream_plugin_manager/Cargo.toml

@@ -0,0 +1,34 @@
+[package]
+name = "snarkvm-slipstream-plugin-manager"
+version = "4.5.0"
+authors = [ "The Aleo Team <hello@aleo.org>" ]
+description = "The SnarkVM Slipstream plugin manager."
+homepage = "https://aleo.org"
+repository = "https://github.com/ProvableHQ/snarkVM"
+license = "Apache-2.0"
+edition = "2024"
+
+[dependencies.snarkvm-slipstream-plugin-interface]
+workspace = true
+
+[dependencies.anyhow]
+workspace = true
+
+[dependencies.thiserror]
+workspace = true
+
+[dependencies.tracing]
+workspace = true
+
+[dependencies.serde_json]
+workspace = true
+
+[dependencies.libloading]
+version = "0.8"
+
+[dependencies.json5]
+version = "0.4"
+
+[dependencies.tokio]
+version = "1"
+features = [ "sync" ]