Repository: borgbackup/borg

Issue:
borg create without syncing cache
code exists, needs testing, integration, but not a priority now

Discussion:
codecov-io: # [Codecov](https://codecov.io/gh/borgbackup/borg/pull/2654?src=pr&el=h1) Report
> Merging [#2654](https://codecov.io/gh/borgbackup/borg/pull/2654?src=pr&el=desc) into [master](https://codecov.io/gh/borgbackup/borg/commit/86363dcd4b218ed2d81686d2fc5ff1f064b8a07a?src=pr&el=desc) will **decrease** coverage by `0.06%`.
> The diff coverage is `78.66%`.

[![Impacted file tree graph](https://codecov.io/gh/borgbackup/borg/pull/2654/graphs/tree.svg?width=650&height=150&src=pr&token=LssEpmW3o4)](https://codecov.io/gh/borgbackup/borg/pull/2654?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master    #2654      +/-   ##
==========================================
- Coverage   83.99%   83.92%   -0.07%     
==========================================
  Files          23       23              
  Lines        8565     8696     +131     
  Branches     1445     1462      +17     
==========================================
+ Hits         7194     7298     +104     
- Misses        981     1003      +22     
- Partials      390      395       +5
```


| [Impacted Files](https://codecov.io/gh/borgbackup/borg/pull/2654?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [src/borg/repository.py](https://codecov.io/gh/borgbackup/borg/pull/2654?src=pr&el=tree#diff-c3JjL2JvcmcvcmVwb3NpdG9yeS5weQ==) | `87.51% <100%> (ø)` | :arrow_up: |
| [src/borg/archiver.py](https://codecov.io/gh/borgbackup/borg/pull/2654?src=pr&el=tree#diff-c3JjL2JvcmcvYXJjaGl2ZXIucHk=) | `82.99% <100%> (ø)` | :arrow_up: |
| [src/borg/helpers.py](https://codecov.io/gh/borgbackup/borg/pull/2654?src=pr&el=tree#diff-c3JjL2JvcmcvaGVscGVycy5weQ==) | `86.97% <60%> (+0.02%)` | :arrow_up: |
| [src/borg/cache.py](https://codecov.io/gh/borgbackup/borg/pull/2654?src=pr&el=tree#diff-c3JjL2JvcmcvY2FjaGUucHk=) | `85.97% <79.02%> (-1.87%)` | :arrow_down: |
| [src/borg/remote.py](https://codecov.io/gh/borgbackup/borg/pull/2654?src=pr&el=tree#diff-c3JjL2JvcmcvcmVtb3RlLnB5) | `80.06% <0%> (+0.33%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/borgbackup/borg/pull/2654?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/borgbackup/borg/pull/2654?src=pr&el=footer). Last update [86363dc...77bfc45](https://codecov.io/gh/borgbackup/borg/pull/2654?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

ThomasWaldmann: See also #2357.
ThomasWaldmann: hmm, if there is no files cache, that means all backup data set files will have to be read, chunked, all chunks id-hashed and looked up, right?

this will have quite an impact for all bigger files (maybe not so much for tiny files where other processing overhead is bigger than the time needed to read/chunk/hash).
enkore: It would be possible to introduce a new files cache format (`files.v2`?) that addresses it by including full ChunkListEntries, which won't be in this PR.

See https://github.com/borgbackup/borg/issues/2357#issuecomment-292162083
textshell: I think this should at least not memorize broken csize into the archive items. (i.e. Item.get_size should ignore memorize=True if any chunk has a 0 csize.
textshell: I'm not really happy with something in a release creating archives with somewhat bogus metadata without a clear plan forward. This could mean that we are forced to carry code to resolve missing csize for a long time.
Also i think this really should not disable the files cache. Using the files cache or not should be an orthogonal choice. It feels it shouldn't be hard to make that possible. 
enkore: I'd recommend you read #2313 and #2357.
enkore: Anything left to do here?
ThomasWaldmann: did you see my comment about "32" hardcoded?
enkore: That's in a different patch.
ThomasWaldmann: I'll have to look it through again.

But my impression was that the discussion with @textshell was not finished, was it?
enkore: I wrote all the answers done months ago. I don't see anything left to discuss.
enkore: Just for the sake of closure...

> I think this should at least not memorize broken csize into the archive items. (i.e. Item.get_size should ignore memorize=True if any chunk has a 0 csize.

That never happened. 36fcbd6

> This could mean that we are forced to carry code to resolve missing csize for a long time.

(1) Borg already generates inconsistent statistics (see recreate warnings) (2) it's literally just statistics. I do not see Borg carrying the extra code around to accommodate this, should it be dropped again.

>  Also i think this really should not disable the files cache. Using the files cache or not should be an orthogonal choice. It feels it shouldn't be hard to make that possible.

https://github.com/borgbackup/borg/issues/2313#issuecomment-300438677

Patch:
diff --git a/docs/internals/frontends.rst b/docs/internals/frontends.rst
index 4000bede..c41d427e 100644
--- a/docs/internals/frontends.rst
+++ b/docs/internals/frontends.rst
@@ -504,6 +504,7 @@ Errors
 
 Operations
     - cache.begin_transaction
+    - cache.download_chunks, appears with ``borg create --no-cache-sync``
     - cache.commit
     - cache.sync
 
diff --git a/src/borg/archiver.py b/src/borg/archiver.py
index 4536b83d..3275bf0b 100644
--- a/src/borg/archiver.py
+++ b/src/borg/archiver.py
@@ -504,7 +504,7 @@ def create_inner(archive, cache):
         t0_monotonic = time.monotonic()
         if not dry_run:
             with Cache(repository, key, manifest, do_files=args.cache_files, progress=args.progress,
-                       lock_wait=self.lock_wait) as cache:
+                       lock_wait=self.lock_wait, permit_adhoc_cache=args.no_cache_sync) as cache:
                 archive = Archive(repository, key, manifest, args.location.archive, cache=cache,
                                   create=True, checkpoint_interval=args.checkpoint_interval,
                                   numeric_owner=args.numeric_owner, noatime=args.noatime, noctime=args.noctime,
@@ -2826,6 +2826,8 @@ def define_common_options(add_common_option):
                                help='only display items with the given status characters')
         subparser.add_argument('--json', action='store_true',
                                help='output stats as JSON (implies --stats)')
+        subparser.add_argument('--no-cache-sync', dest='no_cache_sync', action='store_true',
+                               help='experimental: do not synchronize the cache. Implies --no-files-cache.')
 
         exclude_group = subparser.add_argument_group('Exclusion options')
         exclude_group.add_argument('-e', '--exclude', dest='patterns',
diff --git a/src/borg/cache.py b/src/borg/cache.py
index 70d6f029..47b53deb 100644
--- a/src/borg/cache.py
+++ b/src/borg/cache.py
@@ -4,6 +4,7 @@
 import stat
 from binascii import unhexlify
 from collections import namedtuple
+from time import perf_counter
 
 import msgpack
 
@@ -30,6 +31,7 @@
 from .locking import Lock
 from .platform import SaveFile
 from .remote import cache_if_remote
+from .repository import LIST_SCAN_LIMIT
 
 FileCacheEntry = namedtuple('FileCacheEntry', 'age inode size mtime chunk_ids')
 
@@ -347,6 +349,69 @@ def destroy(repository, path=None):
             os.remove(config)  # kill config first
             shutil.rmtree(path)
 
+    def __new__(cls, repository, key, manifest, path=None, sync=True, do_files=False, warn_if_unencrypted=True,
+                progress=False, lock_wait=None, permit_adhoc_cache=False):
+        def local():
+            return LocalCache(repository=repository, key=key, manifest=manifest, path=path, sync=sync,
+                              do_files=do_files, warn_if_unencrypted=warn_if_unencrypted, progress=progress,
+                              lock_wait=lock_wait)
+
+        def adhoc():
+            return AdHocCache(repository=repository, key=key, manifest=manifest)
+
+        if not permit_adhoc_cache:
+            return local()
+
+        # ad-hoc cache may be permitted, but if the local cache is in sync it'd be stupid to invalidate
+        # it by needlessly using the ad-hoc cache.
+        # Check if the local cache exists and is in sync.
+
+        cache_config = CacheConfig(repository, path, lock_wait)
+        if cache_config.exists():
+            with cache_config:
+                cache_in_sync = cache_config.manifest_id == manifest.id
+            # Don't nest cache locks
+            if cache_in_sync:
+                # Local cache is in sync, use it
+                logger.debug('Cache: choosing local cache (in sync)')
+                return local()
+        logger.debug('Cache: choosing ad-hoc cache (local cache does not exist or is not in sync)')
+        return adhoc()
+
+
+class CacheStatsMixin:
+    str_format = """\
+All archives:   {0.total_size:>20s} {0.total_csize:>20s} {0.unique_csize:>20s}
+
+                       Unique chunks         Total chunks
+Chunk index:    {0.total_unique_chunks:20d} {0.total_chunks:20d}"""
+
+    def __str__(self):
+        return self.str_format.format(self.format_tuple())
+
+    Summary = namedtuple('Summary', ['total_size', 'total_csize', 'unique_size', 'unique_csize', 'total_unique_chunks',
+                                     'total_chunks'])
+
+    def stats(self):
+        # XXX: this should really be moved down to `hashindex.pyx`
+        stats = self.Summary(*self.chunks.summarize())._asdict()
+        return stats
+
+    def format_tuple(self):
+        stats = self.stats()
+        for field in ['total_size', 'total_csize', 'unique_csize']:
+            stats[field] = format_file_size(stats[field])
+        return self.Summary(**stats)
+
+    def chunks_stored_size(self):
+        return self.stats()['unique_csize']
+
+
+class LocalCache(CacheStatsMixin):
+    """
+    Persistent, local (client-side) cache.
+    """
+
     def __init__(self, repository, key, manifest, path=None, sync=True, do_files=False, warn_if_unencrypted=True,
                  progress=False, lock_wait=None):
         """
@@ -394,31 +459,6 @@ def __enter__(self):
     def __exit__(self, exc_type, exc_val, exc_tb):
         self.close()
 
-    def __str__(self):
-        fmt = """\
-All archives:   {0.total_size:>20s} {0.total_csize:>20s} {0.unique_csize:>20s}
-
-                       Unique chunks         Total chunks
-Chunk index:    {0.total_unique_chunks:20d} {0.total_chunks:20d}"""
-        return fmt.format(self.format_tuple())
-
-    Summary = namedtuple('Summary', ['total_size', 'total_csize', 'unique_size', 'unique_csize', 'total_unique_chunks',
-                                     'total_chunks'])
-
-    def stats(self):
-        # XXX: this should really be moved down to `hashindex.pyx`
-        stats = self.Summary(*self.chunks.summarize())._asdict()
-        return stats
-
-    def format_tuple(self):
-        stats = self.stats()
-        for field in ['total_size', 'total_csize', 'unique_csize']:
-            stats[field] = format_file_size(stats[field])
-        return self.Summary(**stats)
-
-    def chunks_stored_size(self):
-        return self.stats()['unique_csize']
-
     def create(self):
         """Create a new empty cache at `self.path`
         """
@@ -547,10 +587,14 @@ def sync(self):
         archive indexes.
         """
         archive_path = os.path.join(self.path, 'chunks.archive.d')
+        # An index of chunks whose size had to be fetched
+        chunks_fetched_size_index = ChunkIndex()
         # Instrumentation
         processed_item_metadata_bytes = 0
         processed_item_metadata_chunks = 0
         compact_chunks_archive_saved_space = 0
+        fetched_chunks_for_csize = 0
+        fetched_bytes_for_csize = 0
 
         def mkpath(id, suffix=''):
             id_hex = bin_to_hex(id)
@@ -588,6 +632,39 @@ def cleanup_cached_archive(id, cleanup_compact=True):
             except FileNotFoundError:
                 pass
 
+        def fetch_missing_csize(chunk_idx):
+            """
+            Archives created with AdHocCache will have csize=0 in all chunk list entries whose
+            chunks were already in the repository.
+
+            Scan *chunk_idx* for entries where csize=0 and fill in the correct information.
+            """
+            nonlocal fetched_chunks_for_csize
+            nonlocal fetched_bytes_for_csize
+
+            all_missing_ids = chunk_idx.zero_csize_ids()
+            fetch_ids = []
+            if len(chunks_fetched_size_index):
+                for id_ in all_missing_ids:
+                    already_fetched_entry = chunks_fetched_size_index.get(id_)
+                    if already_fetched_entry:
+                        entry = chunk_idx[id_]._replace(csize=already_fetched_entry.csize)
+                        assert entry.size == already_fetched_entry.size, 'Chunk size mismatch'
+                        chunk_idx[id_] = entry
+                    else:
+                        fetch_ids.append(id_)
+            else:
+                fetch_ids = all_missing_ids
+
+            # This is potentially a rather expensive operation, but it's hard to tell at this point
+            # if it's a problem in practice (hence the experimental status of --no-cache-sync).
+            for id_, data in zip(fetch_ids, decrypted_repository.repository.get_many(fetch_ids)):
+                entry = chunk_idx[id_]._replace(csize=len(data))
+                chunk_idx[id_] = entry
+                chunks_fetched_size_index[id_] = entry
+                fetched_chunks_for_csize += 1
+                fetched_bytes_for_csize += len(data)
+
         def fetch_and_build_idx(archive_id, decrypted_repository, chunk_idx):
             nonlocal processed_item_metadata_bytes
             nonlocal processed_item_metadata_chunks
@@ -603,6 +680,7 @@ def fetch_and_build_idx(archive_id, decrypted_repository, chunk_idx):
                 processed_item_metadata_chunks += 1
                 sync.feed(data)
             if self.do_cache:
+                fetch_missing_csize(chunk_idx)
                 write_archive_index(archive_id, chunk_idx)
 
         def write_archive_index(archive_id, chunk_idx):
@@ -698,8 +776,13 @@ def create_master_idx(chunk_idx):
                         chunk_idx = chunk_idx or ChunkIndex(master_index_capacity)
                         logger.info('Fetching archive index for %s ...', archive_name)
                         fetch_and_build_idx(archive_id, decrypted_repository, chunk_idx)
+                if not self.do_cache:
+                    fetch_missing_csize(chunk_idx)
                 pi.finish()
-                logger.debug('Cache sync: processed %s bytes (%d chunks) of metadata',
+                logger.debug('Cache sync: had to fetch %s (%d chunks) because no archive had a csize set for them '
+                             '(due to --no-cache-sync)',
+                             format_file_size(fetched_bytes_for_csize), fetched_chunks_for_csize)
+                logger.debug('Cache sync: processed %s (%d chunks) of metadata',
                              format_file_size(processed_item_metadata_bytes), processed_item_metadata_chunks)
                 logger.debug('Cache sync: compact chunks.archive.d storage saved %s bytes',
                              format_file_size(compact_chunks_archive_saved_space))
@@ -843,3 +926,143 @@ def memorize_file(self, path_hash, st, ids):
         entry = FileCacheEntry(age=0, inode=st.st_ino, size=st.st_size, mtime=int_to_bigint(mtime_ns), chunk_ids=ids)
         self.files[path_hash] = msgpack.packb(entry)
         self._newest_mtime = max(self._newest_mtime or 0, mtime_ns)
+
+
+class AdHocCache(CacheStatsMixin):
+    """
+    Ad-hoc, non-persistent cache.
+
+    Compared to the standard LocalCache the AdHocCache does not maintain accurate reference count,
+    nor does it provide a files cache (which would require persistence). Chunks that were not added
+    during the current AdHocCache lifetime won't have correct size/csize set (0 bytes) and will
+    have an infinite reference count (MAX_VALUE).
+    """
+
+    str_format = """\
+All archives:                unknown              unknown              unknown
+
+                       Unique chunks         Total chunks
+Chunk index:    {0.total_unique_chunks:20d}             unknown"""
+
+    def __init__(self, repository, key, manifest, warn_if_unencrypted=True):
+        self.repository = repository
+        self.key = key
+        self.manifest = manifest
+        self._txn_active = False
+
+        self.security_manager = SecurityManager(repository)
+        self.security_manager.assert_secure(manifest, key)
+
+        logger.warning('Note: --no-cache-sync is an experimental feature.')
+
+    # Public API
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        pass
+
+    files = None
+    do_files = False
+
+    def file_known_and_unchanged(self, path_hash, st, ignore_inode=False):
+        return None
+
+    def memorize_file(self, path_hash, st, ids):
+        pass
+
+    def add_chunk(self, id, chunk, stats, overwrite=False, wait=True):
+        assert not overwrite, 'AdHocCache does not permit overwrites — trying to use it for recreate?'
+        if not self._txn_active:
+            self._begin_txn()
+        size = len(chunk)
+        refcount = self.seen_chunk(id, size)
+        if refcount:
+            return self.chunk_incref(id, stats, size_=size)
+        data = self.key.encrypt(chunk)
+        csize = len(data)
+        self.repository.put(id, data, wait=wait)
+        self.chunks.add(id, 1, size, csize)
+        stats.update(size, csize, not refcount)
+        return ChunkListEntry(id, size, csize)
+
+    def seen_chunk(self, id, size=None):
+        if not self._txn_active:
+            self._begin_txn()
+        entry = self.chunks.get(id, ChunkIndexEntry(0, None, None))
+        if entry.refcount and size and not entry.size:
+            # The LocalCache has existing size information and uses *size* to make an effort at detecting collisions.
+            # This is of course not possible for the AdHocCache.
+            # Here *size* is used to update the chunk's size information, which will be zero for existing chunks.
+            self.chunks[id] = entry._replace(size=size)
+        return entry.refcount
+
+    def chunk_incref(self, id, stats, size_=None):
+        if not self._txn_active:
+            self._begin_txn()
+        count, size, csize = self.chunks.incref(id)
+        stats.update(size or size_, csize, False)
+        # When size is 0 and size_ is not given, then this chunk has not been locally visited yet (seen_chunk with
+        # size or add_chunk); we can't add references to those (size=0 is invalid) and generally don't try to.
+        assert size or size_
+        return ChunkListEntry(id, size or size_, csize)
+
+    def chunk_decref(self, id, stats, wait=True):
+        if not self._txn_active:
+            self._begin_txn()
+        count, size, csize = self.chunks.decref(id)
+        if count == 0:
+            del self.chunks[id]
+            self.repository.delete(id, wait=wait)
+            stats.update(-size, -csize, True)
+        else:
+            stats.update(-size, -csize, False)
+
+    def commit(self):
+        if not self._txn_active:
+            return
+        self.security_manager.save(self.manifest, self.key)
+        self._txn_active = False
+
+    def rollback(self):
+        self._txn_active = False
+        del self.chunks
+
+    # Private API
+
+    def _begin_txn(self):
+        self._txn_active = True
+        # Explicitly set the initial hash table capacity to avoid performance issues
+        # due to hash table "resonance".
+        # Since we're creating an archive, add 10 % from the start.
+        num_chunks = len(self.repository)
+        capacity = int(num_chunks / ChunkIndex.MAX_LOAD_FACTOR * 1.1)
+        self.chunks = ChunkIndex(capacity)
+        pi = ProgressIndicatorPercent(total=num_chunks, msg='Downloading chunk list... %3.0f%%',
+                                      msgid='cache.download_chunks')
+        t0 = perf_counter()
+        num_requests = 0
+        marker = None
+        while True:
+            result = self.repository.list(limit=LIST_SCAN_LIMIT, marker=marker)
+            num_requests += 1
+            if not result:
+                break
+            pi.show(increase=len(result))
+            marker = result[-1]
+            # All chunks from the repository have a refcount of MAX_VALUE, which is sticky,
+            # therefore we can't/won't delete them. Chunks we added ourselves in this transaction
+            # (e.g. checkpoint archives) are tracked correctly.
+            init_entry = ChunkIndexEntry(refcount=ChunkIndex.MAX_VALUE, size=0, csize=0)
+            for id_ in result:
+                self.chunks[id_] = init_entry
+        assert len(self.chunks) == num_chunks
+        # LocalCache does not contain the manifest, either.
+        del self.chunks[self.manifest.MANIFEST_ID]
+        duration = perf_counter() - t0
+        pi.finish()
+        logger.debug('AdHocCache: downloaded %d chunk IDs in %.2f s (%d requests), ~%s/s',
+                     num_chunks, duration, num_requests, format_file_size(num_chunks * 34 / duration))
+        # Chunk IDs in a list are encoded in 34 bytes: 1 byte msgpack header, 1 byte length, 32 ID bytes.
+        # Protocol overhead is neglected in this calculation.
diff --git a/src/borg/hashindex.pyx b/src/borg/hashindex.pyx
index b8e86d14..0d271ad6 100644
--- a/src/borg/hashindex.pyx
+++ b/src/borg/hashindex.pyx
@@ -8,8 +8,9 @@ from libc.stdint cimport uint32_t, UINT32_MAX, uint64_t
 from libc.errno cimport errno
 from cpython.exc cimport PyErr_SetFromErrnoWithFilename
 from cpython.buffer cimport PyBUF_SIMPLE, PyObject_GetBuffer, PyBuffer_Release
+from cpython.bytes cimport PyBytes_FromStringAndSize
 
-API_VERSION = '1.1_05'
+API_VERSION = '1.1_06'
 
 
 cdef extern from "_hashindex.c":
@@ -410,6 +411,22 @@ cdef class ChunkIndex(IndexBase):
                 break
             self._add(key, <uint32_t*> (key + self.key_size))
 
+    def zero_csize_ids(self):
+        cdef void *key = NULL
+        cdef uint32_t *values
+        entries = []
+        while True:
+            key = hashindex_next_key(self.index, key)
+            if not key:
+                break
+            values = <uint32_t*> (key + self.key_size)
+            refcount = _le32toh(values[0])
+            assert refcount <= _MAX_VALUE, "invalid reference count"
+            if _le32toh(values[2]) == 0:
+                # csize == 0
+                entries.append(PyBytes_FromStringAndSize(<char*> key, self.key_size))
+        return entries
+
 
 cdef class ChunkKeyIterator:
     cdef ChunkIndex idx
diff --git a/src/borg/helpers.py b/src/borg/helpers.py
index 9100bedf..870d2021 100644
--- a/src/borg/helpers.py
+++ b/src/borg/helpers.py
@@ -131,7 +131,7 @@ class MandatoryFeatureUnsupported(Error):
 
 def check_extension_modules():
     from . import platform, compress, item
-    if hashindex.API_VERSION != '1.1_05':
+    if hashindex.API_VERSION != '1.1_06':
         raise ExtensionModuleError
     if chunker.API_VERSION != '1.1_01':
         raise ExtensionModuleError
@@ -2010,7 +2010,7 @@ def default(self, o):
         from .repository import Repository
         from .remote import RemoteRepository
         from .archive import Archive
-        from .cache import Cache
+        from .cache import LocalCache, AdHocCache
         if isinstance(o, Repository) or isinstance(o, RemoteRepository):
             return {
                 'id': bin_to_hex(o.id),
@@ -2018,11 +2018,15 @@ def default(self, o):
             }
         if isinstance(o, Archive):
             return o.info()
-        if isinstance(o, Cache):
+        if isinstance(o, LocalCache):
             return {
                 'path': o.path,
                 'stats': o.stats(),
             }
+        if isinstance(o, AdHocCache):
+            return {
+                'stats': o.stats(),
+            }
         return super().default(o)
 
 
diff --git a/src/borg/item.pyx b/src/borg/item.pyx
index 5ca93404..91fe57ee 100644
--- a/src/borg/item.pyx
+++ b/src/borg/item.pyx
@@ -189,6 +189,7 @@ class Item(PropDict):
         If memorize is True, the computed size value will be stored into the item.
         """
         attr = 'csize' if compressed else 'size'
+        assert not (compressed and memorize), 'Item does not have a csize field.'
         try:
             if from_chunks:
                 raise AttributeError
diff --git a/src/borg/repository.py b/src/borg/repository.py
index d15d51b1..f73e9cf5 100644
--- a/src/borg/repository.py
+++ b/src/borg/repository.py
@@ -34,7 +34,7 @@
 TAG_DELETE = 1
 TAG_COMMIT = 2
 
-LIST_SCAN_LIMIT = 10000  # repo.list() / .scan() result count limit the borg client uses
+LIST_SCAN_LIMIT = 100000  # repo.list() / .scan() result count limit the borg client uses
 
 FreeSpace = partial(defaultdict, int)
 


Test patch:
diff --git a/conftest.py b/conftest.py
index e85ae6ef..cc428be1 100644
--- a/conftest.py
+++ b/conftest.py
@@ -62,16 +62,16 @@ def pytest_report_header(config, startdir):
 
 class DefaultPatches:
     def __init__(self, request):
-        self.org_cache_wipe_cache = borg.cache.Cache.wipe_cache
+        self.org_cache_wipe_cache = borg.cache.LocalCache.wipe_cache
 
         def wipe_should_not_be_called(*a, **kw):
             raise AssertionError("Cache wipe was triggered, if this is part of the test add @pytest.mark.allow_cache_wipe")
         if 'allow_cache_wipe' not in request.keywords:
-            borg.cache.Cache.wipe_cache = wipe_should_not_be_called
+            borg.cache.LocalCache.wipe_cache = wipe_should_not_be_called
         request.addfinalizer(self.undo)
 
     def undo(self):
-        borg.cache.Cache.wipe_cache = self.org_cache_wipe_cache
+        borg.cache.LocalCache.wipe_cache = self.org_cache_wipe_cache
 
 
 @pytest.fixture(autouse=True)
diff --git a/src/borg/testsuite/archiver.py b/src/borg/testsuite/archiver.py
index e9bcef5e..b0b0a9b7 100644
--- a/src/borg/testsuite/archiver.py
+++ b/src/borg/testsuite/archiver.py
@@ -34,7 +34,7 @@
 from .. import xattr, helpers, platform
 from ..archive import Archive, ChunkBuffer, flags_noatime, flags_normal
 from ..archiver import Archiver, parse_storage_quota
-from ..cache import Cache
+from ..cache import Cache, LocalCache
 from ..constants import *  # NOQA
 from ..crypto.low_level import bytes_to_long, num_aes_blocks
 from ..crypto.key import KeyfileKeyBase, RepoKey, KeyfileKey, Passphrase, TAMRequiredError
@@ -1031,6 +1031,21 @@ def test_create_pattern_intermediate_folders_first(self):
         assert out_list.index('d x/a') < out_list.index('- x/a/foo_a')
         assert out_list.index('d x/b') < out_list.index('- x/b/foo_b')
 
+    def test_create_no_cache_sync(self):
+        self.create_test_files()
+        self.cmd('init', '--encryption=repokey', self.repository_location)
+        self.cmd('delete', '--cache-only', self.repository_location)
+        create_json = json.loads(self.cmd('create', '--no-cache-sync', self.repository_location + '::test', 'input',
+                                          '--json', '--error'))  # ignore experimental warning
+        info_json = json.loads(self.cmd('info', self.repository_location + '::test', '--json'))
+        create_stats = create_json['cache']['stats']
+        info_stats = info_json['cache']['stats']
+        assert create_stats == info_stats
+        self.cmd('delete', '--cache-only', self.repository_location)
+        self.cmd('create', '--no-cache-sync', self.repository_location + '::test2', 'input')
+        self.cmd('info', self.repository_location)
+        self.cmd('check', self.repository_location)
+
     def test_extract_pattern_opt(self):
         self.cmd('init', '--encryption=repokey', self.repository_location)
         self.create_regular_file('file1', size=1024 * 80)
@@ -1509,14 +1524,14 @@ def test_unknown_mandatory_feature_in_cache(self):
             self.cmd('create', self.repository_location + '::test', 'input')
         else:
             called = False
-            wipe_cache_safe = Cache.wipe_cache
+            wipe_cache_safe = LocalCache.wipe_cache
 
             def wipe_wrapper(*args):
                 nonlocal called
                 called = True
                 wipe_cache_safe(*args)
 
-            with patch.object(Cache, 'wipe_cache', wipe_wrapper):
+            with patch.object(LocalCache, 'wipe_cache', wipe_wrapper):
                 self.cmd('create', self.repository_location + '::test', 'input')
 
             assert called
@@ -2223,7 +2238,7 @@ def check_cache(self):
             manifest, key = Manifest.load(repository, Manifest.NO_OPERATION_CHECK)
             with Cache(repository, key, manifest, sync=False) as cache:
                 original_chunks = cache.chunks
-            cache.destroy(repository)
+            Cache.destroy(repository)
             with Cache(repository, key, manifest) as cache:
                 correct_chunks = cache.chunks
         assert original_chunks is not correct_chunks
diff --git a/src/borg/testsuite/cache.py b/src/borg/testsuite/cache.py
index 6f6452a1..6cce0cb7 100644
--- a/src/borg/testsuite/cache.py
+++ b/src/borg/testsuite/cache.py
@@ -1,11 +1,19 @@
 import io
+import os.path
 
 from msgpack import packb
 
 import pytest
 
-from ..hashindex import ChunkIndex, CacheSynchronizer
 from .hashindex import H
+from .key import TestKey
+from ..archive import Statistics
+from ..cache import AdHocCache
+from ..compress import CompressionSpec
+from ..crypto.key import RepoKey
+from ..hashindex import ChunkIndex, CacheSynchronizer
+from ..helpers import Manifest
+from ..repository import Repository
 
 
 class TestCacheSynchronizer:
@@ -196,3 +204,76 @@ def test_refcount_one_below_max_value(self):
         assert index[H(0)] == (ChunkIndex.MAX_VALUE, 1234, 5678)
         sync.feed(data)
         assert index[H(0)] == (ChunkIndex.MAX_VALUE, 1234, 5678)
+
+
+class TestAdHocCache:
+    @pytest.yield_fixture
+    def repository(self, tmpdir):
+        self.repository_location = os.path.join(str(tmpdir), 'repository')
+        with Repository(self.repository_location, exclusive=True, create=True) as repository:
+            repository.put(H(1), b'1234')
+            repository.put(Manifest.MANIFEST_ID, b'5678')
+            yield repository
+
+    @pytest.fixture
+    def key(self, repository, monkeypatch):
+        monkeypatch.setenv('BORG_PASSPHRASE', 'test')
+        key = RepoKey.create(repository, TestKey.MockArgs())
+        key.compressor = CompressionSpec('none').compressor
+        return key
+
+    @pytest.fixture
+    def manifest(self, repository, key):
+        Manifest(key, repository).write()
+        return Manifest.load(repository, key=key, operations=Manifest.NO_OPERATION_CHECK)[0]
+
+    @pytest.fixture
+    def cache(self, repository, key, manifest):
+        return AdHocCache(repository, key, manifest)
+
+    def test_does_not_contain_manifest(self, cache):
+        assert not cache.seen_chunk(Manifest.MANIFEST_ID)
+
+    def test_does_not_delete_existing_chunks(self, repository, cache):
+        assert cache.seen_chunk(H(1)) == ChunkIndex.MAX_VALUE
+        cache.chunk_decref(H(1), Statistics())
+        assert repository.get(H(1)) == b'1234'
+
+    def test_does_not_overwrite(self, cache):
+        with pytest.raises(AssertionError):
+            cache.add_chunk(H(1), b'5678', Statistics(), overwrite=True)
+
+    def test_seen_chunk_add_chunk_size(self, cache):
+        assert cache.add_chunk(H(1), b'5678', Statistics()) == (H(1), 4, 0)
+
+    def test_deletes_chunks_during_lifetime(self, cache, repository):
+        """E.g. checkpoint archives"""
+        cache.add_chunk(H(5), b'1010', Statistics())
+        assert cache.seen_chunk(H(5)) == 1
+        cache.chunk_decref(H(5), Statistics())
+        assert not cache.seen_chunk(H(5))
+        with pytest.raises(Repository.ObjectNotFound):
+            repository.get(H(5))
+
+    def test_files_cache(self, cache):
+        assert cache.file_known_and_unchanged(bytes(32), None) is None
+        assert not cache.do_files
+        assert cache.files is None
+
+    def test_txn(self, cache):
+        assert not cache._txn_active
+        cache.seen_chunk(H(5))
+        assert cache._txn_active
+        assert cache.chunks
+        cache.rollback()
+        assert not cache._txn_active
+        assert not hasattr(cache, 'chunks')
+
+    def test_incref_after_add_chunk(self, cache):
+        assert cache.add_chunk(H(3), b'5678', Statistics()) == (H(3), 4, 47)
+        assert cache.chunk_incref(H(3), Statistics()) == (H(3), 4, 47)
+
+    def test_existing_incref_after_add_chunk(self, cache):
+        """This case occurs with part files, see Archive.chunk_file."""
+        assert cache.add_chunk(H(1), b'5678', Statistics()) == (H(1), 4, 0)
+        assert cache.chunk_incref(H(1), Statistics()) == (H(1), 4, 0)
diff --git a/src/borg/testsuite/item.py b/src/borg/testsuite/item.py
index 785a962c..f9d72f87 100644
--- a/src/borg/testsuite/item.py
+++ b/src/borg/testsuite/item.py
@@ -154,6 +154,11 @@ def test_item_file_size():
         ChunkListEntry(csize=1, size=2000, id=None),
     ])
     assert item.get_size() == 3000
+    with pytest.raises(AssertionError):
+        item.get_size(compressed=True, memorize=True)
+    assert item.get_size(compressed=True) == 2
+    item.get_size(memorize=True)
+    assert item.size == 3000
 
 
 def test_item_file_size_no_chunks():