Add Active -> Offline -> Faulted tests (#1257)

This is (hopefully) the final prep PR before addressing #1252.

There was an untested path of `Active` -(timeout)-> `Offline` -(too many
jobs)-> `Faulted`; this PR adds tests for both the job and byte-based
fault conditions. In addition, it confirms that live-repair works after
all fault-inducing tests, moving that logic to a new `async fn
run_live_repair(mut harness: TestHarness)`.

Making these tests pass required fixing a (probably innocuous) race
condition:

- Downstairs is marked as offline, begins to reconnect (with a 10s
pause)
- Too many jobs accumulate, and we try to stop the downstairs (to
restart it again)
- `ClientIoTask::run_inner` ignores the `stop` message during the
initial 10s pause
- After the pause completes, the client IO task connects then
immediately disconnects, because it finally looks at the stop message

I think this was only an issue in unit tests, where we manage
reconnections by hand; in the real system, the second reconnection
should work fine. Anyhow, I fixed it by adding the `stop` condition to
the initial client sleep and connection calls, so that it can interrupt
the client at any point.

Author: mkeeter

upstairs/src/client.rs

@@ -37,6 +37,10 @@ const TIMEOUT_SECS: f32 = 15.0;
 const TIMEOUT_LIMIT: usize = 3;
 const PING_INTERVAL_SECS: f32 = 5.0;
 
+/// Total time before a client is timed out
+#[cfg(test)]
+pub const CLIENT_TIMEOUT_SECS: f32 = TIMEOUT_SECS * TIMEOUT_LIMIT as f32;
+
 /// Handle to a running I/O task
 ///
 /// The I/O task is "thin"; it simply forwards messages around.  The task
@@ -2503,7 +2507,25 @@ impl ClientIoTask {
         // spinning (e.g. if something is fundamentally wrong with the
         // Downstairs)
         if self.delay {
-            tokio::time::sleep(std::time::Duration::from_secs(10)).await;
+            tokio::select! {
+                s = &mut self.stop => {
+                    warn!(self.log, "client IO task stopped during sleep");
+                    return match s {
+                        Ok(s) =>
+                            ClientRunResult::RequestedStop(s),
+                        Err(e) => {
+                            warn!(
+                                self.log,
+                               "client_stop_rx closed unexpectedly: {e:?}"
+                            );
+                            ClientRunResult::QueueClosed
+                        }
+                    }
+                }
+                _ = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
+                    // this is fine
+                },
+            }
         }
 
         // Wait for the start oneshot to fire.  This may happen immediately, but
@@ -2551,7 +2573,11 @@ impl ClientIoTask {
             tcp = sock.connect(self.target) => {
                 match tcp {
                     Ok(tcp) => {
-                        info!(self.log, "ds_connection connected");
+                        info!(
+                            self.log,
+                            "ds_connection connected from {:?}",
+                            tcp.local_addr()
+                        );
                         tcp
                     }
                     Err(e) => {
@@ -2564,6 +2590,20 @@ impl ClientIoTask {
                     }
                 }
             }
+            s = &mut self.stop => {
+                warn!(self.log, "client IO task stopped during connection");
+                return match s {
+                    Ok(s) =>
+                        ClientRunResult::RequestedStop(s),
+                    Err(e) => {
+                        warn!(
+                            self.log,
+                           "client_stop_rx closed unexpectedly: {e:?}"
+                        );
+                        ClientRunResult::QueueClosed
+                    }
+                }
+            }
         };
 
         // We're connected; before we wrap it, set TCP_NODELAY to assure