updateHLV updates the HLV in the sync data appropriately based on what type of document update event we are encountering. mouMatch represents if the _mou.cas == doc.cas
(ctx context.Context, d *Document, docUpdateEvent DocUpdateType, mouMatch bool)
| 1003 | |
| 1004 | // updateHLV updates the HLV in the sync data appropriately based on what type of document update event we are encountering. mouMatch represents if the _mou.cas == doc.cas |
| 1005 | func (db *DatabaseCollectionWithUser) updateHLV(ctx context.Context, d *Document, docUpdateEvent DocUpdateType, mouMatch bool) (*Document, error) { |
| 1006 | |
| 1007 | hasHLV := d.HLV != nil |
| 1008 | if d.HLV == nil { |
| 1009 | d.HLV = &HybridLogicalVector{} |
| 1010 | base.DebugfCtx(ctx, base.KeyVV, "No existing HLV for doc %s", base.UD(d.ID)) |
| 1011 | } else { |
| 1012 | base.DebugfCtx(ctx, base.KeyVV, "Existing HLV for doc %s before modification %#v", base.UD(d.ID), d.HLV) |
| 1013 | } |
| 1014 | switch docUpdateEvent { |
| 1015 | case ExistingVersion: |
| 1016 | // preserve any other logic on the HLV that has been done by the client, only update to cvCAS will be needed |
| 1017 | d.HLV.CurrentVersionCAS = expandMacroCASValueUint64 |
| 1018 | case Import: |
| 1019 | // Do not update HLV if the current document version (cas) is already included in the existing HLV, as either: |
| 1020 | // 1. _vv.cvCAS == document.cas (current mutation is already present as cv), or |
| 1021 | // 2. _mou.cas == document.cas (current mutation is already present as cv, and was imported on a different cluster) |
| 1022 | |
| 1023 | cvCASMatch := hasHLV && d.HLV.CurrentVersionCAS == d.Cas |
| 1024 | if !hasHLV || (!cvCASMatch && !mouMatch) { |
| 1025 | // Otherwise this is an SDK mutation made by the local cluster that should be added to HLV. |
| 1026 | newVVEntry := Version{} |
| 1027 | sourceID := db.dbCtx.EncodedSourceID |
| 1028 | // use unknown source ID when CCV is enabled but doc cas is less than CCV CAS |
| 1029 | if db.dbCtx.CachedCCVEnabled.Load() { |
| 1030 | vbNo := sgbucket.VBHash(d.ID, db.dbCtx.numVBuckets) |
| 1031 | ccvStartingCas := db.dbCtx.CachedCCVStartingCas.Load(ctx, base.VBNo(vbNo)) |
| 1032 | if d.Cas <= ccvStartingCas { |
| 1033 | sourceID = unknownSourceID |
| 1034 | } |
| 1035 | } |
| 1036 | newVVEntry.SourceID = sourceID |
| 1037 | newVVEntry.Value = d.Cas |
| 1038 | err := d.HLV.AddVersion(newVVEntry) |
| 1039 | if err != nil { |
| 1040 | return nil, err |
| 1041 | } |
| 1042 | d.HLV.CurrentVersionCAS = d.Cas |
| 1043 | base.DebugfCtx(ctx, base.KeyVV, "Adding new version to HLV due to import for doc %s, updated HLV %#v", base.UD(d.ID), d.HLV) |
| 1044 | } else { |
| 1045 | base.DebugfCtx(ctx, base.KeyVV, "Not updating HLV due to _mou.cas == doc.cas for doc %s, extant HLV %#v", base.UD(d.ID), d.HLV) |
| 1046 | } |
| 1047 | case NewVersion, ExistingVersionWithUpdateToHLV: |
| 1048 | // add a new entry to the version vector |
| 1049 | newVVEntry := Version{} |
| 1050 | newVVEntry.SourceID = db.dbCtx.EncodedSourceID |
| 1051 | newVVEntry.Value = expandMacroCASValueUint64 |
| 1052 | err := d.HLV.AddVersion(newVVEntry) |
| 1053 | if err != nil { |
| 1054 | return nil, err |
| 1055 | } |
| 1056 | // update the cvCAS on the SGWrite event too |
| 1057 | d.HLV.CurrentVersionCAS = expandMacroCASValueUint64 |
| 1058 | case ExistingVersionLegacyRev: |
| 1059 | revTreeEncodedCV, err := LegacyRevToRevTreeEncodedVersion(d.GetRevTreeID()) |
| 1060 | if err != nil { |
| 1061 | return nil, err |
| 1062 | } |
no test coverage detected