Skip to content

Commit 69e9bf5

Browse files
davidkellydavidkelly
committed
NCBC-3999: Zone Aware read from replica for txns
Motivation ========== Nice to have since we have this capability in the sdk outside transactions. Modification ============ New AttemptContext call, which uses the lookupInAnyReplica functionality. Mostly just sprinkling in a boolean allowReplica in existing get logic, then using it deep inside. Small error handling update so we can return DocumentUnretrievableException or FeatureNotSupported when appropriate. Results ======= Fit tests specific to the extReplicaFromPreferredGroup all pass. Change-Id: I34cc1ba412677bc84b0c98d2d9a91c1326ee851d Reviewed-on: https://review.couchbase.org/c/couchbase-net-client/+/229386 Reviewed-by: Michael Reiche <michael.reiche@couchbase.com> Tested-by: Build Bot <build@couchbase.com>
1 parent 79a8bda commit 69e9bf5

4 files changed

Lines changed: 79 additions & 25 deletions

File tree

‎src/Couchbase/Client/Transactions/AttemptContext.cs‎

Lines changed: 37 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
using System.Text.Json;
1010
using System.Threading;
1111
using System.Threading.Tasks;
12+
using System.Transactions;
1213
using Couchbase.Core;
1314
using Couchbase.Core.Compatibility;
1415
using Couchbase.Core.Diagnostics.Tracing;
@@ -187,9 +188,24 @@ public async Task<TransactionGetResult> GetAsync(ICouchbaseCollection collection
187188
return GetOptionalAsync(collection, id, options);
188189
}
189190

191+
public async Task<TransactionGetResult?> GetReplicaFromPreferredServerGroup(
192+
ICouchbaseCollection collection, string id,
193+
TransactionGetOptionsBuilder? options = null)
194+
{
195+
var opt = (options ??= TransactionGetOptionsBuilder.Default).Build();
196+
197+
return await _opWrapper.WrapOperationAsync(
198+
() => GetWithKv(collection, id, opt, allowReplica: true),
199+
() => throw CreateError(this, ErrorClass.FailOther,
200+
new FeatureNotAvailableException(
201+
"GetReplicaFromPreferredServerGroup cannot be mixed with query operations"))
202+
.Build()).CAF();
203+
204+
}
205+
190206

191207
private async Task<TransactionGetResult?> GetWithKv(ICouchbaseCollection collection, string id,
192-
TransactionGetOptions options)
208+
TransactionGetOptions options, bool allowReplica = false)
193209
{
194210
using var traceSpan = TraceSpan(parent: options.Span);
195211
DoneCheck();
@@ -226,13 +242,19 @@ public async Task<TransactionGetResult> GetAsync(ICouchbaseCollection collection
226242
{
227243
await _testHooks.BeforeDocGet(this, id).CAF();
228244

229-
var result = await GetWithMav(collection, id, options.Transcoder, parentSpan: traceSpan.Item).CAF();
245+
var result = await GetWithMav(collection, id, options.Transcoder,
246+
parentSpan: traceSpan.Item, allowReplica: allowReplica).CAF();
230247

231248
await _testHooks.AfterGetComplete(this, id).CAF();
232249
await ForwardCompatibility.Check(this, ForwardCompatibility.Gets,
233250
result?.TransactionXattrs?.ForwardCompatibility).CAF();
234251
return result;
235252
}
253+
catch (Exception ex) when (ex is FeatureNotAvailableException ||
254+
ex is DocumentUnretrievableException)
255+
{
256+
throw;
257+
}
236258
catch (Exception ex)
237259
{
238260
var tr = _triage.TriageGetErrors(ex);
@@ -295,7 +317,8 @@ await ForwardCompatibility.Check(this, ForwardCompatibility.Gets,
295317
}
296318

297319
private async Task<TransactionGetResult?> GetWithMav(ICouchbaseCollection collection, string id,
298-
ITypeTranscoder? transcoder, string? resolveMissingAtrEntry = null, IRequestSpan? parentSpan = null)
320+
ITypeTranscoder? transcoder, string? resolveMissingAtrEntry = null, IRequestSpan? parentSpan = null,
321+
bool allowReplica = false)
299322
{
300323
using var traceSpan = TraceSpan(parent: parentSpan);
301324

@@ -304,7 +327,7 @@ await ForwardCompatibility.Check(this, ForwardCompatibility.Gets,
304327
{
305328
// Do a Sub-Document lookup, getting all transactional metadata, the “$document” virtual xattr,
306329
// and the document’s body. Timeout is set as in Timeouts.
307-
var docLookupResult = await _docs.LookupDocumentAsync(collection, id, fullDocument: true, transcoder: transcoder).CAF();
330+
var docLookupResult = await _docs.LookupDocumentAsync(collection, id, fullDocument: true, transcoder: transcoder, allowReplica: allowReplica).CAF();
308331
Logger.LogDebug("{method} for {redactedId}, attemptId={attemptId}, postCas={postCas}",
309332
nameof(GetWithMav), Redactor.UserData(id), AttemptId, docLookupResult.Cas);
310333
if (docLookupResult == null)
@@ -360,9 +383,18 @@ await ForwardCompatibility.Check(this, ForwardCompatibility.Gets,
360383
catch (ActiveTransactionRecordEntryNotFoundException)
361384
{
362385
// Recursively call this section from the top, passing resolvingMissingATREntry set to the attemptId of the blocking transaction.
363-
return await GetWithMav(collection, id, transcoder, resolveMissingAtrEntry = blockingTxn.Id!.AttemptId,
386+
return await GetWithMav(collection, id, transcoder,
387+
resolveMissingAtrEntry = blockingTxn.Id!.AttemptId,
364388
traceSpan.Item).ConfigureAwait(false);
365389
}
390+
catch (Exception ex) when (ex is FeatureNotAvailableException ||
391+
ex is DocumentUnretrievableException)
392+
{
393+
// seems you did a zone-aware get and it failed, we don't make TransactionOperationFailed
394+
// exceptions for these, they go back to the lambda directly.
395+
Logger.LogInformation("Got {ex} during GetWithMav", ex);
396+
throw;
397+
}
366398
catch (Exception atrLookupException)
367399
{
368400
var atrLookupTriage = _triage.TriageAtrLookupInMavErrors(atrLookupException);

‎src/Couchbase/Client/Transactions/DataAccess/DocumentRepository.cs‎

Lines changed: 39 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
using Couchbase.Client.Transactions.Support;
1515
using Couchbase.Core.Configuration.Server;
1616
using Couchbase.Core.IO.Operations;
17+
using Couchbase.KeyValue.ZoneAware;
1718
using Newtonsoft.Json;
1819
using Newtonsoft.Json.Linq;
1920
using JsonSerializer = Newtonsoft.Json.JsonSerializer;
@@ -217,8 +218,8 @@ public async Task ClearTransactionMetadata(ICouchbaseCollection collection, stri
217218
_ = await collection.MutateInAsync(docId, specs, opts).CAF();
218219
}
219220

220-
public async Task<DocumentLookupResult> LookupDocumentAsync(ICouchbaseCollection collection, string docId, bool fullDocument = true, ITypeTranscoder? transcoder = null) => await LookupDocumentAsync(collection, docId, _keyValueTimeout, fullDocument, transcoder).CAF();
221-
internal static async Task<DocumentLookupResult> LookupDocumentAsync(ICouchbaseCollection collection, string docId, TimeSpan? keyValueTimeout, bool fullDocument = true, ITypeTranscoder? transcoder = null)
221+
public async Task<DocumentLookupResult> LookupDocumentAsync(ICouchbaseCollection collection, string docId, bool fullDocument = true, ITypeTranscoder? transcoder = null, bool allowReplica = false) => await LookupDocumentAsync(collection, docId, _keyValueTimeout, fullDocument, transcoder, allowReplica).CAF();
222+
internal static async Task<DocumentLookupResult> LookupDocumentAsync(ICouchbaseCollection collection, string docId, TimeSpan? keyValueTimeout, bool fullDocument = true, ITypeTranscoder? transcoder = null, bool allowReplica = false)
222223
{
223224
var specs = new List<LookupInSpec>()
224225
{
@@ -227,33 +228,52 @@ internal static async Task<DocumentLookupResult> LookupDocumentAsync(ICouchbaseC
227228
LookupInSpec.Get(TransactionFields.StagedData, isXattr: true),
228229
LookupInSpec.Get(TransactionFields.BinStagedData, isXattr: true, isBinary: true),
229230
};
230-
231-
var opts = new LookupInOptions().Defaults(keyValueTimeout).AccessDeleted(true).Transcoder(transcoder ?? new JsonTranscoder(Transactions.MetadataSerializer));
232-
233-
int? txnIndex = 0;
234-
int docMetaIndex = 1;
235-
int? stagedDataIndex = 2;
236-
int? stagedBinDataIndex = 3;
237-
238231
int? fullDocIndex = null;
239232
if (fullDocument)
240233
{
241234
specs.Add(LookupInSpec.GetFull());
242235
fullDocIndex = specs.Count - 1;
243236
}
237+
ILookupInResult? lookupInResult;
238+
if (allowReplica)
239+
{
240+
var opts = new LookupInAnyReplicaOptions()
241+
.Timeout(keyValueTimeout)
242+
.Transcoder(transcoder ?? new JsonTranscoder(Transactions.MetadataSerializer))
243+
.ReadPreference(ReadPreference.SelectedServerGroup);
244+
lookupInResult = await collection.LookupInAnyReplicaAsync(docId, specs, opts).CAF();
245+
246+
}
247+
else
248+
{
249+
var opts = new LookupInOptions().Defaults(keyValueTimeout)
250+
.AccessDeleted(true)
251+
.Transcoder(transcoder ?? new JsonTranscoder(Transactions.MetadataSerializer));
252+
lookupInResult = await collection.LookupInAsync(docId, specs, opts).CAF();
253+
}
244254

245-
var lookupInResult = await collection.LookupInAsync(docId, specs, opts).CAF();
255+
int txnIndex = 0;
256+
int docMetaIndex = 1;
257+
int stagedDataIndex = 2;
258+
int stagedBinDataIndex = 3;
246259

247260
var docMeta = lookupInResult.ContentAs<DocumentMetadata>(docMetaIndex);
261+
int dataIdx = stagedDataIndex; // just need a default
262+
if (lookupInResult.Exists(txnIndex))
263+
{
264+
// only StageData or BinStagedData - figure out which...
265+
dataIdx = lookupInResult.Exists(stagedDataIndex)
266+
? stagedDataIndex
267+
: stagedBinDataIndex;
248268

249-
// only StageData or BinStagedData - figure out which...
250-
var dataIdx = lookupInResult.Exists(stagedDataIndex.Value) ? stagedDataIndex.Value : stagedBinDataIndex.Value;
269+
// several possibilities for the transcoder...
270+
if (dataIdx == stagedBinDataIndex && transcoder == null)
271+
{
272+
transcoder = new RawBinaryTranscoder();
273+
}
251274

252-
// several possibilities for the transcoder...
253-
if (dataIdx == stagedBinDataIndex.Value && transcoder == null)
254-
{
255-
transcoder = new RawBinaryTranscoder();
256275
}
276+
257277
transcoder ??= new JsonTranscoder(
258278
NonStreamingSerializerWrapper.FromCluster(collection.Scope.Bucket.Cluster));
259279

@@ -272,9 +292,9 @@ internal static async Task<DocumentLookupResult> LookupDocumentAsync(ICouchbaseC
272292
docMeta,
273293
collection);
274294

275-
if (txnIndex.HasValue && lookupInResult.Exists(txnIndex.Value))
295+
if (lookupInResult.Exists(txnIndex))
276296
{
277-
result.TransactionXattrs = lookupInResult.ContentAs<TransactionXattrs>(txnIndex.Value);
297+
result.TransactionXattrs = lookupInResult.ContentAs<TransactionXattrs>(txnIndex);
278298
}
279299

280300
return result;

‎src/Couchbase/Client/Transactions/DataAccess/IDocumentRepository.cs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,8 @@ internal interface IDocumentRepository
2222
Task ClearTransactionMetadata(ICouchbaseCollection collection, string docId, ulong cas, bool isDeleted);
2323

2424
Task<DocumentLookupResult> LookupDocumentAsync(ICouchbaseCollection collection,
25-
string docId, bool fullDocument = true, ITypeTranscoder? transcoder = null);
25+
string docId, bool fullDocument = true, ITypeTranscoder? transcoder = null,
26+
bool allowReplica = false);
2627
bool SupportsReplaceBodyWithXattr(ICouchbaseCollection collection);
2728

2829
}

‎src/Couchbase/Client/Transactions/ProtocolVersion.cs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@ public static IEnumerable<ExtensionName> ExtensionsSupported()
3535
yield return new ExtensionName("ExtParallelUnstaging", "EXT_PARALLEL_UNSTAGING", "PU");
3636
yield return new ExtensionName("ExtBinarySupport", "EXT_BINARY_SUPPORT", "BS");
3737
yield return new ExtensionName("ExtMemoryOptUnstaging", "EXT_MEMORY_OPT_UNSTAGING", "MO");
38+
yield return new ExtensionName("ExtReplicaFromPreferredGroup", "EXT_REPLICA_FROM_PREFERRED_GROUP", "RP");
3839
}
3940

4041
internal static bool Supported(string shortCode) => SupportedShortCodes.Value.Contains(shortCode);

0 commit comments

Comments
 (0)