source4/cluster and source4/ntvfs: convert to dbwrap, add ntdb option.

This makes the code use dbwrap_local_open(), so it can handle
NTDB.

brlock.tdb, notify.tdb and openfiles.tdb can now be brlock.ntdb,
notify.ntdb and openfiles.ntdb, if 'use ntdb' is set.

Cc: Andrew Bartlett <abartlet@samba.org>
Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>
Reviewed-by: Jeremy Allison <jra@samba.org>

3 files changed, 129 insertions, 160 deletions

diff --git a/source4/ntvfs/common/brlock_tdb.c b/source4/ntvfs/common/brlock_tdb.c
index c6d736e4b84..56cf26c70b9 100644
--- a/source4/ntvfs/common/brlock_tdb.c
+++ b/source4/ntvfs/common/brlock_tdb.c
@@ -26,15 +26,14 @@
 
 #include "includes.h"
 #include "system/filesys.h"
-#include "tdb_compat.h"
 #include "messaging/messaging.h"
-#include "lib/tdb_wrap/tdb_wrap.h"
 #include "lib/messaging/irpc.h"
 #include "libcli/libcli.h"
 #include "cluster/cluster.h"
 #include "ntvfs/common/brlock.h"
 #include "ntvfs/ntvfs.h"
 #include "param/param.h"
+#include "dbwrap/dbwrap.h"
 
 /*
   in this module a "DATA_BLOB *file_key" is a blob that uniquely identifies
@@ -46,7 +45,7 @@
 
 /* this struct is typicaly attached to tcon */
 struct brl_context {
-	struct tdb_wrap *w;
+	struct db_context *db;
 	struct server_id server;
 	struct imessaging_context *imessaging_ctx;
 };
@@ -103,8 +102,8 @@ static struct brl_context *brl_tdb_init(TALLOC_CTX *mem_ctx, struct server_id se
 		return NULL;
 	}
 
-	brl->w = cluster_tdb_tmp_open(brl, lp_ctx, "brlock.tdb", TDB_DEFAULT);
-	if (brl->w == NULL) {
+	brl->db = cluster_db_tmp_open(brl, lp_ctx, "brlock", TDB_DEFAULT);
+	if (brl->db == NULL) {
 		talloc_free(brl);
 		return NULL;
 	}
@@ -302,6 +301,7 @@ static NTSTATUS brl_tdb_lock(struct brl_context *brl,
 	int count=0, i;
 	struct lock_struct lock, *locks=NULL;
 	NTSTATUS status;
+	struct db_record *locked;
 
 	kbuf.dptr = brlh->key.data;
 	kbuf.dsize = brlh->key.length;
@@ -310,7 +310,8 @@ static NTSTATUS brl_tdb_lock(struct brl_context *brl,
 		return NT_STATUS_INVALID_LOCK_RANGE;
 	}
 
-	if (tdb_chainlock(brl->w->tdb, kbuf) != 0) {
+	locked = dbwrap_fetch_locked(brl->db, brl, kbuf);
+	if (!locked) {
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
 	}
 
@@ -328,12 +329,12 @@ static NTSTATUS brl_tdb_lock(struct brl_context *brl,
 		brlh->last_lock = lock;
 
 		if (NT_STATUS_IS_OK(status)) {
-			tdb_chainunlock(brl->w->tdb, kbuf);
+			talloc_free(locked);
 			return NT_STATUS_OK;
 		}
 	}
 
-	dbuf = tdb_fetch_compat(brl->w->tdb, kbuf);
+	dbuf = dbwrap_record_get_value(locked);
 
 	lock.context.smbpid = smbpid;
 	lock.context.server = brl->server;
@@ -358,23 +359,24 @@ static NTSTATUS brl_tdb_lock(struct brl_context *brl,
 	}
 
 	/* no conflicts - add it to the list of locks */
-	locks = realloc_p(locks, struct lock_struct, count+1);
+	/* FIXME: a dbwrap_record_append() would help here! */
+	locks = talloc_array(locked, struct lock_struct, count+1);
 	if (!locks) {
 		status = NT_STATUS_NO_MEMORY;
 		goto fail;
-	} else {
-		dbuf.dptr = (uint8_t *)locks;
 	}
+	memcpy(locks, dbuf.dptr, dbuf.dsize);
 	locks[count] = lock;
+
+	dbuf.dptr = (unsigned char *)locks;
 	dbuf.dsize += sizeof(lock);
 
-	if (tdb_store(brl->w->tdb, kbuf, dbuf, TDB_REPLACE) != 0) {
-		status = NT_STATUS_INTERNAL_DB_CORRUPTION;
+	status = dbwrap_record_store(locked, dbuf, TDB_REPLACE);
+	if (!NT_STATUS_IS_OK(status)) {
 		goto fail;
 	}
 
-	free(dbuf.dptr);
-	tdb_chainunlock(brl->w->tdb, kbuf);
+	talloc_free(locked);
 
 	/* the caller needs to know if the real lock was granted. If
 	   we have reached here then it must be a pending lock that
@@ -386,9 +388,7 @@ static NTSTATUS brl_tdb_lock(struct brl_context *brl,
 	return NT_STATUS_OK;
 
  fail:
-
-	free(dbuf.dptr);
-	tdb_chainunlock(brl->w->tdb, kbuf);
+	talloc_free(locked);
 	return status;
 }
 
@@ -455,6 +455,7 @@ static NTSTATUS brl_tdb_unlock(struct brl_context *brl,
 	int count, i;
 	struct lock_struct *locks, *lock;
 	struct lock_context context;
+	struct db_record *locked;
 	NTSTATUS status;
 
 	kbuf.dptr = brlh->key.data;
@@ -464,15 +465,11 @@ static NTSTATUS brl_tdb_unlock(struct brl_context *brl,
 		return NT_STATUS_INVALID_LOCK_RANGE;
 	}
 
-	if (tdb_chainlock(brl->w->tdb, kbuf) != 0) {
+	locked = dbwrap_fetch_locked(brl->db, brl, kbuf);
+	if (!locked) {
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
 	}
-
-	dbuf = tdb_fetch_compat(brl->w->tdb, kbuf);
-	if (!dbuf.dptr) {
-		tdb_chainunlock(brl->w->tdb, kbuf);
-		return NT_STATUS_RANGE_NOT_LOCKED;
-	}
+	dbuf = dbwrap_record_get_value(locked);
 
 	context.smbpid = smbpid;
 	context.server = brl->server;
@@ -509,8 +506,8 @@ found:
 	if (i < count) {
 		/* found it - delete it */
 		if (count == 1) {
-			if (tdb_delete(brl->w->tdb, kbuf) != 0) {
-				status = NT_STATUS_INTERNAL_DB_CORRUPTION;
+			status = dbwrap_record_delete(locked);
+			if (!NT_STATUS_IS_OK(status)) {
 				goto fail;
 			}
 		} else {
@@ -525,15 +522,14 @@ found:
 			brl_tdb_notify_unlock(brl, locks, count, &removed_lock);
 			
 			dbuf.dsize = count * sizeof(*locks);
-			
-			if (tdb_store(brl->w->tdb, kbuf, dbuf, TDB_REPLACE) != 0) {
-				status = NT_STATUS_INTERNAL_DB_CORRUPTION;
+
+			status = dbwrap_record_store(locked, dbuf, TDB_REPLACE);
+			if (!NT_STATUS_IS_OK(status)) {
 				goto fail;
 			}
 		}
-		
-		free(dbuf.dptr);
-		tdb_chainunlock(brl->w->tdb, kbuf);
+
+		talloc_free(locked);
 		return NT_STATUS_OK;
 	}
 	
@@ -541,8 +537,7 @@ found:
 	status = NT_STATUS_RANGE_NOT_LOCKED;
 
  fail:
-	free(dbuf.dptr);
-	tdb_chainunlock(brl->w->tdb, kbuf);
+	talloc_free(locked);
 	return status;
 }
 
@@ -560,17 +555,19 @@ static NTSTATUS brl_tdb_remove_pending(struct brl_context *brl,
 	int count, i;
 	struct lock_struct *locks;
 	NTSTATUS status;
+	struct db_record *locked;
 
 	kbuf.dptr = brlh->key.data;
 	kbuf.dsize = brlh->key.length;
 
-	if (tdb_chainlock(brl->w->tdb, kbuf) != 0) {
+	locked = dbwrap_fetch_locked(brl->db, brl, kbuf);
+	if (!locked) {
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
 	}
 
-	dbuf = tdb_fetch_compat(brl->w->tdb, kbuf);
+	dbuf = dbwrap_record_get_value(locked);
 	if (!dbuf.dptr) {
-		tdb_chainunlock(brl->w->tdb, kbuf);
+		talloc_free(locked);
 		return NT_STATUS_RANGE_NOT_LOCKED;
 	}
 
@@ -586,8 +583,8 @@ static NTSTATUS brl_tdb_remove_pending(struct brl_context *brl,
 		    cluster_id_equal(&lock->context.server, &brl->server)) {
 			/* found it - delete it */
 			if (count == 1) {
-				if (tdb_delete(brl->w->tdb, kbuf) != 0) {
-					status = NT_STATUS_INTERNAL_DB_CORRUPTION;
+				status = dbwrap_record_delete(locked);
+				if (!NT_STATUS_IS_OK(status)) {
 					goto fail;
 				}
 			} else {
@@ -597,14 +594,14 @@ static NTSTATUS brl_tdb_remove_pending(struct brl_context *brl,
 				}
 				count--;
 				dbuf.dsize = count * sizeof(*locks);
-				if (tdb_store(brl->w->tdb, kbuf, dbuf, TDB_REPLACE) != 0) {
-					status = NT_STATUS_INTERNAL_DB_CORRUPTION;
+				status = dbwrap_record_store(locked, dbuf,
+							     TDB_REPLACE);
+				if (!NT_STATUS_IS_OK(status)) {
 					goto fail;
 				}
 			}
 			
-			free(dbuf.dptr);
-			tdb_chainunlock(brl->w->tdb, kbuf);
+			talloc_free(locked);
 			return NT_STATUS_OK;
 		}
 	}
@@ -613,8 +610,7 @@ static NTSTATUS brl_tdb_remove_pending(struct brl_context *brl,
 	status = NT_STATUS_RANGE_NOT_LOCKED;
 
  fail:
-	free(dbuf.dptr);
-	tdb_chainunlock(brl->w->tdb, kbuf);
+	talloc_free(locked);
 	return status;
 }
 
@@ -631,6 +627,7 @@ static NTSTATUS brl_tdb_locktest(struct brl_context *brl,
 	TDB_DATA kbuf, dbuf;
 	int count, i;
 	struct lock_struct lock, *locks;
+	NTSTATUS status;
 
 	kbuf.dptr = brlh->key.data;
 	kbuf.dsize = brlh->key.length;
@@ -639,9 +636,11 @@ static NTSTATUS brl_tdb_locktest(struct brl_context *brl,
 		return NT_STATUS_INVALID_LOCK_RANGE;
 	}
 
-	dbuf = tdb_fetch_compat(brl->w->tdb, kbuf);
-	if (dbuf.dptr == NULL) {
+	status = dbwrap_fetch(brl->db, brl, kbuf, &dbuf);
+	if (NT_STATUS_EQUAL(status, NT_STATUS_NOT_FOUND)) {
 		return NT_STATUS_OK;
+	} else if (!NT_STATUS_IS_OK(status)) {
+		return status;
 	}
 
 	lock.context.smbpid = smbpid;
@@ -658,12 +657,12 @@ static NTSTATUS brl_tdb_locktest(struct brl_context *brl,
 
 	for (i=0; i<count; i++) {
 		if (brl_tdb_conflict_other(&locks[i], &lock)) {
-			free(dbuf.dptr);
+			talloc_free(dbuf.dptr);
 			return NT_STATUS_FILE_LOCK_CONFLICT;
 		}
 	}
 
-	free(dbuf.dptr);
+	talloc_free(dbuf.dptr);
 	return NT_STATUS_OK;
 }
 
@@ -677,18 +676,19 @@ static NTSTATUS brl_tdb_close(struct brl_context *brl,
 	TDB_DATA kbuf, dbuf;
 	int count, i, dcount=0;
 	struct lock_struct *locks;
+	struct db_record *locked;
 	NTSTATUS status;
 
 	kbuf.dptr = brlh->key.data;
 	kbuf.dsize = brlh->key.length;
 
-	if (tdb_chainlock(brl->w->tdb, kbuf) != 0) {
+	locked = dbwrap_fetch_locked(brl->db, brl, kbuf);
+	if (!locked) {
 		return NT_STATUS_INTERNAL_DB_CORRUPTION;
 	}
-
-	dbuf = tdb_fetch_compat(brl->w->tdb, kbuf);
+	dbuf = dbwrap_record_get_value(locked);
 	if (!dbuf.dptr) {
-		tdb_chainunlock(brl->w->tdb, kbuf);
+		talloc_free(locked);
 		return NT_STATUS_OK;
 	}
 
@@ -716,9 +716,7 @@ static NTSTATUS brl_tdb_close(struct brl_context *brl,
 	status = NT_STATUS_OK;
 
 	if (count == 0) {
-		if (tdb_delete(brl->w->tdb, kbuf) != 0) {
-			status = NT_STATUS_INTERNAL_DB_CORRUPTION;
-		}
+		status = dbwrap_record_delete(locked);
 	} else if (dcount != 0) {
 		/* tell all pending lock holders for this file that
 		   they have a chance now. This is a bit indiscriminant,
@@ -727,13 +725,9 @@ static NTSTATUS brl_tdb_close(struct brl_context *brl,
 
 		dbuf.dsize = count * sizeof(*locks);
 
-		if (tdb_store(brl->w->tdb, kbuf, dbuf, TDB_REPLACE) != 0) {
-			status = NT_STATUS_INTERNAL_DB_CORRUPTION;
-		}
+		status = dbwrap_record_store(locked, dbuf, TDB_REPLACE);
 	}
-
-	free(dbuf.dptr);
-	tdb_chainunlock(brl->w->tdb, kbuf);
+	talloc_free(locked);
 
 	return status;
 }
@@ -742,25 +736,21 @@ static NTSTATUS brl_tdb_count(struct brl_context *brl, struct brl_handle *brlh,
 			      int *count)
 {
 	TDB_DATA kbuf, dbuf;
+	NTSTATUS status;
 
 	kbuf.dptr = brlh->key.data;
 	kbuf.dsize = brlh->key.length;
 	*count = 0;
 
-	if (tdb_chainlock(brl->w->tdb, kbuf) != 0) {
-		return NT_STATUS_INTERNAL_DB_CORRUPTION;
-	}
-
-	dbuf = tdb_fetch_compat(brl->w->tdb, kbuf);
-	if (!dbuf.dptr) {
-		tdb_chainunlock(brl->w->tdb, kbuf);
+	status = dbwrap_fetch(brl->db, brl, kbuf, &dbuf);
+	if (NT_STATUS_EQUAL(status, NT_STATUS_NOT_FOUND)) {
 		return NT_STATUS_OK;
+	} else if (!NT_STATUS_IS_OK(status)) {
+		return status;
 	}
-
 	*count = dbuf.dsize / sizeof(struct lock_struct);
 
-	free(dbuf.dptr);
-	tdb_chainunlock(brl->w->tdb, kbuf);
+	talloc_free(dbuf.dptr);
 
 	return NT_STATUS_OK;
 }
diff --git a/source4/ntvfs/common/notify.c b/source4/ntvfs/common/notify.c
index 6b5ece9e4ec..445f38f742b 100644
--- a/source4/ntvfs/common/notify.c
+++ b/source4/ntvfs/common/notify.c
@@ -25,10 +25,7 @@
 
 #include "includes.h"
 #include "system/filesys.h"
-#include "../lib/tdb_compat/tdb_compat.h"
-#include "../lib/util/util_tdb.h"
 #include "messaging/messaging.h"
-#include "lib/tdb_wrap/tdb_wrap.h"
 #include "lib/messaging/irpc.h"
 #include "librpc/gen_ndr/ndr_notify.h"
 #include "../lib/util/dlinklist.h"
@@ -37,14 +34,16 @@
 #include "cluster/cluster.h"
 #include "param/param.h"
 #include "lib/util/tsort.h"
+#include "lib/dbwrap/dbwrap.h"
+#include "../lib/util/util_tdb.h"
 
 struct notify_context {
-	struct tdb_wrap *w;
+	struct db_context *db;
 	struct server_id server;
 	struct imessaging_context *imessaging_ctx;
 	struct notify_list *list;
 	struct notify_array *array;
-	int seqnum;
+	int64_t seqnum;
 	struct sys_notify_context *sys_notify_ctx;
 };
 
@@ -102,8 +101,8 @@ struct notify_context *notify_init(TALLOC_CTX *mem_ctx, struct server_id server,
 		return NULL;
 	}
 
-	notify->w = cluster_tdb_tmp_open(notify, lp_ctx, "notify.tdb", TDB_SEQNUM);
-	if (notify->w == NULL) {
+	notify->db = cluster_db_tmp_open(notify, lp_ctx, "notify", TDB_SEQNUM);
+	if (notify->db == NULL) {
 		talloc_free(notify);
 		return NULL;
 	}
@@ -112,7 +111,7 @@ struct notify_context *notify_init(TALLOC_CTX *mem_ctx, struct server_id server,
 	notify->imessaging_ctx = imessaging_ctx;
 	notify->list = NULL;
 	notify->array = NULL;
-	notify->seqnum = tdb_get_seqnum(notify->w->tdb);
+	notify->seqnum = dbwrap_get_seqnum(notify->db);
 
 	talloc_set_destructor(notify, notify_destructor);
 
@@ -130,20 +129,16 @@ struct notify_context *notify_init(TALLOC_CTX *mem_ctx, struct server_id server,
 /*
   lock the notify db
 */
-static NTSTATUS notify_lock(struct notify_context *notify)
+static struct db_record *notify_lock(struct notify_context *notify)
 {
-	if (tdb_lock_bystring(notify->w->tdb, NOTIFY_KEY) != 0) {
-		return NT_STATUS_INTERNAL_DB_CORRUPTION;
-	}
-	return NT_STATUS_OK;
+	TDB_DATA key = string_term_tdb_data(NOTIFY_KEY);
+
+	return dbwrap_fetch_locked(notify->db, notify, key);
 }
 
-/*
-  unlock the notify db
-*/
-static void notify_unlock(struct notify_context *notify)
+static void notify_unlock(struct db_record *lock)
 {
-	tdb_unlock_bystring(notify->w->tdb, NOTIFY_KEY);
+	talloc_free(lock);
 }
 
 /*
@@ -155,8 +150,9 @@ static NTSTATUS notify_load(struct notify_context *notify)
 	DATA_BLOB blob;
 	enum ndr_err_code ndr_err;
 	int seqnum;
+	NTSTATUS status;
 
-	seqnum = tdb_get_seqnum(notify->w->tdb);
+	seqnum = dbwrap_get_seqnum(notify->db);
 
 	if (seqnum == notify->seqnum && notify->array != NULL) {
 		return NT_STATUS_OK;
@@ -168,8 +164,8 @@ static NTSTATUS notify_load(struct notify_context *notify)
 	notify->array = talloc_zero(notify, struct notify_array);
 	NT_STATUS_HAVE_NO_MEMORY(notify->array);
 
-	dbuf = tdb_fetch_bystring(notify->w->tdb, NOTIFY_KEY);
-	if (dbuf.dptr == NULL) {
+	status = dbwrap_fetch_bystring(notify->db, notify, NOTIFY_KEY, &dbuf);
+	if (!NT_STATUS_IS_OK(status)) {
 		return NT_STATUS_OK;
 	}
 
@@ -178,7 +174,7 @@ static NTSTATUS notify_load(struct notify_context *notify)
 
 	ndr_err = ndr_pull_struct_blob(&blob, notify->array, notify->array,
 				       (ndr_pull_flags_fn_t)ndr_pull_notify_array);
-	free(dbuf.dptr);
+	talloc_free(dbuf.dptr);
 	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
 		return ndr_map_error2ntstatus(ndr_err);
 	}
@@ -203,8 +199,8 @@ static NTSTATUS notify_save(struct notify_context *notify)
 	TDB_DATA dbuf;
 	DATA_BLOB blob;
 	enum ndr_err_code ndr_err;
-	int ret;
 	TALLOC_CTX *tmp_ctx;
+	NTSTATUS status;
 
 	/* if possible, remove some depth arrays */
 	while (notify->array->num_depths > 0 &&
@@ -214,11 +210,7 @@ static NTSTATUS notify_save(struct notify_context *notify)
 
 	/* we might just be able to delete the record */
 	if (notify->array->num_depths == 0) {
-		ret = tdb_delete_bystring(notify->w->tdb, NOTIFY_KEY);
-		if (ret != 0) {
-			return NT_STATUS_INTERNAL_DB_CORRUPTION;
-		}
-		return NT_STATUS_OK;
+		return dbwrap_delete_bystring(notify->db, NOTIFY_KEY);
 	}
 
 	tmp_ctx = talloc_new(notify);
@@ -233,14 +225,11 @@ static NTSTATUS notify_save(struct notify_context *notify)
 
 	dbuf.dptr = blob.data;
 	dbuf.dsize = blob.length;
-		
-	ret = tdb_store_bystring(notify->w->tdb, NOTIFY_KEY, dbuf, TDB_REPLACE);
-	talloc_free(tmp_ctx);
-	if (ret != 0) {
-		return NT_STATUS_INTERNAL_DB_CORRUPTION;
-	}
 
-	return NT_STATUS_OK;
+	status = dbwrap_store_bystring(notify->db, NOTIFY_KEY, dbuf,
+				       TDB_REPLACE);
+	talloc_free(tmp_ctx);
+	return status;
 }
 
 
@@ -354,14 +343,17 @@ NTSTATUS notify_add(struct notify_context *notify, struct notify_entry *e0,
 	struct notify_list *listel;
 	size_t len;
 	int depth;
+	struct db_record *locked;
 
 	/* see if change notify is enabled at all */
 	if (notify == NULL) {
 		return NT_STATUS_NOT_IMPLEMENTED;
 	}
 
-	status = notify_lock(notify);
-	NT_STATUS_NOT_OK_RETURN(status);
+	locked = notify_lock(notify);
+	if (!locked) {
+		return NT_STATUS_INTERNAL_DB_CORRUPTION;
+	}
 
 	status = notify_load(notify);
 	if (!NT_STATUS_IS_OK(status)) {
@@ -415,7 +407,7 @@ NTSTATUS notify_add(struct notify_context *notify, struct notify_entry *e0,
 	}
 
 done:
-	notify_unlock(notify);
+	notify_unlock(locked);
 	talloc_free(tmp_path);
 
 	return status;
@@ -430,6 +422,7 @@ NTSTATUS notify_remove(struct notify_context *notify, void *private_data)
 	struct notify_list *listel;
 	int i, depth;
 	struct notify_depth *d;
+	struct db_record *locked;
 
 	/* see if change notify is enabled at all */
 	if (notify == NULL) {
@@ -450,17 +443,19 @@ NTSTATUS notify_remove(struct notify_context *notify, void *private_data)
 
 	talloc_free(listel);
 
-	status = notify_lock(notify);
-	NT_STATUS_NOT_OK_RETURN(status);
+	locked = notify_lock(notify);
+	if (!locked) {
+		return NT_STATUS_INTERNAL_DB_CORRUPTION;
+	}
 
 	status = notify_load(notify);
 	if (!NT_STATUS_IS_OK(status)) {
-		notify_unlock(notify);
+		notify_unlock(locked);
 		return status;
 	}
 
 	if (depth >= notify->array->num_depths) {
-		notify_unlock(notify);
+		notify_unlock(locked);
 		return NT_STATUS_OBJECT_NAME_NOT_FOUND;
 	}
 
@@ -474,7 +469,7 @@ NTSTATUS notify_remove(struct notify_context *notify, void *private_data)
 		}
 	}
 	if (i == d->num_entries) {
-		notify_unlock(notify);
+		notify_unlock(locked);
 		return NT_STATUS_OBJECT_NAME_NOT_FOUND;
 	}
 
@@ -486,7 +481,7 @@ NTSTATUS notify_remove(struct notify_context *notify, void *private_data)
 
 	status = notify_save(notify);
 
-	notify_unlock(notify);
+	notify_unlock(locked);
 
 	return status;
 }
@@ -498,17 +493,20 @@ static NTSTATUS notify_remove_all(struct notify_context *notify)
 {
 	NTSTATUS status;
 	int i, depth, del_count=0;
+	struct db_record *locked;
 
 	if (notify->list == NULL) {
 		return NT_STATUS_OK;
 	}
 
-	status = notify_lock(notify);
-	NT_STATUS_NOT_OK_RETURN(status);
+	locked = notify_lock(notify);
+	if (!locked) {
+		return NT_STATUS_INTERNAL_DB_CORRUPTION;
+	}
 
 	status = notify_load(notify);
 	if (!NT_STATUS_IS_OK(status)) {
-		notify_unlock(notify);
+		notify_unlock(locked);
 		return status;
 	}
 
@@ -533,7 +531,7 @@ static NTSTATUS notify_remove_all(struct notify_context *notify)
 		status = notify_save(notify);
 	}
 
-	notify_unlock(notify);
+	notify_unlock(locked);
 
 	return status;
 }
diff --git a/source4/ntvfs/common/opendb_tdb.c b/source4/ntvfs/common/opendb_tdb.c
index ed8fb9032a0..2047a0b3145 100644
--- a/source4/ntvfs/common/opendb_tdb.c
+++ b/source4/ntvfs/common/opendb_tdb.c
@@ -40,9 +40,8 @@
 
 #include "includes.h"
 #include "system/filesys.h"
-#include "../lib/tdb_compat/tdb_compat.h"
+#include "lib/dbwrap/dbwrap.h"
 #include "messaging/messaging.h"
-#include "lib/tdb_wrap/tdb_wrap.h"
 #include "lib/messaging/irpc.h"
 #include "librpc/gen_ndr/ndr_opendb.h"
 #include "ntvfs/ntvfs.h"
@@ -52,7 +51,7 @@
 #include "ntvfs/sysdep/sys_lease.h"
 
 struct odb_context {
-	struct tdb_wrap *w;
+	struct db_context *db;
 	struct ntvfs_context *ntvfs_ctx;
 	bool oplocks;
 	struct sys_lease_context *lease_ctx;
@@ -64,7 +63,7 @@ struct odb_context {
 */
 struct odb_lock {
 	struct odb_context *odb;
-	TDB_DATA key;
+	struct db_record *locked;
 
 	struct opendb_file file;
 
@@ -93,8 +92,9 @@ static struct odb_context *odb_tdb_init(TALLOC_CTX *mem_ctx,
 		return NULL;
 	}
 
-	odb->w = cluster_tdb_tmp_open(odb, ntvfs_ctx->lp_ctx, "openfiles.tdb", TDB_DEFAULT);
-	if (odb->w == NULL) {
+	odb->db = cluster_db_tmp_open(odb, ntvfs_ctx->lp_ctx,
+				      "openfiles", TDB_DEFAULT);
+	if (odb->db == NULL) {
 		talloc_free(odb);
 		return NULL;
 	}
@@ -111,15 +111,6 @@ static struct odb_context *odb_tdb_init(TALLOC_CTX *mem_ctx,
 	return odb;
 }
 
-/*
-  destroy a lock on the database
-*/
-static int odb_lock_destructor(struct odb_lock *lck)
-{
-	tdb_chainunlock(lck->odb->w->tdb, lck->key);
-	return 0;
-}
-
 static NTSTATUS odb_pull_record(struct odb_lock *lck, struct opendb_file *file);
 
 /*
@@ -131,6 +122,7 @@ static struct odb_lock *odb_tdb_lock(TALLOC_CTX *mem_ctx,
 {
 	struct odb_lock *lck;
 	NTSTATUS status;
+	TDB_DATA key;
 
 	lck = talloc(mem_ctx, struct odb_lock);
 	if (lck == NULL) {
@@ -138,22 +130,21 @@ static struct odb_lock *odb_tdb_lock(TALLOC_CTX *mem_ctx,
 	}
 
 	lck->odb = talloc_reference(lck, odb);
-	lck->key.dptr = talloc_memdup(lck, file_key->data, file_key->length);
-	lck->key.dsize = file_key->length;
-	if (lck->key.dptr == NULL) {
+	key.dptr = talloc_memdup(lck, file_key->data, file_key->length);
+	key.dsize = file_key->length;
+	if (key.dptr == NULL) {
 		talloc_free(lck);
 		return NULL;
 	}
 
-	if (tdb_chainlock(odb->w->tdb, lck->key) != 0) {
+	lck->locked = dbwrap_fetch_locked(odb->db, lck, key);
+	if (!lck->locked) {
 		talloc_free(lck);
 		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 */
@@ -168,7 +159,8 @@ static struct odb_lock *odb_tdb_lock(TALLOC_CTX *mem_ctx,
 
 static DATA_BLOB odb_tdb_get_key(TALLOC_CTX *mem_ctx, struct odb_lock *lck)
 {
-	return data_blob_talloc(mem_ctx, lck->key.dptr, lck->key.dsize);
+	TDB_DATA key = dbwrap_record_get_key(lck->locked);
+	return data_blob_talloc(mem_ctx, key.dptr, key.dsize);
 }
 
 
@@ -233,13 +225,12 @@ static NTSTATUS share_conflict(struct opendb_entry *e1,
 */
 static NTSTATUS odb_pull_record(struct odb_lock *lck, struct opendb_file *file)
 {
-	struct odb_context *odb = lck->odb;
 	TDB_DATA dbuf;
 	DATA_BLOB blob;
 	enum ndr_err_code ndr_err;
 
-	dbuf = tdb_fetch_compat(odb->w->tdb, lck->key);
-	if (dbuf.dptr == NULL) {
+	dbuf = dbwrap_record_get_value(lck->locked);
+	if (!dbuf.dptr) {
 		return NT_STATUS_OBJECT_NAME_NOT_FOUND;
 	}
 
@@ -247,7 +238,6 @@ static NTSTATUS odb_pull_record(struct odb_lock *lck, struct opendb_file *file)
 	blob.length = dbuf.dsize;
 
 	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);
 	}
@@ -260,18 +250,13 @@ static NTSTATUS odb_pull_record(struct odb_lock *lck, struct opendb_file *file)
 */
 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;
 	enum ndr_err_code ndr_err;
-	int ret;
+	NTSTATUS status;
 
 	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;
+		return dbwrap_record_delete(lck->locked);
 	}
 
 	ndr_err = ndr_push_struct_blob(&blob, lck, file, (ndr_push_flags_fn_t)ndr_push_opendb_file);
@@ -282,13 +267,9 @@ static NTSTATUS odb_push_record(struct odb_lock *lck, struct opendb_file *file)
 	dbuf.dptr = blob.data;
 	dbuf.dsize = blob.length;
 		
-	ret = tdb_store(odb->w->tdb, lck->key, dbuf, TDB_REPLACE);
+	status = dbwrap_record_store(lck->locked, dbuf, TDB_REPLACE);
 	data_blob_free(&blob);
-	if (ret != 0) {
-		return NT_STATUS_INTERNAL_DB_CORRUPTION;
-	}
-
-	return NT_STATUS_OK;
+	return status;
 }
 
 /*