tdb2: stricter ordering on expansion lock

It's problematic for transaction commit to get the expansion lock, but
in fact we always grab a hash lock before the transaction lock, so it
doesn't really need it (the transaction locks the entire database).

Assert that this is true, and fix up a few lowlevel tests where it wasn't.

ccan/tdb2/free.c

@@ -530,6 +530,14 @@ static int tdb_expand(struct tdb_context *tdb, tdb_len_t size)
 	/* We need room for the record header too. */
 	wanted = sizeof(struct tdb_used_record) + size;
 
+	/* Need to hold a hash lock to expand DB: transactions rely on it. */
+	if (!(tdb->flags & TDB_NOLOCK)
+	    && !tdb->allrecord_lock.count && !tdb_has_hash_locks(tdb)) {
+		tdb->log(tdb, TDB_DEBUG_FATAL, tdb->log_priv,
+			 "tdb_expand: must hold lock during expand\n");
+		return -1;
+	}
+
 	/* Only one person can expand file at a time. */
 	if (tdb_lock_expand(tdb, F_WRLCK) != 0)
 		return -1;

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

@@ -17,7 +17,7 @@ int main(int argc, char *argv[])
 			TDB_INTERNAL|TDB_CONVERT, TDB_CONVERT,
 			TDB_NOMMAP|TDB_CONVERT };
 
-	plan_tests(sizeof(flags) / sizeof(flags[0]) * 7 + 1);
+	plan_tests(sizeof(flags) / sizeof(flags[0]) * 11 + 1);
 
 	for (i = 0; i < sizeof(flags) / sizeof(flags[0]); i++) {
 		tdb = tdb_open("run-expand.tdb", flags[i],
@@ -27,12 +27,17 @@ int main(int argc, char *argv[])
 			continue;
 
 		val = tdb->map_size;
+		/* Need some hash lock for expand. */
+		ok1(tdb_lock_hashes(tdb, 0, 1, F_WRLCK, TDB_LOCK_WAIT) == 0);
 		ok1(tdb_expand(tdb, 1) == 0);
 		ok1(tdb->map_size >= val + 1 * TDB_EXTENSION_FACTOR);
+		ok1(tdb_unlock_hashes(tdb, 0, 1, F_WRLCK) == 0);
 		ok1(tdb_check(tdb, NULL, NULL) == 0);
 
 		val = tdb->map_size;
+		ok1(tdb_lock_hashes(tdb, 0, 1, F_WRLCK, TDB_LOCK_WAIT) == 0);
 		ok1(tdb_expand(tdb, 1024) == 0);
+		ok1(tdb_unlock_hashes(tdb, 0, 1, F_WRLCK) == 0);
 		ok1(tdb->map_size >= val + 1024 * TDB_EXTENSION_FACTOR);
 		ok1(tdb_check(tdb, NULL, NULL) == 0);
 		tdb_close(tdb);

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

@@ -34,7 +34,7 @@ int main(int argc, char *argv[])
 			TDB_INTERNAL|TDB_CONVERT, TDB_CONVERT,
 			TDB_NOMMAP|TDB_CONVERT };
 
-	plan_tests(sizeof(flags) / sizeof(flags[0]) * 7 + 1);
+	plan_tests(sizeof(flags) / sizeof(flags[0]) * 9 + 1);
 
 	for (i = 0; i < sizeof(flags) / sizeof(flags[0]); i++) {
 		TDB_DATA k;
@@ -51,8 +51,11 @@ int main(int argc, char *argv[])
 			continue;
 
 		ok1(empty_freelist(tdb));
+		/* Need some hash lock for expand. */
+		ok1(tdb_lock_hashes(tdb, 0, 1, F_WRLCK, TDB_LOCK_WAIT) == 0);
 		/* Create some free space. */
 		ok1(tdb_expand(tdb, 1) == 0);
+		ok1(tdb_unlock_hashes(tdb, 0, 1, F_WRLCK) == 0);
 		ok1(tdb_check(tdb, NULL, NULL) == 0);
 		ok1(!empty_freelist(tdb));
 

ccan/tdb2/transaction.c

@@ -454,7 +454,6 @@ static void _tdb_transaction_cancel(struct tdb_context *tdb)
 	tdb->methods = tdb->transaction->io_methods;
 
 	tdb_transaction_unlock(tdb, F_WRLCK);
-	tdb_unlock_expand(tdb, F_WRLCK);
 
 	if (tdb_has_open_lock(tdb))
 		tdb_unlock_open(tdb);
@@ -516,10 +515,6 @@ int tdb_transaction_start(struct tdb_context *tdb)
 		goto fail_allrecord_lock;
 	}
 
-	if (tdb_lock_expand(tdb, F_WRLCK) != 0) {
-		goto fail_expand_lock;
-	}
-
 	/* make sure we know about any file expansions already done by
 	   anyone else */
 	tdb->methods->oob(tdb, tdb->map_size + 1, true);
@@ -531,8 +526,6 @@ int tdb_transaction_start(struct tdb_context *tdb)
 	tdb->methods = &transaction_methods;
 	return 0;
 
-fail_expand_lock:
-	tdb_allrecord_unlock(tdb, F_RDLCK);
 fail_allrecord_lock:
 	tdb_transaction_unlock(tdb, F_WRLCK);
 	SAFE_FREE(tdb->transaction->blocks);
@@ -884,6 +877,7 @@ static int _tdb_transaction_prepare_commit(struct tdb_context *tdb)
 		return -1;
 	}
 
+	/* Since we have whole db locked, we don't need the expansion lock. */
 	if (!(tdb->flags & TDB_NOSYNC)) {
 		/* write the recovery data to the end of the file */
 		if (transaction_setup_recovery(tdb, &tdb->transaction->magic_offset) == -1) {