diff --git a/src/borg/helpers.py b/src/borg/helpers.py
index 5e6b74cdc..e6c805bc6 100644
--- a/src/borg/helpers.py
+++ b/src/borg/helpers.py
@@ -269,6 +269,17 @@ def get_keys_dir():
     return keys_dir
 
 
+def get_nonces_dir():
+    """Determine where to store the local nonce high watermark"""
+
+    xdg_config = os.environ.get('XDG_CONFIG_HOME', os.path.join(get_home_dir(), '.config'))
+    nonces_dir = os.environ.get('BORG_NONCES_DIR', os.path.join(xdg_config, 'borg', 'key-nonces'))
+    if not os.path.exists(nonces_dir):
+        os.makedirs(nonces_dir)
+        os.chmod(nonces_dir, stat.S_IRWXU)
+    return nonces_dir
+
+
 def get_cache_dir():
     """Determine where to repository keys and cache"""
     xdg_cache = os.environ.get('XDG_CACHE_HOME', os.path.join(get_home_dir(), '.cache'))
diff --git a/src/borg/key.py b/src/borg/key.py
index 11a50f80e..849b99e4a 100644
--- a/src/borg/key.py
+++ b/src/borg/key.py
@@ -3,7 +3,7 @@ import getpass
 import os
 import sys
 import textwrap
-from binascii import a2b_base64, b2a_base64, hexlify
+from binascii import a2b_base64, b2a_base64, hexlify, unhexlify
 from hashlib import sha256, pbkdf2_hmac
 from hmac import compare_digest
 
@@ -23,6 +23,7 @@ from .helpers import bin_to_hex
 from .helpers import CompressionDecider2, CompressionSpec
 from .item import Key, EncryptedKey
 from .platform import SaveFile
+from .nonces import NonceManager
 
 
 PREFIX = b'\0' * 8
@@ -169,6 +170,7 @@ class AESKeyBase(KeyBase):
 
     def encrypt(self, chunk):
         chunk = self.compress(chunk)
+        self.nonce_manager.ensure_reservation(num_aes_blocks(len(chunk.data)))
         self.enc_cipher.reset()
         data = b''.join((self.enc_cipher.iv[8:], self.enc_cipher.encrypt(chunk.data)))
         hmac = hmac_sha256(self.enc_hmac_key, data)
@@ -207,8 +209,9 @@ class AESKeyBase(KeyBase):
         if self.chunk_seed & 0x80000000:
             self.chunk_seed = self.chunk_seed - 0xffffffff - 1
 
-    def init_ciphers(self, enc_iv=b''):
-        self.enc_cipher = AES(is_encrypt=True, key=self.enc_key, iv=enc_iv)
+    def init_ciphers(self, manifest_nonce=0):
+        self.enc_cipher = AES(is_encrypt=True, key=self.enc_key, iv=manifest_nonce.to_bytes(16, byteorder='big'))
+        self.nonce_manager = NonceManager(self.repository, self.enc_cipher, manifest_nonce)
         self.dec_cipher = AES(is_encrypt=False, key=self.enc_key)
 
 
@@ -299,7 +302,7 @@ class PassphraseKey(AESKeyBase):
             try:
                 key.decrypt(None, manifest_data)
                 num_blocks = num_aes_blocks(len(manifest_data) - 41)
-                key.init_ciphers(PREFIX + long_to_bytes(key.extract_nonce(manifest_data) + num_blocks))
+                key.init_ciphers(key.extract_nonce(manifest_data) + num_blocks)
                 return key
             except IntegrityError:
                 passphrase = Passphrase.getpass(prompt)
@@ -337,7 +340,7 @@ class KeyfileKeyBase(AESKeyBase):
             if not key.load(target, passphrase):
                 raise PassphraseWrong
         num_blocks = num_aes_blocks(len(manifest_data) - 41)
-        key.init_ciphers(PREFIX + long_to_bytes(key.extract_nonce(manifest_data) + num_blocks))
+        key.init_ciphers(key.extract_nonce(manifest_data) + num_blocks)
         return key
 
     def find_key(self):
diff --git a/src/borg/nonces.py b/src/borg/nonces.py
new file mode 100644
index 000000000..4f9299581
--- /dev/null
+++ b/src/borg/nonces.py
@@ -0,0 +1,87 @@
+import os
+import sys
+from binascii import unhexlify
+
+from .crypto import bytes_to_long, long_to_bytes
+from .helpers import get_nonces_dir
+from .helpers import bin_to_hex
+from .platform import SaveFile
+from .remote import InvalidRPCMethod
+
+
+MAX_REPRESENTABLE_NONCE = 2**64 - 1
+NONCE_SPACE_RESERVATION = 2**28  # This in units of AES blocksize (16 bytes)
+
+
+class NonceManager:
+    def __init__(self, repository, enc_cipher, manifest_nonce):
+        self.repository = repository
+        self.enc_cipher = enc_cipher
+        self.end_of_nonce_reservation = None
+        self.manifest_nonce = manifest_nonce
+        self.nonce_file = os.path.join(get_nonces_dir(), self.repository.id_str)
+
+    def get_local_free_nonce(self):
+        try:
+            with open(self.nonce_file, 'r') as fd:
+                return bytes_to_long(unhexlify(fd.read()))
+        except FileNotFoundError:
+            return None
+
+    def commit_local_nonce_reservation(self, next_unreserved, start_nonce):
+        if self.get_local_free_nonce() != start_nonce:
+            raise Exception("nonce space reservation with mismatched previous state")
+        with SaveFile(self.nonce_file, binary=False) as fd:
+            fd.write(bin_to_hex(long_to_bytes(next_unreserved)))
+
+    def get_repo_free_nonce(self):
+        try:
+            return self.repository.get_free_nonce()
+        except InvalidRPCMethod as error:
+            # old server version, suppress further calls
+            sys.stderr.write("Please upgrade to borg version 1.1+ on the server for safer AES-CTR nonce handling.\n")
+            self.get_repo_free_nonce = lambda: None
+            self.commit_repo_nonce_reservation = lambda next_unreserved, start_nonce: None
+            return None
+
+    def commit_repo_nonce_reservation(self, next_unreserved, start_nonce):
+        self.repository.commit_nonce_reservation(next_unreserved, start_nonce)
+
+    def ensure_reservation(self, nonce_space_needed):
+        # Nonces may never repeat, even if a transaction aborts or the system crashes.
+        # Therefore a part of the nonce space is reserved before any nonce is used for encryption.
+        # As these reservations are commited to permanent storage before any nonce is used, this protects
+        # against nonce reuse in crashes and transaction aborts. In that case the reservation still
+        # persists and the whole reserved space is never reused.
+        #
+        # Local storage on the client is used to protect against an attacker that is able to rollback the
+        # state of the server or can do arbitrary modifications to the repository.
+        # Storage on the server is used for the multi client use case where a transaction on client A is
+        # aborted and later client B writes to the repository.
+        #
+        # This scheme does not protect against attacker who is able to rollback the state of the server
+        # or can do arbitrary modifications to the repository in the multi client usecase.
+
+        if self.end_of_nonce_reservation:
+            # we already got a reservation, if nonce_space_needed still fits everything is ok
+            next_nonce = int.from_bytes(self.enc_cipher.iv, byteorder='big')
+            assert next_nonce <= self.end_of_nonce_reservation
+            if next_nonce + nonce_space_needed <= self.end_of_nonce_reservation:
+                return
+
+        repo_free_nonce = self.get_repo_free_nonce()
+        local_free_nonce = self.get_local_free_nonce()
+        free_nonce_space = max(x for x in (repo_free_nonce, local_free_nonce, self.manifest_nonce, self.end_of_nonce_reservation) if x is not None)
+        reservation_end = free_nonce_space + nonce_space_needed + NONCE_SPACE_RESERVATION
+        assert reservation_end < MAX_REPRESENTABLE_NONCE
+        if self.end_of_nonce_reservation is None:
+            # initialization, reset the encryption cipher to the start of the reservation
+            self.enc_cipher.reset(None, free_nonce_space.to_bytes(16, byteorder='big'))
+        else:
+            # expand existing reservation if possible
+            if free_nonce_space != self.end_of_nonce_reservation:
+                # some other client got an interleaved reservation, skip partial space in old reservation to avoid overlap
+                self.enc_cipher.reset(None, free_nonce_space.to_bytes(16, byteorder='big'))
+        self.commit_repo_nonce_reservation(reservation_end, repo_free_nonce)
+        self.commit_local_nonce_reservation(reservation_end, local_free_nonce)
+        self.end_of_nonce_reservation = reservation_end
diff --git a/src/borg/remote.py b/src/borg/remote.py
index 604506cd0..4632a50a5 100644
--- a/src/borg/remote.py
+++ b/src/borg/remote.py
@@ -66,6 +66,8 @@ class RepositoryServer:  # pragma: no cover
         'save_key',
         'load_key',
         'break_lock',
+        'get_free_nonce',
+        'commit_nonce_reservation'
     )
 
     def __init__(self, restrict_to_paths, append_only):
@@ -450,6 +452,12 @@ This problem will go away as soon as the server has been upgraded to 1.0.7+.
     def load_key(self):
         return self.call('load_key')
 
+    def get_free_nonce(self):
+        return self.call('get_free_nonce')
+
+    def commit_nonce_reservation(self, next_unreserved, start_nonce):
+        return self.call('commit_nonce_reservation', next_unreserved, start_nonce)
+
     def break_lock(self):
         return self.call('break_lock')
 
diff --git a/src/borg/repository.py b/src/borg/repository.py
index 9eebd90e6..6c0159a72 100644
--- a/src/borg/repository.py
+++ b/src/borg/repository.py
@@ -189,6 +189,27 @@ class Repository:
         keydata = self.config.get('repository', 'key')
         return keydata.encode('utf-8')  # remote repo: msgpack issue #99, returning bytes
 
+    def get_free_nonce(self):
+        if not self.lock.got_exclusive_lock():
+            raise AssertionError("bug in code, exclusive lock should exist here")
+
+        nonce_path = os.path.join(self.path, 'nonce')
+        try:
+            with open(nonce_path, 'r') as fd:
+                return int.from_bytes(unhexlify(fd.read()), byteorder='big')
+        except FileNotFoundError:
+            return None
+
+    def commit_nonce_reservation(self, next_unreserved, start_nonce):
+        if not self.lock.got_exclusive_lock():
+            raise AssertionError("bug in code, exclusive lock should exist here")
+
+        if self.get_free_nonce() != start_nonce:
+            raise Exception("nonce space reservation with mismatched previous state")
+        nonce_path = os.path.join(self.path, 'nonce')
+        with SaveFile(nonce_path, binary=False) as fd:
+            fd.write(bin_to_hex(next_unreserved.to_bytes(8, byteorder='big')))
+
     def destroy(self):
         """Destroy the repository at `self.path`
         """
diff --git a/src/borg/testsuite/archiver.py b/src/borg/testsuite/archiver.py
index dfa540e01..d5d6f5a59 100644
--- a/src/borg/testsuite/archiver.py
+++ b/src/borg/testsuite/archiver.py
@@ -1493,7 +1493,6 @@ class ArchiverTestCase(ArchiverTestCaseBase):
         verify_uniqueness()
         self.cmd('delete', self.repository_location + '::test.2')
         verify_uniqueness()
-        self.assert_equal(used, set(range(len(used))))
 
     def test_aes_counter_uniqueness_keyfile(self):
         self.verify_aes_counter_uniqueness('keyfile')
diff --git a/src/borg/testsuite/helpers.py b/src/borg/testsuite/helpers.py
index 5f5a3806c..6583b4ead 100644
--- a/src/borg/testsuite/helpers.py
+++ b/src/borg/testsuite/helpers.py
@@ -14,7 +14,7 @@ from ..helpers import Buffer
 from ..helpers import partial_format, format_file_size, parse_file_size, format_timedelta, format_line, PlaceholderError
 from ..helpers import make_path_safe, clean_lines
 from ..helpers import prune_within, prune_split
-from ..helpers import get_cache_dir, get_keys_dir
+from ..helpers import get_cache_dir, get_keys_dir, get_nonces_dir
 from ..helpers import is_slow_msgpack
 from ..helpers import yes, TRUISH, FALSISH, DEFAULTISH
 from ..helpers import StableDict, int_to_bigint, bigint_to_int, bin_to_hex
@@ -636,6 +636,17 @@ def test_get_keys_dir():
         os.environ['BORG_KEYS_DIR'] = old_env
 
 
+def test_get_nonces_dir(monkeypatch):
+    """test that get_nonces_dir respects environment"""
+    monkeypatch.delenv('XDG_CONFIG_HOME', raising=False)
+    monkeypatch.delenv('BORG_NONCES_DIR', raising=False)
+    assert get_nonces_dir() == os.path.join(os.path.expanduser('~'), '.config', 'borg', 'key-nonces')
+    monkeypatch.setenv('XDG_CONFIG_HOME', '/var/tmp/.config')
+    assert get_nonces_dir() == os.path.join('/var/tmp/.config', 'borg', 'key-nonces')
+    monkeypatch.setenv('BORG_NONCES_DIR', '/var/tmp')
+    assert get_nonces_dir() == '/var/tmp'
+
+
 def test_file_size():
     """test the size formatting routines"""
     si_size_map = {
diff --git a/src/borg/testsuite/key.py b/src/borg/testsuite/key.py
index b85650a4f..94b455396 100644
--- a/src/borg/testsuite/key.py
+++ b/src/borg/testsuite/key.py
@@ -1,6 +1,7 @@
 import getpass
 import re
 import tempfile
+import os.path
 from binascii import hexlify, unhexlify
 
 import pytest
@@ -9,6 +10,7 @@ from ..crypto import bytes_to_long, num_aes_blocks
 from ..helpers import Location
 from ..helpers import Chunk
 from ..helpers import IntegrityError
+from ..helpers import get_nonces_dir
 from ..key import PlaintextKey, PassphraseKey, KeyfileKey, Passphrase, PasswordRetriesExceeded, bin_to_hex
 
 
@@ -18,6 +20,11 @@ def clean_env(monkeypatch):
     monkeypatch.delenv('BORG_PASSPHRASE', False)
 
 
+@pytest.fixture(autouse=True)
+def nonce_dir(tmpdir_factory, monkeypatch):
+    monkeypatch.setenv('XDG_CONFIG_HOME', tmpdir_factory.mktemp('xdg-config-home'))
+
+
 class TestKey:
     class MockArgs:
         location = Location(tempfile.mkstemp()[1])
@@ -59,6 +66,12 @@ class TestKey:
         id = bytes(32)
         id_str = bin_to_hex(id)
 
+        def get_free_nonce(self):
+            return None
+
+        def commit_nonce_reservation(self, next_unreserved, start_nonce):
+            pass
+
     def test_plaintext(self):
         key = PlaintextKey.create(None, None)
         chunk = Chunk(b'foo')
@@ -77,13 +90,23 @@ class TestKey:
         assert key.extract_nonce(manifest2) == 1
         iv = key.extract_nonce(manifest)
         key2 = KeyfileKey.detect(self.MockRepository(), manifest)
-        assert bytes_to_long(key2.enc_cipher.iv, 8) == iv + num_aes_blocks(len(manifest) - KeyfileKey.PAYLOAD_OVERHEAD)
+        assert bytes_to_long(key2.enc_cipher.iv, 8) >= iv + num_aes_blocks(len(manifest) - KeyfileKey.PAYLOAD_OVERHEAD)
         # Key data sanity check
         assert len({key2.id_key, key2.enc_key, key2.enc_hmac_key}) == 3
         assert key2.chunk_seed != 0
         chunk = Chunk(b'foo')
         assert chunk == key2.decrypt(key.id_hash(chunk.data), key.encrypt(chunk))
 
+    def test_keyfile_nonce_rollback_protection(self, monkeypatch, keys_dir):
+        monkeypatch.setenv('BORG_PASSPHRASE', 'test')
+        repository = self.MockRepository()
+        with open(os.path.join(get_nonces_dir(), repository.id_str), "w") as fd:
+            fd.write("0000000000002000")
+        key = KeyfileKey.create(repository, self.MockArgs())
+        data = key.encrypt(Chunk(b'ABC'))
+        assert key.extract_nonce(data) == 0x2000
+        assert key.decrypt(None, data).data == b'ABC'
+
     def test_keyfile_kfenv(self, tmpdir, monkeypatch):
         keyfile = tmpdir.join('keyfile')
         monkeypatch.setenv('BORG_KEY_FILE', str(keyfile))
diff --git a/src/borg/testsuite/nonces.py b/src/borg/testsuite/nonces.py
new file mode 100644
index 000000000..88405f56d
--- /dev/null
+++ b/src/borg/testsuite/nonces.py
@@ -0,0 +1,242 @@
+import os.path
+
+import pytest
+
+from ..helpers import get_nonces_dir
+from ..key import bin_to_hex
+from ..nonces import NonceManager
+from ..remote import InvalidRPCMethod
+
+from .. import nonces  # for monkey patching NONCE_SPACE_RESERVATION
+
+
+@pytest.fixture(autouse=True)
+def clean_env(monkeypatch):
+    # Workaround for some tests (testsuite/archiver) polluting the environment
+    monkeypatch.delenv('BORG_PASSPHRASE', False)
+
+
+@pytest.fixture(autouse=True)
+def nonce_dir(tmpdir_factory, monkeypatch):
+    monkeypatch.setenv('XDG_CONFIG_HOME', tmpdir_factory.mktemp('xdg-config-home'))
+
+
+class TestNonceManager:
+
+    class MockRepository:
+        class _Location:
+            orig = '/some/place'
+
+        _location = _Location()
+        id = bytes(32)
+        id_str = bin_to_hex(id)
+
+        def get_free_nonce(self):
+            return self.next_free
+
+        def commit_nonce_reservation(self, next_unreserved, start_nonce):
+            assert start_nonce == self.next_free
+            self.next_free = next_unreserved
+
+    class MockOldRepository(MockRepository):
+        def get_free_nonce(self):
+            raise InvalidRPCMethod("")
+
+        def commit_nonce_reservation(self, next_unreserved, start_nonce):
+            pytest.fail("commit_nonce_reservation should never be called on an old repository")
+
+    class MockEncCipher:
+        def __init__(self, iv):
+            self.iv_set = False  # placeholder, this is never a valid iv
+            self.iv = iv
+
+        def reset(self, key, iv):
+            assert key is None
+            assert iv is not False
+            self.iv_set = iv
+            self.iv = iv
+
+        def expect_iv_and_advance(self, expected_iv, advance):
+            expected_iv = expected_iv.to_bytes(16, byteorder='big')
+            iv_set = self.iv_set
+            assert iv_set == expected_iv
+            self.iv_set = False
+            self.iv = advance.to_bytes(16, byteorder='big')
+
+        def expect_no_reset_and_advance(self, advance):
+            iv_set = self.iv_set
+            assert iv_set is False
+            self.iv = advance.to_bytes(16, byteorder='big')
+
+    def setUp(self):
+        self.repository = None
+
+    def cache_nonce(self):
+        with open(os.path.join(get_nonces_dir(), self.repository.id_str), "r") as fd:
+            return fd.read()
+
+    def set_cache_nonce(self, nonce):
+        with open(os.path.join(get_nonces_dir(), self.repository.id_str), "w") as fd:
+            assert fd.write(nonce)
+
+    def test_empty_cache_and_old_server(self, monkeypatch):
+        monkeypatch.setattr(nonces, 'NONCE_SPACE_RESERVATION', 0x20)
+
+        enc_cipher = self.MockEncCipher(0x2000)
+        self.repository = self.MockOldRepository()
+        manager = NonceManager(self.repository, enc_cipher, 0x2000)
+        manager.ensure_reservation(19)
+        enc_cipher.expect_iv_and_advance(0x2000, 0x2013)
+
+        assert self.cache_nonce() == "0000000000002033"
+
+    def test_empty_cache(self, monkeypatch):
+        monkeypatch.setattr(nonces, 'NONCE_SPACE_RESERVATION', 0x20)
+
+        enc_cipher = self.MockEncCipher(0x2000)
+        self.repository = self.MockRepository()
+        self.repository.next_free = 0x2000
+        manager = NonceManager(self.repository, enc_cipher, 0x2000)
+        manager.ensure_reservation(19)
+        enc_cipher.expect_iv_and_advance(0x2000, 0x2013)
+
+        assert self.cache_nonce() == "0000000000002033"
+
+    def test_empty_nonce(self, monkeypatch):
+        monkeypatch.setattr(nonces, 'NONCE_SPACE_RESERVATION', 0x20)
+
+        enc_cipher = self.MockEncCipher(0x2000)
+        self.repository = self.MockRepository()
+        self.repository.next_free = None
+        manager = NonceManager(self.repository, enc_cipher, 0x2000)
+        manager.ensure_reservation(19)
+        enc_cipher.expect_iv_and_advance(0x2000, 0x2000 + 19)
+
+        assert self.cache_nonce() == "0000000000002033"
+        assert self.repository.next_free == 0x2033
+
+        # enough space in reservation
+        manager.ensure_reservation(13)
+        enc_cipher.expect_no_reset_and_advance(0x2000 + 19 + 13)
+        assert self.cache_nonce() == "0000000000002033"
+        assert self.repository.next_free == 0x2033
+
+        # just barely enough space in reservation
+        manager.ensure_reservation(19)
+        enc_cipher.expect_no_reset_and_advance(0x2000 + 19 + 13 + 19)
+        assert self.cache_nonce() == "0000000000002033"
+        assert self.repository.next_free == 0x2033
+
+        # no space in reservation
+        manager.ensure_reservation(16)
+        enc_cipher.expect_no_reset_and_advance(0x2000 + 19 + 13 + 19 + 16)
+        assert self.cache_nonce() == "0000000000002063"
+        assert self.repository.next_free == 0x2063
+
+        # spans reservation boundary
+        manager.ensure_reservation(64)
+        enc_cipher.expect_no_reset_and_advance(0x2000 + 19 + 13 + 19 + 16 + 64)
+        assert self.cache_nonce() == "00000000000020c3"
+        assert self.repository.next_free == 0x20c3
+
+    def test_sync_nonce(self, monkeypatch):
+        monkeypatch.setattr(nonces, 'NONCE_SPACE_RESERVATION', 0x20)
+
+        enc_cipher = self.MockEncCipher(0x2000)
+        self.repository = self.MockRepository()
+        self.repository.next_free = 0x2000
+        self.set_cache_nonce("0000000000002000")
+
+        manager = NonceManager(self.repository, enc_cipher, 0x2000)
+        manager.ensure_reservation(19)
+        enc_cipher.expect_iv_and_advance(0x2000, 0x2000 + 19)
+
+        assert self.cache_nonce() == "0000000000002033"
+        assert self.repository.next_free == 0x2033
+
+    def test_server_just_upgraded(self, monkeypatch):
+        monkeypatch.setattr(nonces, 'NONCE_SPACE_RESERVATION', 0x20)
+
+        enc_cipher = self.MockEncCipher(0x2000)
+        self.repository = self.MockRepository()
+        self.repository.next_free = None
+        self.set_cache_nonce("0000000000002000")
+
+        manager = NonceManager(self.repository, enc_cipher, 0x2000)
+        manager.ensure_reservation(19)
+        enc_cipher.expect_iv_and_advance(0x2000, 0x2000 + 19)
+
+        assert self.cache_nonce() == "0000000000002033"
+        assert self.repository.next_free == 0x2033
+
+    def test_transaction_abort_no_cache(self, monkeypatch):
+        monkeypatch.setattr(nonces, 'NONCE_SPACE_RESERVATION', 0x20)
+
+        enc_cipher = self.MockEncCipher(0x1000)
+        self.repository = self.MockRepository()
+        self.repository.next_free = 0x2000
+
+        manager = NonceManager(self.repository, enc_cipher, 0x2000)
+        manager.ensure_reservation(19)
+        enc_cipher.expect_iv_and_advance(0x2000, 0x2000 + 19)
+
+        assert self.cache_nonce() == "0000000000002033"
+        assert self.repository.next_free == 0x2033
+
+    def test_transaction_abort_old_server(self, monkeypatch):
+        monkeypatch.setattr(nonces, 'NONCE_SPACE_RESERVATION', 0x20)
+
+        enc_cipher = self.MockEncCipher(0x1000)
+        self.repository = self.MockOldRepository()
+        self.set_cache_nonce("0000000000002000")
+
+        manager = NonceManager(self.repository, enc_cipher, 0x2000)
+        manager.ensure_reservation(19)
+        enc_cipher.expect_iv_and_advance(0x2000, 0x2000 + 19)
+
+        assert self.cache_nonce() == "0000000000002033"
+
+    def test_transaction_abort_on_other_client(self, monkeypatch):
+        monkeypatch.setattr(nonces, 'NONCE_SPACE_RESERVATION', 0x20)
+
+        enc_cipher = self.MockEncCipher(0x1000)
+        self.repository = self.MockRepository()
+        self.repository.next_free = 0x2000
+        self.set_cache_nonce("0000000000001000")
+
+        manager = NonceManager(self.repository, enc_cipher, 0x2000)
+        manager.ensure_reservation(19)
+        enc_cipher.expect_iv_and_advance(0x2000, 0x2000 + 19)
+
+        assert self.cache_nonce() == "0000000000002033"
+        assert self.repository.next_free == 0x2033
+
+    def test_interleaved(self, monkeypatch):
+        monkeypatch.setattr(nonces, 'NONCE_SPACE_RESERVATION', 0x20)
+
+        enc_cipher = self.MockEncCipher(0x2000)
+        self.repository = self.MockRepository()
+        self.repository.next_free = 0x2000
+        self.set_cache_nonce("0000000000002000")
+
+        manager = NonceManager(self.repository, enc_cipher, 0x2000)
+        manager.ensure_reservation(19)
+        enc_cipher.expect_iv_and_advance(0x2000, 0x2000 + 19)
+
+        assert self.cache_nonce() == "0000000000002033"
+        assert self.repository.next_free == 0x2033
+
+        # somehow the clients unlocks, another client reserves and this client relocks
+        self.repository.next_free = 0x4000
+
+        # enough space in reservation
+        manager.ensure_reservation(12)
+        enc_cipher.expect_no_reset_and_advance(0x2000 + 19 + 12)
+        assert self.cache_nonce() == "0000000000002033"
+        assert self.repository.next_free == 0x4000
+
+        # spans reservation boundary
+        manager.ensure_reservation(21)
+        enc_cipher.expect_iv_and_advance(0x4000, 0x4000 + 21)
+        assert self.cache_nonce() == "0000000000004035"
+        assert self.repository.next_free == 0x4035
diff --git a/src/borg/testsuite/repository.py b/src/borg/testsuite/repository.py
index 620bcaf3e..90c46f4d7 100644
--- a/src/borg/testsuite/repository.py
+++ b/src/borg/testsuite/repository.py
@@ -390,6 +390,48 @@ class RepositoryFreeSpaceTestCase(RepositoryTestCaseBase):
                 self.repository.commit()
 
 
+class NonceReservation(RepositoryTestCaseBase):
+    def test_get_free_nonce_asserts(self):
+        self.reopen(exclusive=False)
+        with pytest.raises(AssertionError):
+            with self.repository:
+                self.repository.get_free_nonce()
+
+    def test_get_free_nonce(self):
+        with self.repository:
+            assert self.repository.get_free_nonce() is None
+
+            with open(os.path.join(self.repository.path, "nonce"), "w") as fd:
+                fd.write("0000000000000000")
+            assert self.repository.get_free_nonce() == 0
+
+            with open(os.path.join(self.repository.path, "nonce"), "w") as fd:
+                fd.write("5000000000000000")
+            assert self.repository.get_free_nonce() == 0x5000000000000000
+
+    def test_commit_nonce_reservation_asserts(self):
+        self.reopen(exclusive=False)
+        with pytest.raises(AssertionError):
+            with self.repository:
+                self.repository.commit_nonce_reservation(0x200, 0x100)
+
+    def test_commit_nonce_reservation(self):
+        with self.repository:
+            with pytest.raises(Exception):
+                self.repository.commit_nonce_reservation(0x200, 15)
+
+            self.repository.commit_nonce_reservation(0x200, None)
+            with open(os.path.join(self.repository.path, "nonce"), "r") as fd:
+                assert fd.read() == "0000000000000200"
+
+            with pytest.raises(Exception):
+                self.repository.commit_nonce_reservation(0x200, 15)
+
+            self.repository.commit_nonce_reservation(0x400, 0x200)
+            with open(os.path.join(self.repository.path, "nonce"), "r") as fd:
+                assert fd.read() == "0000000000000400"
+
+
 class RepositoryAuxiliaryCorruptionTestCase(RepositoryTestCaseBase):
     def setUp(self):
         super().setUp()