WIRE-344: Slash noncanonical delivery before termination

contracts/sysio.epoch/src/sysio.epoch.cpp

@@ -18,6 +18,7 @@
 #include <sysio.system/emissions.hpp>
 #include <sysio.chains/sysio.chains.hpp>
 #include <sysio/opp/attestations/attestations.pb.hpp>
+#include <magic_enum/magic_enum.hpp>
 
 namespace sysio {
 
@@ -45,6 +46,21 @@ namespace {
 constexpr name SYSTEM_ACCOUNT     = "sysio"_n;
 constexpr name TOKEN_ACCOUNT      = "sysio.token"_n;
 
+/// Action identifiers owned by sysio.chalg and invoked by epoch close.
+namespace chalg_actions {
+constexpr name SLASHOP = "slashop"_n;
+} // namespace chalg_actions
+
+/// Action identifiers owned by sysio.opreg and invoked while closing an epoch.
+namespace opreg_actions {
+constexpr name RECORD_DELIVERY = "recorddel"_n;
+constexpr name TERMINATION_CHECK = "termcheck"_n;
+} // namespace opreg_actions
+
+/// Durable reason prefix for an epoch-delivery classification slash.
+constexpr const char* NON_CANONICAL_DELIVERY_REASON_PREFIX =
+   "non-canonical OPP envelope delivery, epoch ";
+
 // System-owned rows are billed to the sysio RAM pool rather than to this contract account (the
 // privileged-contract model sysio.token uses): the contract account stays finite at its code+abi
 // size while table growth draws from sysio's pool. Permitted because the contract is privileged.
@@ -440,9 +456,9 @@ void epoch::advance() {
    // For each (outpost × member of the expiring group):
    //   - scan `msgch::envelopes` (`byoutepoch` index) for any row matching
    //     (chain_code, current_epoch_index, batch_op_name == member)
-   //   - inline `opreg::recorddel(member, current_epoch_index, did_deliver)`
-   //   - inline `opreg::termcheck(member)` — the threshold + window come
-   //     from `op_config`, so tests dial the thresholds via setconfig
+   //   - collect the delivery result and every non-canonical deliverer
+   //   - slash all non-canonical deliverers before delivery accounting can
+   //     terminate them, then record the result and run `termcheck`
    //
    // The outpost set is sourced via a cross-contract read of
    // `sysio.chains::chains` (no local mirror) filtered to
@@ -469,6 +485,15 @@ void epoch::advance() {
       // operator, which would abort advance and stall the chain).
       std::vector<name> to_slash;
 
+      /// A delivery result retained until non-canonical offenders have been
+      /// slashed. `recorddel` remains an audit record even for a newly
+      /// slashed operator, while `termcheck` safely skips non-ACTIVE rows.
+      struct delivery_observation {
+         name member;
+         bool did_deliver;
+      };
+      std::vector<delivery_observation> observations;
+
       sysio::chains::chains_t chains_tbl(CHAINS_ACCOUNT);
       for (auto op_it = chains_tbl.begin(); op_it != chains_tbl.end(); ++op_it) {
          if (!is_active_outpost(*op_it)) continue;
@@ -520,22 +545,11 @@ void epoch::advance() {
                   break;
                }
             }
-            action(
-               permission_level{get_self(), "owner"_n},
-               OPREG_ACCOUNT,
-               "recorddel"_n,
-               std::make_tuple(member, state.current_epoch_index, did_deliver)
-            ).send();
-            action(
-               permission_level{get_self(), "owner"_n},
-               OPREG_ACCOUNT,
-               "termcheck"_n,
-               std::make_tuple(member)
-            ).send();
+            observations.push_back({member, did_deliver});
 
             // Single slash path (dispute-vote design, per-operator outcome table): a delivered
             // NON-canonical checksum is a fault -> slash. Silence (no delivery) is never slashed; it
-            // stays on the recorddel/termcheck miss ladder above. Collect here; flush once below.
+            // stays on the recorddel/termcheck miss ladder below. Collect here; flush once below.
             if (did_deliver && have_winner && member_checksum != winner) {
                bool queued = false;
                for (const auto& s : to_slash) {
@@ -551,27 +565,43 @@ void epoch::advance() {
       // operator is marked SLASHED.
       //
       // Invariant — no cross-epoch double slash: opreg::slash THROWS on an already-SLASHED operator,
-      // which would abort advance and stall OPP epoch advancement. A slashed operator is guaranteed
-      // never to reappear in a later expiring group, so advance never attempts a second slash of it:
-      //   1. this flush runs BEFORE the window-slide below, so the operator is already SLASHED when
-      //      the next tail group is formed;
-      //   2. the new-tail filter pulls OPERATOR_STATUS_ACTIVE operators only (see the schedule slide
-      //      below), so a SLASHED operator is excluded from every newly-formed group; and
-      //   3. resident-exclusion keeps an operator in at most one group within the window, so the
-      //      operator slashed for THIS (expiring) group is not also sitting in a future
-      //      already-scheduled group.
-      // If any of those three scheduling facts change, this single-slash path must be revisited.
+      // which would abort advance and stall OPP epoch advancement. These inline slashes execute only
+      // after advance returns, so the schedule slide below can temporarily place a just-slashed
+      // operator in its new tail while the operator still reads ACTIVE. That member cannot create a
+      // later non-canonical observation: sysio.msgch::deliver requires its current sysio.opreg status
+      // to be ACTIVE before accepting delivery. Once the slash has executed, the scheduled SLASHED
+      // member cannot deliver or be queued for another non-canonical-delivery slash. The collection
+      // above also deduplicates multiple non-canonical observations for one member in this advance.
+      // Keep the deliver status gate and this single-slash path in sync if either behavior changes.
       for (const auto& member : to_slash) {
          action(
             permission_level{get_self(), "owner"_n},
             CHALG_ACCOUNT,
-            "slashop"_n,
+            chalg_actions::SLASHOP,
             std::make_tuple(member,
-                            std::string("non-canonical OPP envelope delivery, epoch ")
+                            std::string(NON_CANONICAL_DELIVERY_REASON_PREFIX)
                                + std::to_string(state.current_epoch_index))
          ).send();
       }
 
+      // Preserve the delivery history after slashing. A non-canonical operator is already
+      // SLASHED here, so opreg::termcheck returns without converting the punitive outcome into a
+      // termination/remit. Other group members retain their normal delivery accounting.
+      for (const auto& observation : observations) {
+         action(
+            permission_level{get_self(), "owner"_n},
+            OPREG_ACCOUNT,
+            opreg_actions::RECORD_DELIVERY,
+            std::make_tuple(observation.member, state.current_epoch_index, observation.did_deliver)
+         ).send();
+         action(
+            permission_level{get_self(), "owner"_n},
+            OPREG_ACCOUNT,
+            opreg_actions::TERMINATION_CHECK,
+            std::make_tuple(observation.member)
+         ).send();
+      }
+
       // NOTE: we intentionally do NOT erase the per-batch-op envelope
       // metadata rows here. `evalcons` already cleared their heavy
       // `raw_data` (1-2 KB → 0 bytes) at consensus reach, so the residual
@@ -625,7 +655,7 @@ void epoch::advance() {
       auto status_idx = opreg_ops.get_index<"bystatus"_n>();
       std::vector<std::pair<name, bool>> pool;
       for (auto it = status_idx.lower_bound(
-              static_cast<uint64_t>(OperatorStatus::OPERATOR_STATUS_ACTIVE));
+              magic_enum::enum_integer(OperatorStatus::OPERATOR_STATUS_ACTIVE));
            it != status_idx.end() &&
            it->status == OperatorStatus::OPERATOR_STATUS_ACTIVE; ++it) {
          if (it->type == OperatorType::OPERATOR_TYPE_BATCH && !is_resident(it->account)) {
@@ -705,7 +735,7 @@ void epoch::advance() {
          auto link_it = links_by_name.lower_bound(it->account.value);
          while (link_it != links_by_name.end() && link_it->username == it->account) {
             opp::types::ChainAddress chain_addr;
-            chain_addr.kind = static_cast<opp::types::ChainKind>(link_it->chain_kind);
+            chain_addr.kind = link_it->chain_kind;
 
             std::visit([&](const auto& key_data) {
                using T = std::decay_t<decltype(key_data)>;
@@ -889,7 +919,7 @@ void epoch::schbatchgps() {
    auto status_idx = opreg_ops.get_index<"bystatus"_n>();
    std::vector<std::pair<name, bool>> available_batch; // (account, is_bootstrapped)
    for (auto it = status_idx.lower_bound(
-           static_cast<uint64_t>(OperatorStatus::OPERATOR_STATUS_ACTIVE));
+           magic_enum::enum_integer(OperatorStatus::OPERATOR_STATUS_ACTIVE));
         it != status_idx.end() &&
         it->status == OperatorStatus::OPERATOR_STATUS_ACTIVE; ++it) {
       if (it->type == OperatorType::OPERATOR_TYPE_BATCH) {