Proper PoT-based slot production with real proofs stored in the pre-d…

…igest, no verification yet

crates/pallet-subspace/src/lib.rs

@@ -199,6 +199,10 @@ mod pallet {
         // TODO: Remove when switching to PoT by default
         type GlobalRandomnessUpdateInterval: Get<Self::BlockNumber>;
 
+        /// The amount of time, in blocks, between updates of global randomness.
+        #[pallet::constant]
+        type BlockAuthoringDelay: Get<Slot>;
+
         /// The amount of time, in blocks, that each era should last.
         /// NOTE: Currently it is not possible to change the era duration after
         /// the chain has started. Attempting to do so will brick block production.
@@ -1081,6 +1085,8 @@ impl<T: Config> Pallet<T> {
             global_randomness_interval: T::GlobalRandomnessUpdateInterval::get()
                 .try_into()
                 .unwrap_or_else(|_| panic!("Block number always fits in BlockNumber; qed")),
+            #[cfg(feature = "pot")]
+            block_authoring_delay: T::BlockAuthoringDelay::get(),
             era_duration: T::EraDuration::get()
                 .try_into()
                 .unwrap_or_else(|_| panic!("Block number always fits in BlockNumber; qed")),

crates/pallet-subspace/src/mock.rs

@@ -45,7 +45,7 @@ use subspace_core_primitives::crypto::Scalar;
 use subspace_core_primitives::{
     ArchivedBlockProgress, ArchivedHistorySegment, Blake2b256Hash, HistorySize, LastArchivedBlock,
     Piece, PieceOffset, PublicKey, Randomness, RecordedHistorySegment, SegmentCommitment,
-    SegmentHeader, SegmentIndex, Solution, SolutionRange,
+    SegmentHeader, SegmentIndex, SlotNumber, Solution, SolutionRange,
 };
 use subspace_erasure_coding::ErasureCoding;
 use subspace_farmer_components::auditing::audit_sector;
@@ -159,6 +159,7 @@ pub const INITIAL_SOLUTION_RANGE: SolutionRange =
 parameter_types! {
     #[cfg(not(feature = "pot"))]
     pub const GlobalRandomnessUpdateInterval: u64 = 10;
+    pub const BlockAuthoringDelay: SlotNumber = 2;
     pub const EraDuration: u32 = 4;
     // 1GB
     pub const InitialSolutionRange: SolutionRange = INITIAL_SOLUTION_RANGE;
@@ -180,6 +181,7 @@ parameter_types! {
 impl Config for Test {
     type RuntimeEvent = RuntimeEvent;
     type GlobalRandomnessUpdateInterval = GlobalRandomnessUpdateInterval;
+    type BlockAuthoringDelay = BlockAuthoringDelay;
     type EraDuration = EraDuration;
     type InitialSolutionRange = InitialSolutionRange;
     type SlotProbability = SlotProbability;

crates/sc-consensus-subspace-rpc/Cargo.toml

@@ -17,6 +17,7 @@ async-oneshot = "0.5.0"
 futures = "0.3.28"
 futures-timer = "3.0.2"
 jsonrpsee = { version = "0.16.2", features = ["server", "macros"] }
+lru = "0.10.0"
 parity-scale-codec = "3.6.3"
 parking_lot = "0.12.1"
 sc-client-api = { version = "4.0.0-dev", git = "https://github.com/subspace/substrate", rev = "55c157cff49b638a59d81a9f971f0f9a66829c71" }

crates/sc-consensus-subspace-rpc/src/lib.rs

@@ -24,6 +24,7 @@ use jsonrpsee::core::{async_trait, Error as JsonRpseeError, RpcResult};
 use jsonrpsee::proc_macros::rpc;
 use jsonrpsee::types::{SubscriptionEmptyError, SubscriptionResult};
 use jsonrpsee::SubscriptionSink;
+use lru::LruCache;
 use parity_scale_codec::{Decode, Encode};
 use parking_lot::Mutex;
 use sc_client_api::{AuxStore, BlockBackend};
@@ -37,7 +38,6 @@ use sc_utils::mpsc::TracingUnboundedSender;
 use sp_api::{ApiError, ProvideRuntimeApi};
 use sp_blockchain::HeaderBackend;
 use sp_consensus::SyncOracle;
-use sp_consensus_slots::Slot;
 use sp_consensus_subspace::{FarmerPublicKey, FarmerSignature, SubspaceApi as SubspaceRuntimeApi};
 use sp_core::crypto::ByteArray;
 use sp_core::H256;
@@ -46,12 +46,13 @@ use sp_runtime::traits::Block as BlockT;
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use std::marker::PhantomData;
+use std::num::NonZeroUsize;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Weak};
 use std::time::Duration;
 use subspace_archiving::archiver::NewArchivedSegment;
 use subspace_core_primitives::crypto::kzg::Kzg;
-use subspace_core_primitives::{PieceIndex, SegmentHeader, SegmentIndex, Solution};
+use subspace_core_primitives::{PieceIndex, SegmentHeader, SegmentIndex, SlotNumber, Solution};
 use subspace_farmer_components::FarmerProtocolInfo;
 use subspace_networking::libp2p::Multiaddr;
 use subspace_rpc_primitives::{
@@ -134,12 +135,6 @@ struct ArchivedSegmentHeaderAcknowledgementSenders {
     senders: HashMap<u64, TracingUnboundedSender<()>>,
 }
 
-#[derive(Default)]
-struct SolutionResponseSenders {
-    current_slot: Slot,
-    senders: Vec<async_oneshot::Sender<SolutionResponse>>,
-}
-
 #[derive(Default)]
 struct BlockSignatureSenders {
     current_hash: H256,
@@ -206,7 +201,8 @@ where
     new_slot_notification_stream: SubspaceNotificationStream<NewSlotNotification>,
     reward_signing_notification_stream: SubspaceNotificationStream<RewardSigningNotification>,
     archived_segment_notification_stream: SubspaceNotificationStream<ArchivedSegmentNotification>,
-    solution_response_senders: Arc<Mutex<SolutionResponseSenders>>,
+    solution_response_senders:
+        Arc<Mutex<LruCache<SlotNumber, Vec<async_oneshot::Sender<SolutionResponse>>>>>,
     reward_signature_senders: Arc<Mutex<BlockSignatureSenders>>,
     dsn_bootstrap_nodes: Vec<Multiaddr>,
     segment_headers_store: SegmentHeadersStore<AS>,
@@ -230,18 +226,39 @@ where
 impl<Block, Client, SO, AS> SubspaceRpc<Block, Client, SO, AS>
 where
     Block: BlockT,
+    Client: ProvideRuntimeApi<Block> + HeaderBackend<Block>,
+    Client::Api: SubspaceRuntimeApi<Block, FarmerPublicKey>,
     SO: SyncOracle + Send + Sync + Clone + 'static,
     AS: AuxStore + Send + Sync + 'static,
 {
     /// Creates a new instance of the `SubspaceRpc` handler.
-    pub fn new(config: SubspaceRpcConfig<Client, SO, AS>) -> Self {
-        Self {
+    pub fn new(config: SubspaceRpcConfig<Client, SO, AS>) -> Result<Self, ApiError> {
+        #[cfg(feature = "pot")]
+        let best_hash = config.client.info().best_hash;
+        #[cfg(feature = "pot")]
+        let runtime_api = config.client.runtime_api();
+        #[cfg(feature = "pot")]
+        let chain_constants = runtime_api.chain_constants(best_hash)?;
+        #[cfg(feature = "pot")]
+        let block_authoring_delay = u64::from(chain_constants.block_authoring_delay());
+        #[cfg(feature = "pot")]
+        let block_authoring_delay = usize::try_from(block_authoring_delay)
+            .expect("Block authoring delay will never exceed usize on any platform; qed");
+        #[cfg(feature = "pot")]
+        let solution_response_senders_capacity =
+            NonZeroUsize::try_from(block_authoring_delay).unwrap_or(NonZeroUsize::MIN);
+        #[cfg(not(feature = "pot"))]
+        let solution_response_senders_capacity = NonZeroUsize::MIN;
+
+        Ok(Self {
             client: config.client,
             subscription_executor: config.subscription_executor,
             new_slot_notification_stream: config.new_slot_notification_stream,
             reward_signing_notification_stream: config.reward_signing_notification_stream,
             archived_segment_notification_stream: config.archived_segment_notification_stream,
-            solution_response_senders: Arc::default(),
+            solution_response_senders: Arc::new(Mutex::new(LruCache::new(
+                solution_response_senders_capacity,
+            ))),
             reward_signature_senders: Arc::default(),
             dsn_bootstrap_nodes: config.dsn_bootstrap_nodes,
             segment_headers_store: config.segment_headers_store,
@@ -252,7 +269,7 @@ where
             kzg: config.kzg,
             deny_unsafe: config.deny_unsafe,
             _block: PhantomData,
-        }
+        })
     }
 }
 
@@ -261,8 +278,8 @@ impl<Block, Client, SO, AS> SubspaceRpcApiServer for SubspaceRpc<Block, Client,
 where
     Block: BlockT,
     Client: ProvideRuntimeApi<Block>
-        + BlockBackend<Block>
         + HeaderBackend<Block>
+        + BlockBackend<Block>
         + Send
         + Sync
         + 'static,
@@ -318,10 +335,19 @@ where
 
         let mut solution_response_senders = solution_response_senders.lock();
 
-        if *solution_response_senders.current_slot == solution_response.slot_number {
-            if let Some(mut sender) = solution_response_senders.senders.pop() {
-                let _ = sender.send(solution_response);
-            }
+        let slot = solution_response.slot_number;
+
+        let success = solution_response_senders
+            .peek_mut(&slot)
+            .and_then(|senders| senders.pop())
+            .and_then(|mut sender| sender.send(solution_response).ok())
+            .is_some();
+
+        if !success {
+            warn!(
+                %slot,
+                "Solution was ignored, likely because farmer was too slow"
+            );
         }
 
         Ok(())
@@ -338,7 +364,7 @@ where
                 .map(move |new_slot_notification| {
                     let NewSlotNotification {
                         new_slot_info,
-                        solution_sender,
+                        mut solution_sender,
                     } = new_slot_notification;
 
                     // Only handle solution responses in case unsafe APIs are allowed
@@ -350,12 +376,9 @@ where
                         {
                             let mut solution_response_senders = solution_response_senders.lock();
 
-                            if solution_response_senders.current_slot != new_slot_info.slot {
-                                solution_response_senders.current_slot = new_slot_info.slot;
-                                solution_response_senders.senders.clear();
-                            }
-
-                            solution_response_senders.senders.push(response_sender);
+                            solution_response_senders
+                                .get_or_insert_mut(SlotNumber::from(new_slot_info.slot), Vec::new)
+                                .push(response_sender);
                         }
 
                         // Wait for solutions and transform proposed proof of space solutions into
@@ -371,10 +394,12 @@ where
                                     )
                                     .expect("Always correct length; qed");
 
+                                    let sector_index = solution.sector_index;
+
                                     let solution = Solution {
-                                        public_key,
+                                        public_key: public_key.clone(),
                                         reward_address,
-                                        sector_index: solution.sector_index,
+                                        sector_index,
                                         history_size: solution.history_size,
                                         piece_offset: solution.piece_offset,
                                         record_commitment: solution.record_commitment,
@@ -385,7 +410,15 @@ where
                                         proof_of_space: solution.proof_of_space,
                                     };
 
-                                    let _ = solution_sender.unbounded_send(solution);
+                                    if solution_sender.try_send(solution).is_err() {
+                                        warn!(
+                                            slot = %solution_response.slot_number,
+                                            %sector_index,
+                                            %public_key,
+                                            "Solution receiver is closed, likely because farmer \
+                                            was too slow"
+                                        );
+                                    }
                                 }
                             }
                         };

crates/sc-consensus-subspace/src/lib.rs

@@ -108,7 +108,7 @@ pub struct NewSlotNotification {
     /// New slot information.
     pub new_slot_info: NewSlotInfo,
     /// Sender that can be used to send solutions for the slot.
-    pub solution_sender: TracingUnboundedSender<Solution<FarmerPublicKey, FarmerPublicKey>>,
+    pub solution_sender: mpsc::Sender<Solution<FarmerPublicKey, FarmerPublicKey>>,
 }
 
 /// Notification with a hash that needs to be signed to receive reward and sender for signature.
@@ -349,12 +349,13 @@ where
 }
 
 /// Parameters for Subspace.
-pub struct SubspaceParams<B: BlockT, C, SC, E, I, SO, L, CIDP, BS, AS>
+pub struct SubspaceParams<Block, Client, SC, E, I, SO, L, CIDP, BS, AS>
 where
+    Block: BlockT,
     SO: SyncOracle + Send + Sync,
 {
     /// The client to use
-    pub client: Arc<C>,
+    pub client: Arc<Client>,
 
     /// The SelectChain Strategy
     pub select_chain: SC,
@@ -383,7 +384,7 @@ where
     pub backoff_authoring_blocks: Option<BS>,
 
     /// The source of timestamps for relative slots
-    pub subspace_link: SubspaceLink<B>,
+    pub subspace_link: SubspaceLink<Block>,
 
     /// Persistent storage of segment headers
     pub segment_headers_store: SegmentHeadersStore<AS>,
@@ -458,7 +459,7 @@ where
     BlockNumber: From<<<Block as BlockT>::Header as HeaderT>::Number>,
 {
     let worker = SubspaceSlotWorker {
-        client,
+        client: client.clone(),
         block_import,
         env,
         sync_oracle: sync_oracle.clone(),
@@ -471,6 +472,10 @@ where
         max_block_proposal_slot_portion,
         telemetry,
         segment_headers_store,
+        #[cfg(feature = "pot")]
+        pending_solutions: Default::default(),
+        #[cfg(feature = "pot")]
+        pot_checkpoints: Default::default(),
         _pos_table: PhantomData::<PosTable>,
     };
 
@@ -486,6 +491,7 @@ where
     #[cfg(feature = "pot")]
     let inner = sc_proof_of_time::start_slot_worker(
         subspace_link.slot_duration(),
+        client,
         select_chain,
         worker,
         sync_oracle,