fix(linux-session): replay a read the reconnect stopped, test a restart

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.
 

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();

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<Vec<LinkGeneration>>,
+) -> Result<Value, OperationError> {
+    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<Vec<LinkGeneration>>,
+) -> 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);
+}

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,