diff --git a/OVERVIEW.md b/OVERVIEW.md index 592b8d5af9..8a71afb209 100644 --- a/OVERVIEW.md +++ b/OVERVIEW.md @@ -55,8 +55,8 @@ This allows a single unified message format across all supported chains. ### Outbound (Wire → External Chain) -1. System contracts queue attestations for a target outpost via `sysio.msgch::queueout()`. -2. When the epoch advances, `buildenv()` packs the `READY` attestations into a protobuf-encoded outbound envelope; anything that does not fit stays queued for the next epoch. +1. System contracts queue attestations for a target outpost via `sysio.msgch::queueout()`. Once an outpost has 512 non-roster attestations waiting, `queueout` refuses another. +2. When the epoch advances, `buildenv()` packs the outpost's `READY` attestations, rosters first, into a protobuf-encoded outbound envelope of at most 32,768 bytes and 64 attestations; the rest stays queued for the next epoch. 3. Batch operators read the outbound envelope and submit it to the target chain's Outpost — chunked on both chains (`OPPInbound.epochIn` on Ethereum, `epoch_in` on Solana). 4. The Outpost verifies the envelope chain and group consensus, then routes attestations to their registered handlers. An unhandled type is skipped, never reverted. diff --git a/contracts/sysio.msgch/README.md b/contracts/sysio.msgch/README.md index b835b5dee5..bff4c68217 100644 --- a/contracts/sysio.msgch/README.md +++ b/contracts/sysio.msgch/README.md @@ -11,8 +11,8 @@ Inbound/outbound OPP message chain management and consensus tracking contract. post-boundary split with no strict majority opens a `sysio.chalg` dispute - Validates the per-outpost envelope chain and message sequence, stores the accepted messages and attestations, and routes each attestation to its depot handler -- Queues outbound attestations and packs them into one OPP envelope per outpost per epoch; whatever - does not fit under the envelope cap stays queued for the next epoch +- Queues outbound attestations and packs them into one OPP envelope per outpost per epoch, schedule + attestations first; whatever does not fit under the envelope caps stays queued for the next epoch - Drives the epoch: `chkcons` advances `sysio.epoch` once every active outpost has reached consensus and the wall-clock boundary has passed @@ -22,7 +22,8 @@ Inbound/outbound OPP message chain management and consensus tracking contract. |-------|------|-------------| | `envelopes` | `kv::table` | Inbound deliveries, one row per batch operator per outpost per epoch (working state, pruned to the current and previous epoch) | | `messages` | `kv::table` | Individual OPP messages extracted from consensus-verified envelopes | -| `attestations` | `kv::table` | Attestations extracted inbound, and outbound attestations queued (`READY`) for envelope packing | +| `attestations` | `kv::table` | Attestations extracted inbound, and outbound attestations queued (`READY`) for envelope packing. The `byqueue` index orders them by outpost, status and lane | +| `outqueue` | `kv::table` | Per-outpost count of queued general-lane attestations, capped at `MAX_QUEUED_ATTESTATIONS` | | `outenvelopes` | `kv::table` | Built outbound envelopes awaiting relay, with their canonical epoch digest | | `outpcons` | `kv::table` | Per-outpost consensus record: accepted envelope digest and inbound message-chain tip | | `attseq` | `kv::table` | Attestation sequence counter | @@ -37,8 +38,8 @@ Inbound/outbound OPP message chain management and consensus tracking contract. | `evalcons` | `sysio.msgch` | Evaluate consensus for an (outpost, epoch); on consensus store and dispatch the attestations; on an eligible split open a `sysio.chalg` dispute | | `chkcons` | permissionless | If no dispute is open, every active outpost has consensus and the boundary has passed, call `sysio.epoch::advance` | | `resolvedisp` | `sysio.chalg` | Dispatch the winning envelope of a resolved dispute | -| `queueout` | `sysio.epoch`, `sysio.opreg`, `sysio.synd`, or `sysio.msgch` | Queue an outbound attestation for an outpost. Rejects `data` larger than `MAX_ATTESTATION_DATA_BYTES`, so every queued row fits an envelope | -| `buildenv` | `sysio.epoch` | Pack `READY` attestations into the outpost's outbound envelope for the epoch | +| `queueout` | `sysio.epoch`, `sysio.opreg`, `sysio.synd`, or `sysio.msgch` | Queue an outbound attestation for an outpost. Rejects `data` larger than `MAX_ATTESTATION_DATA_BYTES`, or `MAX_GENERAL_ATTESTATION_DATA_BYTES` (7,656) for a general-lane row, which must fit beside a full schedule lane, so every queued row ships. Also rejects a general-lane row once the outpost already holds `MAX_QUEUED_ATTESTATIONS` (512) of them | +| `buildenv` | `sysio.epoch` | Pack the outpost's `READY` attestations into its outbound envelope for the epoch: the schedule lane (`OPERATORS`, `BATCH_OPERATOR_GROUPS`) first, then the general lane in `queueout` order, up to `MAX_ENVELOPE_BYTES` and `MAX_ENVELOPE_ATTESTATIONS` (64). It reads only that outpost's rows | ## Attestation routing @@ -55,8 +56,10 @@ Inbound (outpost → depot): Outbound (depot → outpost), queued through `queueout`: -- `OPERATORS` and `BATCH_OPERATOR_GROUPS` from `sysio.epoch::advance`, to every active outpost -- `DESYNDICATE_LIQ` from `sysio.synd::desyndicate` +- `OPERATORS` and `BATCH_OPERATOR_GROUPS` from `sysio.epoch::advance`, to every active outpost. They + form the schedule lane, which ships first and which the queue cap never refuses. +- `DESYNDICATE_LIQ` from `sysio.synd::desyndicate`. A full queue refuses it, and the whole + `desyndicate` reverts. ## Dependencies diff --git a/contracts/sysio.msgch/include/sysio.msgch/sysio.msgch.hpp b/contracts/sysio.msgch/include/sysio.msgch/sysio.msgch.hpp index 576b704e34..e11d93ed00 100644 --- a/contracts/sysio.msgch/include/sysio.msgch/sysio.msgch.hpp +++ b/contracts/sysio.msgch/include/sysio.msgch/sysio.msgch.hpp @@ -9,6 +9,7 @@ #include #include #include +#include namespace sysio { @@ -40,6 +41,55 @@ namespace sysio { static constexpr size_t MAX_ATTESTATION_DATA_BYTES = MAX_ENVELOPE_BYTES - ENVELOPE_BASELINE_BYTES - ATTESTATION_OVERHEAD_BYTES; + /// Most attestations one envelope carries, so the rows `buildenv` reads and erases per outpost + /// per epoch stay fixed whatever their sizes. Neither outpost needs it: both resume dispatch + /// across calls. + static constexpr size_t MAX_ENVELOPE_ATTESTATIONS = 64; + + // ----------------------------------------------------------------------- + // Outbound queue + // ----------------------------------------------------------------------- + + /// Most general-lane attestations an outpost's READY queue holds. `queueout` refuses another + /// until `buildenv` ships some, so the producer's transaction fails instead of the queue growing + /// without bound. Schedule-lane rows are exempt: `sysio.epoch` queues them inside `advance`. + static constexpr uint32_t MAX_QUEUED_ATTESTATIONS = 512; + + /// Lane of an outbound attestation. `buildenv` ships the schedule lane first, so a backlog never + /// delays the rosters an outpost needs to admit the next epoch's deliveries. + enum class queue_lane : uint8_t { + schedule = 0, ///< OPERATORS and BATCH_OPERATOR_GROUPS + general = 1, ///< every other type, in `queueout` order + }; + + /// The lane `type` queues in. + static constexpr queue_lane lane_of(opp::types::AttestationType type) { + return type == opp::types::ATTESTATION_TYPE_OPERATORS || + type == opp::types::ATTESTATION_TYPE_BATCH_OPERATOR_GROUPS + ? queue_lane::schedule + : queue_lane::general; + } + + /// Most envelope bytes one epoch's schedule-lane rows take, counted as `buildenv` estimates them + /// (`data` plus ATTESTATION_OVERHEAD_BYTES each). The largest window's OPERATORS and + /// BATCH_OPERATOR_GROUPS come to about 6.8 KB, and 15.4 KB even at Solana's 16 per group; 24 KiB also + /// matches Solana's OPERATORS payload cap. + static constexpr size_t SCHEDULE_LANE_BYTES = 24 * 1024; + + /// Largest general-lane `data` that fits beside a full schedule lane, which every envelope carries + /// first. `queueout` rejects anything larger, so no general row can stall the rows behind it. + static constexpr size_t MAX_GENERAL_ATTESTATION_DATA_BYTES = + MAX_ENVELOPE_BYTES - ENVELOPE_BASELINE_BYTES - SCHEDULE_LANE_BYTES - ATTESTATION_OVERHEAD_BYTES; + + /// `byqueue` key: outpost, then status, then lane. Rows with equal keys sort by id, so each lane + /// is FIFO. 128 bits keep the full slug range of `chain_code` (see `opp::outpost_epoch_key`). + static constexpr uint128_t queue_key(uint64_t chain_code, opp::types::AttestationStatus status, + queue_lane lane) { + return (static_cast(chain_code) << 16) | + (static_cast(magic_enum::enum_integer(status)) << 8) | + static_cast(magic_enum::enum_integer(lane)); + } + // ----------------------------------------------------------------------- // Actions // ----------------------------------------------------------------------- @@ -84,7 +134,9 @@ namespace sysio { /// Queue an outbound attestation for an outpost. /// Writes to the attestations table with status READY. Rejects `data` larger than - /// MAX_ATTESTATION_DATA_BYTES, so every READY row fits an envelope. + /// MAX_ATTESTATION_DATA_BYTES (MAX_GENERAL_ATTESTATION_DATA_BYTES for a general-lane row), so every + /// READY row ships, and a general-lane row once the outpost already holds MAX_QUEUED_ATTESTATIONS + /// of them. /// /// `chain_code` is the destination outpost's chain slug_name value /// (uint64). Called by sibling system contracts that need to send @@ -107,7 +159,9 @@ namespace sysio { std::vector data); /// Build outbound envelope from READY attestations for an outpost. - /// Collects attestations, packs into OPP Envelope, stores in outenvelopes. + /// Walks only this outpost's READY rows, schedule lane first, and stops at MAX_ENVELOPE_BYTES or + /// MAX_ENVELOPE_ATTESTATIONS, so its cost does not grow with the backlog. Packs the rows into an + /// OPP Envelope, stores it in outenvelopes and erases them. /// Stamps the envelope's route endpoints (start = WIRE, end = the /// destination's `sysio.chains` `{kind, external_chain_id}`) so the /// receiving outpost can reject an envelope a misconfigured relay @@ -202,9 +256,7 @@ namespace sysio { uint64_t ready_timestamp; uint64_t processed_timestamp; - uint64_t by_status() const { return static_cast(status); } - uint64_t by_type() const { return static_cast(type); } - uint64_t by_epoch() const { return epoch_index; } + uint128_t by_queue() const { return queue_key(chain_code, status, lane_of(type)); } SYSLIB_SERIALIZE(attestation_entry, (id)(chain_code)(epoch_index)(type)(status)(data) @@ -212,12 +264,8 @@ namespace sysio { }; using attestations_t = sysio::kv::table<"attestations"_n, id_key, attestation_entry, - sysio::kv::index<"bystatus"_n, - sysio::const_mem_fun>, - sysio::kv::index<"bytype"_n, - sysio::const_mem_fun>, - sysio::kv::index<"byepoch"_n, - sysio::const_mem_fun> + sysio::kv::index<"byqueue"_n, + sysio::const_mem_fun> >; /// Outbound envelope table. `buildenv` erases a row only once the outpost has consumed it @@ -305,16 +353,33 @@ namespace sysio { using outpost_consensus_t = sysio::kv::table<"outpcons"_n, outpost_consensus_key, outpost_consensus_entry>; + /// Per-outpost outbound queue primary key. + struct outbound_queue_key { + uint64_t chain_code; + SYSLIB_SERIALIZE(outbound_queue_key, (chain_code)) + }; + + /// The general-lane READY attestations queued for an outpost. `queueout` counts them in and + /// refuses one past MAX_QUEUED_ATTESTATIONS; `buildenv` counts the shipped ones out. + struct [[sysio::table("outqueue")]] outbound_queue_entry { + uint64_t chain_code; + uint32_t queued; + + SYSLIB_SERIALIZE(outbound_queue_entry, (chain_code)(queued)) + }; + + using outqueue_t = sysio::kv::table<"outqueue"_n, outbound_queue_key, outbound_queue_entry>; + /// Singleton holding the monotonic next-attestation-id counter. /// - /// Why a singleton instead of `atts.available_primary_key()`: the - /// `buildenv` outbound-bundle cleanup at the bottom of this file - /// erases the just-bundled `ATTESTATION_STATUS_PROCESSED` rows for - /// the destination `chain_code`. Inbound `deliver()`-inserted rows - /// also carry `status = PROCESSED` (they go straight to dispatch), - /// so the cleanup drains both. Once the atts table is empty, - /// `available_primary_key()` resets to 0, our `std::max(1, ...)` - /// floor bumps it to 1, reusing a prior attestation's audit identity. + /// Why a singleton instead of `atts.available_primary_key()`: + /// `buildenv` erases the rows it ships for the destination + /// `chain_code`, and that outpost's inbound `deliver()`-inserted rows, + /// which carry `status = PROCESSED` (they go straight to dispatch). + /// Once the atts table is empty, `available_primary_key()` resets to + /// 0, our `std::max(1, ...)` floor bumps it to 1, reusing a prior + /// attestation's audit identity. Monotonic ids also keep each + /// `byqueue` lane FIFO. /// /// `att_seq` is a one-row table holding `next` — the next id to /// mint. `mint_att_id()` reads + bumps it atomically. Cleanup diff --git a/contracts/sysio.msgch/src/sysio.msgch.cpp b/contracts/sysio.msgch/src/sysio.msgch.cpp index 941554ca3c..2f333c1caa 100644 --- a/contracts/sysio.msgch/src/sysio.msgch.cpp +++ b/contracts/sysio.msgch/src/sysio.msgch.cpp @@ -108,11 +108,11 @@ constexpr const char* DISPUTE_NO_TIER_ONE_ELECTORATE_LOG = /// Mint the next attestation id from the `attseq` singleton. /// /// Replaces a `std::max(1, atts.available_primary_key())` call -/// at every `attestations_t` insertion site. The `attseq` singleton survives the -/// `buildenv` cleanup of `ATTESTATION_STATUS_PROCESSED` rows, so the -/// monotonic counter keeps advancing across phases even when the atts -/// table is drained. Reusing an erased row's id would give a later attestation -/// the same identity as an earlier audit record. +/// at every `attestations_t` insertion site. The `attseq` singleton survives +/// `buildenv` erasing shipped and inbound rows, so the monotonic counter +/// keeps advancing across phases even when the atts table is drained. +/// Reusing an erased row's id would give a later attestation the same +/// identity as an earlier audit record. /// /// First call materialises the row at `next = 2` and returns `1`. /// Subsequent calls return the current `next` and post-increment. @@ -129,6 +129,12 @@ uint64_t mint_att_id(name self) { return out; } +/// The outpost's `outqueue` row, or an empty one before its first general-lane attestation. +msgch::outbound_queue_entry outbound_queue_of(const msgch::outqueue_t& queues, uint64_t chain_code) { + return queues.try_get(msgch::outbound_queue_key{chain_code}) + .value_or(msgch::outbound_queue_entry{.chain_code = chain_code, .queued = 0}); +} + /// Size of the ACTIVE batch-operator group -- the set that can actually deliver for the current /// epoch. Every consensus threshold derives from THIS, never from the configured /// `operators_per_epoch`. @@ -1430,6 +1436,11 @@ void msgch::queueout(uint64_t chain_code, // the row for. check(data.size() <= MAX_ATTESTATION_DATA_BYTES, "sysio.msgch::queueout: attestation exceeds MAX_ATTESTATION_DATA_BYTES"); + // A general-lane row must also fit beside the schedule lane every envelope carries first, or it + // would stall the general lane behind it for good. + const queue_lane lane = lane_of(attest_type); + check(lane == queue_lane::schedule || data.size() <= MAX_GENERAL_ATTESTATION_DATA_BYTES, + "sysio.msgch::queueout: attestation exceeds MAX_GENERAL_ATTESTATION_DATA_BYTES"); // The chains registry is the ONLY authority on which chain codes exist. // @@ -1438,10 +1449,8 @@ void msgch::queueout(uint64_t chain_code, // sysio.chains, but a governance msig holding one of those authorities can // call queueout directly, and a typo'd code used to abort here. Without it // the row is created READY and is then unreachable forever: epoch::advance - // only fans buildenv to registered active outposts, the PROCESSED drain - // skips mismatched codes, and no prune action exists -- so it is permanent - // sysio-pool RAM, re-walked by the Phase-1 bystatus scan on every buildenv, - // for every outpost, every epoch. + // only fans buildenv to registered active outposts and no prune action + // exists -- so it is permanent sysio-pool RAM. // // The SVM terminal-account estimator that used to live beside this lookup // was deliberately deleted (it made the depot model another chain's packet @@ -1451,6 +1460,17 @@ void msgch::queueout(uint64_t chain_code, chains_tbl.get(sysio::chains::chain_key{sysio::slug_name{chain_code}}, "sysio.msgch::queueout: chain_code is not registered"); + // Backpressure: past the cap, a general-lane row fails its producer's transaction (a `desyndicate` + // reverts whole, burn included) instead of growing the queue. Schedule rows are exempt because + // `sysio.epoch` queues them inside `advance`, which must not fail. + if (lane == queue_lane::general) { + outqueue_t queues(get_self()); + auto queue = outbound_queue_of(queues, chain_code); + check(queue.queued < MAX_QUEUED_ATTESTATIONS, "sysio.msgch::queueout: outbound queue is full for this outpost"); + ++queue.queued; + queues.set(ram_payer, outbound_queue_key{chain_code}, queue); + } + auto now_sec = static_cast(current_time_point().sec_since_epoch()); attestations_t atts(get_self()); @@ -1472,11 +1492,12 @@ void msgch::queueout(uint64_t chain_code, // --------------------------------------------------------------------------- // buildenv — build outbound envelope from READY attestations // -// Packs as many READY attestations as fit under MAX_ENVELOPE_BYTES into a -// single outbound envelope; any that don't fit stay in the table with -// status = READY and ride the next epoch's `buildenv` call. Mirrors the -// Solana (`emit_outbound_inner`) and Ethereum (`emitOutboundEnvelope`) -// packing-loop pattern — never drop, never refuse, always emit what fits. +// Packs this outpost's READY attestations in queue order (schedule lane +// first, FIFO within a lane) into a single outbound envelope, up to +// MAX_ENVELOPE_BYTES and MAX_ENVELOPE_ATTESTATIONS; the rest stay READY and +// ride the next epoch's `buildenv` call. Mirrors the Solana +// (`emit_outbound_inner`) and Ethereum (`emitOutboundEnvelope`) packing-loop +// pattern — never drop, never refuse, always emit what fits. // // Each envelope chains from its predecessor on the same outpost stream: // `previous_envelope_hash` carries the prior emit's canonical epoch digest @@ -1497,36 +1518,57 @@ void msgch::buildenv(uint64_t chain_code) { uint32_t epoch = epoch::current_epoch_index(); attestations_t atts(get_self()); - auto now_sec = static_cast(current_time_point().sec_since_epoch()); // OPP wire timestamps (`MessageHeader.timestamp`, `Envelope.epoch_timestamp`) are milliseconds // since the Unix epoch (opp.proto). WIRE's half-second blocks make sub-second resolution - // meaningful, so emit real milliseconds; `now_sec` stays for the seconds-unit table columns. + // meaningful, so emit real milliseconds. const uint64_t now_ms = static_cast(current_time_point().time_since_epoch().count()) / 1000; - // ── Phase 1: collect candidate READY attestations for this outpost. - // Order is the secondary index's natural order, which is stable across - // epochs — preserves cross-epoch attestation ordering for the receiving - // chain. - std::vector candidate_entries; - std::vector candidate_ids; + // Phase 1: take this outpost's READY rows in queue order, stopping at the first one that would push + // the estimate past MAX_ENVELOPE_BYTES, or at MAX_ENVELOPE_ATTESTATIONS. The walk reads nothing it + // does not ship, so its cost does not grow with the backlog, and the rows it leaves keep their place. + // + // The estimate is conservative. The trim loop below is the source of truth for the encoded size + // invariant. + // + // This pass deliberately does NOT bound the envelope by any destination chain's transaction + // capacity. Solana dispatch is resumable: the outpost settles `[cursor, cursor + dispatch_limit)` + // per terminal call and the relay loops until the cursor drains, so an envelope that needs more + // effect accounts than one Solana transaction can carry is settled across several calls rather + // than refused here. Sizing a WIRE consensus envelope against another chain's packet limit put + // that chain's MTU inside depot consensus; the cursor removes the need. + auto by_queue = atts.get_index<"byqueue"_n>(); + const auto in_ready_queue = [&](const auto& row) { + return row != by_queue.end() && row->chain_code == chain_code && + row->status == AttestationStatus::ATTESTATION_STATUS_READY; + }; - auto status_idx = atts.get_index<"bystatus"_n>(); - for (auto it = status_idx.lower_bound( - static_cast(AttestationStatus::ATTESTATION_STATUS_READY)); - it != status_idx.end() && - it->status == AttestationStatus::ATTESTATION_STATUS_READY; ++it) { - if (it->chain_code != chain_code) continue; + auto ready = by_queue.lower_bound( + queue_key(chain_code, AttestationStatus::ATTESTATION_STATUS_READY, queue_lane::schedule)); + if (!in_ready_queue(ready)) return; + + std::vector entries; + std::vector included_ids; + size_t estimated_bytes = ENVELOPE_BASELINE_BYTES; + for (; in_ready_queue(ready) && entries.size() < MAX_ENVELOPE_ATTESTATIONS; ++ready) { + const size_t entry_bytes = ATTESTATION_OVERHEAD_BYTES + ready->data.size(); + if (estimated_bytes + entry_bytes > MAX_ENVELOPE_BYTES) { + break; + } + estimated_bytes += entry_bytes; opp::AttestationEntry entry; - entry.type = it->type; - entry.data_size = zpp::bits::vuint32_t{static_cast(it->data.size())}; - entry.data = it->data; - candidate_entries.push_back(std::move(entry)); - candidate_ids.push_back(it->id); + entry.type = ready->type; + entry.data_size = zpp::bits::vuint32_t{static_cast(ready->data.size())}; + entry.data = ready->data; + entries.push_back(std::move(entry)); + included_ids.push_back(ready->id); } - if (candidate_entries.empty()) return; + // Invariant, not a reachable failure: `queueout` caps every row at MAX_ATTESTATION_DATA_BYTES, which + // is what the estimate leaves an otherwise empty envelope, so the first row always fits. + check(!entries.empty(), + "sysio.msgch::buildenv: a single READY attestation exceeds the outbound envelope"); const auto op_row = [&]() { sysio::chains::chains_t chains_tbl(CHAINS_ACCOUNT); @@ -1546,39 +1588,6 @@ void msgch::buildenv(uint64_t chain_code) { route_endpoints.end.kind = op_row.kind; route_endpoints.end.id = op_row.external_chain_id; - // Phase 2: estimator-based initial pick. Walk candidates in order, accumulating a conservative byte - // estimate; stop once the next one would push the envelope over MAX_ENVELOPE_BYTES. The trim loop - // below is the source of truth for the encoded size invariant. - // - // This pass deliberately does NOT bound the envelope by any destination chain's transaction - // capacity. Solana dispatch is resumable: the outpost settles `[cursor, cursor + dispatch_limit)` - // per terminal call and the relay loops until the cursor drains, so an envelope that needs more - // effect accounts than one Solana transaction can carry is settled across several calls rather - // than refused here. Sizing a WIRE consensus envelope against another chain's packet limit put - // that chain's MTU inside depot consensus; the cursor removes the need. - size_t included_count = 0; - size_t estimated_bytes = ENVELOPE_BASELINE_BYTES; - for (const auto& entry : candidate_entries) { - const size_t entry_bytes = ATTESTATION_OVERHEAD_BYTES + entry.data.size(); - if (estimated_bytes + entry_bytes > MAX_ENVELOPE_BYTES) { - break; - } - estimated_bytes += entry_bytes; - ++included_count; - } - - // Invariant, not a reachable failure: `queueout` caps every row at MAX_ATTESTATION_DATA_BYTES, which - // is what the estimate leaves an otherwise empty envelope, so the first candidate always fits. - check(included_count > 0, - "sysio.msgch::buildenv: a single READY attestation exceeds the outbound envelope"); - - std::vector entries( - std::make_move_iterator(candidate_entries.begin()), - std::make_move_iterator(candidate_entries.begin() + included_count)); - std::vector included_ids( - candidate_ids.begin(), - candidate_ids.begin() + included_count); - // Chain links come from the previous emit, the newest row for this outpost: one step back from the // first `byoutepoch` entry past it, so retained rows are never walked. No row means genesis: both // links chain from empty. @@ -1664,10 +1673,10 @@ void msgch::buildenv(uint64_t chain_code) { // under-counted relative to the actual zpp::bits encoding for this candidate set), pop the last // entry and retry. This loop -- not the estimator -- is the sole authority on the size invariant. It // replaces the previous post-build hard abort, which left this action in a dead-letter loop on - // estimator drift: the PROCESSED markings would roll back with the aborted tx, the same set would - // re-pack against the same estimator next epoch, and the same check would trip again. Popped entries - // stay READY and ride the next `buildenv` call. PROCESSED marking is deferred until after the loop - // converges so a popped entry never needs a status revert. + // estimator drift: the erasures would roll back with the aborted tx, the same set would re-pack + // against the same estimator next epoch, and the same check would trip again. Popped entries stay + // READY and ride the next `buildenv` call. Erasure waits until the loop converges, so a popped entry + // is never touched. std::vector packed = build_packed(entries); while (packed.size() > MAX_ENVELOPE_BYTES) { // Invariant: the estimate over-counts the real encoding, so one row within the `queueout` cap @@ -1680,16 +1689,20 @@ void msgch::buildenv(uint64_t chain_code) { packed = build_packed(entries); } - // Mark the surviving attestations as PROCESSED. Remaining candidates (popped by the trim loop or - // never picked by the estimator) stay READY for the next epoch's `buildenv` call. - for (uint64_t aid : included_ids) { - auto att_pk = id_key{aid}; - if (atts.contains(att_pk)) { - atts.modify(same_payer, att_pk, [&](auto& a) { - a.status = AttestationStatus::ATTESTATION_STATUS_PROCESSED; - a.processed_timestamp = now_sec; - }); - } + // The shipped rows' bytes are in `packed` now: erase them, and count the general-lane ones out of the + // outpost's queue. Rows the walk stopped before, or the trim loop popped, stay READY for the next + // epoch's `buildenv` call. + uint32_t general_shipped = 0; + for (size_t i = 0; i < included_ids.size(); ++i) { + if (lane_of(entries[i].type) == queue_lane::general) ++general_shipped; + atts.erase(id_key{included_ids[i]}); + } + if (general_shipped > 0) { + outqueue_t queues(get_self()); + auto queue = outbound_queue_of(queues, chain_code); + // Saturating: the count only bounds `queueout`, and a short one must not abort `advance`. + queue.queued -= std::min(queue.queued, general_shipped); + queues.set(ram_payer, outbound_queue_key{chain_code}, queue); } // This envelope's canonical epoch digest: keccak256 over the packed bytes, which equal the @@ -1714,7 +1727,7 @@ void msgch::buildenv(uint64_t chain_code) { // === AUDIT LOG + INLINE CLEANUP OF WORKING STATE === // // Audit-log row mirrors the outbound emit (WIRE → outpost). Then drain the outenvelopes rows the - // outpost has consumed and its PROCESSED attestations (now baked into `packed`). + // outpost has consumed and this outpost's inbound attestations, already dispatched. { // Same endpoints the wire envelope carries (derived above from the // destination's `sysio.chains` row). @@ -1742,16 +1755,13 @@ void msgch::buildenv(uint64_t chain_code) { it->epoch_index, static_cast(chain_code)); } - // Drop the attestations we just consumed (status PROCESSED rows - // for this outpost). They've been bundled into `packed`; the - // bytes are dead weight on chain. - auto processed_idx = atts.get_index<"bystatus"_n>(); - for (auto it = processed_idx.lower_bound( - static_cast(AttestationStatus::ATTESTATION_STATUS_PROCESSED)); - it != processed_idx.end() && - it->status == AttestationStatus::ATTESTATION_STATUS_PROCESSED; ) { - if (it->chain_code != chain_code) { ++it; continue; } - it = processed_idx.erase(std::move(it)); + // Drop this outpost's inbound attestations: `evalcons` stores them PROCESSED and dispatches them + // inline, so their bytes are dead weight on chain. The range holds this outpost's rows only. + for (auto inbound = by_queue.lower_bound( + queue_key(chain_code, AttestationStatus::ATTESTATION_STATUS_PROCESSED, queue_lane::schedule)); + inbound != by_queue.end() && inbound->chain_code == chain_code && + inbound->status == AttestationStatus::ATTESTATION_STATUS_PROCESSED; ) { + inbound = by_queue.erase(std::move(inbound)); } } } diff --git a/contracts/sysio.msgch/sysio.msgch.abi b/contracts/sysio.msgch/sysio.msgch.abi index 91c63abca7..0951fc3e09 100644 --- a/contracts/sysio.msgch/sysio.msgch.abi +++ b/contracts/sysio.msgch/sysio.msgch.abi @@ -308,6 +308,30 @@ } ] }, + { + "name": "outbound_queue_entry", + "base": "", + "fields": [ + { + "name": "chain_code", + "type": "uint64" + }, + { + "name": "queued", + "type": "uint32" + } + ] + }, + { + "name": "outbound_queue_key", + "base": "", + "fields": [ + { + "name": "chain_code", + "type": "uint64" + } + ] + }, { "name": "outpost_consensus_entry", "base": "", @@ -442,19 +466,9 @@ "table_id": 4978, "secondary_indexes": [ { - "name": "bystatus", - "key_type": "uint64", - "table_id": 15348 - }, - { - "name": "bytype", - "key_type": "uint64", - "table_id": 31638 - }, - { - "name": "byepoch", - "key_type": "uint64", - "table_id": 21249 + "name": "byqueue", + "key_type": "uint128", + "table_id": 11308 } ] }, @@ -541,6 +555,14 @@ "key_names": ["chain_code"], "key_types": ["uint64"], "table_id": 51836 + }, + { + "name": "outqueue", + "type": "outbound_queue_entry", + "index_type": "i64", + "key_names": ["chain_code"], + "key_types": ["uint64"], + "table_id": 52512 } ], "ricardian_clauses": [], diff --git a/contracts/sysio.msgch/sysio.msgch.wasm b/contracts/sysio.msgch/sysio.msgch.wasm index cb7d13c961..2c2224c232 100755 Binary files a/contracts/sysio.msgch/sysio.msgch.wasm and b/contracts/sysio.msgch/sysio.msgch.wasm differ diff --git a/contracts/tests/sysio.msgch_chain_tests.cpp b/contracts/tests/sysio.msgch_chain_tests.cpp index e03fc9a5b7..ddb557868f 100644 --- a/contracts/tests/sysio.msgch_chain_tests.cpp +++ b/contracts/tests/sysio.msgch_chain_tests.cpp @@ -141,6 +141,7 @@ constexpr name CHECK_CONSENSUS = "chkcons"_n; /// sysio.msgch table identifiers used by the WNS-16 fixture. namespace msgch_tables { +constexpr name ATTESTATIONS = "attestations"_n; constexpr name ENVELOPES = "envelopes"_n; constexpr name OUTBOUND_ENVELOPES = "outenvelopes"_n; } // namespace msgch_tables @@ -152,6 +153,7 @@ constexpr name BY_OUTPOST_EPOCH = "byoutepoch"_n; /// sysio.msgch ABI type identifiers used by the WNS-16 fixture. namespace msgch_abi_types { +constexpr const char* ATTESTATION_ENTRY = "attestation_entry"; constexpr const char* ENVELOPE_ENTRY = "envelope_entry"; constexpr const char* OUTBOUND_ENVELOPE = "outbound_envelope"; } // namespace msgch_abi_types @@ -166,6 +168,7 @@ constexpr const char* BATCH_OP_NAME = "batch_op_name"; constexpr const char* WINNING_CHECKSUM = "winning_checksum"; constexpr const char* CHECKSUM = "checksum"; constexpr const char* CONSENSUS_REACHED = "consensus_reached"; +constexpr const char* STATUS = "status"; } // namespace msgch_fields /// sysio.chalg table identifiers used by the split-consensus regressions. @@ -773,15 +776,31 @@ class sysio_msgch_chain_tester : public tester { return n; } + /// Attestation rows for `chain_code` with `status`, over the whole table. + uint32_t attestation_status_count(uint64_t chain_code, opp::types::AttestationStatus status) { + uint32_t n = 0; + for (const auto& row : table_rows(MSGCH_ACCOUNT, msgch_tables::ATTESTATIONS, msgch_abi, + msgch_abi_types::ATTESTATION_ENTRY)) { + if (row[msgch_fields::CHAIN_CODE].as_uint64() == chain_code && + row[msgch_fields::STATUS].as() == status) ++n; + } + return n; + } + // -- Inbound envelope builder -- - /// Encode a deliverable envelope carrying one OPERATORS roster echo (dispatch drops - /// the attestation silently; acceptance is still fully observable via `outpcons` and the - /// stored attestation row). The semantic header is derived per the spec — `apply_consensus` - /// drops envelopes whose header fields do not recompute or whose message does not continue the - /// per-outpost message chain. `prev` (previous_envelope_hash), `prev_message_id`, and - /// `env_hash` are raw 32-byte strings (or empty for stream genesis). - std::vector encode_delivery(uint32_t epoch_index, const std::string& att_data, + /// One attestation of an inbound envelope. + struct inbound_attestation { + sysio::opp::types::AttestationType type; + std::string data; + }; + + /// Encode a deliverable envelope carrying `attestations` in one message. The semantic header is + /// derived per the spec — `apply_consensus` drops envelopes whose header fields do not recompute + /// or whose message does not continue the per-outpost message chain. `prev` + /// (previous_envelope_hash), `prev_message_id`, and `env_hash` are raw 32-byte strings (or empty + /// for stream genesis). + std::vector encode_delivery(uint32_t epoch_index, const std::vector& attestations, const std::string& prev = {}, const std::string& prev_message_id = {}, const std::string& env_hash = {}) { @@ -791,16 +810,31 @@ class sysio_msgch_chain_tester : public tester { env.set_epoch_timestamp(1'775'612'516'983ULL); if (!prev.empty()) env.set_previous_envelope_hash(prev); if (!env_hash.empty()) env.set_envelope_hash(env_hash); - auto* att = env.add_messages()->mutable_payload()->add_attestations(); - att->set_type(sysio::opp::types::ATTESTATION_TYPE_OPERATORS); - att->set_data(att_data); - att->set_data_size(static_cast(att_data.size())); + auto* payload = env.add_messages()->mutable_payload(); + for (const auto& attestation : attestations) { + auto* att = payload->add_attestations(); + att->set_type(attestation.type); + att->set_data(attestation.data); + att->set_data_size(static_cast(attestation.data.size())); + } oracle::finalize_header(*env.mutable_messages(0), prev_message_id, 1'775'612'516'983ULL); std::vector out(env.ByteSizeLong()); env.SerializeToArray(out.data(), static_cast(out.size())); return out; } + /// Encode a deliverable envelope carrying one OPERATORS roster echo (dispatch drops + /// the attestation silently; acceptance is still fully observable via `outpcons` and the + /// stored attestation row). + std::vector encode_delivery(uint32_t epoch_index, const std::string& att_data, + const std::string& prev = {}, + const std::string& prev_message_id = {}, + const std::string& env_hash = {}) { + const std::vector attestations{ + {sysio::opp::types::ATTESTATION_TYPE_OPERATORS, att_data}}; + return encode_delivery(epoch_index, attestations, prev, prev_message_id, env_hash); + } + /// Decode the raw bytes back into the pb Envelope (asserts success). sysio::opp::Envelope decode_envelope(const std::vector& raw) { sysio::opp::Envelope env; @@ -818,8 +852,8 @@ class sysio_msgch_chain_tester : public tester { /// `chain_code`. Read from the surviving `outenvelopes` row, NOT the msgch `attestations` /// working table — `advance` queues its schedule attestations via `queueout` and then fans out /// `buildenv` inline in the SAME transaction, and `buildenv` packs the READY rows into the - /// outbound envelope, marks them PROCESSED, and ERASES them before the transaction ends. The - /// packed envelope (one-deep per outpost) is therefore the only durable copy of what `advance` + /// outbound envelope and ERASES them before the transaction ends. The + /// newest `outenvelopes` row is therefore the only durable copy of what `advance` /// emitted; scanning `attestations` after an advance always finds zero rows. sysio::opp::attestations::BatchOperatorGroups shipped_batch_operator_groups(uint64_t chain_code) { auto row = find_outbound_envelope(chain_code); @@ -1514,6 +1548,53 @@ BOOST_FIXTURE_TEST_CASE(buildenv_handles_unacknowledged_backlog, sysio_msgch_cha BOOST_REQUIRE(!outbound_envelope_for_epoch(ETH_OUTPOST_ID, next_epoch).is_null()); } FC_LOG_AND_RETHROW() } +// --------------------------------------------------------------------------- +// Inbound cleanup: buildenv erases only its own outpost's dispatched rows. +// --------------------------------------------------------------------------- + +/// WIRE-415: `evalcons` stores every inbound attestation PROCESSED, and `buildenv` erases the +/// destination outpost's PROCESSED rows through the `byqueue` range for that outpost. Each envelope +/// carries one attestation per lane, so a range that misses a lane or reads another status leaves a +/// row behind. Another outpost's rows, PROCESSED or READY, survive until its own `buildenv`. +BOOST_FIXTURE_TEST_CASE(buildenv_erases_only_its_outposts_processed_rows, sysio_msgch_chain_tester) { try { + constexpr auto processed = opp::types::AttestationStatus::ATTESTATION_STATUS_PROCESSED; + constexpr auto ready = opp::types::AttestationStatus::ATTESTATION_STATUS_READY; + bootstrap(); + + // Each outpost's first envelope reaches consensus on the single bootstrapped operator's delivery. + // OPERATORS is a schedule-lane type and OPERATOR_ACTION a general-lane one; dispatch drops both. + const uint32_t epoch = current_epoch(); + const std::vector attestations{ + {opp::types::ATTESTATION_TYPE_OPERATORS, "schedule-lane"}, + {opp::types::ATTESTATION_TYPE_OPERATOR_ACTION, "general-lane"}}; + for (const uint64_t outpost : {ETH_OUTPOST_ID, SOL_OUTPOST_ID}) { + BOOST_REQUIRE_EQUAL(success(), deliver(outpost, encode_delivery(epoch, attestations))); + } + produce_blocks(); + for (const uint64_t outpost : {ETH_OUTPOST_ID, SOL_OUTPOST_ID}) { + BOOST_REQUIRE_EQUAL(attestations.size(), attestation_status_count(outpost, processed)); + // A READY row on each outpost: `buildenv` builds nothing without one. + BOOST_REQUIRE_EQUAL(success(), queueout(outpost, opp::types::ATTESTATION_TYPE_DESYNDICATE_LIQ, + std::vector(8, '\x01'))); + } + produce_blocks(); + + // ETH's build ships ETH's READY row and erases both of its inbound rows. SOL's are untouched. + BOOST_REQUIRE_EQUAL(success(), buildenv(ETH_OUTPOST_ID)); + produce_blocks(); + BOOST_REQUIRE_EQUAL(0u, attestation_status_count(ETH_OUTPOST_ID, processed)); + BOOST_REQUIRE_EQUAL(0u, attestation_status_count(ETH_OUTPOST_ID, ready)); + BOOST_REQUIRE_EQUAL(attestations.size(), attestation_status_count(SOL_OUTPOST_ID, processed)); + BOOST_REQUIRE_EQUAL(1u, attestation_status_count(SOL_OUTPOST_ID, ready)); + + // The real advance path builds SOL's envelope, which erases SOL's inbound rows in turn. + advance_one_epoch(); + for (const uint64_t outpost : {ETH_OUTPOST_ID, SOL_OUTPOST_ID}) { + BOOST_REQUIRE_EQUAL(0u, attestation_status_count(outpost, processed)); + BOOST_REQUIRE_EQUAL(0u, attestation_status_count(outpost, ready)); + } +} FC_LOG_AND_RETHROW() } + // --------------------------------------------------------------------------- // Inbound: apply_consensus records and verifies the per-outpost chain tip. // --------------------------------------------------------------------------- diff --git a/contracts/tests/sysio.msgch_tests.cpp b/contracts/tests/sysio.msgch_tests.cpp index 98d7e64ebd..e777fb8464 100644 --- a/contracts/tests/sysio.msgch_tests.cpp +++ b/contracts/tests/sysio.msgch_tests.cpp @@ -3,6 +3,7 @@ #include #include #include +#include #include #include @@ -19,6 +20,20 @@ using mvo = fc::mutable_variant_object; namespace { +/// A one-of-a-kind attestation payload carrying `index`, so repeated `queueout`s never collapse as +/// duplicate transactions and a shipped row can be traced back to its `queueout`. +std::vector indexed_payload(uint32_t index) { + return {static_cast(index >> 24), static_cast(index >> 16), + static_cast(index >> 8), static_cast(index)}; +} + +/// The index an `indexed_payload` carries. +uint32_t payload_index(const std::string& data) { + BOOST_REQUIRE_EQUAL(data.size(), 4u); + return (uint32_t{static_cast(data[0])} << 24) | (uint32_t{static_cast(data[1])} << 16) | + (uint32_t{static_cast(data[2])} << 8) | uint32_t{static_cast(data[3])}; +} + } // anonymous namespace class sysio_msgch_tester : public tester { @@ -317,6 +332,45 @@ class sysio_msgch_envlog_tester : public tester { ); } + /// Queue `count` general-lane rows for `chain_code`, with payloads `indexed_payload(first)` onward, + /// producing a block every `per_block` so the pending block stays inside its limits. + void queue_general(uint64_t chain_code, uint32_t first, uint32_t count, uint32_t per_block = 20) { + for (uint32_t i = 0; i < count; ++i) { + BOOST_REQUIRE_EQUAL(success(), queueout_with_data(chain_code, opp::types::ATTESTATION_TYPE_DESYNDICATE_LIQ, + indexed_payload(first + i))); + if ((i + 1) % per_block == 0) produce_block(); + } + produce_block(); + } + + /// The outpost's general-lane READY count in `outqueue`; 0 before its first such row. + uint32_t queued_attestations(uint64_t chain_code) { + const auto data = get_row_by_id(MSGCH_ACCOUNT, MSGCH_ACCOUNT, "outqueue"_n, chain_code); + if (data.empty()) return 0; + const auto row = msgch_abi.binary_to_variant( + "outbound_queue_entry", data, fixture_abi_yield_function()); + return row["queued"].as(); + } + + /// The attestations of the envelope `buildenv` last built for `chain_code`, in shipped order. Earlier + /// emits the outpost has not acknowledged stay in `outenvelopes`, so this reads the newest row. + std::vector shipped_attestations(uint64_t chain_code, uint64_t scan_until = 16) { + fc::variant newest; + for (uint64_t id = 0; id < scan_until; ++id) { + const 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, fixture_abi_yield_function()); + if (row["chain_code"].as_uint64() == chain_code) newest = std::move(row); + } + BOOST_REQUIRE_MESSAGE(!newest.is_null(), "no outbound envelope for the outpost"); + const auto raw = newest["raw_envelope"].as>(); + opp::Envelope env; + BOOST_REQUIRE(env.ParseFromArray(raw.data(), static_cast(raw.size()))); + BOOST_REQUIRE_EQUAL(env.messages_size(), 1); + const auto& shipped = env.messages(0).payload().attestations(); + return {shipped.begin(), shipped.end()}; + } + /// Return the newest outbound envelope row in a small id scan (the chain tip). fc::variant find_outbound_envelope(uint64_t scan_until = 16) { fc::variant newest; @@ -373,6 +427,26 @@ constexpr size_t ATTESTATION_OVERHEAD_BYTES = 24; constexpr size_t MAX_ATTESTATION_DATA_BYTES = MAX_ENVELOPE_BYTES - ENVELOPE_BASELINE_BYTES - ATTESTATION_OVERHEAD_BYTES; +/// Mirrors of the outbound queue bounds in `sysio.msgch.hpp`. +constexpr size_t MAX_ENVELOPE_ATTESTATIONS = 64; +constexpr uint32_t MAX_QUEUED_ATTESTATIONS = 512; +constexpr size_t SCHEDULE_LANE_BYTES = 24 * 1024; +constexpr size_t MAX_GENERAL_ATTESTATION_DATA_BYTES = + MAX_ENVELOPE_BYTES - ENVELOPE_BASELINE_BYTES - SCHEDULE_LANE_BYTES - ATTESTATION_OVERHEAD_BYTES; + +/// Mirrors of the window ceilings `sysio.epoch::setconfig` enforces in `sysio.epoch.hpp`. +constexpr uint32_t MAX_BATCH_OP_GROUPS = 8; +constexpr uint32_t MAX_OPERATORS_PER_EPOCH = 7; + +/// `register_outpost(CHAIN_KIND_SVM, ...)` registers the spelling `"SOL"`. +constexpr uint64_t SOL_OUTPOST_ID = "SOL"_s.value; + +/// A general-lane type (anything but OPERATORS and BATCH_OPERATOR_GROUPS), as `queue_general` queues. +constexpr auto GENERAL_LANE_TYPE = opp::types::ATTESTATION_TYPE_DESYNDICATE_LIQ; + +constexpr auto QUEUE_FULL_ERROR = + "assertion failure with message: sysio.msgch::queueout: outbound queue is full for this outpost"; + } // anonymous namespace /// Smoke: queueout + buildenv writes one row to `envlog` with the @@ -668,6 +742,225 @@ BOOST_FIXTURE_TEST_CASE(buildenv_ships_an_attestation_at_the_cap_alone, sysio_ms BOOST_REQUIRE_EQUAL(1u, count_ready_attestations(ETH_OUTPOST_ID, /*scan_until=*/4)); } FC_LOG_AND_RETHROW() } +/// WIRE-415: OPERATORS and BATCH_OPERATOR_GROUPS ship ahead of a general-lane backlog queued before +/// them, so the rosters an outpost needs for the next epoch never wait behind it. The backlog keeps +/// its `queueout` order. +BOOST_FIXTURE_TEST_CASE(buildenv_ships_the_schedule_lane_ahead_of_a_backlog, sysio_msgch_envlog_tester) { try { + bootstrap_epoch_config(/*retention=*/200); + register_outpost(opp::types::CHAIN_KIND_EVM, 31337); + produce_blocks(); + + constexpr uint32_t BACKLOG = 4; + queue_general(ETH_OUTPOST_ID, /*first=*/0, BACKLOG); + BOOST_REQUIRE_EQUAL(success(), queueout_with_data(ETH_OUTPOST_ID, opp::types::ATTESTATION_TYPE_OPERATORS, + indexed_payload(100))); + BOOST_REQUIRE_EQUAL(success(), queueout_with_data(ETH_OUTPOST_ID, + opp::types::ATTESTATION_TYPE_BATCH_OPERATOR_GROUPS, + indexed_payload(101))); + produce_blocks(); + + BOOST_REQUIRE_EQUAL(success(), buildenv(ETH_OUTPOST_ID)); + produce_blocks(); + + const auto shipped = shipped_attestations(ETH_OUTPOST_ID); + BOOST_REQUIRE_EQUAL(shipped.size(), BACKLOG + 2); + BOOST_REQUIRE_EQUAL(shipped[0].type(), opp::types::ATTESTATION_TYPE_OPERATORS); + BOOST_REQUIRE_EQUAL(shipped[1].type(), opp::types::ATTESTATION_TYPE_BATCH_OPERATOR_GROUPS); + for (uint32_t i = 0; i < BACKLOG; ++i) { + BOOST_REQUIRE_EQUAL(shipped[2 + i].type(), GENERAL_LANE_TYPE); + BOOST_REQUIRE_EQUAL(payload_index(shipped[2 + i].data()), i); + } + BOOST_REQUIRE_EQUAL(0u, queued_attestations(ETH_OUTPOST_ID)); +} FC_LOG_AND_RETHROW() } + +/// WIRE-415: an envelope carries at most MAX_ENVELOPE_ATTESTATIONS rows however small they are, so +/// `buildenv`'s work per outpost stays fixed. The rest ship next time, in order. +BOOST_FIXTURE_TEST_CASE(buildenv_stops_at_the_attestation_cap, sysio_msgch_envlog_tester) { try { + bootstrap_epoch_config(/*retention=*/200); + register_outpost(opp::types::CHAIN_KIND_EVM, 31337); + produce_blocks(); + + constexpr uint32_t LEFT_OVER = 3; + constexpr uint32_t QUEUED = MAX_ENVELOPE_ATTESTATIONS + LEFT_OVER; + queue_general(ETH_OUTPOST_ID, /*first=*/0, QUEUED); + BOOST_REQUIRE_EQUAL(QUEUED, queued_attestations(ETH_OUTPOST_ID)); + + BOOST_REQUIRE_EQUAL(success(), buildenv(ETH_OUTPOST_ID)); + produce_blocks(); + const auto first = shipped_attestations(ETH_OUTPOST_ID); + BOOST_REQUIRE_EQUAL(first.size(), MAX_ENVELOPE_ATTESTATIONS); + for (uint32_t i = 0; i < first.size(); ++i) { + BOOST_REQUIRE_EQUAL(payload_index(first[i].data()), i); + } + BOOST_REQUIRE_EQUAL(LEFT_OVER, count_ready_attestations(ETH_OUTPOST_ID, /*scan_until=*/QUEUED + 2)); + BOOST_REQUIRE_EQUAL(LEFT_OVER, queued_attestations(ETH_OUTPOST_ID)); + + BOOST_REQUIRE_EQUAL(success(), buildenv(ETH_OUTPOST_ID)); + produce_blocks(); + const auto second = shipped_attestations(ETH_OUTPOST_ID); + BOOST_REQUIRE_EQUAL(second.size(), LEFT_OVER); + for (uint32_t i = 0; i < second.size(); ++i) { + BOOST_REQUIRE_EQUAL(payload_index(second[i].data()), MAX_ENVELOPE_ATTESTATIONS + i); + } + BOOST_REQUIRE_EQUAL(0u, count_ready_attestations(ETH_OUTPOST_ID, /*scan_until=*/QUEUED + 2)); + BOOST_REQUIRE_EQUAL(0u, queued_attestations(ETH_OUTPOST_ID)); +} FC_LOG_AND_RETHROW() } + +/// WIRE-415: `buildenv` takes only its own outpost's rows; another outpost's queue and count are +/// untouched until its own `buildenv`. +BOOST_FIXTURE_TEST_CASE(buildenv_ships_only_its_own_outposts_rows, sysio_msgch_envlog_tester) { try { + bootstrap_epoch_config(/*retention=*/200); + register_outpost(opp::types::CHAIN_KIND_EVM, 31337); + register_outpost(opp::types::CHAIN_KIND_SVM, 1); + produce_blocks(); + + queue_general(SOL_OUTPOST_ID, /*first=*/0, /*count=*/2); + queue_general(ETH_OUTPOST_ID, /*first=*/2, /*count=*/1); + queue_general(SOL_OUTPOST_ID, /*first=*/3, /*count=*/1); + + BOOST_REQUIRE_EQUAL(success(), buildenv(ETH_OUTPOST_ID)); + produce_blocks(); + const auto eth = shipped_attestations(ETH_OUTPOST_ID); + BOOST_REQUIRE_EQUAL(eth.size(), 1u); + BOOST_REQUIRE_EQUAL(payload_index(eth[0].data()), 2u); + BOOST_REQUIRE_EQUAL(3u, count_ready_attestations(SOL_OUTPOST_ID, /*scan_until=*/8)); + BOOST_REQUIRE_EQUAL(3u, queued_attestations(SOL_OUTPOST_ID)); + BOOST_REQUIRE_EQUAL(0u, queued_attestations(ETH_OUTPOST_ID)); + + BOOST_REQUIRE_EQUAL(success(), buildenv(SOL_OUTPOST_ID)); + produce_blocks(); + const auto sol = shipped_attestations(SOL_OUTPOST_ID); + BOOST_REQUIRE_EQUAL(sol.size(), 3u); + BOOST_REQUIRE_EQUAL(payload_index(sol[0].data()), 0u); + BOOST_REQUIRE_EQUAL(payload_index(sol[1].data()), 1u); + BOOST_REQUIRE_EQUAL(payload_index(sol[2].data()), 3u); + BOOST_REQUIRE_EQUAL(0u, queued_attestations(SOL_OUTPOST_ID)); +} FC_LOG_AND_RETHROW() } + +/// WIRE-415: the largest window's schedule rows fit SCHEDULE_LANE_BYTES, so a general row at +/// MAX_GENERAL_ATTESTATION_DATA_BYTES always ships beside them. Built at the extremes `sysio.epoch` allows: +/// 8 groups of 7, 13-character names, and an EVM and an SVM link on every operator. +BOOST_AUTO_TEST_CASE(the_largest_windows_schedule_fits_its_lane_budget) { try { + constexpr size_t EVM_KEY_BYTES = 33; // a compressed EM key, as authex stores it + constexpr size_t SVM_KEY_BYTES = 32; // an ED key + const std::string longest_name = "zzzzzzzzzzzzj"; + + opp::attestations::Operators roster; + opp::attestations::BatchOperatorGroups groups; + groups.set_active_group_index(MAX_BATCH_OP_GROUPS - 1); + groups.set_epoch_index(std::numeric_limits::max()); + groups.set_epoch_duration_sec(std::numeric_limits::max()); + for (uint32_t g = 0; g < MAX_BATCH_OP_GROUPS; ++g) { + auto* group = groups.add_groups(); + for (uint32_t m = 0; m < MAX_OPERATORS_PER_EPOCH; ++m) { + auto* member = group->add_operators(); + member->set_kind(opp::types::CHAIN_KIND_WIRE); + member->set_address(longest_name); + + auto* entry = roster.add_operators(); + entry->mutable_account()->set_name(longest_name); + entry->set_type(opp::types::OPERATOR_TYPE_BATCH); + entry->set_status(opp::types::OPERATOR_STATUS_SLASHED); // the widest status varint + auto* evm = entry->add_addresses(); + evm->set_kind(opp::types::CHAIN_KIND_EVM); + evm->set_address(std::string(EVM_KEY_BYTES, '\xff')); + auto* svm = entry->add_addresses(); + svm->set_kind(opp::types::CHAIN_KIND_SVM); + svm->set_address(std::string(SVM_KEY_BYTES, '\xff')); + } + } + + const size_t schedule_bytes = + 2 * ATTESTATION_OVERHEAD_BYTES + roster.ByteSizeLong() + groups.ByteSizeLong(); + BOOST_TEST_MESSAGE("largest window's schedule lane = " << schedule_bytes << " bytes"); + BOOST_REQUIRE_LE(schedule_bytes, SCHEDULE_LANE_BYTES); +} FC_LOG_AND_RETHROW() } + +/// WIRE-415: a general-lane row may carry at most MAX_GENERAL_ATTESTATION_DATA_BYTES, what fits beside a +/// full schedule lane. A schedule row keeps the whole-envelope cap. +BOOST_FIXTURE_TEST_CASE(queueout_caps_general_rows_to_fit_beside_the_schedule, sysio_msgch_envlog_tester) { try { + bootstrap_epoch_config(/*retention=*/200); + register_outpost(opp::types::CHAIN_KIND_EVM, 31337); + produce_blocks(); + + const std::vector over_general_cap(MAX_GENERAL_ATTESTATION_DATA_BYTES + 1, 0x42); + BOOST_REQUIRE_EQUAL( + error("assertion failure with message: " + "sysio.msgch::queueout: attestation exceeds MAX_GENERAL_ATTESTATION_DATA_BYTES"), + queueout_with_data(ETH_OUTPOST_ID, GENERAL_LANE_TYPE, over_general_cap)); + BOOST_REQUIRE_EQUAL(success(), queueout_with_data(ETH_OUTPOST_ID, GENERAL_LANE_TYPE, + std::vector(MAX_GENERAL_ATTESTATION_DATA_BYTES, 0x42))); + BOOST_REQUIRE_EQUAL(success(), queueout_with_data(ETH_OUTPOST_ID, opp::types::ATTESTATION_TYPE_OPERATORS, + over_general_cap)); +} FC_LOG_AND_RETHROW() } + +/// WIRE-415: a general row at its cap ships in the same envelope as a schedule lane at its budget, so +/// the general lane moves every epoch. +BOOST_FIXTURE_TEST_CASE(buildenv_ships_a_general_row_at_its_cap_beside_a_full_schedule, + sysio_msgch_envlog_tester) { try { + bootstrap_epoch_config(/*retention=*/200); + register_outpost(opp::types::CHAIN_KIND_EVM, 31337); + produce_blocks(); + + // The schedule budget split between the two schedule rows, each charged its overhead. + constexpr size_t BATCH_OPERATOR_GROUPS_BYTES = 4 * 1024; + constexpr size_t OPERATORS_BYTES = + SCHEDULE_LANE_BYTES - 2 * ATTESTATION_OVERHEAD_BYTES - BATCH_OPERATOR_GROUPS_BYTES; + BOOST_REQUIRE_EQUAL(success(), queueout_with_data(ETH_OUTPOST_ID, GENERAL_LANE_TYPE, + std::vector(MAX_GENERAL_ATTESTATION_DATA_BYTES, 0x42))); + BOOST_REQUIRE_EQUAL(success(), queueout_with_data(ETH_OUTPOST_ID, opp::types::ATTESTATION_TYPE_OPERATORS, + std::vector(OPERATORS_BYTES, 0x43))); + BOOST_REQUIRE_EQUAL(success(), queueout_with_data(ETH_OUTPOST_ID, + opp::types::ATTESTATION_TYPE_BATCH_OPERATOR_GROUPS, + std::vector(BATCH_OPERATOR_GROUPS_BYTES, 0x44))); + produce_blocks(); + + BOOST_REQUIRE_EQUAL(success(), buildenv(ETH_OUTPOST_ID)); + produce_blocks(); + + const auto shipped = shipped_attestations(ETH_OUTPOST_ID); + BOOST_REQUIRE_EQUAL(shipped.size(), 3u); + BOOST_REQUIRE_EQUAL(shipped[0].type(), opp::types::ATTESTATION_TYPE_OPERATORS); + BOOST_REQUIRE_EQUAL(shipped[1].type(), opp::types::ATTESTATION_TYPE_BATCH_OPERATOR_GROUPS); + BOOST_REQUIRE_EQUAL(shipped[2].type(), GENERAL_LANE_TYPE); + BOOST_REQUIRE_EQUAL(shipped[2].data().size(), MAX_GENERAL_ATTESTATION_DATA_BYTES); + BOOST_REQUIRE_EQUAL(0u, count_ready_attestations(ETH_OUTPOST_ID, /*scan_until=*/6)); +} FC_LOG_AND_RETHROW() } + +/// WIRE-415: an outpost holds at most MAX_QUEUED_ATTESTATIONS general-lane rows; the next one fails its +/// producer's transaction. Schedule rows and other outposts are unaffected, and shipping rows frees +/// exactly that many places. +BOOST_FIXTURE_TEST_CASE(queueout_refuses_general_rows_past_the_queue_cap, sysio_msgch_envlog_tester) { try { + bootstrap_epoch_config(/*retention=*/200); + register_outpost(opp::types::CHAIN_KIND_EVM, 31337); + register_outpost(opp::types::CHAIN_KIND_SVM, 1); + produce_blocks(); + + uint32_t next_index = 0; + queue_general(ETH_OUTPOST_ID, next_index, MAX_QUEUED_ATTESTATIONS); + next_index += MAX_QUEUED_ATTESTATIONS; + BOOST_REQUIRE_EQUAL(MAX_QUEUED_ATTESTATIONS, queued_attestations(ETH_OUTPOST_ID)); + + BOOST_REQUIRE_EQUAL(error(QUEUE_FULL_ERROR), + queueout_with_data(ETH_OUTPOST_ID, GENERAL_LANE_TYPE, indexed_payload(next_index++))); + BOOST_REQUIRE_EQUAL(success(), queueout_with_data(ETH_OUTPOST_ID, opp::types::ATTESTATION_TYPE_OPERATORS, + indexed_payload(next_index++))); + queue_general(SOL_OUTPOST_ID, next_index++, /*count=*/1); + BOOST_REQUIRE_EQUAL(MAX_QUEUED_ATTESTATIONS, queued_attestations(ETH_OUTPOST_ID)); + + // The OPERATORS row ships first, then MAX_ENVELOPE_ATTESTATIONS - 1 general rows. + BOOST_REQUIRE_EQUAL(success(), buildenv(ETH_OUTPOST_ID)); + produce_blocks(); + constexpr uint32_t FREED = MAX_ENVELOPE_ATTESTATIONS - 1; + BOOST_REQUIRE_EQUAL(MAX_QUEUED_ATTESTATIONS - FREED, queued_attestations(ETH_OUTPOST_ID)); + + queue_general(ETH_OUTPOST_ID, next_index, FREED); + next_index += FREED; + BOOST_REQUIRE_EQUAL(MAX_QUEUED_ATTESTATIONS, queued_attestations(ETH_OUTPOST_ID)); + BOOST_REQUIRE_EQUAL(error(QUEUE_FULL_ERROR), + queueout_with_data(ETH_OUTPOST_ID, GENERAL_LANE_TYPE, indexed_payload(next_index++))); +} FC_LOG_AND_RETHROW() } + // queueout carries no ABI-level auth. Without the depot-contract gate, any account could call it // directly and inject a forged READY attestation that buildenv() then packs into the depot's // group-signed outbound envelope (a forged DESYNDICATE_LIQ the outpost executes). diff --git a/contracts/tests/sysio.synd_tests.cpp b/contracts/tests/sysio.synd_tests.cpp index bd4d4b93aa..ea86b06249 100644 --- a/contracts/tests/sysio.synd_tests.cpp +++ b/contracts/tests/sysio.synd_tests.cpp @@ -1472,6 +1472,32 @@ BOOST_FIXTURE_TEST_CASE(desyndicate_burns_and_queues_the_attestation, sysio_synd BOOST_REQUIRE_EQUAL(static_cast(150 * UNIT), supply()); } FC_LOG_AND_RETHROW() +// WIRE-415: once the outpost's outbound queue is full, `queueout` refuses the DESYNDICATE_LIQ and the +// whole `desyndicate` reverts: the holder keeps the shadow, nothing burns, and no `returns` row is +// written. The holder retries once `buildenv` frees a place. +BOOST_FIXTURE_TEST_CASE(desyndicate_reverts_whole_when_the_outbound_queue_is_full, sysio_synd_tester) try { + constexpr uint32_t MAX_QUEUED_ATTESTATIONS = 512; // mirrors sysio.msgch.hpp + constexpr uint32_t QUEUEOUTS_PER_BLOCK = 20; + BOOST_REQUIRE_EQUAL(success(), liq_mint("alice"_n, LIQSOL, 100 * UNIT)); + BOOST_REQUIRE_EQUAL(success(), setconfig(SOLANA, LIQSOL, {})); + BOOST_REQUIRE_EQUAL(success(), link_svm("alice"_n, ed_key())); + + for (uint32_t i = 0; i < MAX_QUEUED_ATTESTATIONS; ++i) { + BOOST_REQUIRE_EQUAL(success(), push(MSGCH_ACCOUNT, msgch_abi_ser, MSGCH_ACCOUNT, "queueout"_n, mvo() + ("chain_code", slug_value(SOLANA)) + ("attest_type", static_cast(sysio::opp::types::ATTESTATION_TYPE_DESYNDICATE_LIQ)) + ("data", std::vector{static_cast(i >> 8), static_cast(i)}))); + if ((i + 1) % QUEUEOUTS_PER_BLOCK == 0) produce_block(); + } + produce_block(); + + BOOST_REQUIRE_EQUAL(wasm_assert_msg("sysio.msgch::queueout: outbound queue is full for this outpost"), + desyndicate("alice"_n, 40 * UNIT)); + BOOST_REQUIRE_EQUAL(static_cast(100 * UNIT), liq_balance("alice"_n)); + BOOST_REQUIRE_EQUAL(static_cast(100 * UNIT), supply()); + BOOST_REQUIRE(return_row(1).is_null()); +} FC_LOG_AND_RETHROW() + /// Outstanding returns retain the original beneficiary and exact obligation, without lifetime totals. BOOST_FIXTURE_TEST_CASE(outstanding_returns_record_the_original_obligation, sysio_synd_tester) try { BOOST_REQUIRE_EQUAL(success(), liq_mint("alice"_n, LIQSOL, 100 * UNIT)); diff --git a/docs/contract-upgrade-order.md b/docs/contract-upgrade-order.md index 5af5d5bddc..8bd5d7cb70 100644 --- a/docs/contract-upgrade-order.md +++ b/docs/contract-upgrade-order.md @@ -356,3 +356,11 @@ LIQ earned-yield distribution and their normal claim actions remain. `sysio.synd::setconfig` requires `min_desyndicate` as an ordinary uint64, and `syndconfig` stores it as an ordinary field. Bootstrap schema version 1 contains chains, tokens, LIQ pools and syndications; it has no external-reserve or underwriting configuration. + +## WIRE-415 pre-launch deployment + +Deploy this `sysio.msgch` on fresh state and regenerate the SDK contract types with it. The +`attestations` table's `bystatus`, `bytype` and `byepoch` indexes are replaced by `byqueue`, and a row +written before the change has no `byqueue` entry, so `buildenv` never finds it: a queued +`DESYNDICATE_LIQ` would never ship, and an inbound row would never be erased. There is no migration +path. diff --git a/docs/sysio-synd.md b/docs/sysio-synd.md index b9f4109858..fa72602d0e 100644 --- a/docs/sysio-synd.md +++ b/docs/sysio-synd.md @@ -154,6 +154,8 @@ banked, and a later delivery pays it. Parked funds have already paid the fee and the next request id from `syndcounters`. `returns` retains the original holder, destination key, outpost, token and exact net amount until its external outcome is established. The attestation carries `total_syndicated`, the live outstanding shadow after the burn; no historical total is stored. + While the outpost already has `sysio.msgch`'s `MAX_QUEUED_ATTESTATIONS` (512) outbound attestations + waiting, `queueout` refuses it and the whole `desyndicate` reverts, burn included. The burn is final. An outpost either pays the release, stores it and pays it later, or skips it: @@ -526,7 +528,8 @@ desyndication. The row must exist, with the buckets set, before the pair carries delivered, and `sysio.liq::claim` pays it. - `desyndicate` takes the whole quantity from the holder's balance and releases the quantity less the desyndication fee on the outpost. It is refused when the pair's desyndication budget is too low now; - it refills every depot epoch. + it refills every depot epoch. It is also refused while the outpost's outbound queue is full, which + drains every depot epoch. ## What an operator sees