market actor events

actors/market/src/emit.rs

@@ -0,0 +1,76 @@
+use fil_actors_runtime::runtime::Runtime;
+use fil_actors_runtime::{ActorError, EventBuilder};
+use fvm_shared::deal::DealID;
+use fvm_shared::ActorID;
+
+/// Indicates a deal has been published.
+pub fn deal_published(
+    rt: &impl Runtime,
+    client: ActorID,
+    provider: ActorID,
+    deal_id: DealID,
+) -> Result<(), ActorError> {
+    rt.emit_event(
+        &EventBuilder::new()
+            .typ("deal-published")
+            .with_parties(deal_id, client, provider)
+            .build()?,
+    )
+}
+
+/// Indicates a deal has been activated.
+pub fn deal_activated(
+    rt: &impl Runtime,
+    deal_id: DealID,
+    client: ActorID,
+    provider: ActorID,
+) -> Result<(), ActorError> {
+    rt.emit_event(
+        &EventBuilder::new()
+            .typ("deal-activated")
+            .with_parties(deal_id, client, provider)
+            .build()?,
+    )
+}
+
+/// Indicates a deal has been terminated.
+pub fn deal_terminated(
+    rt: &impl Runtime,
+    deal_id: DealID,
+    client: ActorID,
+    provider: ActorID,
+) -> Result<(), ActorError> {
+    rt.emit_event(
+        &EventBuilder::new()
+            .typ("deal-terminated")
+            .with_parties(deal_id, client, provider)
+            .build()?,
+    )
+}
+
+/// Indicates a deal has been completed successfully.
+pub fn deal_completed(
+    rt: &impl Runtime,
+    deal_id: DealID,
+    client: ActorID,
+    provider: ActorID,
+) -> Result<(), ActorError> {
+    rt.emit_event(
+        &EventBuilder::new()
+            .typ("deal-completed")
+            .with_parties(deal_id, client, provider)
+            .build()?,
+    )
+}
+
+trait WithParties {
+    fn with_parties(self, id: DealID, client: ActorID, provider: ActorID) -> EventBuilder;
+}
+
+impl WithParties for EventBuilder {
+    fn with_parties(self, id: DealID, client: ActorID, provider: ActorID) -> EventBuilder {
+        self.field_indexed("id", &id)
+            .field_indexed("client", &client)
+            .field_indexed("provider", &provider)
+    }
+}

actors/market/src/lib.rs

@@ -54,6 +54,7 @@ pub mod policy;
 pub mod testing;
 
 mod deal;
+pub mod emit;
 mod state;
 mod types;
 
@@ -482,6 +483,13 @@ impl Actor {
             .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(),
+                new_deal_ids[i],
+            )?;
         }
 
         Ok(PublishStorageDealsReturn { ids: new_deal_ids, valid_deals: valid_input_bf })
@@ -640,6 +648,16 @@ impl Actor {
                     verified_infos,
                     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();
             }
 
@@ -833,6 +851,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)?;
@@ -919,7 +943,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() {
@@ -958,6 +982,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/src/state.rs

@@ -740,7 +740,7 @@ impl State {
         state: &DealState,
         deal: &DealProposal,
         epoch: ChainEpoch,
-    ) -> Result<(TokenAmount, bool), ActorError>
+    ) -> Result<(TokenAmount, bool, bool), ActorError>
     where
         BS: Blockstore,
     {
@@ -759,7 +759,7 @@ impl State {
         // This would be the case that the first callback somehow triggers before it is scheduled to
         // This is expected not to be able to happen
         if deal.start_epoch > epoch {
-            return Ok((TokenAmount::zero(), false));
+            return Ok((TokenAmount::zero(), false, false));
         }
 
         let payment_end_epoch = if ever_slashed {
@@ -818,14 +818,14 @@ impl State {
             self.slash_balance(store, &deal.provider, &slashed, Reason::ProviderCollateral)
                 .context("slashing balance")?;
 
-            return Ok((slashed, true));
+            return Ok((slashed, true, false));
         }
 
         if epoch >= deal.end_epoch {
             self.process_deal_expired(store, deal, state)?;
-            return Ok((TokenAmount::zero(), true));
+            return Ok((TokenAmount::zero(), true, true));
         }
-        Ok((TokenAmount::zero(), false))
+        Ok((TokenAmount::zero(), false, false))
     }
 
     /// Deal start deadline elapsed without appearing in a proven sector.

actors/market/tests/activate_deal_failures.rs

@@ -131,6 +131,7 @@ fn assert_activation_failure(
             deal_ids: vec![deal_id],
         }],
         false,
+        vec![],
     )
     .unwrap();
     let res: BatchActivateDealsResult =

actors/market/tests/batch_activate_deals.rs

@@ -168,7 +168,8 @@ fn sectors_fail_and_succeed_independently_during_batch_activation() {
         },
     ];
 
-    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, vec![id_4]).unwrap();
     let res: BatchActivateDealsResult =
         res.unwrap().deserialize().expect("VerifyDealsForActivation failed!");
 
@@ -227,7 +228,8 @@ fn handles_sectors_empty_of_deals_gracefully() {
         SectorDeals { sector_number: 3, deal_ids: vec![], sector_type, 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, vec![id_1]).unwrap();
     let res: BatchActivateDealsResult =
         res.unwrap().deserialize().expect("VerifyDealsForActivation failed!");
 
@@ -273,7 +275,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, vec![]).unwrap();
     let res: BatchActivateDealsResult =
         res.unwrap().deserialize().expect("VerifyDealsForActivation failed!");
 
@@ -328,7 +330,8 @@ fn fails_to_activate_cross_sector_duplicate_deals() {
         },
     ];
 
-    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, vec![id_1, id_3])
+        .unwrap();
     let res: BatchActivateDealsResult =
         res.unwrap().deserialize().expect("VerifyDealsForActivation failed!");
 

actors/market/tests/cron_tick_deal_expiry.rs

@@ -183,6 +183,13 @@ fn expired_deal_should_unlock_the_remaining_client_and_provider_locked_balance_a
     let p_escrow = get_balance(&rt, &PROVIDER_ADDR).balance;
 
     // move the current epoch so that deal is expired
+    expect_emitted(
+        &rt,
+        "deal-completed",
+        deal_id,
+        CLIENT_ADDR.id().unwrap(),
+        PROVIDER_ADDR.id().unwrap(),
+    );
     rt.set_epoch(END_EPOCH + 1000);
     cron_tick(&rt);
 
@@ -220,6 +227,13 @@ fn all_payments_are_made_for_a_deal_client_withdraws_collateral_and_client_accou
 
     // move the current epoch so that deal is expired
     rt.set_epoch(END_EPOCH + 100);
+    expect_emitted(
+        &rt,
+        "deal-completed",
+        deal_id,
+        CLIENT_ADDR.id().unwrap(),
+        PROVIDER_ADDR.id().unwrap(),
+    );
     cron_tick(&rt);
     assert_eq!(deal_proposal.client_collateral, get_balance(&rt, &CLIENT_ADDR).balance);
 

actors/market/tests/cron_tick_deal_slashing.rs

@@ -136,7 +136,7 @@ fn deal_is_slashed_at_the_end_epoch_should_not_be_slashed_and_should_be_consider
     // as deal is considered to be expired.
 
     rt.set_epoch(END_EPOCH);
-    terminate_deals(&rt, PROVIDER_ADDR, &[SECTOR_NUMBER]);
+    terminate_deals_for(&rt, PROVIDER_ADDR, &[SECTOR_NUMBER], vec![]);
 
     // on the next cron tick, it will be processed as expired
     let current = END_EPOCH + 300;
@@ -365,7 +365,7 @@ fn regular_payments_till_deal_expires_and_then_we_attempt_to_slash_it_but_it_wil
     // as deal is considered to be expired.
     let duration = END_EPOCH - current;
     rt.set_epoch(END_EPOCH);
-    terminate_deals(&rt, PROVIDER_ADDR, &[SECTOR_NUMBER]);
+    terminate_deals_for(&rt, PROVIDER_ADDR, &[SECTOR_NUMBER], vec![]);
 
     // next epoch for cron schedule is endEpoch + 300 ->
     // setting epoch to higher than that will cause deal to be expired, payment will be made