Start to rework provision for LDAP backends
[samba.git] / source4 / ntvfs / common / opendb_tdb.c
1 /* 
2    Unix SMB/CIFS implementation.
3
4    Copyright (C) Andrew Tridgell 2004
5    Copyright (C) Stefan Metzmacher 2008
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   this is the open files database, tdb backend. It implements shared
23   storage of what files are open between server instances, and
24   implements the rules of shared access to files.
25
26   The caller needs to provide a file_key, which specifies what file
27   they are talking about. This needs to be a unique key across all
28   filesystems, and is usually implemented in terms of a device/inode
29   pair.
30
31   Before any operations can be performed the caller needs to establish
32   a lock on the record associated with file_key. That is done by
33   calling odb_lock(). The caller releases this lock by calling
34   talloc_free() on the returned handle.
35
36   All other operations on a record are done by passing the odb_lock()
37   handle back to this module. The handle contains internal
38   information about what file_key is being operated on.
39 */
40
41 #include "includes.h"
42 #include "system/filesys.h"
43 #include "lib/tdb/include/tdb.h"
44 #include "messaging/messaging.h"
45 #include "tdb_wrap.h"
46 #include "lib/messaging/irpc.h"
47 #include "librpc/gen_ndr/ndr_opendb.h"
48 #include "ntvfs/ntvfs.h"
49 #include "ntvfs/common/ntvfs_common.h"
50 #include "cluster/cluster.h"
51 #include "param/param.h"
52
53 struct odb_context {
54         struct tdb_wrap *w;
55         struct ntvfs_context *ntvfs_ctx;
56         bool oplocks;
57 };
58
59 /*
60   an odb lock handle. You must obtain one of these using odb_lock() before doing
61   any other operations. 
62 */
63 struct odb_lock {
64         struct odb_context *odb;
65         TDB_DATA key;
66 };
67
68 /*
69   Open up the openfiles.tdb database. Close it down using
70   talloc_free(). We need the messaging_ctx to allow for pending open
71   notifications.
72 */
73 static struct odb_context *odb_tdb_init(TALLOC_CTX *mem_ctx, 
74                                         struct ntvfs_context *ntvfs_ctx)
75 {
76         struct odb_context *odb;
77
78         odb = talloc(mem_ctx, struct odb_context);
79         if (odb == NULL) {
80                 return NULL;
81         }
82
83         odb->w = cluster_tdb_tmp_open(odb, ntvfs_ctx->lp_ctx, "openfiles.tdb", TDB_DEFAULT);
84         if (odb->w == NULL) {
85                 talloc_free(odb);
86                 return NULL;
87         }
88
89         odb->ntvfs_ctx = ntvfs_ctx;
90
91         /* leave oplocks disabled by default until the code is working */
92         odb->oplocks = lp_parm_bool(ntvfs_ctx->lp_ctx, NULL, "opendb", "oplocks", false);
93
94         return odb;
95 }
96
97 /*
98   destroy a lock on the database
99 */
100 static int odb_lock_destructor(struct odb_lock *lck)
101 {
102         tdb_chainunlock(lck->odb->w->tdb, lck->key);
103         return 0;
104 }
105
106 /*
107   get a lock on a entry in the odb. This call returns a lock handle,
108   which the caller should unlock using talloc_free().
109 */
110 static struct odb_lock *odb_tdb_lock(TALLOC_CTX *mem_ctx,
111                                      struct odb_context *odb, DATA_BLOB *file_key)
112 {
113         struct odb_lock *lck;
114
115         lck = talloc(mem_ctx, struct odb_lock);
116         if (lck == NULL) {
117                 return NULL;
118         }
119
120         lck->odb = talloc_reference(lck, odb);
121         lck->key.dptr = talloc_memdup(lck, file_key->data, file_key->length);
122         lck->key.dsize = file_key->length;
123         if (lck->key.dptr == NULL) {
124                 talloc_free(lck);
125                 return NULL;
126         }
127
128         if (tdb_chainlock(odb->w->tdb, lck->key) != 0) {
129                 talloc_free(lck);
130                 return NULL;
131         }
132
133         talloc_set_destructor(lck, odb_lock_destructor);
134         
135         return lck;
136 }
137
138 static DATA_BLOB odb_tdb_get_key(TALLOC_CTX *mem_ctx, struct odb_lock *lck)
139 {
140         return data_blob_talloc(mem_ctx, lck->key.dptr, lck->key.dsize);
141 }
142
143
144 /*
145   determine if two odb_entry structures conflict
146
147   return NT_STATUS_OK on no conflict
148 */
149 static NTSTATUS share_conflict(struct opendb_entry *e1,
150                                uint32_t stream_id,
151                                uint32_t share_access,
152                                uint32_t access_mask)
153 {
154         /* if either open involves no read.write or delete access then
155            it can't conflict */
156         if (!(e1->access_mask & (SEC_FILE_WRITE_DATA |
157                                  SEC_FILE_APPEND_DATA |
158                                  SEC_FILE_READ_DATA |
159                                  SEC_FILE_EXECUTE |
160                                  SEC_STD_DELETE))) {
161                 return NT_STATUS_OK;
162         }
163         if (!(access_mask & (SEC_FILE_WRITE_DATA |
164                              SEC_FILE_APPEND_DATA |
165                              SEC_FILE_READ_DATA |
166                              SEC_FILE_EXECUTE |
167                              SEC_STD_DELETE))) {
168                 return NT_STATUS_OK;
169         }
170
171         /* data IO access masks. This is skipped if the two open handles
172            are on different streams (as in that case the masks don't
173            interact) */
174         if (e1->stream_id != stream_id) {
175                 return NT_STATUS_OK;
176         }
177
178 #define CHECK_MASK(am, right, sa, share) \
179         if (((am) & (right)) && !((sa) & (share))) return NT_STATUS_SHARING_VIOLATION
180
181         CHECK_MASK(e1->access_mask, SEC_FILE_WRITE_DATA | SEC_FILE_APPEND_DATA,
182                    share_access, NTCREATEX_SHARE_ACCESS_WRITE);
183         CHECK_MASK(access_mask, SEC_FILE_WRITE_DATA | SEC_FILE_APPEND_DATA,
184                    e1->share_access, NTCREATEX_SHARE_ACCESS_WRITE);
185         
186         CHECK_MASK(e1->access_mask, SEC_FILE_READ_DATA | SEC_FILE_EXECUTE,
187                    share_access, NTCREATEX_SHARE_ACCESS_READ);
188         CHECK_MASK(access_mask, SEC_FILE_READ_DATA | SEC_FILE_EXECUTE,
189                    e1->share_access, NTCREATEX_SHARE_ACCESS_READ);
190
191         CHECK_MASK(e1->access_mask, SEC_STD_DELETE,
192                    share_access, NTCREATEX_SHARE_ACCESS_DELETE);
193         CHECK_MASK(access_mask, SEC_STD_DELETE,
194                    e1->share_access, NTCREATEX_SHARE_ACCESS_DELETE);
195 #undef CHECK_MASK
196         return NT_STATUS_OK;
197 }
198
199 /*
200   pull a record, translating from the db format to the opendb_file structure defined
201   in opendb.idl
202 */
203 static NTSTATUS odb_pull_record(struct odb_lock *lck, struct opendb_file *file)
204 {
205         struct odb_context *odb = lck->odb;
206         TDB_DATA dbuf;
207         DATA_BLOB blob;
208         enum ndr_err_code ndr_err;
209
210         dbuf = tdb_fetch(odb->w->tdb, lck->key);
211         if (dbuf.dptr == NULL) {
212                 return NT_STATUS_OBJECT_NAME_NOT_FOUND;
213         }
214
215         blob.data = dbuf.dptr;
216         blob.length = dbuf.dsize;
217
218         ndr_err = ndr_pull_struct_blob(&blob, lck, lp_iconv_convenience(lck->odb->ntvfs_ctx->lp_ctx), file, (ndr_pull_flags_fn_t)ndr_pull_opendb_file);
219         free(dbuf.dptr);
220         if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
221                 return ndr_map_error2ntstatus(ndr_err);
222         }
223
224         return NT_STATUS_OK;
225 }
226
227 /*
228   push a record, translating from the opendb_file structure defined in opendb.idl
229 */
230 static NTSTATUS odb_push_record(struct odb_lock *lck, struct opendb_file *file)
231 {
232         struct odb_context *odb = lck->odb;
233         TDB_DATA dbuf;
234         DATA_BLOB blob;
235         enum ndr_err_code ndr_err;
236         int ret;
237
238         if (file->num_entries == 0) {
239                 ret = tdb_delete(odb->w->tdb, lck->key);
240                 if (ret != 0) {
241                         return NT_STATUS_INTERNAL_DB_CORRUPTION;
242                 }
243                 return NT_STATUS_OK;
244         }
245
246         ndr_err = ndr_push_struct_blob(&blob, lck, lp_iconv_convenience(lck->odb->ntvfs_ctx->lp_ctx), file, (ndr_push_flags_fn_t)ndr_push_opendb_file);
247         if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
248                 return ndr_map_error2ntstatus(ndr_err);
249         }
250
251         dbuf.dptr = blob.data;
252         dbuf.dsize = blob.length;
253                 
254         ret = tdb_store(odb->w->tdb, lck->key, dbuf, TDB_REPLACE);
255         data_blob_free(&blob);
256         if (ret != 0) {
257                 return NT_STATUS_INTERNAL_DB_CORRUPTION;
258         }
259
260         return NT_STATUS_OK;
261 }
262
263 /*
264   send an oplock break to a client
265 */
266 static NTSTATUS odb_oplock_break_send(struct odb_context *odb,
267                                       struct opendb_entry *e,
268                                       uint8_t level)
269 {
270         NTSTATUS status;
271         struct opendb_oplock_break op_break;
272         DATA_BLOB blob;
273
274         ZERO_STRUCT(op_break);
275
276         /* tell the server handling this open file about the need to send the client
277            a break */
278         op_break.file_handle    = e->file_handle;
279         op_break.level          = level;
280
281         blob = data_blob_const(&op_break, sizeof(op_break));
282
283         status = messaging_send(odb->ntvfs_ctx->msg_ctx, e->server,
284                                 MSG_NTVFS_OPLOCK_BREAK, &blob);
285         NT_STATUS_NOT_OK_RETURN(status);
286
287         return NT_STATUS_OK;
288 }
289
290 static bool access_attributes_only(uint32_t access_mask,
291                                    uint32_t open_disposition,
292                                    bool break_to_none)
293 {
294         switch (open_disposition) {
295         case NTCREATEX_DISP_SUPERSEDE:
296         case NTCREATEX_DISP_OVERWRITE_IF:
297         case NTCREATEX_DISP_OVERWRITE:
298                 return false;
299         default:
300                 break;
301         }
302
303         if (break_to_none) {
304                 return false;
305         }
306
307 #define CHECK_MASK(m,g) ((m) && (((m) & ~(g))==0) && (((m) & (g)) != 0))
308         return CHECK_MASK(access_mask,
309                           SEC_STD_SYNCHRONIZE |
310                           SEC_FILE_READ_ATTRIBUTE |
311                           SEC_FILE_WRITE_ATTRIBUTE);
312 #undef CHECK_MASK
313 }
314
315 static NTSTATUS odb_tdb_open_can_internal(struct odb_context *odb,
316                                           const struct opendb_file *file,
317                                           uint32_t stream_id, uint32_t share_access,
318                                           uint32_t access_mask, bool delete_on_close,
319                                           uint32_t open_disposition, bool break_to_none,
320                                           bool *_attrs_only)
321 {
322         NTSTATUS status;
323         uint32_t i;
324         bool attrs_only = false;
325
326         /* see if anyone has an oplock, which we need to break */
327         for (i=0;i<file->num_entries;i++) {
328                 if (file->entries[i].oplock_level == OPLOCK_BATCH) {
329                         bool oplock_return = OPLOCK_BREAK_TO_LEVEL_II;
330                         /* if this is an attribute only access
331                          * it doesn't conflict with a BACTCH oplock
332                          * but we'll not grant the oplock below
333                          */
334                         attrs_only = access_attributes_only(access_mask,
335                                                             open_disposition,
336                                                             break_to_none);
337                         if (attrs_only) {
338                                 break;
339                         }
340                         /* a batch oplock caches close calls, which
341                            means the client application might have
342                            already closed the file. We have to allow
343                            this close to propogate by sending a oplock
344                            break request and suspending this call
345                            until the break is acknowledged or the file
346                            is closed */
347                         if (break_to_none) {
348                                 oplock_return = OPLOCK_BREAK_TO_NONE;
349                         }
350                         odb_oplock_break_send(odb, &file->entries[i],
351                                               oplock_return);
352                         return NT_STATUS_OPLOCK_NOT_GRANTED;
353                 }
354         }
355
356         if (file->delete_on_close) {
357                 /* while delete on close is set, no new opens are allowed */
358                 return NT_STATUS_DELETE_PENDING;
359         }
360
361         if (file->num_entries != 0 && delete_on_close) {
362                 return NT_STATUS_SHARING_VIOLATION;
363         }
364
365         /* check for sharing violations */
366         for (i=0;i<file->num_entries;i++) {
367                 status = share_conflict(&file->entries[i], stream_id,
368                                         share_access, access_mask);
369                 NT_STATUS_NOT_OK_RETURN(status);
370         }
371
372         /* we now know the open could succeed, but we need to check
373            for any exclusive oplocks. We can't grant a second open
374            till these are broken. Note that we check for batch oplocks
375            before checking for sharing violations, and check for
376            exclusive oplocks afterwards. */
377         for (i=0;i<file->num_entries;i++) {
378                 if (file->entries[i].oplock_level == OPLOCK_EXCLUSIVE) {
379                         bool oplock_return = OPLOCK_BREAK_TO_LEVEL_II;
380                         /* if this is an attribute only access
381                          * it doesn't conflict with an EXCLUSIVE oplock
382                          * but we'll not grant the oplock below
383                          */
384                         attrs_only = access_attributes_only(access_mask,
385                                                             open_disposition,
386                                                             break_to_none);
387                         if (attrs_only) {
388                                 break;
389                         }
390                         /*
391                          * send an oplock break to the holder of the
392                          * oplock and tell caller to retry later
393                          */
394                         if (break_to_none) {
395                                 oplock_return = OPLOCK_BREAK_TO_NONE;
396                         }
397                         odb_oplock_break_send(odb, &file->entries[i],
398                                               oplock_return);
399                         return NT_STATUS_OPLOCK_NOT_GRANTED;
400                 }
401         }
402
403         if (_attrs_only) {
404                 *_attrs_only = attrs_only;
405         }
406         return NT_STATUS_OK;
407 }
408
409 /*
410   register an open file in the open files database. This implements the share_access
411   rules
412
413   Note that the path is only used by the delete on close logic, not
414   for comparing with other filenames
415 */
416 static NTSTATUS odb_tdb_open_file(struct odb_lock *lck,
417                                   void *file_handle, const char *path,
418                                   uint32_t stream_id, uint32_t share_access,
419                                   uint32_t access_mask, bool delete_on_close,
420                                   uint32_t open_disposition, bool break_to_none,
421                                   uint32_t oplock_level, uint32_t *oplock_granted)
422 {
423         struct odb_context *odb = lck->odb;
424         struct opendb_entry e;
425         struct opendb_file file;
426         NTSTATUS status;
427         bool attrs_only = false;
428
429         if (odb->oplocks == false) {
430                 oplock_level = OPLOCK_NONE;
431         }
432
433         status = odb_pull_record(lck, &file);
434         if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
435                 /* initialise a blank structure */
436                 ZERO_STRUCT(file);
437                 file.path = path;
438         } else {
439                 NT_STATUS_NOT_OK_RETURN(status);
440         }
441
442         /* see if it conflicts */
443         status = odb_tdb_open_can_internal(odb, &file, stream_id,
444                                            share_access, access_mask,
445                                            delete_on_close, open_disposition,
446                                            break_to_none, &attrs_only);
447         NT_STATUS_NOT_OK_RETURN(status);
448
449         /* see if it conflicts */
450         e.server          = odb->ntvfs_ctx->server_id;
451         e.file_handle     = file_handle;
452         e.stream_id       = stream_id;
453         e.share_access    = share_access;
454         e.access_mask     = access_mask;
455         e.delete_on_close = delete_on_close;
456         e.oplock_level    = OPLOCK_NONE;
457
458         /*
459           possibly grant an exclusive, batch or level2 oplock
460         */
461         if (oplock_granted) {
462                 if (attrs_only) {
463                         e.oplock_level  = OPLOCK_NONE;
464                         *oplock_granted = NO_OPLOCK_RETURN;
465                 } else if (oplock_level == OPLOCK_EXCLUSIVE) {
466                         if (file.num_entries == 0) {
467                                 e.oplock_level  = OPLOCK_EXCLUSIVE;
468                                 *oplock_granted = EXCLUSIVE_OPLOCK_RETURN;
469                         } else {
470                                 e.oplock_level  = OPLOCK_LEVEL_II;
471                                 *oplock_granted = LEVEL_II_OPLOCK_RETURN;
472                         }
473                 } else if (oplock_level == OPLOCK_BATCH) {
474                         if (file.num_entries == 0) {
475                                 e.oplock_level  = OPLOCK_BATCH;
476                                 *oplock_granted = BATCH_OPLOCK_RETURN;
477                         } else {
478                                 e.oplock_level  = OPLOCK_LEVEL_II;
479                                 *oplock_granted = LEVEL_II_OPLOCK_RETURN;
480                         }
481                 } else if (oplock_level == OPLOCK_LEVEL_II) {
482                         e.oplock_level  = OPLOCK_LEVEL_II;
483                         *oplock_granted = LEVEL_II_OPLOCK_RETURN;
484                 } else {
485                         e.oplock_level  = OPLOCK_NONE;
486                         *oplock_granted = NO_OPLOCK_RETURN;
487                 }
488         }
489
490         /* it doesn't conflict, so add it to the end */
491         file.entries = talloc_realloc(lck, file.entries, struct opendb_entry, 
492                                       file.num_entries+1);
493         NT_STATUS_HAVE_NO_MEMORY(file.entries);
494
495         file.entries[file.num_entries] = e;
496         file.num_entries++;
497
498         return odb_push_record(lck, &file);
499 }
500
501
502 /*
503   register a pending open file in the open files database
504 */
505 static NTSTATUS odb_tdb_open_file_pending(struct odb_lock *lck, void *private)
506 {
507         struct odb_context *odb = lck->odb;
508         struct opendb_file file;
509         NTSTATUS status;
510                 
511         status = odb_pull_record(lck, &file);
512         NT_STATUS_NOT_OK_RETURN(status);
513
514         file.pending = talloc_realloc(lck, file.pending, struct opendb_pending, 
515                                       file.num_pending+1);
516         NT_STATUS_HAVE_NO_MEMORY(file.pending);
517
518         file.pending[file.num_pending].server = odb->ntvfs_ctx->server_id;
519         file.pending[file.num_pending].notify_ptr = private;
520
521         file.num_pending++;
522
523         return odb_push_record(lck, &file);
524 }
525
526
527 /*
528   remove a opendb entry
529 */
530 static NTSTATUS odb_tdb_close_file(struct odb_lock *lck, void *file_handle,
531                                    const char **_delete_path)
532 {
533         struct odb_context *odb = lck->odb;
534         struct opendb_file file;
535         const char *delete_path = NULL;
536         int i;
537         NTSTATUS status;
538
539         status = odb_pull_record(lck, &file);
540         NT_STATUS_NOT_OK_RETURN(status);
541
542         /* find the entry, and delete it */
543         for (i=0;i<file.num_entries;i++) {
544                 if (file_handle == file.entries[i].file_handle &&
545                     cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.entries[i].server)) {
546                         if (file.entries[i].delete_on_close) {
547                                 file.delete_on_close = true;
548                         }
549                         if (i < file.num_entries-1) {
550                                 memmove(file.entries+i, file.entries+i+1, 
551                                         (file.num_entries - (i+1)) * 
552                                         sizeof(struct opendb_entry));
553                         }
554                         break;
555                 }
556         }
557
558         if (i == file.num_entries) {
559                 return NT_STATUS_UNSUCCESSFUL;
560         }
561
562         /* send any pending notifications, removing them once sent */
563         for (i=0;i<file.num_pending;i++) {
564                 messaging_send_ptr(odb->ntvfs_ctx->msg_ctx, file.pending[i].server, 
565                                    MSG_PVFS_RETRY_OPEN, 
566                                    file.pending[i].notify_ptr);
567         }
568         file.num_pending = 0;
569
570         file.num_entries--;
571
572         if (file.num_entries == 0 && file.delete_on_close) {
573                 delete_path = talloc_strdup(lck, file.path);
574                 NT_STATUS_HAVE_NO_MEMORY(delete_path);
575         }
576
577         if (_delete_path) {
578                 *_delete_path = delete_path;
579         }
580
581         return odb_push_record(lck, &file);
582 }
583
584 /*
585   update the oplock level of the client
586 */
587 static NTSTATUS odb_tdb_update_oplock(struct odb_lock *lck, void *file_handle,
588                                       uint32_t oplock_level)
589 {
590         struct odb_context *odb = lck->odb;
591         struct opendb_file file;
592         int i;
593         NTSTATUS status;
594
595         status = odb_pull_record(lck, &file);
596         NT_STATUS_NOT_OK_RETURN(status);
597
598         /* find the entry, and update it */
599         for (i=0;i<file.num_entries;i++) {
600                 if (file_handle == file.entries[i].file_handle &&
601                     cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.entries[i].server)) {
602                         file.entries[i].oplock_level = oplock_level;
603                         break;
604                 }
605         }
606
607         if (i == file.num_entries) {
608                 return NT_STATUS_UNSUCCESSFUL;
609         }
610
611         /* send any pending notifications, removing them once sent */
612         for (i=0;i<file.num_pending;i++) {
613                 messaging_send_ptr(odb->ntvfs_ctx->msg_ctx,
614                                    file.pending[i].server,
615                                    MSG_PVFS_RETRY_OPEN,
616                                    file.pending[i].notify_ptr);
617         }
618         file.num_pending = 0;
619
620         return odb_push_record(lck, &file);
621 }
622
623 /*
624   send oplocks breaks to none to all level2 holders
625 */
626 static NTSTATUS odb_tdb_break_oplocks(struct odb_lock *lck)
627 {
628         struct odb_context *odb = lck->odb;
629         NTSTATUS status;
630         struct opendb_file file;
631         int i;
632         bool modified = true;
633
634         status = odb_pull_record(lck, &file);
635         if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
636                 return NT_STATUS_OK;
637         }
638         NT_STATUS_NOT_OK_RETURN(status);
639
640         /* see if anyone has an oplock, which we need to break */
641         for (i=0;i<file.num_entries;i++) {
642                 if (file.entries[i].oplock_level == OPLOCK_LEVEL_II) {
643                         /*
644                          * there could be multiple level2 oplocks
645                          * and we just send a break to none to all of them
646                          * without waiting for a release
647                          */
648                         odb_oplock_break_send(odb, &file.entries[i],
649                                               OPLOCK_BREAK_TO_NONE);
650                         file.entries[i].oplock_level = OPLOCK_NONE;
651                         modified = true;
652                 }
653         }
654
655         if (modified) {
656                 return odb_push_record(lck, &file);
657         }
658         return NT_STATUS_OK;
659 }
660
661 /*
662   remove a pending opendb entry
663 */
664 static NTSTATUS odb_tdb_remove_pending(struct odb_lock *lck, void *private)
665 {
666         struct odb_context *odb = lck->odb;
667         int i;
668         NTSTATUS status;
669         struct opendb_file file;
670
671         status = odb_pull_record(lck, &file);
672         NT_STATUS_NOT_OK_RETURN(status);
673
674         /* find the entry, and delete it */
675         for (i=0;i<file.num_pending;i++) {
676                 if (private == file.pending[i].notify_ptr &&
677                     cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.pending[i].server)) {
678                         if (i < file.num_pending-1) {
679                                 memmove(file.pending+i, file.pending+i+1, 
680                                         (file.num_pending - (i+1)) * 
681                                         sizeof(struct opendb_pending));
682                         }
683                         break;
684                 }
685         }
686
687         if (i == file.num_pending) {
688                 return NT_STATUS_UNSUCCESSFUL;
689         }
690
691         file.num_pending--;
692         
693         return odb_push_record(lck, &file);
694 }
695
696
697 /*
698   rename the path in a open file
699 */
700 static NTSTATUS odb_tdb_rename(struct odb_lock *lck, const char *path)
701 {
702         struct opendb_file file;
703         NTSTATUS status;
704
705         status = odb_pull_record(lck, &file);
706         if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
707                 /* not having the record at all is OK */
708                 return NT_STATUS_OK;
709         }
710         NT_STATUS_NOT_OK_RETURN(status);
711
712         file.path = path;
713         return odb_push_record(lck, &file);
714 }
715
716 /*
717   get the path of an open file
718 */
719 static NTSTATUS odb_tdb_get_path(struct odb_lock *lck, const char **path)
720 {
721         struct opendb_file file;
722         NTSTATUS status;
723
724         *path = NULL;
725
726         status = odb_pull_record(lck, &file);
727         /* we don't ignore NT_STATUS_OBJECT_NAME_NOT_FOUND here */
728         NT_STATUS_NOT_OK_RETURN(status);
729
730         *path = file.path;
731
732         return NT_STATUS_OK;
733 }
734
735 /*
736   update delete on close flag on an open file
737 */
738 static NTSTATUS odb_tdb_set_delete_on_close(struct odb_lock *lck, bool del_on_close)
739 {
740         NTSTATUS status;
741         struct opendb_file file;
742
743         status = odb_pull_record(lck, &file);
744         NT_STATUS_NOT_OK_RETURN(status);
745
746         file.delete_on_close = del_on_close;
747
748         return odb_push_record(lck, &file);
749 }
750
751 /*
752   return the current value of the delete_on_close bit, and how many
753   people still have the file open
754 */
755 static NTSTATUS odb_tdb_get_delete_on_close(struct odb_context *odb, 
756                                             DATA_BLOB *key, bool *del_on_close)
757 {
758         NTSTATUS status;
759         struct opendb_file file;
760         struct odb_lock *lck;
761
762         (*del_on_close) = false;
763
764         lck = odb_lock(odb, odb, key);
765         NT_STATUS_HAVE_NO_MEMORY(lck);
766
767         status = odb_pull_record(lck, &file);
768         if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
769                 talloc_free(lck);
770                 return NT_STATUS_OK;
771         }
772         if (!NT_STATUS_IS_OK(status)) {
773                 talloc_free(lck);
774                 return status;
775         }
776
777         (*del_on_close) = file.delete_on_close;
778
779         talloc_free(lck);
780
781         return NT_STATUS_OK;
782 }
783
784
785 /*
786   determine if a file can be opened with the given share_access,
787   create_options and access_mask
788 */
789 static NTSTATUS odb_tdb_can_open(struct odb_lock *lck,
790                                  uint32_t stream_id, uint32_t share_access,
791                                  uint32_t access_mask, bool delete_on_close,
792                                  uint32_t open_disposition, bool break_to_none)
793 {
794         struct odb_context *odb = lck->odb;
795         NTSTATUS status;
796         struct opendb_file file;
797         bool attrs_only = false;
798
799         status = odb_pull_record(lck, &file);
800         if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
801                 return NT_STATUS_OK;
802         }
803         NT_STATUS_NOT_OK_RETURN(status);
804
805         status = odb_tdb_open_can_internal(odb, &file, stream_id,
806                                            share_access, access_mask,
807                                            delete_on_close, open_disposition,
808                                            break_to_none, &attrs_only);
809         NT_STATUS_NOT_OK_RETURN(status);
810
811         return NT_STATUS_OK;
812 }
813
814
815 static const struct opendb_ops opendb_tdb_ops = {
816         .odb_init                = odb_tdb_init,
817         .odb_lock                = odb_tdb_lock,
818         .odb_get_key             = odb_tdb_get_key,
819         .odb_open_file           = odb_tdb_open_file,
820         .odb_open_file_pending   = odb_tdb_open_file_pending,
821         .odb_close_file          = odb_tdb_close_file,
822         .odb_remove_pending      = odb_tdb_remove_pending,
823         .odb_rename              = odb_tdb_rename,
824         .odb_get_path            = odb_tdb_get_path,
825         .odb_set_delete_on_close = odb_tdb_set_delete_on_close,
826         .odb_get_delete_on_close = odb_tdb_get_delete_on_close,
827         .odb_can_open            = odb_tdb_can_open,
828         .odb_update_oplock       = odb_tdb_update_oplock,
829         .odb_break_oplocks       = odb_tdb_break_oplocks
830 };
831
832
833 void odb_tdb_init_ops(void)
834 {
835         odb_set_ops(&opendb_tdb_ops);
836 }