diff --git a/linux/ARCHITECTURE.md b/linux/ARCHITECTURE.md index 903242489a..6e6afd9029 100644 --- a/linux/ARCHITECTURE.md +++ b/linux/ARCHITECTURE.md @@ -259,7 +259,7 @@ On Linux the system store is the distribution's CA bundle. Under Flatpak it is t ### Sessions and health -A window holds one `Session`. Every use of its connection goes through the `OperationRunner`, which adds no timer of its own: the deadline travels in `CallOptions` so the driver stops the statement on the server. The session's monitor probes every 30 s, retries an unanswered probe after 10 s, and calls the connection lost after three unanswered probes in a row, when a probe finds the link gone, or as soon as `Connection::lost()` fires, which a driver resolves only when the link is gone, never for one dead backend in its pool. It then reconnects on a backoff from 5 s to 60 s. A failure that a retry cannot fix (authentication, TLS, configuration, a host key to review) parks the session until the user acts. A read that failed because its connection was lost runs once more on the replacement link; a save never does. +A window holds one `Session`. Every use of its connection goes through the `OperationRunner`, which adds no timer of its own: the deadline travels in `CallOptions` so the driver stops the statement on the server. The session's monitor probes every 30 s, retries an unanswered probe after 10 s, and calls the connection lost after three unanswered probes in a row, when a probe finds the link gone, or as soon as `Connection::lost()` fires, which a driver resolves only when the link is gone, never for one dead backend in its pool. It then reconnects on a backoff from 5 s to 60 s. A failure that a retry cannot fix (authentication, TLS, configuration, a host key to review) parks the session until the user acts. A read that failed because its connection was lost runs once more on the replacement link, as does one stopped when the reconnect closed the connection it ran on; a save never does. The application tells each session what its connection cannot see. When `GNetworkMonitor` reports a change, and when a window adopts a session, it asks whether that session's server, or its first SSH hop, is reachable; a file, a socket and a loopback server are never asked about. A session that loses a server the network cannot reach waits offline, checking every 60 s, and reconnects as soon as the server is reachable again. While a session is open, a 5 s timer compares `CLOCK_BOOTTIME` with `CLOCK_MONOTONIC`, and after a suspend each server is checked again and a session that slept more than 30 s probes its connection. A session parked on an SSH host key keeps the error that parked it, and its banner's Review Key shows the fingerprint. Trusting it retries with ssh's question about that one key answered yes, so ssh still checks the key and writes known_hosts; a changed key is first removed with `ssh-keygen -R`, since ssh never accepts one. When the review cannot accept the key, because ssh's configuration takes keys only from known_hosts or the key is revoked, its dialog offers Retry for after the user fixes the key outside TablePro. diff --git a/linux/crates/session/src/operation/operation_runner.rs b/linux/crates/session/src/operation/operation_runner.rs index f005da7f9e..4fee973cdd 100644 --- a/linux/crates/session/src/operation/operation_runner.rs +++ b/linux/crates/session/src/operation/operation_runner.rs @@ -1,7 +1,7 @@ use std::future::Future; use std::sync::Arc; -use tablepro_core::{CallOptions, DriverError, ErrorCategory}; +use tablepro_core::{CallOptions, DriverError, ErrorCategory, LossPhase}; use tokio::sync::watch; use tokio::time::Instant; use tokio_util::sync::CancellationToken; @@ -136,6 +136,13 @@ impl OperationRunner { let error = match error { OperationError::Driver(DriverError::Cancelled) if cancel.is_cancelled() => OperationError::Cancelled, OperationError::Driver(DriverError::Cancelled) if self.cancel.is_cancelled() => OperationError::Closed, + // Nobody here asked for the stop: a reconnect retired the + // connection, and closing it stopped the call. + OperationError::Driver(DriverError::Cancelled) if self.link.generation() > generation => { + OperationError::Driver(DriverError::ConnectionLost { + during: LossPhase::Statement, + }) + } other => other, }; if error.category() == Some(ErrorCategory::ConnectionLost) { @@ -158,7 +165,7 @@ mod tests { use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; - use tablepro_core::{LivenessPolicy, LossPhase, WriteKind}; + use tablepro_core::{LivenessPolicy, WriteKind}; use tablepro_test_support::{ConnectionScript, FakeConnection, FakeReply}; use super::*; @@ -268,6 +275,38 @@ mod tests { assert_eq!(calls, 2); } + #[tokio::test(start_paused = true)] + async fn a_read_its_retired_connection_stopped_replays_on_the_new_one() { + let harness = harness(); + let stopped = OperationError::Driver(DriverError::Cancelled); + + let (read, read_calls) = run_failing_after_replace(&harness, OperationKind::Count, stopped.clone()).await; + let (save, save_calls) = + run_failing_after_replace(&harness, OperationKind::Save(WriteKind::RowChanges), stopped).await; + + assert_eq!(read, Ok(LinkGeneration::FIRST.next())); + assert_eq!(read_calls, 2); + assert_eq!(save, Err(lost()), "the save was not told its connection went away"); + assert_eq!(save_calls, 1); + } + + #[tokio::test(start_paused = true)] + async fn a_stop_on_the_current_connection_is_not_replayed() { + let harness = harness(); + let calls = AtomicUsize::new(0); + + let outcome = harness + .runner + .run_cancellable(OperationKind::Page, &CancellationToken::new(), |_, _| { + calls.fetch_add(1, Ordering::SeqCst); + async { Err::<(), _>(OperationError::Driver(DriverError::Cancelled)) } + }) + .await; + + assert_eq!(outcome, Err(OperationError::Driver(DriverError::Cancelled))); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + #[tokio::test(start_paused = true)] async fn a_replay_runs_once_even_when_it_fails_the_same_way() { let harness = harness(); diff --git a/linux/crates/session/tests/session_postgres.rs b/linux/crates/session/tests/session_postgres.rs index 726d76161a..8ab79beddb 100644 --- a/linux/crates/session/tests/session_postgres.rs +++ b/linux/crates/session/tests/session_postgres.rs @@ -6,23 +6,24 @@ )] use std::path::{Path, PathBuf}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::Duration; use secrecy::SecretString; use tablepro_core::credentials::CredentialInteraction; -use tablepro_core::statement::{Statement, StatementBuilder}; +use tablepro_core::statement::{GuardedStatement, Statement, StatementBuilder}; use tablepro_core::{ CallOptions, ConnectEnvironment, ConnectOptions, ConnectTarget, Connection, Credentials, DatabaseDriver, - DriverError, LivenessPolicy, NetworkEndpoint, NetworkTarget, RowLimit, TlsConfig, Value, WorkspaceSettings, + DriverError, ErrorCategory, ForeignKeyMode, LivenessPolicy, NetworkEndpoint, NetworkTarget, RowLimit, TableRef, + TlsConfig, Value, WorkspaceSettings, WriteBatch, WriteKind, }; use tablepro_driver_postgres::{PgDriver, WORKSPACE_POOL_MAX_SIZE}; use tablepro_net::DirectTcpTransport; use tablepro_session::runtime::Tasks; use tablepro_session::{ ConnectRequest, ConnectServices, ConnectionLease, ConnectionMetadata, Connector, DefaultDeadlines, DriverConnector, - MonitorSignal, NoStoredCredentials, OperationError, OperationKind, PendingSession, RequestTarget, Session, - SessionDeps, SessionHealth, + LinkGeneration, MonitorSignal, NoStoredCredentials, OperationError, OperationKind, PendingSession, RequestTarget, + Session, SessionDeps, SessionHealth, }; use tablepro_ssh::{SshAuth, SshConfig, SshDestination, SshRuntimeCell, SshServices}; use tablepro_test_fixtures::{ @@ -417,3 +418,173 @@ async fn session_through_openssh_reconnects_after_the_master_dies() { let report = session.begin_shutdown().await.expect("the close task"); assert_eq!(report, tablepro_session::CloseReport::Closed); } + +/// The table the restart test reads and writes. +fn notes() -> TableRef { + TableRef::new(Some("public".to_owned()), "notes") +} + +/// The notes table with one row, written outside the session. +async fn create_notes(fixture: &PostgresFixture) { + let setup = admin(fixture).await; + for sql in [ + "CREATE TABLE notes (id integer PRIMARY KEY, body text NOT NULL)", + "INSERT INTO notes VALUES (1, 'before the restart')", + ] { + let batch = WriteBatch::new( + WriteKind::SchemaChanges, + notes(), + vec![GuardedStatement::unguarded(statement(sql))], + ForeignKeyMode::Enforced, + ); + setup.write_batch(&batch, within(WAIT)).await.expect("the setup"); + } + setup.close().await.expect("the setup connection closed"); +} + +/// Count the notes. On the connection the session had before the +/// restart the count waits 30 seconds first, as a read still running +/// when the reconnect lands. +async fn count_notes( + lease: ConnectionLease, + options: CallOptions, + before: LinkGeneration, + leases: &Mutex>, +) -> Result { + leases.lock().expect("the lease log").push(lease.generation()); + let sql = match lease.generation() == before { + true => "SELECT count(*) FROM notes, pg_sleep(30)", + false => "SELECT count(*) FROM notes", + }; + let result = lease + .query_bound(&statement(sql), RowLimit::EDITOR_DEFAULT, options) + .await?; + Ok(result + .rows + .into_iter() + .next() + .and_then(|row| row.into_iter().next()) + .unwrap_or(Value::Null)) +} + +/// Add a note. On the connection from before the restart the save +/// holds its transaction open for 30 seconds before it commits. +async fn add_note( + lease: ConnectionLease, + options: CallOptions, + before: LinkGeneration, + leases: &Mutex>, +) -> Result<(), OperationError> { + leases.lock().expect("the lease log").push(lease.generation()); + let mut steps = vec![GuardedStatement::unguarded(statement( + "INSERT INTO notes (id, body) VALUES (2, 'saved across the restart')", + ))]; + if lease.generation() == before { + steps.push(GuardedStatement::unguarded(statement("SELECT pg_sleep(30)"))); + } + let batch = WriteBatch::new(WriteKind::RowChanges, notes(), steps, ForeignKeyMode::Enforced); + lease.write_batch(&batch, options).await?; + Ok(()) +} + +/// The database restarts under a session. A read made while it is down +/// fails, since no other connection exists yet to run it on. A read +/// still running on the old connection when the reconnect lands runs +/// once more on the new one; a save does not, so its row is never +/// written. The server comes back on the same host port with its data, +/// and the session is online on the new connection. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires docker"] +async fn a_database_restart_replays_reads_and_never_writes() { + let fixture = PostgresFixture::start_restartable(HbaMode::Password) + .await + .expect("PostgreSQL"); + let connector = connector(isolated_ssh().services); + let session = start( + connector.clone(), + establish(&connector, request(direct_target(&fixture))).await, + ); + create_notes(&fixture).await; + let before = session.link_generation(); + let mut health = session.health(); + + fixture.stop().await.expect("the server stopped"); + let during = scalar(&session, OperationKind::Count, "SELECT count(*) FROM notes").await; + assert!( + matches!( + during.as_ref().map_err(OperationError::category), + Err(Some(ErrorCategory::ConnectionLost | ErrorCategory::Network)) + ), + "a read while the server was down gave {during:?}" + ); + // Two failed attempts put the next one ten seconds out, which + // leaves the server time to come back before the reconnect does. + tokio::time::timeout( + WAIT, + health.wait_for(|health| matches!(health, SessionHealth::Reconnecting { attempt } if *attempt >= 3)), + ) + .await + .expect("the session kept retrying") + .expect("the health sender is alive"); + fixture.start_again().await.expect("the server started again"); + assert_eq!( + session.link_generation(), + before, + "the reconnect landed before the reads were made" + ); + + let reads = Arc::new(Mutex::new(Vec::new())); + let reading = { + let session = session.clone(); + let reads = reads.clone(); + tasks().spawn_task(async move { + session + .run(OperationKind::Count, |lease, options| { + count_notes(lease, options, before, &reads) + }) + .await + }) + }; + let saves = Arc::new(Mutex::new(Vec::new())); + let saving = { + let session = session.clone(); + let saves = saves.clone(); + tasks().spawn_task(async move { + session + .run(OperationKind::Save(WriteKind::RowChanges), |lease, options| { + add_note(lease, options, before, &saves) + }) + .await + }) + }; + let read = reading.await.expect("the read task"); + let saved = saving.await.expect("the save task"); + + let after = session.link_generation(); + assert!(after > before, "the session never reconnected"); + assert_eq!(read, Ok(Value::Int(1)), "the read was not answered"); + assert_eq!( + *reads.lock().expect("the lease log"), + [before, after], + "the read did not run once more on the new connection" + ); + assert!(saved.is_err(), "a save stopped by the reconnect reported {saved:?}"); + assert_eq!(*saves.lock().expect("the lease log"), [before], "the save ran again"); + assert_eq!( + scalar( + &session, + OperationKind::Count, + "SELECT count(*) FROM notes WHERE id = 2" + ) + .await, + Ok(Value::Int(0)), + "the save's row was written" + ); + assert_eq!( + scalar(&session, OperationKind::Metadata, "SELECT body FROM notes WHERE id = 1").await, + Ok(Value::Text("before the restart".to_owned())) + ); + assert_eq!(*session.health().borrow(), SessionHealth::Online); + let report = session.begin_shutdown().await.expect("the close task"); + assert_eq!(report, tablepro_session::CloseReport::Closed); +} diff --git a/linux/crates/test-fixtures/src/fixture_error.rs b/linux/crates/test-fixtures/src/fixture_error.rs index 46cfa6b571..3e025139aa 100644 --- a/linux/crates/test-fixtures/src/fixture_error.rs +++ b/linux/crates/test-fixtures/src/fixture_error.rs @@ -16,6 +16,12 @@ pub enum FixtureError { }, #[error("{service} did not accept connections within {seconds} seconds")] NotReady { service: &'static str, seconds: u64 }, + #[error("{service} came back on host port {after}, not {before}")] + PortMoved { + service: &'static str, + before: u16, + after: u16, + }, #[error("{service} was not ready: {reason}; last reply: {last_reply}; container {state}\n{logs}")] ContainerNotReady { service: &'static str, diff --git a/linux/crates/test-fixtures/src/postgres_fixture.rs b/linux/crates/test-fixtures/src/postgres_fixture.rs index f6ee0b2a30..e9c47e3d1f 100644 --- a/linux/crates/test-fixtures/src/postgres_fixture.rs +++ b/linux/crates/test-fixtures/src/postgres_fixture.rs @@ -1,6 +1,11 @@ +use std::net::TcpListener; +use std::time::Duration; + +use testcontainers::core::IntoContainerPort; use testcontainers::runners::AsyncRunner; use testcontainers::{ContainerAsync, ImageExt}; use testcontainers_modules::postgres::Postgres; +use tokio::time::Instant; use crate::container_file::with_files; use crate::{ @@ -15,6 +20,8 @@ const INSTALLED_KEY: &str = "/var/lib/postgresql/server.key"; const CERTIFICATE_ROLE_SQL: &str = "CREATE ROLE tablepro_client LOGIN;\n\ GRANT CONNECT ON DATABASE tablepro TO tablepro_client;\n\ GRANT USAGE, CREATE ON SCHEMA public TO tablepro_client;\n"; +const READY_TIMEOUT: Duration = Duration::from_secs(60); +const READY_RETRY: Duration = Duration::from_millis(250); pub struct PostgresFixture { container: ContainerAsync, @@ -32,22 +39,45 @@ impl PostgresFixture { pub const NETWORK_ROLE: &'static str = "postgres"; pub async fn start(hba: HbaMode) -> Result { - Self::launch(hba, PostgresVersion::default(), None, PkiAlgorithm::default()).await + Self::launch(hba, PostgresVersion::default(), None, PkiAlgorithm::default(), None).await } pub async fn start_in(hba: HbaMode, network: &TestNetwork) -> Result { - Self::launch(hba, PostgresVersion::default(), Some(network), PkiAlgorithm::default()).await + Self::launch( + hba, + PostgresVersion::default(), + Some(network), + PkiAlgorithm::default(), + None, + ) + .await } pub async fn start_version(version: PostgresVersion, hba: HbaMode) -> Result { - Self::launch(hba, version, None, PkiAlgorithm::default()).await + Self::launch(hba, version, None, PkiAlgorithm::default(), None).await + } + + /// A server on a host port picked here rather than by Docker, which + /// maps a published port anew each time a container starts. After + /// `stop` and `start_again` it answers where a connection made + /// before already points. + pub async fn start_restartable(hba: HbaMode) -> Result { + let port = TcpListener::bind(("0.0.0.0", 0))?.local_addr()?.port(); + Self::launch( + hba, + PostgresVersion::default(), + None, + PkiAlgorithm::default(), + Some(port), + ) + .await } /// A server whose whole PKI is of one key type, for the channel /// binding cases: the server hashes its certificate by the /// algorithm that signed it. pub async fn start_with_pki(hba: HbaMode, algorithm: PkiAlgorithm) -> Result { - Self::launch(hba, PostgresVersion::default(), None, algorithm).await + Self::launch(hba, PostgresVersion::default(), None, algorithm, None).await } async fn launch( @@ -55,6 +85,7 @@ impl PostgresFixture { version: PostgresVersion, network: Option<&TestNetwork>, algorithm: PkiAlgorithm, + host_port: Option, ) -> Result { let pki = TestPki::generate_with(algorithm)?; let password = FixtureCredentials::generate_password(); @@ -72,6 +103,9 @@ impl PostgresFixture { if let (Some(network), Some(alias)) = (network, &network_alias) { request = request.with_network(network.name()).with_container_name(alias); } + if let Some(host_port) = host_port { + request = request.with_mapped_port(host_port, CONTAINER_PORT.tcp()); + } let container = with_files(request, Self::files(hba, &pki)?).start().await?; let host = container.get_host().await?.to_string(); let port = container.get_host_port_ipv4(CONTAINER_PORT).await?; @@ -165,6 +199,53 @@ impl PostgresFixture { pub async fn exec(&self, command: &[&str]) -> Result { ExecOutput::run(&self.container, command).await } + + /// Shut the server down the way `docker stop` does. The container + /// and its data stay. + pub async fn stop(&self) -> Result<(), FixtureError> { + Ok(self.container.stop().await?) + } + + /// Start a stopped server again and wait until it accepts + /// connections. Fails when Docker mapped it to another host port, as + /// it does for a server not started by `start_restartable`. + pub async fn start_again(&self) -> Result<(), FixtureError> { + self.container.start().await?; + let port = self.container.get_host_port_ipv4(CONTAINER_PORT).await?; + if port != self.port { + return Err(FixtureError::PortMoved { + service: "PostgreSQL", + before: self.port, + after: port, + }); + } + let deadline = Instant::now() + READY_TIMEOUT; + loop { + let ready = self + .exec(&[ + "pg_isready", + "-h", + "127.0.0.1", + "-p", + "5432", + "-U", + PASSWORD_USER, + "-d", + DATABASE, + ]) + .await?; + if ready.succeeded() { + return Ok(()); + } + if Instant::now() >= deadline { + return Err(FixtureError::NotReady { + service: "PostgreSQL", + seconds: READY_TIMEOUT.as_secs(), + }); + } + tokio::time::sleep(READY_RETRY).await; + } + } } #[cfg(test)] diff --git a/linux/crates/test-fixtures/tests/fixtures.rs b/linux/crates/test-fixtures/tests/fixtures.rs index c809b011fb..f9eb029387 100644 --- a/linux/crates/test-fixtures/tests/fixtures.rs +++ b/linux/crates/test-fixtures/tests/fixtures.rs @@ -505,6 +505,28 @@ async fn postgres_fixture_runs_the_requested_major() -> TestResult { Ok(()) } +#[tokio::test] +#[ignore = "requires docker"] +async fn postgres_restartable_fixture_comes_back_on_its_port_with_its_data() -> TestResult { + let fixture = PostgresFixture::start_restartable(HbaMode::Password).await?; + let port = fixture.port(); + let psql = |sql: &'static str| ["psql", "-U", "tablepro", "-d", "tablepro", "-tAc", sql]; + let created = fixture + .exec(&psql("CREATE TABLE kept (n integer); INSERT INTO kept VALUES (7)")) + .await?; + assert!(created.succeeded(), "{}", created.stderr_text()); + + fixture.stop().await?; + assert!(!fixture.container().is_running().await?); + fixture.start_again().await?; + + let read = fixture.exec(&psql("SELECT n FROM kept")).await?; + assert_eq!(read.stdout_text().trim(), "7", "{}", read.stderr_text()); + assert_eq!(fixture.container().get_host_port_ipv4(5432).await?, port); + TcpStream::connect((fixture.host(), port)).await?; + Ok(()) +} + #[tokio::test] #[ignore = "requires docker"] async fn openssh_fixture_on_a_network_forwards_to_postgres_by_alias() -> TestResult { diff --git a/linux/docs/testing.md b/linux/docs/testing.md index 91ddab6297..609bfdd5f1 100644 --- a/linux/docs/testing.md +++ b/linux/docs/testing.md @@ -112,7 +112,7 @@ Every TLS fixture has two principals: SQL Server has only the `sa` principal, because its TLS contract carries no client identity. -`HbaMode` picks what the PostgreSQL fixture accepts: `Password` (the image defaults, no TLS), `HostSslOnly`, or `ClientCertificate`. `OpenSshFixture` takes an `SshAuthVariant`: `Password`, `PublicKey`, `KeyboardInteractive` or `HostCertificate`. `HostKeyRevocation` builds a KRL from the running server's host key, and `ScriptedAskpass` writes a 0700 script for `SSH_ASKPASS`. +`HbaMode` picks what the PostgreSQL fixture accepts: `Password` (the image defaults, no TLS), `HostSslOnly`, or `ClientCertificate`. Docker maps a published port anew each time a container starts, so a test that restarts the server starts it with `PostgresFixture::start_restartable`, which picks the host port itself; `stop` and `start_again` then bring it back where the session's connection points, with its data. `OpenSshFixture` takes an `SshAuthVariant`: `Password`, `PublicKey`, `KeyboardInteractive` or `HostCertificate`. `HostKeyRevocation` builds a KRL from the running server's host key, and `ScriptedAskpass` writes a 0700 script for `SSH_ASKPASS`. ### Conformance