fix(cluster): serve metadata reads at or above the client's own writes

bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java

@@ -142,8 +142,16 @@ public void deleteStreamByNumericId() {
 
     @Then("getting the stream by its numeric ID should return no stream")
     public void getStreamReturnsNoStream() {
+        // The assertion is "not the stream we deleted", not "nothing at this id":
+        // the server hands out the lowest free stream id, so once these scenarios
+        // run concurrently against one server a fresh create can legitimately
+        // occupy the deleted stream's id. Named, so a missing pre-value fails
+        // here instead of making the comparison below vacuously true.
+        assertNotNull(context.lastStreamName, "Stream should have been created");
         Optional<StreamDetails> stream = getClient().streams().getStream(context.lastStreamId);
-        assertTrue(stream.isEmpty(), "Deleted stream should not be returned");
+        assertTrue(
+                stream.isEmpty() || !stream.get().name().equals(context.lastStreamName),
+                "Deleted stream should not be returned");
     }
 
     @When("I create a topic with name {string} in stream {int} with {int} partitions")

bdd/rust/tests/steps/streams.rs

@@ -18,11 +18,6 @@
 use crate::common::global_context::GlobalContext;
 use cucumber::{given, then, when};
 use iggy::prelude::{Identifier, StreamClient, StreamUpdateOptions};
-use std::time::Duration;
-use tokio::time::{Instant, sleep};
-
-const METADATA_CONVERGENCE_TIMEOUT: Duration = Duration::from_secs(2);
-const METADATA_CONVERGENCE_POLL: Duration = Duration::from_millis(10);
 
 #[given("I have no streams in the system")]
 pub async fn given_no_streams(world: &mut GlobalContext) {
@@ -128,18 +123,20 @@ pub async fn when_delete_stream_by_numeric_id(world: &mut GlobalContext) {
 
 #[then("getting the stream by its numeric ID should return no stream")]
 pub async fn then_get_stream_returns_no_stream(world: &mut GlobalContext) {
-    let deadline = Instant::now() + METADATA_CONVERGENCE_TIMEOUT;
-    loop {
-        get_stream_by_numeric_id(world).await;
-        if world.last_stream_name.is_none() {
-            return;
-        }
-        assert!(
-            Instant::now() < deadline,
-            "Deleted stream should not be returned after {METADATA_CONVERGENCE_TIMEOUT:?}"
-        );
-        sleep(METADATA_CONVERGENCE_POLL).await;
-    }
+    // Read before the get overwrites it. The assertion is "not the stream we
+    // deleted", not "nothing at this id": `IdSlab::insert` hands out the lowest
+    // free key and these scenarios share one server, so a concurrent create can
+    // legitimately occupy the deleted stream's id.
+    let deleted = world
+        .last_stream_name
+        .clone()
+        .expect("Stream should have been created");
+    get_stream_by_numeric_id(world).await;
+    assert_ne!(
+        world.last_stream_name.as_ref(),
+        Some(&deleted),
+        "Deleted stream should not be returned"
+    );
 }
 
 async fn create_stream(world: &mut GlobalContext, stream_name: &str) {

core/integration/tests/server/http_client.rs

@@ -16,16 +16,20 @@
 // under the License.
 
 //! Shared HTTP transport plumbing for the server REST suites (`http_vsr`,
-//! `http_rbac`): one authenticated `reqwest` session with the login-retry gate
-//! and the generic verb helpers. Each suite keeps its own request shapes and
-//! assertions as extension methods on [`HttpClient`], so the wire-contract and
-//! listener-behavior separation between the suites stays intact.
+//! `http_rbac`): one authenticated `reqwest` session with the login-retry gate,
+//! the generic verb helpers, and the cluster-shaped helpers the multi-node
+//! suites share (which node is the leader, which is a follower, and the retry
+//! a follower needs before it can resolve the primary). Each suite keeps its
+//! own request shapes and assertions as extension methods on [`HttpClient`], so
+//! the wire-contract and listener-behavior separation between the suites stays
+//! intact.
 
+use std::future::Future;
 use std::time::{Duration, Instant};
 
 use iggy::prelude::*;
 use integration::harness::TestHarness;
-use reqwest::Response;
+use reqwest::{Response, StatusCode};
 use serde_json::{Value, json};
 use tokio::time::sleep;
 
@@ -196,6 +200,67 @@ impl HttpClient {
     }
 }
 
+/// `http://host:port` of a harness node's HTTP listener.
+pub fn node_url(harness: &TestHarness, node: usize) -> String {
+    let addr = harness.node(node).http_addr().expect("node http address");
+    format!("http://{addr}")
+}
+
+/// Harness indexes of the node the roster marks `Leader` and of one it marks
+/// `Follower`. The harness emits the roster in node order, so a roster
+/// position is a harness index. Every node reads `Follower` until shard 0
+/// publishes its first view, so the roster is polled within the shared
+/// warmup budget until it marks a leader.
+pub async fn leader_and_follower(harness: &TestHarness) -> (usize, usize) {
+    let client = harness
+        .root_client_for_node(0)
+        .await
+        .expect("connect to node 0");
+    let deadline = Instant::now() + LOGIN_TIMEOUT;
+    loop {
+        let metadata = client
+            .get_cluster_metadata()
+            .await
+            .expect("get cluster metadata");
+        let position =
+            |role: ClusterNodeRole| metadata.nodes.iter().position(|node| node.role == role);
+        if let (Some(leader), Some(follower)) = (
+            position(ClusterNodeRole::Leader),
+            position(ClusterNodeRole::Follower),
+        ) {
+            return (leader, follower);
+        }
+        assert!(
+            Instant::now() < deadline,
+            "the roster did not mark a leader within {LOGIN_TIMEOUT:?}, got {metadata}"
+        );
+        sleep(LOGIN_RETRY_INTERVAL).await;
+    }
+}
+
+/// Repeat `request` while the follower answers 503, which it does until it
+/// can resolve the primary from its own view; bounded by the shared warmup
+/// budget, as cluster_metadata_vsr does. A 503 is the retry-safe class: the
+/// request provably never entered a pipeline.
+pub async fn until_primary_resolved<F, Fut>(request: F) -> Response
+where
+    F: Fn() -> Fut,
+    Fut: Future<Output = Response>,
+{
+    let deadline = Instant::now() + LOGIN_TIMEOUT;
+    loop {
+        let response = request().await;
+        if response.status() != StatusCode::SERVICE_UNAVAILABLE {
+            return response;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "follower did not resolve the primary within {LOGIN_TIMEOUT:?}"
+        );
+        sleep(LOGIN_RETRY_INTERVAL).await;
+    }
+}
+
 /// Extract the JWT from a successful login response.
 pub async fn access_token(response: Response) -> String {
     let identity: IdentityInfo = response.json().await.expect("decode IdentityInfo");

core/integration/tests/server/http_read_your_writes.rs

@@ -0,0 +1,219 @@
+// 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.
+
+//! Read-your-writes over the REST listener, end to end: an unqualified read
+//! must never answer below the metadata op the same caller was already told
+//! committed.
+//!
+//! The window is a node that HANDED OUT a committed op it has not applied yet.
+//! On the REST plane that node is a follower running a `Register`: the session
+//! its first authenticated request mints is forwarded to the primary
+//! (`dispatch::submit_register_local_or_forward`), so the follower answers with
+//! an epoch its own commit walk can still be behind. Everything the metadata
+//! group committed below that epoch is therefore state the caller has been
+//! promised and this follower may not have applied.
+//!
+//! Each round seeds a stream through the primary over TCP, then authenticates
+//! on the follower, which binds an epoch above that create. The follower's next
+//! read is the assertion: it must not answer from before the stream existed.
+//! Logout is the authenticated request that binds it, deliberately - it tears
+//! the session entry down again, so the floor the read waits on has to outlive
+//! the session that established it.
+//!
+//! The seeding runs over TCP rather than the primary's own REST listener so the
+//! only HTTP sessions in play are the follower's: a second long-lived REST
+//! session would be competing for VSR client ids with the fresh register each
+//! round mints, which is a different subject.
+//!
+//! What each case can and cannot prove, stated plainly:
+//!
+//! - The wiring is proved deterministically. The forwarding case reads the
+//!   serving primary's applied op off the relayed write's `iggy-applied-op`
+//!   header and then off the follower's OWN answer to the following read: the
+//!   second can only be at or above the first if this node held that read until
+//!   its commit walk covered the op the primary handed the caller. That is the
+//!   invariant itself, in op numbers, not a proxy for it.
+//! - Whether the gate actually PARKED is invisible from outside, and on a fast
+//!   local cluster the follower often applies within the same tick, so neither
+//!   case can force the park. The park, the wake and the expiry are pinned
+//!   deterministically next to the gate instead - `dispatch::reads`,
+//!   `metadata::applied_frontier`, and the simulator's
+//!   `metadata_read_frontier_tests`, which cuts replication to force the lag
+//!   the binary plane sees.
+//! - A stale answer still fails loudly if the race does land: a 404, a list
+//!   missing the stream, or an applied op below the one the caller was handed.
+//!   A 503 fails too, on purpose - that is what the gate answers when the
+//!   follower never catches up inside its budget.
+
+use iggy::prelude::*;
+use integration::iggy_harness;
+use reqwest::{Response, StatusCode};
+use serde_json::{Value, json};
+
+use crate::server::http_client::{
+    HttpClient, leader_and_follower, node_url, until_primary_resolved,
+};
+
+/// Seed / read-back rounds. More than one because the lag is a race the test
+/// cannot force: each round re-runs it with the follower's commit walk in a
+/// different position relative to the epoch it just handed out.
+const ROUNDS: u32 = 4;
+
+/// The serving node's applied metadata op, as stamped on every authenticated
+/// success. A response without it is a contract violation, not an absence: the
+/// relay records this value as the caller's read-your-writes floor, so a
+/// missing header would silently reopen the stale read it exists to close.
+fn applied_op(response: &Response) -> u64 {
+    response
+        .headers()
+        .get("iggy-applied-op")
+        .expect("every authenticated success carries iggy-applied-op")
+        .to_str()
+        .expect("iggy-applied-op must be ASCII")
+        .parse()
+        .expect("iggy-applied-op must be an op number")
+}
+
+/// The `name` of every stream in a `GET /streams` list body.
+fn stream_names(body: &Value) -> Vec<String> {
+    body.as_array()
+        .expect("the stream list is a JSON array")
+        .iter()
+        .map(|stream| {
+            stream["name"]
+                .as_str()
+                .expect("every stream carries a name")
+                .to_owned()
+        })
+        .collect()
+}
+
+/// Three nodes, the smallest cluster with a quorum, one shard each so every
+/// request is served by shard 0 where the metadata consensus lives. No
+/// `http.jwt` secret and no `cluster.auth`: bearers are node-local and
+/// follower-to-primary forwarding is off, so the follower answers its own
+/// requests instead of relaying them (see `http_view_header`, which pins both
+/// halves of that switch).
+#[iggy_harness(cluster_nodes = 3, server(system.sharding.cpu_allocation = "0..1"))]
+async fn given_a_follower_when_its_register_binds_a_committed_epoch_should_not_read_below_it(
+    harness: &TestHarness,
+) {
+    let (leader, follower) = leader_and_follower(harness).await;
+    let seeder = harness
+        .root_client_for_node(leader)
+        .await
+        .expect("connect to the primary");
+
+    for round in 0..ROUNDS {
+        let stream = format!("read-your-writes-{round}");
+        seeder
+            .create_stream(&stream)
+            .await
+            .expect("the primary must commit the stream this round reads back");
+
+        // Fresh bearer, then one authenticated request on the follower: it is
+        // what forwards the `Register` and binds an epoch above the create.
+        let http = HttpClient::login_root_no_redirect(node_url(harness, follower)).await;
+        let logout = until_primary_resolved(|| http.delete("/users/logout")).await;
+        assert_eq!(
+            logout.status(),
+            StatusCode::NO_CONTENT,
+            "the follower must bind and end a forwarded session"
+        );
+
+        // The list read resolves nothing, so it cannot 404 its way into looking
+        // correct: a stale answer here is a short list.
+        let read = http.get("/streams").await;
+        assert_eq!(read.status(), StatusCode::OK, "the stream list must serve");
+        let names = stream_names(&read.json().await.expect("the stream list is JSON"));
+        assert!(
+            names.contains(&stream),
+            "the follower listed streams from before the epoch it had just handed out: {names:?}"
+        );
+
+        // The entity read, where the stale answer is a 404 instead.
+        let read = http.get(&format!("/streams/{stream}")).await;
+        assert_eq!(
+            read.status(),
+            StatusCode::OK,
+            "the follower answered a read below the epoch it had just handed out"
+        );
+        let body: Value = read.json().await.expect("stream details are JSON");
+        assert_eq!(
+            body["name"].as_str(),
+            Some(stream.as_str()),
+            "the read answered with another stream's state"
+        );
+    }
+}
+
+/// The same three nodes WITH cluster-wide bearer key material, which switches
+/// follower-to-primary forwarding on. That is the other half of the window and
+/// the one with a deterministic proof: the write is relayed, so this follower
+/// never runs the local write path and learns what the caller was told
+/// committed only from the primary's `iggy-applied-op`. Its own answer to the
+/// read that follows carries its own applied op, which must be at or above it.
+#[iggy_harness(
+    cluster_nodes = 3,
+    server(
+        system.sharding.cpu_allocation = "0..1",
+        http.jwt.encoding_secret = "0123456789abcdef0123456789abcdef",
+        http.jwt.decoding_secret = "0123456789abcdef0123456789abcdef"
+    )
+)]
+async fn given_a_forwarding_follower_when_it_relays_a_write_should_not_read_below_the_primary_op(
+    harness: &TestHarness,
+) {
+    let (_leader, follower) = leader_and_follower(harness).await;
+    let http = HttpClient::login_root_no_redirect(node_url(harness, follower)).await;
+
+    for round in 0..ROUNDS {
+        let stream = format!("relayed-read-your-writes-{round}");
+        let create = json!({ "name": stream });
+        let created = until_primary_resolved(|| http.post_json("/streams", &create)).await;
+        assert_eq!(
+            created.status(),
+            StatusCode::OK,
+            "the follower must relay the write and answer the primary's reply"
+        );
+        // The primary applies a metadata op before it replies, so this is at or
+        // above the op the caller now holds.
+        let committed_at = applied_op(&created);
+        assert!(
+            committed_at > 0,
+            "the relayed reply carried no applied op, so this node recorded no floor"
+        );
+
+        let read = http.get("/streams").await;
+        assert_eq!(
+            read.status(),
+            StatusCode::OK,
+            "the follower must serve the read rather than refuse it"
+        );
+        assert!(
+            applied_op(&read) >= committed_at,
+            "the follower answered a read at op {} after handing the caller {committed_at}: \
+             the relayed floor was not recorded or not waited for",
+            applied_op(&read)
+        );
+        let names = stream_names(&read.json().await.expect("the stream list is JSON"));
+        assert!(
+            names.contains(&stream),
+            "the follower listed streams without the one it had just relayed: {names:?}"
+        );
+    }
+}