Sql server agent check

doc/user/layouts/shortcodes/sql-server-direct/before-you-begin.html

@@ -5,3 +5,14 @@
 
 - Ensure you have access to your SQL Server instance via the [`sqlcmd` client](https://learn.microsoft.com/en-us/sql/tools/sqlcmd/sqlcmd-utility),
   or your preferred SQL client.
+
+- Ensure SQL Server Agent is running.
+  ```mzsql
+  USE msdb;
+  SELECT
+    servicename,
+    status_desc,
+    startup_type_desc
+  FROM sys.dm_server_services
+  WHERE servicename LIKE 'SQL Server Agent%';
+  ```

misc/python/materialize/mzcompose/services/sql_server.py

@@ -25,6 +25,7 @@ def __init__(
         name: str = "sql-server",
         environment_extra: list[str] = [],
         volumes_extra: list[str] = [],
+        enable_agent: bool = True,
     ) -> None:
         super().__init__(
             name=name,
@@ -35,7 +36,7 @@ def __init__(
                 "environment": [
                     "ACCEPT_EULA=Y",
                     "MSSQL_PID=Developer",
-                    "MSSQL_AGENT_ENABLED=True",
+                    f"MSSQL_AGENT_ENABLED={enable_agent}",
                     f"SA_PASSWORD={sa_password}",
                     *environment_extra,
                 ],

src/sql-server-util/src/inspect.rs

@@ -642,6 +642,17 @@ pub async fn ensure_snapshot_isolation_enabled(client: &mut Client) -> Result<()
     Ok(())
 }
 
+/// Ensure the SQL Server Agent is running.
+///
+/// See: <https://learn.microsoft.com/en-us/sql/relational-databases/system-dynamic-management-views/sys-dm-server-services-transact-sql?view=azuresqldb-current&viewFallbackFrom=sql-server-ver17>
+pub async fn ensure_sql_server_agent_running(client: &mut Client) -> Result<(), SqlServerError> {
+    static AGENT_STATUS_QUERY: &str = "SELECT status_desc FROM sys.dm_server_services WHERE servicename LIKE 'SQL Server Agent%';";
+    let result = client.simple_query(AGENT_STATUS_QUERY).await?;
+
+    check_system_result(&result, "SQL Server Agent status".to_string(), "Running")?;
+    Ok(())
+}
+
 pub async fn get_tables(client: &mut Client) -> Result<Vec<SqlServerTableRaw>, SqlServerError> {
     let result = client
         .simple_query(&format!("{GET_COLUMNS_FOR_TABLES_WITH_CDC_QUERY};"))

src/storage-types/src/connections.rs

@@ -2106,6 +2106,7 @@ impl SqlServerConnectionDetails<InlinedConnection> {
         for error in [
             mz_sql_server_util::inspect::ensure_database_cdc_enabled(&mut client).await,
             mz_sql_server_util::inspect::ensure_snapshot_isolation_enabled(&mut client).await,
+            mz_sql_server_util::inspect::ensure_sql_server_agent_running(&mut client).await,
         ] {
             match error {
                 Err(mz_sql_server_util::SqlServerError::InvalidSystemSetting {

test/sql-server-cdc/21-privileges.td

@@ -34,9 +34,13 @@ DENY SELECT ON OBJECT::cdc.dbo_t1_privileges_ct(__$start_lsn) TO authorization_u
 CREATE LOGIN authorization_user_table_perms WITH PASSWORD = '${arg.default-sql-server-password}';
 CREATE USER authorization_user_table_perms FOR LOGIN authorization_user_table_perms;
 ALTER ROLE db_datareader ADD MEMBER authorization_user_table_perms;
-
 DENY SELECT ON OBJECT::dbo.t1_privileges TO authorization_user_table_perms;
 
+USE master;
+GRANT VIEW SERVER STATE TO authorization_user_column_perms;
+GRANT VIEW SERVER STATE TO authorization_user_column_perms_capture_instance;
+GRANT VIEW SERVER STATE TO authorization_user_table_perms;
+
 > CREATE SECRET IF NOT EXISTS sql_server_pass AS '${arg.default-sql-server-password}'
 
 > CREATE CONNECTION sql_server_privileges_connection_table TO SQL SERVER (

test/sql-server-cdc/mzcompose.py

@@ -48,6 +48,8 @@
     ),
 ]
 
+VOLUMES = {"ms_scratch": {}}
+
 
 #
 # Test that SQL Server ingestion works
@@ -124,6 +126,57 @@ def run(file: pathlib.Path | str) -> None:
     c.test_parts(sharded_files, run)
 
 
+def workflow_no_agent(c: Composition, parser: WorkflowArgumentParser) -> None:
+    """
+    Ensures that MZ detects that the SQL Server Agent is not running at purification and produces
+    an error to the user.
+    """
+    # Start with a fresh state
+    c.kill("sql-server")
+    c.rm("sql-server")
+    c.kill("materialized")
+    c.rm("materialized")
+
+    try:
+        with c.override(
+            SqlServer(volumes_extra=["ms_scratch:/var/opt/mssql"]),
+        ):
+            c.up("materialized", "sql-server", Service("testdrive", idle=True))
+            c.run_testdrive_files(
+                "setup/setup.td",
+                f"--var=default-sql-server-user={SqlServer.DEFAULT_USER}",
+                f"--var=default-sql-server-password={SqlServer.DEFAULT_SA_PASSWORD}",
+            )
+            c.kill("sql-server")
+
+            with c.override(
+                SqlServer(
+                    enable_agent=False, volumes_extra=["ms_scratch:/var/opt/mssql"]
+                )
+            ):
+                c.up("sql-server")
+
+                c.testdrive(
+                    dedent(
+                        f"""
+                        > CREATE SECRET IF NOT EXISTS sql_server_pass AS '{SqlServer.DEFAULT_SA_PASSWORD}'
+
+                        ! CREATE CONNECTION sql_server_conn TO SQL SERVER (
+                            HOST 'sql-server',
+                            PORT 1433,
+                            DATABASE test,
+                            USER '{SqlServer.DEFAULT_USER}',
+                            PASSWORD = SECRET sql_server_pass);
+                        contains:Invalid SQL Server system replication settings
+                        """
+                    )
+                )
+    finally:
+        c.kill("sql-server")
+        c.rm("sql-server")
+        c.rm_volumes("ms_scratch")
+
+
 def workflow_snapshot_consistency(
     c: Composition, parser: WorkflowArgumentParser
 ) -> None: