s3:passdb: Remove unused function secrets_fetch_trust_account_password()
[samba.git] / source3 / lib / dbwrap / dbwrap_ctdb.c
1 /*
2    Unix SMB/CIFS implementation.
3    Database interface wrapper around ctdbd
4    Copyright (C) Volker Lendecke 2007-2009
5    Copyright (C) Michael Adam 2009
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 #include "includes.h"
22 #include "system/filesys.h"
23 #include "lib/tdb_wrap/tdb_wrap.h"
24 #include "util_tdb.h"
25 #include "dbwrap/dbwrap.h"
26 #include "dbwrap/dbwrap_ctdb.h"
27 #include "dbwrap/dbwrap_rbt.h"
28 #include "lib/param/param.h"
29
30 #include "ctdb/include/ctdb_protocol.h"
31 #include "ctdbd_conn.h"
32 #include "dbwrap/dbwrap.h"
33 #include "dbwrap/dbwrap_private.h"
34 #include "dbwrap/dbwrap_ctdb.h"
35 #include "g_lock.h"
36 #include "messages.h"
37 #include "messages_ctdb.h"
38 #include "lib/cluster_support.h"
39 #include "lib/util/tevent_ntstatus.h"
40
41 struct db_ctdb_transaction_handle {
42         struct db_ctdb_ctx *ctx;
43         /*
44          * we store the writes done under a transaction:
45          */
46         struct ctdb_marshall_buffer *m_write;
47         uint32_t nesting;
48         bool nested_cancel;
49         char *lock_name;
50 };
51
52 struct db_ctdb_ctx {
53         struct db_context *db;
54         struct tdb_wrap *wtdb;
55         uint32_t db_id;
56         struct db_ctdb_transaction_handle *transaction;
57         struct g_lock_ctx *lock_ctx;
58
59         /* thresholds for warning messages */
60         int warn_unlock_msecs;
61         int warn_migrate_msecs;
62         int warn_migrate_attempts;
63         int warn_locktime_msecs;
64 };
65
66 struct db_ctdb_rec {
67         struct db_ctdb_ctx *ctdb_ctx;
68         struct ctdb_ltdb_header header;
69         struct timeval lock_time;
70 };
71
72 struct ctdb_async_ctx {
73         bool initialized;
74         struct ctdbd_connection *async_conn;
75 };
76
77 static struct ctdb_async_ctx ctdb_async_ctx;
78
79 static int ctdb_async_ctx_init_internal(TALLOC_CTX *mem_ctx,
80                                         struct tevent_context *ev,
81                                         bool reinit)
82 {
83         int ret;
84
85         if (reinit) {
86                 TALLOC_FREE(ctdb_async_ctx.async_conn);
87                 ctdb_async_ctx.initialized = false;
88         }
89
90         if (ctdb_async_ctx.initialized) {
91                 return 0;
92         }
93
94         become_root();
95         ret = ctdbd_init_async_connection(
96                 mem_ctx,
97                 lp_ctdbd_socket(),
98                 lp_ctdb_timeout(),
99                 &ctdb_async_ctx.async_conn);
100         unbecome_root();
101
102         if (ret != 0) {
103                 DBG_ERR("ctdbd_init_async_connection(%s, timeout=%d) "
104                         "failed: ret=%d %s\n",
105                         lp_ctdbd_socket(),
106                         lp_ctdb_timeout(),
107                         ret, strerror(ret));
108                 return ret;
109         }
110
111         SMB_ASSERT(ctdb_async_ctx.async_conn != NULL);
112
113         ctdb_async_ctx.initialized = true;
114         return 0;
115 }
116
117 static int ctdb_async_ctx_init(TALLOC_CTX *mem_ctx, struct tevent_context *ev)
118 {
119         return ctdb_async_ctx_init_internal(mem_ctx, ev, false);
120 }
121
122 int ctdb_async_ctx_reinit(TALLOC_CTX *mem_ctx, struct tevent_context *ev)
123 {
124         return ctdb_async_ctx_init_internal(mem_ctx, ev, true);
125 }
126
127 static NTSTATUS tdb_error_to_ntstatus(struct tdb_context *tdb)
128 {
129         enum TDB_ERROR tret = tdb_error(tdb);
130
131         return map_nt_error_from_tdb(tret);
132 }
133
134 struct db_ctdb_ltdb_parse_state {
135         void (*parser)(TDB_DATA key, struct ctdb_ltdb_header *header,
136                        TDB_DATA data, void *private_data);
137         void *private_data;
138 };
139
140 static int db_ctdb_ltdb_parser(TDB_DATA key, TDB_DATA data,
141                                void *private_data)
142 {
143         struct db_ctdb_ltdb_parse_state *state =
144                 (struct db_ctdb_ltdb_parse_state *)private_data;
145
146         if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
147                 return -1;
148         }
149
150         state->parser(
151                 key, (struct ctdb_ltdb_header *)data.dptr,
152                 make_tdb_data(data.dptr + sizeof(struct ctdb_ltdb_header),
153                               data.dsize - sizeof(struct ctdb_ltdb_header)),
154                 state->private_data);
155         return 0;
156 }
157
158 static NTSTATUS db_ctdb_ltdb_parse(
159         struct db_ctdb_ctx *db, TDB_DATA key,
160         void (*parser)(TDB_DATA key, struct ctdb_ltdb_header *header,
161                        TDB_DATA data, void *private_data),
162         void *private_data)
163 {
164         struct db_ctdb_ltdb_parse_state state;
165         int ret;
166
167         state.parser = parser;
168         state.private_data = private_data;
169
170         ret = tdb_parse_record(db->wtdb->tdb, key, db_ctdb_ltdb_parser,
171                                &state);
172         if (ret == -1) {
173                 return NT_STATUS_NOT_FOUND;
174         }
175         return NT_STATUS_OK;
176 }
177
178 /*
179  * Store a record together with the ctdb record header
180  * in the local copy of the database.
181  */
182 static NTSTATUS db_ctdb_ltdb_store(struct db_ctdb_ctx *db,
183                                    TDB_DATA key,
184                                    struct ctdb_ltdb_header *header,
185                                    const TDB_DATA *dbufs, int num_dbufs)
186 {
187         TDB_DATA recs[num_dbufs+1];
188         int ret;
189
190         recs[0] = (TDB_DATA) { .dptr = (uint8_t *)header,
191                                .dsize = sizeof(struct ctdb_ltdb_header) };
192         memcpy(&recs[1], dbufs, sizeof(TDB_DATA) * num_dbufs);
193
194         ret = tdb_storev(db->wtdb->tdb, key, recs, num_dbufs + 1, TDB_REPLACE);
195
196         return (ret == 0) ? NT_STATUS_OK
197                           : tdb_error_to_ntstatus(db->wtdb->tdb);
198
199 }
200
201 /*
202   form a ctdb_rec_data record from a key/data pair
203  */
204 static struct ctdb_rec_data_old *db_ctdb_marshall_record(TALLOC_CTX *mem_ctx, uint32_t reqid,
205                                                   TDB_DATA key,
206                                                   struct ctdb_ltdb_header *header,
207                                                   TDB_DATA data)
208 {
209         size_t length;
210         struct ctdb_rec_data_old *d;
211
212         length = offsetof(struct ctdb_rec_data_old, data) + key.dsize +
213                 data.dsize + sizeof(*header);
214         d = (struct ctdb_rec_data_old *)talloc_size(mem_ctx, length);
215         if (d == NULL) {
216                 return NULL;
217         }
218         d->length = length;
219         d->reqid = reqid;
220         d->keylen = key.dsize;
221         memcpy(&d->data[0], key.dptr, key.dsize);
222
223         d->datalen = data.dsize + sizeof(*header);
224         memcpy(&d->data[key.dsize], header, sizeof(*header));
225         memcpy(&d->data[key.dsize+sizeof(*header)], data.dptr, data.dsize);
226         return d;
227 }
228
229
230 /* helper function for marshalling multiple records */
231 static struct ctdb_marshall_buffer *db_ctdb_marshall_add(TALLOC_CTX *mem_ctx,
232                                                struct ctdb_marshall_buffer *m,
233                                                uint32_t db_id,
234                                                uint32_t reqid,
235                                                TDB_DATA key,
236                                                struct ctdb_ltdb_header *header,
237                                                TDB_DATA data)
238 {
239         struct ctdb_rec_data_old *r;
240         size_t m_size, r_size;
241         struct ctdb_marshall_buffer *m2 = NULL;
242
243         r = db_ctdb_marshall_record(talloc_tos(), reqid, key, header, data);
244         if (r == NULL) {
245                 talloc_free(m);
246                 return NULL;
247         }
248
249         if (m == NULL) {
250                 m = (struct ctdb_marshall_buffer *)talloc_zero_size(
251                         mem_ctx, offsetof(struct ctdb_marshall_buffer, data));
252                 if (m == NULL) {
253                         goto done;
254                 }
255                 m->db_id = db_id;
256         }
257
258         m_size = talloc_get_size(m);
259         r_size = talloc_get_size(r);
260
261         m2 = (struct ctdb_marshall_buffer *)talloc_realloc_size(
262                 mem_ctx, m,  m_size + r_size);
263         if (m2 == NULL) {
264                 talloc_free(m);
265                 goto done;
266         }
267
268         memcpy(m_size + (uint8_t *)m2, r, r_size);
269
270         m2->count++;
271
272 done:
273         talloc_free(r);
274         return m2;
275 }
276
277 /* we've finished marshalling, return a data blob with the marshalled records */
278 static TDB_DATA db_ctdb_marshall_finish(struct ctdb_marshall_buffer *m)
279 {
280         TDB_DATA data;
281         data.dptr = (uint8_t *)m;
282         data.dsize = talloc_get_size(m);
283         return data;
284 }
285
286 /*
287    loop over a marshalling buffer
288
289      - pass r==NULL to start
290      - loop the number of times indicated by m->count
291 */
292 static struct ctdb_rec_data_old *db_ctdb_marshall_loop_next_key(
293         struct ctdb_marshall_buffer *m, struct ctdb_rec_data_old *r, TDB_DATA *key)
294 {
295         if (r == NULL) {
296                 r = (struct ctdb_rec_data_old *)&m->data[0];
297         } else {
298                 r = (struct ctdb_rec_data_old *)(r->length + (uint8_t *)r);
299         }
300
301         key->dptr   = &r->data[0];
302         key->dsize  = r->keylen;
303         return r;
304 }
305
306 static bool db_ctdb_marshall_buf_parse(
307         struct ctdb_rec_data_old *r, uint32_t *reqid,
308         struct ctdb_ltdb_header **header, TDB_DATA *data)
309 {
310         if (r->datalen < sizeof(struct ctdb_ltdb_header)) {
311                 return false;
312         }
313
314         *reqid = r->reqid;
315
316         data->dptr  = &r->data[r->keylen] + sizeof(struct ctdb_ltdb_header);
317         data->dsize = r->datalen - sizeof(struct ctdb_ltdb_header);
318
319         *header = (struct ctdb_ltdb_header *)&r->data[r->keylen];
320
321         return true;
322 }
323
324 /**
325  * CTDB transaction destructor
326  */
327 static int db_ctdb_transaction_destructor(struct db_ctdb_transaction_handle *h)
328 {
329         NTSTATUS status;
330
331         status = g_lock_unlock(h->ctx->lock_ctx,
332                                string_term_tdb_data(h->lock_name));
333         if (!NT_STATUS_IS_OK(status)) {
334                 DEBUG(0, ("g_lock_unlock failed for %s: %s\n", h->lock_name,
335                           nt_errstr(status)));
336                 return -1;
337         }
338         return 0;
339 }
340
341 /**
342  * CTDB dbwrap API: transaction_start function
343  * starts a transaction on a persistent database
344  */
345 static int db_ctdb_transaction_start(struct db_context *db)
346 {
347         struct db_ctdb_transaction_handle *h;
348         NTSTATUS status;
349         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
350                                                         struct db_ctdb_ctx);
351
352         if (!db->persistent) {
353                 DEBUG(0,("transactions not supported on non-persistent database 0x%08x\n", 
354                          ctx->db_id));
355                 return -1;
356         }
357
358         if (ctx->transaction) {
359                 ctx->transaction->nesting++;
360                 DEBUG(5, (__location__ " transaction start on db 0x%08x: nesting %d -> %d\n",
361                           ctx->db_id, ctx->transaction->nesting - 1, ctx->transaction->nesting));
362                 return 0;
363         }
364
365         h = talloc_zero(db, struct db_ctdb_transaction_handle);
366         if (h == NULL) {
367                 DEBUG(0,(__location__ " oom for transaction handle\n"));
368                 return -1;
369         }
370
371         h->ctx = ctx;
372
373         h->lock_name = talloc_asprintf(h, "transaction_db_0x%08x",
374                                        (unsigned int)ctx->db_id);
375         if (h->lock_name == NULL) {
376                 DEBUG(0, ("talloc_asprintf failed\n"));
377                 TALLOC_FREE(h);
378                 return -1;
379         }
380
381         /*
382          * Wait a day, i.e. forever...
383          */
384         status = g_lock_lock(ctx->lock_ctx, string_term_tdb_data(h->lock_name),
385                              G_LOCK_WRITE, timeval_set(86400, 0));
386         if (!NT_STATUS_IS_OK(status)) {
387                 DEBUG(0, ("g_lock_lock failed: %s\n", nt_errstr(status)));
388                 TALLOC_FREE(h);
389                 return -1;
390         }
391
392         talloc_set_destructor(h, db_ctdb_transaction_destructor);
393
394         ctx->transaction = h;
395
396         DEBUG(5,(__location__ " transaction started on db 0x%08x\n", ctx->db_id));
397
398         return 0;
399 }
400
401 static bool parse_newest_in_marshall_buffer(
402         struct ctdb_marshall_buffer *buf, TDB_DATA key,
403         void (*parser)(TDB_DATA key, struct ctdb_ltdb_header *header,
404                        TDB_DATA data, void *private_data),
405         void *private_data)
406 {
407         struct ctdb_rec_data_old *rec = NULL;
408         struct ctdb_ltdb_header *h = NULL;
409         TDB_DATA data;
410         uint32_t i;
411
412         if (buf == NULL) {
413                 return false;
414         }
415
416         /*
417          * Walk the list of records written during this
418          * transaction. If we want to read one we have already
419          * written, return the last written sample. Thus we do not do
420          * a "break;" for the first hit, this record might have been
421          * overwritten later.
422          */
423
424         for (i=0; i<buf->count; i++) {
425                 TDB_DATA tkey;
426                 uint32_t reqid;
427
428                 rec = db_ctdb_marshall_loop_next_key(buf, rec, &tkey);
429                 if (rec == NULL) {
430                         return false;
431                 }
432
433                 if (!tdb_data_equal(key, tkey)) {
434                         continue;
435                 }
436
437                 if (!db_ctdb_marshall_buf_parse(rec, &reqid, &h, &data)) {
438                         return false;
439                 }
440         }
441
442         if (h == NULL) {
443                 return false;
444         }
445
446         parser(key, h, data, private_data);
447
448         return true;
449 }
450
451 struct pull_newest_from_marshall_buffer_state {
452         struct ctdb_ltdb_header *pheader;
453         TALLOC_CTX *mem_ctx;
454         TDB_DATA *pdata;
455 };
456
457 static void pull_newest_from_marshall_buffer_parser(
458         TDB_DATA key, struct ctdb_ltdb_header *header,
459         TDB_DATA data, void *private_data)
460 {
461         struct pull_newest_from_marshall_buffer_state *state =
462                 (struct pull_newest_from_marshall_buffer_state *)private_data;
463
464         if (state->pheader != NULL) {
465                 memcpy(state->pheader, header, sizeof(*state->pheader));
466         }
467         if (state->pdata != NULL) {
468                 state->pdata->dsize = data.dsize;
469                 state->pdata->dptr = (uint8_t *)talloc_memdup(
470                         state->mem_ctx, data.dptr, data.dsize);
471         }
472 }
473
474 static bool pull_newest_from_marshall_buffer(struct ctdb_marshall_buffer *buf,
475                                              TDB_DATA key,
476                                              struct ctdb_ltdb_header *pheader,
477                                              TALLOC_CTX *mem_ctx,
478                                              TDB_DATA *pdata)
479 {
480         struct pull_newest_from_marshall_buffer_state state;
481
482         state.pheader = pheader;
483         state.mem_ctx = mem_ctx;
484         state.pdata = pdata;
485
486         if (!parse_newest_in_marshall_buffer(
487                     buf, key, pull_newest_from_marshall_buffer_parser,
488                     &state)) {
489                 return false;
490         }
491         if ((pdata != NULL) && (pdata->dsize != 0) && (pdata->dptr == NULL)) {
492                 /* ENOMEM */
493                 return false;
494         }
495         return true;
496 }
497
498 static NTSTATUS db_ctdb_storev_transaction(struct db_record *rec,
499                                            const TDB_DATA *dbufs, int num_dbufs,
500                                            int flag);
501 static NTSTATUS db_ctdb_delete_transaction(struct db_record *rec);
502
503 static struct db_record *db_ctdb_fetch_locked_transaction(struct db_ctdb_ctx *ctx,
504                                                           TALLOC_CTX *mem_ctx,
505                                                           TDB_DATA key)
506 {
507         struct db_record *result;
508         TDB_DATA ctdb_data;
509
510         if (!(result = talloc(mem_ctx, struct db_record))) {
511                 DEBUG(0, ("talloc failed\n"));
512                 return NULL;
513         }
514
515         result->db = ctx->db;
516         result->private_data = ctx->transaction;
517
518         result->key.dsize = key.dsize;
519         result->key.dptr = (uint8_t *)talloc_memdup(result, key.dptr,
520                                                     key.dsize);
521         if (result->key.dptr == NULL) {
522                 DEBUG(0, ("talloc failed\n"));
523                 TALLOC_FREE(result);
524                 return NULL;
525         }
526
527         result->storev = db_ctdb_storev_transaction;
528         result->delete_rec = db_ctdb_delete_transaction;
529
530         if (ctx->transaction == NULL) {
531                 DEBUG(0, ("no transaction available\n"));
532                 TALLOC_FREE(result);
533                 return NULL;
534         }
535         if (pull_newest_from_marshall_buffer(ctx->transaction->m_write, key,
536                                              NULL, result, &result->value)) {
537                 result->value_valid = true;
538                 return result;
539         }
540
541         ctdb_data = tdb_fetch(ctx->wtdb->tdb, key);
542         if (ctdb_data.dptr == NULL) {
543                 /* create the record */
544                 result->value = tdb_null;
545                 result->value_valid = true;
546                 return result;
547         }
548
549         result->value.dsize = ctdb_data.dsize - sizeof(struct ctdb_ltdb_header);
550         result->value.dptr = NULL;
551
552         if ((result->value.dsize != 0)
553             && !(result->value.dptr = (uint8_t *)talloc_memdup(
554                          result, ctdb_data.dptr + sizeof(struct ctdb_ltdb_header),
555                          result->value.dsize))) {
556                 DEBUG(0, ("talloc failed\n"));
557                 TALLOC_FREE(result);
558                 return NULL;
559         }
560         result->value_valid = true;
561
562         SAFE_FREE(ctdb_data.dptr);
563
564         return result;
565 }
566
567 static int db_ctdb_record_destructor(struct db_record **recp)
568 {
569         struct db_record *rec = talloc_get_type_abort(*recp, struct db_record);
570         struct db_ctdb_transaction_handle *h = talloc_get_type_abort(
571                 rec->private_data, struct db_ctdb_transaction_handle);
572         int ret = h->ctx->db->transaction_commit(h->ctx->db);
573         if (ret != 0) {
574                 DEBUG(0,(__location__ " transaction_commit failed\n"));
575         }
576         return 0;
577 }
578
579 /*
580   auto-create a transaction for persistent databases
581  */
582 static struct db_record *db_ctdb_fetch_locked_persistent(struct db_ctdb_ctx *ctx,
583                                                          TALLOC_CTX *mem_ctx,
584                                                          TDB_DATA key)
585 {
586         int res;
587         struct db_record *rec, **recp;
588
589         res = db_ctdb_transaction_start(ctx->db);
590         if (res == -1) {
591                 return NULL;
592         }
593
594         rec = db_ctdb_fetch_locked_transaction(ctx, mem_ctx, key);
595         if (rec == NULL) {
596                 ctx->db->transaction_cancel(ctx->db);
597                 return NULL;
598         }
599
600         /* destroy this transaction when we release the lock */
601         recp = talloc(rec, struct db_record *);
602         if (recp == NULL) {
603                 ctx->db->transaction_cancel(ctx->db);
604                 talloc_free(rec);
605                 return NULL;
606         }
607         *recp = rec;
608         talloc_set_destructor(recp, db_ctdb_record_destructor);
609         return rec;
610 }
611
612
613 /*
614   stores a record inside a transaction
615  */
616 static NTSTATUS db_ctdb_transaction_store(struct db_ctdb_transaction_handle *h,
617                                           TDB_DATA key, TDB_DATA data)
618 {
619         TALLOC_CTX *tmp_ctx = talloc_new(h);
620         TDB_DATA rec;
621         struct ctdb_ltdb_header header;
622
623         ZERO_STRUCT(header);
624
625         /* we need the header so we can update the RSN */
626
627         if (!pull_newest_from_marshall_buffer(h->m_write, key, &header,
628                                               NULL, NULL)) {
629
630                 rec = tdb_fetch(h->ctx->wtdb->tdb, key);
631
632                 if (rec.dptr != NULL) {
633                         memcpy(&header, rec.dptr,
634                                sizeof(struct ctdb_ltdb_header));
635                         rec.dsize -= sizeof(struct ctdb_ltdb_header);
636
637                         /*
638                          * a special case, we are writing the same
639                          * data that is there now
640                          */
641                         if (data.dsize == rec.dsize &&
642                             memcmp(data.dptr,
643                                    rec.dptr + sizeof(struct ctdb_ltdb_header),
644                                    data.dsize) == 0) {
645                                 SAFE_FREE(rec.dptr);
646                                 talloc_free(tmp_ctx);
647                                 return NT_STATUS_OK;
648                         }
649                 }
650                 SAFE_FREE(rec.dptr);
651         }
652
653         header.dmaster = get_my_vnn();
654         header.rsn++;
655
656         h->m_write = db_ctdb_marshall_add(h, h->m_write, h->ctx->db_id, 0, key, &header, data);
657         if (h->m_write == NULL) {
658                 DEBUG(0,(__location__ " Failed to add to marshalling record\n"));
659                 talloc_free(tmp_ctx);
660                 return NT_STATUS_NO_MEMORY;
661         }
662
663         talloc_free(tmp_ctx);
664         return NT_STATUS_OK;
665 }
666
667
668 /* 
669    a record store inside a transaction
670  */
671 static NTSTATUS db_ctdb_storev_transaction(
672         struct db_record *rec, const TDB_DATA *dbufs, int num_dbufs, int flag)
673 {
674         struct db_ctdb_transaction_handle *h = talloc_get_type_abort(
675                 rec->private_data, struct db_ctdb_transaction_handle);
676         NTSTATUS status;
677         TDB_DATA data;
678
679         data = dbwrap_merge_dbufs(rec, dbufs, num_dbufs);
680         if (data.dptr == NULL) {
681                 return NT_STATUS_NO_MEMORY;
682         }
683
684         status = db_ctdb_transaction_store(h, rec->key, data);
685
686         TALLOC_FREE(data.dptr);
687
688         return status;
689 }
690
691 /*
692    a record delete inside a transaction
693  */
694 static NTSTATUS db_ctdb_delete_transaction(struct db_record *rec)
695 {
696         struct db_ctdb_transaction_handle *h = talloc_get_type_abort(
697                 rec->private_data, struct db_ctdb_transaction_handle);
698         NTSTATUS status;
699
700         status =  db_ctdb_transaction_store(h, rec->key, tdb_null);
701         return status;
702 }
703
704 static void db_ctdb_fetch_db_seqnum_parser(
705         TDB_DATA key, struct ctdb_ltdb_header *header,
706         TDB_DATA data, void *private_data)
707 {
708         uint64_t *seqnum = (uint64_t *)private_data;
709
710         if (data.dsize != sizeof(uint64_t)) {
711                 *seqnum = 0;
712                 return;
713         }
714         memcpy(seqnum, data.dptr, sizeof(*seqnum));
715 }
716
717 /**
718  * Fetch the db sequence number of a persistent db directly from the db.
719  */
720 static NTSTATUS db_ctdb_fetch_db_seqnum_from_db(struct db_ctdb_ctx *db,
721                                                 uint64_t *seqnum)
722 {
723         NTSTATUS status;
724         TDB_DATA key;
725
726         if (seqnum == NULL) {
727                 return NT_STATUS_INVALID_PARAMETER;
728         }
729
730         key = string_term_tdb_data(CTDB_DB_SEQNUM_KEY);
731
732         status = db_ctdb_ltdb_parse(
733                 db, key, db_ctdb_fetch_db_seqnum_parser, seqnum);
734
735         if (NT_STATUS_IS_OK(status)) {
736                 return NT_STATUS_OK;
737         }
738         if (NT_STATUS_EQUAL(status, NT_STATUS_NOT_FOUND)) {
739                 *seqnum = 0;
740                 return NT_STATUS_OK;
741         }
742         return status;
743 }
744
745 /**
746  * Store the database sequence number inside a transaction.
747  */
748 static NTSTATUS db_ctdb_store_db_seqnum(struct db_ctdb_transaction_handle *h,
749                                         uint64_t seqnum)
750 {
751         NTSTATUS status;
752         TDB_DATA key = string_term_tdb_data(CTDB_DB_SEQNUM_KEY);
753         TDB_DATA data = { .dptr=(uint8_t *)&seqnum, .dsize=sizeof(seqnum) };
754
755         status = db_ctdb_transaction_store(h, key, data);
756
757         return status;
758 }
759
760 /*
761   commit a transaction
762  */
763 static int db_ctdb_transaction_commit(struct db_context *db)
764 {
765         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
766                                                         struct db_ctdb_ctx);
767         NTSTATUS rets;
768         int32_t status;
769         struct db_ctdb_transaction_handle *h = ctx->transaction;
770         uint64_t old_seqnum, new_seqnum;
771         int ret;
772
773         if (h == NULL) {
774                 DEBUG(0,(__location__ " transaction commit with no open transaction on db 0x%08x\n", ctx->db_id));
775                 return -1;
776         }
777
778         if (h->nested_cancel) {
779                 db->transaction_cancel(db);
780                 DEBUG(5,(__location__ " Failed transaction commit after nested cancel\n"));
781                 return -1;
782         }
783
784         if (h->nesting != 0) {
785                 h->nesting--;
786                 DEBUG(5, (__location__ " transaction commit on db 0x%08x: nesting %d -> %d\n",
787                           ctx->db_id, ctx->transaction->nesting + 1, ctx->transaction->nesting));
788                 return 0;
789         }
790
791         if (h->m_write == NULL) {
792                 /*
793                  * No changes were made, so don't change the seqnum,
794                  * don't push to other node, just exit with success.
795                  */
796                 ret = 0;
797                 goto done;
798         }
799
800         DEBUG(5,(__location__ " transaction commit on db 0x%08x\n", ctx->db_id));
801
802         /*
803          * As the last db action before committing, bump the database sequence
804          * number. Note that this undoes all changes to the seqnum records
805          * performed under the transaction. This record is not meant to be
806          * modified by user interaction. It is for internal use only...
807          */
808         rets = db_ctdb_fetch_db_seqnum_from_db(ctx, &old_seqnum);
809         if (!NT_STATUS_IS_OK(rets)) {
810                 DEBUG(1, (__location__ " failed to fetch the db sequence number "
811                           "in transaction commit on db 0x%08x\n", ctx->db_id));
812                 ret = -1;
813                 goto done;
814         }
815
816         new_seqnum = old_seqnum + 1;
817
818         rets = db_ctdb_store_db_seqnum(h, new_seqnum);
819         if (!NT_STATUS_IS_OK(rets)) {
820                 DEBUG(1, (__location__ "failed to store the db sequence number "
821                           " in transaction commit on db 0x%08x\n", ctx->db_id));
822                 ret = -1;
823                 goto done;
824         }
825
826 again:
827         /* tell ctdbd to commit to the other nodes */
828         ret = ctdbd_control_local(messaging_ctdb_connection(),
829                                   CTDB_CONTROL_TRANS3_COMMIT,
830                                   h->ctx->db_id, 0,
831                                   db_ctdb_marshall_finish(h->m_write),
832                                   NULL, NULL, &status);
833         if ((ret != 0) || status != 0) {
834                 /*
835                  * The TRANS3_COMMIT control should only possibly fail when a
836                  * recovery has been running concurrently. In any case, the db
837                  * will be the same on all nodes, either the new copy or the
838                  * old copy.  This can be detected by comparing the old and new
839                  * local sequence numbers.
840                  */
841                 rets = db_ctdb_fetch_db_seqnum_from_db(ctx, &new_seqnum);
842                 if (!NT_STATUS_IS_OK(rets)) {
843                         DEBUG(1, (__location__ " failed to refetch db sequence "
844                                   "number after failed TRANS3_COMMIT\n"));
845                         ret = -1;
846                         goto done;
847                 }
848
849                 if (new_seqnum == old_seqnum) {
850                         /* Recovery prevented all our changes: retry. */
851                         goto again;
852                 }
853                 if (new_seqnum != (old_seqnum + 1)) {
854                         DEBUG(0, (__location__ " ERROR: new_seqnum[%lu] != "
855                                   "old_seqnum[%lu] + (0 or 1) after failed "
856                                   "TRANS3_COMMIT - this should not happen!\n",
857                                   (unsigned long)new_seqnum,
858                                   (unsigned long)old_seqnum));
859                         ret = -1;
860                         goto done;
861                 }
862                 /*
863                  * Recovery propagated our changes to all nodes, completing
864                  * our commit for us - succeed.
865                  */
866         }
867
868         ret = 0;
869
870 done:
871         h->ctx->transaction = NULL;
872         talloc_free(h);
873         return ret;
874 }
875
876
877 /*
878   cancel a transaction
879  */
880 static int db_ctdb_transaction_cancel(struct db_context *db)
881 {
882         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
883                                                         struct db_ctdb_ctx);
884         struct db_ctdb_transaction_handle *h = ctx->transaction;
885
886         if (h == NULL) {
887                 DEBUG(0,(__location__ " transaction cancel with no open transaction on db 0x%08x\n", ctx->db_id));
888                 return -1;
889         }
890
891         if (h->nesting != 0) {
892                 h->nesting--;
893                 h->nested_cancel = true;
894                 DEBUG(5, (__location__ " transaction cancel on db 0x%08x: nesting %d -> %d\n",
895                           ctx->db_id, ctx->transaction->nesting + 1, ctx->transaction->nesting));
896                 return 0;
897         }
898
899         DEBUG(5,(__location__ " Cancel transaction on db 0x%08x\n", ctx->db_id));
900
901         ctx->transaction = NULL;
902         talloc_free(h);
903         return 0;
904 }
905
906
907 static NTSTATUS db_ctdb_storev(struct db_record *rec,
908                                const TDB_DATA *dbufs, int num_dbufs, int flag)
909 {
910         struct db_ctdb_rec *crec = talloc_get_type_abort(
911                 rec->private_data, struct db_ctdb_rec);
912         NTSTATUS status;
913
914         status = db_ctdb_ltdb_store(crec->ctdb_ctx, rec->key, &(crec->header),
915                                     dbufs, num_dbufs);
916         return status;
917 }
918
919
920
921 static NTSTATUS db_ctdb_send_schedule_for_deletion(struct db_record *rec)
922 {
923         NTSTATUS status = NT_STATUS_OK;
924         int ret;
925         struct ctdb_control_schedule_for_deletion *dd;
926         TDB_DATA indata;
927         int32_t cstatus;
928         struct db_ctdb_rec *crec = talloc_get_type_abort(
929                 rec->private_data, struct db_ctdb_rec);
930         struct db_ctdb_ctx *ctx = crec->ctdb_ctx;
931
932         indata.dsize = offsetof(struct ctdb_control_schedule_for_deletion, key) + rec->key.dsize;
933         indata.dptr = talloc_zero_array(crec, uint8_t, indata.dsize);
934         if (indata.dptr == NULL) {
935                 DEBUG(0, (__location__ " talloc failed!\n"));
936                 return NT_STATUS_NO_MEMORY;
937         }
938
939         dd = (struct ctdb_control_schedule_for_deletion *)(void *)indata.dptr;
940         dd->db_id = ctx->db_id;
941         dd->hdr = crec->header;
942         dd->keylen = rec->key.dsize;
943         memcpy(dd->key, rec->key.dptr, rec->key.dsize);
944
945         ret = ctdbd_control_local(messaging_ctdb_connection(),
946                                   CTDB_CONTROL_SCHEDULE_FOR_DELETION,
947                                   crec->ctdb_ctx->db_id,
948                                   CTDB_CTRL_FLAG_NOREPLY, /* flags */
949                                   indata,
950                                   NULL, /* mem_ctx */
951                                   NULL, /* outdata */
952                                   &cstatus);
953         talloc_free(indata.dptr);
954
955         if ((ret != 0) || cstatus != 0) {
956                 DEBUG(1, (__location__ " Error sending local control "
957                           "SCHEDULE_FOR_DELETION: %s, cstatus = %"PRIi32"\n",
958                           strerror(ret), cstatus));
959                 if (ret != 0) {
960                         status = map_nt_error_from_unix(ret);
961                 } else {
962                         status = NT_STATUS_UNSUCCESSFUL;
963                 }
964         }
965
966         return status;
967 }
968
969 static NTSTATUS db_ctdb_delete(struct db_record *rec)
970 {
971         NTSTATUS status;
972
973         /*
974          * We have to store the header with empty data. TODO: Fix the
975          * tdb-level cleanup
976          */
977
978         status = db_ctdb_storev(rec, &tdb_null, 1, 0);
979         if (!NT_STATUS_IS_OK(status)) {
980                 return status;
981         }
982
983         status = db_ctdb_send_schedule_for_deletion(rec);
984         return status;
985 }
986
987 static int db_ctdb_record_destr(struct db_record* data)
988 {
989         struct db_ctdb_rec *crec = talloc_get_type_abort(
990                 data->private_data, struct db_ctdb_rec);
991         int threshold;
992         int ret;
993         struct timeval before;
994         double timediff;
995
996         DEBUG(10, (DEBUGLEVEL > 10
997                    ? "Unlocking db %u key %s\n"
998                    : "Unlocking db %u key %.20s\n",
999                    (int)crec->ctdb_ctx->db_id,
1000                    hex_encode_talloc(data, (unsigned char *)data->key.dptr,
1001                               data->key.dsize)));
1002
1003         before = timeval_current();
1004
1005         ret = tdb_chainunlock(crec->ctdb_ctx->wtdb->tdb, data->key);
1006
1007         timediff = timeval_elapsed(&before);
1008         timediff *= 1000;       /* get us milliseconds */
1009
1010         if (timediff > crec->ctdb_ctx->warn_unlock_msecs) {
1011                 char *key;
1012                 key = hex_encode_talloc(talloc_tos(),
1013                                         (unsigned char *)data->key.dptr,
1014                                         data->key.dsize);
1015                 DEBUG(0, ("tdb_chainunlock on db %s, key %s took %f milliseconds\n",
1016                           tdb_name(crec->ctdb_ctx->wtdb->tdb), key,
1017                           timediff));
1018                 TALLOC_FREE(key);
1019         }
1020
1021         if (ret != 0) {
1022                 DEBUG(0, ("tdb_chainunlock failed\n"));
1023                 return -1;
1024         }
1025
1026         threshold = crec->ctdb_ctx->warn_locktime_msecs;
1027         if (threshold != 0) {
1028                 timediff = timeval_elapsed(&crec->lock_time) * 1000;
1029                 if (timediff > threshold) {
1030                         const char *key;
1031
1032                         key = hex_encode_talloc(data,
1033                                                 (unsigned char *)data->key.dptr,
1034                                                 data->key.dsize);
1035                         DEBUG(0, ("Held tdb lock on db %s, key %s "
1036                                   "%f milliseconds\n",
1037                                   tdb_name(crec->ctdb_ctx->wtdb->tdb),
1038                                   key, timediff));
1039                 }
1040         }
1041
1042         return 0;
1043 }
1044
1045 /**
1046  * Check whether we have a valid local copy of the given record,
1047  * either for reading or for writing.
1048  */
1049 static bool db_ctdb_can_use_local_hdr(const struct ctdb_ltdb_header *hdr,
1050                                       uint32_t my_vnn, bool read_only)
1051 {
1052         if (hdr->dmaster != my_vnn) {
1053                 /* If we're not dmaster, it must be r/o copy. */
1054                 return read_only && (hdr->flags & CTDB_REC_RO_HAVE_READONLY);
1055         }
1056
1057         /*
1058          * If we want write access, no one may have r/o copies.
1059          */
1060         return read_only || !(hdr->flags & CTDB_REC_RO_HAVE_DELEGATIONS);
1061 }
1062
1063 static bool db_ctdb_can_use_local_copy(TDB_DATA ctdb_data, uint32_t my_vnn,
1064                                        bool read_only)
1065 {
1066         if (ctdb_data.dptr == NULL) {
1067                 return false;
1068         }
1069
1070         if (ctdb_data.dsize < sizeof(struct ctdb_ltdb_header)) {
1071                 return false;
1072         }
1073
1074         return db_ctdb_can_use_local_hdr(
1075                 (struct ctdb_ltdb_header *)ctdb_data.dptr, my_vnn, read_only);
1076 }
1077
1078 static struct db_record *fetch_locked_internal(struct db_ctdb_ctx *ctx,
1079                                                TALLOC_CTX *mem_ctx,
1080                                                TDB_DATA key)
1081 {
1082         struct db_record *result;
1083         struct db_ctdb_rec *crec;
1084         TDB_DATA ctdb_data;
1085         int migrate_attempts;
1086         struct timeval migrate_start;
1087         struct timeval chainlock_start;
1088         struct timeval ctdb_start_time;
1089         double chainlock_time = 0;
1090         double ctdb_time = 0;
1091         int duration_msecs;
1092         int lockret;
1093         int ret;
1094
1095         if (!(result = talloc(mem_ctx, struct db_record))) {
1096                 DEBUG(0, ("talloc failed\n"));
1097                 return NULL;
1098         }
1099
1100         if (!(crec = talloc_zero(result, struct db_ctdb_rec))) {
1101                 DEBUG(0, ("talloc failed\n"));
1102                 TALLOC_FREE(result);
1103                 return NULL;
1104         }
1105
1106         result->db = ctx->db;
1107         result->private_data = (void *)crec;
1108         crec->ctdb_ctx = ctx;
1109
1110         result->key.dsize = key.dsize;
1111         result->key.dptr = (uint8_t *)talloc_memdup(result, key.dptr,
1112                                                     key.dsize);
1113         if (result->key.dptr == NULL) {
1114                 DEBUG(0, ("talloc failed\n"));
1115                 TALLOC_FREE(result);
1116                 return NULL;
1117         }
1118
1119         migrate_attempts = 0;
1120         GetTimeOfDay(&migrate_start);
1121
1122         /*
1123          * Do a blocking lock on the record
1124          */
1125 again:
1126
1127         if (DEBUGLEVEL >= 10) {
1128                 char *keystr = hex_encode_talloc(result, key.dptr, key.dsize);
1129                 DEBUG(10, (DEBUGLEVEL > 10
1130                            ? "Locking db %u key %s\n"
1131                            : "Locking db %u key %.20s\n",
1132                            (int)crec->ctdb_ctx->db_id, keystr));
1133                 TALLOC_FREE(keystr);
1134         }
1135
1136         GetTimeOfDay(&chainlock_start);
1137         lockret = tdb_chainlock(ctx->wtdb->tdb, key);
1138         chainlock_time += timeval_elapsed(&chainlock_start);
1139
1140         if (lockret != 0) {
1141                 DEBUG(3, ("tdb_chainlock failed\n"));
1142                 TALLOC_FREE(result);
1143                 return NULL;
1144         }
1145
1146         result->storev = db_ctdb_storev;
1147         result->delete_rec = db_ctdb_delete;
1148         talloc_set_destructor(result, db_ctdb_record_destr);
1149
1150         ctdb_data = tdb_fetch(ctx->wtdb->tdb, key);
1151
1152         /*
1153          * See if we have a valid record and we are the dmaster. If so, we can
1154          * take the shortcut and just return it.
1155          */
1156
1157         if (!db_ctdb_can_use_local_copy(ctdb_data, get_my_vnn(), false)) {
1158                 SAFE_FREE(ctdb_data.dptr);
1159                 tdb_chainunlock(ctx->wtdb->tdb, key);
1160                 talloc_set_destructor(result, NULL);
1161
1162                 migrate_attempts += 1;
1163
1164                 DEBUG(10, ("ctdb_data.dptr = %p, dmaster = %"PRIu32" "
1165                            "(%"PRIu32") %"PRIu32"\n",
1166                            ctdb_data.dptr, ctdb_data.dptr ?
1167                            ((struct ctdb_ltdb_header *)ctdb_data.dptr)->dmaster :
1168                            UINT32_MAX,
1169                            get_my_vnn(),
1170                            ctdb_data.dptr ?
1171                            ((struct ctdb_ltdb_header *)ctdb_data.dptr)->flags : 0));
1172
1173                 GetTimeOfDay(&ctdb_start_time);
1174                 ret = ctdbd_migrate(messaging_ctdb_connection(), ctx->db_id,
1175                                     key);
1176                 ctdb_time += timeval_elapsed(&ctdb_start_time);
1177
1178                 if (ret != 0) {
1179                         DEBUG(5, ("ctdbd_migrate failed: %s\n",
1180                                   strerror(ret)));
1181                         TALLOC_FREE(result);
1182                         return NULL;
1183                 }
1184                 /* now its migrated, try again */
1185                 goto again;
1186         }
1187
1188         {
1189                 double duration;
1190                 duration = timeval_elapsed(&migrate_start);
1191
1192                 /*
1193                  * Convert the duration to milliseconds to avoid a
1194                  * floating-point division of
1195                  * lp_parm_int("migrate_duration") by 1000.
1196                  */
1197                 duration_msecs = duration * 1000;
1198         }
1199
1200         if ((migrate_attempts > ctx->warn_migrate_attempts) ||
1201             (duration_msecs > ctx->warn_migrate_msecs)) {
1202                 int chain = 0;
1203
1204                 if (tdb_get_flags(ctx->wtdb->tdb) & TDB_INCOMPATIBLE_HASH) {
1205                         chain = tdb_jenkins_hash(&key) %
1206                                 tdb_hash_size(ctx->wtdb->tdb);
1207                 }
1208
1209                 DEBUG(0, ("db_ctdb_fetch_locked for %s key %s, chain %d "
1210                           "needed %d attempts, %d milliseconds, "
1211                           "chainlock: %f ms, CTDB %f ms\n",
1212                           tdb_name(ctx->wtdb->tdb),
1213                           hex_encode_talloc(talloc_tos(),
1214                                             (unsigned char *)key.dptr,
1215                                             key.dsize),
1216                           chain,
1217                           migrate_attempts, duration_msecs,
1218                           chainlock_time * 1000.0,
1219                           ctdb_time * 1000.0));
1220         }
1221
1222         GetTimeOfDay(&crec->lock_time);
1223
1224         memcpy(&crec->header, ctdb_data.dptr, sizeof(crec->header));
1225
1226         result->value.dsize = ctdb_data.dsize - sizeof(crec->header);
1227         result->value.dptr = NULL;
1228
1229         if (result->value.dsize != 0) {
1230                 result->value.dptr = talloc_memdup(
1231                         result, ctdb_data.dptr + sizeof(crec->header),
1232                         result->value.dsize);
1233                 if (result->value.dptr == NULL) {
1234                         DBG_ERR("talloc failed\n");
1235                         TALLOC_FREE(result);
1236                         return NULL;
1237                 }
1238         }
1239         result->value_valid = true;
1240
1241         SAFE_FREE(ctdb_data.dptr);
1242
1243         return result;
1244 }
1245
1246 static struct db_record *db_ctdb_fetch_locked(struct db_context *db,
1247                                               TALLOC_CTX *mem_ctx,
1248                                               TDB_DATA key)
1249 {
1250         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1251                                                         struct db_ctdb_ctx);
1252
1253         if (ctx->transaction != NULL) {
1254                 return db_ctdb_fetch_locked_transaction(ctx, mem_ctx, key);
1255         }
1256
1257         if (db->persistent) {
1258                 return db_ctdb_fetch_locked_persistent(ctx, mem_ctx, key);
1259         }
1260
1261         return fetch_locked_internal(ctx, mem_ctx, key);
1262 }
1263
1264 struct db_ctdb_parse_record_state {
1265         void (*parser)(TDB_DATA key, TDB_DATA data, void *private_data);
1266         void *private_data;
1267         uint32_t my_vnn;
1268         bool ask_for_readonly_copy;
1269         bool done;
1270         bool empty_record;
1271 };
1272
1273 static void db_ctdb_parse_record_parser(
1274         TDB_DATA key, struct ctdb_ltdb_header *header,
1275         TDB_DATA data, void *private_data)
1276 {
1277         struct db_ctdb_parse_record_state *state =
1278                 (struct db_ctdb_parse_record_state *)private_data;
1279         state->parser(key, data, state->private_data);
1280 }
1281
1282 static void db_ctdb_parse_record_parser_nonpersistent(
1283         TDB_DATA key, struct ctdb_ltdb_header *header,
1284         TDB_DATA data, void *private_data)
1285 {
1286         struct db_ctdb_parse_record_state *state =
1287                 (struct db_ctdb_parse_record_state *)private_data;
1288
1289         if (db_ctdb_can_use_local_hdr(header, state->my_vnn, true)) {
1290                 /*
1291                  * A record consisting only of the ctdb header can be
1292                  * a validly created empty record or a tombstone
1293                  * record of a deleted record (not vacuumed yet). Mark
1294                  * it accordingly.
1295                  */
1296                 state->empty_record = (data.dsize == 0);
1297                 if (!state->empty_record) {
1298                         state->parser(key, data, state->private_data);
1299                 }
1300                 state->done = true;
1301         } else {
1302                 /*
1303                  * We found something in the db, so it seems that this record,
1304                  * while not usable locally right now, is popular. Ask for a
1305                  * R/O copy.
1306                  */
1307                 state->ask_for_readonly_copy = true;
1308         }
1309 }
1310
1311 static NTSTATUS db_ctdb_try_parse_local_record(struct db_ctdb_ctx *ctx,
1312                                                TDB_DATA key,
1313                                                struct db_ctdb_parse_record_state *state)
1314 {
1315         NTSTATUS status;
1316
1317         if (ctx->transaction != NULL) {
1318                 struct db_ctdb_transaction_handle *h = ctx->transaction;
1319                 bool found;
1320
1321                 /*
1322                  * Transactions only happen for persistent db's.
1323                  */
1324
1325                 found = parse_newest_in_marshall_buffer(
1326                         h->m_write, key, db_ctdb_parse_record_parser, state);
1327
1328                 if (found) {
1329                         return NT_STATUS_OK;
1330                 }
1331         }
1332
1333         if (ctx->db->persistent) {
1334                 /*
1335                  * Persistent db, but not found in the transaction buffer
1336                  */
1337                 return db_ctdb_ltdb_parse(
1338                         ctx, key, db_ctdb_parse_record_parser, state);
1339         }
1340
1341         state->done = false;
1342         state->ask_for_readonly_copy = false;
1343
1344         status = db_ctdb_ltdb_parse(
1345                 ctx, key, db_ctdb_parse_record_parser_nonpersistent, state);
1346         if (NT_STATUS_IS_OK(status) && state->done) {
1347                 if (state->empty_record) {
1348                         /*
1349                          * We know authoritatively, that this is an empty
1350                          * record. Since ctdb does not distinguish between empty
1351                          * and deleted records, this can be a record stored as
1352                          * empty or a not-yet-vacuumed tombstone record of a
1353                          * deleted record. Now Samba right now can live without
1354                          * empty records, so we can safely report this record
1355                          * as non-existing.
1356                          *
1357                          * See bugs 10008 and 12005.
1358                          */
1359                         return NT_STATUS_NOT_FOUND;
1360                 }
1361                 return NT_STATUS_OK;
1362         }
1363
1364         return NT_STATUS_MORE_PROCESSING_REQUIRED;
1365 }
1366
1367 static NTSTATUS db_ctdb_parse_record(struct db_context *db, TDB_DATA key,
1368                                      void (*parser)(TDB_DATA key,
1369                                                     TDB_DATA data,
1370                                                     void *private_data),
1371                                      void *private_data)
1372 {
1373         struct db_ctdb_ctx *ctx = talloc_get_type_abort(
1374                 db->private_data, struct db_ctdb_ctx);
1375         struct db_ctdb_parse_record_state state;
1376         NTSTATUS status;
1377         int ret;
1378
1379         state.parser = parser;
1380         state.private_data = private_data;
1381         state.my_vnn = get_my_vnn();
1382         state.empty_record = false;
1383
1384         status = db_ctdb_try_parse_local_record(ctx, key, &state);
1385         if (!NT_STATUS_EQUAL(status, NT_STATUS_MORE_PROCESSING_REQUIRED)) {
1386                 return status;
1387         }
1388
1389         ret = ctdbd_parse(messaging_ctdb_connection(), ctx->db_id, key,
1390                           state.ask_for_readonly_copy, parser, private_data);
1391         if (ret != 0) {
1392                 if (ret == ENOENT) {
1393                         /*
1394                          * This maps to
1395                          * NT_STATUS_OBJECT_NAME_NOT_FOUND. Our upper
1396                          * layers expect NT_STATUS_NOT_FOUND for "no
1397                          * record around". We need to convert dbwrap
1398                          * to 0/errno away from NTSTATUS ... :-)
1399                          */
1400                         return NT_STATUS_NOT_FOUND;
1401                 }
1402                 return map_nt_error_from_unix(ret);
1403         }
1404         return NT_STATUS_OK;
1405 }
1406
1407 static void db_ctdb_parse_record_done(struct tevent_req *subreq);
1408
1409 static struct tevent_req *db_ctdb_parse_record_send(
1410         TALLOC_CTX *mem_ctx,
1411         struct tevent_context *ev,
1412         struct db_context *db,
1413         TDB_DATA key,
1414         void (*parser)(TDB_DATA key,
1415                        TDB_DATA data,
1416                        void *private_data),
1417         void *private_data,
1418         enum dbwrap_req_state *req_state)
1419 {
1420         struct db_ctdb_ctx *ctx = talloc_get_type_abort(
1421                 db->private_data, struct db_ctdb_ctx);
1422         struct tevent_req *req = NULL;
1423         struct tevent_req *subreq = NULL;
1424         struct db_ctdb_parse_record_state *state = NULL;
1425         NTSTATUS status;
1426
1427         req = tevent_req_create(mem_ctx, &state,
1428                                 struct db_ctdb_parse_record_state);
1429         if (req == NULL) {
1430                 *req_state = DBWRAP_REQ_ERROR;
1431                 return NULL;
1432
1433         }
1434
1435         *state = (struct db_ctdb_parse_record_state) {
1436                 .parser = parser,
1437                 .private_data = private_data,
1438                 .my_vnn = get_my_vnn(),
1439                 .empty_record = false,
1440         };
1441
1442         status = db_ctdb_try_parse_local_record(ctx, key, state);
1443         if (!NT_STATUS_EQUAL(status, NT_STATUS_MORE_PROCESSING_REQUIRED)) {
1444                 if (tevent_req_nterror(req, status)) {
1445                         *req_state = DBWRAP_REQ_ERROR;
1446                         return tevent_req_post(req, ev);
1447                 }
1448                 *req_state = DBWRAP_REQ_DONE;
1449                 tevent_req_done(req);
1450                 return tevent_req_post(req, ev);
1451         }
1452
1453         subreq = ctdbd_parse_send(state,
1454                                   ev,
1455                                   ctdb_async_ctx.async_conn,
1456                                   ctx->db_id,
1457                                   key,
1458                                   state->ask_for_readonly_copy,
1459                                   parser,
1460                                   private_data,
1461                                   req_state);
1462         if (tevent_req_nomem(subreq, req)) {
1463                 *req_state = DBWRAP_REQ_ERROR;
1464                 return tevent_req_post(req, ev);
1465         }
1466         tevent_req_set_callback(subreq, db_ctdb_parse_record_done, req);
1467
1468         return req;
1469 }
1470
1471 static void db_ctdb_parse_record_done(struct tevent_req *subreq)
1472 {
1473         struct tevent_req *req = tevent_req_callback_data(
1474                 subreq, struct tevent_req);
1475         int ret;
1476
1477         ret = ctdbd_parse_recv(subreq);
1478         TALLOC_FREE(subreq);
1479         if (ret != 0) {
1480                 if (ret == ENOENT) {
1481                         /*
1482                          * This maps to NT_STATUS_OBJECT_NAME_NOT_FOUND. Our
1483                          * upper layers expect NT_STATUS_NOT_FOUND for "no
1484                          * record around". We need to convert dbwrap to 0/errno
1485                          * away from NTSTATUS ... :-)
1486                          */
1487                         tevent_req_nterror(req, NT_STATUS_NOT_FOUND);
1488                         return;
1489                 }
1490                 tevent_req_nterror(req, map_nt_error_from_unix(ret));
1491                 return;
1492         }
1493
1494         tevent_req_done(req);
1495         return;
1496 }
1497
1498 static NTSTATUS db_ctdb_parse_record_recv(struct tevent_req *req)
1499 {
1500         return tevent_req_simple_recv_ntstatus(req);
1501 }
1502
1503 struct traverse_state {
1504         struct db_context *db;
1505         int (*fn)(struct db_record *rec, void *private_data);
1506         void *private_data;
1507         int count;
1508 };
1509
1510 static void traverse_callback(TDB_DATA key, TDB_DATA data, void *private_data)
1511 {
1512         struct traverse_state *state = (struct traverse_state *)private_data;
1513         struct db_record *rec = NULL;
1514         TALLOC_CTX *tmp_ctx = NULL;
1515
1516         tmp_ctx = talloc_new(state->db);
1517         if (tmp_ctx == NULL) {
1518                 DBG_ERR("talloc_new failed\n");
1519                 return;
1520         }
1521
1522         /* we have to give them a locked record to prevent races */
1523         rec = db_ctdb_fetch_locked(state->db, tmp_ctx, key);
1524         if (rec != NULL && rec->value.dsize > 0) {
1525                 state->fn(rec, state->private_data);
1526                 state->count++;
1527         }
1528         talloc_free(tmp_ctx);
1529 }
1530
1531 static int traverse_persistent_callback(TDB_CONTEXT *tdb, TDB_DATA kbuf, TDB_DATA dbuf,
1532                                         void *private_data)
1533 {
1534         struct traverse_state *state = (struct traverse_state *)private_data;
1535         struct db_record *rec;
1536         TALLOC_CTX *tmp_ctx = talloc_new(state->db);
1537         int ret = 0;
1538
1539         /*
1540          * Skip the __db_sequence_number__ key:
1541          * This is used for persistent transactions internally.
1542          */
1543         if (kbuf.dsize == strlen(CTDB_DB_SEQNUM_KEY) + 1 &&
1544             strcmp((const char*)kbuf.dptr, CTDB_DB_SEQNUM_KEY) == 0)
1545         {
1546                 goto done;
1547         }
1548
1549         /* we have to give them a locked record to prevent races */
1550         rec = db_ctdb_fetch_locked(state->db, tmp_ctx, kbuf);
1551         if (rec && rec->value.dsize > 0) {
1552                 ret = state->fn(rec, state->private_data);
1553         }
1554
1555 done:
1556         talloc_free(tmp_ctx);
1557         return ret;
1558 }
1559
1560 /* wrapper to use traverse_persistent_callback with dbwrap */
1561 static int traverse_persistent_callback_dbwrap(struct db_record *rec, void* data)
1562 {
1563         return traverse_persistent_callback(NULL, rec->key, rec->value, data);
1564 }
1565
1566 static int db_ctdbd_traverse(uint32_t db_id,
1567                              void (*fn)(TDB_DATA key, TDB_DATA data,
1568                                         void *private_data),
1569                              void *private_data)
1570 {
1571         struct ctdbd_connection *conn;
1572         int ret;
1573
1574         become_root();
1575         ret = ctdbd_init_connection(talloc_tos(), lp_ctdbd_socket(),
1576                                     lp_ctdb_timeout(), &conn);
1577         unbecome_root();
1578         if (ret != 0) {
1579                 DBG_WARNING("ctdbd_init_connection failed: %s\n",
1580                             strerror(ret));
1581                 return ret;
1582         }
1583
1584         ret = ctdbd_traverse(conn, db_id, fn, private_data);
1585         TALLOC_FREE(conn);
1586
1587         if (ret != 0) {
1588                 DBG_WARNING("ctdbd_traverse failed: %s\n",
1589                             strerror(ret));
1590                 return ret;
1591         }
1592
1593         return 0;
1594 }
1595
1596
1597 static int db_ctdb_traverse(struct db_context *db,
1598                             int (*fn)(struct db_record *rec,
1599                                       void *private_data),
1600                             void *private_data)
1601 {
1602         int ret;
1603         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1604                                                         struct db_ctdb_ctx);
1605         struct traverse_state state;
1606
1607         state = (struct traverse_state) {
1608                 .db = db,
1609                 .fn = fn,
1610                 .private_data = private_data,
1611         };
1612
1613         if (db->persistent) {
1614                 struct tdb_context *ltdb = ctx->wtdb->tdb;
1615
1616                 /* for persistent databases we don't need to do a ctdb traverse,
1617                    we can do a faster local traverse */
1618                 ret = tdb_traverse(ltdb, traverse_persistent_callback, &state);
1619                 if (ret < 0) {
1620                         return ret;
1621                 }
1622                 if (ctx->transaction && ctx->transaction->m_write) {
1623                         /*
1624                          * we now have to handle keys not yet
1625                          * present at transaction start
1626                          */
1627                         struct db_context *newkeys = db_open_rbt(talloc_tos());
1628                         struct ctdb_marshall_buffer *mbuf = ctx->transaction->m_write;
1629                         struct ctdb_rec_data_old *rec=NULL;
1630                         uint32_t i;
1631                         int count = 0;
1632                         NTSTATUS status;
1633
1634                         if (newkeys == NULL) {
1635                                 return -1;
1636                         }
1637
1638                         for (i=0; i<mbuf->count; i++) {
1639                                 TDB_DATA key;
1640                                 rec = db_ctdb_marshall_loop_next_key(
1641                                         mbuf, rec, &key);
1642                                 SMB_ASSERT(rec != NULL);
1643
1644                                 if (!tdb_exists(ltdb, key)) {
1645                                         dbwrap_store(newkeys, key, tdb_null, 0);
1646                                 }
1647                         }
1648                         status = dbwrap_traverse(newkeys,
1649                                                  traverse_persistent_callback_dbwrap,
1650                                                  &state,
1651                                                  &count);
1652                         talloc_free(newkeys);
1653                         if (!NT_STATUS_IS_OK(status)) {
1654                                 return -1;
1655                         }
1656                         ret += count;
1657                 }
1658                 return ret;
1659         }
1660
1661         ret = db_ctdbd_traverse(ctx->db_id, traverse_callback, &state);
1662         if (ret != 0) {
1663                 return -1;
1664         }
1665         return state.count;
1666 }
1667
1668 static NTSTATUS db_ctdb_storev_deny(struct db_record *rec,
1669                                     const TDB_DATA *dbufs, int num_dbufs, int flag)
1670 {
1671         return NT_STATUS_MEDIA_WRITE_PROTECTED;
1672 }
1673
1674 static NTSTATUS db_ctdb_delete_deny(struct db_record *rec)
1675 {
1676         return NT_STATUS_MEDIA_WRITE_PROTECTED;
1677 }
1678
1679 static void traverse_read_callback(TDB_DATA key, TDB_DATA data, void *private_data)
1680 {
1681         struct traverse_state *state = (struct traverse_state *)private_data;
1682         struct db_record rec;
1683
1684         ZERO_STRUCT(rec);
1685         rec.db = state->db;
1686         rec.key = key;
1687         rec.value = data;
1688         rec.storev = db_ctdb_storev_deny;
1689         rec.delete_rec = db_ctdb_delete_deny;
1690         rec.private_data = NULL;
1691         rec.value_valid = true;
1692         state->fn(&rec, state->private_data);
1693         state->count++;
1694 }
1695
1696 static int traverse_persistent_callback_read(TDB_CONTEXT *tdb, TDB_DATA kbuf, TDB_DATA dbuf,
1697                                         void *private_data)
1698 {
1699         struct traverse_state *state = (struct traverse_state *)private_data;
1700         struct db_record rec;
1701
1702         /*
1703          * Skip the __db_sequence_number__ key:
1704          * This is used for persistent transactions internally.
1705          */
1706         if (kbuf.dsize == strlen(CTDB_DB_SEQNUM_KEY) + 1 &&
1707             strcmp((const char*)kbuf.dptr, CTDB_DB_SEQNUM_KEY) == 0)
1708         {
1709                 return 0;
1710         }
1711
1712         ZERO_STRUCT(rec);
1713         rec.db = state->db;
1714         rec.key = kbuf;
1715         rec.value = dbuf;
1716         rec.value_valid = true;
1717         rec.storev = db_ctdb_storev_deny;
1718         rec.delete_rec = db_ctdb_delete_deny;
1719         rec.private_data = NULL;
1720
1721         if (rec.value.dsize <= sizeof(struct ctdb_ltdb_header)) {
1722                 /* a deleted record */
1723                 return 0;
1724         }
1725         rec.value.dsize -= sizeof(struct ctdb_ltdb_header);
1726         rec.value.dptr += sizeof(struct ctdb_ltdb_header);
1727
1728         state->count++;
1729         return state->fn(&rec, state->private_data);
1730 }
1731
1732 static int db_ctdb_traverse_read(struct db_context *db,
1733                                  int (*fn)(struct db_record *rec,
1734                                            void *private_data),
1735                                  void *private_data)
1736 {
1737         int ret;
1738         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1739                                                         struct db_ctdb_ctx);
1740         struct traverse_state state;
1741
1742         state = (struct traverse_state) {
1743                 .db = db,
1744                 .fn = fn,
1745                 .private_data = private_data,
1746         };
1747
1748         if (db->persistent) {
1749                 /* for persistent databases we don't need to do a ctdb traverse,
1750                    we can do a faster local traverse */
1751                 int nrecs;
1752
1753                 nrecs = tdb_traverse_read(ctx->wtdb->tdb,
1754                                           traverse_persistent_callback_read,
1755                                           &state);
1756                 if (nrecs == -1) {
1757                         return -1;
1758                 }
1759                 return state.count;
1760         }
1761
1762         ret = db_ctdbd_traverse(ctx->db_id, traverse_read_callback, &state);
1763         if (ret != 0) {
1764                 return -1;
1765         }
1766         return state.count;
1767 }
1768
1769 static int db_ctdb_get_seqnum(struct db_context *db)
1770 {
1771         struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
1772                                                         struct db_ctdb_ctx);
1773         return tdb_get_seqnum(ctx->wtdb->tdb);
1774 }
1775
1776 static size_t db_ctdb_id(struct db_context *db, uint8_t *id, size_t idlen)
1777 {
1778         struct db_ctdb_ctx *ctx = talloc_get_type_abort(
1779                 db->private_data, struct db_ctdb_ctx);
1780
1781         if (idlen >= sizeof(ctx->db_id)) {
1782                 memcpy(id, &ctx->db_id, sizeof(ctx->db_id));
1783         }
1784
1785         return sizeof(ctx->db_id);
1786 }
1787
1788 struct db_context *db_open_ctdb(TALLOC_CTX *mem_ctx,
1789                                 struct messaging_context *msg_ctx,
1790                                 const char *name,
1791                                 int hash_size, int tdb_flags,
1792                                 int open_flags, mode_t mode,
1793                                 enum dbwrap_lock_order lock_order,
1794                                 uint64_t dbwrap_flags)
1795 {
1796         struct db_context *result;
1797         struct db_ctdb_ctx *db_ctdb;
1798         char *db_path;
1799         struct loadparm_context *lp_ctx;
1800         TDB_DATA data;
1801         TDB_DATA outdata = {0};
1802         bool persistent = (tdb_flags & TDB_CLEAR_IF_FIRST) == 0;
1803         int32_t cstatus;
1804         int ret;
1805
1806         if (!lp_clustering()) {
1807                 DEBUG(10, ("Clustering disabled -- no ctdb\n"));
1808                 return NULL;
1809         }
1810
1811         if (!(result = talloc_zero(mem_ctx, struct db_context))) {
1812                 DEBUG(0, ("talloc failed\n"));
1813                 TALLOC_FREE(result);
1814                 return NULL;
1815         }
1816
1817         if (!(db_ctdb = talloc(result, struct db_ctdb_ctx))) {
1818                 DEBUG(0, ("talloc failed\n"));
1819                 TALLOC_FREE(result);
1820                 return NULL;
1821         }
1822
1823         result->name = talloc_strdup(result, name);
1824         if (result->name == NULL) {
1825                 DEBUG(0, ("talloc failed\n"));
1826                 TALLOC_FREE(result);
1827                 return NULL;
1828         }
1829
1830         db_ctdb->transaction = NULL;
1831         db_ctdb->db = result;
1832
1833         ret = ctdbd_db_attach(messaging_ctdb_connection(), name,
1834                               &db_ctdb->db_id, persistent);
1835         if (ret != 0) {
1836                 DEBUG(0, ("ctdbd_db_attach failed for %s: %s\n", name,
1837                           strerror(ret)));
1838                 TALLOC_FREE(result);
1839                 return NULL;
1840         }
1841
1842         if (tdb_flags & TDB_SEQNUM) {
1843                 data.dptr = (uint8_t *)&db_ctdb->db_id;
1844                 data.dsize = sizeof(db_ctdb->db_id);
1845
1846                 ret = ctdbd_control_local(messaging_ctdb_connection(),
1847                                           CTDB_CONTROL_ENABLE_SEQNUM,
1848                                           0, 0, data,
1849                                           NULL, NULL, &cstatus);
1850                 if ((ret != 0) || cstatus != 0) {
1851                         DBG_ERR("ctdb_control for enable seqnum "
1852                                 "failed: %s\n", strerror(ret));
1853                         TALLOC_FREE(result);
1854                         return NULL;
1855                 }
1856         }
1857
1858         db_path = ctdbd_dbpath(messaging_ctdb_connection(), db_ctdb,
1859                                db_ctdb->db_id);
1860         if (db_path == NULL) {
1861                 DBG_ERR("ctdbd_dbpath failed\n");
1862                 TALLOC_FREE(result);
1863                 return NULL;
1864         }
1865
1866         result->persistent = persistent;
1867         result->lock_order = lock_order;
1868
1869         data.dptr = (uint8_t *)&db_ctdb->db_id;
1870         data.dsize = sizeof(db_ctdb->db_id);
1871
1872         ret = ctdbd_control_local(messaging_ctdb_connection(),
1873                                   CTDB_CONTROL_DB_OPEN_FLAGS,
1874                                   0, 0, data, NULL, &outdata, &cstatus);
1875         if (ret != 0) {
1876                 DBG_ERR(" ctdb control for db_open_flags "
1877                          "failed: %s\n", strerror(ret));
1878                 TALLOC_FREE(result);
1879                 return NULL;
1880         }
1881
1882         if (cstatus != 0 || outdata.dsize != sizeof(int)) {
1883                 DBG_ERR("ctdb_control for db_open_flags failed\n");
1884                 TALLOC_FREE(outdata.dptr);
1885                 TALLOC_FREE(result);
1886                 return NULL;
1887         }
1888
1889         tdb_flags = *(int *)outdata.dptr;
1890         TALLOC_FREE(outdata.dptr);
1891
1892         if (!result->persistent) {
1893                 ret = ctdb_async_ctx_init(NULL, messaging_tevent_context(msg_ctx));
1894                 if (ret != 0) {
1895                         DBG_ERR("ctdb_async_ctx_init failed: %s\n", strerror(ret));
1896                         TALLOC_FREE(result);
1897                         return NULL;
1898                 }
1899         }
1900
1901         if (!result->persistent &&
1902             (dbwrap_flags & DBWRAP_FLAG_OPTIMIZE_READONLY_ACCESS))
1903         {
1904                 TDB_DATA indata;
1905
1906                 indata = make_tdb_data((uint8_t *)&db_ctdb->db_id,
1907                                        sizeof(db_ctdb->db_id));
1908
1909                 ret = ctdbd_control_local(
1910                         messaging_ctdb_connection(),
1911                         CTDB_CONTROL_SET_DB_READONLY, 0, 0,
1912                         indata, NULL, NULL, &cstatus);
1913                 if ((ret != 0) || (cstatus != 0)) {
1914                         DEBUG(1, ("CTDB_CONTROL_SET_DB_READONLY failed: "
1915                                   "%s, %"PRIi32"\n", strerror(ret), cstatus));
1916                         TALLOC_FREE(result);
1917                         return NULL;
1918                 }
1919         }
1920
1921         lp_ctx = loadparm_init_s3(db_path, loadparm_s3_helpers());
1922
1923         if (hash_size == 0) {
1924                 hash_size = lpcfg_tdb_hash_size(lp_ctx, db_path);
1925         }
1926
1927         db_ctdb->wtdb = tdb_wrap_open(db_ctdb, db_path, hash_size,
1928                                       lpcfg_tdb_flags(lp_ctx, tdb_flags),
1929                                       O_RDWR, 0);
1930         talloc_unlink(db_path, lp_ctx);
1931         if (db_ctdb->wtdb == NULL) {
1932                 DEBUG(0, ("Could not open tdb %s: %s\n", db_path, strerror(errno)));
1933                 TALLOC_FREE(result);
1934                 return NULL;
1935         }
1936         talloc_free(db_path);
1937
1938         /* honor permissions if user has specified O_CREAT */
1939         if (open_flags & O_CREAT) {
1940                 int fd;
1941                 fd = tdb_fd(db_ctdb->wtdb->tdb);
1942                 ret = fchmod(fd, mode);
1943                 if (ret == -1) {
1944                         DBG_WARNING("fchmod failed: %s\n",
1945                                     strerror(errno));
1946                         TALLOC_FREE(result);
1947                         return NULL;
1948                 }
1949         }
1950
1951         if (result->persistent) {
1952                 db_ctdb->lock_ctx = g_lock_ctx_init(db_ctdb, msg_ctx);
1953                 if (db_ctdb->lock_ctx == NULL) {
1954                         DEBUG(0, ("g_lock_ctx_init failed\n"));
1955                         TALLOC_FREE(result);
1956                         return NULL;
1957                 }
1958         }
1959
1960         db_ctdb->warn_unlock_msecs = lp_parm_int(-1, "ctdb",
1961                                                  "unlock_warn_threshold", 5);
1962         db_ctdb->warn_migrate_attempts = lp_parm_int(-1, "ctdb",
1963                                                      "migrate_attempts", 10);
1964         db_ctdb->warn_migrate_msecs = lp_parm_int(-1, "ctdb",
1965                                                   "migrate_duration", 5000);
1966         db_ctdb->warn_locktime_msecs = lp_ctdb_locktime_warn_threshold();
1967
1968         result->private_data = (void *)db_ctdb;
1969         result->fetch_locked = db_ctdb_fetch_locked;
1970         result->parse_record = db_ctdb_parse_record;
1971         result->parse_record_send = db_ctdb_parse_record_send;
1972         result->parse_record_recv = db_ctdb_parse_record_recv;
1973         result->traverse = db_ctdb_traverse;
1974         result->traverse_read = db_ctdb_traverse_read;
1975         result->get_seqnum = db_ctdb_get_seqnum;
1976         result->transaction_start = db_ctdb_transaction_start;
1977         result->transaction_commit = db_ctdb_transaction_commit;
1978         result->transaction_cancel = db_ctdb_transaction_cancel;
1979         result->id = db_ctdb_id;
1980
1981         DEBUG(3,("db_open_ctdb: opened database '%s' with dbid 0x%x\n",
1982                  name, db_ctdb->db_id));
1983
1984         return result;
1985 }