tuwunel_database/map/
get.rs

1use std::{fmt::Debug, sync::Arc};
2
3use futures::{FutureExt, TryFutureExt, future::ready};
4use rocksdb::{DBPinnableSlice, ReadOptions};
5use tokio::task::consume_budget;
6use tuwunel_core::{Err, Result, err, implement, utils::result::MapExpect};
7
8use crate::{
9	Handle,
10	util::{is_incomplete, map_err, or_else},
11};
12
13/// Fetches a raw key asynchronously and returns a pinned value handle.
14///
15/// Cache results consume cooperative scheduler budget, while misses run on the
16/// engine's blocking pool. The returned handle keeps its RocksDB value storage
17/// pinned for the handle's lifetime.
18#[implement(super::Map)]
19#[tracing::instrument(skip(self, key), fields(%self), level = "trace")]
20pub fn get<K>(
21	self: &Arc<Self>,
22	key: &K,
23) -> impl Future<Output = Result<Handle<'_>>> + Send + use<'_, K>
24where
25	K: AsRef<[u8]> + Debug + ?Sized,
26{
27	use crate::pool::Get;
28
29	let cached = self.get_cached(key);
30	if matches!(cached, Err(_) | Ok(Some(_))) {
31		return consume_budget()
32			.map(move |()| cached.map_expect("data found in cache"))
33			.left_future();
34	}
35
36	debug_assert!(matches!(cached, Ok(None)), "expected status Incomplete");
37	let cmd = Get {
38		map: self.clone(),
39		key: [key.as_ref().into()].into(),
40		res: None,
41	};
42
43	self.engine
44		.pool
45		.execute_get(cmd)
46		.and_then(|mut res| ready(res.remove(0)))
47		.right_future()
48}
49
50/// Fetches a raw key from block cache without storage I/O.
51///
52/// A cache miss returns `Ok(None)`, while a cached absence or database failure
53/// remains an error.
54#[implement(super::Map)]
55#[tracing::instrument(skip(self, key), name = "cache", level = "trace")]
56pub(crate) fn get_cached<K>(&self, key: &K) -> Result<Option<Handle<'_>>>
57where
58	K: AsRef<[u8]> + Debug + ?Sized,
59{
60	let res = self.get_blocking_opts(key, &self.cache_read_options);
61	cached_handle_from(res)
62}
63
64/// Fetches a raw key synchronously and returns a pinned value handle.
65///
66/// The call may block on storage and populate RocksDB caches. The returned
67/// handle keeps its value storage pinned for the handle's lifetime.
68#[implement(super::Map)]
69#[tracing::instrument(skip(self, key), name = "blocking", level = "trace")]
70pub fn get_blocking<K>(&self, key: &K) -> Result<Handle<'_>>
71where
72	K: AsRef<[u8]> + ?Sized,
73{
74	let res = self.get_blocking_opts(key, &self.read_options);
75	handle_from(res)
76}
77
78/// Performs a pinned point read with explicit RocksDB read options.
79///
80/// The raw RocksDB result distinguishes absence from storage failure for the
81/// caller to classify.
82#[implement(super::Map)]
83fn get_blocking_opts<K>(
84	&self,
85	key: &K,
86	read_options: &ReadOptions,
87) -> Result<Option<DBPinnableSlice<'_>>, rocksdb::Error>
88where
89	K: AsRef<[u8]> + ?Sized,
90{
91	self.engine
92		.db
93		.get_pinned_cf_opt(&self.cf(), key, read_options)
94}
95
96/// Converts a RocksDB point-read result into a required value handle.
97///
98/// Missing values become the database not-found error, while RocksDB failures
99/// use the shared error mapping.
100#[inline]
101pub(super) fn handle_from(
102	result: Result<Option<DBPinnableSlice<'_>>, rocksdb::Error>,
103) -> Result<Handle<'_>> {
104	result
105		.map_err(map_err)?
106		.map(Handle::from)
107		.ok_or(err!(Request(NotFound("Not found in database"))))
108}
109
110/// Classifies a block-cache point-read result.
111///
112/// `Ok(None)` represents a cache miss, a cached absence becomes not-found, and
113/// other RocksDB failures use the shared error mapping.
114#[inline]
115pub(super) fn cached_handle_from(
116	result: Result<Option<DBPinnableSlice<'_>>, rocksdb::Error>,
117) -> Result<Option<Handle<'_>>> {
118	match result {
119		// cache hit; not found
120		| Ok(None) => Err!(Request(NotFound("Not found in database"))),
121
122		// cache hit; value found
123		| Ok(Some(result)) => Ok(Some(Handle::from(result))),
124
125		// cache miss; unknown
126		| Err(error) if is_incomplete(&error) => Ok(None),
127
128		// some other error occurred
129		| Err(error) => or_else(error),
130	}
131}
tuwunel_database/map/get.rs

tuwunel_database/map/
get.rs
