hubcio commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4270145159
##########
core/server/src/dispatch/mod.rs:
##########
@@ -256,25 +440,50 @@ async fn drain_client_requests<B, MJ, S, SB>(
SB: SuperblockStore + 'static,
{
loop {
- let Some(message) = pop_next_client_request(&queues, client_id) else {
+ let metadata_frontier =
shard.plane.metadata().applied_frontier().get();
+ let next = pop_next_client_request(&queues, client_id, |queued| {
+ if is_ping(&queued.message) {
+ return true;
+ }
+ if drain.metadata_write.get() {
+ return false;
+ }
+ queued
+ .progress_key(metadata_frontier, |message|
progress_key(&shard, message))
+ .is_none_or(|key| {
+ drain.in_flight.get() < drain.capacity
+ && !drain.progress.borrow().contains(&key)
Review Comment:
PR4 now tests a held keyed Store followed by queued Next(auto=false) and
GetConsumerOffset: both remain behind its progress guard. Optional shared
read-only lanes are deferred to PR6 and require explicit Store/Delete ordering
regressions before changing that serialization. This split is recorded in B7
and the PR6 plan.
##########
foreign/go/client/tcp/tcp_core.go:
##########
@@ -860,6 +857,9 @@ func (c *IggyTcpClient) attempt(
if captured, ok :=
ctx.Value(capturedPartitionContext{}).(iggcon.PartitionContext); ok {
Review Comment:
PR4 removes the duplicate context lookup. The remaining context.WithValue
boxing/allocation work is deferred to PR6, recorded with Go RTT, copies, key
reuse and locking measurements in B7. Those measurements follow PR5's managed
link/heartbeat lifecycle gates; they do not replace the lifecycle proof.
##########
foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs:
##########
@@ -201,105 +120,142 @@ private static IggyInvalidStatusCodeException
TruncatedSend()
return VsrError.Exception(VsrError.INVALID_COMMAND, "The SendMessages
body is truncated.");
}
- private static ContextQuery? TryBuildContextQuery(int code,
ReadOnlySpan<byte> body)
+ private static SendTarget PartitionTarget(int code, ReadOnlySpan<byte>
body)
{
- if (code is CommandCodes.STORE_CONSUMER_OFFSET_CODE or
CommandCodes.DELETE_CONSUMER_OFFSET_CODE)
+ if (code == CommandCodes.SEND_MESSAGES_CODE)
{
- var suffixSize = code == CommandCodes.STORE_CONSUMER_OFFSET_CODE
- ? StoreOffsetSuffixSize : DeleteOffsetSuffixSize;
- if (body.Length <= suffixSize)
- {
- return null;
- }
- var query = body[..^suffixSize].ToArray();
- return new
ContextQuery(CommandCodes.GET_CONSUMER_OFFSET_ROUTING_CODE, query,
ConsumerSessionSize,
- Convert.ToBase64String(query));
+ return ParseSendTarget(body);
}
- if (code != CommandCodes.POLL_MESSAGES_CODE || body.Length <=
PollOptionsSize)
+
+ if (body.IsEmpty)
{
- return null;
+ throw TruncatedSend();
}
- return new ContextQuery(CommandCodes.GET_POLL_ROUTING_CODE,
body.ToArray(), ConsumerSessionSize,
- Convert.ToBase64String(body[..^PollOptionsSize]));
+ var group = body[0] == ConsumerGroupWireKind;
+ var position = 1;
+ _ = ReadIdentifier(body, ref position);
+ var identifiersStart = position;
+ var streamId = ReadIdentifier(body, ref position);
+ var topicId = ReadIdentifier(body, ref position);
+ var identifiersLength = position - identifiersStart;
+ if (position + 1 + sizeof(uint) > body.Length)
+ {
+ throw TruncatedSend();
+ }
+ var present = body[position++] != 0;
+ if (!present && (code is CommandCodes.STORE_CONSUMER_OFFSET_CODE or
CommandCodes.DELETE_CONSUMER_OFFSET_CODE
+ || code == CommandCodes.POLL_MESSAGES_CODE && group))
+ {
+ throw VsrError.Exception(VsrError.INVALID_COMMAND, "This partition
request requires an explicit partition id.");
+ }
+ var partitionId = present ?
BinaryPrimitives.ReadUInt32LittleEndian(body[position..]) : 0;
+ var query = new byte[identifiersLength + sizeof(uint)];
+ body.Slice(identifiersStart, identifiersLength).CopyTo(query);
+
BinaryPrimitives.WriteUInt32LittleEndian(query.AsSpan(identifiersLength),
partitionId);
+ return new SendTarget(new PartitionRouteKey(streamId, topicId,
partitionId), query);
+ }
+
+ private static byte[] RouteQuery(PartitionRouteKey key)
+ {
+ var query = new byte[IdentifierPrefixSize * 2 + key.StreamId.Length +
key.TopicId.Length + sizeof(uint)];
+ query.AsSpan().WriteBytesFromStreamAndTopicIdentifiers(key.StreamId,
key.TopicId);
+ BinaryPrimitives.WriteUInt32LittleEndian(query.AsSpan(query.Length -
sizeof(uint)), key.PartitionId);
+ return query;
}
- /// <summary>
- /// Sends a poll or an offset write to the partition primary, over a
data connection that leaves the
- /// membership connection where it is. A caller context, or else the
first context a route reports, is
- /// retained across every retry, so a retry never moves the request
onto a newer incarnation or owner.
- /// </summary>
- /// <param name="code">The poll or offset write command code.</param>
- /// <param name="key">The cached route of the request.</param>
- /// <param name="routing">The body of the routing request.</param>
- /// <param name="payload">The body of the request.</param>
- /// <param name="context">The caller context, or null to take the context
the route reports.</param>
- /// <param name="token">The cancellation token.</param>
- private async Task<IMemoryOwner<byte>> SendRoutedAsync(int code,
PollRouteKey key, ReadOnlyMemory<byte> routing,
+ private async Task<IMemoryOwner<byte>> SendRoutedAsync(int code,
SendTarget target,
ReadOnlyMemory<byte> payload, PartitionContext? context,
CancellationToken token)
{
using var cancellation =
CancellationTokenSource.CreateLinkedTokenSource(token);
cancellation.CancelAfter(VsrRequestTimeoutMs);
var deadline = Environment.TickCount64 + VsrRequestTimeoutMs;
+ // A new route can change the destination, but not the request's
identity or captured position.
var captured = context;
+ SessionFrame? frame = null;
+ ulong? generation = null;
+ var refreshHistory = code == CommandCodes.SEND_MESSAGES_CODE &&
context is null;
+ var refusedPrimary = false;
+ var lifecycleInterval = LifecycleRetryIntervalMs;
try
{
while (true)
{
try
{
- if (Volatile.Read(ref _clusterNodeCount) == 0)
+ var route = await GetPartitionRouteAsync(target.Key,
target.Query, deadline, cancellation.Token);
+ captured ??= route.Route.Context;
+ if (generation is not null && generation !=
route.Generation)
{
- using var metadata = await
SendPollControlAsync(CommandCodes.GET_CLUSTER_METADATA_CODE,
- ReadOnlyMemory<byte>.Empty, deadline,
cancellation.Token);
- if (metadata.Memory.Length == 0)
- {
- throw new MalformedResponseException("Poll routing
requires a nonempty cluster roster.");
- }
-
-
RememberRoster(BinaryMapper.MapClusterMetadata(metadata.Memory.Span));
+ throw PollNotAccepted();
}
-
+ generation ??= route.Generation;
+ frame ??=
_consensusSession.Resolve(VsrOperations.ForCode(code));
if (Volatile.Read(ref _clusterNodeCount) <= 1)
{
- captured ??= await CapturePartitionContextAsync(code,
payload, cancellation.Token);
- return await SendWithResponseAsync(code, payload,
token: cancellation.Token,
- capturedContext: captured);
+ var attempt = await SendVsrAttemptAsync(code, payload,
Environment.TickCount64, deadline,
+ false, cancellation.Token, captured.Value, frame,
+ retryTransient: VsrOperations.ForCode(code) !=
VsrOperation.NonReplicated);
+ if (attempt.Error is not null)
+ {
+ if (attempt.RequestStarted &&
!VsrOperations.IsReplaySafeRead(code, false, payload.Span)
+ && !IsDefinitiveVerdict(attempt.Error))
+ {
+ throw new
VsrRequestOutcomeUnknownException(attempt.Error);
+ }
+ ExceptionDispatchInfo.Throw(attempt.Error);
+ }
+ return attempt.Response!;
}
- var route = await GetPollRouteAsync(key, routing,
deadline, cancellation.Token);
- captured ??= route.Context;
- return await SendPrimaryOnceAsync(key,
- code == CommandCodes.POLL_MESSAGES_CODE ?
CommandCodes.POLL_MESSAGES_ON_PRIMARY_CODE : code,
- payload, route, captured.Value, deadline,
cancellation.Token);
+ return await SendPrimaryOnceAsync(target.Key, code,
payload, route, captured.Value,
+ frame.Value, deadline, cancellation.Token);
}
- catch (IggyInvalidStatusCodeException error) when
(error.StatusCode == VsrError.TRANSIENT_NOT_ACCEPTED)
+ catch (IggyInvalidStatusCodeException error) when
(error.StatusCode is
+ VsrError.TRANSIENT_NOT_ACCEPTED or VsrError.NOT_PRIMARY)
{
- RemovePollRoute(key);
- if (Environment.TickCount64 + PollRoutingRetryIntervalMs
>= deadline)
+ RemovePartitionRoute(target.Key);
+ if ((generation is not null && generation !=
_consensusSession.Generation)
Review Comment:
Deferred to PR5 native route-activation acceptance. B7 and the PR5 plan
require LifecycleBusy or HistoryUnavailable followed by relogin so the
post-refresh generation guard is exercised independently. Removing either guard
alone must fail its own regression while the other remains enabled. The
existing dual-guard mutation test is not claimed as independent proof.
##########
core/shard/src/lib.rs:
##########
@@ -5197,7 +5345,18 @@ where
let Some(partition) = planes.1.0.get_mut_by_ns(&namespace) else {
return;
};
- if !partition.consensus().is_normal() {
+ if !self.partition_incarnation_matches(
+ partition,
+ header.partition_incarnation,
+ Command::RequestPrepares,
+ ) {
+ return;
+ }
+ if !matches!(
Review Comment:
The new ViewChange donor exposure is fixed in b048d0263: requests carry the
requester view; a ViewChange donor serves its uncommitted suffix only to that
view's primary, while ordinary backups and other views receive at most its
locally applied prefix. RepairDone/RangeEvicted must also match the selected
peer and nonce. The regression captures real repair frames and preserves
primary-elect missing-body serving.
The broader pre-existing requester-side trust problem is deferred to PR6 as
a correctness/release gate in B5 and the plan. Repair bodies still lack
sender/nonce attribution. Closure requires canonical or committed evidence for
unanchored history and end-to-end five-/seven-replica commit, failover and
readback tests, including stale normal donors, blank canonical heads and
delayed bodies after peer rotation. The donor regression and terminator check
do not claim that broader proof.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
