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

fix(datatables): wait out live listeners, and resolve stored names containing ?

Browse Source 
Turning roles on counted a trigger as gone once disabled, and a capture
once its client stopped pinging, but the listener keeps its replication
connection until its next heartbeat notices. A trigger or capture whose
listener pinged in the last 15 seconds, the window a server holds a
listener for, now still counts as streaming.

Data table names could contain `?` before they were restricted, and such
entries are still stored. Splitting `?role=` off a reference misread them:
`a?b` became `a` with an unknown parameter, and the clone checks looked at
a different entry than the one copied. An entry stored under the whole
reference is now looked up first, in the Postgres executor, DuckDB ATTACH
and the clone checks. Agent workers cannot read the workspace and keep
the strict parse, which refuses such a name rather than misreading it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Diego Imbert
2026-09-17 10:01:15 +02:00
committed by GitHub
co-authored by Claude Opus 5
parent 122a3bd164
commit df471e37d8
7 changed files with 164 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-bc87664b87a84e1033589f729148961cfbd0171f835faa756762919c021bfd2b.json
Generated
-24
View File
@@ -1,24 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT workspace_id || '/' || path AS \"stream!\" FROM postgres_trigger\n WHERE workspace_id = $1 AND mode <> 'disabled'::TRIGGER_MODE\n AND (postgres_resource_path = $2 OR starts_with(postgres_resource_path, $3))\n UNION ALL\n SELECT workspace_id || '/' || path || ' (capture)' FROM capture_config\n WHERE workspace_id = $1 AND trigger_kind = 'postgres'\n AND last_client_ping > now() - interval '10 seconds'\n AND (trigger_config->>'postgres_resource_path' = $2\n OR starts_with(trigger_config->>'postgres_resource_path', $3))",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "stream!",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": [
null
]
},
"hash": "bc87664b87a84e1033589f729148961cfbd0171f835faa756762919c021bfd2b"
}
backend/windmill-api-integration-tests/tests/datatable_roles.rs
+53 -1
View File
@@ -786,8 +786,27 @@ async fn roles_cannot_be_turned_on_while_a_trigger_streams_the_data_table(
"the refusal does not name the trigger to disable"
);
// Disabled, but its listener pinged just now and stops only at its next heartbeat.
sqlx::query(
"UPDATE postgres_trigger SET mode = 'disabled' WHERE path = 'u/test-user-2/fork_stream'",
"UPDATE postgres_trigger SET mode = 'disabled', server_id = NULL, last_server_ping = now()
WHERE path = 'u/test-user-2/fork_stream'",
)
.execute(&db)
.await?;
let resp = authed(client().post(&url), "SECRET_TOKEN")
.json(&turn_on)
.send()
.await?;
assert_eq!(
resp.status(),
400,
"roles went on while a disabled trigger's listener was still attached: {}",
resp.text().await?
);
sqlx::query(
"UPDATE postgres_trigger SET last_server_ping = now() - interval '20 seconds'
WHERE path = 'u/test-user-2/fork_stream'",
)
.execute(&db)
.await?;
@@ -852,3 +871,36 @@ async fn roles_going_on_wait_for_a_trigger_being_enabled(db: Pool<Postgres>) ->
);
Ok(())
}
#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
async fn a_stored_name_containing_a_question_mark_resolves_as_itself(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
initialize_tracing().await;
// Names could contain `?` before they were restricted, and such an entry is still stored.
sqlx::query(
"UPDATE workspace_settings
SET datatable = jsonb_set(datatable, '{datatables,legacy?dt}', datatable->'datatables'->'main')
WHERE workspace_id = 'test-workspace'",
)
.execute(&db)
.await?;
let resolve = |reference: &'static str| {
let db = db.clone();
async move {
windmill_common::workspaces::parse_datatable_ref_for(&db, "test-workspace", reference)
.await
}
};
assert_eq!(resolve("legacy?dt").await?, ("legacy?dt".to_string(), None));
assert_eq!(
resolve("main?role=analytics").await?,
("main".to_string(), Some("analytics".to_string()))
);
assert!(
resolve("main?dt").await.is_err(),
"an unknown parameter was ignored"
);
Ok(())
}
backend/windmill-api-workspaces/src/datatable_permissions.rs
+16 -11
View File
@@ -481,23 +481,28 @@ async fn ensure_no_streams_reaching(db: &DB, governing: &GoverningDatatable) ->
for (w_id, name) in reached {
let reference = format!("datatable://{name}");
let with_query = format!("{reference}?");
// A suspended trigger keeps its listener, so only a disabled one is not streaming; a
// capture streams for as long as its client keeps pinging.
// A suspended trigger keeps its listener, and a capture streams while its client pings. A
// listener also outlives its trigger being disabled, or its capture's client going quiet,
// until its next heartbeat notices; one that pinged within the 15 seconds a server holds a
// listener for may still be dispatching.
streams.extend(
sqlx::query_scalar!(
r#"SELECT workspace_id || '/' || path AS "stream!" FROM postgres_trigger
WHERE workspace_id = $1 AND mode <> 'disabled'::TRIGGER_MODE
sqlx::query_scalar::<_, String>(
r#"SELECT workspace_id || '/' || path FROM postgres_trigger
WHERE workspace_id = $1
AND (mode <> 'disabled'::TRIGGER_MODE
OR last_server_ping > now() - interval '15 seconds')
AND (postgres_resource_path = $2 OR starts_with(postgres_resource_path, $3))
UNION ALL
SELECT workspace_id || '/' || path || ' (capture)' FROM capture_config
WHERE workspace_id = $1 AND trigger_kind = 'postgres'
AND last_client_ping > now() - interval '10 seconds'
AND (last_client_ping > now() - interval '10 seconds'
OR last_server_ping > now() - interval '15 seconds')
AND (trigger_config->>'postgres_resource_path' = $2
OR starts_with(trigger_config->>'postgres_resource_path', $3))"#,
&w_id,
&reference,
&with_query,
)
.bind(&w_id)
.bind(&reference)
.bind(&with_query)
.fetch_all(db)
.await?,
);
@@ -505,8 +510,8 @@ async fn ensure_no_streams_reaching(db: &DB, governing: &GoverningDatatable) ->
if !streams.is_empty() {
return Err(Error::BadRequest(format!(
"Data table '{}' cannot be put under roles while a Postgres trigger or capture streams \
it: a replication stream reads every row whatever the roles grant. Disable them \
first: {}",
it: a replication stream reads every row whatever the roles grant. Disable them, then \
allow their listeners up to 15 seconds to stop: {}",
governing.name,
streams.join(", ")
)));
backend/windmill-api-workspaces/src/workspaces.rs
+10 -10
View File
@@ -45,12 +45,12 @@ use windmill_common::workspaces::GitRepositorySettings;
#[cfg(feature = "enterprise")]
use windmill_common::workspaces::WorkspaceDeploymentUISettings;
use windmill_common::workspaces::{
check_deploy_rules, check_user_against_rule, datatable_ref_name,
get_datatable_resource_from_db, get_datatable_resource_from_db_unchecked,
resolve_governing_datatable, validate_dev_workspace_id, validate_fork_workspace_id,
validate_workspace_name, DataTable, DataTableCatalogResourceType, DataTableForkBehavior,
DatatableAccess, GoverningDatatable, ProtectionRuleKind, ProtectionRules, ProtectionRuleset,
RuleCheckResult, WorkspaceGitSyncSettings, DEV_WORKSPACE_LOCK_RULE_NAME,
check_deploy_rules, check_user_against_rule, get_datatable_resource_from_db,
get_datatable_resource_from_db_unchecked, parse_datatable_ref_for, resolve_governing_datatable,
validate_dev_workspace_id, validate_fork_workspace_id, validate_workspace_name, DataTable,
DataTableCatalogResourceType, DataTableForkBehavior, DatatableAccess, GoverningDatatable,
ProtectionRuleKind, ProtectionRules, ProtectionRuleset, RuleCheckResult,
WorkspaceGitSyncSettings, DEV_WORKSPACE_LOCK_RULE_NAME,
};
use windmill_common::workspaces::{Ducklake, DucklakeCatalogResourceType};
use windmill_common::PgDatabase;
@@ -3313,8 +3313,8 @@ async fn create_pg_database(
// database that no data table entry names. Refuse here too, so the clone stops before one
// exists rather than leaving an empty registered `wm_fork_…` behind.
if let Some(reference) = req.source.strip_prefix("datatable://") {
let name = datatable_ref_name(reference);
ensure_datatable_is_clonable(&db, &w_id, name).await?;
let (name, _) = parse_datatable_ref_for(&db, &w_id, reference).await?;
ensure_datatable_is_clonable(&db, &w_id, &name).await?;
}
// Non-superadmin: restrict dbname to wm_fork_ prefix
@@ -3457,8 +3457,8 @@ async fn import_pg_database(
}
if let Some(reference) = req.source.strip_prefix("datatable://") {
let name = datatable_ref_name(reference);
ensure_datatable_is_clonable(&db, &w_id, name).await?;
let (name, _) = parse_datatable_ref_for(&db, &w_id, reference).await?;
ensure_datatable_is_clonable(&db, &w_id, &name).await?;
}
if req.fork_behavior == DataTableForkBehavior::SchemaAndData {
backend/windmill-common/src/workspaces.rs
+24 -10
View File
@@ -2106,13 +2106,30 @@ pub fn strip_datatable_permissions(
Some(datatable)
}
/// The data table a `datatable://` reference names, ignoring its query string. For callers that
/// only need to find the entry; use [`parse_datatable_ref`] wherever the role is acted on.
pub fn datatable_ref_name(reference: &str) -> &str {
reference
.split_once('?')
.map(|(name, _)| name)
.unwrap_or(reference)
/// As [`parse_datatable_ref`], except that an entry whose stored name itself contains `?` — which
/// names could before they were restricted — resolves by that exact name, without a role. It is
/// looked up first, so `sales?role=x` never reaches a different entry than the one stored so.
pub async fn parse_datatable_ref_for(
db: &DB,
w_id: &str,
reference: &str,
) -> Result<(String, Option<String>)> {
if reference.contains('?') {
let exists = sqlx::query_scalar::<_, Option<bool>>(
"SELECT (datatable->'datatables') ? $2 FROM workspace_settings WHERE workspace_id = $1",
)
.bind(w_id)
.bind(reference)
.fetch_optional(db)
.await?
.flatten()
.unwrap_or(false);
if exists {
return Ok((reference.to_string(), None));
}
}
let (name, role) = parse_datatable_ref(reference)?;
Ok((name.to_string(), role.map(str::to_string)))
}
/// Split a `datatable://` reference into its name and the role its query string names.
@@ -3431,9 +3448,6 @@ mod tests {
"silently ignored: {malformed}"
);
}
// The name-only helper stays lenient — it is used where the role is never acted on.
assert_eq!(datatable_ref_name("sales?role="), "sales");
}
#[test]
backend/windmill-worker/src/duckdb_executor.rs
+53 -57
View File
@@ -2602,31 +2602,28 @@ fn fork_defer_statements(
}
struct AttachedDatatable<'a> {
name: &'a str,
role: Option<&'a str>,
/// The data table reference, query string included; a bare `datatable` is `main`.
reference: String,
alias: &'a str,
}
/// `ATTACH 'datatable[://<name>][?role=<role>]' AS <alias>`. A bare `datatable` names the default
/// data table, so the role query string has to be accepted with and without an explicit name.
fn parse_attach_datatable(query: &str) -> Result<Option<AttachedDatatable<'_>>> {
/// data table, so the role query string has to be accepted with and without an explicit name. The
/// reference is split only once the workspace can be read, because a stored name may contain `?`.
fn parse_attach_datatable(query: &str) -> Option<AttachedDatatable<'_>> {
lazy_static::lazy_static! {
static ref RE: regex::Regex = regex::Regex::new(
r"(?i)ATTACH\s*'datatable(://[^'?:]+)?(\?[^':]*)?'\s*AS\s+([^ ;]+)"
r"(?i)ATTACH\s*'datatable(://[^':]+|\?[^':]*)?'\s*AS\s+([^ ;]+)"
).unwrap();
}
let Some(cap) = RE.captures(query) else {
return Ok(None);
let cap = RE.captures(query)?;
let reference = match cap.get(1).map(|m| m.as_str()) {
Some(named) if named.starts_with("://") => named[3..].to_string(),
Some(query) => format!("main{query}"),
None => "main".to_string(),
};
let name = cap.get(1).map(|m| &m.as_str()[3..]).unwrap_or("main");
// A query string that does not parse is refused rather than dropped: attaching under the
// default role when the statement asked for another one is the failure this guards.
let role = match cap.get(2) {
Some(m) => windmill_common::workspaces::parse_datatable_ref(m.as_str())?.1,
None => None,
};
let alias = cap.get(3).map(|m| m.as_str()).unwrap_or("");
Ok(Some(AttachedDatatable { name, role, alias }))
let alias = cap.get(2).map(|m| m.as_str()).unwrap_or("");
Some(AttachedDatatable { reference, alias })
}
async fn transform_attach_datatable(
@@ -2635,27 +2632,31 @@ async fn transform_attach_datatable(
hidden_passwords: &mut Arc<Mutex<Vec<String>>>,
job: &MiniPulledJob,
) -> Result<Option<Vec<String>>> {
let Some(attached) = parse_attach_datatable(query)? else {
let Some(attached) = parse_attach_datatable(query) else {
return Ok(None);
};
// A query string that does not parse is refused rather than dropped: attaching under the
// default role when the statement asked for another one is the failure this guards.
let db_resource = match conn {
Connection::Http(client) => {
get_datatable_resource_from_agent_http(
client,
attached.name,
&job.workspace_id,
attached.role,
&job.id,
)
.await?
let (name, role) =
windmill_common::workspaces::parse_datatable_ref(&attached.reference)?;
get_datatable_resource_from_agent_http(client, name, &job.workspace_id, role, &job.id)
.await?
}
Connection::Sql(db) => {
let (name, role) = windmill_common::workspaces::parse_datatable_ref_for(
db,
&job.workspace_id,
&attached.reference,
)
.await?;
get_datatable_resource_from_db(
db,
&job.workspace_id,
attached.name,
attached.role,
&name,
role.as_deref(),
DatatableAccess::PermissionedAs {
permissioned_as: &job.permissioned_as,
email: &job.permissioned_as_email,
@@ -2792,45 +2793,40 @@ mod tests {
#[test]
fn attach_datatable_parses_name_and_role() {
let named = parse_attach_datatable("ATTACH 'datatable://sales?role=analytics' AS dt")
.unwrap()
.unwrap();
let reference_of = |q: &str| parse_attach_datatable(q).unwrap().reference;
let named =
parse_attach_datatable("ATTACH 'datatable://sales?role=analytics' AS dt").unwrap();
assert_eq!(
(named.name, named.role, named.alias),
("sales", Some("analytics"), "dt")
(named.reference.as_str(), named.alias),
("sales?role=analytics", "dt")
);
// A bare `datatable` is the default one, and still takes a role.
let default = parse_attach_datatable("ATTACH 'datatable?role=analytics' AS dt")
.unwrap()
.unwrap();
assert_eq!((default.name, default.role), ("main", Some("analytics")));
let no_role = parse_attach_datatable("ATTACH 'datatable://sales' AS dt")
.unwrap()
.unwrap();
assert_eq!((no_role.name, no_role.role), ("sales", None));
let bare = parse_attach_datatable("ATTACH 'datatable' AS dt")
.unwrap()
.unwrap();
assert_eq!((bare.name, bare.role), ("main", None));
assert!(parse_attach_datatable("SELECT 1").unwrap().is_none());
assert_eq!(
reference_of("ATTACH 'datatable?role=analytics' AS dt"),
"main?role=analytics"
);
assert_eq!(reference_of("ATTACH 'datatable://sales' AS dt"), "sales");
assert_eq!(reference_of("ATTACH 'datatable' AS dt"), "main");
assert!(parse_attach_datatable("SELECT 1").is_none());
// A stored name can contain `?`, so that is left to the workspace lookup to split.
assert_eq!(reference_of("ATTACH 'datatable://a?b' AS dt"), "a?b");
// The key matches case-insensitively, as the `-- role` annotation does.
let cased = parse_attach_datatable("ATTACH 'datatable://sales?Role=analytics' AS dt")
.unwrap()
.unwrap();
assert_eq!(cased.role, Some("analytics"));
// A query string that does not parse is refused rather than attached under the default
// role: the statement asked for a specific one.
// The key matches case-insensitively, as the `-- role` annotation does, and a query string
// that does not parse is refused rather than attached under the default role.
let parse = |q: &str| {
windmill_common::workspaces::parse_datatable_ref(&reference_of(q))
.map(|(name, role)| (name.to_string(), role.map(str::to_string)))
};
assert_eq!(
parse("ATTACH 'datatable://sales?Role=analytics' AS dt").unwrap(),
("sales".to_string(), Some("analytics".to_string()))
);
for malformed in [
"ATTACH 'datatable://sales?role=' AS dt",
"ATTACH 'datatable://sales?role=an;alytics' AS dt",
"ATTACH 'datatable://sales?x=1&role=analytics' AS dt",
] {
assert!(
parse_attach_datatable(malformed).is_err(),
"silently ignored: {malformed}"
);
assert!(parse(malformed).is_err(), "silently ignored: {malformed}");
}
}
backend/windmill-worker/src/pg_executor.rs
+8 -7
View File
@@ -29,7 +29,7 @@ use windmill_common::worker::{
to_raw_value, Connection, SqlAnnotations, SqlResultCollectionStrategy, CLOUD_HOSTED,
};
use windmill_common::workspaces::{
get_datatable_resource_from_db, parse_datatable_ref, DatatableAccess,
get_datatable_resource_from_db, parse_datatable_ref, parse_datatable_ref_for, DatatableAccess,
};
use windmill_common::{PgDatabase, PrepareQueryColumnInfo, PrepareQueryResult, DB};
use windmill_parser::{Arg, Typ};
@@ -683,28 +683,29 @@ pub async fn do_postgresql(
match pg_args.get("database").cloned() {
Some(Value::String(db_str)) if db_str.starts_with("datatable://") => {
let reference = db_str.trim_start_matches("datatable://");
let (db_str, uri_role) = parse_datatable_ref(reference)?;
// The annotation wins: a generated query can carry a `?role=` in the reference it
// was handed, but only the script's author writes the leading comment block.
let annotated = SqlAnnotations::datatable_role(&query)?;
let role = annotated.as_deref().or(uri_role);
Some(match conn {
Connection::Http(client) => {
let (name, uri_role) = parse_datatable_ref(reference)?;
get_datatable_resource_from_agent_http(
client,
db_str,
name,
&job.workspace_id,
role,
annotated.as_deref().or(uri_role),
&job.id,
)
.await?
}
Connection::Sql(db) => {
let (name, uri_role) =
parse_datatable_ref_for(db, &job.workspace_id, reference).await?;
get_datatable_resource_from_db(
db,
&job.workspace_id,
db_str,
role,
&name,
annotated.as_deref().or(uri_role.as_deref()),
DatatableAccess::PermissionedAs {
permissioned_as: &job.permissioned_as,
email: &job.permissioned_as_email,