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)]
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 : /// Checks whether a relation exists, at the given LSN. Only valid on shard 0, other shards error.
143 : #[derive(Clone, Copy, Debug)]
144 : pub struct CheckRelExistsRequest {
145 : pub read_lsn: ReadLsn,
146 : pub rel: RelTag,
147 : }
148 :
149 : impl TryFrom<proto::CheckRelExistsRequest> for CheckRelExistsRequest {
150 : type Error = ProtocolError;
151 :
152 0 : fn try_from(pb: proto::CheckRelExistsRequest) -> Result<Self, Self::Error> {
153 : Ok(Self {
154 0 : read_lsn: pb
155 0 : .read_lsn
156 0 : .ok_or(ProtocolError::Missing("read_lsn"))?
157 0 : .try_into()?,
158 0 : rel: pb.rel.ok_or(ProtocolError::Missing("rel"))?.try_into()?,
159 : })
160 0 : }
161 : }
162 :
163 : impl From<CheckRelExistsRequest> for proto::CheckRelExistsRequest {
164 0 : fn from(request: CheckRelExistsRequest) -> Self {
165 0 : Self {
166 0 : read_lsn: Some(request.read_lsn.into()),
167 0 : rel: Some(request.rel.into()),
168 0 : }
169 0 : }
170 : }
171 :
172 : pub type CheckRelExistsResponse = bool;
173 :
174 : impl From<proto::CheckRelExistsResponse> for CheckRelExistsResponse {
175 0 : fn from(pb: proto::CheckRelExistsResponse) -> Self {
176 0 : pb.exists
177 0 : }
178 : }
179 :
180 : impl From<CheckRelExistsResponse> for proto::CheckRelExistsResponse {
181 0 : fn from(exists: CheckRelExistsResponse) -> Self {
182 0 : Self { exists }
183 0 : }
184 : }
185 :
186 : /// Requests a base backup.
187 : #[derive(Clone, Copy, Debug)]
188 : pub struct GetBaseBackupRequest {
189 : /// The LSN to fetch a base backup at. If None, uses the latest LSN known to the Pageserver.
190 : pub lsn: Option<Lsn>,
191 : /// If true, logical replication slots will not be created.
192 : pub replica: bool,
193 : /// If true, include relation files in the base backup. Mainly for debugging and tests.
194 : pub full: bool,
195 : /// Compression algorithm to use. Base backups send a compressed payload instead of using gRPC
196 : /// compression, so that we can cache compressed backups on the server.
197 : pub compression: BaseBackupCompression,
198 : }
199 :
200 : impl TryFrom<proto::GetBaseBackupRequest> for GetBaseBackupRequest {
201 : type Error = ProtocolError;
202 :
203 0 : fn try_from(pb: proto::GetBaseBackupRequest) -> Result<Self, Self::Error> {
204 : Ok(Self {
205 0 : lsn: (pb.lsn != 0).then_some(Lsn(pb.lsn)),
206 0 : replica: pb.replica,
207 0 : full: pb.full,
208 0 : compression: pb.compression.try_into()?,
209 : })
210 0 : }
211 : }
212 :
213 : impl From<GetBaseBackupRequest> for proto::GetBaseBackupRequest {
214 0 : fn from(request: GetBaseBackupRequest) -> Self {
215 0 : Self {
216 0 : lsn: request.lsn.unwrap_or_default().0,
217 0 : replica: request.replica,
218 0 : full: request.full,
219 0 : compression: request.compression.into(),
220 0 : }
221 0 : }
222 : }
223 :
224 : /// Base backup compression algorithm.
225 : #[derive(Clone, Copy, Debug)]
226 : pub enum BaseBackupCompression {
227 : None,
228 : Gzip,
229 : }
230 :
231 : impl TryFrom<proto::BaseBackupCompression> for BaseBackupCompression {
232 : type Error = ProtocolError;
233 :
234 0 : fn try_from(pb: proto::BaseBackupCompression) -> Result<Self, Self::Error> {
235 0 : match pb {
236 0 : proto::BaseBackupCompression::Unknown => Err(ProtocolError::invalid("compression", pb)),
237 0 : proto::BaseBackupCompression::None => Ok(Self::None),
238 0 : proto::BaseBackupCompression::Gzip => Ok(Self::Gzip),
239 : }
240 0 : }
241 : }
242 :
243 : impl TryFrom<i32> for BaseBackupCompression {
244 : type Error = ProtocolError;
245 :
246 0 : fn try_from(compression: i32) -> Result<Self, Self::Error> {
247 0 : proto::BaseBackupCompression::try_from(compression)
248 0 : .map_err(|_| ProtocolError::invalid("compression", compression))
249 0 : .and_then(Self::try_from)
250 0 : }
251 : }
252 :
253 : impl From<BaseBackupCompression> for proto::BaseBackupCompression {
254 0 : fn from(compression: BaseBackupCompression) -> Self {
255 0 : match compression {
256 0 : BaseBackupCompression::None => Self::None,
257 0 : BaseBackupCompression::Gzip => Self::Gzip,
258 : }
259 0 : }
260 : }
261 :
262 : impl From<BaseBackupCompression> for i32 {
263 0 : fn from(compression: BaseBackupCompression) -> Self {
264 0 : proto::BaseBackupCompression::from(compression).into()
265 0 : }
266 : }
267 :
268 : pub type GetBaseBackupResponseChunk = Bytes;
269 :
270 : impl TryFrom<proto::GetBaseBackupResponseChunk> for GetBaseBackupResponseChunk {
271 : type Error = ProtocolError;
272 :
273 0 : fn try_from(pb: proto::GetBaseBackupResponseChunk) -> Result<Self, Self::Error> {
274 0 : if pb.chunk.is_empty() {
275 0 : return Err(ProtocolError::Missing("chunk"));
276 0 : }
277 0 : Ok(pb.chunk)
278 0 : }
279 : }
280 :
281 : impl From<GetBaseBackupResponseChunk> for proto::GetBaseBackupResponseChunk {
282 0 : fn from(chunk: GetBaseBackupResponseChunk) -> Self {
283 0 : Self { chunk }
284 0 : }
285 : }
286 :
287 : /// Requests the size of a database, as # of bytes. Only valid on shard 0, other shards will error.
288 : #[derive(Clone, Copy, Debug)]
289 : pub struct GetDbSizeRequest {
290 : pub read_lsn: ReadLsn,
291 : pub db_oid: Oid,
292 : }
293 :
294 : impl TryFrom<proto::GetDbSizeRequest> for GetDbSizeRequest {
295 : type Error = ProtocolError;
296 :
297 0 : fn try_from(pb: proto::GetDbSizeRequest) -> Result<Self, Self::Error> {
298 : Ok(Self {
299 0 : read_lsn: pb
300 0 : .read_lsn
301 0 : .ok_or(ProtocolError::Missing("read_lsn"))?
302 0 : .try_into()?,
303 0 : db_oid: pb.db_oid,
304 : })
305 0 : }
306 : }
307 :
308 : impl From<GetDbSizeRequest> for proto::GetDbSizeRequest {
309 0 : fn from(request: GetDbSizeRequest) -> Self {
310 0 : Self {
311 0 : read_lsn: Some(request.read_lsn.into()),
312 0 : db_oid: request.db_oid,
313 0 : }
314 0 : }
315 : }
316 :
317 : pub type GetDbSizeResponse = u64;
318 :
319 : impl From<proto::GetDbSizeResponse> for GetDbSizeResponse {
320 0 : fn from(pb: proto::GetDbSizeResponse) -> Self {
321 0 : pb.num_bytes
322 0 : }
323 : }
324 :
325 : impl From<GetDbSizeResponse> for proto::GetDbSizeResponse {
326 0 : fn from(num_bytes: GetDbSizeResponse) -> Self {
327 0 : Self { num_bytes }
328 0 : }
329 : }
330 :
331 : /// Requests one or more pages.
332 : #[derive(Clone, Debug)]
333 : pub struct GetPageRequest {
334 : /// A request ID. Will be included in the response. Should be unique for in-flight requests on
335 : /// the stream.
336 : pub request_id: RequestID,
337 : /// The request class.
338 : pub request_class: GetPageClass,
339 : /// The LSN to read at.
340 : pub read_lsn: ReadLsn,
341 : /// The relation to read from.
342 : pub rel: RelTag,
343 : /// Page numbers to read. Must belong to the remote shard.
344 : ///
345 : /// Multiple pages will be executed as a single batch by the Pageserver, amortizing layer access
346 : /// costs and parallelizing them. This may increase the latency of any individual request, but
347 : /// improves the overall latency and throughput of the batch as a whole.
348 : pub block_numbers: Vec<u32>,
349 : }
350 :
351 : impl TryFrom<proto::GetPageRequest> for GetPageRequest {
352 : type Error = ProtocolError;
353 :
354 0 : fn try_from(pb: proto::GetPageRequest) -> Result<Self, Self::Error> {
355 0 : if pb.block_number.is_empty() {
356 0 : return Err(ProtocolError::Missing("block_number"));
357 0 : }
358 : Ok(Self {
359 0 : request_id: pb.request_id,
360 0 : request_class: pb.request_class.into(),
361 0 : read_lsn: pb
362 0 : .read_lsn
363 0 : .ok_or(ProtocolError::Missing("read_lsn"))?
364 0 : .try_into()?,
365 0 : rel: pb.rel.ok_or(ProtocolError::Missing("rel"))?.try_into()?,
366 0 : block_numbers: pb.block_number,
367 : })
368 0 : }
369 : }
370 :
371 : impl From<GetPageRequest> for proto::GetPageRequest {
372 0 : fn from(request: GetPageRequest) -> Self {
373 0 : Self {
374 0 : request_id: request.request_id,
375 0 : request_class: request.request_class.into(),
376 0 : read_lsn: Some(request.read_lsn.into()),
377 0 : rel: Some(request.rel.into()),
378 0 : block_number: request.block_numbers,
379 0 : }
380 0 : }
381 : }
382 :
383 : /// A GetPage request ID.
384 : pub type RequestID = u64;
385 :
386 : /// A GetPage request class.
387 : #[derive(Clone, Copy, Debug)]
388 : pub enum GetPageClass {
389 : /// Unknown class. For backwards compatibility: used when an older client version sends a class
390 : /// that a newer server version has removed.
391 : Unknown,
392 : /// A normal request. This is the default.
393 : Normal,
394 : /// A prefetch request. NB: can only be classified on pg < 18.
395 : Prefetch,
396 : /// A background request (e.g. vacuum).
397 : Background,
398 : }
399 :
400 : impl From<proto::GetPageClass> for GetPageClass {
401 0 : fn from(pb: proto::GetPageClass) -> Self {
402 0 : match pb {
403 0 : proto::GetPageClass::Unknown => Self::Unknown,
404 0 : proto::GetPageClass::Normal => Self::Normal,
405 0 : proto::GetPageClass::Prefetch => Self::Prefetch,
406 0 : proto::GetPageClass::Background => Self::Background,
407 : }
408 0 : }
409 : }
410 :
411 : impl From<i32> for GetPageClass {
412 0 : fn from(class: i32) -> Self {
413 0 : proto::GetPageClass::try_from(class)
414 0 : .unwrap_or(proto::GetPageClass::Unknown)
415 0 : .into()
416 0 : }
417 : }
418 :
419 : impl From<GetPageClass> for proto::GetPageClass {
420 0 : fn from(class: GetPageClass) -> Self {
421 0 : match class {
422 0 : GetPageClass::Unknown => Self::Unknown,
423 0 : GetPageClass::Normal => Self::Normal,
424 0 : GetPageClass::Prefetch => Self::Prefetch,
425 0 : GetPageClass::Background => Self::Background,
426 : }
427 0 : }
428 : }
429 :
430 : impl From<GetPageClass> for i32 {
431 0 : fn from(class: GetPageClass) -> Self {
432 0 : proto::GetPageClass::from(class).into()
433 0 : }
434 : }
435 :
436 : /// A GetPage response.
437 : ///
438 : /// A batch response will contain all of the requested pages. We could eagerly emit individual pages
439 : /// as soon as they are ready, but on a readv() Postgres holds buffer pool locks on all pages in the
440 : /// batch and we'll only return once the entire batch is ready, so no one can make use of the
441 : /// individual pages.
442 : #[derive(Clone, Debug)]
443 : pub struct GetPageResponse {
444 : /// The original request's ID.
445 : pub request_id: RequestID,
446 : /// The response status code.
447 : pub status_code: GetPageStatusCode,
448 : /// A string describing the status, if any.
449 : pub reason: Option<String>,
450 : /// The 8KB page images, in the same order as the request. Empty if status != OK.
451 : pub page_images: Vec<Bytes>,
452 : }
453 :
454 : impl From<proto::GetPageResponse> for GetPageResponse {
455 0 : fn from(pb: proto::GetPageResponse) -> Self {
456 : Self {
457 0 : request_id: pb.request_id,
458 0 : status_code: pb.status_code.into(),
459 0 : reason: Some(pb.reason).filter(|r| !r.is_empty()),
460 0 : page_images: pb.page_image,
461 : }
462 0 : }
463 : }
464 :
465 : impl From<GetPageResponse> for proto::GetPageResponse {
466 0 : fn from(response: GetPageResponse) -> Self {
467 0 : Self {
468 0 : request_id: response.request_id,
469 0 : status_code: response.status_code.into(),
470 0 : reason: response.reason.unwrap_or_default(),
471 0 : page_image: response.page_images,
472 0 : }
473 0 : }
474 : }
475 :
476 : impl GetPageResponse {
477 : /// Attempts to represent a tonic::Status as a GetPageResponse if appropriate. Returning a
478 : /// tonic::Status will terminate the GetPage stream, so per-request errors are emitted as a
479 : /// GetPageResponse with a non-OK status code instead.
480 : #[allow(clippy::result_large_err)]
481 0 : pub fn try_from_status(
482 0 : status: tonic::Status,
483 0 : request_id: RequestID,
484 0 : ) -> Result<Self, tonic::Status> {
485 : // We shouldn't see an OK status here, because we're emitting an error.
486 0 : debug_assert_ne!(status.code(), tonic::Code::Ok);
487 0 : if status.code() == tonic::Code::Ok {
488 0 : return Err(tonic::Status::internal(format!(
489 0 : "unexpected OK status: {status:?}",
490 0 : )));
491 0 : }
492 :
493 : // If we can't convert the tonic::Code to a GetPageStatusCode, this is not a per-request
494 : // error and we should return a tonic::Status to terminate the stream.
495 0 : let Ok(status_code) = status.code().try_into() else {
496 0 : return Err(status);
497 : };
498 :
499 : // Return a GetPageResponse for the status.
500 0 : Ok(Self {
501 0 : request_id,
502 0 : status_code,
503 0 : reason: Some(status.message().to_string()),
504 0 : page_images: Vec::new(),
505 0 : })
506 0 : }
507 : }
508 :
509 : /// A GetPage response status code.
510 : ///
511 : /// These are effectively equivalent to gRPC statuses. However, we use a bidirectional stream
512 : /// (potentially shared by many backends), and a gRPC status response would terminate the stream so
513 : /// we send GetPageResponse messages with these codes instead.
514 : #[derive(Clone, Copy, Debug, PartialEq, strum_macros::Display)]
515 : pub enum GetPageStatusCode {
516 : /// Unknown status. For forwards compatibility: used when an older client version receives a new
517 : /// status code from a newer server version.
518 : Unknown,
519 : /// The request was successful.
520 : Ok,
521 : /// The page did not exist. The tenant/timeline/shard has already been validated during stream
522 : /// setup.
523 : NotFound,
524 : /// The request was invalid.
525 : InvalidRequest,
526 : /// The request failed due to an internal server error.
527 : InternalError,
528 : /// The tenant is rate limited. Slow down and retry later.
529 : SlowDown,
530 : }
531 :
532 : impl From<proto::GetPageStatusCode> for GetPageStatusCode {
533 0 : fn from(pb: proto::GetPageStatusCode) -> Self {
534 0 : match pb {
535 0 : proto::GetPageStatusCode::Unknown => Self::Unknown,
536 0 : proto::GetPageStatusCode::Ok => Self::Ok,
537 0 : proto::GetPageStatusCode::NotFound => Self::NotFound,
538 0 : proto::GetPageStatusCode::InvalidRequest => Self::InvalidRequest,
539 0 : proto::GetPageStatusCode::InternalError => Self::InternalError,
540 0 : proto::GetPageStatusCode::SlowDown => Self::SlowDown,
541 : }
542 0 : }
543 : }
544 :
545 : impl From<i32> for GetPageStatusCode {
546 0 : fn from(status_code: i32) -> Self {
547 0 : proto::GetPageStatusCode::try_from(status_code)
548 0 : .unwrap_or(proto::GetPageStatusCode::Unknown)
549 0 : .into()
550 0 : }
551 : }
552 :
553 : impl From<GetPageStatusCode> for proto::GetPageStatusCode {
554 0 : fn from(status_code: GetPageStatusCode) -> Self {
555 0 : match status_code {
556 0 : GetPageStatusCode::Unknown => Self::Unknown,
557 0 : GetPageStatusCode::Ok => Self::Ok,
558 0 : GetPageStatusCode::NotFound => Self::NotFound,
559 0 : GetPageStatusCode::InvalidRequest => Self::InvalidRequest,
560 0 : GetPageStatusCode::InternalError => Self::InternalError,
561 0 : GetPageStatusCode::SlowDown => Self::SlowDown,
562 : }
563 0 : }
564 : }
565 :
566 : impl From<GetPageStatusCode> for i32 {
567 0 : fn from(status_code: GetPageStatusCode) -> Self {
568 0 : proto::GetPageStatusCode::from(status_code).into()
569 0 : }
570 : }
571 :
572 : impl TryFrom<tonic::Code> for GetPageStatusCode {
573 : type Error = tonic::Code;
574 :
575 0 : fn try_from(code: tonic::Code) -> Result<Self, Self::Error> {
576 : use tonic::Code;
577 :
578 0 : let status_code = match code {
579 0 : Code::Ok => Self::Ok,
580 :
581 : // These are per-request errors, which should be returned as GetPageResponses.
582 0 : Code::AlreadyExists => Self::InvalidRequest,
583 0 : Code::DataLoss => Self::InternalError,
584 0 : Code::FailedPrecondition => Self::InvalidRequest,
585 0 : Code::InvalidArgument => Self::InvalidRequest,
586 0 : Code::Internal => Self::InternalError,
587 0 : Code::NotFound => Self::NotFound,
588 0 : Code::OutOfRange => Self::InvalidRequest,
589 0 : Code::ResourceExhausted => Self::SlowDown,
590 :
591 : // These should terminate the stream by returning a tonic::Status.
592 : Code::Aborted
593 : | Code::Cancelled
594 : | Code::DeadlineExceeded
595 : | Code::PermissionDenied
596 : | Code::Unauthenticated
597 : | Code::Unavailable
598 : | Code::Unimplemented
599 0 : | Code::Unknown => return Err(code),
600 : };
601 0 : Ok(status_code)
602 0 : }
603 : }
604 :
605 : // Fetches the size of a relation at a given LSN, as # of blocks. Only valid on shard 0, other
606 : // shards will error.
607 : #[derive(Clone, Copy, Debug)]
608 : pub struct GetRelSizeRequest {
609 : pub read_lsn: ReadLsn,
610 : pub rel: RelTag,
611 : }
612 :
613 : impl TryFrom<proto::GetRelSizeRequest> for GetRelSizeRequest {
614 : type Error = ProtocolError;
615 :
616 0 : fn try_from(proto: proto::GetRelSizeRequest) -> Result<Self, Self::Error> {
617 : Ok(Self {
618 0 : read_lsn: proto
619 0 : .read_lsn
620 0 : .ok_or(ProtocolError::Missing("read_lsn"))?
621 0 : .try_into()?,
622 0 : rel: proto.rel.ok_or(ProtocolError::Missing("rel"))?.try_into()?,
623 : })
624 0 : }
625 : }
626 :
627 : impl From<GetRelSizeRequest> for proto::GetRelSizeRequest {
628 0 : fn from(request: GetRelSizeRequest) -> Self {
629 0 : Self {
630 0 : read_lsn: Some(request.read_lsn.into()),
631 0 : rel: Some(request.rel.into()),
632 0 : }
633 0 : }
634 : }
635 :
636 : pub type GetRelSizeResponse = u32;
637 :
638 : impl From<proto::GetRelSizeResponse> for GetRelSizeResponse {
639 0 : fn from(proto: proto::GetRelSizeResponse) -> Self {
640 0 : proto.num_blocks
641 0 : }
642 : }
643 :
644 : impl From<GetRelSizeResponse> for proto::GetRelSizeResponse {
645 0 : fn from(num_blocks: GetRelSizeResponse) -> Self {
646 0 : Self { num_blocks }
647 0 : }
648 : }
649 :
650 : /// Requests an SLRU segment. Only valid on shard 0, other shards will error.
651 : #[derive(Clone, Copy, Debug)]
652 : pub struct GetSlruSegmentRequest {
653 : pub read_lsn: ReadLsn,
654 : pub kind: SlruKind,
655 : pub segno: u32,
656 : }
657 :
658 : impl TryFrom<proto::GetSlruSegmentRequest> for GetSlruSegmentRequest {
659 : type Error = ProtocolError;
660 :
661 0 : fn try_from(pb: proto::GetSlruSegmentRequest) -> Result<Self, Self::Error> {
662 : Ok(Self {
663 0 : read_lsn: pb
664 0 : .read_lsn
665 0 : .ok_or(ProtocolError::Missing("read_lsn"))?
666 0 : .try_into()?,
667 0 : kind: u8::try_from(pb.kind)
668 0 : .ok()
669 0 : .and_then(SlruKind::from_repr)
670 0 : .ok_or_else(|| ProtocolError::invalid("slru_kind", pb.kind))?,
671 0 : segno: pb.segno,
672 : })
673 0 : }
674 : }
675 :
676 : impl From<GetSlruSegmentRequest> for proto::GetSlruSegmentRequest {
677 0 : fn from(request: GetSlruSegmentRequest) -> Self {
678 0 : Self {
679 0 : read_lsn: Some(request.read_lsn.into()),
680 0 : kind: request.kind as u32,
681 0 : segno: request.segno,
682 0 : }
683 0 : }
684 : }
685 :
686 : pub type GetSlruSegmentResponse = Bytes;
687 :
688 : impl TryFrom<proto::GetSlruSegmentResponse> for GetSlruSegmentResponse {
689 : type Error = ProtocolError;
690 :
691 0 : fn try_from(pb: proto::GetSlruSegmentResponse) -> Result<Self, Self::Error> {
692 0 : if pb.segment.is_empty() {
693 0 : return Err(ProtocolError::Missing("segment"));
694 0 : }
695 0 : Ok(pb.segment)
696 0 : }
697 : }
698 :
699 : impl From<GetSlruSegmentResponse> for proto::GetSlruSegmentResponse {
700 0 : fn from(segment: GetSlruSegmentResponse) -> Self {
701 0 : Self { segment }
702 0 : }
703 : }
704 :
705 : // SlruKind is defined in pageserver_api::reltag.
706 : pub type SlruKind = pageserver_api::reltag::SlruKind;
707 :
708 : /// Acquires or extends a lease on the given LSN. This guarantees that the Pageserver won't garbage
709 : /// collect the LSN until the lease expires.
710 : pub struct LeaseLsnRequest {
711 : /// The LSN to lease.
712 : pub lsn: Lsn,
713 : }
714 :
715 : impl TryFrom<proto::LeaseLsnRequest> for LeaseLsnRequest {
716 : type Error = ProtocolError;
717 :
718 0 : fn try_from(pb: proto::LeaseLsnRequest) -> Result<Self, Self::Error> {
719 0 : if pb.lsn == 0 {
720 0 : return Err(ProtocolError::Missing("lsn"));
721 0 : }
722 0 : Ok(Self { lsn: Lsn(pb.lsn) })
723 0 : }
724 : }
725 :
726 : impl From<LeaseLsnRequest> for proto::LeaseLsnRequest {
727 0 : fn from(request: LeaseLsnRequest) -> Self {
728 0 : Self { lsn: request.lsn.0 }
729 0 : }
730 : }
731 :
732 : /// Lease expiration time. If the lease could not be granted because the LSN has already been
733 : /// garbage collected, a FailedPrecondition status will be returned instead.
734 : pub type LeaseLsnResponse = SystemTime;
735 :
736 : impl TryFrom<proto::LeaseLsnResponse> for LeaseLsnResponse {
737 : type Error = ProtocolError;
738 :
739 0 : fn try_from(pb: proto::LeaseLsnResponse) -> Result<Self, Self::Error> {
740 0 : let expires = pb.expires.ok_or(ProtocolError::Missing("expires"))?;
741 0 : UNIX_EPOCH
742 0 : .checked_add(Duration::new(expires.seconds as u64, expires.nanos as u32))
743 0 : .ok_or_else(|| ProtocolError::invalid("expires", expires))
744 0 : }
745 : }
746 :
747 : impl From<LeaseLsnResponse> for proto::LeaseLsnResponse {
748 0 : fn from(response: LeaseLsnResponse) -> Self {
749 0 : let expires = response.duration_since(UNIX_EPOCH).unwrap_or_default();
750 0 : Self {
751 0 : expires: Some(prost_types::Timestamp {
752 0 : seconds: expires.as_secs() as i64,
753 0 : nanos: expires.subsec_nanos() as i32,
754 0 : }),
755 0 : }
756 0 : }
757 : }
|
