Refactor archiving code for incremental mapping generation

crates/pallet-subspace/src/mock.rs

@@ -287,11 +287,8 @@ pub fn create_archived_segment() -> &'static NewArchivedSegment {
 
         let mut block = vec![0u8; RecordedHistorySegment::SIZE];
         rand::thread_rng().fill(block.as_mut_slice());
-        archiver
-            .add_block(block, Default::default(), true)
-            .into_iter()
-            .next()
-            .unwrap()
+        let block_outcome = archiver.add_block(block, Default::default(), true);
+        block_outcome.archived_segments.into_iter().next().unwrap()
     })
 }
 

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

@@ -28,7 +28,8 @@ use jsonrpsee::{Extensions, PendingSubscriptionSink};
 use parking_lot::Mutex;
 use sc_client_api::{AuxStore, BlockBackend};
 use sc_consensus_subspace::archiver::{
-    recreate_genesis_segment, ArchivedSegmentNotification, SegmentHeadersStore,
+    recreate_genesis_segment, ArchivedMappingNotification, ArchivedSegmentNotification,
+    SegmentHeadersStore,
 };
 use sc_consensus_subspace::notification::SubspaceNotificationStream;
 use sc_consensus_subspace::slot_worker::{
@@ -234,6 +235,9 @@ where
     pub new_slot_notification_stream: SubspaceNotificationStream<NewSlotNotification>,
     /// Reward signing notification stream
     pub reward_signing_notification_stream: SubspaceNotificationStream<RewardSigningNotification>,
+    /// Archived mapping notification stream
+    pub archived_mapping_notification_stream:
+        SubspaceNotificationStream<ArchivedMappingNotification>,
     /// Archived segment notification stream
     pub archived_segment_notification_stream:
         SubspaceNotificationStream<ArchivedSegmentNotification>,
@@ -259,6 +263,7 @@ where
     subscription_executor: SubscriptionTaskExecutor,
     new_slot_notification_stream: SubspaceNotificationStream<NewSlotNotification>,
     reward_signing_notification_stream: SubspaceNotificationStream<RewardSigningNotification>,
+    archived_mapping_notification_stream: SubspaceNotificationStream<ArchivedMappingNotification>,
     archived_segment_notification_stream: SubspaceNotificationStream<ArchivedSegmentNotification>,
     #[allow(clippy::type_complexity)]
     solution_response_senders: Arc<Mutex<LruMap<SlotNumber, mpsc::Sender<Solution<PublicKey>>>>>,
@@ -316,6 +321,7 @@ where
             subscription_executor: config.subscription_executor,
             new_slot_notification_stream: config.new_slot_notification_stream,
             reward_signing_notification_stream: config.reward_signing_notification_stream,
+            archived_mapping_notification_stream: config.archived_mapping_notification_stream,
             archived_segment_notification_stream: config.archived_segment_notification_stream,
             solution_response_senders: Arc::new(Mutex::new(LruMap::new(ByLength::new(
                 solution_response_senders_capacity,
@@ -844,17 +850,11 @@ where
     fn subscribe_archived_object_mappings(&self, pending: PendingSubscriptionSink) {
         // TODO: deny unsafe subscriptions?
 
-        // The genesis segment isn't included in this stream. In other methods we recreate is as the first segment,
-        // but there aren't any mappings in it, so we don't need to recreate it as part of this subscription.
-
         let mapping_stream = self
-            .archived_segment_notification_stream
+            .archived_mapping_notification_stream
             .subscribe()
-            .flat_map(|archived_segment_notification| {
-                let objects = archived_segment_notification
-                    .archived_segment
-                    .global_object_mappings();
-
+            .flat_map(|archived_mapping_notification| {
+                let objects = archived_mapping_notification.object_mapping;
                 stream::iter(objects)
             })
             .ready_chunks(OBJECT_MAPPING_BATCH_SIZE)
@@ -904,12 +904,12 @@ where
         // The genesis segment isn't included in this stream, see
         // `subscribe_archived_object_mappings` for details.
         let mapping_stream = self
-            .archived_segment_notification_stream
+            .archived_mapping_notification_stream
             .subscribe()
-            .flat_map(move |archived_segment_notification| {
-                let objects = archived_segment_notification
-                    .archived_segment
-                    .global_object_mappings()
+            .flat_map(move |archived_mapping_notification| {
+                let objects = archived_mapping_notification
+                    .object_mapping
+                    .into_iter()
                     .filter(|object| hashes.remove(&object.hash))
                     .collect::<Vec<_>>();
 

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

@@ -76,7 +76,7 @@ use std::sync::atomic::{AtomicU16, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 use subspace_archiving::archiver::{Archiver, NewArchivedSegment};
-use subspace_core_primitives::objects::BlockObjectMapping;
+use subspace_core_primitives::objects::{BlockObjectMapping, GlobalObject};
 use subspace_core_primitives::segments::{RecordedHistorySegment, SegmentHeader, SegmentIndex};
 use subspace_core_primitives::{BlockNumber, PublicKey};
 use subspace_erasure_coding::ErasureCoding;
@@ -345,6 +345,17 @@ pub struct ArchivedSegmentNotification {
     pub acknowledgement_sender: TracingUnboundedSender<()>,
 }
 
+/// Notification with incrementally generated object mappings for a block (and any previous block
+/// continuation)
+#[derive(Debug, Clone)]
+pub struct ArchivedMappingNotification {
+    /// Incremental archived object mappings for a block (and any previous block continuation).
+    ///
+    /// The archived data won't be available in pieces until the entire segment is full and archived.
+    pub object_mapping: Vec<GlobalObject>,
+    // TODO: add an acknowledgement_sender for backpressure if needed
+}
+
 fn find_last_archived_block<Block, Client, AS>(
     client: &Client,
     segment_headers_store: &SegmentHeadersStore<AS>,
@@ -432,8 +443,11 @@ where
 
     let encoded_block = encode_block(signed_block);
 
-    let new_archived_segment = Archiver::new(kzg, erasure_coding)
-        .add_block(encoded_block, block_object_mappings, false)
+    // There are no mappings in the genesis block, so they can be ignored
+    let block_outcome =
+        Archiver::new(kzg, erasure_coding).add_block(encoded_block, block_object_mappings, false);
+    let new_archived_segment = block_outcome
+        .archived_segments
         .into_iter()
         .next()
         .expect("Genesis block always results in exactly one archived segment; qed");
@@ -671,9 +685,17 @@ where
                     encoded_block.len() as f32 / 1024.0
                 );
 
-                let archived_segments =
-                    archiver.add_block(encoded_block, block_object_mappings, false);
-                let new_segment_headers: Vec<SegmentHeader> = archived_segments
+                let block_outcome = archiver.add_block(encoded_block, block_object_mappings, false);
+                // RPC clients only want these mappings in full mapping mode
+                // TODO: turn this into a command-line argument named `--full-mapping`
+                if cfg!(feature = "full-archive") {
+                    send_archived_mapping_notification(
+                        &subspace_link.archived_mapping_notification_sender.clone(),
+                        block_outcome.object_mapping,
+                    );
+                }
+                let new_segment_headers: Vec<SegmentHeader> = block_outcome
+                    .archived_segments
                     .iter()
                     .map(|archived_segment| archived_segment.segment_header)
                     .collect();
@@ -753,6 +775,9 @@ fn finalize_block<Block, Backend, Client>(
 /// processing, which is necessary for ensuring that when the next block is imported, inherents will
 /// contain segment header of newly archived block (must happen exactly in the next block).
 ///
+/// When a block with object mappings is archived, notification ([`SubspaceLink::archived_mapping_notification_stream`])
+/// will be sent. If any mappings spill over to the next segment, they will be sent when the next block is archived.
+///
 /// Once segment header is archived, notification ([`SubspaceLink::archived_segment_notification_stream`])
 /// will be sent and archiver will be paused until all receivers have provided an acknowledgement
 /// for it.
@@ -811,6 +836,8 @@ where
         } = archiver;
         let (mut best_archived_block_hash, mut best_archived_block_number) = best_archived_block;
 
+        let archived_mapping_notification_sender =
+            subspace_link.archived_mapping_notification_sender.clone();
         let archived_segment_notification_sender =
             subspace_link.archived_segment_notification_sender.clone();
 
@@ -842,7 +869,7 @@ where
                 "Checking if block needs to be skipped"
             );
 
-            // TODO: replace this cfg! with a CLI option
+            // TODO: turn this into a command-line argument named `--full-mapping`
             let skip_last_archived_blocks = last_archived_block_number > block_number_to_archive
                 && !cfg!(feature = "full-archive");
             if best_archived_block_number >= block_number_to_archive || skip_last_archived_blocks {
@@ -902,6 +929,7 @@ where
                 &*client,
                 &sync_oracle,
                 telemetry.clone(),
+                archived_mapping_notification_sender.clone(),
                 archived_segment_notification_sender.clone(),
                 best_archived_block_hash,
                 block_number_to_archive,
@@ -921,6 +949,7 @@ async fn archive_block<Block, Backend, Client, AS, SO>(
     client: &Client,
     sync_oracle: &SubspaceSyncOracle<SO>,
     telemetry: Option<TelemetryHandle>,
+    archived_mapping_notification_sender: SubspaceNotificationSender<ArchivedMappingNotification>,
     archived_segment_notification_sender: SubspaceNotificationSender<ArchivedSegmentNotification>,
     best_archived_block_hash: Block::Hash,
     block_number_to_archive: NumberFor<Block>,
@@ -985,11 +1014,16 @@ where
     );
 
     let mut new_segment_headers = Vec::new();
-    for archived_segment in archiver.add_block(
+    let block_outcome = archiver.add_block(
         encoded_block,
         block_object_mappings,
         !sync_oracle.is_major_syncing(),
-    ) {
+    );
+    send_archived_mapping_notification(
+        &archived_mapping_notification_sender,
+        block_outcome.object_mapping,
+    );
+    for archived_segment in block_outcome.archived_segments {
         let segment_header = archived_segment.segment_header;
 
         segment_headers_store.add_segment_headers(slice::from_ref(&segment_header))?;
@@ -1031,6 +1065,15 @@ where
     Ok((block_hash_to_archive, block_number_to_archive))
 }
 
+fn send_archived_mapping_notification(
+    archived_mapping_notification_sender: &SubspaceNotificationSender<ArchivedMappingNotification>,
+    object_mapping: Vec<GlobalObject>,
+) {
+    let archived_mapping_notification = ArchivedMappingNotification { object_mapping };
+
+    archived_mapping_notification_sender.notify(move || archived_mapping_notification);
+}
+
 async fn send_archived_segment_notification(
     archived_segment_notification_sender: &SubspaceNotificationSender<ArchivedSegmentNotification>,
     archived_segment: NewArchivedSegment,

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

@@ -36,7 +36,7 @@ pub mod slot_worker;
 mod tests;
 pub mod verifier;
 
-use crate::archiver::ArchivedSegmentNotification;
+use crate::archiver::{ArchivedMappingNotification, ArchivedSegmentNotification};
 use crate::block_import::BlockImportingNotification;
 use crate::notification::{SubspaceNotificationSender, SubspaceNotificationStream};
 use crate::slot_worker::{NewSlotNotification, RewardSigningNotification};
@@ -52,6 +52,8 @@ pub struct SubspaceLink<Block: BlockT> {
     new_slot_notification_stream: SubspaceNotificationStream<NewSlotNotification>,
     reward_signing_notification_sender: SubspaceNotificationSender<RewardSigningNotification>,
     reward_signing_notification_stream: SubspaceNotificationStream<RewardSigningNotification>,
+    archived_mapping_notification_sender: SubspaceNotificationSender<ArchivedMappingNotification>,
+    archived_mapping_notification_stream: SubspaceNotificationStream<ArchivedMappingNotification>,
     archived_segment_notification_sender: SubspaceNotificationSender<ArchivedSegmentNotification>,
     archived_segment_notification_stream: SubspaceNotificationStream<ArchivedSegmentNotification>,
     block_importing_notification_sender:
@@ -70,6 +72,8 @@ impl<Block: BlockT> SubspaceLink<Block> {
             notification::channel("subspace_new_slot_notification_stream");
         let (reward_signing_notification_sender, reward_signing_notification_stream) =
             notification::channel("subspace_reward_signing_notification_stream");
+        let (archived_mapping_notification_sender, archived_mapping_notification_stream) =
+            notification::channel("subspace_archived_mapping_notification_stream");
         let (archived_segment_notification_sender, archived_segment_notification_stream) =
             notification::channel("subspace_archived_segment_notification_stream");
         let (block_importing_notification_sender, block_importing_notification_stream) =
@@ -80,6 +84,8 @@ impl<Block: BlockT> SubspaceLink<Block> {
             new_slot_notification_stream,
             reward_signing_notification_sender,
             reward_signing_notification_stream,
+            archived_mapping_notification_sender,
+            archived_mapping_notification_stream,
             archived_segment_notification_sender,
             archived_segment_notification_stream,
             block_importing_notification_sender,
@@ -103,6 +109,13 @@ impl<Block: BlockT> SubspaceLink<Block> {
         self.reward_signing_notification_stream.clone()
     }
 
+    /// Get stream with notifications about archived mappings
+    pub fn archived_mapping_notification_stream(
+        &self,
+    ) -> SubspaceNotificationStream<ArchivedMappingNotification> {
+        self.archived_mapping_notification_stream.clone()
+    }
+
     /// Get stream with notifications about archived segment creation
     pub fn archived_segment_notification_stream(
         &self,

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

@@ -452,8 +452,9 @@
 //     let mut archiver = Archiver::new(kzg).expect("Incorrect parameters for archiver");
 //
 //     let genesis_block = client.block(client.info().genesis_hash).unwrap().unwrap();
-//     archiver
-//         .add_block(genesis_block.encode(), BlockObjectMapping::default())
+//     let block_outcome = archiver.add_block(genesis_block.encode(), BlockObjectMapping::default(), true);
+//     block_outcome
+//         .archived_segments
 //         .into_iter()
 //         .map(|archived_segment| archived_segment.pieces)
 //         .collect()

crates/subspace-archiving/Cargo.toml

@@ -17,6 +17,7 @@ include = [
 bench = false
 
 [dependencies]
+hex = { version = "0.4.3", default-features = false, features = ["alloc"] }
 parity-scale-codec = { version = "3.6.12", default-features = false, features = ["derive"] }
 rayon = { version = "1.10.0", optional = true }
 serde = { version = "1.0.110", optional = true, features = ["derive"] }