Finish removal of iconv_convenience in public API's.
[bbaumbach/samba-autobuild/.git] / source4 / ntvfs / common / opendb_tdb.c
index 3d4a760e2e4adcb8d8e9aaa35e52d3571f586f07..944ec866314fb0ea4c6d7eb6315b489053d90cb2 100644 (file)
@@ -40,7 +40,7 @@
 
 #include "includes.h"
 #include "system/filesys.h"
-#include "lib/tdb/include/tdb.h"
+#include "../tdb/include/tdb.h"
 #include "messaging/messaging.h"
 #include "tdb_wrap.h"
 #include "lib/messaging/irpc.h"
 #include "ntvfs/common/ntvfs_common.h"
 #include "cluster/cluster.h"
 #include "param/param.h"
+#include "ntvfs/sysdep/sys_lease.h"
 
 struct odb_context {
        struct tdb_wrap *w;
        struct ntvfs_context *ntvfs_ctx;
        bool oplocks;
+       struct sys_lease_context *lease_ctx;
 };
 
 /*
@@ -63,8 +65,19 @@ struct odb_context {
 struct odb_lock {
        struct odb_context *odb;
        TDB_DATA key;
+
+       struct opendb_file file;
+
+       struct {
+               struct opendb_entry *e;
+               bool attrs_only;
+       } can_open;
 };
 
+static NTSTATUS odb_oplock_break_send(struct messaging_context *msg_ctx,
+                                     struct opendb_entry *e,
+                                     uint8_t level);
+
 /*
   Open up the openfiles.tdb database. Close it down using
   talloc_free(). We need the messaging_ctx to allow for pending open
@@ -88,8 +101,12 @@ static struct odb_context *odb_tdb_init(TALLOC_CTX *mem_ctx,
 
        odb->ntvfs_ctx = ntvfs_ctx;
 
-       /* leave oplocks disabled by default until the code is working */
-       odb->oplocks = lp_parm_bool(ntvfs_ctx->lp_ctx, NULL, "opendb", "oplocks", false);
+       odb->oplocks = share_bool_option(ntvfs_ctx->config, SHARE_OPLOCKS, SHARE_OPLOCKS_DEFAULT);
+
+       odb->lease_ctx = sys_lease_context_create(ntvfs_ctx->config, odb,
+                                                 ntvfs_ctx->event_ctx,
+                                                 ntvfs_ctx->msg_ctx,
+                                                 odb_oplock_break_send);
 
        return odb;
 }
@@ -103,6 +120,8 @@ static int odb_lock_destructor(struct odb_lock *lck)
        return 0;
 }
 
+static NTSTATUS odb_pull_record(struct odb_lock *lck, struct opendb_file *file);
+
 /*
   get a lock on a entry in the odb. This call returns a lock handle,
   which the caller should unlock using talloc_free().
@@ -111,6 +130,7 @@ static struct odb_lock *odb_tdb_lock(TALLOC_CTX *mem_ctx,
                                     struct odb_context *odb, DATA_BLOB *file_key)
 {
        struct odb_lock *lck;
+       NTSTATUS status;
 
        lck = talloc(mem_ctx, struct odb_lock);
        if (lck == NULL) {
@@ -130,7 +150,18 @@ static struct odb_lock *odb_tdb_lock(TALLOC_CTX *mem_ctx,
                return NULL;
        }
 
+       ZERO_STRUCT(lck->can_open);
+
        talloc_set_destructor(lck, odb_lock_destructor);
+
+       status = odb_pull_record(lck, &lck->file);
+       if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
+               /* initialise a blank structure */
+               ZERO_STRUCT(lck->file);
+       } else if (!NT_STATUS_IS_OK(status)) {
+               talloc_free(lck);
+               return NULL;
+       }
        
        return lck;
 }
@@ -146,7 +177,10 @@ static DATA_BLOB odb_tdb_get_key(TALLOC_CTX *mem_ctx, struct odb_lock *lck)
 
   return NT_STATUS_OK on no conflict
 */
-static NTSTATUS share_conflict(struct opendb_entry *e1, struct opendb_entry *e2)
+static NTSTATUS share_conflict(struct opendb_entry *e1,
+                              uint32_t stream_id,
+                              uint32_t share_access,
+                              uint32_t access_mask)
 {
        /* if either open involves no read.write or delete access then
           it can't conflict */
@@ -157,18 +191,18 @@ static NTSTATUS share_conflict(struct opendb_entry *e1, struct opendb_entry *e2)
                                 SEC_STD_DELETE))) {
                return NT_STATUS_OK;
        }
-       if (!(e2->access_mask & (SEC_FILE_WRITE_DATA |
-                                SEC_FILE_APPEND_DATA |
-                                SEC_FILE_READ_DATA |
-                                SEC_FILE_EXECUTE |
-                                SEC_STD_DELETE))) {
+       if (!(access_mask & (SEC_FILE_WRITE_DATA |
+                            SEC_FILE_APPEND_DATA |
+                            SEC_FILE_READ_DATA |
+                            SEC_FILE_EXECUTE |
+                            SEC_STD_DELETE))) {
                return NT_STATUS_OK;
        }
 
        /* data IO access masks. This is skipped if the two open handles
           are on different streams (as in that case the masks don't
           interact) */
-       if (e1->stream_id != e2->stream_id) {
+       if (e1->stream_id != stream_id) {
                return NT_STATUS_OK;
        }
 
@@ -176,20 +210,20 @@ static NTSTATUS share_conflict(struct opendb_entry *e1, struct opendb_entry *e2)
        if (((am) & (right)) && !((sa) & (share))) return NT_STATUS_SHARING_VIOLATION
 
        CHECK_MASK(e1->access_mask, SEC_FILE_WRITE_DATA | SEC_FILE_APPEND_DATA,
-                  e2->share_access, NTCREATEX_SHARE_ACCESS_WRITE);
-       CHECK_MASK(e2->access_mask, SEC_FILE_WRITE_DATA | SEC_FILE_APPEND_DATA,
+                  share_access, NTCREATEX_SHARE_ACCESS_WRITE);
+       CHECK_MASK(access_mask, SEC_FILE_WRITE_DATA | SEC_FILE_APPEND_DATA,
                   e1->share_access, NTCREATEX_SHARE_ACCESS_WRITE);
        
        CHECK_MASK(e1->access_mask, SEC_FILE_READ_DATA | SEC_FILE_EXECUTE,
-                  e2->share_access, NTCREATEX_SHARE_ACCESS_READ);
-       CHECK_MASK(e2->access_mask, SEC_FILE_READ_DATA | SEC_FILE_EXECUTE,
+                  share_access, NTCREATEX_SHARE_ACCESS_READ);
+       CHECK_MASK(access_mask, SEC_FILE_READ_DATA | SEC_FILE_EXECUTE,
                   e1->share_access, NTCREATEX_SHARE_ACCESS_READ);
 
        CHECK_MASK(e1->access_mask, SEC_STD_DELETE,
-                  e2->share_access, NTCREATEX_SHARE_ACCESS_DELETE);
-       CHECK_MASK(e2->access_mask, SEC_STD_DELETE,
+                  share_access, NTCREATEX_SHARE_ACCESS_DELETE);
+       CHECK_MASK(access_mask, SEC_STD_DELETE,
                   e1->share_access, NTCREATEX_SHARE_ACCESS_DELETE);
-
+#undef CHECK_MASK
        return NT_STATUS_OK;
 }
 
@@ -212,7 +246,7 @@ static NTSTATUS odb_pull_record(struct odb_lock *lck, struct opendb_file *file)
        blob.data = dbuf.dptr;
        blob.length = dbuf.dsize;
 
-       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);
+       ndr_err = ndr_pull_struct_blob(&blob, lck, file, (ndr_pull_flags_fn_t)ndr_pull_opendb_file);
        free(dbuf.dptr);
        if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
                return ndr_map_error2ntstatus(ndr_err);
@@ -240,7 +274,7 @@ static NTSTATUS odb_push_record(struct odb_lock *lck, struct opendb_file *file)
                return NT_STATUS_OK;
        }
 
-       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);
+       ndr_err = ndr_push_struct_blob(&blob, lck, file, (ndr_push_flags_fn_t)ndr_push_opendb_file);
        if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
                return ndr_map_error2ntstatus(ndr_err);
        }
@@ -260,7 +294,7 @@ static NTSTATUS odb_push_record(struct odb_lock *lck, struct opendb_file *file)
 /*
   send an oplock break to a client
 */
-static NTSTATUS odb_oplock_break_send(struct odb_context *odb,
+static NTSTATUS odb_oplock_break_send(struct messaging_context *msg_ctx,
                                      struct opendb_entry *e,
                                      uint8_t level)
 {
@@ -277,57 +311,63 @@ static NTSTATUS odb_oplock_break_send(struct odb_context *odb,
 
        blob = data_blob_const(&op_break, sizeof(op_break));
 
-       status = messaging_send(odb->ntvfs_ctx->msg_ctx, e->server,
+       status = messaging_send(msg_ctx, e->server,
                                MSG_NTVFS_OPLOCK_BREAK, &blob);
        NT_STATUS_NOT_OK_RETURN(status);
 
        return NT_STATUS_OK;
 }
 
-/*
-  register an open file in the open files database. This implements the share_access
-  rules
-
-  Note that the path is only used by the delete on close logic, not
-  for comparing with other filenames
-*/
-static NTSTATUS odb_tdb_open_file(struct odb_lock *lck, void *file_handle,
-                                 uint32_t stream_id, uint32_t share_access, 
-                                 uint32_t access_mask, bool delete_on_close,
-                                 const char *path, 
-                                 uint32_t oplock_level, uint32_t *oplock_granted)
+static bool access_attributes_only(uint32_t access_mask,
+                                  uint32_t open_disposition,
+                                  bool break_to_none)
 {
-       struct odb_context *odb = lck->odb;
-       struct opendb_entry e;
-       int i;
-       struct opendb_file file;
-       NTSTATUS status;
-
-       if (odb->oplocks == false) {
-               oplock_level = OPLOCK_NONE;
+       switch (open_disposition) {
+       case NTCREATEX_DISP_SUPERSEDE:
+       case NTCREATEX_DISP_OVERWRITE_IF:
+       case NTCREATEX_DISP_OVERWRITE:
+               return false;
+       default:
+               break;
        }
 
-       status = odb_pull_record(lck, &file);
-       if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
-               /* initialise a blank structure */
-               ZERO_STRUCT(file);
-               file.path = path;
-       } else {
-               NT_STATUS_NOT_OK_RETURN(status);
+       if (break_to_none) {
+               return false;
        }
 
-       /* see if it conflicts */
-       e.server          = odb->ntvfs_ctx->server_id;
-       e.file_handle     = file_handle;
-       e.stream_id       = stream_id;
-       e.share_access    = share_access;
-       e.access_mask     = access_mask;
-       e.delete_on_close = delete_on_close;
-       e.oplock_level    = OPLOCK_NONE;
-               
+#define CHECK_MASK(m,g) ((m) && (((m) & ~(g))==0) && (((m) & (g)) != 0))
+       return CHECK_MASK(access_mask,
+                         SEC_STD_SYNCHRONIZE |
+                         SEC_FILE_READ_ATTRIBUTE |
+                         SEC_FILE_WRITE_ATTRIBUTE);
+#undef CHECK_MASK
+}
+
+static NTSTATUS odb_tdb_open_can_internal(struct odb_context *odb,
+                                         const struct opendb_file *file,
+                                         uint32_t stream_id, uint32_t share_access,
+                                         uint32_t access_mask, bool delete_on_close,
+                                         uint32_t open_disposition, bool break_to_none,
+                                         bool *_attrs_only)
+{
+       NTSTATUS status;
+       uint32_t i;
+       bool attrs_only = false;
+
        /* see if anyone has an oplock, which we need to break */
-       for (i=0;i<file.num_entries;i++) {
-               if (file.entries[i].oplock_level == OPLOCK_BATCH) {
+       for (i=0;i<file->num_entries;i++) {
+               if (file->entries[i].oplock_level == OPLOCK_BATCH) {
+                       bool oplock_return = OPLOCK_BREAK_TO_LEVEL_II;
+                       /* if this is an attribute only access
+                        * it doesn't conflict with a BACTCH oplock
+                        * but we'll not grant the oplock below
+                        */
+                       attrs_only = access_attributes_only(access_mask,
+                                                           open_disposition,
+                                                           break_to_none);
+                       if (attrs_only) {
+                               break;
+                       }
                        /* a batch oplock caches close calls, which
                           means the client application might have
                           already closed the file. We have to allow
@@ -335,21 +375,30 @@ static NTSTATUS odb_tdb_open_file(struct odb_lock *lck, void *file_handle,
                           break request and suspending this call
                           until the break is acknowledged or the file
                           is closed */
-                       odb_oplock_break_send(odb, &file.entries[i],
-                                             OPLOCK_BREAK_TO_LEVEL_II/*TODO*/);
+                       if (break_to_none ||
+                           !file->entries[i].allow_level_II_oplock) {
+                               oplock_return = OPLOCK_BREAK_TO_NONE;
+                       }
+                       odb_oplock_break_send(odb->ntvfs_ctx->msg_ctx,
+                                             &file->entries[i],
+                                             oplock_return);
                        return NT_STATUS_OPLOCK_NOT_GRANTED;
                }
        }
 
-       if (file.delete_on_close || 
-           (file.num_entries != 0 && delete_on_close)) {
+       if (file->delete_on_close) {
                /* while delete on close is set, no new opens are allowed */
                return NT_STATUS_DELETE_PENDING;
        }
 
+       if (file->num_entries != 0 && delete_on_close) {
+               return NT_STATUS_SHARING_VIOLATION;
+       }
+
        /* check for sharing violations */
-       for (i=0;i<file.num_entries;i++) {
-               status = share_conflict(&file.entries[i], &e);
+       for (i=0;i<file->num_entries;i++) {
+               status = share_conflict(&file->entries[i], stream_id,
+                                       share_access, access_mask);
                NT_STATUS_NOT_OK_RETURN(status);
        }
 
@@ -358,110 +407,228 @@ static NTSTATUS odb_tdb_open_file(struct odb_lock *lck, void *file_handle,
           till these are broken. Note that we check for batch oplocks
           before checking for sharing violations, and check for
           exclusive oplocks afterwards. */
-       for (i=0;i<file.num_entries;i++) {
-               if (file.entries[i].oplock_level == OPLOCK_EXCLUSIVE) {
-                       odb_oplock_break_send(odb, &file.entries[i],
-                                             OPLOCK_BREAK_TO_NONE/*TODO*/);
+       for (i=0;i<file->num_entries;i++) {
+               if (file->entries[i].oplock_level == OPLOCK_EXCLUSIVE) {
+                       bool oplock_return = OPLOCK_BREAK_TO_LEVEL_II;
+                       /* if this is an attribute only access
+                        * it doesn't conflict with an EXCLUSIVE oplock
+                        * but we'll not grant the oplock below
+                        */
+                       attrs_only = access_attributes_only(access_mask,
+                                                           open_disposition,
+                                                           break_to_none);
+                       if (attrs_only) {
+                               break;
+                       }
+                       /*
+                        * send an oplock break to the holder of the
+                        * oplock and tell caller to retry later
+                        */
+                       if (break_to_none ||
+                           !file->entries[i].allow_level_II_oplock) {
+                               oplock_return = OPLOCK_BREAK_TO_NONE;
+                       }
+                       odb_oplock_break_send(odb->ntvfs_ctx->msg_ctx,
+                                             &file->entries[i],
+                                             oplock_return);
                        return NT_STATUS_OPLOCK_NOT_GRANTED;
                }
        }
 
+       if (_attrs_only) {
+               *_attrs_only = attrs_only;
+       }
+       return NT_STATUS_OK;
+}
+
+/*
+  register an open file in the open files database.
+  The share_access rules are implemented by odb_can_open()
+  and it's needed to call odb_can_open() before
+  odb_open_file() otherwise NT_STATUS_INTERNAL_ERROR is returned
+
+  Note that the path is only used by the delete on close logic, not
+  for comparing with other filenames
+*/
+static NTSTATUS odb_tdb_open_file(struct odb_lock *lck,
+                                 void *file_handle, const char *path,
+                                 int *fd, NTTIME open_write_time,
+                                 bool allow_level_II_oplock,
+                                 uint32_t oplock_level, uint32_t *oplock_granted)
+{
+       struct odb_context *odb = lck->odb;
+
+       if (!lck->can_open.e) {
+               return NT_STATUS_INTERNAL_ERROR;
+       }
+
+       if (odb->oplocks == false) {
+               oplock_level = OPLOCK_NONE;
+       }
+
+       if (!oplock_granted) {
+               oplock_level = OPLOCK_NONE;
+       }
+
+       if (lck->file.path == NULL) {
+               lck->file.path = talloc_strdup(lck, path);
+               NT_STATUS_HAVE_NO_MEMORY(lck->file.path);
+       }
+
+       if (lck->file.open_write_time == 0) {
+               lck->file.open_write_time = open_write_time;
+       }
+
        /*
-         possibly grant an exclusive or batch oplock if this is the only client
-         with the file open. We don't yet grant levelII oplocks.
+         possibly grant an exclusive, batch or level2 oplock
        */
-       if (oplock_granted != NULL) {
-               if ((oplock_level == OPLOCK_BATCH ||
-                    oplock_level == OPLOCK_EXCLUSIVE) &&
-                   file.num_entries == 0) {
-                       (*oplock_granted) = oplock_level;
+       if (lck->can_open.attrs_only) {
+               oplock_level    = OPLOCK_NONE;
+       } else if (oplock_level == OPLOCK_EXCLUSIVE) {
+               if (lck->file.num_entries == 0) {
+                       oplock_level    = OPLOCK_EXCLUSIVE;
+               } else if (allow_level_II_oplock) {
+                       oplock_level    = OPLOCK_LEVEL_II;
+               } else {
+                       oplock_level    = OPLOCK_NONE;
+               }
+       } else if (oplock_level == OPLOCK_BATCH) {
+               if (lck->file.num_entries == 0) {
+                       oplock_level    = OPLOCK_BATCH;
+               } else if (allow_level_II_oplock) {
+                       oplock_level    = OPLOCK_LEVEL_II;
+               } else {
+                       oplock_level    = OPLOCK_NONE;
+               }
+       } else if (oplock_level == OPLOCK_LEVEL_II) {
+               oplock_level    = OPLOCK_LEVEL_II;
+       } else {
+               oplock_level    = OPLOCK_NONE;
+       }
+
+       lck->can_open.e->file_handle            = file_handle;
+       lck->can_open.e->fd                     = fd;
+       lck->can_open.e->allow_level_II_oplock  = allow_level_II_oplock;
+       lck->can_open.e->oplock_level           = oplock_level;
+
+       if (odb->lease_ctx && fd) {
+               NTSTATUS status;
+               status = sys_lease_setup(odb->lease_ctx, lck->can_open.e);
+               NT_STATUS_NOT_OK_RETURN(status);
+       }
+
+       if (oplock_granted) {
+               if (lck->can_open.e->oplock_level == OPLOCK_EXCLUSIVE) {
+                       *oplock_granted = EXCLUSIVE_OPLOCK_RETURN;
+               } else if (lck->can_open.e->oplock_level == OPLOCK_BATCH) {
+                       *oplock_granted = BATCH_OPLOCK_RETURN;
+               } else if (lck->can_open.e->oplock_level == OPLOCK_LEVEL_II) {
+                       *oplock_granted = LEVEL_II_OPLOCK_RETURN;
                } else {
-                       (*oplock_granted) = OPLOCK_NONE;
+                       *oplock_granted = NO_OPLOCK_RETURN;
                }
-               e.oplock_level = (*oplock_granted);
        }
 
        /* it doesn't conflict, so add it to the end */
-       file.entries = talloc_realloc(lck, file.entries, struct opendb_entry, 
-                                     file.num_entries+1);
-       NT_STATUS_HAVE_NO_MEMORY(file.entries);
+       lck->file.entries = talloc_realloc(lck, lck->file.entries,
+                                          struct opendb_entry,
+                                          lck->file.num_entries+1);
+       NT_STATUS_HAVE_NO_MEMORY(lck->file.entries);
 
-       file.entries[file.num_entries] = e;
-       file.num_entries++;
+       lck->file.entries[lck->file.num_entries] = *lck->can_open.e;
+       lck->file.num_entries++;
 
-       return odb_push_record(lck, &file);
+       talloc_free(lck->can_open.e);
+       lck->can_open.e = NULL;
+
+       return odb_push_record(lck, &lck->file);
 }
 
 
 /*
   register a pending open file in the open files database
 */
-static NTSTATUS odb_tdb_open_file_pending(struct odb_lock *lck, void *private)
+static NTSTATUS odb_tdb_open_file_pending(struct odb_lock *lck, void *private_data)
 {
        struct odb_context *odb = lck->odb;
-       struct opendb_file file;
-       NTSTATUS status;
-               
-       status = odb_pull_record(lck, &file);
-       NT_STATUS_NOT_OK_RETURN(status);
 
-       file.pending = talloc_realloc(lck, file.pending, struct opendb_pending, 
-                                     file.num_pending+1);
-       NT_STATUS_HAVE_NO_MEMORY(file.pending);
+       if (lck->file.path == NULL) {
+               return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+       }
+
+       lck->file.pending = talloc_realloc(lck, lck->file.pending,
+                                          struct opendb_pending,
+                                          lck->file.num_pending+1);
+       NT_STATUS_HAVE_NO_MEMORY(lck->file.pending);
 
-       file.pending[file.num_pending].server = odb->ntvfs_ctx->server_id;
-       file.pending[file.num_pending].notify_ptr = private;
+       lck->file.pending[lck->file.num_pending].server = odb->ntvfs_ctx->server_id;
+       lck->file.pending[lck->file.num_pending].notify_ptr = private_data;
 
-       file.num_pending++;
+       lck->file.num_pending++;
 
-       return odb_push_record(lck, &file);
+       return odb_push_record(lck, &lck->file);
 }
 
 
 /*
   remove a opendb entry
 */
-static NTSTATUS odb_tdb_close_file(struct odb_lock *lck, void *file_handle)
+static NTSTATUS odb_tdb_close_file(struct odb_lock *lck, void *file_handle,
+                                  const char **_delete_path)
 {
        struct odb_context *odb = lck->odb;
-       struct opendb_file file;
+       const char *delete_path = NULL;
        int i;
-       NTSTATUS status;
 
-       status = odb_pull_record(lck, &file);
-       NT_STATUS_NOT_OK_RETURN(status);
+       if (lck->file.path == NULL) {
+               return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+       }
 
        /* find the entry, and delete it */
-       for (i=0;i<file.num_entries;i++) {
-               if (file_handle == file.entries[i].file_handle &&
-                   cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.entries[i].server)) {
-                       if (file.entries[i].delete_on_close) {
-                               file.delete_on_close = true;
+       for (i=0;i<lck->file.num_entries;i++) {
+               if (file_handle == lck->file.entries[i].file_handle &&
+                   cluster_id_equal(&odb->ntvfs_ctx->server_id, &lck->file.entries[i].server)) {
+                       if (lck->file.entries[i].delete_on_close) {
+                               lck->file.delete_on_close = true;
+                       }
+                       if (odb->lease_ctx && lck->file.entries[i].fd) {
+                               NTSTATUS status;
+                               status = sys_lease_remove(odb->lease_ctx, &lck->file.entries[i]);
+                               NT_STATUS_NOT_OK_RETURN(status);
                        }
-                       if (i < file.num_entries-1) {
-                               memmove(file.entries+i, file.entries+i+1, 
-                                       (file.num_entries - (i+1)) * 
+                       if (i < lck->file.num_entries-1) {
+                               memmove(lck->file.entries+i, lck->file.entries+i+1,
+                                       (lck->file.num_entries - (i+1)) *
                                        sizeof(struct opendb_entry));
                        }
                        break;
                }
        }
 
-       if (i == file.num_entries) {
+       if (i == lck->file.num_entries) {
                return NT_STATUS_UNSUCCESSFUL;
        }
 
        /* send any pending notifications, removing them once sent */
-       for (i=0;i<file.num_pending;i++) {
-               messaging_send_ptr(odb->ntvfs_ctx->msg_ctx, file.pending[i].server, 
-                                  MSG_PVFS_RETRY_OPEN, 
-                                  file.pending[i].notify_ptr);
+       for (i=0;i<lck->file.num_pending;i++) {
+               messaging_send_ptr(odb->ntvfs_ctx->msg_ctx,
+                                  lck->file.pending[i].server,
+                                  MSG_PVFS_RETRY_OPEN,
+                                  lck->file.pending[i].notify_ptr);
        }
-       file.num_pending = 0;
+       lck->file.num_pending = 0;
 
-       file.num_entries--;
-       
-       return odb_push_record(lck, &file);
+       lck->file.num_entries--;
+
+       if (lck->file.num_entries == 0 && lck->file.delete_on_close) {
+               delete_path = lck->file.path;
+       }
+
+       if (_delete_path) {
+               *_delete_path = delete_path;
+       }
+
+       return odb_push_record(lck, &lck->file);
 }
 
 /*
@@ -471,72 +638,107 @@ static NTSTATUS odb_tdb_update_oplock(struct odb_lock *lck, void *file_handle,
                                      uint32_t oplock_level)
 {
        struct odb_context *odb = lck->odb;
-       struct opendb_file file;
        int i;
-       NTSTATUS status;
 
-       status = odb_pull_record(lck, &file);
-       NT_STATUS_NOT_OK_RETURN(status);
+       if (lck->file.path == NULL) {
+               return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+       }
 
        /* find the entry, and update it */
-       for (i=0;i<file.num_entries;i++) {
-               if (file_handle == file.entries[i].file_handle &&
-                   cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.entries[i].server)) {
-                       file.entries[i].oplock_level = oplock_level;
+       for (i=0;i<lck->file.num_entries;i++) {
+               if (file_handle == lck->file.entries[i].file_handle &&
+                   cluster_id_equal(&odb->ntvfs_ctx->server_id, &lck->file.entries[i].server)) {
+                       lck->file.entries[i].oplock_level = oplock_level;
+
+                       if (odb->lease_ctx && lck->file.entries[i].fd) {
+                               NTSTATUS status;
+                               status = sys_lease_update(odb->lease_ctx, &lck->file.entries[i]);
+                               NT_STATUS_NOT_OK_RETURN(status);
+                       }
+
                        break;
                }
        }
 
-       if (i == file.num_entries) {
+       if (i == lck->file.num_entries) {
                return NT_STATUS_UNSUCCESSFUL;
        }
 
        /* send any pending notifications, removing them once sent */
-       for (i=0;i<file.num_pending;i++) {
+       for (i=0;i<lck->file.num_pending;i++) {
                messaging_send_ptr(odb->ntvfs_ctx->msg_ctx,
-                                  file.pending[i].server,
+                                  lck->file.pending[i].server,
                                   MSG_PVFS_RETRY_OPEN,
-                                  file.pending[i].notify_ptr);
+                                  lck->file.pending[i].notify_ptr);
        }
-       file.num_pending = 0;
+       lck->file.num_pending = 0;
 
-       return odb_push_record(lck, &file);
+       return odb_push_record(lck, &lck->file);
 }
 
+/*
+  send oplocks breaks to none to all level2 holders
+*/
+static NTSTATUS odb_tdb_break_oplocks(struct odb_lock *lck)
+{
+       struct odb_context *odb = lck->odb;
+       int i;
+       bool modified = false;
+
+       /* see if anyone has an oplock, which we need to break */
+       for (i=0;i<lck->file.num_entries;i++) {
+               if (lck->file.entries[i].oplock_level == OPLOCK_LEVEL_II) {
+                       /*
+                        * there could be multiple level2 oplocks
+                        * and we just send a break to none to all of them
+                        * without waiting for a release
+                        */
+                       odb_oplock_break_send(odb->ntvfs_ctx->msg_ctx,
+                                             &lck->file.entries[i],
+                                             OPLOCK_BREAK_TO_NONE);
+                       lck->file.entries[i].oplock_level = OPLOCK_NONE;
+                       modified = true;
+               }
+       }
+
+       if (modified) {
+               return odb_push_record(lck, &lck->file);
+       }
+       return NT_STATUS_OK;
+}
 
 /*
   remove a pending opendb entry
 */
-static NTSTATUS odb_tdb_remove_pending(struct odb_lock *lck, void *private)
+static NTSTATUS odb_tdb_remove_pending(struct odb_lock *lck, void *private_data)
 {
        struct odb_context *odb = lck->odb;
        int i;
-       NTSTATUS status;
-       struct opendb_file file;
 
-       status = odb_pull_record(lck, &file);
-       NT_STATUS_NOT_OK_RETURN(status);
+       if (lck->file.path == NULL) {
+               return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+       }
 
        /* find the entry, and delete it */
-       for (i=0;i<file.num_pending;i++) {
-               if (private == file.pending[i].notify_ptr &&
-                   cluster_id_equal(&odb->ntvfs_ctx->server_id, &file.pending[i].server)) {
-                       if (i < file.num_pending-1) {
-                               memmove(file.pending+i, file.pending+i+1, 
-                                       (file.num_pending - (i+1)) * 
+       for (i=0;i<lck->file.num_pending;i++) {
+               if (private_data == lck->file.pending[i].notify_ptr &&
+                   cluster_id_equal(&odb->ntvfs_ctx->server_id, &lck->file.pending[i].server)) {
+                       if (i < lck->file.num_pending-1) {
+                               memmove(lck->file.pending+i, lck->file.pending+i+1,
+                                       (lck->file.num_pending - (i+1)) *
                                        sizeof(struct opendb_pending));
                        }
                        break;
                }
        }
 
-       if (i == file.num_pending) {
+       if (i == lck->file.num_pending) {
                return NT_STATUS_UNSUCCESSFUL;
        }
 
-       file.num_pending--;
+       lck->file.num_pending--;
        
-       return odb_push_record(lck, &file);
+       return odb_push_record(lck, &lck->file);
 }
 
 
@@ -545,18 +747,32 @@ static NTSTATUS odb_tdb_remove_pending(struct odb_lock *lck, void *private)
 */
 static NTSTATUS odb_tdb_rename(struct odb_lock *lck, const char *path)
 {
-       struct opendb_file file;
-       NTSTATUS status;
-
-       status = odb_pull_record(lck, &file);
-       if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
+       if (lck->file.path == NULL) {
                /* not having the record at all is OK */
                return NT_STATUS_OK;
        }
-       NT_STATUS_NOT_OK_RETURN(status);
 
-       file.path = path;
-       return odb_push_record(lck, &file);
+       lck->file.path = talloc_strdup(lck, path);
+       NT_STATUS_HAVE_NO_MEMORY(lck->file.path);
+
+       return odb_push_record(lck, &lck->file);
+}
+
+/*
+  get the path of an open file
+*/
+static NTSTATUS odb_tdb_get_path(struct odb_lock *lck, const char **path)
+{
+       *path = NULL;
+
+       /* we don't ignore NT_STATUS_OBJECT_NAME_NOT_FOUND here */
+       if (lck->file.path == NULL) {
+               return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+       }
+
+       *path = lck->file.path;
+
+       return NT_STATUS_OK;
 }
 
 /*
@@ -564,52 +780,61 @@ static NTSTATUS odb_tdb_rename(struct odb_lock *lck, const char *path)
 */
 static NTSTATUS odb_tdb_set_delete_on_close(struct odb_lock *lck, bool del_on_close)
 {
-       NTSTATUS status;
-       struct opendb_file file;
+       if (lck->file.path == NULL) {
+               return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+       }
 
-       status = odb_pull_record(lck, &file);
-       NT_STATUS_NOT_OK_RETURN(status);
+       lck->file.delete_on_close = del_on_close;
+
+       return odb_push_record(lck, &lck->file);
+}
+
+/*
+  update the write time on an open file
+*/
+static NTSTATUS odb_tdb_set_write_time(struct odb_lock *lck,
+                                      NTTIME write_time, bool force)
+{
+       if (lck->file.path == NULL) {
+               return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+       }
+
+       if (lck->file.changed_write_time != 0 && !force) {
+               return NT_STATUS_OK;
+       }
 
-       file.delete_on_close = del_on_close;
+       lck->file.changed_write_time = write_time;
 
-       return odb_push_record(lck, &file);
+       return odb_push_record(lck, &lck->file);
 }
 
 /*
   return the current value of the delete_on_close bit, and how many
   people still have the file open
 */
-static NTSTATUS odb_tdb_get_delete_on_close(struct odb_context *odb, 
-                                           DATA_BLOB *key, bool *del_on_close, 
-                                           int *open_count, char **path)
+static NTSTATUS odb_tdb_get_file_infos(struct odb_context *odb, DATA_BLOB *key,
+                                      bool *del_on_close, NTTIME *write_time)
 {
-       NTSTATUS status;
-       struct opendb_file file;
        struct odb_lock *lck;
 
-       lck = odb_lock(odb, odb, key);
-       NT_STATUS_HAVE_NO_MEMORY(lck);
-
-       status = odb_pull_record(lck, &file);
-       if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
-               talloc_free(lck);
-               (*del_on_close) = false;
-               return NT_STATUS_OK;
+       if (del_on_close) {
+               *del_on_close = false;
        }
-       if (!NT_STATUS_IS_OK(status)) {
-               talloc_free(lck);
-               return status;
+       if (write_time) {
+               *write_time = 0;
        }
 
-       (*del_on_close) = file.delete_on_close;
-       if (open_count != NULL) {
-               (*open_count) = file.num_entries;
+       lck = odb_lock(odb, odb, key);
+       NT_STATUS_HAVE_NO_MEMORY(lck);
+
+       if (del_on_close) {
+               *del_on_close = lck->file.delete_on_close;
        }
-       if (path != NULL) {
-               *path = talloc_strdup(odb, file.path);
-               NT_STATUS_HAVE_NO_MEMORY(*path);
-               if (file.num_entries == 1 && file.entries[0].delete_on_close) {
-                       (*del_on_close) = true;
+       if (write_time) {
+               if (lck->file.changed_write_time == 0) {
+                       *write_time = lck->file.open_write_time;
+               } else {
+                       *write_time = lck->file.changed_write_time;
                }
        }
 
@@ -624,47 +849,31 @@ static NTSTATUS odb_tdb_get_delete_on_close(struct odb_context *odb,
   create_options and access_mask
 */
 static NTSTATUS odb_tdb_can_open(struct odb_lock *lck,
-                                uint32_t share_access, uint32_t create_options, 
-                                uint32_t access_mask)
+                                uint32_t stream_id, uint32_t share_access,
+                                uint32_t access_mask, bool delete_on_close,
+                                uint32_t open_disposition, bool break_to_none)
 {
        struct odb_context *odb = lck->odb;
        NTSTATUS status;
-       struct opendb_file file;
-       struct opendb_entry e;
-       int i;
 
-       status = odb_pull_record(lck, &file);
-       if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
-               return NT_STATUS_OK;
-       }
+       status = odb_tdb_open_can_internal(odb, &lck->file, stream_id,
+                                          share_access, access_mask,
+                                          delete_on_close, open_disposition,
+                                          break_to_none, &lck->can_open.attrs_only);
        NT_STATUS_NOT_OK_RETURN(status);
 
-       if ((create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) && 
-           file.num_entries != 0) {
-               return NT_STATUS_SHARING_VIOLATION;
-       }
+       lck->can_open.e = talloc(lck, struct opendb_entry);
+       NT_STATUS_HAVE_NO_MEMORY(lck->can_open.e);
 
-       if (file.delete_on_close) {
-               return NT_STATUS_DELETE_PENDING;
-       }
-
-       e.server       = odb->ntvfs_ctx->server_id;
-       e.file_handle  = NULL;
-       e.stream_id    = 0;
-       e.share_access = share_access;
-       e.access_mask  = access_mask;
-               
-       for (i=0;i<file.num_entries;i++) {
-               status = share_conflict(&file.entries[i], &e);
-               if (!NT_STATUS_IS_OK(status)) {
-                       /* note that we discard the error code
-                          here. We do this as unless we are actually
-                          doing an open (which comes via a different
-                          function), we need to return a sharing
-                          violation */
-                       return NT_STATUS_SHARING_VIOLATION;
-               }
-       }
+       lck->can_open.e->server                 = odb->ntvfs_ctx->server_id;
+       lck->can_open.e->file_handle            = NULL;
+       lck->can_open.e->fd                     = NULL;
+       lck->can_open.e->stream_id              = stream_id;
+       lck->can_open.e->share_access           = share_access;
+       lck->can_open.e->access_mask            = access_mask;
+       lck->can_open.e->delete_on_close        = delete_on_close;
+       lck->can_open.e->allow_level_II_oplock  = false;
+       lck->can_open.e->oplock_level           = OPLOCK_NONE;
 
        return NT_STATUS_OK;
 }
@@ -679,14 +888,18 @@ static const struct opendb_ops opendb_tdb_ops = {
        .odb_close_file          = odb_tdb_close_file,
        .odb_remove_pending      = odb_tdb_remove_pending,
        .odb_rename              = odb_tdb_rename,
+       .odb_get_path            = odb_tdb_get_path,
        .odb_set_delete_on_close = odb_tdb_set_delete_on_close,
-       .odb_get_delete_on_close = odb_tdb_get_delete_on_close,
+       .odb_set_write_time      = odb_tdb_set_write_time,
+       .odb_get_file_infos      = odb_tdb_get_file_infos,
        .odb_can_open            = odb_tdb_can_open,
-       .odb_update_oplock       = odb_tdb_update_oplock
+       .odb_update_oplock       = odb_tdb_update_oplock,
+       .odb_break_oplocks       = odb_tdb_break_oplocks
 };
 
 
 void odb_tdb_init_ops(void)
 {
+       sys_lease_init();
        odb_set_ops(&opendb_tdb_ops);
 }