diff --git a/benchmarks/minimal/src/store.rs b/benchmarks/minimal/src/store.rs index c1ba4acfe..68c08b3c7 100644 --- a/benchmarks/minimal/src/store.rs +++ b/benchmarks/minimal/src/store.rs @@ -314,6 +314,7 @@ impl RaftStateMachine for Arc { while let Some((entry, responder)) = entries.try_next().await? { sm.last_applied_log = Some(entry.log_id); + let is_normal = matches!(&entry.payload, EntryPayload::Normal(_)); match entry.payload { EntryPayload::Blank | EntryPayload::Normal(_) => {} EntryPayload::Membership(ref mem) => { @@ -322,7 +323,7 @@ impl RaftStateMachine for Arc { }; if let Some(responder) = responder { - responder.send(ClientResponse {}); + responder.send(is_normal.then_some(ClientResponse {})); } } Ok(()) diff --git a/examples/raft-kv-log-wal-sm-mem/tests/cluster/test_cluster.rs b/examples/raft-kv-log-wal-sm-mem/tests/cluster/test_cluster.rs index 4b59e7724..2855c8759 100644 --- a/examples/raft-kv-log-wal-sm-mem/tests/cluster/test_cluster.rs +++ b/examples/raft-kv-log-wal-sm-mem/tests/cluster/test_cluster.rs @@ -111,7 +111,7 @@ async fn test_cluster_inner() -> Result<(), Box Result<(), Box> // next open needs. raft.shutdown().await?; - resp.data + resp.data.unwrap() }; tracing::info!("reopen the same WAL directory with a fresh state machine"); diff --git a/examples/raft-kv-memstore-grpc/src/grpc/app_service.rs b/examples/raft-kv-memstore-grpc/src/grpc/app_service.rs index 24f3039ba..123dfa007 100644 --- a/examples/raft-kv-memstore-grpc/src/grpc/app_service.rs +++ b/examples/raft-kv-memstore-grpc/src/grpc/app_service.rs @@ -62,7 +62,10 @@ impl AppService for AppServiceImpl { let res = self.raft_node.client_write(req.clone()).await; match res.decompose() { - Ok(Ok(resp)) => Ok(Response::new(resp.data)), + Ok(Ok(resp)) => { + let data = resp.data.ok_or_else(|| Status::internal("normal write returned no data"))?; + Ok(Response::new(data)) + } Ok(Err(ClientWriteError::ForwardToLeader(forward))) => Err(status_forward_to_leader(&forward)), Ok(Err(e)) => Err(Status::internal(e.to_string())), Err(fatal) => Err(Status::internal(fatal.to_string())), diff --git a/examples/raft-kv-memstore-grpc/src/pb_impl/impl_client_write_response.rs b/examples/raft-kv-memstore-grpc/src/pb_impl/impl_client_write_response.rs index 1e8995e9c..610f068a8 100644 --- a/examples/raft-kv-memstore-grpc/src/pb_impl/impl_client_write_response.rs +++ b/examples/raft-kv-memstore-grpc/src/pb_impl/impl_client_write_response.rs @@ -5,7 +5,7 @@ impl From for ClientWriteResponse { fn from(r: pb::ClientWriteResponse) -> Self { ClientWriteResponse { log_id: r.log_id.unwrap().into(), - data: r.data.unwrap(), + data: r.data, membership: r.membership.map(|mem| mem.into()), } } @@ -15,7 +15,7 @@ impl From for pb::ClientWriteResponse { fn from(r: ClientWriteResponse) -> Self { pb::ClientWriteResponse { log_id: Some(r.log_id.into()), - data: Some(r.data), + data: r.data, membership: r.membership.map(|mem| mem.into()), } } diff --git a/examples/raft-kv-memstore-grpc/src/store/mod.rs b/examples/raft-kv-memstore-grpc/src/store/mod.rs index 41ebcea90..6d3e96317 100644 --- a/examples/raft-kv-memstore-grpc/src/store/mod.rs +++ b/examples/raft-kv-memstore-grpc/src/store/mod.rs @@ -108,9 +108,9 @@ impl RaftStateMachine for Arc { sm.last_applied = Some(log_id.into()); - let value = if let Some(req) = entry.app_data { + let response = if let Some(req) = entry.app_data { sm.data.insert(req.key, req.value.clone()); - Some(req.value) + Some(Response { value: Some(req.value) }) } else if let Some(mem) = entry.membership { sm.last_membership_log_id = Some(log_id.into()); sm.last_membership = Some(mem); @@ -120,7 +120,7 @@ impl RaftStateMachine for Arc { }; if let Some(responder) = responder { - responder.send(Response { value }); + responder.send(response); } } Ok(()) diff --git a/examples/raft-kv-memstore-network-v1/tests/cluster/test_cluster.rs b/examples/raft-kv-memstore-network-v1/tests/cluster/test_cluster.rs index fe3298eea..86641f6b2 100644 --- a/examples/raft-kv-memstore-network-v1/tests/cluster/test_cluster.rs +++ b/examples/raft-kv-memstore-network-v1/tests/cluster/test_cluster.rs @@ -98,8 +98,11 @@ async fn run_test(rafts: &[Raft]) { println!("write resp: {:#?}", resp); let resp = client.write(&types_kv::Request::set("foo2", "bar2")).await.unwrap().unwrap(); println!("write resp: {:#?}", resp); - assert_eq!(resp.log_id.index(), resp.data.value.as_ref().unwrap().version); - resp.data + assert_eq!( + resp.log_id.index(), + resp.data.as_ref().unwrap().value.as_ref().unwrap().version + ); + resp.data.unwrap() }; println!("=== let node-1 take a snapshot"); diff --git a/examples/raft-kv-memstore-opendal-snapshot-data/src/store.rs b/examples/raft-kv-memstore-opendal-snapshot-data/src/store.rs index 882b69a39..4513382eb 100644 --- a/examples/raft-kv-memstore-opendal-snapshot-data/src/store.rs +++ b/examples/raft-kv-memstore-opendal-snapshot-data/src/store.rs @@ -140,8 +140,8 @@ impl RaftStateMachine for Arc { sm.last_applied = Some(entry.log_id); let response = match entry.payload { - EntryPayload::Blank => types_kv::Response::none(), - EntryPayload::Normal(ref req) => match req { + EntryPayload::Blank => None, + EntryPayload::Normal(ref req) => Some(match req { types_kv::Request::Set { key, value, .. } => { sm.data.insert(key.clone(), types_kv::VersionedValue { value: value.clone(), @@ -166,10 +166,10 @@ impl RaftStateMachine for Arc { types_kv::Response::none() } } - }, + }), EntryPayload::Membership(ref mem) => { sm.last_membership = StoredMembership::new(Some(entry.log_id), mem.clone()); - types_kv::Response::none() + None } }; diff --git a/examples/raft-kv-memstore-single-threaded/src/store.rs b/examples/raft-kv-memstore-single-threaded/src/store.rs index 0624078d4..f582c7f92 100644 --- a/examples/raft-kv-memstore-single-threaded/src/store.rs +++ b/examples/raft-kv-memstore-single-threaded/src/store.rs @@ -196,18 +196,18 @@ impl RaftStateMachine for Rc { sm.last_applied = Some(entry.log_id); let response = match entry.payload { - EntryPayload::Blank => Response { value: None }, + EntryPayload::Blank => None, EntryPayload::Normal(ref req) => match req { Request::Set { key, value, .. } => { sm.data.insert(key.clone(), value.clone()); - Response { + Some(Response { value: Some(value.clone()), - } + }) } }, EntryPayload::Membership(ref mem) => { sm.last_membership = StoredMembership::new(Some(entry.log_id), mem.clone()); - Response { value: None } + None } }; diff --git a/examples/raft-kv-memstore/tests/cluster/test_basic.rs b/examples/raft-kv-memstore/tests/cluster/test_basic.rs index 09f1f59d9..1bf40ea2c 100644 --- a/examples/raft-kv-memstore/tests/cluster/test_basic.rs +++ b/examples/raft-kv-memstore/tests/cluster/test_basic.rs @@ -24,7 +24,7 @@ async fn test_basic_inner() -> anyhow::Result<()> { value: "bar".to_string(), }) .await??; - let first_version = first.data.value.as_ref().unwrap().version; + let first_version = first.data.as_ref().unwrap().value.as_ref().unwrap().version; println!("=== write foo=bar again"); let second = client @@ -33,42 +33,42 @@ async fn test_basic_inner() -> anyhow::Result<()> { value: "bar".to_string(), }) .await??; - let second_version = second.data.value.as_ref().unwrap().version; + let second_version = second.data.as_ref().unwrap().value.as_ref().unwrap().version; assert!(second_version > first_version); TypeConfig::sleep(Duration::from_millis(500)).await; println!("=== read foo"); let got = client.read(&"foo".to_string()).await?; - assert_eq!(second.data, got); + assert_eq!(second.data.as_ref(), Some(&got)); println!("=== compare and set foo=bar to foo=baz"); let cas = client.write(&types_kv::Request::compare_and_set("foo", second_version, "baz")).await??; - let cas_value = cas.data.value.as_ref().unwrap(); + let cas_value = cas.data.as_ref().unwrap().value.as_ref().unwrap(); assert_eq!("baz", cas_value.value); assert!(cas_value.version > second_version); println!("=== reject a stale compare and set"); let stale = client.write(&types_kv::Request::compare_and_set("foo", second_version, "stale")).await??; - assert_eq!(types_kv::Response::none(), stale.data); - assert_eq!(cas.data, client.read(&"foo".to_string()).await?); + assert_eq!(Some(types_kv::Response::none()), stale.data); + assert_eq!(cas.data.as_ref(), Some(&client.read(&"foo".to_string()).await?)); println!("=== reject a compare and set for a missing key"); let missing = client.write(&types_kv::Request::compare_and_set("missing", 0, "value")).await??; - assert_eq!(types_kv::Response::none(), missing.data); + assert_eq!(Some(types_kv::Response::none()), missing.data); assert_eq!(types_kv::Response::none(), client.read(&"missing".to_string()).await?); println!("=== reject a compare and set after ABA"); let original = client.write(&types_kv::Request::set("aba", "A")).await??; - let original_version = original.data.value.as_ref().unwrap().version; + let original_version = original.data.as_ref().unwrap().value.as_ref().unwrap().version; client.write(&types_kv::Request::set("aba", "B")).await??; let restored = client.write(&types_kv::Request::set("aba", "A")).await??; - let restored_version = restored.data.value.as_ref().unwrap().version; + let restored_version = restored.data.as_ref().unwrap().value.as_ref().unwrap().version; assert!(restored_version > original_version); let stale = client.write(&types_kv::Request::compare_and_set("aba", original_version, "C")).await??; - assert_eq!(types_kv::Response::none(), stale.data); - assert_eq!(restored.data, client.read(&"aba".to_string()).await?); + assert_eq!(Some(types_kv::Response::none()), stale.data); + assert_eq!(restored.data.as_ref(), Some(&client.read(&"aba".to_string()).await?)); Ok(()) } diff --git a/examples/raft-kv-memstore/tests/cluster/test_read_modes.rs b/examples/raft-kv-memstore/tests/cluster/test_read_modes.rs index a30b47015..9596a5575 100644 --- a/examples/raft-kv-memstore/tests/cluster/test_read_modes.rs +++ b/examples/raft-kv-memstore/tests/cluster/test_read_modes.rs @@ -26,7 +26,7 @@ async fn test_read_modes_inner() -> anyhow::Result<()> { value: "bar".to_string(), }) .await??; - let expected = write_response.data; + let expected = write_response.data.unwrap(); TypeConfig::sleep(Duration::from_millis(1_000)).await; let follower2 = Client::::new(2, util::api_addr(PORT_BASE, 2)); diff --git a/examples/raft-kv-memstore/tests/cluster/test_snapshot.rs b/examples/raft-kv-memstore/tests/cluster/test_snapshot.rs index b15207895..1b5dfe8e0 100644 --- a/examples/raft-kv-memstore/tests/cluster/test_snapshot.rs +++ b/examples/raft-kv-memstore/tests/cluster/test_snapshot.rs @@ -40,7 +40,7 @@ async fn test_snapshot_replication_inner() -> anyhow::Result<()> { println!("=== write 2 logs on the leader"); leader.write(&types_kv::Request::set("foo1", "bar1")).await??; - let expected_foo2 = leader.write(&types_kv::Request::set("foo2", "bar2")).await??.data; + let expected_foo2 = leader.write(&types_kv::Request::set("foo2", "bar2")).await??.data.unwrap(); println!("=== leader snapshots and purges the logs it covers"); // Building the snapshot and purging the logs it covers are two steps, so wait for diff --git a/examples/raft-kv-rocksdb/tests/cluster/test_cluster.rs b/examples/raft-kv-rocksdb/tests/cluster/test_cluster.rs index 3e26594cc..55cceb806 100644 --- a/examples/raft-kv-rocksdb/tests/cluster/test_cluster.rs +++ b/examples/raft-kv-rocksdb/tests/cluster/test_cluster.rs @@ -165,7 +165,7 @@ async fn test_cluster_inner() -> Result<(), Box Result<(), Box bar_version); - let expected_wow = write_wow.data; + let expected_wow = write_wow.data.unwrap(); TypeConfig::sleep(Duration::from_millis(500)).await; @@ -237,16 +237,16 @@ async fn test_cluster_inner() -> Result<(), Box wow_version); println!("=== reject a stale compare and set"); let stale = leader.write(&types_kv::Request::compare_and_set("foo", wow_version, "stale")).await??; - assert_eq!(types_kv::Response::none(), stale.data); + assert_eq!(Some(types_kv::Response::none()), stale.data); let got = leader.linearizable_read(&"foo".to_string()).await??; - assert_eq!(cas.data, got); + assert_eq!(cas.data.as_ref(), Some(&got)); Ok(()) } diff --git a/examples/raft-kv-rocksdb/tests/cluster/test_snapshot.rs b/examples/raft-kv-rocksdb/tests/cluster/test_snapshot.rs index 52640563c..c232ce6e0 100644 --- a/examples/raft-kv-rocksdb/tests/cluster/test_snapshot.rs +++ b/examples/raft-kv-rocksdb/tests/cluster/test_snapshot.rs @@ -42,7 +42,7 @@ async fn test_snapshot_replication_inner() -> anyhow::Result<()> { println!("=== write 2 logs on the leader"); leader.write(&types_kv::Request::set("foo1", "bar1")).await??; - let expected_foo2 = leader.write(&types_kv::Request::set("foo2", "bar2")).await??.data; + let expected_foo2 = leader.write(&types_kv::Request::set("foo2", "bar2")).await??.data.unwrap(); println!("=== leader snapshots and purges the logs it covers"); // Building the snapshot and purging the logs it covers are two steps, so wait for diff --git a/examples/raft-kv-rocksdb/tests/cluster/test_snapshot_recovery.rs b/examples/raft-kv-rocksdb/tests/cluster/test_snapshot_recovery.rs index 8e0078f92..5afda2701 100644 --- a/examples/raft-kv-rocksdb/tests/cluster/test_snapshot_recovery.rs +++ b/examples/raft-kv-rocksdb/tests/cluster/test_snapshot_recovery.rs @@ -35,9 +35,9 @@ fn test_snapshot_followed_by_logs() -> anyhow::Result<()> { let installed = metrics.borrow_watched().snapshot; assert_eq!(snapshot.meta.last_log_id, installed); let shared = learner.read("shared").await?; - assert_eq!(overwritten.data, shared); + assert_eq!(overwritten.data.as_ref(), Some(&shared)); let tail = learner.read("tail").await?; - assert_eq!(added.data, tail); + assert_eq!(added.data.as_ref(), Some(&tail)); let expected_state = leader.applied_state().await?; let actual_state = learner.applied_state().await?; assert_eq!(expected_state, actual_state); @@ -45,13 +45,13 @@ fn test_snapshot_followed_by_logs() -> anyhow::Result<()> { tracing::info!("Continue writing through the writable database created from the checkpoint"); { - let value = overwritten.data.value.context("overwritten value exists")?; + let value = overwritten.data.unwrap().value.context("overwritten value exists")?; let request = types_kv::Request::compare_and_set("shared", value.version, "cas"); let response = leader.raft().client_write(request).await?; let index = response.log_id.index(); learner.wait_applied(index).await?; let expected = types_kv::Response::new("cas", index); - assert_eq!(expected, response.data); + assert_eq!(Some(&expected), response.data.as_ref()); let actual = learner.read("shared").await?; assert_eq!(expected, actual); } @@ -76,7 +76,7 @@ fn test_restart_after_snapshot_installation() -> anyhow::Result<()> { let snapshot = leader.build_snapshot().await?; leader.purge(&snapshot).await?; leader.add_learner(&learner).await?; - written.data + written.data.unwrap() }; let expected_state = learner.applied_state().await?; let expected_current = learner.current()?; @@ -98,7 +98,7 @@ fn test_restart_after_snapshot_installation() -> anyhow::Result<()> { let index = response.log_id.index(); learner.wait_applied(index).await?; let actual = learner.read("key").await?; - assert_eq!(response.data, actual); + assert_eq!(response.data.as_ref(), Some(&actual)); let expected_state = leader.applied_state().await?; let actual_state = learner.applied_state().await?; assert_eq!(expected_state, actual_state); @@ -123,7 +123,7 @@ fn test_restarted_leader_rebuilds_snapshot_for_learner() -> anyhow::Result<()> { let written = leader.write("key", "survives purge").await?; let snapshot = leader.build_snapshot().await?; leader.purge(&snapshot).await?; - (written.data, snapshot) + (written.data.unwrap(), snapshot) }; let expected_current = leader.current()?; drop(snapshot.snapshot); diff --git a/examples/raft-kv-rocksdb/tests/cluster/test_snapshot_transport.rs b/examples/raft-kv-rocksdb/tests/cluster/test_snapshot_transport.rs index 32ceddfd5..4a2c41b7c 100644 --- a/examples/raft-kv-rocksdb/tests/cluster/test_snapshot_transport.rs +++ b/examples/raft-kv-rocksdb/tests/cluster/test_snapshot_transport.rs @@ -39,7 +39,7 @@ fn test_interrupted_snapshot_preserves_state_and_retries() -> anyhow::Result<()> let response = transfer.send(&learner).await?; let expected_response = SnapshotResponse::new(transfer.vote); assert_eq!(expected_response, response); - written.data + written.data.unwrap() }; let state_before = learner.applied_state().await?; let current_before = learner.current()?; @@ -51,7 +51,7 @@ fn test_interrupted_snapshot_preserves_state_and_retries() -> anyhow::Result<()> let written = leader.write("key", &value).await?; let snapshot = leader.build_snapshot().await?; let transfer = SnapshotTransfer::new(&leader, &snapshot, CHUNK_SIZE).await?; - (written.data, transfer) + (written.data.unwrap(), transfer) }; tracing::info!(dir = %learner.dir().display(), "Disconnect after the receiver writes part of a file"); @@ -150,7 +150,7 @@ fn test_large_snapshot_with_fragmented_http_body() -> anyhow::Result<()> { } let spans_multiple_chunks = full_chunks > 1; assert!(spans_multiple_chunks); - ((written.data, second.data), transfer) + ((written.data.unwrap(), second.data.unwrap()), transfer) }; tracing::info!("Split length prefixes and payloads across HTTP chunks, combining message boundaries"); @@ -201,7 +201,7 @@ fn test_invalid_snapshot_payloads_preserve_state() -> anyhow::Result<()> { let snapshot = leader.build_snapshot().await?; let transfer = SnapshotTransfer::new(&leader, &snapshot, CHUNK_SIZE).await?; transfer.send(&learner).await?; - (written.data, transfer) + (written.data.unwrap(), transfer) }; let state_before = learner.applied_state().await?; let current_before = learner.current()?; @@ -278,7 +278,7 @@ fn test_stale_and_duplicate_snapshots_are_discarded() -> anyhow::Result<()> { let snapshot = leader.build_snapshot().await?; let current = SnapshotTransfer::new(&leader, &snapshot, CHUNK_SIZE).await?; current.send(&learner).await?; - (older, current, written.data) + (older, current, written.data.unwrap()) }; let state_before = learner.applied_state().await?; let current_before = learner.current()?; diff --git a/examples/sm-mem/src/lib.rs b/examples/sm-mem/src/lib.rs index f896de2e2..5f9a607c1 100644 --- a/examples/sm-mem/src/lib.rs +++ b/examples/sm-mem/src/lib.rs @@ -150,8 +150,8 @@ where C: RaftTypeConfig< inner.last_applied_log = Some(entry.log_id.clone()); let response = match &entry.payload { - EntryPayload::Blank => types_kv::Response::none(), - EntryPayload::Normal(req) => match req { + EntryPayload::Blank => None, + EntryPayload::Normal(req) => Some(match req { types_kv::Request::Set { key, value } => { inner.state_machine.data.insert(key.clone(), types_kv::VersionedValue { value: value.clone(), @@ -180,10 +180,10 @@ where C: RaftTypeConfig< types_kv::Response::none() } } - }, + }), EntryPayload::Membership(mem) => { inner.last_membership = StoredMembershipOf::::new(Some(entry.log_id.clone()), mem.clone()); - types_kv::Response::none() + None } }; diff --git a/examples/sm-rocks/src/state_machine.rs b/examples/sm-rocks/src/state_machine.rs index e5eeccb67..863dd634b 100644 --- a/examples/sm-rocks/src/state_machine.rs +++ b/examples/sm-rocks/src/state_machine.rs @@ -125,6 +125,7 @@ where C: RaftTypeConfig< let version = entry.log_id().index(); last_applied_log = Some(entry.log_id()); + let is_normal = matches!(&entry.payload, EntryPayload::Normal(_)); let put = match entry.payload { EntryPayload::Blank => None, EntryPayload::Normal(ref req) => match req { @@ -168,7 +169,7 @@ where C: RaftTypeConfig< }; if let Some(responder) = responder { - responses.push((responder, response)); + responses.push((responder, is_normal.then_some(response))); } } diff --git a/jepsen/openraft-test-app/src/store.rs b/jepsen/openraft-test-app/src/store.rs index 7f3094cf3..053387181 100644 --- a/jepsen/openraft-test-app/src/store.rs +++ b/jepsen/openraft-test-app/src/store.rs @@ -148,8 +148,8 @@ impl RaftStateMachine for StateMachineStore { self.data.last_applied_log_id = Some(entry.log_id.clone()); let response = match entry.payload { - EntryPayload::Blank => types_kv::Response::none(), - EntryPayload::Normal(req) => match req { + EntryPayload::Blank => None, + EntryPayload::Normal(req) => Some(match req { types_kv::Request::Set { key, value } => { let mut st = self.data.kvs.lock().await; st.insert(key, types_kv::VersionedValue { @@ -176,10 +176,10 @@ impl RaftStateMachine for StateMachineStore { types_kv::Response::none() } } - }, + }), EntryPayload::Membership(mem) => { self.data.last_membership = StoredMembership::new(Some(entry.log_id), mem); - types_kv::Response::none() + None } }; diff --git a/openraft/src/raft/message/client_write.rs b/openraft/src/raft/message/client_write.rs index debc7f622..1ae971b8c 100644 --- a/openraft/src/raft/message/client_write.rs +++ b/openraft/src/raft/message/client_write.rs @@ -18,12 +18,22 @@ pub type ClientWriteResult = Result, ClientWriteError< derive(serde::Deserialize, serde::Serialize), serde(bound = "C::R: crate::AppDataResponse") )] +#[since] pub struct ClientWriteResponse { /// The id of the log that is applied. pub log_id: LogIdOf, /// Application specific response data. - pub data: C::R, + /// + /// It is [`Some`] only when the applied log entry is a normal application entry + /// ([`EntryPayload::Normal`](crate::EntryPayload::Normal)). A blank entry does not carry + /// application data, and a membership entry reports its outcome via + /// [`membership`](Self::membership); both yield `None`. + #[since( + version = "0.10.0", + change = "`Option`; `None` when the entry has no response data" + )] + pub data: Option, /// If the log entry is a change-membership entry. pub membership: Option>, @@ -38,7 +48,7 @@ where C: RaftTypeConfig pub(crate) fn new_app_response(log_id: LogIdOf, data: C::R) -> Self { Self { log_id, - data, + data: Some(data), membership: None, } } @@ -48,9 +58,13 @@ where C: RaftTypeConfig &self.log_id } + /// Return the application response data, if the applied entry carries any. + /// + /// Returns [`None`] for blank and membership entries. + #[since(version = "0.10.0", change = "returns `Option<&C::R>`")] #[since(version = "0.9.5")] - pub fn response(&self) -> &C::R { - &self.data + pub fn response(&self) -> Option<&C::R> { + self.data.as_ref() } /// Return membership config if the log entry is a change-membership entry. diff --git a/openraft/src/raft/message/write.rs b/openraft/src/raft/message/write.rs index 62d19f63b..10bc05c09 100644 --- a/openraft/src/raft/message/write.rs +++ b/openraft/src/raft/message/write.rs @@ -38,9 +38,12 @@ pub struct WriteResponse { impl From> for WriteResponse { fn from(resp: ClientWriteResponse) -> Self { + // `client_write_many()` only submits normal payloads, so every response carries data. + let response = resp.data.expect("client_write_many() only produces normal entries"); + WriteResponse { log_id: resp.log_id, - response: resp.data, + response, } } } diff --git a/openraft/src/storage/v2/apply_responder.rs b/openraft/src/storage/v2/apply_responder.rs index cd45152c0..040dd715e 100644 --- a/openraft/src/storage/v2/apply_responder.rs +++ b/openraft/src/storage/v2/apply_responder.rs @@ -1,3 +1,5 @@ +use openraft_macros::since; + use crate::RaftTypeConfig; use crate::storage::v2::apply_responder_inner::ApplyResponderInner; @@ -27,16 +29,17 @@ use crate::storage::v2::apply_responder_inner::ApplyResponderInner; /// I::IntoIter: Send, /// { /// for (entry, responder) in entries { -/// // Compute response based on entry type +/// // Compute response based on entry type. +/// // Blank and membership entries carry no application response data. /// let response = match entry.payload { -/// EntryPayload::Blank => Response::default(), +/// EntryPayload::Blank => None, /// EntryPayload::Normal(ref data) => { /// self.apply_normal_entry(data)?; -/// self.compute_response(data)? +/// Some(self.compute_response(data)?) /// } /// EntryPayload::Membership(ref mem) => { /// self.apply_membership_change(mem)?; -/// Response::default() +/// None /// } /// }; /// @@ -55,10 +58,14 @@ pub struct ApplyResponder { impl ApplyResponder { /// Send the response after applying an entry. /// + /// `response` is [`Some`] only when the applied entry is a normal application entry; + /// blank and membership entries pass [`None`]. + /// /// The response will be returned by [`Raft::client_write`](crate::Raft::client_write). /// This method must be called by [`RaftStateMachine::apply`](super::RaftStateMachine::apply), /// otherwise [`Raft::client_write`](crate::Raft::client_write) will never return. - pub fn send(self, response: C::R) { + #[since(version = "0.10.0", change = "accepts `Option`")] + pub fn send(self, response: Option) { self.inner.send(response) } } diff --git a/openraft/src/storage/v2/apply_responder_inner.rs b/openraft/src/storage/v2/apply_responder_inner.rs index 570de1dea..88809de16 100644 --- a/openraft/src/storage/v2/apply_responder_inner.rs +++ b/openraft/src/storage/v2/apply_responder_inner.rs @@ -21,7 +21,10 @@ pub(crate) enum ApplyResponderInner { impl ApplyResponderInner { /// Send the response after applying an entry. - pub(crate) fn send(self, response: C::R) { + /// + /// `response` is [`Some`] only for a normal application entry; blank and membership entries + /// pass [`None`]. + pub(crate) fn send(self, response: Option) { use crate::raft::ClientWriteResponse; use crate::raft::responder::Responder as ResponderTrait; diff --git a/stores/memstore-custom-node-id/src/lib.rs b/stores/memstore-custom-node-id/src/lib.rs index 314c5fa46..d48d268a7 100644 --- a/stores/memstore-custom-node-id/src/lib.rs +++ b/stores/memstore-custom-node-id/src/lib.rs @@ -262,11 +262,12 @@ impl RaftStateMachine for Arc { let mut sm = self.sm.write().await; while let Some((entry, responder)) = entries.try_next().await? { sm.last_applied_log = Some(entry.log_id); + let is_normal = matches!(&entry.payload, EntryPayload::Normal(_)); if let EntryPayload::Membership(ref mem) = entry.payload { sm.last_membership = StoredMembershipOf::::new(Some(entry.log_id), mem.clone()); } if let Some(r) = responder { - r.send(()); + r.send(is_normal.then_some(())); } } Ok(()) diff --git a/stores/memstore/src/lib.rs b/stores/memstore/src/lib.rs index c8b2ea1e2..c64c01ba7 100644 --- a/stores/memstore/src/lib.rs +++ b/stores/memstore/src/lib.rs @@ -552,14 +552,14 @@ impl RaftStateMachine for Arc { sm.last_applied_log = Some(entry.log_id); let response = match entry.payload { - EntryPayload::Blank => ClientResponse(None), + EntryPayload::Blank => None, EntryPayload::Normal(ref data) => { let previous = sm.client_status.insert(data.client.clone(), data.status.clone()); - ClientResponse(previous) + Some(ClientResponse(previous)) } EntryPayload::Membership(ref mem) => { sm.last_membership = StoredMembershipOf::::new(Some(entry.log_id), mem.clone()); - ClientResponse(None) + None } }; diff --git a/tests-turmoil/src/bin/fuzz.rs b/tests-turmoil/src/bin/fuzz.rs index 0cae665c6..f3d646389 100644 --- a/tests-turmoil/src/bin/fuzz.rs +++ b/tests-turmoil/src/bin/fuzz.rs @@ -1027,7 +1027,11 @@ async fn driver_loop(queue: DriverQueue, history: Arc>) -> if dbg_ops() { println!("DBG ack serial={serial} log={}", resp.log_id); } - history.lock().unwrap().record_write_acked(serial, resp.log_id, resp.data.prev) + history.lock().unwrap().record_write_acked( + serial, + resp.log_id, + resp.data.expect("client write always carries a response").prev, + ) } _ => { if dbg_ops() { diff --git a/tests-turmoil/src/store.rs b/tests-turmoil/src/store.rs index 82bfca241..1d3aec8ba 100644 --- a/tests-turmoil/src/store.rs +++ b/tests-turmoil/src/store.rs @@ -241,18 +241,18 @@ impl RaftStateMachine for Arc { data.last_applied = Some(entry.log_id); let response = match entry.payload { - EntryPayload::Blank => Response { prev: None }, + EntryPayload::Blank => None, EntryPayload::Normal(ref req) => { let prev = data.data.insert(req.key.clone(), ValueMeta { value: req.value.clone(), serial: req.serial, log_id: entry.log_id, }); - Response { prev } + Some(Response { prev }) } EntryPayload::Membership(ref mem) => { data.last_membership = StoredMembership::new(Some(entry.log_id), mem.clone()); - Response { prev: None } + None } }; diff --git a/tests/tests/client_api/t10_client_writes.rs b/tests/tests/client_api/t10_client_writes.rs index 35a0bb6d4..4c5d475ce 100644 --- a/tests/tests/client_api/t10_client_writes.rs +++ b/tests/tests/client_api/t10_client_writes.rs @@ -115,7 +115,7 @@ async fn client_write_ff() -> Result<()> { let (responder, complete_rx) = ProgressResponder::complete_only(); n0.client_write_ff(ClientRequest::make_request("foo", 2), Some(responder)).await?; let got: ClientWriteResponse = complete_rx.await??; - assert_eq!(None, got.response().0.as_deref()); + assert_eq!(None, got.response().unwrap().0.as_deref()); tracing::info!("--- omit one responder and observe that write through the next response"); { @@ -124,7 +124,7 @@ async fn client_write_ff() -> Result<()> { let (responder, complete_rx) = ProgressResponder::complete_only(); n0.client_write_ff(ClientRequest::make_request("foo", 4), Some(responder)).await?; let got: ClientWriteResponse = complete_rx.await??; - assert_eq!(Some("request-3"), got.response().0.as_deref()); + assert_eq!(Some("request-3"), got.response().unwrap().0.as_deref()); } Ok(()) @@ -156,7 +156,7 @@ async fn write_builder() -> Result<()> { let (responder, complete_rx) = ProgressResponder::complete_only(); n0.write(ClientRequest::make_request("foo", 2)).responder(responder).await?; let got: ClientWriteResponse = complete_rx.await??; - assert_eq!(None, got.response().0.as_deref()); + assert_eq!(None, got.response().unwrap().0.as_deref()); // Test write without responder (fire-and-forget) n0.write(ClientRequest::make_request("foo", 3)).await?; @@ -165,7 +165,7 @@ async fn write_builder() -> Result<()> { let (responder, complete_rx) = ProgressResponder::complete_only(); n0.write(ClientRequest::make_request("foo", 4)).responder(responder).await?; let got: ClientWriteResponse = complete_rx.await??; - assert_eq!(Some("request-3"), got.response().0.as_deref()); + assert_eq!(Some("request-3"), got.response().unwrap().0.as_deref()); Ok(()) } @@ -204,7 +204,7 @@ async fn write_with_leader() -> Result<()> { router.wait(&0, None).applied_index(Some(log_index), "write with correct leader").await?; let got = complete_rx.await??; - assert_eq!(&ClientResponse(None), got.response()); + assert_eq!(Some(&ClientResponse(None)), got.response()); assert_eq!(&log_id(1, 0, log_index), got.log_id()); // Test 2: Write with incorrect leader should fail with ForwardToLeader @@ -234,7 +234,7 @@ async fn write_with_leader() -> Result<()> { let got = complete_rx.await??; // Should see request-2 (request-3 failed) - assert_eq!(&ClientResponse(Some("request-2".to_string())), got.response()); + assert_eq!(Some(&ClientResponse(Some("request-2".to_string()))), got.response()); assert_eq!(&log_id(1, 0, log_index), got.log_id()); Ok(()) diff --git a/tests/tests/client_api/t15_client_write_with_twoshot.rs b/tests/tests/client_api/t15_client_write_with_twoshot.rs index a39e5dba6..742bd9bd3 100644 --- a/tests/tests/client_api/t15_client_write_with_twoshot.rs +++ b/tests/tests/client_api/t15_client_write_with_twoshot.rs @@ -52,7 +52,7 @@ async fn client_write_ff_with_progress_responder() -> Result<()> { let response = result?; assert_eq!(log_id(1, 0, log_index), response.log_id); // First write returns None (no previous value) - assert_eq!(None, response.response().0.as_deref()); + assert_eq!(None, response.response().unwrap().0.as_deref()); // Test another write - pattern follows t10: write("foo", 11) with no responder, then write("foo", // 12) @@ -72,7 +72,7 @@ async fn client_write_ff_with_progress_responder() -> Result<()> { let response = result?; assert_eq!(log_id(1, 0, log_index), response.log_id); // Should return the value from the previous write (11) - assert_eq!(Some("request-11"), response.response().0.as_deref()); + assert_eq!(Some("request-11"), response.response().unwrap().0.as_deref()); Ok(()) } diff --git a/tests/tests/fixtures/mod.rs b/tests/tests/fixtures/mod.rs index 281d091e9..e7dfd4595 100644 --- a/tests/tests/fixtures/mod.rs +++ b/tests/tests/fixtures/mod.rs @@ -810,7 +810,7 @@ impl TypedRaftRouter { &self, target: MemNodeId, req: ClientRequest, - ) -> Result>> { + ) -> Result, RaftError>> { let node = { let rt = self.nodes.lock().unwrap(); rt.get(&target) diff --git a/tests/tests/membership/t23_custom_payload.rs b/tests/tests/membership/t23_custom_payload.rs index 12fac40c3..cf7a685df 100644 --- a/tests/tests/membership/t23_custom_payload.rs +++ b/tests/tests/membership/t23_custom_payload.rs @@ -275,9 +275,10 @@ impl RaftStateMachine for TestStateMachine { async fn apply(&mut self, mut entries: Strm) -> Result<(), io::Error> where Strm: Stream, io::Error>> + Unpin + OptionalSend { while let Some((entry, responder)) = entries.try_next().await? { + let has_normal = entry.payload.normal.is_some(); self.apply_entry(entry); if let Some(responder) = responder { - responder.send(()); + responder.send(has_normal.then_some(())); } }