Merge branch 'consdiffmgr_squashed'

src/common/confline.c

@@ -30,6 +30,24 @@ config_line_append(config_line_t **lst,
   (*lst) = newline;
 }
 
+/** Helper: allocate a new configuration option mapping 'key' to 'val',
+ * and prepend it to *<b>lst</b> */
+void
+config_line_prepend(config_line_t **lst,
+                    const char *key,
+                    const char *val)
+{
+  tor_assert(lst);
+
+  config_line_t *newline;
+
+  newline = tor_malloc_zero(sizeof(config_line_t));
+  newline->key = tor_strdup(key);
+  newline->value = tor_strdup(val);
+  newline->next = *lst;
+  *lst = newline;
+}
+
 /** Return the first line in <b>lines</b> whose key is exactly <b>key</b>, or
  * NULL if no such key exists.
  *

src/common/confline.h

@@ -31,6 +31,8 @@ typedef struct config_line_t {
 
 void config_line_append(config_line_t **lst,
                         const char *key, const char *val);
+void config_line_prepend(config_line_t **lst,
+                         const char *key, const char *val);
 config_line_t *config_lines_dup(const config_line_t *inp);
 config_line_t *config_lines_dup_and_filter(const config_line_t *inp,
                                            const char *key);

src/common/workqueue.h

@@ -16,7 +16,7 @@ typedef struct threadpool_s threadpool_t;
 typedef struct workqueue_entry_s workqueue_entry_t;
 
 /** Possible return value from a work function: */
-typedef enum {
+typedef enum workqueue_reply_t {
   WQ_RPL_REPLY = 0, /** indicates success */
   WQ_RPL_ERROR = 1, /** indicates fatal error */
   WQ_RPL_SHUTDOWN = 2, /** indicates thread is shutting down */

src/or/conscache.c

@@ -16,6 +16,7 @@
  */
 struct consensus_cache_entry_t {
   uint32_t magic; /**< Must be set to CCE_MAGIC */
+  HANDLE_ENTRY(consensus_cache_entry, consensus_cache_entry_t);
   int32_t refcnt; /**< Reference count. */
   unsigned can_remove : 1; /**< If true, we want to delete this file. */
   /** If true, we intend to unmap this file as soon as we're done with it. */
@@ -174,6 +175,8 @@ consensus_cache_find_first(consensus_cache_t *cache,
  * Given a <b>cache</b>, add every entry to <b>out<b> for which
  * <b>key</b>=<b>value</b>.  If <b>key</b> is NULL, add every entry.
  *
+ * Do not add any entry that has been marked for removal.
+ *
  * Does not adjust reference counts.
  */
 void
@@ -182,12 +185,15 @@ consensus_cache_find_all(smartlist_t *out,
                          const char *key,
                          const char *value)
 {
-  if (! key) {
-    smartlist_add_all(out, cache->entries);
-    return;
-  }
-
   SMARTLIST_FOREACH_BEGIN(cache->entries, consensus_cache_entry_t *, ent) {
+    if (ent->can_remove == 1) {
+      /* We want to delete this; pretend it isn't there. */
+      continue;
+    }
+    if (! key) {
+      smartlist_add(out, ent);
+      continue;
+    }
     const char *found_val = consensus_cache_entry_get_value(ent, key);
     if (found_val && !strcmp(value, found_val)) {
       smartlist_add(out, ent);
@@ -300,6 +306,7 @@ consensus_cache_entry_decref(consensus_cache_entry_t *ent)
   }
   tor_free(ent->fname);
   config_free_lines(ent->labels);
+  consensus_cache_entry_handles_clear(ent);
   memwipe(ent, 0, sizeof(consensus_cache_entry_t));
   tor_free(ent);
 }
@@ -485,6 +492,8 @@ consensus_cache_entry_unmap(consensus_cache_entry_t *ent)
   ent->unused_since = TIME_MAX;
 }
 
+HANDLE_IMPL(consensus_cache_entry, consensus_cache_entry_t, )
+
 #ifdef TOR_UNIT_TESTS
 /**
  * Testing only: Return true iff <b>ent</b> is mapped into memory.

src/or/conscache.h

@@ -4,9 +4,13 @@
 #ifndef TOR_CONSCACHE_H
 #define TOR_CONSCACHE_H
 
+#include "handles.h"
+
 typedef struct consensus_cache_entry_t consensus_cache_entry_t;
 typedef struct consensus_cache_t consensus_cache_t;
 
+HANDLE_DECL(consensus_cache_entry, consensus_cache_entry_t, )
+
 consensus_cache_t *consensus_cache_open(const char *subdir, int max_entries);
 void consensus_cache_free(consensus_cache_t *cache);
 void consensus_cache_unmap_lazy(consensus_cache_t *cache, time_t cutoff);

src/or/consdiffmgr.c

@@ -0,0 +1,1116 @@
+/* Copyright (c) 2017, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+/**
+ * \file consdiffmsr.c
+ *
+ * \brief consensus diff manager functions
+ *
+ * This module is run by directory authorities and caches in order
+ * to remember a number of past consensus documents, and to generate
+ * and serve the diffs from those documents to the latest consensus.
+ */
+
+#define CONSDIFFMGR_PRIVATE
+
+#include "or.h"
+#include "conscache.h"
+#include "consdiff.h"
+#include "consdiffmgr.h"
+#include "cpuworker.h"
+#include "networkstatus.h"
+#include "workqueue.h"
+
+/* XXXX support compression */
+
+/**
+ * Labels to apply to items in the conscache object.
+ *
+ * @{
+ */
+/* One of DOCTYPE_CONSENSUS or DOCTYPE_CONSENSUS_DIFF */
+#define LABEL_DOCTYPE "document-type"
+/* The valid-after time for a consensus (or for the target consensus of a
+ * diff), encoded as ISO UTC. */
+#define LABEL_VALID_AFTER "consensus-valid-after"
+/* A hex encoded SHA3 digest of the object after decompression. */
+#define LABEL_SHA3_DIGEST "sha3-digest"
+/* The flavor of the consensus or consensuses diff */
+#define LABEL_FLAVOR "consensus-flavor"
+/* Diff only: the SHA3 digest of the source consensus. */
+#define LABEL_FROM_SHA3_DIGEST "from-sha3-digest"
+/* Diff only: the SHA3 digest of the target consensus. */
+#define LABEL_TARGET_SHA3_DIGEST "target-sha3-digest"
+/* Diff only: the valid-after date of the source consensus. */
+#define LABEL_FROM_VALID_AFTER "from-valid-after"
+/** @} */
+
+#define DOCTYPE_CONSENSUS "consensus"
+#define DOCTYPE_CONSENSUS_DIFF "consensus-diff"
+
+/**
+ * Underlying directory that stores consensuses and consensus diffs.  Don't
+ * use this directly: use cdm_cache_get() instead.
+ */
+static consensus_cache_t *cons_diff_cache = NULL;
+/**
+ * If true, we have learned at least one new consensus since the
+ * consensus cache was last up-to-date.
+ */
+static int cdm_cache_dirty = 0;
+/**
+ * If true, we have scanned the cache to update our hashtable of diffs.
+ */
+static int cdm_cache_loaded = 0;
+
+/**
+ * Possible status values for cdm_diff_t.cdm_diff_status
+ **/
+typedef enum cdm_diff_status_t {
+  CDM_DIFF_PRESENT=1,
+  CDM_DIFF_IN_PROGRESS=2,
+  CDM_DIFF_ERROR=3,
+} cdm_diff_status_t;
+
+/** Hashtable node used to remember the current status of the diff
+ * from a given sha3 digest to the current consensus.  */
+typedef struct cdm_diff_t {
+  HT_ENTRY(cdm_diff_t) node;
+
+  /** Consensus flavor for this diff (part of ht key) */
+  consensus_flavor_t flavor;
+  /** SHA3-256 digest of the consensus that this diff is _from_. (part of the
+   * ht key) */
+  uint8_t from_sha3[DIGEST256_LEN];
+
+  /** One of the CDM_DIFF_* values, depending on whether this diff
+   * is available, in progress, or impossible to compute. */
+  cdm_diff_status_t cdm_diff_status;
+  /** SHA3-256 digest of the consensus that this diff is _to. */
+  uint8_t target_sha3[DIGEST256_LEN];
+  /** Handle to the cache entry for this diff, if any.  We use a handle here
+   * to avoid thinking too hard about cache entry lifetime issues. */
+  consensus_cache_entry_handle_t *entry;
+} cdm_diff_t;
+
+/** Hashtable mapping flavor and source consensus digest to status. */
+static HT_HEAD(cdm_diff_ht, cdm_diff_t) cdm_diff_ht = HT_INITIALIZER();
+
+/**
+ * Configuration for this module
+ */
+static consdiff_cfg_t consdiff_cfg = {
+  /* .cache_max_age_hours = */ 24 * 90,
+  /* .cache_max_num = */ 1440
+};
+
+static int consensus_diff_queue_diff_work(consensus_cache_entry_t *diff_from,
+                                          consensus_cache_entry_t *diff_to);
+static void consdiffmgr_set_cache_flags(void);
+
+/* =====
+ * Hashtable setup
+ * ===== */
+
+/** Helper: hash the key of a cdm_diff_t. */
+static unsigned
+cdm_diff_hash(const cdm_diff_t *diff)
+{
+  uint8_t tmp[DIGEST256_LEN + 1];
+  memcpy(tmp, diff->from_sha3, DIGEST256_LEN);
+  tmp[DIGEST256_LEN] = (uint8_t) diff->flavor;
+  return (unsigned) siphash24g(tmp, sizeof(tmp));
+}
+/** Helper: compare two cdm_diff_t objects for key equality */
+static int
+cdm_diff_eq(const cdm_diff_t *diff1, const cdm_diff_t *diff2)
+{
+  return fast_memeq(diff1->from_sha3, diff2->from_sha3, DIGEST256_LEN) &&
+    diff1->flavor == diff2->flavor;
+}
+
+HT_PROTOTYPE(cdm_diff_ht, cdm_diff_t, node, cdm_diff_hash, cdm_diff_eq)
+HT_GENERATE2(cdm_diff_ht, cdm_diff_t, node, cdm_diff_hash, cdm_diff_eq,
+             0.6, tor_reallocarray, tor_free_)
+
+/** Release all storage held in <b>diff</b>. */
+static void
+cdm_diff_free(cdm_diff_t *diff)
+{
+  if (!diff)
+    return;
+  consensus_cache_entry_handle_free(diff->entry);
+  tor_free(diff);
+}
+
+/** Create and return a new cdm_diff_t with the given values.  Does not
+ * add it to the hashtable. */
+static cdm_diff_t *
+cdm_diff_new(consensus_flavor_t flav,
+             const uint8_t *from_sha3,
+             const uint8_t *target_sha3)
+{
+  cdm_diff_t *ent;
+  ent = tor_malloc_zero(sizeof(cdm_diff_t));
+  ent->flavor = flav;
+  memcpy(ent->from_sha3, from_sha3, DIGEST256_LEN);
+  memcpy(ent->target_sha3, target_sha3, DIGEST256_LEN);
+  return ent;
+}
+
+/**
+ * Examine the diff hashtable to see whether we know anything about computing
+ * a diff of type <b>flav</b> between consensuses with the two provided
+ * SHA3-256 digests.  If a computation is in progress, or if the computation
+ * has already been tried and failed, return 1.  Otherwise, note the
+ * computation as "in progress" so that we don't reattempt it later, and
+ * return 0.
+ */
+static int
+cdm_diff_ht_check_and_note_pending(consensus_flavor_t flav,
+                                   const uint8_t *from_sha3,
+                                   const uint8_t *target_sha3)
+{
+  struct cdm_diff_t search, *ent;
+  memset(&search, 0, sizeof(cdm_diff_t));
+  search.flavor = flav;
+  memcpy(search.from_sha3, from_sha3, DIGEST256_LEN);
+  ent = HT_FIND(cdm_diff_ht, &cdm_diff_ht, &search);
+  if (ent) {
+    tor_assert_nonfatal(ent->cdm_diff_status != CDM_DIFF_PRESENT);
+    return 1;
+  }
+  ent = cdm_diff_new(flav, from_sha3, target_sha3);
+  ent->cdm_diff_status = CDM_DIFF_IN_PROGRESS;
+  HT_INSERT(cdm_diff_ht, &cdm_diff_ht, ent);
+  return 0;
+}
+
+/**
+ * Update the status of the diff of type <b>flav</b> between consensuses with
+ * the two provided SHA3-256 digests, so that its status becomes
+ * <b>status</b>, and its value becomes the <b>handle</b>.  If <b>handle</b>
+ * is NULL, then the old handle (if any) is freed, and replaced with NULL.
+ */
+static void
+cdm_diff_ht_set_status(consensus_flavor_t flav,
+                       const uint8_t *from_sha3,
+                       const uint8_t *to_sha3,
+                       int status,
+                       consensus_cache_entry_handle_t *handle)
+{
+  struct cdm_diff_t search, *ent;
+  memset(&search, 0, sizeof(cdm_diff_t));
+  search.flavor = flav;
+  memcpy(search.from_sha3, from_sha3, DIGEST256_LEN);
+  ent = HT_FIND(cdm_diff_ht, &cdm_diff_ht, &search);
+  if (!ent) {
+    ent = cdm_diff_new(flav, from_sha3, to_sha3);
+    ent->cdm_diff_status = CDM_DIFF_IN_PROGRESS;
+    HT_INSERT(cdm_diff_ht, &cdm_diff_ht, ent);
+  } else if (fast_memneq(ent->target_sha3, to_sha3, DIGEST256_LEN)) {
+    // This can happen under certain really pathological conditions
+    // if we decide we don't care about a diff before it is actually
+    // done computing.
+    return;
+  }
+
+  tor_assert_nonfatal(ent->cdm_diff_status == CDM_DIFF_IN_PROGRESS);
+
+  ent->cdm_diff_status = status;
+  consensus_cache_entry_handle_free(ent->entry);
+  ent->entry = handle;
+}
+
+/**
+ * Helper: Remove from the hash table every present (actually computed) diff
+ * of type <b>flav</b> whose target digest does not match
+ * <b>unless_target_sha3_matches</b>.
+ *
+ * This function is used for the hash table to throw away references to diffs
+ * that do not lead to the most given consensus of a given flavor.
+ */
+static void
+cdm_diff_ht_purge(consensus_flavor_t flav,
+                  const uint8_t *unless_target_sha3_matches)
+{
+  cdm_diff_t **diff, **next;
+  for (diff = HT_START(cdm_diff_ht, &cdm_diff_ht); diff; diff = next) {
+    cdm_diff_t *this = *diff;
+
+    if ((*diff)->cdm_diff_status == CDM_DIFF_PRESENT &&
+        flav == (*diff)->flavor) {
+
+      if (consensus_cache_entry_handle_get((*diff)->entry) == NULL) {
+        /* the underlying entry has gone away; drop this. */
+        next = HT_NEXT_RMV(cdm_diff_ht, &cdm_diff_ht, diff);
+        cdm_diff_free(this);
+        continue;
+      }
+
+      if (unless_target_sha3_matches &&
+          fast_memneq(unless_target_sha3_matches, (*diff)->target_sha3,
+                      DIGEST256_LEN)) {
+        /* target hash doesn't match; drop this. */
+        next = HT_NEXT_RMV(cdm_diff_ht, &cdm_diff_ht, diff);
+        cdm_diff_free(this);
+        continue;
+      }
+    }
+    next = HT_NEXT(cdm_diff_ht, &cdm_diff_ht, diff);
+  }
+}
+
+/**
+ * Helper: initialize <b>cons_diff_cache</b>.
+ */
+static void
+cdm_cache_init(void)
+{
+  unsigned n_entries = consdiff_cfg.cache_max_num * 2;
+
+  tor_assert(cons_diff_cache == NULL);
+  cons_diff_cache = consensus_cache_open("diff-cache", n_entries);
+  if (cons_diff_cache == NULL) {
+    // LCOV_EXCL_START
+    log_err(LD_FS, "Error: Couldn't open storage for consensus diffs.");
+    tor_assert_unreached();
+    // LCOV_EXCL_STOP
+  } else {
+    consdiffmgr_set_cache_flags();
+  }
+  cdm_cache_dirty = 1;
+  cdm_cache_loaded = 0;
+}
+
+/**
+ * Helper: return the consensus_cache_t * that backs this manager,
+ * initializing it if needed.
+ */
+STATIC consensus_cache_t *
+cdm_cache_get(void)
+{
+  if (PREDICT_UNLIKELY(cons_diff_cache == NULL)) {
+    cdm_cache_init();
+  }
+  return cons_diff_cache;
+}
+
+/**
+ * Helper: given a list of labels, prepend the hex-encoded SHA3 digest
+ * of the <b>bodylen</b>-byte object at <b>body</b> to those labels,
+ * with LABEL_SHA3_DIGEST as its label.
+ */
+static void
+cdm_labels_prepend_sha3(config_line_t **labels,
+                        const uint8_t *body,
+                        size_t bodylen)
+{
+  uint8_t sha3_digest[DIGEST256_LEN];
+  char hexdigest[HEX_DIGEST256_LEN+1];
+  crypto_digest256((char *)sha3_digest,
+                   (const char *)body, bodylen, DIGEST_SHA3_256);
+  base16_encode(hexdigest, sizeof(hexdigest),
+                (const char *)sha3_digest, sizeof(sha3_digest));
+
+  config_line_prepend(labels, LABEL_SHA3_DIGEST, hexdigest);
+}
+
+/** Helper: if there is a sha3-256 hex-encoded digest in <b>ent</b> with the
+ * given label, set <b>digest_out</b> to that value (decoded), and return 0.
+ *
+ * Return -1 if there is no such label, and -2 if it is badly formatted. */
+STATIC int
+cdm_entry_get_sha3_value(uint8_t *digest_out,
+                         consensus_cache_entry_t *ent,
+                         const char *label)
+{
+  if (ent == NULL)
+    return -1;
+
+  const char *hex = consensus_cache_entry_get_value(ent, label);
+  if (hex == NULL)
+    return -1;
+
+  int n = base16_decode((char*)digest_out, DIGEST256_LEN, hex, strlen(hex));
+  if (n != DIGEST256_LEN)
+    return -2;
+  else
+    return 0;
+}
+
+/**
+ * Helper: look for a consensus with the given <b>flavor</b> and
+ * <b>valid_after</b> time in the cache. Return that consensus if it's
+ * present, or NULL if it's missing.
+ */
+STATIC consensus_cache_entry_t *
+cdm_cache_lookup_consensus(consensus_flavor_t flavor, time_t valid_after)
+{
+  char formatted_time[ISO_TIME_LEN+1];
+  format_iso_time_nospace(formatted_time, valid_after);
+  const char *flavname = networkstatus_get_flavor_name(flavor);
+
+  /* We'll filter by valid-after time first, since that should
+   * match the fewest documents. */
+  /* We could add an extra hashtable here, but since we only do this scan
+   * when adding a new consensus, it probably doesn't matter much. */
+  smartlist_t *matches = smartlist_new();
+  consensus_cache_find_all(matches, cdm_cache_get(),
+                           LABEL_VALID_AFTER, formatted_time);
+  consensus_cache_filter_list(matches, LABEL_FLAVOR, flavname);
+  consensus_cache_filter_list(matches, LABEL_DOCTYPE, DOCTYPE_CONSENSUS);
+
+  consensus_cache_entry_t *result = NULL;
+  if (smartlist_len(matches) > 1) {
+    log_warn(LD_BUG, "How odd; there appear to be two matching consensuses "
+             "with flavor %s published at %s.",
+             flavname, formatted_time);
+  }
+  if (smartlist_len(matches)) {
+    result = smartlist_get(matches, 0);
+  }
+  smartlist_free(matches);
+
+  return result;
+}
+
+/**
+ * Given a string containing a networkstatus consensus, and the results of
+ * having parsed that consensus, add that consensus to the cache if it is not
+ * already present and not too old.  Create new consensus diffs from or to
+ * that consensus as appropriate.
+ *
+ * Return 0 on success and -1 on failure.
+ */
+int
+consdiffmgr_add_consensus(const char *consensus,
+                          const networkstatus_t *as_parsed)
+{
+  if (BUG(consensus == NULL) || BUG(as_parsed == NULL))
+    return -1; // LCOV_EXCL_LINE
+  if (BUG(as_parsed->type != NS_TYPE_CONSENSUS))
+    return -1; // LCOV_EXCL_LINE
+
+  const consensus_flavor_t flavor = as_parsed->flavor;
+  const time_t valid_after = as_parsed->valid_after;
+
+  if (valid_after < approx_time() - 3600 * consdiff_cfg.cache_max_age_hours) {
+    log_info(LD_DIRSERV, "We don't care about this consensus document; it's "
+             "too old.");
+    return -1;
+  }
+
+  /* Do we already have this one? */
+  consensus_cache_entry_t *entry =
+    cdm_cache_lookup_consensus(flavor, valid_after);
+  if (entry) {
+    log_info(LD_DIRSERV, "We already have a copy of that consensus");
+    return -1;
+  }
+
+  /* We don't have it. Add it to the cache. */
+  {
+    size_t bodylen = strlen(consensus);
+    config_line_t *labels = NULL;
+    char formatted_time[ISO_TIME_LEN+1];
+    format_iso_time_nospace(formatted_time, valid_after);
+    const char *flavname = networkstatus_get_flavor_name(flavor);
+
+    cdm_labels_prepend_sha3(&labels, (const uint8_t *)consensus, bodylen);
+    config_line_prepend(&labels, LABEL_FLAVOR, flavname);
+    config_line_prepend(&labels, LABEL_VALID_AFTER, formatted_time);
+    config_line_prepend(&labels, LABEL_DOCTYPE, DOCTYPE_CONSENSUS);
+
+    entry = consensus_cache_add(cdm_cache_get(),
+                                labels,
+                                (const uint8_t *)consensus,
+                                bodylen);
+    config_free_lines(labels);
+  }
+
+  if (entry) {
+    consensus_cache_entry_mark_for_aggressive_release(entry);
+    consensus_cache_entry_decref(entry);
+  }
+
+  cdm_cache_dirty = 1;
+  return entry ? 0 : -1;
+}
+
+/**
+ * Helper: used to sort two smartlists of consensus_cache_entry_t by their
+ * LABEL_VALID_AFTER labels.
+ */
+static int
+compare_by_valid_after_(const void **a, const void **b)
+{
+  const consensus_cache_entry_t *e1 = *a;
+  const consensus_cache_entry_t *e2 = *b;
+  /* We're in luck here: sorting UTC iso-encoded values lexically will work
+   * fine (until 9999). */
+  return strcmp_opt(consensus_cache_entry_get_value(e1, LABEL_VALID_AFTER),
+                    consensus_cache_entry_get_value(e2, LABEL_VALID_AFTER));
+}
+
+/**
+ * Helper: Sort <b>lst</b> by LABEL_VALID_AFTER and return the most recent
+ * entry.
+ */
+static consensus_cache_entry_t *
+sort_and_find_most_recent(smartlist_t *lst)
+{
+  smartlist_sort(lst, compare_by_valid_after_);
+  if (smartlist_len(lst)) {
+    return smartlist_get(lst, smartlist_len(lst) - 1);
+  } else {
+    return NULL;
+  }
+}
+
+/**
+ * Look up consensus_cache_entry_t for the consensus of type <b>flavor</b>,
+ * from the source consensus with the specified digest (which must be SHA3).
+ *
+ * If the diff is present, store it into *<b>entry_out</b> and return
+ * CONSDIFF_AVAILABLE. Otherwise return CONSDIFF_NOT_FOUND or
+ * CONSDIFF_IN_PROGRESS.
+ */
+consdiff_status_t
+consdiffmgr_find_diff_from(consensus_cache_entry_t **entry_out,
+                           consensus_flavor_t flavor,
+                           int digest_type,
+                           const uint8_t *digest,
+                           size_t digestlen)
+{
+  if (BUG(digest_type != DIGEST_SHA3_256) ||
+      BUG(digestlen != DIGEST256_LEN)) {
+    return CONSDIFF_NOT_FOUND; // LCOV_EXCL_LINE
+  }
+
+  // Try to look up the entry in the hashtable.
+  cdm_diff_t search, *ent;
+  memset(&search, 0, sizeof(search));
+  search.flavor = flavor;
+  memcpy(search.from_sha3, digest, DIGEST256_LEN);
+  ent = HT_FIND(cdm_diff_ht, &cdm_diff_ht, &search);
+
+  if (ent == NULL ||
+      ent->cdm_diff_status == CDM_DIFF_ERROR) {
+    return CONSDIFF_NOT_FOUND;
+  } else if (ent->cdm_diff_status == CDM_DIFF_IN_PROGRESS) {
+    return CONSDIFF_IN_PROGRESS;
+  } else if (BUG(ent->cdm_diff_status != CDM_DIFF_PRESENT)) {
+    return CONSDIFF_IN_PROGRESS;
+  }
+
+  *entry_out = consensus_cache_entry_handle_get(ent->entry);
+  return (*entry_out) ? CONSDIFF_AVAILABLE : CONSDIFF_NOT_FOUND;
+
+#if 0
+  // XXXX Remove this.  I'm keeping it around for now in case we need to
+  // XXXX debug issues in the hashtable.
+  char hex[HEX_DIGEST256_LEN+1];
+  base16_encode(hex, sizeof(hex), (const char *)digest, digestlen);
+  const char *flavname = networkstatus_get_flavor_name(flavor);
+
+  smartlist_t *matches = smartlist_new();
+  consensus_cache_find_all(matches, cdm_cache_get(),
+                           LABEL_FROM_SHA3_DIGEST, hex);
+  consensus_cache_filter_list(matches, LABEL_FLAVOR, flavname);
+  consensus_cache_filter_list(matches, LABEL_DOCTYPE, DOCTYPE_CONSENSUS_DIFF);
+
+  *entry_out = sort_and_find_most_recent(matches);
+  consdiff_status_t result =
+    (*entry_out) ? CONSDIFF_AVAILABLE : CONSDIFF_NOT_FOUND;
+  smartlist_free(matches);
+
+  return result;
+#endif
+}
+
+/**
+ * Perform periodic cleanup tasks on the consensus diff cache.  Return
+ * the number of objects marked for deletion.
+ */
+int
+consdiffmgr_cleanup(void)
+{
+  smartlist_t *objects = smartlist_new();
+  smartlist_t *consensuses = smartlist_new();
+  smartlist_t *diffs = smartlist_new();
+  int n_to_delete = 0;
+
+  log_debug(LD_DIRSERV, "Looking for consdiffmgr entries to remove");
+
+  // 1. Delete any consensus or diff or anything whose valid_after is too old.
+  const time_t valid_after_cutoff =
+    approx_time() - 3600 * consdiff_cfg.cache_max_age_hours;
+
+  consensus_cache_find_all(objects, cdm_cache_get(),
+                           NULL, NULL);
+  SMARTLIST_FOREACH_BEGIN(objects, consensus_cache_entry_t *, ent) {
+    const char *lv_valid_after =
+      consensus_cache_entry_get_value(ent, LABEL_VALID_AFTER);
+    if (! lv_valid_after) {
+      log_debug(LD_DIRSERV, "Ignoring entry because it had no %s label",
+                LABEL_VALID_AFTER);
+      continue;
+    }
+    time_t valid_after = 0;
+    if (parse_iso_time_nospace(lv_valid_after, &valid_after) < 0) {
+      log_debug(LD_DIRSERV, "Ignoring entry because its %s value (%s) was "
+                "unparseable", LABEL_VALID_AFTER, escaped(lv_valid_after));
+      continue;
+    }
+    if (valid_after < valid_after_cutoff) {
+      log_debug(LD_DIRSERV, "Deleting entry because its %s value (%s) was "
+                "too old", LABEL_VALID_AFTER, lv_valid_after);
+      consensus_cache_entry_mark_for_removal(ent);
+      ++n_to_delete;
+    }
+  } SMARTLIST_FOREACH_END(ent);
+
+  // 2. Delete all diffs that lead to a consensus whose valid-after is not the
+  // latest.
+  for (int flav = 0; flav < N_CONSENSUS_FLAVORS; ++flav) {
+    const char *flavname = networkstatus_get_flavor_name(flav);
+    /* Determine the most recent consensus of this flavor */
+    consensus_cache_find_all(consensuses, cdm_cache_get(),
+                             LABEL_DOCTYPE, DOCTYPE_CONSENSUS);
+    consensus_cache_filter_list(consensuses, LABEL_FLAVOR, flavname);
+    consensus_cache_entry_t *most_recent =
+      sort_and_find_most_recent(consensuses);
+    if (most_recent == NULL)
+      continue;
+    const char *most_recent_sha3 =
+      consensus_cache_entry_get_value(most_recent, LABEL_SHA3_DIGEST);
+    if (BUG(most_recent_sha3 == NULL))
+      continue; // LCOV_EXCL_LINE
+
+    /* consider all such-flavored diffs, and look to see if they match. */
+    consensus_cache_find_all(diffs, cdm_cache_get(),
+                             LABEL_DOCTYPE, DOCTYPE_CONSENSUS_DIFF);
+    consensus_cache_filter_list(diffs, LABEL_FLAVOR, flavname);
+    SMARTLIST_FOREACH_BEGIN(diffs, consensus_cache_entry_t *, diff) {
+      const char *this_diff_target_sha3 =
+        consensus_cache_entry_get_value(diff, LABEL_TARGET_SHA3_DIGEST);
+      if (!this_diff_target_sha3)
+        continue;
+      if (strcmp(this_diff_target_sha3, most_recent_sha3)) {
+        consensus_cache_entry_mark_for_removal(diff);
+        ++n_to_delete;
+      }
+    } SMARTLIST_FOREACH_END(diff);
+    smartlist_clear(consensuses);
+    smartlist_clear(diffs);
+  }
+
+  smartlist_free(objects);
+  smartlist_free(consensuses);
+  smartlist_free(diffs);
+
+  // Actually remove files, if they're not used.
+  consensus_cache_delete_pending(cdm_cache_get());
+  return n_to_delete;
+}
+
+/**
+ * Initialize the consensus diff manager and its cache, and configure
+ * its parameters based on the latest torrc and networkstatus parameters.
+ */
+void
+consdiffmgr_configure(const consdiff_cfg_t *cfg)
+{
+  memcpy(&consdiff_cfg, cfg, sizeof(consdiff_cfg));
+
+  (void) cdm_cache_get();
+}
+
+/**
+ * Scan the consensus diff manager's cache for any grossly malformed entries,
+ * and mark them as deletable.  Return 0 if no problems were found; 1
+ * if problems were found and fixed.
+ */
+int
+consdiffmgr_validate(void)
+{
+  /* Right now, we only check for entries that have bad sha3 values */
+  int problems = 0;
+
+  smartlist_t *objects = smartlist_new();
+  consensus_cache_find_all(objects, cdm_cache_get(),
+                           NULL, NULL);
+  SMARTLIST_FOREACH_BEGIN(objects, consensus_cache_entry_t *, obj) {
+    uint8_t sha3_expected[DIGEST256_LEN];
+    uint8_t sha3_received[DIGEST256_LEN];
+    int r = cdm_entry_get_sha3_value(sha3_expected, obj, LABEL_SHA3_DIGEST);
+    if (r == -1) {
+      /* digest isn't there; that's allowed */
+      continue;
+    } else if (r == -2) {
+      /* digest is malformed; that's not allowed */
+      problems = 1;
+      consensus_cache_entry_mark_for_removal(obj);
+      continue;
+    }
+    const uint8_t *body;
+    size_t bodylen;
+    consensus_cache_entry_incref(obj);
+    r = consensus_cache_entry_get_body(obj, &body, &bodylen);
+    if (r == 0) {
+      crypto_digest256((char *)sha3_received, (const char *)body, bodylen,
+                       DIGEST_SHA3_256);
+    }
+    consensus_cache_entry_decref(obj);
+    if (r < 0)
+      continue;
+
+    if (fast_memneq(sha3_received, sha3_expected, DIGEST256_LEN)) {
+      problems = 1;
+      consensus_cache_entry_mark_for_removal(obj);
+      continue;
+    }
+
+  } SMARTLIST_FOREACH_END(obj);
+  smartlist_free(objects);
+  return problems;
+}
+
+/**
+ * Helper: build new diffs of <b>flavor</b> as needed
+ */
+static void
+consdiffmgr_rescan_flavor_(consensus_flavor_t flavor)
+{
+  smartlist_t *matches = NULL;
+  smartlist_t *diffs = NULL;
+  smartlist_t *compute_diffs_from = NULL;
+  strmap_t *have_diff_from = NULL;
+
+  // look for the most recent consensus, and for all previous in-range
+  // consensuses.  Do they all have diffs to it?
+  const char *flavname = networkstatus_get_flavor_name(flavor);
+
+  // 1. find the most recent consensus, and the ones that we might want
+  //    to diff to it.
+  matches = smartlist_new();
+  consensus_cache_find_all(matches, cdm_cache_get(),
+                           LABEL_FLAVOR, flavname);
+  consensus_cache_filter_list(matches, LABEL_DOCTYPE, DOCTYPE_CONSENSUS);
+  consensus_cache_entry_t *most_recent = sort_and_find_most_recent(matches);
+  if (!most_recent) {
+    log_info(LD_DIRSERV, "No 'most recent' %s consensus found; "
+             "not making diffs", flavname);
+    goto done;
+  }
+  tor_assert(smartlist_len(matches));
+  smartlist_del(matches, smartlist_len(matches) - 1);
+
+  const char *most_recent_valid_after =
+    consensus_cache_entry_get_value(most_recent, LABEL_VALID_AFTER);
+  if (BUG(most_recent_valid_after == NULL))
+    goto done; //LCOV_EXCL_LINE
+  uint8_t most_recent_sha3[DIGEST256_LEN];
+  if (BUG(cdm_entry_get_sha3_value(most_recent_sha3, most_recent,
+                                   LABEL_SHA3_DIGEST) < 0))
+    goto done; //LCOV_EXCL_LINE
+
+  // 2. Find all the relevant diffs _to_ this consensus. These are ones
+  //    that we don't need to compute.
+  diffs = smartlist_new();
+  consensus_cache_find_all(diffs, cdm_cache_get(),
+                           LABEL_VALID_AFTER, most_recent_valid_after);
+  consensus_cache_filter_list(diffs, LABEL_DOCTYPE, DOCTYPE_CONSENSUS_DIFF);
+  consensus_cache_filter_list(diffs, LABEL_FLAVOR, flavname);
+  have_diff_from = strmap_new();
+  SMARTLIST_FOREACH_BEGIN(diffs, consensus_cache_entry_t *, diff) {
+    const char *va = consensus_cache_entry_get_value(diff,
+                                                     LABEL_FROM_VALID_AFTER);
+    if (BUG(va == NULL))
+      continue; // LCOV_EXCL_LINE
+    strmap_set(have_diff_from, va, diff);
+  } SMARTLIST_FOREACH_END(diff);
+
+  // 3. See which consensuses in 'matches' don't have diffs yet.
+  smartlist_reverse(matches); // from newest to oldest.
+  compute_diffs_from = smartlist_new();
+  SMARTLIST_FOREACH_BEGIN(matches, consensus_cache_entry_t *, ent) {
+    const char *va = consensus_cache_entry_get_value(ent, LABEL_VALID_AFTER);
+    if (BUG(va == NULL))
+      continue; // LCOV_EXCL_LINE
+    if (strmap_get(have_diff_from, va) != NULL)
+      continue; /* we already have this one. */
+    smartlist_add(compute_diffs_from, ent);
+  } SMARTLIST_FOREACH_END(ent);
+
+  log_info(LD_DIRSERV,
+           "The most recent %s consensus is valid-after %s. We have diffs to "
+           "this consensus for %d/%d older %s consensuses. Generating diffs "
+           "for the other %d.",
+           flavname,
+           most_recent_valid_after,
+           smartlist_len(matches) - smartlist_len(compute_diffs_from),
+           smartlist_len(matches),
+           flavname,
+           smartlist_len(compute_diffs_from));
+
+  // 4. Update the hashtable; remove entries in this flavor to other
+  //    target consensuses.
+  cdm_diff_ht_purge(flavor, most_recent_sha3);
+
+  // 5. Actually launch the requests.
+  SMARTLIST_FOREACH_BEGIN(compute_diffs_from, consensus_cache_entry_t *, c) {
+    if (BUG(c == most_recent))
+      continue; // LCOV_EXCL_LINE
+
+    uint8_t this_sha3[DIGEST256_LEN];
+    if (BUG(cdm_entry_get_sha3_value(this_sha3, c, LABEL_SHA3_DIGEST)<0))
+      continue; // LCOV_EXCL_LINE
+    if (cdm_diff_ht_check_and_note_pending(flavor,
+                                           this_sha3, most_recent_sha3)) {
+      // This is already pending, or we encountered an error.
+      continue;
+    }
+    consensus_diff_queue_diff_work(c, most_recent);
+  } SMARTLIST_FOREACH_END(c);
+
+ done:
+  smartlist_free(matches);
+  smartlist_free(diffs);
+  smartlist_free(compute_diffs_from);
+  strmap_free(have_diff_from, NULL);
+}
+
+/**
+ * Scan the cache for diffs, and add them to the hashtable.
+ */
+static void
+consdiffmgr_diffs_load(void)
+{
+  smartlist_t *diffs = smartlist_new();
+  consensus_cache_find_all(diffs, cdm_cache_get(),
+                           LABEL_DOCTYPE, DOCTYPE_CONSENSUS_DIFF);
+  SMARTLIST_FOREACH_BEGIN(diffs, consensus_cache_entry_t *, diff) {
+    const char *lv_flavor =
+      consensus_cache_entry_get_value(diff, LABEL_FLAVOR);
+    if (!lv_flavor)
+      continue;
+    int flavor = networkstatus_parse_flavor_name(lv_flavor);
+    if (flavor < 0)
+      continue;
+    uint8_t from_sha3[DIGEST256_LEN];
+    uint8_t to_sha3[DIGEST256_LEN];
+    if (cdm_entry_get_sha3_value(from_sha3, diff, LABEL_FROM_SHA3_DIGEST)<0)
+      continue;
+    if (cdm_entry_get_sha3_value(to_sha3, diff, LABEL_TARGET_SHA3_DIGEST)<0)
+      continue;
+
+    cdm_diff_ht_set_status(flavor, from_sha3, to_sha3,
+                           CDM_DIFF_PRESENT,
+                           consensus_cache_entry_handle_new(diff));
+  } SMARTLIST_FOREACH_END(diff);
+  smartlist_free(diffs);
+}
+
+/**
+ * Build new diffs as needed.
+ */
+void
+consdiffmgr_rescan(void)
+{
+  if (cdm_cache_dirty == 0)
+    return;
+
+  // Clean up here to make room for new diffs, and to ensure that older
+  // consensuses do not have any entries.
+  consdiffmgr_cleanup();
+
+  if (cdm_cache_loaded == 0) {
+    consdiffmgr_diffs_load();
+    cdm_cache_loaded = 1;
+  }
+
+  for (int flav = 0; flav < N_CONSENSUS_FLAVORS; ++flav) {
+    consdiffmgr_rescan_flavor_((consensus_flavor_t) flav);
+  }
+
+  cdm_cache_dirty = 0;
+}
+
+/**
+ * Set consensus cache flags on the objects in this consdiffmgr.
+ */
+static void
+consdiffmgr_set_cache_flags(void)
+{
+  /* Right now, we just mark the consensus objects for aggressive release,
+   * so that they get mmapped for as little time as possible. */
+  smartlist_t *objects = smartlist_new();
+  consensus_cache_find_all(objects, cdm_cache_get(), LABEL_DOCTYPE,
+                           DOCTYPE_CONSENSUS);
+  SMARTLIST_FOREACH_BEGIN(objects, consensus_cache_entry_t *, ent) {
+    consensus_cache_entry_mark_for_aggressive_release(ent);
+  } SMARTLIST_FOREACH_END(ent);
+  smartlist_free(objects);
+}
+
+/**
+ * Called before shutdown: drop all storage held by the consdiffmgr.c module.
+ */
+void
+consdiffmgr_free_all(void)
+{
+  cdm_diff_t **diff, **next;
+  for (diff = HT_START(cdm_diff_ht, &cdm_diff_ht); diff; diff = next) {
+    cdm_diff_t *this = *diff;
+    next = HT_NEXT_RMV(cdm_diff_ht, &cdm_diff_ht, diff);
+    cdm_diff_free(this);
+  }
+  consensus_cache_free(cons_diff_cache);
+  cons_diff_cache = NULL;
+}
+
+/* =====
+   Thread workers
+   =====*/
+
+/**
+ * An object passed to a worker thread that will try to produce a consensus
+ * diff.
+ */
+typedef struct consensus_diff_worker_job_t {
+  /**
+   * Input: The consensus to compute the diff from.  Holds a reference to the
+   * cache entry, which must not be released until the job is passed back to
+   * the main thread. The body must be mapped into memory in the main thread.
+   */
+  consensus_cache_entry_t *diff_from;
+  /**
+   * Input: The consensus to compute the diff to.  Holds a reference to the
+   * cache entry, which must not be released until the job is passed back to
+   * the main thread. The body must be mapped into memory in the main thread.
+   */
+  consensus_cache_entry_t *diff_to;
+
+  /**
+   * Output: Labels to store in the cache associated with this diff.
+   */
+  config_line_t *labels_out;
+  /**
+   * Output: Body of the diff
+   */
+  uint8_t *body_out;
+  /**
+   * Output: length of body_out
+   */
+  size_t bodylen_out;
+} consensus_diff_worker_job_t;
+
+/**
+ * Worker function. This function runs inside a worker thread and receives
+ * a consensus_diff_worker_job_t as its input.
+ */
+static workqueue_reply_t
+consensus_diff_worker_threadfn(void *state_, void *work_)
+{
+  (void)state_;
+  consensus_diff_worker_job_t *job = work_;
+  const uint8_t *diff_from, *diff_to;
+  size_t len_from, len_to;
+  int r;
+  /* We need to have the body already mapped into RAM here.
+   */
+  r = consensus_cache_entry_get_body(job->diff_from, &diff_from, &len_from);
+  if (BUG(r < 0))
+    return WQ_RPL_REPLY; // LCOV_EXCL_LINE
+  r = consensus_cache_entry_get_body(job->diff_to, &diff_to, &len_to);
+  if (BUG(r < 0))
+    return WQ_RPL_REPLY; // LCOV_EXCL_LINE
+
+  const char *lv_to_valid_after =
+    consensus_cache_entry_get_value(job->diff_to, LABEL_VALID_AFTER);
+  const char *lv_from_valid_after =
+    consensus_cache_entry_get_value(job->diff_from, LABEL_VALID_AFTER);
+  const char *lv_from_digest =
+    consensus_cache_entry_get_value(job->diff_from, LABEL_SHA3_DIGEST);
+  const char *lv_from_flavor =
+    consensus_cache_entry_get_value(job->diff_from, LABEL_FLAVOR);
+  const char *lv_to_flavor =
+    consensus_cache_entry_get_value(job->diff_to, LABEL_FLAVOR);
+  const char *lv_to_digest =
+    consensus_cache_entry_get_value(job->diff_to, LABEL_SHA3_DIGEST);
+
+  /* All these values are mandatory on the input */
+  if (BUG(!lv_to_valid_after) ||
+      BUG(!lv_from_valid_after) ||
+      BUG(!lv_from_digest) ||
+      BUG(!lv_from_flavor) ||
+      BUG(!lv_to_flavor)) {
+    return WQ_RPL_REPLY; // LCOV_EXCL_LINE
+  }
+  /* The flavors need to match */
+  if (BUG(strcmp(lv_from_flavor, lv_to_flavor))) {
+    return WQ_RPL_REPLY; // LCOV_EXCL_LINE
+  }
+
+  char *consensus_diff;
+  {
+    // XXXX the input might not be nul-terminated. And also we wanted to
+    // XXXX support compression later I guess. So, we need to copy here.
+    char *diff_from_nt, *diff_to_nt;
+    diff_from_nt = tor_memdup_nulterm(diff_from, len_from);
+    diff_to_nt = tor_memdup_nulterm(diff_to, len_to);
+
+    // XXXX ugh; this is going to calculate the SHA3 of both its
+    // XXXX inputs again, even though we already have that. Maybe it's time
+    // XXXX to change the API here?
+    consensus_diff = consensus_diff_generate(diff_from_nt, diff_to_nt);
+    tor_free(diff_from_nt);
+    tor_free(diff_to_nt);
+  }
+  if (!consensus_diff) {
+    /* Couldn't generate consensus; we'll leave the reply blank. */
+    return WQ_RPL_REPLY;
+  }
+
+  /* Send the reply */
+  job->body_out = (uint8_t *) consensus_diff;
+  job->bodylen_out = strlen(consensus_diff);
+
+  cdm_labels_prepend_sha3(&job->labels_out, job->body_out, job->bodylen_out);
+  config_line_prepend(&job->labels_out, LABEL_FROM_VALID_AFTER,
+                      lv_from_valid_after);
+  config_line_prepend(&job->labels_out, LABEL_VALID_AFTER, lv_to_valid_after);
+  config_line_prepend(&job->labels_out, LABEL_FLAVOR, lv_from_flavor);
+  config_line_prepend(&job->labels_out, LABEL_FROM_SHA3_DIGEST,
+                      lv_from_digest);
+  config_line_prepend(&job->labels_out, LABEL_TARGET_SHA3_DIGEST,
+                      lv_to_digest);
+  config_line_prepend(&job->labels_out, LABEL_DOCTYPE, DOCTYPE_CONSENSUS_DIFF);
+  return WQ_RPL_REPLY;
+}
+
+/**
+ * Helper: release all storage held in <b>job</b>.
+ */
+static void
+consensus_diff_worker_job_free(consensus_diff_worker_job_t *job)
+{
+  if (!job)
+    return;
+  tor_free(job->body_out);
+  config_free_lines(job->labels_out);
+  consensus_cache_entry_decref(job->diff_from);
+  consensus_cache_entry_decref(job->diff_to);
+  tor_free(job);
+}
+
+/**
+ * Worker function: This function runs in the main thread, and receives
+ * a consensus_diff_worker_job_t that the worker thread has already
+ * processed.
+ */
+static void
+consensus_diff_worker_replyfn(void *work_)
+{
+  tor_assert(in_main_thread());
+  tor_assert(work_);
+
+  consensus_diff_worker_job_t *job = work_;
+
+  const char *lv_from_digest =
+    consensus_cache_entry_get_value(job->diff_from, LABEL_SHA3_DIGEST);
+  const char *lv_to_digest =
+    consensus_cache_entry_get_value(job->diff_to, LABEL_SHA3_DIGEST);
+  const char *lv_flavor =
+    consensus_cache_entry_get_value(job->diff_to, LABEL_FLAVOR);
+  if (BUG(lv_from_digest == NULL))
+    lv_from_digest = "???"; // LCOV_EXCL_LINE
+  if (BUG(lv_to_digest == NULL))
+    lv_to_digest = "???"; // LCOV_EXCL_LINE
+
+  uint8_t from_sha3[DIGEST256_LEN];
+  uint8_t to_sha3[DIGEST256_LEN];
+  int flav = -1;
+  int cache = 1;
+  if (BUG(cdm_entry_get_sha3_value(from_sha3, job->diff_from,
+                                   LABEL_SHA3_DIGEST) < 0))
+    cache = 0;
+  if (BUG(cdm_entry_get_sha3_value(to_sha3, job->diff_to,
+                                   LABEL_SHA3_DIGEST) < 0))
+    cache = 0;
+  if (BUG(lv_flavor == NULL)) {
+    cache = 0;
+  } else if ((flav = networkstatus_parse_flavor_name(lv_flavor)) < 0) {
+    cache = 0;
+  }
+
+  int status;
+  consensus_cache_entry_handle_t *handle = NULL;
+  if (job->body_out && job->bodylen_out && job->labels_out) {
+    /* Success! Store the results */
+    log_info(LD_DIRSERV, "Adding consensus diff from %s to %s",
+             lv_from_digest, lv_to_digest);
+    consensus_cache_entry_t *ent =
+      consensus_cache_add(cdm_cache_get(), job->labels_out,
+                          job->body_out,
+                          job->bodylen_out);
+    status = CDM_DIFF_PRESENT;
+    handle = consensus_cache_entry_handle_new(ent);
+    consensus_cache_entry_decref(ent);
+  } else {
+    /* Failure! Nothing to do but complain */
+    log_warn(LD_DIRSERV,
+             "Worker was unable to compute consensus diff "
+             "from %s to %s", lv_from_digest, lv_to_digest);
+    /* Cache this error so we don't try to compute this one again. */
+    status = CDM_DIFF_ERROR;
+  }
+
+  if (cache)
+    cdm_diff_ht_set_status(flav, from_sha3, to_sha3, status, handle);
+  else
+    consensus_cache_entry_handle_free(handle);
+
+  consensus_diff_worker_job_free(job);
+}
+
+/**
+ * Queue the job of computing the diff from <b>diff_from</b> to <b>diff_to</b>
+ * in a worker thread.
+ */
+static int
+consensus_diff_queue_diff_work(consensus_cache_entry_t *diff_from,
+                               consensus_cache_entry_t *diff_to)
+{
+  tor_assert(in_main_thread());
+
+  consensus_cache_entry_incref(diff_from);
+  consensus_cache_entry_incref(diff_to);
+
+  consensus_diff_worker_job_t *job = tor_malloc_zero(sizeof(*job));
+  job->diff_from = diff_from;
+  job->diff_to = diff_to;
+
+  /* Make sure body is mapped. */
+  const uint8_t *body;
+  size_t bodylen;
+  int r1 = consensus_cache_entry_get_body(diff_from, &body, &bodylen);
+  int r2 = consensus_cache_entry_get_body(diff_to, &body, &bodylen);
+  if (r1 < 0 || r2 < 0)
+    goto err;
+
+  workqueue_entry_t *work;
+  work = cpuworker_queue_work(consensus_diff_worker_threadfn,
+                              consensus_diff_worker_replyfn,
+                              job);
+  if (!work)
+    goto err;
+
+  return 0;
+ err:
+  consensus_diff_worker_job_free(job); // includes decrefs.
+  return -1;
+}
+

src/or/consdiffmgr.h

@@ -0,0 +1,48 @@
+/* Copyright (c) 2017, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+#ifndef TOR_CONSDIFFMGR_H
+#define TOR_CONSDIFFMGR_H
+
+/**
+ * Possible outcomes from trying to look up a given consensus diff.
+ */
+typedef enum consdiff_status_t {
+  CONSDIFF_AVAILABLE,
+  CONSDIFF_NOT_FOUND,
+  CONSDIFF_IN_PROGRESS,
+} consdiff_status_t;
+
+typedef struct consdiff_cfg_t {
+  uint32_t cache_max_age_hours;
+  uint32_t cache_max_num;
+} consdiff_cfg_t;
+
+struct consensus_cache_entry_t; // from conscache.h
+
+int consdiffmgr_add_consensus(const char *consensus,
+                              const networkstatus_t *as_parsed);
+
+consdiff_status_t consdiffmgr_find_diff_from(
+                           struct consensus_cache_entry_t **entry_out,
+                           consensus_flavor_t flavor,
+                           int digest_type,
+                           const uint8_t *digest,
+                           size_t digestlen);
+void consdiffmgr_rescan(void);
+int consdiffmgr_cleanup(void);
+void consdiffmgr_configure(const consdiff_cfg_t *cfg);
+void consdiffmgr_free_all(void);
+int consdiffmgr_validate(void);
+
+#ifdef CONSDIFFMGR_PRIVATE
+STATIC consensus_cache_t *cdm_cache_get(void);
+STATIC consensus_cache_entry_t *cdm_cache_lookup_consensus(
+                          consensus_flavor_t flavor, time_t valid_after);
+STATIC int cdm_entry_get_sha3_value(uint8_t *digest_out,
+                                    consensus_cache_entry_t *ent,
+                                    const char *label);
+#endif
+
+#endif
+

src/or/cpuworker.c

@@ -479,6 +479,20 @@ queue_pending_tasks(void)
   }
 }
 
+/** DOCDOC */
+MOCK_IMPL(workqueue_entry_t *,
+cpuworker_queue_work,(workqueue_reply_t (*fn)(void *, void *),
+                      void (*reply_fn)(void *),
+                      void *arg))
+{
+  tor_assert(threadpool);
+
+  return threadpool_queue_work(threadpool,
+                               fn,
+                               reply_fn,
+                               arg);
+}
+
 /** Try to tell a cpuworker to perform the public key operations necessary to
  * respond to <b>onionskin</b> for the circuit <b>circ</b>.
  *

src/or/cpuworker.h

@@ -14,6 +14,12 @@
 
 void cpu_init(void);
 void cpuworkers_rotate_keyinfo(void);
+struct workqueue_entry_s;
+enum workqueue_reply_t;
+MOCK_DECL(struct workqueue_entry_s *, cpuworker_queue_work, (
+                    enum workqueue_reply_t (*fn)(void *, void *),
+                    void (*reply_fn)(void *),
+                    void *arg));
 
 struct create_cell_t;
 int assign_onionskin_to_cpuworker(or_circuit_t *circ,

src/or/include.am

@@ -38,6 +38,7 @@ LIBTOR_A_SOURCES = \
 	src/or/connection_or.c				\
 	src/or/conscache.c				\
 	src/or/consdiff.c				\
+	src/or/consdiffmgr.c				\
 	src/or/control.c				\
 	src/or/cpuworker.c				\
 	src/or/dircollate.c				\
@@ -156,6 +157,7 @@ ORHEADERS = \
 	src/or/connection_or.h				\
 	src/or/conscache.h				\
 	src/or/consdiff.h				\
+	src/or/consdiffmgr.h				\
 	src/or/control.h				\
 	src/or/cpuworker.h				\
 	src/or/dircollate.h				\

src/test/include.am

@@ -89,6 +89,7 @@ src_test_test_SOURCES = \
 	src/test/test_connection.c \
 	src/test/test_conscache.c \
 	src/test/test_consdiff.c \
+	src/test/test_consdiffmgr.c \
 	src/test/test_containers.c \
 	src/test/test_controller.c \
 	src/test/test_controller_events.c \

src/test/test.c

@@ -1197,6 +1197,7 @@ struct testgroup_t testgroups[] = {
   { "connection/", connection_tests },
   { "conscache/", conscache_tests },
   { "consdiff/", consdiff_tests },
+  { "consdiffmgr/", consdiffmgr_tests },
   { "container/", container_tests },
   { "control/", controller_tests },
   { "control/event/", controller_event_tests },

src/test/test.h

@@ -192,6 +192,7 @@ extern struct testcase_t config_tests[];
 extern struct testcase_t connection_tests[];
 extern struct testcase_t conscache_tests[];
 extern struct testcase_t consdiff_tests[];
+extern struct testcase_t consdiffmgr_tests[];
 extern struct testcase_t container_tests[];
 extern struct testcase_t controller_tests[];
 extern struct testcase_t controller_event_tests[];

src/test/test_conscache.c

@@ -200,8 +200,7 @@ test_conscache_cleanup(void *arg)
   tt_assert(e_tmp);
   tt_assert(consensus_cache_entry_is_mapped(e_tmp));
   e_tmp = consensus_cache_find_first(cache, "index", "7");
-  tt_assert(e_tmp);
-  tt_assert(consensus_cache_entry_is_mapped(e_tmp));
+  tt_assert(e_tmp == NULL); // not found because pending deletion.
 
   /* Delete the pending-deletion items. */
   consensus_cache_delete_pending(cache);
@@ -210,12 +209,12 @@ test_conscache_cleanup(void *arg)
     consensus_cache_find_all(entries, cache, NULL, NULL);
     int n = smartlist_len(entries);
     smartlist_free(entries);
-    tt_int_op(n, OP_EQ, 20 - 1); /* 1 entry was deleted */
+    tt_int_op(n, OP_EQ, 20 - 2); /* 1 entry was deleted; 1 is not-found. */
   }
   e_tmp = consensus_cache_find_first(cache, "index", "7"); // refcnt == 1...
   tt_assert(e_tmp == NULL); // so deleted.
   e_tmp = consensus_cache_find_first(cache, "index", "14"); // refcnt == 2
-  tt_assert(e_tmp); // so, not deleted.
+  tt_assert(e_tmp == NULL); // not deleted; but not found.
 
   /* Now do lazy unmapping. */
   // should do nothing.

src/test/test_consdiffmgr.c

@@ -0,0 +1,883 @@
+/* Copyright (c) 2017, The Tor Project, Inc. */
+/* See LICENSE for licensing information */
+
+#define CONSDIFFMGR_PRIVATE
+
+#include "or.h"
+#include "config.h"
+#include "conscache.h"
+#include "consdiff.h"
+#include "consdiffmgr.h"
+#include "cpuworker.h"
+#include "networkstatus.h"
+#include "workqueue.h"
+
+#include "test.h"
+#include "log_test_helpers.h"
+
+// ============================== Setup/teardown the consdiffmgr
+// These functions get run before/after each test in this module
+
+static void *
+consdiffmgr_test_setup(const struct testcase_t *arg)
+{
+  (void)arg;
+  char *ddir_fname = tor_strdup(get_fname_rnd("datadir_cdm"));
+  tor_free(get_options_mutable()->DataDirectory);
+  get_options_mutable()->DataDirectory = ddir_fname; // now owns the pointer.
+  check_private_dir(ddir_fname, CPD_CREATE, NULL);
+
+  consdiff_cfg_t consdiff_cfg = { 7200, 300 };
+  consdiffmgr_configure(&consdiff_cfg);
+  return (void *)1; // must return something non-null.
+}
+static int
+consdiffmgr_test_teardown(const struct testcase_t *arg, void *ignore)
+{
+  (void)arg;
+  (void)ignore;
+  consdiffmgr_free_all();
+  return 1;
+}
+static struct testcase_setup_t setup_diffmgr = {
+  consdiffmgr_test_setup,
+  consdiffmgr_test_teardown
+};
+
+// ============================== NS faking functions
+// These functions are for making quick fake consensus objects and
+// strings that are just good enough for consdiff and consdiffmgr.
+
+static networkstatus_t *
+fake_ns_new(consensus_flavor_t flav, time_t valid_after)
+{
+  networkstatus_t *ns = tor_malloc_zero(sizeof(networkstatus_t));
+  ns->type = NS_TYPE_CONSENSUS;
+  ns->flavor = flav;
+  ns->valid_after = valid_after;
+  return ns;
+}
+
+static char *
+fake_ns_body_new(consensus_flavor_t flav, time_t valid_after)
+{
+  const char *flavor_string = flav == FLAV_NS ? "" : " microdesc";
+  char valid_after_string[ISO_TIME_LEN+1];
+
+  format_iso_time(valid_after_string, valid_after);
+  char *random_stuff = crypto_random_hostname(3, 25, "junk ", "");
+
+  char *consensus;
+  tor_asprintf(&consensus,
+               "network-status-version 3%s\n"
+               "vote-status consensus\n"
+               "valid-after %s\n"
+               "r name ccccccccccccccccc etc\nsample\n"
+               "r name eeeeeeeeeeeeeeeee etc\nbar\n"
+               "%s\n",
+               flavor_string,
+               valid_after_string,
+               random_stuff);
+  tor_free(random_stuff);
+  return consensus;
+}
+
+// ============================== Cpuworker mocking code
+// These mocking functions and types capture the cpuworker calls
+// so we can inspect them and run them in the main thread.
+static smartlist_t *fake_cpuworker_queue = NULL;
+typedef struct fake_work_queue_ent_t {
+  enum workqueue_reply_t (*fn)(void *, void *);
+  void (*reply_fn)(void *);
+  void *arg;
+} fake_work_queue_ent_t;
+static struct workqueue_entry_s *
+mock_cpuworker_queue_work(enum workqueue_reply_t (*fn)(void *, void *),
+                          void (*reply_fn)(void *),
+                          void *arg)
+{
+  if (! fake_cpuworker_queue)
+    fake_cpuworker_queue = smartlist_new();
+
+  fake_work_queue_ent_t *ent = tor_malloc_zero(sizeof(*ent));
+  ent->fn = fn;
+  ent->reply_fn = reply_fn;
+  ent->arg = arg;
+  smartlist_add(fake_cpuworker_queue, ent);
+  return (struct workqueue_entry_s *)ent;
+}
+static int
+mock_cpuworker_run_work(void)
+{
+  if (! fake_cpuworker_queue)
+    return 0;
+  SMARTLIST_FOREACH(fake_cpuworker_queue, fake_work_queue_ent_t *, ent, {
+      enum workqueue_reply_t r = ent->fn(NULL, ent->arg);
+      if (r != WQ_RPL_REPLY)
+        return -1;
+  });
+  return 0;
+}
+static void
+mock_cpuworker_handle_replies(void)
+{
+  if (! fake_cpuworker_queue)
+    return;
+  SMARTLIST_FOREACH(fake_cpuworker_queue, fake_work_queue_ent_t *, ent, {
+      ent->reply_fn(ent->arg);
+  });
+  smartlist_free(fake_cpuworker_queue);
+  fake_cpuworker_queue = NULL;
+}
+
+// ==============================  Other helpers
+
+static consdiff_status_t
+lookup_diff_from(consensus_cache_entry_t **out,
+                 consensus_flavor_t flav,
+                 const char *str1)
+{
+  uint8_t digest[DIGEST256_LEN];
+  crypto_digest256((char*)digest, str1, strlen(str1), DIGEST_SHA3_256);
+  return consdiffmgr_find_diff_from(out, flav,
+                                    DIGEST_SHA3_256, digest, sizeof(digest));
+}
+
+static int
+lookup_apply_and_verify_diff(consensus_flavor_t flav,
+                             const char *str1,
+                             const char *str2)
+{
+  char *diff_string = NULL;
+  consensus_cache_entry_t *ent = NULL;
+  consdiff_status_t status = lookup_diff_from(&ent, flav, str1);
+  if (ent == NULL || status != CONSDIFF_AVAILABLE)
+    return -1;
+
+  consensus_cache_entry_incref(ent);
+  size_t size;
+  const uint8_t *body;
+  int r = consensus_cache_entry_get_body(ent, &body, &size);
+  if (r == 0)
+    diff_string = tor_memdup_nulterm(body, size);
+  consensus_cache_entry_decref(ent);
+  if (diff_string == NULL)
+    return -1;
+
+  char *applied = consensus_diff_apply(str1, diff_string);
+  tor_free(diff_string);
+  if (applied == NULL)
+    return -1;
+
+  int match = !strcmp(applied, str2);
+  tor_free(applied);
+  return match ? 0 : -1;
+}
+
+static void
+cdm_reload(void)
+{
+  consdiffmgr_free_all();
+  cdm_cache_get();
+  consdiffmgr_rescan();
+}
+
+// ==============================  Beginning of tests
+
+#if 0
+static int got_failure = 0;
+static void
+got_assertion_failure(void)
+{
+  ++got_failure;
+}
+
+/* XXXX This test won't work, because there is currently no way to actually
+ * XXXX capture a real assertion failure. */
+static void
+test_consdiffmgr_init_failure(void *arg)
+{
+  (void)arg;
+  // Capture assertions and bugs.
+
+  /* As in ...test_setup, but do not create the datadir. The missing directory
+   * will cause a failure. */
+  char *ddir_fname = tor_strdup(get_fname_rnd("datadir_cdm"));
+  tor_free(get_options_mutable()->DataDirectory);
+  get_options_mutable()->DataDirectory = ddir_fname; // now owns the pointer.
+
+  consdiff_cfg_t consdiff_cfg = { 7200, 300 };
+
+  tor_set_failed_assertion_callback(got_assertion_failure);
+  tor_capture_bugs_(1);
+  consdiffmgr_configure(&consdiff_cfg); // This should fail.
+  tt_int_op(got_failure, OP_EQ, 1);
+  const smartlist_t *bugs = tor_get_captured_bug_log_();
+  tt_int_op(smartlist_len(bugs), OP_EQ, 1);
+
+ done:
+  tor_end_capture_bugs_();
+}
+#endif
+
+static void
+test_consdiffmgr_sha3_helper(void *arg)
+{
+  (void) arg;
+  consensus_cache_t *cache = cdm_cache_get(); // violate abstraction barrier
+  config_line_t *lines = NULL;
+  char *mem_op_hex_tmp = NULL;
+  config_line_prepend(&lines, "good-sha",
+                      "F00DF00DF00DF00DF00DF00DF00DF00D"
+                      "F00DF00DF00DF00DF00DF00DF00DF00D");
+  config_line_prepend(&lines, "short-sha",
+                      "F00DF00DF00DF00DF00DF00DF00DF00D"
+                      "F00DF00DF00DF00DF00DF00DF00DF0");
+  config_line_prepend(&lines, "long-sha",
+                      "F00DF00DF00DF00DF00DF00DF00DF00D"
+                      "F00DF00DF00DF00DF00DF00DF00DF00DF00D");
+  config_line_prepend(&lines, "not-sha",
+                      "F00DF00DF00DF00DF00DF00DF00DF00D"
+                      "F00DF00DF00DF00DF00DF00DF00DXXXX");
+  consensus_cache_entry_t *ent =
+    consensus_cache_add(cache, lines, (const uint8_t *)"Hi there", 8);
+
+  uint8_t buf[DIGEST256_LEN];
+  tt_int_op(-1, OP_EQ, cdm_entry_get_sha3_value(buf, NULL, "good-sha"));
+  tt_int_op(0, OP_EQ, cdm_entry_get_sha3_value(buf, ent, "good-sha"));
+  test_memeq_hex(buf, "F00DF00DF00DF00DF00DF00DF00DF00D"
+                      "F00DF00DF00DF00DF00DF00DF00DF00D");
+
+  tt_int_op(-1, OP_EQ, cdm_entry_get_sha3_value(buf, ent, "missing-sha"));
+  tt_int_op(-2, OP_EQ, cdm_entry_get_sha3_value(buf, ent, "short-sha"));
+  tt_int_op(-2, OP_EQ, cdm_entry_get_sha3_value(buf, ent, "long-sha"));
+  tt_int_op(-2, OP_EQ, cdm_entry_get_sha3_value(buf, ent, "not-sha"));
+
+ done:
+  consensus_cache_entry_decref(ent);
+  config_free_lines(lines);
+  tor_free(mem_op_hex_tmp);
+}
+
+static void
+test_consdiffmgr_add(void *arg)
+{
+  (void) arg;
+  time_t now = approx_time();
+
+  consensus_cache_entry_t *ent = NULL;
+  networkstatus_t *ns_tmp = fake_ns_new(FLAV_NS, now);
+  const char *dummy = "foo";
+  int r = consdiffmgr_add_consensus(dummy, ns_tmp);
+  tt_int_op(r, OP_EQ, 0);
+
+  /* If we add it again, it won't work */
+  setup_capture_of_logs(LOG_INFO);
+  dummy = "bar";
+  r = consdiffmgr_add_consensus(dummy, ns_tmp);
+  tt_int_op(r, OP_EQ, -1);
+  expect_single_log_msg_containing("We already have a copy of that "
+                                   "consensus");
+  mock_clean_saved_logs();
+
+  /* But it will work fine if the flavor is different */
+  dummy = "baz";
+  ns_tmp->flavor = FLAV_MICRODESC;
+  r = consdiffmgr_add_consensus(dummy, ns_tmp);
+  tt_int_op(r, OP_EQ, 0);
+
+  /* And it will work fine if the time is different */
+  dummy = "quux";
+  ns_tmp->flavor = FLAV_NS;
+  ns_tmp->valid_after = now - 60;
+  r = consdiffmgr_add_consensus(dummy, ns_tmp);
+  tt_int_op(r, OP_EQ, 0);
+
+  /* If we add one a long long time ago, it will fail. */
+  dummy = "xyzzy";
+  ns_tmp->valid_after = 86400 * 100; /* A few months into 1970 */
+  r = consdiffmgr_add_consensus(dummy, ns_tmp);
+  tt_int_op(r, OP_EQ, -1);
+  expect_single_log_msg_containing("it's too old.");
+
+  /* Try looking up a consensuses. */
+  ent = cdm_cache_lookup_consensus(FLAV_NS, now-60);
+  tt_assert(ent);
+  consensus_cache_entry_incref(ent);
+  size_t s;
+  const uint8_t *body;
+  r = consensus_cache_entry_get_body(ent, &body, &s);
+  tt_int_op(r, OP_EQ, 0);
+  tt_int_op(s, OP_EQ, 4);
+  tt_mem_op(body, OP_EQ, "quux", 4);
+
+  /* Try looking up another entry, but fail */
+  tt_assert(NULL == cdm_cache_lookup_consensus(FLAV_MICRODESC, now-60));
+  tt_assert(NULL == cdm_cache_lookup_consensus(FLAV_NS, now-61));
+
+ done:
+  networkstatus_vote_free(ns_tmp);
+  teardown_capture_of_logs();
+  consensus_cache_entry_decref(ent);
+}
+
+static void
+test_consdiffmgr_make_diffs(void *arg)
+{
+  (void)arg;
+  networkstatus_t *ns = NULL;
+  char *ns_body = NULL, *md_ns_body = NULL, *md_ns_body_2 = NULL;
+  char *applied = NULL, *diff_text = NULL;
+  time_t now = approx_time();
+  int r;
+  consensus_cache_entry_t *diff = NULL;
+  uint8_t md_ns_sha3[DIGEST256_LEN];
+  consdiff_status_t diff_status;
+
+  MOCK(cpuworker_queue_work, mock_cpuworker_queue_work);
+
+  // Try rescan with no consensuses: shouldn't crash or queue work.
+  consdiffmgr_rescan();
+  tt_ptr_op(NULL, OP_EQ, fake_cpuworker_queue);
+
+  // Make two consensuses, 1 hour sec ago.
+  ns = fake_ns_new(FLAV_NS, now-3600);
+  ns_body = fake_ns_body_new(FLAV_NS, now-3600);
+  r = consdiffmgr_add_consensus(ns_body, ns);
+  networkstatus_vote_free(ns);
+  tor_free(ns_body);
+  tt_int_op(r, OP_EQ, 0);
+
+  ns = fake_ns_new(FLAV_MICRODESC, now-3600);
+  md_ns_body = fake_ns_body_new(FLAV_MICRODESC, now-3600);
+  r = consdiffmgr_add_consensus(md_ns_body, ns);
+  crypto_digest256((char*)md_ns_sha3, md_ns_body, strlen(md_ns_body),
+                   DIGEST_SHA3_256);
+  networkstatus_vote_free(ns);
+  tt_int_op(r, OP_EQ, 0);
+
+  // No diffs will be generated.
+  consdiffmgr_rescan();
+  tt_ptr_op(NULL, OP_EQ, fake_cpuworker_queue);
+
+  // Add a MD consensus from 45 minutes ago. This should cause one diff
+  // worth of work to get queued.
+  ns = fake_ns_new(FLAV_MICRODESC, now-45*60);
+  md_ns_body_2 = fake_ns_body_new(FLAV_MICRODESC, now-45*60);
+  r = consdiffmgr_add_consensus(md_ns_body_2, ns);
+  networkstatus_vote_free(ns);
+  tt_int_op(r, OP_EQ, 0);
+
+  consdiffmgr_rescan();
+  tt_ptr_op(NULL, OP_NE, fake_cpuworker_queue);
+  tt_int_op(1, OP_EQ, smartlist_len(fake_cpuworker_queue));
+  diff_status = consdiffmgr_find_diff_from(&diff, FLAV_MICRODESC,
+                                           DIGEST_SHA3_256,
+                                           md_ns_sha3, DIGEST256_LEN);
+  tt_int_op(CONSDIFF_IN_PROGRESS, OP_EQ, diff_status);
+
+  // Now run that process and get the diff.
+  r = mock_cpuworker_run_work();
+  tt_int_op(r, OP_EQ, 0);
+  mock_cpuworker_handle_replies();
+
+  // At this point we should be able to get that diff.
+  diff_status = consdiffmgr_find_diff_from(&diff, FLAV_MICRODESC,
+                                           DIGEST_SHA3_256,
+                                           md_ns_sha3, DIGEST256_LEN);
+  tt_int_op(CONSDIFF_AVAILABLE, OP_EQ, diff_status);
+  tt_assert(diff);
+
+  /* Make sure applying the diff actually works */
+  const uint8_t *diff_body;
+  size_t diff_size;
+  r = consensus_cache_entry_get_body(diff, &diff_body, &diff_size);
+  tt_int_op(r, OP_EQ, 0);
+  diff_text = tor_memdup_nulterm(diff_body, diff_size);
+  applied = consensus_diff_apply(md_ns_body, diff_text);
+  tt_assert(applied);
+  tt_str_op(applied, OP_EQ, md_ns_body_2);
+
+  /* Rescan again: no more work to do. */
+  consdiffmgr_rescan();
+  tt_ptr_op(NULL, OP_EQ, fake_cpuworker_queue);
+
+ done:
+  tor_free(md_ns_body);
+  tor_free(md_ns_body_2);
+  tor_free(diff_text);
+  tor_free(applied);
+}
+
+static void
+test_consdiffmgr_diff_rules(void *arg)
+{
+  (void)arg;
+#define N 6
+  char *md_body[N], *ns_body[N];
+  networkstatus_t *md_ns[N], *ns_ns[N];
+  uint8_t md_ns_sha3[N][DIGEST256_LEN], ns_ns_sha3[N][DIGEST256_LEN];
+  int i;
+
+  MOCK(cpuworker_queue_work, mock_cpuworker_queue_work);
+
+  /* Create a bunch of consensus things at 15-second intervals. */
+  time_t start = approx_time() - 120;
+  for (i = 0; i < N; ++i) {
+    time_t when = start + i * 15;
+    md_body[i] = fake_ns_body_new(FLAV_MICRODESC, when);
+    ns_body[i] = fake_ns_body_new(FLAV_NS, when);
+    md_ns[i] = fake_ns_new(FLAV_MICRODESC, when);
+    ns_ns[i] = fake_ns_new(FLAV_NS, when);
+    crypto_digest256((char *)md_ns_sha3[i], md_body[i], strlen(md_body[i]),
+                     DIGEST_SHA3_256);
+    crypto_digest256((char *)ns_ns_sha3[i], ns_body[i], strlen(ns_body[i]),
+                     DIGEST_SHA3_256);
+  }
+
+  /* For the MD consensuses: add 4 of them, and make sure that
+   * diffs are created to one consensus (the most recent) only. */
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(md_body[1], md_ns[1]));
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(md_body[2], md_ns[2]));
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(md_body[3], md_ns[3]));
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(md_body[4], md_ns[4]));
+  consdiffmgr_rescan();
+  tt_ptr_op(NULL, OP_NE, fake_cpuworker_queue);
+  tt_int_op(3, OP_EQ, smartlist_len(fake_cpuworker_queue));
+  tt_int_op(0, OP_EQ, mock_cpuworker_run_work());
+  mock_cpuworker_handle_replies();
+  tt_ptr_op(NULL, OP_EQ, fake_cpuworker_queue);
+
+  /* For the NS consensuses: add 3, generate, and add one older one and
+   * make sure that older one is the only one whose diff is generated */
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(ns_body[0], ns_ns[0]));
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(ns_body[1], ns_ns[1]));
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(ns_body[5], ns_ns[5]));
+  consdiffmgr_rescan();
+  tt_ptr_op(NULL, OP_NE, fake_cpuworker_queue);
+  tt_int_op(2, OP_EQ, smartlist_len(fake_cpuworker_queue));
+  tt_int_op(0, OP_EQ, mock_cpuworker_run_work());
+  mock_cpuworker_handle_replies();
+
+  /* At this point, we should actually have working diffs! */
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_NS, ns_body[0], ns_body[5]));
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_NS, ns_body[1], ns_body[5]));
+
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_MICRODESC, md_body[1], md_body[4]));
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_MICRODESC, md_body[2], md_body[4]));
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_MICRODESC, md_body[3], md_body[4]));
+
+  /* Self-to-self diff won't be present */
+  consensus_cache_entry_t *ent;
+  tt_int_op(CONSDIFF_NOT_FOUND, OP_EQ,
+       lookup_diff_from(&ent, FLAV_NS, ns_body[5]));
+  /* No diff from 2 has been added yet */
+  tt_int_op(CONSDIFF_NOT_FOUND, OP_EQ,
+       lookup_diff_from(&ent, FLAV_NS, ns_body[2]));
+  /* No diff arriving at old things. */
+  tt_int_op(-1, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_MICRODESC, md_body[1], md_body[2]));
+  /* No backwards diff */
+  tt_int_op(-1, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_MICRODESC, md_body[4], md_body[3]));
+
+  /* Now, an update: add number 2 and make sure it's the only one whose diff
+   * is regenerated. */
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(ns_body[2], ns_ns[2]));
+  consdiffmgr_rescan();
+  tt_ptr_op(NULL, OP_NE, fake_cpuworker_queue);
+  tt_int_op(1, OP_EQ, smartlist_len(fake_cpuworker_queue));
+  tt_int_op(0, OP_EQ, mock_cpuworker_run_work());
+  mock_cpuworker_handle_replies();
+
+  tt_int_op(0, OP_EQ,
+            lookup_apply_and_verify_diff(FLAV_NS, ns_body[2], ns_body[5]));
+
+  /* Finally: reload, and make sure that the information is still indexed */
+  cdm_reload();
+
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_NS, ns_body[0], ns_body[5]));
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_NS, ns_body[2], ns_body[5]));
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_NS, ns_body[1], ns_body[5]));
+
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_MICRODESC, md_body[1], md_body[4]));
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_MICRODESC, md_body[2], md_body[4]));
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_MICRODESC, md_body[3], md_body[4]));
+
+ done:
+  for (i = 0; i < N; ++i) {
+    tor_free(md_body[i]);
+    tor_free(ns_body[i]);
+    networkstatus_vote_free(md_ns[i]);
+    networkstatus_vote_free(ns_ns[i]);
+  }
+  UNMOCK(cpuworker_queue_work);
+#undef N
+}
+
+static void
+test_consdiffmgr_diff_failure(void *arg)
+{
+  (void)arg;
+  MOCK(cpuworker_queue_work, mock_cpuworker_queue_work);
+
+  /* We're going to make sure that if we have a bogus request where
+   * we can't actually compute a diff, the world must not end. */
+  networkstatus_t *ns1 = NULL;
+  networkstatus_t *ns2 = NULL;
+  int r;
+
+  ns1 = fake_ns_new(FLAV_NS, approx_time()-100);
+  ns2 = fake_ns_new(FLAV_NS, approx_time()-50);
+  r = consdiffmgr_add_consensus("foo bar baz\n", ns1);
+  tt_int_op(r, OP_EQ, 0);
+  // We refuse to compute a diff to or from a line holding only a single dot.
+  // We can add it here, though.
+  r = consdiffmgr_add_consensus("foo bar baz\n.\n.\n", ns2);
+  tt_int_op(r, OP_EQ, 0);
+
+  consdiffmgr_rescan();
+  tt_ptr_op(NULL, OP_NE, fake_cpuworker_queue);
+  setup_capture_of_logs(LOG_WARN);
+  tt_int_op(1, OP_EQ, smartlist_len(fake_cpuworker_queue));
+  tt_int_op(0, OP_EQ, mock_cpuworker_run_work());
+  expect_single_log_msg_containing("one of the lines to be added is \".\".");
+  mock_clean_saved_logs();
+  mock_cpuworker_handle_replies();
+  expect_single_log_msg_containing("Worker was unable to compute consensus "
+                                   "diff from ");
+
+  /* Make sure the diff is not present */
+  consensus_cache_entry_t *ent;
+  tt_int_op(CONSDIFF_NOT_FOUND, OP_EQ,
+            lookup_diff_from(&ent, FLAV_NS, "foo bar baz\n"));
+
+ done:
+  teardown_capture_of_logs();
+  UNMOCK(cpuworker_queue_work);
+  networkstatus_vote_free(ns1);
+  networkstatus_vote_free(ns2);
+}
+
+static void
+test_consdiffmgr_diff_pending(void *arg)
+{
+#define N 3
+  (void)arg;
+  char *md_body[N];
+  networkstatus_t *md_ns[N];
+  time_t start = approx_time() - 120;
+  int i;
+  for (i = 0; i < N; ++i) {
+    time_t when = start + i * 30;
+    md_body[i] = fake_ns_body_new(FLAV_MICRODESC, when);
+    md_ns[i] = fake_ns_new(FLAV_MICRODESC, when);
+  }
+
+  MOCK(cpuworker_queue_work, mock_cpuworker_queue_work);
+
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(md_body[1], md_ns[1]));
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(md_body[2], md_ns[2]));
+  /* Make a diff */
+  consdiffmgr_rescan();
+  tt_int_op(1, OP_EQ, smartlist_len(fake_cpuworker_queue));
+
+  /* Look it up.  Is it pending? */
+  consensus_cache_entry_t *ent = NULL;
+  consdiff_status_t diff_status;
+  diff_status = lookup_diff_from(&ent, FLAV_MICRODESC, md_body[1]);
+  tt_int_op(CONSDIFF_IN_PROGRESS, OP_EQ, diff_status);
+  tt_ptr_op(ent, OP_EQ, NULL);
+
+  /* Add another old consensus.  only one new diff should launch! */
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(md_body[0], md_ns[0]));
+  consdiffmgr_rescan();
+  tt_int_op(2, OP_EQ, smartlist_len(fake_cpuworker_queue));
+
+  tt_int_op(0, OP_EQ, mock_cpuworker_run_work());
+  mock_cpuworker_handle_replies();
+
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_MICRODESC, md_body[0], md_body[2]));
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_MICRODESC, md_body[1], md_body[2]));
+
+ done:
+  UNMOCK(cpuworker_queue_work);
+#undef N
+}
+
+static void
+test_consdiffmgr_cleanup_old(void *arg)
+{
+  (void)arg;
+  config_line_t *labels = NULL;
+  consensus_cache_entry_t *ent = NULL;
+  consensus_cache_t *cache = cdm_cache_get(); // violate abstraction barrier
+
+  /* This item will be will be cleanable because it has a valid-after
+   * time far in the past. */
+  config_line_prepend(&labels, "document-type", "confribble-blarg");
+  config_line_prepend(&labels, "consensus-valid-after",
+                      "1980-10-10T10:10:10");
+  ent = consensus_cache_add(cache, labels, (const uint8_t*)"Foo", 3);
+  tt_assert(ent);
+  consensus_cache_entry_decref(ent);
+
+  setup_capture_of_logs(LOG_DEBUG);
+  tt_int_op(1, OP_EQ, consdiffmgr_cleanup());
+  expect_log_msg_containing("Deleting entry because its consensus-valid-"
+                            "after value (1980-10-10T10:10:10) was too old");
+
+ done:
+  teardown_capture_of_logs();
+  config_free_lines(labels);
+}
+
+static void
+test_consdiffmgr_cleanup_bad_valid_after(void *arg)
+{
+  /* This will seem cleanable, but isn't, because its valid-after time is
+   * misformed. */
+
+  (void)arg;
+  config_line_t *labels = NULL;
+  consensus_cache_entry_t *ent = NULL;
+  consensus_cache_t *cache = cdm_cache_get(); // violate abstraction barrier
+
+  config_line_prepend(&labels, "document-type", "consensus");
+  config_line_prepend(&labels, "consensus-valid-after",
+                      "whan that aprille with his shoures soote"); // (~1385?)
+  ent = consensus_cache_add(cache, labels, (const uint8_t*)"Foo", 3);
+  tt_assert(ent);
+  consensus_cache_entry_decref(ent);
+
+  setup_capture_of_logs(LOG_DEBUG);
+  tt_int_op(0, OP_EQ, consdiffmgr_cleanup());
+  expect_log_msg_containing("Ignoring entry because its consensus-valid-"
+                            "after value (\"whan that aprille with his "
+                            "shoures soote\") was unparseable");
+
+ done:
+  teardown_capture_of_logs();
+  config_free_lines(labels);
+}
+
+static void
+test_consdiffmgr_cleanup_no_valid_after(void *arg)
+{
+  (void)arg;
+  config_line_t *labels = NULL;
+  consensus_cache_entry_t *ent = NULL;
+  consensus_cache_t *cache = cdm_cache_get(); // violate abstraction barrier
+
+  /* This item will be will be uncleanable because it has no recognized
+   * valid-after. */
+  config_line_prepend(&labels, "document-type", "consensus");
+  config_line_prepend(&labels, "confrooble-voolid-oofter",
+                      "2010-10-10T09:08:07");
+  ent = consensus_cache_add(cache, labels, (const uint8_t*)"Foo", 3);
+  tt_assert(ent);
+  consensus_cache_entry_decref(ent);
+
+  setup_capture_of_logs(LOG_DEBUG);
+  tt_int_op(0, OP_EQ, consdiffmgr_cleanup());
+  expect_log_msg_containing("Ignoring entry because it had no consensus-"
+                            "valid-after label");
+
+ done:
+  teardown_capture_of_logs();
+  config_free_lines(labels);
+}
+
+static void
+test_consdiffmgr_cleanup_old_diffs(void *arg)
+{
+  (void)arg;
+#define N 4
+  char *md_body[N];
+  networkstatus_t *md_ns[N];
+  uint8_t md_ns_sha3[N][DIGEST256_LEN];
+  int i;
+  consensus_cache_entry_t *hold_ent = NULL, *ent;
+
+  /* Make sure that the cleanup function removes diffs to the not-most-recent
+   * consensus. */
+
+  MOCK(cpuworker_queue_work, mock_cpuworker_queue_work);
+
+  /* Create a bunch of consensus things at 15-second intervals. */
+  time_t start = approx_time() - 120;
+  for (i = 0; i < N; ++i) {
+    time_t when = start + i * 15;
+    md_body[i] = fake_ns_body_new(FLAV_MICRODESC, when);
+    md_ns[i] = fake_ns_new(FLAV_MICRODESC, when);
+    crypto_digest256((char *)md_ns_sha3[i], md_body[i], strlen(md_body[i]),
+                     DIGEST_SHA3_256);
+  }
+
+  /* add the first 3. */
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(md_body[0], md_ns[0]));
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(md_body[1], md_ns[1]));
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(md_body[2], md_ns[2]));
+  /* Make diffs. */
+  consdiffmgr_rescan();
+  tt_ptr_op(NULL, OP_NE, fake_cpuworker_queue);
+  tt_int_op(2, OP_EQ, smartlist_len(fake_cpuworker_queue));
+  tt_int_op(0, OP_EQ, mock_cpuworker_run_work());
+  mock_cpuworker_handle_replies();
+  tt_ptr_op(NULL, OP_EQ, fake_cpuworker_queue);
+
+  /* Nothing is deletable now */
+  tt_int_op(0, OP_EQ, consdiffmgr_cleanup());
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_MICRODESC, md_body[0], md_body[2]));
+  tt_int_op(0, OP_EQ,
+       lookup_apply_and_verify_diff(FLAV_MICRODESC, md_body[1], md_body[2]));
+
+  tt_int_op(CONSDIFF_AVAILABLE, OP_EQ,
+            lookup_diff_from(&hold_ent, FLAV_MICRODESC, md_body[1]));
+  consensus_cache_entry_incref(hold_ent); // incref, so it is preserved.
+
+  /* Now add an even-more-recent consensus; this should make all previous
+   * diffs deletable */
+  tt_int_op(0, OP_EQ, consdiffmgr_add_consensus(md_body[3], md_ns[3]));
+  tt_int_op(2, OP_EQ, consdiffmgr_cleanup());
+
+  tt_int_op(CONSDIFF_NOT_FOUND, OP_EQ,
+            lookup_diff_from(&ent, FLAV_MICRODESC, md_body[0]));
+  /* This one is marked deletable but still in the hashtable */
+  tt_int_op(CONSDIFF_AVAILABLE, OP_EQ,
+            lookup_diff_from(&ent, FLAV_MICRODESC, md_body[1]));
+  tt_int_op(CONSDIFF_NOT_FOUND, OP_EQ,
+            lookup_diff_from(&ent, FLAV_MICRODESC, md_body[2]));
+
+  /* Everything should be valid at this point */
+  tt_int_op(0, OP_EQ, consdiffmgr_validate());
+
+  /* And if we recan NOW, we'll purge the hashtable of the entries,
+   * and launch attempts to generate new ones */
+  consdiffmgr_rescan();
+  tt_int_op(CONSDIFF_IN_PROGRESS, OP_EQ,
+            lookup_diff_from(&ent, FLAV_MICRODESC, md_body[0]));
+  tt_int_op(CONSDIFF_IN_PROGRESS, OP_EQ,
+            lookup_diff_from(&ent, FLAV_MICRODESC, md_body[1]));
+  tt_int_op(CONSDIFF_IN_PROGRESS, OP_EQ,
+            lookup_diff_from(&ent, FLAV_MICRODESC, md_body[2]));
+
+  /* We're still holding on to this, though, so we can still map it! */
+  const uint8_t *t1 = NULL;
+  size_t s;
+  int r = consensus_cache_entry_get_body(hold_ent, &t1, &s);
+  tt_int_op(r, OP_EQ, 0);
+  tt_assert(t1);
+
+ done:
+  for (i = 0; i < N; ++i) {
+    tor_free(md_body[i]);
+    networkstatus_vote_free(md_ns[i]);
+  }
+  consensus_cache_entry_decref(hold_ent);
+  UNMOCK(cpuworker_queue_work);
+#undef N
+}
+
+static void
+test_consdiffmgr_validate(void *arg)
+{
+  (void)arg;
+  config_line_t *lines = NULL;
+  consensus_cache_entry_t *ent = NULL;
+  consensus_cache_t *cache = cdm_cache_get(); // violate abstraction barrier
+  smartlist_t *vals = smartlist_new();
+
+  /* Put these: objects in the cache: one with a good sha3, one with bad sha3,
+   * one with a wrong sha3, and one with no sha3. */
+  config_line_prepend(&lines, "id", "wrong sha3");
+  config_line_prepend(&lines, "sha3-digest",
+                      "F00DF00DF00DF00DF00DF00DF00DF00D"
+                      "F00DF00DF00DF00DF00DF00DF00DF00D");
+  ent = consensus_cache_add(cache, lines, (const uint8_t *)"Hi there", 8);
+  consensus_cache_entry_decref(ent);
+  config_free_lines(lines);
+  lines = NULL;
+
+  config_line_prepend(&lines, "id", "bad sha3");
+  config_line_prepend(&lines, "sha3-digest",
+                      "now is the winter of our dicotheque");
+  ent = consensus_cache_add(cache, lines, (const uint8_t *)"Hi there", 8);
+  consensus_cache_entry_decref(ent);
+  config_free_lines(lines);
+  lines = NULL;
+
+  config_line_prepend(&lines, "id", "no sha3");
+  ent = consensus_cache_add(cache, lines, (const uint8_t *)"Hi there", 8);
+  consensus_cache_entry_decref(ent);
+  config_free_lines(lines);
+  lines = NULL;
+
+  config_line_prepend(&lines, "id", "good sha3");
+  config_line_prepend(&lines, "sha3-digest",
+                      "8d8b1998616cd6b4c4055da8d38728dc"
+                      "93c758d4131a53c7d81aa6337dee1c05");
+  ent = consensus_cache_add(cache, lines, (const uint8_t *)"Hi there", 8);
+  consensus_cache_entry_decref(ent);
+  config_free_lines(lines);
+  lines = NULL;
+
+  cdm_reload();
+  cache = cdm_cache_get();
+  tt_int_op(1, OP_EQ, consdiffmgr_validate());
+
+  consensus_cache_find_all(vals, cache, "id", "good sha3");
+  tt_int_op(smartlist_len(vals), OP_EQ, 1);
+  smartlist_clear(vals);
+
+  consensus_cache_find_all(vals, cache, "id", "no sha3");
+  tt_int_op(smartlist_len(vals), OP_EQ, 1);
+  smartlist_clear(vals);
+
+  consensus_cache_find_all(vals, cache, "id", "wrong sha3");
+  tt_int_op(smartlist_len(vals), OP_EQ, 0);
+  consensus_cache_find_all(vals, cache, "id", "bad sha3");
+  tt_int_op(smartlist_len(vals), OP_EQ, 0);
+
+ done:
+  smartlist_free(vals);
+}
+
+#define TEST(name)                                      \
+  { #name, test_consdiffmgr_ ## name , TT_FORK, &setup_diffmgr, NULL }
+
+struct testcase_t consdiffmgr_tests[] = {
+#if 0
+  { "init_failure", test_consdiffmgr_init_failure, TT_FORK, NULL, NULL },
+#endif
+  TEST(sha3_helper),
+  TEST(add),
+  TEST(make_diffs),
+  TEST(diff_rules),
+  TEST(diff_failure),
+  TEST(diff_pending),
+  TEST(cleanup_old),
+  TEST(cleanup_bad_valid_after),
+  TEST(cleanup_no_valid_after),
+  TEST(cleanup_old_diffs),
+  TEST(validate),
+
+  // XXXX Test: non-cacheing cases of replyfn().
+
+  END_OF_TESTCASES
+};
+