Record a metric for staleness of query invalidations (#39205)

This PR plumbs the timestamp of the invalidating write, if known, all the way through `extend_validity`.
Then, in the sync worker, we can measure the real time delay between the write and the subscribed re-executing query.
We can monitor this to make sure it is a reasonable value.

GitOrigin-RevId: 2210b600a96656f489e685553a49f11abfda7fcd

Author: goffrie

crates/application/src/api.rs

@@ -575,14 +575,25 @@ impl<RT: Runtime> SubscriptionClient for ApplicationSubscriptionClient<RT> {
     }
 }
 
+pub enum ExtendValidityResult {
+    /// The subscription's validity can be extended to the requested timestamp.
+    Extended,
+    /// The subscription may no longer be valid at the requested timestamp.
+    /// This result can be returned spuriously even if there were no conflicting
+    /// writes.
+    Invalid {
+        /// The earliest conflicting timestamp, if known. This is not guaranteed
+        /// to be known.
+        invalid_ts: Option<Timestamp>,
+    },
+}
+
 #[async_trait]
 pub trait SubscriptionTrait: Send + Sync {
     fn wait_for_invalidation(&self) -> BoxFuture<'static, anyhow::Result<()>>;
 
-    // Returns true if the subscription validity can be extended to new_ts. Note
-    // that extend_validity might return false even if the subscription can be
-    // extended, but will never return true if it can't.
-    async fn extend_validity(&self, new_ts: Timestamp) -> anyhow::Result<bool>;
+    /// See comments on [`ExtendValidityResult`]
+    async fn extend_validity(&self, new_ts: Timestamp) -> anyhow::Result<ExtendValidityResult>;
 }
 
 struct ApplicationSubscription {
@@ -602,10 +613,10 @@ impl SubscriptionTrait for ApplicationSubscription {
     }
 
     #[fastrace::trace]
-    async fn extend_validity(&self, new_ts: Timestamp) -> anyhow::Result<bool> {
+    async fn extend_validity(&self, new_ts: Timestamp) -> anyhow::Result<ExtendValidityResult> {
         if new_ts < self.initial_ts {
             // new_ts is before the initial subscription timestamp.
-            return Ok(false);
+            return Ok(ExtendValidityResult::Invalid { invalid_ts: None });
         }
 
         // The inner subscription is periodically updated by the subscription
@@ -614,15 +625,19 @@ impl SubscriptionTrait for ApplicationSubscription {
             // Subscription is no longer valid. We could check validity from end_ts
             // to new_ts, but this is likely to fail and is potentially unbounded amount of
             // work, so we return false here. This is valid per the function contract.
-            return Ok(false);
+            return Ok(ExtendValidityResult::Invalid {
+                invalid_ts: self.inner.invalid_ts(),
+            });
         };
 
         let current_token = Token::new(self.reads.clone(), current_ts);
-        let Some(_new_token) = self.log.refresh_token(current_token, new_ts)? else {
-            // Subscription validity can't be extended. Note that returning false
-            // here also doesn't mean there is a conflict.
-            return Ok(false);
-        };
-        return Ok(true);
+        Ok(match self.log.refresh_token(current_token, new_ts)? {
+            Ok(_new_token) => ExtendValidityResult::Extended,
+            Err(invalid_ts) => {
+                // Subscription validity can't be extended. Note that returning false
+                // here also doesn't mean there is a conflict.
+                ExtendValidityResult::Invalid { invalid_ts }
+            },
+        })
     }
 }

crates/application/src/cache/mod.rs

@@ -665,8 +665,8 @@ impl<RT: Runtime> CacheManager<RT> {
             return Ok(None);
         }
         result.token = match self.database.refresh_token(result.token, ts).await? {
-            Some(t) => t,
-            None => {
+            Ok(t) => t,
+            Err(_invalid_ts) => {
                 tracing::debug!(
                     "Couldn't refresh cache entry from {} to {}, retrying...",
                     result.original_ts,

crates/database/benches/subscriptions.rs

@@ -267,11 +267,9 @@ fn bench_query(c: &mut Criterion) {
                     b.to_async(&rt).iter(|| async {
                         for (doc_id, doc_index_keys) in documents {
                             let mut to_notify = BTreeSet::new();
-                            subscription_manager.overlapping(
-                                doc_id,
-                                doc_index_keys,
-                                &mut to_notify,
-                            );
+                            subscription_manager.overlapping(doc_id, doc_index_keys, &mut |id| {
+                                to_notify.insert(id);
+                            });
                         }
                     })
                 },

crates/database/src/database.rs

@@ -1883,14 +1883,14 @@ impl<RT: Runtime> Database<RT> {
             .collect())
     }
 
-    /// Attempt to pull a token forward to a given timestamp, returning `None`
+    /// Attempt to pull a token forward to a given timestamp, returning `Err`
     /// if there have been overlapping writes between the token's original
     /// timestamp and `ts`.
     pub async fn refresh_token(
         &self,
         token: Token,
         ts: Timestamp,
-    ) -> anyhow::Result<Option<Token>> {
+    ) -> anyhow::Result<Result<Token, Option<Timestamp>>> {
         let _timer = metrics::refresh_token_timer();
         self.log.refresh_token(token, ts)
     }
@@ -2195,6 +2195,11 @@ fn occ_write_source_string(
 pub struct ConflictingReadWithWriteSource {
     pub(crate) read: ConflictingRead,
     pub(crate) write_source: WriteSource,
+    /// The timestamp of the conflicting write.
+    ///
+    /// N.B.: this may be a non-repeatable timestamp, if this conflict occurred
+    /// against a pending write!
+    pub(crate) write_ts: Timestamp,
 }
 
 impl ConflictingReadWithWriteSource {

crates/database/src/reads.rs

@@ -283,7 +283,7 @@ impl ReadSet {
         persistence_version: PersistenceVersion,
     ) -> Option<ConflictingReadWithWriteSource> {
         let mut buffer = IndexKeyBuffer::new();
-        for (_ts, updates, write_source) in updates {
+        for (update_ts, updates, write_source) in updates {
             for (_, update) in updates {
                 if let Some(ref document) = update.new_document {
                     if let Some(conflicting_read) =
@@ -292,6 +292,7 @@ impl ReadSet {
                         return Some(ConflictingReadWithWriteSource {
                             read: conflicting_read,
                             write_source: write_source.clone(),
+                            write_ts: *update_ts,
                         });
                     }
                 }
@@ -302,6 +303,7 @@ impl ReadSet {
                         return Some(ConflictingReadWithWriteSource {
                             read: conflicting_read,
                             write_source: write_source.clone(),
+                            write_ts: *update_ts,
                         });
                     }
                 }
@@ -323,13 +325,14 @@ impl ReadSet {
             ),
         >,
     ) -> Option<ConflictingReadWithWriteSource> {
-        for (_ts, updates, write_source) in updates {
+        for (update_ts, updates, write_source) in updates {
             for (id, update) in updates {
                 if let Some(ref document) = update.new_document_keys {
                     if let Some(conflicting_read) = self.overlaps_index_keys(*id, document) {
                         return Some(ConflictingReadWithWriteSource {
                             read: conflicting_read,
                             write_source: write_source.clone(),
+                            write_ts: *update_ts,
                         });
                     }
                 }
@@ -338,6 +341,7 @@ impl ReadSet {
                         return Some(ConflictingReadWithWriteSource {
                             read: conflicting_read,
                             write_source: write_source.clone(),
+                            write_ts: *update_ts,
                         });
                     }
                 }