Cherry picks from 4.1 to 3.4

C/tests/c4DatabaseTest.cc

@@ -780,7 +780,20 @@ N_WAY_TEST_CASE_METHOD(C4DatabaseTest, "Document expiration torture test", "[Dat
         int     setCount = 1;
         C4Error error{};
         for ( ; setCount <= total; ++setCount ) {
-            if ( !c4coll_setDocExpiration(collection, c4str(docID(setCount)), expire, &error) ) {
+            // CBL-8202: Under this artificial contention (a no-pause reader on another connection
+            // plus the background purge) setDocExpiration can momentarily fail with a transient
+            // "busy" error when it can't grab the write lock. That is expected and retryable, so
+            // retry rather than aborting the loop and leaving documents without an expiration.
+            constexpr int kMaxBusyRetries = 200;
+            bool          ok              = false;
+            for ( int busyRetries = 0; !ok; ) {
+                ok = c4coll_setDocExpiration(collection, c4str(docID(setCount)), expire, &error);
+                if ( ok ) break;
+                if ( error.domain != LiteCoreDomain || error.code != kC4ErrorBusy || ++busyRetries > kMaxBusyRetries )
+                    break;
+                std::this_thread::sleep_for(5ms);
+            }
+            if ( !ok ) {
                 stop.store(true);
                 break;
             }

LiteCore/Database/Housekeeper.cc

@@ -16,6 +16,7 @@
 #include "SequenceTracker.hh"
 #include "BackgroundDB.hh"
 #include "DataFile.hh"
+#include "Error.hh"
 #include "Logging.hh"
 #include "SQLiteKeyStore.hh"
 #include "StringUtil.hh"
@@ -98,18 +99,39 @@ namespace litecore {
         enqueue(FUNCTION_TO_QUEUE(Housekeeper::_doExpiration));
     }
 
+    // CBL-8202: How long to wait before retrying expiration after a transient "database busy"
+    // error. Another connection (or thread) held the write lock while we tried to purge; we just
+    // need to try again shortly.
+    static constexpr chrono::milliseconds kRetryExpirationAfterBusyMS{100};
+
     void Housekeeper::_doExpiration() {
         if ( _isStopped() ) return;
 
-        logInfo("Housekeeper: expiring documents...");
-        _bgdb->useInTransaction(_keyStoreName, [&](KeyStore& keyStore, SequenceTracker* sequenceTracker) -> bool {
-            if ( sequenceTracker ) {
-                keyStore.expireRecords([&](slice docID) { sequenceTracker->documentPurged(docID); });
-            } else {
-                keyStore.expireRecords();
+        logVerbose("Housekeeper: expiring documents...");
+        try {
+            _bgdb->useInTransaction(_keyStoreName, [&](KeyStore& keyStore, SequenceTracker* sequenceTracker) -> bool {
+                if ( sequenceTracker ) {
+                    keyStore.expireRecords([&](slice docID) { sequenceTracker->documentPurged(docID); });
+                } else {
+                    keyStore.expireRecords();
+                }
+                return true;
+            });
+        } catch ( const exception& x ) {
+            // CBL-8202: A transient lock conflict (another connection holding the write lock) must
+            // not abort the Housekeeper. If we let the exception propagate, the actor swallows it
+            // and the _scheduleExpiration() below is skipped, so the timer never fires again and
+            // expired documents are never purged once the app stops calling setDocExpiration.
+            // Instead, reschedule a near-term retry of the purge.
+            error e = error::convertException(x).standardized();
+            if ( e.domain == error::LiteCore && e.code == error::Busy ) {
+                logVerbose("Housekeeper: expiration deferred (database busy); retrying in %lldms",
+                           (long long)kRetryExpirationAfterBusyMS.count());
+                _expiryTimer->fireAfter(kRetryExpirationAfterBusyMS);
+                return;
             }
-            return true;
-        });
+            throw;
+        }
 
         _scheduleExpiration(false);
     }

LiteCore/Support/RingBuffer.hh

@@ -58,7 +58,7 @@ namespace litecore {
             if ( end >= _capacity ) end -= _capacity;
             size_t n1 = std::min(n, _capacity - end);
             ::memcpy(&_buffer[end], data.buf, n1);
-            if ( n1 > n ) ::memcpy(&_buffer[0], &data[n1], n - n1);
+            if ( n1 < n ) ::memcpy(&_buffer[0], &data[n1], n - n1);
             _size += n;
             return n;
         }

Replicator/Replicator.cc

@@ -252,9 +252,13 @@ namespace litecore::repl {
 
     // Called after the checkpoint is established.
     void Replicator::startReplicating(CollectionIndex coll) {
-        if ( _options->push(coll) > kC4Passive ) _subRepls[coll].pusher->start();
-        if ( _options->pull(coll) > kC4Passive )
-            _subRepls[coll].puller->start(_subRepls[coll].checkpointer->remoteMinSequence());
+        if ( _options->push(coll) > kC4Passive ) {
+            if ( auto pusher = _subRepls[coll].pusher.get() ) pusher->start();
+        }
+        if ( _options->pull(coll) > kC4Passive ) {
+            if ( auto puller = _subRepls[coll].puller.get() )
+                puller->start(_subRepls[coll].checkpointer->remoteMinSequence());
+        }
     }
 
     pair<int, websocket::Headers> Replicator::httpResponse() const {
@@ -379,9 +383,9 @@ namespace litecore::repl {
 
         CollectionIndex coll = task->collectionIndex();
         if ( coll != kNotCollectionIndex ) {
-            if ( task == _subRepls[coll].pusher ) {
+            if ( auto pusher = _subRepls[coll].pusher.get(); pusher == task ) {
                 updatePushStatus(coll, taskStatus);
-            } else if ( task == _subRepls[coll].puller ) {
+            } else if ( auto puller = _subRepls[coll].puller.get(); puller == task ) {
                 updatePullStatus(coll, taskStatus);
             }
         }
@@ -656,8 +660,8 @@ namespace litecore::repl {
         // Clear connection() and notify the other agents to do the same:
         _connectionClosed();
         for ( auto& sub : _subRepls ) {
-            if ( sub.pusher ) sub.pusher->connectionClosed();
-            if ( sub.puller ) sub.puller->connectionClosed();
+            if ( auto pusher = sub.pusher.get() ) pusher->connectionClosed();
+            if ( auto puller = sub.puller.get() ) puller->connectionClosed();
         }
 
         if ( status.isNormal() && closedByPeer && _options->isActive() ) {
@@ -718,9 +722,10 @@ namespace litecore::repl {
                 cLogInfo(coll, "No local checkpoint '%.*s'", SPLAT(sub.checkpointer->initialCheckpointID()));
                 // If pulling into an empty db with no checkpoint, it's safe to skip deleted
                 // revisions as an optimization.
-                if ( _options->pull(coll) > kC4Passive && sub.puller
+                if ( auto puller = sub.puller.get();
+                     puller && _options->pull(coll) > kC4Passive
                      && _db->useCollection(sub.collectionSpec)->getLastSequence() == 0_seq )
-                    sub.puller->setSkipDeleted();
+                    puller->setSkipDeleted();
             }
             return true;
         } catch ( ... ) {
@@ -767,8 +772,9 @@ namespace litecore::repl {
 
             if ( !refresh && sub.hadLocalCheckpoint ) {
                 // Compare checkpoints, reset if mismatched:
-                bool valid = sub.checkpointer->validateWith(remoteCheckpoint);
-                if ( !valid && sub.pusher ) sub.pusher->checkpointIsInvalid();
+                if ( !sub.checkpointer->validateWith(remoteCheckpoint) ) {
+                    if ( auto pusher = sub.pusher.get() ) pusher->checkpointIsInvalid();
+                }
 
                 if ( !refresh ) {
                     // Now we have the checkpoints! Time to start replicating:
@@ -898,8 +904,9 @@ namespace litecore::repl {
 
                     if ( _subRepls[i].hadLocalCheckpoint ) {
                         // Compare checkpoints, reset if mismatched:
-                        bool valid = _subRepls[i].checkpointer->validateWith(remoteCheckpoints[i]);
-                        if ( !valid && _subRepls[i].pusher ) _subRepls[i].pusher->checkpointIsInvalid();
+                        if ( !_subRepls[i].checkpointer->validateWith(remoteCheckpoints[i]) ) {
+                            if ( auto pusher = _subRepls[i].pusher.get() ) pusher->checkpointIsInvalid();
+                        }
                     }
                     // Now we have the checkpoints! Time to start replicating:
                     startReplicating(i);

Replicator/Replicator.hh

@@ -11,14 +11,14 @@
 //
 
 #pragma once
+#include "AtomicRetained.hh"
 #include "Worker.hh"
 #include "Checkpointer.hh"
 #include "BLIPConnection.hh"
 #include "Batcher.hh"
 #include "Stopwatch.hh"
 #include "c4DatabaseTypes.h"
 #include <access_lock.hh>
-#include <array>
 #include <optional>
 #include <utility>
 
@@ -232,8 +232,8 @@ namespace litecore::repl {
         // Member variables:
 
         struct SubReplicator {
-            Retained<Pusher>         pusher;
-            Retained<Puller>         puller;
+            AtomicRetained<Pusher>   pusher;
+            AtomicRetained<Puller>   puller;
             Status                   pushStatus;                        // Current status of Pusher
             Status                   pullStatus;                        // Current status of Puller
             unique_ptr<Checkpointer> checkpointer;                      // Object that manages checkpoints