Refactor：Refactor curp fast read implementation

[Phoenix500526]

Currently, on performance grounds, xline provides a fast-read mechanism(ReadState) to a client, which can prevent a leader node from becoming a bottleneck in an xline cluster. Thanks to the implementation of the read-state part, a command has to go through three stages from prepare to execute and then to after_sync, just for some notification reasons, even though it fails in its earlier stage. This doesn't make sense. We should refactor the implementation of the ReadState part.

[themanforfree]

Passing LogIndex to the prepare and execute stage of a command can trigger barriers in advance when failures occur, and it is no longer necessary to call other stages for failed commands.

impl<S> CurpCommandExecutor<Command> for CommandExecutor<S>
where
    S: StorageApi,
{
    fn prepare(
        &self,
        cmd: &Command,
        index: LogIndex,
    ) -> Result<<Command as CurpCommand>::PR, Self::Error> {
        let wrapper = cmd.request();
        if let Err(e) = self.auth_storage.check_permission(wrapper) {
            self.id_barrier.trigger(cmd.id());
            self.index_barrier.trigger(index);
            return Err(e);
        }
        let revision = match wrapper.request.backend() {
            RequestBackend::Auth => {
                if wrapper.request.skip_auth_revision() {
                    -1
                } else {
                    self.auth_rev.next()
                }
            }
            RequestBackend::Kv | RequestBackend::Lease => {
                if wrapper.request.skip_general_revision() {
                    -1
                } else {
                    self.general_rev.next()
                }
            }
        };
        Ok(revision)
    }

    async fn execute(
        &self,
        cmd: &Command,
        index: LogIndex,
    ) -> Result<<Command as CurpCommand>::ER, Self::Error> {
        let wrapper = cmd.request();
        let res = match wrapper.request.backend() {
            RequestBackend::Kv => self.kv_storage.execute(wrapper),
            RequestBackend::Auth => self.auth_storage.execute(wrapper),
            RequestBackend::Lease => self.lease_storage.execute(wrapper),
        };
        match res {
            Ok(res) => Ok(res),
            Err(e) => {
                self.id_barrier.trigger(cmd.id());
                self.index_barrier.trigger(index);
                Err(e)
            }
        }
    }

    ...
}

[themanforfree]

state transition diagram
#254 (comment)