Changeset bf670c0 in trunk

Timestamp:

2021-10-12T21:42:59Z (4 years ago)

Author:

fenn-cs <fenn25.fn@…>

Branches:

master

Children:

26419c4

Parents:

49b6080 (diff), be5f583 (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

Merge branch 'master' of github.com:tahoe-lafs/tahoe-lafs into 3788.refactor-src.test.mutable

Files:

• .circleci/config.yml (modified) (4 diffs)
• .github/workflows/ci.yml (modified) (4 diffs)
• docs/proposed/http-storage-node-protocol.rst (modified) (11 diffs)
• newsfragments/3786.feature (added)
• newsfragments/3793.minor (added)
• newsfragments/3795.minor (added)
• newsfragments/3797.minor (added)
• newsfragments/3798.minor (added)
• newsfragments/3799.minor (added)
• newsfragments/3801.bugfix (added)
• newsfragments/3805.minor (added)
• newsfragments/3806.minor (added)
• newsfragments/3808.installation (added)
• newsfragments/3810.minor (added)
• nix/collections-extended.nix (added)
• nix/overlays.nix (modified) (1 diff)
• nix/py3.nix (added)
• nix/tahoe-lafs.nix (modified) (2 diffs)
• setup.py (modified) (2 diffs)
• src/allmydata/interfaces.py (modified) (3 diffs)
• src/allmydata/mutable/layout.py (modified) (6 diffs)
• src/allmydata/storage/common.py (modified) (1 diff)
• src/allmydata/storage/immutable.py (modified) (7 diffs)
• src/allmydata/storage/mutable.py (modified) (1 diff)
• src/allmydata/storage/server.py (modified) (9 diffs)
• src/allmydata/storage_client.py (modified) (1 diff)
• src/allmydata/test/common_util.py (modified) (1 diff)
• src/allmydata/test/mutable/util.py (modified) (1 diff)
• src/allmydata/test/test_istorageserver.py (modified) (13 diffs)
• src/allmydata/test/test_openmetrics.py (added)
• src/allmydata/test/test_storage.py (modified) (15 diffs)
• src/allmydata/web/status.py (modified) (2 diffs)

.circleci/config.yml ¶

@@ -41 +41 @@
 
       - "nixos-19-09":
+          {}
+
+      - "nixos-21-05":
           {}
 
@@ -439 +442 @@
         user: "nobody"
 
-
-  nixos-19-09:
+  nixos-19-09: &NIXOS
     docker:
       # Run in a highly Nix-capable environment.
@@ -448 +450 @@
     environment:
       NIX_PATH: "nixpkgs=https://github.com/NixOS/nixpkgs-channels/archive/nixos-19.09-small.tar.gz"
+      SOURCE: "nix/"
 
     steps:
@@ -464 +467 @@
             # advantage of multiple cores and we get a little speedup by doing
             # them in parallel.
-            nix-build --cores 3 --max-jobs 2 nix/
+            nix-build --cores 3 --max-jobs 2 "$SOURCE"
+
+  nixos-21-05:
+    <<: *NIXOS
+
+    environment:
+      # Note this doesn't look more similar to the 19.09 NIX_PATH URL because
+      # there was some internal shuffling by the NixOS project about how they
+      # publish stable revisions.
+      NIX_PATH: "nixpkgs=https://github.com/NixOS/nixpkgs/archive/d32b07e6df276d78e3640eb43882b80c9b2b3459.tar.gz"
+      SOURCE: "nix/py3.nix"
 
   typechecks:

.github/workflows/ci.yml ¶

@@ -29 +29 @@
         include:
           # On macOS don't bother with 3.6-3.8, just to get faster builds.
-          - os: macos-latest
+          - os: macos-10.15
             python-version: 2.7
           - os: macos-latest
@@ -169 +169 @@
         include:
           # On macOS don't bother with 3.6, just to get faster builds.
-          - os: macos-latest
+          - os: macos-10.15
             python-version: 2.7
           - os: macos-latest
@@ -184 +184 @@
       # tests on macOS.
       - name: Install Tor [macOS, ${{ matrix.python-version }} ]
-        if: ${{ matrix.os == 'macos-latest' }}
+        if: ${{ contains(matrix.os, 'macos') }}
         run: |
           brew extract --version 0.4.5.8 tor homebrew/cask
@@ -248 +248 @@
       matrix:
         os:
-          - macos-latest
+          - macos-10.15
           - windows-latest
           - ubuntu-latest

docs/proposed/http-storage-node-protocol.rst ¶

@@ -483 +483 @@
   It could also just be that the client's preferred servers have changed.
 
-``PUT /v1/immutable/:storage_index/:share_number``
-!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+``PATCH /v1/immutable/:storage_index/:share_number``
+!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
 
 Write data for the indicated share.
@@ -498 +498 @@
 The server must recognize when all of the data has been received and mark the share as complete
 (which it can do because it was informed of the size when the storage index was initialized).
-Clients should upload chunks in re-assembly order.
 
 * When a chunk that does not complete the share is successfully uploaded the response is ``OK``.
-* When the chunk that completes the share is successfully uploaded the response is ``CREATED``.
-* If the *Content-Range* for a request covers part of the share that has already been uploaded the response is ``CONFLICT``.
   The response body indicates the range of share data that has yet to be uploaded.
   That is::
@@ -514 +511 @@
       ]
     }
+
+* When the chunk that completes the share is successfully uploaded the response is ``CREATED``.
+* If the *Content-Range* for a request covers part of the share that has already,
+  and the data does not match already written data,
+  the response is ``CONFLICT``.
+  At this point the only thing to do is abort the upload and start from scratch (see below).
+
+``PUT /v1/immutable/:storage_index/:share_number/abort``
+!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+
+This cancels an *in-progress* upload.
+
+The response code:
+
+* When the upload is still in progress and therefore the abort has succeeded,
+  the response is ``OK``.
+  Future uploads can start from scratch with no pre-existing upload state stored on the server.
+* If the uploaded has already finished, the response is 405 (Method Not Allowed)
+  and no change is made.
+
+
+Discussion
+``````````
+
+``PUT`` verbs are only supposed to be used to replace the whole resource,
+thus the use of ``PATCH``.
+From RFC 7231::
+
+   An origin server that allows PUT on a given target resource MUST send
+   a 400 (Bad Request) response to a PUT request that contains a
+   Content-Range header field (Section 4.2 of [RFC7233]), since the
+   payload is likely to be partial content that has been mistakenly PUT
+   as a full representation.  Partial content updates are possible by
+   targeting a separately identified resource with state that overlaps a
+   portion of the larger resource, or by using a different method that
+   has been specifically defined for partial updates (for example, the
+   PATCH method defined in [RFC5789]).
 
 
@@ -601 +635 @@
                    "offset": 3,
                    "size": 5,
-                   "operator": "eq",
                    "specimen": "hello"
                }, ...],
@@ -627 +660 @@
   }
 
+A test vector or read vector that read beyond the boundaries of existing data will return nothing for any bytes past the end.
+As a result, if there is no data at all, an empty bytestring is returned no matter what the offset or length.
+
 Reading
 ~~~~~~~
@@ -667 +703 @@
 #. Upload the content for immutable share ``7``::
 
-     PUT /v1/immutable/AAAAAAAAAAAAAAAA/7
+     PATCH /v1/immutable/AAAAAAAAAAAAAAAA/7
      Content-Range: bytes 0-15/48
      <first 16 bytes of share data>
@@ -673 +709 @@
      200 OK
 
-     PUT /v1/immutable/AAAAAAAAAAAAAAAA/7
+     PATCH /v1/immutable/AAAAAAAAAAAAAAAA/7
      Content-Range: bytes 16-31/48
      <second 16 bytes of share data>
@@ -679 +715 @@
      200 OK
 
-     PUT /v1/immutable/AAAAAAAAAAAAAAAA/7
+     PATCH /v1/immutable/AAAAAAAAAAAAAAAA/7
      Content-Range: bytes 32-47/48
      <final 16 bytes of share data>
@@ -702 +738 @@
 ~~~~~~~~~~~~
 
-1. Create mutable share number ``3`` with ``10`` bytes of data in slot ``BBBBBBBBBBBBBBBB``::
+1. Create mutable share number ``3`` with ``10`` bytes of data in slot ``BBBBBBBBBBBBBBBB``.
+The special test vector of size 1 but empty bytes will only pass
+if there is no existing share,
+otherwise it will read a byte which won't match `b""`::
 
      POST /v1/mutable/BBBBBBBBBBBBBBBB/read-test-write
@@ -716 +755 @@
                      "offset": 0,
                      "size": 1,
-                     "operator": "eq",
                      "specimen": ""
                  }],
@@ -748 +786 @@
                  "test": [{
                      "offset": 0,
-                     "size": <checkstring size>,
-                     "operator": "eq",
+                     "size": <length of checkstring>,
                      "specimen": "<checkstring>"
                  }],

nix/overlays.nix ¶

@@ -3 +3 @@
     packageOverrides = python-self: python-super: {
       # eliot is not part of nixpkgs at all at this time.
-      eliot = python-self.callPackage ./eliot.nix { };
+      eliot = python-self.pythonPackages.callPackage ./eliot.nix { };
 
       # NixOS autobahn package has trollius as a dependency, although
       # it is optional. Trollius is unmaintained and fails on CI.
-      autobahn = python-super.callPackage ./autobahn.nix { };
+      autobahn = python-super.pythonPackages.callPackage ./autobahn.nix { };
 
       # Porting to Python 3 is greatly aided by the future package.  A
       # slightly newer version than appears in nixos 19.09 is helpful.
-      future = python-super.callPackage ./future.nix { };
+      future = python-super.pythonPackages.callPackage ./future.nix { };
 
       # Need version of pyutil that supports Python 3. The version in 19.09
       # is too old.
-      pyutil = python-super.callPackage ./pyutil.nix { };
+      pyutil = python-super.pythonPackages.callPackage ./pyutil.nix { };
 
       # Need a newer version of Twisted, too.
-      twisted = python-super.callPackage ./twisted.nix { };
+      twisted = python-super.pythonPackages.callPackage ./twisted.nix { };
+
+      # collections-extended is not part of nixpkgs at this time.
+      collections-extended = python-super.pythonPackages.callPackage ./collections-extended.nix { };
+    };
+  };
+
+  python39 = super.python39.override {
+    packageOverrides = python-self: python-super: {
+      # collections-extended is not part of nixpkgs at this time.
+      collections-extended = python-super.pythonPackages.callPackage ./collections-extended.nix { };
     };
   };

nix/tahoe-lafs.nix ¶

@@ -98 +98 @@
     service-identity pyyaml magic-wormhole treq
     eliot autobahn cryptography netifaces setuptools
-    future pyutil distro configparser
+    future pyutil distro configparser collections-extended
   ];
 
@@ -108 +108 @@
     html5lib
     tenacity
+    prometheus_client
   ];
 

setup.py ¶

@@ -138 +138 @@
     # Backported configparser for Python 2:
     "configparser ; python_version < '3.0'",
+
+    # For the RangeMap datastructure.
+    "collections-extended",
 ]
 
@@ -405 +408 @@
               "paramiko",
               "pytest-timeout",
+              # Does our OpenMetrics endpoint adhere to the spec:
+              "prometheus-client == 0.11.0",
           ] + tor_requires + i2p_requires,
           "tor": tor_requires,

src/allmydata/interfaces.py ¶

@@ -54 +54 @@
 
 
+class DataTooLargeError(Exception):
+    """The write went past the expected size of the bucket."""
+
+
+class ConflictingWriteError(Exception):
+    """Two writes happened to same immutable with different data."""
+
+
 class RIBucketWriter(RemoteInterface):
     """ Objects of this kind live on the server side. """
@@ -92 +100 @@
 TestVector = ListOf(TupleOf(Offset, ReadSize, bytes, bytes))
 # elements are (offset, length, operator, specimen)
-# operator is one of "lt, le, eq, ne, ge, gt"
-# nop always passes and is used to fetch data while writing.
-# you should use length==len(specimen) for everything except nop
+# operator must be b"eq", typically length==len(specimen), but one can ensure
+# writes don't happen to empty shares by setting length to 1 and specimen to
+# b"". The operator is still used for wire compatibility with old versions.
 DataVector = ListOf(TupleOf(Offset, ShareData))
 # (offset, data). This limits us to 30 writes of 1MiB each per call
@@ -352 +360 @@
         """
         :see: ``RIStorageServer.slot_testv_readv_and_writev``
+
+        While the interface mostly matches, test vectors are simplified.
+        Instead of a tuple ``(offset, read_size, operator, expected_data)`` in
+        the original, for this method you need only pass in
+        ``(offset, read_size, expected_data)``, with the operator implicitly
+        being ``b"eq"``.
         """
 

src/allmydata/mutable/layout.py ¶

@@ -310 +310 @@
         else:
             checkstring = checkstring_or_seqnum
-        self._testvs = [(0, len(checkstring), b"eq", checkstring)]
+        self._testvs = [(0, len(checkstring), checkstring)]
 
 
@@ -319 +319 @@
         """
         if self._testvs:
-            return self._testvs[0][3]
+            return self._testvs[0][2]
         return b""
 
@@ -549 +549 @@
             # Our caller has not provided us with another checkstring
             # yet, so we assume that we are writing a new share, and set
-            # a test vector that will allow a new share to be written.
+            # a test vector that will only allow a new share to be written.
             self._testvs = []
-            self._testvs.append(tuple([0, 1, b"eq", b""]))
+            self._testvs.append(tuple([0, 1, b""]))
 
         tw_vectors = {}
@@ -890 +890 @@
         else:
             self._testvs = []
-            self._testvs.append((0, len(checkstring), b"eq", checkstring))
+            self._testvs.append((0, len(checkstring), checkstring))
 
 
@@ -1162 +1162 @@
         tw_vectors = {}
         if not self._testvs:
+            # Make sure we will only successfully write if the share didn't
+            # previously exist.
             self._testvs = []
-            self._testvs.append(tuple([0, 1, b"eq", b""]))
+            self._testvs.append(tuple([0, 1, b""]))
         if not self._written:
             # Write a new checkstring to the share when we write it, so
@@ -1171 +1173 @@
             def _first_write():
                 self._written = True
-                self._testvs = [(0, len(new_checkstring), b"eq", new_checkstring)]
+                self._testvs = [(0, len(new_checkstring), new_checkstring)]
             on_success = _first_write
         tw_vectors[self.shnum] = (self._testvs, datavs, None)

src/allmydata/storage/common.py ¶

@@ -14 +14 @@
 from allmydata.util import base32
 
-class DataTooLargeError(Exception):
-    pass
+# Backwards compatibility.
+from allmydata.interfaces import DataTooLargeError  # noqa: F401
+
 class UnknownMutableContainerVersionError(Exception):
     pass

src/allmydata/storage/immutable.py ¶

@@ -14 +14 @@
 import os, stat, struct, time
 
+from collections_extended import RangeMap
+
 from foolscap.api import Referenceable
 
 from zope.interface import implementer
-from allmydata.interfaces import RIBucketWriter, RIBucketReader
+from allmydata.interfaces import (
+    RIBucketWriter, RIBucketReader, ConflictingWriteError,
+    DataTooLargeError,
+)
 from allmydata.util import base32, fileutil, log
 from allmydata.util.assertutil import precondition
 from allmydata.util.hashutil import timing_safe_compare
 from allmydata.storage.lease import LeaseInfo
-from allmydata.storage.common import UnknownImmutableContainerVersionError, \
-     DataTooLargeError
+from allmydata.storage.common import UnknownImmutableContainerVersionError
 
 # each share file (in storage/shares/$SI/$SHNUM) contains lease information
@@ -205 +209 @@
 class BucketWriter(Referenceable):  # type: ignore # warner/foolscap#78
 
-    def __init__(self, ss, incominghome, finalhome, max_size, lease_info, canary):
+    def __init__(self, ss, incominghome, finalhome, max_size, lease_info):
         self.ss = ss
         self.incominghome = incominghome
         self.finalhome = finalhome
         self._max_size = max_size # don't allow the client to write more than this
-        self._canary = canary
-        self._disconnect_marker = canary.notifyOnDisconnect(self._disconnected)
         self.closed = False
         self.throw_out_all_data = False
@@ -218 +220 @@
         # added by simultaneous uploaders
         self._sharefile.add_lease(lease_info)
+        self._already_written = RangeMap()
 
     def allocated_size(self):
@@ -227 +230 @@
         if self.throw_out_all_data:
             return
+
+        # Make sure we're not conflicting with existing data:
+        end = offset + len(data)
+        for (chunk_start, chunk_stop, _) in self._already_written.ranges(offset, end):
+            chunk_len = chunk_stop - chunk_start
+            actual_chunk = self._sharefile.read_share_data(chunk_start, chunk_len)
+            writing_chunk = data[chunk_start - offset:chunk_stop - offset]
+            if actual_chunk != writing_chunk:
+                raise ConflictingWriteError(
+                    "Chunk {}-{} doesn't match already written data.".format(chunk_start, chunk_stop)
+                )
         self._sharefile.write_share_data(offset, data)
+
+        self._already_written.set(True, offset, end)
         self.ss.add_latency("write", time.time() - start)
         self.ss.count("write")
@@ -263 +279 @@
         self._sharefile = None
         self.closed = True
-        self._canary.dontNotifyOnDisconnect(self._disconnect_marker)
 
         filelen = os.stat(self.finalhome)[stat.ST_SIZE]
@@ -270 +285 @@
         self.ss.count("close")
 
-    def _disconnected(self):
+    def disconnected(self):
         if not self.closed:
             self._abort()
@@ -277 +292 @@
         log.msg("storage: aborting sharefile %s" % self.incominghome,
                 facility="tahoe.storage", level=log.UNUSUAL)
-        if not self.closed:
-            self._canary.dontNotifyOnDisconnect(self._disconnect_marker)
         self._abort()
         self.ss.count("abort")

src/allmydata/storage/mutable.py ¶

@@ -435 +435 @@
 
 def testv_compare(a, op, b):
-    assert op in (b"lt", b"le", b"eq", b"ne", b"ge", b"gt")
-    if op == b"lt":
-        return a < b
-    if op == b"le":
-        return a <= b
-    if op == b"eq":
-        return a == b
-    if op == b"ne":
-        return a != b
-    if op == b"ge":
-        return a >= b
-    if op == b"gt":
-        return a > b
-    # never reached
+    assert op == b"eq"
+    return a == b
+
 
 class EmptyShare(object):

src/allmydata/storage/server.py ¶

@@ -12 +12 @@
     # strings. Omit bytes so we don't leak future's custom bytes.
     from future.builtins import filter, map, zip, ascii, chr, hex, input, next, oct, pow, round, super, dict, list, object, range, str, max, min  # noqa: F401
-
+else:
+    from typing import Dict
 
 import os, re, struct, time
-import weakref
 import six
 
 from foolscap.api import Referenceable
+from foolscap.ipb import IRemoteReference
 from twisted.application import service
 
@@ -90 +91 @@
         self._clean_incomplete()
         fileutil.make_dirs(self.incomingdir)
-        self._active_writers = weakref.WeakKeyDictionary()
         log.msg("StorageServer created", facility="tahoe.storage")
 
@@ -121 +121 @@
         self.lease_checker.setServiceParent(self)
         self._get_current_time = get_current_time
+
+        # Currently being-written Bucketwriters. For Foolscap, lifetime is tied
+        # to connection: when disconnection happens, the BucketWriters are
+        # removed. For HTTP, this makes no sense, so there will be
+        # timeout-based cleanup; see
+        # https://tahoe-lafs.org/trac/tahoe-lafs/ticket/3807.
+
+        # Map in-progress filesystem path -> BucketWriter:
+        self._bucket_writers = {}  # type: Dict[str,BucketWriter]
+        # Canaries and disconnect markers for BucketWriters created via Foolscap:
+        self._bucket_writer_disconnect_markers = {}  # type: Dict[BucketWriter,(IRemoteReference, object)]
 
     def __repr__(self):
@@ -239 +250 @@
     def allocated_size(self):
         space = 0
-        for bw in self._active_writers:
+        for bw in self._bucket_writers.values():
             space += bw.allocated_size()
         return space
@@ -264 +275 @@
         return version
 
-    def remote_allocate_buckets(self, storage_index,
-                                renew_secret, cancel_secret,
-                                sharenums, allocated_size,
-                                canary, owner_num=0):
+    def _allocate_buckets(self, storage_index,
+                          renew_secret, cancel_secret,
+                          sharenums, allocated_size,
+                          owner_num=0):
+        """
+        Generic bucket allocation API.
+        """
         # owner_num is not for clients to set, but rather it should be
         # curried into the PersonalStorageServer instance that is dedicated
@@ -316 +330 @@
                 pass
             elif os.path.exists(incominghome):
-                # Note that we don't create BucketWriters for shnums that
+                # For Foolscap we don't create BucketWriters for shnums that
                 # have a partial share (in incoming/), so if a second upload
                 # occurs while the first is still in progress, the second
@@ -324 +338 @@
                 # ok! we need to create the new share file.
                 bw = BucketWriter(self, incominghome, finalhome,
-                                  max_space_per_bucket, lease_info, canary)
+                                  max_space_per_bucket, lease_info)
                 if self.no_storage:
                     bw.throw_out_all_data = True
                 bucketwriters[shnum] = bw
-                self._active_writers[bw] = 1
+                self._bucket_writers[incominghome] = bw
                 if limited:
                     remaining_space -= max_space_per_bucket
@@ -339 +353 @@
 
         self.add_latency("allocate", self._get_current_time() - start)
+        return alreadygot, bucketwriters
+
+    def remote_allocate_buckets(self, storage_index,
+                                renew_secret, cancel_secret,
+                                sharenums, allocated_size,
+                                canary, owner_num=0):
+        """Foolscap-specific ``allocate_buckets()`` API."""
+        alreadygot, bucketwriters = self._allocate_buckets(
+            storage_index, renew_secret, cancel_secret, sharenums, allocated_size,
+            owner_num=owner_num,
+        )
+        # Abort BucketWriters if disconnection happens.
+        for bw in bucketwriters.values():
+            disconnect_marker = canary.notifyOnDisconnect(bw.disconnected)
+            self._bucket_writer_disconnect_markers[bw] = (canary, disconnect_marker)
         return alreadygot, bucketwriters
 
@@ -384 +413 @@
         if self.stats_provider:
             self.stats_provider.count('storage_server.bytes_added', consumed_size)
-        del self._active_writers[bw]
+        del self._bucket_writers[bw.incominghome]
+        if bw in self._bucket_writer_disconnect_markers:
+            canary, disconnect_marker = self._bucket_writer_disconnect_markers.pop(bw)
+            canary.dontNotifyOnDisconnect(disconnect_marker)
 
     def _get_bucket_shares(self, storage_index):

src/allmydata/storage_client.py ¶

@@ -995 +995 @@
             r_vector,
     ):
+        # Match the wire protocol, which requires 4-tuples for test vectors.
+        wire_format_tw_vectors = {
+            key: (
+                [(start, length, b"eq", data) for (start, length, data) in value[0]],
+                value[1],
+                value[2],
+            ) for (key, value) in tw_vectors.items()
+        }
         return self._rref.callRemote(
             "slot_testv_and_readv_and_writev",
             storage_index,
             secrets,
-            tw_vectors,
+            wire_format_tw_vectors,
             r_vector,
         )

src/allmydata/test/common_util.py ¶

@@ -315 +315 @@
         return "<fake>"
 
+    def disconnected(self):
+        """Disconnect the canary, to be called by test code.
+
+        Can only  happen once.
+        """
+        if self.disconnectors is not None:
+            for (f, args, kwargs) in list(self.disconnectors.values()):
+                f(*args, **kwargs)
+            self.disconnectors = None
+
 
 class ShouldFailMixin(object):

src/allmydata/test/mutable/util.py ¶

@@ -150 +150 @@
         for shnum, (testv, writev, new_length) in list(tw_vectors.items()):
             for (offset, length, op, specimen) in testv:
-                assert op in (b"le", b"eq", b"ge")
+                assert op == b"eq"
             # TODO: this isn't right, the read is controlled by read_vector,
             # not by testv

src/allmydata/test/test_istorageserver.py ¶

@@ -21 +21 @@
 from random import Random
 
-from testtools import skipIf
-
 from twisted.internet.defer import inlineCallbacks
 
-from foolscap.api import Referenceable
+from foolscap.api import Referenceable, RemoteException
 
 from allmydata.interfaces import IStorageServer
@@ -78 +76 @@
 
     ``self.storage_server`` is expected to provide ``IStorageServer``.
+
+    ``self.disconnect()`` should disconnect and then reconnect, creating a new
+    ``self.storage_server``.  Some implementations may wish to skip tests using
+    this; HTTP has no notion of disconnection.
     """
 
@@ -99 +101 @@
 
     @inlineCallbacks
-    @skipIf(True, "https://tahoe-lafs.org/trac/tahoe-lafs/ticket/3793")
     def test_allocate_buckets_repeat(self):
         """
-        allocate_buckets() with the same storage index returns the same result,
-        because the shares have not been written to.
-
-        This fails due to https://tahoe-lafs.org/trac/tahoe-lafs/ticket/3793
+        ``IStorageServer.allocate_buckets()`` with the same storage index does not return
+        work-in-progress buckets, but will add any newly added buckets.
         """
         storage_index, renew_secret, cancel_secret = (
@@ -116 +115 @@
             renew_secret,
             cancel_secret,
-            sharenums=set(range(5)),
+            sharenums=set(range(4)),
             allocated_size=1024,
             canary=Referenceable(),
@@ -129 +128 @@
         )
         self.assertEqual(already_got, already_got2)
-        self.assertEqual(set(allocated.keys()), set(allocated2.keys()))
-
-    @skipIf(True, "https://tahoe-lafs.org/trac/tahoe-lafs/ticket/3793")
-    @inlineCallbacks
-    def test_allocate_buckets_more_sharenums(self):
-        """
-        allocate_buckets() with the same storage index but more sharenums
-        acknowledges the extra shares don't exist.
-
-        Fails due to https://tahoe-lafs.org/trac/tahoe-lafs/ticket/3793
+        self.assertEqual(set(allocated2.keys()), {4})
+
+    @inlineCallbacks
+    def abort_or_disconnect_half_way(self, abort_or_disconnect):
+        """
+        If we disconnect/abort in the middle of writing to a bucket, all data
+        is wiped, and it's even possible to write different data to the bucket.
+
+        (In the real world one shouldn't do that, but writing different data is
+        a good way to test that the original data really was wiped.)
+
+        ``abort_or_disconnect`` is a callback that takes a bucket and aborts up
+        load, or perhaps disconnects the whole connection.
         """
         storage_index, renew_secret, cancel_secret = (
@@ -145 +147 @@
             new_secret(),
         )
-        yield self.storage_server.allocate_buckets(
+        (_, allocated) = yield self.storage_server.allocate_buckets(
+            storage_index,
+            renew_secret,
+            cancel_secret,
+            sharenums={0},
+            allocated_size=1024,
+            canary=Referenceable(),
+        )
+
+        # Bucket 1 is fully written in one go.
+        yield allocated[0].callRemote("write", 0, b"1" * 1024)
+
+        # Disconnect or abort, depending on the test:
+        yield abort_or_disconnect(allocated[0])
+
+        # Write different data with no complaint:
+        (_, allocated) = yield self.storage_server.allocate_buckets(
+            storage_index,
+            renew_secret,
+            cancel_secret,
+            sharenums={0},
+            allocated_size=1024,
+            canary=Referenceable(),
+        )
+        yield allocated[0].callRemote("write", 0, b"2" * 1024)
+
+    def test_disconnection(self):
+        """
+        If we disconnect in the middle of writing to a bucket, all data is
+        wiped, and it's even possible to write different data to the bucket.
+
+        (In the real world one shouldn't do that, but writing different data is
+        a good way to test that the original data really was wiped.)
+
+        HTTP protocol should skip this test, since disconnection is meaningless
+        concept; this is more about testing implicit contract the Foolscap
+        implementation depends on doesn't change as we refactor things.
+        """
+        return self.abort_or_disconnect_half_way(lambda _: self.disconnect())
+
+    @inlineCallbacks
+    def test_written_shares_are_allocated(self):
+        """
+        Shares that are fully written to show up as allocated in result from
+        ``IStorageServer.allocate_buckets()``.  Partially-written or empty
+        shares don't.
+        """
+        storage_index, renew_secret, cancel_secret = (
+            new_storage_index(),
+            new_secret(),
+            new_secret(),
+        )
+        (_, allocated) = yield self.storage_server.allocate_buckets(
             storage_index,
             renew_secret,
@@ -153 +207 @@
             canary=Referenceable(),
         )
-        (already_got2, allocated2) = yield self.storage_server.allocate_buckets(
-            storage_index,
-            renew_secret,
-            cancel_secret,
-            sharenums=set(range(7)),
-            allocated_size=1024,
-            canary=Referenceable(),
-        )
-        self.assertEqual(already_got2, set())  # none were fully written
-        self.assertEqual(set(allocated2.keys()), set(range(7)))
-
-    @inlineCallbacks
-    def test_written_shares_are_allocated(self):
-        """
-        Shares that are fully written to show up as allocated in result from
-        ``IStorageServer.allocate_buckets()``.  Partially-written or empty
-        shares don't.
-        """
-        storage_index, renew_secret, cancel_secret = (
-            new_storage_index(),
-            new_secret(),
-            new_secret(),
-        )
-        (_, allocated) = yield self.storage_server.allocate_buckets(
+
+        # Bucket 1 is fully written in one go.
+        yield allocated[1].callRemote("write", 0, b"1" * 1024)
+        yield allocated[1].callRemote("close")
+
+        # Bucket 2 is fully written in two steps.
+        yield allocated[2].callRemote("write", 0, b"1" * 512)
+        yield allocated[2].callRemote("write", 512, b"2" * 512)
+        yield allocated[2].callRemote("close")
+
+        # Bucket 0 has partial write.
+        yield allocated[0].callRemote("write", 0, b"1" * 512)
+
+        (already_got, _) = yield self.storage_server.allocate_buckets(
             storage_index,
             renew_secret,
@@ -184 +228 @@
             canary=Referenceable(),
         )
-
-        # Bucket 1 is fully written in one go.
-        yield allocated[1].callRemote("write", 0, b"1" * 1024)
-        yield allocated[1].callRemote("close")
-
-        # Bucket 2 is fully written in two steps.
-        yield allocated[2].callRemote("write", 0, b"1" * 512)
-        yield allocated[2].callRemote("write", 512, b"2" * 512)
-        yield allocated[2].callRemote("close")
-
-        # Bucket 0 has partial write.
-        yield allocated[0].callRemote("write", 0, b"1" * 512)
-
-        (already_got, _) = yield self.storage_server.allocate_buckets(
+        self.assertEqual(already_got, {1, 2})
+
+    @inlineCallbacks
+    def test_written_shares_are_readable(self):
+        """
+        Shares that are fully written to can be read.
+
+        The result is not affected by the order in which writes
+        happened, only by their offsets.
+        """
+        storage_index, renew_secret, cancel_secret = (
+            new_storage_index(),
+            new_secret(),
+            new_secret(),
+        )
+        (_, allocated) = yield self.storage_server.allocate_buckets(
             storage_index,
             renew_secret,
@@ -205 +251 @@
             canary=Referenceable(),
         )
-        self.assertEqual(already_got, {1, 2})
-
-    @inlineCallbacks
-    def test_written_shares_are_readable(self):
-        """
-        Shares that are fully written to can be read.
-
-        The result is not affected by the order in which writes
-        happened, only by their offsets.
-        """
-        storage_index, renew_secret, cancel_secret = (
-            new_storage_index(),
-            new_secret(),
-            new_secret(),
-        )
-        (_, allocated) = yield self.storage_server.allocate_buckets(
-            storage_index,
-            renew_secret,
-            cancel_secret,
-            sharenums=set(range(5)),
-            allocated_size=1024,
-            canary=Referenceable(),
-        )
 
         # Bucket 1 is fully written in order
@@ -249 +272 @@
         )
 
-    @skipIf(True, "https://tahoe-lafs.org/trac/tahoe-lafs/ticket/3801")
-    def test_overlapping_writes(self):
-        """
-        The policy for overlapping writes is TBD:
-        https://tahoe-lafs.org/trac/tahoe-lafs/ticket/3801
-        """
+    @inlineCallbacks
+    def test_non_matching_overlapping_writes(self):
+        """
+        When doing overlapping writes in immutable uploads, non-matching writes
+        fail.
+        """
+        storage_index, renew_secret, cancel_secret = (
+            new_storage_index(),
+            new_secret(),
+            new_secret(),
+        )
+        (_, allocated) = yield self.storage_server.allocate_buckets(
+            storage_index,
+            renew_secret,
+            cancel_secret,
+            sharenums={0},
+            allocated_size=30,
+            canary=Referenceable(),
+        )
+
+        yield allocated[0].callRemote("write", 0, b"1" * 25)
+        # Overlapping write that doesn't match:
+        with self.assertRaises(RemoteException):
+            yield allocated[0].callRemote("write", 20, b"2" * 10)
+
+    @inlineCallbacks
+    def test_matching_overlapping_writes(self):
+        """
+        When doing overlapping writes in immutable uploads, matching writes
+        succeed.
+        """
+        storage_index, renew_secret, cancel_secret = (
+            new_storage_index(),
+            new_secret(),
+            new_secret(),
+        )
+        (_, allocated) = yield self.storage_server.allocate_buckets(
+            storage_index,
+            renew_secret,
+            cancel_secret,
+            sharenums={0},
+            allocated_size=25,
+            canary=Referenceable(),
+        )
+
+        yield allocated[0].callRemote("write", 0, b"1" * 10)
+        # Overlapping write that matches:
+        yield allocated[0].callRemote("write", 5, b"1" * 20)
+        yield allocated[0].callRemote("close")
+
+        buckets = yield self.storage_server.get_buckets(storage_index)
+        self.assertEqual(set(buckets.keys()), {0})
+
+        self.assertEqual((yield buckets[0].callRemote("read", 0, 25)), b"1" * 25)
+
+    def test_abort(self):
+        """
+        If we call ``abort`` on the ``RIBucketWriter`` to disconnect in the
+        middle of writing to a bucket, all data is wiped, and it's even
+        possible to write different data to the bucket.
+
+        (In the real world one probably wouldn't do that, but writing different
+        data is a good way to test that the original data really was wiped.)
+        """
+        return self.abort_or_disconnect_half_way(
+            lambda bucket: bucket.callRemote("abort")
+        )
+
+    @inlineCallbacks
+    def test_get_buckets_skips_unfinished_buckets(self):
+        """
+        Buckets that are not fully written are not returned by
+        ``IStorageServer.get_buckets()`` implementations.
+        """
+        storage_index = new_storage_index()
+        (_, allocated) = yield self.storage_server.allocate_buckets(
+            storage_index,
+            renew_secret=new_secret(),
+            cancel_secret=new_secret(),
+            sharenums=set(range(5)),
+            allocated_size=10,
+            canary=Referenceable(),
+        )
+
+        # Bucket 1 is fully written
+        yield allocated[1].callRemote("write", 0, b"1" * 10)
+        yield allocated[1].callRemote("close")
+
+        # Bucket 2 is partially written
+        yield allocated[2].callRemote("write", 0, b"1" * 5)
+
+        buckets = yield self.storage_server.get_buckets(storage_index)
+        self.assertEqual(set(buckets.keys()), {1})
+
+    @inlineCallbacks
+    def test_read_bucket_at_offset(self):
+        """
+        Given a read bucket returned from ``IStorageServer.get_buckets()``, it
+        is possible to read at different offsets and lengths, with reads past
+        the end resulting in empty bytes.
+        """
+        length = 256 * 17
+
+        storage_index = new_storage_index()
+        (_, allocated) = yield self.storage_server.allocate_buckets(
+            storage_index,
+            renew_secret=new_secret(),
+            cancel_secret=new_secret(),
+            sharenums=set(range(1)),
+            allocated_size=length,
+            canary=Referenceable(),
+        )
+
+        total_data = _randbytes(256 * 17)
+        yield allocated[0].callRemote("write", 0, total_data)
+        yield allocated[0].callRemote("close")
+
+        buckets = yield self.storage_server.get_buckets(storage_index)
+        bucket = buckets[0]
+        for start, to_read in [
+            (0, 250),  # fraction
+            (0, length),  # whole thing
+            (100, 1024),  # offset fraction
+            (length + 1, 100),  # completely out of bounds
+            (length - 100, 200),  # partially out of bounds
+        ]:
+            data = yield bucket.callRemote("read", start, to_read)
+            self.assertEqual(
+                data,
+                total_data[start : start + to_read],
+                "Didn't match for start {}, length {}".format(start, to_read),
+            )
+
+    @inlineCallbacks
+    def test_bucket_advise_corrupt_share(self):
+        """
+        Calling ``advise_corrupt_share()`` on a bucket returned by
+        ``IStorageServer.get_buckets()`` does not result in error (other
+        behavior is opaque at this level of abstraction).
+        """
+        storage_index = new_storage_index()
+        (_, allocated) = yield self.storage_server.allocate_buckets(
+            storage_index,
+            renew_secret=new_secret(),
+            cancel_secret=new_secret(),
+            sharenums=set(range(1)),
+            allocated_size=10,
+            canary=Referenceable(),
+        )
+
+        yield allocated[0].callRemote("write", 0, b"0123456789")
+        yield allocated[0].callRemote("close")
+
+        buckets = yield self.storage_server.get_buckets(storage_index)
+        yield buckets[0].callRemote("advise_corrupt_share", b"OH NO")
+
+
+class IStorageServerMutableAPIsTestsMixin(object):
+    """
+    Tests for ``IStorageServer``'s mutable APIs.
+
+    ``self.storage_server`` is expected to provide ``IStorageServer``.
+
+    ``STARAW`` is short for ``slot_testv_and_readv_and_writev``.
+    """
+
+    def new_secrets(self):
+        """Return a 3-tuple of secrets for STARAW calls."""
+        return (new_secret(), new_secret(), new_secret())
+
+    def staraw(self, *args, **kwargs):
+        """Like ``slot_testv_and_readv_and_writev``, but less typing."""
+        return self.storage_server.slot_testv_and_readv_and_writev(*args, **kwargs)
+
+    @inlineCallbacks
+    def test_STARAW_reads_after_write(self):
+        """
+        When data is written with
+        ``IStorageServer.slot_testv_and_readv_and_writev``, it can then be read
+        by a separate call using that API.
+        """
+        secrets = self.new_secrets()
+        storage_index = new_storage_index()
+        (written, _) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: ([], [(0, b"abcdefg")], 7),
+                1: ([], [(0, b"0123"), (4, b"456")], 7),
+            },
+            r_vector=[],
+        )
+        self.assertEqual(written, True)
+
+        (_, reads) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={},
+            # Whole thing, partial, going beyond the edge, completely outside
+            # range:
+            r_vector=[(0, 7), (2, 3), (6, 8), (100, 10)],
+        )
+        self.assertEqual(
+            reads,
+            {0: [b"abcdefg", b"cde", b"g", b""], 1: [b"0123456", b"234", b"6", b""]},
+        )
+
+    @inlineCallbacks
+    def test_SATRAW_reads_happen_before_writes_in_single_query(self):
+        """
+        If a ``IStorageServer.slot_testv_and_readv_and_writev`` command
+        contains both reads and writes, the read returns results that precede
+        the write.
+        """
+        secrets = self.new_secrets()
+        storage_index = new_storage_index()
+        (written, _) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: ([], [(0, b"abcdefg")], 7),
+            },
+            r_vector=[],
+        )
+        self.assertEqual(written, True)
+
+        # Read and write in same command; read happens before write:
+        (written, reads) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: ([], [(0, b"X" * 7)], 7),
+            },
+            r_vector=[(0, 7)],
+        )
+        self.assertEqual(written, True)
+        self.assertEqual(reads, {0: [b"abcdefg"]})
+
+        # The write is available in next read:
+        (_, reads) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={},
+            r_vector=[(0, 7)],
+        )
+        self.assertEqual(reads, {0: [b"X" * 7]})
+
+    @inlineCallbacks
+    def test_SATRAW_writes_happens_only_if_test_matches(self):
+        """
+        If a ``IStorageServer.slot_testv_and_readv_and_writev`` includes both a
+        test and a write, the write succeeds if the test matches, and fails if
+        the test does not match.
+        """
+        secrets = self.new_secrets()
+        storage_index = new_storage_index()
+        (written, _) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: ([], [(0, b"1" * 7)], 7),
+            },
+            r_vector=[],
+        )
+        self.assertEqual(written, True)
+
+        # Test matches, so write happens:
+        (written, _) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: (
+                    [(0, 3, b"1" * 3), (3, 4, b"1" * 4)],
+                    [(0, b"2" * 7)],
+                    7,
+                ),
+            },
+            r_vector=[],
+        )
+        self.assertEqual(written, True)
+        (_, reads) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={},
+            r_vector=[(0, 7)],
+        )
+        self.assertEqual(reads, {0: [b"2" * 7]})
+
+        # Test does not match, so write does not happen:
+        (written, _) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: ([(0, 7, b"1" * 7)], [(0, b"3" * 7)], 7),
+            },
+            r_vector=[],
+        )
+        self.assertEqual(written, False)
+        (_, reads) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={},
+            r_vector=[(0, 7)],
+        )
+        self.assertEqual(reads, {0: [b"2" * 7]})
+
+    @inlineCallbacks
+    def test_SATRAW_tests_past_end_of_data(self):
+        """
+        If a ``IStorageServer.slot_testv_and_readv_and_writev`` includes a test
+        vector that reads past the end of the data, the result is limited to
+        actual available data.
+        """
+        secrets = self.new_secrets()
+        storage_index = new_storage_index()
+
+        # Since there is no data on server, the test vector will return empty
+        # string, which matches expected result, so write will succeed.
+        (written, _) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: ([(0, 10, b"")], [(0, b"1" * 7)], 7),
+            },
+            r_vector=[],
+        )
+        self.assertEqual(written, True)
+
+        # Now the test vector is a 10-read off of a 7-byte value, but expected
+        # value is still 7 bytes, so the write will again succeed.
+        (written, _) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: ([(0, 10, b"1" * 7)], [(0, b"2" * 7)], 7),
+            },
+            r_vector=[],
+        )
+        self.assertEqual(written, True)
+
+    @inlineCallbacks
+    def test_SATRAW_reads_past_end_of_data(self):
+        """
+        If a ``IStorageServer.slot_testv_and_readv_and_writev`` reads past the
+        end of the data, the result is limited to actual available data.
+        """
+        secrets = self.new_secrets()
+        storage_index = new_storage_index()
+
+        # Write some data
+        (written, _) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: ([], [(0, b"12345")], 5),
+            },
+            r_vector=[],
+        )
+        self.assertEqual(written, True)
+
+        # Reads past end.
+        (_, reads) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={},
+            r_vector=[(0, 100), (2, 50)],
+        )
+        self.assertEqual(reads, {0: [b"12345", b"345"]})
+
+    @inlineCallbacks
+    def test_STARAW_write_enabler_must_match(self):
+        """
+        If the write enabler secret passed to
+        ``IStorageServer.slot_testv_and_readv_and_writev`` doesn't match
+        previous writes, the write fails.
+        """
+        secrets = self.new_secrets()
+        storage_index = new_storage_index()
+        (written, _) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: ([], [(0, b"1" * 7)], 7),
+            },
+            r_vector=[],
+        )
+        self.assertEqual(written, True)
+
+        # Write enabler secret does not match, so write does not happen:
+        bad_secrets = (new_secret(),) + secrets[1:]
+        with self.assertRaises(RemoteException):
+            yield self.staraw(
+                storage_index,
+                bad_secrets,
+                tw_vectors={
+                    0: ([], [(0, b"2" * 7)], 7),
+                },
+                r_vector=[],
+            )
+        (_, reads) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={},
+            r_vector=[(0, 7)],
+        )
+        self.assertEqual(reads, {0: [b"1" * 7]})
+
+    @inlineCallbacks
+    def test_STARAW_zero_new_length_deletes(self):
+        """
+        A zero new length passed to
+        ``IStorageServer.slot_testv_and_readv_and_writev`` deletes the share.
+        """
+        secrets = self.new_secrets()
+        storage_index = new_storage_index()
+        (written, _) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: ([], [(0, b"1" * 7)], 7),
+            },
+            r_vector=[],
+        )
+        self.assertEqual(written, True)
+
+        # Write with new length of 0:
+        (written, _) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: ([], [(0, b"1" * 7)], 0),
+            },
+            r_vector=[],
+        )
+        self.assertEqual(written, True)
+
+        # It's gone!
+        (_, reads) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={},
+            r_vector=[(0, 7)],
+        )
+        self.assertEqual(reads, {})
+
+    @inlineCallbacks
+    def test_slot_readv(self):
+        """
+        Data written with ``IStorageServer.slot_testv_and_readv_and_writev()``
+        can be read using ``IStorageServer.slot_readv()``.  Reads can't go past
+        the end of the data.
+        """
+        secrets = self.new_secrets()
+        storage_index = new_storage_index()
+        (written, _) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: ([], [(0, b"abcdefg")], 7),
+                1: ([], [(0, b"0123"), (4, b"456")], 7),
+                # This will never get read from, just here to show we only read
+                # from shares explicitly requested by slot_readv:
+                2: ([], [(0, b"XYZW")], 4),
+            },
+            r_vector=[],
+        )
+        self.assertEqual(written, True)
+
+        reads = yield self.storage_server.slot_readv(
+            storage_index,
+            shares=[0, 1],
+            # Whole thing, partial, going beyond the edge, completely outside
+            # range:
+            readv=[(0, 7), (2, 3), (6, 8), (100, 10)],
+        )
+        self.assertEqual(
+            reads,
+            {0: [b"abcdefg", b"cde", b"g", b""], 1: [b"0123456", b"234", b"6", b""]},
+        )
+
+    @inlineCallbacks
+    def test_slot_readv_no_shares(self):
+        """
+        With no shares given, ``IStorageServer.slot_readv()`` reads from all shares.
+        """
+        secrets = self.new_secrets()
+        storage_index = new_storage_index()
+        (written, _) = yield self.staraw(
+            storage_index,
+            secrets,
+            tw_vectors={
+                0: ([], [(0, b"abcdefg")], 7),
+                1: ([], [(0, b"0123456")], 7),
+                2: ([], [(0, b"9876543")], 7),
+            },
+            r_vector=[],
+        )
+        self.assertEqual(written, True)
+
+        reads = yield self.storage_server.slot_readv(
+            storage_index,
+            shares=[],
+            readv=[(0, 7)],
+        )
+        self.assertEqual(
+            reads,
+            {0: [b"abcdefg"], 1: [b"0123456"], 2: [b"9876543"]},
+        )
 
 
 class _FoolscapMixin(SystemTestMixin):
     """Run tests on Foolscap version of ``IStorageServer."""
+
+    def _get_native_server(self):
+        return next(iter(self.clients[0].storage_broker.get_known_servers()))
 
     @inlineCallbacks
@@ -266 +794 @@
         yield SystemTestMixin.setUp(self)
         yield self.set_up_nodes(1)
-        self.storage_server = next(
-            iter(self.clients[0].storage_broker.get_known_servers())
-        ).get_storage_server()
+        self.storage_server = self._get_native_server().get_storage_server()
         self.assertTrue(IStorageServer.providedBy(self.storage_server))
 
@@ -275 +801 @@
         AsyncTestCase.tearDown(self)
         yield SystemTestMixin.tearDown(self)
+
+    @inlineCallbacks
+    def disconnect(self):
+        """
+        Disconnect and then reconnect with a new ``IStorageServer``.
+        """
+        current = self.storage_server
+        yield self.bounce_client(0)
+        self.storage_server = self._get_native_server().get_storage_server()
+        assert self.storage_server is not current
 
 
@@ -287 +823 @@
 ):
     """Foolscap-specific tests for immutable ``IStorageServer`` APIs."""
+
+
+class FoolscapMutableAPIsTests(
+    _FoolscapMixin, IStorageServerMutableAPIsTestsMixin, AsyncTestCase
+):
+    """Foolscap-specific tests for immutable ``IStorageServer`` APIs."""

src/allmydata/test/test_storage.py ¶

@@ -9 +9 @@
 from __future__ import unicode_literals
 
-from future.utils import native_str, PY2, bytes_to_native_str
+from future.utils import native_str, PY2, bytes_to_native_str, bchr
 if PY2:
     from future.builtins import filter, map, zip, ascii, chr, hex, input, next, oct, open, pow, round, super, bytes, dict, list, object, range, str, max, min  # noqa: F401
@@ -20 +20 @@
 import struct
 import shutil
-import gc
+from uuid import uuid4
 
 from twisted.trial import unittest
@@ -26 +26 @@
 from twisted.internet import defer
 from twisted.internet.task import Clock
+
+from hypothesis import given, strategies
 
 import itertools
@@ -34 +36 @@
 from allmydata.storage.mutable import MutableShareFile
 from allmydata.storage.immutable import BucketWriter, BucketReader, ShareFile
-from allmydata.storage.common import DataTooLargeError, storage_index_to_dir, \
+from allmydata.storage.common import storage_index_to_dir, \
      UnknownMutableContainerVersionError, UnknownImmutableContainerVersionError, \
      si_b2a, si_a2b
@@ -48 +50 @@
                                      VERIFICATION_KEY_SIZE, \
                                      SHARE_HASH_CHAIN_SIZE
-from allmydata.interfaces import BadWriteEnablerError
+from allmydata.interfaces import (
+    BadWriteEnablerError, DataTooLargeError, ConflictingWriteError,
+)
 from allmydata.test.no_network import NoNetworkServer
 from allmydata.storage_client import (
@@ -125 +129 @@
     def test_create(self):
         incoming, final = self.make_workdir("test_create")
-        bw = BucketWriter(self, incoming, final, 200, self.make_lease(),
-                          FakeCanary())
+        bw = BucketWriter(self, incoming, final, 200, self.make_lease())
         bw.remote_write(0, b"a"*25)
         bw.remote_write(25, b"b"*25)
@@ -135 +138 @@
     def test_readwrite(self):
         incoming, final = self.make_workdir("test_readwrite")
-        bw = BucketWriter(self, incoming, final, 200, self.make_lease(),
-                          FakeCanary())
+        bw = BucketWriter(self, incoming, final, 200, self.make_lease())
         bw.remote_write(0, b"a"*25)
         bw.remote_write(25, b"b"*25)
@@ -147 +149 @@
         self.failUnlessEqual(br.remote_read(25, 25), b"b"*25)
         self.failUnlessEqual(br.remote_read(50, 7), b"c"*7)
+
+    def test_write_past_size_errors(self):
+        """Writing beyond the size of the bucket throws an exception."""
+        for (i, (offset, length)) in enumerate([(0, 201), (10, 191), (202, 34)]):
+            incoming, final = self.make_workdir(
+                "test_write_past_size_errors-{}".format(i)
+            )
+            bw = BucketWriter(self, incoming, final, 200, self.make_lease())
+            with self.assertRaises(DataTooLargeError):
+                bw.remote_write(offset, b"a" * length)
+
+    @given(
+        maybe_overlapping_offset=strategies.integers(min_value=0, max_value=98),
+        maybe_overlapping_length=strategies.integers(min_value=1, max_value=100),
+    )
+    def test_overlapping_writes_ok_if_matching(
+            self, maybe_overlapping_offset, maybe_overlapping_length
+    ):
+        """
+        Writes that overlap with previous writes are OK when the content is the
+        same.
+        """
+        length = 100
+        expected_data = b"".join(bchr(i) for i in range(100))
+        incoming, final = self.make_workdir("overlapping_writes_{}".format(uuid4()))
+        bw = BucketWriter(
+            self, incoming, final, length, self.make_lease(),
+        )
+        # Three writes: 10-19, 30-39, 50-59. This allows for a bunch of holes.
+        bw.remote_write(10, expected_data[10:20])
+        bw.remote_write(30, expected_data[30:40])
+        bw.remote_write(50, expected_data[50:60])
+        # Then, an overlapping write but with matching data:
+        bw.remote_write(
+            maybe_overlapping_offset,
+            expected_data[
+                maybe_overlapping_offset:maybe_overlapping_offset + maybe_overlapping_length
+            ]
+        )
+        # Now fill in the holes:
+        bw.remote_write(0, expected_data[0:10])
+        bw.remote_write(20, expected_data[20:30])
+        bw.remote_write(40, expected_data[40:50])
+        bw.remote_write(60, expected_data[60:])
+        bw.remote_close()
+
+        br = BucketReader(self, bw.finalhome)
+        self.assertEqual(br.remote_read(0, length), expected_data)
+
+
+    @given(
+        maybe_overlapping_offset=strategies.integers(min_value=0, max_value=98),
+        maybe_overlapping_length=strategies.integers(min_value=1, max_value=100),
+    )
+    def test_overlapping_writes_not_ok_if_different(
+            self, maybe_overlapping_offset, maybe_overlapping_length
+    ):
+        """
+        Writes that overlap with previous writes fail with an exception if the
+        contents don't match.
+        """
+        length = 100
+        incoming, final = self.make_workdir("overlapping_writes_{}".format(uuid4()))
+        bw = BucketWriter(
+            self, incoming, final, length, self.make_lease(),
+        )
+        # Three writes: 10-19, 30-39, 50-59. This allows for a bunch of holes.
+        bw.remote_write(10, b"1" * 10)
+        bw.remote_write(30, b"1" * 10)
+        bw.remote_write(50, b"1" * 10)
+        # Then, write something that might overlap with some of them, but
+        # conflicts. Then fill in holes left by first three writes. Conflict is
+        # inevitable.
+        with self.assertRaises(ConflictingWriteError):
+            bw.remote_write(
+                maybe_overlapping_offset,
+                b'X' * min(maybe_overlapping_length, length - maybe_overlapping_offset),
+            )
+            bw.remote_write(0, b"1" * 10)
+            bw.remote_write(20, b"1" * 10)
+            bw.remote_write(40, b"1" * 10)
+            bw.remote_write(60, b"1" * 40)
 
     def test_read_past_end_of_share_data(self):
@@ -229 +313 @@
         fileutil.make_dirs(basedir)
         fileutil.make_dirs(os.path.join(basedir, "tmp"))
-        bw = BucketWriter(self, incoming, final, size, self.make_lease(),
-                          FakeCanary())
+        bw = BucketWriter(self, incoming, final, size, self.make_lease())
         rb = RemoteBucket(bw)
         return bw, rb, final
@@ -580 +663 @@
         OVERHEAD = 3*4
         LEASE_SIZE = 4+32+32+4
-        canary = FakeCanary(True)
+        canary = FakeCanary()
         already, writers = self.allocate(ss, b"vid1", [0,1,2], 1000, canary)
         self.failUnlessEqual(len(writers), 3)
         # now the StorageServer should have 3000 bytes provisionally
         # allocated, allowing only 2000 more to be claimed
-        self.failUnlessEqual(len(ss._active_writers), 3)
+        self.failUnlessEqual(len(ss._bucket_writers), 3)
 
         # allocating 1001-byte shares only leaves room for one
-        already2, writers2 = self.allocate(ss, b"vid2", [0,1,2], 1001, canary)
+        canary2 = FakeCanary()
+        already2, writers2 = self.allocate(ss, b"vid2", [0,1,2], 1001, canary2)
         self.failUnlessEqual(len(writers2), 1)
-        self.failUnlessEqual(len(ss._active_writers), 4)
+        self.failUnlessEqual(len(ss._bucket_writers), 4)
 
         # we abandon the first set, so their provisional allocation should be
         # returned
-
-        del already
-        del writers
-        gc.collect()
-
-        self.failUnlessEqual(len(ss._active_writers), 1)
+        canary.disconnected()
+
+        self.failUnlessEqual(len(ss._bucket_writers), 1)
         # now we have a provisional allocation of 1001 bytes
 
@@ -608 +689 @@
             bw.remote_write(0, b"a"*25)
             bw.remote_close()
-        del already2
-        del writers2
-        del bw
-        self.failUnlessEqual(len(ss._active_writers), 0)
+        self.failUnlessEqual(len(ss._bucket_writers), 0)
 
         # this also changes the amount reported as available by call_get_disk_stats
@@ -618 +696 @@
         # now there should be ALLOCATED=1001+12+72=1085 bytes allocated, and
         # 5000-1085=3915 free, therefore we can fit 39 100byte shares
-        already3, writers3 = self.allocate(ss, b"vid3", list(range(100)), 100, canary)
+        canary3 = FakeCanary()
+        already3, writers3 = self.allocate(ss, b"vid3", list(range(100)), 100, canary3)
         self.failUnlessEqual(len(writers3), 39)
-        self.failUnlessEqual(len(ss._active_writers), 39)
-
-        del already3
-        del writers3
-        gc.collect()
-
-        self.failUnlessEqual(len(ss._active_writers), 0)
+        self.failUnlessEqual(len(ss._bucket_writers), 39)
+
+        canary3.disconnected()
+
+        self.failUnlessEqual(len(ss._bucket_writers), 0)
         ss.disownServiceParent()
         del ss
@@ -1075 +1152 @@
         self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [data]})
 
-        # as should this one
-        answer = write(b"si1", secrets,
-                       {0: ([(10, 5, b"lt", b"11111"),
-                             ],
-                            [(0, b"x"*100)],
-                            None),
-                        },
-                       [(10,5)],
-                       )
-        self.failUnlessEqual(answer, (False,
-                                      {0: [b"11111"],
-                                       1: [b""],
-                                       2: [b""]},
-                                      ))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [data]})
-
-
     def test_operators(self):
         # test operators, the data we're comparing is '11111' in all cases.
@@ -1109 +1169 @@
                   [])
 
-        reset()
-
-        #  lt
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"lt", b"11110"),
-                                             ],
-                                            [(0, b"x"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (False, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [data]})
-        self.failUnlessEqual(read(b"si1", [], [(0,100)]), {0: [data]})
-        reset()
-
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"lt", b"11111"),
-                                             ],
-                                            [(0, b"x"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (False, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [data]})
-        reset()
-
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"lt", b"11112"),
-                                             ],
-                                            [(0, b"y"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (True, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [b"y"*100]})
-        reset()
-
-        #  le
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"le", b"11110"),
-                                             ],
-                                            [(0, b"x"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (False, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [data]})
-        reset()
-
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"le", b"11111"),
-                                             ],
-                                            [(0, b"y"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (True, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [b"y"*100]})
-        reset()
-
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"le", b"11112"),
-                                             ],
-                                            [(0, b"y"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (True, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [b"y"*100]})
         reset()
 
@@ -1185 +1188 @@
         self.failUnlessEqual(answer, (True, {0: [b"11111"]}))
         self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [b"y"*100]})
-        reset()
-
-        #  ne
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"ne", b"11111"),
-                                             ],
-                                            [(0, b"x"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (False, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [data]})
-        reset()
-
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"ne", b"11112"),
-                                             ],
-                                            [(0, b"y"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (True, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [b"y"*100]})
-        reset()
-
-        #  ge
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"ge", b"11110"),
-                                             ],
-                                            [(0, b"y"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (True, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [b"y"*100]})
-        reset()
-
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"ge", b"11111"),
-                                             ],
-                                            [(0, b"y"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (True, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [b"y"*100]})
-        reset()
-
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"ge", b"11112"),
-                                             ],
-                                            [(0, b"y"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (False, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [data]})
-        reset()
-
-        #  gt
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"gt", b"11110"),
-                                             ],
-                                            [(0, b"y"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (True, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [b"y"*100]})
-        reset()
-
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"gt", b"11111"),
-                                             ],
-                                            [(0, b"x"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (False, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [data]})
-        reset()
-
-        answer = write(b"si1", secrets, {0: ([(10, 5, b"gt", b"11112"),
-                                             ],
-                                            [(0, b"x"*100)],
-                                            None,
-                                            )}, [(10,5)])
-        self.failUnlessEqual(answer, (False, {0: [b"11111"]}))
-        self.failUnlessEqual(read(b"si1", [0], [(0,100)]), {0: [data]})
         reset()
 

src/allmydata/web/status.py ¶

@@ -15 +15 @@
 import itertools
 import hashlib
+import re
 from twisted.internet import defer
 from twisted.python.filepath import FilePath
@@ -1552 +1553 @@
         return json.dumps(stats, indent=1) + "\n"
 
+    @render_exception
+    def render_OPENMETRICS(self, req):
+        """
+        Render our stats in `OpenMetrics <https://openmetrics.io/>` format.
+        For example Prometheus and Victoriametrics can parse this.
+        Point the scraper to ``/statistics?t=openmetrics`` (instead of the
+        default ``/metrics``).
+        """
+        req.setHeader("content-type", "application/openmetrics-text; version=1.0.0; charset=utf-8")
+        stats = self._provider.get_stats()
+        ret = []
+
+        def mangle_name(name):
+            return re.sub(
+                u"_(\d\d)_(\d)_percentile",
+                u'{quantile="0.\g<1>\g<2>"}',
+                name.replace(u".", u"_")
+            )
+
+        def mangle_value(val):
+            return str(val) if val is not None else u"NaN"
+
+        for (k, v) in sorted(stats['counters'].items()):
+            ret.append(u"tahoe_counters_%s %s" % (mangle_name(k), mangle_value(v)))
+        for (k, v) in sorted(stats['stats'].items()):
+            ret.append(u"tahoe_stats_%s %s" % (mangle_name(k), mangle_value(v)))
+
+        ret.append(u"# EOF\n")
+
+        return u"\n".join(ret)
+
 class StatisticsElement(Element):