feat(torture):reverted changeset on previous state

Make sure that any reverted changes did not alter
already present values. To do this, compare them with
the values of the previous state maintained by the supervisor.

Author: gabriele-0201

torture/src/supervisor/workload.rs

@@ -404,16 +404,36 @@ impl Workload {
         // is equal to its previous state and that each new key is not avaible.
         for change in changeset {
             match change {
-                KeyValueChange::Insert(key, value)
-                    if rr.send_request_query(*key).await?.as_ref() == Some(value) =>
-                {
-                    return Err(anyhow::anyhow!("Inserted item should be reverted"));
+                KeyValueChange::Insert(key, _value) => {
+                    let Some(prev_value) = self.state.committed.state.get(key) else {
+                        // if the key was not present in the previous state,
+                        // it must be a new key. Thus None is expected
+                        if rr.send_request_query(*key).await?.is_some() {
+                            return Err(anyhow::anyhow!("New inserted item should not be present"));
+                        }
+                        continue;
+                    };
+
+                    if rr.send_request_query(*key).await? != *prev_value {
+                        let err = if prev_value.is_some() {
+                            "Modified item should be reverted to previous state"
+                        } else {
+                            "New inserted item should not be present"
+                        };
+                        return Err(anyhow::anyhow!(err));
+                    }
                 }
-                // This holds as long as we are sure that every deletions is done on present keys.
-                KeyValueChange::Delete(key) if rr.send_request_query(*key).await?.is_none() => {
-                    return Err(anyhow::anyhow!("Deletion should have been reverted"));
+                KeyValueChange::Delete(key) => {
+                    // UNWRAP: Non existing keys are not deleted
+                    let prev_value = self.state.committed.state.get(key).unwrap();
+                    if rr.send_request_query(*key).await? != *prev_value {
+                        // Non existing keys are not deleted
+                        assert!(prev_value.is_some());
+                        return Err(anyhow::anyhow!(
+                            "Deleted item should be reverted to previous state"
+                        ));
+                    }
                 }
-                _ => (),
             }
         }
         Ok(())