spetz commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4269769190
##########
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:
This arm now lets a partition replica in `ViewChange` serve
`RequestPrepares`, and such a replica can hand an uncommitted op from an
abandoned branch to a lagging backup that then commits it. A backup that times
out into `ViewChange` never sets `view_log_unproven`, so
`view_log_entry_proven` passes for every op and `repair_serve_ceiling`
(`core/shard/src/lib.rs:11011`) serves through its own head. The requester
trusts whatever arrives: `on_repair_prepare` checks only frame integrity
(`core/shard/src/lib.rs:5638`), the canonical checksum comes only from
`pending_view_log` and is not required at or below `commit_to_op`
(`core/partitions/src/iggy_partition.rs:11651`), no repair path compares
`parent` with the neighboring op, `RequestPreparesHeader.view` is never set or
read, and `RepairDone` is not checked against `session.peer`
(`core/shard/src/lib.rs:5901`). With 5 replicas: the view 1 primary sends op N
only to R3 and crashes before its WAL write, R3 is cut off and times out into
`ViewChange`, R1 restarts and the others commit a different op at N while R4
drops prepare N, and R4's stalled repair rotates to R3 through
`next_transfer_peer` (`core/shard/src/lib.rs:2855`). R4 journals and commits
the abandoned op at N, later chunks with `parent` of the real op are accepted
on top of it, and if R4 becomes primary its log wins at N, so polls return
messages nobody acknowledged in place of acknowledged ones. A shard unit test
that builds this sequence (R3 in `ViewChange` serving ops 1..=3 to R4 with
`commit_max = 3`) ends with R4 at `commit_min = 3` holding the abandoned
checksum at op 3. The requester-side trust predates this PR and needs 7
replicas at base, but restricting this arm to `is_normal()` is what kept the
5-replica case out.
##########
core/server/src/http/forward.rs:
##########
@@ -241,19 +360,108 @@ pub(in crate::http) async fn forward_to_primary(
SendWrapper::new(forward_or_pass(state, request, next)).await
}
-/// Route-layer fallback for acknowledged partition writes over HTTP.
-///
-/// HTTP has no persistent leader-aware connection to retarget. Execute on the
-/// contacted node first, then retry across the configured HTTP nodes within
-/// the deadline after a typed `TransientNotAccepted` denial. That
-/// denial proves the write never entered a partition pipeline. Every ambiguous
-/// outcome is returned without replay.
-pub(in crate::http) async fn forward_partition_write(
+/// Route layer of the partition routes: buffer the request once and attach it
+/// as a [`PartitionRequest`], so its handler can relay it to the partition
+/// primary with [`route_partition`]. No attempt runs here.
+pub(in crate::http) async fn forward_partition_request(
State(state): State<HttpState>,
request: Request,
next: Next,
) -> Response {
- SendWrapper::new(forward_partition_or_pass(state, request, next)).await
+ SendWrapper::new(attach_partition_request(state, request, next)).await
+}
+
+/// Decide where a partition request runs once its partition and context are
+/// fixed. `None` serves it on this node. `Some` is the final answer: the reply
+/// of the partition primary, or a refusal that proves nothing was admitted.
+///
+/// Until the deadline, every hop waits for a view with a primary, because a
+/// new partition starts on each replica at its own pace. A hop target never
+/// relays: it refuses with `NotPrimary` when another replica is the primary,
+/// so a stale view never chains hops. The contacted node starts another turn
+/// only after such a proven non-admission, a failed connect, or a read that
+/// failed after its send.
+pub(in crate::http) async fn route_partition(
+ state: &HttpInner,
+ request: &PartitionRequest,
+ namespace: IggyNamespace,
+ context: PartitionContext,
+ effect: PartitionEffect,
+) -> Option<Response> {
+ if request.hop == PartitionHop::Unrouted {
+ return None;
+ }
+ let Ok(partition_id) = u32::try_from(namespace.partition_id()) else {
+ return
Some(CustomError::from(IggyError::InvalidCommand).into_response());
+ };
+ let mut guard = None;
+ loop {
+ if let Some(primary) = partition_primary(state, namespace).await {
+ if is_local_replica(state, primary) {
+ return None;
+ }
+ if matches!(request.hop, PartitionHop::Target(_)) {
+ return
Some(CustomError::from(IggyError::NotPrimary).into_response());
+ }
+ if !state.forward.active {
+ return Some(not_admitted());
+ }
+ if let Some(socket) = primary_http_socket(&state.roster, primary) {
+ let remaining =
request.deadline.saturating_duration_since(Instant::now());
+ // A target without time would dispatch a write it cannot wait
+ // for and answer an unknown outcome.
+ if remaining <= RELAY_MARGIN {
+ return Some(not_admitted());
+ }
+ if guard.is_none() {
+ guard = ForwardGuard::admit(&state.forward.in_flight);
+ if guard.is_none() {
+ return Some(forward_busy());
+ }
+ }
+ let route = PartitionRoute {
+ partition_id,
+ context,
+ budget: remaining.saturating_sub(RELAY_MARGIN),
+ };
+ let Ok(route) = HeaderValue::try_from(route.to_string()) else {
+ return Some(not_admitted());
+ };
+ let mut headers = request.headers.clone();
+ headers.insert(PARTITION_ROUTE_HEADER, route);
+ let url = format!(
+ "{}://{socket}{}",
+ state.forward.scheme, request.path_and_query
+ );
+ let outcome = attempt(
+ state,
+ &request.method,
+ headers,
+ &request.body,
+ &url,
+ RelayKind::Partition(effect),
+ remaining,
+ )
+ .await;
+ if let AttemptOutcome::Relay(response) = outcome {
+ return Some(match guard {
+ Some(guard)
+ if request.method == Method::GET &&
response.status().is_success() =>
+ {
+ let remaining =
+
request.deadline.saturating_duration_since(Instant::now());
+ retain_forward_guard(response, guard, remaining)
Review Comment:
The relayed body of a successful poll is now bounded by what is left of the
10s `PARTITION_REQUEST_DEADLINE`, and `retain_forward_guard` aborts the whole
stream with `TimedOut` when it runs out, even while chunks still flow
(`core/server/src/http/forward.rs:821-829`). The base gave the same span
`FORWARD_ATTEMPT_TIMEOUT` (35s), and the local path at
`core/server/src/http/handlers.rs:1340-1375` has no body timer at all. With
`auto_commit` the primary advances the offset inside the poll completion before
the reply leaves (`core/partitions/src/iggy_partition.rs:4042-4052`), so a
client that reads a relayed page slower than the remaining budget loses
messages that are already marked consumed, and the next poll skips them. HTTP
poll pages have no byte cap and carry base64 payloads, so a page of 1000
messages of 10 KB (about 13 MB) is cut for any client below about 1.3 MB/s.
This contradicts the `RESPONSE_BODY_LIMIT` doc
(`core/server/src/http/forward.rs:146-151`), which exempts success
ful polls from the size cap because cutting them discards acknowledged data.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -240,13 +284,153 @@ impl Drop for ActiveDrainSlot {
}
}
+/// The state that the drain task of one connection shares with the requests
+/// it spawned.
+///
+/// The drain takes the frames of the connection in arrival order and runs the
+/// prologue of each. Login, logout, `BIND_SESSION` and partition admission
+/// finish on the drain. Metadata commits run in tasks, but later non-PING
+/// frames wait in the queue until the metadata lane releases, before they
+/// capture session state or resolve a namespace. This orders dispatch, not
local
+/// metadata catch-up: partition writes still resolve from this shard's STM
+/// without waiting for the connection watermark. PING can pass that queue.
+/// Stored-offset reads, Next/auto-commit polls and explicit offset writes
share
+/// a lane per resolved namespace and consumer until their reply or wait
expiry.
+/// Expiry releases only this connection's guard, not an admitted mutation,
which
+/// may still commit after later progress. Independent requests can pass a
queued
+/// conflict, but cannot pass a queued session or metadata barrier. A PING
still waits
+/// behind a frame already suspended in admission or binding.
+///
+/// `capacity` caps those tasks at `message_bus.client_queue_capacity`, the
Review Comment:
The doc says the in-flight cap keeps replies from overflowing the outbound
queue, but a read task drops its `InFlight` slot as soon as `send_to_client`
enqueues the reply (`core/server/src/dispatch/mod.rs:842-853`), so the cap
counts running tasks, not queued replies. Up to 256 reads now run at once per
connection, where the base drain awaited each request before taking the next,
so a client that pipelines past `client_queue_capacity` gets replies dropped as
`Backpressure` by `try_send`, and `send_host_frame` only logs the drop and
keeps the connection open (`core/server/src/dispatch/failure.rs:105`). The Java
async SDK reaches this. It releases the channel lease right after the write for
every command except login and logout
(`foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncTcpConnection.java:626-631`)
and tracks replies in an unbounded map, so a request whose reply was dropped
waits for its timeout, and that timeout closes the channel and fails every ot
her request pending on it.
##########
core/simulator/src/workload/state_checker.rs:
##########
@@ -265,6 +390,85 @@ impl StateChecker {
}
}
+/// Membership is metadata state; the partition's ordered install barrier is
+/// what fences a departed member's later offset writes.
+fn check_partition_owner(
+ owners: &mut PartitionOwners,
+ header: &PrepareHeader,
+ body: &[u8],
+ seed: u64,
+) {
+ if header.operation == Operation::InstallConsumerGroupOwner {
+ let install = InstallConsumerGroupOwnerRequest::decode_from(body)
+ .expect("a committed owner installation decodes");
+ if !owners
+ .retirement
+ .as_ref()
+ .is_some_and(|catalog| catalog.retires(install.group_id))
+ {
+ owners.installed.insert(install.group_id, install.owner);
+ }
+ return;
+ }
+ if header.operation == Operation::RetireConsumerGroupOwners {
Review Comment:
This retirement check never runs end to end. The simulator calls only
`reconcile_pending_revocations` and `reconcile_partition_lifecycles`
(`core/simulator/src/lib.rs:1198`) and never
`reconcile_consumer_group_offsets`, the only producer of
`RetireConsumerGroupOwners` (`core/server/src/partition_reconciler.rs:1108`,
`core/server/src/partition_reconciler.rs:1216`), so the branch is reached only
by its own unit test at `core/simulator/src/workload/state_checker.rs:786`.
Comment 4263203892 asked for this branch and it was added, and commit 8d3154989
says the simulator now checks owner retirement, but no workload drives it.
Owner retirement is one of the #4456 gaps this PR says it closes.
##########
core/server/src/http/forward.rs:
##########
@@ -332,92 +536,69 @@ fn record_relayed_floor(state: &HttpInner, user_id: u32,
response: &Response) {
state.metadata_watermarks.record(user_id, applied);
}
-async fn forward_partition_or_pass(state: HttpState, request: Request, next:
Next) -> Response {
- if !state.forward.active ||
request.headers().contains_key(FORWARDED_HEADER) {
- return next.run(request).await;
- }
- // A read that does not move consumer progress is answered by whichever
- // replica the caller reached, which is the point of reading from one. Only
- // an automatic commit needs a replica that can originate the offset
- // operation, so only that poll pays for the buffering and the credential
- // resolution below.
- if request.method() == Method::GET &&
!wants_auto_commit(request.uri().query()) {
- return next.run(request).await;
- }
+async fn attach_partition_request(state: HttpState, request: Request, next:
Next) -> Response {
+ let received = Instant::now();
+ let hop = match partition_hop(request.headers()) {
+ Ok(hop) => hop,
+ Err(error) => return CustomError::from(error).into_response(),
+ };
+ // Verify-only, like the metadata relay: an unauthenticated caller gets no
+ // buffered body and no relay. The handler authenticates again.
let bearer = match bearer_token(request.headers()) {
Ok(bearer) => bearer,
Err(error) => return CustomError::from(error).into_response(),
};
if let Err(rejection) = resolve_credential(&state, bearer).await {
return rejection.into_response();
}
-
- let (parts, request_body) = request.into_parts();
+ let _guard = if matches!(hop, PartitionHop::Target(_)) {
Review Comment:
Since this admit, a Target hop takes a slot of the shared 128-slot
`forward.in_flight` pool after authentication only, before any stream or topic
permission check, and holds it for up to `PARTITION_REQUEST_DEADLINE` (10s)
while `hold_for_frontier` waits for a caller-chosen `metadata_op`
(`core/server/src/dispatch/reads.rs:249`). Both headers are taken from the
client request with no peer check (`core/server/src/http/forward.rs:943-954`),
and nothing bounds the frontier against the local commit. A user with no
permissions can send 128 concurrent requests with `iggy-forwarded: 1` and
`iggy-partition-route: 0,0,0,18446744073709551615,10000` to any partition path,
and for 10s that node answers 503 `forward_busy` to every metadata and
partition relay it starts (`core/server/src/http/forward.rs:417`,
`core/server/src/http/forward.rs:499`) and to every partition relay other nodes
send to it, repeatable without limit. Comment 4268027195 asked for this
admission, but sharing the pool with
outbound relays turns one user's slow requests into a node-wide stall of HTTP
forwarding for every user.
##########
core/consensus/src/plane_helpers.rs:
##########
@@ -903,6 +903,84 @@ where
.await
}
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ViewLogProof {
+ Proven,
+ /// Fetch `op`; retained entries above `unproven_through` match the view.
+ Missing {
+ op: u64,
+ unproven_through: u64,
+ },
+ Divergent(u64),
+}
+
+/// Prove held entries against the view's hash chain. Canonical headers can
+/// bridge missing bodies; a gap with no header leaves its lower prefix
unknown.
+#[must_use]
+pub fn view_log_proof(
+ commit_floor: u64,
+ canonical: &[PrepareHeader],
+ op_head: u64,
+ header_at: impl Fn(u64) -> Option<PrepareHeader>,
+) -> ViewLogProof {
+ let proven = locally_proven_commit(commit_floor, &header_at);
+ if op_head <= proven {
+ return ViewLogProof::Proven;
+ }
+ let Some(head) = canonical.first().filter(|head| head.op == op_head) else {
+ return ViewLogProof::Missing {
+ op: op_head,
+ unproven_through: op_head,
+ };
+ };
+ let mut expected = head.checksum;
+ let mut missing = None;
+ for op in (proven + 1..=op_head).rev() {
+ let header = if let Some(local) = header_at(op) {
+ local
+ } else if let Some(header) = canonical.iter().find(|header| header.op
== op) {
+ missing.get_or_insert(op);
+ *header
+ } else {
+ return ViewLogProof::Missing {
+ op,
+ unproven_through: op,
+ };
+ };
+ if header.checksum == CHECKSUM_UNSEALED || header.checksum != expected
{
+ return ViewLogProof::Divergent(op);
+ }
+ expected = header.parent;
+ }
+ missing.map_or(ViewLogProof::Proven, |op| ViewLogProof::Missing {
+ op,
+ unproven_through: 0,
+ })
+}
+
+fn locally_proven_commit(
+ commit_floor: u64,
+ header_at: &impl Fn(u64) -> Option<PrepareHeader>,
+) -> u64 {
+ let mut proven = commit_floor;
+ let mut op = commit_floor;
+ let mut parent = None;
+ while let Some(next_op) = op.checked_add(1) {
+ let Some(local) = header_at(next_op) else {
+ break;
+ };
+ if local.checksum == CHECKSUM_UNSEALED
+ || parent.is_some_and(|checksum| local.parent != checksum)
+ {
+ break;
+ }
+ proven = proven.max(local.commit.min(op));
Review Comment:
No unit test reaches the commit-stamp proof any more. The d54da7a3 rewrite
of the proof tests dropped the durable commit stamp case and the unsealed chain
case, and the replacement test at `core/consensus/src/plane_helpers.rs:2603`
stamps every header `commit: 2`, equal to the floor, so this line never raises
`proven`. The shard StartView tests do not cover it either, because
`primary_prepares` (`core/shard/src/partition_io.rs:4815`) never commits and
every stamp is 0. This branch decides which held entries count as committed
before a StartView proof settles.
##########
core/shard/src/lib.rs:
##########
@@ -6214,145 +6425,71 @@ where
.await;
}
- /// Compare this replica's log against the headers the view decided, and
drop or
- /// report where they disagree.
- ///
- /// Without this, divergence is silent and permanent: the replica acks
with its
- /// own checksum, the primary rejects the ack, and journal repair skips an
op
- /// it already has a header for.
- ///
- /// Both roles. A backup runs it against the `StartView` suffix it
adopted, the
- /// primary-elect against the merged log before the coverage scan in
- /// [`Self::advance_pending_metadata_view`]. The merge does NOT reconcile
the
- /// primary's log for it: nothing installs the merged headers into the
journal,
- /// `RebuildPipeline` reads the pipeline back out of it, and
`CommitJournal`
- /// applies whatever sits at each op up to the merged commit point.
- ///
- /// Runs on every adoption, parked suffix or not: an EMPTY `StartView`
suffix
- /// (`commit == op`, the steady case) parks no pending log, yet adoption
still
- /// drops the head under any journaled relics above it, and the primary's
next
- /// prepare would collide with them in `append` and poison the journal.
With
- /// no pending log the divergence scan has nothing to walk and only the
- /// above-head sweep applies, with the head read off the adopted sequencer.
- ///
- /// The split at the announced commit point is what matters. Above it a
- /// disagreement is ordinary, so the entry is dropped and the primary's
- /// retransmission refills the range. At or below it, this replica applied
- /// something the view says was different, which only state transfer
fixes, so
- /// it is reported and left alone.
- ///
- /// Truncation uses `Journal::truncate_from`, not `drain`: `drain` advances
- /// `snapshot_op` past what it removed, marking ops that must stay
refillable
- /// as evictable.
+ // The shard pump serializes these rewrites with append. The journal's
drain
+ // guard excludes concurrent compaction, but does not exclude append
itself.
#[allow(clippy::future_not_send)]
async fn reconcile_metadata_view_divergence(&self)
where
B: MessageBus,
MJ: JournalHandle,
<MJ as JournalHandle>::Target:
Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
- M: MetadataStm,
{
let metadata = self.plane.metadata();
- let Some(ref consensus) = metadata.consensus else {
+ let (Some(consensus), Some(journal)) = (&metadata.consensus,
&metadata.journal) else {
return;
};
- let pending = consensus.pending_view_log();
- let Some(journal) = metadata.journal.as_ref() else {
+ if !consensus.view_log_unproven() {
return;
- };
-
- // Truncation is safe only above what this replica has *applied*,
which is
- // not the view's commit point: `pending.commit_max` is the new
primary's
- // number and a backup can sit above it. Splitting on the view's number
- // would drop already-executed ops with no rollback, and silently.
- let announced_commit = pending.as_ref().map_or(0, |pending|
pending.commit_max);
- let applied_floor = consensus.commit_min();
-
- let mut repairable_from: Option<u64> = None;
- for canonical in pending.as_ref().map_or(&[][..], |pending|
&pending.headers) {
- let Some(local) = usize::try_from(canonical.op)
- .ok()
- .and_then(|slot| journal.handle().header(slot))
- else {
- continue;
- };
- if header_is_view_entry(&local, canonical) {
- continue;
- }
- if canonical.op <= applied_floor {
- tracing::error!(
- shard = self.id,
- op = canonical.op,
- view = consensus.view(),
- commit_max = announced_commit,
- commit_min = consensus.commit_min(),
- local_checksum = local.checksum,
- canonical_checksum = canonical.checksum,
- "committed op {} disagrees with the view that just
started; this replica \
- applied a different op as committed and cannot be
reconciled by log repair",
- canonical.op
- );
- continue;
- }
- repairable_from = Some(repairable_from.map_or(canonical.op, |op|
op.min(canonical.op)));
}
-
- // A suffix ABOVE the announced head is named by nobody, so the loop
cannot
- // see it, and it is exactly the log that AGREES in-window (restart and
- // re-adopt), where `repairable_from` never arms. Adoption drops the
head
- // under it, and the next prepare at `op_head + 1` then collides in
`append`,
- // which refuses the slot even when the ops match. Floored at the
applied
- // point too: an executed op is not rollback-able whatever the head
says.
- // With no parked suffix the adopted sequencer IS the announced head.
+ let pending = consensus.pending_view_log();
let op_head = pending.as_ref().map_or_else(
|| consensus.sequencer().current_sequence(),
|pending| pending.op_head,
);
- let above_head = op_head.max(applied_floor) + 1;
- if journal
- .handle()
- .last_op()
- .is_some_and(|last_op| last_op >= above_head)
+ let journal = journal.handle();
+ let above_head = op_head.max(consensus.commit_min()).saturating_add(1);
+ if consensus.view_log_repair_op().is_none()
+ && journal.last_op().is_some_and(|last| last >= above_head)
{
- repairable_from = Some(repairable_from.map_or(above_head, |op|
op.min(above_head)));
- }
-
- let Some(from_op) = repairable_from else {
- return;
- };
- // SERIALIZATION: the drain guard excludes `truncate_from` against
`drain`,
- // NOT against an append; that is metadata's private `journal_gate`.
It holds
- // by call-site placement, not construction: the pump is
single-threaded, and
- // both callers run with a view change parked, so no submit is
admitted and no
- // repair prepare is in flight for these ops. Routing shard-side
journal
- // mutations through gate-taking metadata methods would make it
structural.
- match journal.handle().truncate_from(from_op).await {
- Ok(removed) => {
- // The snapshot's head and commit point do not move when
entries
- // are removed under them, so without this the next
`DoViewChange`
- // would advertise the dropped headers and offer bodies this
replica
- // cannot serve.
- consensus.note_journal_mutation();
- tracing::warn!(
- shard = self.id,
- from_op,
- removed,
- op_head,
- view = consensus.view(),
- "dropped {removed} uncommitted entries from op {from_op}
that disagreed with \
- the view's log; the primary's retransmission refills the
range"
- );
+ if let Err(error) = journal.truncate_from(above_head).await {
+ tracing::error!(shard = self.id, above_head, %error, "could
not truncate superseded metadata suffix");
+ return;
}
- Err(error) => {
- tracing::error!(
- shard = self.id,
- from_op,
- %error,
- "could not drop the diverging uncommitted entries from op
{from_op}; journal \
- repair skips ops it already holds a header for, so this
replica will not \
- converge at those ops until it is restarted"
- );
+ consensus.note_journal_mutation();
+ }
+ loop {
+ match consensus::view_log_proof(
+ consensus.commit_min(),
+ pending.as_ref().map_or(&[][..], |pending| &pending.headers),
+ op_head,
+ |op| {
+ usize::try_from(op)
+ .ok()
+ .and_then(|slot| journal.header(slot).map(|header|
*header))
+ },
+ ) {
+ consensus::ViewLogProof::Proven => {
+ consensus.settle_view_log_proof();
+ return;
+ }
+ consensus::ViewLogProof::Missing {
+ op,
+ unproven_through,
+ } => {
+ consensus.fence_view_log(op, unproven_through);
+ return;
+ }
+ consensus::ViewLogProof::Divergent(op) => {
+ consensus.fence_view_log(op, op);
+ // The canonical tail may be this replica's only copy.
Remove
+ // the proven mismatch without cutting entries above a gap.
+ if let Err(error) = journal.remove_range(op..=op).await {
Review Comment:
This loop removes one divergent op per pass, and every `remove_range` runs
`rewrite_wal`, which writes every live entry again and fsyncs the temp file and
the parent directory (`core/journal/src/prepare_journal.rs:883`). A stale
branch of K ops (up to `PIPELINE_PREPARE_QUEUE_MAX`, 32) now costs K full WAL
rewrites and 2K fsyncs on the view change path, where the base cut it with one
`truncate_from`. The partition plane has the same shape at
`core/shard/src/lib.rs:11664`. Each divergent op defers its own `op..=op`
transition, which drains persistence and rebuilds the resident index, and
`ViewApplied` re-proves to find the next one, so K ops cost K IO round trips
although `PendingPartitionTransition::Truncate` already takes a range.
--
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]
