From 3e334ceac00e53da8070c3af9237aed19f3f15ee Mon Sep 17 00:00:00 2001 From: echobt <154886644+echobt@users.noreply.github.com> Date: Tue, 11 Aug 2026 14:50:10 +0000 Subject: [PATCH 1/2] feat(weights): tip-track /v1/weights/latest via leaf supersede + reseal Make sealed tip weights follow live Design/Prism scores: supersede leaves on digest change, reseal tip with revision++, and continuously re-emit tip leaves. Validators still Match only sealed:true. --- crates/challenge-common/src/submit.rs | 6 +- ...9819526eca2c8fbf5840d857a0c2aad37edeb.json | 32 --------- .../0017_raw_weight_tip_supersede.sql | 56 +++++++++++++++ crates/db/src/lib.rs | 5 +- crates/db/src/store.rs | 65 +++++++++-------- crates/db/tests/gateway_store.rs | 31 +++++++- crates/design-challenge-task/src/emit.rs | 37 +++++++--- crates/design-challenge/src/lib.rs | 11 ++- crates/design-store/src/store.rs | 25 ++++--- crates/gateway-core/src/weights_store.rs | 40 ++++++++--- crates/gateway-store-pg/src/lib.rs | 11 ++- crates/gateway/src/sealer.rs | 35 ++++++--- crates/gateway/src/weights.rs | 71 ++++++++++++++++--- crates/gateway/tests/raw_weights.rs | 53 ++++++++++++-- crates/gateway/tests/sealer.rs | 40 +++++++++++ crates/prism-emit/src/lib.rs | 50 ++++++++++--- crates/prism-emit/tests/epoch_semantics.rs | 8 +-- deploy/AGENTS.md | 2 +- deploy/scripts/prod-real-seal.sh | 12 +++- deploy/systemd/base-real-seal.timer | 10 +-- docs/ARCHITECTURE.md | 6 +- docs/BUNDLE_SPEC.md | 5 +- docs/PRISM.md | 55 +++++++------- 23 files changed, 484 insertions(+), 182 deletions(-) delete mode 100644 crates/db/.sqlx/query-14604982a6b089a47e32e4fa7189819526eca2c8fbf5840d857a0c2aad37edeb.json create mode 100644 crates/db/migrations/0017_raw_weight_tip_supersede.sql diff --git a/crates/challenge-common/src/submit.rs b/crates/challenge-common/src/submit.rs index a25840989..a0cb39ae8 100644 --- a/crates/challenge-common/src/submit.rs +++ b/crates/challenge-common/src/submit.rs @@ -111,9 +111,9 @@ impl GatewayClient { /// POST one signed leaf. Retries 5xx and transport errors. /// - /// Idempotency: HTTP 409 (already present) is success — never submits a - /// conflicting `ScoreOrAbsence` for the same key from this client path; - /// callers must not change the leaf between retries. + /// Idempotency: HTTP 409 (identical digest already present) is success. + /// Tip re-emits with a changed score/digest return 202 (supersede) and + /// are also success. Callers may change tip leaves between ticks. /// /// # Errors /// diff --git a/crates/db/.sqlx/query-14604982a6b089a47e32e4fa7189819526eca2c8fbf5840d857a0c2aad37edeb.json b/crates/db/.sqlx/query-14604982a6b089a47e32e4fa7189819526eca2c8fbf5840d857a0c2aad37edeb.json deleted file mode 100644 index 69312edba..000000000 --- a/crates/db/.sqlx/query-14604982a6b089a47e32e4fa7189819526eca2c8fbf5840d857a0c2aad37edeb.json +++ /dev/null @@ -1,32 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO raw_weight_snapshot\n (id, challenge_id, epoch, miner_hotkey, kind, score, absence_reason,\n payload, payload_digest, signature, nonce)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)\n ON CONFLICT (challenge_id, epoch, miner_hotkey) DO NOTHING\n RETURNING id\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Uuid" - } - ], - "parameters": { - "Left": [ - "Uuid", - "Text", - "Int8", - "Text", - "Text", - "Int8", - "Text", - "Bytea", - "Bytea", - "Bytea", - "Bytea" - ] - }, - "nullable": [ - false - ] - }, - "hash": "14604982a6b089a47e32e4fa7189819526eca2c8fbf5840d857a0c2aad37edeb" -} diff --git a/crates/db/migrations/0017_raw_weight_tip_supersede.sql b/crates/db/migrations/0017_raw_weight_tip_supersede.sql new file mode 100644 index 000000000..8d32f0f70 --- /dev/null +++ b/crates/db/migrations/0017_raw_weight_tip_supersede.sql @@ -0,0 +1,56 @@ +-- Tip leaf supersede: allow replacing a raw_weight_snapshot row when the +-- payload_digest changes for the same (challenge_id, epoch, miner_hotkey). +-- +-- `base_app` still has no direct UPDATE privilege on append-only tables +-- (schema tests keep that invariant). Tip supersede runs through this +-- SECURITY DEFINER helper owned by the migration role. + +CREATE OR REPLACE FUNCTION upsert_raw_weight_tip( + p_id uuid, + p_challenge_id text, + p_epoch bigint, + p_miner_hotkey text, + p_kind text, + p_score bigint, + p_absence_reason text, + p_payload bytea, + p_payload_digest bytea, + p_signature bytea, + p_nonce bytea +) RETURNS uuid +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = public +AS $$ +DECLARE + result_id uuid; +BEGIN + INSERT INTO raw_weight_snapshot ( + id, challenge_id, epoch, miner_hotkey, kind, score, absence_reason, + payload, payload_digest, signature, nonce + ) VALUES ( + p_id, p_challenge_id, p_epoch, p_miner_hotkey, p_kind, p_score, + p_absence_reason, p_payload, p_payload_digest, p_signature, p_nonce + ) + ON CONFLICT (challenge_id, epoch, miner_hotkey) DO UPDATE SET + id = EXCLUDED.id, + kind = EXCLUDED.kind, + score = EXCLUDED.score, + absence_reason = EXCLUDED.absence_reason, + payload = EXCLUDED.payload, + payload_digest = EXCLUDED.payload_digest, + signature = EXCLUDED.signature, + nonce = EXCLUDED.nonce + WHERE raw_weight_snapshot.payload_digest IS DISTINCT FROM EXCLUDED.payload_digest + RETURNING id INTO result_id; + + RETURN result_id; +END; +$$; + +REVOKE ALL ON FUNCTION upsert_raw_weight_tip( + uuid, text, bigint, text, text, bigint, text, bytea, bytea, bytea, bytea +) FROM PUBLIC; +GRANT EXECUTE ON FUNCTION upsert_raw_weight_tip( + uuid, text, bigint, text, text, bigint, text, bytea, bytea, bytea, bytea +) TO base_app; diff --git a/crates/db/src/lib.rs b/crates/db/src/lib.rs index f05949da6..428a12a8d 100644 --- a/crates/db/src/lib.rs +++ b/crates/db/src/lib.rs @@ -3,9 +3,10 @@ //! # Roles //! //! Migrations run as the database owner (superuser in tests). Application -//! connections should use the `base_app` role, which has **no** `UPDATE` +//! connections should use the `base_app` role, which has **no** direct `UPDATE` //! privilege on the append-only tables `raw_weight_snapshot`, `epoch_bundle`, -//! and `peer_root_statement`. +//! and `peer_root_statement`. Tip leaf supersede uses the +//! `upsert_raw_weight_tip` SECURITY DEFINER helper (migration 0017). //! //! # D18 //! diff --git a/crates/db/src/store.rs b/crates/db/src/store.rs index 249ae8c26..c99e37c64 100644 --- a/crates/db/src/store.rs +++ b/crates/db/src/store.rs @@ -1,10 +1,12 @@ //! Typed persistence for the gateway's append-only tables. //! -//! `raw_weight_snapshot` and `epoch_bundle` are `SELECT`/`INSERT` only for the -//! application role, so every helper here is an insert or a read — never an -//! update. All uniqueness and shape invariants are enforced by the schema -//! (`0001_init.sql`, `0002_epoch_bundle_revision.sql`); the Rust side only -//! feeds them and interprets the conflicts they raise. +//! `epoch_bundle` and `peer_root_statement` stay `SELECT`/`INSERT` only for +//! the application role. `raw_weight_snapshot` inserts go through +//! [`insert_raw_weight`] / tip supersede via the `upsert_raw_weight_tip` +//! SECURITY DEFINER helper (no direct `UPDATE` grant on the table). Bundle +//! reseal appends a new `epoch_bundle.revision`. Schema invariants live in +//! `0001_init.sql`, `0002_epoch_bundle_revision.sql`, +//! `0017_raw_weight_tip_supersede.sql`. use sqlx::PgPool; use uuid::Uuid; @@ -67,11 +69,14 @@ pub struct RawWeightRecord { pub signature: Vec, } -/// Append one raw-weight leaf. +/// Insert or tip-supersede one raw-weight leaf. /// -/// Returns `Ok(None)` when `(challenge_id, epoch, miner_hotkey)` is already -/// stored — `raw_weight_snapshot_challenge_epoch_miner_unique` is what makes a -/// retried submission a conflict instead of a duplicate. +/// Returns `Ok(Some(id))` when a row was inserted or replaced because +/// `payload_digest` changed. Returns `Ok(None)` when the unique key already +/// holds an identical digest (idempotent replay → HTTP 409). +/// +/// Tip supersede runs via `upsert_raw_weight_tip` so `base_app` never needs a +/// direct `UPDATE` grant on `raw_weight_snapshot`. /// /// # Errors /// @@ -81,28 +86,28 @@ pub async fn insert_raw_weight( pool: &PgPool, row: &NewRawWeight<'_>, ) -> Result, DbError> { - let id = sqlx::query_scalar!( - r#" - INSERT INTO raw_weight_snapshot - (id, challenge_id, epoch, miner_hotkey, kind, score, absence_reason, - payload, payload_digest, signature, nonce) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) - ON CONFLICT (challenge_id, epoch, miner_hotkey) DO NOTHING - RETURNING id - "#, - row.id, - row.challenge_id, - row.epoch, - row.miner_hotkey, - row.kind, - row.score, - row.absence_reason, - row.payload, - row.payload_digest, - row.signature, - row.nonce, + // Runtime query: return type is `Option` from the tip-supersede + // helper (NULL = identical digest). Avoids regenerating sqlx offline + // metadata for a SECURITY DEFINER function signature. + let id: Option = sqlx::query_scalar( + r" + SELECT upsert_raw_weight_tip( + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11 + ) + ", ) - .fetch_optional(pool) + .bind(row.id) + .bind(row.challenge_id) + .bind(row.epoch) + .bind(row.miner_hotkey) + .bind(row.kind) + .bind(row.score) + .bind(row.absence_reason) + .bind(row.payload) + .bind(row.payload_digest) + .bind(row.signature) + .bind(row.nonce) + .fetch_one(pool) .await?; Ok(id) } diff --git a/crates/db/tests/gateway_store.rs b/crates/db/tests/gateway_store.rs index 9effe16f8..10a5f3515 100644 --- a/crates/db/tests/gateway_store.rs +++ b/crates/db/tests/gateway_store.rs @@ -2,7 +2,7 @@ //! //! Scenarios: //! - S1 happy: raw-weight insert + read back + list + count -//! - S2 edge: duplicate `(challenge_id, epoch, miner_hotkey)` is a conflict, not a second row +//! - S2 edge: identical digest is a conflict; digest change tip-supersedes in place //! - S3 happy: sealed bundle insert, read by epoch / root, and re-seal bumps `revision` //! - S4 regression: schema `CHECK`s still reject a malformed raw weight @@ -169,9 +169,36 @@ async fn s2_duplicate_raw_weight_conflicts() { ) .await .expect("retry must not error"); - assert!(retry.is_none(), "unique key → no second row"); + assert!(retry.is_none(), "identical digest → no second row"); assert_eq!(count_raw_weights(pool).await.expect("count"), 1); + // Tip supersede: different digest replaces in place. + let digest2 = vec![9u8; 32]; + let payload2 = b"scale-body-v2".to_vec(); + let supersede = insert_raw_weight( + pool, + &score_row( + Uuid::new_v4(), + "c1", + 1, + "aa", + &payload2, + &digest2, + &sig, + &nonce, + ), + ) + .await + .expect("supersede"); + assert!(supersede.is_some()); + assert_eq!(count_raw_weights(pool).await.expect("count"), 1); + let row = get_raw_weight(pool, "c1", 1, "aa") + .await + .expect("get") + .expect("row"); + assert_eq!(row.payload, payload2); + assert_eq!(row.payload_digest, digest2); + tp.drop_schema().await.expect("drop"); } diff --git a/crates/design-challenge-task/src/emit.rs b/crates/design-challenge-task/src/emit.rs index d5fdf4228..e9fb3d80c 100644 --- a/crates/design-challenge-task/src/emit.rs +++ b/crates/design-challenge-task/src/emit.rs @@ -1,10 +1,11 @@ -//! Design leaf-emit scheduling (late-tempo filler + catch-up). +//! Design leaf-emit scheduling (late-tempo filler + catch-up + tip re-emit). -/// How many blocks before epoch end the NotAttempted filler may run. +/// How many blocks before epoch end the NotAttempted filler may first run +/// when the tip has not yet been emitted this process. /// -/// Wider than the historical 48-block window so `base-real-seal` (10 min) still -/// has time to seal after design emits, while leaving most of the epoch for -/// `award_round` to land Score leaves first (first-write-wins). +/// Wider than the historical 48-block window so `base-real-seal` still has +/// time to seal after design emits. Once the tip has been emitted, every +/// emitter tick re-emits so mid-epoch awards tip-supersede gateway leaves. pub const DESIGN_EMIT_LATE_BLOCKS: u64 = 96; /// Planned design leaf emission for one emitter tick. @@ -26,8 +27,11 @@ const MAX_CATCHUP_EPOCHS: u64 = 16; /// epoch 1 pins a pruned block and fails with `SubnetOwnerHotkey not found`. /// - Catch up `last_emitted+1` when behind (capped to [`MAX_CATCHUP_EPOCHS`]) /// so end-of-epoch relabel skips can recover without exceeding prune depth. -/// - Otherwise wait until the last [`DESIGN_EMIT_LATE_BLOCKS`] of the current -/// epoch so admin awards can submit Score leaves first. +/// - **Tip already emitted** (`last_emitted == current`): re-emit every tick so +/// rolling window scores tip-supersede gateway leaves (gateway accepts digest +/// changes; identical digests stay 409-as-ok). +/// - First tip emit in-process: wait until the last [`DESIGN_EMIT_LATE_BLOCKS`] +/// of the current epoch unless cold-start / catch-up already covered it. #[must_use] pub fn design_emit_plan( last_emitted: u64, @@ -47,8 +51,12 @@ pub fn design_emit_plan( pin_block: current_last_epoch_block, }); } + // Tip tracking: re-emit current epoch every tick after the first emit. if last_emitted >= current_epoch { - return None; + return Some(DesignEmitPlan { + epoch: current_epoch, + pin_block: current_last_epoch_block, + }); } // Sequential catch-up for skipped epochs (award path / boundary race). if last_emitted + 1 < current_epoch { @@ -65,7 +73,7 @@ pub fn design_emit_plan( pin_block, }); } - // Current epoch: late-tempo filler only. + // Current epoch not yet emitted this process: late-tempo filler only. if blocks_since_last_step.saturating_add(DESIGN_EMIT_LATE_BLOCKS) < tempo { return None; } @@ -120,7 +128,14 @@ mod tests { } #[test] - fn emit_plan_noop_when_already_emitted_current() { - assert!(design_emit_plan(11, 11, 350, 360, 1000).is_none()); + fn emit_plan_reemits_tip_when_already_emitted_current() { + let p = design_emit_plan(11, 11, 10, 360, 1000).unwrap(); + assert_eq!( + p, + DesignEmitPlan { + epoch: 11, + pin_block: 1000 + } + ); } } diff --git a/crates/design-challenge/src/lib.rs b/crates/design-challenge/src/lib.rs index dab755e1d..12fc97b8c 100644 --- a/crates/design-challenge/src/lib.rs +++ b/crates/design-challenge/src/lib.rs @@ -97,7 +97,14 @@ mod tests { } #[test] - fn emit_plan_noop_when_already_emitted_current() { - assert!(design_emit_plan(11, 11, 350, 360, 1000).is_none()); + fn emit_plan_reemits_tip_when_already_emitted_current() { + let p = design_emit_plan(11, 11, 10, 360, 1000).unwrap(); + assert_eq!( + p, + DesignEmitPlan { + epoch: 11, + pin_block: 1000 + } + ); } } diff --git a/crates/design-store/src/store.rs b/crates/design-store/src/store.rs index e9cd9ebda..f5c0014d5 100644 --- a/crates/design-store/src/store.rs +++ b/crates/design-store/src/store.rs @@ -1018,27 +1018,36 @@ impl DesignStore for MemoryDesignStore { netuid: u16, epoch: u64, ) -> Result, StoreError> { - let rounds: Vec = self + // Match PG `design_scores_for_epoch`: newest rating per miner among + // rounds with `round.epoch <= target` (rolling window projection). + let round_epoch: BTreeMap = self .rounds .lock() .map_err(|_| StoreError::Backend("poison".into()))? .values() - .filter(|r| r.netuid == netuid && r.epoch == epoch) - .map(|r| r.round_id) + .filter(|r| r.netuid == netuid && r.epoch <= epoch) + .map(|r| (r.round_id, r.epoch)) .collect(); - let mut by: BTreeMap = BTreeMap::new(); + let mut by: BTreeMap = BTreeMap::new(); let ratings = self .ratings .lock() .map_err(|_| StoreError::Backend("poison".into()))?; for ((rid, _), row) in ratings.iter() { - if rounds.contains(rid) { - if let Some(fs) = &row.final_score { - by.insert(row.miner_hotkey.clone(), fs.clone()); + if !round_epoch.contains_key(rid) { + continue; + } + let Some(fs) = &row.final_score else { + continue; + }; + match by.get(&row.miner_hotkey) { + Some((prev_rid, _)) if *prev_rid >= *rid => {} + _ => { + by.insert(row.miner_hotkey.clone(), (*rid, fs.clone())); } } } - Ok(by.into_iter().collect()) + Ok(by.into_iter().map(|(hk, (_, fs))| (hk, fs)).collect()) } async fn set_round_award(&self, award: &RoundAward) -> Result<(), StoreError> { diff --git a/crates/gateway-core/src/weights_store.rs b/crates/gateway-core/src/weights_store.rs index 8c40f92cd..2fb5c0bb7 100644 --- a/crates/gateway-core/src/weights_store.rs +++ b/crates/gateway-core/src/weights_store.rs @@ -42,13 +42,18 @@ pub struct RawWeightRow { pub challenge_sig: Vec, } -/// Append-only raw-weight persistence. +/// Raw-weight persistence with tip supersede. +/// +/// Unique key: `(challenge_id, epoch, miner_hotkey)`. A later leaf with a +/// **different** `payload_digest` replaces the stored row (tip tracking). An +/// identical digest is a conflict (idempotent replay). pub trait RawWeightStore: Send + Sync { - /// Insert a new row. Fails with [`StoreError::Conflict`] if the unique key exists. + /// Insert a new row, or replace when the digest changes for the same key. /// /// # Errors /// - /// [`StoreError::Conflict`] when `(challenge_id, epoch, miner_hotkey)` already stored. + /// [`StoreError::Conflict`] when the key exists with the **same** + /// `payload_digest`. fn insert(&self, row: RawWeightRow) -> Result; /// Lookup by unique key. @@ -69,7 +74,7 @@ pub trait RawWeightStore: Send + Sync { /// Store insert failures. #[derive(Debug, Error, Clone, PartialEq, Eq)] pub enum StoreError { - /// Unique key already present; original row is returned for 409 bodies. + /// Unique key already present with the same digest; original for 409 bodies. #[error("raw weight already present for challenge/epoch/miner")] Conflict { /// Unchanged original row. @@ -80,7 +85,7 @@ pub enum StoreError { Backend(String), } -/// In-memory append-only store (tests + default runtime until DB hydrate). +/// In-memory store with tip supersede (tests + default runtime until DB hydrate). #[derive(Debug, Default)] pub struct MemoryRawWeightStore { rows: RwLock>, @@ -103,9 +108,12 @@ impl RawWeightStore for MemoryRawWeightStore { ); let mut guard = self.rows.write(); if let Some(existing) = guard.get(&key) { - return Err(StoreError::Conflict { - original: Box::new(existing.clone()), - }); + if existing.payload_digest == row.payload_digest { + return Err(StoreError::Conflict { + original: Box::new(existing.clone()), + }); + } + // Tip supersede: digest changed → replace in place. } guard.insert(key, row.clone()); Ok(row) @@ -181,6 +189,9 @@ pub struct RawWeightAccepted { /// Absence reason when present. #[serde(skip_serializing_if = "Option::is_none")] pub absence_reason: Option, + /// True when an earlier leaf for the same key was replaced (digest change). + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub superseded: bool, } impl From<&RawWeightRow> for RawWeightAccepted { @@ -193,10 +204,21 @@ impl From<&RawWeightRow> for RawWeightAccepted { kind: row.kind.clone(), score: row.score, absence_reason: row.absence_reason.clone(), + superseded: false, } } } +impl RawWeightAccepted { + /// Build an ack, marking tip supersede when requested. + #[must_use] + pub fn from_row(row: &RawWeightRow, superseded: bool) -> Self { + let mut ack = Self::from(row); + ack.superseded = superseded; + ack + } +} + /// Ingress errors → HTTP. #[derive(Debug, Error)] pub enum IngressError { @@ -209,7 +231,7 @@ pub enum IngressError { /// Challenge id absent from local trust root. #[error("challenge not registered")] UnknownChallenge, - /// Unique key already present. + /// Unique key already present with the same digest. #[error("conflict: raw weight already stored")] Conflict { /// Original row (unchanged). diff --git a/crates/gateway-store-pg/src/lib.rs b/crates/gateway-store-pg/src/lib.rs index 8bada2f51..9b9a23efa 100644 --- a/crates/gateway-store-pg/src/lib.rs +++ b/crates/gateway-store-pg/src/lib.rs @@ -143,12 +143,12 @@ fn record_to_row(rec: RawWeightRecord) -> Result { impl RawWeightStore for PgRawWeightStore { fn insert(&self, row: RawWeightRow) -> Result { + // Tip supersede (digest change) returns Some; identical digest → None. let inserted = self.try_insert(&row).map_err(StoreError::Backend)?; if inserted { return Ok(row); } - // The unique key already holds a row: the 409 body must echo the - // stored original, not the rejected resubmission. + // Same digest already stored: 409 body echoes the original row. match self.get(&row.challenge_id, row.epoch, &row.miner_hotkey) { Some(original) => Err(StoreError::Conflict { original: Box::new(original), @@ -196,7 +196,8 @@ impl RawWeightStore for PgRawWeightStore { } impl PgRawWeightStore { - /// `Ok(true)` when the row was appended, `Ok(false)` on unique conflict. + /// `Ok(true)` when the row was inserted or tip-superseded; `Ok(false)` when + /// the unique key already holds an identical `payload_digest`. fn try_insert(&self, row: &RawWeightRow) -> Result { let epoch = epoch_i64(row.epoch)?; let score = row @@ -269,6 +270,10 @@ impl BundleStore for PgBundleStore { if let Some(existing) = self.get_by_epoch(epoch) { return existing; } + self.put_revision(epoch, bytes) + } + + fn put_revision(&self, epoch: u64, bytes: Vec) -> Vec { match self.seal(epoch, &bytes) { Ok(()) => bytes, Err(e) => { diff --git a/crates/gateway/src/sealer.rs b/crates/gateway/src/sealer.rs index a1c199faa..4f94c9a64 100644 --- a/crates/gateway/src/sealer.rs +++ b/crates/gateway/src/sealer.rs @@ -31,6 +31,8 @@ pub type SharedBundleStore = Arc; pub trait BundleStore: Send + Sync { /// Insert or return existing sealed bytes for `epoch` (idempotent). fn put_if_absent(&self, epoch: u64, bytes: Vec) -> Vec; + /// Append a new seal revision for `epoch` (tip reseal). Returns stored bytes. + fn put_revision(&self, epoch: u64, bytes: Vec) -> Vec; /// Lookup by epoch. fn get_by_epoch(&self, epoch: u64) -> Option>; /// Lookup by merkle root. @@ -68,20 +70,19 @@ impl BundleStore for MemoryBundleStore { if let Some(existing) = self.by_epoch.read().get(&epoch) { return existing.clone(); } + self.put_revision(epoch, bytes) + } + + fn put_revision(&self, epoch: u64, bytes: Vec) -> Vec { let root = match EpochBundleV1::decode_bytes(&bytes) { Ok(b) => b.body.merkle_root, Err(_) => [0u8; 32], }; - let mut by_epoch = self.by_epoch.write(); - if let Some(existing) = by_epoch.get(&epoch) { - return existing.clone(); - } - by_epoch.insert(epoch, bytes.clone()); - drop(by_epoch); + self.by_epoch.write().insert(epoch, bytes.clone()); self.by_root.write().insert(root, bytes.clone()); let mut seals = self.seals.write(); // Revision counts accepted seals for the epoch; the first seal is 1 and - // only a replacing seal of the same epoch can raise it. + // tip reseal raises it when merkle/vector change. let revision = seals.get(&epoch).map_or(1, |prev| prev.revision + 1); seals.insert(epoch, SealRecord::now(revision)); bytes @@ -151,7 +152,11 @@ fn map_bundle_err(e: bundle::BundleError) -> SealError { } } -/// Seal epoch: gather leaves, D24+aggregate+sign, persist (idempotent). +/// Seal epoch: gather leaves, D24+aggregate+sign, persist. +/// +/// Tip reseal: when a bundle already exists, rebuild from current leaves. If +/// `merkle_root` and `final_vector` are unchanged, return the existing seal +/// (no-op). Otherwise append the next `epoch_bundle.revision`. /// /// # Errors /// @@ -163,9 +168,6 @@ pub fn seal_epoch( bundles: &dyn BundleStore, params: &SealParams, ) -> Result { - if let Some(existing) = bundles.get_by_epoch(params.epoch) { - return EpochBundleV1::decode_bytes(&existing).map_err(|e| SealError::Codec(e.to_string())); - } let leaves = rows_to_leaves(&weights.list_for_epoch(params.epoch))?; let trust = LocalTrustRoot { challenges: challenges.clone(), @@ -178,6 +180,17 @@ pub fn seal_epoch( gateway_secret: params.gateway_secret, }; let bundle = build_sealed_bundle(chain, &trust, leaves, &bparams).map_err(map_bundle_err)?; + if let Some(existing) = bundles.get_by_epoch(params.epoch) { + let old = + EpochBundleV1::decode_bytes(&existing).map_err(|e| SealError::Codec(e.to_string()))?; + if old.body.merkle_root == bundle.body.merkle_root + && old.body.final_vector == bundle.body.final_vector + { + return Ok(old); + } + let stored = bundles.put_revision(params.epoch, bundle.encode_bytes()); + return EpochBundleV1::decode_bytes(&stored).map_err(|e| SealError::Codec(e.to_string())); + } let stored = bundles.put_if_absent(params.epoch, bundle.encode_bytes()); EpochBundleV1::decode_bytes(&stored).map_err(|e| SealError::Codec(e.to_string())) } diff --git a/crates/gateway/src/weights.rs b/crates/gateway/src/weights.rs index ba5061644..35703983e 100644 --- a/crates/gateway/src/weights.rs +++ b/crates/gateway/src/weights.rs @@ -1,9 +1,11 @@ //! //! Raw-weight HTTP router, verification pipeline and request/response shapes. //! Challenge leaves are verified against the **local** owner-signed trust root -//! (D18 defence in depth) under domain tag `base-rawweight-v1`, then appended -//! to an append-only store. Unique key: `(challenge_id, epoch, miner_hotkey)`. -//! The store plane itself lives in [`crate::weights_store`]. +//! (D18 defence in depth) under domain tag `base-rawweight-v1`, then stored +//! under unique key `(challenge_id, epoch, miner_hotkey)`. A later leaf with a +//! different `payload_digest` tip-supersedes the prior row (202 + +//! `superseded: true`); identical digest remains 409. The store plane itself +//! lives in [`crate::weights_store`]. use std::sync::Arc; @@ -34,8 +36,11 @@ async fn post_raw_weight( State(st): State, Json(req): Json, ) -> Result<(StatusCode, Json), IngressError> { - let row = accept_raw_weight(st.challenges.as_ref(), st.weights.as_ref(), &req)?; - Ok((StatusCode::ACCEPTED, Json(RawWeightAccepted::from(&row)))) + let (row, superseded) = accept_raw_weight(st.challenges.as_ref(), st.weights.as_ref(), &req)?; + Ok(( + StatusCode::ACCEPTED, + Json(RawWeightAccepted::from_row(&row, superseded)), + )) } /// SCALE enum matching `BUNDLE_SPEC` §3.3 (`0 = Score`, `1 = NoScore`). @@ -55,7 +60,10 @@ struct RawWeightBodyV1 { score_or_absence: ScoreOrAbsenceScale, } -/// Verify + append a single raw-weight leaf. +/// Verify + store a single raw-weight leaf (insert or tip supersede). +/// +/// Returns `(row, superseded)` where `superseded` is true when an earlier +/// leaf for the same key was replaced because `payload_digest` changed. /// /// # Errors /// @@ -64,7 +72,7 @@ pub fn accept_raw_weight( challenges: &ChallengesBody, store: &dyn RawWeightStore, req: &RawWeightRequest, -) -> Result { +) -> Result<(RawWeightRow, bool), IngressError> { if req.challenge_id.is_empty() { return Err(IngressError::BadRequest( "challenge_id must be non-empty".into(), @@ -110,11 +118,16 @@ pub fn accept_raw_weight( let mut digest = [0u8; 32]; digest.copy_from_slice(&payload_digest); + let miner_hex = hex::encode(miner); + let superseded = store + .get(&req.challenge_id, req.epoch, &miner_hex) + .is_some_and(|prev| prev.payload_digest != digest); + let row = RawWeightRow { id: Uuid::new_v4(), challenge_id: req.challenge_id.clone(), epoch: req.epoch, - miner_hotkey: hex::encode(miner), + miner_hotkey: miner_hex, kind, score: score_value, absence_reason, @@ -124,7 +137,7 @@ pub fn accept_raw_weight( }; match store.insert(row) { - Ok(stored) => Ok(stored), + Ok(stored) => Ok((stored, superseded)), Err(StoreError::Conflict { original }) => Err(IngressError::Conflict { original }), Err(StoreError::Backend(msg)) => Err(IngressError::Backend(msg)), } @@ -207,8 +220,46 @@ mod unit_tests { score_or_absence: ScoreOrAbsenceWire::Score { value: 7 }, challenge_sig: hex::encode(sig), }; - let row = accept_raw_weight(&body(pk), &store, &req).unwrap(); + let (row, superseded) = accept_raw_weight(&body(pk), &store, &req).unwrap(); assert_eq!(row.score, Some(7)); + assert!(!superseded); assert_eq!(store.len(), 1); } + + #[test] + fn unit_accept_supersedes_when_digest_changes() { + let (sk, pk) = kp(); + let miner = [9u8; 32]; + let store = MemoryRawWeightStore::new(); + let challenges = body(pk); + let mk = |value: u64| { + let scale = RawWeightBodyV1 { + challenge_id: b"c1".to_vec(), + miner_hotkey: miner, + epoch: 1, + score_or_absence: ScoreOrAbsenceScale::Score { value }, + }; + let payload = scale.encode(); + let sig = sign_raw(&sk, domain::RAW_WEIGHT, &payload).unwrap(); + RawWeightRequest { + challenge_id: "c1".into(), + miner_hotkey: hex::encode(miner), + epoch: 1, + score_or_absence: ScoreOrAbsenceWire::Score { value }, + challenge_sig: hex::encode(sig), + } + }; + let (first, _) = accept_raw_weight(&challenges, &store, &mk(7)).unwrap(); + assert_eq!(first.score, Some(7)); + let (second, superseded) = accept_raw_weight(&challenges, &store, &mk(99)).unwrap(); + assert!(superseded); + assert_eq!(second.score, Some(99)); + assert_eq!( + store.get("c1", 1, &hex::encode(miner)).unwrap().score, + Some(99) + ); + // Identical digest replay → conflict. + let err = accept_raw_weight(&challenges, &store, &mk(99)).unwrap_err(); + assert!(matches!(err, IngressError::Conflict { .. })); + } } diff --git a/crates/gateway/tests/raw_weights.rs b/crates/gateway/tests/raw_weights.rs index 929a24e70..def71f03f 100644 --- a/crates/gateway/tests/raw_weights.rs +++ b/crates/gateway/tests/raw_weights.rs @@ -8,7 +8,8 @@ //! S1 valid → 202 + row //! S2 wrong key → 401 + NO row //! S3 unknown challenge → 404 + no row -//! S4 replay (challenge, epoch, miner) → 409 + original unchanged +//! S4 replay identical digest → 409 + original unchanged +//! S5 digest change for same key → 202 tip supersede use std::net::SocketAddr; use std::sync::Arc; @@ -341,7 +342,35 @@ async fn s4_replay_challenge_epoch_miner_returns_409_original_unchanged() { assert_eq!(after.score, original.score); assert_eq!(after.id, original.id); - // Conflicting score for same key also 409 without mutation. + let _ = shutdown.send(()); +} + +#[tokio::test] +async fn s5_digest_change_tip_supersedes() { + let (sk, pk) = mini_keypair(); + let cid = "dummy"; + let miner = [0x66u8; 32]; + let epoch = 12u64; + let (_payload, sig) = sign_leaf( + &sk, + cid, + miner, + epoch, + ScoreOrAbsenceScale::Score { value: 100 }, + ); + + let store = Arc::new(MemoryRawWeightStore::new()); + let (addr, shutdown) = spawn_gateway(challenges_body(cid, pk), Arc::clone(&store)).await; + let client = reqwest::Client::new(); + + let first = client + .post(format!("http://{addr}/v1/weights/raw")) + .json(&json_score(cid, miner, epoch, 100, &sig)) + .send() + .await + .expect("first"); + assert_eq!(first.status().as_u16(), 202); + let (_p2, sig2) = sign_leaf( &sk, cid, @@ -349,16 +378,28 @@ async fn s4_replay_challenge_epoch_miner_returns_409_original_unchanged() { epoch, ScoreOrAbsenceScale::Score { value: 999 }, ); + let second = client + .post(format!("http://{addr}/v1/weights/raw")) + .json(&json_score(cid, miner, epoch, 999, &sig2)) + .send() + .await + .expect("supersede"); + let status = second.status().as_u16(); + let body = second.text().await.unwrap_or_default(); + assert_eq!(status, 202, "body={body}"); + assert!(body.contains("\"superseded\":true"), "body={body}"); + let final_row = store.get(cid, epoch, &hex::encode(miner)).expect("final"); + assert_eq!(final_row.score, Some(999)); + assert_eq!(store.len(), 1); + + // Identical digest after supersede → 409. let third = client .post(format!("http://{addr}/v1/weights/raw")) .json(&json_score(cid, miner, epoch, 999, &sig2)) .send() .await - .expect("third"); + .expect("replay"); assert_eq!(third.status().as_u16(), 409); - let final_row = store.get(cid, epoch, &hex::encode(miner)).expect("final"); - assert_eq!(final_row.score, Some(value)); - assert_eq!(final_row.id, original.id); let _ = shutdown.send(()); } diff --git a/crates/gateway/tests/sealer.rs b/crates/gateway/tests/sealer.rs index 72cc089b4..0f8209d0e 100644 --- a/crates/gateway/tests/sealer.rs +++ b/crates/gateway/tests/sealer.rs @@ -237,6 +237,46 @@ fn s2_reseal_idempotent_identical_bytes_and_signature() { assert_eq!(bytes1, bytes2, "re-seal must be byte-identical"); assert_eq!(b1.gateway_sig, b2.gateway_sig); assert_eq!(bundles.get_by_epoch(params.epoch).unwrap(), bytes1); + assert_eq!(bundles.seal_record(params.epoch).unwrap().revision, 1); +} + +#[test] +fn s2b_tip_reseal_bumps_revision_when_leaf_digest_changes() { + let (chain, challenges, weights, bundles, params, trust, _gsk) = seal_fixture(); + let b1 = seal_epoch( + &chain, + &challenges, + weights.as_ref(), + bundles.as_ref(), + ¶ms, + ) + .expect("seal1"); + assert_eq!(bundles.seal_record(params.epoch).unwrap().revision, 1); + + // Supersede one tip leaf with a different score → merkle/vector change. + let csk = sk(1); + let cid = b"dummy"; + let miners = [hk(0xA1), hk(0xB2), hk(0xC3)]; + seed_leaf_row( + weights.as_ref(), + &csk, + cid, + miners[0], + params.epoch, + ScoreOrAbsence::Score { value: 90 }, + ); + let b2 = seal_epoch( + &chain, + &challenges, + weights.as_ref(), + bundles.as_ref(), + ¶ms, + ) + .expect("reseal"); + assert_ne!(b1.body.merkle_root, b2.body.merkle_root); + assert_ne!(b1.body.final_vector, b2.body.final_vector); + assert_eq!(bundles.seal_record(params.epoch).unwrap().revision, 2); + verify_bundle(&b2, &chain, &trust).expect("verify reseal"); } #[test] diff --git a/crates/prism-emit/src/lib.rs b/crates/prism-emit/src/lib.rs index aed4e8536..372698cba 100644 --- a/crates/prism-emit/src/lib.rs +++ b/crates/prism-emit/src/lib.rs @@ -18,9 +18,9 @@ //! assigned (`emitted_epoch = E`, sticky) before the submit, and the //! per-netuid emit cursor advances only after the full set landed. A crash //! between submit and cursor advance replays the sticky assigned set on -//! the next tick; gateway leaves are first-write-wins with identical -//! values under replay, so the retry converges. Retried/re-scored rows -//! re-enter the outbox (`reset_for_retry` clears the watermark). +//! the next tick; gateway leaves tip-supersede on digest change and treat +//! identical digests as 409-as-ok, so the retry converges. Retried/re-scored +//! rows re-enter the outbox (`reset_for_retry` clears the watermark). //! - **Positive scores carry forward**: after outbox assignment, a //! `Score(v>0)` row keeps participating in every later epoch's //! competition set until a better/valid score supersedes it via `max`. @@ -30,6 +30,10 @@ //! ([`prism_registry::OWNER_ARCH_CREDIT_ENABLED`] = `false`) — WTA follows //! the miner hotkey that posted the best-BPB run. Empty or reject-only //! fresh batches therefore do not burn the prism share (active carry). +//! - **Tip refresh**: once the cursor reaches the live epoch, later ticks +//! re-submit the current WTA projection so a mid-epoch champion change +//! tip-supersedes gateway leaves (then `prod-real-seal` reseals). The +//! cursor does not advance again on tip refresh. //! - Epochs during a master outage carry no *new* outbox rows; the first //! epoch after recovery still includes active positive scores plus any //! backlog (the seal always pins fresh epochs — stale ones can never @@ -106,8 +110,8 @@ impl EpochEmitter { /// Drive emission up to `current_epoch`: replay any assigned-but- /// incomplete epoch (crash recovery), then emit the set for - /// `current_epoch` unless it already landed. Returns the summary when a - /// new set was emitted. + /// `current_epoch` (first land) or tip-refresh when the cursor already + /// covers the tip. Returns the summary when leaves were submitted. /// /// # Errors /// Store / sign / submit failures (the caller retries next tick; the @@ -126,7 +130,11 @@ impl EpochEmitter { } let done = self.store.emit_cursor(self.netuid).await?; match done { - Some(d) if d >= current_epoch => Ok(None), + Some(d) if d >= current_epoch => { + // Tip tracking: re-submit WTA so mid-epoch champion changes + // supersede gateway leaves. Cursor stays put. + self.refresh_tip(current_epoch, expected).await.map(Some) + } // First boot or catch-up after downtime: emit once at the live // epoch with the whole backlog; skipped gap epochs get no set // (seals always pin fresh epochs). @@ -134,6 +142,20 @@ impl EpochEmitter { } } + /// Re-submit the tip epoch's current WTA projection without re-assigning + /// the outbox or advancing the emit cursor. + /// + /// # Errors + /// See [`EmitError`]. + pub async fn refresh_tip( + &self, + epoch: u64, + expected: &ExpectedSet, + ) -> Result { + let batch = self.store.emit_batch(self.netuid, epoch).await?; + self.submit_rows(epoch, batch, expected, false).await + } + /// Assign the pending batch to `epoch` and emit its set. /// /// # Errors @@ -166,6 +188,16 @@ impl EpochEmitter { epoch: u64, batch: Vec, expected: &ExpectedSet, + ) -> Result { + self.submit_rows(epoch, batch, expected, true).await + } + + async fn submit_rows( + &self, + epoch: u64, + batch: Vec, + expected: &ExpectedSet, + advance_cursor: bool, ) -> Result { let n = batch.len(); let active = self.store.active_score_rows(self.netuid).await?; @@ -176,8 +208,10 @@ impl EpochEmitter { challenge_common::submit_signed_leaf_set(&self.gateway, &signed) .await .map_err(|e| EmitError::Submit(e.to_string()))?; - // Cursor advances only after the full set landed. - self.store.set_emit_cursor(self.netuid, epoch).await?; + if advance_cursor { + // Cursor advances only after the first full set for the epoch landed. + self.store.set_emit_cursor(self.netuid, epoch).await?; + } Ok(EmitSummary { epoch, leaves: signed.len(), diff --git a/crates/prism-emit/tests/epoch_semantics.rs b/crates/prism-emit/tests/epoch_semantics.rs index 27185cf5e..91c80b31a 100644 --- a/crates/prism-emit/tests/epoch_semantics.rs +++ b/crates/prism-emit/tests/epoch_semantics.rs @@ -211,10 +211,10 @@ async fn no_double_emission_across_epochs() { let s = em.tick(7, &exp).await.unwrap().expect("emit"); assert_eq!(s.batch, 1); - assert!( - em.tick(7, &exp).await.unwrap().is_none(), - "same epoch is a no-op" - ); + // Tip refresh re-submits WTA for the same epoch; cursor stays put. + let tip = em.tick(7, &exp).await.unwrap().expect("tip refresh"); + assert_eq!(tip.epoch, 7); + assert_eq!(store.emit_cursor(541).await.unwrap(), Some(7)); // A second scorer finalizes during epoch 7; only that row is freshly assigned. store diff --git a/deploy/AGENTS.md b/deploy/AGENTS.md index 6e6188a92..ab285878b 100644 --- a/deploy/AGENTS.md +++ b/deploy/AGENTS.md @@ -96,7 +96,7 @@ cargo run -q --release -p weights-smoke -- \ A seal older than ~256 blocks can never be verified by the validator (public RPC prunes state) — if `GET /v1/weights/latest` shows `metagraph_block` lagging tip by thousands of blocks, check `systemctl status base-burn-seal.timer` and `/var/log/base-burn-seal.log` on the master. -**Real-epoch sealer (post burn-seal retirement):** `base-real-seal.timer` (every 10 min) drives [`scripts/prod-real-seal.sh`](scripts/prod-real-seal.sh), which walks **current … current−N** chain epochs (`REAL_SEAL_WALK_BACK`, default 16) with `block_b = LastEpochBlock − k×tempo` so a skipped design/prism leaf epoch does not pin `/v1/weights/latest` on a stale real seal forever (burn seals cannot outrank chain-scale bundles). 409 `incomplete_participant_set` on a candidate is expected and the script continues walking. The gateway prefers chain-scale bundles over the reserved smoke range (`>= 8_000_000`) — retire the burn timer (`systemctl disable --now base-burn-seal.timer`) after the first real seal verifies end-to-end. Install: +**Real-epoch sealer (post burn-seal retirement):** `base-real-seal.timer` (every **2 min**) drives [`scripts/prod-real-seal.sh`](scripts/prod-real-seal.sh), which walks **current … current−N** chain epochs (`REAL_SEAL_WALK_BACK`, default 16) with `block_b = LastEpochBlock − k×tempo`. Tip reseal is expected: when design/prism tip-supersede leaves mid-epoch, seal rebuilds and appends `epoch_bundle.revision` so `/v1/weights/latest` tracks live scores; identical merkle/vector is a no-op 200. Walk-back still recovers when tip is incomplete D24 (409 continues). The gateway prefers chain-scale bundles over the reserved smoke range (`>= 8_000_000`) — retire the burn timer (`systemctl disable --now base-burn-seal.timer`) after the first real seal verifies end-to-end. Install: ```bash install -m 0755 deploy/scripts/prod-real-seal.sh /opt/base/deploy/scripts/prod-real-seal.sh diff --git a/deploy/scripts/prod-real-seal.sh b/deploy/scripts/prod-real-seal.sh index 9a44f8261..d625045ea 100755 --- a/deploy/scripts/prod-real-seal.sh +++ b/deploy/scripts/prod-real-seal.sh @@ -1,12 +1,18 @@ #!/usr/bin/env bash -# Prod real-epoch sealer: seal the newest chain epoch on the master gateway -# that has a complete D24 participant set from every >0-bps challenge. +# Prod real-epoch sealer: seal (or tip-reseal) the newest chain epoch on the +# master gateway that has a complete D24 participant set from every >0-bps +# challenge. +# +# Tip reseal: `POST /v1/admin/seal` rebuilds from current leaves. When the tip +# is already sealed but leaves tip-superseded (design/prism re-emit), a new +# `epoch_bundle.revision` is appended. Identical merkle/vector → no-op 200. # # Why a walk-back: design historically emitted only in a tight late-tempo # window and could skip alternate epochs (end-of-epoch relabel race). Waiting # solely on *current* epoch then 409s forever while `/v1/weights/latest` stays # pinned on an older real seal (burn seals cannot outrank it). Trying current, -# then current-1 … recovers the newest sealable epoch. +# then current-1 … recovers the newest sealable epoch. Walk-back remains for +# incomplete D24; tip reseal success on current stops the walk (expected). # # block_b pins the bundle metagraph to that epoch's start block # (LastEpochBlock − k×tempo) so D24 participant matching holds. diff --git a/deploy/systemd/base-real-seal.timer b/deploy/systemd/base-real-seal.timer index 12fd3f877..ebcde99da 100644 --- a/deploy/systemd/base-real-seal.timer +++ b/deploy/systemd/base-real-seal.timer @@ -2,11 +2,11 @@ Description=Seal the current chain epoch on the master gateway (real weights) [Timer] -# 10 min cadence: both challenges emit within minutes of an epoch boundary / -# round close, so the current epoch becomes sealable well before it ends -# (~72 min at tempo 360). The endpoint 409s harmlessly until both sets land. -OnBootSec=5min -OnUnitActiveSec=10min +# 2 min cadence: tip leaf supersede (design awards / prism WTA changes) should +# land in `/v1/weights/latest` quickly via tip reseal. The endpoint 409s +# harmlessly until both D24 sets land; identical tip reseal is a no-op 200. +OnBootSec=2min +OnUnitActiveSec=2min Persistent=true [Install] diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 03ed30b20..cead37ce8 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -75,9 +75,9 @@ Miner-facing docs (version-pinned): [`external-miner/`](./external-miner/). ## 3. Data flow (one epoch) 1. **Pin.** Gateway (or seal path) pins `block_hash` / metagraph root at epoch boundary. -2. **Leaves.** Challenge backends produce challenge-signed `Score` or `NoScore` leaves for the **validator-derived** expected set (D24). -3. **Seal.** Gateway builds `EpochBundleV1`, computes merkle root, signs the body. **Does not** put the merkle root into the on-chain weight payload (there is no field; see BUNDLE_SPEC §12 / D5). -4. **Distribute.** Bundle served over HTTPS; validators may also **mirror** from peers (content-addressed by root). +2. **Leaves.** Challenge backends produce challenge-signed `Score` or `NoScore` leaves for the **validator-derived** expected set (D24). Tip epochs may **supersede** a leaf when the signed `payload_digest` changes for the same `(challenge, epoch, miner)`; identical digests stay idempotent. +3. **Seal.** Gateway builds `EpochBundleV1`, computes merkle root, signs the body. Tip reseal appends `epoch_bundle.revision` when leaves/merkle change; no-op if identical. **Does not** put the merkle root into the on-chain weight payload (there is no field; see BUNDLE_SPEC §12 / D5). +4. **Distribute.** `GET /v1/weights/latest` serves the newest revision of the highest chain-scale sealed epoch (`sealed: true` only for Match). Validators may also **mirror** from peers (content-addressed by root). 5. **Verify.** Each validator loads **local** `challenges.toml` + `measurements.toml` (owner-signed). Rejects leaves whose keys are not in the local trust root (D18). 6. **Cross-check.** Hotkey-authenticated peer root exchange; minimum sample (D26). Persist signed bundle + peer statements as local evidence. 7. **Recompute.** Integer aggregation per BUNDLE_SPEC. Compare to gateway `final_vector`. diff --git a/docs/BUNDLE_SPEC.md b/docs/BUNDLE_SPEC.md index 6fa9c5289..bdde5c6a2 100644 --- a/docs/BUNDLE_SPEC.md +++ b/docs/BUNDLE_SPEC.md @@ -517,9 +517,10 @@ Validators compare: |----------|----------| | `GET /v1/bundle/{epoch}` | Returns the sealed `EpochBundleV1` SCALE bytes (content-type `application/octet-stream`) or 404 | | `GET /v1/bundle/root/{root}` | Lookup by `merkle_root` hex (64 lowercase hex chars); returns same body or 404 | -| `GET /v1/weights/latest` | Sealed projection when a bundle exists (`sealed: true`). If no sealed bundle is available or the stored bytes cannot be decoded, MUST return **200** with the fail-closed **burn vector** under `burn-uid0.v1` (`uids: [0]`, `weights: [1.0]`, `sealed: false`) — never 404. Validators MUST NOT treat `sealed: false` as a Match / submit path | +| `GET /v1/weights/latest` | Sealed projection of the **newest revision** of the highest chain-scale sealed epoch (`sealed: true`, with `revision` / `vector_digest`). If no sealed bundle is available or the stored bytes cannot be decoded, MUST return **200** with the fail-closed **burn vector** under `burn-uid0.v1` (`uids: [0]`, `weights: [1.0]`, `sealed: false`) — never 404. Validators MUST NOT treat `sealed: false` as a Match / submit path — Match only on `sealed: true` | +| Tip reseal | Within the live tip epoch, when challenge leaves tip-supersede (`POST /v1/weights/raw` with a changed `payload_digest` for the same `(challenge_id, epoch, miner_hotkey)`), the gateway MAY append a new `epoch_bundle.revision` whose merkle/vector reflect the updated leaves. Re-seal with unchanged merkle root and `final_vector` is a no-op (returns the existing seal). Older epochs are not rewritten by the tip sealer walk | | Validator mirroring | Validators MAY re-serve a bundle they have verified and persisted; peers SHOULD prefer multi-source fetch | -| **No last-known-good** | MUST NOT fall back to a previous epoch's bundle, root, or vector when the current epoch fetch/verify fails. Failure → class B / degraded path, not stale success (aligns with D13 spirit for attestation). The unsealed burn response on `/v1/weights/latest` is an operator-safety default, not a last-known-good seal | +| **No last-known-good** | MUST NOT fall back to a previous epoch's bundle, root, or vector when the current epoch fetch/verify fails. Failure → class B / degraded path, not stale success (aligns with D13 spirit for attestation). The unsealed burn response on `/v1/weights/latest` is an operator-safety default, not a last-known-good seal. Mid-epoch tip revision bumps are intentional tip-tracking, not last-known-good | Gateway signature and leaf signatures are always verified against local trust roots after fetch. diff --git a/docs/PRISM.md b/docs/PRISM.md index e9eb35f33..440636c12 100644 --- a/docs/PRISM.md +++ b/docs/PRISM.md @@ -121,40 +121,41 @@ the telemetry-hooks rule and metrics forge checks still apply. Gating: `prism:train:` as above — one accepted entry per `(hotkey, arch_id)`, retries same rules. -**Leaf emission (epoch-close, exactly-once outbox + score carry).** A +**Leaf emission (epoch-close, exactly-once outbox + score carry + tip refresh).** A submission row's acceptance epoch (`prism_submission.epoch`) is intake -metadata only. A dedicated emitter loop (`prism-emit`, one tick per chain -epoch) emits **one D24-complete leaf set per chain epoch**: the first tick -that observes epoch `E` assigns every submission finalized since the -previously emitted epoch — the outbox batch, +metadata only. A dedicated emitter loop (`prism-emit`) emits a +**D24-complete leaf set** for the live chain epoch: the first tick that +observes epoch `E` assigns every submission finalized since the previously +emitted epoch — the outbox batch, `kind IS NOT NULL AND emitted_epoch IS NULL` — to `E`, competition-aggregates that batch **unioned with every still-active positive lattice score** (`kind = 'score' AND score > 0`), signs the full expected set (`NoScore(NotAttempted)` for everyone else), submits it, and advances the -per-netuid emit cursor (`prism_emit_cursor`, migration 0012). This fixes the -two acceptance-epoch bugs: independent scorers finalized in the same epoch -used to lock each other out (gateway leaves are append-only first-write-wins -per `(challenge, epoch, hotkey)`), and a submission accepted in epoch `X` but -finalized in `X+k` (prod trains up to 6h ≫ 72-min epochs) never scored at -all. +per-netuid emit cursor (`prism_emit_cursor`, migration 0012). Later ticks on +the same tip **re-submit** the current WTA projection so a mid-epoch champion +change tip-supersedes gateway leaves (`payload_digest` change → 202; identical +digest → 409-as-ok). Cursor does not advance again on tip refresh. This fixes +the acceptance-epoch bugs (a submission accepted in epoch `X` but finalized in +`X+k` never scored) while keeping `/v1/weights/latest` aligned with live WTA +after tip reseal. Architecture-owner credit stays off +(`OWNER_ARCH_CREDIT_ENABLED = false`). Exactly-once **outbox assignment** per scoring run: batch assignment is sticky -before submit, the cursor advances only after the full set landed, and a crash -mid-submit replays the identical assigned set on the next tick -(first-write-wins with identical values converges). After assignment, a -positive `Score(v>0)` keeps participating in every later epoch's competition -set until a better/valid score supersedes it via lattice `max` — so an empty -or reject-only fresh batch does not burn the prism share. Leaf emission then -applies **winner-take-all** (`prism_registry::apply_wta`): only the single -highest positive credit (lexicographically smallest hotkey on ties) receives a -positive Score leaf; every other positive credit is zeroed. `Score(0)` rejects -and `NoScore` absences do not carry. A manually retried + re-scored row -re-enters the outbox (`reset_for_retry` clears the watermark); its old leaf -stays immutable history in its original epoch. Epochs during a master outage -carry no *new* outbox rows; the first epoch after recovery still includes -active positive scores plus any backlog (seals always pin fresh epochs — -stale bundles can never Match on-chain). Run **exactly one** prism-challenge -emitter instance per netuid (single master topology). +before submit, the cursor advances only after the first full set for an epoch +landed, and a crash mid-submit replays the identical assigned set on the next +tick. After assignment, a positive `Score(v>0)` keeps participating in every +later epoch's competition set until a better/valid score supersedes it via +lattice `max` — so an empty or reject-only fresh batch does not burn the prism +share. Leaf emission then applies **winner-take-all** +(`prism_registry::apply_wta`): only the single highest positive credit +(lexicographically smallest hotkey on ties) receives a positive Score leaf; +every other positive credit is zeroed. `Score(0)` rejects and `NoScore` +absences do not carry. A manually retried + re-scored row re-enters the outbox +(`reset_for_retry` clears the watermark). Epochs during a master outage carry +no *new* outbox rows; the first epoch after recovery still includes active +positive scores plus any backlog (seals always pin fresh epochs — stale +bundles can never Match on-chain). Run **exactly one** prism-challenge emitter +instance per netuid (single master topology). **Competition scoring (epoch-local, SCORE_MAX lattice preserved; prism `SCORING_VERSION` stays 2 — the competition reallocates credits inside the From 7c169752ab6ca43cee4feee0360a5ed8cf99ec43 Mon Sep 17 00:00:00 2001 From: echobt <154886644+echobt@users.noreply.github.com> Date: Tue, 11 Aug 2026 15:00:01 +0000 Subject: [PATCH 2/2] fix(test): expect prism tip refresh on same-epoch tick --- crates/prism-challenge/tests/e2e_orchestrate_sim.rs | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/crates/prism-challenge/tests/e2e_orchestrate_sim.rs b/crates/prism-challenge/tests/e2e_orchestrate_sim.rs index ff8b39028..2ecdb56ae 100644 --- a/crates/prism-challenge/tests/e2e_orchestrate_sim.rs +++ b/crates/prism-challenge/tests/e2e_orchestrate_sim.rs @@ -318,7 +318,14 @@ async fn emit_and_submit_covers_expected_set() { summary.signed.get(&hk).map(|l| &l.score_or_absence), Some(prism_challenge::ScoreOrAbsence::Score { value: 500_000 }) )); - // Cursor advanced; a same-epoch tick is a no-op. + // Cursor advanced; same-epoch tick tip-refreshes WTA (cursor stays put). + assert_eq!(store.emit_cursor(541).await.unwrap(), Some(7)); + let tip = orch + .emitter() + .tick(7, &expected) + .await + .unwrap() + .expect("tip refresh"); + assert_eq!(tip.epoch, 7); assert_eq!(store.emit_cursor(541).await.unwrap(), Some(7)); - assert!(orch.emitter().tick(7, &expected).await.unwrap().is_none()); }