fix #6946: reload merchant instance settings from DB on change

8 files changed, 417 insertions, 81 deletions

diff --git a/src/backend/taler-merchant-httpd.c b/src/backend/taler-merchant-httpd.c
index 104d7435..577672b9 100644
--- a/src/backend/taler-merchant-httpd.c
+++ b/src/backend/taler-merchant-httpd.c
@@ -22,6 +22,7 @@
  * @author Florian Dold
  */
 #include "platform.h"
+#include <taler/taler_dbevents.h>
 #include <taler/taler_bank_service.h>
 #include <taler/taler_exchange_service.h>
 #include "taler-merchant-httpd_auditors.h"
@@ -101,6 +102,11 @@ int TMH_force_audit;
 struct TALER_MERCHANTDB_Plugin *TMH_db;
 
 /**
+ * Event handler for instance settings changes.
+ */
+static struct GNUNET_DB_EventHandler *instance_eh;
+
+/**
  * Hashmap pointing at merchant instances by 'id'. An 'id' is
  * just a string that identifies a merchant instance. When a frontend
  * needs to specify an instance to the backend, it does so by 'id'
@@ -241,22 +247,14 @@ TMH_compute_auth (const char *token,
  *
  * @param[in,out] mi merchant instance to update and possibly free
  */
-void
-TMH_instance_decref (struct TMH_MerchantInstance *mi)
+static void
+instance_decref (struct TMH_MerchantInstance *mi)
 {
   struct TMH_WireMethod *wm;
-  struct GNUNET_HashCode h_instance;
 
   mi->rc--;
   if (0 != mi->rc)
     return;
-  GNUNET_CRYPTO_hash (mi->settings.id,
-                      strlen (mi->settings.id),
-                      &h_instance);
-  GNUNET_assert (GNUNET_OK ==
-                 GNUNET_CONTAINER_multihashmap_remove (TMH_by_id_map,
-                                                       &h_instance,
-                                                       mi));
   TMH_force_get_orders_resume (mi);
   while (NULL != (wm = (mi->wm_head)))
   {
@@ -277,22 +275,27 @@ TMH_instance_decref (struct TMH_MerchantInstance *mi)
 
 
 /**
- * Callback that frees all the instances in the hashmap
+ * Callback that frees an instances removing
+ * it from the global hashmap.
  *
  * @param cls closure, NULL
  * @param key current key
  * @param value a `struct TMH_MerchantInstance`
  */
-static int
-instance_free_cb (void *cls,
-                  const struct GNUNET_HashCode *key,
-                  void *value)
+int
+TMH_instance_free_cb (void *cls,
+                      const struct GNUNET_HashCode *key,
+                      void *value)
 {
   struct TMH_MerchantInstance *mi = value;
 
   (void) cls;
   (void) key;
-  TMH_instance_decref (mi);
+  GNUNET_assert (GNUNET_OK ==
+                 GNUNET_CONTAINER_multihashmap_remove (TMH_by_id_map,
+                                                       &mi->h_instance,
+                                                       mi));
+  instance_decref (mi);
   return GNUNET_YES;
 }
 
@@ -732,6 +735,11 @@ do_shutdown (void *cls)
     mhd = NULL;
   }
   TMH_RESERVES_done ();
+  if (NULL != instance_eh)
+  {
+    TMH_db->event_listen_cancel (instance_eh);
+    instance_eh = NULL;
+  }
   if (NULL != TMH_db)
   {
     TALER_MERCHANTDB_plugin_unload (TMH_db);
@@ -755,7 +763,7 @@ do_shutdown (void *cls)
   if (NULL != TMH_by_id_map)
   {
     GNUNET_CONTAINER_multihashmap_iterate (TMH_by_id_map,
-                                           &instance_free_cb,
+                                           &TMH_instance_free_cb,
                                            NULL);
     GNUNET_CONTAINER_multihashmap_destroy (TMH_by_id_map);
     TMH_by_id_map = NULL;
@@ -799,7 +807,7 @@ handle_mhd_completion_callback (void *cls,
   if (NULL != hc->request_body)
     json_decref (hc->request_body);
   if (NULL != hc->instance)
-    TMH_instance_decref (hc->instance);
+    instance_decref (hc->instance);
   memset (&hc->async_scope_id,
           0,
           sizeof (struct GNUNET_AsyncScopeId));
@@ -914,8 +922,6 @@ struct TMH_MerchantInstance *
 TMH_lookup_instance (const char *instance_id)
 {
   struct GNUNET_HashCode h_instance;
-  struct TMH_MerchantInstance *mi;
-  struct GNUNET_TIME_Relative age;
 
   if (NULL == instance_id)
     instance_id = "default";
@@ -928,16 +934,8 @@ TMH_lookup_instance (const char *instance_id)
               instance_id);
   /* We're fine if that returns NULL, the calling routine knows how
      to handle that */
-  mi = GNUNET_CONTAINER_multihashmap_get (TMH_by_id_map,
-                                          &h_instance);
-  if (NULL == mi)
-    return mi;
-  age = GNUNET_TIME_absolute_get_duration (mi->staleness);
-  if (age.rel_value_us > INSTANCE_STALENESS.rel_value_us)
-  {
-    // FIXME: referesh from DB
-  }
-  return mi;
+  return GNUNET_CONTAINER_multihashmap_get (TMH_by_id_map,
+                                            &h_instance);
 }
 
 
@@ -950,7 +948,6 @@ TMH_lookup_instance (const char *instance_id)
 int
 TMH_add_instance (struct TMH_MerchantInstance *mi)
 {
-  struct GNUNET_HashCode h_instance;
   const char *id;
   int ret;
 
@@ -959,13 +956,13 @@ TMH_add_instance (struct TMH_MerchantInstance *mi)
     id = "default";
   GNUNET_CRYPTO_hash (id,
                       strlen (id),
-                      &h_instance);
+                      &mi->h_instance);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Looking for by-id key %s of `%s' in hashmap\n",
-              GNUNET_h2s (&h_instance),
+              GNUNET_h2s (&mi->h_instance),
               id);
   ret = GNUNET_CONTAINER_multihashmap_put (TMH_by_id_map,
-                                           &h_instance,
+                                           &mi->h_instance,
                                            mi,
                                            GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
   if (GNUNET_OK == ret)
@@ -2043,6 +2040,15 @@ add_instance_cb (void *cls,
   struct TMH_MerchantInstance *mi;
 
   (void) cls;
+  mi = TMH_lookup_instance (is->id);
+  if (NULL != mi)
+  {
+    /* (outdated) entry exists, remove old entry */
+    (void) TMH_instance_free_cb (NULL,
+                                 &mi->h_instance,
+                                 mi);
+  }
+
   mi = GNUNET_new (struct TMH_MerchantInstance);
   mi->settings = *is;
   mi->auth = *ias;
@@ -2055,7 +2061,6 @@ add_instance_cb (void *cls,
   else
     mi->deleted = true;
   mi->merchant_pub = *merchant_pub;
-  mi->staleness = GNUNET_TIME_absolute_get ();
   for (unsigned int i = 0; i<accounts_length; i++)
   {
     const struct TALER_MERCHANTDB_AccountDetails *acc = &accounts[i];
@@ -2078,6 +2083,98 @@ add_instance_cb (void *cls,
 
 
 /**
+ * Trigger (re)loading of instance settings from DB.
+ *
+ * @param cls NULL
+ * @param extra ID of the instance that changed, NULL
+ *              to load all instances (will not handle purges!)
+ * @param extra_len number of bytes in @a extra
+ */
+static void
+load_instances (void *cls,
+                const void *extra,
+                size_t extra_len)
+{
+  enum GNUNET_DB_QueryStatus qs;
+  const char *id = extra;
+
+  (void) cls;
+  (void) extra;
+  (void) extra_len;
+  if ( (NULL != extra) &&
+       ( (0 == extra_len) ||
+         ('\0' != id[extra_len - 1]) ) )
+  {
+    GNUNET_break (0); /* bogus notification */
+    extra = NULL;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Received instance settings notification: reload `%s'\n",
+              id);
+  if (NULL == extra)
+  {
+    qs = TMH_db->lookup_instances (TMH_db->cls,
+                                   false,
+                                   &add_instance_cb,
+                                   NULL);
+  }
+  else
+  {
+    struct TMH_MerchantInstance *mi;
+
+    /* This must be done here to handle instance
+       purging, as for purged instances, the DB
+       lookup below will otherwise do nothing */
+    mi = TMH_lookup_instance (id);
+    if (NULL != mi)
+    {
+      (void) TMH_instance_free_cb (NULL,
+                                   &mi->h_instance,
+                                   mi);
+    }
+    qs = TMH_db->lookup_instance (TMH_db->cls,
+                                  id,
+                                  false,
+                                  &add_instance_cb,
+                                  NULL);
+  }
+  if (0 > qs)
+  {
+    GNUNET_break (0);
+    GNUNET_SCHEDULER_shutdown ();
+    return;
+  }
+}
+
+
+/**
+ * A transaction modified an instance setting
+ * (or created/deleted/purged one). Notify all
+ * backends about the change.
+ *
+ * @param id ID of the instance that changed
+ */
+void
+TMH_reload_instances (const char *id)
+{
+  struct GNUNET_DB_EventHeaderP es = {
+    es.size = ntohs (sizeof (es)),
+    es.type = ntohs (sizeof (TALER_DBEVENT_MERCHANT_INSTANCE_SETTINGS))
+  };
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Generating instance settings notification: reload `%s'\n",
+              id);
+  TMH_db->event_notify (TMH_db->cls,
+                        &es,
+                        id,
+                        (NULL == id)
+                        ? 0
+                        : strlen (id) + 1);
+}
+
+
+/**
  * Main function that will be run by the scheduler.
  *
  * @param cls closure
@@ -2175,8 +2272,8 @@ run (void *cls,
     return;
   }
   if (NULL ==
-      (TMH_by_id_map = GNUNET_CONTAINER_multihashmap_create (1,
-                                                             GNUNET_NO)))
+      (TMH_by_id_map = GNUNET_CONTAINER_multihashmap_create (4,
+                                                             GNUNET_YES)))
   {
     GNUNET_SCHEDULER_shutdown ();
     return;
@@ -2195,21 +2292,20 @@ run (void *cls,
     GNUNET_SCHEDULER_shutdown ();
     return;
   }
-  /* load instances */
   {
-    enum GNUNET_DB_QueryStatus qs;
-
-    qs = TMH_db->lookup_instances (TMH_db->cls,
-                                   false,
-                                   &add_instance_cb,
-                                   NULL);
-    if (0 > qs)
-    {
-      GNUNET_break (0);
-      GNUNET_SCHEDULER_shutdown ();
-      return;
-    }
+    struct GNUNET_DB_EventHeaderP es = {
+      es.size = ntohs (sizeof (es)),
+      es.type = ntohs (sizeof (TALER_DBEVENT_MERCHANT_INSTANCE_SETTINGS))
+    };
+
+    instance_eh = TMH_db->event_listen (TMH_db->cls,
+                                        &es,
+                                        &load_instances,
+                                        NULL);
   }
+  load_instances (NULL,
+                  NULL,
+                  0);
   /* start watching reserves */
   TMH_RESERVES_init ();
   fh = TALER_MHD_bind (cfg,
diff --git a/src/backend/taler-merchant-httpd.h b/src/backend/taler-merchant-httpd.h
index 03e62d2c..64bd8ad1 100644
--- a/src/backend/taler-merchant-httpd.h
+++ b/src/backend/taler-merchant-httpd.h
@@ -1,6 +1,6 @@
 /*
   This file is part of TALER
-  Copyright (C) 2014-2018 Taler Systems SA
+  Copyright (C) 2014-2021 Taler Systems SA
 
   TALER is free software; you can redistribute it and/or modify it under the
   terms of the GNU General Public License as published by the Free Software
@@ -108,6 +108,11 @@ struct TMH_MerchantInstance
   struct TMH_WireMethod *wm_tail;
 
   /**
+   * Hash of the instance ID, key for the DHT.
+   */
+  struct GNUNET_HashCode h_instance;
+
+  /**
    * Head of DLL of long-polling GET /orders requests of this instance.
    */
   struct TMH_PendingOrder *po_head;
@@ -138,12 +143,6 @@ struct TMH_MerchantInstance
   struct TALER_MERCHANTDB_InstanceAuthSettings auth;
 
   /**
-   * Timestamp when we last updated this instance's data
-   * with the information from the database.
-   */
-  struct GNUNET_TIME_Absolute staleness;
-
-  /**
    * Reference counter on this structure. Only destroyed if the
    * counter hits zero.
    */
@@ -512,12 +511,17 @@ TMH_long_poll_resume2 (const char *session_id,
 
 
 /**
- * Decrement reference counter of @a mi, and free if it hits zero.
+ * Callback that frees an instances removing
+ * it from the global hashmap.
  *
- * @param[in,out] mi merchant instance to update and possibly free
+ * @param cls closure, NULL
+ * @param key current key
+ * @param value a `struct TMH_MerchantInstance`
  */
-void
-TMH_instance_decref (struct TMH_MerchantInstance *mi);
+int
+TMH_instance_free_cb (void *cls,
+                      const struct GNUNET_HashCode *key,
+                      void *value);
 
 
 /**
@@ -541,6 +545,17 @@ TMH_lookup_instance (const char *instance_id);
 
 
 /**
+ * A transaction modified an instance setting
+ * (or created/deleted/purged one). Notify all
+ * backends about the change.
+ *
+ * @param id ID of the instance that changed
+ */
+void
+TMH_reload_instances (const char *id);
+
+
+/**
  * Check that @a token hashes to @a hash under @a salt for
  * merchant instance authentication.
  *
diff --git a/src/backend/taler-merchant-httpd_private-delete-instances-ID.c b/src/backend/taler-merchant-httpd_private-delete-instances-ID.c
index 0de88164..bb0c8e56 100644
--- a/src/backend/taler-merchant-httpd_private-delete-instances-ID.c
+++ b/src/backend/taler-merchant-httpd_private-delete-instances-ID.c
@@ -1,6 +1,6 @@
 /*
   This file is part of TALER
-  (C) 2020 Taler Systems SA
+  (C) 2020-2021 Taler Systems SA
 
   TALER is free software; you can redistribute it and/or modify it under the
   terms of the GNU Affero General Public License as published by the Free Software
@@ -73,8 +73,11 @@ delete_instances_ID (struct TMH_MerchantInstance *mi,
                                        ? "Instance unknown"
                                        : "Private key unknown");
   case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
+    TMH_reload_instances (mi->settings.id);
     if (purge)
-      TMH_instance_decref (mi);
+      TMH_instance_free_cb (NULL,
+                            &mi->h_instance,
+                            mi);
     else
       mi->deleted = true;
     return TALER_MHD_reply_static (connection,
diff --git a/src/backend/taler-merchant-httpd_private-patch-instances-ID.c b/src/backend/taler-merchant-httpd_private-patch-instances-ID.c
index 65f1760f..482ef048 100644
--- a/src/backend/taler-merchant-httpd_private-patch-instances-ID.c
+++ b/src/backend/taler-merchant-httpd_private-patch-instances-ID.c
@@ -1,6 +1,6 @@
 /*
   This file is part of TALER
-  (C) 2020 Taler Systems SA
+  (C) 2020-2021 Taler Systems SA
 
   TALER is free software; you can redistribute it and/or modify
   it under the terms of the GNU Affero General Public License as
@@ -370,6 +370,7 @@ giveup:
   }
 
   GNUNET_JSON_parse_free (spec);
+  TMH_reload_instances (mi->settings.id);
   return TALER_MHD_reply_static (connection,
                                  MHD_HTTP_NO_CONTENT,
                                  NULL,
diff --git a/src/backend/taler-merchant-httpd_private-post-instances-ID-auth.c b/src/backend/taler-merchant-httpd_private-post-instances-ID-auth.c
index 17316348..eef710b6 100644
--- a/src/backend/taler-merchant-httpd_private-post-instances-ID-auth.c
+++ b/src/backend/taler-merchant-httpd_private-post-instances-ID-auth.c
@@ -108,25 +108,25 @@ post_instances_ID_auth (struct TMH_MerchantInstance *mi,
 
         switch (qs)
         {
-          case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
-            /* Instance got purged. */
-            TMH_db->rollback (TMH_db->cls);
-            return TALER_MHD_reply_with_error (connection,
-                                              MHD_HTTP_UNAUTHORIZED,
-                                              TALER_EC_GENERIC_DB_COMMIT_FAILED,
-                                              NULL);
-          case GNUNET_DB_STATUS_SOFT_ERROR:
-            TMH_db->rollback (TMH_db->cls);
-            goto retry;
-          case GNUNET_DB_STATUS_HARD_ERROR:
-            TMH_db->rollback (TMH_db->cls);
-            return TALER_MHD_reply_with_error (connection,
+        case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
+          /* Instance got purged. */
+          TMH_db->rollback (TMH_db->cls);
+          return TALER_MHD_reply_with_error (connection,
+                                             MHD_HTTP_UNAUTHORIZED,
+                                             TALER_EC_GENERIC_DB_COMMIT_FAILED,
+                                             NULL);
+        case GNUNET_DB_STATUS_SOFT_ERROR:
+          TMH_db->rollback (TMH_db->cls);
+          goto retry;
+        case GNUNET_DB_STATUS_HARD_ERROR:
+          TMH_db->rollback (TMH_db->cls);
+          return TALER_MHD_reply_with_error (connection,
                                              MHD_HTTP_INTERNAL_SERVER_ERROR,
                                              TALER_EC_GENERIC_DB_FETCH_FAILED,
                                              NULL);
-          case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
-            /* Success! */
-            break;
+        case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
+          /* Success! */
+          break;
         }
 
         if (GNUNET_OK !=
@@ -183,6 +183,7 @@ retry:
        for the default instance. */
     GNUNET_assert (NULL == TMH_default_auth);
   }
+  TMH_reload_instances (mi->settings.id);
   return TALER_MHD_reply_static (connection,
                                  MHD_HTTP_NO_CONTENT,
                                  NULL,
diff --git a/src/backend/taler-merchant-httpd_private-post-instances.c b/src/backend/taler-merchant-httpd_private-post-instances.c
index 2d3f660c..7951652a 100644
--- a/src/backend/taler-merchant-httpd_private-post-instances.c
+++ b/src/backend/taler-merchant-httpd_private-post-instances.c
@@ -459,6 +459,7 @@ retry:
     /* Finally, also update our running process */
     GNUNET_assert (GNUNET_OK ==
                    TMH_add_instance (mi));
+    TMH_reload_instances (mi->settings.id);
   }
   GNUNET_JSON_parse_free (spec);
   GNUNET_free (TMH_default_auth); /* clear it: user just either created default
diff --git a/src/backenddb/plugin_merchantdb_postgres.c b/src/backenddb/plugin_merchantdb_postgres.c
index b882b24b..0cb2149a 100644
--- a/src/backenddb/plugin_merchantdb_postgres.c
+++ b/src/backenddb/plugin_merchantdb_postgres.c
@@ -154,6 +154,84 @@ postgres_create_tables (void *cls)
 
 
 /**
+ * Register callback to be invoked on events of type @a es.
+ *
+ * Unlike many other calls, this function is thread-safe
+ * and may be called from threads that are different
+ * from the one that setup @a db. However, the @a cb
+ * will always be called from the thread that runs
+ * #GNUNET_PQ_event_do_poll() or the GNUnet scheduler.
+ *
+ * @param db database context to use
+ * @param es specification of the event to listen for
+ * @param cb function to call when the event happens, possibly
+ *         multiple times (until #GNUNET_PQ_event_listen_cancel() is invoked)
+ * @param cb_cls closure for @a cb
+ * @return handle useful to cancel the listener
+ */
+static struct GNUNET_DB_EventHandler *
+postgres_event_listen (void *cls,
+                       const struct GNUNET_DB_EventHeaderP *es,
+                       GNUNET_DB_EventCallback cb,
+                       void *cb_cls)
+{
+  struct PostgresClosure *pg = cls;
+
+  return GNUNET_PQ_event_listen (pg->conn,
+                                 es,
+                                 cb,
+                                 cb_cls);
+}
+
+
+/**
+ * Stop notifications.
+ *
+ * Unlike many other calls, this function is thread-safe
+ * and may be called from threads that are different
+ * from the one that setup @a db. However, the @a cb
+ * will always be called from the thread that runs
+ * #GNUNET_PQ_event_do_poll() or the GNUnet scheduler.
+ *
+ * @param eh handle to unregister.
+ */
+static void
+postgres_event_listen_cancel (struct GNUNET_DB_EventHandler *eh)
+{
+  GNUNET_PQ_event_listen_cancel (eh);
+}
+
+
+/**
+ * Notify all that listen on @a es of an event.
+ *
+ * Unlike many other calls, this function is thread-safe
+ * and may be called from threads that are different
+ * from the one that setup @a db. However, the @a cb
+ * will always be called from the thread that runs
+ * #GNUNET_PQ_event_do_poll() or the GNUnet scheduler.
+ *
+ * @param db database context to use
+ * @param es specification of the event to generate
+ * @param extra additional event data provided
+ * @param extra_size number of bytes in @a extra
+ */
+static void
+postgres_event_notify (void *cls,
+                       const struct GNUNET_DB_EventHeaderP *es,
+                       const void *extra,
+                       size_t extra_size)
+{
+  struct PostgresClosure *pg = cls;
+
+  return GNUNET_PQ_event_notify (pg->conn,
+                                 es,
+                                 extra,
+                                 extra_size);
+}
+
+
+/**
  * Do a pre-flight check that we are not in an uncommitted transaction.
  * If we are, die.
  * Does not return anything, as we will continue regardless of the outcome.
@@ -597,6 +675,47 @@ postgres_lookup_instances (void *cls,
 
 
 /**
+ * Lookup all one of the instances this backend has configured.
+ *
+ * @param cls closure
+ * @param id instance ID to resolve
+ * @param active_only only find 'active' instances
+ * @param cb function to call on all instances found
+ * @param cb_cls closure for @a cb
+ */
+static enum GNUNET_DB_QueryStatus
+postgres_lookup_instance (void *cls,
+                          const char *id,
+                          bool active_only,
+                          TALER_MERCHANTDB_InstanceCallback cb,
+                          void *cb_cls)
+{
+  struct PostgresClosure *pg = cls;
+  struct LookupInstancesContext lic = {
+    .cb = cb,
+    .cb_cls = cb_cls,
+    .active_only = active_only,
+    .pg = pg
+  };
+  struct GNUNET_PQ_QueryParam params[] = {
+    GNUNET_PQ_query_param_string (id),
+    GNUNET_PQ_query_param_end
+  };
+  enum GNUNET_DB_QueryStatus qs;
+
+  check_connection (pg);
+  qs = GNUNET_PQ_eval_prepared_multi_select (pg->conn,
+                                             "lookup_instance",
+                                             params,
+                                             &lookup_instances_cb,
+                                             &lic);
+  if (0 > lic.qs)
+    return lic.qs;
+  return qs;
+}
+
+
+/**
  * Lookup authentication data of an instance.
  *
  * @param cls closure
@@ -6419,6 +6538,27 @@ postgres_connect (void *cls)
                             ",default_pay_delay"
                             " FROM merchant_instances",
                             0),
+    /* for postgres_lookup_instance() */
+    GNUNET_PQ_make_prepare ("lookup_instance",
+                            "SELECT"
+                            " merchant_serial"
+                            ",merchant_pub"
+                            ",auth_hash"
+                            ",auth_salt"
+                            ",merchant_id"
+                            ",merchant_name"
+                            ",address"
+                            ",jurisdiction"
+                            ",default_max_deposit_fee_val"
+                            ",default_max_deposit_fee_frac"
+                            ",default_max_wire_fee_val"
+                            ",default_max_wire_fee_frac"
+                            ",default_wire_fee_amortization"
+                            ",default_wire_transfer_delay"
+                            ",default_pay_delay"
+                            " FROM merchant_instances"
+                            " WHERE merchant_id=$1",
+                            1),
     /* for postgres_insert_instance() */
     GNUNET_PQ_make_prepare ("insert_instance",
                             "INSERT INTO merchant_instances"
@@ -8823,6 +8963,7 @@ postgres_connect (void *cls)
                                          ps);
   if (NULL == pg->conn)
     return GNUNET_SYSERR;
+  GNUNET_PQ_event_scheduler_start (pg->conn);
   return GNUNET_OK;
 }
 
@@ -8858,7 +8999,6 @@ libtaler_plugin_merchantdb_postgres_init (void *cls)
       TALER_config_get_currency (cfg,
                                  &pg->currency))
   {
-    GNUNET_PQ_disconnect (pg->conn);
     GNUNET_free (pg->sql_dir);
     GNUNET_free (pg);
     return NULL;
@@ -8868,12 +9008,16 @@ libtaler_plugin_merchantdb_postgres_init (void *cls)
   plugin->connect = &postgres_connect;
   plugin->create_tables = &postgres_create_tables;
   plugin->drop_tables = &postgres_drop_tables;
+  plugin->event_listen = &postgres_event_listen;
+  plugin->event_listen_cancel = &postgres_event_listen_cancel;
+  plugin->event_notify = &postgres_event_notify;
   plugin->preflight = &postgres_preflight;
   plugin->start = &postgres_start;
   plugin->start_read_committed = &postgres_start_read_committed;
   plugin->rollback = &postgres_rollback;
   plugin->commit = &postgres_commit;
   plugin->lookup_instances = &postgres_lookup_instances;
+  plugin->lookup_instance = &postgres_lookup_instance;
   plugin->lookup_instance_auth = &postgres_lookup_instance_auth;
   plugin->insert_instance = &postgres_insert_instance;
   plugin->insert_account = &postgres_insert_account;
@@ -8967,6 +9111,7 @@ libtaler_plugin_merchantdb_postgres_done (void *cls)
 
   if (NULL != pg->conn)
   {
+    GNUNET_PQ_event_scheduler_stop (pg->conn);
     GNUNET_PQ_disconnect (pg->conn);
     pg->conn = NULL;
   }
diff --git a/src/include/taler_merchantdb_plugin.h b/src/include/taler_merchantdb_plugin.h
index f8ac0eaf..d5c4e9b1 100644
--- a/src/include/taler_merchantdb_plugin.h
+++ b/src/include/taler_merchantdb_plugin.h
@@ -717,6 +717,64 @@ struct TALER_MERCHANTDB_Plugin
   (*create_tables) (void *cls);
 
   /**
+   * Register callback to be invoked on events of type @a es.
+   *
+   * Unlike many other calls, this function is thread-safe
+   * and may be called from threads that are different
+   * from the one that setup @a db. However, the @a cb
+   * will always be called from the thread that runs
+   * #GNUNET_PQ_event_do_poll() or the GNUnet scheduler.
+   *
+   * @param db database context to use
+   * @param es specification of the event to listen for
+   * @param cb function to call when the event happens, possibly
+   *         multiple times (until #GNUNET_PQ_event_listen_cancel() is invoked)
+   * @param cb_cls closure for @a cb
+   * @return handle useful to cancel the listener
+   */
+  struct GNUNET_DB_EventHandler *
+  (*event_listen)(void *cls,
+                  const struct GNUNET_DB_EventHeaderP *es,
+                  GNUNET_DB_EventCallback cb,
+                  void *cb_cls);
+
+  /**
+   * Stop notifications.
+   *
+   * Unlike many other calls, this function is thread-safe
+   * and may be called from threads that are different
+   * from the one that setup @a db. However, the @a cb
+   * will always be called from the thread that runs
+   * #GNUNET_PQ_event_do_poll() or the GNUnet scheduler.
+   *
+   * @param eh handle to unregister.
+   */
+  void
+  (*event_listen_cancel)(struct GNUNET_DB_EventHandler *eh);
+
+
+  /**
+   * Notify all that listen on @a es of an event.
+   *
+   * Unlike many other calls, this function is thread-safe
+   * and may be called from threads that are different
+   * from the one that setup @a db. However, the @a cb
+   * will always be called from the thread that runs
+   * #GNUNET_PQ_event_do_poll() or the GNUnet scheduler.
+   *
+   * @param db database context to use
+   * @param es specification of the event to generate
+   * @param extra additional event data provided
+   * @param extra_size number of bytes in @a extra
+   */
+  void
+  (*event_notify)(void *cls,
+                  const struct GNUNET_DB_EventHeaderP *es,
+                  const void *extra,
+                  size_t extra_size);
+
+
+  /**
    * Do a pre-flight check that we are not in an uncommitted transaction.  If
    * we are, die.  Does not return anything, as we will continue regardless of
    * the outcome.
@@ -783,6 +841,22 @@ struct TALER_MERCHANTDB_Plugin
                       void *cb_cls);
 
   /**
+   * Lookup one of the instances this backend has configured.
+   *
+   * @param cls closure
+   * @param id ID of instance to look up
+   * @param active_only only find 'active' instances
+   * @param cb function to call on all instances found
+   * @param cb_cls closure for @a cb
+   */
+  enum GNUNET_DB_QueryStatus
+  (*lookup_instance)(void *cls,
+                     const char *id,
+                     bool active_only,
+                     TALER_MERCHANTDB_InstanceCallback cb,
+                     void *cb_cls);
+
+  /**
    * Lookup authentication data of an instance.
    *
    * @param cls closure