// Copyright 2016 TiKV Project Authors. Licensed under Apache-2.0. // Modifications Copyright 2022 PingCAP, Inc. use std::{ borrow::Cow, cell::UnsafeCell, fmt::{self, Debug, Display, Formatter}, io::Error as IoError, mem, num::NonZeroU64, ops::Deref, pin::Pin, result, sync::{ Arc, atomic::{AtomicU8, Ordering}, }, task::Poll, time::Duration, }; use collections::{HashMap, HashSet}; use concurrency_manager::ConcurrencyManager; use engine_traits::{CF_DEFAULT, CF_LOCK, CF_WRITE}; use futures::{Future, Stream, StreamExt, task::AtomicWaker}; use kvproto::{ errorpb, kvrpcpb::{Context, IsolationLevel}, raft_cmdpb::{ CmdType, CustomRequest, RaftCmdRequest, RaftCmdResponse, RaftRequestHeader, Request, Response, }, }; use raft::{ StateRole, eraftpb::{self, MessageType}, }; use raftstore::{ RegionInfoAccessor, coprocessor::{ Coprocessor, CoprocessorHost, ReadIndexObserver, RegionInfoProvider, dispatcher::BoxReadIndexObserver, }, }; use rfstore::{ Error as RaftServerError, RaftStoreRouter, store, store::{ Callback as StoreCallback, CustomBuilder, LocalReader, ReadIndexContext, ReadResponse, RegionSnapshot, WriteResponse, rlog, }, }; use thiserror::Error; use tikv::{ server::metrics::*, storage::{ self, kv::{ self, Engine, Error as KvError, ErrorInner as KvErrorInner, Modify, SnapContext, WriteData, }, }, }; use tikv_kv::{ExtraRegionOverride, OnAppliedCb, WriteEvent}; use tikv_util::{ callback::must_call, codec::number::NumberEncoder, future::paired_must_called_future_callback, time::Instant, }; use trace_event::types::TraceContext; use tracker::{GLOBAL_TRACKERS, TrackerToken, get_tls_tracker_token, set_tls_trace_ctx}; use txn_types::{ Key, LockType, ReqType, TxnExtra, TxnExtraScheduler, WriteBatchFlags, WriteRef, WriteType, }; #[derive(Debug, Error)] pub enum Error { #[error("{}", .0.get_message())] RequestFailed(Box), #[error("{0}")] Io(#[from] IoError), #[error("{0}")] Server(#[from] RaftServerError), #[error("{0}")] InvalidResponse(String), #[error("{0}")] InvalidRequest(String), #[error("{0}")] Undetermined(String), #[error("timeout after {0:?}")] Timeout(Duration), } fn get_status_kind_from_engine_error(e: &kv::Error) -> RequestStatusKind { match *e { KvError(box KvErrorInner::Request(ref header)) => { RequestStatusKind::from(storage::get_error_kind_from_header(header)) } KvError(box KvErrorInner::KeyIsLocked(_)) => { RequestStatusKind::err_leader_memory_lock_check } KvError(box KvErrorInner::Timeout(_)) => RequestStatusKind::err_timeout, KvError(box KvErrorInner::EmptyRequest) => RequestStatusKind::err_empty_request, KvError(box KvErrorInner::Undetermined(_)) => RequestStatusKind::err_undetermind, KvError(box KvErrorInner::Other(_)) => RequestStatusKind::err_other, } } pub type Result = result::Result; impl From for kv::Error { fn from(e: Error) -> kv::Error { match e { Error::RequestFailed(e) => KvError::from(KvErrorInner::Request(*e)), Error::Undetermined(e) => KvError::from(KvErrorInner::Undetermined(e)), Error::Server(e) => e.into(), e => box_err!(e), } } } #[inline] pub fn new_request_header( ctx: &Context, extra_snap_override: Option<&ExtraRegionOverride>, ) -> RaftRequestHeader { let mut header = RaftRequestHeader::default(); match extra_snap_override { Some(&ExtraRegionOverride { region_id, ref region_epoch, ref peer, check_term, }) => { header.set_region_id(region_id); header.set_peer(peer.clone()); header.set_region_epoch(region_epoch.clone()); if let Some(term) = check_term { header.set_term(term); } } _ => { header.set_region_id(ctx.get_region_id()); header.set_peer(ctx.get_peer().clone()); header.set_region_epoch(ctx.get_region_epoch().clone()); if ctx.get_term() != 0 { header.set_term(ctx.get_term()); } } } header.set_sync_log(ctx.get_sync_log()); header.set_replica_read(ctx.get_replica_read()); header } pub fn drop_snapshot_callback() -> kv::Result { let bt = backtrace::Backtrace::new(); warn!("async snapshot callback is dropped"; "backtrace" => ?bt); let mut err = errorpb::Error::default(); err.set_message("async snapshot callback is dropped".to_string()); Err(kv::Error::from(kv::ErrorInner::Request(err))) } struct WriteResCore { ev: AtomicU8, result: UnsafeCell>>, wake: AtomicWaker, } struct WriteResSub { notified_ev: u8, core: Arc, } unsafe impl Send for WriteResSub {} impl Stream for WriteResSub { type Item = WriteEvent; #[inline] fn poll_next( mut self: Pin<&mut Self>, cx: &mut std::task::Context<'_>, ) -> Poll> { let mut s = self.as_mut(); let mut cur_ev = s.core.ev.load(Ordering::Acquire); if cur_ev == s.notified_ev { s.core.wake.register(cx.waker()); cur_ev = s.core.ev.load(Ordering::Acquire); if cur_ev == s.notified_ev { return Poll::Pending; } } s.notified_ev = cur_ev; match cur_ev { WriteEvent::EVENT_PROPOSED => Poll::Ready(Some(WriteEvent::Proposed)), WriteEvent::EVENT_COMMITTED => Poll::Ready(Some(WriteEvent::Committed)), u8::MAX => { let result = unsafe { (*s.core.result.get()).take().unwrap() }; Poll::Ready(Some(WriteEvent::Finished(result))) } e => panic!("unexpected event {}", e), } } } #[derive(Clone)] struct WriteResFeed { core: Arc, } unsafe impl Send for WriteResFeed {} impl WriteResFeed { fn pair() -> (Self, WriteResSub) { let core = Arc::new(WriteResCore { ev: AtomicU8::new(0), result: UnsafeCell::new(None), wake: AtomicWaker::new(), }); ( Self { core: core.clone() }, WriteResSub { notified_ev: 0, core, }, ) } fn notify_proposed(&self) { self.core .ev .store(WriteEvent::EVENT_PROPOSED, Ordering::Release); self.core.wake.wake(); } fn notify_committed(&self) { self.core .ev .store(WriteEvent::EVENT_COMMITTED, Ordering::Release); self.core.wake.wake(); } fn notify(&self, result: kv::Result<()>) { unsafe { (*self.core.result.get()) = Some(result); } self.core.ev.store(u8::MAX, Ordering::Release); self.core.wake.wake(); } } /// `RaftKv` is a storage engine base on `RaftStore`. #[derive(Clone)] pub struct RaftKv { reader: LocalReader, region_info_accessor: Option, txn_extra_scheduler: Option>, } impl Deref for RaftKv { type Target = LocalReader; fn deref(&self) -> &Self::Target { &self.reader } } pub enum CmdRes { Resp(Vec), Snap(RegionSnapshot), } fn check_raft_cmd_response(resp: &mut RaftCmdResponse) -> Result<()> { if resp.get_header().has_error() { let mut err = resp.take_header().take_error(); if err.get_message() == ASYNC_WRITE_CALLBACK_DROPPED_ERR_MSG { return Err(Error::Undetermined(err.take_message())); } return Err(Error::RequestFailed(Box::from(err))); } Ok(()) } fn on_write_result(mut write_resp: WriteResponse) -> Result { check_raft_cmd_response(&mut write_resp.response)?; let resps = write_resp.response.take_responses(); Ok(CmdRes::Resp(resps.into())) } fn on_read_result(mut read_resp: ReadResponse) -> Result { check_raft_cmd_response(&mut read_resp.response)?; let resps = read_resp.response.take_responses(); if let Some(mut snapshot) = read_resp.snapshot { snapshot.term = NonZeroU64::new(read_resp.response.get_header().get_current_term()); snapshot.txn_extra_op = read_resp.txn_extra_op; Ok(CmdRes::Snap(snapshot)) } else { Ok(CmdRes::Resp(resps.into())) } } impl RaftKv { /// Create a RaftKv using specified configuration. pub fn new(reader: LocalReader, region_info_accessor: Option) -> RaftKv { RaftKv { reader, region_info_accessor, txn_extra_scheduler: None, } } pub fn set_txn_extra_scheduler(&mut self, txn_extra_scheduler: Arc) { self.txn_extra_scheduler = Some(txn_extra_scheduler); } fn new_request_header(&self, ctx: &Context) -> RaftRequestHeader { let mut header = RaftRequestHeader::default(); header.set_region_id(ctx.get_region_id()); header.set_peer(ctx.get_peer().clone()); header.set_region_epoch(ctx.get_region_epoch().clone()); if ctx.get_term() != 0 { header.set_term(ctx.get_term()); } header.set_sync_log(ctx.get_sync_log()); header.set_replica_read(ctx.get_replica_read()); header } } fn invalid_resp_type(exp: CmdType, act: CmdType) -> Error { Error::InvalidResponse(format!( "cmd type not match, want {:?}, got {:?}!", exp, act )) } pub const ASYNC_WRITE_CALLBACK_DROPPED_ERR_MSG: &str = "async write on_applied callback is dropped"; pub fn async_write_callback_dropped_err() -> errorpb::Error { let mut err = errorpb::Error::default(); err.set_message(ASYNC_WRITE_CALLBACK_DROPPED_ERR_MSG.to_string()); err } pub fn drop_on_applied_callback() -> WriteResponse { let bt = backtrace::Backtrace::new(); error!("async write on_applied callback is dropped"; "backtrace" => ?bt); let mut write_resp = WriteResponse { response: Default::default(), key_errors: Default::default(), }; write_resp .response .mut_header() .set_error(async_write_callback_dropped_err()); write_resp } impl Display for RaftKv { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { write!(f, "RaftKv") } } impl Debug for RaftKv { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { write!(f, "RaftKv") } } #[allow(dead_code)] impl Engine for RaftKv { type Snap = RegionSnapshot; type Local = kvengine::Engine; fn kv_engine(&self) -> Option { Some(self.kv_engine.clone()) } fn modify_on_kv_engine(&self, _modifies: HashMap>) -> kv::Result<()> { panic!(); } fn precheck_write_with_ctx(&self, ctx: &Context) -> tikv_kv::Result<()> { let region_id = ctx.get_region_id(); if let Some(region_info_accessor) = self.region_info_accessor.as_ref() { match region_info_accessor.find_region_by_id(region_id) { Some(region_info) if region_info.is_leader() => Ok(()), Some(region_info) => Err(RaftServerError::NotLeader( region_id, region_info.leader_peer().cloned(), ) .into()), None => Err(RaftServerError::NotLeader(region_id, None).into()), } } else { Ok(()) } } type WriteRes = impl Stream + Send + Unpin; fn async_write( &self, ctx: &Context, mut batch: WriteData, subscribed: u8, on_applied: Option, tracker: Option, ) -> Self::WriteRes { let mut res = (|| { fail_point!("raftkv_async_write"); if batch.modifies.is_empty() && batch.txn_file.is_none() { return Err(KvError::from(KvErrorInner::EmptyRequest)); } Ok(()) })(); ASYNC_REQUESTS_COUNTER_VEC.write.all.inc(); let begin_instant = Instant::now_coarse(); if res.is_ok() { // If rid is some, only the specified region reports error. // If rid is None, all regions report error. res = (|| { fail_point!("raftkv_early_error_report", |rid| { let region_id = ctx.get_region_id(); rid.and_then(|rid| { let rid: u64 = rid.parse().unwrap(); if rid == region_id { None } else { Some(()) } }) .ok_or_else(|| RaftServerError::RegionNotFound(region_id, None).into()) }); Ok(()) })(); } let req = modifies_to_requests(ctx, &mut batch); let txn_extra = batch.extra; let mut header = self.new_request_header(ctx); if txn_extra.one_pc { header.set_flags(WriteBatchFlags::ONE_PC.bits()); } let mut cmd = RaftCmdRequest::default(); cmd.set_header(header); cmd.set_custom_request(req); self.schedule_txn_extra(txn_extra); let trace_ctx = TraceContext::from_proto(ctx.get_trace_id(), ctx.get_trace_control_flags()); let (tx, rx) = WriteResFeed::pair(); if res.is_ok() { let region_id = ctx.get_region_id(); let create_cb_for_logging = trace_ctx.category_enabled(trace_event::types::Category::WriteDetails); let proposed_cb = if WriteEvent::subscribed_proposed(subscribed) { let tx = tx.clone(); let trace_ctx1 = trace_ctx.clone(); Some(Box::new(move || { txn_debug!(trace_ctx: trace_ctx1, trace_event::types::Category::WriteDetails, "RaftKv::async_write proposed_cb executed"; "region_id" => region_id, "tracker" => ?tracker ); tx.notify_proposed() }) as store::ExtCallback) } else if create_cb_for_logging { txn_debug!(trace_ctx: trace_ctx.clone(), trace_event::types::Category::WriteDetails, "RaftKv::async_write proposed_cb created (logging-only)"; "region_id" => region_id, "tracker" => ?tracker ); let trace_ctx1 = trace_ctx.clone(); Some(Box::new(move || { txn_debug!(trace_ctx: trace_ctx1, trace_event::types::Category::WriteDetails, "RaftKv::async_write proposed_cb executed (logging-only)"; "region_id" => region_id, "tracker" => ?tracker ); }) as store::ExtCallback) } else { None }; let committed_cb = if WriteEvent::subscribed_committed(subscribed) { let tx = tx.clone(); let trace_ctx1 = trace_ctx.clone(); Some(Box::new(move || { txn_debug!(trace_ctx: trace_ctx1, trace_event::types::Category::WriteDetails, "RaftKv::async_write committed_cb executed"; "region_id" => region_id, "tracker" => ?tracker ); tx.notify_committed() }) as store::ExtCallback) } else if create_cb_for_logging { txn_debug!(trace_ctx: trace_ctx.clone(), trace_event::types::Category::WriteDetails, "RaftKv::async_write committed_cb created (logging-only)"; "region_id" => region_id, "tracker" => ?tracker ); let trace_ctx1 = trace_ctx.clone(); Some(Box::new(move || { txn_debug!(trace_ctx: trace_ctx1, trace_event::types::Category::WriteDetails, "RaftKv::async_write committed_cb executed (logging-only)"; "region_id" => region_id, "tracker" => ?tracker ); }) as store::ExtCallback) } else { None }; let applied_tx = tx.clone(); txn_debug!(trace_event::types::Category::WriteDetails, "RaftKv::async_write applied_cb created"; "region_id" => region_id, "tracker" => ?tracker ); let applied_cb = must_call( Box::new(move |resp: WriteResponse| { // Set trace_ctx to tls so that it's possible to be retrieved in the inner cb. set_tls_trace_ctx(trace_ctx); txn_debug!(trace_event::types::Category::WriteDetails, "RaftKv::async_write applied_cb executed"; "region_id" => region_id, "resp" => ?resp, "tracker" => ?tracker ); fail_point!("applied_cb_return_undetermined_err", |_| { applied_tx.notify(Err(kv::Error::from(Error::Undetermined( ASYNC_WRITE_CALLBACK_DROPPED_ERR_MSG.to_string(), )))); }); let mut res = match on_write_result(resp) { Ok(CmdRes::Resp(_)) => { let elapse = begin_instant.saturating_elapsed_secs(); ASYNC_REQUESTS_COUNTER_VEC.write.success.inc(); ASYNC_REQUESTS_DURATIONS_VEC.write.observe(elapse); fail_point!("raftkv_async_write_finish"); Ok(()) } Ok(CmdRes::Snap(_)) => { Err(box_err!("unexpect snapshot, should mutate instead.")) } Err(e) => Err(kv::Error::from(e)), }; if let Some(cb) = on_applied { cb(&mut res); } applied_tx.notify(res); }), drop_on_applied_callback, ); let cb = StoreCallback::write_ext(applied_cb, proposed_cb, committed_cb); let trace_ctx = TraceContext::from_proto(ctx.get_trace_id(), ctx.get_trace_control_flags()); // TODO: do we need to support deadline? self.router.send_command(cmd, trace_ctx, cb); } else { // Note that `on_applied` is not called in this case. We send a message to the // channel here to notify the caller that the writing ended, like // how the `applied_cb` does. tx.notify(res); } rx.inspect(move |ev| { if let WriteEvent::Finished(Err(e)) = ev { let status_kind = get_status_kind_from_engine_error(e); ASYNC_REQUESTS_COUNTER_VEC.write.get(status_kind).inc(); } }) } type SnapshotRes = impl Future> + Send; fn async_snapshot(&mut self, mut ctx: SnapContext<'_>) -> Self::SnapshotRes { let res: kv::Result<()> = (|| { fail_point!("raftkv_async_snapshot_err", |_| { Err(box_err!("injected error for async_snapshot")) }); Ok(()) })(); let mut req = Request::default(); req.set_cmd_type(CmdType::Snap); if !ctx.key_ranges.is_empty() && ctx.start_ts.is_some_and(|ts| !ts.is_zero()) { req.mut_read_index() .set_start_ts(ctx.start_ts.as_ref().unwrap().into_inner()); req.mut_read_index() .set_key_ranges(mem::take(&mut ctx.key_ranges).into()); } ASYNC_REQUESTS_COUNTER_VEC.snapshot.all.inc(); let begin_instant = Instant::now_coarse(); let (cb, f) = paired_must_called_future_callback(drop_snapshot_callback); let mut header = new_request_header(ctx.pb_ctx, ctx.extra_region_override.as_ref()); let mut flags = 0; if ctx.pb_ctx.get_stale_read() && ctx.start_ts.is_none_or(|ts| !ts.is_zero()) { let mut data = [0u8; 8]; (&mut data[..]) .encode_u64(ctx.start_ts.unwrap_or_default().into_inner()) .unwrap(); flags |= WriteBatchFlags::STALE_READ.bits(); header.set_flag_data(data.into()); } header.set_flags(flags); let mut cmd = RaftCmdRequest::default(); cmd.set_header(header); cmd.set_requests(vec![req].into()); let tracker = get_tls_tracker_token(); if res.is_ok() { self.reader.read( ctx.read_id, cmd, StoreCallback::read(Box::new(move |resp| { let res = on_read_result(resp).map_err(Error::into); if res.is_ok() { let elapse = begin_instant.saturating_elapsed_secs(); GLOBAL_TRACKERS.with_tracker(tracker, |tracker| { if tracker.metrics.read_index_propose_wait_nanos > 0 { ASYNC_REQUESTS_DURATIONS_VEC .snapshot_read_index_propose_wait .observe( tracker.metrics.read_index_propose_wait_nanos as f64 / 1_000_000_000.0, ); // snapshot may be handled by lease read in raftstore if tracker.metrics.read_index_confirm_wait_nanos > 0 { ASYNC_REQUESTS_DURATIONS_VEC .snapshot_read_index_confirm .observe( tracker.metrics.read_index_confirm_wait_nanos as f64 / 1_000_000_000.0, ); } } else if tracker.metrics.local_read { ASYNC_REQUESTS_DURATIONS_VEC .snapshot_local_read .observe(elapse); } }); ASYNC_REQUESTS_DURATIONS_VEC.snapshot.observe(elapse); ASYNC_REQUESTS_COUNTER_VEC.snapshot.success.inc(); } cb(res); })), ); } async move { let res = match res { Ok(()) => f .await .map_err(|_| tikv_kv::ErrorInner::Other(box_err!("canceled")))?, Err(e) => Err(e), }; match res { Ok(CmdRes::Resp(mut r)) => { let e = if r .first() .map(|resp| resp.get_read_index().has_locked()) .unwrap_or(false) { let locked = r[0].take_read_index().take_locked(); KvError::from(KvErrorInner::KeyIsLocked(locked)) } else { invalid_resp_type(CmdType::Snap, r[0].get_cmd_type()).into() }; Err(e) } Ok(CmdRes::Snap(s)) => Ok(s), Err(e) => { let status_kind = get_status_kind_from_engine_error(&e); ASYNC_REQUESTS_COUNTER_VEC.snapshot.get(status_kind).inc(); Err(e) } } } } fn schedule_txn_extra(&self, txn_extra: TxnExtra) { if let Some(tx) = self.txn_extra_scheduler.as_ref() { if !txn_extra.is_empty() { tx.schedule(txn_extra); } } } fn get_kvengine(&self) -> Option { self.kv_engine() } } #[derive(Clone)] pub struct ReplicaReadLockChecker { concurrency_manager: ConcurrencyManager, } impl ReplicaReadLockChecker { pub fn new(concurrency_manager: ConcurrencyManager) -> Self { ReplicaReadLockChecker { concurrency_manager, } } pub fn register(self, host: &mut CoprocessorHost) { host.registry .register_read_index_observer(1, BoxReadIndexObserver::new(self)); } } impl Coprocessor for ReplicaReadLockChecker {} impl ReadIndexObserver for ReplicaReadLockChecker { fn on_step(&self, msg: &mut eraftpb::Message, role: StateRole) { // Only check and return result if the current peer is a leader. // If it's not a leader, the read index request will be redirected to the leader // later. if msg.get_msg_type() != MessageType::MsgReadIndex || role != StateRole::Leader { return; } assert_eq!(msg.get_entries().len(), 1); let mut rctx = ReadIndexContext::parse(msg.get_entries()[0].get_data()).unwrap(); if let Some(mut request) = rctx.request.take() { let begin_instant = Instant::now(); let start_ts = request.get_start_ts().into(); if let Err(e) = self .concurrency_manager .update_max_ts(start_ts, "read_index_observer") { error!("failed to update max ts: {:?}", e); } for range in request.mut_key_ranges().iter_mut() { let key_bound = |key: Vec| { if key.is_empty() { None } else { Some(txn_types::Key::from_encoded(key)) } }; let start_key = key_bound(range.take_start_key()); let end_key = key_bound(range.take_end_key()); let res = self.concurrency_manager.read_range_check( start_key.as_ref(), end_key.as_ref(), |key, lock| { txn_types::Lock::check_ts_conflict( Cow::Borrowed(lock), key, start_ts, &Default::default(), IsolationLevel::Si, ) }, ); if let Err(txn_types::Error(box txn_types::ErrorInner::KeyIsLocked(lock))) = res { rctx.locked = Some(lock); REPLICA_READ_LOCK_CHECK_HISTOGRAM_VEC_STATIC .locked .observe(begin_instant.saturating_elapsed().as_secs_f64()); } else { REPLICA_READ_LOCK_CHECK_HISTOGRAM_VEC_STATIC .unlocked .observe(begin_instant.saturating_elapsed().as_secs_f64()); } } msg.mut_entries()[0].set_data(rctx.to_bytes().into()); } } } /// Reconstruct transaction operations from separated modifications. /// /// pessimistic_lock /// modify::put lock(pessimistic) /// /// prewrite /// optional modify::put cf_default /// | modify::put cf_lock /// /// 1pc /// optional modify::put cf_default /// optional modify::del cf_lock /// | modify::put cf_write:put /// | modify::put cf_write:del /// /// commit /// modify:put cf_write + modify::del cf_lock /// note: if a commit batch also needs to clean up stale pessimistic locks, the /// command is routed through ResolveLock so it can reuse the existing /// delete-lock-only rollback entry format. See upstream tikv/tikv#14557. /// /// rollback /// modify:del cf_lock /// | modify::put cf_write::put overlapped + modify:del cf_lock + /// optional modify::del cf_write(collapse rollback) /// | modify::put cf_write::rollback + /// optional modify::del cf_write(collapse rollback) /// | modify::put cf_write::rollback + modify:del cf_lock + /// optional modify::del cf_write(collapse rollback) /// | modify::put cf_lock(has rollback_ts) + modify:put cf_write:rollback + /// optional modify::del cf_write(collapse rollback) /// /// pessimistic_rollback /// modify::del cf_lock(pessimistic) /// /// check_txn_status /// | pessimistic_lock(push min_commit_ts) /// | prewrite(push min_commit_ts) /// | pessimistic_rollback /// | rollback /// /// CheckSecondaryLocks /// rollback /// /// ResolveLock /// commit and rollback /// /// Heartbeat /// modify::put cf_lock(optimistc or pessimistic) pub fn modifies_to_requests(_ctx: &Context, data: &mut WriteData) -> CustomRequest { let builder = &mut rlog::CustomBuilder::new(); let modifies = std::mem::take(&mut data.modifies); let shared_lock_lifecycle = data.extra.shared_lock_lifecycle; if let Some(txn_file) = data.txn_file.take() { builder.set_txn_file(&txn_file.txn_file_ref); return builder.build(); } if data.extra.one_pc { build_one_pc(builder, modifies); } else { match data.extra.req_type { ReqType::PessimisticLock => build_pessimistic_lock(builder, modifies), ReqType::Prewrite => build_prewrite(builder, modifies), ReqType::Commit => build_commit(builder, modifies, shared_lock_lifecycle), ReqType::PessimisticRollback => { build_pessimistic_rollback(builder, modifies, shared_lock_lifecycle) } ReqType::CheckTxnStatus => { build_check_txn_status(builder, modifies, shared_lock_lifecycle) } ReqType::CheckSecondaryLocks | ReqType::Rollback => { build_rollback(builder, modifies, shared_lock_lifecycle) } ReqType::ResolveLock => build_resolve_lock(builder, modifies, shared_lock_lifecycle), ReqType::Heartbeat => build_heartbeat(builder, modifies, shared_lock_lifecycle), ReqType::Noop => unreachable!("modifies: {:?}", modifies), } } builder.build() } fn append_shared_lock_modify(builder: &mut CustomBuilder, modify: Modify) { match modify { Modify::Put(CF_WRITE, key, value) => { let commit_ts = key.decode_ts().unwrap().into_inner(); let raw_key = key.into_raw().unwrap(); builder.append_shared_lock_modify_write_put(&raw_key, commit_ts, &value); } Modify::Put(CF_LOCK, key, value) => { let raw_key = key.into_raw().unwrap(); builder.append_shared_lock_modify_lock_put(&raw_key, &value); } Modify::Delete(CF_LOCK, key) => { let raw_key = key.into_raw().unwrap(); builder.append_shared_lock_modify_lock_delete(&raw_key); } Modify::Delete(CF_WRITE, key) => { debug_assert!( false, "shared-lock modifies should not contain Delete(CF_WRITE): {:?}", key ); error!( "unexpected shared-lock modify contains Delete(CF_WRITE); cloud engine ignores it because rollback metadata lives in EXTRA_CF"; "key" => %key ); // Cloud engine keeps rollback metadata in EXTRA_CF, so even if a // stray Write CF delete appears here, ignoring it is still safe. } _ => unreachable!("unexpected shared-lock modify: {:?}", modify), } } fn build_shared_lock_modifies(builder: &mut CustomBuilder, modifies: Vec) { builder.set_type(rlog::CustomRaftLogType::SharedLockModifies); let mut lock_cf_modifies = Vec::new(); // SharedLockModifies is a replay format for commands that may update both // Write CF records and the shared-lock wrapper in Lock CF. It is not // classifying whether a user mutation is a "put"; `Put(CF_WRITE)` means an // MVCC write record is being appended. // // Keep all Write CF puts before Lock CF changes. This preserves the apply // order when one batch contains both a commit/rollback write record and a // shared-lock wrapper update for the same raw key. for modify in modifies { match &modify { Modify::Put(CF_WRITE, ..) => { append_shared_lock_modify(builder, modify); } Modify::Put(CF_LOCK, ..) | Modify::Delete(CF_LOCK, ..) => { lock_cf_modifies.push(modify); } Modify::Delete(CF_WRITE, _) => { debug_assert!( false, "shared-lock modifies should not contain Delete(CF_WRITE): {:?}", modify ); append_shared_lock_modify(builder, modify); } _ => { debug_assert!( false, "unexpected shared-lock modify before ordering: {:?}", modify ); append_shared_lock_modify(builder, modify); } } } for modify in lock_cf_modifies { debug_assert!( matches!( &modify, Modify::Put(CF_LOCK, ..) | Modify::Delete(CF_LOCK, ..) ), "shared-lock deferred modifies should only contain Lock CF updates: {:?}", modify ); append_shared_lock_modify(builder, modify); } } fn build_pessimistic_lock(builder: &mut CustomBuilder, modifies: Vec) { builder.set_type(rlog::CustomRaftLogType::PessimisticLock); for m in modifies { match m { Modify::PessimisticLock(key, lock) => { let raw_key = key.into_raw().unwrap(); let val = lock.into_lock().to_bytes(); builder.append_lock(&raw_key, &val); } Modify::Put(CF_LOCK, key, val) => { let raw_key = key.into_raw().unwrap(); builder.append_lock(&raw_key, &val); } _ => unreachable!("unexpected modify: {:?}", m), } } } fn build_prewrite(builder: &mut CustomBuilder, mut modifies: Vec) { builder.set_type(rlog::CustomRaftLogType::Prewrite); let mut default_vals = modifies .iter_mut() .filter_map(|m| { if let Modify::Put(CF_DEFAULT, k, v) = m { Some((k.to_raw().unwrap(), mem::take(v))) } else { None } }) .collect::>(); for m in modifies { match m { Modify::Put(CF_LOCK, key, mut val) => { let raw_key = key.into_raw().unwrap(); if let Some(default_val) = default_vals.remove(&raw_key) { let mut lock = match txn_types::parse_lock(&val).unwrap() { tikv_util::Either::Left(lock) => lock, tikv_util::Either::Right(_) => { panic!( "unexpected shared-lock wrapper in prewrite lock value for key {:?}; shared-lock prewrite should not carry default CF payload", raw_key ); } }; lock.short_value = Some(default_val); val = lock.to_bytes(); } builder.append_lock(&raw_key, &val); } Modify::Put(CF_DEFAULT, ..) => {} _ => unreachable!("unexpected modify {:?}", m), } } } // Append the legacy compact per-key commit entries. A top-level `Commit` log // and a top-level `ResolveLock` log both reuse this exact sub-entry encoding // for real committed writes, so apply sees the same `Commit { key, commit_ts }` // item format in both cases. fn append_compact_commit_entries(builder: &mut CustomBuilder, modifies: &[Modify]) { for modify in modifies { match modify { Modify::Put(CF_WRITE, key, _) => { let commit_ts = key.decode_ts().unwrap(); let raw_key = key.to_raw().unwrap(); builder.append_commit(&raw_key, commit_ts.into_inner()); // value will be fetched during applying. } Modify::Delete(CF_LOCK, _) => { // lock will be deleted during applying. } _ => unreachable!("unexpected modify: {:?}", modify), } } } fn build_commit(builder: &mut CustomBuilder, modifies: Vec, shared_lock_lifecycle: bool) { if shared_lock_lifecycle { build_shared_lock_modifies(builder, modifies); return; } // In the legacy compact `Commit` format, only `Put(CF_WRITE)` is persisted. // The matching `Delete(CF_LOCK)` is implicit because apply replays the entry // through `commit_lock()`, which writes CF_WRITE and deletes CF_LOCK as one // operation. // builder.set_type(rlog::CustomRaftLogType::Commit); append_compact_commit_entries(builder, &modifies); } fn build_one_pc(builder: &mut CustomBuilder, mut modifies: Vec) { builder.set_type(rlog::CustomRaftLogType::OnePc); let deleted_lock = modifies .iter_mut() .filter_map(|m| { if let Modify::Delete(CF_LOCK, k) = m { Some(mem::replace(k, Key::from_encoded(vec![]))) } else { None } }) .collect::>(); let mut default_vals = modifies .iter_mut() .filter_map(|m| { if let Modify::Put(CF_DEFAULT, k, v) = m { Some((k.to_raw().unwrap(), mem::take(v))) } else { None } }) .collect::>(); for m in modifies { match m { Modify::Put(CF_WRITE, key, val) => { let commit_ts = key.decode_ts().unwrap().into_inner(); let key = key.truncate_ts().unwrap(); let del_lock = deleted_lock.contains(&key); let raw_key = key.into_raw().unwrap(); let write = WriteRef::parse(&val).unwrap(); let mut value = vec![]; if write.write_type == WriteType::Put { value = default_vals .remove(&raw_key) .unwrap_or_else(|| write.short_value.unwrap().to_vec()); } let is_extra = write.write_type == WriteType::Lock; let start_ts = write.start_ts.into_inner(); builder.append_one_pc(&raw_key, &value, is_extra, del_lock, start_ts, commit_ts); } Modify::Delete(CF_LOCK, _) | Modify::Put(CF_DEFAULT, ..) => {} _ => unreachable!("unexpected modify: {:?}", m), } } } fn build_pessimistic_rollback( builder: &mut CustomBuilder, modifies: Vec, shared_lock_lifecycle: bool, ) { if shared_lock_lifecycle { build_shared_lock_modifies(builder, modifies); return; } builder.set_type(rlog::CustomRaftLogType::PessimisticRollback); for modify in modifies { match modify { Modify::Delete(CF_LOCK, key) => { let raw_key = key.into_raw().unwrap(); builder.append_del_lock(&raw_key); } _ => unreachable!("unexpected modify: {:?}", modify), } } } fn build_rollback(builder: &mut CustomBuilder, modifies: Vec, shared_lock_lifecycle: bool) { if shared_lock_lifecycle { build_shared_lock_modifies(builder, modifies); return; } builder.set_type(rlog::CustomRaftLogType::Rollback); let mut modifies = modifies; let mut deleted_locks = modifies .iter_mut() .filter_map(|modify| { if let Modify::Delete(CF_LOCK, key) = modify { Some(mem::replace(key, Key::from_encoded(vec![]))) } else { None } }) .collect::>(); for modify in modifies { match modify { Modify::Put(CF_WRITE, key, _) => { let start_ts = key.decode_ts().unwrap().into_inner(); let key = key.truncate_ts().unwrap(); let del_lock = deleted_locks.remove(&key); let raw_key = key.into_raw().unwrap(); builder.append_rollback(&raw_key, start_ts, del_lock); } Modify::Delete(CF_LOCK, _) => { // Lock deletion is carried on putting write. } Modify::Put(CF_LOCK, ..) => { // The EXTRA_CF ensures no overlapped rollback and commit // record, so we ignore them. } Modify::Delete(CF_WRITE, _) => { // No need to collapse rollback write due to the EXTRA_CF. } _ => unreachable!("unexpected modify: {:?}", modify), } } for delete_only in deleted_locks { let raw_key = delete_only.into_raw().unwrap(); builder.append_rollback(&raw_key, 0, true); } } fn build_check_txn_status( builder: &mut CustomBuilder, modifies: Vec, shared_lock_lifecycle: bool, ) { if shared_lock_lifecycle { build_shared_lock_modifies(builder, modifies); return; } let tp = match &modifies[0] { Modify::Put(CF_LOCK, _, v) => match non_shared_lock_type(v, "CheckTxnStatus") { LockType::Pessimistic => rlog::CustomRaftLogType::PessimisticLock, _ => rlog::CustomRaftLogType::Prewrite, }, Modify::Delete(CF_LOCK, _) => { if modifies.len() == 1 { rlog::CustomRaftLogType::PessimisticRollback } else { rlog::CustomRaftLogType::Rollback } } Modify::Put(CF_WRITE, ..) | Modify::Delete(CF_WRITE, ..) => { rlog::CustomRaftLogType::Rollback } _ => unreachable!("unexpected modifies: {:?}", modifies), }; match tp { rlog::CustomRaftLogType::PessimisticLock => build_pessimistic_lock(builder, modifies), rlog::CustomRaftLogType::Prewrite => build_prewrite(builder, modifies), rlog::CustomRaftLogType::PessimisticRollback => { build_pessimistic_rollback(builder, modifies, false) } rlog::CustomRaftLogType::Rollback => build_rollback(builder, modifies, false), _ => unreachable!(), }; } fn build_resolve_lock( builder: &mut CustomBuilder, modifies: Vec, shared_lock_lifecycle: bool, ) { if shared_lock_lifecycle { build_shared_lock_modifies(builder, modifies); return; } let mut grouped = HashMap::, Vec<_>>::default(); for modify in modifies { let raw_key = modify.key().to_raw().unwrap(); grouped.entry(raw_key).or_default().push(modify); } for (_, key_modifies) in grouped { debug_assert!( key_modifies.len() <= 2, "unexpected ResolveLock modifies for one key: {:?}", key_modifies ); debug_assert!( !is_reversed_resolve_lock_commit_group(&key_modifies), "unexpected reversed ResolveLock commit group: {:?}", key_modifies ); if key_modifies.len() == 2 { if let Modify::Put(CF_WRITE, _, value) = &key_modifies[0] { let write = WriteRef::parse(value).unwrap(); if write.write_type != WriteType::Rollback { assert!(matches!(&key_modifies[1], Modify::Delete(CF_LOCK, _))); builder.append_type(rlog::CustomRaftLogType::Commit); // Reuse the compact Commit sub-entry inside ResolveLock so mixed // commit and stale-lock cleanup batches do not need a new persisted // log format. append_compact_commit_entries(builder, &key_modifies); continue; } } } builder.append_type(rlog::CustomRaftLogType::Rollback); build_rollback(builder, key_modifies, false); } builder.set_type(rlog::CustomRaftLogType::ResolveLock); } fn is_reversed_resolve_lock_commit_group(modifies: &[Modify]) -> bool { if modifies.len() != 2 || !matches!(&modifies[0], Modify::Delete(CF_LOCK, _)) { return false; } match &modifies[1] { Modify::Put(CF_WRITE, _, v) => { WriteRef::parse(v).unwrap().write_type != WriteType::Rollback } _ => false, } } fn build_heartbeat( builder: &mut CustomBuilder, modifies: Vec, shared_lock_lifecycle: bool, ) { if shared_lock_lifecycle { build_shared_lock_modifies(builder, modifies); return; } let tp = match &modifies[0] { Modify::Put(CF_LOCK, _, v) => match non_shared_lock_type(v, "Heartbeat") { LockType::Pessimistic => rlog::CustomRaftLogType::PessimisticLock, _ => rlog::CustomRaftLogType::Prewrite, }, _ => unreachable!("unexpected modifies: {:?}", modifies), }; match tp { rlog::CustomRaftLogType::PessimisticLock => build_pessimistic_lock(builder, modifies), rlog::CustomRaftLogType::Prewrite => build_prewrite(builder, modifies), _ => unreachable!(), }; } fn non_shared_lock_type(value: &[u8], req_name: &str) -> LockType { match txn_types::parse_lock(value).unwrap_or_else(|err| { panic!( "failed to parse {} lock bytes for cloud custom log: {:?}", req_name, err ) }) { tikv_util::Either::Left(lock) => lock.lock_type, tikv_util::Either::Right(_) => { panic!( "shared-lock bytes without shared-lock provenance in {} builder", req_name ) } } } #[cfg(test)] mod tests { use std::{ iter::FromIterator, sync::{Mutex, mpsc}, }; use kvproto::kvrpcpb::PrewriteRequestPessimisticAction::*; use tikv_kv::BTreeEngine; use tracker::TrackerToken; use txn_types::{SHORT_VALUE_MAX_LEN, Write}; use uuid::Uuid; use super::*; // This test ensures `ReplicaReadLockChecker` won't change UUID context of read // index. #[test] fn test_replica_read_lock_checker_for_single_uuid() { let cm = ConcurrencyManager::new(1.into()); let checker = ReplicaReadLockChecker::new(cm); let mut m = eraftpb::Message::default(); m.set_msg_type(MessageType::MsgReadIndex); let uuid = Uuid::new_v4(); let mut e = eraftpb::Entry::default(); e.set_data(uuid.as_bytes().to_vec().into()); m.mut_entries().push(e); checker.on_step(&mut m, raft::StateRole::Leader); assert_eq!(m.get_entries()[0].get_data(), uuid.as_bytes()); } // It saves the write data of the last request for testing custom raft log. #[derive(Clone)] struct MockEngine { base: BTreeEngine, last_write_data: Arc>>, } impl MockEngine { fn new(base: BTreeEngine) -> Self { Self { base, last_write_data: Arc::default(), } } fn take_last_write_data(&self) -> Option { self.last_write_data.lock().unwrap().take() } fn set_write_data(&self, data: &WriteData) { let clone = WriteData::new(data.modifies.clone(), data.extra.clone(), None); *self.last_write_data.lock().unwrap() = Some(clone); } } impl Engine for MockEngine { type Snap = ::Snap; type Local = ::Local; fn kv_engine(&self) -> Option { self.base.kv_engine() } fn modify_on_kv_engine(&self, modifies: HashMap>) -> kv::Result<()> { self.base.modify_on_kv_engine(modifies) } type SnapshotRes = ::SnapshotRes; fn async_snapshot(&mut self, ctx: SnapContext<'_>) -> Self::SnapshotRes { self.base.async_snapshot(ctx) } type WriteRes = ::WriteRes; fn async_write( &self, ctx: &Context, batch: WriteData, subscribed: u8, on_applied: Option, tracker: Option, ) -> Self::WriteRes { self.set_write_data(&batch); self.base .async_write(ctx, batch, subscribed, on_applied, tracker) } } #[test] #[ignore] fn test_custom_raft_log() { use kvproto::kvrpcpb::AssertionLevel; use tikv::storage::{ TestStorageBuilderApiV1, kv::TestEngineBuilder, lock_manager::MockLockManager, test_util::expect_ok_callback, txn::commands, }; use txn_types::Mutation; let engine = MockEngine::new(TestEngineBuilder::new().build().unwrap()); let storage = TestStorageBuilderApiV1::from_engine_and_lock_mgr( engine.clone(), MockLockManager::new(), ) .build() .unwrap(); let (tx, rx) = mpsc::channel(); let get_modify_value = |modify: &_| match modify { Modify::Put(_, _, v) => v.clone(), Modify::PessimisticLock(_, lock) => lock.to_lock().to_bytes(), _ => unreachable!("unexpected modify: {:?}", modify), }; let prewrite = |mutations, primary: &[_], ts: u64, secondaries, one_pc, max_commit_ts: u64| { storage .sched_txn_command( commands::Prewrite::new( mutations, primary.to_vec(), ts.into(), 3000, false, 0, (ts + 1).into(), max_commit_ts.into(), secondaries, one_pc, AssertionLevel::Off, vec![], Context::default(), ), expect_ok_callback(tx.clone(), 1), ) .unwrap(); rx.recv().unwrap(); engine.take_last_write_data().unwrap() }; let pessimistic_prewrite = |mutations, primary: &[_], ts: u64, one_pc, max_commit_ts: u64| { storage .sched_txn_command( commands::PrewritePessimistic::new( mutations, primary.to_vec(), ts.into(), 3000, ts.into(), 0, (ts + 1).into(), max_commit_ts.into(), None, one_pc, AssertionLevel::Off, vec![], Context::default(), ), expect_ok_callback(tx.clone(), 1), ) .unwrap(); rx.recv().unwrap(); engine.take_last_write_data().unwrap() }; let pessimistic_lock = |keys: Vec<_>, primary: &[_], ts: u64| { storage .sched_txn_command( commands::AcquirePessimisticLock::new( keys.into_iter() .map(|k| (Key::from_raw(k), false, false)) .collect(), primary.to_vec(), ts.into(), 3000, false, ts.into(), None, false, (ts + 1).into(), Default::default(), false, false, Context::default(), ), expect_ok_callback(tx.clone(), 1), ) .unwrap(); rx.recv().unwrap(); engine.take_last_write_data().unwrap() }; // Prewrite let mut data = prewrite( vec![ Mutation::make_put(Key::from_raw(b"k1"), b"v1".to_vec()), Mutation::make_put(Key::from_raw(b"k2"), b"v".repeat(SHORT_VALUE_MAX_LEN + 1)), ], b"k1", 10, None, false, 0, ); let modifies = data.modifies.clone(); assert_eq!(data.extra.req_type, ReqType::Prewrite); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::Prewrite); custom_log.iterate_lock(|k, v| match k { b"k1" => { assert_eq!(k, modifies[0].key().to_raw().unwrap()); assert_eq!(v, &get_modify_value(&modifies[0])); } b"k2" => { assert_eq!(k, modifies[2].key().to_raw().unwrap()); let default_val = get_modify_value(&modifies[1]); let mut lock = txn_types::parse_non_shared_lock(&get_modify_value(&modifies[2])).unwrap(); lock.short_value = Some(default_val); assert_eq!(v, lock.to_bytes()); } _ => unreachable!(), }); // Commit storage .sched_txn_command( commands::Commit::new( vec![Key::from_raw(b"k1"), Key::from_raw(b"k2")], 10.into(), 20.into(), false, false, Context::default(), ), expect_ok_callback(tx.clone(), 1), ) .unwrap(); rx.recv().unwrap(); let mut data = engine.take_last_write_data().unwrap(); assert_eq!(data.extra.req_type, ReqType::Commit); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::Commit); let mut keys = [b"k1", b"k2"].iter(); custom_log.iterate_commit(|k, ts| { assert_eq!(k, keys.next().unwrap().as_slice()); assert_eq!(ts, 20); }); assert!(keys.next().is_none()); // 1PC let mut data = prewrite( vec![ Mutation::make_put(Key::from_raw(b"k1"), b"v1".to_vec()), Mutation::make_put(Key::from_raw(b"k2"), b"v".repeat(SHORT_VALUE_MAX_LEN + 1)), ], b"k1", 30, None, true, 40, ); assert!(data.extra.one_pc); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::OnePc); let mut cnt = 0; custom_log.iterate_one_pc(|k, v, is_extra, del_lock, start_ts, commit_ts| { match k { b"k1" => assert_eq!(v, b"v1"), b"k2" => assert_eq!(v, &b"v".repeat(SHORT_VALUE_MAX_LEN + 1)), _ => unreachable!(), } assert!(!is_extra); assert!(!del_lock); assert_eq!(start_ts, 30); assert_eq!(commit_ts, 31); cnt += 1; }); assert_eq!(cnt, 2); // PessimisticLock let mut data = pessimistic_lock(vec![b"k1"], b"k1", 40); let modifies = data.modifies.clone(); assert_eq!(data.extra.req_type, ReqType::PessimisticLock); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!( custom_log.get_type(), rlog::CustomRaftLogType::PessimisticLock ); let mut cnt = 0; custom_log.iterate_lock(|k, v| { assert_eq!(k, &modifies[0].key().to_raw().unwrap()); assert_eq!(v, &get_modify_value(&modifies[0])); cnt += 1; }); assert_eq!(cnt, 1); // Pessimistic 1PC let mut data = pessimistic_prewrite( vec![ ( Mutation::make_put(Key::from_raw(b"k1"), b"v1".to_vec()), DoPessimisticCheck, ), ( Mutation::make_lock(Key::from_raw(b"k2")), SkipPessimisticCheck, ), ], b"k1", 40, true, 50, ); assert!(data.extra.one_pc); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::OnePc); let mut cnt = 0; custom_log.iterate_one_pc(|k, v, is_extra, del_lock, start_ts, commit_ts| { match k { b"k1" => { assert_eq!(v, b"v1"); assert!(!is_extra); assert!(del_lock); } b"k2" => { assert!(v.is_empty()); assert!(is_extra); assert!(!del_lock); } _ => unreachable!("unexpected key: {:?}", k), }; assert_eq!(start_ts, 40); assert_eq!(commit_ts, 41); cnt += 1; }); assert_eq!(cnt, 2); // ResolveLock prewrite( vec![ Mutation::make_put(Key::from_raw(b"k1"), b"v1".to_vec()), Mutation::make_put(Key::from_raw(b"k2"), b"v2".to_vec()), ], b"k1", 50, None, false, 0, ); prewrite( vec![ Mutation::make_put(Key::from_raw(b"k3"), b"v3".to_vec()), Mutation::make_put(Key::from_raw(b"k4"), b"v4".to_vec()), ], b"k3", 60, None, false, 0, ); let txn_status = HashMap::from_iter( [(50.into(), 51.into()), (60.into(), 0.into())] .iter() .cloned(), ); storage .sched_txn_command( commands::ResolveLockReadPhase::new(txn_status, None, false, Context::default()), expect_ok_callback(tx.clone(), 1), ) .unwrap(); rx.recv().unwrap(); let mut data = engine.take_last_write_data().unwrap(); assert_eq!(data.extra.req_type, ReqType::ResolveLock); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::ResolveLock); let mut expected = HashMap::from_iter(vec![ ( b"k1".as_slice(), (rlog::CustomRaftLogType::Commit, b"k1".as_slice(), 51, true), ), ( b"k2".as_slice(), (rlog::CustomRaftLogType::Commit, b"k2".as_slice(), 51, true), ), ( b"k3".as_slice(), ( rlog::CustomRaftLogType::Rollback, b"k3".as_slice(), 60, true, ), ), ( b"k4".as_slice(), ( rlog::CustomRaftLogType::Rollback, b"k4".as_slice(), 60, true, ), ), ]); custom_log.iterate_resolve_lock(|tp, k, ts, del_lock| { assert_eq!(expected.remove(k).unwrap(), (tp, k, ts, del_lock)); }); // 1PC fallback let mut data = prewrite( vec![Mutation::make_put(Key::from_raw(b"k1"), b"v1".to_vec())], b"k1", 70, None, true, 1, ); assert_eq!(data.extra.req_type, ReqType::Prewrite); let modifies = data.modifies.clone(); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::Prewrite); let mut cnt = 0; custom_log.iterate_lock(|k, v| { assert_eq!(k, modifies[0].key().to_raw().unwrap()); assert_eq!(v, &get_modify_value(&modifies[0])); cnt += 1; }); assert_eq!(cnt, 1); // Rollback storage .sched_txn_command( commands::Rollback::new( vec![Key::from_raw(b"k1"), Key::from_raw(b"k2")], 70.into(), false, Context::default(), ), expect_ok_callback(tx.clone(), 1), ) .unwrap(); rx.recv().unwrap(); let mut data = engine.take_last_write_data().unwrap(); assert_eq!(data.extra.req_type, ReqType::Rollback); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::Rollback); let mut cnt = 0; custom_log.iterate_rollback(|k, ts, del_lock| { match k { b"k1" => assert!(del_lock), b"k2" => assert!(!del_lock), _ => unreachable!(), }; assert_eq!(ts, 70); cnt += 1 }); assert_eq!(cnt, 2); // PessimisticRollback pessimistic_lock(vec![b"k1"], b"k1", 80); storage .sched_txn_command( commands::PessimisticRollback::new( vec![Key::from_raw(b"k1")], 80.into(), 80.into(), None, Context::default(), ), expect_ok_callback(tx.clone(), 1), ) .unwrap(); rx.recv().unwrap(); let mut data = engine.take_last_write_data().unwrap(); assert_eq!(data.extra.req_type, ReqType::PessimisticRollback); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!( custom_log.get_type(), rlog::CustomRaftLogType::PessimisticRollback ); let mut cnt = 0; custom_log.iterate_del_lock(|k| { assert_eq!(k, b"k1"); cnt += 1 }); assert_eq!(cnt, 1); // Heartbeat pessimistic_lock(vec![b"k1"], b"k1", 90); storage .sched_txn_command( commands::TxnHeartBeat::new( Key::from_raw(b"k1"), 90.into(), 10000, false, Context::default(), ), expect_ok_callback(tx.clone(), 1), ) .unwrap(); rx.recv().unwrap(); let mut data = engine.take_last_write_data().unwrap(); assert_eq!(data.extra.req_type, ReqType::Heartbeat); let modifies = data.modifies.clone(); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!( custom_log.get_type(), rlog::CustomRaftLogType::PessimisticLock ); let mut cnt = 0; custom_log.iterate_lock(|k, v| { assert_eq!(k, &modifies[0].key().to_raw().unwrap()); assert_eq!(v, &get_modify_value(&modifies[0])); cnt += 1 }); assert_eq!(cnt, 1); pessimistic_prewrite( vec![( Mutation::make_delete(Key::from_raw(b"k1")), DoPessimisticCheck, )], b"k1", 90, false, 0, ); storage .sched_txn_command( commands::TxnHeartBeat::new( Key::from_raw(b"k1"), 90.into(), 20000, false, Context::default(), ), expect_ok_callback(tx.clone(), 1), ) .unwrap(); rx.recv().unwrap(); let mut data = engine.take_last_write_data().unwrap(); assert_eq!(data.extra.req_type, ReqType::Heartbeat); let modifies = data.modifies.clone(); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::Prewrite); let mut cnt = 0; custom_log.iterate_lock(|k, v| { assert_eq!(k, &modifies[0].key().to_raw().unwrap()); assert_eq!(v, &get_modify_value(&modifies[0])); cnt += 1 }); assert_eq!(cnt, 1); // CheckTxnStatus let check_txn_status = |key, lock_ts: u64, caller_start_ts: u64, resolving_pessimistic_lock| { storage .sched_txn_command( commands::CheckTxnStatus::new( Key::from_raw(key), lock_ts.into(), caller_start_ts.into(), caller_start_ts.into(), true, true, resolving_pessimistic_lock, false, true, Context::default(), ), expect_ok_callback(tx.clone(), 1), ) .unwrap(); rx.recv().unwrap(); engine.take_last_write_data().unwrap() }; // Push lock's min_commit_ts let mut data = check_txn_status(b"k1", 90, 95, false); assert_eq!(data.extra.req_type, ReqType::CheckTxnStatus); let modifies = data.modifies.clone(); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::Prewrite); let mut cnt = 0; custom_log.iterate_lock(|k, v| { assert_eq!(k, &modifies[0].key().to_raw().unwrap()); assert_eq!(v, &get_modify_value(&modifies[0])); cnt += 1 }); assert_eq!(cnt, 1); // Rollback lock let mut data = check_txn_status(b"k1", 90, u64::MAX, false); assert_eq!(data.extra.req_type, ReqType::CheckTxnStatus); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::Rollback); let mut cnt = 0; custom_log.iterate_rollback(|k, ts, del_lock| { assert_eq!(k, b"k1"); assert!(del_lock); assert_eq!(ts, 90); cnt += 1 }); assert_eq!(cnt, 1); // Rollback a missing lock and there is a lock of another async-commit // transaction. prewrite( vec![Mutation::make_put(Key::from_raw(b"k1"), b"v1".to_vec())], b"k1", 95, Some(vec![b"k2".to_vec()]), false, 100, ); let mut data = check_txn_status(b"k1", 96, 97, false); assert_eq!(data.extra.req_type, ReqType::CheckTxnStatus); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::Rollback); let mut cnt = 0; custom_log.iterate_rollback(|k, ts, del_lock| { assert_eq!(k, b"k1"); assert!(!del_lock); assert_eq!(ts, 96); cnt += 1 }); assert_eq!(cnt, 1); check_txn_status(b"k1", 95, u64::MAX, false); pessimistic_lock(vec![b"k1"], b"k1", 100); // Push pessimistic lock's min_commit_ts let mut data = check_txn_status(b"k1", 100, 110, false); assert_eq!(data.extra.req_type, ReqType::CheckTxnStatus); let modifies = data.modifies.clone(); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!( custom_log.get_type(), rlog::CustomRaftLogType::PessimisticLock ); let mut cnt = 0; custom_log.iterate_lock(|k, v| { assert_eq!(k, &modifies[0].key().to_raw().unwrap()); assert_eq!(v, &get_modify_value(&modifies[0])); cnt += 1 }); assert_eq!(cnt, 1); // Rollback pessimistic lock let mut data = check_txn_status(b"k1", 100, u64::MAX, true); assert_eq!(data.extra.req_type, ReqType::CheckTxnStatus); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!( custom_log.get_type(), rlog::CustomRaftLogType::PessimisticRollback ); let mut cnt = 0; custom_log.iterate_del_lock(|k| { assert_eq!(k, b"k1"); cnt += 1 }); assert_eq!(cnt, 1); // CheckSecondaryLocks storage .sched_txn_command( commands::CheckSecondaryLocks::new( vec![Key::from_raw(b"k1")], 110.into(), Context::default(), ), expect_ok_callback(tx, 1), ) .unwrap(); rx.recv().unwrap(); let mut data = engine.take_last_write_data().unwrap(); assert_eq!(data.extra.req_type, ReqType::CheckSecondaryLocks); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::Rollback); let mut cnt = 0; custom_log.iterate_rollback(|k, ts, del_lock| { assert_eq!(k, b"k1"); assert_eq!(ts, 110); assert!(!del_lock); cnt += 1 }); assert_eq!(cnt, 1); } #[test] fn test_empty_commit_uses_legacy_commit_log() { let mut data = WriteData::from_modifies(vec![]); data.set_req_type(ReqType::Commit); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::Commit); let mut cnt = 0; custom_log.iterate_commit(|_, _| cnt += 1); assert_eq!(cnt, 0); } #[test] fn test_normal_commit_uses_legacy_commit_log() { let mut data = WriteData::from_modifies(vec![ Modify::Put( CF_WRITE, Key::from_raw(b"pk").append_ts(30.into()), txn_types::Write::new(WriteType::Put, 10.into(), None) .as_ref() .to_bytes(), ), Modify::Delete(CF_LOCK, Key::from_raw(b"pk")), ]); data.set_req_type(ReqType::Commit); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::Commit); let mut entries = vec![]; custom_log.iterate_commit(|key, commit_ts| { entries.push((key.to_vec(), commit_ts)); }); assert_eq!(entries, vec![(b"pk".to_vec(), 30)]); } #[test] fn test_delete_lock_only_resolve_lock_log_uses_existing_encoding() { let mut data = WriteData::from_modifies(vec![Modify::Delete(CF_LOCK, Key::from_raw(b"stale"))]); data.set_req_type(ReqType::ResolveLock); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::ResolveLock); let mut entries = vec![]; custom_log.iterate_resolve_lock(|tp, key, ts, del_lock| { entries.push((tp, key.to_vec(), ts, del_lock)); }); assert_eq!( entries, vec![( rlog::CustomRaftLogType::Rollback, b"stale".to_vec(), 0, true )] ); } #[test] fn test_mixed_commit_resolve_lock_log_uses_existing_encoding() { let mut data = WriteData::from_modifies(vec![ Modify::Put( CF_WRITE, Key::from_raw(b"pk").append_ts(30.into()), Write::new(WriteType::Put, 10.into(), None) .as_ref() .to_bytes(), ), Modify::Delete(CF_LOCK, Key::from_raw(b"pk")), Modify::Delete(CF_LOCK, Key::from_raw(b"stale")), ]); data.set_req_type(ReqType::ResolveLock); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!(custom_log.get_type(), rlog::CustomRaftLogType::ResolveLock); let mut entries = vec![]; custom_log.iterate_resolve_lock(|tp, key, ts, del_lock| { entries.push((tp, key.to_vec(), ts, del_lock)); }); entries.sort_by(|left, right| left.1.cmp(&right.1)); assert_eq!( entries, vec![ (rlog::CustomRaftLogType::Commit, b"pk".to_vec(), 30, true), ( rlog::CustomRaftLogType::Rollback, b"stale".to_vec(), 0, true ), ] ); } #[test] #[cfg(debug_assertions)] #[should_panic(expected = "unexpected reversed ResolveLock commit group")] fn test_resolve_lock_debug_asserts_reversed_commit_group() { let mut data = WriteData::from_modifies(vec![ Modify::Delete(CF_LOCK, Key::from_raw(b"pk")), Modify::Put( CF_WRITE, Key::from_raw(b"pk").append_ts(30.into()), Write::new(WriteType::Put, 10.into(), None) .as_ref() .to_bytes(), ), ]); data.set_req_type(ReqType::ResolveLock); let _ = modifies_to_requests(&Context::default(), &mut data); } #[test] #[cfg(debug_assertions)] #[should_panic(expected = "unexpected ResolveLock modifies for one key")] fn test_resolve_lock_debug_asserts_unexpected_three_entry_group() { let mut data = WriteData::from_modifies(vec![ Modify::Put( CF_WRITE, Key::from_raw(b"pk").append_ts(30.into()), Write::new(WriteType::Put, 10.into(), None) .as_ref() .to_bytes(), ), Modify::Delete(CF_LOCK, Key::from_raw(b"pk")), Modify::Put( CF_WRITE, Key::from_raw(b"pk").append_ts(31.into()), Write::new(WriteType::Put, 10.into(), None) .as_ref() .to_bytes(), ), ]); data.set_req_type(ReqType::ResolveLock); let _ = modifies_to_requests(&Context::default(), &mut data); } #[test] fn test_shared_lock_commit_uses_shared_lock_replay_log() { let mut data = WriteData::from_modifies(vec![ Modify::Put( CF_WRITE, Key::from_raw(b"k1").append_ts(20.into()), Write::new(WriteType::Lock, 10.into(), None) .as_ref() .to_bytes(), ), Modify::Delete(CF_LOCK, Key::from_raw(b"k1")), ]); data.set_req_type(ReqType::Commit); data.set_shared_lock_lifecycle(true); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!( custom_log.get_type(), rlog::CustomRaftLogType::SharedLockModifies ); let mut seen_write_put = false; let mut seen_lock_delete = false; custom_log.iterate_shared_lock_modifies(|modify| match modify { rlog::SharedLockModifyRef::WritePut { key, commit_ts, value, } => { seen_write_put = true; assert_eq!(key, b"k1"); assert_eq!(commit_ts, 20); let write = WriteRef::parse(value).unwrap(); assert_eq!(write.write_type, WriteType::Lock); assert_eq!(write.start_ts, 10.into()); } rlog::SharedLockModifyRef::LockDelete { key } => { seen_lock_delete = true; assert_eq!(key, b"k1"); } rlog::SharedLockModifyRef::LockPut { .. } => { unreachable!("final shared-lock holder removal should not rewrite CF_LOCK") } }); assert!(seen_write_put); assert!(seen_lock_delete); } #[test] fn test_mixed_commit_uses_shared_lock_replay_log() { let mut data = WriteData::from_modifies(vec![ Modify::Put( CF_WRITE, Key::from_raw(b"pk").append_ts(30.into()), Write::new(WriteType::Put, 10.into(), None) .as_ref() .to_bytes(), ), Modify::Delete(CF_LOCK, Key::from_raw(b"pk")), Modify::Put( CF_WRITE, Key::from_raw(b"sk").append_ts(30.into()), Write::new(WriteType::Lock, 10.into(), None) .as_ref() .to_bytes(), ), Modify::Delete(CF_LOCK, Key::from_raw(b"sk")), ]); data.set_req_type(ReqType::Commit); data.set_shared_lock_lifecycle(true); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!( custom_log.get_type(), rlog::CustomRaftLogType::SharedLockModifies ); let mut saw_non_shared_commit = false; let mut saw_shared_commit = false; let mut saw_non_shared_lock_delete = false; let mut saw_shared_lock_delete = false; custom_log.iterate_shared_lock_modifies(|modify| match modify { rlog::SharedLockModifyRef::WritePut { key, commit_ts, value, } => { let write = WriteRef::parse(value).unwrap(); if key == b"pk" { saw_non_shared_commit = true; assert_eq!(commit_ts, 30); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 10.into()); } else if key == b"sk" { saw_shared_commit = true; assert_eq!(commit_ts, 30); assert_eq!(write.write_type, WriteType::Lock); assert_eq!(write.start_ts, 10.into()); } else { unreachable!("unexpected mixed commit key") } } rlog::SharedLockModifyRef::LockDelete { key } if key == b"pk" => { saw_non_shared_lock_delete = true; } rlog::SharedLockModifyRef::LockDelete { key } if key == b"sk" => { saw_shared_lock_delete = true; } rlog::SharedLockModifyRef::LockDelete { .. } => unreachable!("unexpected delete key"), rlog::SharedLockModifyRef::LockPut { .. } => unreachable!("unexpected lock put"), }); assert!(saw_non_shared_commit); assert!(saw_shared_commit); assert!(saw_non_shared_lock_delete); assert!(saw_shared_lock_delete); } #[test] fn test_shared_lock_resolve_lock_uses_shared_lock_entries() { let mut data = WriteData::from_modifies(vec![ Modify::Put( CF_WRITE, Key::from_raw(b"k1").append_ts(20.into()), Write::new(WriteType::Lock, 10.into(), None) .as_ref() .to_bytes(), ), Modify::Delete(CF_LOCK, Key::from_raw(b"k1")), ]); data.set_req_type(ReqType::ResolveLock); data.set_shared_lock_lifecycle(true); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!( custom_log.get_type(), rlog::CustomRaftLogType::SharedLockModifies ); let mut seen_write_put = false; let mut seen_lock_delete = false; custom_log.iterate_shared_lock_modifies(|modify| match modify { rlog::SharedLockModifyRef::WritePut { key, commit_ts, value, } => { seen_write_put = true; assert_eq!(key, b"k1"); assert_eq!(commit_ts, 20); let write = WriteRef::parse(value).unwrap(); assert_eq!(write.write_type, WriteType::Lock); assert_eq!(write.start_ts, 10.into()); } rlog::SharedLockModifyRef::LockDelete { key } => { seen_lock_delete = true; assert_eq!(key, b"k1"); } rlog::SharedLockModifyRef::LockPut { .. } => { unreachable!("final shared-lock holder removal should not rewrite CF_LOCK") } }); assert!(seen_write_put); assert!(seen_lock_delete); } #[test] fn test_shared_lock_modifies_orders_same_key_write_puts_before_lock_updates() { let mut data = WriteData::from_modifies(vec![ Modify::Put( CF_WRITE, Key::from_raw(b"k1").append_ts(20.into()), Write::new(WriteType::Lock, 10.into(), None) .as_ref() .to_bytes(), ), Modify::Put(CF_LOCK, Key::from_raw(b"k1"), b"remaining".to_vec()), Modify::Put( CF_WRITE, Key::from_raw(b"k1").append_ts(11.into()), Write::new(WriteType::Rollback, 11.into(), None) .as_ref() .to_bytes(), ), Modify::Delete(CF_LOCK, Key::from_raw(b"k1")), ]); data.set_req_type(ReqType::ResolveLock); data.set_shared_lock_lifecycle(true); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!( custom_log.get_type(), rlog::CustomRaftLogType::SharedLockModifies ); #[derive(Debug, PartialEq)] enum Entry { Write { write_type: WriteType, start_ts: u64, commit_ts: u64, }, LockPut, LockDelete, } let mut entries = Vec::new(); custom_log.iterate_shared_lock_modifies(|modify| match modify { rlog::SharedLockModifyRef::WritePut { key, commit_ts, value, } => { assert_eq!(key, b"k1"); let write = WriteRef::parse(value).unwrap(); entries.push(Entry::Write { write_type: write.write_type, start_ts: write.start_ts.into_inner(), commit_ts, }); } rlog::SharedLockModifyRef::LockPut { key, value } => { assert_eq!(key, b"k1"); assert_eq!(value, b"remaining"); entries.push(Entry::LockPut); } rlog::SharedLockModifyRef::LockDelete { key } => { assert_eq!(key, b"k1"); entries.push(Entry::LockDelete); } }); assert_eq!( entries, vec![ Entry::Write { write_type: WriteType::Lock, start_ts: 10, commit_ts: 20 }, Entry::Write { write_type: WriteType::Rollback, start_ts: 11, commit_ts: 11 }, Entry::LockPut, Entry::LockDelete ] ); } #[test] fn test_mixed_resolve_lock_uses_shared_lock_replay_log() { let mut data = WriteData::from_modifies(vec![ Modify::Put( CF_WRITE, Key::from_raw(b"pk").append_ts(30.into()), Write::new(WriteType::Put, 10.into(), None) .as_ref() .to_bytes(), ), Modify::Delete(CF_LOCK, Key::from_raw(b"pk")), Modify::Put( CF_WRITE, Key::from_raw(b"sk").append_ts(30.into()), Write::new(WriteType::Lock, 10.into(), None) .as_ref() .to_bytes(), ), Modify::Delete(CF_LOCK, Key::from_raw(b"sk")), ]); data.set_req_type(ReqType::ResolveLock); data.set_shared_lock_lifecycle(true); let custom_req = modifies_to_requests(&Context::default(), &mut data); let custom_log = rlog::CustomRaftLog::new_from_data(custom_req.get_data()); assert_eq!( custom_log.get_type(), rlog::CustomRaftLogType::SharedLockModifies ); let mut saw_non_shared_commit = false; let mut saw_shared_commit = false; let mut saw_non_shared_lock_delete = false; let mut saw_shared_lock_delete = false; custom_log.iterate_shared_lock_modifies(|modify| match modify { rlog::SharedLockModifyRef::WritePut { key, commit_ts, value, } => { let write = WriteRef::parse(value).unwrap(); if key == b"pk" { saw_non_shared_commit = true; assert_eq!(commit_ts, 30); assert_eq!(write.write_type, WriteType::Put); assert_eq!(write.start_ts, 10.into()); } else if key == b"sk" { saw_shared_commit = true; assert_eq!(commit_ts, 30); assert_eq!(write.write_type, WriteType::Lock); assert_eq!(write.start_ts, 10.into()); } else { unreachable!("unexpected mixed resolve-lock key") } } rlog::SharedLockModifyRef::LockDelete { key } if key == b"pk" => { saw_non_shared_lock_delete = true; } rlog::SharedLockModifyRef::LockDelete { key } if key == b"sk" => { saw_shared_lock_delete = true; } rlog::SharedLockModifyRef::LockDelete { .. } => unreachable!("unexpected delete key"), rlog::SharedLockModifyRef::LockPut { .. } => unreachable!("unexpected lock put"), }); assert!(saw_non_shared_commit); assert!(saw_shared_commit); assert!(saw_non_shared_lock_delete); assert!(saw_shared_lock_delete); } }