tdb2: transaction support

This adds transactions to tdb2; the code is taken from tdb1 with minimal
modifications, as are the unit 

ccan/tdb2/check.c

@@ -30,7 +30,7 @@ static bool append(tdb_off_t **arr, size_t *num, tdb_off_t off)
 	return true;
 }
 
-static bool check_header(struct tdb_context *tdb)
+static bool check_header(struct tdb_context *tdb, tdb_off_t *recovery)
 {
 	uint64_t hash_test;
 	struct tdb_header hdr;
@@ -57,6 +57,16 @@ static bool check_header(struct tdb_context *tdb)
 		return false;
 	}
 
+	*recovery = hdr.recovery;
+	if (*recovery) {
+		if (*recovery < sizeof(hdr) || *recovery > tdb->map_size) {
+			tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
+				 "tdb_check: invalid recovery offset %zu\n",
+				 (size_t)*recovery);
+			return false;
+		}
+	}
+
 	/* Don't check reserved: they *can* be used later. */
 	return true;
 }
@@ -370,23 +380,73 @@ static bool check_free_list(struct tdb_context *tdb,
 	return true;
 }
 
+/* Slow, but should be very rare. */
+static size_t dead_space(struct tdb_context *tdb, tdb_off_t off)
+{
+	size_t len;
+
+	for (len = 0; off + len < tdb->map_size; len++) {
+		char c;
+		if (tdb->methods->read(tdb, off, &c, 1))
+			return 0;
+		if (c != 0 && c != 0x43)
+			break;
+	}
+	return len;
+}
+
 static bool check_linear(struct tdb_context *tdb,
 			 tdb_off_t **used, size_t *num_used,
-			 tdb_off_t **free, size_t *num_free)
+			 tdb_off_t **free, size_t *num_free,
+			 tdb_off_t recovery)
 {
 	tdb_off_t off;
 	tdb_len_t len;
+	bool found_recovery = false;
 
 	for (off = sizeof(struct tdb_header); off < tdb->map_size; off += len) {
 		union {
 			struct tdb_used_record u;
 			struct tdb_free_record f;
+			struct tdb_recovery_record r;
 		} pad, *p;
 		p = tdb_get(tdb, off, &pad, sizeof(pad));
 		if (!p)
 			return false;
-		if (frec_magic(&p->f) == TDB_FREE_MAGIC
-		    || frec_magic(&p->f) == TDB_COALESCING_MAGIC) {
+
+		/* If we crash after ftruncate, we can get zeroes or fill. */
+		if (p->r.magic == TDB_RECOVERY_INVALID_MAGIC
+		    || p->r.magic ==  0x4343434343434343ULL) {
+			if (recovery == off) {
+				found_recovery = true;
+				len = sizeof(p->r) + p->r.max_len;
+			} else {
+				len = dead_space(tdb, off);
+				if (len < sizeof(p->r)) {
+					tdb->log(tdb, TDB_DEBUG_ERROR,
+						 tdb->log_priv,
+						 "tdb_check: invalid dead space"
+						 " at %zu\n", (size_t)off);
+					return false;
+				}
+
+				tdb->log(tdb, TDB_DEBUG_WARNING, tdb->log_priv,
+					 "Dead space at %zu-%zu (of %zu)\n",
+					 (size_t)off, (size_t)(off + len),
+					 (size_t)tdb->map_size);
+			}
+		} else if (p->r.magic == TDB_RECOVERY_MAGIC) {
+			if (recovery != off) {
+				tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
+					 "tdb_check: unexpected recovery"
+					 " record at offset %zu\n",
+					 (size_t)off);
+				return false;
+			}
+			found_recovery = true;
+			len = sizeof(p->r) + p->r.max_len;
+		} else if (frec_magic(&p->f) == TDB_FREE_MAGIC
+			   || frec_magic(&p->f) == TDB_COALESCING_MAGIC) {
 			len = sizeof(p->u) + p->f.data_len;
 			if (off + len > tdb->map_size) {
 				tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
@@ -437,6 +497,15 @@ static bool check_linear(struct tdb_context *tdb,
 			}
 		}
 	}
+
+	/* We must have found recovery area if there was one. */
+	if (recovery != 0 && !found_recovery) {
+		tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
+			 "tdb_check: expected a recovery area at %zu\n",
+			 (size_t)recovery);
+		return false;
+	}
+
 	return true;
 }
 
@@ -445,7 +514,7 @@ int tdb_check(struct tdb_context *tdb,
 	      int (*check)(TDB_DATA key, TDB_DATA data, void *private_data),
 	      void *private_data)
 {
-	tdb_off_t *free = NULL, *used = NULL, flist;
+	tdb_off_t *free = NULL, *used = NULL, flist, recovery;
 	size_t num_free = 0, num_used = 0, num_found = 0, num_flists = 0;
 
 	if (tdb_allrecord_lock(tdb, F_RDLCK, TDB_LOCK_WAIT, false) != 0)
@@ -456,11 +525,11 @@ int tdb_check(struct tdb_context *tdb,
 		return -1;
 	}
 
-	if (!check_header(tdb))
+	if (!check_header(tdb, &recovery))
 		goto fail;
 
 	/* First we do a linear scan, checking all records. */
-	if (!check_linear(tdb, &used, &num_used, &free, &num_free))
+	if (!check_linear(tdb, &used, &num_used, &free, &num_free, recovery))
 		goto fail;
 
 	for (flist = first_flist(tdb); flist; flist = next_flist(tdb, flist)) {

ccan/tdb2/lock.c

@@ -190,7 +190,6 @@ static int tdb_brunlock(struct tdb_context *tdb,
 	return ret;
 }
 
-#if 0
 /*
   upgrade a read lock to a write lock. This needs to be handled in a
   special way as some OSes (such as solaris) have too conservative
@@ -217,8 +216,7 @@ int tdb_allrecord_upgrade(struct tdb_context *tdb)
 	while (count--) {
 		struct timeval tv;
 		if (tdb_brlock(tdb, F_WRLCK,
-			       TDB_HASH_LOCK_START
-			       + (1ULL << tdb->header.v.hash_bits), 0,
+			       TDB_HASH_LOCK_START, 0,
 			       TDB_LOCK_WAIT|TDB_LOCK_PROBE) == 0) {
 			tdb->allrecord_lock.ltype = F_WRLCK;
 			tdb->allrecord_lock.off = 0;
@@ -236,7 +234,6 @@ int tdb_allrecord_upgrade(struct tdb_context *tdb)
 		 "tdb_allrecord_upgrade failed\n");
 	return -1;
 }
-#endif
 
 static struct tdb_lock_type *find_nestlock(struct tdb_context *tdb,
 					   tdb_off_t offset)
@@ -251,6 +248,27 @@ static struct tdb_lock_type *find_nestlock(struct tdb_context *tdb,
 	return NULL;
 }
 
+int tdb_lock_and_recover(struct tdb_context *tdb)
+{
+	int ret;
+
+	if (tdb_allrecord_lock(tdb, F_WRLCK, TDB_LOCK_WAIT|TDB_LOCK_NOCHECK,
+			       false) == -1) {
+		return -1;
+	}
+
+	if (tdb_lock_open(tdb, TDB_LOCK_WAIT|TDB_LOCK_NOCHECK) == -1) {
+		tdb_allrecord_unlock(tdb, F_WRLCK);
+		return -1;
+	}
+	ret = tdb_transaction_recover(tdb);
+
+	tdb_unlock_open(tdb);
+	tdb_allrecord_unlock(tdb, F_WRLCK);
+
+	return ret;
+}
+
 /* lock an offset in the database. */
 static int tdb_nest_lock(struct tdb_context *tdb, tdb_off_t offset, int ltype,
 			 enum tdb_lock_flags flags)
@@ -310,6 +328,21 @@ static int tdb_nest_lock(struct tdb_context *tdb, tdb_off_t offset, int ltype,
 		return -1;
 	}
 
+	/* First time we grab a lock, perhaps someone died in commit? */
+	if (!(flags & TDB_LOCK_NOCHECK)
+	    && tdb->num_lockrecs == 0
+	    && unlikely(tdb_needs_recovery(tdb))) {
+		tdb_brunlock(tdb, ltype, offset, 1);
+
+		if (tdb_lock_and_recover(tdb) == -1) {
+			return -1;
+		}
+
+		if (tdb_brlock(tdb, ltype, offset, 1, flags)) {
+			return -1;
+		}
+	}
+
 	tdb->lockrecs[tdb->num_lockrecs].off = offset;
 	tdb->lockrecs[tdb->num_lockrecs].count = 1;
 	tdb->lockrecs[tdb->num_lockrecs].ltype = ltype;
@@ -318,40 +351,6 @@ static int tdb_nest_lock(struct tdb_context *tdb, tdb_off_t offset, int ltype,
 	return 0;
 }
 
-static int tdb_lock_and_recover(struct tdb_context *tdb)
-{
-#if 0 /* FIXME */
-
-	int ret;
-
-	/* We need to match locking order in transaction commit. */
-	if (tdb_brlock(tdb, F_WRLCK, FREELIST_TOP, 0, TDB_LOCK_WAIT)) {
-		return -1;
-	}
-
-	if (tdb_brlock(tdb, F_WRLCK, OPEN_LOCK, 1, TDB_LOCK_WAIT)) {
-		tdb_brunlock(tdb, F_WRLCK, FREELIST_TOP, 0);
-		return -1;
-	}
-
-	ret = tdb_transaction_recover(tdb);
-
-	tdb_brunlock(tdb, F_WRLCK, OPEN_LOCK, 1);
-	tdb_brunlock(tdb, F_WRLCK, FREELIST_TOP, 0);
-
-	return ret;
-#else
-	abort();
-	return -1;
-#endif
-}
-
-static bool tdb_needs_recovery(struct tdb_context *tdb)
-{
-	/* FIXME */
-	return false;
-}
-
 static int tdb_nest_unlock(struct tdb_context *tdb, tdb_off_t off, int ltype)
 {
 	int ret = -1;
@@ -390,14 +389,12 @@ static int tdb_nest_unlock(struct tdb_context *tdb, tdb_off_t off, int ltype)
 	return ret;
 }
 
-#if 0
 /*
   get the transaction lock
  */
-int tdb_transaction_lock(struct tdb_context *tdb, int ltype,
-			 enum tdb_lock_flags lockflags)
+int tdb_transaction_lock(struct tdb_context *tdb, int ltype)
 {
-	return tdb_nest_lock(tdb, TRANSACTION_LOCK, ltype, lockflags);
+	return tdb_nest_lock(tdb, TDB_TRANSACTION_LOCK, ltype, TDB_LOCK_WAIT);
 }
 
 /*
@@ -405,9 +402,8 @@ int tdb_transaction_lock(struct tdb_context *tdb, int ltype,
  */
 int tdb_transaction_unlock(struct tdb_context *tdb, int ltype)
 {
-	return tdb_nest_unlock(tdb, TRANSACTION_LOCK, ltype, false);
+	return tdb_nest_unlock(tdb, TDB_TRANSACTION_LOCK, ltype);
 }
-#endif
 
 /* We only need to lock individual bytes, but Linux merges consecutive locks
  * so we lock in contiguous ranges. */
@@ -474,7 +470,7 @@ int tdb_allrecord_lock(struct tdb_context *tdb, int ltype,
 		return -1;
 	}
 
-	if (tdb_has_locks(tdb)) {
+	if (tdb_has_hash_locks(tdb)) {
 		/* can't combine global and chain locks */
 		tdb->ecode = TDB_ERR_LOCK;
 		tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
@@ -522,7 +518,7 @@ again:
 	tdb->allrecord_lock.off = upgradable;
 
 	/* Now check for needing recovery. */
-	if (unlikely(tdb_needs_recovery(tdb))) {
+	if (!(flags & TDB_LOCK_NOCHECK) && unlikely(tdb_needs_recovery(tdb))) {
 		tdb_allrecord_unlock(tdb, ltype);
 		if (tdb_lock_and_recover(tdb) == -1) {
 			return -1;
@@ -533,9 +529,9 @@ again:
 	return 0;
 }
 
-int tdb_lock_open(struct tdb_context *tdb)
+int tdb_lock_open(struct tdb_context *tdb, enum tdb_lock_flags flags)
 {
-	return tdb_nest_lock(tdb, TDB_OPEN_LOCK, F_WRLCK, TDB_LOCK_WAIT);
+	return tdb_nest_lock(tdb, TDB_OPEN_LOCK, F_WRLCK, flags);
 }
 
 void tdb_unlock_open(struct tdb_context *tdb)
@@ -543,9 +539,16 @@ void tdb_unlock_open(struct tdb_context *tdb)
 	tdb_nest_unlock(tdb, TDB_OPEN_LOCK, F_WRLCK);
 }
 
+bool tdb_has_open_lock(struct tdb_context *tdb)
+{
+	return find_nestlock(tdb, TDB_OPEN_LOCK) != NULL;
+}
+
 int tdb_lock_expand(struct tdb_context *tdb, int ltype)
 {
-	return tdb_nest_lock(tdb, TDB_EXPANSION_LOCK, ltype, TDB_LOCK_WAIT);
+	/* Lock doesn't protect data, so don't check (we recurse if we do!) */
+	return tdb_nest_lock(tdb, TDB_EXPANSION_LOCK, ltype,
+			     TDB_LOCK_WAIT | TDB_LOCK_NOCHECK);
 }
 
 void tdb_unlock_expand(struct tdb_context *tdb, int ltype)
@@ -598,9 +601,17 @@ bool tdb_has_expansion_lock(struct tdb_context *tdb)
 	return find_nestlock(tdb, TDB_EXPANSION_LOCK) != NULL;
 }
 
-bool tdb_has_locks(struct tdb_context *tdb)
+bool tdb_has_hash_locks(struct tdb_context *tdb)
 {
-	return tdb->allrecord_lock.count || tdb->num_lockrecs;
+	unsigned int i;
+
+	for (i=0; i<tdb->num_lockrecs; i++) {
+		if (tdb->lockrecs[i].off >= TDB_HASH_LOCK_START
+		    && tdb->lockrecs[i].off < (TDB_HASH_LOCK_START
+					       + TDB_HASH_LOCK_RANGE))
+			return true;
+	}
+	return false;
 }
 
 #if 0

ccan/tdb2/private.h

@@ -67,8 +67,8 @@ typedef uint64_t tdb_off_t;
 #define TDB_FREE_MAGIC ((uint64_t)0xFE)
 #define TDB_COALESCING_MAGIC ((uint64_t)0xFD)
 #define TDB_HASH_MAGIC (0xA1ABE11A01092008ULL)
-#define TDB_RECOVERY_MAGIC (0xf53bc0e7U)
-#define TDB_RECOVERY_INVALID_MAGIC (0x0)
+#define TDB_RECOVERY_MAGIC (0xf53bc0e7ad124589ULL)
+#define TDB_RECOVERY_INVALID_MAGIC (0x0ULL)
 
 #define TDB_OFF_ERR ((tdb_off_t)-1)
 
@@ -189,6 +189,16 @@ static inline uint64_t frec_flist(const struct tdb_free_record *f)
 	return f->magic_and_meta & ((1ULL << (64 - TDB_OFF_UPPER_STEAL)) - 1);
 }
 
+struct tdb_recovery_record {
+	uint64_t magic;
+	/* Length of record. */
+	uint64_t max_len;
+	/* Length used. */
+	uint64_t len;
+	/* Old length of file before transaction. */
+	uint64_t eof;
+};
+
 /* this is stored at the front of every database */
 struct tdb_header {
 	char magic_food[64]; /* for /etc/magic */
@@ -197,8 +207,9 @@ struct tdb_header {
 	uint64_t hash_test; /* result of hashing HASH_MAGIC. */
 	uint64_t hash_seed; /* "random" seed written at creation time. */
 	tdb_off_t free_list; /* (First) free list. */
+	tdb_off_t recovery; /* Transaction recovery area. */
 
-	tdb_off_t reserved[27];
+	tdb_off_t reserved[26];
 
 	/* Top level hash table. */
 	tdb_off_t hashtable[1ULL << TDB_TOPLEVEL_HASH_BITS];
@@ -248,6 +259,8 @@ enum tdb_lock_flags {
 	TDB_LOCK_WAIT = 1,
 	/* If set, don't log an error on failure. */
 	TDB_LOCK_PROBE = 2,
+	/* If set, don't check for recovery (used by recovery code). */
+	TDB_LOCK_NOCHECK = 4,
 };
 
 struct tdb_lock_type {
@@ -456,23 +469,31 @@ int tdb_lock_free_bucket(struct tdb_context *tdb, tdb_off_t b_off,
 			 enum tdb_lock_flags waitflag);
 void tdb_unlock_free_bucket(struct tdb_context *tdb, tdb_off_t b_off);
 
-/* Do we have any locks? */
-bool tdb_has_locks(struct tdb_context *tdb);
+/* Serialize transaction start. */
+int tdb_transaction_lock(struct tdb_context *tdb, int ltype);
+int tdb_transaction_unlock(struct tdb_context *tdb, int ltype);
+
+/* Do we have any hash locks (ie. via tdb_chainlock) ? */
+bool tdb_has_hash_locks(struct tdb_context *tdb);
 
 /* Lock entire database. */
 int tdb_allrecord_lock(struct tdb_context *tdb, int ltype,
 		       enum tdb_lock_flags flags, bool upgradable);
 int tdb_allrecord_unlock(struct tdb_context *tdb, int ltype);
+int tdb_allrecord_upgrade(struct tdb_context *tdb);
 
 /* Serialize db open. */
-int tdb_lock_open(struct tdb_context *tdb);
+int tdb_lock_open(struct tdb_context *tdb, enum tdb_lock_flags flags);
 void tdb_unlock_open(struct tdb_context *tdb);
+bool tdb_has_open_lock(struct tdb_context *tdb);
 
 /* Serialize db expand. */
 int tdb_lock_expand(struct tdb_context *tdb, int ltype);
 void tdb_unlock_expand(struct tdb_context *tdb, int ltype);
 bool tdb_has_expansion_lock(struct tdb_context *tdb);
 
+/* If it needs recovery, grab all the locks and do it. */
+int tdb_lock_and_recover(struct tdb_context *tdb);
 
 /* traverse.c: */
 int first_in_hash(struct tdb_context *tdb, int ltype,
@@ -482,6 +503,9 @@ int next_in_hash(struct tdb_context *tdb, int ltype,
 		 struct traverse_info *tinfo,
 		 TDB_DATA *kbuf, size_t *dlen);
 
+/* transaction.c: */
+int tdb_transaction_recover(struct tdb_context *tdb);
+bool tdb_needs_recovery(struct tdb_context *tdb);
 
 #if 0
 /* Low-level locking primitives. */

ccan/tdb2/tdb.c

@@ -103,6 +103,7 @@ static int tdb_new_database(struct tdb_context *tdb,
 					 sizeof(newdb.hdr.hash_test),
 					 newdb.hdr.hash_seed,
 					 tdb->hash_priv);
+	newdb.hdr.recovery = 0;
 	memset(newdb.hdr.reserved, 0, sizeof(newdb.hdr.reserved));
 	/* Initial hashes are empty. */
 	memset(newdb.hdr.hashtable, 0, sizeof(newdb.hdr.hashtable));
@@ -246,7 +247,7 @@ struct tdb_context *tdb_open(const char *name, int tdb_flags,
         fcntl(tdb->fd, F_SETFD, v | FD_CLOEXEC);
 
 	/* ensure there is only one process initialising at once */
-	if (tdb_lock_open(tdb) == -1) {
+	if (tdb_lock_open(tdb, TDB_LOCK_WAIT|TDB_LOCK_NOCHECK) == -1) {
 		tdb->log(tdb, TDB_DEBUG_ERROR, tdb->log_priv,
 			 "tdb_open: failed to get open lock on %s: %s\n",
 			 name, strerror(errno));
@@ -314,6 +315,12 @@ struct tdb_context *tdb_open(const char *name, int tdb_flags,
 	/* This make sure we have current map_size and mmap. */
 	tdb->methods->oob(tdb, tdb->map_size + 1, true);
 
+	/* Now it's fully formed, recover if necessary. */
+	if (tdb_needs_recovery(tdb) && tdb_lock_and_recover(tdb) == -1) {
+		errno = EIO;
+		goto fail;
+	}
+
 	if (tdb_flist_init(tdb) == -1)
 		goto fail;
 

ccan/tdb2/tdb2.h

@@ -142,6 +142,11 @@ int tdb_check(struct tdb_context *tdb,
 enum TDB_ERROR tdb_error(struct tdb_context *tdb);
 const char *tdb_errorstr(struct tdb_context *tdb);
 
+int tdb_transaction_start(struct tdb_context *tdb);
+void tdb_transaction_cancel(struct tdb_context *tdb);
+int tdb_transaction_prepare_commit(struct tdb_context *tdb);
+int tdb_transaction_commit(struct tdb_context *tdb);
+
 char *tdb_summary(struct tdb_context *tdb, enum tdb_summary_flags flags);
 
 extern struct tdb_data tdb_null;

ccan/tdb2/test/external-agent.c

 #include "external-agent.h"
 #include "logging.h"
+#include "lock-tracking.h"
 #include <sys/types.h>
 #include <sys/wait.h>
 #include <unistd.h>
@@ -16,11 +17,6 @@
 
 static struct tdb_context *tdb;
 
-#if 1 /* FIXME */
-static unsigned int locking_would_block = 0;
-static bool nonblocking_locks = false;
-#endif
-
 static enum agent_return do_operation(enum operation op, const char *name)
 {
 	TDB_DATA k;
@@ -32,6 +28,8 @@ static enum agent_return do_operation(enum operation op, const char *name)
 		return OTHER_FAILURE;
 	}
 
+	diag("external: %s", operation_name(op));
+
 	k.dptr = (void *)name;
 	k.dsize = strlen(name);
 
@@ -46,6 +44,7 @@ static enum agent_return do_operation(enum operation op, const char *name)
 		if (!tdb) {
 			if (!locking_would_block)
 				diag("Opening tdb gave %s", strerror(errno));
+			forget_locking();
 			ret = OTHER_FAILURE;
 		} else
 			ret = SUCCESS;
@@ -68,7 +67,6 @@ static enum agent_return do_operation(enum operation op, const char *name)
 	case STORE:
 		ret = tdb_store(tdb, k, k, 0) == 0 ? SUCCESS : OTHER_FAILURE;
 		break;
-#if 0 /* FIXME */
 	case TRANSACTION_START:
 		ret = tdb_transaction_start(tdb) == 0 ? SUCCESS : OTHER_FAILURE;
 		break;
@@ -78,7 +76,6 @@ static enum agent_return do_operation(enum operation op, const char *name)
 	case NEEDS_RECOVERY:
 		ret = tdb_needs_recovery(tdb) ? SUCCESS : FAILED;
 		break;
-#endif
 	case CHECK:
 		ret = tdb_check(tdb, NULL, NULL) == 0 ? SUCCESS : OTHER_FAILURE;
 		break;
@@ -183,11 +180,9 @@ const char *operation_name(enum operation op)
 	case FETCH: return "FETCH";
 	case STORE: return "STORE";
 	case CHECK: return "CHECK";
-#if 0
 	case TRANSACTION_START: return "TRANSACTION_START";
 	case TRANSACTION_COMMIT: return "TRANSACTION_COMMIT";
 	case NEEDS_RECOVERY: return "NEEDS_RECOVERY";
-#endif
 	case CLOSE: return "CLOSE";
 	}
 	return "**INVALID**";

ccan/tdb2/test/external-agent.h

@@ -7,11 +7,9 @@ enum operation {
 	OPEN,
 	FETCH,
 	STORE,
-#if 0
 	TRANSACTION_START,
 	TRANSACTION_COMMIT,
 	NEEDS_RECOVERY,
-#endif
 	CHECK,
 	CLOSE,
 };

ccan/tdb2/test/lock-tracking.c

+/* We save the locks so we can reaquire them. */
+#include <unistd.h>
+#include <fcntl.h>
+#include <stdarg.h>
+#include <stdlib.h>
+#include <ccan/tap/tap.h>
+#include <ccan/tdb2/private.h>
+#include "lock-tracking.h"
+
+struct lock {
+	struct lock *next;
+	unsigned int off;
+	unsigned int len;
+	int type;
+};
+static struct lock *locks;
+int locking_errors = 0;
+bool suppress_lockcheck = false;
+bool nonblocking_locks;
+int locking_would_block = 0;
+void (*unlock_callback)(int fd);
+
+int fcntl_with_lockcheck(int fd, int cmd, ... /* arg */ )
+{
+	va_list ap;
+	int ret, arg3;
+	struct flock *fl;
+	bool may_block = false;
+
+	if (cmd != F_SETLK && cmd != F_SETLKW) {
+		/* This may be totally bogus, but we don't know in general. */
+		va_start(ap, cmd);
+		arg3 = va_arg(ap, int);
+		va_end(ap);
+
+		return fcntl(fd, cmd, arg3);
+	}
+
+	va_start(ap, cmd);
+	fl = va_arg(ap, struct flock *);
+	va_end(ap);
+
+	if (cmd == F_SETLKW && nonblocking_locks) {
+		cmd = F_SETLK;
+		may_block = true;
+	}
+	ret = fcntl(fd, cmd, fl);
+
+	/* Detect when we failed, but might have been OK if we waited. */
+	if (may_block && ret == -1 && (errno == EAGAIN || errno == EACCES)) {
+		locking_would_block++;
+	}
+
+	if (fl->l_type == F_UNLCK) {
+		struct lock **l;
+		struct lock *old = NULL;
+
+		for (l = &locks; *l; l = &(*l)->next) {
+			if ((*l)->off == fl->l_start
+			    && (*l)->len == fl->l_len) {
+				if (ret == 0) {
+					old = *l;
+					*l = (*l)->next;
+					free(old);
+				}
+				break;
+			}
+		}
+		if (!old && !suppress_lockcheck) {
+			diag("Unknown unlock %u@%u - %i",
+			     (int)fl->l_len, (int)fl->l_start, ret);
+			locking_errors++;
+		}
+	} else {
+		struct lock *new, *i;
+		unsigned int fl_end = fl->l_start + fl->l_len;
+		if (fl->l_len == 0)
+			fl_end = (unsigned int)-1;
+
+		/* Check for overlaps: we shouldn't do this. */
+		for (i = locks; i; i = i->next) {
+			unsigned int i_end = i->off + i->len;
+			if (i->len == 0)
+				i_end = (unsigned int)-1;
+
+			if (fl->l_start >= i->off && fl->l_start < i_end)
+				break;
+			if (fl_end > i->off && fl_end < i_end)
+				break;
+
+			/* tdb_allrecord_lock does this, handle adjacent: */
+			if (fl->l_start > TDB_HASH_LOCK_START
+			    && fl->l_start == i_end && fl->l_type == i->type) {
+				if (ret == 0) {
+					i->len = fl->l_len
+						? i->len + fl->l_len
+						: 0;
+				}
+				goto done;
+			}
+		}
+		if (i) {
+			/* Special case: upgrade of allrecord lock. */
+			if (i->type == F_RDLCK && fl->l_type == F_WRLCK
+			    && i->off == TDB_HASH_LOCK_START
+			    && fl->l_start == TDB_HASH_LOCK_START
+			    && i->len == 0
+			    && fl->l_len == 0) {
+				if (ret == 0)
+					i->type = F_WRLCK;
+				goto done;
+			}
+			if (!suppress_lockcheck) {
+				diag("%s lock %u@%u overlaps %u@%u",
+				     fl->l_type == F_WRLCK ? "write" : "read",
+				     (int)fl->l_len, (int)fl->l_start,
+				     i->len, (int)i->off);
+				locking_errors++;
+			}
+		}
+
+		if (ret == 0) {
+			new = malloc(sizeof *new);
+			new->off = fl->l_start;
+			new->len = fl->l_len;
+			new->type = fl->l_type;
+			new->next = locks;
+			locks = new;
+		}
+	}
+done:
+	if (ret == 0 && fl->l_type == F_UNLCK && unlock_callback)
+		unlock_callback(fd);
+	return ret;
+}
+
+unsigned int forget_locking(void)
+{
+	unsigned int num = 0;
+	while (locks) {
+		struct lock *next = locks->next;
+		free(locks);
+		locks = next;
+		num++;
+	}
+	return num;
+}

ccan/tdb2/test/lock-tracking.h

+#ifndef LOCK_TRACKING_H
+#define LOCK_TRACKING_H
+#include <stdbool.h>
+
+/* Set this if you want a callback after fnctl unlock. */
+extern void (*unlock_callback)(int fd);
+
+/* Replacement fcntl. */
+int fcntl_with_lockcheck(int fd, int cmd, ... /* arg */ );
+
+/* Discard locking info: returns number of locks outstanding. */
+unsigned int forget_locking(void);
+
+/* Number of errors in locking. */
+extern int locking_errors;
+
+/* Suppress lock checking. */
+extern bool suppress_lockcheck;
+
+/* Make all locks non-blocking. */
+extern bool nonblocking_locks;
+
+/* Number of times we failed a lock because we made it non-blocking. */
+extern int locking_would_block;
+#endif /* LOCK_TRACKING_H */

ccan/tdb2/test/run-001-encode.c

@@ -3,6 +3,7 @@
 #include <ccan/tdb2/lock.c>
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/io.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"

ccan/tdb2/test/run-001-fls.c

@@ -3,6 +3,7 @@
 #include <ccan/tdb2/lock.c>
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tap/tap.h>
 

ccan/tdb2/test/run-01-new_database.c

@@ -3,6 +3,7 @@
 #include <ccan/tdb2/lock.c>
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"

ccan/tdb2/test/run-02-expand.c

@@ -3,6 +3,7 @@
 #include <ccan/tdb2/lock.c>
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/check.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tdb2/hash.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"

ccan/tdb2/test/run-03-coalesce.c

@@ -4,6 +4,7 @@
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"
 #include "layout.h"
@@ -88,7 +89,7 @@ int main(int argc, char *argv[])
 	/* Lock and coalesce. */
 	ok1(tdb_lock_free_bucket(tdb, b_off, TDB_LOCK_WAIT) == 0);
 	ok1(coalesce(tdb, layout->elem[1].base.off, b_off, 1024) == 1);
-	ok1(!tdb_has_locks(tdb));
+	ok1(tdb->allrecord_lock.count == 0 && tdb->num_lockrecs == 0);
 	ok1(free_record_length(tdb, layout->elem[1].base.off)
 	    == 1024 + sizeof(struct tdb_used_record) + 2048);
 	ok1(tdb_check(tdb, NULL, NULL) == 0);
@@ -110,7 +111,7 @@ int main(int argc, char *argv[])
 	/* Lock and coalesce. */
 	ok1(tdb_lock_free_bucket(tdb, b_off, TDB_LOCK_WAIT) == 0);
 	ok1(coalesce(tdb, layout->elem[1].base.off, b_off, 1024) == 1);
-	ok1(!tdb_has_locks(tdb));
+	ok1(tdb->allrecord_lock.count == 0 && tdb->num_lockrecs == 0);
 	ok1(free_record_length(tdb, layout->elem[1].base.off)
 	    == 1024 + sizeof(struct tdb_used_record) + 512);
 	ok1(tdb_check(tdb, NULL, NULL) == 0);
@@ -133,7 +134,7 @@ int main(int argc, char *argv[])
 	/* Lock and coalesce. */
 	ok1(tdb_lock_free_bucket(tdb, b_off, TDB_LOCK_WAIT) == 0);
 	ok1(coalesce(tdb, layout->elem[1].base.off, b_off, 1024) == 1);
-	ok1(!tdb_has_locks(tdb));
+	ok1(tdb->allrecord_lock.count == 0 && tdb->num_lockrecs == 0);
 	ok1(free_record_length(tdb, layout->elem[1].base.off)
 	    == 1024 + sizeof(struct tdb_used_record) + 512
 	    + sizeof(struct tdb_used_record) + 256);

ccan/tdb2/test/run-04-basichash.c

@@ -3,6 +3,7 @@
 #include <ccan/tdb2/lock.c>
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"

ccan/tdb2/test/run-10-simple-store.c

@@ -3,6 +3,7 @@
 #include <ccan/tdb2/lock.c>
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"

ccan/tdb2/test/run-11-simple-fetch.c

@@ -3,6 +3,7 @@
 #include <ccan/tdb2/lock.c>
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"

ccan/tdb2/test/run-12-store.c

@@ -4,6 +4,7 @@
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"
 

ccan/tdb2/test/run-13-delete.c

@@ -4,6 +4,7 @@
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"
 
@@ -159,7 +160,7 @@ int main(int argc, char *argv[])
 		/* Check mixed bitpattern. */
 		test_val(tdb, 0x123456789ABCDEF0ULL);
 
-		ok1(!tdb_has_locks(tdb));
+		ok1(tdb->allrecord_lock.count == 0 && tdb->num_lockrecs == 0);
 		tdb_close(tdb);
 
 		/* Deleting these entries in the db gave problems. */

ccan/tdb2/test/run-15-append.c

@@ -4,6 +4,7 @@
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 #include <ccan/ilog/ilog.h>
 #include "logging.h"
@@ -67,7 +68,7 @@ int main(int argc, char *argv[])
 				moves++;
 			oldoff = newoff;
 		}
-		ok1(!tdb_has_locks(tdb));
+		ok1(tdb->allrecord_lock.count == 0 && tdb->num_lockrecs == 0);
 		/* We should increase by 50% each time... */
 		ok(moves <= ilog64(j / SIZE_STEP)*2, "Moved %u times", moves);
 		tdb_close(tdb);
@@ -98,7 +99,7 @@ int main(int argc, char *argv[])
 				moves++;
 			oldoff = newoff;
 		}
-		ok1(!tdb_has_locks(tdb));
+		ok1(tdb->allrecord_lock.count == 0 && tdb->num_lockrecs == 0);
 		/* We should increase by 50% each time... */
 		ok(moves <= ilog64(j / SIZE_STEP)*2, "Moved %u times", moves);
 		tdb_close(tdb);
@@ -120,7 +121,7 @@ int main(int argc, char *argv[])
 		ok1(data.dsize == MAX_SIZE);
 		ok1(memcmp(data.dptr, buffer, data.dsize) == 0);
 		free(data.dptr);
-		ok1(!tdb_has_locks(tdb));
+		ok1(tdb->allrecord_lock.count == 0 && tdb->num_lockrecs == 0);
 		tdb_close(tdb);
 	}
 

ccan/tdb2/test/run-20-growhash.c

@@ -3,6 +3,7 @@
 #include <ccan/tdb2/lock.c>
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"

ccan/tdb2/test/run-30-exhaust-before-expand.c

@@ -4,6 +4,7 @@
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tdb2/hash.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 #include <err.h>
 #include "logging.h"

ccan/tdb2/test/run-50-multiple-freelists.c

@@ -5,6 +5,7 @@
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tap/tap.h>
+#include <ccan/tdb2/transaction.c>
 #include "logging.h"
 #include "layout.h"
 

ccan/tdb2/test/run-55-transaction.c

+#include <ccan/tdb2/tdb.c>
+#include <ccan/tdb2/free.c>
+#include <ccan/tdb2/lock.c>
+#include <ccan/tdb2/io.c>
+#include <ccan/tdb2/hash.c>
+#include <ccan/tdb2/check.c>
+#include <ccan/tdb2/transaction.c>
+#include <ccan/tap/tap.h>
+#include "logging.h"
+
+int main(int argc, char *argv[])
+{
+	unsigned int i;
+	struct tdb_context *tdb;
+	unsigned char *buffer;
+	int flags[] = { TDB_DEFAULT, TDB_NOMMAP,
+			TDB_CONVERT, TDB_NOMMAP|TDB_CONVERT };
+	struct tdb_data key = { (unsigned char *)"key", 3 };
+	struct tdb_data data;
+
+	buffer = malloc(1000);
+	for (i = 0; i < 1000; i++)
+		buffer[i] = i;
+
+	plan_tests(sizeof(flags) / sizeof(flags[0]) * 18 + 1);
+
+	for (i = 0; i < sizeof(flags) / sizeof(flags[0]); i++) {
+		tdb = tdb_open("run-55-transaction.tdb", flags[i],
+			       O_RDWR|O_CREAT|O_TRUNC, 0600, &tap_log_attr);
+		ok1(tdb);
+		if (!tdb)
+			continue;
+
+		ok1(tdb_transaction_start(tdb) == 0);
+		data.dptr = buffer;
+		data.dsize = 1000;
+		ok1(tdb_store(tdb, key, data, TDB_INSERT) == 0);
+		data = tdb_fetch(tdb, key);
+		ok1(data.dsize == 1000);
+		ok1(memcmp(data.dptr, buffer, data.dsize) == 0);
+		free(data.dptr);
+
+		/* Cancelling a transaction means no store */
+		tdb_transaction_cancel(tdb);
+		ok1(tdb->allrecord_lock.count == 0 && tdb->num_lockrecs == 0);
+		ok1(tdb_check(tdb, NULL, NULL) == 0);
+		data = tdb_fetch(tdb, key);
+		ok1(data.dsize == 0);
+		ok1(data.dptr == NULL);
+
+		/* Commit the transaction. */
+		ok1(tdb_transaction_start(tdb) == 0);
+		data.dptr = buffer;
+		data.dsize = 1000;
+		ok1(tdb_store(tdb, key, data, TDB_INSERT) == 0);
+		data = tdb_fetch(tdb, key);
+		ok1(data.dsize == 1000);
+		ok1(memcmp(data.dptr, buffer, data.dsize) == 0);
+		free(data.dptr);
+		ok1(tdb_transaction_commit(tdb) == 0);
+		ok1(tdb->allrecord_lock.count == 0 && tdb->num_lockrecs == 0);
+		ok1(tdb_check(tdb, NULL, NULL) == 0);
+		data = tdb_fetch(tdb, key);
+		ok1(data.dsize == 1000);
+		ok1(memcmp(data.dptr, buffer, data.dsize) == 0);
+		free(data.dptr);
+
+		tdb_close(tdb);
+	}
+
+	ok1(tap_log_messages == 0);
+	return exit_status();
+}

ccan/tdb2/test/run-56-open-during-transaction.c

+#define _XOPEN_SOURCE 500
+#include <unistd.h>
+#include "lock-tracking.h"
+
+static ssize_t pwrite_check(int fd, const void *buf, size_t count, off_t offset);
+static ssize_t write_check(int fd, const void *buf, size_t count);
+static int ftruncate_check(int fd, off_t length);
+
+#define pwrite pwrite_check
+#define write write_check
+#define fcntl fcntl_with_lockcheck
+#define ftruncate ftruncate_check
+
+#include <ccan/tdb2/tdb.c>
+#include <ccan/tdb2/free.c>
+#include <ccan/tdb2/lock.c>
+#include <ccan/tdb2/io.c>
+#include <ccan/tdb2/hash.c>
+#include <ccan/tdb2/check.c>
+#include <ccan/tdb2/transaction.c>
+#include <ccan/tap/tap.h>
+#include <stdlib.h>
+#include <stdbool.h>
+#include <stdarg.h>
+#include <err.h>
+#include "external-agent.h"
+#include "logging.h"
+
+static struct agent *agent;
+static bool opened;
+static int errors = 0;
+#define TEST_DBNAME "run-56-open-during-transaction.tdb"
+
+#undef write
+#undef pwrite
+#undef fcntl
+#undef ftruncate
+
+static bool is_same(const char *snapshot, const char *latest, off_t len)
+{
+	unsigned i;
+
+	for (i = 0; i < len; i++) {
+		if (snapshot[i] != latest[i])
+			return false;
+	}
+	return true;
+}
+
+static bool compare_file(int fd, const char *snapshot, off_t snapshot_len)
+{
+	char *contents;
+	bool same;
+
+	/* over-length read serves as length check. */
+	contents = malloc(snapshot_len+1);
+	same = pread(fd, contents, snapshot_len+1, 0) == snapshot_len
+		&& is_same(snapshot, contents, snapshot_len);
+	free(contents);
+	return same;
+}
+
+static void check_file_intact(int fd)
+{
+	enum agent_return ret;
+	struct stat st;
+	char *contents;
+
+	fstat(fd, &st);
+	contents = malloc(st.st_size);
+	if (pread(fd, contents, st.st_size, 0) != st.st_size) {
+		diag("Read fail");
+		errors++;
+		return;
+	}
+
+	/* Ask agent to open file. */
+	ret = external_agent_operation(agent, OPEN, TEST_DBNAME);
+
+	/* It's OK to open it, but it must not have changed! */
+	if (!compare_file(fd, contents, st.st_size)) {
+		diag("Agent changed file after opening %s",
+		     agent_return_name(ret));
+		errors++;
+	}
+
+	if (ret == SUCCESS) {
+		ret = external_agent_operation(agent, CLOSE, NULL);
+		if (ret != SUCCESS) {
+			diag("Agent failed to close tdb: %s",
+			     agent_return_name(ret));
+			errors++;
+		}
+	} else if (ret != WOULD_HAVE_BLOCKED) {
+		diag("Agent opening file gave %s",
+		     agent_return_name(ret));
+		errors++;
+	}
+
+	free(contents);
+}
+
+static void after_unlock(int fd)
+{
+	if (opened)
+		check_file_intact(fd);
+}
+
+static ssize_t pwrite_check(int fd,
+			    const void *buf, size_t count, off_t offset)
+{
+	if (opened)
+		check_file_intact(fd);
+
+	return pwrite(fd, buf, count, offset);
+}
+
+static ssize_t write_check(int fd, const void *buf, size_t count)
+{
+	if (opened)
+		check_file_intact(fd);
+
+	return write(fd, buf, count);
+}
+
+static int ftruncate_check(int fd, off_t length)
+{
+	if (opened)
+		check_file_intact(fd);
+
+	return ftruncate(fd, length);
+
+}
+
+int main(int argc, char *argv[])
+{
+	const int flags[] = { TDB_DEFAULT,
+			      TDB_NOMMAP,
+			      TDB_CONVERT,
+			      TDB_CONVERT | TDB_NOMMAP };
+	int i;
+	struct tdb_context *tdb;
+	TDB_DATA key, data;
+
+	plan_tests(20);
+	agent = prepare_external_agent();
+	if (!agent)
+		err(1, "preparing agent");
+
+	unlock_callback = after_unlock;
+	for (i = 0; i < sizeof(flags)/sizeof(flags[0]); i++) {
+		diag("Test with %s and %s\n",
+		     (flags[i] & TDB_CONVERT) ? "CONVERT" : "DEFAULT",
+		     (flags[i] & TDB_NOMMAP) ? "no mmap" : "mmap");
+		unlink(TEST_DBNAME);
+		tdb = tdb_open(TEST_DBNAME, flags[i],
+			       O_RDWR|O_CREAT|O_TRUNC, 0600, &tap_log_attr);
+		ok1(tdb);
+
+		opened = true;
+		ok1(tdb_transaction_start(tdb) == 0);
+		key.dsize = strlen("hi");
+		key.dptr = (void *)"hi";
+		data.dptr = (void *)"world";
+		data.dsize = strlen("world");
+
+		ok1(tdb_store(tdb, key, data, TDB_INSERT) == 0);
+		ok1(tdb_transaction_commit(tdb) == 0);
+		ok(!errors, "We had %u open errors", errors);
+
+		opened = false;
+		tdb_close(tdb);
+	}
+
+	return exit_status();
+}

ccan/tdb2/test/run-57-die-during-transaction.c

+#define _XOPEN_SOURCE 500
+#include <unistd.h>
+#include "lock-tracking.h"
+static ssize_t pwrite_check(int fd, const void *buf, size_t count, off_t offset);
+static ssize_t write_check(int fd, const void *buf, size_t count);
+static int ftruncate_check(int fd, off_t length);
+
+#define pwrite pwrite_check
+#define write write_check
+#define fcntl fcntl_with_lockcheck
+#define ftruncate ftruncate_check
+
+#include <ccan/tdb2/tdb.c>
+#include <ccan/tdb2/free.c>
+#include <ccan/tdb2/lock.c>
+#include <ccan/tdb2/io.c>
+#include <ccan/tdb2/hash.c>
+#include <ccan/tdb2/check.c>
+#include <ccan/tdb2/transaction.c>
+#include <ccan/tap/tap.h>
+#include <stdlib.h>
+#include <stdbool.h>
+#include <stdarg.h>
+#include <err.h>
+#include <setjmp.h>
+#include "external-agent.h"
+#include "logging.h"
+
+#undef write
+#undef pwrite
+#undef fcntl
+#undef ftruncate
+
+static bool in_transaction;
+static int target, current;
+static jmp_buf jmpbuf;
+#define TEST_DBNAME "run-57-die-during-transaction.tdb"
+#define KEY_STRING "helloworld"
+
+static void maybe_die(int fd)
+{
+	if (in_transaction && current++ == target) {
+		longjmp(jmpbuf, 1);
+	}
+}
+
+static ssize_t pwrite_check(int fd,
+			    const void *buf, size_t count, off_t offset)
+{
+	ssize_t ret;
+
+	maybe_die(fd);
+
+	ret = pwrite(fd, buf, count, offset);
+	if (ret != count)
+		return ret;
+
+	maybe_die(fd);
+	return ret;
+}
+
+static ssize_t write_check(int fd, const void *buf, size_t count)
+{
+	ssize_t ret;
+
+	maybe_die(fd);
+
+	ret = write(fd, buf, count);
+	if (ret != count)
+		return ret;
+
+	maybe_die(fd);
+	return ret;
+}
+
+static int ftruncate_check(int fd, off_t length)
+{
+	int ret;
+
+	maybe_die(fd);
+
+	ret = ftruncate(fd, length);
+
+	maybe_die(fd);
+	return ret;
+}
+
+static bool test_death(enum operation op, struct agent *agent)
+{
+	struct tdb_context *tdb = NULL;
+	TDB_DATA key;
+	enum agent_return ret;
+	int needed_recovery = 0;
+
+	current = target = 0;
+reset:
+	unlink(TEST_DBNAME);
+	tdb = tdb_open(TEST_DBNAME, TDB_NOMMAP,
+		       O_CREAT|O_TRUNC|O_RDWR, 0600, &tap_log_attr);
+
+	if (setjmp(jmpbuf) != 0) {
+		/* We're partway through.  Simulate our death. */
+		close(tdb->fd);
+		forget_locking();
+		in_transaction = false;
+
+		ret = external_agent_operation(agent, NEEDS_RECOVERY, "");
+		if (ret == SUCCESS)
+			needed_recovery++;
+		else if (ret != FAILED) {
+			diag("Step %u agent NEEDS_RECOVERY = %s", current,
+			     agent_return_name(ret));
+			return false;
+		}
+
+		ret = external_agent_operation(agent, op, KEY_STRING);
+		if (ret != SUCCESS) {
+			diag("Step %u op %s failed = %s", current,
+			     operation_name(op),
+			     agent_return_name(ret));
+			return false;
+		}
+
+		ret = external_agent_operation(agent, NEEDS_RECOVERY, "");
+		if (ret != FAILED) {
+			diag("Still needs recovery after step %u = %s",
+			     current, agent_return_name(ret));
+			return false;
+		}
+
+		ret = external_agent_operation(agent, CHECK, "");
+		if (ret != SUCCESS) {
+			diag("Step %u check failed = %s", current,
+			     agent_return_name(ret));
+			return false;
+		}
+
+		ret = external_agent_operation(agent, CLOSE, "");
+		if (ret != SUCCESS) {
+			diag("Step %u close failed = %s", current,
+			     agent_return_name(ret));
+			return false;
+		}
+
+		/* Suppress logging as this tries to use closed fd. */
+		suppress_logging = true;
+		suppress_lockcheck = true;
+		tdb_close(tdb);
+		suppress_logging = false;
+		suppress_lockcheck = false;
+		target++;
+		current = 0;
+		goto reset;
+	}
+
+	/* Put key for agent to fetch. */
+	key.dsize = strlen(KEY_STRING);
+	key.dptr = (void *)KEY_STRING;
+	if (tdb_store(tdb, key, key, TDB_INSERT) != 0)
+		return false;
+
+	/* This is the key we insert in transaction. */
+	key.dsize--;
+
+	ret = external_agent_operation(agent, OPEN, TEST_DBNAME);
+	if (ret != SUCCESS)
+		errx(1, "Agent failed to open: %s", agent_return_name(ret));
+
+	ret = external_agent_operation(agent, FETCH, KEY_STRING);
+	if (ret != SUCCESS)
+		errx(1, "Agent failed find key: %s", agent_return_name(ret));
+
+	in_transaction = true;
+	if (tdb_transaction_start(tdb) != 0)
+		return false;
+
+	if (tdb_store(tdb, key, key, TDB_INSERT) != 0)
+		return false;
+
+	if (tdb_transaction_commit(tdb) != 0)
+		return false;
+
+	in_transaction = false;
+
+	/* We made it! */
+	diag("Completed %u runs", current);
+	tdb_close(tdb);
+	ret = external_agent_operation(agent, CLOSE, "");
+	if (ret != SUCCESS) {
+		diag("Step %u close failed = %s", current,
+		     agent_return_name(ret));
+		return false;
+	}
+
+	ok1(needed_recovery);
+	ok1(locking_errors == 0);
+	ok1(forget_locking() == 0);
+	locking_errors = 0;
+	return true;
+}
+
+int main(int argc, char *argv[])
+{
+	enum operation ops[] = { FETCH, STORE, TRANSACTION_START };
+	struct agent *agent;
+	int i;
+
+	plan_tests(12);
+	unlock_callback = maybe_die;
+
+	agent = prepare_external_agent();
+	if (!agent)
+		err(1, "preparing agent");
+
+	for (i = 0; i < sizeof(ops)/sizeof(ops[0]); i++) {
+		diag("Testing %s after death", operation_name(ops[i]));
+		ok1(test_death(ops[i], agent));
+	}
+
+	return exit_status();
+}

ccan/tdb2/test/run-firstkey-nextkey.c

@@ -5,6 +5,7 @@
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tdb2/traverse.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"
 

ccan/tdb2/test/run-missing-entries.c

@@ -6,6 +6,7 @@
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tdb2/traverse.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"
 

ccan/tdb2/test/run-record-expand.c

@@ -4,6 +4,7 @@
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"
 

ccan/tdb2/test/run-remap-in-read_traverse.c

@@ -7,6 +7,7 @@
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tdb2/traverse.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 #include "external-agent.h"
 #include "logging.h"

ccan/tdb2/test/run-seed.c

@@ -4,6 +4,7 @@
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"
 

ccan/tdb2/test/run-simple-delete.c

@@ -4,6 +4,7 @@
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"
 

ccan/tdb2/test/run-summary.c

@@ -5,6 +5,7 @@
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tdb2/summary.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"
 

ccan/tdb2/test/run-tdb_errorstr.c

@@ -4,6 +4,7 @@
 #include <ccan/tdb2/io.c>
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 
 int main(int argc, char *argv[])

ccan/tdb2/test/run-traverse.c

@@ -5,6 +5,7 @@
 #include <ccan/tdb2/hash.c>
 #include <ccan/tdb2/check.c>
 #include <ccan/tdb2/traverse.c>
+#include <ccan/tdb2/transaction.c>
 #include <ccan/tap/tap.h>
 #include "logging.h"