diff --git a/C/tests/c4DatabaseTest.cc b/C/tests/c4DatabaseTest.cc index d39af0d48..b2a2b1a4b 100644 --- a/C/tests/c4DatabaseTest.cc +++ b/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; } diff --git a/LiteCore/Database/Housekeeper.cc b/LiteCore/Database/Housekeeper.cc index e52a001af..11dc60179 100644 --- a/LiteCore/Database/Housekeeper.cc +++ b/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); } diff --git a/LiteCore/Support/RingBuffer.hh b/LiteCore/Support/RingBuffer.hh index 3440d2d50..48d236979 100644 --- a/LiteCore/Support/RingBuffer.hh +++ b/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; } diff --git a/Replicator/Replicator.cc b/Replicator/Replicator.cc index a6ca6505f..018d21ae5 100644 --- a/Replicator/Replicator.cc +++ b/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 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); diff --git a/Replicator/Replicator.hh b/Replicator/Replicator.hh index 735ab293b..ba56b2e92 100644 --- a/Replicator/Replicator.hh +++ b/Replicator/Replicator.hh @@ -11,6 +11,7 @@ // #pragma once +#include "AtomicRetained.hh" #include "Worker.hh" #include "Checkpointer.hh" #include "BLIPConnection.hh" @@ -18,7 +19,6 @@ #include "Stopwatch.hh" #include "c4DatabaseTypes.h" #include -#include #include #include @@ -232,8 +232,8 @@ namespace litecore::repl { // Member variables: struct SubReplicator { - Retained pusher; - Retained puller; + AtomicRetained pusher; + AtomicRetained puller; Status pushStatus; // Current status of Pusher Status pullStatus; // Current status of Puller unique_ptr checkpointer; // Object that manages checkpoints diff --git a/vendor/fleece b/vendor/fleece index ace0a6562..4a65a55fe 160000 --- a/vendor/fleece +++ b/vendor/fleece @@ -1 +1 @@ -Subproject commit ace0a65628bc13690efc019a87c7f553cf33eece +Subproject commit 4a65a55fef5c9e7068526ace6c675d5a99ef3c21