Fix SQL quoting

Cargo.toml

@@ -34,6 +34,7 @@ serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0", features = ["preserve_order", "raw_value"] }
 uuid = { version = "1.0", features = ["v4", "serde"] }
 percent-encoding = "2.3"
+postgres-protocol = "0.6"
 
 # duroxide integration
 duroxide = "=0.1.30"

scripts/test_upgrade_lifecycle.py

@@ -44,6 +44,22 @@ def case(shape="seq", waiting=False):
     }
 
 
+class LiteralQuotingTests(unittest.TestCase):
+    def test_quotes_apostrophes_and_backslashes(self):
+        for value, expected in [
+            ("", "''"),
+            ("plain", "'plain'"),
+            ("O'Reilly", "'O''Reilly'"),
+            ("\\", "E'\\\\'"),
+            ("\\' OR TRUE --", "E'\\\\'' OR TRUE --'"),
+            (r"\n\x41\u0041", r"E'\\n\\x41\\u0041'"),
+            ("line one\nline two", "'line one\nline two'"),
+            ("\u00e9\u4e2d\\'", "E'\u00e9\u4e2d\\\\'''"),
+        ]:
+            with self.subTest(value=value):
+                self.assertEqual(lifecycle.literal(value), expected)
+
+
 class ValidationTests(unittest.TestCase):
     def test_every_shape_passes_before_and_after(self):
         for shape in lifecycle.FAMILIES:
@@ -115,6 +131,15 @@ def test_break_requires_first_iteration_before_resume(self):
         self.assertIn("incorrect marker values at r.0",
                       "; ".join(lifecycle.case_problems(value, {})))
 
+    def test_quoting_requires_exact_values_before_and_after_resume(self):
+        for waiting in (True, False):
+            for invalid in (None, 0):
+                value = case("quoting", waiting=waiting)
+                value["marks"][0]["value"] = invalid
+                with self.subTest(waiting=waiting, invalid=invalid):
+                    self.assertIn("incorrect marker values at r.0",
+                                  "; ".join(lifecycle.case_problems(value, {})))
+
     def test_race_requires_exact_loser_and_durable_timer(self):
         for waiting in (True, False):
             for field, invalid in (("race_loser_id", None), ("race_loser_timer_created", False)):
@@ -251,9 +276,16 @@ def validate(names, completed):
             self.assertLess(alter, validations[3])
             self.assertLess(validations[3], binary_wait_release)
             self.assertLess(binary_wait_release, validations[4])
+            quoting_start = calls.index(unittest.mock.call.cluster.start_case(
+                "quoting-binary-b2", "quoting", True))
+            quoting_release = calls.index(unittest.mock.call.cluster.release("quoting-binary-b2"))
+            self.assertLess(new_install, quoting_start)
+            self.assertLess(quoting_start, validations[2])
+            self.assertLess(validations[3], quoting_release)
+            self.assertLess(quoting_release, validations[4])
             baseline = {f"{family}-{boundary}" for family in lifecycle.FAMILIES
                         for boundary in ("b1", "b2")} | {"seq-done-baseline"}
-            binary = baseline | {"seq-done-b1", "seq-binary-b2"}
+            binary = baseline | {"seq-done-b1", "seq-binary-b2", "quoting-binary-b2"}
             self.assertEqual(cohorts, [
                 baseline, baseline, binary, binary, binary | {"seq-done-b2"},
             ])
@@ -275,7 +307,8 @@ def validate(names, completed):
             self.assertEqual(cluster.release.call_args_list,
                 [unittest.mock.call(f"{family}-b1") for family in lifecycle.FAMILIES]
                 + [unittest.mock.call(f"{family}-b2") for family in lifecycle.FAMILIES]
-                + [unittest.mock.call("seq-binary-b2")])
+                + [unittest.mock.call("seq-binary-b2"),
+                   unittest.mock.call("quoting-binary-b2")])
             self.assertEqual(cluster.stop.call_count, 2)
             self.assertEqual(len(list(output.glob("*.json"))), 5)
 

scripts/upgrade_lifecycle.py

@@ -25,7 +25,7 @@
 # structure and resumed under both the B1 (new binary / old schema) and B2
 # (new binary / new schema) boundaries. Every #410 combinator plus the explicit
 # else and break seeds is represented; df.sql is exercised by every shape.
-FAMILIES = ["seq", "if-then", "if-else", "loop", "break", "join", "race"]
+FAMILIES = ["seq", "if-then", "if-else", "loop", "break", "join", "race", "quoting"]
 
 # Exact marker counts per structural path while the instance is still suspended
 # ("before") and after it resumes and completes ("after"). Paths absent from a
@@ -44,6 +44,9 @@
               "before_values": {"r.0": [1]}, "after_values": {"r.0": [1, 2, 3]}},
     "join": {"before": {"r.0": 1, "r.b": 1}, "after": {"r.0": 1, "r.b": 1, "r.1": 1}},
     "race": {"before": {"r.w": 1}, "after": {"r.w": 1, "r.w2": 1}},
+    "quoting": {"before": {"r.0": 1}, "after": {"r.0": 2, "r.1": 2},
+                "before_values": {"r.0": [1]}, "after_values": {"r.0": [1, 1], "r.1": [1, 1]},
+                "result": {"rows": [{"value": 1}], "row_count": 1}},
 }
 
 
@@ -67,7 +70,8 @@ def require_equal(actual, expected):
 
 
 def literal(value):
-    return "'" + value.replace("'", "''") + "'"
+    escaped = value.replace("\\", "\\\\").replace("'", "''")
+    return ("E" if "\\" in value else "") + "'" + escaped + "'"
 
 
 def build(source, output, pg_config, pg_major):
@@ -363,7 +367,8 @@ def check(phase, binary, schema):
             cluster.release(f"{family}-b1")
         cluster.start_sequence("seq-done-b1")
         cluster.start_sequence("seq-binary-b2", waiting=True)
-        names += ["seq-done-b1", "seq-binary-b2"]
+        cluster.start_case("quoting-binary-b2", "quoting", True)
+        names += ["seq-done-b1", "seq-binary-b2", "quoting-binary-b2"]
         check("b1", current, previous)
 
         # B2: new binary, new schema. Resume every -b2 case and start a fresh one.
@@ -372,6 +377,7 @@ def check(phase, binary, schema):
         for family in FAMILIES:
             cluster.release(f"{family}-b2")
         cluster.release("seq-binary-b2")
+        cluster.release("quoting-binary-b2")
         cluster.start_sequence("seq-done-b2")
         names += ["seq-done-b2"]
         check("b2", current, current)

src/activities/execute_sql.rs

@@ -51,7 +51,9 @@ use sqlx::{Column, Row, TypeInfo};
 use std::sync::Arc;
 use tokio::sync::Semaphore;
 
-use crate::types::{connect_as_user, get_execution_acquire_timeout, get_max_user_connections};
+use crate::types::{
+    connect_as_user, get_execution_acquire_timeout, get_max_user_connections, SqlQuotingVersion,
+};
 
 /// Activity name for registration and scheduling
 pub const NAME: &str = "pg_durable::activity::execute-sql";
@@ -61,9 +63,36 @@ pub const NAME: &str = "pg_durable::activity::execute-sql";
 pub struct ExecuteSqlInput {
     pub query: String,
     pub submitted_by: String,
-    /// Target database (None = extension database)
-    #[serde(skip_serializing_if = "Option::is_none")]
+    /// Target database (None = extension database). Historical inputs include the null.
     pub database: Option<String>,
+    #[serde(default, skip_serializing_if = "SqlQuotingVersion::is_legacy")]
+    pub sql_quoting_version: SqlQuotingVersion,
+}
+
+pub(crate) async fn prepare_sql_quoting(
+    conn: &mut sqlx::PgConnection,
+    version: SqlQuotingVersion,
+) -> Result<(), String> {
+    if version.is_legacy() {
+        // Pin an already-enabled setting so a configuration reload cannot make
+        // historical literals unsafe between this check and query parsing.
+        let setting = sqlx::query_scalar::<_, String>(
+            "SELECT pg_catalog.set_config('standard_conforming_strings', 'on', false)
+             WHERE pg_catalog.current_setting('standard_conforming_strings')
+                   OPERATOR(pg_catalog.=) 'on'",
+        )
+        .fetch_optional(conn)
+        .await
+        .map_err(|e| format!("Failed to enforce legacy SQL quoting: {e}"))?;
+        if setting.is_none() {
+            return Err(
+                "Legacy workflow SQL requires standard_conforming_strings=on; \
+                 replay-compatible literals are unsafe when it is off"
+                    .to_string(),
+            );
+        }
+    }
+    Ok(())
 }
 
 /// Decode a single column value from a PostgreSQL row into a `serde_json::Value`.
@@ -237,6 +266,7 @@ pub(crate) async fn execute_in_origin(
     if let Some(origin) = origin {
         crate::origin::validate_execution_connection(&mut conn, &origin).await?;
     }
+    prepare_sql_quoting(&mut conn, input.sql_quoting_version).await?;
     if let Some(source) = source {
         source.validate().await?;
     }
@@ -283,3 +313,32 @@ async fn execute_query(
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn sql_quoting_version_preserves_legacy_activity_bytes() {
+        for legacy in [
+            r#"{"query":"SELECT '\\path'","submitted_by":"reader","database":null}"#,
+            r#"{"query":"SELECT '\\path'","submitted_by":"reader","database":"other"}"#,
+        ] {
+            let input: ExecuteSqlInput = serde_json::from_str(legacy).unwrap();
+            assert_eq!(input.sql_quoting_version, SqlQuotingVersion::Legacy);
+            assert_eq!(serde_json::to_string(&input).unwrap(), legacy);
+        }
+    }
+
+    #[test]
+    fn sql_quoting_version_is_explicit_for_new_activities() {
+        let current = r#"{"query":"SELECT 1","submitted_by":"reader","database":null,"sql_quoting_version":"v1"}"#;
+        let input: ExecuteSqlInput = serde_json::from_str(current).unwrap();
+        assert_eq!(input.sql_quoting_version, SqlQuotingVersion::V1);
+        assert_eq!(serde_json::to_string(&input).unwrap(), current);
+        assert!(serde_json::from_str::<ExecuteSqlInput>(
+            r#"{"query":"SELECT 1","submitted_by":"reader","database":null,"sql_quoting_version":"v2"}"#
+        )
+        .is_err());
+    }
+}

src/activities/load_function_graph.rs

@@ -13,6 +13,7 @@ use std::time::Duration;
 
 use crate::types::{
     is_role_superuser_name, superuser_instances_enabled, FunctionGraph, FunctionNode,
+    SqlQuotingVersion,
 };
 
 /// Activity name for registration and scheduling
@@ -370,6 +371,7 @@ async fn load_visible_graph(
         instance_id,
         root_node_id,
         nodes,
+        sql_quoting_version: SqlQuotingVersion::V1,
     };
 
     ctx.trace_info(format!(