From 768b6b530639688d5f5f3d5ec570e21ab0624497 Mon Sep 17 00:00:00 2001 From: Aldo Borrero <82811+aldoborrero@users.noreply.github.com> Date: Tue, 6 Oct 2026 18:53:08 +0200 Subject: [PATCH 1/2] Revalidate import job commands and paged error downloads --- crates/horae/src/jobs.rs | 102 ++- crates/horae/src/jobs/access.rs | 42 ++ crates/horae/src/jobs/report.rs | 73 +- crates/horae/src/jobs/report/stream_tests.rs | 385 ++++++++++ crates/horae/src/server_fns/importers.rs | 66 +- .../importers/authorization_tests.rs | 125 +++- .../src/server_fns/importers/commands.rs | 145 ++++ .../server_fns/importers/commands/tests.rs | 692 ++++++++++++++++++ 8 files changed, 1519 insertions(+), 111 deletions(-) create mode 100644 crates/horae/src/jobs/access.rs create mode 100644 crates/horae/src/server_fns/importers/commands.rs create mode 100644 crates/horae/src/server_fns/importers/commands/tests.rs diff --git a/crates/horae/src/jobs.rs b/crates/horae/src/jobs.rs index f755a4bd..784cf0dc 100644 --- a/crates/horae/src/jobs.rs +++ b/crates/horae/src/jobs.rs @@ -15,6 +15,7 @@ use crate::{config::JobPolicy, state::AppState}; const LEASE: Duration = Duration::from_secs(300); const POLL: Duration = Duration::from_secs(2); +pub(crate) mod access; mod lease; pub(crate) mod outbox; pub(crate) mod report; @@ -214,8 +215,8 @@ impl JobPayload { #[error("Idempotency key conflicts with another import request")] pub struct RequestConflict; -pub async fn request_exists( - pool: &sqlx::PgPool, +pub async fn request_exists<'e, E: sqlx::PgExecutor<'e>>( + exec: E, org_id: Uuid, kind: &str, key: &str, @@ -227,7 +228,7 @@ pub async fn request_exists( kind, key, ) - .fetch_one(pool) + .fetch_one(exec) .await?) } @@ -250,6 +251,7 @@ pub async fn enqueue( Ok(id) } +#[cfg(test)] pub async fn enqueue_api( pool: &sqlx::PgPool, org_id: Uuid, @@ -258,20 +260,8 @@ pub async fn enqueue_api( policy: JobPolicy, generation: i64, ) -> anyhow::Result { - use crate::importers::harvest::account_switch::{self, ChangeError}; let mut tx = pool.begin().await?; - let current = account_switch::gate(&mut tx, org_id).await?; - anyhow::ensure!( - generation == current.account_generation, - ChangeError::OldImport - ); - let existing = sqlx::query_scalar!("SELECT id FROM horae_jobs WHERE org_id = $1 AND kind = 'harvest_api_import' AND idempotency_key = $2", org_id, idempotency_key) - .fetch_optional(&mut *tx).await?; - if existing.is_none() { - let connection = account_switch::status(&mut *tx, org_id, true).await?; - anyhow::ensure!(connection.connected, ChangeError::NotConnected); - } - let id = enqueue_in( + let id = enqueue_api_in( &mut tx, org_id, payload, @@ -284,6 +274,31 @@ pub async fn enqueue_api( Ok(id) } +/// Enqueue within the caller's authorized transaction; never acquire a worker reservation. +pub async fn enqueue_api_in( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + org_id: Uuid, + payload: &JobPayload, + idempotency_key: &str, + policy: JobPolicy, + generation: i64, +) -> anyhow::Result { + use crate::importers::harvest::account_switch::{self, ChangeError}; + let current = account_switch::gate(tx, org_id).await?; + anyhow::ensure!( + generation == current.account_generation, + ChangeError::OldImport + ); + let existing = sqlx::query_scalar!("SELECT id FROM horae_jobs WHERE org_id = $1 AND kind = 'harvest_api_import' AND idempotency_key = $2", org_id, idempotency_key) + .fetch_optional(&mut **tx).await?; + if existing.is_none() { + let connection = account_switch::status(&mut **tx, org_id, true).await?; + anyhow::ensure!(connection.connected, ChangeError::NotConnected); + } + let id = enqueue_in(tx, org_id, payload, idempotency_key, policy, generation).await?; + Ok(id) +} + async fn enqueue_in( tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, org_id: Uuid, @@ -318,6 +333,7 @@ async fn enqueue_in( Ok(row.id) } +#[cfg(test)] pub async fn enqueue_csv( pool: &sqlx::PgPool, org_id: Uuid, @@ -325,14 +341,28 @@ pub async fn enqueue_csv( body: Vec, idempotency_key: &str, policy: JobPolicy, +) -> anyhow::Result { + let mut tx = pool.begin().await?; + let id = enqueue_csv_in(&mut tx, org_id, mode, body, idempotency_key, policy).await?; + tx.commit().await?; + Ok(id) +} + +/// Persist the accepted upload and job atomically in the authorization transaction. +pub async fn enqueue_csv_in( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + org_id: Uuid, + mode: ImportMode, + body: Vec, + idempotency_key: &str, + policy: JobPolicy, ) -> anyhow::Result { let policy = policy.validate()?; let id = Uuid::now_v7(); let payload = JobPayload::HarvestCsv { mode }; let report = payload.initial_report()?; let payload = encode_payload(&payload)?; - let mut tx = pool.begin().await?; - crate::importers::harvest::account_switch::gate(&mut tx, org_id).await?; + crate::importers::harvest::account_switch::gate(tx, org_id).await?; let row = sqlx::query!( r#"INSERT INTO horae_jobs (id, org_id, kind, payload, idempotency_key, max_attempts, report) VALUES ($1, $2, 'harvest_csv_import', @@ -350,7 +380,7 @@ pub async fn enqueue_csv( report, &body, ) - .fetch_optional(&mut *tx) + .fetch_optional(&mut **tx) .await? .ok_or(RequestConflict)?; // An identical resubmission must not restore an upload already removed by @@ -364,10 +394,9 @@ pub async fn enqueue_csv( org_id, body, ) - .execute(&mut *tx) + .execute(&mut **tx) .await?; } - tx.commit().await?; Ok(row.id) } @@ -398,8 +427,8 @@ fn retry_availability( } } -pub async fn status( - pool: &sqlx::PgPool, +pub async fn status<'e, E: sqlx::PgExecutor<'e>>( + exec: E, org_id: Uuid, id: Uuid, ) -> anyhow::Result> { @@ -418,7 +447,7 @@ pub async fn status( id, org_id, ) - .fetch_optional(pool) + .fetch_optional(exec) .await?; Ok(row.map(|r| crate::models::JobStatus { retry_availability: retry_availability( @@ -442,8 +471,8 @@ pub async fn status( })) } -pub async fn list( - pool: &sqlx::PgPool, +pub async fn list<'e, E: sqlx::PgExecutor<'e>>( + exec: E, org_id: Uuid, limit: i64, before: Option, @@ -469,7 +498,7 @@ pub async fn list( limit.clamp(1, 100), before, ) - .fetch_all(pool) + .fetch_all(exec) .await?; Ok(rows .into_iter() @@ -518,11 +547,23 @@ pub async fn cancel<'e, E: sqlx::PgExecutor<'e>>( Ok(result.rows_affected() == 1) } +#[cfg(test)] pub async fn retry(pool: &sqlx::PgPool, org_id: Uuid, id: Uuid) -> anyhow::Result { let mut tx = pool.begin().await?; - let current = crate::importers::harvest::account_switch::gate(&mut tx, org_id).await?; + let retried = retry_in(&mut tx, org_id, id).await?; + tx.commit().await?; + Ok(retried) +} + +/// Retry within the caller's authorization transaction, retaining generation fencing. +pub async fn retry_in( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + org_id: Uuid, + id: Uuid, +) -> anyhow::Result { + let current = crate::importers::harvest::account_switch::gate(tx, org_id).await?; let generation = sqlx::query_scalar!("SELECT account_generation FROM horae_jobs WHERE id = $1 AND org_id = $2 AND kind = 'harvest_api_import'", id, org_id) - .fetch_optional(&mut *tx).await?; + .fetch_optional(&mut **tx).await?; anyhow::ensure!( generation.is_none_or(|generation| generation == current.account_generation), crate::importers::harvest::account_switch::ChangeError::OldImport @@ -540,9 +581,8 @@ pub async fn retry(pool: &sqlx::PgPool, org_id: Uuid, id: Uuid) -> anyhow::Resul id, org_id, ) - .execute(&mut *tx) + .execute(&mut **tx) .await?; - tx.commit().await?; Ok(result.rows_affected() == 1) } diff --git a/crates/horae/src/jobs/access.rs b/crates/horae/src/jobs/access.rs new file mode 100644 index 00000000..d563d53a --- /dev/null +++ b/crates/horae/src/jobs/access.rs @@ -0,0 +1,42 @@ +//! Current user authority for import commands and retained results, not worker leases. + +use horae_core::types::OrgRole; +use sqlx::{PgPool, Postgres, Transaction}; +use uuid::Uuid; + +#[derive(Debug, thiserror::Error)] +pub(crate) enum AccessError { + #[error("Active administrator access required")] + Forbidden, + #[error(transparent)] + Database(#[from] sqlx::Error), +} + +/// Hold current tenant-bound authority through a bounded operation's commit. +pub(crate) async fn begin_admin_access( + pool: &PgPool, + org_id: Uuid, + actor_id: Uuid, +) -> Result, AccessError> { + let mut tx = pool.begin().await?; + sqlx::query!("SET TRANSACTION ISOLATION LEVEL READ COMMITTED") + .execute(&mut *tx) + .await?; + crate::db::lock_organization(&mut tx, org_id, crate::db::OrganizationLock::Shared) + .await + .map_err(|error| match error { + sqlx::Error::RowNotFound => AccessError::Forbidden, + other => AccessError::Database(other), + })?; + sqlx::query_scalar!( + "SELECT id FROM users WHERE id = $1 AND org_id = $2 + AND active AND org_role = $3 FOR SHARE", + actor_id, + org_id, + OrgRole::Admin as OrgRole, + ) + .fetch_optional(&mut *tx) + .await? + .ok_or(AccessError::Forbidden)?; + Ok(tx) +} diff --git a/crates/horae/src/jobs/report.rs b/crates/horae/src/jobs/report.rs index da8e6253..539365e2 100644 --- a/crates/horae/src/jobs/report.rs +++ b/crates/horae/src/jobs/report.rs @@ -104,8 +104,8 @@ async fn append_report_chunk( } /// A fixed-size read at a captured report boundary, never the archive's live end. -pub(crate) async fn chunks( - pool: &sqlx::PgPool, +pub(crate) async fn chunks<'e, E: sqlx::PgExecutor<'e>>( + exec: E, org_id: Uuid, job_id: Uuid, next: i64, @@ -121,7 +121,7 @@ pub(crate) async fn chunks( next, end, ) - .fetch_all(pool) + .fetch_all(exec) .await?; anyhow::ensure!( rows.len() as i64 == (end - next).min(16), @@ -145,10 +145,7 @@ pub(crate) async fn download( session: tower_sessions::Session, axum::extract::Path(job_id): axum::extract::Path, ) -> Result { - use axum::{ - http::{StatusCode, header}, - response::IntoResponse, - }; + use axum::http::StatusCode; let user_id = crate::auth::session::get_session_user_id(&session) .await @@ -166,7 +163,21 @@ pub(crate) async fn download( if user.org_role != horae_core::types::OrgRole::Admin { return Err(StatusCode::FORBIDDEN); } - let job = super::status(&state.db, user.org_id, job_id) + download_response(&state.db, user.org_id, user_id, job_id).await +} + +async fn download_response( + pool: &sqlx::PgPool, + org_id: Uuid, + actor_id: Uuid, + job_id: Uuid, +) -> Result { + use axum::{ + http::{StatusCode, header}, + response::IntoResponse, + }; + let mut tx = download_access(pool, org_id, actor_id).await?; + let job = super::status(&mut *tx, org_id, job_id) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? .ok_or(StatusCode::NOT_FOUND)?; @@ -179,6 +190,9 @@ pub(crate) async fn download( serde_json::to_writer(&mut tail, &error).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; tail.push(b'\n'); } + tx.commit() + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; Ok(( [ (header::CONTENT_TYPE, "application/x-ndjson".to_owned()), @@ -189,14 +203,30 @@ pub(crate) async fn download( (header::CACHE_CONTROL, "no-store".to_owned()), (header::X_CONTENT_TYPE_OPTIONS, "nosniff".to_owned()), ], - download_body(state.db.clone(), user.org_id, job_id, end, tail), + download_body(pool.clone(), org_id, actor_id, job_id, end, tail), ) .into_response()) } +async fn download_access( + pool: &sqlx::PgPool, + org_id: Uuid, + actor_id: Uuid, +) -> Result, axum::http::StatusCode> { + super::access::begin_admin_access(pool, org_id, actor_id) + .await + .map_err(|error| match error { + super::access::AccessError::Forbidden => axum::http::StatusCode::FORBIDDEN, + super::access::AccessError::Database(_) => { + axum::http::StatusCode::INTERNAL_SERVER_ERROR + } + }) +} + fn download_body( pool: sqlx::PgPool, org_id: Uuid, + actor_id: Uuid, job_id: Uuid, end: i64, tail: Vec, @@ -208,16 +238,25 @@ fn download_body( move |(mut next, mut pending, mut tail)| { let pool = pool.clone(); async move { - if pending.is_empty() && next < end { - let page = chunks(&pool, org_id, job_id, next, end) + if pending.is_empty() && (next < end || tail.is_some()) { + let mut tx = download_access(&pool, org_id, actor_id) + .await + .map_err(|status| std::io::Error::other(status.to_string()))?; + if next < end { + let page = chunks(&mut *tx, org_id, job_id, next, end) + .await + .map_err(|_| std::io::Error::other("report archive is incomplete"))?; + next += page.len() as i64; + pending.extend(page); + } else if let Some(bytes) = tail.take().filter(|bytes| !bytes.is_empty()) { + pending.push_back(bytes); + } + // Release authority and the connection before client-paced output. + tx.commit() .await - .map_err(std::io::Error::other)?; - next += page.len() as i64; - pending.extend(page); + .map_err(|_| std::io::Error::other("report download failed"))?; } - let chunk = pending - .pop_front() - .or_else(|| tail.take().filter(|bytes| !bytes.is_empty())); + let chunk = pending.pop_front(); Ok::<_, std::io::Error>(chunk.map(|chunk| (chunk, (next, pending, tail)))) } }, diff --git a/crates/horae/src/jobs/report/stream_tests.rs b/crates/horae/src/jobs/report/stream_tests.rs index a54579a4..51545f1f 100644 --- a/crates/horae/src/jobs/report/stream_tests.rs +++ b/crates/horae/src/jobs/report/stream_tests.rs @@ -9,6 +9,7 @@ use crate::{config::JobPolicy, jobs}; struct ArchivedJob { org: Uuid, + actor: Uuid, id: Uuid, lease: jobs::JobLease, _stop: tokio::sync::watch::Sender, @@ -26,6 +27,7 @@ impl ArchivedJob { super::download_body( pool.clone(), self.org, + self.actor, self.id, self.report.archived_error_chunks().try_into().unwrap(), tail, @@ -42,6 +44,11 @@ async fn archived_job(pool: &PgPool) -> ArchivedJob { .execute(pool) .await .unwrap(); + let actor = Uuid::now_v7(); + sqlx::query!( + "INSERT INTO users (id, org_id, email, name, org_role) VALUES ($1, $2, $3, 'Report reader', 'admin')", + actor, org, format!("{actor}@example.test"), + ).execute(pool).await.unwrap(); let id = jobs::enqueue( pool, org, @@ -88,6 +95,7 @@ async fn archived_job(pool: &PgPool) -> ArchivedJob { tx.commit().await.unwrap(); ArchivedJob { org, + actor, id, lease, _stop: stop, @@ -96,6 +104,383 @@ async fn archived_job(pool: &PgPool) -> ArchivedJob { } } +#[sqlx::test] +#[serial_test::serial] +async fn download_preparation_denies_revoked_authority(pool: PgPool) { + let job = archived_job(&pool).await; + sqlx::query!( + "UPDATE users SET org_role = 'member' WHERE id = $1", + job.actor + ) + .execute(&pool) + .await + .unwrap(); + let result = super::download_response(&pool, job.org, job.actor, job.id).await; + assert!( + matches!(result, Err(axum::http::StatusCode::FORBIDDEN)), + "{result:?}" + ); +} + +#[sqlx::test] +#[serial_test::serial] +async fn download_denies_revocation_before_first_page_and_inline_tail(pool: PgPool) { + let job = archived_job(&pool).await; + for (end, tail) in [ + (0, Vec::new()), + (0, b"private tail\n".to_vec()), + ( + job.report.archived_error_chunks().try_into().unwrap(), + Vec::new(), + ), + ] { + sqlx::query!( + "UPDATE users SET org_role = 'admin' WHERE id = $1", + job.actor + ) + .execute(&pool) + .await + .unwrap(); + let body = super::download_body(pool.clone(), job.org, job.actor, job.id, end, tail); + sqlx::query!( + "UPDATE users SET org_role = 'member' WHERE id = $1", + job.actor + ) + .execute(&pool) + .await + .unwrap(); + let result = body + .into_data_stream() + .next() + .await + .expect("denial must not become successful EOF"); + assert!(result.is_err(), "revoked reader received report bytes"); + } +} + +#[sqlx::test] +#[serial_test::serial] +async fn download_rechecks_after_buffered_pages_and_before_the_captured_tail(pool: PgPool) { + let job = archived_job(&pool).await; + let end = job.report.archived_error_chunks() as usize; + for consumed in [1_usize, 16, end] { + sqlx::query!("UPDATE users SET active = true WHERE id = $1", job.actor) + .execute(&pool) + .await + .unwrap(); + let mut stream = job.body(&pool).into_data_stream(); + for _ in 0..consumed { + stream.next().await.unwrap().unwrap(); + } + sqlx::query!("UPDATE users SET active = false WHERE id = $1", job.actor) + .execute(&pool) + .await + .unwrap(); + // Only the bounded page already authorized may finish draining. + for _ in consumed..(consumed.div_ceil(16) * 16).min(end) { + stream.next().await.unwrap().unwrap(); + } + let error = stream + .next() + .await + .expect("revocation is not successful EOF") + .unwrap_err(); + assert!(error.to_string().contains("403"), "{error}"); + assert!(stream.next().await.is_none()); + } +} + +#[sqlx::test] +#[serial_test::serial] +async fn download_preparation_is_tenant_bound_and_releases_failed_transactions(pool: PgPool) { + use axum::http::StatusCode; + use horae_core::types::OrgRole; + let job = archived_job(&pool).await; + let foreign = crate::server_fns::test_seed::seed(&pool, OrgRole::Admin).await; + for (org, actor, id, expected) in [ + (job.org, foreign.user_id, job.id, StatusCode::FORBIDDEN), + (job.org, Uuid::now_v7(), job.id, StatusCode::FORBIDDEN), + (Uuid::now_v7(), job.actor, job.id, StatusCode::FORBIDDEN), + ( + foreign.org_id, + foreign.user_id, + job.id, + StatusCode::NOT_FOUND, + ), + (job.org, job.actor, Uuid::now_v7(), StatusCode::NOT_FOUND), + ] { + assert_eq!( + super::download_response(&pool, org, actor, id) + .await + .unwrap_err(), + expected + ); + } + for (role, active) in [ + (OrgRole::Member, true), + (OrgRole::Manager, true), + (OrgRole::Admin, false), + ] { + sqlx::query!( + "UPDATE users SET org_role = $2, active = $3 WHERE id = $1", + job.actor, + role as OrgRole, + active + ) + .execute(&pool) + .await + .unwrap(); + assert_eq!( + super::download_response(&pool, job.org, job.actor, job.id) + .await + .unwrap_err(), + StatusCode::FORBIDDEN + ); + } + sqlx::query!("UPDATE users SET active = true WHERE id = $1", job.actor) + .execute(&pool) + .await + .unwrap(); + let single = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect_with((*pool.connect_options()).clone()) + .await + .unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(5), async { + // Status prefers the checkpoint; remove it to exercise the malformed final report. + sqlx::query!( + "UPDATE horae_jobs SET checkpoint = NULL, report = '{}' WHERE id = $1", + job.id + ) + .execute(&single) + .await + .unwrap(); + assert_eq!( + super::download_response(&single, job.org, job.actor, job.id) + .await + .unwrap_err(), + StatusCode::INTERNAL_SERVER_ERROR + ); + sqlx::query!( + "UPDATE horae_jobs SET report = $2 WHERE id = $1", + job.id, + serde_json::to_value(&job.report).unwrap() + ) + .execute(&single) + .await + .unwrap(); + let response = super::download_response(&single, job.org, job.actor, job.id) + .await + .unwrap(); + let body = axum::body::to_bytes(response.into_body(), super::CHUNK_BYTES * 40) + .await + .unwrap(); + let errors: Vec = serde_json::Deserializer::from_slice(&body) + .into_iter() + .collect::>() + .unwrap(); + assert_eq!(errors, job.expected); + }) + .await + .expect("failed and successful preparation must release the single connection"); + single.close().await; +} + +#[derive(Clone, Copy, Debug)] +enum Boundary { + Preparation, + Page, + Tail, + Empty, +} + +async fn read_boundary( + pool: &PgPool, + org: Uuid, + actor: Uuid, + job: Uuid, + boundary: Boundary, +) -> Result<(), String> { + if matches!(boundary, Boundary::Preparation) { + return super::download_response(pool, org, actor, job) + .await + .map(|_| ()) + .map_err(|status| status.to_string()); + } + let end = i64::from(matches!(boundary, Boundary::Page)); + let tail = if matches!(boundary, Boundary::Tail) { + b"private tail\n".to_vec() + } else { + Vec::new() + }; + let mut stream = + super::download_body(pool.clone(), org, actor, job, end, tail).into_data_stream(); + stream + .next() + .await + .transpose() + .map(|_| ()) + .map_err(|error| error.to_string()) +} + +#[sqlx::test] +#[serial_test::serial] +async fn download_boundaries_recheck_after_organization_and_actor_waits(pool: PgPool) { + use crate::server_fns::test_seed::wait_for_blocked; + use horae_core::types::OrgRole; + let reading = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .after_connect(|connection, _| { + Box::pin(async move { + sqlx::query!( + "SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL REPEATABLE READ" + ) + .execute(connection) + .await?; + Ok(()) + }) + }) + .connect_with((*pool.connect_options()).clone()) + .await + .unwrap(); + let job = archived_job(&pool).await; + for boundary in [ + Boundary::Preparation, + Boundary::Page, + Boundary::Tail, + Boundary::Empty, + ] { + for actor_only in [false, true] { + for deactivate in [false, true] { + sqlx::query!( + "UPDATE users SET org_role = 'admin', active = true WHERE id = $1", + job.actor + ) + .execute(&pool) + .await + .unwrap(); + let mut revoke = pool.begin().await.unwrap(); + let blocker = sqlx::query_scalar!("SELECT pg_backend_pid()") + .fetch_one(&mut *revoke) + .await + .unwrap() + .unwrap(); + if !actor_only { + sqlx::query!( + "SELECT id FROM organizations WHERE id = $1 FOR UPDATE", + job.org + ) + .fetch_one(&mut *revoke) + .await + .unwrap(); + } + sqlx::query!( + "UPDATE users SET org_role = $2, active = $3 WHERE id = $1", + job.actor, + if deactivate { + OrgRole::Admin + } else { + OrgRole::Member + } as OrgRole, + !deactivate + ) + .execute(&mut *revoke) + .await + .unwrap(); + let db = reading.clone(); + let mut pending = tokio::task::JoinSet::new(); + pending.spawn(async move { + read_boundary(&db, job.org, job.actor, job.id, boundary).await + }); + tokio::select! { + result = pending.join_next() => panic!("{boundary:?} bypassed authorization wait: {result:?}"), + () = wait_for_blocked(&pool, blocker) => {}, + } + revoke.commit().await.unwrap(); + let error = + tokio::time::timeout(std::time::Duration::from_secs(5), pending.join_next()) + .await + .unwrap() + .unwrap() + .unwrap() + .unwrap_err(); + assert!(error.contains("403"), "{boundary:?}: {error}"); + } + } + } + reading.close().await; +} + +#[sqlx::test] +#[serial_test::serial] +async fn download_readers_retain_authority_through_their_bounded_reads(pool: PgPool) { + use crate::server_fns::test_seed::wait_for_blocked; + let job = archived_job(&pool).await; + for boundary in [Boundary::Preparation, Boundary::Page] { + sqlx::query!( + "UPDATE users SET org_role = 'admin' WHERE id = $1", + job.actor + ) + .execute(&pool) + .await + .unwrap(); + let mut barrier = pool.begin().await.unwrap(); + let blocker = sqlx::query_scalar!("SELECT pg_backend_pid()") + .fetch_one(&mut *barrier) + .await + .unwrap() + .unwrap(); + match boundary { + Boundary::Preparation => sqlx::query!("LOCK TABLE horae_jobs IN ACCESS EXCLUSIVE MODE") + .execute(&mut *barrier) + .await + .unwrap(), + _ => sqlx::query!("LOCK TABLE horae_job_report_error_chunks IN ACCESS EXCLUSIVE MODE") + .execute(&mut *barrier) + .await + .unwrap(), + }; + let db = pool.clone(); + let mut read = tokio::task::JoinSet::new(); + read.spawn(async move { read_boundary(&db, job.org, job.actor, job.id, boundary).await }); + wait_for_blocked(&pool, blocker).await; + let reader = sqlx::query_scalar!( + "SELECT pid AS \"pid!\" FROM pg_stat_activity WHERE datname=current_database() AND $1=ANY(pg_blocking_pids(pid))", blocker, + ).fetch_one(&pool).await.unwrap(); + let db = pool.clone(); + let mut revoke = tokio::task::JoinSet::new(); + revoke.spawn(async move { + sqlx::query!( + "UPDATE users SET org_role = 'member' WHERE id = $1", + job.actor + ) + .execute(&db) + .await + }); + tokio::select! { + result = revoke.join_next() => panic!("{boundary:?} released authority early: {result:?}"), + () = wait_for_blocked(&pool, reader) => {}, + } + barrier.commit().await.unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(5), read.join_next()) + .await + .unwrap() + .unwrap() + .unwrap() + .unwrap(); + tokio::time::timeout(std::time::Duration::from_secs(5), revoke.join_next()) + .await + .unwrap() + .unwrap() + .unwrap() + .unwrap(); + let error = read_boundary(&pool, job.org, job.actor, job.id, boundary) + .await + .unwrap_err(); + assert!(error.contains("403"), "{boundary:?}: {error}"); + } +} + #[sqlx::test] #[serial_test::serial] async fn download_keeps_its_snapshot_when_later_progress_archives_the_inline_tail(pool: PgPool) { diff --git a/crates/horae/src/server_fns/importers.rs b/crates/horae/src/server_fns/importers.rs index d9f36148..f9cafc9b 100644 --- a/crates/horae/src/server_fns/importers.rs +++ b/crates/horae/src/server_fns/importers.rs @@ -10,6 +10,9 @@ use horae_core::importers::harvest::types::{ConnectionStatus, ImportMode, SyncSc mod csv_upload; pub use csv_upload::CsvUpload; +#[cfg(feature = "server")] +mod commands; + #[cfg(all(test, feature = "server"))] mod authorization_tests; @@ -138,17 +141,16 @@ pub async fn start_harvest_api_import( if state.harvest.is_none() { return Err(err(NOT_FOUND, "Harvest is not configured")); } - let id = crate::jobs::enqueue_api( + commands::start_api( &state.db, admin.org_id, + admin.id, &payload, &key, state.job_policy, generation.unwrap_or(0), ) .await - .map_err(map_enqueue_error)?; - required_import_job(&state.db, admin.org_id, id).await } /// Return a durable import's current state for the current organization. @@ -158,9 +160,7 @@ pub async fn get_harvest_import_job( ) -> Result, ServerFnError> { let admin = require_admin().await?; let state = crate::state::global_state().await; - crate::jobs::status(&state.db, admin.org_id, job_id) - .await - .map_err(server_err) + commands::status(&state.db, admin.org_id, admin.id, job_id).await } #[dioxus_fullstack::post("/api/import/harvest/history")] @@ -170,9 +170,14 @@ pub async fn list_harvest_import_jobs( ) -> Result, ServerFnError> { let admin = require_admin().await?; let state = crate::state::global_state().await; - crate::jobs::list(&state.db, admin.org_id, limit.unwrap_or(20), before) - .await - .map_err(server_err) + commands::history( + &state.db, + admin.org_id, + admin.id, + limit.unwrap_or(20), + before, + ) + .await } /// Buffer and enqueue a CSV import so the request body is not tied to the @@ -188,28 +193,16 @@ pub async fn start_harvest_csv_import( .await .map_err(|_| err(BAD_REQUEST, "CSV upload is incomplete or exceeds 50 MiB"))?; let state = crate::state::global_state().await; - if !crate::jobs::request_exists(&state.db, admin.org_id, "harvest_csv_import", &key) - .await - .map_err(server_err)? - { - crate::importers::harvest::csv_source::validate_upload_headers(&body).map_err(|_| { - err( - BAD_REQUEST, - "Not a recognizable Harvest CSV; check required columns", - ) - })?; - } - let id = crate::jobs::enqueue_csv( + commands::start_csv( &state.db, admin.org_id, + admin.id, mode, body.to_vec(), &key, state.job_policy, ) .await - .map_err(map_enqueue_error)?; - required_import_job(&state.db, admin.org_id, id).await } #[dioxus_fullstack::post("/api/import/harvest/cancel")] @@ -218,11 +211,7 @@ pub async fn cancel_harvest_import_job( ) -> Result { let admin = require_admin().await?; let state = crate::state::global_state().await; - crate::jobs::cancel(&state.db, admin.org_id, job_id) - .await - .map_err(server_err)?; - // Completion can race cancellation; report the actual retained state. - required_import_job(&state.db, admin.org_id, job_id).await + commands::cancel(&state.db, admin.org_id, admin.id, job_id).await } #[dioxus_fullstack::post("/api/import/harvest/retry")] @@ -231,14 +220,7 @@ pub async fn retry_harvest_import_job( ) -> Result { let admin = require_admin().await?; let state = crate::state::global_state().await; - if crate::jobs::retry(&state.db, admin.org_id, job_id) - .await - .map_err(map_enqueue_error)? - { - required_import_job(&state.db, admin.org_id, job_id).await - } else { - Err(err(NOT_FOUND, "Import job not found or is not retryable")) - } + commands::retry(&state.db, admin.org_id, admin.id, job_id).await } #[cfg(feature = "server")] @@ -305,18 +287,6 @@ fn map_enqueue_error(error: anyhow::Error) -> ServerFnError { } } -#[cfg(feature = "server")] -async fn required_import_job( - pool: &sqlx::PgPool, - org_id: uuid::Uuid, - job_id: uuid::Uuid, -) -> Result { - crate::jobs::status(pool, org_id, job_id) - .await - .map_err(server_err)? - .ok_or_else(|| err(NOT_FOUND, "Import job not found")) -} - /// The configured Harvest settings, or a clear error when the importer's API /// source is not configured on this deployment. #[cfg(feature = "server")] diff --git a/crates/horae/src/server_fns/importers/authorization_tests.rs b/crates/horae/src/server_fns/importers/authorization_tests.rs index da65e298..5735e5d8 100644 --- a/crates/horae/src/server_fns/importers/authorization_tests.rs +++ b/crates/horae/src/server_fns/importers/authorization_tests.rs @@ -206,6 +206,7 @@ async fn job_endpoints_enforce_session_role_and_organization(pool: PgPool) { None, ) .await; + let upload_pool = pool.clone(); let router = Router::new() .register_server_functions() .route( @@ -251,21 +252,55 @@ async fn job_endpoints_enforce_session_role_and_organization(pool: PgPool) { }), ) .layer(middleware::from_fn( - |request: Request, next: middleware::Next| async move { - let request = if request.headers().contains_key("X-Test-Unread-Upload") { - request.map(|_| { - Body::from_stream(futures_util::stream::poll_fn( - |_| -> std::task::Poll< - Option>, - > { - panic!("unauthorized CSV job read its upload"); - }, - )) - }) - } else { - request - }; - next.run(request).await + move |request: Request, next: middleware::Next| { + let pool = upload_pool.clone(); + async move { + let revoke_download = request + .headers() + .get("X-Test-Revoke-Download") + .map(|actor| Uuid::parse_str(actor.to_str().unwrap()).unwrap()); + let upload_pool = pool.clone(); + let request = if request.headers().contains_key("X-Test-Unread-Upload") { + request.map(|_| { + Body::from_stream(futures_util::stream::poll_fn( + |_| -> std::task::Poll< + Option>, + > { + panic!("unauthorized CSV job read its upload"); + }, + )) + }) + } else if let Some(actor) = request.headers().get("X-Test-Revoke-Upload") { + let actor = Uuid::parse_str(actor.to_str().unwrap()).unwrap(); + request.map(|_| { + Body::from_stream(futures_util::stream::once(async move { + // Body polling follows endpoint admission, before durable acceptance. + sqlx::query!( + "UPDATE users SET org_role = 'member' WHERE id = $1", + actor + ) + .execute(&upload_pool) + .await + .unwrap(); + Ok::<_, std::io::Error>(axum::body::Bytes::from_static( + b"Date,Client,Project,Task,Hours,Email\n", + )) + })) + }) + } else { + request + }; + let response = next.run(request).await; + if let Some(actor) = revoke_download { + assert_eq!(response.status(), StatusCode::OK); + // Preparation succeeded, but the client has not polled the body yet. + sqlx::query!("UPDATE users SET org_role = 'member' WHERE id = $1", actor) + .execute(&pool) + .await + .unwrap(); + } + response + } }, )) .layer( @@ -304,6 +339,25 @@ async fn job_endpoints_enforce_session_role_and_organization(pool: PgPool) { StatusCode::NO_CONTENT ); let outsider = api.cookie(foreign.user_id).await; + let upload_actor = user(&pool, owner.org_id, OrgRole::Admin).await; + let upload_cookie = api.cookie(upload_actor).await; + let revoked_upload = api + .client + .post(format!("{}/api/import/harvest/csv-job/DryRun", api.base)) + .header("cookie", upload_cookie) + .header("X-Horae-Import", "csv") + .header("X-Test-Revoke-Upload", upload_actor.to_string()) + .body("replaced by the paused-body fixture") + .send() + .await + .unwrap(); + assert_eq!(revoked_upload.status(), StatusCode::FORBIDDEN); + assert!( + crate::jobs::list(&pool, owner.org_id, 100, None) + .await + .unwrap() + .is_empty() + ); let member = api .cookie(user(&pool, owner.org_id, OrgRole::Member).await) .await; @@ -759,12 +813,53 @@ async fn job_endpoints_enforce_session_role_and_organization(pool: PgPool) { assert_eq!(response.status(), StatusCode::OK); assert_eq!(response.headers()["content-type"], "application/x-ndjson"); assert_eq!(response.headers()["cache-control"], "no-store"); + assert_eq!(response.headers()["x-content-type-options"], "nosniff"); + assert_eq!( + response.headers()["content-disposition"], + format!( + "attachment; filename=\"import-{}-errors.jsonl\"", + archived_job.id + ) + ); let bytes = response.bytes().await.unwrap(); let actual = serde_json::Deserializer::from_slice(&bytes) .into_iter::() .collect::, _>>() .unwrap(); assert_eq!(actual, expected); + let download_actor = user(&pool, owner.org_id, OrgRole::Admin).await; + let download_cookie = api.cookie(download_actor).await; + let interrupted = async { + let response = api + .client + .get(format!( + "{}/api/import/harvest/jobs/{}/errors", + api.base, archived_job.id + )) + .header("cookie", &download_cookie) + .header("X-Test-Revoke-Download", download_actor.to_string()) + .send() + .await?; + assert_eq!(response.status(), StatusCode::OK); + response.bytes().await + } + .await + .expect_err("revocation must abort the HTTP body, not return successful EOF"); + assert!(!interrupted.is_timeout(), "{interrupted}"); + assert_eq!( + api.errors(archived_job.id, Some(&download_cookie)) + .await + .status(), + StatusCode::FORBIDDEN + ); + assert_eq!( + api.errors(archived_job.id, Some(&admin)) + .await + .bytes() + .await + .unwrap(), + bytes + ); let end = i64::try_from(report.archived_error_chunks()).unwrap(); assert!( crate::jobs::report::chunks(&pool, owner.org_id, archived_job.id, end, end + 1) diff --git a/crates/horae/src/server_fns/importers/commands.rs b/crates/horae/src/server_fns/importers/commands.rs new file mode 100644 index 00000000..4e327008 --- /dev/null +++ b/crates/horae/src/server_fns/importers/commands.rs @@ -0,0 +1,145 @@ +//! Session-bound importer operations, separate from queue worker mechanics. + +use super::*; +use crate::{config::JobPolicy, jobs, models::JobStatus}; +use sqlx::PgPool; +use uuid::Uuid; + +pub(super) async fn start_api( + pool: &PgPool, + org_id: Uuid, + actor_id: Uuid, + payload: &jobs::JobPayload, + key: &str, + policy: JobPolicy, + generation: i64, +) -> Result { + let mut tx = begin_access(pool, org_id, actor_id).await?; + let id = jobs::enqueue_api_in(&mut tx, org_id, payload, key, policy, generation) + .await + .map_err(map_enqueue_error)?; + let result = required_job(&mut tx, org_id, id).await?; + tx.commit().await.map_err(server_err)?; + Ok(result) +} + +pub(super) async fn start_csv( + pool: &PgPool, + org_id: Uuid, + actor_id: Uuid, + mode: ImportMode, + body: Vec, + key: &str, + policy: JobPolicy, +) -> Result { + let mut tx = begin_access(pool, org_id, actor_id).await?; + if !jobs::request_exists(&mut *tx, org_id, "harvest_csv_import", key) + .await + .map_err(server_err)? + { + crate::importers::harvest::csv_source::validate_upload_headers(&body).map_err(|_| { + err( + BAD_REQUEST, + "Not a recognizable Harvest CSV; check required columns", + ) + })?; + } + let id = jobs::enqueue_csv_in(&mut tx, org_id, mode, body, key, policy) + .await + .map_err(map_enqueue_error)?; + let result = required_job(&mut tx, org_id, id).await?; + tx.commit().await.map_err(server_err)?; + Ok(result) +} + +pub(super) async fn cancel( + pool: &PgPool, + org_id: Uuid, + actor_id: Uuid, + id: Uuid, +) -> Result { + let mut tx = begin_access(pool, org_id, actor_id).await?; + jobs::cancel(&mut *tx, org_id, id) + .await + .map_err(server_err)?; + // Completion can race cancellation; report the actual retained state. + let result = required_job(&mut tx, org_id, id).await?; + tx.commit().await.map_err(server_err)?; + Ok(result) +} + +pub(super) async fn retry( + pool: &PgPool, + org_id: Uuid, + actor_id: Uuid, + id: Uuid, +) -> Result { + let mut tx = begin_access(pool, org_id, actor_id).await?; + if !jobs::retry_in(&mut tx, org_id, id) + .await + .map_err(map_enqueue_error)? + { + return Err(err(NOT_FOUND, "Import job not found or is not retryable")); + } + let result = required_job(&mut tx, org_id, id).await?; + tx.commit().await.map_err(server_err)?; + Ok(result) +} + +pub(super) async fn status( + pool: &PgPool, + org_id: Uuid, + actor_id: Uuid, + id: Uuid, +) -> Result, ServerFnError> { + let mut tx = begin_access(pool, org_id, actor_id).await?; + let result = jobs::status(&mut *tx, org_id, id) + .await + .map_err(server_err)?; + tx.commit().await.map_err(server_err)?; + Ok(result) +} + +pub(super) async fn history( + pool: &PgPool, + org_id: Uuid, + actor_id: Uuid, + limit: i64, + before: Option, +) -> Result, ServerFnError> { + let mut tx = begin_access(pool, org_id, actor_id).await?; + let result = jobs::list(&mut *tx, org_id, limit, before) + .await + .map_err(server_err)?; + tx.commit().await.map_err(server_err)?; + Ok(result) +} + +async fn required_job( + connection: &mut sqlx::PgConnection, + org_id: Uuid, + id: Uuid, +) -> Result { + jobs::status(connection, org_id, id) + .await + .map_err(server_err)? + .ok_or_else(|| err(NOT_FOUND, "Import job not found")) +} + +async fn begin_access( + pool: &PgPool, + org_id: Uuid, + actor_id: Uuid, +) -> Result, ServerFnError> { + jobs::access::begin_admin_access(pool, org_id, actor_id) + .await + .map_err(|error| match error { + jobs::access::AccessError::Forbidden => { + forbidden("Active administrator access required") + } + jobs::access::AccessError::Database(error) => server_err(error), + }) +} + +#[cfg(test)] +mod tests; diff --git a/crates/horae/src/server_fns/importers/commands/tests.rs b/crates/horae/src/server_fns/importers/commands/tests.rs new file mode 100644 index 00000000..c900306e --- /dev/null +++ b/crates/horae/src/server_fns/importers/commands/tests.rs @@ -0,0 +1,692 @@ +use super::*; +use crate::server_fns::test_seed::{SeedIds, seed, wait_for_blocked}; +use std::time::Duration; + +const CSV: &[u8] = b"Date,Client,Project,Task,Hours,Email\n"; + +#[derive(Clone, Copy, Debug)] +enum Operation { + Api, + Csv, + Cancel, + Retry, + Status, + History, +} +const OPERATIONS: [Operation; 6] = [ + Operation::Api, + Operation::Csv, + Operation::Cancel, + Operation::Retry, + Operation::Status, + Operation::History, +]; + +fn payload() -> jobs::JobPayload { + jobs::JobPayload::HarvestApi { + mode: ImportMode::DryRun, + sync: SyncScope::Full, + } +} + +async fn fixture(pool: &PgPool) -> (SeedIds, Uuid) { + let ids = seed(pool, OrgRole::Admin).await; + crate::importers::harvest::credentials::store( + pool, + ids.org_id, + ids.user_id, + &"11".repeat(32), + "original", + "access", + "refresh", + None, + None, + ) + .await + .unwrap(); + let job = jobs::enqueue_csv( + pool, + ids.org_id, + ImportMode::DryRun, + CSV.to_vec(), + "existing", + JobPolicy::default(), + ) + .await + .unwrap(); + jobs::cancel(pool, ids.org_id, job).await.unwrap(); + (ids, job) +} + +async fn run( + pool: &PgPool, + org: Uuid, + actor: Uuid, + job: Uuid, + operation: Operation, +) -> Result<(), ServerFnError> { + match operation { + Operation::Api => start_api(pool, org, actor, &payload(), "new", JobPolicy::default(), 0) + .await + .map(|_| ()), + Operation::Csv => start_csv( + pool, + org, + actor, + ImportMode::DryRun, + CSV.to_vec(), + "new", + JobPolicy::default(), + ) + .await + .map(|_| ()), + Operation::Cancel => cancel(pool, org, actor, job).await.map(|_| ()), + Operation::Retry => retry(pool, org, actor, job).await.map(|_| ()), + Operation::Status => status(pool, org, actor, job).await.map(|_| ()), + Operation::History => history(pool, org, actor, 20, None).await.map(|_| ()), + } +} + +async fn snapshot(pool: &PgPool) -> serde_json::Value { + sqlx::query_scalar!( + r#"SELECT jsonb_build_array( + (SELECT jsonb_agg(to_jsonb(j) ORDER BY id) FROM horae_jobs j), + (SELECT jsonb_agg(to_jsonb(u) ORDER BY job_id) FROM horae_job_uploads u), + (SELECT jsonb_agg(to_jsonb(g) ORDER BY org_id) FROM harvest_connection_generations g) + ) AS "snapshot!""# + ) + .fetch_one(pool) + .await + .unwrap() +} + +fn assert_denied(result: Result<(), ServerFnError>, operation: Operation) { + assert!( + matches!( + result, + Err(ServerFnError::ServerError { + code: FORBIDDEN, + .. + }) + ), + "{operation:?} must deny stale authority: {result:?}" + ); +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn commands_and_status_require_current_active_tenant_administrator(pool: PgPool) { + for operation in OPERATIONS { + let (ids, job) = fixture(&pool).await; + let foreign = seed(&pool, OrgRole::Admin).await; + let before = snapshot(&pool).await; + for (role, active) in [ + (OrgRole::Member, true), + (OrgRole::Manager, true), + (OrgRole::Admin, false), + ] { + sqlx::query!( + "UPDATE users SET org_role = $2, active = $3 WHERE id = $1", + ids.user_id, + role as OrgRole, + active + ) + .execute(&pool) + .await + .unwrap(); + assert_denied( + run(&pool, ids.org_id, ids.user_id, job, operation).await, + operation, + ); + assert_eq!(snapshot(&pool).await, before); + } + for actor in [foreign.user_id, Uuid::now_v7()] { + assert_denied( + run(&pool, ids.org_id, actor, job, operation).await, + operation, + ); + assert_eq!(snapshot(&pool).await, before); + } + } +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn duplicate_submission_and_noop_cancellation_do_not_bypass_revocation(pool: PgPool) { + let (ids, job) = fixture(&pool).await; + sqlx::query!( + "UPDATE users SET org_role = 'member' WHERE id = $1", + ids.user_id + ) + .execute(&pool) + .await + .unwrap(); + let before = snapshot(&pool).await; + assert_denied( + start_csv( + &pool, + ids.org_id, + ids.user_id, + ImportMode::DryRun, + CSV.to_vec(), + "existing", + JobPolicy::default(), + ) + .await + .map(|_| ()), + Operation::Csv, + ); + for key in ["existing", "missing"] { + assert_denied( + start_csv( + &pool, + ids.org_id, + ids.user_id, + ImportMode::DryRun, + b"invalid headers".to_vec(), + key, + JobPolicy::default(), + ) + .await + .map(|_| ()), + Operation::Csv, + ); + } + assert_denied( + cancel(&pool, ids.org_id, ids.user_id, job) + .await + .map(|_| ()), + Operation::Cancel, + ); + assert_eq!(snapshot(&pool).await, before); +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn import_access_rechecks_revocation_after_organization_and_actor_waits(pool: PgPool) { + let run_pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .after_connect(|connection, _| { + Box::pin(async move { + sqlx::query!( + "SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL REPEATABLE READ" + ) + .execute(connection) + .await?; + Ok(()) + }) + }) + .connect_with((*pool.connect_options()).clone()) + .await + .unwrap(); + for operation in OPERATIONS { + for actor_only in [false, true] { + for deactivate in [false, true] { + let (ids, job) = fixture(&pool).await; + let before = snapshot(&pool).await; + let mut revoke = pool.begin().await.unwrap(); + let blocker = sqlx::query_scalar!("SELECT pg_backend_pid()") + .fetch_one(&mut *revoke) + .await + .unwrap() + .unwrap(); + if !actor_only { + sqlx::query!( + "SELECT id FROM organizations WHERE id = $1 FOR UPDATE", + ids.org_id + ) + .fetch_one(&mut *revoke) + .await + .unwrap(); + } + sqlx::query!( + "UPDATE users SET org_role = $2, active = $3 WHERE id = $1", + ids.user_id, + if deactivate { + OrgRole::Admin + } else { + OrgRole::Member + } as OrgRole, + !deactivate + ) + .execute(&mut *revoke) + .await + .unwrap(); + let db = run_pool.clone(); + let mut pending = tokio::task::JoinSet::new(); + pending + .spawn(async move { run(&db, ids.org_id, ids.user_id, job, operation).await }); + tokio::select! { + result = pending.join_next() => panic!("{operation:?} bypassed revocation wait: {result:?}"), + () = wait_for_blocked(&pool, blocker) => {}, + } + revoke.commit().await.unwrap(); + assert_denied( + tokio::time::timeout(Duration::from_secs(5), pending.join_next()) + .await + .unwrap() + .unwrap() + .unwrap(), + operation, + ); + assert_eq!(snapshot(&pool).await, before); + } + } + } + run_pool.close().await; +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn import_commands_and_results_hold_actor_authority_until_commit(pool: PgPool) { + for operation in OPERATIONS { + let (ids, job) = fixture(&pool).await; + let mut barrier = pool.begin().await.unwrap(); + let blocker = sqlx::query_scalar!("SELECT pg_backend_pid()") + .fetch_one(&mut *barrier) + .await + .unwrap() + .unwrap(); + // All command/status paths reach the job table after authorization. + sqlx::query!("LOCK TABLE horae_jobs IN ACCESS EXCLUSIVE MODE") + .execute(&mut *barrier) + .await + .unwrap(); + let db = pool.clone(); + let mut command = tokio::task::JoinSet::new(); + command.spawn(async move { run(&db, ids.org_id, ids.user_id, job, operation).await }); + wait_for_blocked(&pool, blocker).await; + let writer = sqlx::query_scalar!( + "SELECT pid AS \"pid!\" FROM pg_stat_activity WHERE datname=current_database() AND $1=ANY(pg_blocking_pids(pid))", + blocker, + ).fetch_one(&pool).await.unwrap(); + let db = pool.clone(); + let mut revoke = tokio::task::JoinSet::new(); + revoke.spawn(async move { + sqlx::query!( + "UPDATE users SET org_role = 'member' WHERE id = $1", + ids.user_id + ) + .execute(&db) + .await + }); + tokio::select! { + result = revoke.join_next() => panic!("{operation:?} released authority early: {result:?}"), + () = wait_for_blocked(&pool, writer) => {}, + } + barrier.commit().await.unwrap(); + tokio::time::timeout(Duration::from_secs(5), command.join_next()) + .await + .unwrap() + .unwrap() + .unwrap() + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), revoke.join_next()) + .await + .unwrap() + .unwrap() + .unwrap() + .unwrap(); + let before = snapshot(&pool).await; + assert_denied( + run(&pool, ids.org_id, ids.user_id, job, operation).await, + operation, + ); + assert_eq!(snapshot(&pool).await, before); + } +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn foreign_and_missing_jobs_never_reveal_or_change_another_tenant(pool: PgPool) { + let (owner, job) = fixture(&pool).await; + let foreign = seed(&pool, OrgRole::Admin).await; + let before = snapshot(&pool).await; + for id in [job, Uuid::now_v7()] { + assert!( + status(&pool, foreign.org_id, foreign.user_id, id) + .await + .unwrap() + .is_none() + ); + assert!( + history(&pool, foreign.org_id, foreign.user_id, 20, Some(id)) + .await + .unwrap() + .is_empty() + ); + for result in [ + cancel(&pool, foreign.org_id, foreign.user_id, id).await, + retry(&pool, foreign.org_id, foreign.user_id, id).await, + ] { + assert!( + matches!( + result, + Err(ServerFnError::ServerError { + code: NOT_FOUND, + .. + }) + ), + "{result:?}" + ); + } + } + for operation in OPERATIONS { + assert_denied( + run(&pool, Uuid::now_v7(), owner.user_id, job, operation).await, + operation, + ); + } + assert_eq!(snapshot(&pool).await, before); +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn authorized_job_lifecycle_and_duplicate_retention_work_with_one_connection(pool: PgPool) { + let (ids, existing) = fixture(&pool).await; + let single = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect_with((*pool.connect_options()).clone()) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + let api = start_api( + &single, + ids.org_id, + ids.user_id, + &payload(), + "api", + JobPolicy::default(), + 0, + ) + .await + .unwrap(); + assert_eq!(api.status, "queued"); + let duplicate = start_api( + &single, + ids.org_id, + ids.user_id, + &payload(), + "api", + JobPolicy::default(), + 0, + ) + .await + .unwrap(); + assert_eq!(duplicate.id, api.id); + let csv = start_csv( + &single, + ids.org_id, + ids.user_id, + ImportMode::DryRun, + CSV.to_vec(), + "csv", + JobPolicy::default(), + ) + .await + .unwrap(); + assert_eq!(csv.status, "queued"); + assert_eq!( + cancel(&single, ids.org_id, ids.user_id, csv.id) + .await + .unwrap() + .status, + "cancelled" + ); + assert_eq!( + cancel(&single, ids.org_id, ids.user_id, csv.id) + .await + .unwrap() + .status, + "cancelled" + ); + assert_eq!( + retry(&single, ids.org_id, ids.user_id, csv.id) + .await + .unwrap() + .status, + "queued" + ); + assert!(matches!( + retry(&single, ids.org_id, ids.user_id, csv.id).await, + Err(ServerFnError::ServerError { + code: NOT_FOUND, + .. + }) + )); + assert_eq!( + status(&single, ids.org_id, ids.user_id, csv.id) + .await + .unwrap() + .unwrap() + .status, + "queued" + ); + assert_eq!( + history(&single, ids.org_id, ids.user_id, 100, None) + .await + .unwrap() + .len(), + 3 + ); + sqlx::query!("DELETE FROM horae_job_uploads WHERE job_id = $1", existing) + .execute(&single) + .await + .unwrap(); + let replay = start_csv( + &single, + ids.org_id, + ids.user_id, + ImportMode::DryRun, + CSV.to_vec(), + "existing", + JobPolicy::default(), + ) + .await + .unwrap(); + assert_eq!(replay.id, existing); + assert!(matches!( + replay.retry_availability, + crate::models::RetryAvailability::MissingUpload + )); + assert!(matches!( + retry(&single, ids.org_id, ids.user_id, existing).await, + Err(ServerFnError::ServerError { + code: NOT_FOUND, + .. + }) + )); + }) + .await + .expect("authorized commands must not acquire a second connection"); + single.close().await; +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn failed_upload_insert_rolls_back_the_job_and_releases_authority(pool: PgPool) { + let (ids, _) = fixture(&pool).await; + let before = snapshot(&pool).await; + sqlx::query!( + "ALTER TABLE horae_job_uploads ADD CONSTRAINT reject_command_upload CHECK (false) NOT VALID" + ) + .execute(&pool) + .await + .unwrap(); + let failed = start_csv( + &pool, + ids.org_id, + ids.user_id, + ImportMode::DryRun, + CSV.to_vec(), + "failed", + JobPolicy::default(), + ) + .await; + assert!( + matches!( + failed, + Err(ServerFnError::ServerError { + code: INTERNAL_ERROR, + .. + }) + ), + "{failed:?}" + ); + assert_eq!(snapshot(&pool).await, before); + tokio::time::timeout(Duration::from_secs(5), async { + let mut probe = pool.begin().await.unwrap(); + sqlx::query!( + "SELECT id FROM organizations WHERE id = $1 FOR UPDATE", + ids.org_id + ) + .fetch_one(&mut *probe) + .await + .unwrap(); + sqlx::query!("SELECT id FROM users WHERE id = $1 FOR UPDATE", ids.user_id) + .fetch_one(&mut *probe) + .await + .unwrap(); + probe.rollback().await.unwrap(); + }) + .await + .expect("rollback must release authority locks"); + sqlx::query!("ALTER TABLE horae_job_uploads DROP CONSTRAINT reject_command_upload") + .execute(&pool) + .await + .unwrap(); + assert_eq!( + start_csv( + &pool, + ids.org_id, + ids.user_id, + ImportMode::DryRun, + CSV.to_vec(), + "failed", + JobPolicy::default() + ) + .await + .unwrap() + .status, + "queued" + ); +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn conflicting_and_stale_commands_preserve_payload_policy_and_retained_state(pool: PgPool) { + let (ids, csv) = fixture(&pool).await; + let api = start_api( + &pool, + ids.org_id, + ids.user_id, + &payload(), + "api", + JobPolicy::default(), + 0, + ) + .await + .unwrap(); + cancel(&pool, ids.org_id, ids.user_id, api.id) + .await + .unwrap(); + let before = snapshot(&pool).await; + let conflicting = jobs::JobPayload::HarvestApi { + mode: ImportMode::Commit, + sync: SyncScope::Full, + }; + for result in [ + start_api( + &pool, + ids.org_id, + ids.user_id, + &conflicting, + "api", + JobPolicy::default(), + 0, + ) + .await, + start_csv( + &pool, + ids.org_id, + ids.user_id, + ImportMode::Commit, + CSV.to_vec(), + "existing", + JobPolicy::default(), + ) + .await, + ] { + assert!( + matches!( + result, + Err(ServerFnError::ServerError { code: CONFLICT, .. }) + ), + "{result:?}" + ); + } + assert_eq!(snapshot(&pool).await, before); + sqlx::query!("UPDATE harvest_connection_generations SET account_generation = account_generation + 1 WHERE org_id = $1", ids.org_id) + .execute(&pool).await.unwrap(); + let before = snapshot(&pool).await; + for result in [ + start_api( + &pool, + ids.org_id, + ids.user_id, + &payload(), + "api", + JobPolicy::default(), + 0, + ) + .await, + retry(&pool, ids.org_id, ids.user_id, api.id).await, + ] { + assert!( + matches!( + result, + Err(ServerFnError::ServerError { code: CONFLICT, .. }) + ), + "{result:?}" + ); + } + assert_eq!(snapshot(&pool).await, before); + assert_eq!( + retry(&pool, ids.org_id, ids.user_id, csv) + .await + .unwrap() + .status, + "queued" + ); +} + +#[sqlx::test(migrations = "./migrations")] +#[serial_test::serial] +async fn authorized_submissions_do_not_wait_for_the_running_import_reservation(pool: PgPool) { + let (ids, _) = fixture(&pool).await; + let mut reservation = pool.acquire().await.unwrap(); + reservation.close_on_drop(); + assert!( + sqlx::query_scalar!( + r#"SELECT pg_try_advisory_lock(hashtextextended($1, 0)) as "acquired!""#, + format!("horae:harvest-import:{}", ids.org_id), + ) + .fetch_one(&mut *reservation) + .await + .unwrap() + ); + tokio::time::timeout(Duration::from_secs(5), async { + for operation in [Operation::Api, Operation::Csv] { + run(&pool, ids.org_id, ids.user_id, Uuid::now_v7(), operation) + .await + .unwrap(); + } + }) + .await + .expect("job acceptance must not acquire the worker reservation"); + sqlx::query!("SELECT pg_advisory_unlock_all()") + .execute(&mut *reservation) + .await + .unwrap(); + reservation.close().await.unwrap(); +} From 66ff076be1ddf12f1ee84eae05fe669e8c6ec05f Mon Sep 17 00:00:00 2001 From: Aldo Borrero <82811+aldoborrero@users.noreply.github.com> Date: Tue, 6 Oct 2026 19:04:20 +0200 Subject: [PATCH 2/2] Cache import command and download authorization queries --- ...2ec3760046dc2171b2bfee258f443874e46da.json | 20 +++++++++++++++++++ ...e74744b48964237f31b43e51dbb77c034be0b.json | 12 +++++++++++ ...c579a146104533ba981487fcf185afa88a833.json | 12 +++++++++++ ...27c8cd84ff7263f2d32545174a649a5a6889f.json | 12 +++++++++++ ...d07a256121a5fc8ce560814e5a4ce5d8bc8df.json | 14 +++++++++++++ ...a50ffacf1c37642d25b3f0573ef872e566893.json | 12 +++++++++++ ...c2cdf2254fb45d2f2d38ea7497ec2856842f6.json | 16 +++++++++++++++ 7 files changed, 98 insertions(+) create mode 100644 .sqlx/query-2b520c9f8437f15599775ba45322ec3760046dc2171b2bfee258f443874e46da.json create mode 100644 .sqlx/query-2d9e3195dd5416a18fbd5b41953e74744b48964237f31b43e51dbb77c034be0b.json create mode 100644 .sqlx/query-5a6283b83914a7441df61281631c579a146104533ba981487fcf185afa88a833.json create mode 100644 .sqlx/query-909406bb5877f8ac4c2db9f562f27c8cd84ff7263f2d32545174a649a5a6889f.json create mode 100644 .sqlx/query-95a7899261de6057a51077a83f9d07a256121a5fc8ce560814e5a4ce5d8bc8df.json create mode 100644 .sqlx/query-b63be505cada22baad2cc2a979ea50ffacf1c37642d25b3f0573ef872e566893.json create mode 100644 .sqlx/query-ed781289154a0bfc3e2db1b7081c2cdf2254fb45d2f2d38ea7497ec2856842f6.json diff --git a/.sqlx/query-2b520c9f8437f15599775ba45322ec3760046dc2171b2bfee258f443874e46da.json b/.sqlx/query-2b520c9f8437f15599775ba45322ec3760046dc2171b2bfee258f443874e46da.json new file mode 100644 index 00000000..a67465aa --- /dev/null +++ b/.sqlx/query-2b520c9f8437f15599775ba45322ec3760046dc2171b2bfee258f443874e46da.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT jsonb_build_array(\n (SELECT jsonb_agg(to_jsonb(j) ORDER BY id) FROM horae_jobs j),\n (SELECT jsonb_agg(to_jsonb(u) ORDER BY job_id) FROM horae_job_uploads u),\n (SELECT jsonb_agg(to_jsonb(g) ORDER BY org_id) FROM harvest_connection_generations g)\n ) AS \"snapshot!\"", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "snapshot!", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null + ] + }, + "hash": "2b520c9f8437f15599775ba45322ec3760046dc2171b2bfee258f443874e46da" +} diff --git a/.sqlx/query-2d9e3195dd5416a18fbd5b41953e74744b48964237f31b43e51dbb77c034be0b.json b/.sqlx/query-2d9e3195dd5416a18fbd5b41953e74744b48964237f31b43e51dbb77c034be0b.json new file mode 100644 index 00000000..cba33f42 --- /dev/null +++ b/.sqlx/query-2d9e3195dd5416a18fbd5b41953e74744b48964237f31b43e51dbb77c034be0b.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "LOCK TABLE horae_job_report_error_chunks IN ACCESS EXCLUSIVE MODE", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "2d9e3195dd5416a18fbd5b41953e74744b48964237f31b43e51dbb77c034be0b" +} diff --git a/.sqlx/query-5a6283b83914a7441df61281631c579a146104533ba981487fcf185afa88a833.json b/.sqlx/query-5a6283b83914a7441df61281631c579a146104533ba981487fcf185afa88a833.json new file mode 100644 index 00000000..f8597d98 --- /dev/null +++ b/.sqlx/query-5a6283b83914a7441df61281631c579a146104533ba981487fcf185afa88a833.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "LOCK TABLE horae_jobs IN ACCESS EXCLUSIVE MODE", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "5a6283b83914a7441df61281631c579a146104533ba981487fcf185afa88a833" +} diff --git a/.sqlx/query-909406bb5877f8ac4c2db9f562f27c8cd84ff7263f2d32545174a649a5a6889f.json b/.sqlx/query-909406bb5877f8ac4c2db9f562f27c8cd84ff7263f2d32545174a649a5a6889f.json new file mode 100644 index 00000000..0b9f8052 --- /dev/null +++ b/.sqlx/query-909406bb5877f8ac4c2db9f562f27c8cd84ff7263f2d32545174a649a5a6889f.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "ALTER TABLE horae_job_uploads DROP CONSTRAINT reject_command_upload", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "909406bb5877f8ac4c2db9f562f27c8cd84ff7263f2d32545174a649a5a6889f" +} diff --git a/.sqlx/query-95a7899261de6057a51077a83f9d07a256121a5fc8ce560814e5a4ce5d8bc8df.json b/.sqlx/query-95a7899261de6057a51077a83f9d07a256121a5fc8ce560814e5a4ce5d8bc8df.json new file mode 100644 index 00000000..9384f995 --- /dev/null +++ b/.sqlx/query-95a7899261de6057a51077a83f9d07a256121a5fc8ce560814e5a4ce5d8bc8df.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE horae_jobs SET checkpoint = NULL, report = '{}' WHERE id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "95a7899261de6057a51077a83f9d07a256121a5fc8ce560814e5a4ce5d8bc8df" +} diff --git a/.sqlx/query-b63be505cada22baad2cc2a979ea50ffacf1c37642d25b3f0573ef872e566893.json b/.sqlx/query-b63be505cada22baad2cc2a979ea50ffacf1c37642d25b3f0573ef872e566893.json new file mode 100644 index 00000000..9c679310 --- /dev/null +++ b/.sqlx/query-b63be505cada22baad2cc2a979ea50ffacf1c37642d25b3f0573ef872e566893.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "ALTER TABLE horae_job_uploads ADD CONSTRAINT reject_command_upload CHECK (false) NOT VALID", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "b63be505cada22baad2cc2a979ea50ffacf1c37642d25b3f0573ef872e566893" +} diff --git a/.sqlx/query-ed781289154a0bfc3e2db1b7081c2cdf2254fb45d2f2d38ea7497ec2856842f6.json b/.sqlx/query-ed781289154a0bfc3e2db1b7081c2cdf2254fb45d2f2d38ea7497ec2856842f6.json new file mode 100644 index 00000000..94678ebb --- /dev/null +++ b/.sqlx/query-ed781289154a0bfc3e2db1b7081c2cdf2254fb45d2f2d38ea7497ec2856842f6.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO users (id, org_id, email, name, org_role) VALUES ($1, $2, $3, 'Report reader', 'admin')", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Uuid", + "Text" + ] + }, + "nullable": [] + }, + "hash": "ed781289154a0bfc3e2db1b7081c2cdf2254fb45d2f2d38ea7497ec2856842f6" +}