hubcio commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4270146506
##########
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:
Deferred to PR6 as reply-delivery correctness work, not just a cap
benchmark. B7 and the plan require bounded reply ownership through flush,
waiting for bounded delivery, or explicit connection failure on enqueue
refusal; logging and dropping a response while leaving its request pending is
insufficient. Acceptance includes slow readers with Java async pipelines above
queue capacity, bounded retained memory, and a response or explicit connection
failure for every pending call. The ClientDrain comment now states that task
permits end at enqueue and do not reserve outbound slots.
##########
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:
Deferred to PR6 as an admission-isolation correctness gate. The earlier
target cap bounds the number of holds but does not prevent a caller without
topic permissions from starving the shared outbound pool. B7 and the plan
require trusted relay provenance and authorization before accepting untrusted
frontier holds, plus capacity isolation/reservation for legitimate forwarding.
The regression must hold 128 spoofed forwarded requests and prove unrelated
authorized traffic still progresses. Rejecting only u64::MAX or changing the
cap alone will not close this item.
##########
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:
Deferred to PR6 in B5's recovery gates. Batch only a proven contiguous
divergent range into one atomic metadata rewrite or partition continuation,
preserving unknown gaps, canonical tails, commit floors and retry after failed
I/O. Acceptance requires rewrite/fsync counts and
interrupted-publication/restart tests. Dropping unknown entries to reduce the
number of rewrites is not an acceptable shortcut.
##########
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:
Deferred to PR6. The consolidated simulator gates, B6 and the plan now
require calling the actual reconcile_consumer_group_offsets producer in a
group-retirement/recreation workload. Acceptance requires nonzero
RetireConsumerGroupOwners delivery and retirement comparisons, vacuity failure
when either is absent, and an end-to-end mutation that the strict retirement
oracle catches. The PR description now calls the existing retirement check unit
coverage only; session revocation and generation-refusal counters are separate
evidence.
--
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]
