r13739: a fairly major overhaul of the opendb code to allow the BASE-DELETE

test to pass. To try to make the code a bit more understandable, I
moved to using an IDL description of the opendb tdb record format.

One of the larger changes was to make directory opens and creates go
via the opendb code, so directory operations now obey all the share
mode restrictions, as well as delete on close semantics. I also
changed the period over which the opendb locks are held, to try to
minimise races due to two open operations happening at the same time.
(This used to be commit cd2602d05725e1734b0862131dd91601c6b6d51a)

9 files changed, 932 insertions, 324 deletions

diff --git a/source4/librpc/config.mk b/source4/librpc/config.mk
index 131531dad5..a827e481f8 100644
--- a/source4/librpc/config.mk
+++ b/source4/librpc/config.mk
@@ -286,6 +286,11 @@ OBJ_FILES = gen_ndr/ndr_xattr.o
 NOPROTO = YES
 REQUIRED_SUBSYSTEMS = LIBNDR
 
+[SUBSYSTEM::NDR_OPENDB]
+OBJ_FILES = gen_ndr/ndr_opendb.o
+NOPROTO = YES
+REQUIRED_SUBSYSTEMS = LIBNDR
+
 [SUBSYSTEM::NDR_SCHANNEL]
 OBJ_FILES = gen_ndr/ndr_schannel.o
 NOPROTO = YES
@@ -319,7 +324,7 @@ REQUIRED_SUBSYSTEMS = NDR_IFACE_TABLE NDR_AUDIOSRV NDR_ECHO NDR_DCERPC NDR_EXCHA
 	NDR_REMACT NDR_WZCSVC NDR_BROWSER NDR_W32TIME NDR_SCERPC NDR_NTSVCS \
 	NDR_NETLOGON NDR_TRKWKS NDR_KEYSVC NDR_KRB5PAC NDR_XATTR NDR_SCHANNEL \
 	NDR_ROT NDR_DRSBLOBS NDR_SVCCTL NDR_NBT NDR_WINSREPL NDR_SECURITY \
-	NDR_INITSHUTDOWN NDR_DNSSERVER NDR_WINSTATION NDR_IRPC NDR_DCOM \
+	NDR_INITSHUTDOWN NDR_DNSSERVER NDR_WINSTATION NDR_IRPC NDR_DCOM NDR_OPENDB \
 	NDR_SASL_HELPERS
 
 [SUBSYSTEM::RPC_NDR_ROT]
diff --git a/source4/librpc/idl/opendb.idl b/source4/librpc/idl/opendb.idl
new file mode 100644
index 0000000000..3754d233f6
--- /dev/null
+++ b/source4/librpc/idl/opendb.idl
@@ -0,0 +1,39 @@
+#include "idl_types.h"
+
+/*
+   IDL structures for opendb code
+
+   this defines the structures used in the opendb database code, in 
+   ntvfs/common/opendb.c
+*/
+
+[
+  pointer_default(unique)
+]
+interface opendb
+{
+	typedef struct {
+		uint32 server;
+		uint32 stream_id;
+		uint32 share_access;
+		uint32 access_mask;
+		pointer file_handle;
+		/* we need a per-entry delete on close, as well as a per-file
+		   one, to cope with strange semantics on open */
+		bool8 delete_on_close;
+	} opendb_entry;
+
+	typedef struct {
+		uint32 server;
+		pointer notify_ptr;
+	} opendb_pending;
+
+	typedef [public] struct {
+		bool8 delete_on_close;
+		utf8string path;
+		uint32 num_entries;
+		opendb_entry entries[num_entries];
+		uint32 num_pending;
+		opendb_pending pending[num_pending];
+	} opendb_file;
+}
diff --git a/source4/ntvfs/common/opendb.c b/source4/ntvfs/common/opendb.c
index 330af47862..1faa6387b7 100644
--- a/source4/ntvfs/common/opendb.c
+++ b/source4/ntvfs/common/opendb.c
@@ -46,6 +46,7 @@
 #include "db_wrap.h"
 #include "smb_server/smb_server.h"
 #include "lib/messaging/irpc.h"
+#include "librpc/gen_ndr/ndr_opendb.h"
 
 struct odb_context {
 	struct tdb_wrap *w;
@@ -53,22 +54,6 @@ struct odb_context {
 	struct messaging_context *messaging_ctx;
 };
 
-/* 
-   the database is indexed by a file_key, and contains entries of the
-   following form
-*/
-struct odb_entry {
-	uint32_t server;
-	void     *file_handle;
-	uint32_t stream_id;
-	uint32_t share_access;
-	uint32_t create_options;
-	uint32_t access_mask;
-	void	 *notify_ptr;
-	BOOL     pending;
-};
-
-
 /*
   an odb lock handle. You must obtain one of these using odb_lock() before doing
   any other operations. 
@@ -157,10 +142,8 @@ struct odb_lock *odb_lock(TALLOC_CTX *mem_ctx,
 
   return NT_STATUS_OK on no conflict
 */
-static NTSTATUS share_conflict(struct odb_entry *e1, struct odb_entry *e2)
+static NTSTATUS share_conflict(struct opendb_entry *e1, struct opendb_entry *e2)
 {
-	if (e1->pending || e2->pending) return NT_STATUS_OK;
-
 	/* if either open involves no read.write or delete access then
 	   it can't conflict */
 	if (!(e1->access_mask & (SEC_FILE_WRITE_DATA |
@@ -203,187 +186,198 @@ static NTSTATUS share_conflict(struct odb_entry *e1, struct odb_entry *e2)
 	CHECK_MASK(e2->access_mask, SEC_STD_DELETE,
 		   e1->share_access, NTCREATEX_SHARE_ACCESS_DELETE);
 
-	/* if a delete is pending then a second open is not allowed */
-	if ((e1->create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) ||
-	    (e2->create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE)) {
-		return NT_STATUS_DELETE_PENDING;
-	}
-
 	return NT_STATUS_OK;
 }
 
 /*
-  register an open file in the open files database. This implements the share_access
-  rules
+  pull a record, translating from the db format to the opendb_file structure defined
+  in opendb.idl
 */
-NTSTATUS odb_open_file(struct odb_lock *lck, void *file_handle,
-		       uint32_t stream_id,
-		       uint32_t share_access, uint32_t create_options,
-		       uint32_t access_mask)
+static NTSTATUS odb_pull_record(struct odb_lock *lck, struct opendb_file *file)
 {
 	struct odb_context *odb = lck->odb;
 	TDB_DATA dbuf;
-	struct odb_entry e;
-	int i, count;
-	struct odb_entry *elist;
+	DATA_BLOB blob;
+	NTSTATUS status;
 		
 	dbuf = tdb_fetch(odb->w->tdb, lck->key);
-
-	e.server         = odb->server;
-	e.file_handle    = file_handle;
-	e.stream_id      = stream_id;
-	e.share_access   = share_access;
-	e.create_options = create_options;
-	e.access_mask    = access_mask;
-	e.notify_ptr	 = NULL;
-	e.pending        = False;
-
-	/* check the existing file opens to see if they
-	   conflict */
-	elist = (struct odb_entry *)dbuf.dptr;
-	count = dbuf.dsize / sizeof(struct odb_entry);
-
-	for (i=0;i<count;i++) {
-		NTSTATUS status;
-		status = share_conflict(elist+i, &e);
-		if (!NT_STATUS_IS_OK(status)) {
-			if (dbuf.dptr) free(dbuf.dptr);
-			return status;
-		}
+	if (dbuf.dptr == NULL) {
+		return NT_STATUS_OBJECT_NAME_NOT_FOUND;
 	}
 
-	elist = realloc_p(dbuf.dptr, struct odb_entry, count+1);
-	if (elist == NULL) {
-		if (dbuf.dptr) free(dbuf.dptr);
-		return NT_STATUS_NO_MEMORY;
-	}
+	blob.data = dbuf.dptr;
+	blob.length = dbuf.dsize;
 
-	dbuf.dptr = (uint8_t *)elist;
-	dbuf.dsize = (count+1) * sizeof(struct odb_entry);
+	status = ndr_pull_struct_blob(&blob, lck, file, (ndr_pull_flags_fn_t)ndr_pull_opendb_file);
 
-	memcpy(dbuf.dptr + (count*sizeof(struct odb_entry)),
-	       &e, sizeof(struct odb_entry));
+	free(dbuf.dptr);
 
-	if (tdb_store(odb->w->tdb, lck->key, dbuf, TDB_REPLACE) != 0) {
-		free(dbuf.dptr);
+	return status;
+}
+
+/*
+  push a record, translating from the opendb_file structure defined in opendb.idl
+*/
+static NTSTATUS odb_push_record(struct odb_lock *lck, struct opendb_file *file)
+{
+	struct odb_context *odb = lck->odb;
+	TDB_DATA dbuf;
+	DATA_BLOB blob;
+	NTSTATUS status;
+	int ret;
+
+	if (file->num_entries == 0) {
+		ret = tdb_delete(odb->w->tdb, lck->key);
+		if (ret != 0) {
+			return NT_STATUS_INTERNAL_DB_CORRUPTION;
+		}
+		return NT_STATUS_OK;
+	}
+
+	status = ndr_push_struct_blob(&blob, lck, file, (ndr_push_flags_fn_t)ndr_push_opendb_file);
+	NT_STATUS_NOT_OK_RETURN(status);
+
+	dbuf.dptr = blob.data;
+	dbuf.dsize = blob.length;
+		
+	ret = tdb_store(odb->w->tdb, lck->key, dbuf, TDB_REPLACE);
+	data_blob_free(&blob);
+	if (ret != 0) {
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
 	}
 
-	free(dbuf.dptr);
 	return NT_STATUS_OK;
 }
 
 
 /*
-  register a pending open file in the open files database
+  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
 */
-NTSTATUS odb_open_file_pending(struct odb_lock *lck, void *private)
+NTSTATUS odb_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)
 {
 	struct odb_context *odb = lck->odb;
-	TDB_DATA dbuf;
-	struct odb_entry e;
-	struct odb_entry *elist;
-	int count;
-		
-	dbuf = tdb_fetch(odb->w->tdb, lck->key);
+	struct opendb_entry e;
+	int i;
+	struct opendb_file file;
+	NTSTATUS status;
 
-	e.server         = odb->server;
-	e.file_handle    = NULL;
-	e.stream_id      = 0;
-	e.share_access   = 0;
-	e.create_options = 0;
-	e.access_mask    = 0;
-	e.notify_ptr	 = private;
-	e.pending        = True;
-
-	/* check the existing file opens to see if they
-	   conflict */
-	elist = (struct odb_entry *)dbuf.dptr;
-	count = dbuf.dsize / sizeof(struct odb_entry);
-
-	elist = realloc_p(dbuf.dptr, struct odb_entry, count+1);
-	if (elist == NULL) {
-		if (dbuf.dptr) free(dbuf.dptr);
-		return NT_STATUS_NO_MEMORY;
+	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);
 	}
 
-	dbuf.dptr = (uint8_t *)elist;
-	dbuf.dsize = (count+1) * sizeof(struct odb_entry);
 
-	memcpy(dbuf.dptr + (count*sizeof(struct odb_entry)),
-	       &e, sizeof(struct odb_entry));
+	if (file.delete_on_close || 
+	    (file.num_entries != 0 && delete_on_close)) {
+		/* while delete on close is set, no new opens are allowed */
+		return NT_STATUS_DELETE_PENDING;
+	}
 
-	if (tdb_store(odb->w->tdb, lck->key, dbuf, TDB_REPLACE) != 0) {
-		free(dbuf.dptr);
-		return NT_STATUS_INTERNAL_DB_CORRUPTION;
+	/* see if it conflicts */
+	e.server          = odb->server;
+	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;
+		
+	for (i=0;i<file.num_entries;i++) {
+		status = share_conflict(&file.entries[i], &e);
+		NT_STATUS_NOT_OK_RETURN(status);
 	}
 
-	free(dbuf.dptr);
-	return NT_STATUS_OK;
+	/* it doesn't, 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);
+
+	file.entries[file.num_entries] = e;
+	file.num_entries++;
+
+	return odb_push_record(lck, &file);
 }
 
 
 /*
-  remove a opendb entry
+  register a pending open file in the open files database
 */
-NTSTATUS odb_close_file(struct odb_lock *lck, void *file_handle)
+NTSTATUS odb_open_file_pending(struct odb_lock *lck, void *private)
 {
 	struct odb_context *odb = lck->odb;
-	TDB_DATA dbuf;
-	struct odb_entry *elist;
-	int i, count;
+	struct opendb_file file;
 	NTSTATUS status;
+		
+	status = odb_pull_record(lck, &file);
+	NT_STATUS_NOT_OK_RETURN(status);
 
-	dbuf = tdb_fetch(odb->w->tdb, lck->key);
+	file.pending = talloc_realloc(lck, file.pending, struct opendb_pending, 
+				      file.num_pending+1);
+	NT_STATUS_HAVE_NO_MEMORY(file.pending);
 
-	if (dbuf.dptr == NULL) {
-		return NT_STATUS_UNSUCCESSFUL;
-	}
+	file.pending[file.num_pending].server = odb->server;
+	file.pending[file.num_pending].notify_ptr = private;
 
-	elist = (struct odb_entry *)dbuf.dptr;
-	count = dbuf.dsize / sizeof(struct odb_entry);
+	file.num_pending++;
 
-	/* send any pending notifications, removing them once sent */
-	for (i=0;i<count;i++) {
-		if (elist[i].pending) {
-			messaging_send_ptr(odb->messaging_ctx, elist[i].server, 
-					   MSG_PVFS_RETRY_OPEN, elist[i].notify_ptr);
-			memmove(&elist[i], &elist[i+1], sizeof(struct odb_entry)*(count-(i+1)));
-			i--;
-			count--;
-		}
-	}
+	return odb_push_record(lck, &file);
+}
+
+
+/*
+  remove a opendb entry
+*/
+NTSTATUS odb_close_file(struct odb_lock *lck, void *file_handle)
+{
+	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);
 
 	/* find the entry, and delete it */
-	for (i=0;i<count;i++) {
-		if (file_handle == elist[i].file_handle &&
-		    odb->server == elist[i].server) {
-			if (i < count-1) {
-				memmove(elist+i, elist+i+1, 
-					(count - (i+1)) * sizeof(struct odb_entry));
+	for (i=0;i<file.num_entries;i++) {
+		if (file_handle == file.entries[i].file_handle &&
+		    odb->server == file.entries[i].server) {
+			if (file.entries[i].delete_on_close) {
+				file.delete_on_close = True;
+			}
+			if (i < file.num_entries-1) {
+				memmove(file.entries+i, file.entries+i+1, 
+					(file.num_entries - (i+1)) * 
+					sizeof(struct opendb_entry));
 			}
 			break;
 		}
 	}
 
-	status = NT_STATUS_OK;
-
-	if (i == count) {
-		status = NT_STATUS_UNSUCCESSFUL;
-	} else if (count == 1) {
-		if (tdb_delete(odb->w->tdb, lck->key) != 0) {
-			status = NT_STATUS_INTERNAL_DB_CORRUPTION;
-		}
-	} else {
-		dbuf.dsize = (count-1) * sizeof(struct odb_entry);
-		if (tdb_store(odb->w->tdb, lck->key, dbuf, TDB_REPLACE) != 0) {
-			status = NT_STATUS_INTERNAL_DB_CORRUPTION;
-		}
+	if (i == file.num_entries) {
+		return NT_STATUS_UNSUCCESSFUL;
 	}
 
-	free(dbuf.dptr);
+	/* send any pending notifications, removing them once sent */
+	for (i=0;i<file.num_pending;i++) {
+		messaging_send_ptr(odb->messaging_ctx, file.pending[i].server, 
+				   MSG_PVFS_RETRY_OPEN, 
+				   file.pending[i].notify_ptr);
+	}
+	file.num_pending = 0;
 
-	return status;
+	file.num_entries--;
+	
+	return odb_push_record(lck, &file);
 }
 
 
@@ -393,91 +387,112 @@ NTSTATUS odb_close_file(struct odb_lock *lck, void *file_handle)
 NTSTATUS odb_remove_pending(struct odb_lock *lck, void *private)
 {
 	struct odb_context *odb = lck->odb;
-	TDB_DATA dbuf;
-	struct odb_entry *elist;
-	int i, count;
+	int i;
 	NTSTATUS status;
+	struct opendb_file file;
 
-	dbuf = tdb_fetch(odb->w->tdb, lck->key);
-
-	if (dbuf.dptr == NULL) {
-		return NT_STATUS_UNSUCCESSFUL;
-	}
-
-	elist = (struct odb_entry *)dbuf.dptr;
-	count = dbuf.dsize / sizeof(struct odb_entry);
+	status = odb_pull_record(lck, &file);
+	NT_STATUS_NOT_OK_RETURN(status);
 
 	/* find the entry, and delete it */
-	for (i=0;i<count;i++) {
-		if (private == elist[i].notify_ptr &&
-		    odb->server == elist[i].server) {
-			if (i < count-1) {
-				memmove(elist+i, elist+i+1, 
-					(count - (i+1)) * sizeof(struct odb_entry));
+	for (i=0;i<file.num_pending;i++) {
+		if (private == file.pending[i].notify_ptr &&
+		    odb->server == file.pending[i].server) {
+			if (i < file.num_pending-1) {
+				memmove(file.pending+i, file.pending+i+1, 
+					(file.num_pending - (i+1)) * 
+					sizeof(struct opendb_pending));
 			}
 			break;
 		}
 	}
 
-	status = NT_STATUS_OK;
-
-	if (i == count) {
-		status = NT_STATUS_UNSUCCESSFUL;
-	} else if (count == 1) {
-		if (tdb_delete(odb->w->tdb, lck->key) != 0) {
-			status = NT_STATUS_INTERNAL_DB_CORRUPTION;
-		}
-	} else {
-		dbuf.dsize = (count-1) * sizeof(struct odb_entry);
-		if (tdb_store(odb->w->tdb, lck->key, dbuf, TDB_REPLACE) != 0) {
-			status = NT_STATUS_INTERNAL_DB_CORRUPTION;
-		}
+	if (i == file.num_pending) {
+		return NT_STATUS_UNSUCCESSFUL;
 	}
 
-	free(dbuf.dptr);
-
-	return status;
+	file.num_pending--;
+	
+	return odb_push_record(lck, &file);
 }
 
 
 /*
-  update create options on an open file
+  rename the path in a open file
 */
-NTSTATUS odb_set_create_options(struct odb_lock *lck, 
-				void *file_handle, uint32_t create_options)
+NTSTATUS odb_rename(struct odb_lock *lck, const char *path)
 {
-	struct odb_context *odb = lck->odb;
-	TDB_DATA dbuf;
-	struct odb_entry *elist;
-	int i, count;
+	struct opendb_file file;
 	NTSTATUS status;
 
-	dbuf = tdb_fetch(odb->w->tdb, lck->key);
-	if (dbuf.dptr == NULL) {
-		return NT_STATUS_UNSUCCESSFUL;
+	status = odb_pull_record(lck, &file);
+	if (NT_STATUS_EQUAL(NT_STATUS_OBJECT_NAME_NOT_FOUND, status)) {
+		/* 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);
+}
+
+/*
+  update delete on close flag on an open file
+*/
+NTSTATUS odb_set_delete_on_close(struct odb_lock *lck, BOOL del_on_close)
+{
+	NTSTATUS status;
+	struct opendb_file file;
 
-	elist = (struct odb_entry *)dbuf.dptr;
-	count = dbuf.dsize / sizeof(struct odb_entry);
+	status = odb_pull_record(lck, &file);
+	NT_STATUS_NOT_OK_RETURN(status);
 
-	/* find the entry, and modify it */
-	for (i=0;i<count;i++) {
-		if (file_handle == elist[i].file_handle &&
-		    odb->server == elist[i].server) {
-			elist[i].create_options = create_options;
-			break;
-		}
+	file.delete_on_close = del_on_close;
+
+	return odb_push_record(lck, &file);
+}
+
+/*
+  return the current value of the delete_on_close bit, and how many
+  people still have the file open
+*/
+NTSTATUS odb_get_delete_on_close(struct odb_context *odb, 
+				 DATA_BLOB *key, BOOL *del_on_close, 
+				 int *open_count, char **path)
+{
+	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 (!NT_STATUS_IS_OK(status)) {
+		talloc_free(lck);
+		return status;
 	}
 
-	if (tdb_store(odb->w->tdb, lck->key, dbuf, TDB_REPLACE) != 0) {
-		status = NT_STATUS_INTERNAL_DB_CORRUPTION;
-	} else {
-		status = NT_STATUS_OK;
+	(*del_on_close) = file.delete_on_close;
+	if (open_count != NULL) {
+		(*open_count) = file.num_entries;
+	}
+	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;
+		}
 	}
 
-	free(dbuf.dptr);
+	talloc_free(lck);
 
-	return status;
+	return NT_STATUS_OK;
 }
 
 
@@ -485,46 +500,40 @@ NTSTATUS odb_set_create_options(struct odb_lock *lck,
   determine if a file can be opened with the given share_access,
   create_options and access_mask
 */
-NTSTATUS odb_can_open(struct odb_context *odb, DATA_BLOB *key, 
+NTSTATUS odb_can_open(struct odb_lock *lck,
 		      uint32_t share_access, uint32_t create_options, 
 		      uint32_t access_mask)
 {
-	TDB_DATA dbuf;
-	TDB_DATA kbuf;
-	struct odb_entry *elist;
-	int i, count;
-	struct odb_entry e;
-
-	kbuf.dptr = key->data;
-	kbuf.dsize = key->length;
+	struct odb_context *odb = lck->odb;
+	NTSTATUS status;
+	struct opendb_file file;
+	struct opendb_entry e;
+	int i;
 
-	dbuf = tdb_fetch(odb->w->tdb, kbuf);
-	if (dbuf.dptr == NULL) {
+	status = odb_pull_record(lck, &file);
+	if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND)) {
 		return NT_STATUS_OK;
 	}
+	NT_STATUS_NOT_OK_RETURN(status);
 
-	elist = (struct odb_entry *)dbuf.dptr;
-	count = dbuf.dsize / sizeof(struct odb_entry);
+	if ((create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) && 
+	    file.num_entries != 0) {
+		return NT_STATUS_SHARING_VIOLATION;
+	}
 
-	if (count == 0) {
-		free(dbuf.dptr);
-		return NT_STATUS_OK;
+	if (file.delete_on_close) {
+		return NT_STATUS_DELETE_PENDING;
 	}
 
-	e.server         = odb->server;
-	e.file_handle    = NULL;
-	e.stream_id      = 0;
-	e.share_access   = share_access;
-	e.create_options = create_options;
-	e.access_mask    = access_mask;
-	e.notify_ptr	 = NULL;
-	e.pending	 = False;
-
-	for (i=0;i<count;i++) {
-		NTSTATUS status;
-		status = share_conflict(elist+i, &e);
+	e.server       = odb->server;
+	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)) {
-			if (dbuf.dptr) free(dbuf.dptr);
 			/* note that we discard the error code
 			   here. We do this as unless we are actually
 			   doing an open (which comes via a sdifferent
@@ -534,6 +543,5 @@ NTSTATUS odb_can_open(struct odb_context *odb, DATA_BLOB *key,
 		}
 	}
 
-	free(dbuf.dptr);
 	return NT_STATUS_OK;
 }
diff --git a/source4/ntvfs/posix/pvfs_fileinfo.c b/source4/ntvfs/posix/pvfs_fileinfo.c
index 027118e7d9..d5f9156ec2 100644
--- a/source4/ntvfs/posix/pvfs_fileinfo.c
+++ b/source4/ntvfs/posix/pvfs_fileinfo.c
@@ -67,9 +67,10 @@ static uint32_t dos_mode_from_stat(struct pvfs_state *pvfs, struct stat *st)
 */
 NTSTATUS pvfs_fill_dos_info(struct pvfs_state *pvfs, struct pvfs_filename *name, int fd)
 {
-	/* make directories appear as size 0 */
+	/* make directories appear as size 0 with 1 link */
 	if (S_ISDIR(name->st.st_mode)) {
 		name->st.st_size = 0;
+		name->st.st_nlink = 1;
 	}
 
 	/* for now just use the simple samba mapping */
diff --git a/source4/ntvfs/posix/pvfs_open.c b/source4/ntvfs/posix/pvfs_open.c
index 5aeb5eb7e8..a8c0bba97f 100644
--- a/source4/ntvfs/posix/pvfs_open.c
+++ b/source4/ntvfs/posix/pvfs_open.c
@@ -68,17 +68,41 @@ struct pvfs_file *pvfs_find_fd(struct pvfs_state *pvfs,
 static int pvfs_dir_handle_destructor(void *p)
 {
 	struct pvfs_file_handle *h = p;
+	int open_count;
+	char *path;
 
-	if (h->create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) {
-		NTSTATUS status = pvfs_xattr_unlink_hook(h->pvfs, h->name->full_name);
+	if (h->name->stream_name == NULL && 
+	    pvfs_delete_on_close_set(h->pvfs, h, &open_count, &path) &&
+	    open_count == 1) {
+		NTSTATUS status;
+		status = pvfs_xattr_unlink_hook(h->pvfs, path);
 		if (!NT_STATUS_IS_OK(status)) {
-			DEBUG(0,("Warning: xattr rmdir hook failed for '%s' - %s\n",
-				 h->name->full_name, nt_errstr(status)));
+			DEBUG(0,("Warning: xattr unlink hook failed for '%s' - %s\n",
+				 path, nt_errstr(status)));
+		}
+		if (rmdir(path) != 0) {
+			DEBUG(0,("pvfs_dir_handle_destructor: failed to rmdir '%s' - %s\n", 
+				 path, strerror(errno)));
+		}
+	}
+
+	if (h->have_opendb_entry) {
+		struct odb_lock *lck;
+		NTSTATUS status;
+
+		lck = odb_lock(h, h->pvfs->odb_context, &h->odb_locking_key);
+		if (lck == NULL) {
+			DEBUG(0,("Unable to lock opendb for close\n"));
+			return 0;
 		}
-		if (rmdir(h->name->full_name) != 0 && errno != ENOTEMPTY) {
-			DEBUG(0,("pvfs_close: failed to rmdir '%s' - %s\n", 
-				 h->name->full_name, strerror(errno)));
+
+		status = odb_close_file(lck, h);
+		if (!NT_STATUS_IS_OK(status)) {
+			DEBUG(0,("Unable to remove opendb entry for '%s' - %s\n", 
+				 h->name->full_name, nt_errstr(status)));
 		}
+
+		talloc_free(lck);
 	}
 
 	return 0;
@@ -134,6 +158,379 @@ static NTSTATUS pvfs_open_setup_eas_acl(struct pvfs_state *pvfs,
 }
 
 /*
+  form the lock context used for opendb locking. Note that we must
+  zero here to take account of possible padding on some architectures
+*/
+static NTSTATUS pvfs_locking_key(struct pvfs_filename *name, 
+				 TALLOC_CTX *mem_ctx, DATA_BLOB *key)
+{
+	struct {
+		dev_t device;
+		ino_t inode;
+	} lock_context;
+	ZERO_STRUCT(lock_context);
+
+	lock_context.device = name->st.st_dev;
+	lock_context.inode = name->st.st_ino;
+
+	*key = data_blob_talloc(mem_ctx, &lock_context, sizeof(lock_context));
+	if (key->data == NULL) {
+		return NT_STATUS_NO_MEMORY;
+	}
+	
+	return NT_STATUS_OK;
+}
+
+
+/*
+  create a new directory
+*/
+static NTSTATUS pvfs_create_directory(struct pvfs_state *pvfs, 
+				      struct smbsrv_request *req, 
+				      struct pvfs_filename *name, 
+				      union smb_open *io)
+{
+	struct pvfs_file *f;
+	NTSTATUS status;
+	int fnum, ret;
+	struct odb_lock *lck;
+	uint32_t create_options = io->generic.in.create_options;
+	uint32_t share_access = io->generic.in.share_access;
+	uint32_t access_mask = io->generic.in.access_mask;
+	mode_t mode;
+	uint32_t attrib;
+	BOOL del_on_close;
+
+	if ((io->ntcreatex.in.file_attr & FILE_ATTRIBUTE_READONLY) &&
+	    (create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE)) {
+		return NT_STATUS_CANNOT_DELETE;
+	}
+	
+	status = pvfs_access_check_create(pvfs, req, name, &access_mask);
+	if (!NT_STATUS_IS_OK(status)) {
+		return status;
+	}
+
+	f = talloc(req, struct pvfs_file);
+	if (f == NULL) {
+		return NT_STATUS_NO_MEMORY;
+	}
+
+	f->handle = talloc(f, struct pvfs_file_handle);
+	if (f->handle == NULL) {
+		return NT_STATUS_NO_MEMORY;
+	}
+
+	fnum = idr_get_new_above(pvfs->idtree_fnum, f, PVFS_MIN_NEW_FNUM, UINT16_MAX);
+	if (fnum == -1) {
+		return NT_STATUS_TOO_MANY_OPENED_FILES;
+	}
+
+	attrib = io->ntcreatex.in.file_attr | FILE_ATTRIBUTE_DIRECTORY;
+	mode = pvfs_fileperms(pvfs, attrib);
+
+	/* create the directory */
+	ret = mkdir(name->full_name, mode);
+	if (ret == -1) {
+		idr_remove(pvfs->idtree_fnum, fnum);
+		return pvfs_map_errno(pvfs, errno);
+	}
+
+	pvfs_xattr_unlink_hook(pvfs, name->full_name);
+
+	/* re-resolve the new directory */
+	status = pvfs_resolve_name_fd(pvfs, -1, name);
+	if (!NT_STATUS_IS_OK(status)) {
+		idr_remove(pvfs->idtree_fnum, fnum);
+		rmdir(name->full_name);
+		return status;
+	}
+
+	name->dos.attrib = attrib;
+
+	status = pvfs_open_setup_eas_acl(pvfs, req, name, -1, fnum, io);
+	if (!NT_STATUS_IS_OK(status)) {
+		goto cleanup_delete;
+	}
+
+	/* form the lock context used for opendb locking */
+	status = pvfs_locking_key(name, f->handle, &f->handle->odb_locking_key);
+	if (!NT_STATUS_IS_OK(status)) {
+		goto cleanup_delete;
+	}
+
+	/* grab a lock on the open file record */
+	lck = odb_lock(req, pvfs->odb_context, &f->handle->odb_locking_key);
+	if (lck == NULL) {
+		DEBUG(0,("pvfs_open: failed to lock file '%s' in opendb\n",
+			 name->full_name));
+		/* we were supposed to do a blocking lock, so something
+		   is badly wrong! */
+		status = NT_STATUS_INTERNAL_DB_CORRUPTION;
+		goto cleanup_delete;
+	}
+
+	if (create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) {
+		del_on_close = True;
+	} else {
+		del_on_close = False;
+	}
+
+	status = odb_open_file(lck, f->handle, name->stream_id,
+			       share_access, access_mask, del_on_close, name->full_name);
+	talloc_free(lck);
+	if (!NT_STATUS_IS_OK(status)) {
+		/* bad news, we must have hit a race */
+		idr_remove(pvfs->idtree_fnum, fnum);
+		return status;
+	}
+
+	f->fnum              = fnum;
+	f->session           = req->session;
+	f->smbpid            = req->smbpid;
+	f->pvfs              = pvfs;
+	f->pending_list      = NULL;
+	f->lock_count        = 0;
+	f->share_access      = io->generic.in.share_access;
+	f->access_mask       = access_mask;
+	f->impersonation     = io->generic.in.impersonation;
+
+	f->handle->pvfs              = pvfs;
+	f->handle->name              = talloc_steal(f->handle, name);
+	f->handle->fd                = -1;
+	f->handle->create_options    = io->generic.in.create_options;
+	f->handle->seek_offset       = 0;
+	f->handle->position          = 0;
+	f->handle->mode              = 0;
+	f->handle->have_opendb_entry = True;
+	f->handle->sticky_write_time = False;
+
+	DLIST_ADD(pvfs->open_files, f);
+
+	/* setup a destructor to avoid file descriptor leaks on
+	   abnormal termination */
+	talloc_set_destructor(f, pvfs_dir_fnum_destructor);
+	talloc_set_destructor(f->handle, pvfs_dir_handle_destructor);
+	
+	io->generic.out.oplock_level  = OPLOCK_NONE;
+	io->generic.out.fnum          = f->fnum;
+	io->generic.out.create_action = NTCREATEX_ACTION_CREATED;
+	io->generic.out.create_time   = name->dos.create_time;
+	io->generic.out.access_time   = name->dos.access_time;
+	io->generic.out.write_time    = name->dos.write_time;
+	io->generic.out.change_time   = name->dos.change_time;
+	io->generic.out.attrib        = name->dos.attrib;
+	io->generic.out.alloc_size    = name->dos.alloc_size;
+	io->generic.out.size          = name->st.st_size;
+	io->generic.out.file_type     = FILE_TYPE_DISK;
+	io->generic.out.ipc_state     = 0;
+	io->generic.out.is_directory  = 0;
+
+	/* success - keep the file handle */
+	talloc_steal(pvfs, f);
+
+	return NT_STATUS_OK;
+
+cleanup_delete:
+	idr_remove(pvfs->idtree_fnum, fnum);
+	rmdir(name->full_name);
+	return status;
+}
+
+#if 0
+/*
+  open a directory
+*/
+static NTSTATUS pvfs_open_directory(struct pvfs_state *pvfs, 
+				    struct smbsrv_request *req, 
+				    struct pvfs_filename *name, 
+				    union smb_open *io)
+{
+	struct pvfs_file *f;
+	NTSTATUS status;
+	int fnum;
+	struct odb_lock *lck;
+	uint32_t create_options;
+	uint32_t share_access;
+	uint32_t access_mask;
+	BOOL del_on_close;
+
+	create_options = io->generic.in.create_options;
+	share_access   = io->generic.in.share_access;
+	access_mask    = io->generic.in.access_mask;
+
+	/* certain create options are not allowed */
+	if ((create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) &&
+	    !(access_mask & SEC_STD_DELETE)) {
+		return NT_STATUS_INVALID_PARAMETER;
+	}
+
+	switch (io->generic.in.open_disposition) {
+	case NTCREATEX_DISP_SUPERSEDE:
+	case NTCREATEX_DISP_OVERWRITE_IF:
+	case NTCREATEX_DISP_OPEN_IF:
+		break;
+
+	case NTCREATEX_DISP_OPEN:
+	case NTCREATEX_DISP_OVERWRITE:
+		if (!name->exists) {
+			return NT_STATUS_OBJECT_NAME_NOT_FOUND;
+		}
+		break;
+
+	case NTCREATEX_DISP_CREATE:
+		if (name->exists) {
+			return NT_STATUS_OBJECT_NAME_COLLISION;
+		}
+		break;
+
+	default:
+		return NT_STATUS_INVALID_PARAMETER;
+	}
+
+	/* handle creating a new directory separately */
+	if (!name->exists) {
+		status = pvfs_create_directory(pvfs, req, name, io);
+		if (!NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_COLLISION)) {
+			return status;
+		}
+
+		/* we've hit a race - the directory was created during this call */
+		if (io->generic.in.open_disposition == NTCREATEX_DISP_CREATE) {
+			return status;
+		}
+
+		/* try re-resolving the name */
+		status = pvfs_resolve_name(pvfs, req, io->ntcreatex.in.fname, 0, &name);
+		if (!NT_STATUS_IS_OK(status)) {
+			return status;
+		}
+		/* fall through to a normal open */
+	}
+
+	if ((name->dos.attrib & FILE_ATTRIBUTE_READONLY) &&
+	    (create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE)) {
+		return NT_STATUS_CANNOT_DELETE;
+	}
+
+	/* check the security descriptor */
+	status = pvfs_access_check(pvfs, req, name, &access_mask);
+	if (!NT_STATUS_IS_OK(status)) {
+		return status;
+	}
+
+	f = talloc(req, struct pvfs_file);
+	if (f == NULL) {
+		return NT_STATUS_NO_MEMORY;
+	}
+
+	f->handle = talloc(f, struct pvfs_file_handle);
+	if (f->handle == NULL) {
+		return NT_STATUS_NO_MEMORY;
+	}
+
+	/* allocate a fnum */
+	fnum = idr_get_new_above(pvfs->idtree_fnum, f, PVFS_MIN_DIR_FNUM, UINT16_MAX);
+	if (fnum == -1) {
+		return NT_STATUS_TOO_MANY_OPENED_FILES;
+	}
+
+	f->fnum          = fnum;
+	f->session       = req->session;
+	f->smbpid        = req->smbpid;
+	f->pvfs          = pvfs;
+	f->pending_list  = NULL;
+	f->lock_count    = 0;
+	f->share_access  = io->generic.in.share_access;
+	f->access_mask   = access_mask;
+	f->impersonation = io->generic.in.impersonation;
+
+	f->handle->pvfs              = pvfs;
+	f->handle->fd                = -1;
+	f->handle->name              = talloc_steal(f->handle, name);
+	f->handle->create_options    = io->generic.in.create_options;
+	f->handle->seek_offset       = 0;
+	f->handle->position          = 0;
+	f->handle->mode              = 0;
+	f->handle->have_opendb_entry = False;
+	f->handle->sticky_write_time = False;
+
+	/* form the lock context used for opendb locking */
+	status = pvfs_locking_key(name, f->handle, &f->handle->odb_locking_key);
+	if (!NT_STATUS_IS_OK(status)) {
+		idr_remove(pvfs->idtree_fnum, f->fnum);
+		return status;
+	}
+
+	/* get a lock on this file before the actual open */
+	lck = odb_lock(req, pvfs->odb_context, &f->handle->odb_locking_key);
+	if (lck == NULL) {
+		DEBUG(0,("pvfs_open: failed to lock file '%s' in opendb\n",
+			 name->full_name));
+		/* we were supposed to do a blocking lock, so something
+		   is badly wrong! */
+		idr_remove(pvfs->idtree_fnum, fnum);
+		return NT_STATUS_INTERNAL_DB_CORRUPTION;
+	}
+
+	DLIST_ADD(pvfs->open_files, f);
+
+	/* setup a destructor to avoid file descriptor leaks on
+	   abnormal termination */
+	talloc_set_destructor(f, pvfs_dir_fnum_destructor);
+	talloc_set_destructor(f->handle, pvfs_dir_handle_destructor);
+
+	if ((create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) &&
+	    pvfs_directory_empty(pvfs, f->handle->name)) {
+		del_on_close = True;
+	} else {
+		del_on_close = False;
+	}
+
+	/* see if we are allowed to open at the same time as existing opens */
+	status = odb_open_file(lck, f->handle, f->handle->name->stream_id,
+			       share_access, access_mask, del_on_close, name->full_name);
+
+#if 0
+	/* we don't do async open retries on directories yet */
+	if (NT_STATUS_EQUAL(status, NT_STATUS_SHARING_VIOLATION) &&
+	    (req->async_states->state & NTVFS_ASYNC_STATE_MAY_ASYNC)) {
+		return pvfs_open_setup_retry(ntvfs, req, io, f, lck);
+	}
+#endif
+
+	if (!NT_STATUS_IS_OK(status)) {
+		talloc_free(lck);
+		return status;
+	}
+
+	f->handle->have_opendb_entry = True;
+
+	talloc_free(lck);
+
+	io->generic.out.oplock_level  = OPLOCK_NONE;
+	io->generic.out.fnum	      = f->fnum;
+	io->generic.out.create_action = NTCREATEX_ACTION_EXISTED;
+	io->generic.out.create_time   = name->dos.create_time;
+	io->generic.out.access_time   = name->dos.access_time;
+	io->generic.out.write_time    = name->dos.write_time;
+	io->generic.out.change_time   = name->dos.change_time;
+	io->generic.out.attrib        = name->dos.attrib;
+	io->generic.out.alloc_size    = name->dos.alloc_size;
+	io->generic.out.size          = name->st.st_size;
+	io->generic.out.file_type     = FILE_TYPE_DISK;
+	io->generic.out.ipc_state     = 0;
+	io->generic.out.is_directory  = 0;
+
+	/* success - keep the file handle */
+	talloc_steal(f->pvfs, f);
+
+	return NT_STATUS_OK;
+}
+#endif
+
+#if 1
+/*
   open a directory
 */
 static NTSTATUS pvfs_open_directory(struct pvfs_state *pvfs, 
@@ -146,6 +543,13 @@ static NTSTATUS pvfs_open_directory(struct pvfs_state *pvfs,
 	NTSTATUS status;
 	uint32_t create_action;
 	uint32_t access_mask = io->generic.in.access_mask;
+	struct odb_lock *lck;
+	BOOL del_on_close;
+	uint32_t create_options;
+	uint32_t share_access;
+
+	create_options = io->generic.in.create_options;
+	share_access   = io->generic.in.share_access;
 
 	if (name->stream_name) {
 		return NT_STATUS_NOT_A_DIRECTORY;
@@ -227,10 +631,47 @@ static NTSTATUS pvfs_open_directory(struct pvfs_state *pvfs,
 	f->handle->mode              = 0;
 	f->handle->sticky_write_time = False;
 
-	DLIST_ADD(pvfs->open_files, f);
+	if ((create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) &&
+	    pvfs_directory_empty(pvfs, f->handle->name)) {
+		del_on_close = True;
+	} else {
+		del_on_close = False;
+	}
+
+
+	if (name->exists) {
+		/* form the lock context used for opendb locking */
+		status = pvfs_locking_key(name, f->handle, &f->handle->odb_locking_key);
+		if (!NT_STATUS_IS_OK(status)) {
+			idr_remove(pvfs->idtree_fnum, f->fnum);
+			return status;
+		}
+
+		/* get a lock on this file before the actual open */
+		lck = odb_lock(req, pvfs->odb_context, &f->handle->odb_locking_key);
+		if (lck == NULL) {
+			DEBUG(0,("pvfs_open: failed to lock file '%s' in opendb\n",
+				 name->full_name));
+			/* we were supposed to do a blocking lock, so something
+			   is badly wrong! */
+			idr_remove(pvfs->idtree_fnum, fnum);
+			return NT_STATUS_INTERNAL_DB_CORRUPTION;
+		}
+		
+		/* see if we are allowed to open at the same time as existing opens */
+		status = odb_open_file(lck, f->handle, f->handle->name->stream_id,
+				       share_access, access_mask, del_on_close, name->full_name);
+
+		if (!NT_STATUS_IS_OK(status)) {
+			idr_remove(pvfs->idtree_fnum, f->fnum);
+			talloc_free(lck);
+			return status;
+		}
+
+		f->handle->have_opendb_entry = True;
+	}
 
-	/* TODO: should we check in the opendb? Do directory opens 
-	   follow the share_access rules? */
+	DLIST_ADD(pvfs->open_files, f);
 
 	/* setup destructors to avoid leaks on abnormal termination */
 	talloc_set_destructor(f->handle, pvfs_dir_handle_destructor);
@@ -239,6 +680,7 @@ static NTSTATUS pvfs_open_directory(struct pvfs_state *pvfs,
 	if (!name->exists) {
 		uint32_t attrib = io->generic.in.file_attr | FILE_ATTRIBUTE_DIRECTORY;
 		mode_t mode = pvfs_fileperms(pvfs, attrib);
+
 		if (mkdir(name->full_name, mode) == -1) {
 			idr_remove(pvfs->idtree_fnum, fnum);
 			return pvfs_map_errno(pvfs,errno);
@@ -256,6 +698,32 @@ static NTSTATUS pvfs_open_directory(struct pvfs_state *pvfs,
 			goto cleanup_delete;
 		}
 
+		/* form the lock context used for opendb locking */
+		status = pvfs_locking_key(name, f->handle, &f->handle->odb_locking_key);
+		if (!NT_STATUS_IS_OK(status)) {
+			idr_remove(pvfs->idtree_fnum, f->fnum);
+			return status;
+		}
+
+		lck = odb_lock(req, pvfs->odb_context, &f->handle->odb_locking_key);
+		if (lck == NULL) {
+			DEBUG(0,("pvfs_open: failed to lock file '%s' in opendb\n",
+				 name->full_name));
+			/* we were supposed to do a blocking lock, so something
+			   is badly wrong! */
+			idr_remove(pvfs->idtree_fnum, fnum);
+			return NT_STATUS_INTERNAL_DB_CORRUPTION;
+		}
+
+		status = odb_open_file(lck, f->handle, f->handle->name->stream_id,
+				       share_access, access_mask, del_on_close, name->full_name);
+
+		if (!NT_STATUS_IS_OK(status)) {
+			goto cleanup_delete;
+		}
+
+		f->handle->have_opendb_entry = True;
+
 		create_action = NTCREATEX_ACTION_CREATED;
 	} else {
 		create_action = NTCREATEX_ACTION_EXISTED;
@@ -290,6 +758,7 @@ cleanup_delete:
 	rmdir(name->full_name);
 	return status;
 }
+#endif
 
 /*
   destroy a struct pvfs_file_handle
@@ -297,6 +766,8 @@ cleanup_delete:
 static int pvfs_handle_destructor(void *p)
 {
 	struct pvfs_file_handle *h = p;
+	int open_count;
+	char *path;
 
 	/* the write time is no longer sticky */
 	if (h->sticky_write_time) {
@@ -326,17 +797,18 @@ static int pvfs_handle_destructor(void *p)
 		h->fd = -1;
 	}
 
-	if ((h->create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) &&
-	    h->name->stream_name == NULL) {
+	if (h->name->stream_name == NULL && 
+	    pvfs_delete_on_close_set(h->pvfs, h, &open_count, &path) &&
+	    open_count == 1) {
 		NTSTATUS status;
-		status = pvfs_xattr_unlink_hook(h->pvfs, h->name->full_name);
+		status = pvfs_xattr_unlink_hook(h->pvfs, path);
 		if (!NT_STATUS_IS_OK(status)) {
 			DEBUG(0,("Warning: xattr unlink hook failed for '%s' - %s\n",
-				 h->name->full_name, nt_errstr(status)));
+				 path, nt_errstr(status)));
 		}
-		if (unlink(h->name->full_name) != 0) {
+		if (unlink(path) != 0) {
 			DEBUG(0,("pvfs_close: failed to delete '%s' - %s\n", 
-				 h->name->full_name, strerror(errno)));
+				 path, strerror(errno)));
 		}
 	}
 
@@ -379,30 +851,6 @@ static int pvfs_fnum_destructor(void *p)
 
 
 /*
-  form the lock context used for opendb locking. Note that we must
-  zero here to take account of possible padding on some architectures
-*/
-static NTSTATUS pvfs_locking_key(struct pvfs_filename *name, 
-				 TALLOC_CTX *mem_ctx, DATA_BLOB *key)
-{
-	struct {
-		dev_t device;
-		ino_t inode;
-	} lock_context;
-	ZERO_STRUCT(lock_context);
-
-	lock_context.device = name->st.st_dev;
-	lock_context.inode = name->st.st_ino;
-
-	*key = data_blob_talloc(mem_ctx, &lock_context, sizeof(lock_context));
-	if (key->data == NULL) {
-		return NT_STATUS_NO_MEMORY;
-	}
-	
-	return NT_STATUS_OK;
-}
-
-/*
   form the lock context used for byte range locking. This is separate
   from the locking key used for opendb locking as it needs to take
   account of file streams (each stream is a separate byte range
@@ -451,6 +899,8 @@ static NTSTATUS pvfs_create_file(struct pvfs_state *pvfs,
 	uint32_t access_mask = io->generic.in.access_mask;
 	mode_t mode;
 	uint32_t attrib;
+	BOOL del_on_close;
+	struct pvfs_filename *parent;
 
 	if ((io->ntcreatex.in.file_attr & FILE_ATTRIBUTE_READONLY) &&
 	    (create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE)) {
@@ -458,8 +908,20 @@ static NTSTATUS pvfs_create_file(struct pvfs_state *pvfs,
 	}
 	
 	status = pvfs_access_check_create(pvfs, req, name, &access_mask);
-	if (!NT_STATUS_IS_OK(status)) {
-		return status;
+	NT_STATUS_NOT_OK_RETURN(status);
+
+	/* check that the parent isn't opened with delete on close set */
+	status = pvfs_resolve_parent(pvfs, req, name, &parent);
+	if (NT_STATUS_IS_OK(status)) {
+		DATA_BLOB locking_key;
+		status = pvfs_locking_key(parent, req, &locking_key);
+		NT_STATUS_NOT_OK_RETURN(status);
+		status = odb_get_delete_on_close(pvfs->odb_context, &locking_key, 
+						 &del_on_close, NULL, NULL);
+		NT_STATUS_NOT_OK_RETURN(status);
+		if (del_on_close) {
+			return NT_STATUS_DELETE_PENDING;
+		}
 	}
 
 	if (access_mask & (SEC_FILE_WRITE_DATA | SEC_FILE_APPEND_DATA)) {
@@ -548,8 +1010,14 @@ static NTSTATUS pvfs_create_file(struct pvfs_state *pvfs,
 		goto cleanup_delete;
 	}
 
+	if (create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) {
+		del_on_close = True;
+	} else {
+		del_on_close = False;
+	}
+
 	status = odb_open_file(lck, f->handle, name->stream_id,
-			       share_access, create_options, access_mask);
+			       share_access, access_mask, del_on_close, name->full_name);
 	talloc_free(lck);
 	if (!NT_STATUS_IS_OK(status)) {
 		/* bad news, we must have hit a race - we don't delete the file
@@ -1037,10 +1505,9 @@ NTSTATUS pvfs_open(struct ntvfs_module_context *ntvfs,
 	talloc_set_destructor(f, pvfs_fnum_destructor);
 	talloc_set_destructor(f->handle, pvfs_handle_destructor);
 
-
 	/* see if we are allowed to open at the same time as existing opens */
 	status = odb_open_file(lck, f->handle, f->handle->name->stream_id,
-			       share_access, create_options, access_mask);
+			       share_access, access_mask, False, name->full_name);
 
 	/* on a sharing violation we need to retry when the file is closed by 
 	   the other user, or after 1 second */
@@ -1222,41 +1689,36 @@ NTSTATUS pvfs_exit(struct ntvfs_module_context *ntvfs,
 
 
 /*
-  change the create options on an already open file
+  change the delete on close flag on an already open file
 */
-NTSTATUS pvfs_change_create_options(struct pvfs_state *pvfs,
-				    struct smbsrv_request *req, 
-				    struct pvfs_file *f, uint32_t create_options)
+NTSTATUS pvfs_set_delete_on_close(struct pvfs_state *pvfs,
+				  struct smbsrv_request *req, 
+				  struct pvfs_file *f, BOOL del_on_close)
 {
 	struct odb_lock *lck;
 	NTSTATUS status;
 
-	if (f->handle->create_options == create_options) {
-		return NT_STATUS_OK;
-	}
-
-	if ((f->handle->name->dos.attrib & FILE_ATTRIBUTE_READONLY) &&
-	    (create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE)) {
+	if ((f->handle->name->dos.attrib & FILE_ATTRIBUTE_READONLY) && del_on_close) {
 		return NT_STATUS_CANNOT_DELETE;
 	}
-
-	if (f->handle->name->dos.attrib & FILE_ATTRIBUTE_DIRECTORY) {
-		if (!pvfs_directory_empty(pvfs, f->handle->name)) {
-			return NT_STATUS_DIRECTORY_NOT_EMPTY;
-		}
-		f->handle->create_options = create_options;
-		return NT_STATUS_OK;
+	
+	if ((f->handle->name->dos.attrib & FILE_ATTRIBUTE_DIRECTORY) &&
+	    !pvfs_directory_empty(pvfs, f->handle->name)) {
+		return NT_STATUS_DIRECTORY_NOT_EMPTY;
 	}
 
+	if (del_on_close) {
+		f->handle->create_options |= NTCREATEX_OPTIONS_DELETE_ON_CLOSE;
+	} else {
+		f->handle->create_options &= ~NTCREATEX_OPTIONS_DELETE_ON_CLOSE;
+	}
+	
 	lck = odb_lock(req, pvfs->odb_context, &f->handle->odb_locking_key);
 	if (lck == NULL) {
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
 	}
 
-	status = odb_set_create_options(lck, f->handle, create_options);
-	if (NT_STATUS_IS_OK(status)) {
-		f->handle->create_options = create_options;
-	}
+	status = odb_set_delete_on_close(lck, del_on_close);
 
 	talloc_free(lck);
 
@@ -1270,17 +1732,25 @@ NTSTATUS pvfs_change_create_options(struct pvfs_state *pvfs,
 */
 NTSTATUS pvfs_can_delete(struct pvfs_state *pvfs, 
 			 struct smbsrv_request *req,
-			 struct pvfs_filename *name)
+			 struct pvfs_filename *name,
+			 struct odb_lock **lckp)
 {
 	NTSTATUS status;
 	DATA_BLOB key;
+	struct odb_lock *lck;
 
 	status = pvfs_locking_key(name, name, &key);
 	if (!NT_STATUS_IS_OK(status)) {
 		return NT_STATUS_NO_MEMORY;
 	}
 
-	status = odb_can_open(pvfs->odb_context, &key, 
+	lck = odb_lock(req, pvfs->odb_context, &key);
+	if (lck == NULL) {
+		DEBUG(0,("Unable to lock opendb for can_delete\n"));
+		return NT_STATUS_INTERNAL_DB_CORRUPTION;
+	}
+
+	status = odb_can_open(lck,
 			      NTCREATEX_SHARE_ACCESS_READ |
 			      NTCREATEX_SHARE_ACCESS_WRITE | 
 			      NTCREATEX_SHARE_ACCESS_DELETE, 
@@ -1291,6 +1761,13 @@ NTSTATUS pvfs_can_delete(struct pvfs_state *pvfs,
 		status = pvfs_access_check_simple(pvfs, req, name, SEC_STD_DELETE);
 	}
 
+	if (!NT_STATUS_IS_OK(status)) {
+		talloc_free(lck);
+		*lckp = lck;
+	} else if (lckp != NULL) {
+		*lckp = lck;
+	}
+
 	return status;
 }
 
@@ -1298,21 +1775,89 @@ NTSTATUS pvfs_can_delete(struct pvfs_state *pvfs,
   determine if a file can be renamed, or if it is prevented by an
   already open file
 */
-NTSTATUS pvfs_can_rename(struct pvfs_state *pvfs, struct pvfs_filename *name)
+NTSTATUS pvfs_can_rename(struct pvfs_state *pvfs, 
+			 struct smbsrv_request *req,
+			 struct pvfs_filename *name,
+			 struct odb_lock **lckp)
 {
 	NTSTATUS status;
 	DATA_BLOB key;
+	struct odb_lock *lck;
 
 	status = pvfs_locking_key(name, name, &key);
 	if (!NT_STATUS_IS_OK(status)) {
 		return NT_STATUS_NO_MEMORY;
 	}
 
-	status = odb_can_open(pvfs->odb_context, &key, 
+	lck = odb_lock(req, pvfs->odb_context, &key);
+	if (lck == NULL) {
+		DEBUG(0,("Unable to lock opendb for can_stat\n"));
+		return NT_STATUS_INTERNAL_DB_CORRUPTION;
+	}
+
+	status = odb_can_open(lck,
 			      NTCREATEX_SHARE_ACCESS_READ |
 			      NTCREATEX_SHARE_ACCESS_WRITE,
 			      0,
 			      SEC_STD_DELETE);
 
+	if (!NT_STATUS_IS_OK(status)) {
+		talloc_free(lck);
+		*lckp = lck;
+	} else if (lckp != NULL) {
+		*lckp = lck;
+	}
+
 	return status;
 }
+
+/*
+  determine if file meta data can be accessed, or if it is prevented by an
+  already open file
+*/
+NTSTATUS pvfs_can_stat(struct pvfs_state *pvfs, 
+		       struct smbsrv_request *req,
+		       struct pvfs_filename *name)
+{
+	NTSTATUS status;
+	DATA_BLOB key;
+	struct odb_lock *lck;
+
+	status = pvfs_locking_key(name, name, &key);
+	if (!NT_STATUS_IS_OK(status)) {
+		return NT_STATUS_NO_MEMORY;
+	}
+
+	lck = odb_lock(req, pvfs->odb_context, &key);
+	if (lck == NULL) {
+		DEBUG(0,("Unable to lock opendb for can_stat\n"));
+		return NT_STATUS_INTERNAL_DB_CORRUPTION;
+	}
+
+	status = odb_can_open(lck,
+			      NTCREATEX_SHARE_ACCESS_READ |
+			      NTCREATEX_SHARE_ACCESS_WRITE,
+			      0, 0);
+
+	return status;
+}
+
+
+/*
+  determine if delete on close is set on 
+*/
+BOOL pvfs_delete_on_close_set(struct pvfs_state *pvfs, struct pvfs_file_handle *h, 
+			      int *open_count, char **path)
+{
+	NTSTATUS status;
+	BOOL del_on_close;
+
+	status = odb_get_delete_on_close(pvfs->odb_context, &h->odb_locking_key, 
+					 &del_on_close, open_count, path);
+	if (!NT_STATUS_IS_OK(status)) {
+		DEBUG(1,("WARNING: unable to determine delete on close status for open file\n"));
+		return False;
+	}
+
+	return del_on_close;
+}
diff --git a/source4/ntvfs/posix/pvfs_qfileinfo.c b/source4/ntvfs/posix/pvfs_qfileinfo.c
index 1f2f86e953..ca26331373 100644
--- a/source4/ntvfs/posix/pvfs_qfileinfo.c
+++ b/source4/ntvfs/posix/pvfs_qfileinfo.c
@@ -185,7 +185,7 @@ static NTSTATUS pvfs_map_fileinfo(struct pvfs_state *pvfs,
 	case RAW_FILEINFO_STANDARD_INFORMATION:
 		info->standard_info.out.alloc_size     = name->dos.alloc_size;
 		info->standard_info.out.size           = name->st.st_size;
-		info->standard_info.out.nlink          = name->st.st_nlink;
+		info->standard_info.out.nlink          = name->dos.nlink;
 		info->standard_info.out.delete_pending = 0;
 		info->standard_info.out.directory   = 
 			(name->dos.attrib & FILE_ATTRIBUTE_DIRECTORY)? 1 : 0;
@@ -210,7 +210,7 @@ static NTSTATUS pvfs_map_fileinfo(struct pvfs_state *pvfs,
 		info->all_info.out.attrib         = name->dos.attrib;
 		info->all_info.out.alloc_size     = name->dos.alloc_size;
 		info->all_info.out.size           = name->st.st_size;
-		info->all_info.out.nlink          = name->st.st_nlink;
+		info->all_info.out.nlink          = name->dos.nlink;
 		info->all_info.out.delete_pending = 0;
 		info->all_info.out.directory      = 
 			(name->dos.attrib & FILE_ATTRIBUTE_DIRECTORY)? 1 : 0;
@@ -298,6 +298,11 @@ NTSTATUS pvfs_qpathinfo(struct ntvfs_module_context *ntvfs,
 		return NT_STATUS_OBJECT_NAME_NOT_FOUND;
 	}
 
+	status = pvfs_can_stat(pvfs, req, name);
+	if (!NT_STATUS_IS_OK(status)) {
+		return NT_STATUS_DELETE_PENDING;
+	}
+
 	status = pvfs_access_check_simple(pvfs, req, name, 
 					  pvfs_fileinfo_access(info->generic.level));
 	if (!NT_STATUS_IS_OK(status)) {
@@ -345,7 +350,7 @@ NTSTATUS pvfs_qfileinfo(struct ntvfs_module_context *ntvfs,
 	switch (info->generic.level) {
 	case RAW_FILEINFO_STANDARD_INFO:
 	case RAW_FILEINFO_STANDARD_INFORMATION:
-		if (h->create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) {
+		if (pvfs_delete_on_close_set(pvfs, h, NULL, NULL)) {
 			info->standard_info.out.delete_pending = 1;
 			info->standard_info.out.nlink--;
 		}
@@ -353,7 +358,7 @@ NTSTATUS pvfs_qfileinfo(struct ntvfs_module_context *ntvfs,
 
 	case RAW_FILEINFO_ALL_INFO:
 	case RAW_FILEINFO_ALL_INFORMATION:
-		if (h->create_options & NTCREATEX_OPTIONS_DELETE_ON_CLOSE) {
+		if (pvfs_delete_on_close_set(pvfs, h, NULL, NULL)) {
 			info->all_info.out.delete_pending = 1;
 			info->all_info.out.nlink--;
 		}
diff --git a/source4/ntvfs/posix/pvfs_rename.c b/source4/ntvfs/posix/pvfs_rename.c
index b70f129888..7f1f43e719 100644
--- a/source4/ntvfs/posix/pvfs_rename.c
+++ b/source4/ntvfs/posix/pvfs_rename.c
@@ -130,6 +130,7 @@ static NTSTATUS pvfs_rename_one(struct pvfs_state *pvfs,
 	struct pvfs_filename *name1, *name2;
 	TALLOC_CTX *mem_ctx = talloc_new(req);
 	NTSTATUS status;
+	struct odb_lock *lck, *lck2;
 
 	/* resolve the wildcard pattern for this name */
 	fname2 = pvfs_resolve_wildcard(mem_ctx, fname1, fname2);
@@ -150,7 +151,7 @@ static NTSTATUS pvfs_rename_one(struct pvfs_state *pvfs,
 		goto failed;
 	}
 
-	status = pvfs_can_rename(pvfs, name1);
+	status = pvfs_can_rename(pvfs, req, name1, &lck);
 	if (!NT_STATUS_IS_OK(status)) {
 		goto failed;
 	}
@@ -159,7 +160,7 @@ static NTSTATUS pvfs_rename_one(struct pvfs_state *pvfs,
 	status = pvfs_resolve_partial(pvfs, mem_ctx, 
 				      dir_path, fname2, &name2);
 	if (NT_STATUS_IS_OK(status)) {
-		status = pvfs_can_delete(pvfs, req, name2);
+		status = pvfs_can_delete(pvfs, req, name2, &lck2);
 		if (!NT_STATUS_IS_OK(status)) {
 			goto failed;
 		}
@@ -177,6 +178,8 @@ static NTSTATUS pvfs_rename_one(struct pvfs_state *pvfs,
 		return pvfs_map_errno(pvfs, errno);
 	}
 
+	status = odb_rename(lck, fname2);
+
 failed:
 	talloc_free(mem_ctx);
 	return status;
@@ -246,6 +249,7 @@ static NTSTATUS pvfs_rename_mv(struct ntvfs_module_context *ntvfs,
 	struct pvfs_state *pvfs = ntvfs->private_data;
 	NTSTATUS status;
 	struct pvfs_filename *name1, *name2;
+	struct odb_lock *lck;
 
 	/* resolve the cifs name to a posix name */
 	status = pvfs_resolve_name(pvfs, req, ren->rename.in.pattern1, 
@@ -286,7 +290,7 @@ static NTSTATUS pvfs_rename_mv(struct ntvfs_module_context *ntvfs,
 		return status;
 	}
 
-	status = pvfs_can_rename(pvfs, name1);
+	status = pvfs_can_rename(pvfs, req, name1, &lck);
 	if (!NT_STATUS_IS_OK(status)) {
 		return status;
 	}
@@ -294,6 +298,8 @@ static NTSTATUS pvfs_rename_mv(struct ntvfs_module_context *ntvfs,
 	if (rename(name1->full_name, name2->full_name) == -1) {
 		return pvfs_map_errno(pvfs, errno);
 	}
+
+	status = odb_rename(lck, name2->full_name);
 	
 	return NT_STATUS_OK;
 }
@@ -308,6 +314,7 @@ static NTSTATUS pvfs_rename_nt(struct ntvfs_module_context *ntvfs,
 	struct pvfs_state *pvfs = ntvfs->private_data;
 	NTSTATUS status;
 	struct pvfs_filename *name1, *name2;
+	struct odb_lock *lck;
 
 	switch (ren->ntrename.in.flags) {
 	case RENAME_FLAG_RENAME:
@@ -353,7 +360,7 @@ static NTSTATUS pvfs_rename_nt(struct ntvfs_module_context *ntvfs,
 		return status;
 	}
 
-	status = pvfs_can_rename(pvfs, name1);
+	status = pvfs_can_rename(pvfs, req, name1, &lck);
 	if (!NT_STATUS_IS_OK(status)) {
 		return status;
 	}
diff --git a/source4/ntvfs/posix/pvfs_setfileinfo.c b/source4/ntvfs/posix/pvfs_setfileinfo.c
index 0753ccb40b..69c9cd5e4a 100644
--- a/source4/ntvfs/posix/pvfs_setfileinfo.c
+++ b/source4/ntvfs/posix/pvfs_setfileinfo.c
@@ -121,6 +121,8 @@ static NTSTATUS pvfs_setfileinfo_rename(struct pvfs_state *pvfs,
 
 	/* if the destination exists, then check the rename is allowed */
 	if (name2->exists) {
+		struct odb_lock *lck;
+
 		if (strcmp(name2->full_name, name->full_name) == 0) {
 			/* rename to same name is null-op */
 			return NT_STATUS_OK;
@@ -130,7 +132,7 @@ static NTSTATUS pvfs_setfileinfo_rename(struct pvfs_state *pvfs,
 			return NT_STATUS_OBJECT_NAME_COLLISION;
 		}
 
-		status = pvfs_can_delete(pvfs, req, name2);
+		status = pvfs_can_delete(pvfs, req, name2, &lck);
 		if (NT_STATUS_EQUAL(status, NT_STATUS_SHARING_VIOLATION)) {
 			return NT_STATUS_ACCESS_DENIED;
 		}
@@ -243,7 +245,6 @@ NTSTATUS pvfs_setfileinfo(struct ntvfs_module_context *ntvfs,
 	struct utimbuf unix_times;
 	struct pvfs_file *f;
 	struct pvfs_file_handle *h;
-	uint32_t create_options;
 	struct pvfs_filename newstats;
 	NTSTATUS status;
 	uint32_t access_needed;
@@ -322,13 +323,8 @@ NTSTATUS pvfs_setfileinfo(struct ntvfs_module_context *ntvfs,
 
 	case RAW_SFILEINFO_DISPOSITION_INFO:
 	case RAW_SFILEINFO_DISPOSITION_INFORMATION:
-		create_options = h->create_options;
-		if (info->disposition_info.in.delete_on_close) {
-			create_options |= NTCREATEX_OPTIONS_DELETE_ON_CLOSE;
-		} else {
-			create_options &= ~NTCREATEX_OPTIONS_DELETE_ON_CLOSE;
-		}
-		return pvfs_change_create_options(pvfs, req, f, create_options);
+		return pvfs_set_delete_on_close(pvfs, req, f, 
+						info->disposition_info.in.delete_on_close);
 
 	case RAW_SFILEINFO_ALLOCATION_INFO:
 	case RAW_SFILEINFO_ALLOCATION_INFORMATION:
diff --git a/source4/ntvfs/posix/pvfs_unlink.c b/source4/ntvfs/posix/pvfs_unlink.c
index 5f8a828f15..8eea2c47b8 100644
--- a/source4/ntvfs/posix/pvfs_unlink.c
+++ b/source4/ntvfs/posix/pvfs_unlink.c
@@ -33,6 +33,7 @@ static NTSTATUS pvfs_unlink_stream(struct pvfs_state *pvfs,
 				   uint16_t attrib)
 {
 	NTSTATUS status;
+	struct odb_lock *lck;
 
 	if (!name->stream_exists) {
 		return NT_STATUS_OBJECT_NAME_NOT_FOUND;
@@ -44,7 +45,7 @@ static NTSTATUS pvfs_unlink_stream(struct pvfs_state *pvfs,
 		return status;
 	}
 
-	status = pvfs_can_delete(pvfs, req, name);
+	status = pvfs_can_delete(pvfs, req, name, &lck);
 	if (!NT_STATUS_IS_OK(status)) {
 		return status;
 	}
@@ -63,6 +64,7 @@ static NTSTATUS pvfs_unlink_one(struct pvfs_state *pvfs,
 {
 	struct pvfs_filename *name;
 	NTSTATUS status;
+	struct odb_lock *lck;
 
 	/* get a pvfs_filename object */
 	status = pvfs_resolve_partial(pvfs, req, 
@@ -78,7 +80,7 @@ static NTSTATUS pvfs_unlink_one(struct pvfs_state *pvfs,
 		return status;
 	}
 
-	status = pvfs_can_delete(pvfs, req, name);
+	status = pvfs_can_delete(pvfs, req, name, &lck);
 	if (!NT_STATUS_IS_OK(status)) {
 		talloc_free(name);
 		return status;