ctdbd_conn: move CTDB_CONTROL_ENABLE_SEQNUM control to db_open_ctdb
[vlendec/samba-autobuild/.git] / source3 / lib / messages.c
index ef8e83dd6e61fa6b13133dd6ffde4cec8d5e97e3..b94a6965eb888ea4df0bc35e755f63b71a04f2d2 100644 (file)
 */
 
 #include "includes.h"
+#include "lib/util/server_id.h"
 #include "dbwrap/dbwrap.h"
 #include "serverid.h"
 #include "messages.h"
 #include "lib/util/tevent_unix.h"
 #include "lib/background.h"
 #include "lib/messages_dgm.h"
+#include "lib/messages_ctdbd.h"
 #include "lib/util/iov_buf.h"
 #include "lib/util/server_id_db.h"
 #include "lib/messages_dgm_ref.h"
@@ -72,10 +74,10 @@ struct messaging_context {
        struct messaging_callback *callbacks;
 
        struct tevent_req **new_waiters;
-       unsigned num_new_waiters;
+       size_t num_new_waiters;
 
        struct tevent_req **waiters;
-       unsigned num_waiters;
+       size_t num_waiters;
 
        void *msg_dgm_ref;
        struct messaging_backend *remote;
@@ -83,7 +85,10 @@ struct messaging_context {
        struct server_id_db *names_db;
 };
 
+static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
+                                              struct messaging_rec *rec);
 static void messaging_dispatch_rec(struct messaging_context *msg_ctx,
+                                  struct tevent_context *ev,
                                   struct messaging_rec *rec);
 
 /****************************************************************************
@@ -105,109 +110,55 @@ static void ping_message(struct messaging_context *msg_ctx,
        messaging_send(msg_ctx, src, MSG_PONG, data);
 }
 
-/****************************************************************************
- Register/replace a dispatch function for a particular message type.
- JRA changed Dec 13 2006. Only one message handler now permitted per type.
- *NOTE*: Dispatch functions must be able to cope with incoming
- messages on an *odd* byte boundary.
-****************************************************************************/
-
-struct msg_all {
-       struct messaging_context *msg_ctx;
-       int msg_type;
-       uint32_t msg_flag;
-       const void *buf;
-       size_t len;
-       int n_sent;
-};
-
-/****************************************************************************
- Send one of the messages for the broadcast.
-****************************************************************************/
-
-static int traverse_fn(struct db_record *rec, const struct server_id *id,
-                      uint32_t msg_flags, void *state)
+struct messaging_rec *messaging_rec_create(
+       TALLOC_CTX *mem_ctx, struct server_id src, struct server_id dst,
+       uint32_t msg_type, const struct iovec *iov, int iovlen,
+       const int *fds, size_t num_fds)
 {
-       struct msg_all *msg_all = (struct msg_all *)state;
-       NTSTATUS status;
-
-       /* Don't send if the receiver hasn't registered an interest. */
+       ssize_t buflen;
+       uint8_t *buf;
+       struct messaging_rec *result;
 
-       if((msg_flags & msg_all->msg_flag) == 0) {
-               return 0;
+       if (num_fds > INT8_MAX) {
+               return NULL;
        }
 
-       /* If the msg send fails because the pid was not found (i.e. smbd died), 
-        * the msg has already been deleted from the messages.tdb.*/
-
-       status = messaging_send_buf(msg_all->msg_ctx, *id, msg_all->msg_type,
-                                   (const uint8_t *)msg_all->buf, msg_all->len);
+       buflen = iov_buflen(iov, iovlen);
+       if (buflen == -1) {
+               return NULL;
+       }
+       buf = talloc_array(mem_ctx, uint8_t, buflen);
+       if (buf == NULL) {
+               return NULL;
+       }
+       iov_buf(iov, iovlen, buf, buflen);
 
-       if (NT_STATUS_EQUAL(status, NT_STATUS_INVALID_HANDLE)) {
-               struct server_id_buf idbuf;
+       {
+               struct messaging_rec rec;
+               int64_t fds64[num_fds];
+               size_t i;
 
-               /*
-                * If the pid was not found delete the entry from
-                * serverid.tdb
-                */
+               for (i=0; i<num_fds; i++) {
+                       fds64[i] = fds[i];
+               }
 
-               DEBUG(2, ("pid %s doesn't exist\n",
-                         server_id_str_buf(*id, &idbuf)));
+               rec = (struct messaging_rec) {
+                       .msg_version = MESSAGE_VERSION, .msg_type = msg_type,
+                       .src = src, .dest = dst,
+                       .buf.data = buf, .buf.length = buflen,
+                       .num_fds = num_fds, .fds = fds64,
+               };
 
-               dbwrap_record_delete(rec);
+               result = messaging_rec_dup(mem_ctx, &rec);
        }
-       msg_all->n_sent++;
-       return 0;
-}
 
-/**
- * Send a message to all smbd processes.
- *
- * It isn't very efficient, but should be OK for the sorts of
- * applications that use it. When we need efficient broadcast we can add
- * it.
- *
- * @param n_sent Set to the number of messages sent.  This should be
- * equal to the number of processes, but be careful for races.
- *
- * @retval True for success.
- **/
-bool message_send_all(struct messaging_context *msg_ctx,
-                     int msg_type,
-                     const void *buf, size_t len,
-                     int *n_sent)
-{
-       struct msg_all msg_all;
-
-       msg_all.msg_type = msg_type;
-       if (msg_type < 0x100) {
-               msg_all.msg_flag = FLAG_MSG_GENERAL;
-       } else if (msg_type > 0x100 && msg_type < 0x200) {
-               msg_all.msg_flag = FLAG_MSG_NMBD;
-       } else if (msg_type > 0x200 && msg_type < 0x300) {
-               msg_all.msg_flag = FLAG_MSG_PRINT_GENERAL;
-       } else if (msg_type > 0x300 && msg_type < 0x400) {
-               msg_all.msg_flag = FLAG_MSG_SMBD;
-       } else if (msg_type > 0x400 && msg_type < 0x600) {
-               msg_all.msg_flag = FLAG_MSG_WINBIND;
-       } else if (msg_type > 4000 && msg_type < 5000) {
-               msg_all.msg_flag = FLAG_MSG_DBWRAP;
-       } else {
-               return false;
-       }
+       TALLOC_FREE(buf);
 
-       msg_all.buf = buf;
-       msg_all.len = len;
-       msg_all.n_sent = 0;
-       msg_all.msg_ctx = msg_ctx;
-
-       serverid_traverse(traverse_fn, &msg_all);
-       if (n_sent)
-               *n_sent = msg_all.n_sent;
-       return true;
+       return result;
 }
 
-static void messaging_recv_cb(const uint8_t *msg, size_t msg_len,
+static void messaging_recv_cb(struct tevent_context *ev,
+                             const uint8_t *msg, size_t msg_len,
                              int *fds, size_t num_fds,
                              void *private_data)
 {
@@ -219,12 +170,12 @@ static void messaging_recv_cb(const uint8_t *msg, size_t msg_len,
        size_t i;
 
        if (msg_len < MESSAGE_HDR_LENGTH) {
-               DEBUG(1, ("message too short: %u\n", (unsigned)msg_len));
+               DBG_WARNING("message too short: %zu\n", msg_len);
                goto close_fail;
        }
 
        if (num_fds > INT8_MAX) {
-               DEBUG(1, ("too many fds: %u\n", (unsigned)num_fds));
+               DBG_WARNING("too many fds: %zu\n", num_fds);
                goto close_fail;
        }
 
@@ -247,13 +198,11 @@ static void messaging_recv_cb(const uint8_t *msg, size_t msg_len,
 
        message_hdr_get(&rec.msg_type, &rec.src, &rec.dest, msg);
 
-       DEBUG(10, ("%s: Received message 0x%x len %u (num_fds:%u) from %s\n",
-                  __func__, (unsigned)rec.msg_type,
-                  (unsigned)rec.buf.length,
-                  (unsigned)num_fds,
-                  server_id_str_buf(rec.src, &idbuf)));
+       DBG_DEBUG("Received message 0x%x len %zu (num_fds:%zu) from %s\n",
+                 (unsigned)rec.msg_type, rec.buf.length, num_fds,
+                 server_id_str_buf(rec.src, &idbuf));
 
-       messaging_dispatch_rec(msg_ctx, &rec);
+       messaging_dispatch_rec(msg_ctx, ev, &rec);
        return;
 
 close_fail:
@@ -264,7 +213,7 @@ close_fail:
 
 static int messaging_context_destructor(struct messaging_context *ctx)
 {
-       unsigned i;
+       size_t i;
 
        for (i=0; i<ctx->num_new_waiters; i++) {
                if (ctx->new_waiters[i] != NULL) {
@@ -287,67 +236,77 @@ static const char *private_path(const char *name)
        return talloc_asprintf(talloc_tos(), "%s/%s", lp_private_dir(), name);
 }
 
-struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx, 
-                                        struct tevent_context *ev)
+static NTSTATUS messaging_init_internal(TALLOC_CTX *mem_ctx,
+                                       struct tevent_context *ev,
+                                       struct messaging_context **pmsg_ctx)
 {
+       TALLOC_CTX *frame;
        struct messaging_context *ctx;
+       NTSTATUS status = NT_STATUS_UNSUCCESSFUL;
        int ret;
        const char *lck_path;
        const char *priv_path;
        bool ok;
 
-       if (!(ctx = talloc_zero(mem_ctx, struct messaging_context))) {
-               return NULL;
-       }
-
-       ctx->id = (struct server_id) {
-               .pid = getpid(), .vnn = NONCLUSTER_VNN
-       };
-
-       ctx->event_ctx = ev;
-
-       sec_init();
-
        lck_path = lock_path("msg.lock");
        if (lck_path == NULL) {
-               TALLOC_FREE(ctx);
-               return NULL;
+               return NT_STATUS_NO_MEMORY;
        }
 
-       ok = directory_create_or_exist_strict(lck_path, sec_initial_uid(),
+       ok = directory_create_or_exist_strict(lck_path,
+                                             sec_initial_uid(),
                                              0755);
        if (!ok) {
-               DEBUG(10, ("%s: Could not create lock directory: %s\n",
-                          __func__, strerror(errno)));
-               TALLOC_FREE(ctx);
-               return NULL;
+               DBG_DEBUG("Could not create lock directory: %s\n",
+                         strerror(errno));
+               return NT_STATUS_ACCESS_DENIED;
        }
 
        priv_path = private_path("msg.sock");
        if (priv_path == NULL) {
-               TALLOC_FREE(ctx);
-               return NULL;
+               return NT_STATUS_NO_MEMORY;
        }
 
        ok = directory_create_or_exist_strict(priv_path, sec_initial_uid(),
                                              0700);
        if (!ok) {
-               DEBUG(10, ("%s: Could not create msg directory: %s\n",
-                          __func__, strerror(errno)));
-               TALLOC_FREE(ctx);
-               return NULL;
+               DBG_DEBUG("Could not create msg directory: %s\n",
+                         strerror(errno));
+               return NT_STATUS_ACCESS_DENIED;
        }
 
-       ctx->msg_dgm_ref = messaging_dgm_ref(
-               ctx, ctx->event_ctx, &ctx->id.unique_id,
-               priv_path, lck_path, messaging_recv_cb, ctx, &ret);
+       frame = talloc_stackframe();
+       if (frame == NULL) {
+               return NT_STATUS_NO_MEMORY;
+       }
 
+       ctx = talloc_zero(frame, struct messaging_context);
+       if (ctx == NULL) {
+               status = NT_STATUS_NO_MEMORY;
+               goto done;
+       }
+
+       ctx->id = (struct server_id) {
+               .pid = getpid(), .vnn = NONCLUSTER_VNN
+       };
+
+       ctx->event_ctx = ev;
+
+       sec_init();
+
+       ctx->msg_dgm_ref = messaging_dgm_ref(ctx,
+                                            ctx->event_ctx,
+                                            &ctx->id.unique_id,
+                                            priv_path,
+                                            lck_path,
+                                            messaging_recv_cb,
+                                            ctx,
+                                            &ret);
        if (ctx->msg_dgm_ref == NULL) {
                DEBUG(2, ("messaging_dgm_ref failed: %s\n", strerror(ret)));
-               TALLOC_FREE(ctx);
-               return NULL;
+               status = map_nt_error_from_unix(ret);
+               goto done;
        }
-
        talloc_set_destructor(ctx, messaging_context_destructor);
 
        if (lp_clustering()) {
@@ -356,19 +315,21 @@ struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
                if (ret != 0) {
                        DEBUG(2, ("messaging_ctdbd_init failed: %s\n",
                                  strerror(ret)));
-                       TALLOC_FREE(ctx);
-                       return NULL;
+                       status = map_nt_error_from_unix(ret);
+                       goto done;
                }
        }
        ctx->id.vnn = get_my_vnn();
 
-       ctx->names_db = server_id_db_init(
-               ctx, ctx->id, lp_lock_directory(), 0,
-               TDB_INCOMPATIBLE_HASH|TDB_CLEAR_IF_FIRST);
+       ctx->names_db = server_id_db_init(ctx,
+                                         ctx->id,
+                                         lp_lock_directory(),
+                                         0,
+                                         TDB_INCOMPATIBLE_HASH|TDB_CLEAR_IF_FIRST);
        if (ctx->names_db == NULL) {
-               DEBUG(10, ("%s: server_id_db_init failed\n", __func__));
-               TALLOC_FREE(ctx);
-               return NULL;
+               DBG_DEBUG("server_id_db_init failed\n");
+               status = NT_STATUS_NO_MEMORY;
+               goto done;
        }
 
        messaging_register(ctx, NULL, MSG_PING, ping_message);
@@ -379,9 +340,45 @@ struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
        register_dmalloc_msgs(ctx);
        debug_register_msgs(ctx);
 
+       {
+               struct server_id_buf tmp;
+               DBG_DEBUG("my id: %s\n", server_id_str_buf(ctx->id, &tmp));
+       }
+
+       *pmsg_ctx = talloc_steal(mem_ctx, ctx);
+
+       status = NT_STATUS_OK;
+done:
+       TALLOC_FREE(frame);
+
+       return status;
+}
+
+struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
+                                        struct tevent_context *ev)
+{
+       struct messaging_context *ctx = NULL;
+       NTSTATUS status;
+
+       status = messaging_init_internal(mem_ctx,
+                                        ev,
+                                        &ctx);
+       if (!NT_STATUS_IS_OK(status)) {
+               return NULL;
+       }
+
        return ctx;
 }
 
+NTSTATUS messaging_init_client(TALLOC_CTX *mem_ctx,
+                              struct tevent_context *ev,
+                              struct messaging_context **pmsg_ctx)
+{
+       return messaging_init_internal(mem_ctx,
+                                       ev,
+                                       pmsg_ctx);
+}
+
 struct server_id messaging_server_id(const struct messaging_context *msg_ctx)
 {
        return msg_ctx->id;
@@ -393,6 +390,7 @@ struct server_id messaging_server_id(const struct messaging_context *msg_ctx)
 NTSTATUS messaging_reinit(struct messaging_context *msg_ctx)
 {
        int ret;
+       char *lck_path;
 
        TALLOC_FREE(msg_ctx->msg_dgm_ref);
 
@@ -400,9 +398,14 @@ NTSTATUS messaging_reinit(struct messaging_context *msg_ctx)
                .pid = getpid(), .vnn = msg_ctx->id.vnn
        };
 
+       lck_path = lock_path("msg.lock");
+       if (lck_path == NULL) {
+               return NT_STATUS_NO_MEMORY;
+       }
+
        msg_ctx->msg_dgm_ref = messaging_dgm_ref(
                msg_ctx, msg_ctx->event_ctx, &msg_ctx->id.unique_id,
-               private_path("msg.sock"), lock_path("msg.lock"),
+               private_path("msg.sock"), lck_path,
                messaging_recv_cb, msg_ctx, &ret);
 
        if (msg_ctx->msg_dgm_ref == NULL) {
@@ -410,11 +413,9 @@ NTSTATUS messaging_reinit(struct messaging_context *msg_ctx)
                return map_nt_error_from_unix(ret);
        }
 
-       TALLOC_FREE(msg_ctx->remote);
-
        if (lp_clustering()) {
-               ret = messaging_ctdbd_init(msg_ctx, msg_ctx,
-                                          &msg_ctx->remote);
+               ret = messaging_ctdbd_reinit(msg_ctx, msg_ctx,
+                                            msg_ctx->remote);
 
                if (ret != 0) {
                        DEBUG(1, ("messaging_ctdbd_init failed: %s\n",
@@ -507,10 +508,12 @@ NTSTATUS messaging_send(struct messaging_context *msg_ctx,
                        struct server_id server, uint32_t msg_type,
                        const DATA_BLOB *data)
 {
-       struct iovec iov;
+       struct iovec iov = {0};
 
-       iov.iov_base = data->data;
-       iov.iov_len = data->length;
+       if (data != NULL) {
+               iov.iov_base = data->data;
+               iov.iov_len = data->length;
+       };
 
        return messaging_send_iov(msg_ctx, server, msg_type, &iov, 1, NULL, 0);
 }
@@ -523,6 +526,59 @@ NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
        return messaging_send(msg_ctx, server, msg_type, &blob);
 }
 
+struct messaging_post_state {
+       struct messaging_context *msg_ctx;
+       struct messaging_rec *rec;
+};
+
+static void messaging_post_handler(struct tevent_context *ev,
+                                  struct tevent_immediate *ti,
+                                  void *private_data);
+
+static int messaging_post_self(struct messaging_context *msg_ctx,
+                              struct server_id src, struct server_id dst,
+                              uint32_t msg_type,
+                              const struct iovec *iov, int iovlen,
+                              const int *fds, size_t num_fds)
+{
+       struct tevent_immediate *ti;
+       struct messaging_post_state *state;
+
+       state = talloc(msg_ctx, struct messaging_post_state);
+       if (state == NULL) {
+               return ENOMEM;
+       }
+       state->msg_ctx = msg_ctx;
+
+       ti = tevent_create_immediate(state);
+       if (ti == NULL) {
+               goto fail;
+       }
+       state->rec = messaging_rec_create(
+               state, src, dst, msg_type, iov, iovlen, fds, num_fds);
+       if (state->rec == NULL) {
+               goto fail;
+       }
+
+       tevent_schedule_immediate(ti, msg_ctx->event_ctx,
+                                 messaging_post_handler, state);
+       return 0;
+
+fail:
+       TALLOC_FREE(state);
+       return ENOMEM;
+}
+
+static void messaging_post_handler(struct tevent_context *ev,
+                                  struct tevent_immediate *ti,
+                                  void *private_data)
+{
+       struct messaging_post_state *state = talloc_get_type_abort(
+               private_data, struct messaging_post_state);
+       messaging_dispatch_rec(state->msg_ctx, ev, state->rec);
+       TALLOC_FREE(state);
+}
+
 int messaging_send_iov_from(struct messaging_context *msg_ctx,
                            struct server_id src, struct server_id dst,
                            uint32_t msg_type,
@@ -541,7 +597,7 @@ int messaging_send_iov_from(struct messaging_context *msg_ctx,
                return EINVAL;
        }
 
-       if (!procid_is_local(&dst)) {
+       if (dst.vnn != msg_ctx->id.vnn) {
                if (num_fds > 0) {
                        return ENOSYS;
                }
@@ -553,13 +609,35 @@ int messaging_send_iov_from(struct messaging_context *msg_ctx,
                return ret;
        }
 
+       if (server_id_equal(&dst, &msg_ctx->id)) {
+               ret = messaging_post_self(msg_ctx, src, dst, msg_type,
+                                         iov, iovlen, fds, num_fds);
+               return ret;
+       }
+
        message_hdr_put(hdr, msg_type, src, dst);
        iov2[0] = (struct iovec){ .iov_base = hdr, .iov_len = sizeof(hdr) };
        memcpy(&iov2[1], iov, iovlen * sizeof(*iov));
 
-       become_root();
        ret = messaging_dgm_send(dst.pid, iov2, iovlen+1, fds, num_fds);
-       unbecome_root();
+
+       if (ret == EACCES) {
+               become_root();
+               ret = messaging_dgm_send(dst.pid, iov2, iovlen+1,
+                                        fds, num_fds);
+               unbecome_root();
+       }
+
+       if (ret == ECONNREFUSED) {
+               /*
+                * Linux returns this when a socket exists in the file
+                * system without a listening process. This is not
+                * documented in susv4 or the linux manpages, but it's
+                * easily testable. For the higher levels this is the
+                * same as "destination does not exist"
+                */
+               ret = ENOENT;
+       }
 
        return ret;
 }
@@ -584,9 +662,16 @@ static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
 {
        struct messaging_rec *result;
        size_t fds_size = sizeof(int64_t) * rec->num_fds;
+       size_t payload_len;
+
+       payload_len = rec->buf.length + fds_size;
+       if (payload_len < rec->buf.length) {
+               /* overflow */
+               return NULL;
+       }
 
        result = talloc_pooled_object(mem_ctx, struct messaging_rec, 2,
-                                     rec->buf.length + fds_size);
+                                     payload_len);
        if (result == NULL) {
                return NULL;
        }
@@ -608,7 +693,7 @@ static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
 struct messaging_filtered_read_state {
        struct tevent_context *ev;
        struct messaging_context *msg_ctx;
-       void *tevent_handle;
+       struct messaging_dgm_fde *fde;
 
        bool (*filter)(struct messaging_rec *rec, void *private_data);
        void *private_data;
@@ -645,9 +730,8 @@ struct tevent_req *messaging_filtered_read_send(
         */
        tevent_req_defer_callback(req, state->ev);
 
-       state->tevent_handle = messaging_dgm_register_tevent_context(
-               state, ev);
-       if (tevent_req_nomem(state->tevent_handle, req)) {
+       state->fde = messaging_dgm_register_tevent_context(state, ev);
+       if (tevent_req_nomem(state->fde, req)) {
                return tevent_req_post(req, ev);
        }
 
@@ -685,11 +769,11 @@ static void messaging_filtered_read_cleanup(struct tevent_req *req,
        struct messaging_filtered_read_state *state = tevent_req_data(
                req, struct messaging_filtered_read_state);
        struct messaging_context *msg_ctx = state->msg_ctx;
-       unsigned i;
+       size_t i;
 
        tevent_req_set_cleanup_fn(req, NULL);
 
-       TALLOC_FREE(state->tevent_handle);
+       TALLOC_FREE(state->fde);
 
        /*
         * Just set the [new_]waiters entry to NULL, be careful not to mess
@@ -738,7 +822,9 @@ int messaging_filtered_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
                tevent_req_received(req);
                return err;
        }
-       *presult = talloc_move(mem_ctx, &state->rec);
+       if (presult != NULL) {
+               *presult = talloc_move(mem_ctx, &state->rec);
+       }
        return 0;
 }
 
@@ -820,87 +906,6 @@ int messaging_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
        return 0;
 }
 
-struct messaging_handler_state {
-       struct tevent_context *ev;
-       struct messaging_context *msg_ctx;
-       uint32_t msg_type;
-       bool (*handler)(struct messaging_context *msg_ctx,
-                       struct messaging_rec **rec, void *private_data);
-       void *private_data;
-};
-
-static void messaging_handler_got_msg(struct tevent_req *subreq);
-
-struct tevent_req *messaging_handler_send(
-       TALLOC_CTX *mem_ctx, struct tevent_context *ev,
-       struct messaging_context *msg_ctx, uint32_t msg_type,
-       bool (*handler)(struct messaging_context *msg_ctx,
-                       struct messaging_rec **rec, void *private_data),
-       void *private_data)
-{
-       struct tevent_req *req, *subreq;
-       struct messaging_handler_state *state;
-
-       req = tevent_req_create(mem_ctx, &state,
-                               struct messaging_handler_state);
-       if (req == NULL) {
-               return NULL;
-       }
-       state->ev = ev;
-       state->msg_ctx = msg_ctx;
-       state->msg_type = msg_type;
-       state->handler = handler;
-       state->private_data = private_data;
-
-       subreq = messaging_read_send(state, state->ev, state->msg_ctx,
-                                    state->msg_type);
-       if (tevent_req_nomem(subreq, req)) {
-               return tevent_req_post(req, ev);
-       }
-       tevent_req_set_callback(subreq, messaging_handler_got_msg, req);
-       return req;
-}
-
-static void messaging_handler_got_msg(struct tevent_req *subreq)
-{
-       struct tevent_req *req = tevent_req_callback_data(
-               subreq, struct tevent_req);
-       struct messaging_handler_state *state = tevent_req_data(
-               req, struct messaging_handler_state);
-       struct messaging_rec *rec;
-       int ret;
-       bool ok;
-
-       ret = messaging_read_recv(subreq, state, &rec);
-       TALLOC_FREE(subreq);
-       if (tevent_req_error(req, ret)) {
-               return;
-       }
-
-       subreq = messaging_read_send(state, state->ev, state->msg_ctx,
-                                    state->msg_type);
-       if (tevent_req_nomem(subreq, req)) {
-               return;
-       }
-       tevent_req_set_callback(subreq, messaging_handler_got_msg, req);
-
-       ok = state->handler(state->msg_ctx, &rec, state->private_data);
-       TALLOC_FREE(rec);
-       if (ok) {
-               /*
-                * Next round
-                */
-               return;
-       }
-       TALLOC_FREE(subreq);
-       tevent_req_done(req);
-}
-
-int messaging_handler_recv(struct tevent_req *req)
-{
-       return tevent_req_simple_recv_unix(req);
-}
-
 static bool messaging_append_new_waiters(struct messaging_context *msg_ctx)
 {
        if (msg_ctx->num_new_waiters == 0) {
@@ -929,17 +934,14 @@ static bool messaging_append_new_waiters(struct messaging_context *msg_ctx)
        return true;
 }
 
-/*
-  Dispatch one messaging_rec
-*/
-static void messaging_dispatch_rec(struct messaging_context *msg_ctx,
-                                  struct messaging_rec *rec)
+static bool messaging_dispatch_classic(struct messaging_context *msg_ctx,
+                                      struct messaging_rec *rec)
 {
        struct messaging_callback *cb, *next;
-       unsigned i;
-       size_t j;
 
        for (cb = msg_ctx->callbacks; cb != NULL; cb = next) {
+               size_t j;
+
                next = cb->next;
                if (cb->msg_type != rec->msg_type) {
                        continue;
@@ -958,15 +960,31 @@ static void messaging_dispatch_rec(struct messaging_context *msg_ctx,
                cb->fn(msg_ctx, cb->private_data, rec->msg_type,
                       rec->src, &rec->buf);
 
-               /*
-                * we continue looking for matching messages after finding
-                * one. This matters for subsystems like the internal notify
-                * code which register more than one handler for the same
-                * message type
-                */
+               return true;
+       }
+
+       return false;
+}
+
+/*
+  Dispatch one messaging_rec
+*/
+static void messaging_dispatch_rec(struct messaging_context *msg_ctx,
+                                  struct tevent_context *ev,
+                                  struct messaging_rec *rec)
+{
+       size_t i;
+       bool consumed;
+
+       if (ev == msg_ctx->event_ctx) {
+               consumed = messaging_dispatch_classic(msg_ctx, rec);
+               if (consumed) {
+                       return;
+               }
        }
 
        if (!messaging_append_new_waiters(msg_ctx)) {
+               size_t j;
                for (j=0; j < rec->num_fds; j++) {
                        int fd = rec->fds[j];
                        close(fd);
@@ -1001,24 +1019,46 @@ static void messaging_dispatch_rec(struct messaging_context *msg_ctx,
 
                state = tevent_req_data(
                        req, struct messaging_filtered_read_state);
-               if (state->filter(rec, state->private_data)) {
+               if ((ev == state->ev) &&
+                   state->filter(rec, state->private_data)) {
                        messaging_filtered_read_done(req, rec);
-
-                       /*
-                        * Only the first one gets the fd-array
-                        */
-                       rec->num_fds = 0;
-                       rec->fds = NULL;
+                       return;
                }
 
                i += 1;
        }
 
+       if (ev != msg_ctx->event_ctx) {
+               struct iovec iov;
+               int fds[rec->num_fds];
+               int ret;
+
+               /*
+                * We've been listening on a nested event
+                * context. Messages need to be handled in the main
+                * event context, so post to ourselves
+                */
+
+               iov.iov_base = rec->buf.data;
+               iov.iov_len = rec->buf.length;
+
+               for (i=0; i<rec->num_fds; i++) {
+                       fds[i] = rec->fds[i];
+               }
+
+               ret = messaging_post_self(
+                       msg_ctx, rec->src, rec->dest, rec->msg_type,
+                       &iov, 1, fds, rec->num_fds);
+               if (ret == 0) {
+                       return;
+               }
+       }
+
        /*
         * If the fd-array isn't used, just close it.
         */
-       for (j=0; j < rec->num_fds; j++) {
-               int fd = rec->fds[j];
+       for (i=0; i < rec->num_fds; i++) {
+               int fd = rec->fds[i];
                close(fd);
        }
        rec->num_fds = 0;