Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions LiteCore/tests/CMakeLists.txt
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,7 @@ add_executable(
${TOP}Crypto/CertificateTest.cc
${TOP}Networking/tests/CookieStoreTest.cc
${TOP}Replicator/tests/DBAccessTestWrapper.cc
${TOP}Replicator/tests/ParsedSequenceIDTest.cc
${TOP}Replicator/tests/PropertyEncryptionTests.cc
${TOP}Replicator/tests/ReplicatorLoopbackTest.cc
${TOP}Replicator/tests/ReplicatorAPITest.cc
Expand Down
18 changes: 12 additions & 6 deletions Replicator/Checkpoint.cc
Original file line number Diff line number Diff line change
Expand Up @@ -114,19 +114,25 @@ namespace litecore::repl {
if ( _remote && _remote != remoteSequences._remote ) {
LogTo(SyncLog, "Remote sequence mismatch: I had '%s', remote had '%s'", _remote.toJSONString().c_str(),
remoteSequences._remote.toJSONString().c_str());
if ( _remote.isInt() && remoteSequences._remote.isInt() ) {
if ( _remote.intValue() > remoteSequences._remote.intValue() ) {

ParsedSequenceID localParsed, remoteParsed;
bool localParseable = _remote.toParsedSequenceID(localParsed);
bool remoteParseable = remoteSequences._remote.toParsedSequenceID(remoteParsed);
if ( localParseable && remoteParseable ) {
if ( remoteParsed.before(localParsed) ) {
LogTo(SyncLog, "Rolling back to earlier remote sequence from server, some redundant changes may be "
"proposed...");
_remote = remoteSequences._remote;
match = false;
} else {
LogTo(SyncLog, "Ignoring remote sequence on server since client side is older, some redundant "
"changes may be proposed...");
LogTo(SyncLog, "Ignoring remote sequence on server since client side is older or equal, some "
"redundant changes may be proposed...");
}
} else {
Warn("Non-numeric remote sequence detected, resetting replication back to start. Redundant changes "
"will be proposed...");
Warn("Unparseable remote sequence: locally-saved remote seq '%s' is %s, "
"server-side remote seq '%s' is %s. Resetting replication.",
_remote.toJSONString().c_str(), localParseable ? "parseable" : "UNPARSEABLE",
remoteSequences._remote.toJSONString().c_str(), remoteParseable ? "parseable" : "UNPARSEABLE");
_remote = {};
match = false;
}
Expand Down
180 changes: 180 additions & 0 deletions Replicator/ParsedSequenceID.hh
Original file line number Diff line number Diff line change
@@ -0,0 +1,180 @@
//
// ParsedSequenceID.hh
//
// Copyright 2026-Present Couchbase, Inc.
//
// Use of this software is governed by the Business Source License included
// in the file licenses/BSL-Couchbase.txt. As of the Change Date specified
// in that file, in accordance with the Business Source License, use of this
// software will be governed by the Apache License, Version 2.0, included in
// the file licenses/APL2.txt.
//

#pragma once
#include "StringUtil.hh"
#include "slice_stream.hh"
#include <cstdint>
#include <string>

namespace litecore::repl {

/**
* Represents a parsed Sync Gateway compound sequence ID.
*
* Sync Gateway emits sequence IDs in three formats on the _changes feed:
*
* - Simple: "Seq" e.g. "42"
* A plain database sequence number.
*
* - Backfill: "TriggeredBy:Seq" e.g. "100:35"
* A revision (seq 35) is being sent retroactively because an access-grant
* change at sequence 100 gave the user access to its channel.
*
* - LowSeq: "LowSeq:TriggeredBy:Seq" e.g. "20:100:35"
* Same as backfill, but also records the lowest contiguous sequence (20)
* that the client has fully processed on the feed. TriggeredBy may be
* omitted (e.g. "20::35") when there is no access-grant trigger.
*
* For checkpoint comparison the "comparison value" of each format is:
* - Simple → seq
* - Backfill → triggeredBy
* - LowSeq → lowSeq
*
* The before() method implements the ordering rules that match Sync Gateway's
* SequenceID.Before(), so that LiteCore and SG always agree on which checkpoint is older.
*/

class ParsedSequenceID {
public:
ParsedSequenceID() = default;

ParsedSequenceID(uint64_t seq, uint64_t triggeredBy, uint64_t lowSeq)
: _seq(seq), _triggeredBy(triggeredBy), _lowSeq(lowSeq) {}

[[nodiscard]] uint64_t seq() const { return _seq; }

[[nodiscard]] uint64_t triggeredBy() const { return _triggeredBy; }

[[nodiscard]] uint64_t lowSeq() const { return _lowSeq; }

/// "Seq" — plain sequence, no trigger, no lowSeq.
[[nodiscard]] bool isSimpleRevision() const { return _lowSeq == 0 && _triggeredBy == 0; }

/// "TriggeredBy:Seq" — sent retroactively due to an access-grant, no lowSeq tracking.
/// triggeredBy is the primary comparison value in before().
[[nodiscard]] bool isTriggeredRevision() const { return _lowSeq == 0 && _triggeredBy > 0; }

/// "LowSeq:TriggeredBy:Seq" or "LowSeq::Seq" — includes lowSeq feed-position tracking.
/// lowSeq is the primary comparison value in before(). May or may not include a trigger.
[[nodiscard]] bool isLowSeqRevision() const { return _lowSeq > 0; }

/**
* Parse a sequence string into a ParsedSequenceID.
*
* Accepted formats:
* "42" → {seq=42, triggeredBy=0, lowSeq=0}
* "100:35" → {seq=35, triggeredBy=100, lowSeq=0}
* "20:100:35" → {seq=35, triggeredBy=100, lowSeq=20}
* "20::35" → {seq=35, triggeredBy=0, lowSeq=20}
*
* @param str The string to parse.
* @param out Receives the parsed result on success.
* @return true on success, false if the string is empty, malformed, or has
* an unexpected number of components.
*/
static bool parse(const std::string& str, ParsedSequenceID& out) {
if ( str.empty() || str.back() == ':' ) return false; // litecore::split drops trailing empty tokens

auto parts = litecore::split(str, ":");
if ( parts.size() < 1 || parts.size() > 3 ) return false;

// Parse a single component as uint64. Empty is only valid for middle part of "LowSeq::Seq".
auto readUInt = [](std::string_view sv, uint64_t& val) -> bool {
if ( sv.empty() ) return false;
fleece::slice_istream stream(sv.data(), sv.size());
val = stream.readDecimal();
return stream.eof();
};

uint64_t v[3] = {};
switch ( parts.size() ) {
case 1:
if ( !readUInt(parts[0], v[0]) ) return false;
out = ParsedSequenceID(v[0], 0, 0); // Seq
return true;
case 2:
if ( !readUInt(parts[0], v[0]) || !readUInt(parts[1], v[1]) ) return false;
out = ParsedSequenceID(v[1], v[0], 0); // TriggeredBy:Seq
return true;
case 3:
if ( !readUInt(parts[0], v[0]) ) return false;
if ( !parts[1].empty() && !readUInt(parts[1], v[1]) ) return false; // empty → 0 for "LowSeq::Seq"
if ( !readUInt(parts[2], v[2]) ) return false;
out = ParsedSequenceID(v[2], v[1], v[0]); // LowSeq:TriggeredBy:Seq
return true;
default:
return false;
}
}

/**
* Returns true if this sequence is strictly before \p seqID2.
*
* Implements the same ordering as Sync Gateway's SequenceID.Before().
* Each format has a "comparison value" (CV):
* - Simple → CV = seq
* - Backfill → CV = triggeredBy
* - LowSeq → CV = lowSeq
*
* Cross-format rules (a.before(b)):
*
* a \ b | Simple | Backfill | LowSeq
* ───────────────|──────────────|─────────────────|────────────────
* Simple | seq < seq | seq < trig | seq <= low
* Backfill | trig <= seq | trig < trig | trig <= low
* | | (tie: seq<seq) |
* LowSeq | low < seq | low < trig | low < low
* | | | (tie: recurse
* | | | on inner pair)
*
*/
[[nodiscard]] bool before(const ParsedSequenceID& seqID2) const {
const auto& a = *this;
const auto& b = seqID2;

if ( a.isLowSeqRevision() ) {
if ( b.isLowSeqRevision() ) {
if ( a.lowSeq() != b.lowSeq() ) return a.lowSeq() < b.lowSeq();
ParsedSequenceID aInner(a.seq(), a.triggeredBy(), 0);
ParsedSequenceID bInner(b.seq(), b.triggeredBy(), 0);
return aInner.before(bInner);
}
if ( b.isTriggeredRevision() ) return a.lowSeq() < b.triggeredBy();
/* b is simple seq*/
return a.lowSeq() < b.seq();
}

if ( a.isTriggeredRevision() ) {
if ( b.isLowSeqRevision() ) return a.triggeredBy() <= b.lowSeq();
if ( b.isTriggeredRevision() ) {
if ( a.triggeredBy() != b.triggeredBy() ) return a.triggeredBy() < b.triggeredBy();
return a.seq() < b.seq();
}
/* b is simple */
return a.triggeredBy() <= b.seq();
}

// a is simple revision
if ( b.isLowSeqRevision() ) return a.seq() <= b.lowSeq();
if ( b.isTriggeredRevision() ) return a.seq() < b.triggeredBy();
/* both simple */
return a.seq() < b.seq();
}

private:
uint64_t _seq{0}; ///< The actual internal database sequence number.
uint64_t _triggeredBy{0}; ///< Sequence# of the access-grant that triggered backfill (0 = none).
uint64_t _lowSeq{0}; ///< Lowest contiguous sequence seen on the feed (0 = not present).
};

} // namespace litecore::repl
17 changes: 17 additions & 0 deletions Replicator/RemoteSequence.hh
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
//

#pragma once
#include "ParsedSequenceID.hh"
#include "StringUtil.hh"
#include "fleece/Fleece.hh"
#include "slice_stream.hh"
Expand All @@ -19,6 +20,7 @@

namespace litecore::repl {


/** A sequence received from a remote peer. Can be any JSON value, but optimized for positive ints. */
class RemoteSequence {
public:
Expand Down Expand Up @@ -73,6 +75,21 @@ namespace litecore::repl {

bool operator!=(const RemoteSequence& other) const noexcept FLPURE { return _value != other._value; }

/** Convert this RemoteSequence to a ParsedSequenceID. Returns false if unparseable.
Note: sliceValue() may contain JSON string quotes (added by Value::toJSON() in the
Value constructor path). These are stripped before parsing so that compound Sync
Gateway sequences like "20:100:35" are handled correctly whether quoted or not. */
[[nodiscard]] bool toParsedSequenceID(ParsedSequenceID& out) const {
if ( !*this ) return false;
if ( isInt() ) {
out = {intValue(), 0, 0};
return true;
}
std::string s(sliceValue());
if ( s.size() >= 2 && s.front() == '"' && s.back() == '"' ) s = s.substr(1, s.size() - 2);
return ParsedSequenceID::parse(s, out);
}

bool operator<(const RemoteSequence& other) const noexcept FLPURE {
if ( isInt() ) return !other.isInt() || intValue() < other.intValue();
else
Expand Down
Loading
Loading