changes

actors/market/src/lib.rs

@@ -54,6 +54,7 @@ pub mod policy;
 pub mod testing;
 
 mod deal;
+mod emit;
 mod state;
 mod types;
 
@@ -469,19 +470,26 @@ impl Actor {
         // notify clients, any failures cause the entire publish_storage_deals method to fail
         // it's unsafe to ignore errors here, since that could be used to attack storage contract clients
         // that might be unaware they're making storage deals
-        for (i, valid_deal) in valid_deals.iter().enumerate() {
+        for (valid_deal, &deal_id) in valid_deals.iter().zip(&new_deal_ids) {
             _ = extract_send_result(rt.send_simple(
                 &valid_deal.proposal.client,
                 MARKET_NOTIFY_DEAL_METHOD,
                 IpldBlock::serialize_cbor(&MarketNotifyDealParams {
                     proposal: valid_deal.serialized_proposal.to_vec(),
-                    deal_id: new_deal_ids[i],
+                    deal_id,
                 })?,
                 TokenAmount::zero(),
             ))
             .with_context_code(ExitCode::USR_ILLEGAL_ARGUMENT, || {
                 format!("failed to notify deal with proposal cid {}", valid_deal.cid)
             })?;
+
+            emit::deal_published(
+                rt,
+                valid_deal.proposal.client.id().unwrap(),
+                valid_deal.proposal.provider.id().unwrap(),
+                deal_id,
+            )?;
         }
 
         Ok(PublishStorageDealsReturn { ids: new_deal_ids, valid_deals: valid_input_bf })
@@ -554,7 +562,7 @@ impl Actor {
             let mut deal_states: Vec<(DealID, DealState)> = vec![];
             let mut batch_gen = BatchReturnGen::new(params.sectors.len());
             let mut activations: Vec<SectorDealActivation> = vec![];
-            let mut activated_deals = BTreeSet::<DealID>::new();
+            let mut activated_deals: HashSet<DealID> = HashSet::new();
             let mut sectors_deals: Vec<(SectorNumber, SectorDealIDs)> = vec![];
 
             'sector: for sector in params.sectors {
@@ -597,8 +605,7 @@ impl Actor {
                     validated_proposals.push(proposal);
                 }
 
-                let mut verified_infos = vec![];
-                let mut nonverified_deal_space = BigInt::zero();
+                let mut activated = vec![];
                 // Given that all deals validated, prepare the state updates for them all.
                 // There's no continue below here to ensure updates are consistent.
                 // Any error must abort.
@@ -608,16 +615,12 @@ impl Actor {
                     let alloc_id =
                         pending_deal_allocation_ids.delete(deal_id)?.unwrap_or(NO_ALLOCATION_ID);
 
-                    if alloc_id != NO_ALLOCATION_ID {
-                        verified_infos.push(VerifiedDealInfo {
-                            client: proposal.client.id().unwrap(),
-                            allocation_id: alloc_id,
-                            data: proposal.piece_cid,
-                            size: proposal.piece_size,
-                        })
-                    } else {
-                        nonverified_deal_space += proposal.piece_size.0;
-                    }
+                    activated.push(ActivatedDeal {
+                        client: proposal.client.id().unwrap(),
+                        allocation_id: alloc_id,
+                        data: proposal.piece_cid,
+                        size: proposal.piece_size,
+                    });
 
                     // Prepare initial deal state.
                     deal_states.push((
@@ -639,11 +642,17 @@ impl Actor {
 
                 sectors_deals
                     .push((sector.sector_number, SectorDealIDs { deals: sector.deal_ids.clone() }));
-                activations.push(SectorDealActivation {
-                    nonverified_deal_space,
-                    verified_infos,
-                    unsealed_cid: data_commitment,
-                });
+                activations.push(SectorDealActivation { activated, unsealed_cid: data_commitment });
+
+                for (deal_id, proposal) in sector.deal_ids.iter().zip(&validated_proposals) {
+                    emit::deal_activated(
+                        rt,
+                        *deal_id,
+                        proposal.client.id().unwrap(),
+                        proposal.provider.id().unwrap(),
+                    )?;
+                }
+
                 batch_gen.add_success();
             }
 
@@ -830,6 +839,12 @@ impl Actor {
                 state.slash_epoch = params.epoch;
 
                 deal_states.push((id, state));
+                emit::deal_terminated(
+                    rt,
+                    id,
+                    deal.client.id().unwrap(),
+                    deal.provider.id().unwrap(),
+                )?;
             }
 
             st.put_deal_states(rt.store(), &deal_states)?;
@@ -916,7 +931,7 @@ impl Actor {
                         })?;
                     }
 
-                    let (slash_amount, remove_deal) =
+                    let (slash_amount, remove_deal, complete_success) =
                         st.process_deal_update(rt.store(), &state, &deal, curr_epoch)?;
 
                     if slash_amount.is_negative() {
@@ -957,6 +972,15 @@ impl Actor {
                                 "failed to delete deal proposal: does not exist"
                             ));
                         }
+
+                        if complete_success {
+                            emit::deal_completed(
+                                rt,
+                                deal_id,
+                                deal.client.id().unwrap(),
+                                deal.provider.id().unwrap(),
+                            )?;
+                        }
                     } else {
                         if !slash_amount.is_zero() {
                             return Err(actor_error!(

actors/market/tests/batch_activate_deals.rs

@@ -273,7 +273,7 @@ fn fails_to_activate_single_sector_duplicate_deals() {
             sector_expiry: END_EPOCH,
         },
     ];
-    let res = batch_activate_deals_raw(&rt, PROVIDER_ADDR, sectors_deals, false).unwrap();
+    let res = batch_activate_deals_raw(&rt, PROVIDER_ADDR, sectors_deals, false, &[]).unwrap();
     let res: BatchActivateDealsResult =
         res.unwrap().deserialize().expect("VerifyDealsForActivation failed!");
 

actors/market/tests/on_miner_sectors_terminate.rs

@@ -290,7 +290,7 @@ fn terminating_a_deal_the_second_time_does_not_change_its_slash_epoch() {
 
     // set a new epoch and terminate again -> however slash epoch will still be the old epoch.
     rt.set_epoch(current_epoch + 1);
-    terminate_deals(&rt, PROVIDER_ADDR, &[sector_number],  &[]);
+    terminate_deals(&rt, PROVIDER_ADDR, &[sector_number], &[]);
     let s = get_deal_state(&rt, deal1);
     assert_eq!(s.slash_epoch, current_epoch);
     check_state(&rt);

actors/market/tests/verify_deals_for_activation_test.rs

@@ -8,9 +8,11 @@ use fvm_shared::econ::TokenAmount;
 use fvm_shared::error::ExitCode;
 use fvm_shared::piece::PieceInfo;
 use fvm_shared::sector::RegisteredSealProof;
-use num_traits::Zero;
 
-use fil_actor_market::{Actor as MarketActor, Method, SectorDeals, VerifyDealsForActivationParams};
+use fil_actor_market::{
+    ActivatedDeal, Actor as MarketActor, Method, SectorDeals, VerifyDealsForActivationParams,
+    NO_ALLOCATION_ID,
+};
 use fil_actors_runtime::runtime::builtins::Type;
 use fil_actors_runtime::test_utils::{
     expect_abort, expect_abort_contains_message, make_piece_cid, ACCOUNT_ACTOR_CODE_ID,
@@ -133,7 +135,7 @@ fn verification_and_weights_for_verified_and_unverified_deals() {
     assert_eq!(4, deal_ids.len());
 
     let sector_number = 7;
-    let response = verify_deals_for_activation(
+    verify_deals_for_activation(
         &rt,
         PROVIDER_ADDR,
         vec![SectorDeals {
@@ -152,7 +154,6 @@ fn verification_and_weights_for_verified_and_unverified_deals() {
         },
     );
 
-
     let a_response =
         activate_deals(&rt, SECTOR_EXPIRY, PROVIDER_ADDR, CURR_EPOCH, sector_number, &deal_ids);
     let s_response = a_response.activations.get(0).unwrap();