starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(backend): gate asset producer-change event on write-set changes (#9672)

Browse Source 
* perf(backend): only emit asset producer-change event on write-set changes

Data Pipelines (#9193) made every script deploy emit a
`notify_asset_producer_change` event: `clear_static_asset_usage` inserted
into `notify_event` unconditionally on every clear, and the per-asset
insert path emitted nothing. So a plain script with no assets — the
overwhelming majority of deploys — wrote a `notify_event` row that made
every worker drop its `ASSET_PRODUCER_WRITES_CACHE` entry instance-wide,
needlessly thrashing the cache the feature added.

That cache only tracks script rows with write access (`usage_access_type
IN ('w','rw')`), so a deploy changes it only when the script gains or
loses a write producer. Gate the event on exactly that:

- `clear_static_asset_usage` / `clear_static_asset_usage_by_script_hash`
  emit only when the delete removed a 'w'/'rw' row (via `RETURNING`).
- `insert_static_asset_usage` emits only when it actually inserts a
  'w'/'rw' script row (no-op `ON CONFLICT`, read-only, and flow usage
  stay silent).

Plain, read-only, and flow deploys now emit nothing; producer-changing
deploys still invalidate, atomically and visible-only-on-commit as
before. Adds a test asserting the emit/no-emit matrix.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* perf(backend): dedup producer-change notify on write-asset redeploys

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(backend): derive replace write-set from persisted rows; document auth contract

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* docs(backend): correct replace_static_asset_usage call-site comment

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-06-19 15:49:32 +02:00
committed by GitHub
parent 6833a554ae
commit 5970510cf6
9 changed files with 759 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-7300eb89029e0863241087fc10df7616db640054e804d4ca66958cad06008fdb.json → backend/.sqlx/query-1acfeed9c7a5b1e3d2da262d338655dba6e43067a9912cc2b775830856390c5d.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "WITH del AS (\n DELETE FROM asset WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3\n )\n INSERT INTO notify_event (channel, payload)\n SELECT 'notify_asset_producer_change', $1 WHERE $3 = 'script'",
"query": "WITH del AS (\n DELETE FROM asset WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3\n RETURNING usage_access_type\n )\n INSERT INTO notify_event (channel, payload)\n SELECT 'notify_asset_producer_change', $1\n WHERE $3 = 'script'\n AND EXISTS (SELECT 1 FROM del WHERE usage_access_type IN ('w', 'rw'))",
"describe": {
"columns": [],
"parameters": {
@@ -23,5 +23,5 @@
},
"nullable": []
},
"hash": "7300eb89029e0863241087fc10df7616db640054e804d4ca66958cad06008fdb"
"hash": "1acfeed9c7a5b1e3d2da262d338655dba6e43067a9912cc2b775830856390c5d"
}
backend/.sqlx/query-6a1998cb3a9a0898c0fd21c35cc41e309d7e066ee4869880c90e90dd66dc9d4d.json → backend/.sqlx/query-1f375b37ff9f6f01972e284e84a7b2f9d2d323a3da55f20ff6671e8eba510043.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "WITH del AS (\n DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'script'\n AND usage_path = (SELECT path FROM script WHERE hash = $2 AND workspace_id = $1)\n )\n INSERT INTO notify_event (channel, payload)\n VALUES ('notify_asset_producer_change', $1)",
"query": "WITH del AS (\n DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'script'\n AND usage_path = (SELECT path FROM script WHERE hash = $2 AND workspace_id = $1)\n RETURNING usage_access_type\n )\n INSERT INTO notify_event (channel, payload)\n SELECT 'notify_asset_producer_change', $1\n WHERE EXISTS (SELECT 1 FROM del WHERE usage_access_type IN ('w', 'rw'))",
"describe": {
"columns": [],
"parameters": {
@@ -11,5 +11,5 @@
},
"nullable": []
},
"hash": "6a1998cb3a9a0898c0fd21c35cc41e309d7e066ee4869880c90e90dd66dc9d4d"
"hash": "1f375b37ff9f6f01972e284e84a7b2f9d2d323a3da55f20ff6671e8eba510043"
}
backend/.sqlx/query-ab18d8765d6795eaa8035a8ac902790ff61549c5dd76e5b5cfb14b110a98abf2.json
Generated
+60
View File
@@ -0,0 +1,60 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM asset\n WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = 'script'\n RETURNING kind AS \"kind!: AssetKind\", path,\n usage_access_type AS \"usage_access_type: AssetUsageAccessType\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "kind!: AssetKind",
"type_info": {
"Custom": {
"name": "asset_kind",
"kind": {
"Enum": [
"s3object",
"resource",
"variable",
"ducklake",
"datatable",
"volume"
]
}
}
}
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "usage_access_type: AssetUsageAccessType",
"type_info": {
"Custom": {
"name": "asset_access_type",
"kind": {
"Enum": [
"r",
"w",
"rw"
]
}
}
}
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false,
false,
true
]
},
"hash": "ab18d8765d6795eaa8035a8ac902790ff61549c5dd76e5b5cfb14b110a98abf2"
}
backend/.sqlx/query-bfb97d2f48157a1575b7b6f3e64e0d075701d541a44dd7a0b586d1f337ced5e1.json
Generated
+63
View File
@@ -0,0 +1,63 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns)\n VALUES ($1, $2, $3, $4, $5, 'script', $6) ON CONFLICT DO NOTHING\n RETURNING usage_access_type AS \"usage_access_type: AssetUsageAccessType\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "usage_access_type: AssetUsageAccessType",
"type_info": {
"Custom": {
"name": "asset_access_type",
"kind": {
"Enum": [
"r",
"w",
"rw"
]
}
}
}
}
],
"parameters": {
"Left": [
"Varchar",
"Varchar",
{
"Custom": {
"name": "asset_kind",
"kind": {
"Enum": [
"s3object",
"resource",
"variable",
"ducklake",
"datatable",
"volume"
]
}
}
},
{
"Custom": {
"name": "asset_access_type",
"kind": {
"Enum": [
"r",
"w",
"rw"
]
}
}
},
"Varchar",
"Jsonb"
]
},
"nullable": [
true
]
},
"hash": "bfb97d2f48157a1575b7b6f3e64e0d075701d541a44dd7a0b586d1f337ced5e1"
}
backend/.sqlx/query-a9e29764b5b9d94269e2b8aa755c71b61774c8ff8ae218d7a8d6ed0ac0169366.json → backend/.sqlx/query-e041b20c1c4166b30ced8c6a0f50bcf0691ab2554ead6b489a8441a32fc0af82.json
Generated
+2 -2
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns)\n VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING",
"query": "WITH ins AS (\n INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns)\n VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING\n RETURNING usage_kind, usage_access_type\n )\n INSERT INTO notify_event (channel, payload)\n SELECT 'notify_asset_producer_change', $1\n FROM ins WHERE usage_kind = 'script' AND usage_access_type IN ('w', 'rw')",
"describe": {
"columns": [],
"parameters": {
@@ -52,5 +52,5 @@
},
"nullable": []
},
"hash": "a9e29764b5b9d94269e2b8aa755c71b61774c8ff8ae218d7a8d6ed0ac0169366"
"hash": "e041b20c1c4166b30ced8c6a0f50bcf0691ab2554ead6b489a8441a32fc0af82"
}
backend/.sqlx/query-eeaad5c2284c1856cfc64ae0bc0dfc79b283c06987a6de399b6c1aaca94a2b7a.json
Generated
+14
View File
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO notify_event (channel, payload)\n VALUES ('notify_asset_producer_change', $1)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text"
]
},
"nullable": []
},
"hash": "eeaad5c2284c1856cfc64ae0bc0dfc79b283c06987a6de399b6c1aaca94a2b7a"
}
backend/windmill-api-scripts/src/scripts.rs
+12 -7
View File
@@ -45,8 +45,8 @@ use windmill_dep_map::scoped_dependency_map::ScopedDependencyMap;
use windmill_common::{
assets::{
clear_script_triggers, clear_static_asset_usage, clear_static_asset_usage_by_script_hash,
insert_script_trigger, insert_static_asset_usage, parse_duration_secs,
parse_pipeline_annotations, trigger_spec_to_row, AssetUsageKind, TriggerSpec,
insert_script_trigger, parse_duration_secs, parse_pipeline_annotations,
replace_static_asset_usage, trigger_spec_to_row, AssetUsageKind, TriggerSpec,
},
error::{self, to_anyhow},
min_version::{MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2},
@@ -1609,11 +1609,16 @@ async fn create_script_internal<'c>(
);
}
clear_static_asset_usage(&mut *tx, &w_id, &script_path, AssetUsageKind::Script).await?;
for asset in effective_assets.as_ref().into_iter().flatten() {
insert_static_asset_usage(&mut *tx, &w_id, &asset, &ns.path, AssetUsageKind::Script)
.await?;
}
// Clear + reinsert this script's producer rows at script_path (== ns.path),
// invalidating the producer-writes cache once iff the write-producer set
// changed (see replace_static_asset_usage).
replace_static_asset_usage(
&mut tx,
&w_id,
&script_path,
effective_assets.as_deref().unwrap_or(&[]),
)
.await?;
// Pipeline trigger edges: wipe-and-reinsert per deploy so removing an
// `// on ...` annotation drops the edge. Only Asset / Schedule produce
backend/windmill-common/src/assets.rs
+130 -15
View File
@@ -1,4 +1,6 @@
use sqlx::PgExecutor;
use std::collections::HashSet;
use sqlx::{PgExecutor, Postgres, Transaction};
use crate::{error, scripts::ScriptHash};
@@ -33,9 +35,22 @@ pub async fn insert_static_asset_usage<'e>(
.as_ref()
.map(|cols| serde_json::to_value(cols).unwrap_or(serde_json::Value::Null));
// Invalidate the per-workspace producer-writes cache only when this insert
// actually adds a write producer: the cache (asset_dispatch::
// ASSET_PRODUCER_WRITES_CACHE) tracks script rows with 'w'/'rw' access, so
// a row that was a no-op (ON CONFLICT skipped), a flow usage, or read-only
// can't change it. Emitting in the same statement keeps the notify atomic
// with the insert and visible to pollers only on commit. See the matching
// delete-side guard in clear_static_asset_usage.
sqlx::query!(
r#"INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns)
VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING"#,
r#"WITH ins AS (
INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns)
VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT DO NOTHING
RETURNING usage_kind, usage_access_type
)
INSERT INTO notify_event (channel, payload)
SELECT 'notify_asset_producer_change', $1
FROM ins WHERE usage_kind = 'script' AND usage_access_type IN ('w', 'rw')"#,
workspace_id,
asset.path,
asset.kind as AssetKind,
@@ -58,19 +73,23 @@ pub async fn clear_static_asset_usage<'e>(
) -> error::Result<()> {
// Invalidate the per-workspace producer-writes cache that gates the
// asset-trigger dispatch hook (windmill-queue
// asset_dispatch::ASSET_PRODUCER_WRITES_CACHE), but only for script
// usage. The notify is unconditional (not gated on rows deleted) so a
// deploy where the script *gains* its first asset still invalidates;
// emitting it in the same statement keeps it atomic with the change and
// visible to pollers only on commit. Flow usage doesn't affect the
// script producer set, so the INSERT is skipped via the `$3 = 'script'`
// guard.
// asset_dispatch::ASSET_PRODUCER_WRITES_CACHE). That cache only tracks
// script rows with 'w'/'rw' access, so emit the notify only when the
// delete actually removed such a write producer: flow usage, read-only
// usage, and deletes that matched no producer row leave the cache
// unchanged (the common case — most deploys touch no write asset). The
// matching add side lives in insert_static_asset_usage. Emitting in the
// same statement keeps the notify atomic with the delete and visible to
// pollers only on commit.
sqlx::query!(
r#"WITH del AS (
DELETE FROM asset WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = $3
RETURNING usage_access_type
)
INSERT INTO notify_event (channel, payload)
SELECT 'notify_asset_producer_change', $1 WHERE $3 = 'script'"#,
SELECT 'notify_asset_producer_change', $1
WHERE $3 = 'script'
AND EXISTS (SELECT 1 FROM del WHERE usage_access_type IN ('w', 'rw'))"#,
workspace_id,
usage_path,
usage_kind as AssetUsageKind
@@ -85,16 +104,18 @@ pub async fn clear_static_asset_usage_by_script_hash<'e>(
workspace_id: &str,
script_hash: ScriptHash,
) -> error::Result<()> {
// Always script usage → unconditionally invalidate the producer-writes
// cache for this workspace (see clear_static_asset_usage), atomically
// with the delete.
// Always script usage → invalidate the producer-writes cache for this
// workspace, but only when the delete actually removed a write producer
// ('w'/'rw'); see clear_static_asset_usage. Atomic with the delete.
sqlx::query!(
r#"WITH del AS (
DELETE FROM asset WHERE workspace_id = $1 AND usage_kind = 'script'
AND usage_path = (SELECT path FROM script WHERE hash = $2 AND workspace_id = $1)
RETURNING usage_access_type
)
INSERT INTO notify_event (channel, payload)
VALUES ('notify_asset_producer_change', $1)"#,
SELECT 'notify_asset_producer_change', $1
WHERE EXISTS (SELECT 1 FROM del WHERE usage_access_type IN ('w', 'rw'))"#,
workspace_id,
script_hash.0
)
@@ -103,6 +124,100 @@ pub async fn clear_static_asset_usage_by_script_hash<'e>(
Ok(())
}
fn is_write_access(access: Option<AssetUsageAccessType>) -> bool {
matches!(
access,
Some(AssetUsageAccessType::W) | Some(AssetUsageAccessType::RW)
)
}
/// Clear and reinsert the full static-asset usage set of a script in one tx,
/// invalidating the producer-writes cache at most once and only on a real
/// change. The cache (asset_dispatch::ASSET_PRODUCER_WRITES_CACHE) keys a
/// workspace by the set of (kind, path) script rows with 'w'/'rw' access, so a
/// redeploy that keeps the same write producers must leave it untouched. We
/// diff the old write set (captured from the clearing DELETE's RETURNING)
/// against the new one and emit a single notify only when they differ —
/// emitting per statement, as clear/insert_static_asset_usage do, would fire
/// twice on every write-asset redeploy (the clear removes the row, the reinsert
/// adds it back). The notify rides the deploy tx, so it stays atomic with the
/// DML and visible to pollers only on commit.
///
/// DELETE and INSERT of the same primary key cannot share a single statement
/// (both would read the pre-statement snapshot, so the reinsert's ON CONFLICT
/// would silently drop the row), which is why this clears and reinserts as
/// separate statements rather than one CTE.
///
/// Performs no authorization itself: `tx` must already be scoped to a caller
/// authorized for `workspace_id`/`usage_path` (e.g. `user_db.begin(&authed)`,
/// which applies RLS), exactly like the sibling clear/insert helpers.
pub async fn replace_static_asset_usage(
tx: &mut Transaction<'_, Postgres>,
workspace_id: &str,
usage_path: &str,
assets: &[AssetWithAltAccessType],
) -> error::Result<()> {
let cleared = sqlx::query!(
r#"DELETE FROM asset
WHERE workspace_id = $1 AND usage_path = $2 AND usage_kind = 'script'
RETURNING kind AS "kind!: AssetKind", path,
usage_access_type AS "usage_access_type: AssetUsageAccessType""#,
workspace_id,
usage_path,
)
.fetch_all(&mut **tx)
.await?;
let old_writes: HashSet<(AssetKind, String)> = cleared
.into_iter()
.filter(|r| is_write_access(r.usage_access_type))
.map(|r| (r.kind, r.path))
.collect();
// Build new_writes from rows actually inserted (RETURNING), not the
// requested slice: ON CONFLICT DO NOTHING is first-writer-wins, so a payload
// with duplicate (kind, path) entries at conflicting access types persists
// only the first. Since the DELETE above emptied this usage_path, every
// non-conflicting insert lands, so the inserted rows are exactly the new
// persisted set.
let mut new_writes: HashSet<(AssetKind, String)> = HashSet::new();
for asset in assets {
let access = asset.access_type.or(asset.alt_access_type);
let columns_json = asset
.columns
.as_ref()
.map(|cols| serde_json::to_value(cols).unwrap_or(serde_json::Value::Null));
let inserted = sqlx::query!(
r#"INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind, columns)
VALUES ($1, $2, $3, $4, $5, 'script', $6) ON CONFLICT DO NOTHING
RETURNING usage_access_type AS "usage_access_type: AssetUsageAccessType""#,
workspace_id,
asset.path,
asset.kind as AssetKind,
access as Option<AssetUsageAccessType>,
usage_path,
columns_json as Option<serde_json::Value>,
)
.fetch_optional(&mut **tx)
.await?;
if let Some(row) = inserted {
if is_write_access(row.usage_access_type) {
new_writes.insert((asset.kind, asset.path.clone()));
}
}
}
if old_writes != new_writes {
sqlx::query!(
r#"INSERT INTO notify_event (channel, payload)
VALUES ('notify_asset_producer_change', $1)"#,
workspace_id,
)
.execute(&mut **tx)
.await?;
}
Ok(())
}
// Wipe all pipeline trigger declarations held by the given runnable. Used at
// deploy time: redeploying a script wipes its prior `// on` annotations so
// removing them implicitly un-declares those edges.
backend/windmill-common/tests/asset_producer_notify.rs
+474
View File
@@ -0,0 +1,474 @@
/*!
* Tests that the asset producer-writes cache invalidation
* (`notify_asset_producer_change`) is emitted only when a deploy actually
* changes the set of script write-producers ('w'/'rw' asset usage), not on
* every deploy. The cache (asset_dispatch::ASSET_PRODUCER_WRITES_CACHE) only
* tracks script rows with write access, so read-only usage, flow usage, and
* deploys touching no write asset must NOT emit an event.
*/
use sqlx::{Pool, Postgres};
use windmill_common::assets::{
clear_static_asset_usage, clear_static_asset_usage_by_script_hash, insert_static_asset_usage,
replace_static_asset_usage, AssetKind, AssetUsageAccessType, AssetUsageKind,
AssetWithAltAccessType,
};
use windmill_common::scripts::ScriptHash;
const WS: &str = "test-workspace";
/// Run the deploy-time clear+reinsert against `usage_path` in its own tx,
/// mirroring how create_script_internal calls it.
async fn replace(db: &Pool<Postgres>, usage_path: &str, assets: &[AssetWithAltAccessType]) {
let mut tx = db.begin().await.expect("begin");
replace_static_asset_usage(&mut tx, WS, usage_path, assets)
.await
.expect("replace static asset usage");
tx.commit().await.expect("commit");
}
fn asset_at(
path: &str,
kind: AssetKind,
access: Option<AssetUsageAccessType>,
) -> AssetWithAltAccessType {
AssetWithAltAccessType {
path: path.to_string(),
kind,
access_type: access,
alt_access_type: None,
columns: None,
}
}
async fn producer_notify_count(db: &Pool<Postgres>) -> i64 {
sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*) FROM notify_event WHERE channel = 'notify_asset_producer_change'",
)
.fetch_one(db)
.await
.expect("count notify events")
}
async fn reset_notify(db: &Pool<Postgres>) {
sqlx::query("DELETE FROM notify_event WHERE channel = 'notify_asset_producer_change'")
.execute(db)
.await
.expect("reset notify events");
}
fn asset(access: Option<AssetUsageAccessType>) -> AssetWithAltAccessType {
AssetWithAltAccessType {
path: "u/test-user/res".to_string(),
kind: AssetKind::Resource,
access_type: access,
alt_access_type: None,
columns: None,
}
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn insert_write_script_asset_emits(db: Pool<Postgres>) {
insert_static_asset_usage(
&db,
WS,
&asset(Some(AssetUsageAccessType::W)),
"u/test-user/script",
AssetUsageKind::Script,
)
.await
.unwrap();
assert_eq!(producer_notify_count(&db).await, 1, "write asset must emit");
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn insert_readonly_script_asset_does_not_emit(db: Pool<Postgres>) {
insert_static_asset_usage(
&db,
WS,
&asset(Some(AssetUsageAccessType::R)),
"u/test-user/script",
AssetUsageKind::Script,
)
.await
.unwrap();
assert_eq!(
producer_notify_count(&db).await,
0,
"read-only asset is not a write producer"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn insert_write_flow_asset_does_not_emit(db: Pool<Postgres>) {
insert_static_asset_usage(
&db,
WS,
&asset(Some(AssetUsageAccessType::W)),
"u/test-user/flow",
AssetUsageKind::Flow,
)
.await
.unwrap();
assert_eq!(
producer_notify_count(&db).await,
0,
"flow usage does not affect the script producer cache"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn clear_plain_script_does_not_emit(db: Pool<Postgres>) {
// No asset rows for this path: a plain script deploy must not emit.
clear_static_asset_usage(&db, WS, "u/test-user/plain", AssetUsageKind::Script)
.await
.unwrap();
assert_eq!(
producer_notify_count(&db).await,
0,
"plain script deploy must not emit"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn clear_removes_write_producer_emits(db: Pool<Postgres>) {
insert_static_asset_usage(
&db,
WS,
&asset(Some(AssetUsageAccessType::RW)),
"u/test-user/script",
AssetUsageKind::Script,
)
.await
.unwrap();
reset_notify(&db).await; // ignore the insert-side event; isolate the clear
clear_static_asset_usage(&db, WS, "u/test-user/script", AssetUsageKind::Script)
.await
.unwrap();
assert_eq!(
producer_notify_count(&db).await,
1,
"removing a write producer must emit"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn clear_removes_only_readonly_does_not_emit(db: Pool<Postgres>) {
insert_static_asset_usage(
&db,
WS,
&asset(Some(AssetUsageAccessType::R)),
"u/test-user/script",
AssetUsageKind::Script,
)
.await
.unwrap();
reset_notify(&db).await;
clear_static_asset_usage(&db, WS, "u/test-user/script", AssetUsageKind::Script)
.await
.unwrap();
assert_eq!(
producer_notify_count(&db).await,
0,
"removing only read-only usage must not emit"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn clear_flow_usage_does_not_emit(db: Pool<Postgres>) {
insert_static_asset_usage(
&db,
WS,
&asset(Some(AssetUsageAccessType::W)),
"u/test-user/flow",
AssetUsageKind::Flow,
)
.await
.unwrap();
reset_notify(&db).await;
clear_static_asset_usage(&db, WS, "u/test-user/flow", AssetUsageKind::Flow)
.await
.unwrap();
assert_eq!(
producer_notify_count(&db).await,
0,
"clearing flow usage must not emit"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn clear_by_script_hash_gated_on_write(db: Pool<Postgres>) {
let hash = 123456789_i64;
sqlx::query(
r#"INSERT INTO script (workspace_id, hash, path, summary, description, content,
created_by, language, lock)
VALUES ($1, $2, 'u/test-user/byhash', '', '', '', 'test-user', 'deno', '')"#,
)
.bind(WS)
.bind(hash)
.execute(&db)
.await
.unwrap();
// Read-only usage on that script path → clear must not emit.
insert_static_asset_usage(
&db,
WS,
&asset(Some(AssetUsageAccessType::R)),
"u/test-user/byhash",
AssetUsageKind::Script,
)
.await
.unwrap();
reset_notify(&db).await;
clear_static_asset_usage_by_script_hash(&db, WS, ScriptHash(hash))
.await
.unwrap();
assert_eq!(
producer_notify_count(&db).await,
0,
"by-hash clear of read-only usage must not emit"
);
// Now a write usage → by-hash clear must emit.
insert_static_asset_usage(
&db,
WS,
&asset(Some(AssetUsageAccessType::W)),
"u/test-user/byhash",
AssetUsageKind::Script,
)
.await
.unwrap();
reset_notify(&db).await;
clear_static_asset_usage_by_script_hash(&db, WS, ScriptHash(hash))
.await
.unwrap();
assert_eq!(
producer_notify_count(&db).await,
1,
"by-hash clear of write usage must emit"
);
}
const SP: &str = "u/test-user/script";
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn replace_plain_deploy_does_not_emit(db: Pool<Postgres>) {
replace(&db, SP, &[]).await;
assert_eq!(
producer_notify_count(&db).await,
0,
"deploy with no assets must not emit"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn replace_gaining_write_emits_once(db: Pool<Postgres>) {
replace(
&db,
SP,
&[asset_at(
"u/test-user/res",
AssetKind::Resource,
Some(AssetUsageAccessType::W),
)],
)
.await;
assert_eq!(
producer_notify_count(&db).await,
1,
"gaining a write producer must emit exactly once"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn replace_unchanged_write_set_does_not_emit(db: Pool<Postgres>) {
let assets = [asset_at(
"u/test-user/res",
AssetKind::Resource,
Some(AssetUsageAccessType::W),
)];
replace(&db, SP, &assets).await;
reset_notify(&db).await; // isolate the redeploy
// Redeploy with the identical write-producer set: clear removes the row and
// the reinsert adds it back, but the cache value is unchanged → no emit.
replace(&db, SP, &assets).await;
assert_eq!(
producer_notify_count(&db).await,
0,
"redeploy keeping the same write producers must not emit"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn replace_dropping_write_emits(db: Pool<Postgres>) {
let assets = [asset_at(
"u/test-user/res",
AssetKind::Resource,
Some(AssetUsageAccessType::RW),
)];
replace(&db, SP, &assets).await;
reset_notify(&db).await;
// Redeploy with no assets: the write producer disappears → emit.
replace(&db, SP, &[]).await;
assert_eq!(
producer_notify_count(&db).await,
1,
"dropping the last write producer must emit"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn replace_readonly_to_write_emits(db: Pool<Postgres>) {
replace(
&db,
SP,
&[asset_at(
"u/test-user/res",
AssetKind::Resource,
Some(AssetUsageAccessType::R),
)],
)
.await;
reset_notify(&db).await;
// Same asset path, access flips read-only → write: cache gains a row → emit.
replace(
&db,
SP,
&[asset_at(
"u/test-user/res",
AssetKind::Resource,
Some(AssetUsageAccessType::W),
)],
)
.await;
assert_eq!(
producer_notify_count(&db).await,
1,
"read-only → write transition must emit"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn replace_duplicate_entries_use_persisted_state(db: Pool<Postgres>) {
replace(
&db,
SP,
&[asset_at(
"u/test-user/res",
AssetKind::Resource,
Some(AssetUsageAccessType::W),
)],
)
.await;
reset_notify(&db).await;
// Redeploy with conflicting duplicates for the same (kind, path), read-only
// first: ON CONFLICT DO NOTHING persists the read-only row and drops the
// write one, so the write producer is really gone → must emit (the diff must
// follow the persisted state, not the requested slice).
replace(
&db,
SP,
&[
asset_at(
"u/test-user/res",
AssetKind::Resource,
Some(AssetUsageAccessType::R),
),
asset_at(
"u/test-user/res",
AssetKind::Resource,
Some(AssetUsageAccessType::W),
),
],
)
.await;
assert_eq!(
producer_notify_count(&db).await,
1,
"duplicate entries losing the persisted write producer must emit"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn replace_duplicate_entries_keeping_write_does_not_emit(db: Pool<Postgres>) {
replace(
&db,
SP,
&[asset_at(
"u/test-user/res",
AssetKind::Resource,
Some(AssetUsageAccessType::W),
)],
)
.await;
reset_notify(&db).await;
// Same duplicates but write first: the write row persists, so the write set
// is unchanged → no emit.
replace(
&db,
SP,
&[
asset_at(
"u/test-user/res",
AssetKind::Resource,
Some(AssetUsageAccessType::W),
),
asset_at(
"u/test-user/res",
AssetKind::Resource,
Some(AssetUsageAccessType::R),
),
],
)
.await;
assert_eq!(
producer_notify_count(&db).await,
0,
"duplicate entries keeping the persisted write producer must not emit"
);
}
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn replace_changing_only_readonly_does_not_emit(db: Pool<Postgres>) {
replace(
&db,
SP,
&[asset_at(
"u/test-user/res",
AssetKind::Resource,
Some(AssetUsageAccessType::R),
)],
)
.await;
reset_notify(&db).await;
// Swap one read-only producer for another: no write producer either side, so
// the write-set cache is untouched → no emit.
replace(
&db,
SP,
&[asset_at(
"u/test-user/other",
AssetKind::Resource,
Some(AssetUsageAccessType::R),
)],
)
.await;
assert_eq!(
producer_notify_count(&db).await,
0,
"changes confined to read-only producers must not emit"
);
}