source4/dsdb/samdb/ldb_modules/partition.c

   1 /*
   2    Partitions ldb module
   3
   4    Copyright (C) Andrew Bartlett <abartlet@samba.org> 2006
   5    Copyright (C) Stefan Metzmacher <metze@samba.org> 2007
   6
   7    This program is free software; you can redistribute it and/or modify
   8    it under the terms of the GNU General Public License as published by
   9    the Free Software Foundation; either version 3 of the License, or
  10    (at your option) any later version.
  11
  12    This program is distributed in the hope that it will be useful,
  13    but WITHOUT ANY WARRANTY; without even the implied warranty of
  14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15    GNU General Public License for more details.
  16
  17    You should have received a copy of the GNU General Public License
  18    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19 */
  20
  21 /*
  22  *  Name: ldb
  23  *
  24  *  Component: ldb partitions module
  25  *
  26  *  Description: Implement LDAP partitions
  27  *
  28  *  Author: Andrew Bartlett
  29  *  Author: Stefan Metzmacher
  30  */
  31
  32 #include "dsdb/samdb/ldb_modules/partition.h"
  33
  34 struct part_request {
  35         struct ldb_module *module;
  36         struct ldb_request *req;
  37 };
  38
  39 struct partition_context {
  40         struct ldb_module *module;
  41         struct ldb_request *req;
  42
  43         struct part_request *part_req;
  44         unsigned int num_requests;
  45         unsigned int finished_requests;
  46
  47         const char **referrals;
  48 };
  49
  50 static struct partition_context *partition_init_ctx(struct ldb_module *module, struct ldb_request *req)
  51 {
  52         struct partition_context *ac;
  53
  54         ac = talloc_zero(req, struct partition_context);
  55         if (ac == NULL) {
  56                 ldb_set_errstring(ldb_module_get_ctx(module), "Out of Memory");
  57                 return NULL;
  58         }
  59
  60         ac->module = module;
  61         ac->req = req;
  62
  63         return ac;
  64 }
  65
  66 /*
  67  * helper functions to call the next module in chain
  68  */
  69 int partition_request(struct ldb_module *module, struct ldb_request *request)
  70 {
  71         if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) { \
  72                 const struct dsdb_control_current_partition *partition = NULL;
  73                 struct ldb_control *partition_ctrl = ldb_request_get_control(request, DSDB_CONTROL_CURRENT_PARTITION_OID);
  74                 if (partition_ctrl) {
  75                         partition = talloc_get_type(partition_ctrl->data,
  76                                                     struct dsdb_control_current_partition);
  77                 }
  78
  79                 if (partition != NULL) {
  80                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_request() -> %s",
  81                                   ldb_dn_get_linearized(partition->dn));
  82                 } else {
  83                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_request() -> (metadata partition)");
  84                 }
  85         }
  86
  87         return ldb_next_request(module, request);
  88 }
  89
  90 static struct dsdb_partition *find_partition(struct partition_private_data *data,
  91                                              struct ldb_dn *dn,
  92                                              struct ldb_request *req)
  93 {
  94         unsigned int i;
  95         struct ldb_control *partition_ctrl;
  96
  97         /* see if the request has the partition DN specified in a
  98          * control. The repl_meta_data module can specify this to
  99          * ensure that replication happens to the right partition
 100          */
 101         partition_ctrl = ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID);
 102         if (partition_ctrl) {
 103                 const struct dsdb_control_current_partition *partition;
 104                 partition = talloc_get_type(partition_ctrl->data,
 105                                             struct dsdb_control_current_partition);
 106                 if (partition != NULL) {
 107                         dn = partition->dn;
 108                 }
 109         }
 110
 111         if (dn == NULL) {
 112                 return NULL;
 113         }
 114
 115         /* Look at base DN */
 116         /* Figure out which partition it is under */
 117         /* Skip the lot if 'data' isn't here yet (initialisation) */
 118         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 119                 if (ldb_dn_compare_base(data->partitions[i]->ctrl->dn, dn) == 0) {
 120                         return data->partitions[i];
 121                 }
 122         }
 123
 124         return NULL;
 125 }
 126
 127 /**
 128  * fire the caller's callback for every entry, but only send 'done' once.
 129  */
 130 static int partition_req_callback(struct ldb_request *req,
 131                                   struct ldb_reply *ares)
 132 {
 133         struct partition_context *ac;
 134         struct ldb_module *module;
 135         struct ldb_request *nreq;
 136         int ret;
 137         struct partition_private_data *data;
 138         struct ldb_control *partition_ctrl;
 139
 140         ac = talloc_get_type(req->context, struct partition_context);
 141         data = talloc_get_type(ldb_module_get_private(ac->module), struct partition_private_data);
 142
 143         if (!ares) {
 144                 return ldb_module_done(ac->req, NULL, NULL,
 145                                         LDB_ERR_OPERATIONS_ERROR);
 146         }
 147
 148         partition_ctrl = ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID);
 149         if (partition_ctrl && (ac->num_requests == 1 || ares->type == LDB_REPLY_ENTRY)) {
 150                 /* If we didn't fan this request out to mulitple partitions,
 151                  * or this is an individual search result, we can
 152                  * deterministically tell the caller what partition this was
 153                  * written to (repl_meta_data likes to know) */
 154                 ret = ldb_reply_add_control(ares,
 155                                             DSDB_CONTROL_CURRENT_PARTITION_OID,
 156                                             false, partition_ctrl->data);
 157                 if (ret != LDB_SUCCESS) {
 158                         return ldb_module_done(ac->req, NULL, NULL,
 159                                                ret);
 160                 }
 161         }
 162
 163         if (ares->error != LDB_SUCCESS) {
 164                 return ldb_module_done(ac->req, ares->controls,
 165                                         ares->response, ares->error);
 166         }
 167
 168         switch (ares->type) {
 169         case LDB_REPLY_REFERRAL:
 170                 return ldb_module_send_referral(ac->req, ares->referral);
 171
 172         case LDB_REPLY_ENTRY:
 173                 if (ac->req->operation != LDB_SEARCH) {
 174                         ldb_set_errstring(ldb_module_get_ctx(ac->module),
 175                                 "partition_req_callback:"
 176                                 " Unsupported reply type for this request");
 177                         return ldb_module_done(ac->req, NULL, NULL,
 178                                                 LDB_ERR_OPERATIONS_ERROR);
 179                 }
 180
 181                 return ldb_module_send_entry(ac->req, ares->message, ares->controls);
 182
 183         case LDB_REPLY_DONE:
 184                 if (ac->req->operation == LDB_EXTENDED) {
 185                         /* FIXME: check for ares->response, replmd does not fill it ! */
 186                         if (ares->response) {
 187                                 if (strcmp(ares->response->oid, LDB_EXTENDED_START_TLS_OID) != 0) {
 188                                         ldb_set_errstring(ldb_module_get_ctx(ac->module),
 189                                                           "partition_req_callback:"
 190                                                           " Unknown extended reply, "
 191                                                           "only supports START_TLS");
 192                                         talloc_free(ares);
 193                                         return ldb_module_done(ac->req, NULL, NULL,
 194                                                                 LDB_ERR_OPERATIONS_ERROR);
 195                                 }
 196                         }
 197                 }
 198
 199                 ac->finished_requests++;
 200                 if (ac->finished_requests == ac->num_requests) {
 201                         /* Send back referrals if they do exist (search ops) */
 202                         if (ac->referrals != NULL) {
 203                                 const char **ref;
 204                                 for (ref = ac->referrals; *ref != NULL; ++ref) {
 205                                         ret = ldb_module_send_referral(ac->req,
 206                                                                        talloc_strdup(ac->req, *ref));
 207                                         if (ret != LDB_SUCCESS) {
 208                                                 return ldb_module_done(ac->req, NULL, NULL,
 209                                                                        ret);
 210                                         }
 211                                 }
 212                         }
 213
 214                         /* this was the last one, call callback */
 215                         return ldb_module_done(ac->req, ares->controls,
 216                                                ares->response,
 217                                                ares->error);
 218                 }
 219
 220                 /* not the last, now call the next one */
 221                 module = ac->part_req[ac->finished_requests].module;
 222                 nreq = ac->part_req[ac->finished_requests].req;
 223
 224                 ret = partition_request(module, nreq);
 225                 if (ret != LDB_SUCCESS) {
 226                         talloc_free(ares);
 227                         return ldb_module_done(ac->req, NULL, NULL, ret);
 228                 }
 229
 230                 break;
 231         }
 232
 233         talloc_free(ares);
 234         return LDB_SUCCESS;
 235 }
 236
 237 static int partition_prep_request(struct partition_context *ac,
 238                                   struct dsdb_partition *partition)
 239 {
 240         int ret;
 241         struct ldb_request *req;
 242         struct ldb_control *partition_ctrl = NULL;
 243
 244         ac->part_req = talloc_realloc(ac, ac->part_req,
 245                                         struct part_request,
 246                                         ac->num_requests + 1);
 247         if (ac->part_req == NULL) {
 248                 return ldb_oom(ldb_module_get_ctx(ac->module));
 249         }
 250
 251         switch (ac->req->operation) {
 252         case LDB_SEARCH:
 253                 ret = ldb_build_search_req_ex(&req, ldb_module_get_ctx(ac->module),
 254                                         ac->part_req,
 255                                         ac->req->op.search.base,
 256                                         ac->req->op.search.scope,
 257                                         ac->req->op.search.tree,
 258                                         ac->req->op.search.attrs,
 259                                         ac->req->controls,
 260                                         ac, partition_req_callback,
 261                                         ac->req);
 262                 LDB_REQ_SET_LOCATION(req);
 263                 break;
 264         case LDB_ADD:
 265                 ret = ldb_build_add_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
 266                                         ac->req->op.add.message,
 267                                         ac->req->controls,
 268                                         ac, partition_req_callback,
 269                                         ac->req);
 270                 LDB_REQ_SET_LOCATION(req);
 271                 break;
 272         case LDB_MODIFY:
 273                 ret = ldb_build_mod_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
 274                                         ac->req->op.mod.message,
 275                                         ac->req->controls,
 276                                         ac, partition_req_callback,
 277                                         ac->req);
 278                 LDB_REQ_SET_LOCATION(req);
 279                 break;
 280         case LDB_DELETE:
 281                 ret = ldb_build_del_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
 282                                         ac->req->op.del.dn,
 283                                         ac->req->controls,
 284                                         ac, partition_req_callback,
 285                                         ac->req);
 286                 LDB_REQ_SET_LOCATION(req);
 287                 break;
 288         case LDB_RENAME:
 289                 ret = ldb_build_rename_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
 290                                         ac->req->op.rename.olddn,
 291                                         ac->req->op.rename.newdn,
 292                                         ac->req->controls,
 293                                         ac, partition_req_callback,
 294                                         ac->req);
 295                 LDB_REQ_SET_LOCATION(req);
 296                 break;
 297         case LDB_EXTENDED:
 298                 ret = ldb_build_extended_req(&req, ldb_module_get_ctx(ac->module),
 299                                         ac->part_req,
 300                                         ac->req->op.extended.oid,
 301                                         ac->req->op.extended.data,
 302                                         ac->req->controls,
 303                                         ac, partition_req_callback,
 304                                         ac->req);
 305                 LDB_REQ_SET_LOCATION(req);
 306                 break;
 307         default:
 308                 ldb_set_errstring(ldb_module_get_ctx(ac->module),
 309                                   "Unsupported request type!");
 310                 ret = LDB_ERR_UNWILLING_TO_PERFORM;
 311         }
 312
 313         if (ret != LDB_SUCCESS) {
 314                 return ret;
 315         }
 316
 317         ac->part_req[ac->num_requests].req = req;
 318
 319         if (ac->req->controls) {
 320                 /* Duplicate everything beside the current partition control */
 321                 partition_ctrl = ldb_request_get_control(ac->req,
 322                                                          DSDB_CONTROL_CURRENT_PARTITION_OID);
 323                 if (!ldb_save_controls(partition_ctrl, req, NULL)) {
 324                         return ldb_module_oom(ac->module);
 325                 }
 326         }
 327
 328         if (partition) {
 329                 void *part_data = partition->ctrl;
 330
 331                 ac->part_req[ac->num_requests].module = partition->module;
 332
 333                 if (partition_ctrl != NULL) {
 334                         if (partition_ctrl->data != NULL) {
 335                                 part_data = partition_ctrl->data;
 336                         }
 337
 338                         /*
 339                          * If the provided current partition control is without
 340                          * data then use the calculated one.
 341                          */
 342                         ret = ldb_request_add_control(req,
 343                                                       DSDB_CONTROL_CURRENT_PARTITION_OID,
 344                                                       false, part_data);
 345                         if (ret != LDB_SUCCESS) {
 346                                 return ret;
 347                         }
 348                 }
 349
 350                 if (req->operation == LDB_SEARCH) {
 351                         /* If the search is for 'more' than this partition,
 352                          * then change the basedn, so a remote LDAP server
 353                          * doesn't object */
 354                         if (ldb_dn_compare_base(partition->ctrl->dn,
 355                                                 req->op.search.base) != 0) {
 356                                 req->op.search.base = partition->ctrl->dn;
 357                         }
 358                 }
 359
 360         } else {
 361                 /* make sure you put the module here, or
 362                  * or ldb_next_request() will skip a module */
 363                 ac->part_req[ac->num_requests].module = ac->module;
 364         }
 365
 366         ac->num_requests++;
 367
 368         return LDB_SUCCESS;
 369 }
 370
 371 static int partition_call_first(struct partition_context *ac)
 372 {
 373         return partition_request(ac->part_req[0].module, ac->part_req[0].req);
 374 }
 375
 376 /**
 377  * Send a request down to all the partitions
 378  */
 379 static int partition_send_all(struct ldb_module *module,
 380                               struct partition_context *ac,
 381                               struct ldb_request *req)
 382 {
 383         unsigned int i;
 384         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 385                                                               struct partition_private_data);
 386         int ret = partition_prep_request(ac, NULL);
 387         if (ret != LDB_SUCCESS) {
 388                 return ret;
 389         }
 390         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 391                 ret = partition_prep_request(ac, data->partitions[i]);
 392                 if (ret != LDB_SUCCESS) {
 393                         return ret;
 394                 }
 395         }
 396
 397         /* fire the first one */
 398         return partition_call_first(ac);
 399 }
 400
 401
 402 /**
 403  * send an operation to the top partition, then copy the resulting
 404  * object to all other partitions
 405  */
 406 static int partition_copy_all(struct ldb_module *module,
 407                               struct partition_context *ac,
 408                               struct ldb_request *req,
 409                               struct ldb_dn *dn)
 410 {
 411         unsigned int i;
 412         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 413                                                               struct partition_private_data);
 414         int ret, search_ret;
 415         struct ldb_result *res;
 416
 417         /* do the request on the top level sam.ldb synchronously */
 418         ret = ldb_next_request(module, req);
 419         if (ret != LDB_SUCCESS) {
 420                 return ret;
 421         }
 422         ret = ldb_wait(req->handle, LDB_WAIT_ALL);
 423         if (ret != LDB_SUCCESS) {
 424                 return ret;
 425         }
 426
 427         /* now fetch the resulting object, and then copy it to all the
 428          * other partitions. We need this approach to cope with the
 429          * partitions getting out of sync. If for example the
 430          * @ATTRIBUTES object exists on one partition but not the
 431          * others then just doing each of the partitions in turn will
 432          * lead to an error
 433          */
 434         search_ret = dsdb_module_search_dn(module, ac, &res, dn, NULL, DSDB_FLAG_NEXT_MODULE, req);
 435         if (search_ret != LDB_SUCCESS && ret != LDB_ERR_NO_SUCH_OBJECT) {
 436                 return search_ret;
 437         }
 438
 439         /* now delete the object in the other partitions. Once that is
 440            done we will re-add the object, if search_ret was not
 441            LDB_ERR_NO_SUCH_OBJECT
 442         */
 443         for (i=0; data->partitions && data->partitions[i]; i++) {
 444                 int pret;
 445                 pret = dsdb_module_del(data->partitions[i]->module, dn, DSDB_FLAG_NEXT_MODULE, req);
 446                 if (pret != LDB_SUCCESS && pret != LDB_ERR_NO_SUCH_OBJECT) {
 447                         /* we should only get success or no
 448                            such object from the other partitions */
 449                         return pret;
 450                 }
 451         }
 452
 453
 454         if (search_ret != LDB_ERR_NO_SUCH_OBJECT) {
 455                 /* now re-add in the other partitions */
 456                 for (i=0; data->partitions && data->partitions[i]; i++) {
 457                         int pret;
 458                         pret = dsdb_module_add(data->partitions[i]->module, res->msgs[0], DSDB_FLAG_NEXT_MODULE, req);
 459                         if (pret != LDB_SUCCESS) {
 460                                 return pret;
 461                         }
 462                 }
 463         }
 464
 465         return ldb_module_done(req, NULL, NULL, LDB_SUCCESS);
 466 }
 467
 468 /**
 469  * Figure out which backend a request needs to be aimed at.  Some
 470  * requests must be replicated to all backends
 471  */
 472 static int partition_replicate(struct ldb_module *module, struct ldb_request *req, struct ldb_dn *dn)
 473 {
 474         struct partition_context *ac;
 475         unsigned int i;
 476         int ret;
 477         struct dsdb_partition *partition;
 478         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 479                                                               struct partition_private_data);
 480
 481         /* if we aren't initialised yet go further */
 482         if (!data || !data->partitions) {
 483                 return ldb_next_request(module, req);
 484         }
 485
 486         if (ldb_dn_is_special(dn)) {
 487                 /* Is this a special DN, we need to replicate to every backend? */
 488                 for (i=0; data->replicate && data->replicate[i]; i++) {
 489                         if (ldb_dn_compare(data->replicate[i],
 490                                            dn) == 0) {
 491
 492                                 ac = partition_init_ctx(module, req);
 493                                 if (!ac) {
 494                                         return ldb_operr(ldb_module_get_ctx(module));
 495                                 }
 496
 497                                 return partition_copy_all(module, ac, req, dn);
 498                         }
 499                 }
 500         }
 501
 502         /* Otherwise, we need to find the partition to fire it to */
 503
 504         /* Find partition */
 505         partition = find_partition(data, dn, req);
 506         if (!partition) {
 507                 /*
 508                  * if we haven't found a matching partition
 509                  * pass the request to the main ldb
 510                  *
 511                  * TODO: we should maybe return an error here
 512                  *       if it's not a special dn
 513                  */
 514
 515                 return ldb_next_request(module, req);
 516         }
 517
 518         ac = partition_init_ctx(module, req);
 519         if (!ac) {
 520                 return ldb_operr(ldb_module_get_ctx(module));
 521         }
 522
 523         /* we need to add a control but we never touch the original request */
 524         ret = partition_prep_request(ac, partition);
 525         if (ret != LDB_SUCCESS) {
 526                 return ret;
 527         }
 528
 529         /* fire the first one */
 530         return partition_call_first(ac);
 531 }
 532
 533 /* search */
 534 static int partition_search(struct ldb_module *module, struct ldb_request *req)
 535 {
 536         struct ldb_control **saved_controls;
 537         /* Find backend */
 538         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 539                                                               struct partition_private_data);
 540         struct partition_context *ac;
 541         struct ldb_context *ldb;
 542         struct loadparm_context *lp_ctx;
 543
 544         struct ldb_control *search_control = ldb_request_get_control(req, LDB_CONTROL_SEARCH_OPTIONS_OID);
 545         struct ldb_control *domain_scope_control = ldb_request_get_control(req, LDB_CONTROL_DOMAIN_SCOPE_OID);
 546         struct ldb_control *no_gc_control = ldb_request_get_control(req, DSDB_CONTROL_NO_GLOBAL_CATALOG);
 547
 548         struct ldb_search_options_control *search_options = NULL;
 549         struct dsdb_partition *p;
 550         unsigned int i, j;
 551         int ret;
 552         bool domain_scope = false, phantom_root = false;
 553
 554         /* see if we are still up-to-date */
 555         ret = partition_reload_if_required(module, data, req);
 556         if (ret != LDB_SUCCESS) {
 557                 return ret;
 558         }
 559
 560         p = find_partition(data, NULL, req);
 561         if (p != NULL) {
 562                 /* the caller specified what partition they want the
 563                  * search - just pass it on
 564                  */
 565                 return ldb_next_request(p->module, req);
 566         }
 567
 568         /* Get back the search options from the search control, and mark it as
 569          * non-critical (to make backends and also dcpromo happy).
 570          */
 571         if (search_control) {
 572                 search_options = talloc_get_type(search_control->data, struct ldb_search_options_control);
 573                 search_control->critical = 0;
 574
 575         }
 576
 577         /* Remove the "domain_scope" control, so we don't confuse a backend
 578          * server */
 579         if (domain_scope_control && !ldb_save_controls(domain_scope_control, req, &saved_controls)) {
 580                 return ldb_oom(ldb_module_get_ctx(module));
 581         }
 582
 583         /* if we aren't initialised yet go further */
 584         if (!data || !data->partitions) {
 585                 return ldb_next_request(module, req);
 586         }
 587
 588         /* Locate the options */
 589         domain_scope = (search_options
 590                 && (search_options->search_options & LDB_SEARCH_OPTION_DOMAIN_SCOPE))
 591                 || domain_scope_control;
 592         phantom_root = search_options
 593                 && (search_options->search_options & LDB_SEARCH_OPTION_PHANTOM_ROOT);
 594
 595         /* Remove handled options from the search control flag */
 596         if (search_options) {
 597                 search_options->search_options = search_options->search_options
 598                         & ~LDB_SEARCH_OPTION_DOMAIN_SCOPE
 599                         & ~LDB_SEARCH_OPTION_PHANTOM_ROOT;
 600         }
 601
 602         ac = partition_init_ctx(module, req);
 603         if (!ac) {
 604                 return ldb_operr(ldb_module_get_ctx(module));
 605         }
 606
 607         ldb = ldb_module_get_ctx(ac->module);
 608         lp_ctx = talloc_get_type(ldb_get_opaque(ldb, "loadparm"),
 609                                                 struct loadparm_context);
 610
 611         /* Search from the base DN */
 612         if (ldb_dn_is_null(req->op.search.base)) {
 613                 if (!phantom_root) {
 614                         return ldb_error(ldb, LDB_ERR_NO_SUCH_OBJECT, "empty base DN");
 615                 }
 616                 return partition_send_all(module, ac, req);
 617         }
 618
 619         for (i=0; data->partitions[i]; i++) {
 620                 bool match = false, stop = false;
 621
 622                 if (data->partitions[i]->partial_replica && no_gc_control != NULL) {
 623                         if (ldb_dn_compare_base(data->partitions[i]->ctrl->dn,
 624                                                 req->op.search.base) == 0) {
 625                                 /* base DN is in a partial replica
 626                                    with the NO_GLOBAL_CATALOG
 627                                    control. This partition is invisible */
 628                                 /* DEBUG(0,("DENYING NON-GC OP: %s\n", ldb_module_call_chain(req, req))); */
 629                                 continue;
 630                         }
 631                 }
 632
 633                 if (phantom_root) {
 634                         /* Phantom root: Find all partitions under the
 635                          * search base. We match if:
 636                          *
 637                          * 1) the DN we are looking for exactly matches a
 638                          *    certain partition and always stop
 639                          * 2) the DN we are looking for is a parent of certain
 640                          *    partitions and it isn't a scope base search
 641                          * 3) the DN we are looking for is a child of a certain
 642                          *    partition and always stop
 643                          *    - we don't need to go any further up in the
 644                          *    hierarchy!
 645                          */
 646                         if (ldb_dn_compare(data->partitions[i]->ctrl->dn,
 647                                            req->op.search.base) == 0) {
 648                                 match = true;
 649                                 stop = true;
 650                         }
 651                         if (!match &&
 652                             (ldb_dn_compare_base(req->op.search.base,
 653                                                  data->partitions[i]->ctrl->dn) == 0 &&
 654                              req->op.search.scope != LDB_SCOPE_BASE)) {
 655                                 match = true;
 656                         }
 657                         if (!match &&
 658                             ldb_dn_compare_base(data->partitions[i]->ctrl->dn,
 659                                                 req->op.search.base) == 0) {
 660                                 match = true;
 661                                 stop = true; /* note that this relies on partition ordering */
 662                         }
 663                 } else {
 664                         /* Domain scope: Find all partitions under the search
 665                          * base.
 666                          *
 667                          * We generate referral candidates if we haven't
 668                          * specified the domain scope control, haven't a base
 669                          * search* scope and the DN we are looking for is a real
 670                          * predecessor of certain partitions. When a new
 671                          * referral candidate is nearer to the DN than an
 672                          * existing one delete the latter (we want to have only
 673                          * the closest ones). When we checked this for all
 674                          * candidates we have the final referrals.
 675                          *
 676                          * We match if the DN we are looking for is a child of
 677                          * a certain partition or the partition
 678                          * DN itself - we don't need to go any further
 679                          * up in the hierarchy!
 680                          */
 681                         if ((!domain_scope) &&
 682                             (req->op.search.scope != LDB_SCOPE_BASE) &&
 683                             (ldb_dn_compare_base(req->op.search.base,
 684                                                  data->partitions[i]->ctrl->dn) == 0) &&
 685                             (ldb_dn_compare(req->op.search.base,
 686                                             data->partitions[i]->ctrl->dn) != 0)) {
 687                                 char *ref = talloc_asprintf(ac,
 688                                                             "ldap://%s/%s%s",
 689                                                             lpcfg_dnsdomain(lp_ctx),
 690                                                             ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
 691                                                             req->op.search.scope == LDB_SCOPE_ONELEVEL ? "??base" : "");
 692
 693                                 if (ref == NULL) {
 694                                         return ldb_oom(ldb);
 695                                 }
 696
 697                                 /* Initialise the referrals list */
 698                                 if (ac->referrals == NULL) {
 699                                         ac->referrals = (const char **) str_list_make_empty(ac);
 700                                         if (ac->referrals == NULL) {
 701                                                 return ldb_oom(ldb);
 702                                         }
 703                                 }
 704
 705                                 /* Check if the new referral candidate is
 706                                  * closer to the base DN than already
 707                                  * saved ones and delete the latters */
 708                                 j = 0;
 709                                 while (ac->referrals[j] != NULL) {
 710                                         if (strstr(ac->referrals[j],
 711                                                    ldb_dn_get_linearized(data->partitions[i]->ctrl->dn)) != NULL) {
 712                                                 str_list_remove(ac->referrals,
 713                                                                 ac->referrals[j]);
 714                                         } else {
 715                                                 ++j;
 716                                         }
 717                                 }
 718
 719                                 /* Add our new candidate */
 720                                 ac->referrals = str_list_add(ac->referrals, ref);
 721
 722                                 talloc_free(ref);
 723
 724                                 if (ac->referrals == NULL) {
 725                                         return ldb_oom(ldb);
 726                                 }
 727                         }
 728                         if (ldb_dn_compare_base(data->partitions[i]->ctrl->dn, req->op.search.base) == 0) {
 729                                 match = true;
 730                                 stop = true; /* note that this relies on partition ordering */
 731                         }
 732                 }
 733
 734                 if (match) {
 735                         ret = partition_prep_request(ac, data->partitions[i]);
 736                         if (ret != LDB_SUCCESS) {
 737                                 return ret;
 738                         }
 739                 }
 740
 741                 if (stop) break;
 742         }
 743
 744         /* Perhaps we didn't match any partitions. Try the main partition */
 745         if (ac->num_requests == 0) {
 746                 talloc_free(ac);
 747                 return ldb_next_request(module, req);
 748         }
 749
 750         /* fire the first one */
 751         return partition_call_first(ac);
 752 }
 753
 754 /* add */
 755 static int partition_add(struct ldb_module *module, struct ldb_request *req)
 756 {
 757         return partition_replicate(module, req, req->op.add.message->dn);
 758 }
 759
 760 /* modify */
 761 static int partition_modify(struct ldb_module *module, struct ldb_request *req)
 762 {
 763         return partition_replicate(module, req, req->op.mod.message->dn);
 764 }
 765
 766 /* delete */
 767 static int partition_delete(struct ldb_module *module, struct ldb_request *req)
 768 {
 769         return partition_replicate(module, req, req->op.del.dn);
 770 }
 771
 772 /* rename */
 773 static int partition_rename(struct ldb_module *module, struct ldb_request *req)
 774 {
 775         /* Find backend */
 776         struct dsdb_partition *backend, *backend2;
 777
 778         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 779                                                               struct partition_private_data);
 780
 781         /* Skip the lot if 'data' isn't here yet (initialisation) */
 782         if (!data) {
 783                 return ldb_operr(ldb_module_get_ctx(module));
 784         }
 785
 786         backend = find_partition(data, req->op.rename.olddn, req);
 787         backend2 = find_partition(data, req->op.rename.newdn, req);
 788
 789         if ((backend && !backend2) || (!backend && backend2)) {
 790                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
 791         }
 792
 793         if (backend != backend2) {
 794                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
 795                                        "Cannot rename from %s in %s to %s in %s: %s",
 796                                        ldb_dn_get_linearized(req->op.rename.olddn),
 797                                        ldb_dn_get_linearized(backend->ctrl->dn),
 798                                        ldb_dn_get_linearized(req->op.rename.newdn),
 799                                        ldb_dn_get_linearized(backend2->ctrl->dn),
 800                                        ldb_strerror(LDB_ERR_AFFECTS_MULTIPLE_DSAS));
 801                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
 802         }
 803
 804         return partition_replicate(module, req, req->op.rename.olddn);
 805 }
 806
 807 /* start a transaction */
 808 static int partition_start_trans(struct ldb_module *module)
 809 {
 810         unsigned int i;
 811         int ret;
 812         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 813                                                               struct partition_private_data);
 814         /* Look at base DN */
 815         /* Figure out which partition it is under */
 816         /* Skip the lot if 'data' isn't here yet (initialization) */
 817         if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
 818                 ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_start_trans() -> (metadata partition)");
 819         }
 820         ret = ldb_next_start_trans(module);
 821         if (ret != LDB_SUCCESS) {
 822                 return ret;
 823         }
 824
 825         ret = partition_reload_if_required(module, data, NULL);
 826         if (ret != LDB_SUCCESS) {
 827                 return ret;
 828         }
 829
 830         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 831                 if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
 832                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_start_trans() -> %s",
 833                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
 834                 }
 835                 ret = ldb_next_start_trans(data->partitions[i]->module);
 836                 if (ret != LDB_SUCCESS) {
 837                         /* Back it out, if it fails on one */
 838                         for (i--; i >= 0; i--) {
 839                                 ldb_next_del_trans(data->partitions[i]->module);
 840                         }
 841                         ldb_next_del_trans(module);
 842                         return ret;
 843                 }
 844         }
 845
 846         data->in_transaction++;
 847
 848         return LDB_SUCCESS;
 849 }
 850
 851 /* prepare for a commit */
 852 static int partition_prepare_commit(struct ldb_module *module)
 853 {
 854         unsigned int i;
 855         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 856                                                               struct partition_private_data);
 857
 858         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 859                 int ret;
 860
 861                 if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
 862                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_prepare_commit() -> %s",
 863                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
 864                 }
 865                 ret = ldb_next_prepare_commit(data->partitions[i]->module);
 866                 if (ret != LDB_SUCCESS) {
 867                         ldb_asprintf_errstring(ldb_module_get_ctx(module), "prepare_commit error on %s: %s",
 868                                                ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
 869                                                ldb_errstring(ldb_module_get_ctx(module)));
 870                         return ret;
 871                 }
 872         }
 873
 874         if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
 875                 ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_prepare_commit() -> (metadata partition)");
 876         }
 877         return ldb_next_prepare_commit(module);
 878 }
 879
 880
 881 /* end a transaction */
 882 static int partition_end_trans(struct ldb_module *module)
 883 {
 884         int ret, ret2;
 885         unsigned int i;
 886         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 887                                                               struct partition_private_data);
 888
 889         ret = LDB_SUCCESS;
 890
 891         if (data->in_transaction == 0) {
 892                 DEBUG(0,("partition end transaction mismatch\n"));
 893                 ret = LDB_ERR_OPERATIONS_ERROR;
 894         } else {
 895                 data->in_transaction--;
 896         }
 897
 898         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 899                 if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
 900                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_end_trans() -> %s",
 901                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
 902                 }
 903                 ret2 = ldb_next_end_trans(data->partitions[i]->module);
 904                 if (ret2 != LDB_SUCCESS) {
 905                         ldb_asprintf_errstring(ldb_module_get_ctx(module), "end_trans error on %s: %s",
 906                                                ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
 907                                                ldb_errstring(ldb_module_get_ctx(module)));
 908                         ret = ret2;
 909                 }
 910         }
 911
 912         if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
 913                 ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_end_trans() -> (metadata partition)");
 914         }
 915         ret2 = ldb_next_end_trans(module);
 916         if (ret2 != LDB_SUCCESS) {
 917                 ret = ret2;
 918         }
 919         return ret;
 920 }
 921
 922 /* delete a transaction */
 923 static int partition_del_trans(struct ldb_module *module)
 924 {
 925         int ret, final_ret = LDB_SUCCESS;
 926         unsigned int i;
 927         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
 928                                                               struct partition_private_data);
 929         for (i=0; data && data->partitions && data->partitions[i]; i++) {
 930                 if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
 931                         ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_del_trans() -> %s",
 932                                   ldb_dn_get_linearized(data->partitions[i]->ctrl->dn));
 933                 }
 934                 ret = ldb_next_del_trans(data->partitions[i]->module);
 935                 if (ret != LDB_SUCCESS) {
 936                         ldb_asprintf_errstring(ldb_module_get_ctx(module), "del_trans error on %s: %s",
 937                                                ldb_dn_get_linearized(data->partitions[i]->ctrl->dn),
 938                                                ldb_errstring(ldb_module_get_ctx(module)));
 939                         final_ret = ret;
 940                 }
 941         }
 942
 943         if (data->in_transaction == 0) {
 944                 DEBUG(0,("partition del transaction mismatch\n"));
 945                 return ldb_operr(ldb_module_get_ctx(module));
 946         }
 947         data->in_transaction--;
 948
 949         if ((module && ldb_module_flags(ldb_module_get_ctx(module)) & LDB_FLG_ENABLE_TRACING)) {
 950                 ldb_debug(ldb_module_get_ctx(module), LDB_DEBUG_TRACE, "partition_del_trans() -> (metadata partition)");
 951         }
 952         ret = ldb_next_del_trans(module);
 953         if (ret != LDB_SUCCESS) {
 954                 final_ret = ret;
 955         }
 956         return final_ret;
 957 }
 958
 959 int partition_primary_sequence_number(struct ldb_module *module, TALLOC_CTX *mem_ctx,
 960                                      enum ldb_sequence_type type, uint64_t *seq_number)
 961 {
 962         int ret;
 963         struct ldb_result *res;
 964         struct ldb_seqnum_request *tseq;
 965         struct ldb_request *treq;
 966         struct ldb_seqnum_result *seqr;
 967         res = talloc_zero(mem_ctx, struct ldb_result);
 968         if (res == NULL) {
 969                 return ldb_oom(ldb_module_get_ctx(module));
 970         }
 971         tseq = talloc_zero(res, struct ldb_seqnum_request);
 972         if (tseq == NULL) {
 973                 talloc_free(res);
 974                 return ldb_oom(ldb_module_get_ctx(module));
 975         }
 976         tseq->type = type;
 977
 978         ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
 979                                      LDB_EXTENDED_SEQUENCE_NUMBER,
 980                                      tseq,
 981                                      NULL,
 982                                      res,
 983                                      ldb_extended_default_callback,
 984                                      NULL);
 985         LDB_REQ_SET_LOCATION(treq);
 986         if (ret != LDB_SUCCESS) {
 987                 talloc_free(res);
 988                 return ret;
 989         }
 990
 991         ret = ldb_next_request(module, treq);
 992         if (ret != LDB_SUCCESS) {
 993                 talloc_free(res);
 994                 return ret;
 995         }
 996         ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
 997         if (ret != LDB_SUCCESS) {
 998                 talloc_free(res);
 999                 return ret;
1000         }
1001
1002         seqr = talloc_get_type(res->extended->data,
1003                                struct ldb_seqnum_result);
1004         if (seqr->flags & LDB_SEQ_TIMESTAMP_SEQUENCE) {
1005                 ret = LDB_ERR_OPERATIONS_ERROR;
1006                 ldb_set_errstring(ldb_module_get_ctx(module), "Primary backend in partitions module returned a timestamp based seq number (must return a normal number)");
1007                 talloc_free(res);
1008                 return ret;
1009         } else {
1010                 *seq_number = seqr->seq_num;
1011         }
1012         talloc_free(res);
1013         return LDB_SUCCESS;
1014 }
1015
1016 /* FIXME: This function is still semi-async */
1017 static int partition_sequence_number(struct ldb_module *module, struct ldb_request *req)
1018 {
1019         int ret;
1020         unsigned int i;
1021         uint64_t seq_number = 0;
1022         uint64_t timestamp_sequence = 0;
1023         uint64_t timestamp = 0;
1024         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
1025                                                               struct partition_private_data);
1026         struct ldb_seqnum_request *seq;
1027         struct ldb_seqnum_result *seqr;
1028         struct ldb_request *treq;
1029         struct ldb_seqnum_request *tseq;
1030         struct ldb_seqnum_result *tseqr;
1031         struct ldb_extended *ext;
1032         struct ldb_result *res;
1033         struct dsdb_partition *p;
1034
1035         p = find_partition(data, NULL, req);
1036         if (p != NULL) {
1037                 /* the caller specified what partition they want the
1038                  * sequence number operation on - just pass it on
1039                  */
1040                 return ldb_next_request(p->module, req);
1041         }
1042
1043         seq = talloc_get_type(req->op.extended.data, struct ldb_seqnum_request);
1044
1045         switch (seq->type) {
1046         case LDB_SEQ_NEXT:
1047         case LDB_SEQ_HIGHEST_SEQ:
1048
1049                 ret = partition_primary_sequence_number(module, req, seq->type, &seq_number);
1050                 if (ret != LDB_SUCCESS) {
1051                         return ret;
1052                 }
1053
1054                 /* Skip the lot if 'data' isn't here yet (initialisation) */
1055                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
1056
1057                         res = talloc_zero(req, struct ldb_result);
1058                         if (res == NULL) {
1059                                 return ldb_oom(ldb_module_get_ctx(module));
1060                         }
1061                         tseq = talloc_zero(res, struct ldb_seqnum_request);
1062                         if (tseq == NULL) {
1063                                 talloc_free(res);
1064                                 return ldb_oom(ldb_module_get_ctx(module));
1065                         }
1066                         tseq->type = seq->type;
1067
1068                         ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
1069                                                      LDB_EXTENDED_SEQUENCE_NUMBER,
1070                                                      tseq,
1071                                                      NULL,
1072                                                      res,
1073                                                      ldb_extended_default_callback,
1074                                                      req);
1075                         LDB_REQ_SET_LOCATION(treq);
1076                         if (ret != LDB_SUCCESS) {
1077                                 talloc_free(res);
1078                                 return ret;
1079                         }
1080
1081                         ret = ldb_request_add_control(treq,
1082                                                       DSDB_CONTROL_CURRENT_PARTITION_OID,
1083                                                       false, data->partitions[i]->ctrl);
1084                         if (ret != LDB_SUCCESS) {
1085                                 talloc_free(res);
1086                                 return ret;
1087                         }
1088
1089                         ret = partition_request(data->partitions[i]->module, treq);
1090                         if (ret != LDB_SUCCESS) {
1091                                 talloc_free(res);
1092                                 return ret;
1093                         }
1094                         ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
1095                         if (ret != LDB_SUCCESS) {
1096                                 talloc_free(res);
1097                                 return ret;
1098                         }
1099                         tseqr = talloc_get_type(res->extended->data,
1100                                                 struct ldb_seqnum_result);
1101                         if (tseqr->flags & LDB_SEQ_TIMESTAMP_SEQUENCE) {
1102                                 timestamp_sequence = MAX(timestamp_sequence,
1103                                                          tseqr->seq_num);
1104                         } else {
1105                                 seq_number += tseqr->seq_num;
1106                         }
1107                         talloc_free(res);
1108                 }
1109                 /* fall through */
1110         case LDB_SEQ_HIGHEST_TIMESTAMP:
1111
1112                 res = talloc_zero(req, struct ldb_result);
1113                 if (res == NULL) {
1114                         return ldb_oom(ldb_module_get_ctx(module));
1115                 }
1116
1117                 tseq = talloc_zero(res, struct ldb_seqnum_request);
1118                 if (tseq == NULL) {
1119                         talloc_free(res);
1120                         return ldb_oom(ldb_module_get_ctx(module));
1121                 }
1122                 tseq->type = LDB_SEQ_HIGHEST_TIMESTAMP;
1123
1124                 ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
1125                                              LDB_EXTENDED_SEQUENCE_NUMBER,
1126                                              tseq,
1127                                              NULL,
1128                                              res,
1129                                              ldb_extended_default_callback,
1130                                              req);
1131                 LDB_REQ_SET_LOCATION(treq);
1132                 if (ret != LDB_SUCCESS) {
1133                         talloc_free(res);
1134                         return ret;
1135                 }
1136
1137                 ret = ldb_next_request(module, treq);
1138                 if (ret != LDB_SUCCESS) {
1139                         talloc_free(res);
1140                         return ret;
1141                 }
1142                 ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
1143                 if (ret != LDB_SUCCESS) {
1144                         talloc_free(res);
1145                         return ret;
1146                 }
1147
1148                 tseqr = talloc_get_type(res->extended->data,
1149                                            struct ldb_seqnum_result);
1150                 timestamp = tseqr->seq_num;
1151
1152                 talloc_free(res);
1153
1154                 /* Skip the lot if 'data' isn't here yet (initialisation) */
1155                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
1156
1157                         res = talloc_zero(req, struct ldb_result);
1158                         if (res == NULL) {
1159                                 return ldb_oom(ldb_module_get_ctx(module));
1160                         }
1161
1162                         tseq = talloc_zero(res, struct ldb_seqnum_request);
1163                         if (tseq == NULL) {
1164                                 talloc_free(res);
1165                                 return ldb_oom(ldb_module_get_ctx(module));
1166                         }
1167                         tseq->type = LDB_SEQ_HIGHEST_TIMESTAMP;
1168
1169                         ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
1170                                                      LDB_EXTENDED_SEQUENCE_NUMBER,
1171                                                      tseq,
1172                                                      NULL,
1173                                                      res,
1174                                                      ldb_extended_default_callback,
1175                                                      req);
1176                         LDB_REQ_SET_LOCATION(treq);
1177                         if (ret != LDB_SUCCESS) {
1178                                 talloc_free(res);
1179                                 return ret;
1180                         }
1181
1182                         ret = ldb_request_add_control(treq,
1183                                                       DSDB_CONTROL_CURRENT_PARTITION_OID,
1184                                                       false, data->partitions[i]->ctrl);
1185                         if (ret != LDB_SUCCESS) {
1186                                 talloc_free(res);
1187                                 return ret;
1188                         }
1189
1190                         ret = partition_request(data->partitions[i]->module, treq);
1191                         if (ret != LDB_SUCCESS) {
1192                                 talloc_free(res);
1193                                 return ret;
1194                         }
1195                         ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
1196                         if (ret != LDB_SUCCESS) {
1197                                 talloc_free(res);
1198                                 return ret;
1199                         }
1200
1201                         tseqr = talloc_get_type(res->extended->data,
1202                                                   struct ldb_seqnum_result);
1203                         timestamp = MAX(timestamp, tseqr->seq_num);
1204
1205                         talloc_free(res);
1206                 }
1207
1208                 break;
1209         }
1210
1211         ext = talloc_zero(req, struct ldb_extended);
1212         if (!ext) {
1213                 return ldb_oom(ldb_module_get_ctx(module));
1214         }
1215         seqr = talloc_zero(ext, struct ldb_seqnum_result);
1216         if (seqr == NULL) {
1217                 talloc_free(ext);
1218                 return ldb_oom(ldb_module_get_ctx(module));
1219         }
1220         ext->oid = LDB_EXTENDED_SEQUENCE_NUMBER;
1221         ext->data = seqr;
1222
1223         switch (seq->type) {
1224         case LDB_SEQ_NEXT:
1225         case LDB_SEQ_HIGHEST_SEQ:
1226
1227                 /* Has someone above set a timebase sequence? */
1228                 if (timestamp_sequence) {
1229                         seqr->seq_num = (((unsigned long long)timestamp << 24) | (seq_number & 0xFFFFFF));
1230                 } else {
1231                         seqr->seq_num = seq_number;
1232                 }
1233
1234                 if (timestamp_sequence > seqr->seq_num) {
1235                         seqr->seq_num = timestamp_sequence;
1236                         seqr->flags |= LDB_SEQ_TIMESTAMP_SEQUENCE;
1237                 }
1238
1239                 seqr->flags |= LDB_SEQ_GLOBAL_SEQUENCE;
1240                 break;
1241         case LDB_SEQ_HIGHEST_TIMESTAMP:
1242                 seqr->seq_num = timestamp;
1243                 break;
1244         }
1245
1246         if (seq->type == LDB_SEQ_NEXT) {
1247                 seqr->seq_num++;
1248         }
1249
1250         /* send request done */
1251         return ldb_module_done(req, NULL, ext, LDB_SUCCESS);
1252 }
1253
1254 /* extended */
1255 static int partition_extended(struct ldb_module *module, struct ldb_request *req)
1256 {
1257         struct partition_private_data *data = talloc_get_type(ldb_module_get_private(module),
1258                                                               struct partition_private_data);
1259         struct partition_context *ac;
1260         int ret;
1261
1262         /* if we aren't initialised yet go further */
1263         if (!data) {
1264                 return ldb_next_request(module, req);
1265         }
1266
1267         /* see if we are still up-to-date */
1268         ret = partition_reload_if_required(module, data, req);
1269         if (ret != LDB_SUCCESS) {
1270                 return ret;
1271         }
1272
1273         if (strcmp(req->op.extended.oid, LDB_EXTENDED_SEQUENCE_NUMBER) == 0) {
1274                 return partition_sequence_number(module, req);
1275         }
1276
1277         if (strcmp(req->op.extended.oid, DSDB_EXTENDED_CREATE_PARTITION_OID) == 0) {
1278                 return partition_create(module, req);
1279         }
1280
1281         /*
1282          * as the extended operation has no dn
1283          * we need to send it to all partitions
1284          */
1285
1286         ac = partition_init_ctx(module, req);
1287         if (!ac) {
1288                 return ldb_operr(ldb_module_get_ctx(module));
1289         }
1290
1291         return partition_send_all(module, ac, req);
1292 }
1293
1294 static const struct ldb_module_ops ldb_partition_module_ops = {
1295         .name              = "partition",
1296         .init_context      = partition_init,
1297         .search            = partition_search,
1298         .add               = partition_add,
1299         .modify            = partition_modify,
1300         .del               = partition_delete,
1301         .rename            = partition_rename,
1302         .extended          = partition_extended,
1303         .start_transaction = partition_start_trans,
1304         .prepare_commit    = partition_prepare_commit,
1305         .end_transaction   = partition_end_trans,
1306         .del_transaction   = partition_del_trans,
1307 };
1308
1309 int ldb_partition_module_init(const char *version)
1310 {
1311         LDB_MODULE_CHECK_VERSION(version);
1312         return ldb_register_module(&ldb_partition_module_ops);
1313 }