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