feat(prop-grid-rs): surface grid_tasks.kind in claim_next
Introduces TaskKind { Forecast, Analysis } and threads it through the
claim flow. Forecast-only filter on the claim query keeps the claim
lane clean while Elixir still owns f00 — a half-deployed cluster won't
deadlock on an unfulfillable analysis row.
Prep for Stream A of Phase 3: once the Rust-side native-duct + NEXRAD +
commercial pipeline lands, a sibling claim_next_analysis function pulls
kind='analysis' rows and dispatches to the new pipeline.
This commit is contained in:
parent
14d3e6ae5b
commit
7b1de3f995
1 changed files with 36 additions and 2 deletions
|
|
@ -15,6 +15,30 @@ use uuid::Uuid;
|
||||||
|
|
||||||
pub const NOTIFY_CHANNEL: &str = "propagation_ready";
|
pub const NOTIFY_CHANNEL: &str = "propagation_ready";
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||||
|
pub enum TaskKind {
|
||||||
|
/// f01..f18 — fetch + decode + score the forecast-hour grid. This is
|
||||||
|
/// what Rust has owned since Phase 2 cutover.
|
||||||
|
Forecast,
|
||||||
|
/// f00 enrichment — adds native-level duct merge, NEXRAD composite,
|
||||||
|
/// and commercial-link degradation on top of the forecast pipeline,
|
||||||
|
/// and writes the ProfilesFile consumed by /weather. Introduced in
|
||||||
|
/// Phase 3 Stream A.
|
||||||
|
Analysis,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl TaskKind {
|
||||||
|
fn from_str(s: &str) -> Self {
|
||||||
|
match s {
|
||||||
|
"analysis" => TaskKind::Analysis,
|
||||||
|
// Default to Forecast so older producers (which wrote rows
|
||||||
|
// before the kind column existed, defaulted to 'forecast'
|
||||||
|
// by migration) keep working.
|
||||||
|
_ => TaskKind::Forecast,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub struct ClaimedTask {
|
pub struct ClaimedTask {
|
||||||
pub id: Uuid,
|
pub id: Uuid,
|
||||||
|
|
@ -22,6 +46,7 @@ pub struct ClaimedTask {
|
||||||
pub forecast_hour: i16,
|
pub forecast_hour: i16,
|
||||||
pub valid_time: DateTime<Utc>,
|
pub valid_time: DateTime<Utc>,
|
||||||
pub attempt: i32,
|
pub attempt: i32,
|
||||||
|
pub kind: TaskKind,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, thiserror::Error)]
|
#[derive(Debug, thiserror::Error)]
|
||||||
|
|
@ -44,11 +69,16 @@ pub async fn connect(url: &str, max_connections: u32) -> Result<PgPool, DbError>
|
||||||
/// stamps claimed_at. Returns `None` if the queue is empty.
|
/// stamps claimed_at. Returns `None` if the queue is empty.
|
||||||
pub async fn claim_next(pool: &PgPool) -> Result<Option<ClaimedTask>, DbError> {
|
pub async fn claim_next(pool: &PgPool) -> Result<Option<ClaimedTask>, DbError> {
|
||||||
let mut tx = pool.begin().await?;
|
let mut tx = pool.begin().await?;
|
||||||
|
// Forecast-only claim. Analysis rows (kind='analysis') go through
|
||||||
|
// `claim_next_analysis` once the Rust-side f00 pipeline is wired;
|
||||||
|
// keeping them out of the forecast lane means a half-deployed cluster
|
||||||
|
// (Elixir still owns f00, Rust pulls forecasts) doesn't deadlock on
|
||||||
|
// an unfulfillable analysis row.
|
||||||
let row = sqlx::query(
|
let row = sqlx::query(
|
||||||
r#"
|
r#"
|
||||||
SELECT id, run_time, forecast_hour, valid_time, attempt
|
SELECT id, run_time, forecast_hour, valid_time, attempt, kind
|
||||||
FROM grid_tasks
|
FROM grid_tasks
|
||||||
WHERE status = 'queued' AND forecast_hour > 0
|
WHERE status = 'queued' AND forecast_hour > 0 AND kind = 'forecast'
|
||||||
ORDER BY run_time ASC, forecast_hour ASC
|
ORDER BY run_time ASC, forecast_hour ASC
|
||||||
FOR UPDATE SKIP LOCKED
|
FOR UPDATE SKIP LOCKED
|
||||||
LIMIT 1
|
LIMIT 1
|
||||||
|
|
@ -74,6 +104,8 @@ pub async fn claim_next(pool: &PgPool) -> Result<Option<ClaimedTask>, DbError> {
|
||||||
let valid_time_naive: NaiveDateTime = row.try_get("valid_time")?;
|
let valid_time_naive: NaiveDateTime = row.try_get("valid_time")?;
|
||||||
let valid_time = DateTime::<Utc>::from_naive_utc_and_offset(valid_time_naive, Utc);
|
let valid_time = DateTime::<Utc>::from_naive_utc_and_offset(valid_time_naive, Utc);
|
||||||
let attempt: i32 = row.try_get("attempt")?;
|
let attempt: i32 = row.try_get("attempt")?;
|
||||||
|
let kind_str: String = row.try_get("kind")?;
|
||||||
|
let kind = TaskKind::from_str(&kind_str);
|
||||||
|
|
||||||
sqlx::query(
|
sqlx::query(
|
||||||
r#"
|
r#"
|
||||||
|
|
@ -97,6 +129,7 @@ pub async fn claim_next(pool: &PgPool) -> Result<Option<ClaimedTask>, DbError> {
|
||||||
forecast_hour,
|
forecast_hour,
|
||||||
valid_time,
|
valid_time,
|
||||||
attempt: attempt + 1,
|
attempt: attempt + 1,
|
||||||
|
kind,
|
||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -204,6 +237,7 @@ mod tests {
|
||||||
let claimed = claim_next(&pool).await.unwrap().expect("had a row");
|
let claimed = claim_next(&pool).await.unwrap().expect("had a row");
|
||||||
assert!(claimed.forecast_hour > 0);
|
assert!(claimed.forecast_hour > 0);
|
||||||
assert_eq!(claimed.attempt, 1);
|
assert_eq!(claimed.attempt, 1);
|
||||||
|
assert_eq!(claimed.kind, TaskKind::Forecast);
|
||||||
|
|
||||||
complete(&pool, &claimed).await.unwrap();
|
complete(&pool, &claimed).await.unwrap();
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue