Merge pull request #3114 from steemit/2718-event-driven-automated-act…

…ions

Event Driven Automated Actions

libraries/chain/database.cpp

@@ -919,29 +919,46 @@ struct action_validate_visitor
    }
 };
 
-void database::push_required_action( const required_automated_action& a )
+void database::push_required_action( const required_automated_action& a, time_point_sec execution_time )
 {
+   FC_ASSERT( execution_time >= head_block_time(), "Cannot push required action to execute in the past. head_block_time: ${h} execution_time: ${e}",
+      ("h", head_block_time())("e", execution_time) );
+
    static const action_validate_visitor validate_visitor;
    a.visit( validate_visitor );
 
    create< pending_required_action_object >( [&]( pending_required_action_object& pending_action )
    {
       pending_action.action = a;
+      pending_action.execution_time = execution_time;
    });
 }
 
-void database::push_optional_action( const optional_automated_action& a )
+void database::push_required_action( const required_automated_action& a )
 {
+   push_required_action( a, head_block_time() );
+}
+
+void database::push_optional_action( const optional_automated_action& a, time_point_sec execution_time )
+{
+   FC_ASSERT( execution_time >= head_block_time(), "Cannot push optional action to execute in the past. head_block_time: ${h} execution_time: ${e}",
+      ("h", head_block_time())("e", execution_time) );
+
    static const action_validate_visitor validate_visitor;
    a.visit( validate_visitor );
 
    create< pending_optional_action_object >( [&]( pending_optional_action_object& pending_action )
    {
       pending_action.action = a;
-      pending_action.pushed_block_num = head_block_num();
+      pending_action.execution_time = execution_time;
    });
 }
 
+void database::push_optional_action( const optional_automated_action& a )
+{
+   push_optional_action( a, head_block_time() );
+}
+
 void database::notify_pre_apply_required_action( const required_action_notification& note )
 {
    STEEM_TRY_NOTIFY( _pre_apply_required_action_signal, note );
@@ -3037,9 +3054,6 @@ void database::_apply_block( const signed_block& next_block )
       );
    }
 
-   process_required_actions( req_actions );
-   process_optional_actions( opt_actions );
-
    for( const auto& trx : next_block.transactions )
    {
       /* We do not need to push the undo state for each transaction
@@ -3087,6 +3101,9 @@ void database::_apply_block( const signed_block& next_block )
    generate_required_actions();
    generate_optional_actions();
 
+   process_required_actions( req_actions );
+   process_optional_actions( opt_actions );
+
    process_hardforks();
 
    // notify observers that the block has been applied
@@ -3386,6 +3403,7 @@ void database::process_required_actions( const required_automated_actions& actio
          // We're done processing actions in the block.
          if( pending_itr != pending_action_idx.end() )
          {
+            FC_ASSERT( pending_itr->execution_time > head_block_time(), "Expected action was not included in block. ${a}", ("a", pending_itr->action) );
             FC_TODO("Check that the block producer stopped including required actions for a good reason, such as running out of space #2722");
          }
          break;
@@ -3424,20 +3442,29 @@ void database::process_optional_actions( const optional_automated_actions& actio
    {
       actions_itr->visit( validate_visitor );
 
+      // There is no execution check because we don't have a good way of indexing into local
+      // optional actions from those contained in a block. It is the responsibility of the
+      // action evaluator to prevent early execution.
       apply_optional_action( *actions_itr );
    }
 
    // Clear out "expired" optional_actions. If the block when an optional action was generated
    // has become irreversible then a super majority of witnesses have chosen to not include it
    // and it is safe to delete.
-   const auto& pending_action_idx = get_index< pending_optional_action_index, by_id >();
+   const auto& pending_action_idx = get_index< pending_optional_action_index, by_execution >();
    auto pending_itr = pending_action_idx.begin();
-   auto lib = get_dynamic_global_properties().last_irreversible_block_num;
+   auto lib = fetch_block_by_number( get_dynamic_global_properties().last_irreversible_block_num );
 
-   while( pending_itr != pending_action_idx.end() && pending_itr->pushed_block_num <= lib )
+   // This is always valid when running on mainnet because there are irreversible blocks
+   // Testnet and unit tests, not so much. Could be ifdeffed with IS_TEST_NET, but seems
+   // like a reasonable check and will be optimized via speculative execution.
+   if( lib.valid() )
    {
-      remove( *pending_itr );
-      pending_itr = pending_action_idx.begin();
+      while( pending_itr != pending_action_idx.end() && pending_itr->execution_time <= lib->timestamp )
+      {
+         remove( *pending_itr );
+         pending_itr = pending_action_idx.begin();
+      }
    }
 }
 

libraries/chain/include/steem/chain/database.hpp

@@ -236,7 +236,14 @@ namespace steem { namespace chain {
          void pre_push_virtual_operation( const operation& op );
          void post_push_virtual_operation( const operation& op );
 
+         /*
+          * Pushing an action without specifying an execution time will execute at head block.
+          * The execution time must be greater than or equal to head block.
+          */
+         void push_required_action( const required_automated_action& a, time_point_sec execution_time );
          void push_required_action( const required_automated_action& a );
+
+         void push_optional_action( const optional_automated_action& a, time_point_sec execution_time );
          void push_optional_action( const optional_automated_action& a );
 
          void notify_pre_apply_required_action( const required_action_notification& note );

libraries/chain/include/steem/chain/pending_optional_action_object.hpp

@@ -20,22 +20,28 @@ class pending_optional_action_object : public object< pending_optional_action_ob
          c( *this );
       }
 
-      id_type           id;
+      id_type                    id;
 
+      time_point_sec             execution_time;
       optional_automated_action  action;
-      uint32_t                   pushed_block_num = 0;
 };
 
 typedef multi_index_container<
    pending_optional_action_object,
    indexed_by<
-      ordered_unique< tag< by_id >, member< pending_optional_action_object, pending_optional_action_id_type, &pending_optional_action_object::id > >
+      ordered_unique< tag< by_id >, member< pending_optional_action_object, pending_optional_action_id_type, &pending_optional_action_object::id > >,
+      ordered_unique< tag< by_execution >,
+         composite_key< pending_optional_action_object,
+            member< pending_optional_action_object, time_point_sec, &pending_optional_action_object::execution_time >,
+            member< pending_optional_action_object, pending_optional_action_id_type, &pending_optional_action_object::id >
+         >
+      >
    >,
    allocator< pending_optional_action_object >
 > pending_optional_action_index;
 
 } } //steem::chain
 
 FC_REFLECT( steem::chain::pending_optional_action_object,
-            (id)(action)(pushed_block_num) )
+            (id)(execution_time)(action) )
 CHAINBASE_SET_INDEX_TYPE( steem::chain::pending_optional_action_object, steem::chain::pending_optional_action_index )

libraries/chain/include/steem/chain/pending_required_action_object.hpp

@@ -20,21 +20,30 @@ class pending_required_action_object : public object< pending_required_action_ob
          c( *this );
       }
 
-      id_type           id;
+      id_type                    id;
 
+      time_point_sec             execution_time;
       required_automated_action  action;
 };
 
+struct by_execution;
+
 typedef multi_index_container<
    pending_required_action_object,
    indexed_by<
-      ordered_unique< tag< by_id >, member< pending_required_action_object, pending_required_action_id_type, &pending_required_action_object::id > >
+      ordered_unique< tag< by_id >, member< pending_required_action_object, pending_required_action_id_type, &pending_required_action_object::id > >,
+      ordered_unique< tag< by_execution >,
+         composite_key< pending_required_action_object,
+            member< pending_required_action_object, time_point_sec, &pending_required_action_object::execution_time >,
+            member< pending_required_action_object, pending_required_action_id_type, &pending_required_action_object::id >
+         >
+      >
    >,
    allocator< pending_required_action_object >
 > pending_required_action_index;
 
 } } //steem::chain
 
 FC_REFLECT( steem::chain::pending_required_action_object,
-            (id)(action) )
+            (id)(execution_time)(action) )
 CHAINBASE_SET_INDEX_TYPE( steem::chain::pending_required_action_object, steem::chain::pending_required_action_index )

libraries/plugins/witness/block_producer.cpp

@@ -172,11 +172,11 @@ void block_producer::apply_pending_transactions(
       wlog( "Postponed ${n} transactions due to block size limit", ("n", postponed_tx_count) );
    }
 
-   const auto& pending_required_action_idx = _db.get_index< chain::pending_required_action_index, chain::by_id >();
+   const auto& pending_required_action_idx = _db.get_index< chain::pending_required_action_index, chain::by_execution >();
    auto pending_required_itr = pending_required_action_idx.begin();
    chain::required_automated_actions required_actions;
 
-   while( pending_required_itr != pending_required_action_idx.end() )
+   while( pending_required_itr != pending_required_action_idx.end() && pending_required_itr->execution_time <= when )
    {
       uint64_t new_total_size = total_block_size + fc::raw::pack_size( pending_required_itr->action );
 
@@ -203,11 +203,11 @@ void block_producer::apply_pending_transactions(
       pending_block.extensions.insert( required_actions );
    }
 
-   const auto& pending_optional_action_idx = _db.get_index< chain::pending_optional_action_index, chain::by_id >();
+   const auto& pending_optional_action_idx = _db.get_index< chain::pending_optional_action_index, chain::by_execution >();
    auto pending_optional_itr = pending_optional_action_idx.begin();
    chain::optional_automated_actions optional_actions;
 
-   while( pending_optional_itr != pending_optional_action_idx.end() )
+   while( pending_optional_itr != pending_optional_action_idx.end() && pending_optional_itr->execution_time <= when )
    {
       uint64_t new_total_size = total_block_size + fc::raw::pack_size( pending_optional_itr->action );