feat(weights): tip-track /v1/weights/latest via leaf supersede + reseal (#121)

* 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.

* fix(test): expect prism tip refresh on same-epoch tick

Author: echobt

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
     ///

crates/db/.sqlx/query-14604982a6b089a47e32e4fa7189819526eca2c8fbf5840d857a0c2aad37edeb.json

@@ -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;

crates/db/migrations/0017_raw_weight_tip_supersede.sql

@@ -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
 //!

crates/db/src/lib.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<u8>,
 }
 
-/// 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<Option<Uuid>, 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<Uuid>` from the tip-supersede
+    // helper (NULL = identical digest). Avoids regenerating sqlx offline
+    // metadata for a SECURITY DEFINER function signature.
+    let id: Option<Uuid> = 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)
 }

crates/db/src/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");
 }
 

crates/db/tests/gateway_store.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
+            }
+        );
     }
 }

crates/design-challenge-task/src/emit.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
+            }
+        );
     }
 }

crates/design-challenge/src/lib.rs

@@ -1018,27 +1018,36 @@ impl DesignStore for MemoryDesignStore {
         netuid: u16,
         epoch: u64,
     ) -> Result<Vec<(String, FinalScore)>, StoreError> {
-        let rounds: Vec<u64> = self
+        // Match PG `design_scores_for_epoch`: newest rating per miner among
+        // rounds with `round.epoch <= target` (rolling window projection).
+        let round_epoch: BTreeMap<u64, u64> = 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<String, FinalScore> = BTreeMap::new();
+        let mut by: BTreeMap<String, (u64, FinalScore)> = 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> {