CBL-2720 : Remove kC4ReplicatorOptionProgressLevel, simplify setting progress level (#1322)

* repl::Worker cleanup and documentation

- Lots of doc-comments in Worker.hh
- Changed progressNotificationLevel from int to C4ReplicatorProgressLevel.
- Renamed _important to _importance.

* Put C4ReplicatorProgressLevel in Options struct, not Fleece Dict

* Removed the already-deprecated `kC4ReplicatorOptionProgressLevel`.
* Added a `progressLevel` field to litecore::repl::Options.
* Made Options ref-counted.
* Worker objects store a shared RetainedConst<Options> instead of their
  own copies of the object.
* Removed Worker::progressNotificationLevel(), its setter, and the
  several overrides of its setter.
* C4Replicator::setProgressLevel() now just sets Options::progressLevel.

- Do not wipe Options' data and remove the related test case.

CBL-2720

Co-authored-by: Jianmin Zhao <[email protected]>

Author: snej

C/include/c4ReplicatorTypes.h

@@ -213,7 +213,6 @@ C4API_BEGIN_DECLS
     #define kC4ReplicatorOptionNoIncomingConflicts "noIncomingConflicts" ///< Reject incoming conflicts (bool)
     #define kC4ReplicatorCheckpointInterval     "checkpointInterval" ///< How often to checkpoint, in seconds (number)
     #define kC4ReplicatorOptionRemoteDBUniqueID "remoteDBUniqueID" ///< Stable ID for remote db with unstable URL (string)
-    #define kC4ReplicatorOptionProgressLevel    "progress"  ///< If >=1, notify on every doc; if >=2, on every attachment (int)
     #define kC4ReplicatorOptionDisableDeltas    "noDeltas"   ///< Disables delta sync (bool)
     #define kC4ReplicatorOptionDisablePropertyDecryption "noDecryption" ///< Disables property decryption (bool)
     #define kC4ReplicatorOptionMaxRetries       "maxRetries" ///< Max number of retry attempts (int)

Replicator/ChangesFeed.cc

@@ -36,18 +36,18 @@ using namespace fleece;
 namespace litecore { namespace repl {
 
 
-    ChangesFeed::ChangesFeed(Delegate &delegate, Options &options,
+    ChangesFeed::ChangesFeed(Delegate &delegate, const Options *options,
                              DBAccess &db, Checkpointer *checkpointer)
     :Logging(SyncLog)
     ,_delegate(delegate)
     ,_options(options)
     ,_db(db)
     ,_checkpointer(checkpointer)
-    ,_continuous(_options.push == kC4Continuous)
-    ,_passive(_options.push <= kC4Passive)
-    ,_skipDeleted(_options.skipDeleted())
+    ,_continuous(_options->push == kC4Continuous)
+    ,_passive(_options->push <= kC4Passive)
+    ,_skipDeleted(_options->skipDeleted())
     {
-        filterByDocIDs(_options.docIDs());
+        filterByDocIDs(_options->docIDs());
     }
 
 
@@ -113,7 +113,7 @@ namespace litecore { namespace repl {
 
         // Run a by-sequence enumerator to find the changed docs:
         C4EnumeratorOptions options = kC4DefaultEnumeratorOptions;
-        if (!_getForeignAncestors && !_options.pushFilter)
+        if (!_getForeignAncestors && !_options->pushFilter)
             options.flags &= ~kC4IncludeBodies;
         if (!_skipDeleted)
             options.flags |= kC4IncludeDeleted;
@@ -255,7 +255,7 @@ namespace litecore { namespace repl {
     bool ChangesFeed::shouldPushRev(RevToSend *rev, C4DocEnumerator *e) const {
         bool needRemoteRevID = _getForeignAncestors && !rev->remoteAncestorRevID
                                                     && _isCheckpointValid;
-        if (needRemoteRevID || _options.pushFilter) {
+        if (needRemoteRevID || _options->pushFilter) {
             C4Error error;
             Retained<C4Document> doc;
             try {
@@ -284,14 +284,14 @@ namespace litecore { namespace repl {
                 if (!getRemoteRevID(rev, doc))
                     return false;     // skip or fail rev: it's already on the peer
             }
-            if (_options.pushFilter) {
+            if (_options->pushFilter) {
                 // If there's a push filter, ask it whether to push the doc:
-                if (!_options.pushFilter(nullslice,     // TODO: Collection support
+                if (!_options->pushFilter(nullslice,     // TODO: Collection support
                                          doc->docID(),
                                          doc->selectedRev().revID,
                                          doc->selectedRev().flags,
                                          doc->getProperties(),
-                                         _options.callbackContext)) {
+                                         _options->callbackContext)) {
                     logVerbose("Doc '%.*s' rejected by push filter", SPLAT(doc->docID()));
                     return false;     // skip rev: rejected by push filter
                 }
@@ -310,7 +310,8 @@ namespace litecore { namespace repl {
 #pragma mark - REPLICATOR CHANGES FEED:
 
 
-    ReplicatorChangesFeed::ReplicatorChangesFeed(Delegate &delegate, Options &options, DBAccess &db, Checkpointer *cp)
+    ReplicatorChangesFeed::ReplicatorChangesFeed(Delegate &delegate, const Options *options,
+                                                 DBAccess &db, Checkpointer *cp)
     :ChangesFeed(delegate, options, db, cp)     // DBAccess is a subclass of access_lock<C4Database*>
     ,_usingVersionVectors(db.usingVersionVectors())
     { }
@@ -329,7 +330,7 @@ namespace litecore { namespace repl {
         if (foreignAncestor && !_usingVersionVectors
                     && C4Document::getRevIDGeneration(foreignAncestor)
                         >= C4Document::getRevIDGeneration(doc->revID())) {
-            if (_options.pull <= kC4Passive) {
+            if (_options->pull <= kC4Passive) {
                 C4Error error = C4Error::make(WebSocketDomain, 409,
                                      "conflicts with newer server revision"_sl);
                 _delegate.failedToGetChange(rev, error, false);

Replicator/ChangesFeed.hh

@@ -26,7 +26,7 @@ namespace fleece {
 
 namespace litecore::repl {
     class DBAccess;
-    struct Options;
+    class Options;
     class Checkpointer;
 
     using DocIDSet = std::shared_ptr<std::unordered_set<std::string>>;
@@ -47,7 +47,7 @@ namespace litecore::repl {
             virtual void failedToGetChange(ReplicatedRev *rev, C4Error error, bool transient) =0;
         };
 
-        ChangesFeed(Delegate&, Options&, DBAccess &db, Checkpointer*);
+        ChangesFeed(Delegate&, const Options* NONNULL, DBAccess &db, Checkpointer*);
         ~ChangesFeed();
 
         // Setup:
@@ -94,7 +94,7 @@ namespace litecore::repl {
 
     protected:
         Delegate& _delegate;
-        Options &_options;
+        RetainedConst<Options> _options;
         DBAccess& _db;
         bool _getForeignAncestors {false};                  // True in propose-changes mode
     private:
@@ -114,7 +114,8 @@ namespace litecore::repl {
 
     class ReplicatorChangesFeed final : public ChangesFeed {
     public:
-        ReplicatorChangesFeed(Delegate &delegate, Options &options, DBAccess &db, Checkpointer *cp);
+        ReplicatorChangesFeed(Delegate &delegate, const Options *options,
+                              DBAccess &db, Checkpointer *cp);
 
         void setFindForeignAncestors(bool use)      {_getForeignAncestors = use;}
 

Replicator/Checkpointer.cc

@@ -37,7 +37,7 @@ namespace litecore { namespace repl {
 #pragma mark - CHECKPOINT ACCESSORS:
 
 
-    Checkpointer::Checkpointer(const Options &opt, fleece::slice remoteURL)
+    Checkpointer::Checkpointer(const Options *opt, fleece::slice remoteURL)
     :_options(opt)
     ,_remoteURL(remoteURL)
     { }
@@ -204,7 +204,7 @@ namespace litecore { namespace repl {
 
 
     slice Checkpointer::remoteDBIDString() const {
-        return _options.remoteDBIDString(_remoteURL);
+        return _options->remoteDBIDString(_remoteURL);
     }
 
 
@@ -221,10 +221,10 @@ namespace litecore { namespace repl {
     // Computes the ID of the checkpoint document.
     string Checkpointer::docIDForUUID(const C4UUID &localUUID, URLTransformStrategy urlStrategy) {
         // Derive docID from from db UUID, remote URL, channels, filter, and docIDs.
-        Array channels = _options.channels();
-        Value filter = _options.properties[kC4ReplicatorOptionFilter];
-        const Value filterParams = _options.properties[kC4ReplicatorOptionFilterParams];
-        Array docIDs = _options.docIDs();
+        Array channels = _options->channels();
+        Value filter = _options->properties[kC4ReplicatorOptionFilter];
+        const Value filterParams = _options->properties[kC4ReplicatorOptionFilterParams];
+        Array docIDs = _options->docIDs();
 
         // Compute the ID by writing the values to a Fleece array, then taking a SHA1 digest:
         fleece::Encoder enc;
@@ -326,11 +326,11 @@ namespace litecore { namespace repl {
 
 
     void Checkpointer::initializeDocIDs() {
-        if(!_docIDs.empty() || !_options.docIDs() || _options.docIDs().empty()) {
+        if(!_docIDs.empty() || !_options->docIDs() || _options->docIDs().empty()) {
             return;
         }
 
-        Array::iterator i(_options.docIDs());
+        Array::iterator i(_options->docIDs());
         while(i) {
             string docID = i.value().asString().asString();
             if(!docID.empty()) {
@@ -344,12 +344,12 @@ namespace litecore { namespace repl {
 
     bool Checkpointer::isDocumentAllowed(C4Document* doc) {
         return isDocumentIDAllowed(doc->docID())
-            && (!_options.pushFilter || _options.pushFilter(nullslice,   // TODO: Collection support
+            && (!_options->pushFilter || _options->pushFilter(nullslice,   // TODO: Collection support
                                                             doc->docID(),
                                                             doc->selectedRev().revID,
                                                             doc->selectedRev().flags,
                                                             doc->getProperties(),
-                                                            _options.callbackContext));
+                                                            _options->callbackContext));
     }
 
 
@@ -363,7 +363,7 @@ namespace litecore { namespace repl {
 
 
     void Checkpointer::pendingDocumentIDs(C4Database* db, PendingDocCallback callback) {
-        if(_options.push < kC4OneShot) {
+        if(_options->push < kC4OneShot) {
             // Couchbase Lite should not allow this case
             C4Error::raise(LiteCoreDomain, kC4ErrorUnsupported);
         }
@@ -377,8 +377,8 @@ namespace litecore { namespace repl {
         }
 
         C4EnumeratorOptions opts { kC4IncludeNonConflicted | kC4IncludeDeleted };
-        const auto hasDocIDs = bool(_options.docIDs());
-        if(!hasDocIDs && _options.pushFilter) {
+        const auto hasDocIDs = bool(_options->docIDs());
+        if(!hasDocIDs && _options->pushFilter) {
             // docIDs has precedence over push filter
             opts.flags |= kC4IncludeBodies;
         }
@@ -393,7 +393,7 @@ namespace litecore { namespace repl {
             if(!isDocumentIDAllowed(info.docID))
                 continue;
 
-            if (!hasDocIDs && _options.pushFilter) {
+            if (!hasDocIDs && _options->pushFilter) {
                 // If there is a push filter, we have to get the doc body for it to peruse:
                 Retained<C4Document> nextDoc = e.getDocument();
                 if(!nextDoc) {
@@ -416,7 +416,7 @@ namespace litecore { namespace repl {
 
 
     bool Checkpointer::isDocumentPending(C4Database* db, slice docId) {
-        if(_options.push < kC4OneShot) {
+        if(_options->push < kC4OneShot) {
             // Couchbase Lite should not allow this case
             C4Error::raise(LiteCoreDomain, kC4ErrorUnsupported);
         }

Replicator/Checkpointer.hh

@@ -35,7 +35,7 @@ namespace litecore { namespace repl {
     using namespace fleece;
 
     class Checkpoint;
-    struct Options;
+    class Options;
     class RemoteSequence;
 
 
@@ -44,7 +44,7 @@ namespace litecore { namespace repl {
               Replicator, Pusher and Puller. */
     class Checkpointer {
     public:
-        Checkpointer(const Options&, fleece::slice remoteURL);
+        Checkpointer(const Options* NONNULL, fleece::slice remoteURL);
 
         ~Checkpointer();
 
@@ -170,7 +170,7 @@ namespace litecore { namespace repl {
         void saveSoon();
 
         Logging*                        _logger;
-        const Options&                  _options;
+        RetainedConst<Options>          _options;
         alloc_slice const               _remoteURL;
         std::unordered_set<std::string> _docIDs;
 

Replicator/IncomingRev.cc

@@ -38,8 +38,8 @@ namespace litecore { namespace repl {
     :Worker(puller, "inc")
     ,_puller(puller)
     {
-        _passive = _options.pull <= kC4Passive;
-        _important = false;
+        _passive = _options->pull <= kC4Passive;
+        _importance = false;
         static atomic<uint32_t> sRevSignpostCount {0};
         _serialNumber = ++sRevSignpostCount;
     }
@@ -71,7 +71,7 @@ namespace litecore { namespace repl {
                                _revMessage->property("history"_sl),
                                _revMessage->boolProperty("deleted"_sl),
                                _revMessage->boolProperty("noconflicts"_sl)
-                                   || _options.noIncomingConflicts());
+                                   || _options->noIncomingConflicts());
         _rev->deltaSrcRevID = _revMessage->property("deltaSrc"_sl);
         slice sequenceStr = _revMessage->property(slice("sequence"));
         _remoteSequence = RemoteSequence(sequenceStr);
@@ -121,11 +121,11 @@ namespace litecore { namespace repl {
             _revMessage = nullptr;
 
         _mayContainBlobs = jsonBody.containsBytes("\"digest\""_sl);
-        _mayContainEncryptedProperties = !_options.disablePropertyDecryption()
+        _mayContainEncryptedProperties = !_options->disablePropertyDecryption()
                                          && MayContainPropertiesToDecrypt(jsonBody);
 
         // Decide whether to continue now (on the Puller thread) or asynchronously on my own:
-        if (_options.pullValidator|| jsonBody.size > kMaxImmediateParseSize
+        if (_options->pullValidator|| jsonBody.size > kMaxImmediateParseSize
                                   || _mayContainBlobs || _mayContainEncryptedProperties)
             enqueue(FUNCTION_TO_QUEUE(IncomingRev::parseAndInsert), move(jsonBody));
         else
@@ -140,13 +140,13 @@ namespace litecore { namespace repl {
         rev->owner = this;
 
         // Do not purge if the auto-purge is not enabled:
-        if (!_options.enableAutoPurge()) {
+        if (!_options->enableAutoPurge()) {
             finish();
             return;
         }
 
         // Call the custom validation function if any:
-        if (_options.pullValidator) {
+        if (_options->pullValidator) {
             // Revoked rev body is empty when sent to the filter:
             auto body = Dict::emptyDict();
             if (!performPullValidation(body))
@@ -168,7 +168,7 @@ namespace litecore { namespace repl {
             if (!fleeceDoc)
                 err = C4Error::make(FleeceDomain, (int)encodeErr, "Incoming rev failed to encode"_sl);
 
-        } else if (_options.pullValidator || _mayContainBlobs || _mayContainEncryptedProperties) {
+        } else if (_options->pullValidator || _mayContainBlobs || _mayContainEncryptedProperties) {
             // It's a delta, but we need the entire document body now because either it has to be
             // passed to the validation function, it may contain new blobs to download, or it may
             // have properties to decrypt.
@@ -178,7 +178,7 @@ namespace litecore { namespace repl {
             if (!fleeceDoc) {
                 // Don't have the body of the source revision. This might be because I'm in
                 // no-conflict mode and the peer is trying to push me a now-obsolete revision.
-                if (_options.noIncomingConflicts())
+                if (_options->noIncomingConflicts())
                     err = {WebSocketDomain, 409};
                 else
                     err = C4Error::printf(LiteCoreDomain, kC4ErrorDeltaBaseUnknown,
@@ -208,7 +208,7 @@ namespace litecore { namespace repl {
         // no longer accessible (not in any channel the client has access to.)
         if (root["_removed"_sl].asBool()) {
             _rev->flags |= kRevPurged;
-            if (!_options.enableAutoPurge()) {
+            if (!_options->enableAutoPurge()) {
                 finish();
                 return;
             }
@@ -220,8 +220,8 @@ namespace litecore { namespace repl {
             C4Error error;
             decryptedRoot = DecryptDocumentProperties(_rev->docID,
                                                       root,
-                                                      _options.propertyDecryptor,
-                                                      _options.callbackContext,
+                                                      _options->propertyDecryptor,
+                                                      _options->callbackContext,
                                                       &error);
             if (decryptedRoot) {
                 root = decryptedRoot;
@@ -277,10 +277,10 @@ namespace litecore { namespace repl {
 
     // Calls the custom pull validator if available.
     bool IncomingRev::performPullValidation(Dict body) {
-        if (_options.pullValidator) {
-            if (!_options.pullValidator(nullslice,     // TODO: Collection support
+        if (_options->pullValidator) {
+            if (!_options->pullValidator(nullslice,     // TODO: Collection support
                                         _rev->docID, _rev->revID, _rev->flags, body,
-                                        _options.callbackContext)) {
+                                        _options->callbackContext)) {
                 failWithError(WebSocketDomain, 403, "rejected by validation function"_sl);
                 return false;
             }
@@ -372,12 +372,6 @@ namespace litecore { namespace repl {
     }
 
 
-    int IncomingRev::progressNotificationLevel() const {
-        return _puller ? _puller->progressNotificationLevel() : 0;
-    }
-
-
-
     Worker::ActivityLevel IncomingRev::computeActivityLevel() const {
         if (Worker::computeActivityLevel() == kC4Busy || _pendingCallbacks > 0
                                                       || (_blob != _pendingBlobs.end())) {

Replicator/IncomingRev.hh

@@ -40,15 +40,13 @@ namespace litecore { namespace repl {
         void revisionProvisionallyInserted();
         void revisionInserted();
 
-        int progressNotificationLevel() const override;
-
     protected:
         ActivityLevel computeActivityLevel() const override;
 
     private:
         void reinitialize();
         void parseAndInsert(alloc_slice jsonBody);
-        bool nonPassive() const                 {return _options.pull > kC4Passive;}
+        bool nonPassive() const                 {return _options->pull > kC4Passive;}
         void _handleRev(Retained<blip::MessageIn>);
         void gotDeltaSrc(alloc_slice deltaSrcBody);
         fleece::Doc parseBody(alloc_slice jsonBody);