fix: skip republishing unchanged watch interest digests

Author: St4NNi

core/src/structs/notification_watch.rs

@@ -291,11 +291,11 @@ pub fn parse_watch_subscription_key(key: &[u8]) -> Result<(UserId, Ulid), Conver
 
 /// Keys in the watch-interest keyspace. Per-node digests use fixed-length keys
 /// so their prefixes are unambiguous: `n/` + realm id + node id (realm first so
-/// all nodes' digests for one realm form a single scan range). Local-only dirty
-/// markers use a text prefix (`dirty/`) that never collides with the binary
-/// digest prefix.
+/// all nodes' digests for one realm form a single scan range). Local-only
+/// markers use text prefixes that never collide with the binary digest prefix.
 pub const WATCH_INTEREST_NODE_PREFIX: &[u8] = b"n/";
 pub const WATCH_INTEREST_DIRTY_PREFIX: &[u8] = b"dirty/";
+const WATCH_INTEREST_PENDING_PREFIX: &[u8] = b"pending/";
 
 /// One coalesced interest entry: the union of every subscription a node holds
 /// for a realm that shares this path prefix.
@@ -390,6 +390,13 @@ pub fn watch_interest_dirty_key(realm_id: RealmId) -> Vec<u8> {
     key
 }
 
+pub fn watch_interest_pending_key(realm_id: RealmId) -> Vec<u8> {
+    let mut key = Vec::with_capacity(WATCH_INTEREST_PENDING_PREFIX.len() + 32);
+    key.extend_from_slice(WATCH_INTEREST_PENDING_PREFIX);
+    key.extend_from_slice(realm_id.as_bytes());
+    key
+}
+
 /// Recovers the realm id from a `dirty/<realm>` marker key.
 pub fn watch_interest_dirty_realm_id(key: &[u8]) -> Option<RealmId> {
     let tail = key.strip_prefix(WATCH_INTEREST_DIRTY_PREFIX)?;

operations/src/notifications/watch/interest.rs

@@ -12,7 +12,8 @@ use aruna_core::structs::{
     RealmConfigDocument, RealmId, WATCH_INTEREST_DIRTY_PREFIX, WatchInterestDigest,
     WatchInterestEntry, WatchInterestTable, watch_interest_dirty_key,
     watch_interest_dirty_realm_id, watch_interest_key_node_id, watch_interest_key_realm_id,
-    watch_interest_node_key, watch_interest_node_prefix, watch_interest_realm_prefix,
+    watch_interest_node_key, watch_interest_node_prefix, watch_interest_pending_key,
+    watch_interest_realm_prefix,
 };
 use aruna_core::task::{TaskEffect, TaskEvent, TaskKey};
 use aruna_core::types::{Key, KeySpace, Value};
@@ -179,7 +180,7 @@ pub async fn publish_watch_interest(ctx: &DriverContext, node_id: NodeId) -> Res
         return Ok(false);
     }
 
-    let mut writes: Vec<(KeySpace, Key, Value)> = Vec::with_capacity(realms.len());
+    let mut writes: Vec<(KeySpace, Key, Value)> = Vec::with_capacity(realms.len() * 2);
     let mut targets: Vec<(RealmId, DocumentSyncTarget)> = Vec::with_capacity(realms.len());
     let mut authorization_retries = Vec::new();
     for realm_id in &realms {
@@ -191,11 +192,48 @@ pub async fn publish_watch_interest(ctx: &DriverContext, node_id: NodeId) -> Res
         if check_failed {
             authorization_retries.push(*realm_id);
         }
-        writes.push((
-            NOTIFICATION_WATCH_INTEREST_KEYSPACE.to_string(),
-            Key::from(watch_interest_node_key(*realm_id, node_id)),
-            Value::from(digest.to_bytes().map_err(|e| e.to_string())?),
-        ));
+        let digest_key = Key::from(watch_interest_node_key(*realm_id, node_id));
+        let pending_key = Key::from(watch_interest_pending_key(*realm_id));
+        let digest_value = Value::from(digest.to_bytes().map_err(|e| e.to_string())?);
+        let current = match storage
+            .send_storage_effect(StorageEffect::BatchRead {
+                reads: vec![
+                    (
+                        NOTIFICATION_WATCH_INTEREST_KEYSPACE.to_string(),
+                        digest_key.clone(),
+                    ),
+                    (
+                        NOTIFICATION_WATCH_INTEREST_KEYSPACE.to_string(),
+                        pending_key.clone(),
+                    ),
+                ],
+                txn_id: None,
+            })
+            .await
+        {
+            Event::Storage(StorageEvent::BatchReadResult { values }) => values,
+            Event::Storage(StorageEvent::Error { error }) => return Err(error.to_string()),
+            other => return Err(format!("watch interest digest read failed: {other:?}")),
+        };
+        let [(_, current_digest), (_, pending)] = current.as_slice() else {
+            return Err("watch interest digest read count mismatch".to_string());
+        };
+        let changed = current_digest.as_ref() != Some(&digest_value);
+        if !changed && pending.is_none() {
+            continue;
+        }
+        if changed {
+            writes.push((
+                NOTIFICATION_WATCH_INTEREST_KEYSPACE.to_string(),
+                digest_key,
+                digest_value,
+            ));
+            writes.push((
+                NOTIFICATION_WATCH_INTEREST_KEYSPACE.to_string(),
+                pending_key,
+                Value::from(Vec::new()),
+            ));
+        }
         targets.push((
             *realm_id,
             DocumentSyncTarget::WatchInterest {
@@ -205,11 +243,12 @@ pub async fn publish_watch_interest(ctx: &DriverContext, node_id: NodeId) -> Res
         ));
     }
 
-    // Persist the refreshed digests but keep the dirty markers until the
-    // documents have been durably handed to replication.
+    // Persist changed digests with pending markers, but keep the dirty markers
+    // until the documents have been durably handed to replication.
     write_documents(storage, writes).await?;
 
     // Each realm rides its own realm-scoped topic, so replicate per realm.
+    let published = !targets.is_empty();
     for (realm_id, target) in targets {
         drive(
             ReplicateDocumentsOperation::new(ReplicateDocumentsConfig {
@@ -225,6 +264,22 @@ pub async fn publish_watch_interest(ctx: &DriverContext, node_id: NodeId) -> Res
         )
         .await
         .map_err(|error| format!("watch interest replication failed: {error}"))?;
+        match storage
+            .send_storage_effect(StorageEffect::Delete {
+                key_space: NOTIFICATION_WATCH_INTEREST_KEYSPACE.to_string(),
+                key: Key::from(watch_interest_pending_key(realm_id)),
+                txn_id: None,
+            })
+            .await
+        {
+            Event::Storage(StorageEvent::DeleteResult { .. }) => {}
+            Event::Storage(StorageEvent::Error { error }) => return Err(error.to_string()),
+            other => {
+                return Err(format!(
+                    "watch interest pending marker delete failed: {other:?}"
+                ));
+            }
+        }
     }
 
     // Preserve a retry generation when a permission lookup failed. The current
@@ -237,7 +292,7 @@ pub async fn publish_watch_interest(ctx: &DriverContext, node_id: NodeId) -> Res
     // those a concurrent CRUD did not re-dirty in the meantime.
     clear_consumed_markers(storage, observed_markers).await?;
 
-    Ok(true)
+    Ok(published)
 }
 
 /// Builds this node's digest for one realm from subscriptions whose owners are
@@ -267,8 +322,7 @@ async fn build_realm_digest(
     ))
 }
 
-/// Persists the refreshed digest documents. Each digest key has a single writer
-/// (this node), so a plain batch write is enough.
+/// Persists changed digests and their pending markers atomically.
 async fn write_documents(
     storage: &StorageHandle,
     writes: Vec<(KeySpace, Key, Value)>,
@@ -820,6 +874,21 @@ mod tests {
         }
     }
 
+    async fn read_pending(ctx: &DriverContext, realm_id: RealmId) -> Option<Value> {
+        match ctx
+            .storage_handle
+            .send_storage_effect(StorageEffect::Read {
+                key_space: NOTIFICATION_WATCH_INTEREST_KEYSPACE.to_string(),
+                key: Key::from(watch_interest_pending_key(realm_id)),
+                txn_id: None,
+            })
+            .await
+        {
+            Event::Storage(StorageEvent::ReadResult { value, .. }) => value,
+            other => panic!("unexpected pending marker read event: {other:?}"),
+        }
+    }
+
     async fn read_digest(
         ctx: &DriverContext,
         realm_id: RealmId,
@@ -924,6 +993,106 @@ mod tests {
         );
     }
 
+    // An acknowledged unchanged digest consumes a fresh dirty marker without
+    // another replication handoff.
+    #[tokio::test]
+    async fn unchanged_skips_publish() {
+        let temp = tempdir().unwrap();
+        let ctx = test_ctx(temp.path().to_str().unwrap());
+        let realm_id = RealmId([7u8; 32]);
+        let node_id = node(5);
+        let owner = user(7, 2);
+        let group_id = Ulid::r#gen();
+        install_realm_config(&ctx, realm_id, &[node_id]).await;
+        install_authorization(&ctx, realm_id, node_id, group_id, owner, &[]).await;
+        create_watch_subscription(
+            &ctx.storage_handle,
+            owner,
+            metadata_prefix(group_id, "a"),
+            mask(),
+            1,
+        )
+        .await
+        .expect("create");
+
+        assert!(
+            publish_watch_interest(&ctx, node_id)
+                .await
+                .expect("publish")
+        );
+        let before = read_digest(&ctx, realm_id, node_id)
+            .await
+            .expect("digest")
+            .to_bytes()
+            .unwrap();
+        assert!(read_pending(&ctx, realm_id).await.is_none());
+
+        mark_watch_interest_dirty(&ctx, realm_id)
+            .await
+            .expect("mark dirty");
+        assert!(
+            !publish_watch_interest(&ctx, node_id)
+                .await
+                .expect("republish")
+        );
+
+        let after = read_digest(&ctx, realm_id, node_id)
+            .await
+            .expect("digest")
+            .to_bytes()
+            .unwrap();
+        assert_eq!(after, before);
+        assert!(read_marker(&ctx, realm_id).await.is_none());
+        assert!(read_pending(&ctx, realm_id).await.is_none());
+    }
+
+    // Persistent authorization unavailability retains retries while identical
+    // empty digests stop producing replicated events.
+    #[tokio::test]
+    async fn retry_stays_dirty() {
+        let temp = tempdir().unwrap();
+        let ctx = test_ctx(temp.path().to_str().unwrap());
+        let realm_id = RealmId([8u8; 32]);
+        let node_id = node(6);
+        let owner = user(8, 2);
+        let group_id = Ulid::r#gen();
+        install_realm_config(&ctx, realm_id, &[node_id]).await;
+        create_watch_subscription(
+            &ctx.storage_handle,
+            owner,
+            metadata_prefix(group_id, "a"),
+            mask(),
+            1,
+        )
+        .await
+        .expect("create");
+
+        assert!(
+            publish_watch_interest(&ctx, node_id)
+                .await
+                .expect("publish")
+        );
+        assert!(read_marker(&ctx, realm_id).await.is_some());
+        let digest = read_digest(&ctx, realm_id, node_id)
+            .await
+            .expect("empty digest");
+        assert!(digest.entries.is_empty());
+        let before = digest.to_bytes().unwrap();
+
+        assert!(
+            !publish_watch_interest(&ctx, node_id)
+                .await
+                .expect("republish")
+        );
+        let after = read_digest(&ctx, realm_id, node_id)
+            .await
+            .expect("empty digest")
+            .to_bytes()
+            .unwrap();
+        assert_eq!(after, before);
+        assert!(read_marker(&ctx, realm_id).await.is_some());
+    }
+
     #[tokio::test]
     async fn deleting_last_watch_publishes_empty_digest() {
         let temp = tempdir().unwrap();