Add "history" mode to versioned_writes middleware

This change introduces the concept of a "versioning mode" for
versioned_writes. The following modes are supported:

 * stack

    When deleting, check whether any previous versions exist in the
    versions container. If none is found, the object is deleted. If the
    most-recent version in the versions container is not a delete
    marker, it is copied into the versioned container (overwriting the
    current version if one exists) and then deleted from the versions
    container. This preserves the previous behavior.

    If the most-recent version in the versions container is a delete
    marker and a current version exists in the versioned container, the
    current version is deleted. If the most-recent version in the
    versions container is a delete marker and no current version exists
    in the versioned container, we copy the next-most-recent version
    from the versions container into the versioned container (assuming
    it exists and is not a delete marker) and delete both the
    most-recent version (i.e., the delete marker) and the just-copied
    next-most-recent version from the versions container.

    With this mode, DELETEs to versioned containers "undo" operations
    on containers. Previously this was limited to undoing PUTs, but now
    it will also undo DELETEs performed while in "history" mode.

 * history

    When deleting, check whether a current version exists in the
    versioned container. If one is found, it is copied to the versions
    container. Then an empty "delete marker" object is also put into the
    versions container; this records when the object was deleted.
    Finally, the original current version is deleted from the versioned
    container. As a result, subsequent GETs or HEADs will return a 404,
    and container listings for the versioned container do not include
    the object.

    With this mode, DELETEs to versioned containers behave like DELETEs
    to other containers, but with a history of what has happened.

Clients may specify (via a new X-Versions-Mode header) which mode a
container should use. By default, the existing "stack" mode is used.

Upgrade consideration:
======================

Clients should not use the "history" mode until all proxies in the
cluster have been upgraded. Attempting to use the "history" mode during
a rolling upgrade may result in some requests being served by proxies
running old code (which necessarily uses the "stack" mode), leading to
data loss.

Change-Id: I555dc17fefd0aa9ade681aa156da24e018ebe74b

api-ref/source/parameters.yaml

@@ -720,21 +720,24 @@ X-Trans-Id-Extra:
   type: string
 X-Versions-Location:
   description: |
-    Enables versioning on this container. The value
+    The URL-encoded UTF-8 representation of the container that stores
-    is the name of another container. You must UTF-8-encode and then
+    previous versions of objects. If not set, versioning is disabled
-    URL-encode the name before you include it in the header. To
+    for this container. For more information about object versioning,
-    disable versioning, set the header to an empty string.
+    see `Object versioning <http://docs.openstack.org/developer/
+    swift/api/object_versioning.html>`_.
   in: header
   required: false
   type: string
-X-Versions-Location_1:
+X-Versions-Mode:
   description: |
-    Enables versioning on this container. The value
+    The versioning mode for this container. The value must be either
-    is the name of another container. You must UTF-8-encode and then
+    ``stack`` or ``history``. If not set, ``stack`` mode will be used.
-    URL-encode the name before you include it in the header. To
+    This setting has no impact unless ``X-Versions-Location`` is set
-    disable versioning, set the header to an empty string.
+    for the container. For more information about object versioning,
+    see `Object versioning <http://docs.openstack.org/developer/
+    swift/api/object_versioning.html>`_.
   in: header
-  required: true
+  required: false
   type: string
 
 # variables in path

api-ref/source/storage-container-services.inc

@@ -172,6 +172,7 @@ Request
    - X-Container-Sync-To: X-Container-Sync-To
    - X-Container-Sync-Key: X-Container-Sync-Key
    - X-Versions-Location: X-Versions-Location
+   - X-Versions-Mode: X-Versions-Mode
    - X-Container-Meta-name: X-Container-Meta-name
    - X-Container-Meta-Access-Control-Allow-Origin: X-Container-Meta-Access-Control-Allow-Origin
    - X-Container-Meta-Access-Control-Max-Age: X-Container-Meta-Access-Control-Max-Age
@@ -302,6 +303,7 @@ Request
    - X-Container-Sync-To: X-Container-Sync-To
    - X-Container-Sync-Key: X-Container-Sync-Key
    - X-Versions-Location: X-Versions-Location
+   - X-Versions-Mode: X-Versions-Mode
    - X-Remove-Versions-Location: X-Remove-Versions-Location
    - X-Container-Meta-name: X-Container-Meta-name
    - X-Container-Meta-Access-Control-Allow-Origin: X-Container-Meta-Access-Control-Allow-Origin
@@ -409,6 +411,7 @@ Response Parameters
    - Content-Type: Content-Type
    - X-Container-Meta-Quota-Bytes: X-Container-Meta-Quota-Bytes
    - X-Versions-Location: X-Versions-Location
+   - X-Versions-Mode: X-Versions-Mode
 
 
 

doc/source/api/object_versioning.rst

@@ -6,19 +6,19 @@ You can store multiple versions of your content so that you can recover
 from unintended overwrites. Object versioning is an easy way to
 implement version control, which you can use with any type of content.
 
-Note
+.. note::
-~~~~
+    You cannot version a large-object manifest file, but the large-object
+    manifest file can point to versioned segments.
 
-You cannot version a large-object manifest file, but the large-object
+.. note::
-manifest file can point to versioned segments.
+    It is strongly recommended that you put non-current objects in a
+    different container than the container where current object versions
+    reside.
 
-It is strongly recommended that you put non-current objects in a
+To allow object versioning within a cluster, the cloud provider should add the
-different container than the container where current object versions
+``versioned_writes`` filter to the pipeline and set the
-reside.
+``allow_versioned_writes`` option to ``true`` in the
-
+``[filter:versioned_writes]`` section of the proxy-server configuration file.
-To enable object versioning, the cloud provider sets the
-``allow_versions`` option to ``TRUE`` in the container configuration
-file.
 
 The ``X-Versions-Location`` header defines the
 container that holds the non-current versions of your objects. You
@@ -29,13 +29,21 @@ object versioning for all objects in the container. With a comparable
 container automatically create non-current versions in the ``archive``
 container.
 
-Here's an example:
+The ``X-Versions-Mode`` header defines the behavior of ``DELETE`` requests to
+objects in the versioned container. In the default ``stack`` mode, deleting an
+object will restore the most-recent version from the ``archive`` container,
+overwriting the curent version. Alternatively you may specify ``history``
+mode, where deleting an object will copy the current version to the
+``archive`` then remove it from the ``current`` container.
+
+Example Using ``stack`` Mode
+----------------------------
 
 #.   Create the ``current`` container:
 
    .. code::
 
-       # curl -i $publicURL/current -X PUT -H "Content-Length: 0" -H "X-Auth-Token: $token" -H "X-Versions-Location: archive"
+       # curl -i $publicURL/current -X PUT -H "Content-Length: 0" -H "X-Auth-Token: $token" -H "X-Versions-Location: archive" -H "X-Versions-Mode: stack"
 
    .. code::
 
@@ -70,7 +78,7 @@ Here's an example:
 
    .. code::
 
-       <length><object_name><timestamp>
+       <length><object_name>/<timestamp>
 
    Where ``length`` is the 3-character, zero-padded hexadecimal
    character length of the object, ``<object_name>`` is the object name,
@@ -117,9 +125,7 @@ Here's an example:
 
        009my_object/1390512682.92052
 
-Note
+   .. note::
-~~~~
-
       A **POST** request to a versioned object updates only the metadata
       for the object and does not create a new version of the object. New
       versions are created only when the content of the object changes.
@@ -163,15 +169,157 @@ Note
    on it. If want to completely remove an object and you have five
    versions of it, you must **DELETE** it five times.
 
-#. To disable object versioning for the ``current`` container, remove
+Example Using ``history`` Mode
-   its ``X-Versions-Location`` metadata header by sending an empty key
+----------------------------
-   value.
+
+#.   Create the ``current`` container:
 
    .. code::
 
+       # curl -i $publicURL/current -X PUT -H "Content-Length: 0" -H "X-Auth-Token: $token" -H "X-Versions-Location: archive" -H "X-Versions-Mode: history"
+
+   .. code::
+
+       HTTP/1.1 201 Created
+       Content-Length: 0
+       Content-Type: text/html; charset=UTF-8
+       X-Trans-Id: txb91810fb717347d09eec8-0052e18997
+       Date: Thu, 23 Jan 2014 21:28:55 GMT
+
+#. Create the first version of an object in the ``current`` container:
+
+   .. code::
+
+       # curl -i $publicURL/current/my_object --data-binary 1 -X PUT -H "Content-Length: 0" -H "X-Auth-Token: $token"
+
+   .. code::
+
+       HTTP/1.1 201 Created
+       Last-Modified: Thu, 23 Jan 2014 21:31:22 GMT
+       Content-Length: 0
+       Etag: d41d8cd98f00b204e9800998ecf8427e
+       Content-Type: text/html; charset=UTF-8
+       X-Trans-Id: tx5992d536a4bd4fec973aa-0052e18a2a
+       Date: Thu, 23 Jan 2014 21:31:22 GMT
+
+   Nothing is written to the non-current version container when you
+   initially **PUT** an object in the ``current`` container. However,
+   subsequent **PUT** requests that edit an object trigger the creation
+   of a version of that object in the ``archive`` container.
+
+   These non-current versions are named as follows:
+
+   .. code::
+
+       <length><object_name>/<timestamp>
+
+   Where ``length`` is the 3-character, zero-padded hexadecimal
+   character length of the object, ``<object_name>`` is the object name,
+   and ``<timestamp>`` is the time when the object was initially created
+   as a current version.
+
+#. Create a second version of the object in the ``current`` container:
+
+   .. code::
+
+       # curl -i $publicURL/current/my_object --data-binary 2 -X PUT -H "Content-Length: 0" -H "X-Auth-Token: $token"
+
+   .. code::
+
+       HTTP/1.1 201 Created
+       Last-Modified: Thu, 23 Jan 2014 21:41:32 GMT
+       Content-Length: 0
+       Etag: d41d8cd98f00b204e9800998ecf8427e
+       Content-Type: text/html; charset=UTF-8
+       X-Trans-Id: tx468287ce4fc94eada96ec-0052e18c8c
+       Date: Thu, 23 Jan 2014 21:41:32 GMT
+
+#. Issue a **GET** request to a versioned object to get the current
+   version of the object. You do not have to do any request redirects or
+   metadata lookups.
+
+   List older versions of the object in the ``archive`` container:
+
+   .. code::
+
+       # curl -i $publicURL/archive?prefix=009my_object -X GET -H "X-Auth-Token: $token"
+
+   .. code::
+
+       HTTP/1.1 200 OK
+       Content-Length: 30
+       X-Container-Object-Count: 1
+       Accept-Ranges: bytes
+       X-Timestamp: 1390513280.79684
+       X-Container-Bytes-Used: 0
+       Content-Type: text/plain; charset=utf-8
+       X-Trans-Id: tx9a441884997542d3a5868-0052e18d8e
+       Date: Thu, 23 Jan 2014 21:45:50 GMT
+
+       009my_object/1390512682.92052
+
+   .. note::
+      A **POST** request to a versioned object updates only the metadata
+      for the object and does not create a new version of the object. New
+      versions are created only when the content of the object changes.
+
+#. Issue a **DELETE** request to a versioned object to copy the
+   current version of the object to the archive container then delete it from
+   the current container. Subsequent **GET** requests to the object in the
+   current container will return 404 Not Found.
+
+   .. code::
+
+       # curl -i $publicURL/current/my_object -X DELETE -H "X-Auth-Token: $token"
+
+   .. code::
+
+       HTTP/1.1 204 No Content
+       Content-Length: 0
+       Content-Type: text/html; charset=UTF-8
+       X-Trans-Id: tx006d944e02494e229b8ee-0052e18edd
+       Date: Thu, 23 Jan 2014 21:51:25 GMT
+
+   List older versions of the object in the ``archive`` container::
+
+   .. code::
+
+       # curl -i $publicURL/archive?prefix=009my_object -X GET -H "X-Auth-Token: $token"
+
+   .. code::
+
+       HTTP/1.1 200 OK
+       Content-Length: 90
+       X-Container-Object-Count: 3
+       Accept-Ranges: bytes
+       X-Timestamp: 1390513280.79684
+       X-Container-Bytes-Used: 0
+       Content-Type: text/html; charset=UTF-8
+       X-Trans-Id: tx044f2a05f56f4997af737-0052e18eed
+       Date: Thu, 23 Jan 2014 21:51:41 GMT
+
+       009my_object/1390512682.92052
+       009my_object/1390512692.23062
+       009my_object/1390513885.67732
+
+   In addition to the two previous versions of the object, the archive
+   container has a "delete marker" to record when the object was deleted.
+
+   To permanently delete a previous version, issue a **DELETE** to the version
+   in the archive container.
+
+Disabling Object Versioning
+---------------------------
+
+To disable object versioning for the ``current`` container, remove
+its ``X-Versions-Location`` metadata header by sending an empty key
+value.
+
+.. code::
+
     # curl -i $publicURL/current -X PUT -H "Content-Length: 0" -H "X-Auth-Token: $token" -H "X-Versions-Location: "
 
-   .. code::
+.. code::
 
     HTTP/1.1 202 Accepted
     Content-Length: 76

swift/common/middleware/versioned_writes.py

@@ -17,14 +17,17 @@
 Object versioning in swift is implemented by setting a flag on the container
 to tell swift to version all objects in the container. The flag is the
 ``X-Versions-Location`` header on the container, and its value is the
-container where the versions are stored. It is recommended to use a different
+container where the versions are stored.
-``X-Versions-Location`` container for each container that is being versioned.
+
+.. note::
+    It is recommended to use a different ``X-Versions-Location`` container for
+    each container that is being versioned.
 
 When data is ``PUT`` into a versioned container (a container with the
 versioning flag turned on), the existing data in the file is redirected to a
 new object and the data in the ``PUT`` request is saved as the data for the
 versioned object. The new object name (for the previous version) is
-``<versions_container>/<length><object_name>/<timestamp>``, where ``length``
+``<archive_container>/<length><object_name>/<timestamp>``, where ``length``
 is the 3-character zero-padded hexadecimal length of the ``<object_name>`` and
 ``<timestamp>`` is the timestamp of when the previous version was created.
 
@@ -35,9 +38,39 @@ A ``POST`` to a versioned object will update the object metadata as normal,
 but will not create a new version of the object. In other words, new versions
 are only created when the content of the object changes.
 
-A ``DELETE`` to a versioned object will only remove the current version of the
+A ``DELETE`` to a versioned object will be handled in one of two ways,
-object. If you have 5 total versions of the object, you must delete the
+depending on the value of a ``X-Versions-Mode`` header set on the container.
-object 5 times to completely remove the object.
+The available modes are:
+
+  * ``stack``
+
+    Only remove the current version of the object. If any previous versions
+    exist in the archive container, the most recent one is copied over the
+    current version, and the copy in the archive container is deleted. As a
+    result, if you have 5 total versions of the object, you must delete the
+    object 5 times to completely remove the object. This is the default
+    behavior if ``X-Versions-Mode`` has not been set for the container.
+
+  * ``history``
+
+    Copy the current version of the object to the archive container, write
+    a zero-byte "delete marker" object that notes when the delete took place,
+    and delete the object from the versioned container. The object will no
+    longer appear in container listings for the versioned container and future
+    requests there will return 404 Not Found. However, the content will still
+    be recoverable from the archive container.
+
+.. note::
+    While it is possible to switch between 'stack' and 'history' mode on a
+    container, it is not recommended.
+
+To restore a previous version of an object, find the desired version in the
+archive container then issue a ``COPY`` with a ``Destination`` header
+indicating the original location. This will retain a copy of the current
+version similar to a ``PUT`` over the versioned object. Additionally, if the
+container is in ``stack`` mode and the client wishes to permanently delete the
+current version, it may issue a ``DELETE`` to the versioned object as
+described above.
 
 --------------------------------------------------
 How to Enable Object Versioning in a Swift Cluster
@@ -57,23 +90,31 @@ set ``allow_versioned_writes`` to ``True`` in the middleware options
 to enable the information about this middleware to be returned in a /info
 request.
 
-Upgrade considerations: If ``allow_versioned_writes`` is set in the filter
+Upgrade considerations:
-configuration, you can leave the ``allow_versions`` flag in the container
++++++++++++++++++++++++
-server configuration files untouched. If you decide to disable or remove the
-``allow_versions`` flag, you must re-set any existing containers that had
-the 'X-Versions-Location' flag configured so that it can now be tracked by the
-versioned_writes middleware.
 
------------------------
+If ``allow_versioned_writes`` is set in the filter configuration, you can leave
-Examples Using ``curl``
+the ``allow_versions`` flag in the container server configuration files
------------------------
+untouched. If you decide to disable or remove the ``allow_versions`` flag, you
+must re-set any existing containers that had the 'X-Versions-Location' flag
+configured so that it can now be tracked by the versioned_writes middleware.
+
+Clients should not use the 'history' mode until all proxies in the cluster
+have been upgraded to a version of Swift that supports it. Attempting to use
+the 'history' mode during a rolling upgrade may result in some requests being
+served by proxies running old code (which necessarily uses the 'stack' mode),
+leading to data loss.
+
+-------------------------------------------
+Examples Using ``curl`` with ``stack`` Mode
+-------------------------------------------
 
 First, create a container with the ``X-Versions-Location`` header or add the
 header to an existing container. Also make sure the container referenced by
 the ``X-Versions-Location`` exists. In this example, the name of that
 container is "versions"::
 
-    curl -i -XPUT -H "X-Auth-Token: <token>" \
+    curl -i -XPUT -H "X-Auth-Token: <token>" -H "X-Versions-Mode: stack" \
 -H "X-Versions-Location: versions" http://<storage_url>/container
     curl -i -XPUT -H "X-Auth-Token: <token>" http://<storage_url>/versions
 
@@ -102,6 +143,59 @@ http://<storage_url>/versions?prefix=008myobject/
     curl -i -XGET -H "X-Auth-Token: <token>" \
 http://<storage_url>/container/myobject
 
+---------------------------------------------
+Examples Using ``curl`` with ``history`` Mode
+---------------------------------------------
+
+As above, create a container with the ``X-Versions-Location`` header and ensure
+that the container referenced by the ``X-Versions-Location`` exists. In this
+example, the name of that container is "versions"::
+
+    curl -i -XPUT -H "X-Auth-Token: <token>" -H "X-Versions-Mode: history" \
+-H "X-Versions-Location: versions" http://<storage_url>/container
+    curl -i -XPUT -H "X-Auth-Token: <token>" http://<storage_url>/versions
+
+Create an object (the first version)::
+
+    curl -i -XPUT --data-binary 1 -H "X-Auth-Token: <token>" \
+http://<storage_url>/container/myobject
+
+Now create a new version of that object::
+
+    curl -i -XPUT --data-binary 2 -H "X-Auth-Token: <token>" \
+http://<storage_url>/container/myobject
+
+Now delete the current version of the object. Subsequent requests will 404::
+
+    curl -i -XDELETE -H "X-Auth-Token: <token>" \
+http://<storage_url>/container/myobject
+    curl -i -H "X-Auth-Token: <token>" \
+http://<storage_url>/container/myobject
+
+A listing of the older versions of the object will include both the first and
+second versions of the object, as well as a "delete marker" object::
+
+    curl -i -H "X-Auth-Token: <token>" \
+http://<storage_url>/versions?prefix=008myobject/
+
+To restore a previous version, simply ``COPY`` it from the archive container::
+
+    curl -i -XCOPY -H "X-Auth-Token: <token>" \
+http://<storage_url>/versions/008myobject/<timestamp> \
+-H "Destination: container/myobject"
+
+Note that the archive container still has all previous versions of the object,
+including the source for the restore::
+
+    curl -i -H "X-Auth-Token: <token>" \
+http://<storage_url>/versions?prefix=008myobject/
+
+To permanently delete a previous version, ``DELETE`` it from the archive
+container::
+
+    curl -i -XDELETE -H "X-Auth-Token: <token>" \
+http://<storage_url>/versions/008myobject/<timestamp> \
+
 ---------------------------------------------------
 How to Disable Object Versioning in a Swift Cluster
 ---------------------------------------------------
@@ -132,11 +226,19 @@ from swift.proxy.controllers.base import get_container_info
 from swift.common.http import (
     is_success, is_client_error, HTTP_NOT_FOUND)
 from swift.common.swob import HTTPPreconditionFailed, HTTPServiceUnavailable, \
-    HTTPServerError
+    HTTPServerError, HTTPBadRequest
 from swift.common.exceptions import (
     ListingIterNotFound, ListingIterError)
 
 
+VERSIONING_MODES = ('stack', 'history')
+DELETE_MARKER_CONTENT_TYPE = 'application/x-deleted;swift_versions_deleted=1'
+VERSIONS_LOC_CLIENT = 'x-versions-location'
+VERSIONS_LOC_SYSMETA = get_sys_meta_prefix('container') + 'versions-location'
+VERSIONS_MODE_CLIENT = 'x-versions-mode'
+VERSIONS_MODE_SYSMETA = get_sys_meta_prefix('container') + 'versions-mode'
+
+
 class VersionedWritesContext(WSGIContext):
 
     def __init__(self, wsgi_app, logger):
@@ -298,6 +400,48 @@ class VersionedWritesContext(WSGIContext):
         # could not version the data, bail
         raise HTTPServiceUnavailable(request=req)
 
+    def _build_versions_object_prefix(self, object_name):
+        return '%03x%s/' % (
+            len(object_name),
+            object_name)
+
+    def _build_versions_object_name(self, object_name, ts):
+        return ''.join((
+            self._build_versions_object_prefix(object_name),
+            Timestamp(ts).internal))
+
+    def _copy_current(self, req, versions_cont, api_version, account_name,
+                      object_name):
+        get_resp = self._get_source_object(req, req.path_info)
+
+        if 'X-Object-Manifest' in get_resp.headers:
+            # do not version DLO manifest, proceed with original request
+            close_if_possible(get_resp.app_iter)
+            return
+        if get_resp.status_int == HTTP_NOT_FOUND:
+            # nothing to version, proceed with original request
+            close_if_possible(get_resp.app_iter)
+            return
+
+        # check for any other errors
+        self._check_response_error(req, get_resp)
+
+        # if there's an existing object, then copy it to
+        # X-Versions-Location
+        ts_source = get_resp.headers.get(
+            'x-timestamp',
+            calendar.timegm(time.strptime(
+                get_resp.headers['last-modified'],
+                '%a, %d %b %Y %H:%M:%S GMT')))
+        vers_obj_name = self._build_versions_object_name(
+            object_name, ts_source)
+
+        put_path_info = "/%s/%s/%s/%s" % (
+            api_version, account_name, versions_cont, vers_obj_name)
+        put_resp = self._put_versioned_obj(req, put_path_info, get_resp)
+
+        self._check_response_error(req, put_resp)
+
     def handle_obj_versions_put(self, req, versions_cont, api_version,
                                 account_name, object_name):
         """
@@ -315,41 +459,77 @@ class VersionedWritesContext(WSGIContext):
             # do not version DLO manifest, proceed with original request
             return self.app
 
-        get_resp = self._get_source_object(req, req.path_info)
+        self._copy_current(req, versions_cont, api_version, account_name,
-
+                           object_name)
-        if 'X-Object-Manifest' in get_resp.headers:
-            # do not version DLO manifest, proceed with original request
-            close_if_possible(get_resp.app_iter)
         return self.app
+
+    def handle_obj_versions_delete_push(self, req, versions_cont, api_version,
+                                        account_name, container_name,
+                                        object_name):
+        """
+        Handle DELETE requests when in history mode.
+
+        Copy current version of object to versions_container and write a
+        delete marker before proceding with original request.
+
+        :param req: original request.
+        :param versions_cont: container where previous versions of the object
+                              are stored.
+        :param api_version: api version.
+        :param account_name: account name.
+        :param object_name: name of object of original request
+        """
+        self._copy_current(req, versions_cont, api_version, account_name,
+                           object_name)
+
+        marker_path = "/%s/%s/%s/%s" % (
+            api_version, account_name, versions_cont,
+            self._build_versions_object_name(object_name, time.time()))
+        marker_headers = {
+            # Definitive source of truth is Content-Type, and since we add
+            # a swift_* param, we know users haven't set it themselves.
+            # This is still open to users POSTing to update the content-type
+            # but they're just shooting themselves in the foot then.
+            'content-type': DELETE_MARKER_CONTENT_TYPE,
+            'content-length': '0',
+            'x-auth-token': req.headers.get('x-auth-token')}
+        marker_req = make_pre_authed_request(
+            req.environ, path=marker_path,
+            headers=marker_headers, method='PUT', swift_source='VW')
+        marker_req.environ['swift.content_type_overridden'] = True
+        marker_resp = marker_req.get_response(self.app)
+        self._check_response_error(req, marker_resp)
+
+        # successfully copied and created delete marker; safe to delete
+        return self.app
+
+    def _restore_data(self, req, versions_cont, api_version, account_name,
+                      container_name, object_name, prev_obj_name):
+        get_path = "/%s/%s/%s/%s" % (
+            api_version, account_name, versions_cont, prev_obj_name)
+
+        get_resp = self._get_source_object(req, get_path)
+
+        # if the version isn't there, keep trying with previous version
         if get_resp.status_int == HTTP_NOT_FOUND:
-            # nothing to version, proceed with original request
+            return False
-            close_if_possible(get_resp.app_iter)
-            return self.app
 
-        # check for any other errors
         self._check_response_error(req, get_resp)
 
-        # if there's an existing object, then copy it to
-        # X-Versions-Location
-        prefix_len = '%03x' % len(object_name)
-        lprefix = prefix_len + object_name + '/'
-        ts_source = get_resp.headers.get(
-            'x-timestamp',
-            calendar.timegm(time.strptime(
-                get_resp.headers['last-modified'],
-                '%a, %d %b %Y %H:%M:%S GMT')))
-        vers_obj_name = lprefix + Timestamp(ts_source).internal
-
         put_path_info = "/%s/%s/%s/%s" % (
-            api_version, account_name, versions_cont, vers_obj_name)
+            api_version, account_name, container_name, object_name)
-        put_resp = self._put_versioned_obj(req, put_path_info, get_resp)
+        put_resp = self._put_versioned_obj(
+            req, put_path_info, get_resp)
 
         self._check_response_error(req, put_resp)
-        return self.app
+        return get_path
 
-    def handle_obj_versions_delete(self, req, versions_cont, api_version,
+    def handle_obj_versions_delete_pop(self, req, versions_cont, api_version,
-                                   account_name, container_name, object_name):
+                                       account_name, container_name,
+                                       object_name):
         """
+        Handle DELETE requests when in stack mode.
+
         Delete current version of object and pop previous version in its place.
 
         :param req: original request.
@@ -360,12 +540,11 @@ class VersionedWritesContext(WSGIContext):
         :param container_name: container name.
         :param object_name: object name.
         """
-        prefix_len = '%03x' % len(object_name)
+        listing_prefix = self._build_versions_object_prefix(object_name)
-        lprefix = prefix_len + object_name + '/'
+        item_iter = self._listing_iter(account_name, versions_cont,
-
+                                       listing_prefix, req)
-        item_iter = self._listing_iter(account_name, versions_cont, lprefix,
-                                       req)
 
+        auth_token_header = {'X-Auth-Token': req.headers.get('X-Auth-Token')}
         authed = False
         for previous_version in item_iter:
             if not authed:
@@ -380,33 +559,66 @@ class VersionedWritesContext(WSGIContext):
                         return aresp
                     authed = True
 
+            if previous_version['content_type'] == DELETE_MARKER_CONTENT_TYPE:
+                # check whether we have data in the versioned container
+                obj_head_headers = {'X-Newest': 'True'}
+                obj_head_headers.update(auth_token_header)
+                head_req = make_pre_authed_request(
+                    req.environ, path=req.path_info, method='HEAD',
+                    headers=obj_head_headers, swift_source='VW')
+                hresp = head_req.get_response(self.app)
+
+                if hresp.status_int != HTTP_NOT_FOUND:
+                    self._check_response_error(req, hresp)
+                    # if there's an existing object, then just let the delete
+                    # through (i.e., restore to the delete-marker state):
+                    break
+
+                # no data currently in the container (delete marker is current)
+                for version_to_restore in item_iter:
+                    if version_to_restore['content_type'] == \
+                            DELETE_MARKER_CONTENT_TYPE:
+                        # Nothing to restore
+                        break
+                    prev_obj_name = version_to_restore['name'].encode('utf-8')
+                    restored_path = self._restore_data(
+                        req, versions_cont, api_version, account_name,
+                        container_name, object_name, prev_obj_name)
+                    if not restored_path:
+                        continue
+
+                    old_del_req = make_pre_authed_request(
+                        req.environ, path=restored_path, method='DELETE',
+                        headers=auth_token_header, swift_source='VW')
+                    del_resp = old_del_req.get_response(self.app)
+                    if del_resp.status_int != HTTP_NOT_FOUND:
+                        self._check_response_error(req, del_resp)
+                        # else, well, it existed long enough to do the
+                        # copy; we won't worry too much
+                    break
+                marker_path = "/%s/%s/%s/%s" % (
+                    api_version, account_name, versions_cont,
+                    previous_version['name'].encode('utf-8'))
+                # done restoring, redirect the delete to the marker
+                req = make_pre_authed_request(
+                    req.environ, path=marker_path, method='DELETE',
+                    headers=auth_token_header, swift_source='VW')
+            else:
                 # there are older versions so copy the previous version to the
                 # current object and delete the previous version
                 prev_obj_name = previous_version['name'].encode('utf-8')
-
+                restored_path = self._restore_data(
-            get_path = "/%s/%s/%s/%s" % (
+                    req, versions_cont, api_version, account_name,
-                api_version, account_name, versions_cont, prev_obj_name)
+                    container_name, object_name, prev_obj_name)
-
+                if not restored_path:
-            get_resp = self._get_source_object(req, get_path)
-
-            # if the version isn't there, keep trying with previous version
-            if get_resp.status_int == HTTP_NOT_FOUND:
                     continue
 
-            self._check_response_error(req, get_resp)
-
-            put_path_info = "/%s/%s/%s/%s" % (
-                api_version, account_name, container_name, object_name)
-            put_resp = self._put_versioned_obj(req, put_path_info, get_resp)
-
-            self._check_response_error(req, put_resp)
-
                 # redirect the original DELETE to the source of the reinstated
                 # version object - we already auth'd original req so make a
                 # pre-authed request
                 req = make_pre_authed_request(
-                req.environ, path=get_path, method='DELETE',
+                    req.environ, path=restored_path, method='DELETE',
-                swift_source='VW')
+                    headers=auth_token_header, swift_source='VW')
 
             # remove 'X-If-Delete-At', since it is not for the older copy
             if 'X-If-Delete-At' in req.headers:
@@ -420,15 +632,19 @@ class VersionedWritesContext(WSGIContext):
         app_resp = self._app_call(env)
         if self._response_headers is None:
             self._response_headers = []
-        sysmeta_version_hdr = get_sys_meta_prefix('container') + \
+        mode = location = ''
-            'versions-location'
-        location = ''
         for key, val in self._response_headers:
-            if key.lower() == sysmeta_version_hdr:
+            if key.lower() == VERSIONS_LOC_SYSMETA:
                 location = val
+            elif key.lower() == VERSIONS_MODE_SYSMETA:
+                mode = val
 
         if location:
-            self._response_headers.extend([('X-Versions-Location', location)])
+            self._response_headers.extend([
+                (VERSIONS_LOC_CLIENT.title(), location)])
+        if mode:
+            self._response_headers.extend([
+                (VERSIONS_MODE_CLIENT.title(), mode)])
 
         start_response(self._response_status,
                        self._response_headers,
@@ -444,12 +660,9 @@ class VersionedWritesMiddleware(object):
         self.logger = get_logger(conf, log_route='versioned_writes')
 
     def container_request(self, req, start_response, enabled):
-        sysmeta_version_hdr = get_sys_meta_prefix('container') + \
-            'versions-location'
-
         # set version location header as sysmeta
-        if 'X-Versions-Location' in req.headers:
+        if VERSIONS_LOC_CLIENT in req.headers:
-            val = req.headers.get('X-Versions-Location')
+            val = req.headers.get(VERSIONS_LOC_CLIENT)
             if val:
                 # differently from previous version, we are actually
                 # returning an error if user tries to set versions location
@@ -461,11 +674,11 @@ class VersionedWritesMiddleware(object):
                         body='Versioned Writes is disabled')
 
                 location = check_container_format(req, val)
-                req.headers[sysmeta_version_hdr] = location
+                req.headers[VERSIONS_LOC_SYSMETA] = location
 
                 # reset original header to maintain sanity
                 # now only sysmeta is source of Versions Location
-                req.headers['X-Versions-Location'] = ''
+                req.headers[VERSIONS_LOC_CLIENT] = ''
 
                 # if both headers are in the same request
                 # adding location takes precedence over removing
@@ -478,10 +691,31 @@ class VersionedWritesMiddleware(object):
         # handle removing versions container
         val = req.headers.get('X-Remove-Versions-Location')
         if val:
-            req.headers.update({sysmeta_version_hdr: ''})
+            req.headers.update({VERSIONS_LOC_SYSMETA: '',
-            req.headers.update({'X-Versions-Location': ''})
+                                VERSIONS_LOC_CLIENT: ''})
             del req.headers['X-Remove-Versions-Location']
 
+        # handle versioning mode
+        if VERSIONS_MODE_CLIENT in req.headers:
+            val = req.headers.pop(VERSIONS_MODE_CLIENT)
+            if val:
+                if not config_true_value(enabled) and \
+                        req.method in ('PUT', 'POST'):
+                    raise HTTPPreconditionFailed(
+                        request=req, content_type='text/plain',
+                        body='Versioned Writes is disabled')
+                if val not in VERSIONING_MODES:
+                    raise HTTPBadRequest(
+                        request=req, content_type='text/plain',
+                        body='X-Versions-Mode must be one of %s' % ', '.join(
+                            VERSIONING_MODES))
+                req.headers[VERSIONS_MODE_SYSMETA] = val
+            else:
+                req.headers['X-Remove-Versions-Mode'] = 'x'
+
+        if req.headers.pop('X-Remove-Versions-Mode', None):
+            req.headers.update({VERSIONS_MODE_SYSMETA: ''})
+
         # send request and translate sysmeta headers from response
         vw_ctx = VersionedWritesContext(self.app, self.logger)
         return vw_ctx.handle_container_request(req.environ, start_response)
@@ -503,6 +737,8 @@ class VersionedWritesMiddleware(object):
         # for backwards compatibility feature is enabled.
         versions_cont = container_info.get(
             'sysmeta', {}).get('versions-location')
+        versioning_mode = container_info.get(
+            'sysmeta', {}).get('versions-mode', 'stack')
         if not versions_cont:
             versions_cont = container_info.get('versions')
             # if allow_versioned_writes is not set in the configuration files
@@ -518,8 +754,13 @@ class VersionedWritesMiddleware(object):
                 resp = vw_ctx.handle_obj_versions_put(
                     req, versions_cont, api_version, account_name,
                     object_name)
-            else:  # handle DELETE
+            # handle DELETE
-                resp = vw_ctx.handle_obj_versions_delete(
+            elif versioning_mode == 'history':
+                resp = vw_ctx.handle_obj_versions_delete_push(
+                    req, versions_cont, api_version, account_name,
+                    container_name, object_name)
+            else:
+                resp = vw_ctx.handle_obj_versions_delete_pop(
                     req, versions_cont, api_version, account_name,
                     container_name, object_name)
 
@@ -573,7 +814,8 @@ def filter_factory(global_conf, **local_conf):
     conf = global_conf.copy()
     conf.update(local_conf)
     if config_true_value(conf.get('allow_versioned_writes')):
-        register_swift_info('versioned_writes')
+        register_swift_info('versioned_writes',
+                            allowed_versions_mode=VERSIONING_MODES)
 
     def obj_versions_filter(app):
         return VersionedWritesMiddleware(app, conf)

test/unit/common/middleware/helpers.py

@@ -15,7 +15,7 @@
 
 # This stuff can't live in test/unit/__init__.py due to its swob dependency.
 
-from collections import defaultdict
+from collections import defaultdict, namedtuple
 from hashlib import md5
 from swift.common import swob
 from swift.common.header_key_dict import HeaderKeyDict
@@ -41,6 +41,9 @@ class LeakTrackingIter(object):
         self.fake_swift.mark_closed(self.path)
 
 
+FakeSwiftCall = namedtuple('FakeSwiftCall', ['method', 'path', 'headers'])
+
+
 class FakeSwift(object):
     """
     A good-enough fake Swift proxy server to use in testing middleware.
@@ -148,7 +151,8 @@ class FakeSwift(object):
 
         # note: tests may assume this copy of req_headers is case insensitive
         # so we deliberately use a HeaderKeyDict
-        self._calls.append((method, path, HeaderKeyDict(req.headers)))
+        self._calls.append(
+            FakeSwiftCall(method, path, HeaderKeyDict(req.headers)))
 
         # range requests ought to work, hence conditional_response=True
         if isinstance(body, list):

test/unit/common/middleware/test_versioned_writes.py

@@ -17,8 +17,9 @@ import functools
 import json
 import os
 import time
+import mock
 import unittest
-from swift.common import swob
+from swift.common import swob, utils
 from swift.common.middleware import versioned_writes, copy
 from swift.common.swob import Request
 from test.unit.common.middleware.helpers import FakeSwift
@@ -121,7 +122,31 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         method, path, req_headers = calls[0]
         self.assertEqual('PUT', method)
         self.assertEqual('/v1/a/c', path)
-        self.assertTrue('x-container-sysmeta-versions-location' in req_headers)
+        self.assertIn('x-container-sysmeta-versions-location', req_headers)
+        self.assertNotIn('x-container-sysmeta-versions-mode', req_headers)
+        self.assertEqual(len(self.authorized), 1)
+        self.assertRequestEqual(req, self.authorized[0])
+
+    def test_put_container_history(self):
+        self.app.register('PUT', '/v1/a/c', swob.HTTPOk, {}, 'passed')
+        req = Request.blank('/v1/a/c',
+                            headers={'X-Versions-Location': 'ver_cont',
+                                     'X-Versions-Mode': 'history'},
+                            environ={'REQUEST_METHOD': 'PUT'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '200 OK')
+
+        # check for sysmeta header
+        calls = self.app.calls_with_headers
+        method, path, req_headers = calls[0]
+        self.assertEqual('PUT', method)
+        self.assertEqual('/v1/a/c', path)
+        self.assertIn('x-container-sysmeta-versions-location', req_headers)
+        self.assertEqual('ver_cont',
+                         req_headers['x-container-sysmeta-versions-location'])
+        self.assertIn('x-container-sysmeta-versions-mode', req_headers)
+        self.assertEqual('history',
+                         req_headers['x-container-sysmeta-versions-mode'])
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
 
@@ -160,10 +185,10 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         method, path, req_headers = calls[0]
         self.assertEqual('POST', method)
         self.assertEqual('/v1/a/c', path)
-        self.assertTrue('x-container-sysmeta-versions-location' in req_headers)
+        self.assertIn('x-container-sysmeta-versions-location', req_headers)
         self.assertEqual('',
                          req_headers['x-container-sysmeta-versions-location'])
-        self.assertTrue('x-versions-location' in req_headers)
+        self.assertIn('x-versions-location', req_headers)
         self.assertEqual('', req_headers['x-versions-location'])
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
@@ -181,14 +206,84 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         method, path, req_headers = calls[0]
         self.assertEqual('POST', method)
         self.assertEqual('/v1/a/c', path)
-        self.assertTrue('x-container-sysmeta-versions-location' in req_headers)
+        self.assertIn('x-container-sysmeta-versions-location', req_headers)
         self.assertEqual('',
                          req_headers['x-container-sysmeta-versions-location'])
-        self.assertTrue('x-versions-location' in req_headers)
+        self.assertIn('x-versions-location', req_headers)
         self.assertEqual('', req_headers['x-versions-location'])
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
 
+    def test_post_versions_mode(self):
+        self.app.register('POST', '/v1/a/c', swob.HTTPOk, {}, 'passed')
+        req = Request.blank('/v1/a/c',
+                            headers={'X-Versions-Mode': 'stack'},
+                            environ={'REQUEST_METHOD': 'POST'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '200 OK')
+
+        # check for sysmeta header
+        calls = self.app.calls_with_headers
+        method, path, req_headers = calls[0]
+        self.assertEqual('POST', method)
+        self.assertEqual('/v1/a/c', path)
+        self.assertIn('x-container-sysmeta-versions-mode', req_headers)
+        self.assertEqual('stack',
+                         req_headers['x-container-sysmeta-versions-mode'])
+        self.assertNotIn('x-versions-mode', req_headers)
+        self.assertEqual(len(self.authorized), 1)
+        self.assertRequestEqual(req, self.authorized[0])
+
+    def test_remove_versions_mode(self):
+        self.app.register('POST', '/v1/a/c', swob.HTTPOk, {}, 'passed')
+        req = Request.blank('/v1/a/c',
+                            headers={'X-Remove-Versions-Mode': 'x'},
+                            environ={'REQUEST_METHOD': 'POST'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '200 OK')
+
+        # check for sysmeta header
+        calls = self.app.calls_with_headers
+        method, path, req_headers = calls[0]
+        self.assertEqual('POST', method)
+        self.assertEqual('/v1/a/c', path)
+        self.assertIn('x-container-sysmeta-versions-mode', req_headers)
+        self.assertEqual('',
+                         req_headers['x-container-sysmeta-versions-mode'])
+        self.assertNotIn('x-versions-mode', req_headers)
+        self.assertEqual(len(self.authorized), 1)
+        self.assertRequestEqual(req, self.authorized[0])
+
+    def test_empty_versions_mode(self):
+        self.app.register('POST', '/v1/a/c', swob.HTTPOk, {}, 'passed')
+        req = Request.blank('/v1/a/c',
+                            headers={'X-Versions-Mode': ''},
+                            environ={'REQUEST_METHOD': 'POST'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '200 OK')
+
+        # check for sysmeta header
+        calls = self.app.calls_with_headers
+        method, path, req_headers = calls[0]
+        self.assertEqual('POST', method)
+        self.assertEqual('/v1/a/c', path)
+        self.assertIn('x-container-sysmeta-versions-mode', req_headers)
+        self.assertEqual('',
+                         req_headers['x-container-sysmeta-versions-mode'])
+        self.assertNotIn('x-versions-mode', req_headers)
+        self.assertEqual(len(self.authorized), 1)
+        self.assertRequestEqual(req, self.authorized[0])
+
+    def test_bad_versions_mode(self):
+        self.app.register('POST', '/v1/a/c', swob.HTTPOk, {}, 'passed')
+        req = Request.blank('/v1/a/c',
+                            headers={'X-Versions-Mode': 'foo'},
+                            environ={'REQUEST_METHOD': 'POST'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '400 Bad Request')
+        self.assertEqual(len(self.authorized), 0)
+        self.assertEqual('X-Versions-Mode must be one of stack, history', body)
+
     def test_remove_add_versions_precedence(self):
         self.app.register(
             'POST', '/v1/a/c', swob.HTTPOk,
@@ -201,28 +296,45 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
 
         status, headers, body = self.call_vw(req)
         self.assertEqual(status, '200 OK')
-        self.assertTrue(('X-Versions-Location', 'ver_cont') in headers)
+        self.assertIn(('X-Versions-Location', 'ver_cont'), headers)
 
         # check for sysmeta header
         calls = self.app.calls_with_headers
         method, path, req_headers = calls[0]
         self.assertEqual('POST', method)
         self.assertEqual('/v1/a/c', path)
-        self.assertTrue('x-container-sysmeta-versions-location' in req_headers)
+        self.assertIn('x-container-sysmeta-versions-location', req_headers)
-        self.assertTrue('x-remove-versions-location' not in req_headers)
+        self.assertNotIn('x-remove-versions-location', req_headers)
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
 
     def test_get_container(self):
         self.app.register(
             'GET', '/v1/a/c', swob.HTTPOk,
-            {'x-container-sysmeta-versions-location': 'ver_cont'}, None)
+            {'x-container-sysmeta-versions-location': 'ver_cont',
+             'x-container-sysmeta-versions-mode': 'stack'}, None)
         req = Request.blank(
             '/v1/a/c',
             environ={'REQUEST_METHOD': 'GET'})
         status, headers, body = self.call_vw(req)
         self.assertEqual(status, '200 OK')
-        self.assertTrue(('X-Versions-Location', 'ver_cont') in headers)
+        self.assertIn(('X-Versions-Location', 'ver_cont'), headers)
+        self.assertIn(('X-Versions-Mode', 'stack'), headers)
+        self.assertEqual(len(self.authorized), 1)
+        self.assertRequestEqual(req, self.authorized[0])
+
+    def test_head_container(self):
+        self.app.register(
+            'HEAD', '/v1/a/c', swob.HTTPOk,
+            {'x-container-sysmeta-versions-location': 'other_ver_cont',
+             'x-container-sysmeta-versions-mode': 'history'}, None)
+        req = Request.blank(
+            '/v1/a/c',
+            environ={'REQUEST_METHOD': 'HEAD'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '200 OK')
+        self.assertIn(('X-Versions-Location', 'other_ver_cont'), headers)
+        self.assertIn(('X-Versions-Mode', 'history'), headers)
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
 
@@ -311,7 +423,7 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         self.assertEqual(len(self.authorized), 1)
         self.assertRequestEqual(req, self.authorized[0])
         called_method = [method for (method, path, hdrs) in self.app._calls]
-        self.assertTrue('GET' not in called_method)
+        self.assertNotIn('GET', called_method)
 
     def test_put_request_is_dlo_manifest_with_container_config_true(self):
         # set x-object-manifest on request and expect no versioning occurred
@@ -364,8 +476,8 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         self.assertRequestEqual(req, self.authorized[0])
         called_method = \
             [method for (method, path, rheaders) in self.app._calls]
-        self.assertTrue('PUT' not in called_method)
+        self.assertNotIn('PUT', called_method)
-        self.assertTrue('GET' not in called_method)
+        self.assertNotIn('GET', called_method)
         self.assertEqual(1, self.app.call_count)
 
     def test_new_version_success(self):
@@ -474,7 +586,7 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
         self.assertEqual('PUT', method)
         self.assertEqual('/v1/a/ver_cont/001o/0000000000.00000', path)
 
-    def test_delete_first_object_success(self):
+    def test_delete_no_versions_container_success(self):
         self.app.register(
             'DELETE', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
         self.app.register(
@@ -501,7 +613,31 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
             ('DELETE', '/v1/a/c/o'),
         ])
 
-    def test_delete_latest_version_success(self):
+    def test_delete_first_object_success(self):
+        self.app.register(
+            'DELETE', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
+        self.app.register(
+            'GET',
+            '/v1/a/ver_cont?format=json&prefix=001o/&marker=&reverse=on',
+            swob.HTTPOk, {}, '[]')
+
+        cache = FakeCache({'sysmeta': {'versions-location': 'ver_cont'}})
+        req = Request.blank(
+            '/v1/a/c/o',
+            environ={'REQUEST_METHOD': 'DELETE', 'swift.cache': cache,
+                     'CONTENT_LENGTH': '0'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '200 OK')
+        self.assertEqual(len(self.authorized), 1)
+        self.assertRequestEqual(req, self.authorized[0])
+
+        prefix_listing_prefix = '/v1/a/ver_cont?format=json&prefix=001o/&'
+        self.assertEqual(self.app.calls, [
+            ('GET', prefix_listing_prefix + 'marker=&reverse=on'),
+            ('DELETE', '/v1/a/c/o'),
+        ])
+
+    def test_delete_latest_version_no_marker_success(self):
         self.app.register(
             'GET',
             '/v1/a/ver_cont?format=json&prefix=001o/&marker=&reverse=on',
@@ -551,6 +687,235 @@ class VersionedWritesTestCase(VersionedWritesBaseTestCase):
             ('DELETE', '/v1/a/ver_cont/001o/2'),
         ])
 
+    def test_delete_latest_version_restores_marker_success(self):
+        self.app.register(
+            'GET',
+            '/v1/a/ver_cont?format=json&prefix=001o/&marker=&reverse=on',
+            swob.HTTPOk, {},
+            '[{"hash": "x", '
+            '"last_modified": "2014-11-21T14:23:02.206740", '
+            '"bytes": 3, '
+            '"name": "001o/2", '
+            '"content_type": "application/x-deleted;swift_versions_deleted=1"'
+            '}, {"hash": "y", '
+            '"last_modified": "2014-11-21T14:14:27.409100", '
+            '"bytes": 3, '
+            '"name": "001o/1", '
+            '"content_type": "text/plain"'
+            '}]')
+        self.app.register(
+            'HEAD', '/v1/a/c/o', swob.HTTPOk, {}, 'passed')
+        self.app.register(
+            'DELETE', '/v1/a/c/o', swob.HTTPNoContent, {})
+
+        cache = FakeCache({'sysmeta': {'versions-location': 'ver_cont'}})
+        req = Request.blank(
+            '/v1/a/c/o',
+            headers={'X-If-Delete-At': 1},
+            environ={'REQUEST_METHOD': 'DELETE', 'swift.cache': cache,
+                     'CONTENT_LENGTH': '0'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '204 No Content')
+        self.assertEqual(len(self.authorized), 2)
+        self.assertRequestEqual(req, self.authorized[0])
+        self.assertRequestEqual(req, self.authorized[1])
+
+        calls = self.app.calls_with_headers
+        self.assertEqual(['GET', 'HEAD', 'DELETE'],
+                         [c.method for c in calls])
+
+        self.assertIn('X-Newest', calls[1].headers)
+        self.assertEqual('True', calls[1].headers['X-Newest'])
+
+        method, path, req_headers = calls.pop()
+        self.assertTrue(path.startswith('/v1/a/c/o'))
+        # Since we're deleting the original, this *should* still be present:
+        self.assertEqual('1', req_headers.get('X-If-Delete-At'))
+
+    def test_delete_latest_version_is_marker_success(self):
+        # Test popping a delete marker off the stack. So, there's data in the
+        # versions container, topped by a delete marker, and there's nothing
+        # in the base versioned container.
+        self.app.register(
+            'GET',
+            '/v1/a/ver_cont?format=json&prefix=001o/&marker=&reverse=on',
+            swob.HTTPOk, {},
+            '[{"hash": "y", '
+            '"last_modified": "2014-11-21T14:23:02.206740", '
+            '"bytes": 3, '
+            '"name": "001o/2", '
+            '"content_type": "application/x-deleted;swift_versions_deleted=1"'
+            '},{"hash": "x", '
+            '"last_modified": "2014-11-21T14:14:27.409100", '
+            '"bytes": 3, '
+            '"name": "001o/1", '
+            '"content_type": "text/plain"'
+            '}]')
+        self.app.register(
+            'HEAD', '/v1/a/c/o', swob.HTTPNotFound, {}, 'passed')
+        self.app.register(
+            'GET', '/v1/a/ver_cont/001o/1', swob.HTTPOk, {}, 'passed')
+        self.app.register(
+            'PUT', '/v1/a/c/o', swob.HTTPCreated, {}, None)
+        self.app.register(
+            'DELETE', '/v1/a/ver_cont/001o/2', swob.HTTPOk, {}, 'passed')
+        self.app.register(
+            'DELETE', '/v1/a/ver_cont/001o/1', swob.HTTPOk, {}, 'passed')
+
+        cache = FakeCache({'sysmeta': {'versions-location': 'ver_cont'}})
+        req = Request.blank(
+            '/v1/a/c/o',
+            headers={'X-If-Delete-At': 1},
+            environ={'REQUEST_METHOD': 'DELETE', 'swift.cache': cache,
+                     'CONTENT_LENGTH': '0'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '200 OK')
+        self.assertEqual(len(self.authorized), 1)
+        self.assertRequestEqual(req, self.authorized[0])
+
+        prefix_listing_prefix = '/v1/a/ver_cont?format=json&prefix=001o/&'
+        self.assertEqual(self.app.calls, [
+            ('GET', prefix_listing_prefix + 'marker=&reverse=on'),
+            ('HEAD', '/v1/a/c/o'),
+            ('GET', '/v1/a/ver_cont/001o/1'),
+            ('PUT', '/v1/a/c/o'),
+            ('DELETE', '/v1/a/ver_cont/001o/1'),
+            ('DELETE', '/v1/a/ver_cont/001o/2'),
+        ])
+        self.assertIn('X-Newest', self.app.headers[1])
+        self.assertEqual('True', self.app.headers[1]['X-Newest'])
+        self.assertIn('X-Newest', self.app.headers[2])
+        self.assertEqual('True', self.app.headers[2]['X-Newest'])
+
+        # check that X-If-Delete-At was removed from DELETE request
+        for req_headers in self.app.headers[-2:]:
+            self.assertNotIn('x-if-delete-at',
+                             [h.lower() for h in req_headers])
+
+    def test_delete_latest_version_doubled_up_markers_success(self):
+        self.app.register(
+            'GET', '/v1/a/ver_cont?format=json&prefix=001o/'
+            '&marker=&reverse=on',
+            swob.HTTPOk, {},
+            '[{"hash": "x", '
+            '"last_modified": "2014-11-21T14:23:02.206740", '
+            '"bytes": 3, '
+            '"name": "001o/3", '
+            '"content_type": "application/x-deleted;swift_versions_deleted=1"'
+            '}, {"hash": "y", '
+            '"last_modified": "2014-11-21T14:14:27.409100", '
+            '"bytes": 3, '
+            '"name": "001o/2", '
+            '"content_type": "application/x-deleted;swift_versions_deleted=1"'
+            '}, {"hash": "y", '
+            '"last_modified": "2014-11-20T14:23:02.206740", '
+            '"bytes": 30, '
+            '"name": "001o/1", '
+            '"content_type": "text/plain"'
+            '}]')
+        self.app.register(
+            'HEAD', '/v1/a/c/o', swob.HTTPNotFound, {}, 'passed')
+        self.app.register(
+            'DELETE', '/v1/a/ver_cont/001o/3', swob.HTTPOk, {}, 'passed')
+
+        cache = FakeCache({'sysmeta': {'versions-location': 'ver_cont'}})
+        req = Request.blank(
+            '/v1/a/c/o',
+            headers={'X-If-Delete-At': 1},
+            environ={'REQUEST_METHOD': 'DELETE', 'swift.cache': cache,
+                     'CONTENT_LENGTH': '0'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '200 OK')
+        self.assertEqual(len(self.authorized), 1)
+        self.assertRequestEqual(req, self.authorized[0])
+
+        # check that X-If-Delete-At was removed from DELETE request
+        calls = self.app.calls_with_headers
+        self.assertEqual(['GET', 'HEAD', 'DELETE'],
+                         [c.method for c in calls])
+
+        method, path, req_headers = calls.pop()
+        self.assertTrue(path.startswith('/v1/a/ver_cont/001o/3'))
+        self.assertNotIn('x-if-delete-at', [h.lower() for h in req_headers])
+
+    def test_post_bad_mode(self):
+        req = Request.blank(
+            '/v1/a/c',
+            environ={'REQUEST_METHOD': 'POST',
+                     'CONTENT_LENGTH': '0',
+                     'HTTP_X_VERSIONS_MODE': 'bad-mode'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '400 Bad Request')
+        self.assertEqual('X-Versions-Mode must be one of stack, history', body)
+        self.assertFalse(self.app.calls_with_headers)
+
+    @mock.patch('swift.common.middleware.versioned_writes.time.time',
+                return_value=1234)
+    def test_history_delete_marker_no_object_success(self, mock_time):
+        self.app.register(
+            'GET', '/v1/a/c/o', swob.HTTPNotFound,
+            {}, 'passed')
+        self.app.register(
+            'PUT', '/v1/a/ver_cont/001o/0000001234.00000', swob.HTTPCreated,
+            {}, 'passed')
+        self.app.register(
+            'DELETE', '/v1/a/c/o', swob.HTTPNotFound, {}, None)
+
+        cache = FakeCache({'sysmeta': {'versions-location': 'ver_cont',
+                                       'versions-mode': 'history'}})
+        req = Request.blank(
+            '/v1/a/c/o',
+            environ={'REQUEST_METHOD': 'DELETE', 'swift.cache': cache,
+                     'CONTENT_LENGTH': '0'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '404 Not Found')
+        self.assertEqual(len(self.authorized), 1)
+
+        req.environ['REQUEST_METHOD'] = 'PUT'
+        self.assertRequestEqual(req, self.authorized[0])
+
+        calls = self.app.calls_with_headers
+        self.assertEqual(['GET', 'PUT', 'DELETE'], [c.method for c in calls])
+        self.assertEqual('application/x-deleted;swift_versions_deleted=1',
+                         calls[1].headers.get('Content-Type'))
+
+    @mock.patch('swift.common.middleware.versioned_writes.time.time',
+                return_value=123456789.54321)
+    def test_history_delete_marker_over_object_success(self, mock_time):
+        self.app.register(
+            'GET', '/v1/a/c/o', swob.HTTPOk,
+            {'last-modified': 'Wed, 19 Nov 2014 18:19:02 GMT'}, 'passed')
+        self.app.register(
+            'PUT', '/v1/a/ver_cont/001o/1416421142.00000', swob.HTTPCreated,
+            {}, 'passed')
+        self.app.register(
+            'PUT', '/v1/a/ver_cont/001o/0123456789.54321', swob.HTTPCreated,
+            {}, 'passed')
+        self.app.register(
+            'DELETE', '/v1/a/c/o', swob.HTTPNoContent, {}, None)
+
+        cache = FakeCache({'sysmeta': {'versions-location': 'ver_cont',
+                                       'versions-mode': 'history'}})
+        req = Request.blank(
+            '/v1/a/c/o',
+            environ={'REQUEST_METHOD': 'DELETE', 'swift.cache': cache,
+                     'CONTENT_LENGTH': '0'})
+        status, headers, body = self.call_vw(req)
+        self.assertEqual(status, '204 No Content')
+        self.assertEqual('', body)
+        self.assertEqual(len(self.authorized), 1)
+
+        req.environ['REQUEST_METHOD'] = 'PUT'
+        self.assertRequestEqual(req, self.authorized[0])
+
+        calls = self.app.calls_with_headers
+        self.assertEqual(['GET', 'PUT', 'PUT', 'DELETE'],
+                         [c.method for c in calls])
+        self.assertEqual('/v1/a/ver_cont/001o/1416421142.00000',
+                         calls[1].path)
+        self.assertEqual('application/x-deleted;swift_versions_deleted=1',
+                         calls[2].headers.get('Content-Type'))
+
     def test_delete_single_version_success(self):
         # check that if the first listing page has just a single item then
         # it is not erroneously inferred to be a non-reversed listing
@@ -1098,3 +1463,28 @@ class VersionedWritesCopyingTestCase(VersionedWritesBaseTestCase):
         self.assertEqual('PUT', self.authorized[1].method)
         self.assertEqual('/v1/a/tgt_cont/tgt_obj', self.authorized[1].path)
         self.assertEqual(2, self.app.call_count)
+
+
+class TestSwiftInfo(unittest.TestCase):
+    def setUp(self):
+        utils._swift_info = {}
+        utils._swift_admin_info = {}
+
+    def test_registered_defaults(self):
+        versioned_writes.filter_factory({})('have to pass in an app')
+        swift_info = utils.get_swift_info()
+        # in default, versioned_writes is not in swift_info
+        self.assertNotIn('versioned_writes', swift_info)
+
+    def test_registered_explicitly_set(self):
+        versioned_writes.filter_factory(
+            {'allow_versioned_writes': 'true'})('have to pass in an app')
+        swift_info = utils.get_swift_info()
+        self.assertIn('versioned_writes', swift_info)
+        self.assertEqual(
+            swift_info['versioned_writes'].get('allowed_versions_mode'),
+            ('stack', 'history'))
+
+
+if __name__ == '__main__':
+    unittest.main()