numinnex commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4268027118
##########
core/consensus/src/plane_helpers.rs:
##########
@@ -903,6 +903,62 @@ where
.await
}
+/// Find a held entry below a view's suffix that the new history does not
prove.
+///
+/// `canonical` is ordered by descending op. Entries above the local commit
floor
+/// survive only if their parent chain reaches its lowest header, or a
prepare's
+/// commit stamp proves them through the uninterrupted local chain. This runs
once
+/// per installed log, before repair starts filling any resulting gap.
+#[must_use]
+pub fn first_unproven_op(
+ commit_floor: u64,
+ canonical: &[PrepareHeader],
+ op_head: u64,
+ last_op: Option<u64>,
+ header_at: impl Fn(u64) -> Option<PrepareHeader>,
+) -> Option<u64> {
+ let last_op = last_op?;
+ let proven = locally_proven_commit(commit_floor, &header_at);
+ let (mut op, mut parent) = canonical.last().map_or((op_head, None),
|lowest| {
Review Comment:
**critical** (second trigger): with an empty `canonical` this starts at
op_head and stops on the first step. shard/lib.rs:10656 `respond_start_view`
ships `local_dvc_suffix()` without refreshing it, despite its comment at
:10652. On a Normal primary the cache tag goes stale after every append (the
refresh at :7993 is gated on status != Normal), so impls.rs:2538-2546 returns
an empty suffix.
The stale old primary that receives it reconciles with `canonical = []`, so
`first_unproven_op` stops at op_head on the first step and truncates every held
op above its locally proven commit, acked ones included.
Neither caller (:5076-5089, :5122-5132) refreshes, unlike the view-change
paths (:4692/:4734). Fix: refresh the metadata/partition suffix before building
this StartView, plus the gap fix.
##########
core/shard/src/lib.rs:
##########
@@ -4511,17 +4609,38 @@ where
}
}
+ fn partition_incarnation_matches(
+ &self,
+ partition: &IggyPartition<B, SB>,
+ partition_incarnation: u64,
+ frame: &'static str,
+ ) -> bool {
+ if partition_incarnation == partition.created_revision() {
Review Comment:
nit: frames with incarnation 0, i.e. every
SVC/DVC/SV/Commit/RequestPrepares/repair/state-transfer frame from a pre-PR
replica, are dropped here, so a mixed-version partition view change cannot
complete. Fine if rolling upgrades aren't supported (the boot format gate
already refuses old data dirs), but please state that explicitly in the upgrade
notes.
##########
core/simulator/src/workload/oracle.rs:
##########
@@ -449,14 +500,15 @@ pub fn assert_converged(sim: &Simulator, workload: &mut
Workload) -> Convergence
"committed metadata prefixes agree across every live replica"
);
- let Some(leader) = metadata_leader(sim, &live) else {
+ let Some(leader) = sim.metadata_leader().map(usize::from) else {
panic!("no metadata leader live at quiesce (seed={seed:#x})");
};
// Settlement treats "no live replica hosts this group" as settled, so
losing
// every instance of a LIVE partition reads as converged. The metadata read
// separates that from a group whose stream the workload deleted.
let namespaces_checked = assert_live_namespaces_have_a_primary(sim,
workload, leader, seed);
Review Comment:
nit (pre-existing, not from this PR; the crash campaigns surfaced them and
they're worth their own issues):
1. impls.rs:4180 `RebuildPipeline` starts at `merged.commit_max + 1` even
when the parked-view repair walk already applied that op, so the metadata
pipeline head sits below commit until the next view (invariants.rs:291
AppliedHead; seeds lifecycle 1191/383, strict 536; base seed 5187 too). Fix:
start from `max(commit_min, merged.commit_max) + 1`.
2. shard/lib.rs:4527: a partition created while 2/3 replicas are down
rebuilds with elections forbidden wherever there is no retained state, so it
never elects and its Delete* never completes (#4476 family). Seed 24 `--plane
uniform --shell --clients 2 --journal-slots 96 --faults light --crash-prob
0.003 --restart-prob 0.03 --crash-primary --ticks 30000 --heal-before-quiesce`.
3. Metadata-plane forks under crash-primary swarm exist on base too (base
fails 56/93, 13 "committed different history"), so the plane_helpers metadata
twin can't be attributed by sim alone.
##########
core/server/src/http/forward.rs:
##########
@@ -332,92 +536,61 @@ 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 (mut parts, request_body) = request.into_parts();
let Ok(body) = to_bytes(request_body, state.forward.body_limit).await else
{
return error_response(
StatusCode::PAYLOAD_TOO_LARGE,
"payload_too_large",
"request body exceeds http.max_request_size",
);
};
- let method = parts.method.clone();
- let request_headers = parts.headers.clone();
- let path_and_query = parts
- .uri
- .path_and_query()
- .map_or("/", |path_and_query| path_and_query.as_str())
- .to_owned();
- let local = next
- .run(Request::from_parts(parts, Body::from(body.clone())))
- .await;
- if let AttemptOutcome::Relay(response) =
classify_local_partition_reply(local).await {
- return response;
- }
-
- let Some(guard) = ForwardGuard::admit(&state.forward.in_flight) else {
- return with_retry_after(error_response(
- StatusCode::SERVICE_UNAVAILABLE,
- "forward_busy",
- "node is at its forward budget; retry with backoff",
- ));
- };
- let self_id = state
- .shard
- .plane
- .metadata()
- .consensus
- .as_ref()
- .map(consensus::VsrConsensus::replica);
- let deadline = Instant::now() + FORWARD_RETRY_DEADLINE;
- let mut skip_node = self_id;
- loop {
- for socket in partition_http_sockets(&state.roster, skip_node) {
- if Instant::now() >= deadline {
- break;
- }
- let url = format!("{}://{socket}{path_and_query}",
state.forward.scheme);
- match attempt(&state, &method, &request_headers, &body, &url,
false).await {
- AttemptOutcome::Relay(response) => {
- return if method == Method::GET &&
response.status().is_success() {
- retain_forward_guard(response, guard,
FORWARD_ATTEMPT_TIMEOUT)
- } else {
- response
- };
- }
- AttemptOutcome::Retry => {}
- }
- }
- let remaining = deadline.saturating_duration_since(Instant::now());
- if remaining <= FORWARD_RETRY_INTERVAL {
- break;
+ let deadline = received
+ + match hop {
+ PartitionHop::Target(route) =>
route.budget.min(PARTITION_REQUEST_DEADLINE),
+ PartitionHop::Contacted | PartitionHop::Unrouted =>
PARTITION_REQUEST_DEADLINE,
+ };
+ // The contacted node chose the partition from metadata at this op. Before
+ // that op is applied here, the topic or the partition may not exist yet,
+ // and the handler would answer 404 for a request it can serve.
+ if let PartitionHop::Target(route) = hop {
+ let frontier = state.shard.plane.metadata().applied_frontier();
+ let budget =
compio::time::sleep(deadline.saturating_duration_since(Instant::now()));
+ if hold_for_frontier(frontier, route.context.metadata_op, budget)
Review Comment:
nit: `iggy-forwarded` + `iggy-partition-route` come from the request, so any
authenticated client can make this node hold up to the deadline on an arbitrary
`metadata_op`, outside the `forward.in_flight` cap. Bounded, but a cheap
slow-request lever. Admit Target holds against the in-flight cap.
##########
core/simulator/src/workload/shadow.rs:
##########
@@ -94,7 +315,327 @@ impl Shadow {
consumer_offsets: HashMap::new(),
id_permutation,
next_index: 1,
+ fixtures,
+ fixture_deletes_open: true,
+ deleted_namespaces: Vec::new(),
+ captured_owners: BTreeMap::new(),
+ }
+ }
+
+ pub(crate) fn observe_owner(
+ &mut self,
+ namespace: IggyNamespace,
+ owner: ConsumerGroupOwner,
+ context: PartitionContext,
+ ) {
+ self.captured_owners.retain(|&(ns, _), observed| {
+ ns != namespace || observed.captured.context.incarnation ==
context.incarnation
+ });
+ for (&(ns, _), observed) in &mut self.captured_owners {
+ if ns == namespace {
+ observed.current = owner;
+ }
+ }
+ if !owner.is_unassigned() {
+ self.captured_owners
+ .entry((namespace, owner.client_id))
+ .or_insert(ObservedOwner {
+ captured: CapturedOwner {
+ namespace,
+ owner,
+ context,
+ },
+ current: owner,
+ });
+ }
+ }
+
+ pub(crate) fn pick_stale_owner(&self, prng: &mut Xoshiro256PlusPlus) ->
Option<CapturedOwner> {
+ let stale = || {
+ self.captured_owners.values().filter(|observed| {
+ observed.current.generation >
observed.captured.owner.generation
+ &&
self.namespaces_live.contains(&observed.captured.namespace)
+ })
+ };
+ let count = stale().count();
+ if count == 0 {
+ return None;
+ }
+ stale()
+ .nth(prng.random_range(0..count))
+ .map(|observed| observed.captured)
+ }
+
+ /// Start a toggle of a fixture topic picked uniformly among those that can
+ /// take one. A delete starts a replacement.
+ ///
+ /// After a toggle of unknown fate, the next toggle of its topic is a
probe: the
+ /// same half again, whose reply tells what the first one did.
+ ///
+ /// Topic slabs are reused last-freed-first, so a delete starts only while
every
+ /// fixture topic is live: with one topic away at a time, each recreated
topic
+ /// lands on its own ids.
+ pub fn begin_fixture_toggle(
+ &mut self,
+ prng: &mut Xoshiro256PlusPlus,
+ ) -> Option<(usize, Toggle)> {
+ let deletes_open = self.fixture_deletes_open
+ && self
+ .fixtures
+ .iter()
+ .all(|fixture| fixture.state == FixtureState::Live);
+ let candidates: Vec<usize> = self
+ .fixtures
+ .iter()
+ .enumerate()
+ .filter(|(_, fixture)| fixture.next_toggle(deletes_open).is_some())
+ .map(|(index, _)| index)
+ .collect();
+ if candidates.is_empty() {
+ return None;
+ }
+ let index = candidates[prng.random_range(0..candidates.len())];
+ let fixture = &mut self.fixtures[index];
+ let (state, toggle) = fixture.next_toggle(deletes_open)?;
+ if fixture.state == FixtureState::Live {
+ fixture.replacements.started += 1;
+ }
+ fixture.state = state;
+ Some((index, toggle))
+ }
+
+ /// Whether the toggle in flight on fixture topic `index` is a probe.
+ #[must_use]
+ pub fn fixture_probing(&self, index: usize) -> bool {
+ matches!(
+ self.fixtures[index].state,
+ FixtureState::Deleting { probe: true } | FixtureState::Creating {
probe: true }
+ )
+ }
+
+ /// The seeded stream and topic names of fixture topic `index`.
+ #[must_use]
+ pub fn fixture_names(&self, index: usize) -> (String, String) {
+ let fixture = &self.fixtures[index];
+ (fixture.stream.clone(), fixture.topic.clone())
+ }
+
+ /// Move fixture topic `index` by what its toggle's reply proved.
+ ///
+ /// # Panics
+ /// If the topic has no toggle in flight: every toggle finishes exactly
once.
+ pub fn finish_fixture_toggle(&mut self, index: usize, fate: ToggleFate) {
+ let previous = self.fixtures[index].state;
+ let Some(next) = previous.after(fate) else {
+ panic!(
+ "fixture topic {} finished a toggle with {fate:?} while
{previous:?}",
+ self.fixtures[index].topic,
+ );
+ };
+ self.fixtures[index].state = next;
+ let namespaces = &self.fixtures[index].namespaces;
+ match next {
+ // A refused delete replaced nothing. Counted as a settle instead,
it
+ // would leave a request sampled during it free to see a
replacement.
+ FixtureState::Live => self.fixtures[index].replacements.started -=
1,
+ FixtureState::Deleted if matches!(previous, FixtureState::Deleting
{ .. }) => {
+ self.namespaces_live
+ .retain(|namespace| !namespaces.contains(namespace));
+ self.sends_committed
+ .retain(|namespace, _| !namespaces.contains(namespace));
+ self.consumer_offsets
+ .retain(|(namespace, _, _), _|
!namespaces.contains(namespace));
+ self.captured_owners
+ .retain(|(namespace, _), _|
!namespaces.contains(namespace));
+ self.deleted_namespaces.extend(namespaces);
+ if let Some(group) = &mut self.fixtures[index].group {
+ group.exists = false;
+ group.joined = false;
+ if let ToggleFate::Reached { op } = fate {
+ group.deletes_reached.push(op);
+ }
+ }
+ }
+ FixtureState::Settling { .. } =>
self.namespaces_live.extend(namespaces),
+ _ => {}
+ }
+ }
+
+ /// Settle every fixture topic that `hosted(op, namespaces)` reports as
applied
+ /// at its create op and hosted at the recreated incarnation.
+ pub fn settle_fixtures(&mut self, mut hosted: impl FnMut(u64,
&[IggyNamespace]) -> bool) {
+ for fixture in &mut self.fixtures {
+ if let FixtureState::Settling { op } = fixture.state
+ && hosted(op, &fixture.namespaces)
+ {
+ fixture.state = FixtureState::Live;
+ fixture.replacements.settled += 1;
+ }
+ }
+ }
+
+ /// Take the namespaces of the fixture topics whose delete was reached
since
+ /// the last call. Every replica hosts the deleted incarnation until the
topic
+ /// is recreated.
+ pub fn take_deleted_namespaces(&mut self) -> Vec<IggyNamespace> {
+ std::mem::take(&mut self.deleted_namespaces)
+ }
+
+ /// Start no more fixture deletes, so the toggles left recreate every
topic.
+ pub const fn close_fixture_deletes(&mut self) {
+ self.fixture_deletes_open = false;
+ }
+
+ /// Whether every fixture topic is live with no toggle in flight, and its
+ /// fixture group, if the run has one, was joined since the topic was last
+ /// replaced.
+ #[must_use]
+ pub fn fixtures_restored(&self) -> bool {
+ self.fixtures.iter().all(|fixture| {
+ fixture.state == FixtureState::Live
+ && fixture.group.as_ref().is_none_or(|group| group.joined)
+ })
+ }
+
+ /// Replacement counts of the fixture topic holding `namespace`, zero for
any
+ /// other namespace.
+ #[must_use]
+ pub fn replacements(&self, namespace: IggyNamespace) -> Replacements {
+ self.fixtures
+ .iter()
+ .find(|fixture| fixture.namespaces.contains(&namespace))
+ .map_or_else(Replacements::default, |fixture| fixture.replacements)
+ }
+
+ /// A pending delete can still refuse, so retain committed effects until
its
+ /// success clears them. After a successful delete, late replies belong to
the
+ /// old history until the replacement settles.
+ #[must_use]
+ pub fn effect_is_current(&self, namespace: IggyNamespace, sampled:
Replacements) -> bool {
+ self.fixtures
+ .iter()
+ .find(|fixture| fixture.namespaces.contains(&namespace))
+ .is_none_or(|fixture| {
Review Comment:
nit: the strict offset oracle gives false positives across fixture
replacement. During `Settling` the namespaces are live again (:456), so a
StoreConsumerOffset on the recreated incarnation commits and is acked, but this
returns false and the effect becomes `Effect::None`. At quiesce, oracle.rs:900
panics `left: {(1,2): 0} right: {}`.
12/600 seeds of `--plane lifecycle --shell --clients 1 --ack-quorum-ratio
1.0 --faults swarm --crash-prob 0.002 --restart-prob 0.02 --crash-primary
--ticks 25000 --heal-before-quiesce`: 132, 146, 168, ... The server is correct.
Fix: treat `Settling` as current when `sampled.started ==
fixture.replacements.started`, or gate the strict compare on every fixture
being `Live`.
##########
core/consensus/src/impls.rs:
##########
@@ -3666,6 +3696,9 @@ impl<B: MessageBus, P: Pipeline<Entry = PipelineEntry>>
VsrConsensus<B, P> {
// it: the new primary's numbers are at least as fresh as any probe
// answer.
self.finish_view_probe();
+ let needs_reconciliation = self.view_log_unproven.get()
Review Comment:
nit: if a metadata backup's truncate fails (I/O error) it keeps
`view_log_unproven`, but the same-view early return at :3647-3653 never re-runs
the reconcile, so commit stays fenced (metadata.rs:4227) until the next view
change. Retry the reconcile here when the proof is still unsettled.
##########
core/message_bus/src/transports/quic.rs:
##########
@@ -326,12 +316,205 @@ fn reply_route(frame: &BusMessage) -> ReplyRoute {
if !is_reply {
return ReplyRoute::Unkeyed;
}
+ read_u64(bytes, offset_of!(ReplyHeader,
request)).map_or(ReplyRoute::Unkeyed, ReplyRoute::Keyed)
+}
+
+/// The request id of a client frame, read as raw bytes like [`reply_route`],
+/// so a frame with an undeclared operation still has one.
+fn request_id(request: &Message<GenericHeader>) -> Option<u64> {
+ read_u64(request.as_slice(), offset_of!(RequestHeader, request))
+}
+
+fn read_u64(bytes: &[u8], offset: usize) -> Option<u64> {
bytes
- .get(..HEADER_SIZE)
- .and_then(|header|
bytemuck::checked::try_from_bytes::<ReplyHeader>(header).ok())
- .map_or(ReplyRoute::Unkeyed, |header| {
- ReplyRoute::Keyed(header.request)
- })
+ .get(offset..offset + size_of::<u64>())
+ .and_then(|field| field.try_into().ok())
+ .map(u64::from_le_bytes)
+}
+
+/// Hands the outbound frames of one connection to the streams that wait
+/// for them.
+///
+/// A stream registers its waiter before it forwards its request, so a reply
+/// never comes before the stream it answers.
+struct ReplyRouter {
+ /// Waiting streams by request id, oldest first.
+ waiters: HashMap<u64, VecDeque<oneshot::Sender<BusMessage>>>,
+ /// Frames that came while no stream waited for them, oldest first.
+ unclaimed: VecDeque<BusMessage>,
+ /// When `unclaimed` holds this many frames, it drops its oldest one.
+ unclaimed_max: usize,
+}
+
+impl ReplyRouter {
+ fn new(unclaimed_max: usize) -> Self {
+ Self {
+ waiters: HashMap::new(),
+ unclaimed: VecDeque::new(),
+ unclaimed_max,
+ }
+ }
+
+ /// Register a stream that waits for the reply to `request_id`.
+ ///
+ /// The stream first takes the oldest unclaimed frame that has no request
+ /// id or has its own. The SDK replays an explicit transient with the same
+ /// request id on a new stream, so a late reply of the first attempt
+ /// answers the replay. The replies for other ids ahead of that frame are
+ /// dropped, because their streams already ended.
+ fn register(&mut self, request_id: u64) -> oneshot::Receiver<BusMessage> {
+ let (waiter, reply) = oneshot::channel();
+ while let Some(frame) = self.unclaimed.pop_front() {
+ if !matches!(reply_route(&frame), ReplyRoute::Keyed(id) if id !=
request_id) {
Review Comment:
nit: an unclaimed unkeyed Eviction is handed to whichever stream registers
next, even one opened after the eviction. It is harmless today because Eviction
is session-terminal. Dropping unclaimed Evictions older than the registering
stream would make that explicit.
##########
core/consensus/src/plane_helpers.rs:
##########
@@ -903,6 +903,62 @@ where
.await
}
+/// Find a held entry below a view's suffix that the new history does not
prove.
+///
+/// `canonical` is ordered by descending op. Entries above the local commit
floor
+/// survive only if their parent chain reaches its lowest header, or a
prepare's
+/// commit stamp proves them through the uninterrupted local chain. This runs
once
+/// per installed log, before repair starts filling any resulting gap.
+#[must_use]
+pub fn first_unproven_op(
+ commit_floor: u64,
+ canonical: &[PrepareHeader],
+ op_head: u64,
+ last_op: Option<u64>,
+ header_at: impl Fn(u64) -> Option<PrepareHeader>,
+) -> Option<u64> {
+ let last_op = last_op?;
+ let proven = locally_proven_commit(commit_floor, &header_at);
+ let (mut op, mut parent) = canonical.last().map_or((op_head, None),
|lowest| {
+ (lowest.op.saturating_sub(1), Some(lowest.parent))
+ });
+ while op > proven {
Review Comment:
**critical**: committed ops are lost. Both this walk and
`locally_proven_commit` (:946) stop at the first op this replica doesn't hold,
so a repair gap counts as divergence. Every held entry in `(proven, gap]` is
returned and `truncate_uncommitted_from` (iggy_partition.rs:8727) cuts it,
including acked entries that link to the canonical head and ops at or below the
merged commit_max.
If a later view-change quorum excludes the remaining holder, the committed
op is gone and its slot is re-prepared with different bytes.
Repro: `workload-fuzz --seed 7016 --ticks 60000 --clients 4 --replicas 3
--plane partition --faults swarm --crash-prob 0.005 --restart-prob 0.02
--crash-primary --require-faults` panics at state_checker.rs:230: "replica 2
and witnesses {0: 6} disagree on committed partition op 93". With only the call
at shard/lib.rs:11646 disabled, the seed passes.
Fix: treat MISSING as unknown. Keep held entries above the first gap but
fence them (no commit, PrepareOk, DVC present-bit or serving), let repair
replace a held entry that doesn't match (`apply_repaired_prepare` skips held
ops at iggy_partition.rs:11600), and re-prove once gaps fill. "Never cut <=
commit_max" alone is unsafe: a dead-branch op next to a gap would be applied.
The metadata path (shard/lib.rs:6486) has the same construction.
##########
core/simulator/src/workload/oracle.rs:
##########
@@ -449,14 +500,15 @@ pub fn assert_converged(sim: &Simulator, workload: &mut
Workload) -> Convergence
"committed metadata prefixes agree across every live replica"
);
- let Some(leader) = metadata_leader(sim, &live) else {
+ let Some(leader) = sim.metadata_leader().map(usize::from) else {
panic!("no metadata leader live at quiesce (seed={seed:#x})");
};
// Settlement treats "no live replica hosts this group" as settled, so
losing
// every instance of a LIVE partition reads as converged. The metadata read
// separates that from a group whose stream the workload deleted.
let namespaces_checked = assert_live_namespaces_have_a_primary(sim,
workload, leader, seed);
+ assert_hosted_incarnations_named(sim, workload, &live, seed);
// Safety direction: no live replica may have COMMITTED more of a group
than
Review Comment:
nit: convergence compares prepare identity chains and offset tables, never
applied segment bytes or messages read back across replicas. Divergence
introduced at apply, rebuild or state-transfer install goes unseen. Separately,
partition state transfer is refused in the sim ("no on-disk directory"), so
crash campaigns report false wedges (seed 1078 `--plane mixed`).
Add a per-namespace read-back digest at quiesce. Until then the PR body's
"Simulator checks cover replacement histories" should be softened.
--
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]
