Port nested transaction allow from newer ctdb.

ie.  f1c6f7dd47bb1081781c0a0d567a92bbbc0aa5d5 (Revert "add TDB_NO_NESTING...")
and  3e49e41c21eb8c53084aa8cc7fd3557bdd8eb7b6 (New attempt at TDB transaction nesting allow/disallow).

ccan/tdb/tdb.h

@@ -55,12 +55,13 @@ extern "C" {
 #define TDB_NOSYNC   64 /* don't use synchronous transactions */
 #define TDB_SEQNUM   128 /* maintain a sequence number */
 #define TDB_VOLATILE   256 /* Activate the per-hashchain freelist, default 5 */
-#define TDB_NO_NESTING 512 /* Dont allow transaction nesting */
+#define TDB_ALLOW_NESTING 512 /* Allow transactions to nest */
 
 /* error codes */
 enum TDB_ERROR {TDB_SUCCESS=0, TDB_ERR_CORRUPT, TDB_ERR_IO, TDB_ERR_LOCK, 
 		TDB_ERR_OOM, TDB_ERR_EXISTS, TDB_ERR_NOLOCK, TDB_ERR_LOCK_TIMEOUT,
-		TDB_ERR_NOEXIST, TDB_ERR_EINVAL, TDB_ERR_RDONLY};
+		TDB_ERR_NOEXIST, TDB_ERR_EINVAL, TDB_ERR_RDONLY,
+		TDB_ERR_NESTING};
 
 /* debugging uses one of the following levels */
 enum tdb_debug_level {TDB_DEBUG_FATAL = 0, TDB_DEBUG_ERROR, 

ccan/tdb/test/run-nested-transactions.c

@@ -0,0 +1,73 @@
+#define _XOPEN_SOURCE 500
+#include "tdb/tdb.h"
+#include "tdb/io.c"
+#include "tdb/tdb.c"
+#include "tdb/lock.c"
+#include "tdb/freelist.c"
+#include "tdb/traverse.c"
+#include "tdb/transaction.c"
+#include "tdb/error.c"
+#include "tdb/open.c"
+#include "tap/tap.h"
+#include <stdlib.h>
+#include <stdbool.h>
+#include <err.h>
+int main(int argc, char *argv[])
+{
+	struct tdb_context *tdb;
+	TDB_DATA key, data;
+
+	plan_tests(27);
+	key.dsize = strlen("hi");
+	key.dptr = (void *)"hi";
+
+	tdb = tdb_open("/tmp/test4.tdb", 1024, TDB_CLEAR_IF_FIRST,
+		       O_CREAT|O_TRUNC|O_RDWR, 0600);
+	ok1(tdb);
+
+	/* No nesting by default. */
+	ok1(tdb_transaction_start(tdb) == 0);
+	data.dptr = (void *)"world";
+	data.dsize = strlen("world");
+	ok1(tdb_store(tdb, key, data, TDB_INSERT) == 0);
+	data = tdb_fetch(tdb, key);
+	ok1(data.dsize == strlen("world"));
+	ok1(memcmp(data.dptr, "world", strlen("world")) == 0);
+	free(data.dptr);
+	ok1(tdb_transaction_start(tdb) != 0);
+	ok1(tdb_error(tdb) == TDB_ERR_NESTING);
+
+	data = tdb_fetch(tdb, key);
+	ok1(data.dsize == strlen("world"));
+	ok1(memcmp(data.dptr, "world", strlen("world")) == 0);
+	free(data.dptr);
+	ok1(tdb_transaction_commit(tdb) == 0);
+	data = tdb_fetch(tdb, key);
+	ok1(data.dsize == strlen("world"));
+	ok1(memcmp(data.dptr, "world", strlen("world")) == 0);
+	free(data.dptr);
+	tdb_close(tdb);
+
+	tdb = tdb_open("/tmp/test4.tdb", 1024, TDB_ALLOW_NESTING, O_RDWR, 0);
+	ok1(tdb);
+
+	ok1(tdb_transaction_start(tdb) == 0);
+	ok1(tdb_transaction_start(tdb) == 0);
+	ok1(tdb_delete(tdb, key) == 0);
+	ok1(tdb_transaction_commit(tdb) == 0);
+	ok1(!tdb_exists(tdb, key));
+	ok1(tdb_transaction_cancel(tdb) == 0);
+	/* Surprise! Kills inner "committed" transaction. */
+	ok1(tdb_exists(tdb, key));
+
+	ok1(tdb_transaction_start(tdb) == 0);
+	ok1(tdb_transaction_start(tdb) == 0);
+	ok1(tdb_delete(tdb, key) == 0);
+	ok1(tdb_transaction_commit(tdb) == 0);
+	ok1(!tdb_exists(tdb, key));
+	ok1(tdb_transaction_commit(tdb) == 0);
+	ok1(!tdb_exists(tdb, key));
+	tdb_close(tdb);
+
+	return exit_status();
+}

ccan/tdb/transaction.c

@@ -85,11 +85,13 @@
     still available, but no transaction recovery area is used and no
     fsync/msync calls are made.
 
-  - if TDB_NO_NESTING is passed to flags in tdb open then transaction
-    nesting is disabled. tdb_transaction_start() will then implicitely
-    cancel any pending transactions and always start a new transaction
-    context instead of nesting.
+  - if TDB_ALLOW_NESTING is passed to flags in tdb open, or added using
+    tdb_add_flags() transaction is enabled.
+    The default is that transaction nesting is not allowed and an attempt
+    to create a nested transaction will fail with TDB_ERR_NESTING.
 
+    Beware. when transactions are nested a transaction successfully
+    completed with tdb_transaction_commit() can be silently unrolled later.
 */
 
 
@@ -533,16 +535,15 @@ int tdb_transaction_start(struct tdb_context *tdb)
 
 	/* cope with nested tdb_transaction_start() calls */
 	if (tdb->transaction != NULL) {
-		tdb_trace(tdb, "tdb_transaction_start");
-		if (!tdb->flags & TDB_NO_NESTING) {
-			tdb->transaction->nesting++;
-			TDB_LOG((tdb, TDB_DEBUG_TRACE, "tdb_transaction_start: nesting %d\n", 
-				 tdb->transaction->nesting));
-			return 0;
-		} else {
-			tdb_transaction_cancel_internal(tdb);
-			TDB_LOG((tdb, TDB_DEBUG_TRACE, "tdb_transaction_start: cancelling previous transaction\n"));
+		if (!(tdb->flags & TDB_ALLOW_NESTING)) {
+			tdb->ecode = TDB_ERR_NESTING;
+			return -1;
 		}
+		tdb_trace(tdb, "tdb_transaction_start");
+		tdb->transaction->nesting++;
+		TDB_LOG((tdb, TDB_DEBUG_TRACE, "tdb_transaction_start: nesting %d\n", 
+			 tdb->transaction->nesting));
+		return 0;
 	}
 
 	if (tdb->num_locks != 0 || tdb->global_lock.count) {