CONFIG = None
if not CONFIG:
    CONFIG = koji.read_config_files([(CONFIG_FILE, True)])

Previously, the config was only loaded as needed, and the if not CONFIG check was there to check if it had been loaded yet. Now that we're reading the config at load time, this check seems superfluous -- we've just set the value to None in the line before. That is unless this is about thread-safety, but if that's the case I think we'd want to do something a little different.

        if extra_limit == 0:
            return buildinfo

This appears to be an unrelated behavior change. It certainly seems reasonable, but I think it deserves to be tracked and documented. Perhaps a separate issue?

    if CONFIG.getboolean('broker', 'test_mode', fallback=False):
        LOG.debug("test mode: Would queue msg: %r/%r/%r" % (address, props, data))
        return

This significantly reduces the code coverage of test mode. Previously test mode would take things all the way up to the point of sending the message.

 Currently you need to create table manually by running the following SQL:

I don't really like requiring this and it also seems a little strange for the table to be configurable.

Since this is a supported plugin bundled with the code, perhaps we can just include this in the schema.

We should probably explain the batch_size config in the doc.

the removed message from self.msgs: debug log message should be issued regardless of the queue setting.

in queue_msg we're dumping data to json twice in the non-db queue case.

Ok, after thinking further, I have a fairly fundamental criticism.

The two different queue pathways, context and db, are written as if they are two ways of doing the same thing, but the db pathway is substantially different.

In the db pathway, messages can be buffered across calls. This is a very significant difference. We can have a call handler emitting messages that are completely unrelated to that call. E.g. a tagBuildBypass call could end up emitting messages from an earlier repoInit call.

I'm worried we could run into some very surprising situations where an earlier call could affect or even break a later one.

I'm also worried about shifting from a thread local queue to a one that is not-only not thread-local, but shared across all call handlers.

Two calls running in parallel will both insert their message data into the table, and then they will both query all messages in that table, up to the batch_size limit. It looks like we could easily generate duplicate messages.

I think that we should treat the "buffer-across-calls" case as a clear exception, and we need to think very carefully about how and when we handle emitting messages that we were unable to emit during the call that generated them.

The send_queued_msgs handler is hooked into the postCommit callback which happens after the commit for the call. The DELETE statements in on_settled will end up executing after the commit and will be rolled back when we call context.cnx.close().

I've rewritten it - now it saves only remaining messages. If all messages were sent, it looks to db and locks table for query/deletion of messages.

db_enabled = CONFIG.getboolean('queue', 'test_mode')

I think you want CONFIG.getboolean('queue', 'enabled')

Do we need a BEGIN in store_to_db?

When we have db queue entries and are unable to send, it looks like we're going to query, delete, and re-insert each entry each time. This seems like unnecessary churn, and a case I think we'll encounter regularly with message bus outages.

c.execute('LOCK TABLE proton_queue IN ACCESS EXCLUSIVE MODE NOWAIT')

It doesn't look like we're handling the case where we don't get the lock

In places where we're performing db writes, it's probably worth including a comment to remind the reader that we're running in postCommit.

I started going through this and ended up wanting to make a number of changes. Here is what I came up with.

https://pagure.io/fork/mikem/koji/commits/pr2441updates

There's a lot there. I definitely prefer to avoid the re-insertion churn, which this does. I'm not 100% sure about the test_mode_fail setting, but it enabled me to test out the plugin locally and debug some issues. I think the docs changes are helpful.

WDYT?

I'm also tempted to add a simple query to check to see if there are any messages in the queue before we bother locking. It seems like that would be a common case. But perhaps I'm over-optimizing? It's not like we wait for the lock.

there is no column: address in proton_queue table.

there is no column: address in proton_queue table.

nor is there a created_tz in the current version of the patch. My branch above adds both.

Another issue I addressed on my branch -- psycopg2.errors is new in version 2.8. While the project released this version about 1.5 yrs ago, it's not readily available on all our platforms. I changed the except clause to be more compatible with older versions of the lib.

I'm hesitating about running this inside the transaction. But it is probably not a big problem. Reason for this is that we effectively serialize sending these messages as only one process in time can be sending db-queued messages. Former solution could have sent more, but it still could have mixed the order if broker is not working well (or if one process hits the working broker while other not). So in the end I'm ok with this approach. (BTW, PG 9+ has nice SKIP LOCKED which we can leverage later if we fix the requirements for PG version https://www.2ndquadrant.com/en/blog/what-is-select-skip-locked-for-in-postgresql-9-5/)

LOCK NOWAIT is almost for free, so I would leave it there as second select (we would still need it after acuqiring the lock) would be more work.

:thumbsup:

Commit 729f8476 fixes this pull-request