opp: erase an outbound envelope only once the outpost has consumed it (WIRE-348 / CertiK WNS-18)

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

@@ -198,9 +198,16 @@ namespace sysio {
             sysio::const_mem_fun<attestation_entry, uint64_t, &attestation_entry::by_epoch>>
       >;
 
-      /// Outbound envelope table. One-deep per outpost: `buildenv` erases every older row for the
-      /// destination `chain_code` after inserting the new emit, so the surviving row doubles as the
-      /// per-outpost chain tip.
+      /// Outbound envelope table. After inserting a new emit, `buildenv` erases the older rows for
+      /// the destination `chain_code` that the outpost has CONSUMED -- those whose `epoch_index` is
+      /// covered by `outpcons.epoch_index` -- and retains any it has not. The epoch-advance
+      /// interlock (`chkcons` releases `advance`, and therefore `buildenv`, only once every active
+      /// outpost has reached consensus at the current epoch) makes that the whole table in the
+      /// healthy path, so it is normally one-deep; a retained row means the interlock did not hold
+      /// and its `raw_envelope` is the only remaining copy of an envelope the outpost still needs.
+      ///
+      /// The per-outpost chain tip is therefore the NEWEST row, selected through `byoutepoch` --
+      /// NOT `byoutpost`, whose entries sort by ascending primary key and so yield the oldest.
       ///
       /// `envelope_hash` is the canonical epoch digest: keccak256 over the canonical
       /// field-complete encoding with the in-envelope `envelope_hash` field blanked (equal to

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

@@ -1870,21 +1870,31 @@ void msgch::buildenv(uint64_t chain_code) {
       candidate_ids.begin(),
       candidate_ids.begin() + included_count);
 
-   // Chain links: the previous envelope emitted for this outpost. `outenvelopes` is one-deep per
-   // outpost (see the cleanup below), so the single surviving row is the previous emit; its
-   // `envelope_hash` is that envelope's epoch digest and its `last_message_id` is this outpost's
-   // message-stream tip. The first emit for an outpost has no row and chains both links from
-   // empty (genesis), matching the outpost contracts' zero genesis tip.
+   // Chain links: the previous envelope emitted for this outpost -- the NEWEST row for
+   // `chain_code`, its `envelope_hash` that envelope's epoch digest and its `last_message_id`
+   // this outpost's message-stream tip. The first emit for an outpost has no row and chains both
+   // links from empty (genesis), matching the outpost contracts' zero genesis tip.
+   //
+   // Walked over `byoutepoch` rather than `byoutpost`: KV secondary entries sort by ascending
+   // PRIMARY key within one secondary key, so a `byoutpost` lower_bound yields the OLDEST row for
+   // the outpost. That was the tip only while exactly one row survived; the ack-gated cleanup
+   // below legitimately retains an emit the outpost has not consumed, so the tip must be selected
+   // by epoch instead of by "the only row left".
+   //
+   // `byoutepoch` packs `(chain_code, epoch_index)`, so this outpost's rows are contiguous and
+   // ascending by epoch -- the LAST match in the walk is the tip. Forward-only (`lower_bound` +
+   // `++` + the chain_code guard), the same shape as the sweep below; the walk is one row in the
+   // healthy path and bounded by the retained emits otherwise.
    outenvelopes_t envelopes(get_self());
    std::vector<char> prev_envelope_digest;
    std::vector<char> prev_message_id;
    {
-      auto by_outpost = envelopes.get_index<"byoutpost"_n>();
-      auto prev_it = by_outpost.lower_bound(chain_code);
-      if (prev_it != by_outpost.end() && prev_it->chain_code == chain_code) {
-         const auto digest_bytes = prev_it->envelope_hash.extract_as_byte_array();
+      auto by_epoch = envelopes.get_index<"byoutepoch"_n>();
+      for (auto it = by_epoch.lower_bound(opp::outpost_epoch_key(chain_code, 0));
+           it != by_epoch.end() && it->chain_code == chain_code; ++it) {
+         const auto digest_bytes = it->envelope_hash.extract_as_byte_array();
          prev_envelope_digest.assign(digest_bytes.begin(), digest_bytes.end());
-         const auto tip_bytes = prev_it->last_message_id.extract_as_byte_array();
+         const auto tip_bytes = it->last_message_id.extract_as_byte_array();
          prev_message_id.assign(tip_bytes.begin(), tip_bytes.end());
       }
    }
@@ -2003,20 +2013,56 @@ void msgch::buildenv(uint64_t chain_code) {
    // === AUDIT LOG + INLINE CLEANUP OF WORKING STATE ===
    //
    // Audit-log row mirrors the outbound emit (WIRE → outpost). Followed
-   // by inline drains of the previous-epoch outenvelopes row (one-deep
-   // retention; the batch op only ever reads the most-recent emit) and
-   // the just-PROCESSED attestations for this outpost (their bytes are
-   // now baked into `packed` above).
+   // by inline drains of the outenvelopes rows this outpost has already
+   // CONSUMED (see the ack gate below) and the just-PROCESSED
+   // attestations for this outpost (their bytes are now baked into
+   // `packed` above).
    {
       // Same endpoints the wire envelope carries (derived above from the
       // destination's `sysio.chains` row).
       write_envelope_log(get_self(), route_endpoints, epoch, envelope_digest);
 
-      // Drop previous outpost emits — keep only the row we just inserted.
+      // Drop previous outpost emits the destination has CONSUMED, and only those.
+      //
+      // `outpcons.epoch_index` is the last epoch whose INBOUND envelope from this outpost
+      // reached consensus, and an outpost can only emit its epoch-N envelope after accepting
+      // the depot's epoch-N emit -- Solana asserts it directly
+      // (`emit.rs`: `require!(wire_epoch_index < config.next_epoch_index, EmitBeforeEpochAccepted)`,
+      // which binds the admin recovery instruction too), and Ethereum reaches
+      // `OPP::emitOutboundEnvelope` only from the `OPPInbound` consensus tip under
+      // `OPP_FINALIZER_ROLE`. So `epoch_index <= acked` is PROOF the outpost consumed that emit.
+      //
+      // In the healthy path this erases exactly what the old unconditional sweep did and leaves
+      // the table one-deep: `chkcons` only released the `advance` that reached this `buildenv`
+      // because every active outpost had already reached consensus at the previous epoch, so
+      // `acked` covers the prior emit. A row survives only when that interlock did not hold, in
+      // which case its bytes are the ONLY copy of an envelope the outpost still needs -- the
+      // audit log keeps a checksum, not a payload, and the source attestations are drained just
+      // below. Retaining is what stops a chain break from becoming unrecoverable data loss.
+      //
+      // `has_ack` is load-bearing, not defensive: NO row means the outpost has acknowledged
+      // NOTHING, which a bare `epoch_index` default of 0 would make indistinguishable from
+      // having acknowledged epoch 0 -- and would erase an epoch-0 emit on that reading.
+      outpost_consensus_t opcons(get_self());
+      const auto     opc_pk  = outpost_consensus_key{chain_code};
+      const bool     has_ack = opcons.contains(opc_pk);
+      const uint32_t acked   = has_ack ? opcons.get(opc_pk).epoch_index : 0;
+
       auto by_outpost = envelopes.get_index<"byoutpost"_n>();
       for (auto it = by_outpost.lower_bound(chain_code);
            it != by_outpost.end() && it->chain_code == chain_code; ) {
          if (it->id == out_id) { ++it; continue; }
+         if (!has_ack || it->epoch_index > acked) {
+            // Never expected: reaching this `buildenv` required consensus past this emit.
+            // Logged (visible under --contracts-console) so a broken interlock is greppable
+            // rather than a silently growing table.
+            sysio::print_f("msgch::buildenv: retaining unacknowledged emit chain_code=%llu "
+                           "epoch=%u has_ack=%d acked=%u\n",
+                           static_cast<unsigned long long>(chain_code), it->epoch_index,
+                           static_cast<int>(has_ack), acked);
+            ++it;
+            continue;
+         }
          it = by_outpost.erase(std::move(it));
       }
 

contracts/tests/sysio.msgch_chain_tests.cpp

@@ -492,20 +492,53 @@ class sysio_msgch_chain_tester : public tester {
 
    // -- Table readers --
 
-   /// The single surviving outbound envelope row for `chain_code` (the table is one-deep per
-   /// outpost); null variant when none.
+   /// The NEWEST outbound envelope row for `chain_code` -- the per-outpost chain tip
+   /// `buildenv` chains from; null variant when none. Deliberately not "the first row
+   /// found": `buildenv` retains emits the outpost has not acknowledged, so the lowest
+   /// id can be a retained predecessor rather than the tip.
    fc::variant find_outbound_envelope(uint64_t chain_code, uint64_t scan_until = 32) {
+      fc::variant newest;
       for (uint64_t id = 0; id < scan_until; ++id) {
          auto data = get_row_by_id(MSGCH_ACCOUNT, MSGCH_ACCOUNT, "outenvelopes"_n, id);
          if (data.empty()) continue;
          auto row = msgch_abi.binary_to_variant(
             "outbound_envelope", data,
             abi_serializer::create_yield_function(abi_serializer_max_time));
-         if (row["chain_code"].as_uint64() == chain_code) return row;
+         if (row["chain_code"].as_uint64() == chain_code) newest = row;
+      }
+      return newest;
+   }
+
+   /// The outbound envelope row `chain_code` emitted for `epoch_index`; null when it is
+   /// absent -- either never emitted, or erased because the outpost acknowledged it.
+   fc::variant outbound_envelope_for_epoch(uint64_t chain_code, uint32_t epoch_index,
+                                           uint64_t scan_until = 32) {
+      for (uint64_t id = 0; id < scan_until; ++id) {
+         auto data = get_row_by_id(MSGCH_ACCOUNT, MSGCH_ACCOUNT, "outenvelopes"_n, id);
+         if (data.empty()) continue;
+         auto row = msgch_abi.binary_to_variant(
+            "outbound_envelope", data,
+            abi_serializer::create_yield_function(abi_serializer_max_time));
+         if (row["chain_code"].as_uint64() == chain_code &&
+             row["epoch_index"].as<uint32_t>() == epoch_index) return row;
       }
       return fc::variant{};
    }
 
+   /// Live outbound envelope rows for `chain_code`.
+   uint32_t outbound_envelope_count(uint64_t chain_code, uint64_t scan_until = 32) {
+      uint32_t n = 0;
+      for (uint64_t id = 0; id < scan_until; ++id) {
+         auto data = get_row_by_id(MSGCH_ACCOUNT, MSGCH_ACCOUNT, "outenvelopes"_n, id);
+         if (data.empty()) continue;
+         auto row = msgch_abi.binary_to_variant(
+            "outbound_envelope", data,
+            abi_serializer::create_yield_function(abi_serializer_max_time));
+         if (row["chain_code"].as_uint64() == chain_code) ++n;
+      }
+      return n;
+   }
+
    /// Per-outpost consensus row (`outpcons`, primary key = chain_code); null variant when absent.
    fc::variant get_outpcons(uint64_t chain_code) {
       auto data = get_row_by_id(MSGCH_ACCOUNT, MSGCH_ACCOUNT, "outpcons"_n, chain_code);
@@ -1063,6 +1096,56 @@ BOOST_FIXTURE_TEST_CASE(buildenv_first_emit_chains_from_empty, sysio_msgch_chain
                        fc::to_hex(header.message_id().data(), header.message_id().size()));
 } FC_LOG_AND_RETHROW() }
 
+// ---------------------------------------------------------------------------
+//  Outbound retention: an emit is erased only once the outpost acknowledges it.
+// ---------------------------------------------------------------------------
+
+/// WNS-18 / WIRE-348, the acknowledged half. `buildenv` erases prior emits for an outpost
+/// only up to `outpcons.epoch_index` -- the last epoch whose INBOUND envelope from that
+/// outpost reached consensus, which the outpost can only have produced after consuming the
+/// depot's emit for that epoch. So:
+///
+///   * while nothing is acknowledged, successive emits ACCUMULATE (their `raw_envelope` is
+///     the only surviving copy of an envelope the outpost still needs), and
+///   * one acknowledgement drains every emit at or below it, leaving the table one-deep --
+///     which is what the healthy path always looks like, because `chkcons` releases the
+///     `advance` that reaches `buildenv` only once every active outpost has reached
+///     consensus at the current epoch.
+BOOST_FIXTURE_TEST_CASE(buildenv_erases_acknowledged_outenvelope, sysio_msgch_chain_tester) { try {
+   bootstrap();
+
+   // bootstrap()'s genesis `advance` fanned a buildenv to every registered outpost.
+   const uint32_t epoch_a = current_epoch();
+   BOOST_REQUIRE(!outbound_envelope_for_epoch(ETH_OUTPOST_ID, epoch_a).is_null());
+   BOOST_REQUIRE(get_outpcons(ETH_OUTPOST_ID).is_null());   // nothing acknowledged yet
+
+   // Unacknowledged: the next epoch's emit does NOT displace epoch A's.
+   const uint32_t epoch_b = advance_one_epoch();
+   BOOST_REQUIRE_NE(epoch_a, epoch_b);
+   BOOST_REQUIRE(!outbound_envelope_for_epoch(ETH_OUTPOST_ID, epoch_a).is_null());
+   BOOST_REQUIRE(!outbound_envelope_for_epoch(ETH_OUTPOST_ID, epoch_b).is_null());
+   BOOST_REQUIRE_EQUAL(2u, outbound_envelope_count(ETH_OUTPOST_ID));
+
+   // The tip is the NEWEST row, not the retained predecessor.
+   BOOST_REQUIRE_EQUAL(find_outbound_envelope(ETH_OUTPOST_ID)["epoch_index"].as<uint32_t>(),
+                       epoch_b);
+
+   // An inbound envelope accepted for epoch B is the acknowledgement: the outpost could not
+   // have emitted it without first consuming the depot's epoch-B envelope.
+   BOOST_REQUIRE_EQUAL(success(), deliver(ETH_OUTPOST_ID, encode_delivery(epoch_b, "alpha")));
+   produce_blocks();
+   auto opc = get_outpcons(ETH_OUTPOST_ID);
+   BOOST_REQUIRE(!opc.is_null());
+   BOOST_REQUIRE_EQUAL(opc["epoch_index"].as<uint32_t>(), epoch_b);
+
+   // The next emit now drains everything at or below the acknowledged epoch -- both A and B.
+   const uint32_t epoch_c = advance_one_epoch();
+   BOOST_REQUIRE(outbound_envelope_for_epoch(ETH_OUTPOST_ID, epoch_a).is_null());
+   BOOST_REQUIRE(outbound_envelope_for_epoch(ETH_OUTPOST_ID, epoch_b).is_null());
+   BOOST_REQUIRE(!outbound_envelope_for_epoch(ETH_OUTPOST_ID, epoch_c).is_null());
+   BOOST_REQUIRE_EQUAL(1u, outbound_envelope_count(ETH_OUTPOST_ID));
+} FC_LOG_AND_RETHROW() }
+
 // ---------------------------------------------------------------------------
 //  Inbound: apply_consensus records and verifies the per-outpost chain tip.
 // ---------------------------------------------------------------------------
@@ -1306,7 +1389,7 @@ BOOST_FIXTURE_TEST_CASE(inbound_rejects_cross_stream_link_for_svm_outposts,
 
    // Next epoch: emit a depot outbound envelope, then deliver an inbound envelope whose prev is
    // that emit's digest (NOT the inbound tip). Building the outbound envelope in the delivery
-   // epoch pins the one-deep outenvelopes row so no epoch-advance emission replaces it.
+   // epoch pins the outenvelopes tip so no epoch-advance emission supersedes it.
    epoch = advance_one_epoch();
    BOOST_REQUIRE_EQUAL(success(), queueout(SOL_OUTPOST_ID,
       sysio::opp::types::ATTESTATION_TYPE_OPERATORS, std::vector<char>{0x0a}));