Prune msgch inbound envelopes past a two-epoch window

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

@@ -630,16 +630,8 @@ void epoch::advance() {
          ).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
-      // weight is just the tuple `(id, chain_code, epoch_index,
-      // batch_op_name, checksum, ...)` — small and bounded by group
-      // membership × outposts × retained-epochs. A dedicated bounded-
-      // retention sweep belongs in a separate periodic ix; trying to
-      // erase here races with the permissionless `chkcons` →
-      // inline-`advance` pattern that fires from every batchop every
-      // cron tick and trips kv-index-remove on already-evicted buckets.
+      // The envelope rows read above are left in place: `sysio.msgch::deliver`
+      // erases them once they fall out of its retention window.
    }
 
    const bool had_expiring_group = state.current_epoch_index > 0;
@@ -1052,12 +1044,11 @@ void epoch::advance() {
       ).send();
    }
 
-   // Working tables on `sysio.msgch` (`envelopes` / `messages` /
-   // `attestations` / `outenvelopes`) are now drained inline by the
-   // `evalcons` consensus-reach + `buildenv` write paths. The durable
-   // audit trail lives in the `envelope_log` table on the same contract,
-   // capped at `active_outposts * 2 * cfg.epoch_retention_envelope_log_count`
-   // and pruned head-first on overflow. No scheduled cleanup needed.
+   // No scheduled cleanup of `sysio.msgch::envelopes` is needed here:
+   // `deliver` prunes rows older than the previous epoch. The durable audit
+   // trail lives in the `envelope_log` table on the same contract, capped at
+   // `active_outposts * 2 * cfg.epoch_retention_envelope_log_count`
+   // and pruned head-first on overflow.
 }
 
 // ---------------------------------------------------------------------------

contracts/sysio.msgch/include/sysio.msgch/sysio.msgch.hpp

@@ -109,20 +109,25 @@ namespace sysio {
 
       /// Inbound envelope delivery — one row per batch-op per outpost per epoch.
       /// Consensus is evaluated by comparing checksums across operators.
+      ///
+      /// Working state, not an audit trail (that is `envlog`). `deliver` prunes rows older than the
+      /// current and previous epoch (`INBOUND_ENVELOPE_RETENTION_EPOCHS` in `src/sysio.msgch.cpp`);
+      /// every on-chain reader only touches the current epoch.
       struct [[sysio::table("envelopes")]] envelope_entry {
          uint64_t                  id;
          uint64_t                  chain_code;
          uint32_t                  epoch_index;
          name                      batch_op_name;
          opp::types::ChainKind     chain_kind;
-         checksum256               checksum;        ///< sha256(raw_data)
+         checksum256               checksum;        ///< sha256 of the delivered bytes
+         /// The delivered bytes. Cleared on the rows present when the epoch's winner is accepted and
+         /// never stored for a late confirmation of it; the retention prune bounds any others.
          std::vector<char>         raw_data;
          time_point                received_at{};
 
          uint128_t by_outpost_epoch() const {
             return opp::outpost_epoch_key(chain_code, epoch_index);
          }
-         uint64_t by_batch_op() const { return batch_op_name.value; }
 
          SYSLIB_SERIALIZE(envelope_entry,
             (id)(chain_code)(epoch_index)(batch_op_name)(chain_kind)
@@ -131,9 +136,7 @@ namespace sysio {
 
       using envelopes_t = sysio::kv::table<"envelopes"_n, id_key, envelope_entry,
          sysio::kv::index<"byoutepoch"_n,
-            sysio::const_mem_fun<envelope_entry, uint128_t, &envelope_entry::by_outpost_epoch>>,
-         sysio::kv::index<"bybatchop"_n,
-            sysio::const_mem_fun<envelope_entry, uint64_t, &envelope_entry::by_batch_op>>
+            sysio::const_mem_fun<envelope_entry, uint128_t, &envelope_entry::by_outpost_epoch>>
       >;
 
       /// Individual message extracted from a consensus-verified envelope.

contracts/sysio.msgch/src/sysio.msgch.cpp

@@ -90,6 +90,18 @@ constexpr size_t   ATTESTATION_OVERHEAD_BYTES = 24;
 /// + payload preamble, and a safety margin for `zpp::bits` length prefixes.
 constexpr size_t   ENVELOPE_BASELINE_BYTES    = 512;
 
+/// Epochs of inbound `envelopes` rows kept: the current epoch, which every on-chain reader uses, and
+/// the previous one, so a batch operator whose cached epoch lags by one still finds its delivery.
+constexpr uint32_t INBOUND_ENVELOPE_RETENTION_EPOCHS = 2;
+
+/// Maximum expired `envelopes` rows a single `deliver` erases. Each delivery adds one row, so a
+/// budget above one outpaces inserts and clears an epoch with few deliveries after a full one.
+constexpr uint32_t ENVELOPE_PRUNE_BUDGET = 4;
+
+static_assert(INBOUND_ENVELOPE_RETENTION_EPOCHS >= 1,
+              "the prune must never reach the current epoch, which every reader depends on");
+static_assert(ENVELOPE_PRUNE_BUDGET > 1, "the prune must outpace the one row each delivery adds");
+
 /// Stable audit marker for a UIC rejected before it can reach `rcrdcommit`.
 constexpr const char* UIC_DISPATCH_REJECTED_LOG_PREFIX =
    "UIC_DISPATCH_REJECTED";
@@ -249,6 +261,24 @@ void write_envelope_log(name self,
    }
 }
 
+/// Erase up to `ENVELOPE_PRUNE_BUDGET` inbound `envelopes` rows that fell out of the retention
+/// window, oldest first.
+///
+/// Primary-key order is epoch order: ids come from `available_primary_key()`, `deliver` only
+/// accepts current-epoch envelopes, and rows leave only from the head, so the newest id is never
+/// erased and ids never restart. The walk stops at the first retained row, so a call is O(budget).
+void prune_expired_envelopes(name self, uint32_t current_epoch) {
+   if (current_epoch < INBOUND_ENVELOPE_RETENTION_EPOCHS) return;
+   const uint32_t newest_expired_epoch = current_epoch - INBOUND_ENVELOPE_RETENTION_EPOCHS;
+
+   msgch::envelopes_t envs(self);
+   auto it = envs.begin();
+   for (uint32_t erased = 0; erased < ENVELOPE_PRUNE_BUDGET && it != envs.end() &&
+                             it->epoch_index <= newest_expired_epoch; ++erased) {
+      it = envs.erase(std::move(it));
+   }
+}
+
 /// Resolve `op_address` (chain-kind + raw pubkey bytes) to the operator's
 /// WIRE account name via `sysio.authex::links`'s `bypubkey` index. Returns
 /// `name{}` (zero) on miss — caller treats that as "operator not linked,
@@ -1270,6 +1300,7 @@ void dispatch_attestation(name self, uint64_t attestation_id,
 
       // Drop heavy raw_data from each per-batch-op envelope row but KEEP the metadata tuple so
       // sysio.epoch::advance can still read per-op checksums + delivery for slash classification.
+      // `deliver` erases the rows once they leave the retention window.
       msgch::envelopes_t envs(self);
       std::vector<uint64_t> ids_to_clear;
       auto modify_idx = envs.get_index<"byoutepoch"_n>();
@@ -1515,25 +1546,23 @@ void msgch::deliver(name batch_op_name, uint64_t chain_code, std::vector<char> d
    // re-validation. Anything else -- including a divergent envelope arriving after acceptance --
    // still validates against the advanced tip and reverts (fail closed: post-acceptance divergence
    // cannot open a dispute, so there is nothing to record it for).
+   bool late_confirmation = false;
    {
-      bool late_confirmation = false;
-      {
-         msgch::outpost_consensus_t opcons(get_self());
-         auto opc_pk = msgch::outpost_consensus_key{chain_code};
-         if (opcons.contains(opc_pk)) {
-            const auto row    = opcons.get(opc_pk);
-            late_confirmation = row.epoch_index == epoch && row.consensus_reached &&
-                                row.winning_checksum == cs;
-         }
-      }
-      if (!late_confirmation) {
-         checksum256 ingress_digest{};
-         checksum256 ingress_message_tip{};
-         check(inbound_envelope_valid(get_self(), env_check, chain_code, epoch, ingress_digest,
-                                      ingress_message_tip),
-               "delivered envelope failed inbound-chain or semantic-header validation");
+      msgch::outpost_consensus_t opcons(get_self());
+      auto opc_pk = msgch::outpost_consensus_key{chain_code};
+      if (opcons.contains(opc_pk)) {
+         const auto row    = opcons.get(opc_pk);
+         late_confirmation = row.epoch_index == epoch && row.consensus_reached &&
+                             row.winning_checksum == cs;
       }
    }
+   if (!late_confirmation) {
+      checksum256 ingress_digest{};
+      checksum256 ingress_message_tip{};
+      check(inbound_envelope_valid(get_self(), env_check, chain_code, epoch, ingress_digest,
+                                   ingress_message_tip),
+            "delivered envelope failed inbound-chain or semantic-header validation");
+   }
 
    // Store envelope
    uint64_t env_id = std::max<uint64_t>(1, envs.available_primary_key());
@@ -1550,10 +1579,16 @@ void msgch::deliver(name batch_op_name, uint64_t chain_code, std::vector<char> d
       // is authoritative; this is just the cached projection.
       .chain_kind    = op_row.kind,
       .checksum      = cs,
-      .raw_data      = data,
+      // A late confirmation repeats this epoch's already-applied winner. `outpcons` records that
+      // acceptance, so `apply_consensus` never decodes this bucket's bytes again; `advance` needs
+      // only the metadata.
+      .raw_data      = late_confirmation ? std::vector<char>{} : std::move(data),
       .received_at   = current_time_point(),
    });
 
+   // After the emplace, so the newest id always survives (see prune_expired_envelopes).
+   prune_expired_envelopes(get_self(), epoch);
+
    // Evaluate consensus inline
    action(
       permission_level{get_self(), "active"_n},
@@ -1785,7 +1820,8 @@ void msgch::resolvedisp(uint64_t chain_code, uint32_t epoch_index, checksum256 w
 
    // Locate the winning envelope's raw bytes among this (outpost, epoch)'s deliveries. The dispute
    // path never cleared raw_data (evalcons returned early before the consensus cleanup), so the
-   // bytes are still on file. Copy them out before apply_consensus drains the rows.
+   // bytes are still on file. Copy them out before apply_consensus drains the rows. `deliver`'s
+   // retention prune never reaches them: an open dispute pauses the epoch, so it is still current.
    envelopes_t envs(get_self());
    auto oe_idx = envs.get_index<"byoutepoch"_n>();
    uint128_t composite = opp::outpost_epoch_key(chain_code, epoch_index);

contracts/sysio.msgch/sysio.msgch.abi

@@ -478,11 +478,6 @@
                     "name": "byoutepoch",
                     "key_type": "uint128",
                     "table_id": 57096
-                },
-                {
-                    "name": "bybatchop",
-                    "key_type": "uint64",
-                    "table_id": 5961
                 }
             ]
         },