fix: lock mgr may failed to release lock in time (#16611)

* fix: lock mgr may failed to release lock in time

relaces `tokio::sync::Notify` with `WatchNotify` which is safe to `nofity_one` before
`Notified` future has been being polled.

* fix: typo, revert default log level

src/common/base/src/base/watch_notify.rs

@@ -44,6 +44,10 @@ impl WatchNotify {
     pub fn notify_waiters(&self) {
         let _ = self.tx.send_replace(true);
     }
+
+    pub fn notify_one(&self) {
+        self.notify_waiters()
+    }
 }
 
 #[cfg(test)]
@@ -68,4 +72,15 @@ mod tests {
         let notified = notify.notified();
         notified.await;
     }
+
+    #[tokio::test]
+    async fn test_notify_one() {
+        let notify = WatchNotify::new();
+        // notify_waiters ahead of notified being instantiated and awaited
+        notify.notify_one();
+
+        // this should not await indefinitely
+        let notified = notify.notified();
+        notified.await;
+    }
 }

src/query/service/src/locks/lock_holder.rs

@@ -19,8 +19,8 @@ use std::time::Duration;
 use std::time::Instant;
 
 use backoff::backoff::Backoff;
-use databend_common_base::base::tokio::sync::Notify;
 use databend_common_base::base::tokio::time::sleep;
+use databend_common_base::base::WatchNotify;
 use databend_common_base::runtime::GlobalIORuntime;
 use databend_common_base::runtime::TrySpawn;
 use databend_common_catalog::catalog::Catalog;
@@ -41,7 +41,7 @@ use crate::sessions::SessionManager;
 #[derive(Default)]
 pub struct LockHolder {
     shutdown_flag: AtomicBool,
-    shutdown_notify: Notify,
+    shutdown_notify: WatchNotify,
 }
 
 impl LockHolder {