Sql server agent check (#33719)

Adds a SQL Server check to ensure the SQL Server Agent is running.
Without this check, CREATE SOURCE will succeed, but the hydration never
finishes because the CDC jobs, while they exist, are not being run.

### Motivation

fixes MaterializeInc/database-issues#9731

### Checklist

- [ ] This PR has adequate test coverage / QA involvement has been duly
considered. ([trigger-ci for additional test/nightly
runs](https://trigger-ci.dev.materialize.com/))
- [ ] This PR has an associated up-to-date [design
doc](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/README.md),
is a design doc
([template](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/00000000_template.md)),
or is sufficiently small to not require a design.
  <!-- Reference the design in the description. -->
- [ ] If this PR evolves [an existing `$T ⇔ Proto$T`
mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md)
(possibly in a backwards-incompatible way), then it is tagged with a
`T-proto` label.
- [ ] If this PR will require changes to cloud orchestration or tests,
there is a companion cloud PR to account for those changes that is
tagged with the release-blocker label
([example](MaterializeInc/cloud#5021)).
<!-- Ask in #team-cloud on Slack if you need help preparing the cloud
PR. -->
- [ ] If this PR includes major [user-facing behavior
changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note),
I have pinged the relevant PM to schedule a changelog post.

Author: martykulma

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

@@ -2113,6 +2113,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: