From 50ce0a447d659673b31c15416ee7b8695224bac3 Mon Sep 17 00:00:00 2001 From: lovasoa Date: Mon, 28 Sep 2026 02:40:07 +0200 Subject: [PATCH 1/4] Support native Windows services and systemd lifecycle notifications --- .github/workflows/ci.yml | 12 ++ CHANGELOG.md | 1 + Cargo.lock | 23 ++ Cargo.toml | 10 +- README.md | 2 + configuration.md | 6 + .../sqlpage/migrations/66_log_component.sql | 5 +- .../your-first-sql-website/service.md | 102 +++++++++ .../your-first-sql-website/service.sql | 2 + .../your-first-sql-website/tutorial.md | 2 +- scripts/test-windows-service.ps1 | 84 ++++++++ sqlpage.service | 16 +- src/app_config.rs | 8 + src/cli/arguments.rs | 9 + src/main.rs | 115 ++++++---- src/service/mod.rs | 124 +++++++++++ src/service/windows.rs | 159 ++++++++++++++ src/telemetry.rs | 22 ++ src/telemetry/windows_event_log.rs | 43 ++++ src/webserver/http.rs | 50 ++++- tests/service_lifecycle.rs | 196 ++++++++++++++++++ 21 files changed, 939 insertions(+), 52 deletions(-) create mode 100644 examples/official-site/your-first-sql-website/service.md create mode 100644 examples/official-site/your-first-sql-website/service.sql create mode 100644 scripts/test-windows-service.ps1 create mode 100644 src/service/mod.rs create mode 100644 src/service/windows.rs create mode 100644 src/telemetry/windows_event_log.rs create mode 100644 tests/service_lifecycle.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3e6eb4eba..64662aab3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -117,6 +117,12 @@ jobs: run: | mkdir -p target/sqlpage-test-binaries tar -xzf target/sqlpage-linux-test-binaries.tar.gz -C target/sqlpage-test-binaries + - name: Download server for service lifecycle tests + uses: actions/download-artifact@v8 + with: + name: sqlpage-linux-debug + path: target/debug + - run: chmod +x target/debug/sqlpage - name: Install DuckDB ODBC driver if: matrix.database == 'duckdb' run: | @@ -146,6 +152,7 @@ jobs: DATABASE_URL: ${{ matrix.db_url }} MALLOC_CHECK_: 3 MALLOC_PERTURB_: 10 + SQLPAGE_BINARY: ${{ github.workspace }}/target/debug/sqlpage windows_test: runs-on: windows-latest @@ -173,6 +180,11 @@ jobs: env: CARGO_INCREMENTAL: 1 RUST_BACKTRACE: 1 + - name: Test native Windows service lifecycle + shell: powershell + run: scripts/test-windows-service.ps1 + - name: Lint Windows service code + run: cargo clippy --all-targets -- -D warnings - name: Upload Windows binary uses: actions/upload-artifact@v7 with: diff --git a/CHANGELOG.md b/CHANGELOG.md index 97ad63575..76d92da61 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,7 @@ # CHANGELOG.md ## v0.47.0 (unreleased) +- SQLPage can now run as a native Windows service and report readiness to systemd. Service stops drain active requests, close database connections, and flush telemetry; Windows service logs appear in Event Viewer. - **Mac users:** the downloadable `sqlpage-macos.tgz` now runs natively on Apple silicon (M-series Macs) and no longer runs on Intel Macs. Homebrew remains the recommended and easiest installation method. On an Intel Mac, [install Homebrew](https://brew.sh/) if needed, then run `brew install sqlpage` (or `brew update` followed by `brew upgrade sqlpage` if you already installed it with Homebrew). Open Terminal in your existing website folder and run `sqlpage` instead of `./sqlpage.bin`; keep your SQL files, database, and `sqlpage` configuration folder in place. Intel installations may build from source and take longer; see the [macOS installation guide](https://sql-page.com/your-first-sql-website/?os=macos#download) for setup and older macOS requirements. - Updated sqlx-oldapi to v0.6.57 to fix SQL Server fallback expressions such as `ISNULL($missing, 'default')` truncating defaults or failing for date values when the bound variable is `NULL`. - Fixed MSSQL `JSON_OBJECT('key': value)` expressions being rejected by SQLPage's parser, including when used in `SET` statements or nested in `sqlpage.*` function calls. diff --git a/Cargo.lock b/Cargo.lock index 22edbbf63..0f7293267 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4188,6 +4188,15 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "sd-notify" +version = "0.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b943eadf71d8b69e661330cb0e2656e31040acf21ee7708e2c238a0ec6af2bf4" +dependencies = [ + "libc", +] + [[package]] name = "sec1" version = "0.7.3" @@ -4567,6 +4576,7 @@ dependencies = [ "rustls", "rustls-acme", "rustls-native-certs", + "sd-notify", "serde", "serde_json", "sha2 0.11.0", @@ -4581,6 +4591,8 @@ dependencies = [ "tracing-log", "tracing-opentelemetry", "tracing-subscriber", + "windows-service", + "windows-sys 0.61.2", ] [[package]] @@ -5454,6 +5466,17 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-service" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "857224b3b211c6f3616921f081ee54721ee3ad2ace2fac6a6337e032f7b4dcf2" +dependencies = [ + "bitflags 2.13.2", + "widestring", + "windows-sys 0.61.2", +] + [[package]] name = "windows-strings" version = "0.5.1" diff --git a/Cargo.toml b/Cargo.toml index 7b29ee23c..6cfffae69 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -58,7 +58,7 @@ handlebars = "6.2.0" log = "0.4.17" mime_guess = "2.0.4" futures-util = "0.3.21" -tokio = { version = "1.24.1", features = ["macros", "rt", "process", "sync"] } +tokio = { version = "1.24.1", features = ["macros", "rt", "process", "sync", "signal"] } tokio-stream = "0.1.9" anyhow = "1" serde = "1" @@ -133,6 +133,13 @@ opentelemetry-http = { version = "0.32", default-features = false } opentelemetry-semantic-conventions = { version = "0.32", features = ["semconv_experimental"] } +[target.'cfg(target_os = "linux")'.dependencies] +sd-notify = "0.4" + +[target.'cfg(windows)'.dependencies] +windows-service = "0.8" +windows-sys = { version = "0.61", features = ["Win32_System_EventLog", "Win32_Security"] } + [features] default = [] odbc-static = ["odbc-sys", "odbc-sys/vendored-unix-odbc"] @@ -148,4 +155,3 @@ lambda-web = [ actix-http = "3" tempfile = "3" tokio = { version = "1", features = ["rt", "time", "test-util"] } - diff --git a/README.md b/README.md index fee5dd806..e2e5303a5 100644 --- a/README.md +++ b/README.md @@ -180,6 +180,8 @@ a cheaper ARM cloud instance, using the docker image is the easiest way to do it For managed SQLPage hosting, use [DataPage](https://datapage.app). To run SQLPage yourself on a VPS, [Hostinger](https://www.hostg.xyz/aff_c?offer_id=815&aff_id=243720&url_id=6808) is another option; this is an affiliate link, so we receive a small commission if you buy through it. +To start automatically at boot, see [running SQLPage as a Windows or systemd service](examples/official-site/your-first-sql-website/service.md), including installation, logs, and graceful shutdown. + ### On macOS, with Homebrew [SQLPage's Homebrew package](https://formulae.brew.sh/formula/sqlpage) is the recommended way to install SQLPage on macOS, including Intel Macs. diff --git a/configuration.md b/configuration.md index 9b4e504c8..e875f20cb 100644 --- a/configuration.md +++ b/configuration.md @@ -4,6 +4,12 @@ SQLPage can be configured through either [environment variables](https://en.wiki or a [JSON](https://en.wikipedia.org/wiki/JSON) file placed in `sqlpage/sqlpage.json`. You can find an example configuration file in [`sqlpage/sqlpage.json`](./sqlpage/sqlpage.json). +For automatic startup, service accounts, logging, and shutdown behavior, see +[running SQLPage as a Windows or systemd service](examples/official-site/your-first-sql-website/service.md). +Windows service mode (`--service NAME`) requires an absolute `--web-root`, which also +sets the working directory before loading `.env` and configuration files. Under +systemd, set `WorkingDirectory` in the unit file. + Here are the available configuration options and their default values: | variable | default | description | diff --git a/examples/official-site/sqlpage/migrations/66_log_component.sql b/examples/official-site/sqlpage/migrations/66_log_component.sql index 7109ad285..1b2e21bc0 100644 --- a/examples/official-site/sqlpage/migrations/66_log_component.sql +++ b/examples/official-site/sqlpage/migrations/66_log_component.sql @@ -1,6 +1,6 @@ INSERT INTO component(name, icon, introduced_in_version, description) VALUES ('log', 'logs', '0.37.1', 'A component that writes messages to the server logs. -When a page runs, it prints your message to the terminal/console (standard error). +When a page runs, it writes your message to the server logs. Use it to track what happens and troubleshoot issues. ### Where do the messages appear? @@ -8,7 +8,8 @@ Use it to track what happens and troubleshoot issues. - Running from a terminal (Linux, macOS, or Windows PowerShell/Command Prompt): they show up in the window. - Docker: run `docker logs `. - Linux service (systemd): run `journalctl -u sqlpage`. -- This component''s output is written to [standard error (stderr)](https://en.wikipedia.org/wiki/Standard_streams#Standard_error_(stderr)). SQLPage request access logs are separate and are written to standard output (stdout). +- Native Windows service (`--service NAME`): open Event Viewer → Windows Logs → Application and select the SQLPage source. See the [service setup guide](/your-first-sql-website/service.sql). +- Outside native Windows service mode, this component''s output is written to [standard error (stderr)](https://en.wikipedia.org/wiki/Standard_streams#Standard_error_(stderr)). SQLPage request access logs are separate and are written to standard output (stdout). '); INSERT INTO parameter(component, name, description, type, top_level, optional) SELECT 'log', * FROM (VALUES diff --git a/examples/official-site/your-first-sql-website/service.md b/examples/official-site/your-first-sql-website/service.md new file mode 100644 index 000000000..36b7dc800 --- /dev/null +++ b/examples/official-site/your-first-sql-website/service.md @@ -0,0 +1,102 @@ +# Run SQLPage as a service + +SQLPage can start automatically at boot and run without an open terminal. It reports +readiness after connecting to the database, applying migrations, initializing the +application, and starting its HTTP listeners and workers. A startup failure is +reported to the service manager so it can apply the configured recovery policy. + +When stopped, SQLPage stops accepting new connections and gives active HTTP requests +up to 30 seconds to finish, then closes database connections and flushes telemetry. +Allow extra time for database cleanup and telemetry when configuring service stop +timeouts. Requests that exceed the drain timeout can be interrupted. + +## Linux with systemd + +Install the Linux executable as `/usr/local/bin/sqlpage.bin`, create a dedicated +`sqlpage` user and group, and put your application in `/var/www/sqlpage`. The account +needs read access to the application and write access to its database, uploads, and +configuration directory as appropriate. + +Download the [provided systemd unit](https://github.com/sqlpage/SQLPage/blob/main/sqlpage.service) +to `/etc/systemd/system/sqlpage.service`, adjusting `User`, `Group`, `WorkingDirectory`, +`ExecStart`, and `LISTEN_ON` to match your installation. The example listens on port +80 and grants only the capability needed to bind a privileged port. For port 8080, +change `LISTEN_ON` and remove `AmbientCapabilities`. + +```sh +sudo systemctl daemon-reload +sudo systemctl enable --now sqlpage +sudo systemctl status sqlpage +journalctl -u sqlpage -f +``` + +The unit uses `Type=notify`: `systemctl start` waits for SQLPage's readiness +notification. Startup has a five-minute timeout to allow for migrations. Change +`TimeoutStartSec` if your deployment needs more time. `systemctl stop sqlpage` sends +SIGTERM, which initiates graceful shutdown. The example allows 60 seconds for the +whole stop operation and restarts the process on failure. SIGINT and SIGQUIT also +initiate graceful shutdown when running from a terminal or another supervisor. + +Logs go to the journal. Use `sqlpage/sqlpage.json`, a `.env` file in the working +directory, or systemd `Environment`/`EnvironmentFile` settings for configuration. +SQLPage stays in the foreground; no PID file or daemonization is needed. + +## Windows Service Control Manager + +SQLPage supports native Windows services without a wrapper. Download `sqlpage.exe`, +then prepare an application folder such as `C:\SQLPage\website`, with its +configuration in `C:\SQLPage\website\sqlpage\sqlpage.json`. + +In **Windows PowerShell run as Administrator**, register the event-log source and +create the service (adjust the paths): + +```powershell +New-EventLog -LogName Application -Source SQLPage +$binary = 'C:\SQLPage\sqlpage.exe' +$root = 'C:\SQLPage\website' +$command = '"{0}" --service SQLPage --web-root "{1}"' -f $binary, $root +New-Service -Name SQLPage -BinaryPathName $command -StartupType Automatic ` + -DisplayName 'SQLPage website' +``` + +Skip `New-EventLog` if the `SQLPage` source is already registered. It is shared by +all SQLPage services. This command is available in Windows PowerShell 5.1. + +Before starting, open `services.msc`, find **SQLPage website**, and set its **Log On** +account to a dedicated service account with access to your application, database, +and uploads. `New-Service` defaults to LocalSystem; choose an account with only the +permissions your application needs. Configure automatic recovery on the **Recovery** +tab if desired, including recovery for non-crash failures. + +```powershell +Start-Service SQLPage +Get-Service SQLPage +Get-WinEvent -FilterHashtable @{ LogName = 'Application'; ProviderName = 'SQLPage' } -MaxEvents 20 +Restart-Service SQLPage +Stop-Service SQLPage +``` + +`--service NAME` must match the registered service name and requires an **absolute** +`--web-root`. In service mode this directory is also the working directory, so `.env`, +relative configuration paths, SQLite files, and uploads resolve there instead of +Windows' system directory. You may also pass `--config-dir` or `--config-file` in the +service command. Use distinct names and listening ports to run multiple services. + +SQLPage reports `START_PENDING` during initialization, `RUNNING` once ready, +`STOP_PENDING` while draining requests, and `STOPPED` after cleanup. Startup progress +updates carry a five-minute wait hint. Both service-stop and operating-system +shutdown controls initiate cleanup. Failures report service-specific exit code 1; +details and application logs appear in **Event Viewer → Windows Logs → Application** +under the **SQLPage** source. Existing OpenTelemetry export remains available. + +To remove the service: + +```powershell +Stop-Service SQLPage +sc.exe delete SQLPage +``` + +Running `sqlpage.exe` without `--service` continues to run in a terminal, with normal +console logging and graceful shutdown on Ctrl+C or Ctrl+Break. To diagnose a service +configuration, open a terminal in its web root and run the same command without +`--service NAME`. diff --git a/examples/official-site/your-first-sql-website/service.sql b/examples/official-site/your-first-sql-website/service.sql new file mode 100644 index 000000000..5b9e360e0 --- /dev/null +++ b/examples/official-site/your-first-sql-website/service.sql @@ -0,0 +1,2 @@ +SELECT 'dynamic' AS component, properties FROM example WHERE component = 'shell' LIMIT 1; +SELECT 'text' AS component, sqlpage.read_file_as_text('your-first-sql-website/service.md') AS contents_md; diff --git a/examples/official-site/your-first-sql-website/tutorial.md b/examples/official-site/your-first-sql-website/tutorial.md index 22cfc9f7e..385216db7 100644 --- a/examples/official-site/your-first-sql-website/tutorial.md +++ b/examples/official-site/your-first-sql-website/tutorial.md @@ -188,7 +188,7 @@ Alternatively, you can use a [Hostinger VPS](https://www.hostg.xyz/aff_c?offer_i If you prefer to host your website yourself, you can use a cloud provider or a VPS provider. You will need to: - Configure domain name resolution to point to your server - Open the port you are using (8080 by default) in your server's firewall -- [Setup docker](https://github.com/sqlpage/SQLPage?tab=readme-ov-file#with-docker) or another process manager such as [systemd](https://github.com/sqlpage/SQLPage/blob/main/sqlpage.service) to start SQLPage automatically when your server boots and to keep it running +- [Setup docker](https://github.com/sqlpage/SQLPage?tab=readme-ov-file#with-docker) or [run SQLPage as a Windows or systemd service](service.sql) to start SQLPage automatically when your server boots and to keep it running - Optionally, [setup a reverse proxy](nginx.sql) to avoid exposing SQLPage directly to the internet - Optionally, setup a TLS certificate to enable HTTPS - Configure connection to a cloud database or a database running on your server in [`sqlpage.json`](https://github.com/sqlpage/SQLPage/blob/main/configuration.md#configuring-sqlpage) diff --git a/scripts/test-windows-service.ps1 b/scripts/test-windows-service.ps1 new file mode 100644 index 000000000..62767a139 --- /dev/null +++ b/scripts/test-windows-service.ps1 @@ -0,0 +1,84 @@ +# Run from an elevated Windows PowerShell prompt, or on the Windows CI runner. +param([string]$Binary = "$PSScriptRoot\..\target\debug\sqlpage.exe") +$ErrorActionPreference = 'Stop' +$Binary = (Resolve-Path $Binary).Path +$name = 'SQLPageTest' + [Guid]::NewGuid().ToString('N') +$root = Join-Path ([IO.Path]::GetTempPath()) ("SQLPage service test " + $name) +$created = $false +$listener = $null +$client = $null + +function Wait-State([string]$state) { + $service = Get-Service $name + $service.WaitForStatus($state, [TimeSpan]::FromSeconds(60)) +} + +try { + New-Item -ItemType Directory -Path (Join-Path $root 'sqlpage') -Force | Out-Null + if (-not [Diagnostics.EventLog]::SourceExists('SQLPage')) { + New-EventLog -LogName Application -Source SQLPage + } + # Reserve a port to exercise startup failure before allowing a successful start. + $listener = [Net.Sockets.TcpListener]::new([Net.IPAddress]::Loopback, 0) + $listener.Start() + $port = $listener.LocalEndpoint.Port + $configuration = @{ listen_on = "127.0.0.1:$port"; database_url = 'sqlite::memory:' } | + ConvertTo-Json + [IO.File]::WriteAllText((Join-Path $root 'sqlpage\sqlpage.json'), $configuration) + [IO.File]::WriteAllText((Join-Path $root 'index.sql'), "SELECT 'text' AS component, 'service ready' AS contents;") + $command = '"{0}" --service {1} --web-root "{2}"' -f $Binary, $name, $root + New-Service -Name $name -BinaryPathName $command -StartupType Manual | Out-Null + $created = $true + $failed = $false + try { Start-Service $name } catch { $failed = $true } + if (-not $failed) { throw 'A port conflict must fail service startup' } + Wait-State 'Stopped' + $status = Get-CimInstance Win32_Service -Filter "Name='$name'" + if ($status.ServiceSpecificExitCode -ne 1) { throw 'Startup failure was not reported to SCM' } + $listener.Stop() + + Start-Service $name + Wait-State 'Running' + $response = Invoke-WebRequest "http://127.0.0.1:$port/" -UseBasicParsing + if ($response.Content -notmatch 'service ready') { throw 'The service did not use its web root' } + + # Exercise STOP while a real SQL request is waiting on an HTTP fetch. + $listener = [Net.Sockets.TcpListener]::new([Net.IPAddress]::Loopback, 0) + $listener.Start() + $upstreamPort = $listener.LocalEndpoint.Port + [IO.File]::WriteAllText((Join-Path $root 'slow.sql'), "SELECT 'text' AS component, sqlpage.fetch('http://127.0.0.1:$upstreamPort/') AS contents;") + Add-Type -AssemblyName System.Net.Http + $client = [Net.Http.HttpClient]::new() + $pendingResponse = $client.GetStringAsync("http://127.0.0.1:$port/slow.sql") + $accepted = $listener.AcceptTcpClientAsync() + if (-not $accepted.Wait(20000)) { throw 'SQL request did not reach the upstream server' } + $upstream = $accepted.Result + $stream = $upstream.GetStream() + $stream.ReadTimeout = 20000 + $headers = New-Object byte[] 4096 + if ($stream.Read($headers, 0, $headers.Length) -eq 0) { throw 'Missing upstream request' } + sc.exe stop $name | Out-Null + if ($LASTEXITCODE -ne 0) { throw 'SCM rejected STOP' } + Wait-State 'StopPending' + $bytes = [Text.Encoding]::ASCII.GetBytes("HTTP/1.1 200 OK`r`nContent-Length: 16`r`nConnection: close`r`n`r`nrequest finished") + $stream.Write($bytes, 0, $bytes.Length) + $upstream.Dispose() + if (-not $pendingResponse.Wait(20000)) { throw 'Active request did not finish' } + if ($pendingResponse.Result -notmatch 'request finished') { throw 'Active response was truncated' } + Wait-State 'Stopped' + $status = Get-CimInstance Win32_Service -Filter "Name='$name'" + if ($status.ExitCode -ne 0) { throw 'Clean stop reported a failure' } + Start-Service $name + Wait-State 'Running' + Stop-Service $name + Wait-State 'Stopped' + Write-Host 'Windows service startup failure, readiness, graceful stop, and restart passed.' +} finally { + if ($client) { $client.Dispose() } + if ($listener) { $listener.Stop() } + if ($created) { + Stop-Service $name -ErrorAction SilentlyContinue + sc.exe delete $name | Out-Null + } + Remove-Item -Recurse -Force $root -ErrorAction SilentlyContinue +} diff --git a/sqlpage.service b/sqlpage.service index ea65debbf..38ba692e8 100644 --- a/sqlpage.service +++ b/sqlpage.service @@ -5,9 +5,16 @@ [Unit] Description=SQLPage website Documentation=https://sql-page.com -After=network.target +Wants=network-online.target +After=network-online.target [Service] +Type=notify +NotifyAccess=main +# Allow database connections and migrations to finish during startup. +TimeoutStartSec=300 +# SQLPage drains HTTP requests for up to 30s, then closes the database and logs. +TimeoutStopSec=60 # Define the user and group to run the service User=sqlpage Group=sqlpage @@ -24,12 +31,12 @@ Environment="LISTEN_ON=0.0.0.0:80" AmbientCapabilities=CAP_NET_BIND_SERVICE # Restart options -Restart=always +Restart=on-failure RestartSec=10 # Logging options -#StandardOutput=syslog -#StandardError=syslog +StandardOutput=journal +StandardError=journal SyslogIdentifier=sqlpage # Security options @@ -42,7 +49,6 @@ ProtectKernelTunables=true ProtectClock=true ProtectHostname=true ProtectProc=invisible -ProtectClock=true # Resource limits LimitNOFILE=65536 diff --git a/src/app_config.rs b/src/app_config.rs index 6e9166457..305ffcf84 100644 --- a/src/app_config.rs +++ b/src/app_config.rs @@ -982,6 +982,8 @@ mod test { config_dir: None, config_file: None, command: None, + #[cfg(windows)] + service: None, }; let config = AppConfig::from_cli(&cli).unwrap(); @@ -1027,6 +1029,8 @@ mod test { config_dir: None, config_file: Some(config_file_path.clone()), command: None, + #[cfg(windows)] + service: None, }; let config = AppConfig::from_cli(&cli).unwrap(); @@ -1046,6 +1050,8 @@ mod test { config_dir: None, config_file: Some(config_file_path), command: None, + #[cfg(windows)] + service: None, }; let config = AppConfig::from_cli(&cli_with_web_root).unwrap(); @@ -1082,6 +1088,8 @@ mod test { config_dir: None, config_file: None, command: None, + #[cfg(windows)] + service: None, }; let config = AppConfig::from_cli(&cli).unwrap(); diff --git a/src/cli/arguments.rs b/src/cli/arguments.rs index 5ba52076b..2d08afe22 100644 --- a/src/cli/arguments.rs +++ b/src/cli/arguments.rs @@ -15,6 +15,11 @@ pub struct Cli { #[clap(short = 'c', long)] pub config_file: Option, + /// Run under the Windows Service Control Manager using this registered service name. + #[cfg(windows)] + #[clap(long, value_name = "NAME", requires = "web_root")] + pub service: Option, + /// Subcommands for additional functionality. #[clap(subcommand)] pub command: Option, @@ -22,6 +27,10 @@ pub struct Cli { pub fn parse_cli() -> anyhow::Result { let cli = Cli::parse(); + #[cfg(windows)] + if cli.service.is_some() && cli.command.is_some() { + anyhow::bail!("--service cannot be used with a subcommand"); + } Ok(cli) } diff --git a/src/main.rs b/src/main.rs index b73369228..413c12d32 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,59 +5,102 @@ use sqlpage::{ webserver::{self, Database}, }; -#[actix_web::main] -async fn main() { - let cli = match cli::arguments::parse_cli() { - Ok(cli) => cli, - Err(e) => { - eprintln!("{e:#}"); - std::process::exit(1); - } - }; - - let is_server_mode = cli.command.is_none(); +mod service; - if is_server_mode { - if let Err(e) = init_logging() { - eprintln!("Failed to initialize logging/telemetry: {e:#}"); - std::process::exit(1); +fn main() -> std::process::ExitCode { + match main_result() { + Ok(()) => std::process::ExitCode::SUCCESS, + Err(error) => { + eprintln!("{error:#}"); + std::process::ExitCode::FAILURE } - } else { - let _ = dotenvy::dotenv(); } +} - if let Err(e) = start(cli).await { - if is_server_mode { - log::error!("{e:?}"); - } else { - eprintln!("{e:#}"); +fn main_result() -> anyhow::Result<()> { + let mut cli = cli::arguments::parse_cli()?; + #[cfg(windows)] + if cli.service.is_some() { + return service::windows::dispatch(cli); + } + + actix_web::rt::System::new().block_on(async { + if let Some(command) = cli.command.take() { + let _ = dotenvy::dotenv(); + let config = AppConfig::from_cli(&cli)?; + return command.execute(config).await; } - std::process::exit(1); + let service = service::Service::console()?; + run(cli, service).await + }) +} + +async fn run(cli: cli::arguments::Cli, service: service::Service) -> anyhow::Result<()> { + init_logging(&service)?; + let result = start(cli, &service).await; + if let Err(error) = &result { + log::error!("{error:#}"); } + // Flush on startup failures as well as on normal shutdown, while the runtime lives. + // Exporter shutdown may block, so keep the runtime free to finish exports. + tokio::task::spawn_blocking(telemetry::shutdown_telemetry).await?; + result } -async fn start(cli: cli::arguments::Cli) -> anyhow::Result<()> { +async fn start(cli: cli::arguments::Cli, service: &service::Service) -> anyhow::Result<()> { + service.starting("Loading configuration")?; let app_config = AppConfig::from_cli(&cli)?; - - if let Some(command) = cli.command { - return command.execute(app_config).await; + service.starting("Connecting to the database")?; + let db = tokio::select! { + biased; + () = service.stop.cancelled() => { + service.stopping()?; + return Ok(()); + } + result = Database::init(&app_config) => result?, + }; + let pool = db.connection.clone(); + let initialize = async { + service.starting("Applying database migrations")?; + webserver::database::migrations::apply(&app_config, &db).await?; + service.starting("Initializing application")?; + AppState::init_with_db(&app_config, db).await + }; + let state = tokio::select! { + biased; + () = service.stop.cancelled() => None, + result = initialize => Some(result), + }; + if !matches!(state, Some(Ok(_))) { + pool.close().await; } + let state = state.transpose()?; + let Some(state) = state else { + service.stopping()?; + return Ok(()); + }; - let db = Database::init(&app_config).await?; - webserver::database::migrations::apply(&app_config, &db).await?; - let state = AppState::init_with_db(&app_config, db).await?; - - log::debug!("Starting server..."); - webserver::http::run_server(&app_config, state).await?; + let stopping_service = service.clone(); + let shutdown = async move { + stopping_service.stop.cancelled().await; + if let Err(error) = stopping_service.stopping() { + log::error!("Unable to report service shutdown: {error:#}"); + } + }; + let result = + webserver::http::run_server_with_shutdown(&app_config, state, shutdown, || service.ready()) + .await; + // Also close on bind failures, before the HTTP runner owns the server. + pool.close().await; + result?; log::info!("Server stopped gracefully. Goodbye!"); - telemetry::shutdown_telemetry(); Ok(()) } -fn init_logging() -> anyhow::Result<()> { +fn init_logging(service: &service::Service) -> anyhow::Result<()> { let load_env = dotenvy::dotenv(); - let otel_active = telemetry::init_telemetry()?; + let otel_active = service.init_telemetry()?; match load_env { Ok(path) => log::info!("Loaded environment variables from {}", path.display()), diff --git a/src/service/mod.rs b/src/service/mod.rs new file mode 100644 index 000000000..3f875f181 --- /dev/null +++ b/src/service/mod.rs @@ -0,0 +1,124 @@ +//! Process lifecycle integration. Kept in the executable so embedding `SQLPage` does +//! not install process-wide signal handlers or contact a service manager. + +use tokio_util::sync::CancellationToken; + +#[cfg(windows)] +pub(super) mod windows; + +#[derive(Clone)] +pub(super) struct Service { + pub(super) stop: CancellationToken, + #[cfg(windows)] + pub(super) status: Option, +} + +impl Service { + pub(super) fn console() -> anyhow::Result { + let service = Self { + stop: CancellationToken::new(), + #[cfg(windows)] + status: None, + }; + let stop = service.stop.clone(); + // Register before database connections or migrations can delay startup. + #[cfg(unix)] + { + use tokio::signal::unix::{SignalKind, signal}; + let mut terminate = signal(SignalKind::terminate())?; + let mut interrupt = signal(SignalKind::interrupt())?; + let mut quit = signal(SignalKind::quit())?; + tokio::spawn(async move { + tokio::select! { + _ = terminate.recv() => {}, + _ = interrupt.recv() => {}, + _ = quit.recv() => {}, + } + stop.cancel(); + }); + } + #[cfg(windows)] + { + let mut interrupt = tokio::signal::windows::ctrl_c()?; + let mut terminate = tokio::signal::windows::ctrl_break()?; + tokio::spawn(async move { + tokio::select! { + _ = interrupt.recv() => {}, + _ = terminate.recv() => {}, + } + stop.cancel(); + }); + } + Ok(service) + } + + #[cfg_attr( + not(windows), + expect( + clippy::unused_self, + reason = "Windows service state selects the log destination" + ) + )] + pub(super) fn init_telemetry(&self) -> anyhow::Result { + #[cfg(windows)] + if self.status.is_some() { + return sqlpage::telemetry::init_windows_service_telemetry(); + } + sqlpage::telemetry::init_telemetry() + } + + #[cfg_attr( + not(windows), + expect( + clippy::unused_self, + reason = "Windows services report startup progress" + ) + )] + pub(super) fn starting(&self, message: &str) -> anyhow::Result<()> { + log::info!("{message}"); + #[cfg(target_os = "linux")] + sd_notify::notify(false, &[sd_notify::NotifyState::Status(message)])?; + #[cfg(windows)] + self.set_status(windows_service::service::ServiceState::StartPending, false)?; + Ok(()) + } + + pub(super) fn ready(&self) -> anyhow::Result<()> { + if self.stop.is_cancelled() { + return Ok(()); + } + #[cfg(target_os = "linux")] + sd_notify::notify( + false, + &[ + sd_notify::NotifyState::Ready, + sd_notify::NotifyState::Status("Serving requests"), + ], + )?; + #[cfg(windows)] + self.set_status(windows_service::service::ServiceState::Running, false)?; + Ok(()) + } + + #[cfg_attr( + not(windows), + expect( + clippy::unused_self, + reason = "Windows service state holds the status handle" + ) + )] + pub(super) fn stopping(&self) -> anyhow::Result<()> { + log::info!("Stopping SQLPage; waiting for active requests to finish"); + #[cfg(target_os = "linux")] + sd_notify::notify( + false, + &[ + sd_notify::NotifyState::Stopping, + sd_notify::NotifyState::Status("Draining requests and closing the database"), + ], + )?; + #[cfg(windows)] + self.set_status(windows_service::service::ServiceState::StopPending, false)?; + Ok(()) + } +} diff --git a/src/service/windows.rs b/src/service/windows.rs new file mode 100644 index 000000000..960676d98 --- /dev/null +++ b/src/service/windows.rs @@ -0,0 +1,159 @@ +use super::Service; +use anyhow::Context; +use sqlpage::cli::arguments::Cli; +use std::{ffi::OsString, sync::Mutex, time::Duration}; +use tokio_util::sync::CancellationToken; +use windows_service::{ + define_windows_service, + service::{ + ServiceControl, ServiceControlAccept, ServiceExitCode, ServiceState, ServiceStatus, + ServiceType, + }, + service_control_handler::{self, ServiceControlHandlerResult}, + service_dispatcher, +}; + +static ARGUMENTS: Mutex> = Mutex::new(None); +static RESULT: Mutex>> = Mutex::new(None); +static CHECKPOINT: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(1); + +define_windows_service!(service_main_ffi, service_main); + +pub(crate) fn dispatch(cli: Cli) -> anyhow::Result<()> { + let name = cli.service.clone().context("Missing service name")?; + *ARGUMENTS.lock().unwrap() = Some(cli); + service_dispatcher::start(&name, service_main_ffi) + .context("Unable to connect to the Windows Service Control Manager. Start this service with Start-Service, or omit --service to run in a terminal")?; + RESULT + .lock() + .unwrap() + .take() + .context("Windows service did not start")? +} + +fn service_main(_arguments: Vec) { + if let Err(error) = run_service() { + sqlpage::telemetry::windows_event_log::write(tracing::Level::ERROR, &format!("{error:#}")); + *RESULT.lock().unwrap() = Some(Err(error)); + } +} + +fn run_service() -> anyhow::Result<()> { + let cli = ARGUMENTS + .lock() + .unwrap() + .take() + .context("Missing service arguments")?; + let name = cli.service.as_deref().context("Missing service name")?; + let stop = CancellationToken::new(); + let handler_stop = stop.clone(); + let status = service_control_handler::register(name, move |control| match control { + ServiceControl::Stop | ServiceControl::Shutdown => { + handler_stop.cancel(); + ServiceControlHandlerResult::NoError + } + ServiceControl::Interrogate => ServiceControlHandlerResult::NoError, + _ => ServiceControlHandlerResult::NotImplemented, + })?; + let service = Service { + stop, + status: Some(status), + }; + service.set_status(ServiceState::StartPending, false)?; + + let result = (|| { + // SCM starts services in System32. Set the directory before dotenv, + // configuration loading, or creating any runtime/worker threads. + let root = cli + .web_root + .as_ref() + .context("--service requires --web-root")?; + anyhow::ensure!( + root.is_absolute(), + "--service requires an absolute --web-root" + ); + std::env::set_current_dir(root) + .with_context(|| format!("Unable to use service web root {}", root.display()))?; + actix_web::rt::System::new().block_on(crate::run(cli, service.clone())) + })(); + let failed = result.is_err(); + if let Err(error) = &result { + sqlpage::telemetry::windows_event_log::write(tracing::Level::ERROR, &format!("{error:#}")); + } + // Publish the result before reporting Stopped: the dispatcher may return as + // soon as SCM receives that status. No work may remain after this call. + *RESULT.lock().unwrap() = Some(result); + service.set_status(ServiceState::Stopped, failed) +} + +impl Service { + pub(super) fn set_status(&self, state: ServiceState, failed: bool) -> anyhow::Result<()> { + if let Some(handle) = &self.status { + let mut value = status(state, failed); + if value.checkpoint != 0 { + value.checkpoint = CHECKPOINT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + handle.set_service_status(value)?; + } + Ok(()) + } +} + +fn status(state: ServiceState, failed: bool) -> ServiceStatus { + let pending = matches!( + state, + ServiceState::StartPending | ServiceState::StopPending + ); + ServiceStatus { + service_type: ServiceType::OWN_PROCESS, + current_state: state, + controls_accepted: if state == ServiceState::Running { + ServiceControlAccept::STOP | ServiceControlAccept::SHUTDOWN + } else { + ServiceControlAccept::empty() + }, + exit_code: if failed { + ServiceExitCode::ServiceSpecific(1) + } else { + ServiceExitCode::Win32(0) + }, + checkpoint: u32::from(pending), + wait_hint: match state { + ServiceState::StartPending => Duration::from_secs(300), + ServiceState::StopPending => Duration::from_secs(60), + _ => Duration::ZERO, + }, + process_id: None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn service_status_obeys_scm_state_contract() { + for state in [ + ServiceState::StartPending, + ServiceState::Running, + ServiceState::StopPending, + ServiceState::Stopped, + ] { + let value = status(state, false); + assert_eq!( + value.controls_accepted.is_empty(), + state != ServiceState::Running + ); + let pending = matches!( + state, + ServiceState::StartPending | ServiceState::StopPending + ); + assert_eq!(value.checkpoint > 0, pending); + assert_eq!(!value.wait_hint.is_zero(), pending); + } + assert_eq!( + status(ServiceState::Stopped, true).exit_code, + ServiceExitCode::ServiceSpecific(1) + ); + } +} diff --git a/src/telemetry.rs b/src/telemetry.rs index 41c669edb..4e7bf3d85 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -186,6 +186,15 @@ pub fn init_telemetry() -> anyhow::Result { init_telemetry_with_log_layer(logfmt::LogfmtLayer::new()) } +/// Initializes service logging in the Windows Application event log, with optional OTLP export. +#[cfg(windows)] +pub fn init_windows_service_telemetry() -> anyhow::Result { + init_telemetry_with_log_layer(logfmt::LogfmtLayer::windows_service()) +} + +#[cfg(windows)] +pub mod windows_event_log; + fn init_telemetry_with_log_layer(logfmt_layer: logfmt::LogfmtLayer) -> anyhow::Result { let otel_endpoint = env::var("OTEL_EXPORTER_OTLP_ENDPOINT").ok(); let otel_active = otel_endpoint.as_deref().is_some_and(|v| !v.is_empty()); @@ -441,6 +450,8 @@ mod logfmt { enum OutputMode { StdoutAndStderr, TestWriter, + #[cfg(windows)] + WindowsEventLog, } pub(super) struct LogfmtLayer { @@ -465,6 +476,15 @@ mod logfmt { output_mode: OutputMode::TestWriter, } } + + #[cfg(windows)] + pub(super) fn windows_service() -> Self { + Self { + stdout_colors: false, + stderr_colors: false, + output_mode: OutputMode::WindowsEventLog, + } + } } impl Layer for LogfmtLayer @@ -527,6 +547,8 @@ mod logfmt { OutputMode::TestWriter => { eprint!("{buf}"); } + #[cfg(windows)] + OutputMode::WindowsEventLog => super::windows_event_log::write(level, &buf), } } } diff --git a/src/telemetry/windows_event_log.rs b/src/telemetry/windows_event_log.rs new file mode 100644 index 000000000..736789bd6 --- /dev/null +++ b/src/telemetry/windows_event_log.rs @@ -0,0 +1,43 @@ +//! Native service logs, including failures before the tracing subscriber exists. + +use windows_sys::Win32::System::EventLog::{ + DeregisterEventSource, EVENTLOG_ERROR_TYPE, EVENTLOG_INFORMATION_TYPE, EVENTLOG_WARNING_TYPE, + RegisterEventSourceW, ReportEventW, +}; + +/// Writes one event under the `SQLPage` source in the Application log. +pub fn write(level: tracing::Level, message: &str) { + let event_type = match level { + tracing::Level::ERROR => EVENTLOG_ERROR_TYPE, + tracing::Level::WARN => EVENTLOG_WARNING_TYPE, + _ => EVENTLOG_INFORMATION_TYPE, + }; + // ReportEvent limits insertion strings to 31,839 UTF-16 code units. + let message: Vec = message + .encode_utf16() + .take(31_838) + .map(|unit| if unit == 0 { 0xfffd } else { unit }) + .chain(Some(0)) + .collect(); + let strings = [message.as_ptr()]; + // SAFETY: All pointers refer to live, NUL-terminated strings. The returned + // handle is used only here and is released after the synchronous write. + unsafe { + let handle = RegisterEventSourceW(std::ptr::null(), windows_sys::w!("SQLPage")); + if handle.is_null() { + return; + } + ReportEventW( + handle, + event_type, + 0, + 0, + std::ptr::null(), + 1, + 0, + strings.as_ptr(), + std::ptr::null(), + ); + DeregisterEventSource(handle); + } +} diff --git a/src/webserver/http.rs b/src/webserver/http.rs index 8448b9dc3..4b5c8ac07 100644 --- a/src/webserver/http.rs +++ b/src/webserver/http.rs @@ -624,6 +624,27 @@ fn default_headers() -> middleware::DefaultHeaders { } pub async fn run_server(config: &AppConfig, state: AppState) -> anyhow::Result<()> { + run_server_inner(config, state, None, || Ok(())).await +} + +/// Runs the server with externally managed shutdown and readiness reporting. +/// The readiness callback runs after the listeners and workers have started. +/// Completing `shutdown` drains requests for up to 30 seconds before closing the database. +pub async fn run_server_with_shutdown( + config: &AppConfig, + state: AppState, + shutdown: impl Future + Send + 'static, + on_ready: impl FnOnce() -> anyhow::Result<()>, +) -> anyhow::Result<()> { + run_server_inner(config, state, Some(Box::pin(shutdown)), on_ready).await +} + +async fn run_server_inner( + config: &AppConfig, + state: AppState, + shutdown: Option + Send>>>, + on_ready: impl FnOnce() -> anyhow::Result<()>, +) -> anyhow::Result<()> { let listen_on = config.listen_on(); let state = web::Data::new(state); let final_state = web::Data::clone(&state); @@ -637,6 +658,9 @@ pub async fn run_server(config: &AppConfig, state: AppState) -> anyhow::Result<( return Ok(()); } let mut server = HttpServer::new(factory); + if let Some(shutdown) = shutdown { + server = server.shutdown_signal(shutdown); + } #[cfg_attr( not(target_family = "unix"), expect( @@ -681,15 +705,29 @@ pub async fn run_server(config: &AppConfig, state: AppState) -> anyhow::Result<( } } - log_welcome_message(config, &server.addrs()); - server - .run() - .await - .with_context(|| "Unable to start the application")?; + let addresses = server.addrs(); + let mut server = Box::pin(server.run()); + // Actix starts its workers and accept thread on the first poll, not in run(). + let initial = + futures_util::future::poll_fn(|cx| std::task::Poll::Ready(server.as_mut().poll(cx))).await; + let result = match initial { + std::task::Poll::Ready(result) => result.context("Unable to start the application"), + std::task::Poll::Pending => match on_ready() { + Ok(()) => { + log_welcome_message(config, &addresses); + server.await.context("HTTP server failed") + } + Err(error) => { + let handle = server.handle(); + let _ = tokio::join!(handle.stop(true), server); + Err(error) + } + }, + }; // We are done, we can close the database connection final_state.db.close().await?; - Ok(()) + result } fn website_url(bound_to: SocketAddr) -> String { diff --git a/tests/service_lifecycle.rs b/tests/service_lifecycle.rs new file mode 100644 index 000000000..f88660845 --- /dev/null +++ b/tests/service_lifecycle.rs @@ -0,0 +1,196 @@ +//! Exercise the real executable, including OS signals and the sd-notify socket. +#![cfg(target_os = "linux")] + +use std::{ + io::{Read, Write}, + net::{TcpListener, TcpStream}, + os::unix::net::UnixDatagram, + process::{Child, Command, Stdio}, + time::{Duration, Instant}, +}; + +struct Server { + child: Child, + directory: tempfile::TempDir, + notifications: UnixDatagram, +} + +impl Server { + fn start(configuration: &serde_json::Value, sql: &str) -> Self { + let directory = tempfile::tempdir().unwrap(); + std::fs::create_dir(directory.path().join("sqlpage")).unwrap(); + std::fs::write( + directory.path().join("sqlpage/sqlpage.json"), + configuration.to_string(), + ) + .unwrap(); + std::fs::write(directory.path().join("index.sql"), sql).unwrap(); + let socket_path = directory.path().join("notify.sock"); + let notifications = UnixDatagram::bind(&socket_path).unwrap(); + notifications + .set_read_timeout(Some(Duration::from_secs(20))) + .unwrap(); + let binary = std::env::var_os("SQLPAGE_BINARY") + .unwrap_or_else(|| env!("CARGO_BIN_EXE_sqlpage").into()); + let child = Command::new(binary) + .current_dir(directory.path()) + .env_clear() + .env("NOTIFY_SOCKET", socket_path) + .stdout(Stdio::null()) + .stderr(std::fs::File::create(directory.path().join("stderr.log")).unwrap()) + .spawn() + .unwrap(); + Self { + child, + directory, + notifications, + } + } + + fn wait_for(&self, expected: &str) { + let deadline = Instant::now() + Duration::from_secs(20); + let mut buffer = [0; 2048]; + while Instant::now() < deadline { + let size = self + .notifications + .recv(&mut buffer) + .unwrap_or_else(|error| panic!("Waiting for {expected}: {error}; {}", self.logs())); + if String::from_utf8_lossy(&buffer[..size]) + .lines() + .any(|line| line == expected) + { + return; + } + } + panic!("Missing {expected}: {}", self.logs()); + } + + fn signal(&self, signal: &str) { + assert!( + Command::new("kill") + .args([signal, &self.child.id().to_string()]) + .status() + .unwrap() + .success() + ); + } + + fn wait_for_exit(&mut self) -> std::process::ExitStatus { + let deadline = Instant::now() + Duration::from_secs(20); + loop { + if let Some(status) = self.child.try_wait().unwrap() { + return status; + } + assert!( + Instant::now() < deadline, + "Service did not stop: {}", + self.logs() + ); + std::thread::sleep(Duration::from_millis(20)); + } + } + + fn logs(&self) -> String { + std::fs::read_to_string(self.directory.path().join("stderr.log")).unwrap() + } +} + +impl Drop for Server { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +fn free_address() -> std::net::SocketAddr { + TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() +} + +#[test] +fn readiness_and_signals_drain_active_requests() { + for signal in ["-TERM", "-INT", "-QUIT"] { + let upstream = TcpListener::bind("127.0.0.1:0").unwrap(); + upstream.set_nonblocking(true).unwrap(); + let address = free_address(); + let mut server = Server::start( + &serde_json::json!({"listen_on": address.to_string(), "database_url": "sqlite::memory:"}), + &format!( + "SELECT 'text' AS component, sqlpage.fetch('http://{}') AS contents;", + upstream.local_addr().unwrap() + ), + ); + server.wait_for("READY=1"); + + let mut request = TcpStream::connect(address).unwrap(); + request + .set_read_timeout(Some(Duration::from_secs(20))) + .unwrap(); + request + .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .unwrap(); + // Wait until the real SQL request is blocked on its outbound HTTP fetch. + let deadline = Instant::now() + Duration::from_secs(20); + let mut upstream_request = loop { + match upstream.accept() { + Ok((request, _)) => break request, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { + assert!( + Instant::now() < deadline, + "SQL request did not start: {}", + server.logs() + ); + std::thread::sleep(Duration::from_millis(20)); + } + Err(error) => panic!("{error}"), + } + }; + upstream_request + .set_read_timeout(Some(Duration::from_secs(20))) + .unwrap(); + let mut headers = [0; 4096]; + assert!(upstream_request.read(&mut headers).unwrap() > 0); + server.signal(signal); + server.wait_for("STOPPING=1"); + assert!(server.child.try_wait().unwrap().is_none()); + upstream_request.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 16\r\nConnection: close\r\n\r\nrequest finished").unwrap(); + let mut response = String::new(); + request.read_to_string(&mut response).unwrap(); + assert!(response.starts_with("HTTP/1.1 200"), "{response}"); + assert!(response.contains("request finished"), "{response}"); + assert!(server.wait_for_exit().success(), "{}", server.logs()); + assert!(server.logs().contains("Closing all database connections")); + // A supervisor can restart immediately on the same address. + TcpListener::bind(address).unwrap(); + } +} + +#[test] +fn bind_failure_is_nonzero_and_never_reports_ready() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let mut server = Server::start( + &serde_json::json!({"listen_on": listener.local_addr().unwrap().to_string(), "database_url": "sqlite::memory:"}), + "SELECT 'text' AS component;", + ); + assert!(!server.wait_for_exit().success()); + server.notifications.set_nonblocking(true).unwrap(); + let mut buffer = [0; 2048]; + while let Ok(size) = server.notifications.recv(&mut buffer) { + assert!(!String::from_utf8_lossy(&buffer[..size]).contains("READY=1")); + } +} + +#[test] +fn termination_during_database_startup_exits_cleanly() { + let database = TcpListener::bind("127.0.0.1:0").unwrap(); + let mut server = Server::start( + &serde_json::json!({"listen_on": "127.0.0.1:0", "database_url": format!("postgres://user:pass@{}/test", database.local_addr().unwrap())}), + "SELECT 'text' AS component;", + ); + server.wait_for("STATUS=Connecting to the database"); + server.signal("-TERM"); + server.wait_for("STOPPING=1"); + assert!(server.wait_for_exit().success(), "{}", server.logs()); +} From 130967ecae1d38caeb6b51197949f57082f0f9d7 Mon Sep 17 00:00:00 2001 From: lovasoa Date: Mon, 28 Sep 2026 03:05:35 +0200 Subject: [PATCH 2/4] Fix Windows event log SID pointer type --- src/telemetry/windows_event_log.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/telemetry/windows_event_log.rs b/src/telemetry/windows_event_log.rs index 736789bd6..a637c06b0 100644 --- a/src/telemetry/windows_event_log.rs +++ b/src/telemetry/windows_event_log.rs @@ -32,7 +32,7 @@ pub fn write(level: tracing::Level, message: &str) { event_type, 0, 0, - std::ptr::null(), + std::ptr::null_mut(), 1, 0, strings.as_ptr(), From d6da4c638497621309d1b70f4c8364c955d3965f Mon Sep 17 00:00:00 2001 From: lovasoa Date: Mon, 28 Sep 2026 03:14:24 +0200 Subject: [PATCH 3/4] Keep Windows validation scoped to service integration --- .github/workflows/ci.yml | 2 -- 1 file changed, 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 64662aab3..63a31c1ee 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -183,8 +183,6 @@ jobs: - name: Test native Windows service lifecycle shell: powershell run: scripts/test-windows-service.ps1 - - name: Lint Windows service code - run: cargo clippy --all-targets -- -D warnings - name: Upload Windows binary uses: actions/upload-artifact@v7 with: From e70590791c8d7084dfe23878fb07c8237a17b1a6 Mon Sep 17 00:00:00 2001 From: lovasoa Date: Mon, 28 Sep 2026 03:31:13 +0200 Subject: [PATCH 4/4] Avoid server specialization and queue Windows service logs --- .../sqlpage/migrations/66_log_component.sql | 2 +- .../your-first-sql-website/service.md | 7 + scripts/test-windows-service.ps1 | 8 +- src/service/windows.rs | 6 +- src/telemetry.rs | 6 +- src/telemetry/background_log.rs | 198 ++++++++++++++++++ src/telemetry/windows_event_log.rs | 130 ++++++++---- src/webserver/http.rs | 8 +- 8 files changed, 321 insertions(+), 44 deletions(-) create mode 100644 src/telemetry/background_log.rs diff --git a/examples/official-site/sqlpage/migrations/66_log_component.sql b/examples/official-site/sqlpage/migrations/66_log_component.sql index 1b2e21bc0..c8e2ec6c9 100644 --- a/examples/official-site/sqlpage/migrations/66_log_component.sql +++ b/examples/official-site/sqlpage/migrations/66_log_component.sql @@ -8,7 +8,7 @@ Use it to track what happens and troubleshoot issues. - Running from a terminal (Linux, macOS, or Windows PowerShell/Command Prompt): they show up in the window. - Docker: run `docker logs `. - Linux service (systemd): run `journalctl -u sqlpage`. -- Native Windows service (`--service NAME`): open Event Viewer → Windows Logs → Application and select the SQLPage source. See the [service setup guide](/your-first-sql-website/service.sql). +- Native Windows service (`--service NAME`): open Event Viewer → Windows Logs → Application and select the SQLPage source. Logs are queued in the background; sustained overload can drop records, and messages are limited to 16 KiB. See the [service setup guide](/your-first-sql-website/service.sql). - Outside native Windows service mode, this component''s output is written to [standard error (stderr)](https://en.wikipedia.org/wiki/Standard_streams#Standard_error_(stderr)). SQLPage request access logs are separate and are written to standard output (stdout). '); diff --git a/examples/official-site/your-first-sql-website/service.md b/examples/official-site/your-first-sql-website/service.md index 36b7dc800..cdcb3403e 100644 --- a/examples/official-site/your-first-sql-website/service.md +++ b/examples/official-site/your-first-sql-website/service.md @@ -89,6 +89,13 @@ shutdown controls initiate cleanup. Failures report service-specific exit code 1 details and application logs appear in **Event Viewer → Windows Logs → Application** under the **SQLPage** source. Existing OpenTelemetry export remains available. +Event Viewer logging uses a background writer so Windows log writes do not block +HTTP workers. The queue holds up to 256 records, each truncated to 16 KiB at a UTF-8 +character boundary. If the writer cannot keep up, new records are dropped and a +warning reports the number lost when the writer progresses or during shutdown. +Accepted records are flushed before the service reports `STOPPED`. Use `LOG_LEVEL` +or `RUST_LOG` to reduce log volume if needed. + To remove the service: ```powershell diff --git a/scripts/test-windows-service.ps1 b/scripts/test-windows-service.ps1 index 62767a139..c6710c5c5 100644 --- a/scripts/test-windows-service.ps1 +++ b/scripts/test-windows-service.ps1 @@ -7,6 +7,7 @@ $root = Join-Path ([IO.Path]::GetTempPath()) ("SQLPage service test " + $name) $created = $false $listener = $null $client = $null +$startedAt = Get-Date function Wait-State([string]$state) { $service = Get-Service $name @@ -25,7 +26,7 @@ try { $configuration = @{ listen_on = "127.0.0.1:$port"; database_url = 'sqlite::memory:' } | ConvertTo-Json [IO.File]::WriteAllText((Join-Path $root 'sqlpage\sqlpage.json'), $configuration) - [IO.File]::WriteAllText((Join-Path $root 'index.sql'), "SELECT 'text' AS component, 'service ready' AS contents;") + [IO.File]::WriteAllText((Join-Path $root 'index.sql'), "SELECT 'log' AS component, '$name flush marker' AS message; SELECT 'text' AS component, 'service ready' AS contents;") $command = '"{0}" --service {1} --web-root "{2}"' -f $Binary, $name, $root New-Service -Name $name -BinaryPathName $command -StartupType Manual | Out-Null $created = $true @@ -68,11 +69,14 @@ try { Wait-State 'Stopped' $status = Get-CimInstance Win32_Service -Filter "Name='$name'" if ($status.ExitCode -ne 0) { throw 'Clean stop reported a failure' } + $events = Get-WinEvent -FilterHashtable @{ LogName = 'Application'; ProviderName = 'SQLPage'; StartTime = $startedAt } + $flushed = $events | Where-Object { $_.Properties[0].Value -like "*$name flush marker*" } + if (-not $flushed) { throw 'The service stopped before its queued log record was written' } Start-Service $name Wait-State 'Running' Stop-Service $name Wait-State 'Stopped' - Write-Host 'Windows service startup failure, readiness, graceful stop, and restart passed.' + Write-Host 'Windows service startup failure, readiness, graceful stop, log flush, and restart passed.' } finally { if ($client) { $client.Dispose() } if ($listener) { $listener.Stop() } diff --git a/src/service/windows.rs b/src/service/windows.rs index 960676d98..09a12d8e0 100644 --- a/src/service/windows.rs +++ b/src/service/windows.rs @@ -33,7 +33,8 @@ pub(crate) fn dispatch(cli: Cli) -> anyhow::Result<()> { fn service_main(_arguments: Vec) { if let Err(error) = run_service() { - sqlpage::telemetry::windows_event_log::write(tracing::Level::ERROR, &format!("{error:#}")); + sqlpage::telemetry::windows_event_log::write(tracing::Level::ERROR, format!("{error:#}")); + sqlpage::telemetry::windows_event_log::shutdown(); *RESULT.lock().unwrap() = Some(Err(error)); } } @@ -78,8 +79,9 @@ fn run_service() -> anyhow::Result<()> { })(); let failed = result.is_err(); if let Err(error) = &result { - sqlpage::telemetry::windows_event_log::write(tracing::Level::ERROR, &format!("{error:#}")); + sqlpage::telemetry::windows_event_log::write(tracing::Level::ERROR, format!("{error:#}")); } + sqlpage::telemetry::windows_event_log::shutdown(); // Publish the result before reporting Stopped: the dispatcher may return as // soon as SCM receives that status. No work may remain after this call. *RESULT.lock().unwrap() = Some(result); diff --git a/src/telemetry.rs b/src/telemetry.rs index 4e7bf3d85..5e8dbfe60 100644 --- a/src/telemetry.rs +++ b/src/telemetry.rs @@ -189,9 +189,13 @@ pub fn init_telemetry() -> anyhow::Result { /// Initializes service logging in the Windows Application event log, with optional OTLP export. #[cfg(windows)] pub fn init_windows_service_telemetry() -> anyhow::Result { + windows_event_log::init()?; init_telemetry_with_log_layer(logfmt::LogfmtLayer::windows_service()) } +#[cfg(any(windows, test))] +mod background_log; + #[cfg(windows)] pub mod windows_event_log; @@ -548,7 +552,7 @@ mod logfmt { eprint!("{buf}"); } #[cfg(windows)] - OutputMode::WindowsEventLog => super::windows_event_log::write(level, &buf), + OutputMode::WindowsEventLog => super::windows_event_log::write(level, buf), } } } diff --git a/src/telemetry/background_log.rs b/src/telemetry/background_log.rs new file mode 100644 index 000000000..9ce584f8b --- /dev/null +++ b/src/telemetry/background_log.rs @@ -0,0 +1,198 @@ +//! Bounded, non-blocking submission to a blocking platform log writer. + +use std::{ + io, + sync::{ + Arc, Mutex, + atomic::{AtomicU64, Ordering}, + mpsc::{RecvTimeoutError, SyncSender, sync_channel}, + }, + thread::{self, JoinHandle}, + time::{Duration, Instant}, +}; +use tracing::Level; + +const QUEUE_CAPACITY: usize = 256; +const MAX_MESSAGE_BYTES: usize = 16 * 1024; + +enum Message { + Record(Level, String), + Shutdown, +} + +pub(super) struct BackgroundLog { + sender: SyncSender, + dropped: Arc, + worker: Mutex>>, +} + +impl BackgroundLog { + pub(super) fn start(make_writer: F) -> io::Result + where + F: FnOnce() -> io::Result + Send + 'static, + W: FnMut(Level, &str) + 'static, + { + let (sender, receiver) = sync_channel(QUEUE_CAPACITY); + let (started_tx, started_rx) = sync_channel(1); + let dropped = Arc::new(AtomicU64::new(0)); + let worker_dropped = Arc::clone(&dropped); + let worker = thread::Builder::new() + .name("sqlpage-event-log".into()) + .spawn(move || { + // Construct the writer on its own thread so native handles never + // need to be shared with HTTP workers. + let mut writer = match make_writer() { + Ok(writer) => writer, + Err(error) => { + let _ = started_tx.send(Err(error)); + return; + } + }; + let _ = started_tx.send(Ok(())); + let mut last_report = Instant::now(); + loop { + match receiver.recv_timeout(Duration::from_secs(5)) { + Ok(Message::Record(level, message)) => writer(level, &message), + Ok(Message::Shutdown) | Err(RecvTimeoutError::Disconnected) => break, + Err(RecvTimeoutError::Timeout) => {} + } + if last_report.elapsed() >= Duration::from_secs(5) { + report_dropped(&worker_dropped, &mut writer); + last_report = Instant::now(); + } + } + report_dropped(&worker_dropped, &mut writer); + })?; + if let Err(error) = started_rx + .recv() + .map_err(io::Error::other) + .and_then(|result| result) + { + let _ = worker.join(); + return Err(error); + } + Ok(Self { + sender, + dropped, + worker: Mutex::new(Some(worker)), + }) + } + + pub(super) fn write(&self, level: Level, mut message: String) { + // Bound both the length and retained allocation, including multibyte text. + let mut end = MAX_MESSAGE_BYTES.min(message.len()); + while !message.is_char_boundary(end) { + end -= 1; + } + message.truncate(end); + if message.capacity() > MAX_MESSAGE_BYTES { + message.shrink_to(MAX_MESSAGE_BYTES); + } + if self + .sender + .try_send(Message::Record(level, message)) + .is_err() + { + self.dropped.fetch_add(1, Ordering::Relaxed); + } + } + + /// Call only after producers have stopped. This can block while flushing. + pub(super) fn shutdown(&self) { + let worker = self.worker.lock().unwrap().take(); + if let Some(worker) = worker { + let _ = self.sender.send(Message::Shutdown); + let _ = worker.join(); + } + } +} + +fn report_dropped(dropped: &AtomicU64, writer: &mut impl FnMut(Level, &str)) { + let count = dropped.swap(0, Ordering::Relaxed); + if count != 0 { + writer( + Level::WARN, + &format!("SQLPage dropped {count} log records because the Event Log queue was full"), + ); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn slow_writer_cannot_block_producers_and_shutdown_flushes_the_queue() { + let (entered_tx, entered_rx) = sync_channel(1); + let (release_tx, release_rx) = sync_channel(1); + let records = Arc::new(Mutex::new(Vec::new())); + let saved = Arc::clone(&records); + let logger = Arc::new( + BackgroundLog::start(move || { + Ok(move |_level, message: &str| { + if message == "first" { + entered_tx.send(()).unwrap(); + release_rx.recv().unwrap(); + } + saved.lock().unwrap().push(message.to_owned()); + }) + }) + .unwrap(), + ); + logger.write(Level::INFO, "first".into()); + entered_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + // A timeout makes a regression to blocking submissions fail, not hang. + let producer_log = Arc::clone(&logger); + let (done_tx, done_rx) = sync_channel(1); + let producer = thread::spawn(move || { + for index in 0..QUEUE_CAPACITY + 3 { + producer_log.write(Level::INFO, format!("record {index}")); + } + done_tx.send(()).unwrap(); + }); + let submitted = done_rx.recv_timeout(Duration::from_secs(5)); + release_tx.send(()).unwrap(); + producer.join().unwrap(); + logger.shutdown(); + logger.shutdown(); + submitted.expect("log submission blocked on the writer"); + let records = records.lock().unwrap(); + assert_eq!(records.len(), QUEUE_CAPACITY + 2); + assert_eq!(records[0], "first"); + assert_eq!( + records[QUEUE_CAPACITY], + format!("record {}", QUEUE_CAPACITY - 1) + ); + assert!(records.last().unwrap().contains("dropped 3 log records")); + } + + #[test] + fn messages_are_bounded_without_splitting_utf8() { + let (tx, rx) = sync_channel(1); + let logger = BackgroundLog::start(move || { + Ok(move |_, message: &str| { + tx.send(message.to_owned()).unwrap(); + }) + }) + .unwrap(); + logger.write(Level::INFO, "€".repeat(MAX_MESSAGE_BYTES)); + let message = rx.recv_timeout(Duration::from_secs(5)).unwrap(); + assert_eq!(message.len(), MAX_MESSAGE_BYTES - 1); + assert!(message.chars().all(|character| character == '€')); + logger.shutdown(); + } + + #[test] + fn writer_initialization_errors_are_returned() { + let result = BackgroundLog::start(|| -> io::Result { + Err(io::Error::new( + io::ErrorKind::PermissionDenied, + "access denied", + )) + }); + assert_eq!( + result.err().unwrap().kind(), + io::ErrorKind::PermissionDenied + ); + } +} diff --git a/src/telemetry/windows_event_log.rs b/src/telemetry/windows_event_log.rs index a637c06b0..dad8a3ece 100644 --- a/src/telemetry/windows_event_log.rs +++ b/src/telemetry/windows_event_log.rs @@ -1,43 +1,103 @@ //! Native service logs, including failures before the tracing subscriber exists. -use windows_sys::Win32::System::EventLog::{ - DeregisterEventSource, EVENTLOG_ERROR_TYPE, EVENTLOG_INFORMATION_TYPE, EVENTLOG_WARNING_TYPE, - RegisterEventSourceW, ReportEventW, +use super::background_log::BackgroundLog; +use std::{io, sync::OnceLock}; +use windows_sys::Win32::{ + Foundation::HANDLE, + System::EventLog::{ + DeregisterEventSource, EVENTLOG_ERROR_TYPE, EVENTLOG_INFORMATION_TYPE, + EVENTLOG_WARNING_TYPE, RegisterEventSourceW, ReportEventW, + }, }; -/// Writes one event under the `SQLPage` source in the Application log. -pub fn write(level: tracing::Level, message: &str) { - let event_type = match level { - tracing::Level::ERROR => EVENTLOG_ERROR_TYPE, - tracing::Level::WARN => EVENTLOG_WARNING_TYPE, - _ => EVENTLOG_INFORMATION_TYPE, - }; - // ReportEvent limits insertion strings to 31,839 UTF-16 code units. - let message: Vec = message - .encode_utf16() - .take(31_838) - .map(|unit| if unit == 0 { 0xfffd } else { unit }) - .chain(Some(0)) - .collect(); - let strings = [message.as_ptr()]; - // SAFETY: All pointers refer to live, NUL-terminated strings. The returned - // handle is used only here and is released after the synchronous write. - unsafe { - let handle = RegisterEventSourceW(std::ptr::null(), windows_sys::w!("SQLPage")); +static LOGGER: OnceLock> = OnceLock::new(); + +/// Opens the event source on a dedicated writer thread before serving requests. +pub(super) fn init() -> anyhow::Result<()> { + logger().map(|_| ()).map_err(anyhow::Error::msg) +} + +fn logger() -> Result<&'static BackgroundLog, &'static str> { + LOGGER + .get_or_init(|| { + BackgroundLog::start(|| { + let source = EventSource::open()?; + Ok(move |level, message: &str| source.write(level, message)) + }) + .map_err(|error| error.to_string()) + }) + .as_ref() + .map_err(String::as_str) +} + +/// Queues one event without waiting for the Windows Event Log service. +/// Overload drops records with a warning summary; shutdown flushes accepted records. +pub fn write(level: tracing::Level, message: String) { + match logger() { + Ok(logger) => logger.write(level, message), + Err(error) => eprintln!("Unable to initialize Windows service logging: {error}: {message}"), + } +} + +/// Flushes queued records and releases the event source after producers have stopped. +/// This can block and must be called outside HTTP worker threads. +pub fn shutdown() { + if let Some(Ok(logger)) = LOGGER.get() { + logger.shutdown(); + } +} + +// Created, used, and dropped exclusively on the log writer thread. +struct EventSource(HANDLE); + +impl EventSource { + fn open() -> io::Result { + // SAFETY: Both arguments are valid constant pointers. A successful handle + // is owned by EventSource and released in Drop. + let handle = unsafe { RegisterEventSourceW(std::ptr::null(), windows_sys::w!("SQLPage")) }; if handle.is_null() { - return; + Err(io::Error::last_os_error()) + } else { + Ok(Self(handle)) + } + } + + fn write(&self, level: tracing::Level, message: &str) { + let event_type = match level { + tracing::Level::ERROR => EVENTLOG_ERROR_TYPE, + tracing::Level::WARN => EVENTLOG_WARNING_TYPE, + _ => EVENTLOG_INFORMATION_TYPE, + }; + // Queued messages are bounded to 16 KiB, below ReportEvent's limit. + let message: Vec = message + .encode_utf16() + .map(|unit| if unit == 0 { 0xfffd } else { unit }) + .chain(Some(0)) + .collect(); + let strings = [message.as_ptr()]; + // SAFETY: The event source remains live and all strings are NUL-terminated + // and valid for the duration of this synchronous call. + unsafe { + ReportEventW( + self.0, + event_type, + 0, + 0, + std::ptr::null_mut(), + 1, + 0, + strings.as_ptr(), + std::ptr::null(), + ); + } + } +} + +impl Drop for EventSource { + fn drop(&mut self) { + // SAFETY: This is the sole owner of the handle returned by registration. + unsafe { + DeregisterEventSource(self.0); } - ReportEventW( - handle, - event_type, - 0, - 0, - std::ptr::null_mut(), - 1, - 0, - strings.as_ptr(), - std::ptr::null(), - ); - DeregisterEventSource(handle); } } diff --git a/src/webserver/http.rs b/src/webserver/http.rs index 4b5c8ac07..240295b13 100644 --- a/src/webserver/http.rs +++ b/src/webserver/http.rs @@ -624,7 +624,7 @@ fn default_headers() -> middleware::DefaultHeaders { } pub async fn run_server(config: &AppConfig, state: AppState) -> anyhow::Result<()> { - run_server_inner(config, state, None, || Ok(())).await + run_server_inner(config, state, None, Box::new(|| Ok(()))).await } /// Runs the server with externally managed shutdown and readiness reporting. @@ -636,14 +636,16 @@ pub async fn run_server_with_shutdown( shutdown: impl Future + Send + 'static, on_ready: impl FnOnce() -> anyhow::Result<()>, ) -> anyhow::Result<()> { - run_server_inner(config, state, Some(Box::pin(shutdown)), on_ready).await + run_server_inner(config, state, Some(Box::pin(shutdown)), Box::new(on_ready)).await } async fn run_server_inner( config: &AppConfig, state: AppState, shutdown: Option + Send>>>, - on_ready: impl FnOnce() -> anyhow::Result<()>, + // Erase the callback type at this boundary: otherwise every caller causes + // another specialization of the entire HTTP server and middleware stack. + on_ready: Box anyhow::Result<()> + '_>, ) -> anyhow::Result<()> { let listen_on = config.listen_on(); let state = web::Data::new(state);