Line data Source code
1 : //! Structs representing the canonical page service API.
2 : //!
3 : //! These mirror the autogenerated Protobuf types. The differences are:
4 : //!
5 : //! - Types that are in fact required by the API are not Options. The protobuf "required"
6 : //! attribute is deprecated and 'prost' marks a lot of members as optional because of that.
7 : //! (See <https://github.com/tokio-rs/prost/issues/800> for a gripe on this)
8 : //!
9 : //! - Use more precise datatypes, e.g. Lsn and uints shorter than 32 bits.
10 : //!
11 : //! - Validate protocol invariants, via try_from() and try_into().
12 : //!
13 : //! Validation only happens on the receiver side, i.e. when converting from Protobuf to domain
14 : //! types. This is where it matters -- the Protobuf types are less strict than the domain types, and
15 : //! receivers should expect all sorts of junk from senders. This also allows the sender to use e.g.
16 : //! stream combinators without dealing with errors, and avoids validating the same message twice.
17 :
18 : use std::fmt::Display;
19 : use std::time::{Duration, SystemTime, UNIX_EPOCH};
20 :
21 : use bytes::Bytes;
22 : use postgres_ffi_types::Oid;
23 : // TODO: split out Lsn, RelTag, SlruKind and other basic types to a separate crate, to avoid
24 : // pulling in all of their other crate dependencies when building the client.
25 : use utils::lsn::Lsn;
26 :
27 : use crate::proto;
28 :
29 : /// A protocol error. Typically returned via try_from() or try_into().
30 : #[derive(thiserror::Error, Clone, Debug)]
31 : pub enum ProtocolError {
32 : #[error("field '{0}' has invalid value '{1}'")]
33 : Invalid(&'static str, String),
34 : #[error("required field '{0}' is missing")]
35 : Missing(&'static str),
36 : }
37 :
38 : impl ProtocolError {
39 : /// Helper to generate a new ProtocolError::Invalid for the given field and value.
40 0 : pub fn invalid(field: &'static str, value: impl std::fmt::Debug) -> Self {
41 0 : Self::Invalid(field, format!("{value:?}"))
42 0 : }
43 : }
44 :
45 : impl From<ProtocolError> for tonic::Status {
46 0 : fn from(err: ProtocolError) -> Self {
47 0 : tonic::Status::invalid_argument(format!("{err}"))
48 0 : }
49 : }
50 :
51 : /// The LSN a request should read at.
52 : #[derive(Clone, Copy, Debug, Default)]
53 : pub struct ReadLsn {
54 : /// The request's read LSN.
55 : pub request_lsn: Lsn,
56 : /// If given, the caller guarantees that the page has not been modified since this LSN. Must be
57 : /// smaller than or equal to request_lsn. This allows the Pageserver to serve an old page
58 : /// without waiting for the request LSN to arrive. If not given, the request will read at the
59 : /// request_lsn and wait for it to arrive if necessary. Valid for all request types.
60 : ///
61 : /// It is undefined behaviour to make a request such that the page was, in fact, modified
62 : /// between request_lsn and not_modified_since_lsn. The Pageserver might detect it and return an
63 : /// error, or it might return the old page version or the new page version. Setting
64 : /// not_modified_since_lsn equal to request_lsn is always safe, but can lead to unnecessary
65 : /// waiting.
66 : pub not_modified_since_lsn: Option<Lsn>,
67 : }
68 :
69 : impl Display for ReadLsn {
70 0 : fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
71 0 : let req_lsn = self.request_lsn;
72 0 : if let Some(mod_lsn) = self.not_modified_since_lsn {
73 0 : write!(f, "{req_lsn}>={mod_lsn}")
74 : } else {
75 0 : req_lsn.fmt(f)
76 : }
77 0 : }
78 : }
79 :
80 : impl TryFrom<proto::ReadLsn> for ReadLsn {
81 : type Error = ProtocolError;
82 :
83 0 : fn try_from(pb: proto::ReadLsn) -> Result<Self, Self::Error> {
84 0 : if pb.request_lsn == 0 {
85 0 : return Err(ProtocolError::invalid("request_lsn", pb.request_lsn));
86 0 : }
87 0 : if pb.not_modified_since_lsn > pb.request_lsn {
88 0 : return Err(ProtocolError::invalid(
89 0 : "not_modified_since_lsn",
90 0 : pb.not_modified_since_lsn,
91 0 : ));
92 0 : }
93 : Ok(Self {
94 0 : request_lsn: Lsn(pb.request_lsn),
95 0 : not_modified_since_lsn: match pb.not_modified_since_lsn {
96 0 : 0 => None,
97 0 : lsn => Some(Lsn(lsn)),
98 : },
99 : })
100 0 : }
101 : }
102 :
103 : impl From<ReadLsn> for proto::ReadLsn {
104 0 : fn from(read_lsn: ReadLsn) -> Self {
105 0 : Self {
106 0 : request_lsn: read_lsn.request_lsn.0,
107 0 : not_modified_since_lsn: read_lsn.not_modified_since_lsn.unwrap_or_default().0,
108 0 : }
109 0 : }
110 : }
111 :
112 : // RelTag is defined in pageserver_api::reltag.
113 : pub type RelTag = pageserver_api::reltag::RelTag;
114 :
115 : impl TryFrom<proto::RelTag> for RelTag {
116 : type Error = ProtocolError;
117 :
118 0 : fn try_from(pb: proto::RelTag) -> Result<Self, Self::Error> {
119 : Ok(Self {
120 0 : spcnode: pb.spc_oid,
121 0 : dbnode: pb.db_oid,
122 0 : relnode: pb.rel_number,
123 0 : forknum: pb
124 0 : .fork_number
125 0 : .try_into()
126 0 : .map_err(|_| ProtocolError::invalid("fork_number", pb.fork_number))?,
127 : })
128 0 : }
129 : }
130 :
131 : impl From<RelTag> for proto::RelTag {
132 0 : fn from(rel_tag: RelTag) -> Self {
133 0 : Self {
134 0 : spc_oid: rel_tag.spcnode,
135 0 : db_oid: rel_tag.dbnode,
136 0 : rel_number: rel_tag.relnode,
137 0 : fork_number: rel_tag.forknum as u32,
138 0 : }
139 0 : }
140 : }
141 :
142 : /// Requests a base backup.
143 : #[derive(Clone, Copy, Debug)]
144 : pub struct GetBaseBackupRequest {
145 : /// The LSN to fetch a base backup at. If None, uses the latest LSN known to the Pageserver.
146 : pub lsn: Option<Lsn>,
147 : /// If true, logical replication slots will not be created.
148 : pub replica: bool,
149 : /// If true, include relation files in the base backup. Mainly for debugging and tests.
150 : pub full: bool,
151 : /// Compression algorithm to use. Base backups send a compressed payload instead of using gRPC
152 : /// compression, so that we can cache compressed backups on the server.
153 : pub compression: BaseBackupCompression,
154 : }
155 :
156 : impl TryFrom<proto::GetBaseBackupRequest> for GetBaseBackupRequest {
157 : type Error = ProtocolError;
158 :
159 0 : fn try_from(pb: proto::GetBaseBackupRequest) -> Result<Self, Self::Error> {
160 : Ok(Self {
161 0 : lsn: (pb.lsn != 0).then_some(Lsn(pb.lsn)),
162 0 : replica: pb.replica,
163 0 : full: pb.full,
164 0 : compression: pb.compression.try_into()?,
165 : })
166 0 : }
167 : }
168 :
169 : impl From<GetBaseBackupRequest> for proto::GetBaseBackupRequest {
170 0 : fn from(request: GetBaseBackupRequest) -> Self {
171 0 : Self {
172 0 : lsn: request.lsn.unwrap_or_default().0,
173 0 : replica: request.replica,
174 0 : full: request.full,
175 0 : compression: request.compression.into(),
176 0 : }
177 0 : }
178 : }
179 :
180 : /// Base backup compression algorithm.
181 : #[derive(Clone, Copy, Debug)]
182 : pub enum BaseBackupCompression {
183 : None,
184 : Gzip,
185 : }
186 :
187 : impl TryFrom<proto::BaseBackupCompression> for BaseBackupCompression {
188 : type Error = ProtocolError;
189 :
190 0 : fn try_from(pb: proto::BaseBackupCompression) -> Result<Self, Self::Error> {
191 0 : match pb {
192 0 : proto::BaseBackupCompression::Unknown => Err(ProtocolError::invalid("compression", pb)),
193 0 : proto::BaseBackupCompression::None => Ok(Self::None),
194 0 : proto::BaseBackupCompression::Gzip => Ok(Self::Gzip),
195 : }
196 0 : }
197 : }
198 :
199 : impl TryFrom<i32> for BaseBackupCompression {
200 : type Error = ProtocolError;
201 :
202 0 : fn try_from(compression: i32) -> Result<Self, Self::Error> {
203 0 : proto::BaseBackupCompression::try_from(compression)
204 0 : .map_err(|_| ProtocolError::invalid("compression", compression))
205 0 : .and_then(Self::try_from)
206 0 : }
207 : }
208 :
209 : impl From<BaseBackupCompression> for proto::BaseBackupCompression {
210 0 : fn from(compression: BaseBackupCompression) -> Self {
211 0 : match compression {
212 0 : BaseBackupCompression::None => Self::None,
213 0 : BaseBackupCompression::Gzip => Self::Gzip,
214 : }
215 0 : }
216 : }
217 :
218 : impl From<BaseBackupCompression> for i32 {
219 0 : fn from(compression: BaseBackupCompression) -> Self {
220 0 : proto::BaseBackupCompression::from(compression).into()
221 0 : }
222 : }
223 :
224 : pub type GetBaseBackupResponseChunk = Bytes;
225 :
226 : impl TryFrom<proto::GetBaseBackupResponseChunk> for GetBaseBackupResponseChunk {
227 : type Error = ProtocolError;
228 :
229 0 : fn try_from(pb: proto::GetBaseBackupResponseChunk) -> Result<Self, Self::Error> {
230 0 : if pb.chunk.is_empty() {
231 0 : return Err(ProtocolError::Missing("chunk"));
232 0 : }
233 0 : Ok(pb.chunk)
234 0 : }
235 : }
236 :
237 : impl From<GetBaseBackupResponseChunk> for proto::GetBaseBackupResponseChunk {
238 0 : fn from(chunk: GetBaseBackupResponseChunk) -> Self {
239 0 : Self { chunk }
240 0 : }
241 : }
242 :
243 : /// Requests the size of a database, as # of bytes. Only valid on shard 0, other shards will error.
244 : #[derive(Clone, Copy, Debug)]
245 : pub struct GetDbSizeRequest {
246 : pub read_lsn: ReadLsn,
247 : pub db_oid: Oid,
248 : }
249 :
250 : impl TryFrom<proto::GetDbSizeRequest> for GetDbSizeRequest {
251 : type Error = ProtocolError;
252 :
253 0 : fn try_from(pb: proto::GetDbSizeRequest) -> Result<Self, Self::Error> {
254 : Ok(Self {
255 0 : read_lsn: pb
256 0 : .read_lsn
257 0 : .ok_or(ProtocolError::Missing("read_lsn"))?
258 0 : .try_into()?,
259 0 : db_oid: pb.db_oid,
260 : })
261 0 : }
262 : }
263 :
264 : impl From<GetDbSizeRequest> for proto::GetDbSizeRequest {
265 0 : fn from(request: GetDbSizeRequest) -> Self {
266 0 : Self {
267 0 : read_lsn: Some(request.read_lsn.into()),
268 0 : db_oid: request.db_oid,
269 0 : }
270 0 : }
271 : }
272 :
273 : pub type GetDbSizeResponse = u64;
274 :
275 : impl From<proto::GetDbSizeResponse> for GetDbSizeResponse {
276 0 : fn from(pb: proto::GetDbSizeResponse) -> Self {
277 0 : pb.num_bytes
278 0 : }
279 : }
280 :
281 : impl From<GetDbSizeResponse> for proto::GetDbSizeResponse {
282 0 : fn from(num_bytes: GetDbSizeResponse) -> Self {
283 0 : Self { num_bytes }
284 0 : }
285 : }
286 :
287 : /// Requests one or more pages.
288 : #[derive(Clone, Debug, Default)]
289 : pub struct GetPageRequest {
290 : /// A request ID. Will be included in the response. Should be unique for in-flight requests on
291 : /// the stream.
292 : pub request_id: RequestID,
293 : /// The request class.
294 : pub request_class: GetPageClass,
295 : /// The LSN to read at.
296 : pub read_lsn: ReadLsn,
297 : /// The relation to read from.
298 : pub rel: RelTag,
299 : /// Page numbers to read. Must belong to the remote shard.
300 : ///
301 : /// Multiple pages will be executed as a single batch by the Pageserver, amortizing layer access
302 : /// costs and parallelizing them. This may increase the latency of any individual request, but
303 : /// improves the overall latency and throughput of the batch as a whole.
304 : pub block_numbers: Vec<u32>,
305 : }
306 :
307 : impl TryFrom<proto::GetPageRequest> for GetPageRequest {
308 : type Error = ProtocolError;
309 :
310 0 : fn try_from(pb: proto::GetPageRequest) -> Result<Self, Self::Error> {
311 0 : if pb.block_number.is_empty() {
312 0 : return Err(ProtocolError::Missing("block_number"));
313 0 : }
314 : Ok(Self {
315 0 : request_id: pb
316 0 : .request_id
317 0 : .ok_or(ProtocolError::Missing("request_id"))?
318 0 : .into(),
319 0 : request_class: pb.request_class.into(),
320 0 : read_lsn: pb
321 0 : .read_lsn
322 0 : .ok_or(ProtocolError::Missing("read_lsn"))?
323 0 : .try_into()?,
324 0 : rel: pb.rel.ok_or(ProtocolError::Missing("rel"))?.try_into()?,
325 0 : block_numbers: pb.block_number,
326 : })
327 0 : }
328 : }
329 :
330 : impl From<GetPageRequest> for proto::GetPageRequest {
331 0 : fn from(request: GetPageRequest) -> Self {
332 0 : Self {
333 0 : request_id: Some(request.request_id.into()),
334 0 : request_class: request.request_class.into(),
335 0 : read_lsn: Some(request.read_lsn.into()),
336 0 : rel: Some(request.rel.into()),
337 0 : block_number: request.block_numbers,
338 0 : }
339 0 : }
340 : }
341 :
342 : /// A GetPage request ID and retry attempt. Should be unique for in-flight requests on a stream.
343 : #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash, PartialOrd, Ord)]
344 : pub struct RequestID {
345 : /// The base request ID.
346 : pub id: u64,
347 : // The request attempt. Starts at 0, incremented on each retry.
348 : pub attempt: u32,
349 : }
350 :
351 : impl RequestID {
352 : /// Creates a new RequestID with the given ID and an initial attempt of 0.
353 0 : pub fn new(id: u64) -> Self {
354 0 : Self { id, attempt: 0 }
355 0 : }
356 : }
357 :
358 : impl Display for RequestID {
359 0 : fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
360 0 : write!(f, "{}.{}", self.id, self.attempt)
361 0 : }
362 : }
363 :
364 : impl From<proto::RequestId> for RequestID {
365 0 : fn from(pb: proto::RequestId) -> Self {
366 0 : Self {
367 0 : id: pb.id,
368 0 : attempt: pb.attempt,
369 0 : }
370 0 : }
371 : }
372 :
373 : impl From<u64> for RequestID {
374 0 : fn from(id: u64) -> Self {
375 0 : Self::new(id)
376 0 : }
377 : }
378 :
379 : impl From<RequestID> for proto::RequestId {
380 0 : fn from(request_id: RequestID) -> Self {
381 0 : Self {
382 0 : id: request_id.id,
383 0 : attempt: request_id.attempt,
384 0 : }
385 0 : }
386 : }
387 :
388 : /// A GetPage request class.
389 : #[derive(Clone, Copy, Debug, Default, strum_macros::Display)]
390 : pub enum GetPageClass {
391 : /// Unknown class. For backwards compatibility: used when an older client version sends a class
392 : /// that a newer server version has removed.
393 : Unknown,
394 : /// A normal request. This is the default.
395 : #[default]
396 : Normal,
397 : /// A prefetch request. NB: can only be classified on pg < 18.
398 : Prefetch,
399 : /// A background request (e.g. vacuum).
400 : Background,
401 : }
402 :
403 : impl From<proto::GetPageClass> for GetPageClass {
404 0 : fn from(pb: proto::GetPageClass) -> Self {
405 0 : match pb {
406 0 : proto::GetPageClass::Unknown => Self::Unknown,
407 0 : proto::GetPageClass::Normal => Self::Normal,
408 0 : proto::GetPageClass::Prefetch => Self::Prefetch,
409 0 : proto::GetPageClass::Background => Self::Background,
410 : }
411 0 : }
412 : }
413 :
414 : impl From<i32> for GetPageClass {
415 0 : fn from(class: i32) -> Self {
416 0 : proto::GetPageClass::try_from(class)
417 0 : .unwrap_or(proto::GetPageClass::Unknown)
418 0 : .into()
419 0 : }
420 : }
421 :
422 : impl From<GetPageClass> for proto::GetPageClass {
423 0 : fn from(class: GetPageClass) -> Self {
424 0 : match class {
425 0 : GetPageClass::Unknown => Self::Unknown,
426 0 : GetPageClass::Normal => Self::Normal,
427 0 : GetPageClass::Prefetch => Self::Prefetch,
428 0 : GetPageClass::Background => Self::Background,
429 : }
430 0 : }
431 : }
432 :
433 : impl From<GetPageClass> for i32 {
434 0 : fn from(class: GetPageClass) -> Self {
435 0 : proto::GetPageClass::from(class).into()
436 0 : }
437 : }
438 :
439 : /// A GetPage response.
440 : ///
441 : /// A batch response will contain all of the requested pages. We could eagerly emit individual pages
442 : /// as soon as they are ready, but on a readv() Postgres holds buffer pool locks on all pages in the
443 : /// batch and we'll only return once the entire batch is ready, so no one can make use of the
444 : /// individual pages.
445 : #[derive(Clone, Debug)]
446 : pub struct GetPageResponse {
447 : /// The original request's ID.
448 : pub request_id: RequestID,
449 : /// The response status code. If not OK, the `rel` and `pages` fields will be empty.
450 : pub status_code: GetPageStatusCode,
451 : /// A string describing the status, if any.
452 : pub reason: Option<String>,
453 : /// The relation that the pages belong to.
454 : pub rel: RelTag,
455 : // The page(s), in the same order as the request.
456 : pub pages: Vec<Page>,
457 : }
458 :
459 : impl TryFrom<proto::GetPageResponse> for GetPageResponse {
460 : type Error = ProtocolError;
461 :
462 0 : fn try_from(pb: proto::GetPageResponse) -> Result<Self, ProtocolError> {
463 : Ok(Self {
464 0 : request_id: pb
465 0 : .request_id
466 0 : .ok_or(ProtocolError::Missing("request_id"))?
467 0 : .into(),
468 0 : status_code: pb.status_code.into(),
469 0 : reason: Some(pb.reason).filter(|r| !r.is_empty()),
470 0 : rel: pb.rel.ok_or(ProtocolError::Missing("rel"))?.try_into()?,
471 0 : pages: pb.page.into_iter().map(Page::from).collect(),
472 : })
473 0 : }
474 : }
475 :
476 : impl From<GetPageResponse> for proto::GetPageResponse {
477 0 : fn from(response: GetPageResponse) -> Self {
478 0 : Self {
479 0 : request_id: Some(response.request_id.into()),
480 0 : status_code: response.status_code.into(),
481 0 : reason: response.reason.unwrap_or_default(),
482 0 : rel: Some(response.rel.into()),
483 0 : page: response.pages.into_iter().map(proto::Page::from).collect(),
484 0 : }
485 0 : }
486 : }
487 :
488 : impl GetPageResponse {
489 : /// Attempts to represent a tonic::Status as a GetPageResponse if appropriate. Returning a
490 : /// tonic::Status will terminate the GetPage stream, so per-request errors are emitted as a
491 : /// GetPageResponse with a non-OK status code instead.
492 : #[allow(clippy::result_large_err)]
493 0 : pub fn try_from_status(
494 0 : status: tonic::Status,
495 0 : request_id: RequestID,
496 0 : ) -> Result<Self, tonic::Status> {
497 : // We shouldn't see an OK status here, because we're emitting an error.
498 0 : debug_assert_ne!(status.code(), tonic::Code::Ok);
499 0 : if status.code() == tonic::Code::Ok {
500 0 : return Err(tonic::Status::internal(format!(
501 0 : "unexpected OK status: {status:?}",
502 0 : )));
503 0 : }
504 :
505 : // If we can't convert the tonic::Code to a GetPageStatusCode, this is not a per-request
506 : // error and we should return a tonic::Status to terminate the stream.
507 0 : let Ok(status_code) = status.code().try_into() else {
508 0 : return Err(status);
509 : };
510 :
511 : // Return a GetPageResponse for the status.
512 0 : Ok(Self {
513 0 : request_id,
514 0 : status_code,
515 0 : reason: Some(status.message().to_string()),
516 0 : rel: RelTag::default(),
517 0 : pages: Vec::new(),
518 0 : })
519 0 : }
520 : }
521 :
522 : // A page.
523 : #[derive(Clone, Debug)]
524 : pub struct Page {
525 : /// The page number.
526 : pub block_number: u32,
527 : /// The materialized page image, as an 8KB byte vector.
528 : pub image: Bytes,
529 : }
530 :
531 : impl From<proto::Page> for Page {
532 0 : fn from(pb: proto::Page) -> Self {
533 0 : Self {
534 0 : block_number: pb.block_number,
535 0 : image: pb.image,
536 0 : }
537 0 : }
538 : }
539 :
540 : impl From<Page> for proto::Page {
541 0 : fn from(page: Page) -> Self {
542 0 : Self {
543 0 : block_number: page.block_number,
544 0 : image: page.image,
545 0 : }
546 0 : }
547 : }
548 :
549 : /// A GetPage response status code.
550 : ///
551 : /// These are effectively equivalent to gRPC statuses. However, we use a bidirectional stream
552 : /// (potentially shared by many backends), and a gRPC status response would terminate the stream so
553 : /// we send GetPageResponse messages with these codes instead.
554 : #[derive(Clone, Copy, Debug, PartialEq, strum_macros::Display)]
555 : pub enum GetPageStatusCode {
556 : /// Unknown status. For forwards compatibility: used when an older client version receives a new
557 : /// status code from a newer server version.
558 : Unknown,
559 : /// The request was successful.
560 : Ok,
561 : /// The page did not exist. The tenant/timeline/shard has already been validated during stream
562 : /// setup.
563 : NotFound,
564 : /// The request was invalid.
565 : InvalidRequest,
566 : /// The request failed due to an internal server error.
567 : InternalError,
568 : /// The tenant is rate limited. Slow down and retry later.
569 : SlowDown,
570 : }
571 :
572 : impl From<proto::GetPageStatusCode> for GetPageStatusCode {
573 0 : fn from(pb: proto::GetPageStatusCode) -> Self {
574 0 : match pb {
575 0 : proto::GetPageStatusCode::Unknown => Self::Unknown,
576 0 : proto::GetPageStatusCode::Ok => Self::Ok,
577 0 : proto::GetPageStatusCode::NotFound => Self::NotFound,
578 0 : proto::GetPageStatusCode::InvalidRequest => Self::InvalidRequest,
579 0 : proto::GetPageStatusCode::InternalError => Self::InternalError,
580 0 : proto::GetPageStatusCode::SlowDown => Self::SlowDown,
581 : }
582 0 : }
583 : }
584 :
585 : impl From<i32> for GetPageStatusCode {
586 0 : fn from(status_code: i32) -> Self {
587 0 : proto::GetPageStatusCode::try_from(status_code)
588 0 : .unwrap_or(proto::GetPageStatusCode::Unknown)
589 0 : .into()
590 0 : }
591 : }
592 :
593 : impl From<GetPageStatusCode> for proto::GetPageStatusCode {
594 0 : fn from(status_code: GetPageStatusCode) -> Self {
595 0 : match status_code {
596 0 : GetPageStatusCode::Unknown => Self::Unknown,
597 0 : GetPageStatusCode::Ok => Self::Ok,
598 0 : GetPageStatusCode::NotFound => Self::NotFound,
599 0 : GetPageStatusCode::InvalidRequest => Self::InvalidRequest,
600 0 : GetPageStatusCode::InternalError => Self::InternalError,
601 0 : GetPageStatusCode::SlowDown => Self::SlowDown,
602 : }
603 0 : }
604 : }
605 :
606 : impl From<GetPageStatusCode> for i32 {
607 0 : fn from(status_code: GetPageStatusCode) -> Self {
608 0 : proto::GetPageStatusCode::from(status_code).into()
609 0 : }
610 : }
611 :
612 : impl TryFrom<tonic::Code> for GetPageStatusCode {
613 : type Error = tonic::Code;
614 :
615 0 : fn try_from(code: tonic::Code) -> Result<Self, Self::Error> {
616 : use tonic::Code;
617 :
618 0 : let status_code = match code {
619 0 : Code::Ok => Self::Ok,
620 :
621 : // These are per-request errors, which should be returned as GetPageResponses.
622 0 : Code::AlreadyExists => Self::InvalidRequest,
623 0 : Code::DataLoss => Self::InternalError,
624 0 : Code::FailedPrecondition => Self::InvalidRequest,
625 0 : Code::InvalidArgument => Self::InvalidRequest,
626 0 : Code::Internal => Self::InternalError,
627 0 : Code::NotFound => Self::NotFound,
628 0 : Code::OutOfRange => Self::InvalidRequest,
629 0 : Code::ResourceExhausted => Self::SlowDown,
630 :
631 : // These should terminate the stream by returning a tonic::Status.
632 : Code::Aborted
633 : | Code::Cancelled
634 : | Code::DeadlineExceeded
635 : | Code::PermissionDenied
636 : | Code::Unauthenticated
637 : | Code::Unavailable
638 : | Code::Unimplemented
639 0 : | Code::Unknown => return Err(code),
640 : };
641 0 : Ok(status_code)
642 0 : }
643 : }
644 :
645 : impl From<GetPageStatusCode> for tonic::Code {
646 0 : fn from(status_code: GetPageStatusCode) -> Self {
647 : use tonic::Code;
648 :
649 0 : match status_code {
650 0 : GetPageStatusCode::Unknown => Code::Unknown,
651 0 : GetPageStatusCode::Ok => Code::Ok,
652 0 : GetPageStatusCode::NotFound => Code::NotFound,
653 0 : GetPageStatusCode::InvalidRequest => Code::InvalidArgument,
654 0 : GetPageStatusCode::InternalError => Code::Internal,
655 0 : GetPageStatusCode::SlowDown => Code::ResourceExhausted,
656 : }
657 0 : }
658 : }
659 :
660 : // Fetches the size of a relation at a given LSN, as # of blocks. Only valid on shard 0, other
661 : // shards will error.
662 : #[derive(Clone, Copy, Debug)]
663 : pub struct GetRelSizeRequest {
664 : pub read_lsn: ReadLsn,
665 : pub rel: RelTag,
666 : /// If true, return missing=true for missing relations instead of a NotFound error.
667 : pub allow_missing: bool,
668 : }
669 :
670 : impl TryFrom<proto::GetRelSizeRequest> for GetRelSizeRequest {
671 : type Error = ProtocolError;
672 :
673 0 : fn try_from(proto: proto::GetRelSizeRequest) -> Result<Self, Self::Error> {
674 : Ok(Self {
675 0 : read_lsn: proto
676 0 : .read_lsn
677 0 : .ok_or(ProtocolError::Missing("read_lsn"))?
678 0 : .try_into()?,
679 0 : rel: proto.rel.ok_or(ProtocolError::Missing("rel"))?.try_into()?,
680 0 : allow_missing: proto.allow_missing,
681 : })
682 0 : }
683 : }
684 :
685 : impl From<GetRelSizeRequest> for proto::GetRelSizeRequest {
686 0 : fn from(request: GetRelSizeRequest) -> Self {
687 0 : Self {
688 0 : read_lsn: Some(request.read_lsn.into()),
689 0 : rel: Some(request.rel.into()),
690 0 : allow_missing: request.allow_missing,
691 0 : }
692 0 : }
693 : }
694 :
695 : /// The size of a relation as number of blocks, or None if `allow_missing=true` and the relation
696 : /// does not exist.
697 : ///
698 : /// INVARIANT: never None if `allow_missing=false` (returns `NotFound` error instead).
699 : pub type GetRelSizeResponse = Option<u32>;
700 :
701 : impl From<proto::GetRelSizeResponse> for GetRelSizeResponse {
702 0 : fn from(pb: proto::GetRelSizeResponse) -> Self {
703 0 : (!pb.missing).then_some(pb.num_blocks)
704 0 : }
705 : }
706 :
707 : impl From<GetRelSizeResponse> for proto::GetRelSizeResponse {
708 0 : fn from(resp: GetRelSizeResponse) -> Self {
709 0 : Self {
710 0 : num_blocks: resp.unwrap_or_default(),
711 0 : missing: resp.is_none(),
712 0 : }
713 0 : }
714 : }
715 :
716 : /// Requests an SLRU segment. Only valid on shard 0, other shards will error.
717 : #[derive(Clone, Copy, Debug)]
718 : pub struct GetSlruSegmentRequest {
719 : pub read_lsn: ReadLsn,
720 : pub kind: SlruKind,
721 : pub segno: u32,
722 : }
723 :
724 : impl TryFrom<proto::GetSlruSegmentRequest> for GetSlruSegmentRequest {
725 : type Error = ProtocolError;
726 :
727 0 : fn try_from(pb: proto::GetSlruSegmentRequest) -> Result<Self, Self::Error> {
728 : Ok(Self {
729 0 : read_lsn: pb
730 0 : .read_lsn
731 0 : .ok_or(ProtocolError::Missing("read_lsn"))?
732 0 : .try_into()?,
733 0 : kind: u8::try_from(pb.kind)
734 0 : .ok()
735 0 : .and_then(SlruKind::from_repr)
736 0 : .ok_or_else(|| ProtocolError::invalid("slru_kind", pb.kind))?,
737 0 : segno: pb.segno,
738 : })
739 0 : }
740 : }
741 :
742 : impl From<GetSlruSegmentRequest> for proto::GetSlruSegmentRequest {
743 0 : fn from(request: GetSlruSegmentRequest) -> Self {
744 0 : Self {
745 0 : read_lsn: Some(request.read_lsn.into()),
746 0 : kind: request.kind as u32,
747 0 : segno: request.segno,
748 0 : }
749 0 : }
750 : }
751 :
752 : pub type GetSlruSegmentResponse = Bytes;
753 :
754 : impl TryFrom<proto::GetSlruSegmentResponse> for GetSlruSegmentResponse {
755 : type Error = ProtocolError;
756 :
757 0 : fn try_from(pb: proto::GetSlruSegmentResponse) -> Result<Self, Self::Error> {
758 0 : if pb.segment.is_empty() {
759 0 : return Err(ProtocolError::Missing("segment"));
760 0 : }
761 0 : Ok(pb.segment)
762 0 : }
763 : }
764 :
765 : impl From<GetSlruSegmentResponse> for proto::GetSlruSegmentResponse {
766 0 : fn from(segment: GetSlruSegmentResponse) -> Self {
767 0 : Self { segment }
768 0 : }
769 : }
770 :
771 : // SlruKind is defined in pageserver_api::reltag.
772 : pub type SlruKind = pageserver_api::reltag::SlruKind;
773 :
774 : /// Acquires or extends a lease on the given LSN. This guarantees that the Pageserver won't garbage
775 : /// collect the LSN until the lease expires.
776 : pub struct LeaseLsnRequest {
777 : /// The LSN to lease.
778 : pub lsn: Lsn,
779 : }
780 :
781 : impl TryFrom<proto::LeaseLsnRequest> for LeaseLsnRequest {
782 : type Error = ProtocolError;
783 :
784 0 : fn try_from(pb: proto::LeaseLsnRequest) -> Result<Self, Self::Error> {
785 0 : if pb.lsn == 0 {
786 0 : return Err(ProtocolError::Missing("lsn"));
787 0 : }
788 0 : Ok(Self { lsn: Lsn(pb.lsn) })
789 0 : }
790 : }
791 :
792 : impl From<LeaseLsnRequest> for proto::LeaseLsnRequest {
793 0 : fn from(request: LeaseLsnRequest) -> Self {
794 0 : Self { lsn: request.lsn.0 }
795 0 : }
796 : }
797 :
798 : /// Lease expiration time. If the lease could not be granted because the LSN has already been
799 : /// garbage collected, a FailedPrecondition status will be returned instead.
800 : pub type LeaseLsnResponse = SystemTime;
801 :
802 : impl TryFrom<proto::LeaseLsnResponse> for LeaseLsnResponse {
803 : type Error = ProtocolError;
804 :
805 0 : fn try_from(pb: proto::LeaseLsnResponse) -> Result<Self, Self::Error> {
806 0 : let expires = pb.expires.ok_or(ProtocolError::Missing("expires"))?;
807 0 : UNIX_EPOCH
808 0 : .checked_add(Duration::new(expires.seconds as u64, expires.nanos as u32))
809 0 : .ok_or_else(|| ProtocolError::invalid("expires", expires))
810 0 : }
811 : }
812 :
813 : impl From<LeaseLsnResponse> for proto::LeaseLsnResponse {
814 0 : fn from(response: LeaseLsnResponse) -> Self {
815 0 : let expires = response.duration_since(UNIX_EPOCH).unwrap_or_default();
816 0 : Self {
817 0 : expires: Some(prost_types::Timestamp {
818 0 : seconds: expires.as_secs() as i64,
819 0 : nanos: expires.subsec_nanos() as i32,
820 0 : }),
821 0 : }
822 0 : }
823 : }
|
