diff --git a/backend/.sqlx/query-341491d10649cf76c6d3bce963cf03d449a85e186e4c0a10303f3a043ecb6816.json b/backend/.sqlx/query-341491d10649cf76c6d3bce963cf03d449a85e186e4c0a10303f3a043ecb6816.json deleted file mode 100644 index 35d01aa0b1..0000000000 --- a/backend/.sqlx/query-341491d10649cf76c6d3bce963cf03d449a85e186e4c0a10303f3a043ecb6816.json +++ /dev/null @@ -1,24 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT 1 as _e FROM dbt_environment_state\n WHERE workspace_id = $1 AND script_path = $2 AND environment = $3 FOR UPDATE", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "_e", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [ - "Text", - "Text", - "Text" - ] - }, - "nullable": [ - null - ] - }, - "hash": "341491d10649cf76c6d3bce963cf03d449a85e186e4c0a10303f3a043ecb6816" -} diff --git a/backend/.sqlx/query-a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247.json b/backend/.sqlx/query-a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247.json new file mode 100644 index 0000000000..909e6ad42d --- /dev/null +++ b/backend/.sqlx/query-a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT pg_advisory_xact_lock($1)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "pg_advisory_xact_lock", + "type_info": "Void" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247" +} diff --git a/backend/windmill-worker/src/dbt_state.rs b/backend/windmill-worker/src/dbt_state.rs index d07a91a351..4664dbe6e6 100644 --- a/backend/windmill-worker/src/dbt_state.rs +++ b/backend/windmill-worker/src/dbt_state.rs @@ -141,15 +141,16 @@ pub(crate) async fn publish( // which describes the same project in the same environment — versioned keys // would move that window to a pointer at an object a reader may already have // been about to fetch, and buy a grace period to sweep. + // + // An advisory lock rather than the row's, because the first publish of an + // environment has no row to lock and is exactly when two runs of a newly + // deployed script are most likely to race. let mut tx = db.begin().await?; - sqlx::query!( - "SELECT 1 as _e FROM dbt_environment_state - WHERE workspace_id = $1 AND script_path = $2 AND environment = $3 FOR UPDATE", - w_id, - &p.script_path, - environment + sqlx::query_scalar!( + "SELECT pg_advisory_xact_lock($1)", + publication_lock(w_id, &p.script_path, &environment) ) - .fetch_optional(&mut *tx) + .execute(&mut *tx) .await?; let manifest = store( manifest, @@ -300,6 +301,19 @@ fn split(home: Option) -> (Option, Option) { } } +/// The advisory lock one environment's publishers take, so that only one of them +/// is between its first upload and its row at a time. +/// +/// Derived from the same three components as the row's key. Two environments +/// whose digests collide in 64 bits wait for each other, which costs a moment and +/// nothing else. +fn publication_lock(w_id: &str, script_path: &str, environment: &str) -> i64 { + let d = digest(&format!("{w_id}|{script_path}|{environment}")); + let mut bytes = [0u8; 8]; + bytes.copy_from_slice(&d.as_bytes()[..8]); + i64::from_be_bytes(bytes) +} + /// The object-storage key an artifact takes. /// /// Derived from the row's own key rather than randomly, so a republish