Add withdrawal hashes to publication and rrdp.

svn path=/branches/tk705/; revision=5888

14 files changed, 301 insertions, 216 deletions

diff --git a/ca/tests/publication-protocol-samples.xml b/ca/tests/publication-protocol-samples.xml
index a9a84f73..6d0a99a9 100644
--- a/ca/tests/publication-protocol-samples.xml
+++ b/ca/tests/publication-protocol-samples.xml
@@ -96,7 +96,7 @@
   </msg>
 
   <msg xmlns="http://www.hactrn.net/uris/rpki/publication-spec/" type="query" version="3">
-    <withdraw uri="rsync://wombat.example/Alice/blCrcCp9ltyPDNzYKPfxc.cer"/>
+    <withdraw uri="rsync://wombat.example/Alice/blCrcCp9ltyPDNzYKPfxc.cer" hash="deadf00d"/>
   </msg>
 
   <msg xmlns="http://www.hactrn.net/uris/rpki/publication-spec/" type="reply" version="3">
diff --git a/ca/tests/rrdp-samples.xml b/ca/tests/rrdp-samples.xml
index 563eb73c..0318b169 100644
--- a/ca/tests/rrdp-samples.xml
+++ b/ca/tests/rrdp-samples.xml
@@ -68,20 +68,20 @@
        </publish>
      </delta>
      <delta serial="2">
-       <withdraw uri="http://host.example/foo/bar/cer1.cer"/>
+       <withdraw uri="http://host.example/foo/bar/cer1.cer" hash="deadf00d"/>
        <publish uri="http://host.example/foo/bar/cer2.cer">
          MIIE+jCCA+KgAwIBAgIBDTANBgkqhkiG9w0BAQsFADAzMTEw
          h8zeHoFVu6ghRPy5dbOA4akX/KG6b8XIx0iwPYdLiDbdWFbt
          jRBODAxN0U2NkE5RTkxNzJFNDYxMkQ4Q0Y0QzgzRjIzOERFM
        </publish>
-       <publish uri="http://host.example/foo/bar/cer3.cer">
+       <publish uri="http://host.example/foo/bar/cer3.cer" hash="deadf00d">
          MIIE+jCCA+KgAwIBAgIBDTANBgkqhkiG9w0BAQsFADAzMTEw
          h8zeHoFVu6ghRPy5dbOA4akX/KG6b8XIx0iwPYdLiDbdWFbt
          jRBODAxN0U2NkE5RTkxNzJFNDYxMkQ4Q0Y0QzgzRjIzOERFM
        </publish>
      </delta>
      <delta serial="3">
-       <withdraw uri="http://host.example/foo/bar/cer2.cer"/>
+       <withdraw uri="http://host.example/foo/bar/cer2.cer" hash="deadf00d"/>
      </delta>
    </deltas>
 
diff --git a/rpki/left_right.py b/rpki/left_right.py
index e4b664b9..c90cac29 100644
--- a/rpki/left_right.py
+++ b/rpki/left_right.py
@@ -312,42 +312,38 @@ class self_elt(data_elt):
   def serve_publish_world_now(self, cb, eb):
     """
     Handle a left-right publish_world_now action for this self.
-
-    The publication stuff needs refactoring, right now publication is
-    interleaved with local operations in a way that forces far too
-    many bounces through the task system for any complex update.  The
-    whole thing ought to be rewritten to queue up outgoing publication
-    PDUs and only send them when we're all done or when we need to
-    force publication at a particular point in a multi-phase operation.
-
-    Once that reorganization has been done, this method should be
-    rewritten to reuse the low-level publish() methods that each
-    object will have...but we're not there yet.  So, for now, we just
-    do this via brute force.  Think of it as a trial version to see
-    whether we've identified everything that needs to be republished
-    for this operation.
     """
 
+    publisher = rpki.rpkid.publication_queue()
+
     def loop(iterator, parent):
-      q_msg = rpki.publication.msg.query()
+      repo = parent.repository
       for ca in parent.cas:
         ca_detail = ca.active_ca_detail
         if ca_detail is not None:
-          q_msg.append(rpki.publication.publish_elt.make(
-            ca_detail.crl_uri, ca_detail.latest_crl))
-          q_msg.append(rpki.publication.publish_elt.make(
-            ca_detail.manifest_uri, ca_detail.latest_manifest))
-          q_msg.extend(rpki.publication.publish_elt.make(
-            c.uri, c.cert) for c in ca_detail.child_certs)
-          q_msg.extend(rpki.publication.publish_elt.make(
-            r.uri, r.roa) for r in ca_detail.roas if r.roa is not None)
-          q_msg.extend(rpki.publication.publish_elt.make(
-            g.uri, g.ghostbuster) for g in ca_detail.ghostbusters)
-          q_msg.extend(rpki.publication.publish_elt.make(
-            c.uri, c.cert) for c in ca_detail.ee_certificates)
-      parent.repository.call_pubd(iterator, eb, q_msg)
+          publisher.queue(
+            uri = ca_detail.crl_uri, new_obj = ca_detail.latest_crl, repository = repo)
+          publisher.queue(
+            uri = ca_detail.manifest_uri, new_obj = ca_detail.latest_manifest, repository = repo)
+          for c in ca_detail.child_certs:
+            publisher.queue(
+              uri = c.uri, new_obj = c.cert, repository = repo)
+          for r in ca_detail.roas:
+            if r.roa is not None:
+              publisher.queue(
+                uri = r.uri, new_obj = r.roa, repository = repo)
+          for g in ca_detail.ghostbusters:
+            publisher.queue(
+              uri = g.uri, new_obj = g.ghostbuster, repository = repo)
+          for c in ca_detail.ee_certificates:
+            publisher.queue(
+              uri = c.uri, new_obj = c.cert, repository = repo)
+      iterator()
 
-    rpki.async.iterator(self.parents, loop, cb)
+    def done():
+      publisher.call_pubd(cb, eb)
+
+    rpki.async.iterator(self.parents, loop, done)
 
   def serve_run_now(self, cb, eb):
     """
diff --git a/rpki/pubd.py b/rpki/pubd.py
index 46e431c4..14de1999 100644
--- a/rpki/pubd.py
+++ b/rpki/pubd.py
@@ -188,7 +188,7 @@ class session_obj(rpki.sql.sql_persistent):
     "uuid")
 
   ## @var expiration_interval
-  # How long to wait after retiring a snapshot before purging it from the database. 
+  # How long to wait after retiring a snapshot before purging it from the database.
 
   expiration_interval = rpki.sundial.timedelta(hours = 6)
 
@@ -318,7 +318,7 @@ class snapshot_obj(rpki.sql.sql_persistent):
     obj.delete(self)
 
 
-    
+
 class object_obj(rpki.sql.sql_persistent):
   """
   A published object.
diff --git a/rpki/publication.py b/rpki/publication.py
index 19ab2107..7b5abaf9 100644
--- a/rpki/publication.py
+++ b/rpki/publication.py
@@ -49,11 +49,15 @@ class base_publication_elt(rpki.xml_utils.base_elt, publication_namespace):
   Base element for publication protocol.  Publish and withdraw PDUs subclass this.
   """
 
-  attributes = ("tag", "uri")
+  attributes = ("tag", "uri", "hash")
+
+  tag = None
+  uri = None
+  hash = None
   payload = None
 
   def __repr__(self):
-    return rpki.log.log_repr(self, self.uri, self.payload)
+    return rpki.log.log_repr(self, self.tag, self.uri, self.hash, self.payload)
 
   def serve_dispatch(self, r_msg, snapshot, cb, eb):
     """
@@ -138,15 +142,6 @@ class publish_elt(base_publication_elt):
       f.write(self.payload.get_DER())
     os.rename(filename_tmp, filename)
 
-  @classmethod
-  def make(cls, uri, obj, tag = None):
-    """
-    Construct a publication PDU.
-    """
-
-    assert isinstance(obj, rpki.x509.uri_dispatch(uri))
-    return cls.make_pdu(uri = uri, payload = obj, tag = tag)
-
 
 class withdraw_elt(base_publication_elt):
 
@@ -177,15 +172,6 @@ class withdraw_elt(base_publication_elt):
       else:
         dirname = os.path.dirname(dirname)
 
-  @classmethod
-  def make(cls, uri, obj, tag = None):
-    """
-    Construct a withdrawal PDU.
-    """
-
-    assert isinstance(obj, rpki.x509.uri_dispatch(uri))
-    return cls.make_pdu(uri = uri, tag = tag)
-
 
 class report_error_elt(rpki.xml_utils.text_elt, publication_namespace):
   """
diff --git a/rpki/relaxng.py b/rpki/relaxng.py
index d01f7e28..218edfc4 100644
--- a/rpki/relaxng.py
+++ b/rpki/relaxng.py
@@ -1806,29 +1806,30 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
     <value>3</value>
   </define>
   <!-- Top level PDU is either a query or a reply. -->
-  <start>
+  <start combine="choice">
     <element name="msg">
       <attribute name="version">
         <ref name="version"/>
       </attribute>
-      <choice>
-        <group>
-          <attribute name="type">
-            <value>query</value>
-          </attribute>
-          <zeroOrMore>
-            <ref name="query_elt"/>
-          </zeroOrMore>
-        </group>
-        <group>
-          <attribute name="type">
-            <value>reply</value>
-          </attribute>
-          <zeroOrMore>
-            <ref name="reply_elt"/>
-          </zeroOrMore>
-        </group>
-      </choice>
+      <attribute name="type">
+        <value>query</value>
+      </attribute>
+      <zeroOrMore>
+        <ref name="query_elt"/>
+      </zeroOrMore>
+    </element>
+  </start>
+  <start combine="choice">
+    <element name="msg">
+      <attribute name="version">
+        <ref name="version"/>
+      </attribute>
+      <attribute name="type">
+        <value>reply</value>
+      </attribute>
+      <zeroOrMore>
+        <ref name="reply_elt"/>
+      </zeroOrMore>
     </element>
   </start>
   <!-- PDUs allowed in  queries and replies. -->
@@ -1865,6 +1866,14 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
       </data>
     </attribute>
   </define>
+  <!-- Digest of objects being withdrawn -->
+  <define name="hash">
+    <attribute name="hash">
+      <data type="string">
+        <param name="pattern">[0-9a-fA-F]+</param>
+      </data>
+    </attribute>
+  </define>
   <!-- Error codes. -->
   <define name="error">
     <data type="token">
@@ -1872,16 +1881,19 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
     </data>
   </define>
   <!-- <publish/> element -->
-  <define name="publish_query" combine="choice">
+  <define name="publish_query">
     <element name="publish">
       <optional>
         <ref name="tag"/>
       </optional>
       <ref name="uri"/>
+      <optional>
+        <ref name="hash"/>
+      </optional>
       <ref name="base64"/>
     </element>
   </define>
-  <define name="publish_reply" combine="choice">
+  <define name="publish_reply">
     <element name="publish">
       <optional>
         <ref name="tag"/>
@@ -1890,15 +1902,16 @@ publication = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" e
     </element>
   </define>
   <!-- <withdraw/> element -->
-  <define name="withdraw_query" combine="choice">
+  <define name="withdraw_query">
     <element name="withdraw">
       <optional>
         <ref name="tag"/>
       </optional>
       <ref name="uri"/>
+      <ref name="hash"/>
     </element>
   </define>
-  <define name="withdraw_reply" combine="choice">
+  <define name="withdraw_reply">
     <element name="withdraw">
       <optional>
         <ref name="tag"/>
@@ -2162,24 +2175,35 @@ rrdp = lxml.etree.RelaxNG(lxml.etree.fromstring(r'''<?xml version="1.0" encoding
             <ref name="serial"/>
           </attribute>
           <oneOrMore>
-            <choice>
-              <element name="publish">
-                <attribute name="uri">
-                  <ref name="uri"/>
-                </attribute>
-                <ref name="base64"/>
-              </element>
-              <element name="withdraw">
-                <attribute name="uri">
-                  <ref name="uri"/>
-                </attribute>
-              </element>
-            </choice>
+            <ref name="delta_element"/>
           </oneOrMore>
         </element>
       </oneOrMore>
     </element>
   </start>
+  <define name="delta_element" combine="choice">
+    <element name="publish">
+      <attribute name="uri">
+        <ref name="uri"/>
+      </attribute>
+      <optional>
+        <attribute name="hash">
+          <ref name="hash"/>
+        </attribute>
+      </optional>
+      <ref name="base64"/>
+    </element>
+  </define>
+  <define name="delta_element" combine="choice">
+    <element name="withdraw">
+      <attribute name="uri">
+        <ref name="uri"/>
+      </attribute>
+      <attribute name="hash">
+        <ref name="hash"/>
+      </attribute>
+    </element>
+  </define>
 </grammar>
 <!--
   Local Variables:
diff --git a/rpki/rpkid.py b/rpki/rpkid.py
index ce16d832..45044ab8 100644
--- a/rpki/rpkid.py
+++ b/rpki/rpkid.py
@@ -867,6 +867,14 @@ class ca_detail_obj(rpki.sql.sql_persistent):
 
     return rpki.rpkid.ghostbuster_obj.sql_fetch_where(self.gctx, "ca_detail_id = %s", (self.ca_detail_id,))
 
+  def unpublished_ghostbusters(self, when):
+    """
+    Fetch all unpublished Ghostbusters objects linked to this
+    ca_detail with attempted publication dates older than when.
+    """
+
+    return rpki.rpkid.ghostbuster_obj.sql_fetch_where(self.gctx, "ca_detail_id = %s AND published IS NOT NULL and published < %s", (self.ca_detail_id, when))
+
   @property
   def ee_certificates(self):
     """
@@ -875,13 +883,13 @@ class ca_detail_obj(rpki.sql.sql_persistent):
 
     return rpki.rpkid.ee_cert_obj.sql_fetch_where(self.gctx, "ca_detail_id = %s", (self.ca_detail_id,))
 
-  def unpublished_ghostbusters(self, when):
+  def unpublished_ee_certificates(self, when):
     """
-    Fetch all unpublished Ghostbusters objects linked to this
+    Fetch all unpublished EE certificate objects linked to this
     ca_detail with attempted publication dates older than when.
     """
 
-    return rpki.rpkid.ghostbuster_obj.sql_fetch_where(self.gctx, "ca_detail_id = %s AND published IS NOT NULL and published < %s", (self.ca_detail_id, when))
+    return rpki.rpkid.ee_cert_obj.sql_fetch_where(self.gctx, "ca_detail_id = %s AND published IS NOT NULL and published < %s", (self.ca_detail_id, when))
 
   @property
   def crl_uri(self):
@@ -963,10 +971,10 @@ class ca_detail_obj(rpki.sql.sql_persistent):
     repository = ca.parent.repository
     handler = False if allow_failure else None
     for child_cert in self.child_certs:
-      publisher.withdraw(uri = child_cert.uri,
-                         obj = child_cert.cert,
-                         repository = repository,
-                         handler = handler)
+      publisher.queue(uri = child_cert.uri,
+                      old_obj = child_cert.cert,
+                      repository = repository,
+                      handler = handler)
       child_cert.sql_mark_deleted()
     for roa in self.roas:
       roa.revoke(publisher = publisher, allow_failure = allow_failure, fast = True)
@@ -977,19 +985,19 @@ class ca_detail_obj(rpki.sql.sql_persistent):
     except AttributeError:
       latest_manifest = None
     if latest_manifest is not None:
-      publisher.withdraw(uri = self.manifest_uri,
-                         obj = self.latest_manifest,
-                         repository = repository,
-                         handler = handler)
+      publisher.queue(uri = self.manifest_uri,
+                      old_obj = self.latest_manifest,
+                      repository = repository,
+                      handler = handler)
     try:
       latest_crl = self.latest_crl
     except AttributeError:
       latest_crl = None
     if latest_crl is not None:
-      publisher.withdraw(uri = self.crl_uri,
-                         obj = self.latest_crl,
-                         repository = repository,
-                         handler = handler)
+      publisher.queue(uri = self.crl_uri,
+                      old_obj = self.latest_crl,
+                      repository = repository,
+                      handler = handler)
     self.gctx.sql.sweep()
     for cert in self.revoked_certs:     # + self.child_certs
       logger.debug("Deleting %r", cert)
@@ -1199,6 +1207,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
       notAfter    = resources.valid_until)
 
     if child_cert is None:
+      old_cert = None
       child_cert = rpki.rpkid.child_cert_obj(
         gctx         = child.gctx,
         child_id     = child.child_id,
@@ -1206,6 +1215,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
         cert         = cert)
       logger.debug("Created new child_cert %r", child_cert)
     else:
+      old_cert = child_cert.cert
       child_cert.cert = cert
       del child_cert.ca_detail
       child_cert.ca_detail_id = self.ca_detail_id
@@ -1214,9 +1224,10 @@ class ca_detail_obj(rpki.sql.sql_persistent):
     child_cert.ski = cert.get_SKI()
     child_cert.published = rpki.sundial.now()
     child_cert.sql_store()
-    publisher.publish(
+    publisher.queue(
       uri = child_cert.uri,
-      obj = child_cert.cert,
+      old_obj = old_cert,
+      new_obj = child_cert.cert,
       repository = ca.parent.repository,
       handler = child_cert.published_callback)
     self.generate_manifest(publisher = publisher)
@@ -1247,6 +1258,8 @@ class ca_detail_obj(rpki.sql.sql_persistent):
         certlist.append((revoked_cert.serial, revoked_cert.revoked))
     certlist.sort()
 
+    old_crl = self.latest_crl
+
     self.latest_crl = rpki.x509.CRL.generate(
       keypair             = self.private_key_id,
       issuer              = self.latest_ca_cert,
@@ -1257,9 +1270,10 @@ class ca_detail_obj(rpki.sql.sql_persistent):
 
     self.crl_published = rpki.sundial.now()
     self.sql_mark_dirty()
-    publisher.publish(
+    publisher.queue(
       uri        = self.crl_uri,
-      obj        = self.latest_crl,
+      old_obj    = old_crl,
+      new_obj    = self.latest_crl,
       repository = parent.repository,
       handler    = self.crl_published_callback)
 
@@ -1304,6 +1318,7 @@ class ca_detail_obj(rpki.sql.sql_persistent):
     objs.extend((e.uri_tail, e.cert) for e in self.ee_certificates)
 
     logger.debug("Building manifest object %s", uri)
+    old_manifest = self.latest_manifest
     self.latest_manifest = rpki.x509.SignedManifest.build(
       serial         = ca.next_manifest_number(),
       thisUpdate     = now,
@@ -1316,10 +1331,11 @@ class ca_detail_obj(rpki.sql.sql_persistent):
 
     self.manifest_published = rpki.sundial.now()
     self.sql_mark_dirty()
-    publisher.publish(uri = uri,
-                      obj = self.latest_manifest,
-                      repository = parent.repository,
-                      handler = self.manifest_published_callback)
+    publisher.queue(uri = uri,
+                    old_obj = old_manifest,
+                    new_obj = self.latest_manifest,
+                    repository = parent.repository,
+                    handler = self.manifest_published_callback)
 
   def manifest_published_callback(self, pdu):
     """
@@ -1387,19 +1403,19 @@ class ca_detail_obj(rpki.sql.sql_persistent):
          self.crl_published is not None and \
          self.crl_published < stale:
       logger.debug("Retrying publication for %s", self.crl_uri)
-      publisher.publish(uri = self.crl_uri,
-                        obj = self.latest_crl,
-                        repository = repository,
-                        handler = self.crl_published_callback)
+      publisher.queue(uri = self.crl_uri,
+                      new_obj = self.latest_crl,
+                      repository = repository,
+                      handler = self.crl_published_callback)
 
     if self.latest_manifest is not None and \
          self.manifest_published is not None and \
          self.manifest_published < stale:
       logger.debug("Retrying publication for %s", self.manifest_uri)
-      publisher.publish(uri = self.manifest_uri,
-                        obj = self.latest_manifest,
-                        repository = repository,
-                        handler = self.manifest_published_callback)
+      publisher.queue(uri = self.manifest_uri,
+                      new_obj = self.latest_manifest,
+                      repository = repository,
+                      handler = self.manifest_published_callback)
 
     if not check_all:
       return
@@ -1409,28 +1425,37 @@ class ca_detail_obj(rpki.sql.sql_persistent):
 
     for child_cert in self.unpublished_child_certs(stale):
       logger.debug("Retrying publication for %s", child_cert)
-      publisher.publish(
+      publisher.queue(
         uri = child_cert.uri,
-        obj = child_cert.cert,
+        new_obj = child_cert.cert,
         repository = repository,
         handler = child_cert.published_callback)
 
     for roa in self.unpublished_roas(stale):
       logger.debug("Retrying publication for %s", roa)
-      publisher.publish(
+      publisher.queue(
         uri = roa.uri,
-        obj = roa.roa,
+        new_obj = roa.roa,
         repository = repository,
         handler = roa.published_callback)
 
     for ghostbuster in self.unpublished_ghostbusters(stale):
       logger.debug("Retrying publication for %s", ghostbuster)
-      publisher.publish(
+      publisher.queue(
         uri = ghostbuster.uri,
-        obj = ghostbuster.ghostbuster,
+        new_obj = ghostbuster.ghostbuster,
         repository = repository,
         handler = ghostbuster.published_callback)
 
+    for ee_cert in self.unpublished_ee_certificates(stale):
+      logger.debug("Retrying publication for %s", ee_cert)
+      publisher.queue(
+        uri = ee_cert.uri,
+        new_obj = ee_cert.cert,
+        repository = repository,
+        handler = ee_cert.published_callback)
+
+
 class child_cert_obj(rpki.sql.sql_persistent):
   """
   Certificate that has been issued to a child.
@@ -1517,9 +1542,9 @@ class child_cert_obj(rpki.sql.sql_persistent):
     ca = ca_detail.ca
     logger.debug("Revoking %r %r", self, self.uri)
     revoked_cert_obj.revoke(cert = self.cert, ca_detail = ca_detail)
-    publisher.withdraw(
-      uri        = self.uri,
-      obj        = self.cert,
+    publisher.queue(
+      uri = self.uri,
+      old_obj = self.cert,
       repository = ca.parent.repository)
     self.gctx.sql.sweep()
     self.sql_delete()
@@ -1927,9 +1952,9 @@ class roa_obj(rpki.sql.sql_persistent):
     self.sql_store()
 
     logger.debug("Generating %r URI %s", self, self.uri)
-    publisher.publish(
+    publisher.queue(
       uri = self.uri,
-      obj = self.roa,
+      new_obj = self.roa,
       repository = ca.parent.repository,
       handler = self.published_callback)
     if not fast:
@@ -1974,10 +1999,10 @@ class roa_obj(rpki.sql.sql_persistent):
 
     logger.debug("Withdrawing %r %s and revoking its EE cert", self, uri)
     rpki.rpkid.revoked_cert_obj.revoke(cert = cert, ca_detail = ca_detail)
-    publisher.withdraw(uri = uri,
-                       obj = roa,
-                       repository = ca_detail.ca.parent.repository,
-                       handler = False if allow_failure else None)
+    publisher.queue(uri = uri,
+                    old_obj = roa,
+                    repository = ca_detail.ca.parent.repository,
+                    handler = False if allow_failure else None)
 
     if not regenerate:
       self.sql_mark_deleted()
@@ -2136,9 +2161,9 @@ class ghostbuster_obj(rpki.sql.sql_persistent):
     self.sql_store()
 
     logger.debug("Generating Ghostbuster record %r", self.uri)
-    publisher.publish(
+    publisher.queue(
       uri = self.uri,
-      obj = self.ghostbuster,
+      new_obj = self.ghostbuster,
       repository = ca.parent.repository,
       handler = self.published_callback)
     if not fast:
@@ -2183,10 +2208,10 @@ class ghostbuster_obj(rpki.sql.sql_persistent):
 
     logger.debug("Withdrawing %r %s and revoking its EE cert", self, uri)
     rpki.rpkid.revoked_cert_obj.revoke(cert = cert, ca_detail = ca_detail)
-    publisher.withdraw(uri = uri,
-                       obj = ghostbuster,
-                       repository = ca_detail.ca.parent.repository,
-                       handler = False if allow_failure else None)
+    publisher.queue(uri = uri,
+                    old_obj = ghostbuster,
+                    repository = ca_detail.ca.parent.repository,
+                    handler = False if allow_failure else None)
 
     if not regenerate:
       self.sql_mark_deleted()
@@ -2341,11 +2366,11 @@ class ee_cert_obj(rpki.sql.sql_persistent):
       ca_detail_id = ca_detail.ca_detail_id,
       cert         = cert)
 
-    publisher.publish(
-      uri        = self.uri,
-      obj        = self.cert,
+    publisher.queue(
+      uri = self.uri,
+      new_obj = self.cert,
       repository = ca.parent.repository,
-      handler    = self.published_callback)
+      handler = self.published_callback)
 
     self.sql_store()
 
@@ -2364,9 +2389,9 @@ class ee_cert_obj(rpki.sql.sql_persistent):
     ca = ca_detail.ca
     logger.debug("Revoking %r %r", self, self.uri)
     revoked_cert_obj.revoke(cert = self.cert, ca_detail = ca_detail)
-    publisher.withdraw(uri = self.uri,
-                       obj = self.cert,
-                       repository = ca.parent.repository)
+    publisher.queue(uri = self.uri,
+                    old_obj = self.cert,
+                    repository = ca.parent.repository)
     self.gctx.sql.sweep()
     self.sql_delete()
     if generate_crl_and_manifest:
@@ -2448,11 +2473,12 @@ class ee_cert_obj(rpki.sql.sql_persistent):
 
     self.sql_mark_dirty()
 
-    publisher.publish(
-      uri        = self.uri,
-      obj        = self.cert,
+    publisher.queue(
+      uri = self.uri,
+      old_obj = old_cert,
+      new_obj = self.cert,
       repository = ca_detail.ca.parent.repository,
-      handler    = self.published_callback)
+      handler = self.published_callback)
 
     if must_revoke:
       revoked_cert_obj.revoke(cert = old_cert.cert, ca_detail = old_ca_detail)
@@ -2498,28 +2524,37 @@ class publication_queue(object):
     if self.replace:
       self.uris = {}
 
-  def _add(self, uri, obj, repository, handler, make_pdu):
+  def queue(self, uri, repository, handler = None, old_obj = None, new_obj = None):
+
+    assert old_obj is not None or new_obj is not None
+    assert old_obj is None or isinstance(old_obj, rpki.x509.uri_dispatch(uri))
+    assert new_obj is None or isinstance(new_obj, rpki.x509.uri_dispatch(uri))
+
     rid = id(repository)
     if rid not in self.repositories:
       self.repositories[rid] = repository
       self.msgs[rid] = rpki.publication.msg.query()
+
     if self.replace and uri in self.uris:
       logger.debug("Removing publication duplicate %r", self.uris[uri])
       self.msgs[rid].remove(self.uris.pop(uri))
-    pdu = make_pdu(uri = uri, obj = obj)
+
+    hash = None if old_obj is None else rpki.x509.sha256(old_obj.get_Base64()).encode("hex")
+
+    if new_obj is None:
+      pdu = rpki.publication.withdraw_elt.make_pdu(uri = uri, hash = hash)
+    else:
+      pdu = rpki.publication.publish_elt.make_pdu( uri = uri, hash = hash, payload = new_obj)
+
     if handler is not None:
       self.handlers[id(pdu)] = handler
       pdu.tag = id(pdu)
+
     self.msgs[rid].append(pdu)
+
     if self.replace:
       self.uris[uri] = pdu
 
-  def publish(self,  uri, obj, repository, handler = None):
-    return self._add(uri, obj, repository, handler, rpki.publication.publish_elt.make)
-
-  def withdraw(self, uri, obj, repository, handler = None):
-    return self._add(uri, obj, repository, handler, rpki.publication.withdraw_elt.make)
-
   def call_pubd(self, cb, eb):
     def loop(iterator, rid):
       logger.debug("Calling pubd[%r]", self.repositories[rid])
@@ -2534,5 +2569,5 @@ class publication_queue(object):
     return sum(len(self.msgs[rid]) for rid in self.repositories)
 
   def empty(self):
-    assert (not self.msgs) == (self.size == 0)
+    assert (not self.msgs) == (self.size == 0), "Assertion failure: not self.msgs: %r, self.size %r" % (not self.msgs, self.size)
     return not self.msgs
diff --git a/rpki/rpkid_tasks.py b/rpki/rpkid_tasks.py
index 49b5b968..8f652fa6 100644
--- a/rpki/rpkid_tasks.py
+++ b/rpki/rpkid_tasks.py
@@ -308,9 +308,9 @@ class UpdateChildrenTask(AbstractTask):
                          self.child.child_handle, child_cert.cert.gSKI(),
                          old_resources.valid_until, irdb_resources.valid_until)
             child_cert.sql_delete()
-            self.publisher.withdraw(
+            self.publisher.queue(
               uri = child_cert.uri,
-              obj = child_cert.cert,
+              old_obj = child_cert.cert,
               repository = ca.parent.repository)
             ca_detail.generate_manifest(publisher = self.publisher)
 
diff --git a/rpki/sql_schemas.py b/rpki/sql_schemas.py
index e3b74b52..7c7079c0 100644
--- a/rpki/sql_schemas.py
+++ b/rpki/sql_schemas.py
@@ -245,7 +245,7 @@ CREATE TABLE ee_cert (
 
 ## @var pubd
 ## SQL schema pubd
-pubd = '''-- $Id: pubd.sql 5884 2014-07-04 00:37:08Z sra $
+pubd = '''-- $Id: pubd.sql 5887 2014-07-07 23:13:35Z sra $
 
 -- Copyright (C) 2012--2014  Dragon Research Labs ("DRL")
 -- Portions copyright (C) 2009--2010  Internet Systems Consortium ("ISC")
diff --git a/rpki/x509.py b/rpki/x509.py
index bf0d33f0..44d9484e 100644
--- a/rpki/x509.py
+++ b/rpki/x509.py
@@ -82,24 +82,24 @@ def first_rsync_uri(xia):
         return uri
   return None
 
-def sha1(bytes):
+def sha1(data):
   """
   Calculate SHA-1 digest of some data.
   Convenience wrapper around rpki.POW.Digest class.
   """
 
   d = rpki.POW.Digest(rpki.POW.SHA1_DIGEST)
-  d.update(bytes)
+  d.update(data)
   return d.digest()
 
-def sha256(bytes):
+def sha256(data):
   """
   Calculate SHA-256 digest of some data.
   Convenience wrapper around rpki.POW.Digest class.
   """
 
   d = rpki.POW.Digest(rpki.POW.SHA256_DIGEST)
-  d.update(bytes)
+  d.update(data)
   return d.digest()
 
 
diff --git a/schemas/relaxng/publication.rnc b/schemas/relaxng/publication.rnc
index 3a519543..8c129546 100644
--- a/schemas/relaxng/publication.rnc
+++ b/schemas/relaxng/publication.rnc
@@ -44,10 +44,16 @@ version = "3"
 
 # Top level PDU is either a query or a reply.
 
-start = element msg { 
-  attribute version { version } , 
-  ( ( attribute type { "query" }, query_elt* ) |
-    ( attribute type { "reply" }, reply_elt* ) )
+start |= element msg {
+  attribute version { version },
+  attribute type    { "query" },
+  query_elt*
+}
+
+start |= element msg {
+  attribute version { version },
+  attribute type    { "reply" },
+  reply_elt*
 }
 
 # PDUs allowed in  queries and replies.
@@ -67,19 +73,23 @@ base64 = xsd:base64Binary
 
 uri = attribute uri { xsd:anyURI { maxLength="4096" } }
 
+# Digest of objects being withdrawn
+
+hash = attribute hash { xsd:string { pattern = "[0-9a-fA-F]+" } }
+
 # Error codes.
 
 error = xsd:token { maxLength="1024" }
 
 # <publish/> element
 
-publish_query |= element publish { tag?, uri, base64 }
-publish_reply |= element publish { tag?, uri }
+publish_query = element publish { tag?, uri, hash?, base64 }
+publish_reply = element publish { tag?, uri }
 
 # <withdraw/> element
 
-withdraw_query |= element withdraw { tag?, uri }
-withdraw_reply |= element withdraw { tag?, uri }
+withdraw_query = element withdraw { tag?, uri, hash }
+withdraw_reply = element withdraw { tag?, uri }
 
 # <report_error/> element
 
diff --git a/schemas/relaxng/publication.rng b/schemas/relaxng/publication.rng
index 2990ec97..fed494b2 100644
--- a/schemas/relaxng/publication.rng
+++ b/schemas/relaxng/publication.rng
@@ -43,29 +43,30 @@
     <value>3</value>
   </define>
   <!-- Top level PDU is either a query or a reply. -->
-  <start>
+  <start combine="choice">
     <element name="msg">
       <attribute name="version">
         <ref name="version"/>
       </attribute>
-      <choice>
-        <group>
-          <attribute name="type">
-            <value>query</value>
-          </attribute>
-          <zeroOrMore>
-            <ref name="query_elt"/>
-          </zeroOrMore>
-        </group>
-        <group>
-          <attribute name="type">
-            <value>reply</value>
-          </attribute>
-          <zeroOrMore>
-            <ref name="reply_elt"/>
-          </zeroOrMore>
-        </group>
-      </choice>
+      <attribute name="type">
+        <value>query</value>
+      </attribute>
+      <zeroOrMore>
+        <ref name="query_elt"/>
+      </zeroOrMore>
+    </element>
+  </start>
+  <start combine="choice">
+    <element name="msg">
+      <attribute name="version">
+        <ref name="version"/>
+      </attribute>
+      <attribute name="type">
+        <value>reply</value>
+      </attribute>
+      <zeroOrMore>
+        <ref name="reply_elt"/>
+      </zeroOrMore>
     </element>
   </start>
   <!-- PDUs allowed in  queries and replies. -->
@@ -102,6 +103,14 @@
       </data>
     </attribute>
   </define>
+  <!-- Digest of objects being withdrawn -->
+  <define name="hash">
+    <attribute name="hash">
+      <data type="string">
+        <param name="pattern">[0-9a-fA-F]+</param>
+      </data>
+    </attribute>
+  </define>
   <!-- Error codes. -->
   <define name="error">
     <data type="token">
@@ -109,16 +118,19 @@
     </data>
   </define>
   <!-- <publish/> element -->
-  <define name="publish_query" combine="choice">
+  <define name="publish_query">
     <element name="publish">
       <optional>
         <ref name="tag"/>
       </optional>
       <ref name="uri"/>
+      <optional>
+        <ref name="hash"/>
+      </optional>
       <ref name="base64"/>
     </element>
   </define>
-  <define name="publish_reply" combine="choice">
+  <define name="publish_reply">
     <element name="publish">
       <optional>
         <ref name="tag"/>
@@ -127,15 +139,16 @@
     </element>
   </define>
   <!-- <withdraw/> element -->
-  <define name="withdraw_query" combine="choice">
+  <define name="withdraw_query">
     <element name="withdraw">
       <optional>
         <ref name="tag"/>
       </optional>
       <ref name="uri"/>
+      <ref name="hash"/>
     </element>
   </define>
-  <define name="withdraw_reply" combine="choice">
+  <define name="withdraw_reply">
     <element name="withdraw">
       <optional>
         <ref name="tag"/>
diff --git a/schemas/relaxng/rrdp.rnc b/schemas/relaxng/rrdp.rnc
index 5b47c90b..2829605d 100644
--- a/schemas/relaxng/rrdp.rnc
+++ b/schemas/relaxng/rrdp.rnc
@@ -61,11 +61,21 @@ start |= element deltas {
   attribute to         { serial },
   element delta {
     attribute serial   { serial },
-    ( element publish  { attribute uri { uri }, base64 } |
-      element withdraw { attribute uri { uri }         } )+
+    delta_element+
   }+
 }
 
+delta_element |= element publish  {
+  attribute uri  { uri },
+  attribute hash { hash }?,
+  base64
+}
+
+delta_element |= element withdraw {
+  attribute uri  { uri },
+  attribute hash { hash }
+}
+
 # Local Variables:
 # indent-tabs-mode: nil
 # comment-start: "# "
diff --git a/schemas/relaxng/rrdp.rng b/schemas/relaxng/rrdp.rng
index 197a15b3..27ba741d 100644
--- a/schemas/relaxng/rrdp.rng
+++ b/schemas/relaxng/rrdp.rng
@@ -124,24 +124,35 @@
             <ref name="serial"/>
           </attribute>
           <oneOrMore>
-            <choice>
-              <element name="publish">
-                <attribute name="uri">
-                  <ref name="uri"/>
-                </attribute>
-                <ref name="base64"/>
-              </element>
-              <element name="withdraw">
-                <attribute name="uri">
-                  <ref name="uri"/>
-                </attribute>
-              </element>
-            </choice>
+            <ref name="delta_element"/>
           </oneOrMore>
         </element>
       </oneOrMore>
     </element>
   </start>
+  <define name="delta_element" combine="choice">
+    <element name="publish">
+      <attribute name="uri">
+        <ref name="uri"/>
+      </attribute>
+      <optional>
+        <attribute name="hash">
+          <ref name="hash"/>
+        </attribute>
+      </optional>
+      <ref name="base64"/>
+    </element>
+  </define>
+  <define name="delta_element" combine="choice">
+    <element name="withdraw">
+      <attribute name="uri">
+        <ref name="uri"/>
+      </attribute>
+      <attribute name="hash">
+        <ref name="hash"/>
+      </attribute>
+    </element>
+  </define>
 </grammar>
 <!--
   Local Variables: