CascadedWorker: publish_local_wm() is missing a commit

diff --git a/python/pgq/cascade/worker.py b/python/pgq/cascade/worker.py
index a721eaa082b6bf9bb2abe312880fe83ae038ede0..6e7e80673beaf3e3b5e40931584af8bcb69595ee 100644 (file)
--- a/python/pgq/cascade/worker.py
+++ b/python/pgq/cascade/worker.py
@@ -227,6 +227,7 @@ class CascadedWorker(CascadedConsumer):
         src_curs = src_db.cursor()
         q = "select * from pgq_node.set_subscriber_watermark(%s, %s, %s)"
         src_curs.execute(q, [self.pgq_queue_name, st.node_name, wm])
+        src_db.commit()
 
         # if last part fails, dont repeat it immediately
         self.local_wm_publish_time = t