Skip to content

Commit 7024552

Browse files
authored
CBG-5453: [4.0.6 backport] generate HLV.version from our own HLC for SGW actor writes (#8366)
1 parent eb5f6ab commit 7024552

20 files changed

Lines changed: 699 additions & 372 deletions

base/hlc.go

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
// Copyright 2026-Present Couchbase, Inc.
2+
//
3+
// Use of this software is governed by the Business Source License included
4+
// in the file licenses/BSL-Couchbase.txt. As of the Change Date specified
5+
// in that file, in accordance with the Business Source License, use of this
6+
// software will be governed by the Apache License, Version 2.0, included in
7+
// the file licenses/APL2.txt.
8+
9+
package base
10+
11+
import (
12+
"sync"
13+
"time"
14+
)
15+
16+
// hlcLogicalBits is the number of low-order bits of a Couchbase Server CAS reserved for the logical
17+
// (counter) component of the Hybrid Logical Clock. The remaining high-order bits hold the physical
18+
// (wall-clock nanosecond) component. Clearing these bits when reading the wall clock keeps generated
19+
// values in the same numeric space as a CAS and leaves room for same-instant logical increments.
20+
const hlcLogicalBits = 16
21+
22+
// hlcLogicalMask masks off the logical component, isolating the physical component of a timestamp.
23+
const hlcLogicalMask = (1 << hlcLogicalBits) - 1
24+
25+
// HybridLogicalClock generates monotonically increasing timestamps in the same numeric space as a
26+
// Couchbase Server CAS (a 48-bit nanosecond physical component plus a 16-bit logical counter). Sync
27+
// Gateway uses it to assign HLV current-version values without relying on server-side CAS macro
28+
// expansion, so the value is known before the write and can be written as a literal into every xattr
29+
// field that would otherwise require a per-field subdoc macro-expansion path.
30+
//
31+
// Generated values are CAS-comparable: under synchronised clocks a value generated just before a write
32+
// is below the CAS the server stamps at commit. This is an invariant goxdcr enforces (cv.ver <= cas) until MB-72252.
33+
// Monotonicity per HLV source is the caller's responsibility, supplied via the floor argument to Now.
34+
type HybridLogicalClock struct {
35+
clock func() uint64 // wall-clock source, nanoseconds since the Unix epoch; overridable in tests
36+
highestTime uint64
37+
mutex sync.Mutex
38+
}
39+
40+
// NewHybridLogicalClock returns a HybridLogicalClock backed by the system wall clock.
41+
func NewHybridLogicalClock() *HybridLogicalClock {
42+
return &HybridLogicalClock{clock: func() uint64 { return uint64(time.Now().UnixNano()) }}
43+
}
44+
45+
// SetClockForTest overrides the wall-clock source and resets the clock's high-water mark, so the next
46+
// value returned by Now is determined solely by getTime. Test-only: used to simulate clock skew between
47+
// Sync Gateway and the server.
48+
func (c *HybridLogicalClock) SetClockForTest(getTime func() uint64) {
49+
c.mutex.Lock()
50+
defer c.mutex.Unlock()
51+
c.clock = getTime
52+
c.highestTime = 0
53+
}
54+
55+
// Now returns the next timestamp, guaranteed to be strictly greater than both the previous value
56+
// returned by this clock (monotonic across calls) and the supplied floor, while tracking the wall clock
57+
// where possible. The result is max(physical, highestTime+1, floor+1) where physical is the wall-clock
58+
// time with its logical bits cleared.
59+
//
60+
// floor is the highest existing HLV version value for the caller's source; passing 0 (a brand-new
61+
// document, or a source not yet present in the HLV) imposes no lower bound beyond monotonicity, so Now(0)
62+
// is an ordinary clock tick.
63+
func (c *HybridLogicalClock) Now(floor uint64) uint64 {
64+
c.mutex.Lock()
65+
defer c.mutex.Unlock()
66+
67+
next := c.clock() &^ hlcLogicalMask // physical component, tracks the wall clock
68+
if c.highestTime+1 > next { // ensure strictly greater than the previous value
69+
next = c.highestTime + 1
70+
}
71+
if floor+1 > next { // ensure strictly greater than the caller's floor
72+
next = floor + 1
73+
}
74+
c.highestTime = next
75+
return next
76+
}
77+
78+
// CASToPhysicalNanos returns the physical (wall-clock nanosecond) component of a CAS / HLC value, with the
79+
// logical counter bits cleared. The difference between two such values approximates the elapsed wall-clock
80+
// time between them, which is used to decide how long to wait for a lagging clock to catch up.
81+
func CASToPhysicalNanos(cas uint64) uint64 {
82+
return cas &^ hlcLogicalMask
83+
}

base/hlc_test.go

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
// Copyright 2026-Present Couchbase, Inc.
2+
//
3+
// Use of this software is governed by the Business Source License included
4+
// in the file licenses/BSL-Couchbase.txt. As of the Change Date specified
5+
// in that file, in accordance with the Business Source License, use of this
6+
// software will be governed by the Apache License, Version 2.0, included in
7+
// the file licenses/APL2.txt.
8+
9+
package base
10+
11+
import (
12+
"sync"
13+
"testing"
14+
"time"
15+
16+
"github.com/stretchr/testify/assert"
17+
"github.com/stretchr/testify/require"
18+
)
19+
20+
// hourNanos is an offset used to place test floors a wall-clock hour either side of "now".
21+
const hourNanos = uint64(time.Hour)
22+
23+
// TestHybridLogicalClockMonotonic asserts successive values from the system-backed clock strictly increase.
24+
func TestHybridLogicalClockMonotonic(t *testing.T) {
25+
hlc := NewHybridLogicalClock()
26+
prev := hlc.Now(0)
27+
for i := 0; i < 1000; i++ {
28+
next := hlc.Now(0)
29+
require.Greater(t, next, prev, "value at iteration %d did not strictly increase", i)
30+
prev = next
31+
}
32+
}
33+
34+
// TestHybridLogicalClockTracksWallClock asserts Now(0) reflects the wall clock (physical component) with
35+
// its logical bits cleared, rather than drifting off into logical-counter space from a cold start.
36+
func TestHybridLogicalClockTracksWallClock(t *testing.T) {
37+
hlc := NewHybridLogicalClock()
38+
before := uint64(time.Now().UnixNano()) &^ hlcLogicalMask
39+
got := hlc.Now(0)
40+
after := uint64(time.Now().UnixNano())
41+
require.GreaterOrEqual(t, got, before)
42+
require.LessOrEqual(t, got, after)
43+
}
44+
45+
// TestHybridLogicalClockSameInstant asserts that when the wall clock does not advance, successive values
46+
// still strictly increase via the logical counter.
47+
func TestHybridLogicalClockSameInstant(t *testing.T) {
48+
now := uint64(time.Now().UnixNano())
49+
hlc := &HybridLogicalClock{clock: func() uint64 { return now }}
50+
51+
first := hlc.Now(0)
52+
require.Equal(t, now&^hlcLogicalMask, first)
53+
require.Equal(t, first+1, hlc.Now(0))
54+
require.Equal(t, first+2, hlc.Now(0))
55+
}
56+
57+
// TestHybridLogicalClockPhysicalMask asserts the logical bits of the wall clock are cleared so generated
58+
// values stay in the CAS numeric space.
59+
func TestHybridLogicalClockPhysicalMask(t *testing.T) {
60+
// OR in low bits so masking is observable regardless of the current nanosecond.
61+
now := uint64(time.Now().UnixNano()) | hlcLogicalMask
62+
hlc := &HybridLogicalClock{clock: func() uint64 { return now }}
63+
64+
got := hlc.Now(0)
65+
// Isolates only bits 15–0 of the result. Requires them to be all zero — proving Now() stripped them via &^ hlcLogicalMask.
66+
require.Zero(t, got&hlcLogicalMask, "low %d (logical) bits should be cleared", hlcLogicalBits)
67+
// Both sides clear the low 16 bits of now
68+
require.Equal(t, now&^hlcLogicalMask, got)
69+
}
70+
71+
// TestHybridLogicalClockFloor asserts Now never returns a value <= floor.
72+
func TestHybridLogicalClockFloor(t *testing.T) {
73+
t.Run("floor below physical is ignored", func(t *testing.T) {
74+
now := uint64(time.Now().UnixNano())
75+
hlc := &HybridLogicalClock{clock: func() uint64 { return now }}
76+
got := hlc.Now(now - hourNanos) // a floor an hour in the past
77+
require.Equal(t, now&^hlcLogicalMask, got)
78+
})
79+
80+
t.Run("floor above physical wins", func(t *testing.T) {
81+
now := uint64(time.Now().UnixNano())
82+
hlc := &HybridLogicalClock{clock: func() uint64 { return now }}
83+
floor := now + hourNanos // a floor an hour in the future
84+
got := hlc.Now(floor)
85+
require.Equal(t, floor+1, got)
86+
require.Greater(t, got, floor)
87+
})
88+
}
89+
90+
// TestHybridLogicalClockNoFloor covers the brand-new-document / absent-source case where floor is 0: Now
91+
// behaves as an ordinary tick rather than special-casing.
92+
func TestHybridLogicalClockNoFloor(t *testing.T) {
93+
now := uint64(time.Now().UnixNano())
94+
hlc := &HybridLogicalClock{clock: func() uint64 { return now }}
95+
require.Equal(t, now&^hlcLogicalMask, hlc.Now(0))
96+
}
97+
98+
// TestHybridLogicalClockConcurrent asserts thread-safety: concurrent callers never receive duplicate
99+
// values (strict monotonicity implies uniqueness).
100+
func TestHybridLogicalClockConcurrent(t *testing.T) {
101+
hlc := NewHybridLogicalClock()
102+
103+
const goroutines = 16
104+
const perGoroutine = 500
105+
106+
var wg sync.WaitGroup
107+
results := make([][]uint64, goroutines)
108+
for g := 0; g < goroutines; g++ {
109+
wg.Add(1)
110+
go func(g int) {
111+
defer wg.Done()
112+
values := make([]uint64, perGoroutine)
113+
for i := range values {
114+
values[i] = hlc.Now(0)
115+
}
116+
results[g] = values
117+
}(g)
118+
}
119+
wg.Wait()
120+
121+
seen := make(map[uint64]struct{}, goroutines*perGoroutine)
122+
for _, values := range results {
123+
for _, v := range values {
124+
_, dup := seen[v]
125+
assert.False(t, dup, "duplicate value %d returned to concurrent callers", v)
126+
seen[v] = struct{}{}
127+
}
128+
}
129+
require.Len(t, seen, goroutines*perGoroutine)
130+
}

base/stats.go

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,7 @@ const (
9292
StatAddedVersion3dot2dot4 = "3.2.4"
9393
StatAddedVersion3dot3dot0 = "3.3.0"
9494
StatAddedVersion4dot0dot0 = "4.0.0"
95+
StatAddedVersion4dot0dot6 = "4.0.6"
9596

9697
StatDeprecatedVersionNotDeprecated = ""
9798
StatDeprecatedVersion3dot2dot0 = "3.2.0"
@@ -631,6 +632,8 @@ type DatabaseStats struct {
631632
DocWritesXattrBytes *SgwIntStat `json:"doc_writes_xattr_bytes"`
632633
// Highest sequence number seen on the caching DCP feed.
633634
HighSeqFeed *SgwIntStat `json:"high_seq_feed"`
635+
// The total number of document writes where the Sync Gateway-generated HLV version exceeded the document CAS and required a corrective re-stamp. A non-zero value indicates clock skew between Sync Gateway and Couchbase Server.
636+
HLVVersionCASRetryCount *SgwIntStat `json:"hlv_version_cas_retry_count"`
634637
// The number of attachments compacted
635638
NumAttachmentsCompacted *SgwIntStat `json:"num_attachments_compacted"`
636639
// The total number of documents read via Couchbase Lite 2.x replication since Sync Gateway node startup.
@@ -1732,6 +1735,10 @@ func (d *DbStats) initDatabaseStats() error {
17321735
if err != nil {
17331736
return err
17341737
}
1738+
resUtil.HLVVersionCASRetryCount, err = NewIntStat(SubsystemDatabaseKey, "hlv_version_cas_retry_count", StatUnitNoUnits, HLVVersionCASRetryCountDesc, StatAddedVersion4dot0dot6, StatDeprecatedVersionNotDeprecated, StatStabilityInternal, labelKeys, labelVals, prometheus.CounterValue, 0)
1739+
if err != nil {
1740+
return err
1741+
}
17351742
resUtil.PublicRestBytesWritten, err = NewIntStat(SubsystemDatabaseKey, "http_bytes_written", StatUnitBytes, PublicRestBytesWrittenDesc, StatAddedVersion3dot2dot0, StatDeprecatedVersionNotDeprecated, StatStabilityVolatile, labelKeys, labelVals, prometheus.CounterValue, 0)
17361743
if err != nil {
17371744
return err
@@ -1903,6 +1910,7 @@ func (d *DbStats) unregisterDatabaseStats() {
19031910
prometheus.Unregister(d.DatabaseStats.DocWritesBytes)
19041911
prometheus.Unregister(d.DatabaseStats.DocWritesXattrBytes)
19051912
prometheus.Unregister(d.DatabaseStats.HighSeqFeed)
1913+
prometheus.Unregister(d.DatabaseStats.HLVVersionCASRetryCount)
19061914
prometheus.Unregister(d.DatabaseStats.DocWritesBytesBlip)
19071915
prometheus.Unregister(d.DatabaseStats.NumAttachmentsCompacted)
19081916
prometheus.Unregister(d.DatabaseStats.NumDocReadsBlip)

base/stats_descriptions.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -263,6 +263,8 @@ const (
263263

264264
HighSeqFeedDesc = "Highest sequence number seen on the caching DCP feed."
265265

266+
HLVVersionCASRetryCountDesc = "The total number of document writes where the Sync Gateway-generated HLV version exceeded the document CAS and required a corrective re-stamp. A non-zero value indicates clock skew between Sync Gateway and Couchbase Server."
267+
266268
NumAttachmentsCompactedDesc = "The number of attachments compacted import_feed"
267269

268270
ImportFeedDesc = "Contains low level dcp stats: (a). dcp_backfill_expected - the expected number of sequences in backfill (b). dcp_backfill_completed - the number of backfill items processed (c). dcp_rollback_count - the number of DCP rollbacks"

db/changes_test.go

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -290,7 +290,7 @@ func TestCVPopulationOnChangeEntry(t *testing.T) {
290290
defer db.Close(ctx)
291291
collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
292292
collectionID := collection.GetCollectionID()
293-
bucketUUID := db.EncodedSourceID
293+
sourceID := db.EncodedSourceID
294294

295295
collection.ChannelMapper = channels.NewChannelMapper(ctx, channels.DocChannelsSyncFunction, db.Options.JavascriptTimeout)
296296

@@ -317,8 +317,8 @@ func TestCVPopulationOnChangeEntry(t *testing.T) {
317317

318318
docVersion := GetChangeEntryCV(t, changes[0])
319319
assert.Equal(t, doc.ID, changes[0].ID)
320-
assert.Equal(t, bucketUUID, docVersion.SourceID)
321-
assert.Equal(t, doc.Cas, docVersion.Value)
320+
assert.Equal(t, sourceID, docVersion.SourceID)
321+
assert.Equal(t, doc.HLV.Version, docVersion.Value)
322322
}
323323

324324
func TestDocDeletionFromChannelCoalesced(t *testing.T) {
@@ -583,7 +583,7 @@ func TestCurrentVersionPopulationOnChannelCache(t *testing.T) {
583583
defer db.Close(ctx)
584584
collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
585585
collectionID := collection.GetCollectionID()
586-
bucketUUID := db.EncodedSourceID
586+
sourceID := db.EncodedSourceID
587587
collection.ChannelMapper = channels.NewChannelMapper(ctx, channels.DocChannelsSyncFunction, db.Options.JavascriptTimeout)
588588

589589
// Make channel active
@@ -606,8 +606,8 @@ func TestCurrentVersionPopulationOnChannelCache(t *testing.T) {
606606

607607
// assert that the source and version has been populated with the channel cache entry for the doc
608608
assert.Equal(t, "doc1", entries[0].DocID)
609-
assert.Equal(t, base.HexCasToUint64(doc.SyncData.Cas), entries[0].Version)
610-
assert.Equal(t, bucketUUID, entries[0].SourceID)
609+
require.NotZero(t, entries[0].Version)
610+
assert.Equal(t, sourceID, entries[0].SourceID)
611611
assert.Equal(t, doc.HLV.SourceID, entries[0].SourceID)
612612
assert.Equal(t, doc.HLV.Version, entries[0].Version)
613613
}

0 commit comments

Comments
 (0)