test(subscriber): add initial integration tests

The `console-subscriber` crate has no integration tests. There are some
unit tests, but without very high coverage of features.

Recently, we've found or fixed a few errors which probably could have
been caught by a medium level of integration testing.

However, testing `console-subscriber` isn't straight forward. It is
effectively a tracing subscriber (or layer) on one end, and a gRPC
server on the other end.

This change adds enough of a testing framework to write some initial
integration tests. It is the first step towards closing #450.

Each test comprises 2 parts:
- One or more "expcted tasks"
- A future which will be driven to completion on a dedicated Tokio runtime.

Behind the scenes, a console subscriber layer is created and it's server
part is connected to a duplex stream. The client of the duplex stream
then records incoming updates and reconstructs "actual tasks". The layer
itself is set as the default subscriber for the duration of `block_on`
which is used to drive the provided future to completioin.

The expected tasks have a set of "matches", which is how we find the
actual task that we want to validate against. Currently, the only value
we match on is the task's name.

The expected tasks also have a set of expectations. These are other
fields on the actual task which are validated once a matching task is
found. Currently, the two fields which can have expectations set on them
are the `wakes` and `self_wakes` fields.

So, to construct an expected task, which will match a task with the name
`"my-task"` and then validate that the matched task gets woken once, the
code would be:

```rust
ExpectedTask::default()
    .match_name("my-task")
    .expect_wakes(1);
```

A future which passes this test could be:

```rust
async {
    task::Builder::new()
        .name("my-task")
        .spawn(async {
            tokio::time::sleep(std::time::Duration::ZERO).await
        })
}
```

The full test would then look like:

```rust
fn wakes_once() {
    let expected_task = ExpectedTask::default()
        .match_name("my-task")
        .expect_wakes(1);

    let future = async {
        task::Builder::new()
            .name("my-task")
            .spawn(async {
                tokio::time::sleep(std::time::Duration::ZERO).await
            })
    };

    assert_task(expected_task, future);
}
```

The PR depends on 2 others:
 - #447 which fixes an error in the logic that determines whether a task
   is retained in the aggregator or not.
 - #451 which exposes the server parts and is necessary to allow us to
   connect the instrument server and client via a duplex channel.

This change contains some initial tests for wakes and self wakes which
would have caught the error fixed in #430. Additionally there are tests
for the functionality of the testing framework itself.

console-subscriber/Cargo.toml

@@ -55,6 +55,7 @@ crossbeam-channel = "0.5"
 
 [dev-dependencies]
 tokio = { version = "^1.21", features = ["full", "rt-multi-thread"] }
+tower = "0.4"
 futures = "0.3"
 
 [package.metadata.docs.rs]

console-subscriber/src/aggregator/id_data.rs

@@ -104,18 +104,18 @@ impl<T: Unsent> IdData<T> {
             if let Some(dropped_at) = stats.dropped_at() {
                 let dropped_for = now.checked_duration_since(dropped_at).unwrap_or_default();
                 let dirty = stats.is_unsent();
-                let should_drop =
+                let should_retain =
                         // if there are any clients watching, retain all dirty tasks regardless of age
                         (dirty && has_watchers)
-                        || dropped_for > retention;
+                        || dropped_for <= retention;
                 tracing::trace!(
                     stats.id = ?id,
                     stats.dropped_at = ?dropped_at,
                     stats.dropped_for = ?dropped_for,
                     stats.dirty = dirty,
-                    should_drop,
+                    should_retain,
                 );
-                return !should_drop;
+                return should_retain;
             }
 
             true

console-subscriber/tests/framework.rs

@@ -0,0 +1,184 @@
+//! Framework tests
+//!
+//! The tests in this module are here to verify the testing framework itself.
+//! As such, some of these tests may be repeated elsewhere (where we wish to
+//! actually test the functionality of `console-subscriber`) and others are
+//! negative tests that should panic.
+
+use std::time::Duration;
+
+use tokio::{task, time::sleep};
+
+mod support;
+use support::{assert_task, assert_tasks, ExpectedTask};
+
+#[test]
+fn expect_present() {
+    let expected_task = ExpectedTask::default()
+        .match_default_name()
+        .expect_present();
+
+    let future = async {
+        sleep(Duration::ZERO).await;
+    };
+
+    assert_task(expected_task, future);
+}
+
+#[test]
+#[should_panic(expected = "Test failed: Task validation failed:
+ - Task<name=main>: no expectations set, if you want to just expect that a matching task is present, use `expect_present()`
+")]
+fn fail_no_expectations() {
+    let expected_task = ExpectedTask::default().match_default_name();
+
+    let future = async {
+        sleep(Duration::ZERO).await;
+    };
+
+    assert_task(expected_task, future);
+}
+
+#[test]
+fn wakes() {
+    let expected_task = ExpectedTask::default().match_default_name().expect_wakes(1);
+
+    let future = async {
+        sleep(Duration::ZERO).await;
+    };
+
+    assert_task(expected_task, future);
+}
+
+#[test]
+#[should_panic(expected = "Test failed: Task validation failed:
+ - Task<name=main>: expected `wakes` to be 5, but actual was 1
+")]
+fn fail_wakes() {
+    let expected_task = ExpectedTask::default().match_default_name().expect_wakes(5);
+
+    let future = async {
+        sleep(Duration::ZERO).await;
+    };
+
+    assert_task(expected_task, future);
+}
+
+#[test]
+fn self_wakes() {
+    let expected_task = ExpectedTask::default()
+        .match_default_name()
+        .expect_self_wakes(1);
+
+    let future = async { task::yield_now().await };
+
+    assert_task(expected_task, future);
+}
+
+#[test]
+#[should_panic(expected = "Test failed: Task validation failed:
+ - Task<name=main>: expected `self_wakes` to be 1, but actual was 0
+")]
+fn fail_self_wake() {
+    let expected_task = ExpectedTask::default()
+        .match_default_name()
+        .expect_self_wakes(1);
+
+    let future = async {
+        sleep(Duration::ZERO).await;
+    };
+
+    assert_task(expected_task, future);
+}
+
+#[test]
+fn test_spawned_task() {
+    let expected_task = ExpectedTask::default()
+        .match_name("another-name".into())
+        .expect_present();
+
+    let future = async {
+        task::Builder::new()
+            .name("another-name")
+            .spawn(async { task::yield_now().await })
+    };
+
+    assert_task(expected_task, future);
+}
+
+#[test]
+#[should_panic(expected = "Test failed: Task validation failed:
+ - Task<name=wrong-name>: no matching actual task was found
+")]
+fn fail_wrong_task_name() {
+    let expected_task = ExpectedTask::default().match_name("wrong-name".into());
+
+    let future = async { task::yield_now().await };
+
+    assert_task(expected_task, future);
+}
+
+#[test]
+fn multiple_tasks() {
+    let expected_tasks = vec![
+        ExpectedTask::default()
+            .match_name("task-1".into())
+            .expect_wakes(1),
+        ExpectedTask::default()
+            .match_name("task-2".into())
+            .expect_wakes(1),
+    ];
+
+    let future = async {
+        let task1 = task::Builder::new()
+            .name("task-1")
+            .spawn(async { task::yield_now().await })
+            .unwrap();
+        let task2 = task::Builder::new()
+            .name("task-2")
+            .spawn(async { task::yield_now().await })
+            .unwrap();
+
+        tokio::try_join! {
+            task1,
+            task2,
+        }
+        .unwrap();
+    };
+
+    assert_tasks(expected_tasks, future);
+}
+
+#[test]
+#[should_panic(expected = "Test failed: Task validation failed:
+ - Task<name=task-2>: expected `wakes` to be 2, but actual was 1
+")]
+fn fail_1_of_2_expected_tasks() {
+    let expected_tasks = vec![
+        ExpectedTask::default()
+            .match_name("task-1".into())
+            .expect_wakes(1),
+        ExpectedTask::default()
+            .match_name("task-2".into())
+            .expect_wakes(2),
+    ];
+
+    let future = async {
+        let task1 = task::Builder::new()
+            .name("task-1")
+            .spawn(async { task::yield_now().await })
+            .unwrap();
+        let task2 = task::Builder::new()
+            .name("task-2")
+            .spawn(async { task::yield_now().await })
+            .unwrap();
+
+        tokio::try_join! {
+            task1,
+            task2,
+        }
+        .unwrap();
+    };
+
+    assert_tasks(expected_tasks, future);
+}

console-subscriber/tests/support/mod.rs

@@ -0,0 +1,47 @@
+use futures::Future;
+
+mod state;
+mod subscriber;
+mod task;
+
+use subscriber::run_test;
+
+pub(crate) use subscriber::MAIN_TASK_NAME;
+pub(crate) use task::ExpectedTask;
+
+/// Assert that an `expected_task` is recorded by a console-subscriber
+/// when driving the provided `future` to completion.
+///
+/// This function is equivalent to calling [`assert_tasks`] with a vector
+/// containing a single task.
+///
+/// # Panics
+///
+/// This function will panic if the expectations on the expected task are not
+/// met or if a matching task is not recorded.
+#[track_caller]
+#[allow(dead_code)]
+pub(crate) fn assert_task<Fut>(expected_task: ExpectedTask, future: Fut)
+where
+    Fut: Future + Send + 'static,
+    Fut::Output: Send + 'static,
+{
+    run_test(vec![expected_task], future)
+}
+
+/// Assert that the `expected_tasks` are recorded by a console-subscriber
+/// when driving the provided `future` to completion.
+///
+/// # Panics
+///
+/// This function will panic if the expectations on any of the expected tasks
+/// are not met or if matching tasks are not recorded for all expected tasks.
+#[track_caller]
+#[allow(dead_code)]
+pub(crate) fn assert_tasks<Fut>(expected_tasks: Vec<ExpectedTask>, future: Fut)
+where
+    Fut: Future + Send + 'static,
+    Fut::Output: Send + 'static,
+{
+    run_test(expected_tasks, future)
+}