Cache proof receipts across the proving pipeline

core/types/Cargo.toml

@@ -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/src/batch_metadata.rs

@@ -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
+    }
 }

examples/tn10-flow/src/daemon.rs

@@ -24,7 +24,7 @@ use vprogs_node_framework::{Node, NodeConfig};
 use vprogs_scheduling_scheduler::ExecutionConfig;
 use vprogs_storage_manager::StorageConfig;
 use vprogs_storage_rocksdb_store::RocksDbStore;
-use vprogs_zk_aggregate_prover::{AggregateProverConfig, ScheduledBundle};
+use vprogs_zk_aggregate_prover::{AggregateProverConfig, ScheduledBundle, SettlementArtifact};
 use vprogs_zk_backend_risc0_api::{Backend, ProofType, Receipt};
 use vprogs_zk_batch_prover::{BatchProverConfig, LaneProofRequest, LaneProofSource};
 use vprogs_zk_vm::{ProvingPipeline, Vm};
@@ -37,7 +37,7 @@ pub type V = Vm<Backend, Store>;
 pub type FlowNode = Node<Store, V>;
 /// Queue of bundle handles the aggregate prover publishes to the settlement worker (one per formed
 /// bundle).
-pub type FlowSettlementQueue = AsyncQueue<ScheduledBundle<Receipt>>;
+pub type FlowSettlementQueue = AsyncQueue<ScheduledBundle<SettlementArtifact<Receipt>>>;
 
 /// Everything the bridge needs to follow our lane on the remote node.
 pub struct BridgeParams {

l1/types/Cargo.toml

@@ -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/src/chain_block_metadata.rs

@@ -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.

node/test-utils/src/vm.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/vm/src/lib.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;
 }

scheduling/scheduler/Cargo.toml

@@ -20,11 +20,13 @@ vprogs-core-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]
 tempfile.workspace                     = true

scheduling/scheduler/src/lib.rs

@@ -26,11 +26,11 @@ pub use processor::Processor;
 pub use pruning_worker::PruningWorker;
 pub(crate) use resource::Resource;
 pub(crate) use resource_access::ResourceAccess;
-pub use scheduled_batch::{ScheduledBatch, ScheduledBatchRef};
+pub use scheduled_batch::{AggReceiptCoord, ScheduledBatch, ScheduledBatchRef};
 pub use scheduled_transaction::ScheduledTransaction;
 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/processor.rs

@@ -1,3 +1,4 @@
+use borsh::{BorshDeserialize, BorshSerialize};
 use vprogs_core_types::BatchMetadata;
 use vprogs_storage_types::Store;
 
@@ -24,13 +25,28 @@ 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 aggregate (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 run 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.

scheduling/scheduler/src/pruning_worker.rs

@@ -12,6 +12,7 @@ use tokio::{runtime::Builder, sync::Notify};
 use vprogs_core_types::{Checkpoint, ResourceId};
 use vprogs_state_batch_metadata::BatchMetadata as StoredBatchMetadata;
 use vprogs_state_metadata::StateMetadata;
+use vprogs_state_proof_receipt::{Prefix, invalidate_checkpoint};
 use vprogs_state_ptr_rollback::StatePtrRollback;
 use vprogs_state_version::StateVersion;
 use vprogs_storage_types::Store;
@@ -225,6 +226,15 @@ impl<S: Store, P: Processor<S>> PruningWorker<S, P> {
                 // Delete batch metadata entries for this batch.
                 StoredBatchMetadata::delete(wb, index);
 
+                // Drop every cached proof receipt at this checkpoint index, across all programs
+                // and all competing forks. A pruned batch can never be rolled back to, so no
+                // future replay (even a flip reorg) needs its receipts again.
+                invalidate_checkpoint(
+                    store.as_ref(),
+                    wb,
+                    &Prefix { checkpoint_index: index.into() },
+                );
+
                 // Prune stale SMT nodes for this version.
                 store.prune(wb, index);
             }