s4:dsdb Remove partition_extended_schema_update_now
[ira/wip.git] / 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         bool got_success;
43
44         struct part_request *part_req;
45         int num_requests;
46         int finished_requests;
47 };
48
49 static struct partition_context *partition_init_ctx(struct ldb_module *module, struct ldb_request *req)
50 {
51         struct partition_context *ac;
52
53         ac = talloc_zero(req, struct partition_context);
54         if (ac == NULL) {
55                 ldb_set_errstring(ldb_module_get_ctx(module), "Out of Memory");
56                 return NULL;
57         }
58
59         ac->module = module;
60         ac->req = req;
61
62         return ac;
63 }
64
65 /*
66  *    helper functions to call the next module in chain
67  *    */
68
69 int partition_request(struct ldb_module *module, struct ldb_request *request)
70 {
71         int ret;
72         switch (request->operation) {
73         case LDB_SEARCH:
74                 PARTITION_FIND_OP(module, search);
75                 ret = module->ops->search(module, request);
76                 break;
77         case LDB_ADD:
78                 PARTITION_FIND_OP(module, add);
79                 ret = module->ops->add(module, request);
80                 break;
81         case LDB_MODIFY:
82                 PARTITION_FIND_OP(module, modify);
83                 ret = module->ops->modify(module, request);
84                 break;
85         case LDB_DELETE:
86                 PARTITION_FIND_OP(module, del);
87                 ret = module->ops->del(module, request);
88                 break;
89         case LDB_RENAME:
90                 PARTITION_FIND_OP(module, rename);
91                 ret = module->ops->rename(module, request);
92                 break;
93         case LDB_EXTENDED:
94                 PARTITION_FIND_OP(module, extended);
95                 ret = module->ops->extended(module, request);
96                 break;
97         default:
98                 PARTITION_FIND_OP(module, request);
99                 ret = module->ops->request(module, request);
100                 break;
101         }
102         if (ret == LDB_SUCCESS) {
103                 return ret;
104         }
105         if (!ldb_errstring(ldb_module_get_ctx(module))) {
106                 /* Set a default error string, to place the blame somewhere */
107                 ldb_asprintf_errstring(ldb_module_get_ctx(module),
108                                         "error in module %s: %s (%d)",
109                                         module->ops->name,
110                                         ldb_strerror(ret), ret);
111         }
112         return ret;
113 }
114
115 static struct dsdb_partition *find_partition(struct partition_private_data *data,
116                                              struct ldb_dn *dn,
117                                              struct ldb_request *req)
118 {
119         int i;
120         struct ldb_control *partition_ctrl;
121
122         /* see if the request has the partition DN specified in a
123          * control. The repl_meta_data module can specify this to
124          * ensure that replication happens to the right partition
125          */
126         partition_ctrl = ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID);
127         if (partition_ctrl) {
128                 const struct dsdb_control_current_partition *partition;
129                 partition = talloc_get_type(partition_ctrl->data,
130                                             struct dsdb_control_current_partition);
131                 if (partition != NULL) {
132                         dn = partition->dn;
133                 }
134         }
135
136         if (dn == NULL) {
137                 return NULL;
138         }
139
140         /* Look at base DN */
141         /* Figure out which partition it is under */
142         /* Skip the lot if 'data' isn't here yet (initialisation) */
143         for (i=0; data && data->partitions && data->partitions[i]; i++) {
144                 if (ldb_dn_compare_base(data->partitions[i]->ctrl->dn, dn) == 0) {
145                         return data->partitions[i];
146                 }
147         }
148
149         return NULL;
150 }
151
152 /**
153  * fire the caller's callback for every entry, but only send 'done' once.
154  */
155 static int partition_req_callback(struct ldb_request *req,
156                                   struct ldb_reply *ares)
157 {
158         struct partition_context *ac;
159         struct ldb_module *module;
160         struct ldb_request *nreq;
161         int ret;
162         struct partition_private_data *data;
163         struct ldb_control *partition_ctrl;
164
165         ac = talloc_get_type(req->context, struct partition_context);
166         data = talloc_get_type(ac->module->private_data, struct partition_private_data);
167
168         if (!ares) {
169                 return ldb_module_done(ac->req, NULL, NULL,
170                                         LDB_ERR_OPERATIONS_ERROR);
171         }
172
173         partition_ctrl = ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID);
174         if (partition_ctrl && (ac->num_requests == 1 || ares->type == LDB_REPLY_ENTRY)) {
175                 /* If we didn't fan this request out to mulitple partitions,
176                  * or this is an individual search result, we can
177                  * deterministily tell the caller what partition this was
178                  * written to (repl_meta_data likes to know) */
179                         ret = ldb_reply_add_control(ares,
180                                                     DSDB_CONTROL_CURRENT_PARTITION_OID,
181                                                     false, partition_ctrl->data);
182                         if (ret != LDB_SUCCESS) {
183                                 return ldb_module_done(ac->req, NULL, NULL,
184                                                        ret);
185                         }
186         }
187
188         if (ares->error != LDB_SUCCESS && !ac->got_success) {
189                 return ldb_module_done(ac->req, ares->controls,
190                                         ares->response, ares->error);
191         }
192
193         switch (ares->type) {
194         case LDB_REPLY_REFERRAL:
195                 /* ignore referrals for now */
196                 break;
197
198         case LDB_REPLY_ENTRY:
199                 if (ac->req->operation != LDB_SEARCH) {
200                         ldb_set_errstring(ldb_module_get_ctx(ac->module),
201                                 "partition_req_callback:"
202                                 " Unsupported reply type for this request");
203                         return ldb_module_done(ac->req, NULL, NULL,
204                                                 LDB_ERR_OPERATIONS_ERROR);
205                 }
206                 
207                 return ldb_module_send_entry(ac->req, ares->message, ares->controls);
208
209         case LDB_REPLY_DONE:
210                 if (ares->error == LDB_SUCCESS) {
211                         ac->got_success = true;
212                 }
213                 if (ac->req->operation == LDB_EXTENDED) {
214                         /* FIXME: check for ares->response, replmd does not fill it ! */
215                         if (ares->response) {
216                                 if (strcmp(ares->response->oid, LDB_EXTENDED_START_TLS_OID) != 0) {
217                                         ldb_set_errstring(ldb_module_get_ctx(ac->module),
218                                                           "partition_req_callback:"
219                                                           " Unknown extended reply, "
220                                                           "only supports START_TLS");
221                                         talloc_free(ares);
222                                         return ldb_module_done(ac->req, NULL, NULL,
223                                                                 LDB_ERR_OPERATIONS_ERROR);
224                                 }
225                         }
226                 }
227
228                 ac->finished_requests++;
229                 if (ac->finished_requests == ac->num_requests) {
230                         /* this was the last one, call callback */
231                         return ldb_module_done(ac->req, ares->controls,
232                                                ares->response, 
233                                                ac->got_success?LDB_SUCCESS:ares->error);
234                 }
235
236                 /* not the last, now call the next one */
237                 module = ac->part_req[ac->finished_requests].module;
238                 nreq = ac->part_req[ac->finished_requests].req;
239
240                 ret = partition_request(module, nreq);
241                 if (ret != LDB_SUCCESS) {
242                         talloc_free(ares);
243                         return ldb_module_done(ac->req, NULL, NULL, ret);
244                 }
245
246                 break;
247         }
248
249         talloc_free(ares);
250         return LDB_SUCCESS;
251 }
252
253 static int partition_prep_request(struct partition_context *ac,
254                                   struct dsdb_partition *partition)
255 {
256         int ret;
257         struct ldb_request *req;
258
259         ac->part_req = talloc_realloc(ac, ac->part_req,
260                                         struct part_request,
261                                         ac->num_requests + 1);
262         if (ac->part_req == NULL) {
263                 ldb_oom(ldb_module_get_ctx(ac->module));
264                 return LDB_ERR_OPERATIONS_ERROR;
265         }
266
267         switch (ac->req->operation) {
268         case LDB_SEARCH:
269                 ret = ldb_build_search_req_ex(&req, ldb_module_get_ctx(ac->module),
270                                         ac->part_req,
271                                         ac->req->op.search.base,
272                                         ac->req->op.search.scope,
273                                         ac->req->op.search.tree,
274                                         ac->req->op.search.attrs,
275                                         ac->req->controls,
276                                         ac, partition_req_callback,
277                                         ac->req);
278                 break;
279         case LDB_ADD:
280                 ret = ldb_build_add_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
281                                         ac->req->op.add.message,
282                                         ac->req->controls,
283                                         ac, partition_req_callback,
284                                         ac->req);
285                 break;
286         case LDB_MODIFY:
287                 ret = ldb_build_mod_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
288                                         ac->req->op.mod.message,
289                                         ac->req->controls,
290                                         ac, partition_req_callback,
291                                         ac->req);
292                 break;
293         case LDB_DELETE:
294                 ret = ldb_build_del_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
295                                         ac->req->op.del.dn,
296                                         ac->req->controls,
297                                         ac, partition_req_callback,
298                                         ac->req);
299                 break;
300         case LDB_RENAME:
301                 ret = ldb_build_rename_req(&req, ldb_module_get_ctx(ac->module), ac->part_req,
302                                         ac->req->op.rename.olddn,
303                                         ac->req->op.rename.newdn,
304                                         ac->req->controls,
305                                         ac, partition_req_callback,
306                                         ac->req);
307                 break;
308         case LDB_EXTENDED:
309                 ret = ldb_build_extended_req(&req, ldb_module_get_ctx(ac->module),
310                                         ac->part_req,
311                                         ac->req->op.extended.oid,
312                                         ac->req->op.extended.data,
313                                         ac->req->controls,
314                                         ac, partition_req_callback,
315                                         ac->req);
316                 break;
317         default:
318                 ldb_set_errstring(ldb_module_get_ctx(ac->module),
319                                   "Unsupported request type!");
320                 ret = LDB_ERR_UNWILLING_TO_PERFORM;
321         }
322
323         if (ret != LDB_SUCCESS) {
324                 return ret;
325         }
326
327         ac->part_req[ac->num_requests].req = req;
328
329         if (ac->req->controls) {
330                 req->controls = talloc_memdup(req, ac->req->controls,
331                                         talloc_get_size(ac->req->controls));
332                 if (req->controls == NULL) {
333                         ldb_oom(ldb_module_get_ctx(ac->module));
334                         return LDB_ERR_OPERATIONS_ERROR;
335                 }
336         }
337
338         if (partition) {
339                 ac->part_req[ac->num_requests].module = partition->module;
340
341                 if (!ldb_request_get_control(req, DSDB_CONTROL_CURRENT_PARTITION_OID)) {
342                         ret = ldb_request_add_control(req,
343                                                       DSDB_CONTROL_CURRENT_PARTITION_OID,
344                                                       false, partition->ctrl);
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 NEXT module here, or
362                  * partition_request() will simply loop forever on itself */
363                 ac->part_req[ac->num_requests].module = ac->module->next;
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         int i;
384         struct partition_private_data *data = talloc_get_type(module->private_data, 
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  * Figure out which backend a request needs to be aimed at.  Some
403  * requests must be replicated to all backends
404  */
405 static int partition_replicate(struct ldb_module *module, struct ldb_request *req, struct ldb_dn *dn) 
406 {
407         struct partition_context *ac;
408         unsigned i;
409         int ret;
410         struct dsdb_partition *partition;
411         struct partition_private_data *data = talloc_get_type(module->private_data, 
412                                                               struct partition_private_data);
413         if (!data || !data->partitions) {
414                 return ldb_next_request(module, req);
415         }
416
417         if (req->operation != LDB_SEARCH) {
418                 /* Is this a special DN, we need to replicate to every backend? */
419                 for (i=0; data->replicate && data->replicate[i]; i++) {
420                         if (ldb_dn_compare(data->replicate[i], 
421                                            dn) == 0) {
422                                 
423                                 ac = partition_init_ctx(module, req);
424                                 if (!ac) {
425                                         return LDB_ERR_OPERATIONS_ERROR;
426                                 }
427                                 
428                                 return partition_send_all(module, ac, req);
429                         }
430                 }
431         }
432
433         /* Otherwise, we need to find the partition to fire it to */
434
435         /* Find partition */
436         partition = find_partition(data, dn, req);
437         if (!partition) {
438                 /*
439                  * if we haven't found a matching partition
440                  * pass the request to the main ldb
441                  *
442                  * TODO: we should maybe return an error here
443                  *       if it's not a special dn
444                  */
445
446                 return ldb_next_request(module, req);
447         }
448
449         ac = partition_init_ctx(module, req);
450         if (!ac) {
451                 return LDB_ERR_OPERATIONS_ERROR;
452         }
453
454         /* we need to add a control but we never touch the original request */
455         ret = partition_prep_request(ac, partition);
456         if (ret != LDB_SUCCESS) {
457                 return ret;
458         }
459
460         /* fire the first one */
461         return partition_call_first(ac);
462 }
463
464 /* search */
465 static int partition_search(struct ldb_module *module, struct ldb_request *req)
466 {
467         int ret;
468         struct ldb_control **saved_controls;
469         /* Find backend */
470         struct partition_private_data *data = talloc_get_type(module->private_data, 
471                                                               struct partition_private_data);
472
473         /* issue request */
474
475         /* (later) consider if we should be searching multiple
476          * partitions (for 'invisible' partition behaviour */
477
478         struct ldb_control *search_control = ldb_request_get_control(req, LDB_CONTROL_SEARCH_OPTIONS_OID);
479         struct ldb_control *domain_scope_control = ldb_request_get_control(req, LDB_CONTROL_DOMAIN_SCOPE_OID);
480         
481         struct ldb_search_options_control *search_options = NULL;
482         struct dsdb_partition *p;
483         
484         ret = partition_reload_if_required(module, data);
485         if (ret != LDB_SUCCESS) {
486                 return ret;
487         }
488
489         p = find_partition(data, NULL, req);
490         if (p != NULL) {
491                 /* the caller specified what partition they want the
492                  * search - just pass it on
493                  */
494                 return ldb_next_request(p->module, req);                
495         }
496
497
498         if (search_control) {
499                 search_options = talloc_get_type(search_control->data, struct ldb_search_options_control);
500         }
501
502         /* Remove the domain_scope control, so we don't confuse a backend server */
503         if (domain_scope_control && !save_controls(domain_scope_control, req, &saved_controls)) {
504                 ldb_oom(ldb_module_get_ctx(module));
505                 return LDB_ERR_OPERATIONS_ERROR;
506         }
507
508         /*
509          * for now pass down the LDB_CONTROL_SEARCH_OPTIONS_OID control
510          * down as uncritical to make windows 2008 dcpromo happy.
511          */
512         if (search_control) {
513                 search_control->critical = 0;
514         }
515
516         /* TODO:
517            Generate referrals (look for a partition under this DN) if we don't have the above control specified
518         */
519         
520         if (search_options && (search_options->search_options & LDB_SEARCH_OPTION_PHANTOM_ROOT)) {
521                 int i;
522                 struct partition_context *ac;
523                 if ((search_options->search_options & ~LDB_SEARCH_OPTION_PHANTOM_ROOT) == 0) {
524                         /* We have processed this flag, so we are done with this control now */
525
526                         /* Remove search control, so we don't confuse a backend server */
527                         if (search_control && !save_controls(search_control, req, &saved_controls)) {
528                                 ldb_oom(ldb_module_get_ctx(module));
529                                 return LDB_ERR_OPERATIONS_ERROR;
530                         }
531                 }
532                 ac = partition_init_ctx(module, req);
533                 if (!ac) {
534                         return LDB_ERR_OPERATIONS_ERROR;
535                 }
536
537                 /* Search from the base DN */
538                 if (!req->op.search.base || ldb_dn_is_null(req->op.search.base)) {
539                         return partition_send_all(module, ac, req);
540                 }
541                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
542                         bool match = false, stop = false;
543                         /* Find all partitions under the search base 
544                            
545                            we match if:
546
547                               1) the DN we are looking for exactly matches the partition
548                              or
549                               2) the DN we are looking for is a parent of the partition and it isn't
550                                  a scope base search
551                              or
552                               3) the DN we are looking for is a child of the partition
553                          */
554                         if (ldb_dn_compare(data->partitions[i]->ctrl->dn, req->op.search.base) == 0) {
555                                 match = true;
556                                 if (req->op.search.scope == LDB_SCOPE_BASE) {
557                                         stop = true;
558                                 }
559                         }
560                         if (!match && 
561                             (ldb_dn_compare_base(req->op.search.base, data->partitions[i]->ctrl->dn) == 0 &&
562                              req->op.search.scope != LDB_SCOPE_BASE)) {
563                                 match = true;
564                         }
565                         if (!match &&
566                             ldb_dn_compare_base(data->partitions[i]->ctrl->dn, req->op.search.base) == 0) {
567                                 match = true;
568                                 stop = true; /* note that this relies on partition ordering */
569                         }
570                         if (match) {
571                                 ret = partition_prep_request(ac, data->partitions[i]);
572                                 if (ret != LDB_SUCCESS) {
573                                         return ret;
574                                 }
575                         }
576                         if (stop) break;
577                 }
578
579                 /* Perhaps we didn't match any partitions.  Try the main partition, only */
580                 if (ac->num_requests == 0) {
581                         talloc_free(ac);
582                         return ldb_next_request(module, req);
583                 }
584
585                 /* fire the first one */
586                 return partition_call_first(ac);
587
588         } else {
589                 /* Handle this like all other requests */
590                 if (search_control && (search_options->search_options & ~LDB_SEARCH_OPTION_PHANTOM_ROOT) == 0) {
591                         /* We have processed this flag, so we are done with this control now */
592
593                         /* Remove search control, so we don't confuse a backend server */
594                         if (search_control && !save_controls(search_control, req, &saved_controls)) {
595                                 ldb_oom(ldb_module_get_ctx(module));
596                                 return LDB_ERR_OPERATIONS_ERROR;
597                         }
598                 }
599
600                 return partition_replicate(module, req, req->op.search.base);
601         }
602 }
603
604 /* add */
605 static int partition_add(struct ldb_module *module, struct ldb_request *req)
606 {
607         return partition_replicate(module, req, req->op.add.message->dn);
608 }
609
610 /* modify */
611 static int partition_modify(struct ldb_module *module, struct ldb_request *req)
612 {
613         return partition_replicate(module, req, req->op.mod.message->dn);
614 }
615
616 /* delete */
617 static int partition_delete(struct ldb_module *module, struct ldb_request *req)
618 {
619         return partition_replicate(module, req, req->op.del.dn);
620 }
621
622 /* rename */
623 static int partition_rename(struct ldb_module *module, struct ldb_request *req)
624 {
625         /* Find backend */
626         struct dsdb_partition *backend, *backend2;
627         
628         struct partition_private_data *data = talloc_get_type(module->private_data, 
629                                                               struct partition_private_data);
630
631         /* Skip the lot if 'data' isn't here yet (initialisation) */
632         if (!data) {
633                 return LDB_ERR_OPERATIONS_ERROR;
634         }
635
636         backend = find_partition(data, req->op.rename.olddn, req);
637         backend2 = find_partition(data, req->op.rename.newdn, req);
638
639         if ((backend && !backend2) || (!backend && backend2)) {
640                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
641         }
642
643         if (backend != backend2) {
644                 ldb_asprintf_errstring(ldb_module_get_ctx(module), 
645                                        "Cannot rename from %s in %s to %s in %s: %s",
646                                        ldb_dn_get_linearized(req->op.rename.olddn),
647                                        ldb_dn_get_linearized(backend->ctrl->dn),
648                                        ldb_dn_get_linearized(req->op.rename.newdn),
649                                        ldb_dn_get_linearized(backend2->ctrl->dn),
650                                        ldb_strerror(LDB_ERR_AFFECTS_MULTIPLE_DSAS));
651                 return LDB_ERR_AFFECTS_MULTIPLE_DSAS;
652         }
653
654         return partition_replicate(module, req, req->op.rename.olddn);
655 }
656
657 /* start a transaction */
658 static int partition_start_trans(struct ldb_module *module)
659 {
660         int i, ret;
661         struct partition_private_data *data = talloc_get_type(module->private_data, 
662                                                               struct partition_private_data);
663         /* Look at base DN */
664         /* Figure out which partition it is under */
665         /* Skip the lot if 'data' isn't here yet (initialization) */
666         ret = ldb_next_start_trans(module);
667         if (ret != LDB_SUCCESS) {
668                 return ret;
669         }
670
671         ret = partition_reload_if_required(module, data);
672         if (ret != LDB_SUCCESS) {
673                 return ret;
674         }
675
676         for (i=0; data && data->partitions && data->partitions[i]; i++) {
677                 struct ldb_module *next = data->partitions[i]->module;
678                 PARTITION_FIND_OP(next, start_transaction);
679
680                 ret = next->ops->start_transaction(next);
681                 if (ret != LDB_SUCCESS) {
682                         /* Back it out, if it fails on one */
683                         for (i--; i >= 0; i--) {
684                                 next = data->partitions[i]->module;
685                                 PARTITION_FIND_OP(next, del_transaction);
686
687                                 next->ops->del_transaction(next);
688                         }
689                         ldb_next_del_trans(module);
690                         return ret;
691                 }
692         }
693
694         data->in_transaction++;
695
696         return LDB_SUCCESS;
697 }
698
699 /* prepare for a commit */
700 static int partition_prepare_commit(struct ldb_module *module)
701 {
702         int i;
703         struct partition_private_data *data = talloc_get_type(module->private_data, 
704                                                               struct partition_private_data);
705
706         for (i=0; data && data->partitions && data->partitions[i]; i++) {
707                 struct ldb_module *next_prepare = data->partitions[i]->module;
708                 int ret;
709
710                 PARTITION_FIND_OP_NOERROR(next_prepare, prepare_commit);
711                 if (next_prepare == NULL) {
712                         continue;
713                 }
714
715                 ret = next_prepare->ops->prepare_commit(next_prepare);
716                 if (ret != LDB_SUCCESS) {
717                         return ret;
718                 }
719         }
720
721         return ldb_next_prepare_commit(module);
722 }
723
724
725 /* end a transaction */
726 static int partition_end_trans(struct ldb_module *module)
727 {
728         int i;
729         struct partition_private_data *data = talloc_get_type(module->private_data, 
730                                                               struct partition_private_data);
731         for (i=0; data && data->partitions && data->partitions[i]; i++) {
732                 struct ldb_module *next_end = data->partitions[i]->module;
733                 int ret;
734
735                 PARTITION_FIND_OP(next_end, end_transaction);
736
737                 ret = next_end->ops->end_transaction(next_end);
738                 if (ret != LDB_SUCCESS) {
739                         return ret;
740                 }
741         }
742
743         if (data->in_transaction == 0) {
744                 DEBUG(0,("partition end transaction mismatch\n"));
745                 return LDB_ERR_OPERATIONS_ERROR;
746         }
747         data->in_transaction--;
748
749         return ldb_next_end_trans(module);
750 }
751
752 /* delete a transaction */
753 static int partition_del_trans(struct ldb_module *module)
754 {
755         int i, ret, final_ret = LDB_SUCCESS;
756         struct partition_private_data *data = talloc_get_type(module->private_data, 
757                                                               struct partition_private_data);
758         for (i=0; data && data->partitions && data->partitions[i]; i++) {
759                 struct ldb_module *next = data->partitions[i]->module;
760                 PARTITION_FIND_OP(next, del_transaction);
761
762                 ret = next->ops->del_transaction(next);
763                 if (ret != LDB_SUCCESS) {
764                         final_ret = ret;
765                 }
766         }       
767
768         if (data->in_transaction == 0) {
769                 DEBUG(0,("partition del transaction mismatch\n"));
770                 return LDB_ERR_OPERATIONS_ERROR;
771         }
772         data->in_transaction--;
773
774         ret = ldb_next_del_trans(module);
775         if (ret != LDB_SUCCESS) {
776                 final_ret = ret;
777         }
778         return final_ret;
779 }
780
781 int partition_primary_sequence_number(struct ldb_module *module, TALLOC_CTX *mem_ctx, 
782                                      enum ldb_sequence_type type, uint64_t *seq_number) 
783 {
784         int ret;
785         struct ldb_result *res;
786         struct ldb_seqnum_request *tseq;
787         struct ldb_request *treq;
788         struct ldb_seqnum_result *seqr;
789         res = talloc_zero(mem_ctx, struct ldb_result);
790         if (res == NULL) {
791                 return LDB_ERR_OPERATIONS_ERROR;
792         }
793         tseq = talloc_zero(res, struct ldb_seqnum_request);
794         if (tseq == NULL) {
795                 talloc_free(res);
796                 return LDB_ERR_OPERATIONS_ERROR;
797         }
798         tseq->type = type;
799         
800         ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
801                                      LDB_EXTENDED_SEQUENCE_NUMBER,
802                                      tseq,
803                                      NULL,
804                                      res,
805                                      ldb_extended_default_callback,
806                                      NULL);
807         if (ret != LDB_SUCCESS) {
808                 talloc_free(res);
809                 return ret;
810         }
811         
812         ret = ldb_next_request(module, treq);
813         if (ret != LDB_SUCCESS) {
814                 talloc_free(res);
815                 return ret;
816         }
817         ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
818         if (ret != LDB_SUCCESS) {
819                 talloc_free(res);
820                 return ret;
821         }
822         
823         seqr = talloc_get_type(res->extended->data,
824                                struct ldb_seqnum_result);
825         if (seqr->flags & LDB_SEQ_TIMESTAMP_SEQUENCE) {
826                 ret = LDB_ERR_OPERATIONS_ERROR;
827                 ldb_set_errstring(ldb_module_get_ctx(module), "Primary backend in partitions module returned a timestamp based seq number (must return a normal number)");
828                 talloc_free(res);
829                 return ret;
830         } else {
831                 *seq_number = seqr->seq_num;
832         }
833         talloc_free(res);
834         return LDB_SUCCESS;
835 }
836
837 /* FIXME: This function is still semi-async */
838 static int partition_sequence_number(struct ldb_module *module, struct ldb_request *req)
839 {
840         int i, ret;
841         uint64_t seq_number = 0;
842         uint64_t timestamp_sequence = 0;
843         uint64_t timestamp = 0;
844         struct partition_private_data *data = talloc_get_type(module->private_data, 
845                                                               struct partition_private_data);
846         struct ldb_seqnum_request *seq;
847         struct ldb_seqnum_result *seqr;
848         struct ldb_request *treq;
849         struct ldb_seqnum_request *tseq;
850         struct ldb_seqnum_result *tseqr;
851         struct ldb_extended *ext;
852         struct ldb_result *res;
853         struct dsdb_partition *p;
854
855         p = find_partition(data, NULL, req);
856         if (p != NULL) {
857                 /* the caller specified what partition they want the
858                  * sequence number operation on - just pass it on
859                  */
860                 return ldb_next_request(p->module, req);                
861         }
862
863         seq = talloc_get_type(req->op.extended.data, struct ldb_seqnum_request);
864
865         switch (seq->type) {
866         case LDB_SEQ_NEXT:
867         case LDB_SEQ_HIGHEST_SEQ:
868
869                 ret = partition_primary_sequence_number(module, req, seq->type, &seq_number);
870                 if (ret != LDB_SUCCESS) {
871                         return ret;
872                 }
873
874                 /* Skip the lot if 'data' isn't here yet (initialisation) */
875                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
876
877                         res = talloc_zero(req, struct ldb_result);
878                         if (res == NULL) {
879                                 return LDB_ERR_OPERATIONS_ERROR;
880                         }
881                         tseq = talloc_zero(res, struct ldb_seqnum_request);
882                         if (tseq == NULL) {
883                                 talloc_free(res);
884                                 return LDB_ERR_OPERATIONS_ERROR;
885                         }
886                         tseq->type = seq->type;
887
888                         ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
889                                                      LDB_EXTENDED_SEQUENCE_NUMBER,
890                                                      tseq,
891                                                      NULL,
892                                                      res,
893                                                      ldb_extended_default_callback,
894                                                      NULL);
895                         if (ret != LDB_SUCCESS) {
896                                 talloc_free(res);
897                                 return ret;
898                         }
899
900                         if (!ldb_request_get_control(treq, DSDB_CONTROL_CURRENT_PARTITION_OID)) {
901                                 ret = ldb_request_add_control(treq,
902                                                               DSDB_CONTROL_CURRENT_PARTITION_OID,
903                                                               false, data->partitions[i]->ctrl);
904                                 if (ret != LDB_SUCCESS) {
905                                         talloc_free(res);
906                                         return ret;
907                                 }
908                         }
909
910                         ret = partition_request(data->partitions[i]->module, treq);
911                         if (ret != LDB_SUCCESS) {
912                                 talloc_free(res);
913                                 return ret;
914                         }
915                         ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
916                         if (ret != LDB_SUCCESS) {
917                                 talloc_free(res);
918                                 return ret;
919                         }
920                         tseqr = talloc_get_type(res->extended->data,
921                                                 struct ldb_seqnum_result);
922                         if (tseqr->flags & LDB_SEQ_TIMESTAMP_SEQUENCE) {
923                                 timestamp_sequence = MAX(timestamp_sequence,
924                                                          tseqr->seq_num);
925                         } else {
926                                 seq_number += tseqr->seq_num;
927                         }
928                         talloc_free(res);
929                 }
930                 /* fall through */
931         case LDB_SEQ_HIGHEST_TIMESTAMP:
932
933                 res = talloc_zero(req, struct ldb_result);
934                 if (res == NULL) {
935                         return LDB_ERR_OPERATIONS_ERROR;
936                 }
937
938                 tseq = talloc_zero(res, struct ldb_seqnum_request);
939                 if (tseq == NULL) {
940                         talloc_free(res);
941                         return LDB_ERR_OPERATIONS_ERROR;
942                 }
943                 tseq->type = LDB_SEQ_HIGHEST_TIMESTAMP;
944
945                 ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
946                                              LDB_EXTENDED_SEQUENCE_NUMBER,
947                                              tseq,
948                                              NULL,
949                                              res,
950                                              ldb_extended_default_callback,
951                                              NULL);
952                 if (ret != LDB_SUCCESS) {
953                         talloc_free(res);
954                         return ret;
955                 }
956
957                 ret = ldb_next_request(module, treq);
958                 if (ret != LDB_SUCCESS) {
959                         talloc_free(res);
960                         return ret;
961                 }
962                 ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
963                 if (ret != LDB_SUCCESS) {
964                         talloc_free(res);
965                         return ret;
966                 }
967
968                 tseqr = talloc_get_type(res->extended->data,
969                                            struct ldb_seqnum_result);
970                 timestamp = tseqr->seq_num;
971
972                 talloc_free(res);
973
974                 /* Skip the lot if 'data' isn't here yet (initialisation) */
975                 for (i=0; data && data->partitions && data->partitions[i]; i++) {
976
977                         res = talloc_zero(req, struct ldb_result);
978                         if (res == NULL) {
979                                 return LDB_ERR_OPERATIONS_ERROR;
980                         }
981
982                         tseq = talloc_zero(res, struct ldb_seqnum_request);
983                         if (tseq == NULL) {
984                                 talloc_free(res);
985                                 return LDB_ERR_OPERATIONS_ERROR;
986                         }
987                         tseq->type = LDB_SEQ_HIGHEST_TIMESTAMP;
988
989                         ret = ldb_build_extended_req(&treq, ldb_module_get_ctx(module), res,
990                                                      LDB_EXTENDED_SEQUENCE_NUMBER,
991                                                      tseq,
992                                                      NULL,
993                                                      res,
994                                                      ldb_extended_default_callback,
995                                                      NULL);
996                         if (ret != LDB_SUCCESS) {
997                                 talloc_free(res);
998                                 return ret;
999                         }
1000
1001                         if (!ldb_request_get_control(treq, DSDB_CONTROL_CURRENT_PARTITION_OID)) {
1002                                 ret = ldb_request_add_control(treq,
1003                                                               DSDB_CONTROL_CURRENT_PARTITION_OID,
1004                                                               false, data->partitions[i]->ctrl);
1005                                 if (ret != LDB_SUCCESS) {
1006                                         talloc_free(res);
1007                                         return ret;
1008                                 }
1009                         }
1010
1011                         ret = partition_request(data->partitions[i]->module, treq);
1012                         if (ret != LDB_SUCCESS) {
1013                                 talloc_free(res);
1014                                 return ret;
1015                         }
1016                         ret = ldb_wait(treq->handle, LDB_WAIT_ALL);
1017                         if (ret != LDB_SUCCESS) {
1018                                 talloc_free(res);
1019                                 return ret;
1020                         }
1021
1022                         tseqr = talloc_get_type(res->extended->data,
1023                                                   struct ldb_seqnum_result);
1024                         timestamp = MAX(timestamp, tseqr->seq_num);
1025
1026                         talloc_free(res);
1027                 }
1028
1029                 break;
1030         }
1031
1032         ext = talloc_zero(req, struct ldb_extended);
1033         if (!ext) {
1034                 return LDB_ERR_OPERATIONS_ERROR;
1035         }
1036         seqr = talloc_zero(ext, struct ldb_seqnum_result);
1037         if (seqr == NULL) {
1038                 talloc_free(ext);
1039                 return LDB_ERR_OPERATIONS_ERROR;
1040         }
1041         ext->oid = LDB_EXTENDED_SEQUENCE_NUMBER;
1042         ext->data = seqr;
1043
1044         switch (seq->type) {
1045         case LDB_SEQ_NEXT:
1046         case LDB_SEQ_HIGHEST_SEQ:
1047
1048                 /* Has someone above set a timebase sequence? */
1049                 if (timestamp_sequence) {
1050                         seqr->seq_num = (((unsigned long long)timestamp << 24) | (seq_number & 0xFFFFFF));
1051                 } else {
1052                         seqr->seq_num = seq_number;
1053                 }
1054
1055                 if (timestamp_sequence > seqr->seq_num) {
1056                         seqr->seq_num = timestamp_sequence;
1057                         seqr->flags |= LDB_SEQ_TIMESTAMP_SEQUENCE;
1058                 }
1059
1060                 seqr->flags |= LDB_SEQ_GLOBAL_SEQUENCE;
1061                 break;
1062         case LDB_SEQ_HIGHEST_TIMESTAMP:
1063                 seqr->seq_num = timestamp;
1064                 break;
1065         }
1066
1067         if (seq->type == LDB_SEQ_NEXT) {
1068                 seqr->seq_num++;
1069         }
1070
1071         /* send request done */
1072         return ldb_module_done(req, NULL, ext, LDB_SUCCESS);
1073 }
1074
1075 /* extended */
1076 static int partition_extended(struct ldb_module *module, struct ldb_request *req)
1077 {
1078         struct partition_private_data *data;
1079         struct partition_context *ac;
1080         int ret;
1081
1082         data = talloc_get_type(module->private_data, struct partition_private_data);
1083         if (!data) {
1084                 return ldb_next_request(module, req);
1085         }
1086
1087         ret = partition_reload_if_required(module, data);
1088         if (ret != LDB_SUCCESS) {
1089                 return ret;
1090         }
1091         
1092         if (strcmp(req->op.extended.oid, LDB_EXTENDED_SEQUENCE_NUMBER) == 0) {
1093                 return partition_sequence_number(module, req);
1094         }
1095
1096         if (strcmp(req->op.extended.oid, DSDB_EXTENDED_CREATE_PARTITION_OID) == 0) {
1097                 return partition_create(module, req);
1098         }
1099
1100         /* 
1101          * as the extended operation has no dn
1102          * we need to send it to all partitions
1103          */
1104
1105         ac = partition_init_ctx(module, req);
1106         if (!ac) {
1107                 return LDB_ERR_OPERATIONS_ERROR;
1108         }
1109
1110         return partition_send_all(module, ac, req);
1111 }
1112
1113 _PUBLIC_ const struct ldb_module_ops ldb_partition_module_ops = {
1114         .name              = "partition",
1115         .init_context      = partition_init,
1116         .search            = partition_search,
1117         .add               = partition_add,
1118         .modify            = partition_modify,
1119         .del               = partition_delete,
1120         .rename            = partition_rename,
1121         .extended          = partition_extended,
1122         .start_transaction = partition_start_trans,
1123         .prepare_commit    = partition_prepare_commit,
1124         .end_transaction   = partition_end_trans,
1125         .del_transaction   = partition_del_trans,
1126 };