Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions crates/challenge-common/src/submit.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
///
Expand Down

This file was deleted.

56 changes: 56 additions & 0 deletions crates/db/migrations/0017_raw_weight_tip_supersede.sql
Original file line number Diff line number Diff line change
@@ -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

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

Bind the helper to the migration schema.

Line 23 forces object lookup to public. test_pool_with_url migrates a generated schema after setting <schema>, public in crates/db/src/lib.rs Lines 282-304. The helper can then access public.raw_weight_snapshot instead of the isolated table, or fail when that table is absent.

Capture the trusted migration search path, or schema-qualify the target table without weakening the SECURITY DEFINER boundary.

Proposed fix
-SET search_path = public
+SET search_path FROM CURRENT
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
SET search_path = public
SET search_path FROM CURRENT
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@crates/db/migrations/0017_raw_weight_tip_supersede.sql` at line 23, Update
the migration’s SECURITY DEFINER helper around SET search_path so it remains
bound to the trusted migration schema instead of forcing lookup to public.
Capture and use the migration search path, or schema-qualify the target table,
ensuring the helper reads the isolated raw_weight_snapshot table while
preserving the existing security boundary.

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;
5 changes: 3 additions & 2 deletions crates/db/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
//!
Expand Down
65 changes: 35 additions & 30 deletions crates/db/src/store.rs
Original file line number Diff line number Diff line change
@@ -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;
Expand Down Expand Up @@ -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
///
Expand All @@ -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)
}
Expand Down
31 changes: 29 additions & 2 deletions crates/db/tests/gateway_store.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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");
}

Expand Down
37 changes: 26 additions & 11 deletions crates/design-challenge-task/src/emit.rs
Original file line number Diff line number Diff line change
@@ -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.
Expand All @@ -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,
Expand All @@ -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 {
Expand All @@ -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;
}
Expand Down Expand Up @@ -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
}
);
}
}
11 changes: 9 additions & 2 deletions crates/design-challenge/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
);
}
}
25 changes: 17 additions & 8 deletions crates/design-store/src/store.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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> {
Expand Down
Loading
Loading