change: raft: make ClientWriteResponse.data optional

benchmarks/minimal/src/store.rs

@@ -314,6 +314,7 @@ impl RaftStateMachine<TypeConfig> for Arc<StateMachineStore> {
         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<TypeConfig> for Arc<StateMachineStore> {
             };
 
             if let Some(responder) = responder {
-                responder.send(ClientResponse {});
+                responder.send(is_normal.then_some(ClientResponse {}));
             }
         }
         Ok(())

examples/raft-kv-log-wal-sm-mem/tests/cluster/test_cluster.rs

@@ -111,7 +111,7 @@ async fn test_cluster_inner() -> Result<(), Box<dyn std::error::Error + Send + S
         tracing::info!("write foo=bar on the leader");
 
         let resp = leader.write(&types_kv::Request::set("foo", "bar")).await??;
-        resp.data
+        resp.data.unwrap()
     };
 
     tracing::info!("read foo=bar back on every node, the two followers included");

examples/raft-kv-log-wal-sm-mem/tests/cluster/test_restart.rs

@@ -44,7 +44,7 @@ async fn restart_inner() -> Result<(), Box<dyn std::error::Error + Send + Sync>>
         // next open needs.
         raft.shutdown().await?;
 
-        resp.data
+        resp.data.unwrap()
     };
 
     tracing::info!("reopen the same WAL directory with a fresh state machine");

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())),

examples/raft-kv-memstore-grpc/src/pb_impl/impl_client_write_response.rs

@@ -5,7 +5,7 @@ impl From<pb::ClientWriteResponse> 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<ClientWriteResponse> 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()),
         }
     }

examples/raft-kv-memstore-grpc/src/store/mod.rs

@@ -108,9 +108,9 @@ impl RaftStateMachine<TypeConfig> for Arc<StateMachineStore> {
 
             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<TypeConfig> for Arc<StateMachineStore> {
             };
 
             if let Some(responder) = responder {
-                responder.send(Response { value });
+                responder.send(response);
             }
         }
         Ok(())

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");

examples/raft-kv-memstore-opendal-snapshot-data/src/store.rs

@@ -140,8 +140,8 @@ impl RaftStateMachine<TypeConfig> for Arc<StateMachineStore> {
             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<TypeConfig> for Arc<StateMachineStore> {
                             types_kv::Response::none()
                         }
                     }
-                },
+                }),
                 EntryPayload::Membership(ref mem) => {
                     sm.last_membership = StoredMembership::new(Some(entry.log_id), mem.clone());
-                    types_kv::Response::none()
+                    None
                 }
             };
 

examples/raft-kv-memstore-single-threaded/src/store.rs

@@ -196,18 +196,18 @@ impl RaftStateMachine<TypeConfig> for Rc<StateMachineStore> {
             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
                 }
             };
 

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(())
 }

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::<TypeConfig>::new(2, util::api_addr(PORT_BASE, 2));

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

examples/raft-kv-rocksdb/tests/cluster/test_cluster.rs

@@ -165,7 +165,7 @@ async fn test_cluster_inner() -> Result<(), Box<dyn std::error::Error + Send + S
             value: "bar".to_string(),
         })
         .await??;
-    let expected_bar = write_bar.data;
+    let expected_bar = write_bar.data.unwrap();
 
     // --- Wait for a while to let the replication get done.
 
@@ -195,9 +195,9 @@ async fn test_cluster_inner() -> Result<(), Box<dyn std::error::Error + Send + S
         })
         .await??;
     let bar_version = expected_bar.value.as_ref().unwrap().version;
-    let wow_version = write_wow.data.value.as_ref().unwrap().version;
+    let wow_version = write_wow.data.as_ref().unwrap().value.as_ref().unwrap().version;
     assert!(wow_version > 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<dyn std::error::Error + Send + S
 
     println!("=== compare and set `foo=wow` to `foo=cas`");
     let cas = leader.write(&types_kv::Request::compare_and_set("foo", wow_version, "cas")).await??;
-    let cas_value = cas.data.value.as_ref().unwrap();
+    let cas_value = cas.data.as_ref().unwrap().value.as_ref().unwrap();
     assert_eq!("cas", cas_value.value);
     assert!(cas_value.version > 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(())
 }

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

examples/raft-kv-rocksdb/tests/cluster/test_snapshot_recovery.rs

@@ -35,23 +35,23 @@ 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);
         }
 
         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);

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()?;