Add a simple immutable database implementation

3 files changed, 1088 insertions, 0 deletions

diff --git a/src/libdisfluid/Makefile.am b/src/libdisfluid/Makefile.am
index 01a5945..dc377c9 100644
--- a/src/libdisfluid/Makefile.am
+++ b/src/libdisfluid/Makefile.am
@@ -5,6 +5,7 @@ lib_LTLIBRARIES += %D%/libdisfluid.la
   %D%/disfluid-cache-entry.h \
   %D%/disfluid-cache-entry-key.h \
   %D%/disfluid-cache-entry-hash.h \
+  %D%/disfluid-db.h \
   %D%/disfluid-init.h \
   %D%/disfluid-tests.h \
   %D%/disfluid-ui.h \
diff --git a/src/libdisfluid/disfluid-db.h b/src/libdisfluid/disfluid-db.h
new file mode 100644
index 0000000..3fec521
--- /dev/null
+++ b/src/libdisfluid/disfluid-db.h
@@ -0,0 +1,1072 @@
+#ifndef DISFLUID_DISFLUID_DB_INCLUDED
+# define DISFLUID_DISFLUID_DB_INCLUDED
+
+# include <config.h>
+# include "string-desc.h"
+
+MAYBE_UNUSED static int
+db_mark_leaf (const char *db_root, const string_desc_t * id);
+
+MAYBE_UNUSED static int
+db_read (const char *db_root, const string_desc_t * id, string_desc_t * data);
+
+MAYBE_UNUSED static int
+db_write (const char *db_root, string_desc_t * id,
+	  const string_desc_t * data);
+
+MAYBE_UNUSED static int db_unmark (const char *db_root);
+
+/* WARNING: Between the call to db_unmark and db_collect, the database
+   must have been fully traced at least once, including the unfinished
+   transactions. */
+MAYBE_UNUSED static int db_collect (const char *db_root);
+
+MAYBE_UNUSED static int db_check (void);
+
+# include <gnutls/gnutls.h>
+# include <gnutls/crypto.h>
+# include <assert.h>
+# include <sys/stat.h>
+# include "safe-alloc.h"
+# include <dirent.h>
+# include <unistd.h>
+# include <stdbool.h>
+
+# ifndef STREQ
+#  define STREQ(a, b) (strcmp ((a), (b)) != 0)
+# endif/* not STREQ */
+
+static FILE *
+open_file_for_reading (const char *filename, const char *trash_filename)
+{
+  assert (filename != NULL);
+  assert (trash_filename != NULL);
+  FILE *f = fopen (filename, "rb");
+  if (f == NULL)
+    {
+      if (rename (trash_filename, filename) == 0)
+	{
+	  f = fopen (filename, "rb");
+	}
+    }
+  return f;
+}
+
+static int
+db_file_name (const char *db_root, size_t id_length, const char *id,
+	      char **dirname, char **filename, char **temp_filename,
+	      char **trash_filename)
+{
+  int error = 0;
+  *dirname = NULL;
+  *filename = NULL;
+  *temp_filename = NULL;
+  *trash_filename = NULL;
+  gnutls_datum_t id_datum = {
+    .data = (char *) id,
+    .size = id_length
+  };
+  gnutls_datum_t hex_id = {
+    .data = NULL,
+    .size = 0
+  };
+  if (gnutls_hex_encode2 (&id_datum, &hex_id) < 0)
+    {
+      error = -2;
+      goto cleanup;
+    }
+  assert (hex_id.size >= 2);
+  const size_t dirname_length = strlen (db_root) + strlen ("/") + 2;
+  const size_t filename_length =
+    dirname_length + strlen ("/") + hex_id.size - 2;
+  const size_t temp_filename_length = filename_length + strlen ("-XXXXXX");
+  const size_t trash_filename_length = filename_length + strlen ("-trash");
+  if (ALLOC_N (*filename, filename_length + 1) < 0
+      || ALLOC_N (*temp_filename, temp_filename_length + 1) < 0
+      || ALLOC_N (*dirname, dirname_length + 1) < 0
+      || ALLOC_N (*trash_filename, trash_filename_length + 1) < 0)
+    {
+      error = -2;
+      goto cleanup;
+    }
+  strcpy (*dirname, db_root);
+  strcat (*dirname, "/");
+  strncat (*dirname, hex_id.data, 2);
+  strcpy (*filename, *dirname);
+  strcat (*filename, "/");
+  strncat (*filename, hex_id.data + 2, hex_id.size - 2);
+  strcpy (*temp_filename, *filename);
+  strcat (*temp_filename, "-XXXXXX");
+  strcpy (*trash_filename, *filename);
+  strcat (*trash_filename, "-trash");
+  mkdir (db_root, S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH | S_IXOTH);
+  mkdir (*dirname, S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH | S_IXOTH);
+cleanup:
+  FREE (hex_id.data);
+  if (error != 0)
+    {
+      FREE (*dirname);
+      FREE (*filename);
+      FREE (*temp_filename);
+      FREE (*trash_filename);
+    }
+  else
+    {
+      assert (*dirname != NULL);
+      assert (*filename != NULL);
+      assert (*temp_filename != NULL);
+      assert (*trash_filename != NULL);
+    }
+  return error;
+}
+
+static int
+db_mark_leaf (const char *db_root, const string_desc_t * id)
+{
+  int error = 0;
+  char *dirname = NULL;
+  char *filename = NULL;
+  char *trash_filename = NULL;
+  char *temp_filename = NULL;
+  if (db_file_name
+      (db_root, id->_nbytes, id->_data, &dirname, &filename, &temp_filename,
+       &trash_filename) < 0)
+    {
+      error = -2;
+      goto cleanup;
+    }
+  FILE *f = open_file_for_reading (filename, trash_filename);
+close:
+  if (f != NULL)
+    {
+      fclose (f);
+    }
+cleanup:
+  FREE (dirname);
+  FREE (filename);
+  FREE (temp_filename);
+  FREE (trash_filename);
+  return error;
+}
+
+static int
+db_read (const char *db_root, const string_desc_t * id, string_desc_t * data)
+{
+  int error = 0;
+  assert (data->_nbytes == 0);
+  assert (data->_data == NULL);
+  char *dirname = NULL;
+  char *filename = NULL;
+  char *trash_filename = NULL;
+  char *temp_filename = NULL;
+  if (db_file_name
+      (db_root, id->_nbytes, id->_data, &dirname, &filename, &temp_filename,
+       &trash_filename) < 0)
+    {
+      error = -2;
+      goto cleanup;
+    }
+  FILE *f = open_file_for_reading (filename, trash_filename);
+  if (f == NULL)
+    {
+      error = -1;
+      goto close;
+    }
+  if (fseeko (f, 0, SEEK_END) != 0)
+    {
+      error = -1;
+      goto close;
+    }
+  const size_t length = ftello (f);
+  if (fseeko (f, 0, SEEK_SET) != 0)
+    {
+      error = -1;
+      goto close;
+    }
+  if (ALLOC_N (data->_data, length) < 0)
+    {
+      error = -2;
+      goto close;
+    }
+  data->_nbytes = length;
+  size_t n_read = fread (data->_data, 1, length, f);
+  if (n_read != length || ferror (f))
+    {
+      error = -1;
+      goto close;
+    }
+close:
+  if (f != NULL)
+    {
+      fclose (f);
+    }
+cleanup:
+  FREE (dirname);
+  FREE (filename);
+  FREE (temp_filename);
+  FREE (trash_filename);
+  if (error != 0)
+    {
+      FREE (data->_data);
+      data->_nbytes = 0;
+    }
+  return error;
+}
+
+static FILE *
+open_file_for_writing (const char *filename, const char *dirname,
+		       const char *db_root)
+{
+  assert (filename != NULL);
+  assert (dirname != NULL);
+  assert (db_root != NULL);
+  FILE *f = fopen (filename, "wb");
+  if (f == NULL)
+    {
+      /* Try mkdir and retry. */
+      int error =
+	mkdir (dirname, S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH | S_IXOTH);
+      if (error < 0)
+	{
+	  /* Try also mkdir the root */
+	  error =
+	    mkdir (db_root, S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH | S_IXOTH);
+	  if (error == 0)
+	    {
+	      error =
+		mkdir (dirname,
+		       S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH | S_IXOTH);
+	    }
+	  if (error == 0)
+	    {
+	      return fopen (filename, "wb");
+	    }
+	}
+      return fopen (filename, "wb");
+    }
+  return f;
+}
+
+static int
+db_write (const char *db_root, string_desc_t * id, const string_desc_t * data)
+{
+  assert (id->_nbytes == 0);
+  assert (id->_data == NULL);
+  uint8_t digest[32];
+  const gnutls_digest_algorithm_t algo = GNUTLS_DIG_SHA256;
+  assert (gnutls_hash_get_len (algo) == sizeof (digest));
+  int error =
+    gnutls_hash_fast (GNUTLS_DIG_SHA256, data->_data, data->_nbytes, digest);
+  char *dirname = NULL;
+  char *filename = NULL;
+  char *temp_filename = NULL;
+  char *trash_filename = NULL;
+  if (error == 0)
+    {
+      if (ALLOC_N (id->_data, sizeof (digest)) < 0)
+	{
+	  error = -2;
+	  goto cleanup;
+	}
+      id->_nbytes = sizeof (digest);
+      memcpy (id->_data, digest, sizeof (digest));
+    }
+  if (db_file_name
+      (db_root, id->_nbytes, id->_data, &dirname, &filename, &temp_filename,
+       &trash_filename) < 0)
+    {
+      error = -2;
+      goto cleanup;
+    }
+  FILE *f = NULL;
+  bool temp_filename_unique = false;
+  while (true)
+    {
+      f = open_file_for_reading (filename, trash_filename);
+      if (f == NULL)
+	{
+	  if (!temp_filename_unique)
+	    {
+	      int tmpfd = mkstemp (temp_filename);
+	      if (tmpfd == -1)
+		{
+		  error = -1;
+		  goto cleanup;
+		}
+	      close (tmpfd);
+	    }
+	  f = open_file_for_writing (temp_filename, dirname, db_root);
+	  if (f == NULL)
+	    {
+	      error = -1;
+	      goto close;
+	    }
+	  size_t n_written = fwrite (data->_data, 1, data->_nbytes, f);
+	  if (n_written < data->_nbytes)
+	    {
+	      error = -1;
+	      goto close;
+	    }
+	  if (fflush (f) != 0)
+	    {
+	      error = -1;
+	      goto close;
+	    }
+	  if (rename (temp_filename, filename) < 0)
+	    {
+	      /* The temporary file has been garbage collected before we
+	         could rename it :( Retry */
+	      continue;
+	    }
+	}
+      else
+	{
+	  break;
+	}
+    }
+close:
+  if (error != 0)
+    {
+      remove (temp_filename);
+      FREE (id->_data);
+      id->_nbytes = 0;
+    }
+  if (f != NULL)
+    {
+      fclose (f);
+      /* GCC analyzer still thinks f is allocated here, but no. */
+    }
+cleanup:
+  FREE (dirname);
+  FREE (filename);
+  FREE (temp_filename);
+  FREE (trash_filename);
+  return error;
+}
+
+static int
+db_unmark_entry (const char *dirname, const char *basename)
+{
+  if (!STREQ (basename, ".") && !STREQ (basename, "..")
+      && strstr (basename, "-") == NULL)
+    {
+      /* This a real entry that we need to unmark. */
+      int error = 0;
+      char *abs_filename = NULL;
+      char *trash_filename = NULL;
+      if (ALLOC_N
+	  (abs_filename,
+	   strlen (dirname) + strlen ("/") + strlen (basename) + 1) < 0)
+	{
+	  error = -2;
+	  goto cleanup;
+	}
+      strcpy (abs_filename, dirname);
+      strcat (abs_filename, "/");
+      strcat (abs_filename, basename);
+      if (ALLOC_N
+	  (trash_filename,
+	   strlen (dirname) + strlen ("/") + strlen (basename) +
+	   strlen ("-trash") + 1) < 0)
+	{
+	  error = -2;
+	  goto cleanup;
+	}
+      strcpy (trash_filename, abs_filename);
+      strcat (trash_filename, "-trash");
+      error = rename (abs_filename, trash_filename);
+      if (error != 0)
+	{
+	  error = -1;
+	  goto cleanup;
+	}
+    cleanup:
+      FREE (trash_filename);
+      FREE (abs_filename);
+      return error;
+    }
+  return 0;
+}
+
+static int
+db_unmark_group_entries (const char *dirname, DIR * dir)
+{
+  int error = 0;
+  struct dirent *entry;
+  while ((entry = readdir (dir)) != NULL)
+    {
+      if (!STREQ (entry->d_name, ".") && !STREQ (entry->d_name, ".."))
+	{
+	  int this_error = db_unmark_entry (dirname, entry->d_name);
+	  if (this_error < 0)
+	    {
+	      error = -1;
+	    }
+	}
+    }
+  return error;
+}
+
+static int
+db_unmark_group (const char *dirname, const char *basename)
+{
+  char *absolute_name = NULL;
+  int error = 0;
+  if (ALLOC_N
+      (absolute_name,
+       strlen (dirname) + strlen ("/") + strlen (basename) + 1) < 0)
+    {
+      error = -2;
+      goto cleanup;
+    }
+  strcpy (absolute_name, dirname);
+  strcat (absolute_name, "/");
+  strcat (absolute_name, basename);
+  DIR *group = opendir (absolute_name);
+  if (group == NULL)
+    {
+      error = -1;
+      goto close;
+    }
+  error = db_unmark_group_entries (absolute_name, group);
+close:
+  if (group != NULL)
+    {
+      closedir (group);
+    }
+cleanup:
+  FREE (absolute_name);
+  return error;
+}
+
+static int
+db_unmark_root (const char *root_filename, DIR * root)
+{
+  int error = 0;
+  struct dirent *entry;
+  while ((entry = readdir (root)) != NULL)
+    {
+      if (!STREQ (entry->d_name, ".") && !STREQ (entry->d_name, ".."))
+	{
+	  int this_error = db_unmark_group (root_filename, entry->d_name);
+	  if (this_error < 0)
+	    {
+	      error = -1;
+	    }
+	}
+    }
+  return error;
+}
+
+static int
+db_unmark (const char *db_root)
+{
+  /* Append the -trash suffix to all files. */
+  int error = 0;
+  DIR *root = opendir (db_root);
+  if (root == NULL)
+    {
+      error = -1;
+      goto cleanup;
+    }
+  error = db_unmark_root (db_root, root);
+cleanup:
+  if (root != NULL)
+    {
+      closedir (root);
+    }
+  return error;
+}
+
+static int
+db_collect_entry (const char *dirname, const char *basename)
+{
+  if (!STREQ (basename, ".") && !STREQ (basename, "..")
+      && strstr (basename, "-") != NULL)
+    {
+      /* This a real entry that we need to collect. */
+      int error = 0;
+      char *abs_filename = NULL;
+      if (ALLOC_N
+	  (abs_filename,
+	   strlen (dirname) + strlen ("/") + strlen (basename) + 1) < 0)
+	{
+	  error = -2;
+	  goto cleanup;
+	}
+      strcpy (abs_filename, dirname);
+      strcat (abs_filename, "/");
+      strcat (abs_filename, basename);
+      error = remove (abs_filename);
+      if (error != 0)
+	{
+	  error = -1;
+	  goto cleanup;
+	}
+    cleanup:
+      FREE (abs_filename);
+      return error;
+    }
+  return 0;
+}
+
+static int
+db_collect_group_entries (const char *dirname, DIR * dir)
+{
+  int error = 0;
+  struct dirent *entry;
+  while ((entry = readdir (dir)) != NULL)
+    {
+      if (!STREQ (entry->d_name, ".") && !STREQ (entry->d_name, ".."))
+	{
+	  int this_error = db_collect_entry (dirname, entry->d_name);
+	  if (this_error < 0)
+	    {
+	      error = -1;
+	    }
+	}
+    }
+  return error;
+}
+
+static int
+db_collect_group (const char *dirname, const char *basename)
+{
+  char *absolute_name = NULL;
+  int error = 0;
+  if (ALLOC_N
+      (absolute_name,
+       strlen (dirname) + strlen ("/") + strlen (basename) + 1) < 0)
+    {
+      error = -2;
+      goto cleanup;
+    }
+  strcpy (absolute_name, dirname);
+  strcat (absolute_name, "/");
+  strcat (absolute_name, basename);
+  DIR *group = opendir (absolute_name);
+  if (group == NULL)
+    {
+      error = -1;
+      goto close;
+    }
+  error = db_collect_group_entries (absolute_name, group);
+close:
+  if (group != NULL)
+    {
+      closedir (group);
+    }
+cleanup:
+  FREE (absolute_name);
+  return error;
+}
+
+static int
+db_collect_root (const char *root_filename, DIR * root)
+{
+  int error = 0;
+  struct dirent *entry;
+  while ((entry = readdir (root)) != NULL)
+    {
+      if (!STREQ (entry->d_name, ".") && !STREQ (entry->d_name, ".."))
+	{
+	  int this_error = db_collect_group (root_filename, entry->d_name);
+	  if (this_error < 0)
+	    {
+	      error = -1;
+	    }
+	}
+    }
+  return error;
+}
+
+static int
+db_collect (const char *db_root)
+{
+  /* Delete all files with a dash in name. */
+  int error = 0;
+  DIR *root = opendir (db_root);
+  if (root == NULL)
+    {
+      error = -1;
+      goto cleanup;
+    }
+  error = db_collect_root (db_root, root);
+cleanup:
+  if (root != NULL)
+    {
+      closedir (root);
+    }
+  return error;
+}
+
+static int
+db_read_example_addressbook (const char *db_root, const string_desc_t * id,
+			     size_t *n_people, string_desc_t ** records)
+{
+  /* The example addressbook is a data structure where the first 8
+     bytes are the number of people, big-endian, then each person has
+     a record that consists of: the size of the identifier, on 4
+     bytes, then the identifier. */
+  string_desc_t data = { 0 };
+  *n_people = 0;
+  *records = NULL;
+  int error = db_read (db_root, id, &data);
+  if (error != 0)
+    {
+      goto cleanup;
+    }
+  for (size_t i = 0; i < 8 && i < data._nbytes; i++)
+    {
+      *n_people *= 256;
+      *n_people += (uint8_t) (data._data[i]);
+    }
+  if (ALLOC_N (*records, *n_people) < 0)
+    {
+      error = -2;
+      goto cleanup;
+    }
+  size_t offset = 8;
+  for (size_t i = 0; i < *n_people; i++)
+    {
+      unsigned id_size = 0;
+      for (size_t j = 0; j < 4; j++, offset++)
+	{
+	  id_size *= 256;
+	  id_size += (uint8_t) (data._data[offset]);
+	}
+      (*records)[i]._nbytes = id_size;
+      if (ALLOC_N ((*records)[i]._data, id_size) < 0)
+	{
+	  error = -2;
+	  for (size_t clean = 0; clean < i; clean++)
+	    {
+	      FREE ((*records)[i]._data);
+	    }
+	  goto cleanup;
+	}
+      memcpy ((*records)[i]._data, data._data + offset, id_size);
+      offset += id_size;
+    }
+cleanup:
+  FREE (data._data);
+  if (error != 0)
+    {
+      FREE (*records);
+    }
+  return error;
+}
+
+static int
+db_read_example_person (const char *db_root,
+			const string_desc_t * id,
+			string_desc_t * name, string_desc_t * email)
+{
+  /* An example person file is: 4 bytes for the name id size, the name
+     id, 4 bytes for the email id size, the email id. */
+  name->_nbytes = 0;
+  name->_data = NULL;
+  email->_nbytes = 0;
+  email->_data = NULL;
+  string_desc_t data = { 0 };
+  int error = db_read (db_root, id, &data);
+  if (error != 0)
+    {
+      goto cleanup;
+    }
+  unsigned name_id_size = 0;
+  unsigned email_id_size = 0;
+  for (size_t i = 0; i < 4; i++)
+    {
+      name_id_size *= 256;
+      name_id_size += (uint8_t) (data._data[i]);
+    }
+  if (name_id_size + 8 > data._nbytes)
+    {
+      error = -1;
+      goto cleanup;
+    }
+  for (size_t i = 0; i < 4; i++)
+    {
+      email_id_size *= 256;
+      email_id_size += (uint8_t) (data._data[4 + name_id_size + i]);
+    }
+  if (ALLOC_N (name->_data, name_id_size) < 0
+      || ALLOC_N (email->_data, email_id_size) < 0)
+    {
+      error = -2;
+      goto cleanup;
+    }
+  memcpy (name->_data, data._data + 4, name_id_size);
+  memcpy (email->_data, data._data + 4 + name_id_size + 4, email_id_size);
+  name->_nbytes = name_id_size;
+  email->_nbytes = email_id_size;
+cleanup:
+  FREE (data._data);
+  if (error != 0)
+    {
+      name->_nbytes = 0;
+      email->_nbytes = 0;
+      FREE (name->_data);
+      FREE (email->_data);
+    }
+  return error;
+}
+
+static int
+db_mark_example_person (const char *db_root, const string_desc_t * id)
+{
+  string_desc_t name = { 0 }, email = { 0 };
+  int error = db_read_example_person (db_root, id, &name, &email);
+  if (error != 0)
+    {
+      goto cleanup;
+    }
+  int name_mark_error = db_mark_leaf (db_root, &name);
+  int email_mark_error = db_mark_leaf (db_root, &email);
+  error = error || name_mark_error || email_mark_error;
+cleanup:
+  FREE (name._data);
+  FREE (email._data);
+  return error;
+}
+
+static int
+db_mark_example_addressbook (const char *db_root, const string_desc_t * id)
+{
+  size_t n_people = 0;
+  string_desc_t *people = NULL;
+  int error = db_read_example_addressbook (db_root, id, &n_people, &people);
+  if (error != 0)
+    {
+      goto cleanup;
+    }
+  for (size_t i = 0; i < n_people; i++)
+    {
+      int this_error = db_mark_example_person (db_root, &(people[i]));
+      error = error || this_error;
+    }
+cleanup:
+  if (people != NULL)
+    {
+      for (size_t i = 0; i < n_people; i++)
+	{
+	  FREE (people[i]._data);
+	}
+    }
+  FREE (people);
+  return error;
+}
+
+static int
+db_write_example_person (const char *db_root, string_desc_t * id,
+			 const string_desc_t * name_id,
+			 const string_desc_t * email_id)
+{
+  int error = 0;
+  char *data = NULL;
+  if (ALLOC_N (data, 4 + name_id->_nbytes + 4 + email_id->_nbytes) < 0)
+    {
+      error = -2;
+      goto cleanup;
+    }
+  size_t name_id_length = name_id->_nbytes;
+  size_t email_id_length = name_id->_nbytes;
+  for (size_t i = 4; i-- > 0;)
+    {
+      data[i] = (char) (uint8_t) (name_id_length % 256);
+      name_id_length /= 256;
+    }
+  memcpy (data + 4, name_id->_data, name_id->_nbytes);
+  for (size_t i = 4; i-- > 0;)
+    {
+      data[4 + name_id->_nbytes + i] =
+	(char) (uint8_t) (email_id_length % 256);
+      email_id_length /= 256;
+    }
+  memcpy (data + 4 + name_id->_nbytes + 4, email_id->_data,
+	  email_id->_nbytes);
+  string_desc_t data_desc = {._nbytes =
+      4 + name_id->_nbytes + 4 + email_id->_nbytes,._data = data
+  };
+  error = db_write (db_root, id, &data_desc);
+cleanup:
+  FREE (data);
+  return error;
+}
+
+static int
+db_write_example_addressbook (const char *db_root, string_desc_t * id,
+			      size_t n_people, const string_desc_t ** people)
+{
+  int error = 0;
+  char *data = NULL;
+  size_t data_required = 8;
+  for (size_t i = 0; i < n_people; i++)
+    {
+      data_required += 4 + people[i]->_nbytes;
+    }
+  if (ALLOC_N (data, data_required) < 0)
+    {
+      error = -2;
+      goto cleanup;
+    }
+  size_t n_people_copy = n_people;
+  for (size_t i = 8; i-- > 0;)
+    {
+      data[i] = (char) (uint8_t) (n_people_copy % 256);
+      n_people_copy /= 256;
+    }
+  size_t offset = 8;
+  for (size_t i = 0; i < n_people; i++)
+    {
+      unsigned person_id_size = people[i]->_nbytes;
+      for (size_t i = 4; i-- > 0;)
+	{
+	  data[offset + i] += (char) (uint8_t) (person_id_size % 256);
+	  person_id_size /= 256;
+	}
+      offset += 4;
+      memcpy (data + offset, people[i]->_data, people[i]->_nbytes);
+      offset += people[i]->_nbytes;
+    }
+  string_desc_t data_desc = {._nbytes = data_required,._data = data };
+  error = db_write (db_root, id, &data_desc);
+cleanup:
+  FREE (data);
+  return error;
+}
+
+static int
+db_write_text (const char *db_root, string_desc_t * id, const char *data)
+{
+  string_desc_t bytes = {._nbytes = strlen (data),._data = (char *) data };
+  return db_write (db_root, id, &bytes);
+}
+
+static int
+db_check (void)
+{
+  int error = 0;
+  char *dir;
+  string_desc_t lit_alice = {._nbytes = 0,._data = 0 };
+  string_desc_t lit_bob = {._nbytes = 0,._data = 0 };
+  string_desc_t lit_alice_at_example_dot_com = {._nbytes = 0,._data = 0 };
+  string_desc_t lit_alice_at_example_dot_net = {._nbytes = 0,._data = 0 };
+  string_desc_t lit_bob_at_example_dot_com = {._nbytes = 0,._data = 0 };
+  string_desc_t alice_1_id = {._nbytes = 0,._data = 0 };
+  string_desc_t alice_2_id = {._nbytes = 0,._data = 0 };
+  string_desc_t bob_id = {._nbytes = 0,._data = 0 };
+  string_desc_t v1_id = {._nbytes = 0,._data = 0 };
+  string_desc_t v2_id = {._nbytes = 0,._data = 0 };
+  string_desc_t check_alice_email_1 = {._nbytes = 0,._data = 0 };
+  string_desc_t check_alice_email_2 = {._nbytes = 0,._data = 0 };
+  string_desc_t recheck_alice_email_1 = {._nbytes = 0,._data = 0 };
+  string_desc_t recheck_alice_email_2 = {._nbytes = 0,._data = 0 };
+  if (ALLOC_N (dir, strlen ("/tmp/check-disfluid-db-XXXXXX") + 1) < 0)
+    {
+      error = -2;
+      goto cleanup;
+    }
+  strcpy (dir, "/tmp/check-disfluid-db-XXXXXX");
+  if (mkdtemp (dir) == NULL)
+    {
+      error = -3;
+      goto cleanup;
+    }
+  static const char *_lit_alice = "alice";
+  static const char *_lit_bob = "bob";
+  static const char *_lit_alice_1 = "alice@example.com";
+  static const char *_lit_alice_2 = "alice@example.net";
+  static const char *_lit_bob_email = "bob@example.com";
+  error = db_write_text (dir, &lit_alice, _lit_alice);
+  if (error != 0)
+    {
+      error = -4;
+      goto cleanup;
+    }
+  error = db_write_text (dir, &lit_alice_at_example_dot_com, _lit_alice_1);
+  if (error != 0)
+    {
+      error = -5;
+      goto cleanup;
+    }
+  error = db_write_example_person (dir, &alice_1_id, &lit_alice,
+				   &lit_alice_at_example_dot_com);
+  if (error != 0)
+    {
+      error = -6;
+      goto cleanup;
+    }
+  error = db_write_text (dir, &lit_alice_at_example_dot_net, _lit_alice_2);
+  if (error != 0)
+    {
+      error = -7;
+      goto cleanup;
+    }
+  error = db_write_example_person (dir, &alice_2_id, &lit_alice,
+				   &lit_alice_at_example_dot_net);
+  if (error != 0)
+    {
+      error = -8;
+      goto cleanup;
+    }
+  error = db_write_text (dir, &lit_bob, _lit_bob);
+  if (error != 0)
+    {
+      error = -9;
+      goto cleanup;
+    }
+  error = db_write_text (dir, &lit_bob_at_example_dot_com, _lit_bob_email);
+  if (error != 0)
+    {
+      error = -10;
+      goto cleanup;
+    }
+  error = db_write_example_person (dir, &bob_id, &lit_bob,
+				   &lit_bob_at_example_dot_com);
+  if (error != 0)
+    {
+      error = -11;
+      goto cleanup;
+    }
+  const string_desc_t *people_v1[] = {
+    &alice_1_id, &bob_id
+  };
+  const string_desc_t *people_v2[] = {
+    &alice_2_id, &bob_id
+  };
+  const size_t n_v1 = sizeof (people_v1) / sizeof (people_v1[0]);
+  const size_t n_v2 = sizeof (people_v2) / sizeof (people_v2[0]);
+  error = db_write_example_addressbook (dir, &v1_id, n_v1, people_v1);
+  if (error != 0)
+    {
+      error = -12;
+      goto cleanup;
+    }
+  error = db_write_example_addressbook (dir, &v2_id, n_v2, people_v2);
+  if (error != 0)
+    {
+      error = -13;
+      goto cleanup;
+    }
+  error = db_read (dir, &lit_alice_at_example_dot_com, &check_alice_email_1);
+  if (error != 0)
+    {
+      error = -14;
+      goto cleanup;
+    }
+  error = db_read (dir, &lit_alice_at_example_dot_net, &check_alice_email_2);
+  if (error != 0)
+    {
+      error = -15;
+      goto cleanup;
+    }
+  if (check_alice_email_1._nbytes != strlen ("alice@example.com")
+      || memcmp (check_alice_email_1._data, "alice@example.com",
+		 check_alice_email_1._nbytes) != 0)
+    {
+      error = -16;
+      goto cleanup;
+    }
+  if (check_alice_email_2._nbytes != strlen ("alice@example.net")
+      || memcmp (check_alice_email_2._data, "alice@example.net",
+		 check_alice_email_2._nbytes) != 0)
+    {
+      error = -17;
+      goto cleanup;
+    }
+  /* Now, collect garbage. */
+  error = db_unmark (dir);
+  if (error != 0)
+    {
+      error = -18;
+      goto cleanup;
+    }
+  error = db_mark_example_addressbook (dir, &v2_id);
+  if (error != 0)
+    {
+      error = -19;
+      goto cleanup;
+    }
+  error = db_collect (dir);
+  if (error != 0)
+    {
+      error = -20;
+      goto cleanup;
+    }
+  /* alice@example.net should be readable, but not alice@example.com. */
+  if (db_read (dir, &lit_alice_at_example_dot_com, &recheck_alice_email_1) ==
+      0)
+    {
+      error = -21;
+      goto cleanup;
+    }
+  error =
+    db_read (dir, &lit_alice_at_example_dot_net, &recheck_alice_email_2);
+  if (error != 0)
+    {
+      error = -22;
+      goto cleanup;
+    }
+  if (recheck_alice_email_2._nbytes != strlen ("alice@example.net")
+      || memcmp (recheck_alice_email_2._data, "alice@example.net",
+		 recheck_alice_email_2._nbytes) != 0)
+    {
+      error = -23;
+      goto cleanup;
+    }
+  /* Collect everything */
+  error = db_unmark (dir);
+  if (error != 0)
+    {
+      error = -24;
+      goto cleanup;
+    }
+  error = db_collect (dir);
+  if (error != 0)
+    {
+      error = -25;
+      goto cleanup;
+    }
+  for (size_t i = 0; i < 256; i++)
+    {
+      char *group_name = NULL;
+      if (ALLOC_N (group_name, strlen (dir) + strlen ("/") + 2 + 1) < 0)
+	{
+	  error = -26;
+	  goto cleanup_group_name;
+	}
+      strcpy (group_name, dir);
+      strcat (group_name, "/");
+      static const char *alpha[] = {
+	"0", "1", "2", "3", "4", "5", "6", "7",
+	"8", "9", "a", "b", "c", "d", "e", "f"
+      };
+      strcat (group_name, alpha[i / 16]);
+      strcat (group_name, alpha[i % 16]);
+      rmdir (group_name);
+    cleanup_group_name:
+      FREE (group_name);
+    }
+  rmdir (dir);
+cleanup:
+  FREE (check_alice_email_1._data);
+  FREE (check_alice_email_2._data);
+  FREE (recheck_alice_email_1._data);
+  FREE (recheck_alice_email_2._data);
+  FREE (lit_alice._data);
+  FREE (lit_bob._data);
+  FREE (lit_alice_at_example_dot_com._data);
+  FREE (lit_alice_at_example_dot_net._data);
+  FREE (lit_bob_at_example_dot_com._data);
+  FREE (alice_1_id._data);
+  FREE (alice_2_id._data);
+  FREE (bob_id._data);
+  FREE (v1_id._data);
+  FREE (v2_id._data);
+  FREE (dir);
+  return error;
+}
+
+#endif /* DISFLUID_DISFLUID_DB_INCLUDED */
diff --git a/src/libdisfluid/disfluid-tests.h b/src/libdisfluid/disfluid-tests.h
index df6d534..74b1198 100644
--- a/src/libdisfluid/disfluid-tests.h
+++ b/src/libdisfluid/disfluid-tests.h
@@ -11,6 +11,7 @@ static inline char *run_tests (size_t *n_tests, size_t *n_errors);
 # include "disfluid-version.h"
 # include "disfluid-cache-entry.h"
 # include "disfluid-cache-entry-key.h"
+# include "disfluid-db.h"
 
 # define BYTES * 1
 
@@ -734,6 +735,17 @@ START_TEST (test_cache_hash)
 END_TEST
 /* *INDENT-ON* */
 
+/* *INDENT-OFF* */
+START_TEST (test_db)
+/* *INDENT-ON* */
+
+{
+  ck_assert_int_eq (db_check (), 0);
+}
+/* *INDENT-OFF* */
+END_TEST
+/* *INDENT-ON* */
+
 static inline char *
 tests_read_whole_file (int file)
 {
@@ -800,6 +812,9 @@ run_tests (size_t *n_tests, size_t *n_errors)
   TCase *cache_hash = tcase_create (_("disfluid cache hash key"));
   tcase_add_test (cache_hash, test_cache_hash);
   suite_add_tcase (suite, cache_hash);
+  TCase *db = tcase_create (_("disfluid database"));
+  tcase_add_test (db, test_db);
+  suite_add_tcase (suite, db);
   SRunner *runner = srunner_create (suite);
   char log_file_name[] = "/tmp/disfluid-unit-tests-XXXXXX";
   int log_file = mkstemp (log_file_name);