source3/lib/dbwrap/dbwrap_watch.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3    Watch dbwrap record changes
   4    Copyright (C) Volker Lendecke 2012
   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 "system/filesys.h"
  22 #include "lib/util/server_id.h"
  23 #include "dbwrap/dbwrap.h"
  24 #include "dbwrap_watch.h"
  25 #include "dbwrap_open.h"
  26 #include "lib/util/util_tdb.h"
  27 #include "lib/util/tevent_ntstatus.h"
  28 #include "serverid.h"
  29 #include "server_id_watch.h"
  30 #include "lib/dbwrap/dbwrap_private.h"
  31
  32 struct dbwrap_watcher {
  33         /*
  34          * Process watching this record
  35          */
  36         struct server_id pid;
  37         /*
  38          * Individual instance inside the waiter, incremented each
  39          * time a watcher is created
  40          */
  41         uint64_t instance;
  42 };
  43
  44 #define DBWRAP_WATCHER_BUF_LENGTH (SERVER_ID_BUF_LENGTH + sizeof(uint64_t))
  45 #define DBWRAP_MAX_WATCHERS (INT32_MAX/DBWRAP_WATCHER_BUF_LENGTH)
  46
  47 /*
  48  * Watched records contain a header of:
  49  *
  50  * [uint32] num_records
  51  * 0 [DBWRAP_WATCHER_BUF_LENGTH]              \
  52  * 1 [DBWRAP_WATCHER_BUF_LENGTH]              |
  53  * ..                                         |- Array of watchers
  54  * (num_records-1)[DBWRAP_WATCHER_BUF_LENGTH] /
  55  *
  56  * [Remainder of record....]
  57  *
  58  * If this header is absent then this is a
  59  * fresh record of length zero (no watchers).
  60  */
  61
  62 static bool dbwrap_watch_rec_parse(
  63         TDB_DATA data,
  64         uint8_t **pwatchers,
  65         size_t *pnum_watchers,
  66         TDB_DATA *pdata)
  67 {
  68         size_t num_watchers;
  69
  70         if (data.dsize == 0) {
  71                 /* Fresh record */
  72                 if (pwatchers != NULL) {
  73                         *pwatchers = NULL;
  74                 }
  75                 if (pnum_watchers != NULL) {
  76                         *pnum_watchers = 0;
  77                 }
  78                 if (pdata != NULL) {
  79                         *pdata = (TDB_DATA) { .dptr = NULL };
  80                 }
  81                 return true;
  82         }
  83
  84         if (data.dsize < sizeof(uint32_t)) {
  85                 /* Invalid record */
  86                 return false;
  87         }
  88
  89         num_watchers = IVAL(data.dptr, 0);
  90
  91         data.dptr += sizeof(uint32_t);
  92         data.dsize -= sizeof(uint32_t);
  93
  94         if (num_watchers > data.dsize/DBWRAP_WATCHER_BUF_LENGTH) {
  95                 /* Invalid record */
  96                 return false;
  97         }
  98
  99         if (pwatchers != NULL) {
 100                 *pwatchers = data.dptr;
 101         }
 102         if (pnum_watchers != NULL) {
 103                 *pnum_watchers = num_watchers;
 104         }
 105         if (pdata != NULL) {
 106                 size_t watchers_len = num_watchers * DBWRAP_WATCHER_BUF_LENGTH;
 107                 *pdata = (TDB_DATA) {
 108                         .dptr = data.dptr + watchers_len,
 109                         .dsize = data.dsize - watchers_len
 110                 };
 111         }
 112
 113         return true;
 114 }
 115
 116 static void dbwrap_watcher_get(struct dbwrap_watcher *w,
 117                                const uint8_t buf[DBWRAP_WATCHER_BUF_LENGTH])
 118 {
 119         server_id_get(&w->pid, buf);
 120         w->instance = BVAL(buf, SERVER_ID_BUF_LENGTH);
 121 }
 122
 123 static void dbwrap_watcher_put(uint8_t buf[DBWRAP_WATCHER_BUF_LENGTH],
 124                                const struct dbwrap_watcher *w)
 125 {
 126         server_id_put(buf, w->pid);
 127         SBVAL(buf, SERVER_ID_BUF_LENGTH, w->instance);
 128 }
 129
 130 static void dbwrap_watch_log_invalid_record(
 131         struct db_context *db, TDB_DATA key, TDB_DATA value)
 132 {
 133         DBG_ERR("Found invalid record in %s\n", dbwrap_name(db));
 134         dump_data(1, key.dptr, key.dsize);
 135         dump_data(1, value.dptr, value.dsize);
 136 }
 137
 138 struct db_watched_ctx {
 139         struct db_context *backend;
 140         struct messaging_context *msg;
 141 };
 142
 143 struct db_watched_record {
 144         struct db_record *rec;
 145         struct server_id self;
 146         struct {
 147                 struct db_record *rec;
 148                 TDB_DATA initial_value;
 149                 bool initial_valid;
 150         } backend;
 151         bool force_fini_store;
 152         struct dbwrap_watcher added;
 153         bool removed_first;
 154         struct {
 155                 /*
 156                  * The is the number of watcher records
 157                  * parsed from backend.initial_value
 158                  */
 159                 size_t count;
 160                 /*
 161                  * This is the pointer to
 162                  * the optentially first watcher record
 163                  * parsed from backend.initial_value
 164                  *
 165                  * The pointer actually points to memory
 166                  * in backend.initial_value.
 167                  *
 168                  * Note it might be NULL, if count is 0.
 169                  */
 170                 uint8_t *first;
 171                 /*
 172                  * This remembers if we already
 173                  * notified the watchers.
 174                  *
 175                  * As we only need to do that once during:
 176                  *   do_locked
 177                  * or:
 178                  *   between rec = fetch_locked
 179                  *   and
 180                  *   TALLOC_FREE(rec)
 181                  */
 182                 bool alerted;
 183         } watchers;
 184         struct {
 185                 struct dbwrap_watcher watcher;
 186         } wakeup;
 187 };
 188
 189 static struct db_watched_record *db_record_get_watched_record(struct db_record *rec)
 190 {
 191         /*
 192          * we can't use wrec = talloc_get_type_abort() here!
 193          * because wrec is likely a stack variable in
 194          * dbwrap_watched_do_locked_fn()
 195          *
 196          * In order to have a least some protection
 197          * we verify the cross reference pointers
 198          * between rec and wrec
 199          */
 200         struct db_watched_record *wrec =
 201                 (struct db_watched_record *)rec->private_data;
 202         SMB_ASSERT(wrec->rec == rec);
 203         return wrec;
 204 }
 205
 206 static NTSTATUS dbwrap_watched_record_storev(
 207         struct db_watched_record *wrec,
 208         const TDB_DATA *dbufs, int num_dbufs, int flags);
 209 static NTSTATUS dbwrap_watched_storev(struct db_record *rec,
 210                                       const TDB_DATA *dbufs, int num_dbufs,
 211                                       int flags);
 212 static NTSTATUS dbwrap_watched_delete(struct db_record *rec);
 213 static void dbwrap_watched_trigger_wakeup(struct messaging_context *msg_ctx,
 214                                           struct dbwrap_watcher *watcher);
 215 static int db_watched_record_destructor(struct db_watched_record *wrec);
 216
 217 static void db_watched_record_init(struct db_context *db,
 218                                    struct messaging_context *msg_ctx,
 219                                    struct db_record *rec,
 220                                    struct db_watched_record *wrec,
 221                                    struct db_record *backend_rec,
 222                                    TDB_DATA backend_value)
 223 {
 224         bool ok;
 225
 226         *rec = (struct db_record) {
 227                 .db = db,
 228                 .key = dbwrap_record_get_key(backend_rec),
 229                 .storev = dbwrap_watched_storev,
 230                 .delete_rec = dbwrap_watched_delete,
 231                 .private_data = wrec,
 232         };
 233
 234         *wrec = (struct db_watched_record) {
 235                 .rec = rec,
 236                 .self = messaging_server_id(msg_ctx),
 237                 .backend = {
 238                         .rec = backend_rec,
 239                         .initial_value = backend_value,
 240                         .initial_valid = true,
 241                 },
 242         };
 243
 244         ok = dbwrap_watch_rec_parse(backend_value,
 245                                     &wrec->watchers.first,
 246                                     &wrec->watchers.count,
 247                                     &rec->value);
 248         if (!ok) {
 249                 dbwrap_watch_log_invalid_record(rec->db, rec->key, backend_value);
 250                 /* wipe invalid data */
 251                 rec->value = (TDB_DATA) { .dptr = NULL, .dsize = 0 };
 252         }
 253 }
 254
 255 static struct db_record *dbwrap_watched_fetch_locked(
 256         struct db_context *db, TALLOC_CTX *mem_ctx, TDB_DATA key)
 257 {
 258         struct db_watched_ctx *ctx = talloc_get_type_abort(
 259                 db->private_data, struct db_watched_ctx);
 260         struct db_record *rec = NULL;
 261         struct db_watched_record *wrec = NULL;
 262         struct db_record *backend_rec = NULL;
 263         TDB_DATA backend_value = { .dptr = NULL, };
 264
 265         rec = talloc_zero(mem_ctx, struct db_record);
 266         if (rec == NULL) {
 267                 return NULL;
 268         }
 269         wrec = talloc_zero(rec, struct db_watched_record);
 270         if (wrec == NULL) {
 271                 TALLOC_FREE(rec);
 272                 return NULL;
 273         }
 274
 275         backend_rec = dbwrap_fetch_locked(ctx->backend, wrec, key);
 276         if (backend_rec == NULL) {
 277                 TALLOC_FREE(rec);
 278                 return NULL;
 279         }
 280         backend_value = dbwrap_record_get_value(backend_rec);
 281
 282         db_watched_record_init(db, ctx->msg,
 283                                rec, wrec,
 284                                backend_rec, backend_value);
 285         rec->value_valid = true;
 286         talloc_set_destructor(wrec, db_watched_record_destructor);
 287
 288         return rec;
 289 }
 290
 291 struct db_watched_record_fini_state {
 292         struct db_watched_record *wrec;
 293         TALLOC_CTX *frame;
 294         TDB_DATA dbufs[2];
 295         int num_dbufs;
 296         bool ok;
 297 };
 298
 299 static void db_watched_record_fini_fetcher(TDB_DATA key,
 300                                            TDB_DATA backend_value,
 301                                            void *private_data)
 302 {
 303         struct db_watched_record_fini_state *state =
 304                 (struct db_watched_record_fini_state *)private_data;
 305         struct db_watched_record *wrec = state->wrec;
 306         struct db_record *rec = wrec->rec;
 307         TDB_DATA value = {};
 308         bool ok;
 309         size_t copy_size;
 310
 311         /*
 312          * We're within dbwrap_parse_record()
 313          * and backend_value directly points into
 314          * the mmap'ed tdb, so we need to copy the
 315          * parts we require.
 316          */
 317
 318         ok = dbwrap_watch_rec_parse(backend_value, NULL, NULL, &value);
 319         if (!ok) {
 320                 struct db_context *db = dbwrap_record_get_db(rec);
 321
 322                 dbwrap_watch_log_invalid_record(db, key, backend_value);
 323
 324                 /* wipe invalid data */
 325                 value = (TDB_DATA) { .dptr = NULL, .dsize = 0 };
 326         }
 327
 328         copy_size = MIN(rec->value.dsize, value.dsize);
 329         if (copy_size != 0) {
 330                 /*
 331                  * First reuse the buffer we already had
 332                  * as much as we can.
 333                  */
 334                 memcpy(rec->value.dptr, value.dptr, copy_size);
 335                 state->dbufs[state->num_dbufs++] = rec->value;
 336                 value.dsize -= copy_size;
 337                 value.dptr += copy_size;
 338         }
 339
 340         if (value.dsize != 0) {
 341                 uint8_t *p = NULL;
 342
 343                 /*
 344                  * There's still new data left
 345                  * allocate it on callers stackframe
 346                  */
 347                 p = talloc_memdup(state->frame, value.dptr, value.dsize);
 348                 if (p == NULL) {
 349                         DBG_WARNING("failed to allocate %zu bytes\n",
 350                                     value.dsize);
 351                         return;
 352                 }
 353
 354                 state->dbufs[state->num_dbufs++] = (TDB_DATA) {
 355                         .dptr = p, .dsize = value.dsize,
 356                 };
 357         }
 358
 359         state->ok = true;
 360 }
 361
 362 static void db_watched_record_fini(struct db_watched_record *wrec)
 363 {
 364         struct db_watched_record_fini_state state = { .wrec = wrec, };
 365         struct db_context *backend = dbwrap_record_get_db(wrec->backend.rec);
 366         struct db_record *rec = wrec->rec;
 367         TDB_DATA key = dbwrap_record_get_key(wrec->backend.rec);
 368         NTSTATUS status;
 369
 370         if (!wrec->force_fini_store) {
 371                 return;
 372         }
 373
 374         if (wrec->backend.initial_valid) {
 375                 if (rec->value.dsize != 0) {
 376                         state.dbufs[state.num_dbufs++] = rec->value;
 377                 }
 378         } else {
 379                 /*
 380                  * We need to fetch the current
 381                  * value from the backend again,
 382                  * which may need to allocate memory
 383                  * on the provided stackframe.
 384                  */
 385
 386                 state.frame = talloc_stackframe();
 387
 388                 status = dbwrap_parse_record(backend, key,
 389                                 db_watched_record_fini_fetcher, &state);
 390                 if (!NT_STATUS_IS_OK(status)) {
 391                         DBG_WARNING("dbwrap_parse_record failed: %s\n",
 392                                     nt_errstr(status));
 393                         TALLOC_FREE(state.frame);
 394                         return;
 395                 }
 396                 if (!state.ok) {
 397                         TALLOC_FREE(state.frame);
 398                         return;
 399                 }
 400         }
 401
 402         /*
 403          * We don't want to wake up others just because
 404          * we added ourself as new watcher. But if we
 405          * removed outself from the first position
 406          * we need to alert the next one.
 407          */
 408         if (!wrec->removed_first) {
 409                 dbwrap_watched_watch_skip_alerting(rec);
 410         }
 411
 412         status = dbwrap_watched_record_storev(wrec, state.dbufs, state.num_dbufs, 0);
 413         TALLOC_FREE(state.frame);
 414         if (!NT_STATUS_IS_OK(status)) {
 415                 DBG_WARNING("dbwrap_watched_record_storev failed: %s\n",
 416                             nt_errstr(status));
 417                 return;
 418         }
 419
 420         return;
 421 }
 422
 423 static int db_watched_record_destructor(struct db_watched_record *wrec)
 424 {
 425         struct db_record *rec = wrec->rec;
 426         struct db_watched_ctx *ctx = talloc_get_type_abort(
 427                 rec->db->private_data, struct db_watched_ctx);
 428
 429         db_watched_record_fini(wrec);
 430         TALLOC_FREE(wrec->backend.rec);
 431         dbwrap_watched_trigger_wakeup(ctx->msg, &wrec->wakeup.watcher);
 432         return 0;
 433 }
 434
 435 struct dbwrap_watched_do_locked_state {
 436         struct db_context *db;
 437         struct messaging_context *msg_ctx;
 438         struct db_watched_record *wrec;
 439         struct db_record *rec;
 440         void (*fn)(struct db_record *rec,
 441                    TDB_DATA value,
 442                    void *private_data);
 443         void *private_data;
 444 };
 445
 446 static void dbwrap_watched_do_locked_fn(
 447         struct db_record *backend_rec,
 448         TDB_DATA backend_value,
 449         void *private_data)
 450 {
 451         struct dbwrap_watched_do_locked_state *state =
 452                 (struct dbwrap_watched_do_locked_state *)private_data;
 453
 454         db_watched_record_init(state->db, state->msg_ctx,
 455                                state->rec, state->wrec,
 456                                backend_rec, backend_value);
 457
 458         state->fn(state->rec, state->rec->value, state->private_data);
 459
 460         db_watched_record_fini(state->wrec);
 461 }
 462
 463 static NTSTATUS dbwrap_watched_do_locked(struct db_context *db, TDB_DATA key,
 464                                          void (*fn)(struct db_record *rec,
 465                                                     TDB_DATA value,
 466                                                     void *private_data),
 467                                          void *private_data)
 468 {
 469         struct db_watched_ctx *ctx = talloc_get_type_abort(
 470                 db->private_data, struct db_watched_ctx);
 471         struct db_watched_record wrec;
 472         struct db_record rec;
 473         struct dbwrap_watched_do_locked_state state = {
 474                 .db = db, .msg_ctx = ctx->msg,
 475                 .rec = &rec, .wrec = &wrec,
 476                 .fn = fn, .private_data = private_data,
 477         };
 478         NTSTATUS status;
 479
 480         status = dbwrap_do_locked(
 481                 ctx->backend, key, dbwrap_watched_do_locked_fn, &state);
 482         if (!NT_STATUS_IS_OK(status)) {
 483                 DBG_DEBUG("dbwrap_do_locked returned %s\n", nt_errstr(status));
 484                 return status;
 485         }
 486
 487         DBG_DEBUG("dbwrap_watched_do_locked_fn returned\n");
 488
 489         dbwrap_watched_trigger_wakeup(state.msg_ctx, &wrec.wakeup.watcher);
 490
 491         return NT_STATUS_OK;
 492 }
 493
 494 static void dbwrap_watched_record_prepare_wakeup(
 495         struct db_watched_record *wrec)
 496 {
 497         /*
 498          * Wakeup only needs to happen once (if at all)
 499          */
 500         if (wrec->watchers.alerted) {
 501                 /* already done */
 502                 return;
 503         }
 504         wrec->watchers.alerted = true;
 505
 506         if (wrec->watchers.count == 0) {
 507                 DBG_DEBUG("No watchers\n");
 508                 return;
 509         }
 510
 511         while (wrec->watchers.count != 0) {
 512                 struct server_id_buf tmp;
 513                 bool exists;
 514
 515                 dbwrap_watcher_get(&wrec->wakeup.watcher, wrec->watchers.first);
 516                 exists = serverid_exists(&wrec->wakeup.watcher.pid);
 517                 if (!exists) {
 518                         DBG_DEBUG("Discard non-existing waiter %s:%"PRIu64"\n",
 519                                   server_id_str_buf(wrec->wakeup.watcher.pid, &tmp),
 520                                   wrec->wakeup.watcher.instance);
 521                         wrec->watchers.first += DBWRAP_WATCHER_BUF_LENGTH;
 522                         wrec->watchers.count -= 1;
 523                         continue;
 524                 }
 525
 526                 /*
 527                  * We will only wakeup the first waiter, via
 528                  * dbwrap_watched_trigger_wakeup(), but keep
 529                  * all (including the first one) in the list that
 530                  * will be flushed back to the backend record
 531                  * again. Waiters are removing their entries
 532                  * via dbwrap_watched_watch_remove_instance()
 533                  * when they no longer want to monitor the record.
 534                  */
 535                 DBG_DEBUG("Will alert first waiter %s:%"PRIu64"\n",
 536                           server_id_str_buf(wrec->wakeup.watcher.pid, &tmp),
 537                           wrec->wakeup.watcher.instance);
 538                 break;
 539         }
 540 }
 541
 542 static void dbwrap_watched_trigger_wakeup(struct messaging_context *msg_ctx,
 543                                           struct dbwrap_watcher *watcher)
 544 {
 545         struct server_id_buf tmp;
 546         uint8_t instance_buf[8];
 547         NTSTATUS status;
 548
 549         if (watcher->instance == 0) {
 550                 DBG_DEBUG("No one to wakeup\n");
 551                 return;
 552         }
 553
 554         DBG_DEBUG("Alerting %s:%"PRIu64"\n",
 555                   server_id_str_buf(watcher->pid, &tmp),
 556                   watcher->instance);
 557
 558         SBVAL(instance_buf, 0, watcher->instance);
 559
 560         status = messaging_send_buf(
 561                 msg_ctx,
 562                 watcher->pid,
 563                 MSG_DBWRAP_MODIFIED,
 564                 instance_buf,
 565                 sizeof(instance_buf));
 566         if (!NT_STATUS_IS_OK(status)) {
 567                 DBG_WARNING("messaging_send_buf to %s failed: %s - ignoring...\n",
 568                             server_id_str_buf(watcher->pid, &tmp),
 569                             nt_errstr(status));
 570         }
 571 }
 572
 573 static NTSTATUS dbwrap_watched_record_storev(
 574         struct db_watched_record *wrec,
 575         const TDB_DATA *dbufs, int num_dbufs, int flags)
 576 {
 577         uint8_t num_watchers_buf[4] = { 0 };
 578         uint8_t add_buf[DBWRAP_WATCHER_BUF_LENGTH];
 579         size_t num_store_watchers;
 580         TDB_DATA my_dbufs[num_dbufs+3];
 581         int num_my_dbufs = 0;
 582         NTSTATUS status;
 583         size_t add_count = 0;
 584
 585         dbwrap_watched_record_prepare_wakeup(wrec);
 586
 587         wrec->backend.initial_valid = false;
 588         wrec->force_fini_store = false;
 589
 590         if (wrec->added.pid.pid != 0) {
 591                 dbwrap_watcher_put(add_buf, &wrec->added);
 592                 add_count = 1;
 593         }
 594
 595         num_store_watchers = wrec->watchers.count + add_count;
 596         if (num_store_watchers == 0 && num_dbufs == 0) {
 597                 status = dbwrap_record_delete(wrec->backend.rec);
 598                 return status;
 599         }
 600         if (num_store_watchers >= DBWRAP_MAX_WATCHERS) {
 601                 DBG_WARNING("Can't handle %zu watchers\n",
 602                             num_store_watchers);
 603                 return NT_STATUS_INSUFFICIENT_RESOURCES;
 604         }
 605
 606         SIVAL(num_watchers_buf, 0, num_store_watchers);
 607
 608         my_dbufs[num_my_dbufs++] = (TDB_DATA) {
 609                 .dptr = num_watchers_buf, .dsize = sizeof(num_watchers_buf),
 610         };
 611         if (wrec->watchers.count != 0) {
 612                 my_dbufs[num_my_dbufs++] = (TDB_DATA) {
 613                         .dptr = wrec->watchers.first, .dsize = wrec->watchers.count * DBWRAP_WATCHER_BUF_LENGTH,
 614                 };
 615         }
 616         if (add_count != 0) {
 617                 my_dbufs[num_my_dbufs++] = (TDB_DATA) {
 618                         .dptr = add_buf,
 619                         .dsize = sizeof(add_buf),
 620                 };
 621         }
 622         if (num_dbufs != 0) {
 623                 memcpy(my_dbufs+num_my_dbufs, dbufs, num_dbufs * sizeof(*dbufs));
 624                 num_my_dbufs += num_dbufs;
 625         }
 626
 627         SMB_ASSERT(num_my_dbufs <= ARRAY_SIZE(my_dbufs));
 628
 629         status = dbwrap_record_storev(
 630                 wrec->backend.rec, my_dbufs, num_my_dbufs, flags);
 631         return status;
 632 }
 633
 634 static NTSTATUS dbwrap_watched_storev(struct db_record *rec,
 635                                       const TDB_DATA *dbufs, int num_dbufs,
 636                                       int flags)
 637 {
 638         struct db_watched_record *wrec = db_record_get_watched_record(rec);
 639
 640         return dbwrap_watched_record_storev(wrec, dbufs, num_dbufs, flags);
 641 }
 642
 643 static NTSTATUS dbwrap_watched_delete(struct db_record *rec)
 644 {
 645         struct db_watched_record *wrec = db_record_get_watched_record(rec);
 646
 647         /*
 648          * dbwrap_watched_record_storev() will figure out
 649          * if the record should be deleted or if there are still
 650          * watchers to be stored.
 651          */
 652         return dbwrap_watched_record_storev(wrec, NULL, 0, 0);
 653 }
 654
 655 struct dbwrap_watched_traverse_state {
 656         int (*fn)(struct db_record *rec, void *private_data);
 657         void *private_data;
 658 };
 659
 660 static int dbwrap_watched_traverse_fn(struct db_record *rec,
 661                                       void *private_data)
 662 {
 663         struct dbwrap_watched_traverse_state *state = private_data;
 664         struct db_record prec = *rec;
 665         bool ok;
 666
 667         ok = dbwrap_watch_rec_parse(rec->value, NULL, NULL, &prec.value);
 668         if (!ok) {
 669                 return 0;
 670         }
 671         if (prec.value.dsize == 0) {
 672                 return 0;
 673         }
 674         prec.value_valid = true;
 675
 676         return state->fn(&prec, state->private_data);
 677 }
 678
 679 static int dbwrap_watched_traverse(struct db_context *db,
 680                                    int (*fn)(struct db_record *rec,
 681                                              void *private_data),
 682                                    void *private_data)
 683 {
 684         struct db_watched_ctx *ctx = talloc_get_type_abort(
 685                 db->private_data, struct db_watched_ctx);
 686         struct dbwrap_watched_traverse_state state = {
 687                 .fn = fn, .private_data = private_data };
 688         NTSTATUS status;
 689         int ret;
 690
 691         status = dbwrap_traverse(
 692                 ctx->backend, dbwrap_watched_traverse_fn, &state, &ret);
 693         if (!NT_STATUS_IS_OK(status)) {
 694                 return -1;
 695         }
 696         return ret;
 697 }
 698
 699 static int dbwrap_watched_traverse_read(struct db_context *db,
 700                                         int (*fn)(struct db_record *rec,
 701                                                   void *private_data),
 702                                         void *private_data)
 703 {
 704         struct db_watched_ctx *ctx = talloc_get_type_abort(
 705                 db->private_data, struct db_watched_ctx);
 706         struct dbwrap_watched_traverse_state state = {
 707                 .fn = fn, .private_data = private_data };
 708         NTSTATUS status;
 709         int ret;
 710
 711         status = dbwrap_traverse_read(
 712                 ctx->backend, dbwrap_watched_traverse_fn, &state, &ret);
 713         if (!NT_STATUS_IS_OK(status)) {
 714                 return -1;
 715         }
 716         return ret;
 717 }
 718
 719 static int dbwrap_watched_get_seqnum(struct db_context *db)
 720 {
 721         struct db_watched_ctx *ctx = talloc_get_type_abort(
 722                 db->private_data, struct db_watched_ctx);
 723         return dbwrap_get_seqnum(ctx->backend);
 724 }
 725
 726 static int dbwrap_watched_transaction_start(struct db_context *db)
 727 {
 728         struct db_watched_ctx *ctx = talloc_get_type_abort(
 729                 db->private_data, struct db_watched_ctx);
 730         return dbwrap_transaction_start(ctx->backend);
 731 }
 732
 733 static int dbwrap_watched_transaction_commit(struct db_context *db)
 734 {
 735         struct db_watched_ctx *ctx = talloc_get_type_abort(
 736                 db->private_data, struct db_watched_ctx);
 737         return dbwrap_transaction_commit(ctx->backend);
 738 }
 739
 740 static int dbwrap_watched_transaction_cancel(struct db_context *db)
 741 {
 742         struct db_watched_ctx *ctx = talloc_get_type_abort(
 743                 db->private_data, struct db_watched_ctx);
 744         return dbwrap_transaction_cancel(ctx->backend);
 745 }
 746
 747 struct dbwrap_watched_parse_record_state {
 748         struct db_context *db;
 749         void (*parser)(TDB_DATA key, TDB_DATA data, void *private_data);
 750         void *private_data;
 751         bool ok;
 752 };
 753
 754 static void dbwrap_watched_parse_record_parser(TDB_DATA key, TDB_DATA data,
 755                                                void *private_data)
 756 {
 757         struct dbwrap_watched_parse_record_state *state = private_data;
 758         TDB_DATA userdata;
 759
 760         state->ok = dbwrap_watch_rec_parse(data, NULL, NULL, &userdata);
 761         if (!state->ok) {
 762                 dbwrap_watch_log_invalid_record(state->db, key, data);
 763                 return;
 764         }
 765
 766         state->parser(key, userdata, state->private_data);
 767 }
 768
 769 static NTSTATUS dbwrap_watched_parse_record(
 770         struct db_context *db, TDB_DATA key,
 771         void (*parser)(TDB_DATA key, TDB_DATA data, void *private_data),
 772         void *private_data)
 773 {
 774         struct db_watched_ctx *ctx = talloc_get_type_abort(
 775                 db->private_data, struct db_watched_ctx);
 776         struct dbwrap_watched_parse_record_state state = {
 777                 .db = db,
 778                 .parser = parser,
 779                 .private_data = private_data,
 780         };
 781         NTSTATUS status;
 782
 783         status = dbwrap_parse_record(
 784                 ctx->backend, key, dbwrap_watched_parse_record_parser, &state);
 785         if (!NT_STATUS_IS_OK(status)) {
 786                 return status;
 787         }
 788         if (!state.ok) {
 789                 return NT_STATUS_NOT_FOUND;
 790         }
 791         return NT_STATUS_OK;
 792 }
 793
 794 static void dbwrap_watched_parse_record_done(struct tevent_req *subreq);
 795
 796 static struct tevent_req *dbwrap_watched_parse_record_send(
 797         TALLOC_CTX *mem_ctx,
 798         struct tevent_context *ev,
 799         struct db_context *db,
 800         TDB_DATA key,
 801         void (*parser)(TDB_DATA key, TDB_DATA data, void *private_data),
 802         void *private_data,
 803         enum dbwrap_req_state *req_state)
 804 {
 805         struct db_watched_ctx *ctx = talloc_get_type_abort(
 806                 db->private_data, struct db_watched_ctx);
 807         struct tevent_req *req = NULL;
 808         struct tevent_req *subreq = NULL;
 809         struct dbwrap_watched_parse_record_state *state = NULL;
 810
 811         req = tevent_req_create(mem_ctx, &state,
 812                                 struct dbwrap_watched_parse_record_state);
 813         if (req == NULL) {
 814                 *req_state = DBWRAP_REQ_ERROR;
 815                 return NULL;
 816         }
 817
 818         *state = (struct dbwrap_watched_parse_record_state) {
 819                 .parser = parser,
 820                 .private_data = private_data,
 821                 .ok = true,
 822         };
 823
 824         subreq = dbwrap_parse_record_send(state,
 825                                           ev,
 826                                           ctx->backend,
 827                                           key,
 828                                           dbwrap_watched_parse_record_parser,
 829                                           state,
 830                                           req_state);
 831         if (tevent_req_nomem(subreq, req)) {
 832                 *req_state = DBWRAP_REQ_ERROR;
 833                 return tevent_req_post(req, ev);
 834         }
 835
 836         tevent_req_set_callback(subreq, dbwrap_watched_parse_record_done, req);
 837         return req;
 838 }
 839
 840 static void dbwrap_watched_parse_record_done(struct tevent_req *subreq)
 841 {
 842         struct tevent_req *req = tevent_req_callback_data(
 843                 subreq, struct tevent_req);
 844         struct dbwrap_watched_parse_record_state *state = tevent_req_data(
 845                 req, struct dbwrap_watched_parse_record_state);
 846         NTSTATUS status;
 847
 848         status = dbwrap_parse_record_recv(subreq);
 849         TALLOC_FREE(subreq);
 850         if (tevent_req_nterror(req, status)) {
 851                 return;
 852         }
 853
 854         if (!state->ok) {
 855                 tevent_req_nterror(req, NT_STATUS_NOT_FOUND);
 856                 return;
 857         }
 858
 859         tevent_req_done(req);
 860         return;
 861 }
 862
 863 static NTSTATUS dbwrap_watched_parse_record_recv(struct tevent_req *req)
 864 {
 865         NTSTATUS status;
 866
 867         if (tevent_req_is_nterror(req, &status)) {
 868                 tevent_req_received(req);
 869                 return status;
 870         }
 871
 872         tevent_req_received(req);
 873         return NT_STATUS_OK;
 874 }
 875
 876 static int dbwrap_watched_exists(struct db_context *db, TDB_DATA key)
 877 {
 878         struct db_watched_ctx *ctx = talloc_get_type_abort(
 879                 db->private_data, struct db_watched_ctx);
 880
 881         return dbwrap_exists(ctx->backend, key);
 882 }
 883
 884 static size_t dbwrap_watched_id(struct db_context *db, uint8_t *id,
 885                                 size_t idlen)
 886 {
 887         struct db_watched_ctx *ctx = talloc_get_type_abort(
 888                 db->private_data, struct db_watched_ctx);
 889
 890         return dbwrap_db_id(ctx->backend, id, idlen);
 891 }
 892
 893 struct db_context *db_open_watched(TALLOC_CTX *mem_ctx,
 894                                    struct db_context **backend,
 895                                    struct messaging_context *msg)
 896 {
 897         struct db_context *db;
 898         struct db_watched_ctx *ctx;
 899
 900         db = talloc_zero(mem_ctx, struct db_context);
 901         if (db == NULL) {
 902                 return NULL;
 903         }
 904         ctx = talloc_zero(db, struct db_watched_ctx);
 905         if (ctx == NULL) {
 906                 TALLOC_FREE(db);
 907                 return NULL;
 908         }
 909         db->private_data = ctx;
 910
 911         ctx->msg = msg;
 912
 913         ctx->backend = talloc_move(ctx, backend);
 914         db->lock_order = ctx->backend->lock_order;
 915         ctx->backend->lock_order = DBWRAP_LOCK_ORDER_NONE;
 916
 917         db->fetch_locked = dbwrap_watched_fetch_locked;
 918         db->do_locked = dbwrap_watched_do_locked;
 919         db->traverse = dbwrap_watched_traverse;
 920         db->traverse_read = dbwrap_watched_traverse_read;
 921         db->get_seqnum = dbwrap_watched_get_seqnum;
 922         db->transaction_start = dbwrap_watched_transaction_start;
 923         db->transaction_commit = dbwrap_watched_transaction_commit;
 924         db->transaction_cancel = dbwrap_watched_transaction_cancel;
 925         db->parse_record = dbwrap_watched_parse_record;
 926         db->parse_record_send = dbwrap_watched_parse_record_send;
 927         db->parse_record_recv = dbwrap_watched_parse_record_recv;
 928         db->exists = dbwrap_watched_exists;
 929         db->id = dbwrap_watched_id;
 930         db->name = dbwrap_name(ctx->backend);
 931
 932         return db;
 933 }
 934
 935 uint64_t dbwrap_watched_watch_add_instance(struct db_record *rec)
 936 {
 937         struct db_watched_record *wrec = db_record_get_watched_record(rec);
 938         static uint64_t global_instance = 1;
 939
 940         SMB_ASSERT(wrec->added.instance == 0);
 941
 942         wrec->added = (struct dbwrap_watcher) {
 943                 .pid = wrec->self,
 944                 .instance = global_instance++,
 945         };
 946
 947         wrec->force_fini_store = true;
 948
 949         return wrec->added.instance;
 950 }
 951
 952 void dbwrap_watched_watch_remove_instance(struct db_record *rec, uint64_t instance)
 953 {
 954         struct db_watched_record *wrec = db_record_get_watched_record(rec);
 955         struct dbwrap_watcher clear_watcher = {
 956                 .pid = wrec->self,
 957                 .instance = instance,
 958         };
 959         size_t i;
 960         struct server_id_buf buf;
 961
 962         if (instance == 0) {
 963                 return;
 964         }
 965
 966         if (wrec->added.instance == instance) {
 967                 SMB_ASSERT(server_id_equal(&wrec->added.pid, &wrec->self));
 968                 DBG_DEBUG("Watcher %s:%"PRIu64" reverted from adding\n",
 969                           server_id_str_buf(clear_watcher.pid, &buf),
 970                           clear_watcher.instance);
 971                 ZERO_STRUCT(wrec->added);
 972         }
 973
 974         for (i=0; i < wrec->watchers.count; i++) {
 975                 struct dbwrap_watcher watcher;
 976                 size_t off = i*DBWRAP_WATCHER_BUF_LENGTH;
 977                 size_t next_off;
 978                 size_t full_len;
 979                 size_t move_len;
 980
 981                 dbwrap_watcher_get(&watcher, wrec->watchers.first + off);
 982
 983                 if (clear_watcher.instance != watcher.instance) {
 984                         continue;
 985                 }
 986                 if (!server_id_equal(&clear_watcher.pid, &watcher.pid)) {
 987                         continue;
 988                 }
 989
 990                 wrec->force_fini_store = true;
 991
 992                 if (i == 0) {
 993                         DBG_DEBUG("Watcher %s:%"PRIu64" removed from first position of %zu\n",
 994                                   server_id_str_buf(clear_watcher.pid, &buf),
 995                                   clear_watcher.instance,
 996                                   wrec->watchers.count);
 997                         wrec->watchers.first += DBWRAP_WATCHER_BUF_LENGTH;
 998                         wrec->watchers.count -= 1;
 999                         wrec->removed_first = true;
1000                         return;
1001                 }
1002                 if (i == (wrec->watchers.count-1)) {
1003                         DBG_DEBUG("Watcher %s:%"PRIu64" removed from last position of %zu\n",
1004                                   server_id_str_buf(clear_watcher.pid, &buf),
1005                                   clear_watcher.instance,
1006                                   wrec->watchers.count);
1007                         wrec->watchers.count -= 1;
1008                         return;
1009                 }
1010
1011                 DBG_DEBUG("Watcher %s:%"PRIu64" cleared at position %zu from %zu\n",
1012                           server_id_str_buf(clear_watcher.pid, &buf),
1013                           clear_watcher.instance, i+1,
1014                           wrec->watchers.count);
1015
1016                 next_off = off + DBWRAP_WATCHER_BUF_LENGTH;
1017                 full_len = wrec->watchers.count * DBWRAP_WATCHER_BUF_LENGTH;
1018                 move_len = full_len - next_off;
1019                 memmove(wrec->watchers.first + off,
1020                         wrec->watchers.first + next_off,
1021                         move_len);
1022                 wrec->watchers.count -= 1;
1023                 return;
1024         }
1025
1026         DBG_DEBUG("Watcher %s:%"PRIu64" not found in %zu watchers\n",
1027                   server_id_str_buf(clear_watcher.pid, &buf),
1028                   clear_watcher.instance,
1029                   wrec->watchers.count);
1030         return;
1031 }
1032
1033 void dbwrap_watched_watch_skip_alerting(struct db_record *rec)
1034 {
1035         struct db_watched_record *wrec = db_record_get_watched_record(rec);
1036
1037         wrec->wakeup.watcher = (struct dbwrap_watcher) { .instance = 0, };
1038         wrec->watchers.alerted = true;
1039 }
1040
1041 void dbwrap_watched_watch_reset_alerting(struct db_record *rec)
1042 {
1043         struct db_watched_record *wrec = db_record_get_watched_record(rec);
1044
1045         wrec->wakeup.watcher = (struct dbwrap_watcher) { .instance = 0, };
1046         wrec->watchers.alerted = false;
1047 }
1048
1049 void dbwrap_watched_watch_force_alerting(struct db_record *rec)
1050 {
1051         struct db_watched_record *wrec = db_record_get_watched_record(rec);
1052
1053         dbwrap_watched_record_prepare_wakeup(wrec);
1054 }
1055
1056 struct dbwrap_watched_watch_state {
1057         struct db_context *db;
1058         TDB_DATA key;
1059         struct dbwrap_watcher watcher;
1060         struct server_id blocker;
1061         bool blockerdead;
1062 };
1063
1064 static bool dbwrap_watched_msg_filter(struct messaging_rec *rec,
1065                                       void *private_data);
1066 static void dbwrap_watched_watch_done(struct tevent_req *subreq);
1067 static void dbwrap_watched_watch_blocker_died(struct tevent_req *subreq);
1068 static int dbwrap_watched_watch_state_destructor(
1069         struct dbwrap_watched_watch_state *state);
1070
1071 struct tevent_req *dbwrap_watched_watch_send(TALLOC_CTX *mem_ctx,
1072                                              struct tevent_context *ev,
1073                                              struct db_record *rec,
1074                                              uint64_t resumed_instance,
1075                                              struct server_id blocker)
1076 {
1077         struct db_context *db = dbwrap_record_get_db(rec);
1078         struct db_watched_ctx *ctx = talloc_get_type_abort(
1079                 db->private_data, struct db_watched_ctx);
1080         struct db_watched_record *wrec = db_record_get_watched_record(rec);
1081         struct tevent_req *req, *subreq;
1082         struct dbwrap_watched_watch_state *state;
1083         uint64_t instance;
1084
1085         req = tevent_req_create(mem_ctx, &state,
1086                                 struct dbwrap_watched_watch_state);
1087         if (req == NULL) {
1088                 return NULL;
1089         }
1090         state->db = db;
1091         state->blocker = blocker;
1092
1093         if (ctx->msg == NULL) {
1094                 tevent_req_nterror(req, NT_STATUS_NOT_SUPPORTED);
1095                 return tevent_req_post(req, ev);
1096         }
1097
1098         if (resumed_instance == 0 && wrec->added.instance == 0) {
1099                 /*
1100                  * Adding a new instance
1101                  */
1102                 instance = dbwrap_watched_watch_add_instance(rec);
1103         } else if (resumed_instance != 0 && wrec->added.instance == 0) {
1104                 /*
1105                  * Resuming an existing instance that was
1106                  * already present before do_locked started
1107                  */
1108                 instance = resumed_instance;
1109         } else if (resumed_instance == wrec->added.instance) {
1110                 /*
1111                  * The caller used dbwrap_watched_watch_add_instance()
1112                  * already during this do_locked() invocation.
1113                  */
1114                 instance = resumed_instance;
1115         } else {
1116                 tevent_req_nterror(req, NT_STATUS_REQUEST_NOT_ACCEPTED);
1117                 return tevent_req_post(req, ev);
1118         }
1119
1120         state->watcher = (struct dbwrap_watcher) {
1121                 .pid = messaging_server_id(ctx->msg),
1122                 .instance = instance,
1123         };
1124
1125         state->key = tdb_data_talloc_copy(state, rec->key);
1126         if (tevent_req_nomem(state->key.dptr, req)) {
1127                 return tevent_req_post(req, ev);
1128         }
1129
1130         subreq = messaging_filtered_read_send(
1131                 state, ev, ctx->msg, dbwrap_watched_msg_filter, state);
1132         if (tevent_req_nomem(subreq, req)) {
1133                 return tevent_req_post(req, ev);
1134         }
1135         tevent_req_set_callback(subreq, dbwrap_watched_watch_done, req);
1136
1137         talloc_set_destructor(state, dbwrap_watched_watch_state_destructor);
1138
1139         if (blocker.pid != 0) {
1140                 subreq = server_id_watch_send(state, ev, blocker);
1141                 if (tevent_req_nomem(subreq, req)) {
1142                         return tevent_req_post(req, ev);
1143                 }
1144                 tevent_req_set_callback(
1145                         subreq, dbwrap_watched_watch_blocker_died, req);
1146         }
1147
1148         return req;
1149 }
1150
1151 static void dbwrap_watched_watch_blocker_died(struct tevent_req *subreq)
1152 {
1153         struct tevent_req *req = tevent_req_callback_data(
1154                 subreq, struct tevent_req);
1155         struct dbwrap_watched_watch_state *state = tevent_req_data(
1156                 req, struct dbwrap_watched_watch_state);
1157         int ret;
1158
1159         ret = server_id_watch_recv(subreq, NULL);
1160         TALLOC_FREE(subreq);
1161         if (ret != 0) {
1162                 tevent_req_nterror(req, map_nt_error_from_unix(ret));
1163                 return;
1164         }
1165         state->blockerdead = true;
1166         tevent_req_done(req);
1167 }
1168
1169 static void dbwrap_watched_watch_state_destructor_fn(
1170         struct db_record *rec,
1171         TDB_DATA value,
1172         void *private_data)
1173 {
1174         struct dbwrap_watched_watch_state *state = talloc_get_type_abort(
1175                 private_data, struct dbwrap_watched_watch_state);
1176
1177         /*
1178          * Here we just remove ourself from the in memory
1179          * watchers array and let db_watched_record_fini()
1180          * call dbwrap_watched_record_storev() to do the magic
1181          * of writing back the modified in memory copy.
1182          */
1183         dbwrap_watched_watch_remove_instance(rec, state->watcher.instance);
1184         return;
1185 }
1186
1187 static int dbwrap_watched_watch_state_destructor(
1188         struct dbwrap_watched_watch_state *state)
1189 {
1190         NTSTATUS status;
1191
1192         status = dbwrap_do_locked(
1193                 state->db,
1194                 state->key,
1195                 dbwrap_watched_watch_state_destructor_fn,
1196                 state);
1197         if (!NT_STATUS_IS_OK(status)) {
1198                 DBG_WARNING("dbwrap_do_locked failed: %s\n",
1199                             nt_errstr(status));
1200         }
1201         return 0;
1202 }
1203
1204 static bool dbwrap_watched_msg_filter(struct messaging_rec *rec,
1205                                       void *private_data)
1206 {
1207         struct dbwrap_watched_watch_state *state = talloc_get_type_abort(
1208                 private_data, struct dbwrap_watched_watch_state);
1209         uint64_t instance;
1210
1211         if (rec->msg_type != MSG_DBWRAP_MODIFIED) {
1212                 return false;
1213         }
1214         if (rec->num_fds != 0) {
1215                 return false;
1216         }
1217
1218         if (rec->buf.length != sizeof(instance)) {
1219                 DBG_DEBUG("Got size %zu, expected %zu\n",
1220                           rec->buf.length,
1221                           sizeof(instance));
1222                 return false;
1223         }
1224
1225         instance = BVAL(rec->buf.data, 0);
1226
1227         if (instance != state->watcher.instance) {
1228                 DBG_DEBUG("Got instance %"PRIu64", expected %"PRIu64"\n",
1229                           instance,
1230                           state->watcher.instance);
1231                 return false;
1232         }
1233
1234         return true;
1235 }
1236
1237 static void dbwrap_watched_watch_done(struct tevent_req *subreq)
1238 {
1239         struct tevent_req *req = tevent_req_callback_data(
1240                 subreq, struct tevent_req);
1241         struct dbwrap_watched_watch_state *state = tevent_req_data(
1242                 req, struct dbwrap_watched_watch_state);
1243         struct messaging_rec *rec;
1244         int ret;
1245
1246         ret = messaging_filtered_read_recv(subreq, state, &rec);
1247         TALLOC_FREE(subreq);
1248         if (ret != 0) {
1249                 tevent_req_nterror(req, map_nt_error_from_unix(ret));
1250                 return;
1251         }
1252         tevent_req_done(req);
1253 }
1254
1255 NTSTATUS dbwrap_watched_watch_recv(struct tevent_req *req,
1256                                    uint64_t *pkeep_instance,
1257                                    bool *blockerdead,
1258                                    struct server_id *blocker)
1259 {
1260         struct dbwrap_watched_watch_state *state = tevent_req_data(
1261                 req, struct dbwrap_watched_watch_state);
1262         NTSTATUS status;
1263
1264         if (tevent_req_is_nterror(req, &status)) {
1265                 tevent_req_received(req);
1266                 return status;
1267         }
1268         if (pkeep_instance != NULL) {
1269                 *pkeep_instance = state->watcher.instance;
1270                 /*
1271                  * No need to remove ourselves anymore,
1272                  * the caller will take care of removing itself.
1273                  */
1274                 talloc_set_destructor(state, NULL);
1275         }
1276         if (blockerdead != NULL) {
1277                 *blockerdead = state->blockerdead;
1278         }
1279         if (blocker != NULL) {
1280                 *blocker = state->blocker;
1281         }
1282         tevent_req_received(req);
1283         return NT_STATUS_OK;
1284 }
1285