Cache proof receipts across the proving pipeline

Wire the proof-receipt store into the transaction, batch, and aggregate
provers so a replay (including a flip reorg back onto the same fork) reuses
a cached receipt instead of re-proving.

- Processor gains tx/batch/aggregator image-id accessors and an
  AggregatorArtifact type; artifacts are bounded Clone + Borsh so the
  scheduler can cache and restore them.
- ScheduledTransaction and ScheduledBatch read/write their per-tx and
  per-batch receipts through the storage workers; ScheduledBundle owns the
  aggregate (settlement) receipt, keyed off its start coordinate with a
  batch as the storage gateway.
- Receipts are keyed by (checkpoint, block_hash, image_id, ...) so
  competing forks stay distinct; the pruning worker drops a checkpoint's
  receipts across all forks and programs in one prefix scan.
- The proving workers prove on a cache miss, wait for the receipt to be
  durable, then publish the artifact; the transaction worker proves inline
  so only one proof occupies the GPU at a time.

ChainBlockMetadata supplies the block hash; a test-utils-gated
BatchMetadata for u64 stands in for tests. Round-trips the tx/batch/agg
receipt paths through the storage workers in a new scheduler test.

Author: biryukovmaxim

Cargo.lock

@@ -8,3 +8,7 @@ borsh             = { workspace = true, default-features = false }
 rkyv              = { workspace = true, default-features = false }
 vprogs-core-codec = { workspace = true }
 zerocopy          = { workspace = true, features = ["derive"] }
+
+[features]
+# Stand-in trait impls for tests only (e.g. `BatchMetadata for u64`); never enabled by the node.
+test-utils = []

core/types/Cargo.toml

@@ -4,14 +4,27 @@ use borsh::{BorshDeserialize, BorshSerialize};
 
 /// Opaque metadata attached to each scheduler batch, supporting serialization.
 ///
-/// Implementors derive `BorshSerialize` and `BorshDeserialize`; the trait is implemented
-/// automatically via a blanket impl.
+/// Implementors derive `BorshSerialize` and `BorshDeserialize` and supply the chain block hash
+/// the batch was formed against via [`block_hash`](Self::block_hash), which keys the batch's
+/// proof receipts in the proof-receipt store.
 pub trait BatchMetadata:
     BorshSerialize + BorshDeserialize + Clone + Debug + Default + Send + Sync + 'static
 {
+    /// The chain block hash this batch was formed against, as raw bytes.
+    ///
+    /// The proving workers fold it into each receipt's cache key so the same checkpoint index on
+    /// two competing chains yields distinct keys.
+    fn block_hash(&self) -> [u8; 32];
 }
 
-impl<T> BatchMetadata for T where
-    T: BorshSerialize + BorshDeserialize + Clone + Debug + Default + Send + Sync + 'static
-{
+/// A `u64` batch index doubles as minimal test metadata: its big-endian bytes stand in for a
+/// block hash, keeping per-index receipts distinct without a real chain. Gated behind `test-utils`
+/// so the node never treats a bare index as batch metadata.
+#[cfg(feature = "test-utils")]
+impl BatchMetadata for u64 {
+    fn block_hash(&self) -> [u8; 32] {
+        let mut bytes = [0u8; 32];
+        bytes[..8].copy_from_slice(&self.to_be_bytes());
+        bytes
+    }
 }

core/types/src/batch_metadata.rs

@@ -10,4 +10,5 @@ kaspa-hashes.workspace         = true
 kaspa-rpc-core.workspace       = true
 kaspa-txscript.workspace       = true
 kaspa-wrpc-client.workspace    = true
+vprogs-core-types.workspace    = true
 zerocopy.workspace             = true

l1/types/Cargo.toml

@@ -1,5 +1,6 @@
 use borsh::{BorshDeserialize, BorshSerialize};
 use kaspa_rpc_core::{RpcHeader, RpcOptionalHeader};
+use vprogs_core_types::BatchMetadata;
 
 use crate::{Hash, SettlementInfo};
 
@@ -39,6 +40,12 @@ pub struct ChainBlockMetadata {
     pub last_settlement: Option<SettlementInfo>,
 }
 
+impl BatchMetadata for ChainBlockMetadata {
+    fn block_hash(&self) -> [u8; 32] {
+        self.hash.as_bytes()
+    }
+}
+
 impl From<&RpcHeader> for ChainBlockMetadata {
     /// Builds metadata from a regular RPC header, populating only the header-derived fields and
     /// leaving lane / parent-derived state at its default.

l1/types/src/chain_block_metadata.rs

@@ -22,9 +22,23 @@ impl<S: Store> Processor<S> for TestNodeVm {
         Ok(())
     }
 
+    // This test VM does not prove, so its receipt-cache image ids are unset.
+    fn tx_image_id(&self) -> [u8; 32] {
+        [0u8; 32]
+    }
+
+    fn batch_image_id(&self) -> [u8; 32] {
+        [0u8; 32]
+    }
+
+    fn aggregator_image_id(&self) -> [u8; 32] {
+        [0u8; 32]
+    }
+
     type Transaction = L1Transaction;
     type TransactionArtifact = ();
     type BatchArtifact = ();
+    type AggregatorArtifact = ();
     type BatchMetadata = ChainBlockMetadata;
     type Error = ();
 }

node/test-utils/src/vm.rs

@@ -18,9 +18,23 @@ impl<S: Store> Processor<S> for VM {
         todo!("transaction execution from SchedulerTransaction<L1Transaction>")
     }
 
+    // The node VM does not yet drive proving, so its receipt-cache image ids are unset.
+    fn tx_image_id(&self) -> [u8; 32] {
+        [0u8; 32]
+    }
+
+    fn batch_image_id(&self) -> [u8; 32] {
+        [0u8; 32]
+    }
+
+    fn aggregator_image_id(&self) -> [u8; 32] {
+        [0u8; 32]
+    }
+
     type Transaction = L1Transaction;
     type TransactionArtifact = TransactionEffects;
     type BatchArtifact = ();
+    type AggregatorArtifact = ();
     type BatchMetadata = ChainBlockMetadata;
     type Error = VmError;
 }

node/vm/src/lib.rs

@@ -22,13 +22,16 @@ vprogs-l1-types.workspace                     = true
 vprogs-scheduling-execution-workers.workspace = true
 vprogs-state-batch-metadata.workspace         = true
 vprogs-state-metadata.workspace               = true
+vprogs-state-proof-receipt.workspace          = true
 vprogs-state-ptr-latest.workspace             = true
 vprogs-state-ptr-rollback.workspace           = true
 vprogs-state-version.workspace                = true
 vprogs-storage-manager.workspace              = true
 vprogs-storage-types.workspace                = true
+zerocopy.workspace                            = true
 
 [dev-dependencies]
+kaspa-hashes.workspace                 = true
 tempfile.workspace                     = true
 vprogs-core-test-utils.workspace       = true
 vprogs-core-types.workspace            = true

scheduling/scheduler/Cargo.toml

@@ -34,5 +34,5 @@ pub(crate) use scheduled_transaction::ScheduledTransactionRef;
 pub use scheduler::Scheduler;
 pub use state::SchedulerState;
 pub use state_diff::{StateDiff, StateDiffRef};
-pub use storage_cmd::{Read, Write};
+pub use storage_cmd::{Read, ReadReceipt, ReceiptRead, ReceiptValue, StoreReceipt, Write};
 pub use transaction_context::TransactionContext;

scheduling/scheduler/src/lib.rs

@@ -1,3 +1,4 @@
+use borsh::{BorshDeserialize, BorshSerialize};
 use vprogs_core_types::BatchMetadata;
 use vprogs_storage_types::Store;
 
@@ -24,13 +25,29 @@ pub trait Processor<S: Store>: Clone + Sized + Send + Sync + 'static {
         // Default implementation does nothing (override if needed).
     }
 
+    /// Program identifier keying a per-transaction receipt in the proof-receipt store.
+    fn tx_image_id(&self) -> [u8; 32];
+
+    /// Program identifier keying a per-batch receipt in the proof-receipt store.
+    fn batch_image_id(&self) -> [u8; 32];
+
+    /// Program identifier keying an aggregated-bundle (settlement) receipt in the proof-receipt
+    /// store.
+    fn aggregator_image_id(&self) -> [u8; 32];
+
     /// The transaction payload type (e.g. kaspa `L1Transaction`, `usize` in tests).
     /// The scheduler wraps it in `SchedulerTransaction<Self::Transaction>`.
     type Transaction: Send + Sync + 'static;
     /// Artifact produced by a processed transaction ([`ScheduledTransaction::publish_artifact`]).
-    type TransactionArtifact: Send + Sync + 'static;
-    /// Artifact produced by a processed batch ([`ScheduledBatch::publish_artifact`]).
-    type BatchArtifact: Send + Sync + 'static;
+    /// The `Borsh` bounds let the scheduler cache it in (and restore it from) the proof-receipt
+    /// store; `Clone` lets a served lookup hand the cached value back out of its shared slot.
+    type TransactionArtifact: Clone + BorshSerialize + BorshDeserialize + Send + Sync + 'static;
+    /// Artifact produced by a processed batch ([`ScheduledBatch::publish_artifact`]). Bounded for
+    /// the same proof-receipt caching as [`TransactionArtifact`](Self::TransactionArtifact).
+    type BatchArtifact: Clone + BorshSerialize + BorshDeserialize + Send + Sync + 'static;
+    /// Receipt produced by aggregating a bundle of per-batch receipts into one settlement receipt.
+    /// Bounded for the same proof-receipt caching as the other artifacts.
+    type AggregatorArtifact: Clone + BorshSerialize + BorshDeserialize + Send + Sync + 'static;
     /// Opaque metadata attached to each batch for persistence.
     type BatchMetadata: BatchMetadata;
     /// Error type returned when transaction processing fails.