watch_add_notify gets the GNode with the cache_lock but then relesaes the cache_lock and acquires message_lock. Before that thread can acquire the message_lock another thread can call watch_remove_notify and remove the same prefix causing the GNode that was returned to thread 1 to be stale/bad.
Thread1 would then set the watch->message.watch.cache_node = cache_node; when it sends it's notifications.
This leads to the assert log and subsequently a corruption downstream