numinnex commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4260721798
##########
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 {
+ match header_at(op) {
+ Some(local)
+ if local.checksum != CHECKSUM_UNSEALED && Some(local.checksum)
== parent =>
+ {
+ parent = Some(local.parent);
+ op -= 1;
+ }
+ _ => break,
+ }
+ }
+ (proven.checked_add(1)?..=op.min(last_op)).find(|&held|
header_at(held).is_some())
Review Comment:
nit: the `find` walks op by op through the gap with a hash lookup each; on
the partition path an ordered range over the resident index would be cheaper
for a lagging backup. Once per view install, so low priority.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -11452,46 +11727,36 @@ where
.map(|durable| durable.saturating_add(1))
.max(self.installed_frontier);
let committed_shape = journal.repaired_window_shape(floor,
session.commit_to_op);
- let connected = match (session.first_batch_offset, stand_in) {
+ // Read at check time, not collected per frame: the journal keeps
+ // each batch's own stamps, so a batch journaled before this floor
+ // settled anchors it as well as one served after.
+ let first_batch_offset = journal.first_batch_offset(floor,
session.commit_to_op);
+ let connected = match (first_batch_offset, stand_in) {
(Some(first), Some(bound)) => first <= bound,
(Some(first), None) => first == 0,
- // No repaired batch arrived, so there is no offset anchor to
- // verify the floor's continuum claim against. `None` is only
- // safe when the served window itself proves it carried no
- // messages: every op in `(floor, to_op]` journaled and none
- // of them `SendMessages`. Anything less -- dropped frames, or
- // a fully evicted window -- is indistinguishable from a
- // message range below the floor that this replica does not
- // durably own, and accepting it would serve a holed log.
- // A replica with no committed message transfers too: a jump
- // would leave its counter at 0 while the group may have
- // committed messages below the floor. Such a replica then
- // moves its commit point only by walking each op, by a window
- // whose first batch starts at offset 0, or by an install, so
- // its empty offer as primary is real state.
- (None, _) => {
- self.offset_frontier() > 0
- && floor < session.commit_to_op
- && committed_shape.complete
- && !committed_shape.holds_messages
- }
+ // No batch above the floor, so nothing ties the floor to this
+ // replica's offsets. The floor stands only while this journal
+ // lacks an op up to the reported floor, and no shape of the
+ // window above it shows that the ops the jump skips carried no
+ // messages. Refused, the window goes to state transfer.
+ (None, _) => false,
Review Comment:
nit (accepted cost, just flagging): refusing control-only repair windows
turns a lagging backup on an offset-commit-heavy, low-produce partition into a
full state transfer. Rare, and the safety gain is real.
##########
core/message_bus/src/transports/quic.rs:
##########
@@ -326,12 +316,202 @@ 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();
Review Comment:
nit: each request allocates a waiter `VecDeque` entry, a oneshot, a SipHash
map slot and arms a 60s timer. Low impact (high-RPS traffic stays on TCP), but
an `AHashMap` and a connection-level deadline sweep would trim it.
##########
core/simulator/src/workload/oracle.rs:
##########
@@ -666,17 +765,91 @@ fn assert_committed_metadata_agrees(sim: &Simulator,
live: &[usize], seed: u64)
compared
}
-/// The live replica whose metadata consensus is the current primary, i.e. the
-/// authoritative holder of the committed metadata log.
-fn metadata_leader(sim: &Simulator, live: &[usize]) -> Option<usize> {
- live.iter().copied().find(|&replica_idx| {
- sim.replicas[replica_idx].shards[0]
- .plane
- .metadata()
- .consensus
- .as_ref()
- .is_some_and(|consensus| consensus.is_primary() &&
consensus.status() == Status::Normal)
- })
+/// Require the partition primary of every workload namespace to hold each
owner
Review Comment:
nit: the PR description says simulator checks cover replacement histories,
but owner fencing isn't checked: with `matches_consumer_owner` forced to always
accept, fuzz seeds 1-8 stay OK with identical stats and all sim tests pass.
This compares only the primary's owner, and `shadow.consumer_offsets` is never
compared at quiesce. Suggest comparing owners and offsets on every live
replica, and re-running that mutant.
##########
core/integration/tests/server/pipelined_frames_vsr.rs:
##########
@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Frames a client pipelines on one connection, against the server (vsr). A
+//! read held for the metadata frontier runs in its own task, so a frame sent
+//! behind it on the same connection gets its reply first.
+//!
+//! The frames are hand-crafted on a raw TCP socket, because the SDK waits for
+//! each reply before it sends the next frame.
+
+use iggy::prelude::*;
+use iggy_binary_protocol::codes::{GET_ME_CODE, PING_CODE};
+use iggy_binary_protocol::consensus::ReplyHeader;
+use integration::iggy_harness;
+
+use crate::server::raw_tcp::{
+ connect, non_replicated_header, read_reply, register_root, write_frame,
+};
+
+/// The header validator requires a nonzero client id; the value is otherwise
+/// free since nothing here reconnects.
+const CLIENT_ID: u128 = 0x9_1BE;
+
+/// No commit reaches this op, so a read that waits for it is held for its
+/// whole budget and then fails retryable.
+const UNREACHED_OP: u64 = u64::MAX;
+
+const READ_REQUEST: u64 = 1;
+const PING_REQUEST: u64 = 2;
+
+#[iggy_harness]
+async fn
given_a_held_read_when_a_ping_follows_on_one_tcp_connection_should_answer_the_ping_first(
Review Comment:
nit: this is the only pipelining test and it covers PING only. Nothing pins
the per-connection ordering the dispatch change affects (pipelined write then
partition op/read, two pipelined Next+auto-commit polls, poll then store);
tests for those would guard the fixes above.
##########
core/integration/tests/cluster/partition_primary_routing.rs:
##########
@@ -262,6 +443,306 @@ async fn
given_split_primaries_when_http_auto_commits_on_a_backup_should_replica
assert_replicated_offset(harness, &consumer, Some((GROUP_PAYLOADS.len() -
1) as u64)).await;
}
+/// Every partition request an HTTP caller sends to a backup runs on the
+/// partition primary: the backup chooses the partition and relays the request.
+#[iggy_harness(cluster_nodes = 3, server(
+ metadata.journal_slots = "256",
+ http.jwt.encoding_secret = "0123456789abcdef0123456789abcdef",
+ http.jwt.decoding_secret = "0123456789abcdef0123456789abcdef"
+))]
+async fn
given_split_primaries_when_http_calls_a_partition_backup_should_serve_on_the_primary(
+ harness: &mut TestHarness,
+) {
+ let (partition_primary, backup, _) = seed_split_primaries(harness).await;
+ assert_ne!(partition_primary, backup);
+ let http = HttpClient::login_root_no_redirect(node_url(harness,
backup)).await;
+ let messages = http.url(&format!(
+ "/streams/{STREAM_NAME}/topics/{TOPIC_NAME}/messages"
+ ));
+ // The relay copies the partition primary's applied op onto each response.
+ let mut primary_applied_op = 0;
+ for (payload, query, status) in [
+ (RELAYED_PAYLOADS[0], "", StatusCode::CREATED),
+ (RELAYED_PAYLOADS[1], "?ack=none", StatusCode::ACCEPTED),
+ ] {
+ let response = http
+ .client
+ .post(format!("{messages}{query}"))
+ .bearer_auth(&http.token)
+ .json(&send_body(payload))
+ .send()
+ .await
+ .expect("send request");
+ assert_eq!(response.status(), status, "send{query} on a backup");
+ primary_applied_op = primary_applied_op.max(applied_op(&response));
+ }
+
+ let expected: Vec<&str> =
GROUP_PAYLOADS.into_iter().chain(RELAYED_PAYLOADS).collect();
+ let poll = format!(
+
"{messages}?consumer_id={RELAY_CONSUMER_ID}&partition_id={PARTITION_ID}\
+ &kind=offset&value=0&count={}&auto_commit=false",
+ expected.len() + 1
+ );
+ let polled = timeout(GROUP_POLL_BUDGET, async {
+ loop {
+ let response = http
+ .client
+ .get(&poll)
+ .bearer_auth(&http.token)
+ .send()
+ .await
+ .expect("poll request");
+ assert_eq!(response.status(), StatusCode::OK, "poll on a backup");
+ let polled: PolledMessages = response.json().await.expect("decode
PolledMessages");
+ if polled.messages.len() >= expected.len() {
+ break polled;
+ }
+ sleep(PRECONDITION_POLL).await;
+ }
+ })
+ .await
+ .expect("the unacknowledged send must commit on the partition primary");
+ let payloads: Vec<&[u8]> = polled
+ .messages
+ .iter()
+ .map(|message| message.payload.as_ref())
+ .collect();
+ let expected: Vec<&[u8]> = expected.iter().map(|payload|
payload.as_bytes()).collect();
+ assert_eq!(payloads, expected);
+ assert!(
+ polled.context.metadata_op >= primary_applied_op,
+ "the relayed poll context {:?} must cover the primary's applied op
{primary_applied_op}",
+ polled.context
+ );
+
+ let offsets = http.url(&format!(
+ "/streams/{STREAM_NAME}/topics/{TOPIC_NAME}/consumer-offsets"
+ ));
+ let offset =
format!("{offsets}?consumer_id={RELAY_CONSUMER_ID}&partition_id={PARTITION_ID}");
+ let store = StoreConsumerOffset {
+ context: None,
+ consumer:
Consumer::new(Identifier::numeric(RELAY_CONSUMER_ID).unwrap()),
+ partition_id: Some(PARTITION_ID),
+ offset: 1,
+ };
+ let response = http
+ .client
+ .put(&offsets)
+ .bearer_auth(&http.token)
+ .json(&store)
+ .send()
+ .await
+ .expect("store offset request");
+ assert_eq!(
+ response.status(),
+ StatusCode::NO_CONTENT,
+ "store on a backup"
+ );
+ let response = http
+ .client
+ .get(&offset)
+ .bearer_auth(&http.token)
+ .send()
+ .await
+ .expect("get offset request");
+ assert_eq!(response.status(), StatusCode::OK, "get on a backup");
+ let info: ConsumerOffsetInfo = response.json().await.expect("decode
ConsumerOffsetInfo");
+ assert_eq!(info.stored_offset, store.offset);
+ let response = http
+ .client
+ .delete(format!(
+ "{offsets}/{RELAY_CONSUMER_ID}?partition_id={PARTITION_ID}"
+ ))
+ .bearer_auth(&http.token)
+ .send()
+ .await
+ .expect("delete offset request");
+ assert_eq!(
+ response.status(),
+ StatusCode::NO_CONTENT,
+ "delete on a backup"
+ );
+ let response = http
+ .client
+ .get(&offset)
+ .bearer_auth(&http.token)
+ .send()
+ .await
+ .expect("get offset request after delete");
+ assert_eq!(response.status(), StatusCode::NOT_FOUND, "deleted offset");
+}
+
+/// A relayed request carries the partition the contacted node chose. A backup
+/// refuses it with `NotPrimary` instead of relaying it again, a malformed
route
+/// is the caller's error, and a route without the relay marker is ignored.
+#[iggy_harness(cluster_nodes = 3, server(
+ metadata.journal_slots = "256",
+ http.jwt.encoding_secret = "0123456789abcdef0123456789abcdef",
+ http.jwt.decoding_secret = "0123456789abcdef0123456789abcdef"
+))]
+async fn
given_split_primaries_when_a_relayed_request_reaches_a_backup_should_refuse_not_primary(
+ harness: &mut TestHarness,
+) {
+ let (partition_primary, backup, _) = seed_split_primaries(harness).await;
+ assert_ne!(partition_primary, backup);
+ let http = HttpClient::login_root_no_redirect(node_url(harness,
backup)).await;
+ let messages = http.url(&format!(
+ "/streams/{STREAM_NAME}/topics/{TOPIC_NAME}/messages"
+ ));
+ let poll = format!(
+
"{messages}?consumer_id={RELAY_CONSUMER_ID}&partition_id={PARTITION_ID}\
+ &kind=offset&value=0&count=1&auto_commit=false"
+ );
+ let send = |route: &'static str| {
+ http.client
+ .post(&messages)
+ .bearer_auth(&http.token)
+ .header(PARTITION_ROUTE_HEADER, route)
+ .json(&send_body(RELAYED_PAYLOADS[0]))
+ };
+
+ for (request, name) in [
+ (
+ send(ROUTE_AT_FIRST_OP).header(FORWARDED_HEADER, "1"),
+ "send",
+ ),
+ (
+ http.client
+ .get(&poll)
+ .bearer_auth(&http.token)
+ .header(FORWARDED_HEADER, "1")
+ .header(PARTITION_ROUTE_HEADER, ROUTE_AT_FIRST_OP),
+ "poll",
+ ),
+ ] {
+ let response = request.send().await.expect("relayed request");
+ assert_eq!(
+ response.status(),
+ StatusCode::SERVICE_UNAVAILABLE,
+ "relayed {name} on a backup"
+ );
+ assert!(response.headers().contains_key(RETRY_AFTER));
+ let body: serde_json::Value = response.json().await.expect("decode the
error body");
+ assert_eq!(
+ body["id"],
+ IggyError::NotPrimary.as_code(),
+ "relayed {name}"
+ );
+ }
+
+ let response = send("0,1")
+ .header(FORWARDED_HEADER, "1")
+ .send()
+ .await
+ .expect("malformed relayed send");
+ assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+
+ let response = send(ROUTE_AT_FIRST_OP)
+ .send()
+ .await
+ .expect("send with a route and no relay marker");
+ assert_eq!(response.status(), StatusCode::CREATED);
+}
+
+/// A read changes no state, so a backup resends a relayed read whose partition
+/// primary died before it answered, and the caller gets the answer of the next
+/// primary. A poll with `auto_commit` may have moved its offset, so its
outcome
+/// stays unknown.
+#[iggy_harness(cluster_nodes = 3, server(
+ metadata.journal_slots = "256",
+ http.jwt.encoding_secret = "0123456789abcdef0123456789abcdef",
+ http.jwt.decoding_secret = "0123456789abcdef0123456789abcdef"
+))]
+async fn
given_split_primaries_when_a_relayed_read_loses_its_primary_should_resend_it(
Review Comment:
nit, flake risk: `poll(false)` asserts 200 with no 503 tolerance, but
failover (5s heartbeat timeout + view changes) must fit the 10s relay deadline,
so under contention it can 503. The auto-commit 502 expectation also assumes
the relay was sent within 1s; if it connects after the kill, the retry gets
200. Tolerating a retryable 503 like the offset read, and killing only once the
relay is observed in flight, would make it robust.
##########
core/simulator/src/workload/mod.rs:
##########
@@ -621,12 +758,34 @@ impl Workload {
// Effect-follows-actual: drive the shadow off the committed outcome,
never
// the targeted one. Success mutates; a nonzero code is a committed
no-op
// whose `predicted_effect` is `Effect::None`. Keeps the shadow correct
- // under at-least-once re-execution and races.
- let effect = ops::predicted_effect(&entry.input, &classified);
+ // under at-least-once re-execution and races. A partition-plane
success
+ // sampled across a fixture replacement can belong to the deleted
+ // incarnation, whose state the shadow dropped with it.
+ let current = header.operation.is_metadata()
+ || self
+ .shadow
Review Comment:
nit: a send sampled Live {s,s} that replies while a delete is pending fails
`unchanged_since` and is recorded as no effect; if the delete is then refused,
`started` goes back to s and a committed send is missing from the shadow.
Weakens detection only. Also, the workload has no poll action and the shell sim
doesn't run consumer liveness (`replica.rs:407` TODO), which is why the
RetireSession stall and the duplicate-delivery race aren't reachable here.
##########
core/integration/tests/cluster/partition_primary_routing.rs:
##########
@@ -262,6 +443,306 @@ async fn
given_split_primaries_when_http_auto_commits_on_a_backup_should_replica
assert_replicated_offset(harness, &consumer, Some((GROUP_PAYLOADS.len() -
1) as u64)).await;
}
+/// Every partition request an HTTP caller sends to a backup runs on the
+/// partition primary: the backup chooses the partition and relays the request.
+#[iggy_harness(cluster_nodes = 3, server(
+ metadata.journal_slots = "256",
+ http.jwt.encoding_secret = "0123456789abcdef0123456789abcdef",
+ http.jwt.decoding_secret = "0123456789abcdef0123456789abcdef"
+))]
+async fn
given_split_primaries_when_http_calls_a_partition_backup_should_serve_on_the_primary(
Review Comment:
nit: only the writes prove the relay here; a poll or offset GET returns the
same bytes whether a backup or the primary serves it, and the `metadata_op`
check also passes on a backup (shared metadata plane). A served-by signal on
reads (e.g. a debug header naming the serving replica) would let this catch a
backup-served read.
##########
core/message_bus/src/transports/quic.rs:
##########
@@ -375,9 +558,13 @@ impl QuicTransportConn {
}
impl TransportConn for QuicTransportConn {
- #[allow(clippy::future_not_send, clippy::too_many_lines)]
+ #[allow(clippy::future_not_send)]
async fn run(self, ctx: ActorContext) {
- let connection = self.connection;
+ let Self {
+ connection,
+ reply_backstop,
+ ..
Review Comment:
nit: `close_grace` is discarded by the `..` here, so shutdown closes
immediately and drops every in-flight `serve` future whose request already went
through `in_tx`. With 256 streams per connection that is up to 256 unknown
outcomes per connection on a graceful restart (dedup covers retries). The
module doc promises a drain bounded by `close_grace`; either drain (while
`in_tx` is alive) or fix the doc.
##########
core/simulator/src/workload/shadow.rs:
##########
@@ -36,6 +37,181 @@ use server_common::sharding::IggyNamespace;
use crate::workload::effect::{ApplyResult, Effect};
use crate::workload::ids::IdPermutation;
+/// Name of the consumer group a shell run keeps on every fixture topic.
Outside
+/// the `wl-` prefix, so the entity oracle leaves it out.
+pub const FIXTURE_GROUP_NAME: &str = "sim-fixture-group";
+
+/// Id of the fixture group: the first group of its topic, and every topic
+/// incarnation numbers its groups from 0.
+pub const FIXTURE_GROUP_ID: u32 = 0;
+
+/// The half of a fixture replacement one request performs.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Toggle {
+ Delete,
+ Create { partitions_count: u32 },
+}
+
+/// A sampled join of a fixture group.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct GroupJoin {
+ /// Index of the fixture topic in the shadow.
+ pub fixture: usize,
+ /// Create the group that a replacement dropped with its topic, instead of
+ /// joining it.
+ pub create: bool,
+ /// Replacements of the topic started at sample time. A delete drops the
+ /// group with its topic, so a reply proves the current group only if no
+ /// replacement started since.
+ pub started: u64,
+}
+
+/// What the reply to a fixture toggle or a fixture group join proves.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ToggleFate {
+ /// The request reached its goal at metadata op `op`: the topic is gone or
+ /// present, the client is a member, or the group exists.
+ Reached { op: u64 },
+ /// The request did not take effect.
+ Refused,
+ /// The request may have taken effect.
+ Unknown,
+}
+
+/// Replacement counts of one fixture topic, shared by its namespaces.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct Replacements {
+ /// Deletes sampled, less those refused before they reached the server.
+ pub started: u64,
+ /// Replacements over: every live replica hosts the recreated incarnation.
+ pub settled: u64,
+}
+
+impl Replacements {
+ /// Whether a request that recorded `sampled` may find its history
replaced.
+ #[must_use]
+ pub const fn may_replace(self, sampled: Self) -> bool {
+ self.started > sampled.settled
+ }
+
+ /// Whether a success of a request that recorded `sampled` belongs to the
+ /// current incarnation. A delete replies before its fences install, so the
+ /// success of an earlier request can belong to the deleted one.
+ #[must_use]
+ pub const fn unchanged_since(self, sampled: Self) -> bool {
+ sampled.started == sampled.settled && sampled.started == self.started
+ }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum FixtureState {
+ Live,
+ /// A delete is in flight. A probe asks again after a delete of unknown
fate, so
+ /// its refusal proves nothing.
+ Deleting {
+ probe: bool,
+ },
+ DeleteUnknown,
+ Deleted,
+ Creating {
+ probe: bool,
+ },
+ CreateUnknown,
+ /// The recreated topic exists since metadata op `op`. A replica may still
+ /// host the fenced incarnation, so a request can still meet it.
+ Settling {
+ op: u64,
+ },
+}
+
+impl FixtureState {
+ /// The state a reply moves an in-flight toggle to, or `None` with no
toggle in
+ /// flight. A probe resolves only on `Reached`: its delete refused under
the
+ /// pending one, or its create refused, says nothing about the first
toggle.
+ const fn after(self, fate: ToggleFate) -> Option<Self> {
+ Some(match (self, fate) {
+ (Self::Deleting { .. }, ToggleFate::Reached { .. })
+ | (Self::Creating { probe: false }, ToggleFate::Refused) =>
Self::Deleted,
+ (Self::Deleting { probe: false }, ToggleFate::Refused) =>
Self::Live,
+ (Self::Deleting { .. }, _) => Self::DeleteUnknown,
+ (Self::Creating { .. }, ToggleFate::Reached { op }) =>
Self::Settling { op },
+ (Self::Creating { .. }, _) => Self::CreateUnknown,
+ _ => return None,
+ })
+ }
+}
+
+/// The fixture group of a topic, which a shell run joins and leaves.
+#[derive(Debug, Default)]
+struct FixtureGroup {
+ /// Whether the group exists in the current incarnation of its topic. A
topic
+ /// delete drops it, and the next join creates it again.
+ exists: bool,
+ /// Whether a join committed on the group of the current incarnation, which
+ /// gives the group the assignments that the owner oracle compares.
+ joined: bool,
+ /// Each client whose join may have committed, with the metadata op of its
+ /// newest such join. `None` when the fate of a join is unknown, which can
+ /// still commit on any later incarnation.
Review Comment:
nit: `joined_at` is never cleared, so `may_have_joined` accepts a member
that left after its successor installed, and `assert_group_writer_joined`
(`workload/mod.rs:850`) passes exactly the stale-owner store this PR fences.
Bounding it by the successor's install op would close that.
--
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]
