Have MutableShare file only write a new lease if there is room for it

This is analagous to the earlier ShareFile change.

src/allmydata/storage/mutable.py

@@ -13,7 +13,10 @@ if PY2:
 
 import os, stat, struct
 
-from allmydata.interfaces import BadWriteEnablerError
+from allmydata.interfaces import (
+    BadWriteEnablerError,
+    NoSpace,
+)
 from allmydata.util import idlib, log
 from allmydata.util.assertutil import precondition
 from allmydata.util.hashutil import timing_safe_compare
@@ -289,7 +292,19 @@ class MutableShareFile(object):
             except IndexError:
                 return
 
-    def add_lease(self, lease_info):
+    def add_lease(self, available_space, lease_info):
+        """
+        Add a new lease to this share.
+
+        :param int available_space: The maximum number of bytes of storage to
+            commit in this operation.  If more than this number of bytes is
+            required, raise ``NoSpace`` instead.
+
+        :raise NoSpace: If more than ``available_space`` bytes is required to
+            complete the operation.  In this case, no lease is added.
+
+        :return: ``None``
+        """
         precondition(lease_info.owner_num != 0) # 0 means "no lease here"
         with open(self.home, 'rb+') as f:
             num_lease_slots = self._get_num_lease_slots(f)
@@ -297,6 +312,8 @@ class MutableShareFile(object):
             if empty_slot is not None:
                 self._write_lease_record(f, empty_slot, lease_info)
             else:
+                if lease_info.mutable_size() > available_space:
+                    raise NoSpace()
                 self._write_lease_record(f, num_lease_slots, lease_info)
 
     def renew_lease(self, renew_secret, new_expire_time):
@@ -321,13 +338,13 @@ class MutableShareFile(object):
         msg += " ."
         raise IndexError(msg)
 
-    def add_or_renew_lease(self, lease_info):
+    def add_or_renew_lease(self, available_space, lease_info):
         precondition(lease_info.owner_num != 0) # 0 means "no lease here"
         try:
             self.renew_lease(lease_info.renew_secret,
                              lease_info.expiration_time)
         except IndexError:
-            self.add_lease(lease_info)
+            self.add_lease(available_space, lease_info)
 
     def cancel_lease(self, cancel_secret):
         """Remove any leases with the given cancel_secret. If the last lease

src/allmydata/test/common_storage.py

@@ -31,3 +31,35 @@ def upload_immutable(storage_server, storage_index, renew_secret, cancel_secret,
     for shnum, writer in writers.items():
         writer.remote_write(0, shares[shnum])
         writer.remote_close()
+
+
+def upload_mutable(storage_server, storage_index, secrets, shares):
+    """
+    Synchronously upload some mutable shares to a ``StorageServer``.
+
+    :param allmydata.storage.server.StorageServer storage_server: The storage
+        server object to use to perform the upload.
+
+    :param bytes storage_index: The storage index for the immutable shares.
+
+    :param secrets: A three-tuple of a write enabler, renew secret, and cancel
+        secret.
+
+    :param dict[int, bytes] shares: A mapping from share numbers to share data
+        to upload.
+
+    :return: ``None``
+    """
+    test_and_write_vectors = {
+        sharenum: ([], [(0, data)], None)
+        for sharenum, data
+        in shares.items()
+    }
+    read_vector = []
+
+    storage_server.remote_slot_testv_and_readv_and_writev(
+        storage_index,
+        secrets,
+        test_and_write_vectors,
+        read_vector,
+    )

src/allmydata/test/test_storage.py

@@ -73,6 +73,7 @@ from .common import (
 from .common_util import FakeCanary
 from .common_storage import (
     upload_immutable,
+    upload_mutable,
 )
 from .strategies import (
     offsets,
@@ -698,9 +699,9 @@ class Server(unittest.TestCase):
 
     def test_reserved_space_immutable_lease(self):
         """
-        If there is not enough available space to store an additional lease then
-        ``remote_add_lease`` fails with ``NoSpace`` when an attempt is made to
-        use it to create a new lease.
+        If there is not enough available space to store an additional lease on an
+        immutable share then ``remote_add_lease`` fails with ``NoSpace`` when
+        an attempt is made to use it to create a new lease.
         """
         disk = FakeDisk(total=1024, used=0)
         self.patch(fileutil, "get_disk_stats", disk.get_disk_stats)
@@ -722,6 +723,46 @@ class Server(unittest.TestCase):
         with self.assertRaises(interfaces.NoSpace):
             ss.remote_add_lease(storage_index, renew_secret, cancel_secret)
 
+    def test_reserved_space_mutable_lease(self):
+        """
+        If there is not enough available space to store an additional lease on a
+        mutable share then ``remote_add_lease`` fails with ``NoSpace`` when an
+        attempt is made to use it to create a new lease.
+        """
+        disk = FakeDisk(total=1024, used=0)
+        self.patch(fileutil, "get_disk_stats", disk.get_disk_stats)
+
+        ss = self.create("test_reserved_space_mutable_lease")
+
+        renew_secrets = iter(
+            "{}{}".format("r" * 31, i).encode("ascii")
+            for i
+            in range(5)
+        )
+
+        storage_index = b"x" * 16
+        write_enabler = b"w" * 32
+        cancel_secret = b"c" * 32
+        secrets = (write_enabler, next(renew_secrets), cancel_secret)
+        shares = {0: b"y" * 500}
+        upload_mutable(ss, storage_index, secrets, shares)
+
+        # use up all the available space
+        disk.use(disk.available)
+
+        # The upload created one lease.  There is room for three more leases
+        # in the share header.  Even if we're out of disk space, on a boring
+        # enough filesystem we can write these.
+        for i in range(3):
+            ss.remote_add_lease(storage_index, next(renew_secrets), cancel_secret)
+
+        # Having used all of the space for leases in the header, we would have
+        # to allocate storage for the next lease.  Since there is no space
+        # available, this must fail instead.
+        with self.assertRaises(interfaces.NoSpace):
+            ss.remote_add_lease(storage_index, next(renew_secrets), cancel_secret)
+
+
     def test_reserved_space(self):
         reserved = 10000
         allocated = 0