From 0d61c94250dba67e689b34d1c530b62ce234e7fb Mon Sep 17 00:00:00 2001 From: Mike McLean Date: May 17 2023 09:02:08 +0000 Subject: [PATCH 1/6] add db_lock function and locks table --- diff --git a/docs/schema.sql b/docs/schema.sql index 505f0b8..da3f3ef 100644 --- a/docs/schema.sql +++ b/docs/schema.sql @@ -983,4 +983,10 @@ CREATE TABLE rpm_checksum ( ) WITHOUT OIDS; CREATE INDEX rpm_checksum_rpm_id ON rpm_checksum(rpm_id); + +CREATE TABLE locks ( + name TEXT NOT NULL PRIMARY KEY +) WITHOUT OIDS; + + COMMIT WORK; diff --git a/kojihub/db.py b/kojihub/db.py index 478c30f..d331bee 100644 --- a/kojihub/db.py +++ b/kojihub/db.py @@ -321,6 +321,41 @@ def currval(sequence): return _singleValue("SELECT currval(%(sequence)s)", data, strict=True) +def db_lock(name, wait=True): + """Obtain lock for name + + :param string name: the lock name + :param bool wait: whether to wait for the lock (default: True) + :return: True if locked, False otherwise + + This function is implemented using db row locks and the locks table + """ + # first see if we need to add the row + query = "SELECT name FROM locks WHERE name=%(name)s" + data = {"name": name} + rows =_fetchMulti(query, data) + if not rows: + insert = "INSERT INTO locks (name) VALUES (%(name)s) ON CONFLICT DO NOTHING" + _dml(insert, data) + + # and then actually lock the row + if wait: + query = "SELECT name FROM locks WHERE name=%(name)s FOR UPDATE" + else: + # using SKIP LOCKED rather than NOWAIT to avoid error messages + query = "SELECT name FROM locks WHERE name=%(name)s FOR UPDATE SKIP LOCKED" + rows = _fetchMulti(query, data) + + if rows: + # we have the lock + return True + elif wait: + # should not happen + raise koji.LockError(f"Failed to read lock {name}") + else: + return False + + class Savepoint(object): def __init__(self, name): From 4913bd0c0ab00c8ed515438f97b508d0a19d4d74 Mon Sep 17 00:00:00 2001 From: Mike McLean Date: May 17 2023 09:02:08 +0000 Subject: [PATCH 2/6] use db_lock in protonmsg --- diff --git a/plugins/hub/protonmsg.py b/plugins/hub/protonmsg.py index 3cb125e..44f31da 100644 --- a/plugins/hub/protonmsg.py +++ b/plugins/hub/protonmsg.py @@ -18,7 +18,7 @@ import koji from koji.context import context from koji.plugin import callback, convert_datetime, ignore_error from kojihub import get_build_type -from kojihub.db import QueryProcessor, InsertProcessor, DeleteProcessor +from kojihub.db import QueryProcessor, InsertProcessor, DeleteProcessor, db_lock CONFIG_FILE = '/etc/koji-hub/plugins/protonmsg.conf' CONFIG = None @@ -361,9 +361,7 @@ def handle_db_msgs(urls, CONFIG): c = context.cnx.cursor() # we're running in postCommit, so we need to handle new transaction c.execute('BEGIN') - try: - c.execute('LOCK TABLE proton_queue IN ACCESS EXCLUSIVE MODE NOWAIT', log_errors=False) - except psycopg2.OperationalError: + if not db_lock('protonmsg-plugin', wait=False): LOG.debug('skipping db queue due to lock') return try: From bc5088f88168f7eb9d2a350673c9abf5ec552cc8 Mon Sep 17 00:00:00 2001 From: Mike McLean Date: May 17 2023 09:02:08 +0000 Subject: [PATCH 3/6] add a note in db_lock --- diff --git a/kojihub/db.py b/kojihub/db.py index d331bee..73565e9 100644 --- a/kojihub/db.py +++ b/kojihub/db.py @@ -336,6 +336,8 @@ def db_lock(name, wait=True): rows =_fetchMulti(query, data) if not rows: insert = "INSERT INTO locks (name) VALUES (%(name)s) ON CONFLICT DO NOTHING" + # this could cause us to wait if another transaction is adding the same lock + # however that will only happen the first time _dml(insert, data) # and then actually lock the row From 14c2e1b38ce2df6be449e4810609344e07aeaaf3 Mon Sep 17 00:00:00 2001 From: Mike McLean Date: May 17 2023 09:02:08 +0000 Subject: [PATCH 4/6] avoid upsert in db_lock --- diff --git a/docs/schema.sql b/docs/schema.sql index da3f3ef..61aca04 100644 --- a/docs/schema.sql +++ b/docs/schema.sql @@ -984,9 +984,11 @@ CREATE TABLE rpm_checksum ( CREATE INDEX rpm_checksum_rpm_id ON rpm_checksum(rpm_id); +-- this table is used for locking, see db_lock() CREATE TABLE locks ( name TEXT NOT NULL PRIMARY KEY ) WITHOUT OIDS; +INSERT INTO locks(name) VALUES('protonmsg-plugin') COMMIT WORK; diff --git a/kojihub/db.py b/kojihub/db.py index 73565e9..1a04e21 100644 --- a/kojihub/db.py +++ b/kojihub/db.py @@ -324,23 +324,16 @@ def currval(sequence): def db_lock(name, wait=True): """Obtain lock for name + The named lock must exist in the locks table + :param string name: the lock name :param bool wait: whether to wait for the lock (default: True) :return: True if locked, False otherwise This function is implemented using db row locks and the locks table """ - # first see if we need to add the row - query = "SELECT name FROM locks WHERE name=%(name)s" + # attempt to lock the row data = {"name": name} - rows =_fetchMulti(query, data) - if not rows: - insert = "INSERT INTO locks (name) VALUES (%(name)s) ON CONFLICT DO NOTHING" - # this could cause us to wait if another transaction is adding the same lock - # however that will only happen the first time - _dml(insert, data) - - # and then actually lock the row if wait: query = "SELECT name FROM locks WHERE name=%(name)s FOR UPDATE" else: @@ -351,11 +344,18 @@ def db_lock(name, wait=True): if rows: # we have the lock return True - elif wait: - # should not happen - raise koji.LockError(f"Failed to read lock {name}") - else: - return False + + if not wait: + # in the no-wait case, this could mean either that the row is already locked, or that + # the lock does not exist, so we check + query = "SELECT name FROM locks WHERE name=%(name)s" + rows =_fetchMulti(query, data) + if rows: + # the lock exists, but we did not acquire it + return False + + # otherwise, the lock does not exist + raise koji.LockError(f"Lock not defined: {name}") class Savepoint(object): From b9e9dd6c4b6ce949bda86b1dbc01537dadb4d601 Mon Sep 17 00:00:00 2001 From: Tomas Kopecek Date: May 17 2023 09:02:30 +0000 Subject: [PATCH 5/6] migration script --- diff --git a/docs/schema-upgrade-1.32-1.33.sql b/docs/schema-upgrade-1.32-1.33.sql index c1b42c8..9efb8d7 100644 --- a/docs/schema-upgrade-1.32-1.33.sql +++ b/docs/schema-upgrade-1.32-1.33.sql @@ -8,4 +8,8 @@ BEGIN; INSERT INTO archivetypes (name, description, extensions) VALUES ('packages', 'Kiwi packages listing', 'packages') ON CONFLICT DO NOTHING; INSERT INTO archivetypes (name, description, extensions) VALUES ('verified', 'Kiwi verified package list', 'verified') ON CONFLICT DO NOTHING; ALTER TABLE host ADD COLUMN update_time TIMESTAMPTZ; + CREATE TABLE locks ( + name TEXT NOT NULL PRIMARY KEY + ) WITHOUT OIDS; + INSERT INTO locks(name) VALUES('protonmsg-plugin'); COMMIT; diff --git a/docs/schema.sql b/docs/schema.sql index 61aca04..5fa6a22 100644 --- a/docs/schema.sql +++ b/docs/schema.sql @@ -983,12 +983,10 @@ CREATE TABLE rpm_checksum ( ) WITHOUT OIDS; CREATE INDEX rpm_checksum_rpm_id ON rpm_checksum(rpm_id); - -- this table is used for locking, see db_lock() CREATE TABLE locks ( name TEXT NOT NULL PRIMARY KEY ) WITHOUT OIDS; -INSERT INTO locks(name) VALUES('protonmsg-plugin') - +INSERT INTO locks(name) VALUES('protonmsg-plugin'); COMMIT WORK; From 0a48d6a6883a94f8a1a51b54886ae4e51026cd50 Mon Sep 17 00:00:00 2001 From: Tomas Kopecek Date: May 17 2023 09:02:32 +0000 Subject: [PATCH 6/6] fix flake8 Related: https://pagure.io/koji/issue/3790 --- diff --git a/kojihub/db.py b/kojihub/db.py index 1a04e21..d5e29df 100644 --- a/kojihub/db.py +++ b/kojihub/db.py @@ -349,7 +349,7 @@ def db_lock(name, wait=True): # in the no-wait case, this could mean either that the row is already locked, or that # the lock does not exist, so we check query = "SELECT name FROM locks WHERE name=%(name)s" - rows =_fetchMulti(query, data) + rows = _fetchMulti(query, data) if rows: # the lock exists, but we did not acquire it return False diff --git a/plugins/hub/protonmsg.py b/plugins/hub/protonmsg.py index 44f31da..0489ff4 100644 --- a/plugins/hub/protonmsg.py +++ b/plugins/hub/protonmsg.py @@ -9,7 +9,6 @@ import json import logging import random -import psycopg2 from proton import Message, SSLDomain from proton.handlers import MessagingHandler from proton.reactor import Container