numinnex commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4268040971
##########
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:
nit: lane removal is barely pinned. Ignoring `drain.progress` here, or never
calling `admit_progress` (:458-461), survives all 102 dispatch tests, and
pipelined_frames_vsr.rs:182 still passes 3/3; only :156 catches it. No test
covers Store -> Next(auto_commit=false) or Store -> GetConsumerOffset ordering.
Add a drain-level unit test where a held keyed store keeps a queued Next poll
queued until its guard drops.
Minor, PR6-able: the lane is exclusive, so read-only keyed frames also
serialize among themselves.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -557,25 +814,51 @@ async fn handle_client_request<B, MJ, S, SB>(
// (`submit_register_local_or_forward`). Every other non-replicated
// code MUST go through Register first, which binds the acting user
// the per-op authz gates resolve.
- handle_non_replicated_request(
- shard,
- sessions,
- server_config,
- transport_client_id,
- request,
- ConnectionContext {
- bound,
- user_id,
- address: client_address,
- metadata_watermark,
- },
- )
- .await;
+ let context = ConnectionContext {
+ bound,
+ user_id,
+ address: client_address,
+ metadata_watermark,
+ };
+ // `ClientDrain` says which frames stay on the drain.
+ if matches!(non_replicated_code(header), PING_CODE |
BIND_SESSION_CODE) {
+ handle_non_replicated_request(
+ shard,
+ sessions,
+ server_config,
+ transport_client_id,
+ request,
+ context,
+ )
+ .await;
+ return;
+ }
+ let in_flight = match progress {
+ Some(progress) => progress,
+ None => drain.admit(Admission::Cap).await,
+ };
+ let task_shard = Rc::clone(shard);
+ let task_sessions = Rc::clone(sessions);
+ let task_config = Arc::clone(server_config);
+ shard.bus.spawn(async move {
Review Comment:
**warning** (re the PR6 deferral): the held-poll PING test only refutes
running the whole poll inline. The proposed design is different: run the read
future inline on the drain, select it against the frame-arrival wake
(:178-181), and on wake move the same future, with its `InFlight` and lane
guard, into `shard.bus.spawn`.
PING is then answered before the held owner reply by construction. Cap,
lane, cancellation and unknown-outcome behaviour are unchanged (same future,
same guard drop). Next/auto-commit order is stricter while inline and identical
after promotion. Lockstep connections spawn nothing, which is master's cost.
Master ran reads inline and fails this PING test too, so this is a new
contract the promotion meets. It's about 40 lines here. As it stands the PR
ships -17% poll / -48% consumer for every lockstep SDK through the edge builds.
##########
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:
nit: two `ctx.Value` chain walks per stamp (:857, :860), plus a
`context.WithValue` that boxes `PartitionContext` once per routed op
(tcp_poll_routing.go:244, :406). Do one lookup with if/else.
##########
core/simulator/src/bin/workload-fuzz.rs:
##########
@@ -586,13 +608,19 @@ fn run_quiesce_phase(
};
println!(
"quiesced and converged (leader-relative; entity oracle:
{entity_oracle}; \
- evictions={}; ops_compared={} partition_ops_compared={}
replicas_compared={} \
- namespaces_checked={})",
+ evictions={}; ops_compared={} partition_ops_compared={} \
+ deleted_partition_ops_compared={} replicas_compared={}
namespaces_checked={} \
+ owners_compared={} offsets_compared_with_shadow={}
stale_owner_stores={} owner_fence_refusals={})",
workload.evictions(),
convergence.ops_compared,
convergence.partition_ops_compared,
+ workload.deleted_partition_ops_compared(),
convergence.replicas_compared,
convergence.namespaces_checked,
+ convergence.owners_compared,
+ convergence.offsets_compared_with_shadow,
+ workload.stale_owner_stores(),
+ workload.owner_fence_refusals(),
Review Comment:
nit: `stale_owner_stores` / `owner_fence_refusals` are only printed. Unlike
`min_owners_compared` (:656-659) there is no vacuity floor, and 4/12 lifecycle
seeds submit zero stale stores. Exit non-zero on lifecycle `--shell` runs with
zero refusals.
##########
.github/workflows/pre-merge.yml:
##########
@@ -237,11 +237,17 @@ jobs:
uses: actions/github-script@v9
with:
script: |
- const jobs = await github.rest.actions.listJobsForWorkflowRun({
- owner: context.repo.owner,
- repo: context.repo.repo,
- run_id: context.runId
- });
+ let jobs;
+ try {
+ jobs = await github.rest.actions.listJobsForWorkflowRun({
+ owner: context.repo.owner,
+ repo: context.repo.repo,
+ run_id: context.runId
+ });
+ } catch (error) {
+ core.warning(`Job timings unavailable: ${error.message}`);
Review Comment:
nit: this CI fix is unrelated to partition routing and will ride inside the
`feat(server)!` squash. Better as its own `ci` PR.
##########
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:
nit: the new test only fails if both :189 and :216 are reverted. Reverting
:216 alone still passes 117/117. Add a case that reaches :189
(LifecycleBusy/HistoryUnavailable, then relogin) so each guard is pinned.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -377,6 +381,74 @@ async fn relay_partition_reply<B, MJ, S, SB>(
});
}
+/// The offset key shared by stored-offset reads and progress writes on a
connection.
+/// Failed resolution is left to the normal handler's authorization and error
path.
+pub(super) fn progress_key<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ message: &Message<GenericHeader>,
+) -> Option<ConsumerProgress>
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header =
PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let request = message.try_as_typed::<RequestHeader>().ok()?;
+ let wire = match request.header().operation {
+ Operation::NonReplicated
+ if request.header().reserved[..4] ==
GET_CONSUMER_OFFSET_CODE.to_le_bytes() =>
+ {
+ GetConsumerOffsetRequest::decode_from(request.body()).ok()?
+ }
+ Operation::NonReplicated
+ if request.header().reserved[..4] ==
POLL_MESSAGES_CODE.to_le_bytes()
+ || request.header().reserved[..4]
+ == POLL_MESSAGES_ON_PRIMARY_CODE.to_le_bytes() =>
+ {
+ let poll = PollMessagesRequest::decode_from(request.body()).ok()?;
Review Comment:
nit: polls that are neither `Next` nor auto-commit fully decode
`PollMessagesRequest` (WireName String allocs) just to return `None`, then the
handler decodes again. Return early on the strategy/flag bytes, or carry the
decoded request into the handler.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/cluster/ClusterNode.java:
##########
@@ -19,14 +19,26 @@
package org.apache.iggy.cluster;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonSetter;
+import com.fasterxml.jackson.annotation.Nulls;
+
/**
* A single node in the cluster roster.
*
+ * @param replicaId the configured replica id, independent of roster order
* @param name the node name
* @param ip the node IP address or hostname
* @param endpoints the per-transport ports of the node
* @param role the node role
* @param status the node status
*/
public record ClusterNode(
- String name, String ip, TransportEndpoints endpoints, ClusterNodeRole
role, ClusterNodeStatus status) {}
+ @JsonProperty(value = "replica_id", required = true) @JsonSetter(nulls
= Nulls.FAIL)
+ int replicaId,
Review Comment:
nit (PR description): the leading `replicaId` record component breaks
constructor calls and record patterns. Together with C#'s `required ReplicaId`
and the CLI list's extra column, these public breaks, and the Go/Java/Node/C#
routing migration as a whole, aren't mentioned in the PR body, which becomes
the squash message and release notes. Please add an SDK paragraph listing the
breaking API changes.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -1035,21 +1077,31 @@ where
if kind != ConsumerKind::ConsumerGroup {
return owner_generation == 0;
}
- self.consumer_group_owner(u64::from(consumer_id))
- .is_some_and(|owner| {
- owner.generation == owner_generation
- && (!owner.is_unassigned()
- || (operation == Operation::DeleteConsumerOffset
- && is_auto_commit_client(client_id)))
- && (is_auto_commit_client(client_id)
- || (owner.client_id == client_id && owner.session ==
session))
- })
+ let group_id = u64::from(consumer_id);
+ let Some(owner) = self.consumer_group_owner(group_id) else {
+ // Group ids are never reused, so no owner follows a retired group.
+ // A replica can still hold its offset when the primary that
retired
+ // it had none, and a delete admitted under the fence can commit
+ // after the catalog, so this cleanup passes at any generation.
+ return operation == Operation::DeleteConsumerOffset
+ && is_auto_commit_client(client_id)
+ && self.consumer_group_retired(group_id);
+ };
+ owner.generation == owner_generation
Review Comment:
nit: the auto-commit generation fence is unpinned. Mutating to
`is_auto_commit_client(client_id) || owner.generation == owner_generation`
survives 580 partitions tests, 236 sim tests and 12 lifecycle fuzz seeds,
because the workload never emits a stale auto-commit (state_checker.rs:458
never fires end to end).
Add a partition unit test: auto-commit admitted at generation G, install G+1
commits before it applies, expect refusal.
--
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]
