NOTIFY/LISTEN support

[msolli]

Proletarian currently polls the database table in order to get jobs. This works great and has very little performance impact on the database server. It is a simple approach, both operationally and conceptually.

There are two drawbacks to this approach, though:

1. If the Postgres server has been configured with log_statement = 'all', then the logs will be overwhelmed with statements like this:

2023-03-01 10:36:05.842 CET [78339] LOG:  execute S_5: SELECT job_id, queue, job_type, payload, attempts, enqueued_at, process_at FROM proletarian.job
	         WHERE
	           queue = $1
	           AND process_at <= $2
	         ORDER BY process_at ASC
	         LIMIT 1
	         FOR UPDATE SKIP LOCKED
2023-03-01 10:36:05.842 CET [78339] DETAIL:  parameters: $1 = ':proletarian/default', $2 = '2023-03-01 10:36:05.842265'
2023-03-01 10:36:05.842 CET [78339] LOG:  execute S_3: COMMIT

2. There is going to be a small delay between the enqueueing of the job, and the time when one of the worker threads picks it up. This delay is going to be, on average: poll-interval / (num-threads * num-instances * 2). The delay is inevitable, but it can be made very small indeed. Using four worker threads and a poll-interval of 100 ms, the average wait time for a job would be 12.5 ms. Note that this only holds when the workers are mostly idle. If they are saturated - jobs are enqueued at a pace the same as or greater than the pace with which the workers can perform the jobs - the wait time would be longer, but not because of polling delay.

NOTIFY/LISTEN

Postgres has a pub/sub mechanism for asynchronously notifying sessions of events. This is called NOTIFY/LISTEN (docs). One idea to get rid of the overwhelming logging decribed above is to replace polling with some NOTIFY/LISTEN mechanism.

This thread is for discussing the design and tradeoffs of such a mechanism, with the goal of either implementing the mechanism or ruling out NOTIFY/LISTEN as a viable mechanism for Proletarian.

[p-himik]

Here's a condensed version of something that I'm currently using in one of my projects:

(ns pg-notification-listener
  (:require [clojure.data.json :as json]
            [clojure.java.jdbc :as jdbc])
  (:import (java.sql Connection)
           (org.postgresql PGConnection PGNotification)))

(def notif-name->handler {"something_happened"
                          (fn handle-something [data]
                            (println "Something happened" data))})

(defn listen [{:keys [db-src]}]
  (doto (Thread.
          ^Runnable
          (fn []
            (jdbc/with-db-connection [c db-src]
              (let [pg-conn ^PGConnection (.unwrap ^Connection (:connection c) PGConnection)
                    delta-t-ms 500

                    read-notification-data
                    (fn [^PGNotification notification]
                      ;; Only for notifications that send out proper JSON.
                      (json/read-str (.getParameter notification)
                                     :key-fn keyword))]
                (doseq [notification-name (keys notif-name->handler)]
                  (jdbc/execute! c (str "LISTEN " notification-name)))
                ;; An example implementation:
                ;; https://jdbc.postgresql.org/documentation/server-prepare/#listen--notify.
                ;; Note that it mentions that no other thread will be able to execute statements
                ;; while `.getNotifications` is waiting. This seems to be true only for the
                ;; threads that use the same connection.
                ;;
                ;; An alternative implementation which is more flexible but doesn't have a timeout:
                ;; https://wiki.postgresql.org/wiki/PgNotificationHelper
                ;;
                ;; Yet another implementation, this time it does everything for you but also
                ;; manages its own DB connection:
                ;; https://wiki.postgresql.org/wiki/PgNotificationPoller
                ;;
                ;; An alternative JDBC driver which apparently has async notifications that
                ;; get rid of that problem: http://impossibl.github.io/pgjdbc-ng/
                ;;
                ;; Another potential alternative is Vert.x reactive PostgreSQL client:
                ;; https://vertx.io/docs/vertx-pg-client/java/#_pubsub
                (loop []
                  (doseq [^PGNotification n (.getNotifications pg-conn delta-t-ms)]
                    (when-let [h (notif-name->handler (.getName n))]
                      (h (read-notification-data n))))
                  (when-not (Thread/interrupted)
                    (recur))))))
          "postgresql-notification-listener")
    (.setDaemon true)
    (.start)))

As per the comment, .getNotifications actually "freezes" that connection for all other requests. But since the jdbc/with-connection block is never returned from while the notifications are still being listened to, this particular connection is "owned" by this thread, so that's fine.

[p-himik]

I was just made aware that the LISTEN/NOTIFY mechanism has some limitations that are specified in the source, of which the most salient one seems to be this one:

 * 3. Every backend that is listening on at least one channel registers by
 *	  entering its PID into the array in AsyncQueueControl. It then scans all
 *	  incoming notifications in the central queue and first compares the
 *	  database OID of the notification with its own database OID and then
 *	  compares the notified channel with the list of channels that it listens
 *	  to. In case there is a match it delivers the notification event to its
 *	  frontend.  Non-matching events are simply skipped.

But it doesn't seem like it could be a problem unless someone has lots and lots of databases governed by the same server process and/or lots and lots of notifications they aren't interested in.