tuwunel_database/map/
get.rs

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

tuwunel_database/map/
get.rs
