tdb2: add a capability list from the header.

This allows even more extensibility in future: in particular, the top
bits of each capability tell us what to do if we don't understand it:
fail the open, fail to open for write, or don't try to check the
format.

tdb_check needs to understand the capability list so it can know to
skip over it: each element in the list is prefixed with the type tag
and the length.

Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>
(Imported from CCAN commit 35f198de1851a7d57064546b7ced677b6fabee27)

9 files changed, 410 insertions, 27 deletions

diff --git a/lib/tdb2/check.c b/lib/tdb2/check.c
index 230eaee8e3..238a5b3a46 100644
--- a/lib/tdb2/check.c
+++ b/lib/tdb2/check.c
@@ -31,11 +31,12 @@ static bool append(tdb_off_t **arr, size_t *num, tdb_off_t off)
 }
 
 static enum TDB_ERROR check_header(struct tdb_context *tdb, tdb_off_t *recovery,
-				   uint64_t *features)
+				   uint64_t *features, size_t *num_capabilities)
 {
 	uint64_t hash_test;
 	struct tdb_header hdr;
 	enum TDB_ERROR ecode;
+	tdb_off_t off, next;
 
 	ecode = tdb_read_convert(tdb, 0, &hdr, sizeof(hdr));
 	if (ecode != TDB_SUCCESS) {
@@ -81,6 +82,24 @@ static enum TDB_ERROR check_header(struct tdb_context *tdb, tdb_off_t *recovery,
 		}
 	}
 
+	for (off = hdr.capabilities; off && ecode == TDB_SUCCESS; off = next) {
+		const struct tdb_capability *cap;
+		enum TDB_ERROR err;
+
+		cap = tdb_access_read(tdb, off, sizeof(*cap), true);
+		if (TDB_PTR_IS_ERR(cap)) {
+			return TDB_PTR_ERR(cap);
+		}
+
+		/* All capabilities are unknown. */
+		err = unknown_capability(tdb, "tdb_check", cap->type);
+		next = cap->next;
+		tdb_access_release(tdb, cap);
+		if (err)
+			return err;
+		(*num_capabilities)++;
+	}
+
 	/* Don't check reserved: they *can* be used later. */
 	return TDB_SUCCESS;
 }
@@ -435,12 +454,12 @@ fail:
 
 static enum TDB_ERROR check_hash(struct tdb_context *tdb,
 				 tdb_off_t used[],
-				 size_t num_used, size_t num_ftables,
+				 size_t num_used, size_t num_other_used,
 				 enum TDB_ERROR (*check)(TDB_DATA, TDB_DATA, void *),
 				 void *data)
 {
-	/* Free tables also show up as used. */
-	size_t num_found = num_ftables;
+	/* Free tables and capabilities also show up as used. */
+	size_t num_found = num_other_used;
 	enum TDB_ERROR ecode;
 
 	ecode = check_hash_tree(tdb, offsetof(struct tdb_header, hashtable),
@@ -698,7 +717,8 @@ static enum TDB_ERROR check_linear(struct tdb_context *tdb,
 		} else if (rec_magic(&rec.u) == TDB_USED_MAGIC
 			   || rec_magic(&rec.u) == TDB_CHAIN_MAGIC
 			   || rec_magic(&rec.u) == TDB_HTABLE_MAGIC
-			   || rec_magic(&rec.u) == TDB_FTABLE_MAGIC) {
+			   || rec_magic(&rec.u) == TDB_FTABLE_MAGIC
+			   || rec_magic(&rec.u) == TDB_CAP_MAGIC) {
 			uint64_t klen, dlen, extra;
 
 			/* This record is used! */
@@ -734,7 +754,8 @@ static enum TDB_ERROR check_linear(struct tdb_context *tdb,
 
 			/* Check that records have correct 0 at end (but may
 			 * not in future). */
-			if (extra && !features) {
+			if (extra && !features
+			    && rec_magic(&rec.u) != TDB_CAP_MAGIC) {
 				const char *p;
 				char c;
 				p = tdb_access_read(tdb, off + sizeof(rec.u)
@@ -778,13 +799,14 @@ enum TDB_ERROR tdb_check_(struct tdb_context *tdb,
 			  void *data)
 {
 	tdb_off_t *fr = NULL, *used = NULL, ft, recovery;
-	size_t num_free = 0, num_used = 0, num_found = 0, num_ftables = 0;
+	size_t num_free = 0, num_used = 0, num_found = 0, num_ftables = 0,
+		num_capabilities = 0;
 	uint64_t features;
 	enum TDB_ERROR ecode;
 
 	if (tdb->flags & TDB_CANT_CHECK) {
 		return tdb_logerr(tdb, TDB_SUCCESS, TDB_LOG_WARNING,
-				  "tdb_check: database has unknown features,"
+				  "tdb_check: database has unknown capability,"
 				  " cannot check.");
 	}
 
@@ -805,7 +827,7 @@ enum TDB_ERROR tdb_check_(struct tdb_context *tdb,
 		return tdb->last_error = ecode;
 	}
 
-	ecode = check_header(tdb, &recovery, &features);
+	ecode = check_header(tdb, &recovery, &features, &num_capabilities);
 	if (ecode != TDB_SUCCESS)
 		goto out;
 
@@ -828,7 +850,8 @@ enum TDB_ERROR tdb_check_(struct tdb_context *tdb,
 	}
 
 	/* FIXME: Check key uniqueness? */
-	ecode = check_hash(tdb, used, num_used, num_ftables, check, data);
+	ecode = check_hash(tdb, used, num_used, num_ftables + num_capabilities,
+			   check, data);
 	if (ecode != TDB_SUCCESS)
 		goto out;
 
diff --git a/lib/tdb2/open.c b/lib/tdb2/open.c
index b76bd24b2a..4965d86288 100644
--- a/lib/tdb2/open.c
+++ b/lib/tdb2/open.c
@@ -135,6 +135,7 @@ static enum TDB_ERROR tdb_new_database(struct tdb_context *tdb,
 	newdb.hdr.recovery = 0;
 	newdb.hdr.features_used = newdb.hdr.features_offered = TDB_FEATURE_MASK;
 	newdb.hdr.seqnum = 0;
+	newdb.hdr.capabilities = 0;
 	memset(newdb.hdr.reserved, 0, sizeof(newdb.hdr.reserved));
 	/* Initial hashes are empty. */
 	memset(newdb.hdr.hashtable, 0, sizeof(newdb.hdr.hashtable));
@@ -375,6 +376,54 @@ static bool is_tdb1(struct tdb1_header *hdr, const void *buf, ssize_t rlen)
 		|| hdr->version == TDB1_BYTEREV(TDB1_VERSION);
 }
 
+/* The top three bits of the capability tell us whether it matters. */
+enum TDB_ERROR unknown_capability(struct tdb_context *tdb, const char *caller,
+				  tdb_off_t type)
+{
+	if (type & TDB_CAP_NOOPEN) {
+		return tdb_logerr(tdb, TDB_ERR_IO, TDB_LOG_ERROR,
+				  "%s: file has unknown capability %llu",
+				  caller, type & TDB_CAP_NOOPEN);
+	}
+
+	if ((type & TDB_CAP_NOWRITE) && !(tdb->flags & TDB_RDONLY)) {
+		return tdb_logerr(tdb, TDB_ERR_RDONLY, TDB_LOG_ERROR,
+				  "%s: file has unknown capability %llu"
+				  " (cannot write to it)",
+				  caller, type & TDB_CAP_NOOPEN);
+	}
+
+	if (type & TDB_CAP_NOCHECK) {
+		tdb->flags |= TDB_CANT_CHECK;
+	}
+	return TDB_SUCCESS;
+}
+
+static enum TDB_ERROR capabilities_ok(struct tdb_context *tdb,
+				      tdb_off_t capabilities)
+{
+	tdb_off_t off, next;
+	enum TDB_ERROR ecode = TDB_SUCCESS;
+	const struct tdb_capability *cap;
+
+	/* Check capability list. */
+	for (off = capabilities; off && ecode == TDB_SUCCESS; off = next) {
+		cap = tdb_access_read(tdb, off, sizeof(*cap), true);
+		if (TDB_PTR_IS_ERR(cap)) {
+			return TDB_PTR_ERR(cap);
+		}
+
+		switch (cap->type & TDB_CAP_TYPE_MASK) {
+		/* We don't understand any capabilities (yet). */
+		default:
+			ecode = unknown_capability(tdb, "tdb_open", cap->type);
+		}
+		next = cap->next;
+		tdb_access_release(tdb, cap);
+	}
+	return ecode;
+}
+
 struct tdb_context *tdb_open(const char *name, int tdb_flags,
 			     int open_flags, mode_t mode,
 			     union tdb_attribute *attr)
@@ -667,6 +716,11 @@ struct tdb_context *tdb_open(const char *name, int tdb_flags,
 		goto fail;
 	}
 
+	ecode = capabilities_ok(tdb, hdr.capabilities);
+	if (ecode != TDB_SUCCESS) {
+		goto fail;
+	}
+
 	/* Clear any features we don't understand. */
 	if ((open_flags & O_ACCMODE) != O_RDONLY) {
 		hdr.features_used &= TDB_FEATURE_MASK;
diff --git a/lib/tdb2/private.h b/lib/tdb2/private.h
index 87245a5970..85cd15a817 100644
--- a/lib/tdb2/private.h
+++ b/lib/tdb2/private.h
@@ -72,11 +72,18 @@ typedef uint64_t tdb_off_t;
 #define TDB_HTABLE_MAGIC ((uint64_t)0x1888)
 #define TDB_CHAIN_MAGIC ((uint64_t)0x1777)
 #define TDB_FTABLE_MAGIC ((uint64_t)0x1666)
+#define TDB_CAP_MAGIC ((uint64_t)0x1555)
 #define TDB_FREE_MAGIC ((uint64_t)0xFE)
 #define TDB_HASH_MAGIC (0xA1ABE11A01092008ULL)
 #define TDB_RECOVERY_MAGIC (0xf53bc0e7ad124589ULL)
 #define TDB_RECOVERY_INVALID_MAGIC (0x0ULL)
 
+/* Capability bits. */
+#define TDB_CAP_TYPE_MASK	0x1FFFFFFFFFFFFFFFULL
+#define TDB_CAP_NOCHECK		0x8000000000000000ULL
+#define TDB_CAP_NOWRITE		0x4000000000000000ULL
+#define TDB_CAP_NOOPEN		0x2000000000000000ULL
+
 #define TDB_OFF_IS_ERR(off) unlikely(off >= (tdb_off_t)(long)TDB_ERR_LAST)
 #define TDB_OFF_TO_ERR(off) ((enum TDB_ERROR)(long)(off))
 #define TDB_ERR_TO_OFF(ecode) ((tdb_off_t)(long)(ecode))
@@ -242,7 +249,8 @@ struct tdb_header {
 
 	uint64_t seqnum; /* Sequence number for TDB_SEQNUM */
 
-	tdb_off_t reserved[23];
+	tdb_off_t capabilities; /* Optional linked list of capabilities. */
+	tdb_off_t reserved[22];
 
 	/* Top level hash table. */
 	tdb_off_t hashtable[1ULL << TDB_TOPLEVEL_HASH_BITS];
@@ -254,6 +262,13 @@ struct tdb_freetable {
 	tdb_off_t buckets[TDB_FREE_BUCKETS];
 };
 
+struct tdb_capability {
+	struct tdb_used_record hdr;
+	tdb_off_t type;
+	tdb_off_t next;
+	/* ... */
+};
+
 /* Information about a particular (locked) hash entry. */
 struct hash_info {
 	/* Full hash value of entry. */
@@ -389,6 +404,8 @@ enum TDB_ERROR delete_from_hash(struct tdb_context *tdb, struct hash_info *h);
 
 /* For tdb_check */
 bool is_subhash(tdb_off_t val);
+enum TDB_ERROR unknown_capability(struct tdb_context *tdb, const char *caller,
+				  tdb_off_t type);
 
 /* free.c: */
 enum TDB_ERROR tdb_ftable_init(struct tdb_context *tdb);
@@ -407,7 +424,7 @@ enum TDB_ERROR add_free_record(struct tdb_context *tdb,
 			       enum tdb_lock_flags waitflag,
 			       bool coalesce_ok);
 
-/* Set up header for a used/ftable/htable/chain record. */
+/* Set up header for a used/ftable/htable/chain/capability record. */
 enum TDB_ERROR set_header(struct tdb_context *tdb,
 			  struct tdb_used_record *rec,
 			  unsigned magic, uint64_t keylen, uint64_t datalen,
diff --git a/lib/tdb2/test/failtest_helper.h b/lib/tdb2/test/failtest_helper.h
index a3c680885d..4130aff1eb 100644
--- a/lib/tdb2/test/failtest_helper.h
+++ b/lib/tdb2/test/failtest_helper.h
@@ -4,7 +4,7 @@
 #include <stdbool.h>
 
 /* FIXME: Check these! */
-#define INITIAL_TDB_MALLOC	"open.c", 396, FAILTEST_MALLOC
+#define INITIAL_TDB_MALLOC	"open.c", 445, FAILTEST_MALLOC
 #define URANDOM_OPEN		"open.c", 62, FAILTEST_OPEN
 #define URANDOM_READ		"open.c", 42, FAILTEST_READ
 
diff --git a/lib/tdb2/test/layout.c b/lib/tdb2/test/layout.c
index 6c601ddb85..0bce5dd89e 100644
--- a/lib/tdb2/test/layout.c
+++ b/lib/tdb2/test/layout.c
@@ -39,6 +39,26 @@ void tdb_layout_add_free(struct tdb_layout *layout, tdb_len_t len,
 	add(layout, elem);
 }
 
+void tdb_layout_add_capability(struct tdb_layout *layout,
+			       uint64_t type,
+			       bool write_breaks,
+			       bool check_breaks,
+			       bool open_breaks,
+			       tdb_len_t extra)
+{
+	union tdb_layout_elem elem;
+	elem.base.type = CAPABILITY;
+	elem.capability.type = type;
+	if (write_breaks)
+		elem.capability.type |= TDB_CAP_NOWRITE;
+	if (open_breaks)
+		elem.capability.type |= TDB_CAP_NOOPEN;
+	if (check_breaks)
+		elem.capability.type |= TDB_CAP_NOCHECK;
+	elem.capability.extra = extra;
+	add(layout, elem);
+}
+
 static struct tdb_data dup_key(struct tdb_data key)
 {
 	struct tdb_data ret;
@@ -81,6 +101,11 @@ static tdb_len_t hashtable_len(struct tle_hashtable *htable)
 		+ htable->extra;
 }
 
+static tdb_len_t capability_len(struct tle_capability *cap)
+{
+	return sizeof(struct tdb_capability) + cap->extra;
+}
+
 static tdb_len_t freetable_len(struct tle_freetable *ftable)
 {
 	return sizeof(struct tdb_freetable);
@@ -122,6 +147,26 @@ static void set_hashtable(void *mem, struct tdb_context *tdb,
 	add_zero_pad(u, len, htable->extra);
 }
 
+static void set_capability(void *mem, struct tdb_context *tdb,
+			   struct tle_capability *cap, struct tdb_header *hdr,
+			   tdb_off_t last_cap)
+{
+	struct tdb_capability *c = mem;
+	tdb_len_t len = sizeof(*c) - sizeof(struct tdb_used_record) + cap->extra;
+
+	c->type = cap->type;
+	c->next = 0;
+	set_header(tdb, &c->hdr, TDB_CAP_MAGIC, 0, len, len, 0);
+
+	/* Append to capability list. */
+	if (!last_cap) {
+		hdr->capabilities = cap->base.off;
+	} else {
+		c = (struct tdb_capability *)((char *)hdr + last_cap);
+		c->next = cap->base.off;
+	}
+}
+
 static void set_freetable(void *mem, struct tdb_context *tdb,
 			 struct tle_freetable *freetable, struct tdb_header *hdr,
 			 tdb_off_t last_ftable)
@@ -228,10 +273,11 @@ static struct tle_freetable *find_ftable(struct tdb_layout *layout, unsigned num
 
 /* FIXME: Support TDB_CONVERT */
 struct tdb_context *tdb_layout_get(struct tdb_layout *layout,
+				   void (*freefn)(void *),
 				   union tdb_attribute *attr)
 {
 	unsigned int i;
-	tdb_off_t off, len, last_ftable;
+	tdb_off_t off, len, last_ftable, last_cap;
 	char *mem;
 	struct tdb_context *tdb;
 
@@ -254,6 +300,9 @@ struct tdb_context *tdb_layout_get(struct tdb_layout *layout,
 		case HASHTABLE:
 			len = hashtable_len(&e->hashtable);
 			break;
+		case CAPABILITY:
+			len = capability_len(&e->capability);
+			break;
 		default:
 			abort();
 		}
@@ -268,11 +317,12 @@ struct tdb_context *tdb_layout_get(struct tdb_layout *layout,
 	memcpy(mem, tdb->file->map_ptr, sizeof(struct tdb_header));
 
 	/* Mug the tdb we have to make it use this. */
-	free(tdb->file->map_ptr);
+	freefn(tdb->file->map_ptr);
 	tdb->file->map_ptr = mem;
 	tdb->file->map_size = off;
 
 	last_ftable = 0;
+	last_cap = 0;
 	for (i = 0; i < layout->num_elems; i++) {
 		union tdb_layout_elem *e = &layout->elem[i];
 		switch (e->base.type) {
@@ -290,6 +340,11 @@ struct tdb_context *tdb_layout_get(struct tdb_layout *layout,
 		case HASHTABLE:
 			set_hashtable(mem + e->base.off, tdb, &e->hashtable);
 			break;
+		case CAPABILITY:
+			set_capability(mem + e->base.off, tdb, &e->capability,
+				       (struct tdb_header *)mem, last_cap);
+			last_cap = e->base.off;
+			break;
 		}
 	}
 	/* Must have a free table! */
@@ -316,10 +371,10 @@ struct tdb_context *tdb_layout_get(struct tdb_layout *layout,
 	return tdb;
 }
 
-void tdb_layout_write(struct tdb_layout *layout, union tdb_attribute *attr,
-		      const char *filename)
+void tdb_layout_write(struct tdb_layout *layout, void (*freefn)(void *),
+		       union tdb_attribute *attr, const char *filename)
 {
-	struct tdb_context *tdb = tdb_layout_get(layout, attr);
+	struct tdb_context *tdb = tdb_layout_get(layout, freefn, attr);
 	int fd;
 
 	fd = open(filename, O_WRONLY|O_TRUNC|O_CREAT,  0600);
diff --git a/lib/tdb2/test/layout.h b/lib/tdb2/test/layout.h
index 6649113766..9a71484601 100644
--- a/lib/tdb2/test/layout.h
+++ b/lib/tdb2/test/layout.h
@@ -9,21 +9,30 @@ void tdb_layout_add_free(struct tdb_layout *layout, tdb_len_t len,
 void tdb_layout_add_used(struct tdb_layout *layout,
 			 TDB_DATA key, TDB_DATA data,
 			 tdb_len_t extra);
+void tdb_layout_add_capability(struct tdb_layout *layout,
+			       uint64_t type,
+			       bool write_breaks,
+			       bool check_breaks,
+			       bool open_breaks,
+			       tdb_len_t extra);
+
 #if 0 /* FIXME: Allow allocation of subtables */
 void tdb_layout_add_hashtable(struct tdb_layout *layout,
 			      int htable_parent, /* -1 == toplevel */
 			      unsigned int bucket,
 			      tdb_len_t extra);
 #endif
+/* freefn is needed if we're using failtest_free. */
 struct tdb_context *tdb_layout_get(struct tdb_layout *layout,
+				   void (*freefn)(void *),
 				   union tdb_attribute *attr);
-void tdb_layout_write(struct tdb_layout *layout, union tdb_attribute *attr,
-		      const char *filename);
+void tdb_layout_write(struct tdb_layout *layout, void (*freefn)(void *),
+		       union tdb_attribute *attr, const char *filename);
 
 void tdb_layout_free(struct tdb_layout *layout);
 
 enum layout_type {
-	FREETABLE, FREE, DATA, HASHTABLE,
+	FREETABLE, FREE, DATA, HASHTABLE, CAPABILITY
 };
 
 /* Shared by all union members. */
@@ -56,12 +65,19 @@ struct tle_hashtable {
 	tdb_len_t extra;
 };
 
+struct tle_capability {
+	struct tle_base base;
+	uint64_t type;
+	tdb_len_t extra;
+};
+
 union tdb_layout_elem {
 	struct tle_base base;
 	struct tle_freetable ftable;
 	struct tle_free free;
 	struct tle_used used;
 	struct tle_hashtable hashtable;
+	struct tle_capability capability;
 };
 
 struct tdb_layout {
diff --git a/lib/tdb2/test/run-03-coalesce.c b/lib/tdb2/test/run-03-coalesce.c
index c64b2bc57c..99f94fe16d 100644
--- a/lib/tdb2/test/run-03-coalesce.c
+++ b/lib/tdb2/test/run-03-coalesce.c
@@ -36,7 +36,7 @@ int main(int argc, char *argv[])
 	tdb_layout_add_freetable(layout);
 	len = 1024;
 	tdb_layout_add_free(layout, len, 0);
-	tdb_layout_write(layout, &tap_log_attr, "run-03-coalesce.tdb");
+	tdb_layout_write(layout, free, &tap_log_attr, "run-03-coalesce.tdb");
 	/* NOMMAP is for lockcheck. */
 	tdb = tdb_open("run-03-coalesce.tdb", TDB_NOMMAP, O_RDWR, 0,
 		       &tap_log_attr);
@@ -62,7 +62,7 @@ int main(int argc, char *argv[])
 	tdb_layout_add_freetable(layout);
 	tdb_layout_add_free(layout, 1024, 0);
 	tdb_layout_add_used(layout, key, data, 6);
-	tdb_layout_write(layout, &tap_log_attr, "run-03-coalesce.tdb");
+	tdb_layout_write(layout, free, &tap_log_attr, "run-03-coalesce.tdb");
 	/* NOMMAP is for lockcheck. */
 	tdb = tdb_open("run-03-coalesce.tdb", TDB_NOMMAP, O_RDWR, 0,
 		       &tap_log_attr);
@@ -88,7 +88,7 @@ int main(int argc, char *argv[])
 	tdb_layout_add_freetable(layout);
 	tdb_layout_add_free(layout, 1024, 0);
 	tdb_layout_add_free(layout, 2048, 0);
-	tdb_layout_write(layout, &tap_log_attr, "run-03-coalesce.tdb");
+	tdb_layout_write(layout, free, &tap_log_attr, "run-03-coalesce.tdb");
 	/* NOMMAP is for lockcheck. */
 	tdb = tdb_open("run-03-coalesce.tdb", TDB_NOMMAP, O_RDWR, 0,
 		       &tap_log_attr);
@@ -118,7 +118,7 @@ int main(int argc, char *argv[])
 	tdb_layout_add_free(layout, 1024, 0);
 	tdb_layout_add_free(layout, 512, 0);
 	tdb_layout_add_used(layout, key, data, 6);
-	tdb_layout_write(layout, &tap_log_attr, "run-03-coalesce.tdb");
+	tdb_layout_write(layout, free, &tap_log_attr, "run-03-coalesce.tdb");
 	/* NOMMAP is for lockcheck. */
 	tdb = tdb_open("run-03-coalesce.tdb", TDB_NOMMAP, O_RDWR, 0,
 		       &tap_log_attr);
@@ -147,7 +147,7 @@ int main(int argc, char *argv[])
 	tdb_layout_add_free(layout, 1024, 0);
 	tdb_layout_add_free(layout, 512, 0);
 	tdb_layout_add_free(layout, 256, 0);
-	tdb_layout_write(layout, &tap_log_attr, "run-03-coalesce.tdb");
+	tdb_layout_write(layout, free, &tap_log_attr, "run-03-coalesce.tdb");
 	/* NOMMAP is for lockcheck. */
 	tdb = tdb_open("run-03-coalesce.tdb", TDB_NOMMAP, O_RDWR, 0,
 		       &tap_log_attr);
diff --git a/lib/tdb2/test/run-50-multiple-freelists.c b/lib/tdb2/test/run-50-multiple-freelists.c
index 10eaf41d98..44fee941cf 100644
--- a/lib/tdb2/test/run-50-multiple-freelists.c
+++ b/lib/tdb2/test/run-50-multiple-freelists.c
@@ -35,7 +35,7 @@ int main(int argc, char *argv[])
 	key.dsize--;
 	tdb_layout_add_used(layout, key, data, 8);
 	tdb_layout_add_free(layout, 40, 0);
-	tdb = tdb_layout_get(layout, &seed);
+	tdb = tdb_layout_get(layout, free, &seed);
 	ok1(tdb_check(tdb, NULL, NULL) == 0);
 
 	off = get_free(tdb, 0, 80 - sizeof(struct tdb_used_record), 0,
diff --git a/lib/tdb2/test/run-capabilities.c b/lib/tdb2/test/run-capabilities.c
new file mode 100644
index 0000000000..e302656108
--- /dev/null
+++ b/lib/tdb2/test/run-capabilities.c
@@ -0,0 +1,218 @@
+#include <ccan/failtest/failtest_override.h>
+#include "tdb2-source.h"
+#include <ccan/tap/tap.h>
+#include "logging.h"
+#include "layout.h"
+#include "failtest_helper.h"
+#include <stdarg.h>
+#include <err.h>
+
+static size_t len_of(bool breaks_check, bool breaks_write, bool breaks_open)
+{
+	size_t len = 0;
+	if (breaks_check)
+		len += 8;
+	if (breaks_write)
+		len += 16;
+	if (breaks_open)
+		len += 32;
+	return len;
+}
+
+/* Creates a TDB with various capabilities. */
+static void create_tdb(const char *name,
+		       unsigned int cap,
+		       bool breaks_check,
+		       bool breaks_write,
+		       bool breaks_open, ...)
+{
+	TDB_DATA key, data;
+	va_list ap;
+	struct tdb_layout *layout;
+	struct tdb_context *tdb;
+	int fd;
+
+	key = tdb_mkdata("Hello", 5);
+	data = tdb_mkdata("world", 5);
+
+	/* Create a TDB with some data, and some capabilities */
+	layout = new_tdb_layout();
+	tdb_layout_add_freetable(layout);
+	tdb_layout_add_used(layout, key, data, 6);
+	tdb_layout_add_free(layout, 80, 0);
+	tdb_layout_add_capability(layout, cap,
+				  breaks_write, breaks_check, breaks_open,
+				  len_of(breaks_check, breaks_write, breaks_open));
+
+	va_start(ap, breaks_open);
+	while ((cap = va_arg(ap, int)) != 0) {
+		breaks_check = va_arg(ap, int);
+		breaks_write = va_arg(ap, int);
+		breaks_open = va_arg(ap, int);
+
+		key.dsize--;
+		tdb_layout_add_used(layout, key, data, 11 - key.dsize);
+		tdb_layout_add_free(layout, 80, 0);
+		tdb_layout_add_capability(layout, cap,
+					  breaks_write, breaks_check,
+					  breaks_open,
+					  len_of(breaks_check, breaks_write,
+						 breaks_open));
+	}
+	va_end(ap);
+
+	/* We open-code this, because we need to use the failtest write. */
+	tdb = tdb_layout_get(layout, failtest_free, &tap_log_attr);
+
+	fd = open(name, O_RDWR|O_TRUNC|O_CREAT, 0600);
+	if (fd < 0)
+		err(1, "opening %s for writing", name);
+	if (write(fd, tdb->file->map_ptr, tdb->file->map_size)
+	    != tdb->file->map_size)
+		err(1, "writing %s", name);
+	close(fd);
+	tdb_close(tdb);
+	tdb_layout_free(layout);
+}
+
+/* Note all the "goto out" early exits: they're to shorten failtest time. */
+int main(int argc, char *argv[])
+{
+	struct tdb_context *tdb;
+
+	failtest_init(argc, argv);
+	failtest_hook = block_repeat_failures;
+	failtest_exit_check = exit_check_log;
+	plan_tests(35);
+
+	failtest_suppress = true;
+	/* Capability says you can ignore it? */
+	create_tdb("run-capabilities.tdb", 1, false, false, false, 0);
+
+	failtest_suppress = false;
+	tdb = tdb_open("run-capabilities.tdb", TDB_DEFAULT, O_RDWR, 0,
+		       &tap_log_attr);
+	failtest_suppress = true;
+	if (!ok1(tdb))
+		goto out;
+	ok1(tap_log_messages == 0);
+	ok1(tdb_check(tdb, NULL, NULL) == TDB_SUCCESS);
+	ok1(tap_log_messages == 0);
+	tdb_close(tdb);
+
+	/* Two capabilitues say you can ignore them? */
+	create_tdb("run-capabilities.tdb",
+		   1, false, false, false,
+		   2, false, false, false, 0);
+
+	failtest_suppress = false;
+	tdb = tdb_open("run-capabilities.tdb", TDB_DEFAULT, O_RDWR, 0,
+		       &tap_log_attr);
+	failtest_suppress = true;
+	if (!ok1(tdb))
+		goto out;
+	ok1(tap_log_messages == 0);
+	ok1(tdb_check(tdb, NULL, NULL) == TDB_SUCCESS);
+	ok1(tap_log_messages == 0);
+	tdb_close(tdb);
+
+	/* Capability says you can't check. */
+	create_tdb("run-capabilities.tdb",
+		   1, false, false, false,
+		   2, true, false, false, 0);
+
+	failtest_suppress = false;
+	tdb = tdb_open("run-capabilities.tdb", TDB_DEFAULT, O_RDWR, 0,
+		       &tap_log_attr);
+	failtest_suppress = true;
+	if (!ok1(tdb))
+		goto out;
+	ok1(tap_log_messages == 0);
+	ok1(tdb_get_flags(tdb) & TDB_CANT_CHECK);
+	ok1(tdb_check(tdb, NULL, NULL) == TDB_SUCCESS);
+	/* We expect a warning! */
+	ok1(tap_log_messages == 1);
+	ok1(strstr(log_last, "capabilit"));
+	tdb_close(tdb);
+
+	/* Capability says you can't write. */
+	create_tdb("run-capabilities.tdb",
+		   1, false, false, false,
+		   2, false, true, false, 0);
+
+	failtest_suppress = false;
+	tdb = tdb_open("run-capabilities.tdb", TDB_DEFAULT, O_RDWR, 0,
+		       &tap_log_attr);
+	failtest_suppress = true;
+	/* We expect a message. */
+	ok1(!tdb);
+	if (!ok1(tap_log_messages == 2))
+		goto out;
+	if (!ok1(strstr(log_last, "unknown")))
+		goto out;
+	ok1(strstr(log_last, "write"));
+
+	/* We can open it read-only though! */
+	failtest_suppress = false;
+	tdb = tdb_open("run-capabilities.tdb", TDB_DEFAULT, O_RDONLY, 0,
+		       &tap_log_attr);
+	failtest_suppress = true;
+	if (!ok1(tdb))
+		goto out;
+	ok1(tap_log_messages == 2);
+	ok1(tdb_check(tdb, NULL, NULL) == TDB_SUCCESS);
+	ok1(tap_log_messages == 2);
+	tdb_close(tdb);
+
+	/* Capability says you can't open. */
+	create_tdb("run-capabilities.tdb",
+		   1, false, false, false,
+		   2, false, false, true, 0);
+
+	failtest_suppress = false;
+	tdb = tdb_open("run-capabilities.tdb", TDB_DEFAULT, O_RDWR, 0,
+		       &tap_log_attr);
+	failtest_suppress = true;
+	/* We expect a message. */
+	ok1(!tdb);
+	if (!ok1(tap_log_messages == 3))
+		goto out;
+	if (!ok1(strstr(log_last, "unknown")))
+		goto out;
+
+	/* Combine capabilities correctly. */
+	create_tdb("run-capabilities.tdb",
+		   1, false, false, false,
+		   2, true, false, false,
+		   3, false, true, false, 0);
+
+	failtest_suppress = false;
+	tdb = tdb_open("run-capabilities.tdb", TDB_DEFAULT, O_RDWR, 0,
+		       &tap_log_attr);
+	failtest_suppress = true;
+	/* We expect a message. */
+	ok1(!tdb);
+	if (!ok1(tap_log_messages == 4))
+		goto out;
+	if (!ok1(strstr(log_last, "unknown")))
+		goto out;
+	ok1(strstr(log_last, "write"));
+
+	/* We can open it read-only though! */
+	failtest_suppress = false;
+	tdb = tdb_open("run-capabilities.tdb", TDB_DEFAULT, O_RDONLY, 0,
+		       &tap_log_attr);
+	failtest_suppress = true;
+	if (!ok1(tdb))
+		goto out;
+	ok1(tap_log_messages == 4);
+	ok1(tdb_get_flags(tdb) & TDB_CANT_CHECK);
+	ok1(tdb_check(tdb, NULL, NULL) == TDB_SUCCESS);
+	/* We expect a warning! */
+	ok1(tap_log_messages == 5);
+	ok1(strstr(log_last, "unknown"));
+	tdb_close(tdb);
+
+out:
+	failtest_exit(exit_status());
+}