diff --git a/dashboard/src/App.tsx b/dashboard/src/App.tsx index 7e2a5db..c7be576 100644 --- a/dashboard/src/App.tsx +++ b/dashboard/src/App.tsx @@ -387,22 +387,23 @@ export function App() {
- {/* Verification leaderboard โ€” assigned users ranked by completion speed */} + {/* Verification leaderboard โ€” assigned users ranked by annotation throughput + (most annotations in the least time), not by video count. */}
-

Verification leaderboard ยท fastest first

+

Verification leaderboard ยท most annotations / min

- + {stats?.verifiers.map((r) => ( - - - + + + ))} {(!stats || stats.verifiers.length === 0) && ( diff --git a/dashboard/src/api.ts b/dashboard/src/api.ts index e600e62..55f6b77 100644 --- a/dashboard/src/api.ts +++ b/dashboard/src/api.ts @@ -91,7 +91,8 @@ export interface VerifierRow { total_time_ms: number; avg_time_ms: number; annotations: number; // total annotations this user authored - rank: number; // 1 = fastest; 0 = no completions yet + annotations_per_min: number; // throughput: annotations added per minute (ranking metric) + rank: number; // 1 = highest throughput; 0 = not yet ranked } export interface ProjectTime { @@ -288,12 +289,20 @@ export function fmtBytes(n: number): string { return `${(n / 1024 ** i).toFixed(i ? 1 : 0)} ${u[i]}`; } -/** Readable local date+time for a server ISO timestamp; "" when missing/invalid. */ +/** Readable date+time for a server ISO timestamp, shown in IST (Asia/Kolkata) + * regardless of the viewer's/VM's timezone; "" when missing/invalid. Server + * timestamps are UTC (TIMESTAMPTZ), so the instant is converted, not relabelled. */ export function fmtWhen(iso: string | null): string { if (!iso) return ""; const d = new Date(iso); if (isNaN(d.getTime())) return ""; - return d.toLocaleString([], { month: "short", day: "numeric", hour: "2-digit", minute: "2-digit" }); + return d.toLocaleString("en-IN", { + timeZone: "Asia/Kolkata", + month: "short", + day: "numeric", + hour: "2-digit", + minute: "2-digit", + }) + " IST"; } export function fmtDuration(ms: number): string { diff --git a/server/src/ingest.rs b/server/src/ingest.rs index 2423541..69ecaf4 100644 --- a/server/src/ingest.rs +++ b/server/src/ingest.rs @@ -72,15 +72,27 @@ pub async fn ingest_project(db: &PgPool, project_id: i32, dir: &Path) -> anyhow: // Never clobber a verified push (Phase 3): a worker's pushed annotations + // completer/verify-time are authoritative, so skip re-ingesting this video. - let existing_status: Option = - sqlx::query_scalar("SELECT status FROM videos WHERE project_id=$1 AND rel_path=$2") - .bind(project_id) - .bind(&rel_path) - .fetch_optional(db) - .await?; - if existing_status.as_deref() == Some("verified") { - tracing::debug!("skip re-ingest of verified video: {rel_path}"); - continue; + // Also never clobber a video that is *actively claimed* (a worker is mid-edit) + // โ€” a stale NAS JSON must not overwrite in-progress work. Both checks read the + // current row's status + claim lease. + let existing: Option<(String, Option, Option>)> = sqlx::query_as( + "SELECT status, claimed_by, lease_expires_at FROM videos WHERE project_id=$1 AND rel_path=$2", + ) + .bind(project_id) + .bind(&rel_path) + .fetch_optional(db) + .await?; + if let Some((status, claimed_by, lease)) = existing.as_ref() { + if status == "verified" { + tracing::debug!("skip re-ingest of verified video: {rel_path}"); + continue; + } + let active_claim = claimed_by.is_some() + && lease.map(|l| l > Utc::now()).unwrap_or(false); + if active_claim { + tracing::debug!("skip re-ingest of actively-claimed video: {rel_path}"); + continue; + } } let json_path = sibling_json(path); @@ -138,6 +150,10 @@ pub async fn ingest_project(db: &PgPool, project_id: i32, dir: &Path) -> anyhow: std::fs::read_to_string(jp).ok().and_then(|t| serde_json::from_str(&t).ok()) }); + // Upsert the video + replace its annotation rows atomically: either the new + // metadata AND the new annotation set both land, or neither does. Prevents a + // failed insert mid-loop from leaving the video with partial/zero annotations. + let mut tx = db.begin().await?; // Upsert the video. Keep 'verified' status sticky (Phase 3 pushes). let video_id: i32 = sqlx::query_scalar( r#" @@ -179,18 +195,19 @@ pub async fn ingest_project(db: &PgPool, project_id: i32, dir: &Path) -> anyhow: .bind(latest) .bind(raw_json) .bind(ann_count) - .fetch_one(db) + .fetch_one(&mut *tx) .await?; - // Replace detailed annotation rows for this video. + // Replace detailed annotation rows for this video (same transaction). if let Some(d) = doc.as_ref() { - replace_annotations(db, video_id, d).await?; + replace_annotations(&mut tx, video_id, d).await?; } else { sqlx::query("DELETE FROM annotations WHERE video_id=$1") .bind(video_id) - .execute(db) + .execute(&mut *tx) .await?; } + tx.commit().await?; count += 1; } Ok(count) @@ -199,27 +216,33 @@ pub async fn ingest_project(db: &PgPool, project_id: i32, dir: &Path) -> anyhow: /// Replace all annotation rows for a video with the contents of an export doc. /// Shared by ingest (NAS JSON) and the Phase 3 worker push. Returns the row count /// written (each range annotation contributes 1โ€“2 rows: start, and end if present). +/// +/// Runs on a caller-supplied transaction connection so the delete + all inserts are +/// **atomic** โ€” either every annotation row lands or none do. Callers (`ingest`, worker +/// `push`) own the transaction so the surrounding metadata update commits together with +/// the annotation set. A failed insert mid-loop rolls back the delete, so annotations +/// can never be partially lost. pub async fn replace_annotations( - db: &PgPool, + conn: &mut sqlx::PgConnection, video_id: i32, d: &ExportDoc, ) -> anyhow::Result { sqlx::query("DELETE FROM annotations WHERE video_id=$1") .bind(video_id) - .execute(db) + .execute(&mut *conn) .await?; let mut written = 0i32; for a in &d.fixed_annotations { - insert_ann(db, video_id, a.frame_number, &a.label, &a.side, &a.shape_type, + insert_ann(&mut *conn, video_id, a.frame_number, &a.label, &a.side, &a.shape_type, &a.vertices, &a.annotated_by, &a.review_status, &a.remark, &a.subclass, &a.created_at).await?; written += 1; } for a in &d.range_annotations { - insert_ann(db, video_id, a.start_frame, &a.label, &a.side, &a.shape_type, + insert_ann(&mut *conn, video_id, a.start_frame, &a.label, &a.side, &a.shape_type, &a.start_vertices, &a.annotated_by, &a.review_status, &a.remark, &a.subclass, &a.created_at).await?; written += 1; if let (Some(ef), Some(ev)) = (a.end_frame, a.end_vertices.as_ref()) { - insert_ann(db, video_id, ef, &a.label, &a.side, &a.shape_type, + insert_ann(&mut *conn, video_id, ef, &a.label, &a.side, &a.shape_type, ev, &a.annotated_by, &a.review_status, &a.remark, &a.subclass, &a.created_at).await?; written += 1; } @@ -229,7 +252,7 @@ pub async fn replace_annotations( #[allow(clippy::too_many_arguments)] async fn insert_ann( - db: &PgPool, video_id: i32, frame: i64, label: &str, side: &str, shape: &str, + conn: &mut sqlx::PgConnection, video_id: i32, frame: i64, label: &str, side: &str, shape: &str, vertices: &serde_json::Value, by: &str, review: &str, remark: &str, subclass: &str, created_at: &str, ) -> anyhow::Result<()> { sqlx::query( @@ -248,7 +271,7 @@ async fn insert_ann( .bind(remark) .bind(subclass) .bind(created_at) - .execute(db) + .execute(&mut *conn) .await?; Ok(()) } diff --git a/server/src/main.rs b/server/src/main.rs index eea90b0..8471e7f 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -790,7 +790,7 @@ async fn stats(State(s): State, user: AuthUser, AxPath(id): AxPath '' + WHERE v.project_id=$1 AND a.annotated_by <> '' AND lower(a.annotated_by) <> 'auto' GROUP BY a.annotated_by"#, ) .bind(id) @@ -812,14 +812,17 @@ async fn stats(State(s): State, user: AuthUser, AxPath(id): AxPath = leaderboard.iter().map(|r| (r.user.clone(), r.annotations)).collect(); - // Verification leaderboard: ALL assigned users (members) ranked by completion - // SPEED (avg verify time per video, fastest first). Anyone who has completed - // videos but isn't a member is included too; members with no completions sort - // last (unranked). Refreshed every poll โ†’ "live" ranking. + // Verification leaderboard: ALL assigned users (members) ranked by ANNOTATION + // THROUGHPUT โ€” annotations added per minute of work (most annotations in the least + // time first), not by video count. Anyone who has completed videos but isn't a + // member is included too; users with no annotations/time sort last (unranked). + // The synthetic 'auto' author (auto-segmented/imported) is excluded. Refreshed + // every poll โ†’ "live" ranking. let ver_rows: Vec<(String, i64, i64)> = sqlx::query_as( r#"SELECT completed_by, COUNT(*), COALESCE(SUM(verify_time_ms),0)::bigint FROM videos - WHERE project_id=$1 AND status='verified' AND completed_by IS NOT NULL AND completed_by <> '' + WHERE project_id=$1 AND status='verified' AND completed_by IS NOT NULL + AND completed_by <> '' AND lower(completed_by) <> 'auto' GROUP BY completed_by"#, ) .bind(id) @@ -834,29 +837,48 @@ async fn stats(State(s): State, user: AuthUser, AxPath(id): AxPath = ver_rows.into_iter().map(|(u, v, t)| (u, (v, t))).collect(); for m in &members { + if m.eq_ignore_ascii_case("auto") { + continue; + } stats_by_user.entry(m.clone()).or_insert((0, 0)); } let mut verifiers: Vec = stats_by_user .into_iter() - .map(|(user, (videos, total_time_ms))| VerifierRow { - avg_time_ms: if videos > 0 { total_time_ms / videos } else { 0 }, - videos, - total_time_ms, - annotations: ann_by_user.get(&user).copied().unwrap_or(0), - rank: 0, - user, + .map(|(user, (videos, total_time_ms))| { + let annotations = ann_by_user.get(&user).copied().unwrap_or(0); + // annotations added per minute of work; 0 when no time recorded yet. + let annotations_per_min = if total_time_ms > 0 { + annotations as f64 / (total_time_ms as f64 / 60_000.0) + } else { + 0.0 + }; + VerifierRow { + avg_time_ms: if videos > 0 { total_time_ms / videos } else { 0 }, + videos, + total_time_ms, + annotations, + annotations_per_min, + rank: 0, + user, + } }) .collect(); - // Fastest first: completers by avg ascending; non-completers last by name. - verifiers.sort_by(|a, b| match (a.videos > 0, b.videos > 0) { + // Highest throughput first (most annotations per minute). A user is rankable only + // once they have both annotations AND recorded time; everyone else sorts last by name. + let rankable = |v: &VerifierRow| v.annotations > 0 && v.annotations_per_min > 0.0; + verifiers.sort_by(|a, b| match (rankable(a), rankable(b)) { (true, false) => std::cmp::Ordering::Less, (false, true) => std::cmp::Ordering::Greater, - (true, true) => a.avg_time_ms.cmp(&b.avg_time_ms).then_with(|| a.user.cmp(&b.user)), + (true, true) => b + .annotations_per_min + .partial_cmp(&a.annotations_per_min) + .unwrap_or(std::cmp::Ordering::Equal) + .then_with(|| a.user.cmp(&b.user)), (false, false) => a.user.cmp(&b.user), }); let mut rank = 0i64; for v in verifiers.iter_mut() { - if v.videos > 0 { + if rankable(v) { rank += 1; v.rank = rank; } diff --git a/server/src/models.rs b/server/src/models.rs index 66d4115..3070b90 100644 --- a/server/src/models.rs +++ b/server/src/models.rs @@ -286,7 +286,11 @@ pub struct VerifierRow { pub avg_time_ms: i64, /// Total annotations this user authored (annotated_by) in the project. pub annotations: i64, - /// 1-based rank by completion speed (fastest avg first); 0 = no completions yet. + /// Annotations added per minute of work (annotations รท total_time). This is the + /// ranking metric: "most annotations in the least time". 0 when no time/annotations. + pub annotations_per_min: f64, + /// 1-based rank by annotations-per-minute (highest first); 0 = not yet ranked + /// (no annotations or no recorded time). pub rank: i64, } diff --git a/server/src/worker.rs b/server/src/worker.rs index 8a23719..6736b0b 100644 --- a/server/src/worker.rs +++ b/server/src/worker.rs @@ -218,20 +218,27 @@ pub async fn push( AxPath(id): AxPath, Json(body): Json, ) -> ApiResult { - // The current holder (or an admin) may push. A null holder (auto-released, not - // re-claimed) is tolerated; a different holder is rejected. - let claimed_by: Option> = - sqlx::query_scalar("SELECT claimed_by FROM videos WHERE id=$1") + // A non-admin pusher must currently hold an ACTIVE claim on this video. This blocks + // a stale/expired claim from overwriting newer authoritative server state: if the + // lease lapsed (auto-released, or re-claimed by someone else) the worker must + // re-claim before pushing โ€” their local work is never lost, it just can't clobber a + // claim they no longer hold. Admins may always push. + let row: Option<(Option, Option>)> = + sqlx::query_as("SELECT claimed_by, lease_expires_at FROM videos WHERE id=$1") .bind(id) .fetch_optional(&s.db) .await .map_err(err)?; - let claimed_by = claimed_by.ok_or((StatusCode::NOT_FOUND, "video not found".to_string()))?; + let (claimed_by, lease_expires_at) = + row.ok_or((StatusCode::NOT_FOUND, "video not found".to_string()))?; if !user.is_admin() { - if let Some(holder) = claimed_by.as_deref() { - if holder != user.username { - return Err((StatusCode::CONFLICT, "claim is held by another worker".into())); - } + let holds_active_claim = claimed_by.as_deref() == Some(user.username.as_str()) + && lease_expires_at.map(|l| l > chrono::Utc::now()).unwrap_or(false); + if !holds_active_claim { + return Err(( + StatusCode::CONFLICT, + "your claim has expired or is held by someone else โ€” re-claim the video before pushing".into(), + )); } } @@ -243,9 +250,6 @@ pub async fn push( let doc: ExportDoc = serde_json::from_value(body.clone()) .map_err(|e| (StatusCode::BAD_REQUEST, format!("invalid export doc: {e}")))?; - crate::ingest::replace_annotations(&s.db, id, &doc) - .await - .map_err(err)?; let ann_count = (doc.fixed_annotations.len() + doc.range_annotations.len()) as i32; // Recompute the video-level annotator fields from the pushed doc (same idea as @@ -261,6 +265,15 @@ pub async fn push( let primary_annotator = freq.into_iter().max_by_key(|(_, c)| *c).map(|(u, _)| u).unwrap_or_default(); let ann_time_ms = doc.video.as_ref().map(|v| v.annotation_time_ms).unwrap_or(0); + // Replace the annotation set AND flip the video to verified in ONE transaction: + // either the pushed annotations + the verified status both commit, or neither does. + // This makes a push all-or-nothing โ€” a verifier can never end up "verified" with a + // half-written annotation set, and a failed push leaves the prior pass intact. + let mut tx = s.db.begin().await.map_err(err)?; + crate::ingest::replace_annotations(&mut tx, id, &doc) + .await + .map_err(err)?; + // Accumulate verify_time_ms across passes (re-verification adds up). The list // of verifiers + the pass count are derived from video_events (each push logs one). sqlx::query( @@ -281,9 +294,10 @@ pub async fn push( .bind(&primary_annotator) .bind(ann_time_ms) .bind(imported_count) - .execute(&s.db) + .execute(&mut *tx) .await .map_err(err)?; + tx.commit().await.map_err(err)?; log_event(&s.db, id, &user.username, "push", json!({ "verify_time_ms": verify_time_ms })).await;
RankUserVerifiedSpeed (avg/video)AnnotationsTotal time
RankUserAnnotationsAnno/minVerifiedTotal time
{r.rank > 0 ? (r.rank === 1 ? "๐Ÿฅ‡ 1" : `#${r.rank}`) : โ€”} {r.user}{r.videos > 0 ? r.videos : 0}{r.videos > 0 ? fmtDuration(r.avg_time_ms) : โ€”} {r.annotations > 0 ? r.annotations : 0}{r.videos > 0 ? fmtDuration(r.total_time_ms) : โ€”}{r.annotations_per_min > 0 ? r.annotations_per_min.toFixed(1) : โ€”}{r.videos > 0 ? r.videos : 0}{r.total_time_ms > 0 ? fmtDuration(r.total_time_ms) : โ€”}