source3/smbd/notifyd/notifyd.c

   1 /*
   2  * Unix SMB/CIFS implementation.
   3  *
   4  * Copyright (C) Volker Lendecke 2014
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License as published by
   8  * the Free Software Foundation; either version 3 of the License, or
   9  * (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14  * GNU General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18  */
  19
  20 #include "includes.h"
  21 #include "librpc/gen_ndr/notify.h"
  22 #include "librpc/gen_ndr/messaging.h"
  23 #include "librpc/gen_ndr/server_id.h"
  24 #include "lib/dbwrap/dbwrap.h"
  25 #include "lib/dbwrap/dbwrap_rbt.h"
  26 #include "messages.h"
  27 #include "proto.h"
  28 #include "tdb.h"
  29 #include "util_tdb.h"
  30 #include "notifyd.h"
  31 #include "lib/util/server_id_db.h"
  32 #include "lib/util/tevent_unix.h"
  33 #include "ctdbd_conn.h"
  34 #include "ctdb_srvids.h"
  35 #include "source3/smbd/proto.h"
  36 #include "server_id_db_util.h"
  37 #include "lib/util/iov_buf.h"
  38 #include "messages_util.h"
  39
  40 #ifdef CLUSTER_SUPPORT
  41 #include "ctdb_protocol.h"
  42 #endif
  43
  44 struct notifyd_peer;
  45
  46 /*
  47  * All of notifyd's state
  48  */
  49
  50 struct notifyd_state {
  51         struct tevent_context *ev;
  52         struct messaging_context *msg_ctx;
  53         struct ctdbd_connection *ctdbd_conn;
  54
  55         /*
  56          * Database of everything clients show interest in. Indexed by
  57          * absolute path. The database keys are not 0-terminated
  58          * because the criticial operation, notifyd_trigger, can walk
  59          * the structure from the top without adding intermediate 0s.
  60          * The database records contain an array of
  61          *
  62          * struct notifyd_instance
  63          *
  64          * to be maintained by parsed by notifyd_entry_parse()
  65          */
  66         struct db_context *entries;
  67
  68         /*
  69          * In the cluster case, this is the place where we store a log
  70          * of all MSG_SMB_NOTIFY_REC_CHANGE messages. We just 1:1
  71          * forward them to our peer notifyd's in the cluster once a
  72          * second or when the log grows too large.
  73          */
  74
  75         struct messaging_reclog *log;
  76
  77         /*
  78          * Array of companion notifyd's in a cluster. Every notifyd
  79          * broadcasts its messaging_reclog to every other notifyd in
  80          * the cluster. This is done by making ctdb send a message to
  81          * srvid CTDB_SRVID_SAMBA_NOTIFY_PROXY with destination node
  82          * number CTDB_BROADCAST_VNNMAP. Everybody in the cluster who
  83          * had called register_with_ctdbd this srvid will receive the
  84          * broadcasts.
  85          *
  86          * Database replication happens via these broadcasts. Also,
  87          * they serve as liveness indication. If a notifyd receives a
  88          * broadcast from an unknown peer, it will create one for this
  89          * srvid. Also when we don't hear anything from a peer for a
  90          * while, we will discard it.
  91          */
  92
  93         struct notifyd_peer **peers;
  94         size_t num_peers;
  95
  96         sys_notify_watch_fn sys_notify_watch;
  97         struct sys_notify_context *sys_notify_ctx;
  98 };
  99
 100 /*
 101  * notifyd's representation of a notify instance
 102  */
 103 struct notifyd_instance {
 104         struct server_id client;
 105         struct notify_instance instance;
 106
 107         void *sys_watch; /* inotify/fam/etc handle */
 108
 109         /*
 110          * Filters after sys_watch took responsibility of some bits
 111          */
 112         uint32_t internal_filter;
 113         uint32_t internal_subdir_filter;
 114 };
 115
 116 struct notifyd_peer {
 117         struct notifyd_state *state;
 118         struct server_id pid;
 119         uint64_t rec_index;
 120         struct db_context *db;
 121         time_t last_broadcast;
 122 };
 123
 124 static bool notifyd_rec_change(struct messaging_context *msg_ctx,
 125                                struct messaging_rec **prec,
 126                                void *private_data);
 127 static bool notifyd_trigger(struct messaging_context *msg_ctx,
 128                             struct messaging_rec **prec,
 129                             void *private_data);
 130 static bool notifyd_get_db(struct messaging_context *msg_ctx,
 131                            struct messaging_rec **prec,
 132                            void *private_data);
 133 static bool notifyd_got_db(struct messaging_context *msg_ctx,
 134                            struct messaging_rec **prec,
 135                            void *private_data);
 136
 137 #ifdef CLUSTER_SUPPORT
 138 static void notifyd_broadcast_reclog(struct ctdbd_connection *ctdbd_conn,
 139                                      struct server_id src,
 140                                      struct messaging_reclog *log);
 141 #endif
 142 static void notifyd_sys_callback(struct sys_notify_context *ctx,
 143                                  void *private_data, struct notify_event *ev);
 144
 145 #ifdef CLUSTER_SUPPORT
 146 static struct tevent_req *notifyd_broadcast_reclog_send(
 147         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 148         struct ctdbd_connection *ctdbd_conn, struct server_id src,
 149         struct messaging_reclog *log);
 150 static int notifyd_broadcast_reclog_recv(struct tevent_req *req);
 151
 152 static struct tevent_req *notifyd_clean_peers_send(
 153         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 154         struct notifyd_state *notifyd);
 155 static int notifyd_clean_peers_recv(struct tevent_req *req);
 156 #endif
 157
 158 static int sys_notify_watch_dummy(
 159         TALLOC_CTX *mem_ctx,
 160         struct sys_notify_context *ctx,
 161         const char *path,
 162         uint32_t *filter,
 163         uint32_t *subdir_filter,
 164         void (*callback)(struct sys_notify_context *ctx,
 165                          void *private_data,
 166                          struct notify_event *ev),
 167         void *private_data,
 168         void *handle_p)
 169 {
 170         void **handle = handle_p;
 171         *handle = NULL;
 172         return 0;
 173 }
 174
 175 static void notifyd_handler_done(struct tevent_req *subreq);
 176
 177 #ifdef CLUSTER_SUPPORT
 178 static void notifyd_broadcast_reclog_finished(struct tevent_req *subreq);
 179 static void notifyd_clean_peers_finished(struct tevent_req *subreq);
 180 static int notifyd_snoop_broadcast(uint32_t src_vnn, uint32_t dst_vnn,
 181                                    uint64_t dst_srvid,
 182                                    const uint8_t *msg, size_t msglen,
 183                                    void *private_data);
 184 #endif
 185
 186 struct tevent_req *notifyd_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 187                                 struct messaging_context *msg_ctx,
 188                                 struct ctdbd_connection *ctdbd_conn,
 189                                 sys_notify_watch_fn sys_notify_watch,
 190                                 struct sys_notify_context *sys_notify_ctx)
 191 {
 192         struct tevent_req *req, *subreq;
 193         struct notifyd_state *state;
 194         struct server_id_db *names_db;
 195         int ret;
 196
 197         req = tevent_req_create(mem_ctx, &state, struct notifyd_state);
 198         if (req == NULL) {
 199                 return NULL;
 200         }
 201         state->ev = ev;
 202         state->msg_ctx = msg_ctx;
 203         state->ctdbd_conn = ctdbd_conn;
 204
 205         if (sys_notify_watch == NULL) {
 206                 sys_notify_watch = sys_notify_watch_dummy;
 207         }
 208
 209         state->sys_notify_watch = sys_notify_watch;
 210         state->sys_notify_ctx = sys_notify_ctx;
 211
 212         state->entries = db_open_rbt(state);
 213         if (tevent_req_nomem(state->entries, req)) {
 214                 return tevent_req_post(req, ev);
 215         }
 216
 217         subreq = messaging_handler_send(state, ev, msg_ctx,
 218                                         MSG_SMB_NOTIFY_REC_CHANGE,
 219                                         notifyd_rec_change, state);
 220         if (tevent_req_nomem(subreq, req)) {
 221                 return tevent_req_post(req, ev);
 222         }
 223         tevent_req_set_callback(subreq, notifyd_handler_done, req);
 224
 225         subreq = messaging_handler_send(state, ev, msg_ctx,
 226                                         MSG_SMB_NOTIFY_TRIGGER,
 227                                         notifyd_trigger, state);
 228         if (tevent_req_nomem(subreq, req)) {
 229                 return tevent_req_post(req, ev);
 230         }
 231         tevent_req_set_callback(subreq, notifyd_handler_done, req);
 232
 233         subreq = messaging_handler_send(state, ev, msg_ctx,
 234                                         MSG_SMB_NOTIFY_GET_DB,
 235                                         notifyd_get_db, state);
 236         if (tevent_req_nomem(subreq, req)) {
 237                 return tevent_req_post(req, ev);
 238         }
 239         tevent_req_set_callback(subreq, notifyd_handler_done, req);
 240
 241         subreq = messaging_handler_send(state, ev, msg_ctx,
 242                                         MSG_SMB_NOTIFY_DB,
 243                                         notifyd_got_db, state);
 244         if (tevent_req_nomem(subreq, req)) {
 245                 return tevent_req_post(req, ev);
 246         }
 247         tevent_req_set_callback(subreq, notifyd_handler_done, req);
 248
 249         names_db = messaging_names_db(msg_ctx);
 250
 251         ret = server_id_db_set_exclusive(names_db, "notify-daemon");
 252         if (ret != 0) {
 253                 DEBUG(10, ("%s: server_id_db_add failed: %s\n",
 254                            __func__, strerror(ret)));
 255                 tevent_req_error(req, ret);
 256                 return tevent_req_post(req, ev);
 257         }
 258
 259         /* Block those signals that we are not handling */
 260         BlockSignals(True, SIGHUP);
 261         BlockSignals(True, SIGUSR1);
 262
 263         if (ctdbd_conn == NULL) {
 264                 /*
 265                  * No cluster around, skip the database replication
 266                  * engine
 267                  */
 268                 return req;
 269         }
 270
 271 #ifdef CLUSTER_SUPPORT
 272         state->log = talloc_zero(state, struct messaging_reclog);
 273         if (tevent_req_nomem(state->log, req)) {
 274                 return tevent_req_post(req, ev);
 275         }
 276
 277         subreq = notifyd_broadcast_reclog_send(
 278                 state->log, ev, ctdbd_conn, messaging_server_id(msg_ctx),
 279                 state->log);
 280         if (tevent_req_nomem(subreq, req)) {
 281                 return tevent_req_post(req, ev);
 282         }
 283         tevent_req_set_callback(subreq, notifyd_broadcast_reclog_finished,
 284                                 req);
 285
 286         subreq = notifyd_clean_peers_send(state, ev, state);
 287         if (tevent_req_nomem(subreq, req)) {
 288                 return tevent_req_post(req, ev);
 289         }
 290         tevent_req_set_callback(subreq, notifyd_clean_peers_finished,
 291                                 req);
 292
 293         ret = register_with_ctdbd(ctdbd_conn, CTDB_SRVID_SAMBA_NOTIFY_PROXY,
 294                                   notifyd_snoop_broadcast, state);
 295         if (ret != 0) {
 296                 tevent_req_error(req, ret);
 297                 return tevent_req_post(req, ev);
 298         }
 299 #endif
 300
 301         return req;
 302 }
 303
 304 static void notifyd_handler_done(struct tevent_req *subreq)
 305 {
 306         struct tevent_req *req = tevent_req_callback_data(
 307                 subreq, struct tevent_req);
 308         int ret;
 309
 310         ret = messaging_handler_recv(subreq);
 311         TALLOC_FREE(subreq);
 312         tevent_req_error(req, ret);
 313 }
 314
 315 #ifdef CLUSTER_SUPPORT
 316
 317 static void notifyd_broadcast_reclog_finished(struct tevent_req *subreq)
 318 {
 319         struct tevent_req *req = tevent_req_callback_data(
 320                 subreq, struct tevent_req);
 321         int ret;
 322
 323         ret = notifyd_broadcast_reclog_recv(subreq);
 324         TALLOC_FREE(subreq);
 325         tevent_req_error(req, ret);
 326 }
 327
 328 static void notifyd_clean_peers_finished(struct tevent_req *subreq)
 329 {
 330         struct tevent_req *req = tevent_req_callback_data(
 331                 subreq, struct tevent_req);
 332         int ret;
 333
 334         ret = notifyd_clean_peers_recv(subreq);
 335         TALLOC_FREE(subreq);
 336         tevent_req_error(req, ret);
 337 }
 338
 339 #endif
 340
 341 int notifyd_recv(struct tevent_req *req)
 342 {
 343         return tevent_req_simple_recv_unix(req);
 344 }
 345
 346 /*
 347  * Parse an entry in the notifyd_context->entries database
 348  */
 349
 350 static bool notifyd_parse_entry(uint8_t *buf, size_t buflen,
 351                                 struct notifyd_instance **instances,
 352                                 size_t *num_instances)
 353 {
 354         if ((buflen % sizeof(struct notifyd_instance)) != 0) {
 355                 DEBUG(1, ("%s: invalid buffer size: %u\n",
 356                           __func__, (unsigned)buflen));
 357                 return false;
 358         }
 359
 360         if (instances != NULL) {
 361                 *instances = (struct notifyd_instance *)buf;
 362         }
 363         if (num_instances != NULL) {
 364                 *num_instances = buflen / sizeof(struct notifyd_instance);
 365         }
 366         return true;
 367 }
 368
 369 static bool notifyd_apply_rec_change(
 370         const struct server_id *client,
 371         const char *path, size_t pathlen,
 372         const struct notify_instance *chg,
 373         struct db_context *entries,
 374         sys_notify_watch_fn sys_notify_watch,
 375         struct sys_notify_context *sys_notify_ctx,
 376         struct messaging_context *msg_ctx)
 377 {
 378         struct db_record *rec;
 379         struct notifyd_instance *instances;
 380         size_t num_instances;
 381         size_t i;
 382         struct notifyd_instance *instance;
 383         TDB_DATA value;
 384         NTSTATUS status;
 385         bool ok = false;
 386
 387         if (pathlen == 0) {
 388                 DEBUG(1, ("%s: pathlen==0\n", __func__));
 389                 return false;
 390         }
 391         if (path[pathlen-1] != '\0') {
 392                 DEBUG(1, ("%s: path not 0-terminated\n", __func__));
 393                 return false;
 394         }
 395
 396         DEBUG(10, ("%s: path=%s, filter=%u, subdir_filter=%u, "
 397                    "private_data=%p\n", __func__, path,
 398                    (unsigned)chg->filter, (unsigned)chg->subdir_filter,
 399                    chg->private_data));
 400
 401         rec = dbwrap_fetch_locked(
 402                 entries, entries,
 403                 make_tdb_data((const uint8_t *)path, pathlen-1));
 404
 405         if (rec == NULL) {
 406                 DEBUG(1, ("%s: dbwrap_fetch_locked failed\n", __func__));
 407                 goto fail;
 408         }
 409
 410         num_instances = 0;
 411         value = dbwrap_record_get_value(rec);
 412
 413         if (value.dsize != 0) {
 414                 if (!notifyd_parse_entry(value.dptr, value.dsize, NULL,
 415                                          &num_instances)) {
 416                         goto fail;
 417                 }
 418         }
 419
 420         /*
 421          * Overallocate by one instance to avoid a realloc when adding
 422          */
 423         instances = talloc_array(rec, struct notifyd_instance,
 424                                  num_instances + 1);
 425         if (instances == NULL) {
 426                 DEBUG(1, ("%s: talloc failed\n", __func__));
 427                 goto fail;
 428         }
 429
 430         if (value.dsize != 0) {
 431                 memcpy(instances, value.dptr, value.dsize);
 432         }
 433
 434         for (i=0; i<num_instances; i++) {
 435                 instance = &instances[i];
 436
 437                 if (server_id_equal(&instance->client, client) &&
 438                     (instance->instance.private_data == chg->private_data)) {
 439                         break;
 440                 }
 441         }
 442
 443         if (i < num_instances) {
 444                 instance->instance = *chg;
 445         } else {
 446                 /*
 447                  * We've overallocated for one instance
 448                  */
 449                 instance = &instances[num_instances];
 450
 451                 *instance = (struct notifyd_instance) {
 452                         .client = *client,
 453                         .instance = *chg,
 454                         .internal_filter = chg->filter,
 455                         .internal_subdir_filter = chg->subdir_filter
 456                 };
 457
 458                 num_instances += 1;
 459         }
 460
 461         if ((instance->instance.filter != 0) ||
 462             (instance->instance.subdir_filter != 0)) {
 463                 int ret;
 464
 465                 TALLOC_FREE(instance->sys_watch);
 466
 467                 ret = sys_notify_watch(entries, sys_notify_ctx, path,
 468                                        &instance->internal_filter,
 469                                        &instance->internal_subdir_filter,
 470                                        notifyd_sys_callback, msg_ctx,
 471                                        &instance->sys_watch);
 472                 if (ret != 0) {
 473                         DEBUG(1, ("%s: inotify_watch returned %s\n",
 474                                   __func__, strerror(errno)));
 475                 }
 476         }
 477
 478         if ((instance->instance.filter == 0) &&
 479             (instance->instance.subdir_filter == 0)) {
 480                 /* This is a delete request */
 481                 TALLOC_FREE(instance->sys_watch);
 482                 *instance = instances[num_instances-1];
 483                 num_instances -= 1;
 484         }
 485
 486         DEBUG(10, ("%s: %s has %u instances\n", __func__,
 487                    path, (unsigned)num_instances));
 488
 489         if (num_instances == 0) {
 490                 status = dbwrap_record_delete(rec);
 491                 if (!NT_STATUS_IS_OK(status)) {
 492                         DEBUG(1, ("%s: dbwrap_record_delete returned %s\n",
 493                                   __func__, nt_errstr(status)));
 494                         goto fail;
 495                 }
 496         } else {
 497                 value = make_tdb_data(
 498                         (uint8_t *)instances,
 499                         sizeof(struct notifyd_instance) * num_instances);
 500
 501                 status = dbwrap_record_store(rec, value, 0);
 502                 if (!NT_STATUS_IS_OK(status)) {
 503                         DEBUG(1, ("%s: dbwrap_record_store returned %s\n",
 504                                   __func__, nt_errstr(status)));
 505                         goto fail;
 506                 }
 507         }
 508
 509         ok = true;
 510 fail:
 511         TALLOC_FREE(rec);
 512         return ok;
 513 }
 514
 515 static void notifyd_sys_callback(struct sys_notify_context *ctx,
 516                                  void *private_data, struct notify_event *ev)
 517 {
 518         struct messaging_context *msg_ctx = talloc_get_type_abort(
 519                 private_data, struct messaging_context);
 520         struct notify_trigger_msg msg;
 521         struct iovec iov[4];
 522         char slash = '/';
 523
 524         msg = (struct notify_trigger_msg) {
 525                 .when = timespec_current(),
 526                 .action = ev->action,
 527                 .filter = UINT32_MAX
 528         };
 529
 530         iov[0].iov_base = &msg;
 531         iov[0].iov_len = offsetof(struct notify_trigger_msg, path);
 532         iov[1].iov_base = discard_const_p(char, ev->dir);
 533         iov[1].iov_len = strlen(ev->dir);
 534         iov[2].iov_base = &slash;
 535         iov[2].iov_len = 1;
 536         iov[3].iov_base = discard_const_p(char, ev->path);
 537         iov[3].iov_len = strlen(ev->path)+1;
 538
 539         messaging_send_iov(
 540                 msg_ctx, messaging_server_id(msg_ctx),
 541                 MSG_SMB_NOTIFY_TRIGGER, iov, ARRAY_SIZE(iov), NULL, 0);
 542 }
 543
 544 static bool notifyd_parse_rec_change(uint8_t *buf, size_t bufsize,
 545                                      struct notify_rec_change_msg **pmsg,
 546                                      size_t *pathlen)
 547 {
 548         struct notify_rec_change_msg *msg;
 549
 550         if (bufsize < offsetof(struct notify_rec_change_msg, path) + 1) {
 551                 DEBUG(1, ("%s: message too short, ignoring: %u\n", __func__,
 552                           (unsigned)bufsize));
 553                 return false;
 554         }
 555
 556         *pmsg = msg = (struct notify_rec_change_msg *)buf;
 557         *pathlen = bufsize - offsetof(struct notify_rec_change_msg, path);
 558
 559         DEBUG(10, ("%s: Got rec_change_msg filter=%u, subdir_filter=%u, "
 560                    "private_data=%p, path=%.*s\n",
 561                    __func__, (unsigned)msg->instance.filter,
 562                    (unsigned)msg->instance.subdir_filter,
 563                    msg->instance.private_data, (int)(*pathlen), msg->path));
 564
 565         return true;
 566 }
 567
 568 static bool notifyd_rec_change(struct messaging_context *msg_ctx,
 569                                struct messaging_rec **prec,
 570                                void *private_data)
 571 {
 572         struct notifyd_state *state = talloc_get_type_abort(
 573                 private_data, struct notifyd_state);
 574         struct server_id_buf idbuf;
 575         struct messaging_rec *rec = *prec;
 576         struct notify_rec_change_msg *msg;
 577         size_t pathlen;
 578         bool ok;
 579
 580         DEBUG(10, ("%s: Got %d bytes from %s\n", __func__,
 581                    (unsigned)rec->buf.length,
 582                    server_id_str_buf(rec->src, &idbuf)));
 583
 584         ok = notifyd_parse_rec_change(rec->buf.data, rec->buf.length,
 585                                       &msg, &pathlen);
 586         if (!ok) {
 587                 return true;
 588         }
 589
 590         ok = notifyd_apply_rec_change(
 591                 &rec->src, msg->path, pathlen, &msg->instance,
 592                 state->entries, state->sys_notify_watch, state->sys_notify_ctx,
 593                 state->msg_ctx);
 594         if (!ok) {
 595                 DEBUG(1, ("%s: notifyd_apply_rec_change failed, ignoring\n",
 596                           __func__));
 597                 return true;
 598         }
 599
 600         if ((state->log == NULL) || (state->ctdbd_conn == NULL)) {
 601                 return true;
 602         }
 603
 604 #ifdef CLUSTER_SUPPORT
 605         {
 606
 607         struct messaging_rec **tmp;
 608         struct messaging_reclog *log;
 609
 610         log = state->log;
 611
 612         tmp = talloc_realloc(log, log->recs, struct messaging_rec *,
 613                              log->num_recs+1);
 614         if (tmp == NULL) {
 615                 DEBUG(1, ("%s: talloc_realloc failed, ignoring\n", __func__));
 616                 return true;
 617         }
 618         log->recs = tmp;
 619
 620         log->recs[log->num_recs] = talloc_move(log->recs, prec);
 621         log->num_recs += 1;
 622
 623         if (log->num_recs >= 100) {
 624                 /*
 625                  * Don't let the log grow too large
 626                  */
 627                 notifyd_broadcast_reclog(state->ctdbd_conn,
 628                                          messaging_server_id(msg_ctx), log);
 629         }
 630
 631         }
 632 #endif
 633
 634         return true;
 635 }
 636
 637 struct notifyd_trigger_state {
 638         struct messaging_context *msg_ctx;
 639         struct notify_trigger_msg *msg;
 640         bool recursive;
 641         bool covered_by_sys_notify;
 642 };
 643
 644 static void notifyd_trigger_parser(TDB_DATA key, TDB_DATA data,
 645                                    void *private_data);
 646
 647 static bool notifyd_trigger(struct messaging_context *msg_ctx,
 648                             struct messaging_rec **prec,
 649                             void *private_data)
 650 {
 651         struct notifyd_state *state = talloc_get_type_abort(
 652                 private_data, struct notifyd_state);
 653         struct server_id my_id = messaging_server_id(msg_ctx);
 654         struct messaging_rec *rec = *prec;
 655         struct notifyd_trigger_state tstate;
 656         const char *path;
 657         const char *p, *next_p;
 658
 659         if (rec->buf.length < offsetof(struct notify_trigger_msg, path) + 1) {
 660                 DEBUG(1, ("message too short, ignoring: %u\n",
 661                           (unsigned)rec->buf.length));
 662                 return true;
 663         }
 664         if (rec->buf.data[rec->buf.length-1] != 0) {
 665                 DEBUG(1, ("%s: path not 0-terminated, ignoring\n", __func__));
 666                 return true;
 667         }
 668
 669         tstate.msg_ctx = msg_ctx;
 670
 671         tstate.covered_by_sys_notify = (rec->src.vnn == my_id.vnn);
 672         tstate.covered_by_sys_notify &= !server_id_equal(&rec->src, &my_id);
 673
 674         tstate.msg = (struct notify_trigger_msg *)rec->buf.data;
 675         path = tstate.msg->path;
 676
 677         DEBUG(10, ("%s: Got trigger_msg action=%u, filter=%u, path=%s\n",
 678                    __func__, (unsigned)tstate.msg->action,
 679                    (unsigned)tstate.msg->filter, path));
 680
 681         if (path[0] != '/') {
 682                 DEBUG(1, ("%s: path %s does not start with /, ignoring\n",
 683                           __func__, path));
 684                 return true;
 685         }
 686
 687         for (p = strchr(path+1, '/'); p != NULL; p = next_p) {
 688                 ptrdiff_t path_len = p - path;
 689                 TDB_DATA key;
 690                 uint32_t i;
 691
 692                 next_p = strchr(p+1, '/');
 693                 tstate.recursive = (next_p != NULL);
 694
 695                 DEBUG(10, ("%s: Trying path %.*s\n", __func__,
 696                            (int)path_len, path));
 697
 698                 key = (TDB_DATA) { .dptr = discard_const_p(uint8_t, path),
 699                                    .dsize = path_len };
 700
 701                 dbwrap_parse_record(state->entries, key,
 702                                     notifyd_trigger_parser, &tstate);
 703
 704                 if (state->peers == NULL) {
 705                         continue;
 706                 }
 707
 708                 if (rec->src.vnn != my_id.vnn) {
 709                         continue;
 710                 }
 711
 712                 for (i=0; i<state->num_peers; i++) {
 713                         if (state->peers[i]->db == NULL) {
 714                                 /*
 715                                  * Inactive peer, did not get a db yet
 716                                  */
 717                                 continue;
 718                         }
 719                         dbwrap_parse_record(state->peers[i]->db, key,
 720                                             notifyd_trigger_parser, &tstate);
 721                 }
 722         }
 723
 724         return true;
 725 }
 726
 727 static void notifyd_send_delete(struct messaging_context *msg_ctx,
 728                                 TDB_DATA key,
 729                                 struct notifyd_instance *instance);
 730
 731 static void notifyd_trigger_parser(TDB_DATA key, TDB_DATA data,
 732                                    void *private_data)
 733
 734 {
 735         struct notifyd_trigger_state *tstate = private_data;
 736         struct notify_event_msg msg = { .action = tstate->msg->action };
 737         struct iovec iov[2];
 738         size_t path_len = key.dsize;
 739         struct notifyd_instance *instances = NULL;
 740         size_t num_instances = 0;
 741         size_t i;
 742
 743         if (!notifyd_parse_entry(data.dptr, data.dsize, &instances,
 744                                  &num_instances)) {
 745                 DEBUG(1, ("%s: Could not parse notifyd_entry\n", __func__));
 746                 return;
 747         }
 748
 749         DEBUG(10, ("%s: Found %u instances for %.*s\n", __func__,
 750                    (unsigned)num_instances, (int)key.dsize,
 751                    (char *)key.dptr));
 752
 753         iov[0].iov_base = &msg;
 754         iov[0].iov_len = offsetof(struct notify_event_msg, path);
 755         iov[1].iov_base = tstate->msg->path + path_len + 1;
 756         iov[1].iov_len = strlen((char *)(iov[1].iov_base)) + 1;
 757
 758         for (i=0; i<num_instances; i++) {
 759                 struct notifyd_instance *instance = &instances[i];
 760                 struct server_id_buf idbuf;
 761                 uint32_t i_filter;
 762                 NTSTATUS status;
 763
 764                 if (tstate->covered_by_sys_notify) {
 765                         if (tstate->recursive) {
 766                                 i_filter = instance->internal_subdir_filter;
 767                         } else {
 768                                 i_filter = instance->internal_filter;
 769                         }
 770                 } else {
 771                         if (tstate->recursive) {
 772                                 i_filter = instance->instance.subdir_filter;
 773                         } else {
 774                                 i_filter = instance->instance.filter;
 775                         }
 776                 }
 777
 778                 if ((i_filter & tstate->msg->filter) == 0) {
 779                         continue;
 780                 }
 781
 782                 msg.private_data = instance->instance.private_data;
 783
 784                 status = messaging_send_iov(
 785                         tstate->msg_ctx, instance->client,
 786                         MSG_PVFS_NOTIFY, iov, ARRAY_SIZE(iov), NULL, 0);
 787
 788                 DEBUG(10, ("%s: messaging_send_iov to %s returned %s\n",
 789                            __func__,
 790                            server_id_str_buf(instance->client, &idbuf),
 791                            nt_errstr(status)));
 792
 793                 if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND) &&
 794                     procid_is_local(&instance->client)) {
 795                         /*
 796                          * That process has died
 797                          */
 798                         notifyd_send_delete(tstate->msg_ctx, key, instance);
 799                         continue;
 800                 }
 801
 802                 if (!NT_STATUS_IS_OK(status)) {
 803                         DEBUG(1, ("%s: messaging_send_iov returned %s\n",
 804                                   __func__, nt_errstr(status)));
 805                 }
 806         }
 807 }
 808
 809 /*
 810  * Send a delete request to ourselves to properly discard a notify
 811  * record for an smbd that has died.
 812  */
 813
 814 static void notifyd_send_delete(struct messaging_context *msg_ctx,
 815                                 TDB_DATA key,
 816                                 struct notifyd_instance *instance)
 817 {
 818         struct notify_rec_change_msg msg = {
 819                 .instance.private_data = instance->instance.private_data
 820         };
 821         uint8_t nul = 0;
 822         struct iovec iov[3];
 823         int ret;
 824
 825         /*
 826          * Send a rec_change to ourselves to delete a dead entry
 827          */
 828
 829         iov[0] = (struct iovec) {
 830                 .iov_base = &msg,
 831                 .iov_len = offsetof(struct notify_rec_change_msg, path) };
 832         iov[1] = (struct iovec) { .iov_base = key.dptr, .iov_len = key.dsize };
 833         iov[2] = (struct iovec) { .iov_base = &nul, .iov_len = sizeof(nul) };
 834
 835         ret = messaging_send_iov_from(
 836                 msg_ctx, instance->client, messaging_server_id(msg_ctx),
 837                 MSG_SMB_NOTIFY_REC_CHANGE, iov, ARRAY_SIZE(iov), NULL, 0);
 838
 839         if (ret != 0) {
 840                 DEBUG(10, ("%s: messaging_send_iov_from returned %s\n",
 841                            __func__, strerror(ret)));
 842         }
 843 }
 844
 845 static bool notifyd_get_db(struct messaging_context *msg_ctx,
 846                            struct messaging_rec **prec,
 847                            void *private_data)
 848 {
 849         struct notifyd_state *state = talloc_get_type_abort(
 850                 private_data, struct notifyd_state);
 851         struct messaging_rec *rec = *prec;
 852         struct server_id_buf id1, id2;
 853         NTSTATUS status;
 854         uint64_t rec_index = UINT64_MAX;
 855         uint8_t index_buf[sizeof(uint64_t)];
 856         size_t dbsize;
 857         uint8_t *buf;
 858         struct iovec iov[2];
 859
 860         dbsize = dbwrap_marshall(state->entries, NULL, 0);
 861
 862         buf = talloc_array(rec, uint8_t, dbsize);
 863         if (buf == NULL) {
 864                 DEBUG(1, ("%s: talloc_array(%ju) failed\n",
 865                           __func__, (uintmax_t)dbsize));
 866                 return true;
 867         }
 868
 869         dbsize = dbwrap_marshall(state->entries, buf, dbsize);
 870
 871         if (dbsize != talloc_get_size(buf)) {
 872                 DEBUG(1, ("%s: dbsize changed: %ju->%ju\n", __func__,
 873                           (uintmax_t)talloc_get_size(buf),
 874                           (uintmax_t)dbsize));
 875                 TALLOC_FREE(buf);
 876                 return true;
 877         }
 878
 879         if (state->log != NULL) {
 880                 rec_index = state->log->rec_index;
 881         }
 882         SBVAL(index_buf, 0, rec_index);
 883
 884         iov[0] = (struct iovec) { .iov_base = index_buf,
 885                                   .iov_len = sizeof(index_buf) };
 886         iov[1] = (struct iovec) { .iov_base = buf,
 887                                   .iov_len = dbsize };
 888
 889         DEBUG(10, ("%s: Sending %ju bytes to %s->%s\n", __func__,
 890                    (uintmax_t)iov_buflen(iov, ARRAY_SIZE(iov)),
 891                    server_id_str_buf(messaging_server_id(msg_ctx), &id1),
 892                    server_id_str_buf(rec->src, &id2)));
 893
 894         status = messaging_send_iov(msg_ctx, rec->src, MSG_SMB_NOTIFY_DB,
 895                                     iov, ARRAY_SIZE(iov), NULL, 0);
 896         TALLOC_FREE(buf);
 897         if (!NT_STATUS_IS_OK(status)) {
 898                 DEBUG(1, ("%s: messaging_send_iov failed: %s\n",
 899                           __func__, nt_errstr(status)));
 900         }
 901
 902         return true;
 903 }
 904
 905 static int notifyd_add_proxy_syswatches(struct db_record *rec,
 906                                         void *private_data);
 907
 908 static bool notifyd_got_db(struct messaging_context *msg_ctx,
 909                            struct messaging_rec **prec,
 910                            void *private_data)
 911 {
 912         struct notifyd_state *state = talloc_get_type_abort(
 913                 private_data, struct notifyd_state);
 914         struct messaging_rec *rec = *prec;
 915         struct notifyd_peer *p = NULL;
 916         struct server_id_buf idbuf;
 917         NTSTATUS status;
 918         int count;
 919         size_t i;
 920
 921         for (i=0; i<state->num_peers; i++) {
 922                 if (server_id_equal(&rec->src, &state->peers[i]->pid)) {
 923                         p = state->peers[i];
 924                         break;
 925                 }
 926         }
 927
 928         if (p == NULL) {
 929                 DEBUG(10, ("%s: Did not find peer for db from %s\n",
 930                            __func__, server_id_str_buf(rec->src, &idbuf)));
 931                 return true;
 932         }
 933
 934         if (rec->buf.length < 8) {
 935                 DEBUG(10, ("%s: Got short db length %u from %s\n", __func__,
 936                            (unsigned)rec->buf.length,
 937                            server_id_str_buf(rec->src, &idbuf)));
 938                 TALLOC_FREE(p);
 939                 return true;
 940         }
 941
 942         p->rec_index = BVAL(rec->buf.data, 0);
 943
 944         p->db = db_open_rbt(p);
 945         if (p->db == NULL) {
 946                 DEBUG(10, ("%s: db_open_rbt failed\n", __func__));
 947                 TALLOC_FREE(p);
 948                 return true;
 949         }
 950
 951         status = dbwrap_unmarshall(p->db, rec->buf.data + 8,
 952                                    rec->buf.length - 8);
 953         if (!NT_STATUS_IS_OK(status)) {
 954                 DEBUG(10, ("%s: dbwrap_unmarshall returned %s for db %s\n",
 955                            __func__, nt_errstr(status),
 956                            server_id_str_buf(rec->src, &idbuf)));
 957                 TALLOC_FREE(p);
 958                 return true;
 959         }
 960
 961         dbwrap_traverse_read(p->db, notifyd_add_proxy_syswatches, state,
 962                              &count);
 963
 964         DEBUG(10, ("%s: Database from %s contained %d records\n", __func__,
 965                    server_id_str_buf(rec->src, &idbuf), count));
 966
 967         return true;
 968 }
 969
 970 #ifdef CLUSTER_SUPPORT
 971
 972 static void notifyd_broadcast_reclog(struct ctdbd_connection *ctdbd_conn,
 973                                      struct server_id src,
 974                                      struct messaging_reclog *log)
 975 {
 976         enum ndr_err_code ndr_err;
 977         uint8_t msghdr[MESSAGE_HDR_LENGTH];
 978         DATA_BLOB blob;
 979         struct iovec iov[2];
 980         int ret;
 981
 982         if (log == NULL) {
 983                 return;
 984         }
 985
 986         DEBUG(10, ("%s: rec_index=%ju, num_recs=%u\n", __func__,
 987                    (uintmax_t)log->rec_index, (unsigned)log->num_recs));
 988
 989         message_hdr_put(msghdr, MSG_SMB_NOTIFY_REC_CHANGES, src,
 990                         (struct server_id) {0 });
 991         iov[0] = (struct iovec) { .iov_base = msghdr,
 992                                   .iov_len = sizeof(msghdr) };
 993
 994         ndr_err = ndr_push_struct_blob(
 995                 &blob, log, log,
 996                 (ndr_push_flags_fn_t)ndr_push_messaging_reclog);
 997         if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
 998                 DEBUG(1, ("%s: ndr_push_messaging_recs failed: %s\n",
 999                           __func__, ndr_errstr(ndr_err)));
1000                 goto done;
1001         }
1002         iov[1] = (struct iovec) { .iov_base = blob.data,
1003                                   .iov_len = blob.length };
1004
1005         ret = ctdbd_messaging_send_iov(
1006                 ctdbd_conn, CTDB_BROADCAST_VNNMAP,
1007                 CTDB_SRVID_SAMBA_NOTIFY_PROXY, iov, ARRAY_SIZE(iov));
1008         TALLOC_FREE(blob.data);
1009         if (ret != 0) {
1010                 DEBUG(1, ("%s: ctdbd_messaging_send failed: %s\n",
1011                           __func__, strerror(ret)));
1012                 goto done;
1013         }
1014
1015         log->rec_index += 1;
1016
1017 done:
1018         log->num_recs = 0;
1019         TALLOC_FREE(log->recs);
1020 }
1021
1022 struct notifyd_broadcast_reclog_state {
1023         struct tevent_context *ev;
1024         struct ctdbd_connection *ctdbd_conn;
1025         struct server_id src;
1026         struct messaging_reclog *log;
1027 };
1028
1029 static void notifyd_broadcast_reclog_next(struct tevent_req *subreq);
1030
1031 static struct tevent_req *notifyd_broadcast_reclog_send(
1032         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
1033         struct ctdbd_connection *ctdbd_conn, struct server_id src,
1034         struct messaging_reclog *log)
1035 {
1036         struct tevent_req *req, *subreq;
1037         struct notifyd_broadcast_reclog_state *state;
1038
1039         req = tevent_req_create(mem_ctx, &state,
1040                                 struct notifyd_broadcast_reclog_state);
1041         if (req == NULL) {
1042                 return NULL;
1043         }
1044         state->ev = ev;
1045         state->ctdbd_conn = ctdbd_conn;
1046         state->src = src;
1047         state->log = log;
1048
1049         subreq = tevent_wakeup_send(state, state->ev,
1050                                     timeval_current_ofs_msec(1000));
1051         if (tevent_req_nomem(subreq, req)) {
1052                 return tevent_req_post(req, ev);
1053         }
1054         tevent_req_set_callback(subreq, notifyd_broadcast_reclog_next, req);
1055         return req;
1056 }
1057
1058 static void notifyd_broadcast_reclog_next(struct tevent_req *subreq)
1059 {
1060         struct tevent_req *req = tevent_req_callback_data(
1061                 subreq, struct tevent_req);
1062         struct notifyd_broadcast_reclog_state *state = tevent_req_data(
1063                 req, struct notifyd_broadcast_reclog_state);
1064         bool ok;
1065
1066         ok = tevent_wakeup_recv(subreq);
1067         TALLOC_FREE(subreq);
1068         if (!ok) {
1069                 tevent_req_oom(req);
1070                 return;
1071         }
1072
1073         notifyd_broadcast_reclog(state->ctdbd_conn, state->src, state->log);
1074
1075         subreq = tevent_wakeup_send(state, state->ev,
1076                                     timeval_current_ofs_msec(1000));
1077         if (tevent_req_nomem(subreq, req)) {
1078                 return;
1079         }
1080         tevent_req_set_callback(subreq, notifyd_broadcast_reclog_next, req);
1081 }
1082
1083 static int notifyd_broadcast_reclog_recv(struct tevent_req *req)
1084 {
1085         return tevent_req_simple_recv_unix(req);
1086 }
1087
1088 struct notifyd_clean_peers_state {
1089         struct tevent_context *ev;
1090         struct notifyd_state *notifyd;
1091 };
1092
1093 static void notifyd_clean_peers_next(struct tevent_req *subreq);
1094
1095 static struct tevent_req *notifyd_clean_peers_send(
1096         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
1097         struct notifyd_state *notifyd)
1098 {
1099         struct tevent_req *req, *subreq;
1100         struct notifyd_clean_peers_state *state;
1101
1102         req = tevent_req_create(mem_ctx, &state,
1103                                 struct notifyd_clean_peers_state);
1104         if (req == NULL) {
1105                 return NULL;
1106         }
1107         state->ev = ev;
1108         state->notifyd = notifyd;
1109
1110         subreq = tevent_wakeup_send(state, state->ev,
1111                                     timeval_current_ofs_msec(30000));
1112         if (tevent_req_nomem(subreq, req)) {
1113                 return tevent_req_post(req, ev);
1114         }
1115         tevent_req_set_callback(subreq, notifyd_clean_peers_next, req);
1116         return req;
1117 }
1118
1119 static void notifyd_clean_peers_next(struct tevent_req *subreq)
1120 {
1121         struct tevent_req *req = tevent_req_callback_data(
1122                 subreq, struct tevent_req);
1123         struct notifyd_clean_peers_state *state = tevent_req_data(
1124                 req, struct notifyd_clean_peers_state);
1125         struct notifyd_state *notifyd = state->notifyd;
1126         size_t i;
1127         bool ok;
1128         time_t now = time(NULL);
1129
1130         ok = tevent_wakeup_recv(subreq);
1131         TALLOC_FREE(subreq);
1132         if (!ok) {
1133                 tevent_req_oom(req);
1134                 return;
1135         }
1136
1137         i = 0;
1138         while (i < notifyd->num_peers) {
1139                 struct notifyd_peer *p = notifyd->peers[i];
1140
1141                 if ((now - p->last_broadcast) > 60) {
1142                         struct server_id_buf idbuf;
1143
1144                         /*
1145                          * Haven't heard for more than 60 seconds. Call this
1146                          * peer dead
1147                          */
1148
1149                         DEBUG(10, ("%s: peer %s died\n", __func__,
1150                                    server_id_str_buf(p->pid, &idbuf)));
1151                         /*
1152                          * This implicitly decrements notifyd->num_peers
1153                          */
1154                         TALLOC_FREE(p);
1155                 } else {
1156                         i += 1;
1157                 }
1158         }
1159
1160         subreq = tevent_wakeup_send(state, state->ev,
1161                                     timeval_current_ofs_msec(30000));
1162         if (tevent_req_nomem(subreq, req)) {
1163                 return;
1164         }
1165         tevent_req_set_callback(subreq, notifyd_clean_peers_next, req);
1166 }
1167
1168 static int notifyd_clean_peers_recv(struct tevent_req *req)
1169 {
1170         return tevent_req_simple_recv_unix(req);
1171 }
1172
1173 #endif
1174
1175 static int notifyd_add_proxy_syswatches(struct db_record *rec,
1176                                         void *private_data)
1177 {
1178         struct notifyd_state *state = talloc_get_type_abort(
1179                 private_data, struct notifyd_state);
1180         struct db_context *db = dbwrap_record_get_db(rec);
1181         TDB_DATA key = dbwrap_record_get_key(rec);
1182         TDB_DATA value = dbwrap_record_get_value(rec);
1183         struct notifyd_instance *instances = NULL;
1184         size_t num_instances = 0;
1185         size_t i;
1186         char path[key.dsize+1];
1187         bool ok;
1188
1189         memcpy(path, key.dptr, key.dsize);
1190         path[key.dsize] = '\0';
1191
1192         ok = notifyd_parse_entry(value.dptr, value.dsize, &instances,
1193                                  &num_instances);
1194         if (!ok) {
1195                 DEBUG(1, ("%s: Could not parse notifyd entry for %s\n",
1196                           __func__, path));
1197                 return 0;
1198         }
1199
1200         for (i=0; i<num_instances; i++) {
1201                 struct notifyd_instance *instance = &instances[i];
1202                 uint32_t filter = instance->instance.filter;
1203                 uint32_t subdir_filter = instance->instance.subdir_filter;
1204                 int ret;
1205
1206                 ret = state->sys_notify_watch(
1207                         db, state->sys_notify_ctx, path,
1208                         &filter, &subdir_filter,
1209                         notifyd_sys_callback, state->msg_ctx,
1210                         &instance->sys_watch);
1211                 if (ret != 0) {
1212                         DEBUG(1, ("%s: inotify_watch returned %s\n",
1213                                   __func__, strerror(errno)));
1214                 }
1215         }
1216
1217         return 0;
1218 }
1219
1220 #ifdef CLUSTER_SUPPORT
1221
1222 static int notifyd_db_del_syswatches(struct db_record *rec, void *private_data)
1223 {
1224         TDB_DATA key = dbwrap_record_get_key(rec);
1225         TDB_DATA value = dbwrap_record_get_value(rec);
1226         struct notifyd_instance *instances = NULL;
1227         size_t num_instances = 0;
1228         size_t i;
1229         bool ok;
1230
1231         ok = notifyd_parse_entry(value.dptr, value.dsize, &instances,
1232                                  &num_instances);
1233         if (!ok) {
1234                 DEBUG(1, ("%s: Could not parse notifyd entry for %.*s\n",
1235                           __func__, (int)key.dsize, (char *)key.dptr));
1236                 return 0;
1237         }
1238         for (i=0; i<num_instances; i++) {
1239                 TALLOC_FREE(instances[i].sys_watch);
1240         }
1241         return 0;
1242 }
1243
1244 static int notifyd_peer_destructor(struct notifyd_peer *p)
1245 {
1246         struct notifyd_state *state = p->state;
1247         size_t i;
1248
1249         if (p->db != NULL) {
1250                 dbwrap_traverse_read(p->db, notifyd_db_del_syswatches,
1251                                      NULL, NULL);
1252         }
1253
1254         for (i = 0; i<state->num_peers; i++) {
1255                 if (p == state->peers[i]) {
1256                         state->peers[i] = state->peers[state->num_peers-1];
1257                         state->num_peers -= 1;
1258                         break;
1259                 }
1260         }
1261         return 0;
1262 }
1263
1264 static struct notifyd_peer *notifyd_peer_new(
1265         struct notifyd_state *state, struct server_id pid)
1266 {
1267         struct notifyd_peer *p, **tmp;
1268
1269         tmp = talloc_realloc(state, state->peers, struct notifyd_peer *,
1270                              state->num_peers+1);
1271         if (tmp == NULL) {
1272                 return NULL;
1273         }
1274         state->peers = tmp;
1275
1276         p = talloc_zero(state->peers, struct notifyd_peer);
1277         if (p == NULL) {
1278                 return NULL;
1279         }
1280         p->state = state;
1281         p->pid = pid;
1282
1283         state->peers[state->num_peers] = p;
1284         state->num_peers += 1;
1285
1286         talloc_set_destructor(p, notifyd_peer_destructor);
1287
1288         return p;
1289 }
1290
1291 static void notifyd_apply_reclog(struct notifyd_peer *peer,
1292                                  const uint8_t *msg, size_t msglen)
1293 {
1294         struct notifyd_state *state = peer->state;
1295         DATA_BLOB blob = { .data = discard_const_p(uint8_t, msg),
1296                            .length = msglen };
1297         struct server_id_buf idbuf;
1298         struct messaging_reclog *log;
1299         enum ndr_err_code ndr_err;
1300         uint32_t i;
1301
1302         if (peer->db == NULL) {
1303                 /*
1304                  * No db yet
1305                  */
1306                 return;
1307         }
1308
1309         log = talloc(peer, struct messaging_reclog);
1310         if (log == NULL) {
1311                 DEBUG(10, ("%s: talloc failed\n", __func__));
1312                 return;
1313         }
1314
1315         ndr_err = ndr_pull_struct_blob_all(
1316                 &blob, log, log,
1317                 (ndr_pull_flags_fn_t)ndr_pull_messaging_reclog);
1318         if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
1319                 DEBUG(10, ("%s: ndr_pull_messaging_reclog failed: %s\n",
1320                            __func__, ndr_errstr(ndr_err)));
1321                 goto fail;
1322         }
1323
1324         DEBUG(10, ("%s: Got %u recs index %ju from %s\n", __func__,
1325                    (unsigned)log->num_recs, (uintmax_t)log->rec_index,
1326                    server_id_str_buf(peer->pid, &idbuf)));
1327
1328         if (log->rec_index != peer->rec_index) {
1329                 DEBUG(3, ("%s: Got rec index %ju from %s, expected %ju\n",
1330                           __func__, (uintmax_t)log->rec_index,
1331                           server_id_str_buf(peer->pid, &idbuf),
1332                           (uintmax_t)peer->rec_index));
1333                 goto fail;
1334         }
1335
1336         for (i=0; i<log->num_recs; i++) {
1337                 struct messaging_rec *r = log->recs[i];
1338                 struct notify_rec_change_msg *chg;
1339                 size_t pathlen;
1340                 bool ok;
1341
1342                 ok = notifyd_parse_rec_change(r->buf.data, r->buf.length,
1343                                               &chg, &pathlen);
1344                 if (!ok) {
1345                         DEBUG(3, ("%s: notifyd_parse_rec_change failed\n",
1346                                   __func__));
1347                         goto fail;
1348                 }
1349
1350                 ok = notifyd_apply_rec_change(&r->src, chg->path, pathlen,
1351                                               &chg->instance, peer->db,
1352                                               state->sys_notify_watch,
1353                                               state->sys_notify_ctx,
1354                                               state->msg_ctx);
1355                 if (!ok) {
1356                         DEBUG(3, ("%s: notifyd_apply_rec_change failed\n",
1357                                   __func__));
1358                         goto fail;
1359                 }
1360         }
1361
1362         peer->rec_index += 1;
1363         peer->last_broadcast = time(NULL);
1364
1365         TALLOC_FREE(log);
1366         return;
1367
1368 fail:
1369         DEBUG(10, ("%s: Dropping peer %s\n", __func__,
1370                    server_id_str_buf(peer->pid, &idbuf)));
1371         TALLOC_FREE(peer);
1372 }
1373
1374 /*
1375  * Receive messaging_reclog (log of MSG_SMB_NOTIFY_REC_CHANGE
1376  * messages) broadcasts by other notifyds. Several cases:
1377  *
1378  * We don't know the source. This creates a new peer. Creating a peer
1379  * involves asking the peer for its full database. We assume ordered
1380  * messages, so the new database will arrive before the next broadcast
1381  * will.
1382  *
1383  * We know the source and the log index matches. We will apply the log
1384  * locally to our peer's db as if we had received it from a local
1385  * client.
1386  *
1387  * We know the source but the log index does not match. This means we
1388  * lost a message. We just drop the whole peer and wait for the next
1389  * broadcast, which will then trigger a fresh database pull.
1390  */
1391
1392 static int notifyd_snoop_broadcast(uint32_t src_vnn, uint32_t dst_vnn,
1393                                    uint64_t dst_srvid,
1394                                    const uint8_t *msg, size_t msglen,
1395                                    void *private_data)
1396 {
1397         struct notifyd_state *state = talloc_get_type_abort(
1398                 private_data, struct notifyd_state);
1399         struct server_id my_id = messaging_server_id(state->msg_ctx);
1400         struct notifyd_peer *p;
1401         uint32_t i;
1402         uint32_t msg_type;
1403         struct server_id src, dst;
1404         struct server_id_buf idbuf;
1405         NTSTATUS status;
1406
1407         if (msglen < MESSAGE_HDR_LENGTH) {
1408                 DEBUG(10, ("%s: Got short broadcast\n", __func__));
1409                 return 0;
1410         }
1411         message_hdr_get(&msg_type, &src, &dst, msg);
1412
1413         if (msg_type != MSG_SMB_NOTIFY_REC_CHANGES) {
1414                 DEBUG(10, ("%s Got message %u, ignoring\n", __func__,
1415                            (unsigned)msg_type));
1416                 return 0;
1417         }
1418         if (server_id_equal(&src, &my_id)) {
1419                 DEBUG(10, ("%s: Ignoring my own broadcast\n", __func__));
1420                 return 0;
1421         }
1422
1423         DEBUG(10, ("%s: Got MSG_SMB_NOTIFY_REC_CHANGES from %s\n",
1424                    __func__, server_id_str_buf(src, &idbuf)));
1425
1426         for (i=0; i<state->num_peers; i++) {
1427                 if (server_id_equal(&state->peers[i]->pid, &src)) {
1428
1429                         DEBUG(10, ("%s: Applying changes to peer %u\n",
1430                                    __func__, (unsigned)i));
1431
1432                         notifyd_apply_reclog(state->peers[i],
1433                                              msg + MESSAGE_HDR_LENGTH,
1434                                              msglen - MESSAGE_HDR_LENGTH);
1435                         return 0;
1436                 }
1437         }
1438
1439         DEBUG(10, ("%s: Creating new peer for %s\n", __func__,
1440                    server_id_str_buf(src, &idbuf)));
1441
1442         p = notifyd_peer_new(state, src);
1443         if (p == NULL) {
1444                 DEBUG(10, ("%s: notifyd_peer_new failed\n", __func__));
1445                 return 0;
1446         }
1447
1448         status = messaging_send_buf(state->msg_ctx, src, MSG_SMB_NOTIFY_GET_DB,
1449                                     NULL, 0);
1450         if (!NT_STATUS_IS_OK(status)) {
1451                 DEBUG(10, ("%s: messaging_send_buf failed: %s\n",
1452                            __func__, nt_errstr(status)));
1453                 TALLOC_FREE(p);
1454                 return 0;
1455         }
1456
1457         return 0;
1458 }
1459 #endif
1460
1461 struct notifyd_parse_db_state {
1462         bool (*fn)(const char *path,
1463                    struct server_id server,
1464                    const struct notify_instance *instance,
1465                    void *private_data);
1466         void *private_data;
1467 };
1468
1469 static bool notifyd_parse_db_parser(TDB_DATA key, TDB_DATA value,
1470                                     void *private_data)
1471 {
1472         struct notifyd_parse_db_state *state = private_data;
1473         char path[key.dsize+1];
1474         struct notifyd_instance *instances = NULL;
1475         size_t num_instances = 0;
1476         size_t i;
1477         bool ok;
1478
1479         memcpy(path, key.dptr, key.dsize);
1480         path[key.dsize] = 0;
1481
1482         ok = notifyd_parse_entry(value.dptr, value.dsize, &instances,
1483                                  &num_instances);
1484         if (!ok) {
1485                 DEBUG(10, ("%s: Could not parse entry for path %s\n",
1486                            __func__, path));
1487                 return true;
1488         }
1489
1490         for (i=0; i<num_instances; i++) {
1491                 ok = state->fn(path, instances[i].client,
1492                                &instances[i].instance,
1493                                state->private_data);
1494                 if (!ok) {
1495                         return false;
1496                 }
1497         }
1498
1499         return true;
1500 }
1501
1502 int notifyd_parse_db(const uint8_t *buf, size_t buflen,
1503                      uint64_t *log_index,
1504                      bool (*fn)(const char *path,
1505                                 struct server_id server,
1506                                 const struct notify_instance *instance,
1507                                 void *private_data),
1508                      void *private_data)
1509 {
1510         struct notifyd_parse_db_state state = {
1511                 .fn = fn, .private_data = private_data
1512         };
1513         NTSTATUS status;
1514
1515         if (buflen < 8) {
1516                 return EINVAL;
1517         }
1518         *log_index = BVAL(buf, 0);
1519
1520         buf += 8;
1521         buflen -= 8;
1522
1523         status = dbwrap_parse_marshall_buf(
1524                 buf, buflen, notifyd_parse_db_parser, &state);
1525         if (!NT_STATUS_IS_OK(status)) {
1526                 return map_errno_from_nt_status(status);
1527         }
1528
1529         return 0;
1530 }