smbd: Add the notify daemon

[Samba.git] / source3 / smbd / notifyd / notifyd.c

/*
 * Unix SMB/CIFS implementation.
 *
 * Copyright (C) Volker Lendecke 2014
 *
 * This program 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 Foundation; either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#include "includes.h"
#include "librpc/gen_ndr/notify.h"
#include "librpc/gen_ndr/messaging.h"
#include "librpc/gen_ndr/server_id.h"
#include "lib/dbwrap/dbwrap.h"
#include "lib/dbwrap/dbwrap_rbt.h"
#include "messages.h"
#include "proto.h"
#include "tdb.h"
#include "util_tdb.h"
#include "notifyd.h"
#include "lib/util/server_id_db.h"
#include "lib/util/tevent_unix.h"
#include "ctdbd_conn.h"
#include "ctdb_srvids.h"
#include "source3/smbd/proto.h"
#include "ctdb/include/ctdb_protocol.h"
#include "server_id_db_util.h"
#include "lib/util/iov_buf.h"
#include "messages_util.h"

struct notifyd_peer;

/*
 * All of notifyd's state
 */

struct notifyd_state {
        struct tevent_context *ev;
        struct messaging_context *msg_ctx;
        struct ctdbd_connection *ctdbd_conn;

        /*
         * Database of everything clients show interest in. Indexed by
         * absolute path. The database keys are not 0-terminated
         * because the criticial operation, notifyd_trigger, can walk
         * the structure from the top without adding intermediate 0s.
         * The database records contain an array of
         *
         * struct notifyd_instance
         *
         * to be maintained by parsed by notifyd_entry_parse()
         */
        struct db_context *entries;

        /*
         * In the cluster case, this is the place where we store a log
         * of all MSG_SMB_NOTIFY_REC_CHANGE messages. We just 1:1
         * forward them to our peer notifyd's in the cluster once a
         * second or when the log grows too large.
         */

        struct messaging_reclog *log;

        /*
         * Array of companion notifyd's in a cluster. Every notifyd
         * broadcasts its messaging_reclog to every other notifyd in
         * the cluster. This is done by making ctdb send a message to
         * srvid CTDB_SRVID_SAMBA_NOTIFY_PROXY with destination node
         * number CTDB_BROADCAST_VNNMAP. Everybody in the cluster who
         * had called register_with_ctdbd this srvid will receive the
         * broadcasts.
         *
         * Database replication happens via these broadcasts. Also,
         * they serve as liveness indication. If a notifyd receives a
         * broadcast from an unknown peer, it will create one for this
         * srvid. Also when we don't hear anything from a peer for a
         * while, we will discard it.
         */

        struct notifyd_peer **peers;
        size_t num_peers;

        sys_notify_watch_fn sys_notify_watch;
        struct sys_notify_context *sys_notify_ctx;
};

/*
 * notifyd's representation of a notify instance
 */
struct notifyd_instance {
        struct server_id client;
        struct notify_instance instance;

        void *sys_watch; /* inotify/fam/etc handle */

        /*
         * Filters after sys_watch took responsibility of some bits
         */
        uint32_t internal_filter;
        uint32_t internal_subdir_filter;
};

struct notifyd_peer {
        struct notifyd_state *state;
        struct server_id pid;
        uint64_t rec_index;
        struct db_context *db;
        time_t last_broadcast;
};

static bool notifyd_rec_change(struct messaging_context *msg_ctx,
                               struct messaging_rec **prec,
                               void *private_data);
static bool notifyd_trigger(struct messaging_context *msg_ctx,
                            struct messaging_rec **prec,
                            void *private_data);
static bool notifyd_get_db(struct messaging_context *msg_ctx,
                           struct messaging_rec **prec,
                           void *private_data);
static bool notifyd_got_db(struct messaging_context *msg_ctx,
                           struct messaging_rec **prec,
                           void *private_data);
static void notifyd_broadcast_reclog(struct ctdbd_connection *ctdbd_conn,
                                     struct server_id src,
                                     struct messaging_reclog *log);
static void notifyd_sys_callback(struct sys_notify_context *ctx,
                                 void *private_data, struct notify_event *ev);

static struct tevent_req *notifyd_broadcast_reclog_send(
        TALLOC_CTX *mem_ctx, struct tevent_context *ev,
        struct ctdbd_connection *ctdbd_conn, struct server_id src,
        struct messaging_reclog *log);
static int notifyd_broadcast_reclog_recv(struct tevent_req *req);

static struct tevent_req *notifyd_clean_peers_send(
        TALLOC_CTX *mem_ctx, struct tevent_context *ev,
        struct notifyd_state *notifyd);
static int notifyd_clean_peers_recv(struct tevent_req *req);

static int sys_notify_watch_dummy(
        TALLOC_CTX *mem_ctx,
        struct sys_notify_context *ctx,
        const char *path,
        uint32_t *filter,
        uint32_t *subdir_filter,
        void (*callback)(struct sys_notify_context *ctx,
                         void *private_data,
                         struct notify_event *ev),
        void *private_data,
        void *handle_p)
{
        void **handle = handle_p;
        *handle = NULL;
        return 0;
}

static void notifyd_handler_done(struct tevent_req *subreq);
static void notifyd_broadcast_reclog_finished(struct tevent_req *subreq);
static void notifyd_clean_peers_finished(struct tevent_req *subreq);
static void notifyd_snoop_broadcast(uint32_t src_vnn, uint32_t dst_vnn,
                                    uint64_t dst_srvid,
                                    const uint8_t *msg, size_t msglen,
                                    void *private_data);

struct tevent_req *notifyd_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
                                struct messaging_context *msg_ctx,
                                struct ctdbd_connection *ctdbd_conn,
                                sys_notify_watch_fn sys_notify_watch,
                                struct sys_notify_context *sys_notify_ctx)
{
        struct tevent_req *req, *subreq;
        struct notifyd_state *state;
        struct server_id_db *names_db;
        NTSTATUS status;
        int ret;

        req = tevent_req_create(mem_ctx, &state, struct notifyd_state);
        if (req == NULL) {
                return NULL;
        }
        state->ev = ev;
        state->msg_ctx = msg_ctx;
        state->ctdbd_conn = ctdbd_conn;

        if (sys_notify_watch == NULL) {
                sys_notify_watch = sys_notify_watch_dummy;
        }

        state->sys_notify_watch = sys_notify_watch;
        state->sys_notify_ctx = sys_notify_ctx;

        state->entries = db_open_rbt(state);
        if (tevent_req_nomem(state->entries, req)) {
                return tevent_req_post(req, ev);
        }

        subreq = messaging_handler_send(state, ev, msg_ctx,
                                        MSG_SMB_NOTIFY_REC_CHANGE,
                                        notifyd_rec_change, state);
        if (tevent_req_nomem(subreq, req)) {
                return tevent_req_post(req, ev);
        }
        tevent_req_set_callback(subreq, notifyd_handler_done, req);

        subreq = messaging_handler_send(state, ev, msg_ctx,
                                        MSG_SMB_NOTIFY_TRIGGER,
                                        notifyd_trigger, state);
        if (tevent_req_nomem(subreq, req)) {
                return tevent_req_post(req, ev);
        }
        tevent_req_set_callback(subreq, notifyd_handler_done, req);

        subreq = messaging_handler_send(state, ev, msg_ctx,
                                        MSG_SMB_NOTIFY_GET_DB,
                                        notifyd_get_db, state);
        if (tevent_req_nomem(subreq, req)) {
                return tevent_req_post(req, ev);
        }
        tevent_req_set_callback(subreq, notifyd_handler_done, req);

        subreq = messaging_handler_send(state, ev, msg_ctx,
                                        MSG_SMB_NOTIFY_DB,
                                        notifyd_got_db, state);
        if (tevent_req_nomem(subreq, req)) {
                return tevent_req_post(req, ev);
        }
        tevent_req_set_callback(subreq, notifyd_handler_done, req);

        names_db = messaging_names_db(msg_ctx);

        ret = server_id_db_set_exclusive(names_db, "notify-daemon");
        if (ret != 0) {
                DEBUG(10, ("%s: server_id_db_add failed: %s\n",
                           __func__, strerror(ret)));
                tevent_req_error(req, ret);
                return tevent_req_post(req, ev);
        }

        if (ctdbd_conn == NULL) {
                /*
                 * No cluster around, skip the database replication
                 * engine
                 */
                return req;
        }

        state->log = talloc_zero(state, struct messaging_reclog);
        if (tevent_req_nomem(state->log, req)) {
                return tevent_req_post(req, ev);
        }

        subreq = notifyd_broadcast_reclog_send(
                state->log, ev, ctdbd_conn, messaging_server_id(msg_ctx),
                state->log);
        if (tevent_req_nomem(subreq, req)) {
                return tevent_req_post(req, ev);
        }
        tevent_req_set_callback(subreq, notifyd_broadcast_reclog_finished,
                                req);

        subreq = notifyd_clean_peers_send(state, ev, state);
        if (tevent_req_nomem(subreq, req)) {
                return tevent_req_post(req, ev);
        }
        tevent_req_set_callback(subreq, notifyd_clean_peers_finished,
                                req);

        status = register_with_ctdbd(ctdbd_conn, CTDB_SRVID_SAMBA_NOTIFY_PROXY,
                                     notifyd_snoop_broadcast, state);
        if (!NT_STATUS_IS_OK(status)) {
                tevent_req_error(req, map_errno_from_nt_status(status));
                return tevent_req_post(req, ev);
        }

        return req;
}

static void notifyd_handler_done(struct tevent_req *subreq)
{
        struct tevent_req *req = tevent_req_callback_data(
                subreq, struct tevent_req);
        int ret;

        ret = messaging_handler_recv(subreq);
        TALLOC_FREE(subreq);
        tevent_req_error(req, ret);
}

static void notifyd_broadcast_reclog_finished(struct tevent_req *subreq)
{
        struct tevent_req *req = tevent_req_callback_data(
                subreq, struct tevent_req);
        int ret;

        ret = notifyd_broadcast_reclog_recv(subreq);
        TALLOC_FREE(subreq);
        tevent_req_error(req, ret);
}

static void notifyd_clean_peers_finished(struct tevent_req *subreq)
{
        struct tevent_req *req = tevent_req_callback_data(
                subreq, struct tevent_req);
        int ret;

        ret = notifyd_clean_peers_recv(subreq);
        TALLOC_FREE(subreq);
        tevent_req_error(req, ret);
}

int notifyd_recv(struct tevent_req *req)
{
        return tevent_req_simple_recv_unix(req);
}

/*
 * Parse an entry in the notifyd_context->entries database
 */

static bool notifyd_parse_entry(uint8_t *buf, size_t buflen,
                                struct notifyd_instance **instances,
                                size_t *num_instances)
{
        if ((buflen % sizeof(struct notifyd_instance)) != 0) {
                DEBUG(1, ("%s: invalid buffer size: %u\n",
                          __func__, (unsigned)buflen));
                return false;
        }

        if (instances != NULL) {
                *instances = (struct notifyd_instance *)buf;
        }
        if (num_instances != NULL) {
                *num_instances = buflen / sizeof(struct notifyd_instance);
        }
        return true;
}

static bool notifyd_apply_rec_change(
        const struct server_id *client,
        const char *path, size_t pathlen,
        const struct notify_instance *chg,
        struct db_context *entries,
        sys_notify_watch_fn sys_notify_watch,
        struct sys_notify_context *sys_notify_ctx,
        struct messaging_context *msg_ctx)
{
        struct db_record *rec;
        struct notifyd_instance *instances;
        size_t num_instances;
        size_t i;
        struct notifyd_instance *instance;
        TDB_DATA value;
        NTSTATUS status;
        bool ok = false;

        if (pathlen == 0) {
                DEBUG(1, ("%s: pathlen==0\n", __func__));
                return false;
        }
        if (path[pathlen-1] != '\0') {
                DEBUG(1, ("%s: path not 0-terminated\n", __func__));
                return false;
        }

        DEBUG(10, ("%s: path=%s, filter=%u, subdir_filter=%u, "
                   "private_data=%p\n", __func__, path,
                   (unsigned)chg->filter, (unsigned)chg->subdir_filter,
                   chg->private_data));

        rec = dbwrap_fetch_locked(
                entries, entries,
                make_tdb_data((const uint8_t *)path, pathlen-1));

        if (rec == NULL) {
                DEBUG(1, ("%s: dbwrap_fetch_locked failed\n", __func__));
                goto fail;
        }

        num_instances = 0;
        value = dbwrap_record_get_value(rec);

        if (value.dsize != 0) {
                if (!notifyd_parse_entry(value.dptr, value.dsize, NULL,
                                         &num_instances)) {
                        goto fail;
                }
        }

        /*
         * Overallocate by one instance to avoid a realloc when adding
         */
        instances = talloc_array(rec, struct notifyd_instance,
                                 num_instances + 1);
        if (instances == NULL) {
                DEBUG(1, ("%s: talloc failed\n", __func__));
                goto fail;
        }

        if (value.dsize != 0) {
                memcpy(instances, value.dptr, value.dsize);
        }

        for (i=0; i<num_instances; i++) {
                instance = &instances[i];

                if (server_id_equal(&instance->client, client) &&
                    (instance->instance.private_data == chg->private_data)) {
                        break;
                }
        }

        if (i < num_instances) {
                instance->instance = *chg;
        } else {
                /*
                 * We've overallocated for one instance
                 */
                instance = &instances[num_instances];

                *instance = (struct notifyd_instance) {
                        .client = *client,
                        .instance = *chg,
                        .internal_filter = chg->filter,
                        .internal_subdir_filter = chg->subdir_filter
                };

                num_instances += 1;
        }

        if ((instance->instance.filter != 0) ||
            (instance->instance.subdir_filter != 0)) {
                int ret;

                TALLOC_FREE(instance->sys_watch);

                ret = sys_notify_watch(entries, sys_notify_ctx, path,
                                       &instance->internal_filter,
                                       &instance->internal_subdir_filter,
                                       notifyd_sys_callback, msg_ctx,
                                       &instance->sys_watch);
                if (ret != 0) {
                        DEBUG(1, ("%s: inotify_watch returned %s\n",
                                  __func__, strerror(errno)));
                }
        }

        if ((instance->instance.filter == 0) &&
            (instance->instance.subdir_filter == 0)) {
                /* This is a delete request */
                TALLOC_FREE(instance->sys_watch);
                *instance = instances[num_instances-1];
                num_instances -= 1;
        }

        DEBUG(10, ("%s: %s has %u instances\n", __func__,
                   path, (unsigned)num_instances));

        if (num_instances == 0) {
                status = dbwrap_record_delete(rec);
                if (!NT_STATUS_IS_OK(status)) {
                        DEBUG(1, ("%s: dbwrap_record_delete returned %s\n",
                                  __func__, nt_errstr(status)));
                        goto fail;
                }
        } else {
                value = make_tdb_data(
                        (uint8_t *)instances,
                        sizeof(struct notifyd_instance) * num_instances);

                status = dbwrap_record_store(rec, value, 0);
                if (!NT_STATUS_IS_OK(status)) {
                        DEBUG(1, ("%s: dbwrap_record_store returned %s\n",
                                  __func__, nt_errstr(status)));
                        goto fail;
                }
        }

        ok = true;
fail:
        TALLOC_FREE(rec);
        return ok;
}

static void notifyd_sys_callback(struct sys_notify_context *ctx,
                                 void *private_data, struct notify_event *ev)
{
        struct messaging_context *msg_ctx = talloc_get_type_abort(
                private_data, struct messaging_context);
        struct notify_trigger_msg msg;
        struct iovec iov[4];
        char slash = '/';

        msg = (struct notify_trigger_msg) {
                .when = timespec_current(),
                .action = ev->action,
                .filter = UINT32_MAX
        };

        iov[0].iov_base = &msg;
        iov[0].iov_len = offsetof(struct notify_trigger_msg, path);
        iov[1].iov_base = discard_const_p(char, ev->dir);
        iov[1].iov_len = strlen(ev->dir);
        iov[2].iov_base = &slash;
        iov[2].iov_len = 1;
        iov[3].iov_base = discard_const_p(char, ev->path);
        iov[3].iov_len = strlen(ev->path)+1;

        messaging_send_iov(
                msg_ctx, messaging_server_id(msg_ctx),
                MSG_SMB_NOTIFY_TRIGGER, iov, ARRAY_SIZE(iov), NULL, 0);
}

static bool notifyd_parse_rec_change(uint8_t *buf, size_t bufsize,
                                     struct notify_rec_change_msg **pmsg,
                                     size_t *pathlen)
{
        struct notify_rec_change_msg *msg;

        if (bufsize < offsetof(struct notify_rec_change_msg, path) + 1) {
                DEBUG(1, ("%s: message too short, ignoring: %u\n", __func__,
                          (unsigned)bufsize));
                return false;
        }

        *pmsg = msg = (struct notify_rec_change_msg *)buf;
        *pathlen = bufsize - offsetof(struct notify_rec_change_msg, path);

        DEBUG(10, ("%s: Got rec_change_msg filter=%u, subdir_filter=%u, "
                   "private_data=%p, path=%.*s\n",
                   __func__, (unsigned)msg->instance.filter,
                   (unsigned)msg->instance.subdir_filter,
                   msg->instance.private_data, (int)(*pathlen), msg->path));

        return true;
}

static bool notifyd_rec_change(struct messaging_context *msg_ctx,
                               struct messaging_rec **prec,
                               void *private_data)
{
        struct notifyd_state *state = talloc_get_type_abort(
                private_data, struct notifyd_state);
        struct server_id_buf idbuf;
        struct messaging_rec *rec = *prec;
        struct messaging_rec **tmp;
        struct messaging_reclog *log;
        struct notify_rec_change_msg *msg;
        size_t pathlen;
        bool ok;

        DEBUG(10, ("%s: Got %d bytes from %s\n", __func__,
                   (unsigned)rec->buf.length,
                   server_id_str_buf(rec->src, &idbuf)));

        ok = notifyd_parse_rec_change(rec->buf.data, rec->buf.length,
                                      &msg, &pathlen);
        if (!ok) {
                return true;
        }

        ok = notifyd_apply_rec_change(
                &rec->src, msg->path, pathlen, &msg->instance,
                state->entries, state->sys_notify_watch, state->sys_notify_ctx,
                state->msg_ctx);
        if (!ok) {
                DEBUG(1, ("%s: notifyd_apply_rec_change failed, ignoring\n",
                          __func__));
                return true;
        }

        if ((state->log == NULL) || (state->ctdbd_conn == NULL)) {
                return true;
        }
        log = state->log;

        tmp = talloc_realloc(log, log->recs, struct messaging_rec *,
                             log->num_recs+1);
        if (tmp == NULL) {
                DEBUG(1, ("%s: talloc_realloc failed, ignoring\n", __func__));
                return true;
        }
        log->recs = tmp;

        log->recs[log->num_recs] = talloc_move(log->recs, prec);
        log->num_recs += 1;

        if (log->num_recs >= 100) {
                /*
                 * Don't let the log grow too large
                 */
                notifyd_broadcast_reclog(state->ctdbd_conn,
                                         messaging_server_id(msg_ctx), log);
        }

        return true;
}

struct notifyd_trigger_state {
        struct messaging_context *msg_ctx;
        struct notify_trigger_msg *msg;
        bool recursive;
        bool covered_by_sys_notify;
};

static void notifyd_trigger_parser(TDB_DATA key, TDB_DATA data,
                                   void *private_data);

static bool notifyd_trigger(struct messaging_context *msg_ctx,
                            struct messaging_rec **prec,
                            void *private_data)
{
        struct notifyd_state *state = talloc_get_type_abort(
                private_data, struct notifyd_state);
        struct server_id my_id = messaging_server_id(msg_ctx);
        struct messaging_rec *rec = *prec;
        struct notifyd_trigger_state tstate;
        const char *path;
        const char *p, *next_p;

        if (rec->buf.length < offsetof(struct notify_trigger_msg, path) + 1) {
                DEBUG(1, ("message too short, ignoring: %u\n",
                          (unsigned)rec->buf.length));
                return true;
        }
        if (rec->buf.data[rec->buf.length-1] != 0) {
                DEBUG(1, ("%s: path not 0-terminated, ignoring\n", __func__));
                return true;
        }

        tstate.msg_ctx = msg_ctx;

        tstate.covered_by_sys_notify = (rec->src.vnn == my_id.vnn);
        tstate.covered_by_sys_notify &= !server_id_equal(&rec->src, &my_id);

        tstate.msg = (struct notify_trigger_msg *)rec->buf.data;
        path = tstate.msg->path;

        DEBUG(10, ("%s: Got trigger_msg action=%u, filter=%u, path=%s\n",
                   __func__, (unsigned)tstate.msg->action,
                   (unsigned)tstate.msg->filter, path));

        if (path[0] != '/') {
                DEBUG(1, ("%s: path %s does not start with /, ignoring\n",
                          __func__, path));
                return true;
        }

        for (p = strchr(path+1, '/'); p != NULL; p = next_p) {
                ptrdiff_t path_len = p - path;
                TDB_DATA key;
                uint32_t i;

                next_p = strchr(p+1, '/');
                tstate.recursive = (next_p != NULL);

                DEBUG(10, ("%s: Trying path %.*s\n", __func__,
                           (int)path_len, path));

                key = (TDB_DATA) { .dptr = discard_const_p(uint8_t, path),
                                   .dsize = path_len };

                dbwrap_parse_record(state->entries, key,
                                    notifyd_trigger_parser, &tstate);

                if (state->peers == NULL) {
                        continue;
                }

                if (rec->src.vnn != my_id.vnn) {
                        continue;
                }

                for (i=0; i<state->num_peers; i++) {
                        if (state->peers[i]->db == NULL) {
                                /*
                                 * Inactive peer, did not get a db yet
                                 */
                                continue;
                        }
                        dbwrap_parse_record(state->peers[i]->db, key,
                                            notifyd_trigger_parser, &tstate);
                }
        }

        return true;
}

static void notifyd_send_delete(struct messaging_context *msg_ctx,
                                TDB_DATA key,
                                struct notifyd_instance *instance);

static void notifyd_trigger_parser(TDB_DATA key, TDB_DATA data,
                                   void *private_data)

{
        struct notifyd_trigger_state *tstate = private_data;
        struct notify_event_msg msg = { .action = tstate->msg->action };
        struct iovec iov[2];
        size_t path_len = key.dsize;
        struct notifyd_instance *instances = NULL;
        size_t num_instances = 0;
        size_t i;

        if (!notifyd_parse_entry(data.dptr, data.dsize, &instances,
                                 &num_instances)) {
                DEBUG(1, ("%s: Could not parse notifyd_entry\n", __func__));
                return;
        }

        DEBUG(10, ("%s: Found %u instances for %.*s\n", __func__,
                   (unsigned)num_instances, (int)key.dsize,
                   (char *)key.dptr));

        iov[0].iov_base = &msg;
        iov[0].iov_len = offsetof(struct notify_event_msg, path);
        iov[1].iov_base = tstate->msg->path + path_len + 1;
        iov[1].iov_len = strlen((char *)(iov[1].iov_base)) + 1;

        for (i=0; i<num_instances; i++) {
                struct notifyd_instance *instance = &instances[i];
                struct server_id_buf idbuf;
                uint32_t i_filter;
                NTSTATUS status;

                if (tstate->covered_by_sys_notify) {
                        if (tstate->recursive) {
                                i_filter = instance->internal_subdir_filter;
                        } else {
                                i_filter = instance->internal_filter;
                        }
                } else {
                        if (tstate->recursive) {
                                i_filter = instance->instance.subdir_filter;
                        } else {
                                i_filter = instance->instance.filter;
                        }
                }

                if ((i_filter & tstate->msg->filter) == 0) {
                        continue;
                }

                msg.private_data = instance->instance.private_data;

                status = messaging_send_iov(
                        tstate->msg_ctx, instance->client,
                        MSG_PVFS_NOTIFY, iov, ARRAY_SIZE(iov), NULL, 0);

                DEBUG(10, ("%s: messaging_send_iov to %s returned %s\n",
                           __func__,
                           server_id_str_buf(instance->client, &idbuf),
                           nt_errstr(status)));

                if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND) &&
                    procid_is_local(&instance->client)) {
                        /*
                         * That process has died
                         */
                        notifyd_send_delete(tstate->msg_ctx, key, instance);
                        continue;
                }

                if (!NT_STATUS_IS_OK(status)) {
                        DEBUG(1, ("%s: messaging_send_iov returned %s\n",
                                  __func__, nt_errstr(status)));
                }
        }
}

/*
 * Send a delete request to ourselves to properly discard a notify
 * record for an smbd that has died.
 */

static void notifyd_send_delete(struct messaging_context *msg_ctx,
                                TDB_DATA key,
                                struct notifyd_instance *instance)
{
        struct notify_rec_change_msg msg = {
                .instance.private_data = instance->instance.private_data
        };
        uint8_t nul = 0;
        struct iovec iov[3];
        NTSTATUS status;

        /*
         * Send a rec_change to ourselves to delete a dead entry
         */

        iov[0] = (struct iovec) {
                .iov_base = &msg,
                .iov_len = offsetof(struct notify_rec_change_msg, path) };
        iov[1] = (struct iovec) { .iov_base = key.dptr, .iov_len = key.dsize };
        iov[2] = (struct iovec) { .iov_base = &nul, .iov_len = sizeof(nul) };

        status = messaging_send_iov_from(
                msg_ctx, instance->client, messaging_server_id(msg_ctx),
                MSG_SMB_NOTIFY_REC_CHANGE, iov, ARRAY_SIZE(iov), NULL, 0);

        if (!NT_STATUS_IS_OK(status)) {
                DEBUG(10, ("%s: messaging_send_iov_from returned %s\n",
                           __func__, nt_errstr(status)));
        }
}

static bool notifyd_get_db(struct messaging_context *msg_ctx,
                           struct messaging_rec **prec,
                           void *private_data)
{
        struct notifyd_state *state = talloc_get_type_abort(
                private_data, struct notifyd_state);
        struct messaging_rec *rec = *prec;
        struct server_id_buf id1, id2;
        NTSTATUS status;
        uint64_t rec_index = UINT64_MAX;
        uint8_t index_buf[sizeof(uint64_t)];
        size_t dbsize;
        uint8_t *buf;
        struct iovec iov[2];

        dbsize = dbwrap_marshall(state->entries, NULL, 0);

        buf = talloc_array(rec, uint8_t, dbsize);
        if (buf == NULL) {
                DEBUG(1, ("%s: talloc_array(%ju) failed\n",
                          __func__, (uintmax_t)dbsize));
                return true;
        }

        dbsize = dbwrap_marshall(state->entries, buf, dbsize);

        if (dbsize != talloc_get_size(buf)) {
                DEBUG(1, ("%s: dbsize changed: %ju->%ju\n", __func__,
                          (uintmax_t)talloc_get_size(buf),
                          (uintmax_t)dbsize));
                TALLOC_FREE(buf);
                return true;
        }

        if (state->log != NULL) {
                rec_index = state->log->rec_index;
        }
        SBVAL(index_buf, 0, rec_index);

        iov[0] = (struct iovec) { .iov_base = index_buf,
                                  .iov_len = sizeof(index_buf) };
        iov[1] = (struct iovec) { .iov_base = buf,
                                  .iov_len = dbsize };

        DEBUG(10, ("%s: Sending %ju bytes to %s->%s\n", __func__,
                   (uintmax_t)iov_buflen(iov, ARRAY_SIZE(iov)),
                   server_id_str_buf(messaging_server_id(msg_ctx), &id1),
                   server_id_str_buf(rec->src, &id2)));

        status = messaging_send_iov(msg_ctx, rec->src, MSG_SMB_NOTIFY_DB,
                                    iov, ARRAY_SIZE(iov), NULL, 0);
        TALLOC_FREE(buf);
        if (!NT_STATUS_IS_OK(status)) {
                DEBUG(1, ("%s: messaging_send_iov failed: %s\n",
                          __func__, nt_errstr(status)));
        }

        return true;
}

static int notifyd_add_proxy_syswatches(struct db_record *rec,
                                        void *private_data);

static bool notifyd_got_db(struct messaging_context *msg_ctx,
                           struct messaging_rec **prec,
                           void *private_data)
{
        struct notifyd_state *state = talloc_get_type_abort(
                private_data, struct notifyd_state);
        struct messaging_rec *rec = *prec;
        struct notifyd_peer *p = NULL;
        struct server_id_buf idbuf;
        NTSTATUS status;
        int count;
        size_t i;

        for (i=0; i<state->num_peers; i++) {
                if (server_id_equal(&rec->src, &state->peers[i]->pid)) {
                        p = state->peers[i];
                        break;
                }
        }

        if (p == NULL) {
                DEBUG(10, ("%s: Did not find peer for db from %s\n",
                           __func__, server_id_str_buf(rec->src, &idbuf)));
                return true;
        }

        if (rec->buf.length < 8) {
                DEBUG(10, ("%s: Got short db length %u from %s\n", __func__,
                           (unsigned)rec->buf.length,
                           server_id_str_buf(rec->src, &idbuf)));
                TALLOC_FREE(p);
                return true;
        }

        p->rec_index = BVAL(rec->buf.data, 0);

        p->db = db_open_rbt(p);
        if (p->db == NULL) {
                DEBUG(10, ("%s: db_open_rbt failed\n", __func__));
                TALLOC_FREE(p);
                return true;
        }

        status = dbwrap_unmarshall(p->db, rec->buf.data + 8,
                                   rec->buf.length - 8);
        if (!NT_STATUS_IS_OK(status)) {
                DEBUG(10, ("%s: dbwrap_unmarshall returned %s for db %s\n",
                           __func__, nt_errstr(status),
                           server_id_str_buf(rec->src, &idbuf)));
                TALLOC_FREE(p);
                return true;
        }

        dbwrap_traverse_read(p->db, notifyd_add_proxy_syswatches, state,
                             &count);

        DEBUG(10, ("%s: Database from %s contained %d records\n", __func__,
                   server_id_str_buf(rec->src, &idbuf), count));

        return true;
}

static void notifyd_broadcast_reclog(struct ctdbd_connection *ctdbd_conn,
                                     struct server_id src,
                                     struct messaging_reclog *log)
{
        NTSTATUS status;
        enum ndr_err_code ndr_err;
        uint8_t msghdr[MESSAGE_HDR_LENGTH];
        DATA_BLOB blob;
        struct iovec iov[2];

        if (log == NULL) {
                return;
        }

        DEBUG(10, ("%s: rec_index=%ju, num_recs=%u\n", __func__,
                   (uintmax_t)log->rec_index, (unsigned)log->num_recs));

        message_hdr_put(msghdr, MSG_SMB_NOTIFY_REC_CHANGES, src,
                        (struct server_id) {0 });
        iov[0] = (struct iovec) { .iov_base = msghdr,
                                  .iov_len = sizeof(msghdr) };

        ndr_err = ndr_push_struct_blob(
                &blob, log, log,
                (ndr_push_flags_fn_t)ndr_push_messaging_reclog);
        if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
                DEBUG(1, ("%s: ndr_push_messaging_recs failed: %s\n",
                          __func__, ndr_errstr(ndr_err)));
                goto done;
        }
        iov[1] = (struct iovec) { .iov_base = blob.data,
                                  .iov_len = blob.length };

        status = ctdbd_messaging_send_iov(
                ctdbd_conn, CTDB_BROADCAST_VNNMAP,
                CTDB_SRVID_SAMBA_NOTIFY_PROXY, iov, ARRAY_SIZE(iov));
        TALLOC_FREE(blob.data);
        if (!NT_STATUS_IS_OK(status)) {
                DEBUG(1, ("%s: ctdbd_messaging_send failed: %s\n",
                          __func__, nt_errstr(status)));
                goto done;
        }

        log->rec_index += 1;

done:
        log->num_recs = 0;
        TALLOC_FREE(log->recs);
}

struct notifyd_broadcast_reclog_state {
        struct tevent_context *ev;
        struct ctdbd_connection *ctdbd_conn;
        struct server_id src;
        struct messaging_reclog *log;
};

static void notifyd_broadcast_reclog_next(struct tevent_req *subreq);

static struct tevent_req *notifyd_broadcast_reclog_send(
        TALLOC_CTX *mem_ctx, struct tevent_context *ev,
        struct ctdbd_connection *ctdbd_conn, struct server_id src,
        struct messaging_reclog *log)
{
        struct tevent_req *req, *subreq;
        struct notifyd_broadcast_reclog_state *state;

        req = tevent_req_create(mem_ctx, &state,
                                struct notifyd_broadcast_reclog_state);
        if (req == NULL) {
                return NULL;
        }
        state->ev = ev;
        state->ctdbd_conn = ctdbd_conn;
        state->src = src;
        state->log = log;

        subreq = tevent_wakeup_send(state, state->ev,
                                    timeval_current_ofs_msec(1000));
        if (tevent_req_nomem(subreq, req)) {
                return tevent_req_post(req, ev);
        }
        tevent_req_set_callback(subreq, notifyd_broadcast_reclog_next, req);
        return req;
}

static void notifyd_broadcast_reclog_next(struct tevent_req *subreq)
{
        struct tevent_req *req = tevent_req_callback_data(
                subreq, struct tevent_req);
        struct notifyd_broadcast_reclog_state *state = tevent_req_data(
                req, struct notifyd_broadcast_reclog_state);
        bool ok;

        ok = tevent_wakeup_recv(subreq);
        TALLOC_FREE(subreq);
        if (!ok) {
                tevent_req_oom(req);
                return;
        }

        notifyd_broadcast_reclog(state->ctdbd_conn, state->src, state->log);

        subreq = tevent_wakeup_send(state, state->ev,
                                    timeval_current_ofs_msec(1000));
        if (tevent_req_nomem(subreq, req)) {
                return;
        }
        tevent_req_set_callback(subreq, notifyd_broadcast_reclog_next, req);
}

static int notifyd_broadcast_reclog_recv(struct tevent_req *req)
{
        return tevent_req_simple_recv_unix(req);
}

struct notifyd_clean_peers_state {
        struct tevent_context *ev;
        struct notifyd_state *notifyd;
};

static void notifyd_clean_peers_next(struct tevent_req *subreq);

static struct tevent_req *notifyd_clean_peers_send(
        TALLOC_CTX *mem_ctx, struct tevent_context *ev,
        struct notifyd_state *notifyd)
{
        struct tevent_req *req, *subreq;
        struct notifyd_clean_peers_state *state;

        req = tevent_req_create(mem_ctx, &state,
                                struct notifyd_clean_peers_state);
        if (req == NULL) {
                return NULL;
        }
        state->ev = ev;
        state->notifyd = notifyd;

        subreq = tevent_wakeup_send(state, state->ev,
                                    timeval_current_ofs_msec(30000));
        if (tevent_req_nomem(subreq, req)) {
                return tevent_req_post(req, ev);
        }
        tevent_req_set_callback(subreq, notifyd_clean_peers_next, req);
        return req;
}

static void notifyd_clean_peers_next(struct tevent_req *subreq)
{
        struct tevent_req *req = tevent_req_callback_data(
                subreq, struct tevent_req);
        struct notifyd_clean_peers_state *state = tevent_req_data(
                req, struct notifyd_clean_peers_state);
        struct notifyd_state *notifyd = state->notifyd;
        size_t i;
        bool ok;
        time_t now = time(NULL);

        ok = tevent_wakeup_recv(subreq);
        TALLOC_FREE(subreq);
        if (!ok) {
                tevent_req_oom(req);
                return;
        }

        i = 0;
        while (i < notifyd->num_peers) {
                struct notifyd_peer *p = notifyd->peers[i];

                if ((now - p->last_broadcast) > 60) {
                        struct server_id_buf idbuf;

                        /*
                         * Haven't heard for more than 60 seconds. Call this
                         * peer dead
                         */

                        DEBUG(10, ("%s: peer %s died\n", __func__,
                                   server_id_str_buf(p->pid, &idbuf)));
                        /*
                         * This implicitly decrements notifyd->num_peers
                         */
                        TALLOC_FREE(p);
                } else {
                        i += 1;
                }
        }

        subreq = tevent_wakeup_send(state, state->ev,
                                    timeval_current_ofs_msec(30000));
        if (tevent_req_nomem(subreq, req)) {
                return;
        }
        tevent_req_set_callback(subreq, notifyd_clean_peers_next, req);
}

static int notifyd_clean_peers_recv(struct tevent_req *req)
{
        return tevent_req_simple_recv_unix(req);
}

static int notifyd_add_proxy_syswatches(struct db_record *rec,
                                        void *private_data)
{
        struct notifyd_state *state = talloc_get_type_abort(
                private_data, struct notifyd_state);
        struct db_context *db = dbwrap_record_get_db(rec);
        TDB_DATA key = dbwrap_record_get_key(rec);
        TDB_DATA value = dbwrap_record_get_value(rec);
        struct notifyd_instance *instances = NULL;
        size_t num_instances = 0;
        size_t i;
        char path[key.dsize+1];
        bool ok;

        memcpy(path, key.dptr, key.dsize);
        path[key.dsize] = '\0';

        ok = notifyd_parse_entry(value.dptr, value.dsize, &instances,
                                 &num_instances);
        if (!ok) {
                DEBUG(1, ("%s: Could not parse notifyd entry for %s\n",
                          __func__, path));
                return 0;
        }

        for (i=0; i<num_instances; i++) {
                struct notifyd_instance *instance = &instances[i];
                uint32_t filter = instance->instance.filter;
                uint32_t subdir_filter = instance->instance.subdir_filter;
                int ret;

                ret = state->sys_notify_watch(
                        db, state->sys_notify_ctx, path,
                        &filter, &subdir_filter,
                        notifyd_sys_callback, state->msg_ctx,
                        &instance->sys_watch);
                if (ret != 0) {
                        DEBUG(1, ("%s: inotify_watch returned %s\n",
                                  __func__, strerror(errno)));
                }
        }

        return 0;
}

static int notifyd_db_del_syswatches(struct db_record *rec, void *private_data)
{
        TDB_DATA key = dbwrap_record_get_key(rec);
        TDB_DATA value = dbwrap_record_get_value(rec);
        struct notifyd_instance *instances = NULL;
        size_t num_instances = 0;
        size_t i;
        bool ok;

        ok = notifyd_parse_entry(value.dptr, value.dsize, &instances,
                                 &num_instances);
        if (!ok) {
                DEBUG(1, ("%s: Could not parse notifyd entry for %.*s\n",
                          __func__, (int)key.dsize, (char *)key.dptr));
                return 0;
        }
        for (i=0; i<num_instances; i++) {
                TALLOC_FREE(instances[i].sys_watch);
        }
        return 0;
}

static int notifyd_peer_destructor(struct notifyd_peer *p)
{
        struct notifyd_state *state = p->state;
        size_t i;

        dbwrap_traverse_read(p->db, notifyd_db_del_syswatches, NULL, NULL);

        for (i = 0; i<state->num_peers; i++) {
                if (p == state->peers[i]) {
                        state->peers[i] = state->peers[state->num_peers-1];
                        state->num_peers -= 1;
                        break;
                }
        }
        return 0;
}

static struct notifyd_peer *notifyd_peer_new(
        struct notifyd_state *state, struct server_id pid)
{
        struct notifyd_peer *p, **tmp;

        tmp = talloc_realloc(state, state->peers, struct notifyd_peer *,
                             state->num_peers+1);
        if (tmp == NULL) {
                return NULL;
        }
        state->peers = tmp;

        p = talloc_zero(state->peers, struct notifyd_peer);
        if (p == NULL) {
                return NULL;
        }
        p->state = state;
        p->pid = pid;

        state->peers[state->num_peers] = p;
        state->num_peers += 1;

        talloc_set_destructor(p, notifyd_peer_destructor);

        return p;
}

static void notifyd_apply_reclog(struct notifyd_peer *peer,
                                 const uint8_t *msg, size_t msglen)
{
        struct notifyd_state *state = peer->state;
        DATA_BLOB blob = { .data = discard_const_p(uint8_t, msg),
                           .length = msglen };
        struct server_id_buf idbuf;
        struct messaging_reclog *log;
        enum ndr_err_code ndr_err;
        uint32_t i;

        if (peer->db == NULL) {
                /*
                 * No db yet
                 */
                return;
        }

        log = talloc(peer, struct messaging_reclog);
        if (log == NULL) {
                DEBUG(10, ("%s: talloc failed\n", __func__));
                return;
        }

        ndr_err = ndr_pull_struct_blob_all(
                &blob, log, log,
                (ndr_pull_flags_fn_t)ndr_pull_messaging_reclog);
        if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
                DEBUG(10, ("%s: ndr_pull_messaging_reclog failed: %s\n",
                           __func__, ndr_errstr(ndr_err)));
                goto fail;
        }

        DEBUG(10, ("%s: Got %u recs index %ju from %s\n", __func__,
                   (unsigned)log->num_recs, (uintmax_t)log->rec_index,
                   server_id_str_buf(peer->pid, &idbuf)));

        if (log->rec_index != peer->rec_index) {
                DEBUG(3, ("%s: Got rec index %ju from %s, expected %ju\n",
                          __func__, (uintmax_t)log->rec_index,
                          server_id_str_buf(peer->pid, &idbuf),
                          (uintmax_t)peer->rec_index));
                goto fail;
        }

        for (i=0; i<log->num_recs; i++) {
                struct messaging_rec *r = log->recs[i];
                struct notify_rec_change_msg *chg;
                size_t pathlen;
                bool ok;

                ok = notifyd_parse_rec_change(r->buf.data, r->buf.length,
                                              &chg, &pathlen);
                if (!ok) {
                        DEBUG(3, ("%s: notifyd_parse_rec_change failed\n",
                                  __func__));
                        goto fail;
                }

                ok = notifyd_apply_rec_change(&r->src, chg->path, pathlen,
                                              &chg->instance, peer->db,
                                              state->sys_notify_watch,
                                              state->sys_notify_ctx,
                                              state->msg_ctx);
                if (!ok) {
                        DEBUG(3, ("%s: notifyd_apply_rec_change failed\n",
                                  __func__));
                        goto fail;
                }
        }

        peer->rec_index += 1;
        peer->last_broadcast = time(NULL);

        TALLOC_FREE(log);
        return;

fail:
        DEBUG(10, ("%s: Dropping peer %s\n", __func__,
                   server_id_str_buf(peer->pid, &idbuf)));
        TALLOC_FREE(peer);
}

/*
 * Receive messaging_reclog (log of MSG_SMB_NOTIFY_REC_CHANGE
 * messages) broadcasts by other notifyds. Several cases:
 *
 * We don't know the source. This creates a new peer. Creating a peer
 * involves asking the peer for its full database. We assume ordered
 * messages, so the new database will arrive before the next broadcast
 * will.
 *
 * We know the source and the log index matches. We will apply the log
 * locally to our peer's db as if we had received it from a local
 * client.
 *
 * We know the source but the log index does not match. This means we
 * lost a message. We just drop the whole peer and wait for the next
 * broadcast, which will then trigger a fresh database pull.
 */

static void notifyd_snoop_broadcast(uint32_t src_vnn, uint32_t dst_vnn,
                                    uint64_t dst_srvid,
                                    const uint8_t *msg, size_t msglen,
                                    void *private_data)
{
        struct notifyd_state *state = talloc_get_type_abort(
                private_data, struct notifyd_state);
        struct server_id my_id = messaging_server_id(state->msg_ctx);
        struct notifyd_peer *p;
        uint32_t i;
        uint32_t msg_type;
        struct server_id src, dst;
        struct server_id_buf idbuf;
        NTSTATUS status;

        if (msglen < MESSAGE_HDR_LENGTH) {
                DEBUG(10, ("%s: Got short broadcast\n", __func__));
                return;
        }
        message_hdr_get(&msg_type, &src, &dst, msg);

        if (msg_type != MSG_SMB_NOTIFY_REC_CHANGES) {
                DEBUG(10, ("%s Got message %u, ignoring\n", __func__,
                           (unsigned)msg_type));
                return;
        }
        if (server_id_equal(&src, &my_id)) {
                DEBUG(10, ("%s: Ignoring my own broadcast\n", __func__));
                return;
        }

        DEBUG(10, ("%s: Got MSG_SMB_NOTIFY_REC_CHANGES from %s\n",
                   __func__, server_id_str_buf(src, &idbuf)));

        for (i=0; i<state->num_peers; i++) {
                if (server_id_equal(&state->peers[i]->pid, &src)) {

                        DEBUG(10, ("%s: Applying changes to peer %u\n",
                                   __func__, (unsigned)i));

                        notifyd_apply_reclog(state->peers[i],
                                             msg + MESSAGE_HDR_LENGTH,
                                             msglen - MESSAGE_HDR_LENGTH);
                        return;
                }
        }

        DEBUG(10, ("%s: Creating new peer for %s\n", __func__,
                   server_id_str_buf(src, &idbuf)));

        p = notifyd_peer_new(state, src);
        if (p == NULL) {
                DEBUG(10, ("%s: notifyd_peer_new failed\n", __func__));
                return;
        }

        status = messaging_send_buf(state->msg_ctx, src, MSG_SMB_NOTIFY_GET_DB,
                                    NULL, 0);
        if (!NT_STATUS_IS_OK(status)) {
                DEBUG(10, ("%s: messaging_send_buf failed: %s\n",
                           __func__, nt_errstr(status)));
                TALLOC_FREE(p);
                return;
        }
}