feat(miner): externalise SectorOnChainInfo from AMT

Author: rvagg

actors/miner/src/sectors.rs

@@ -10,18 +10,21 @@ use fil_actors_runtime::{actor_error, ActorDowncast, ActorError, Array, AsActorE
 use fvm_ipld_amt::Error as AmtError;
 use fvm_ipld_bitfield::BitField;
 use fvm_ipld_blockstore::Blockstore;
+use fvm_ipld_encoding::CborStore;
 use fvm_shared::error::ExitCode;
 use fvm_shared::sector::SectorNumber;
+use multihash_codetable::Code;
 
 use super::SectorOnChainInfo;
 
 pub struct Sectors<'db, BS> {
-    pub amt: Array<'db, SectorOnChainInfo, BS>,
+    pub amt: Array<'db, Cid, BS>,
+    store: &'db BS,
 }
 
 impl<'db, BS: Blockstore> Sectors<'db, BS> {
     pub fn load(store: &'db BS, root: &Cid) -> Result<Self, AmtError> {
-        Ok(Self { amt: Array::load(root, store)? })
+        Ok(Self { amt: Array::load(root, store)?, store })
     }
 
     pub fn load_sector(
@@ -30,7 +33,7 @@ impl<'db, BS: Blockstore> Sectors<'db, BS> {
     ) -> Result<Vec<SectorOnChainInfo>, ActorError> {
         let mut sector_infos: Vec<SectorOnChainInfo> = Vec::new();
         for sector_number in sector_numbers.iter() {
-            let sector_on_chain = self
+            let c = self
                 .amt
                 .get(sector_number)
                 .map_err(|e| {
@@ -41,6 +44,16 @@ impl<'db, BS: Blockstore> Sectors<'db, BS> {
                 })?
                 .cloned()
                 .ok_or_else(|| actor_error!(not_found; "sector not found: {}", sector_number))?;
+            let sector_on_chain = self
+                .store
+                .get_cbor::<SectorOnChainInfo>(&c)
+                .map_err(|e| {
+                    e.downcast_default(
+                        ExitCode::USR_ILLEGAL_STATE,
+                        format!("failed to load sector {}", sector_number),
+                    )
+                })?
+                .ok_or_else(|| actor_error!(not_found; "sector not found: {}", sector_number))?;
             sector_infos.push(sector_on_chain);
         }
         Ok(sector_infos)
@@ -50,26 +63,52 @@ impl<'db, BS: Blockstore> Sectors<'db, BS> {
         &self,
         sector_number: SectorNumber,
     ) -> Result<Option<SectorOnChainInfo>, ActorError> {
-        Ok(self
-            .amt
-            .get(sector_number)
-            .with_context_code(ExitCode::USR_ILLEGAL_STATE, || {
-                format!("failed to get sector {}", sector_number)
-            })?
-            .cloned())
+        match self.amt.get(sector_number).with_context_code(ExitCode::USR_ILLEGAL_STATE, || {
+            format!("failed to get sector {}", sector_number)
+        })? {
+            Some(c) => match self.store.get_cbor::<SectorOnChainInfo>(c) {
+                Ok(Some(sector_info)) => Ok(Some(sector_info)),
+                Ok(None) => Ok(None),
+                Err(e) => Err(e.downcast_default(
+                    ExitCode::USR_ILLEGAL_STATE,
+                    format!("failed to load sector {}", sector_number),
+                )),
+            },
+            None => Ok(None),
+        }
+    }
+
+    pub fn for_each<F>(&self, mut f: F) -> anyhow::Result<()>
+    where
+        F: FnMut(SectorNumber, &SectorOnChainInfo) -> anyhow::Result<()>,
+    {
+        self.amt.for_each(|i, c| {
+            let sector_number = i as SectorNumber;
+            let sector_info = self
+                .store
+                .get_cbor::<SectorOnChainInfo>(c)
+                .map_err(|e| anyhow!(e.to_string()))?
+                .ok_or_else(|| {
+                    anyhow!("sector info not found for sector {}", sector_number)
+                })?;
+            f(sector_number, &sector_info)
+        })?;
+        Ok(())
     }
 
     pub fn store(&mut self, infos: Vec<SectorOnChainInfo>) -> anyhow::Result<()> {
         for info in infos {
             let sector_number = info.sector_number;
-
             if sector_number > MAX_SECTOR_NUMBER {
                 return Err(anyhow!("sector number {} out of range", info.sector_number));
             }
 
-            self.amt.set(sector_number, info).map_err(|e| {
-                e.downcast_wrap(format!("failed to store sector {}", sector_number))
-            })?;
+            match self.store.put_cbor(&info, Code::Blake2b256) {
+                Ok(c) => self.amt.set(sector_number, c).map_err(|e| {
+                    e.downcast_wrap(format!("failed to store sector {}", sector_number))
+                })?,
+                Err(e) => return Err(anyhow!("failed to store sector {}: {}", sector_number, e)),
+            }
         }
 
         Ok(())

actors/miner/src/state.rs

@@ -86,7 +86,7 @@ pub struct State {
     ///
     /// Sectors are removed from this AMT when the partition to which the
     /// sector belongs is compacted.
-    pub sectors: Cid, // Array, AMT[SectorNumber]SectorOnChainInfo (sparse)
+    pub sectors: Cid, // Array, AMT[SectorNumber]*SectorOnChainInfo (sparse)
 
     /// The first epoch in this miner's current proving period. This is the first epoch in which a PoSt for a
     /// partition at the miner's first deadline may arrive. Alternatively, it is after the last epoch at which
@@ -142,7 +142,7 @@ impl State {
                     )
                 })?;
         let empty_sectors_array =
-            Array::<SectorOnChainInfo, BS>::new_with_bit_width(store, SECTORS_AMT_BITWIDTH)
+            Array::<Cid, BS>::new_with_bit_width(store, SECTORS_AMT_BITWIDTH)
                 .flush()
                 .map_err(|e| {
                     e.downcast_default(
@@ -422,7 +422,7 @@ impl State {
         F: FnMut(&SectorOnChainInfo) -> anyhow::Result<()>,
     {
         let sectors = Sectors::load(store, &self.sectors)?;
-        sectors.amt.for_each(|_, v| f(v))?;
+        sectors.for_each(|_, v| f(v))?;
         Ok(())
     }
 

actors/miner/src/testing.rs

@@ -67,7 +67,7 @@ pub fn check_state_invariants<BS: Blockstore>(
     let mut all_sectors: BTreeMap<SectorNumber, SectorOnChainInfo> = BTreeMap::new();
     match Sectors::load(&store, &state.sectors) {
         Ok(sectors) => {
-            let ret = sectors.amt.for_each(|sector_number, sector| {
+            let ret = sectors.for_each(|sector_number, sector| {
                 all_sectors.insert(sector_number, sector.clone());
                 acc.require(
                     allocated_sectors.contains(&sector_number),

actors/miner/tests/util.rs

@@ -2174,7 +2174,6 @@ impl ActorHarness {
                             let skipped_proven = &to_skip - &part.unproven;
                             let mut skipped_proven_sector_infos = Vec::new();
                             sector_arr
-                                .amt
                                 .for_each(|i, sector| {
                                     if skipped_proven.get(i) {
                                         skipped_proven_sector_infos.push(sector.clone());
@@ -2188,7 +2187,6 @@ impl ActorHarness {
                             let new_proven = &part.unproven - &to_skip;
                             let mut new_proven_infos = Vec::new();
                             sector_arr
-                                .amt
                                 .for_each(|i, sector| {
                                     if new_proven.get(i) {
                                         new_proven_infos.push(sector.clone());