fix(subscriber): record timestamps for updates last (#289)

* refac(subscriber): factor out update construction

This branch refactors how the aggregator task builds update messages. In
particular, it factors out the shared code between building the initial
update for a subscription and building the subsequent updates in
`publish` into methods that are called with an `Include` enum to
determine whether the update should include all tasks/resources/ops, or
just the ones since the last update. This makes the code in
`add_instrument_subscription` and in `publish` much easier to read.

This is just the refactor part from #289 split out into a separate
branch; this should make no functional change. If we decide that #289 is
the correct way to fix the bug, we can rebase that branch onto this, so
that it _just_ includes the bugfix. This way, we can land the refactor
anyway, because I think it's nice.

* fix(subscriber): record timestamps for updates last

## Motivation

Currently, when constructing an update message to send over the wire, a
timestamp is taken first, and then the protobuf data is constructed.
This can lead to issues where the "now" timestamp is actually _before_
timestamps present in the stats sent in that update, since the stats for
a particular task/resource/async op might be updated on another thread
after taking the update's "now" timestamp. This results in issues like
#266.

## Solution

There's no actual reason to take those timestamps *before* we assemble
the update. This branch changes the aggregator to build all the various
data updates in an update message, and *then* record the update's "now"
timestamp. Any timestamps for tasks/resources/async ops that are
recorded after the update's "now" timestamp will now be included in the
*next* update.

Fixes #266
Depends on #288

console-subscriber/src/aggregator/mod.rs

@@ -260,11 +260,12 @@ impl Aggregator {
     /// Add the task subscription to the watchers after sending the first update
     fn add_instrument_subscription(&mut self, subscription: Watch<proto::instrument::Update>) {
         tracing::debug!("new instrument subscription");
-        let now = Instant::now();
 
         let task_update = Some(self.task_update(Include::All));
         let resource_update = Some(self.resource_update(Include::All));
         let async_op_update = Some(self.async_op_update(Include::All));
+        let now = Instant::now();
+
         let update = &proto::instrument::Update {
             task_update,
             resource_update,
@@ -356,13 +357,12 @@ impl Aggregator {
         } else {
             None
         };
-        let now = Instant::now();
         let task_update = Some(self.task_update(Include::UpdatedOnly));
         let resource_update = Some(self.resource_update(Include::UpdatedOnly));
         let async_op_update = Some(self.async_op_update(Include::UpdatedOnly));
 
         let update = proto::instrument::Update {
-            now: Some(self.base_time.to_timestamp(now)),
+            now: Some(self.base_time.to_timestamp(Instant::now())),
             new_metadata,
             task_update,
             resource_update,
@@ -379,7 +379,7 @@ impl Aggregator {
             if let Some(task_stats) = stats.get(id) {
                 let details = proto::tasks::TaskDetails {
                     task_id: Some(id.clone().into()),
-                    now: Some(self.base_time.to_timestamp(now)),
+                    now: Some(self.base_time.to_timestamp(Instant::now())),
                     poll_times_histogram: task_stats.serialize_histogram(),
                 };
                 watchers.retain(|watch| watch.update(&details));