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

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

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

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

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
+ }
+ );
  }
 }

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
+ }
+ );
  }
 }

crates/design-store/src/store.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> {