• karol retitled this revision from [services] Blob - Add server reactor implementations - put reactor to [draft] [services] Blob - Add server reactor implementations - put reactor.Mar 24 2022, 2:47 AM

• karol mentioned this in D3464: [services] Backup - Add server reactor base classes - bidi reactor.Mar 24 2022, 7:44 AM

ashoat requested changes to this revision.Mar 24 2022, 1:35 PM

ashoat added inline comments.

services/blob/docker-server/contents/server/src/Reactors/server/PutReactor.h
29 ↗	(On Diff #10648)	It looks like first we send a message to set the `holder`, and then we send a message to set the `blobHash`. Can we set these at the same time, and avoid the extra round trips?
81–97 ↗	(On Diff #10648)	Indentation can be reduced here

This revision now requires changes to proceed.Mar 24 2022, 1:35 PM

rebase to master

thx for the comments, I'm gonna address them later, for now, I needed that rebase.

services/blob/docker-server/contents/server/src/Reactors/server/PutReactor.h
29 ↗	(On Diff #10648)	We can indeed.
81–97 ↗	(On Diff #10648)	right, thx

• karol planned changes to this revision.Mar 25 2022, 7:52 AM

Harbormaster completed remote builds in B7614: Diff 10690.Mar 25 2022, 8:51 AM

It looks like this code was moved from D3471 - it would be really helpful if you could mention cases like that, so that the review process is simplified.

In D3521#96040, @palys-swm wrote:

It looks like this code was moved from D3471 - it would be really helpful if you could mention cases like that, so that the review process is simplified.

You're partially right, but please note that the flow changes, instead of the loop we base here on the callback, so I think it is good to check once again if everything is implemented properly. But you're right that I could've said this is based on D3471 so we know the business logic should be the same.

• karol added inline comments.Mar 28 2022, 1:55 AM

services/blob/docker-server/contents/server/src/Reactors/server/PutReactor.h
29 ↗	(On Diff #10648)	Hey, getting back to this. I don't think this is a good place to change that. In this diff, we want to refactor from sync to async API and what you're mentioning is changing the structure/API of the service. Such doubt should have been raised in D2666 and taken care of there. If you feel like changing this as you mentioned, I can create a task for this and do it separately. Please, let me know if it is important to you and should I create a task.

update

Harbormaster completed remote builds in B7632: Diff 10720.Mar 28 2022, 4:09 AM

• karol planned changes to this revision.Mar 28 2022, 5:52 AM

fix build

Harbormaster completed remote builds in B7643: Diff 10731.Mar 28 2022, 5:59 AM

ashoat accepted this revision.Mar 28 2022, 10:40 AM

ashoat added inline comments.

services/blob/docker-server/contents/server/src/Reactors/server/PutReactor.h
29 ↗	(On Diff #10648)	That makes sense – this will require `.proto` changes`, so it should be a separate diff. Yes, I would like to prioritize this – can please create a task? In general, we should be thoughtful about avoiding unnecessary round trips. I think we did a good job with this in the backup service architecture review, but looks like we missed one thing in the blob service architecture review.

This revision is now accepted and ready to land.Mar 28 2022, 10:40 AM

• karol added inline comments.Mar 29 2022, 1:36 AM

services/blob/docker-server/contents/server/src/Reactors/server/PutReactor.h
29 ↗	(On Diff #10648)	https://linear.app/comm/issue/ENG-937/try-to-group-messages-in-the-blob-service

remove logs

removed draft label

Harbormaster completed remote builds in B7659: Diff 10754.Mar 29 2022, 1:48 AM

tomek accepted this revision.Mar 30 2022, 4:26 AM

This revision is now accepted and ready to land.Mar 30 2022, 4:26 AM

• karol planned changes to this revision.Apr 1 2022, 3:20 AM

• karol added inline comments.

services/blob/src/Reactors/server/PutReactor.h
72 ↗	(On Diff #10754)	Wrong condition

fix condition in done callback

This revision is now accepted and ready to land.Apr 1 2022, 3:21 AM

Harbormaster completed remote builds in B7794: Diff 10936.Apr 1 2022, 3:34 AM

rebase?

This revision was landed with ongoing or failed builds.Apr 4 2022, 2:40 AM

Closed by commit rCOMM0f1c274ceed3: [services] Blob - Add server reactor implementations - put reactor (authored by • karol). · Explain Why

This revision was automatically updated to reflect the committed changes.

• karol added a commit: rCOMM0f1c274ceed3: [services] Blob - Add server reactor implementations - put reactor.

Harbormaster completed remote builds in B7814: Diff 10978.Apr 4 2022, 11:08 AM

Revision Contents
Changeset List

Path

Size

services/

blob/

src/

Reactors/

server/

PutReactor.h

102 lines

Diff 10720

View Options

services/blob/src/Reactors/server/PutReactor.h

use aws_sdk_dynamodb::types::AttributeValue;		use aws_sdk_dynamodb::types::AttributeValue;
use aws_sdk_dynamodb::Error as DynamoDBError;		use aws_sdk_dynamodb::Error as DynamoDBError;
use chrono::{DateTime, Utc};		use chrono::{DateTime, Utc};
use std::collections::HashMap;		use std::collections::HashMap;
use std::fmt::{Display, Formatter};		use std::fmt::{Display, Formatter};
use std::num::ParseIntError;		use std::num::ParseIntError;
		use std::str::FromStr;

#[derive(		#[derive(
Debug, derive_more::Display, derive_more::From, derive_more::Error,		Debug, derive_more::Display, derive_more::From, derive_more::Error,
)]		)]
pub enum Error {		pub enum Error {
#[display(...)]		#[display(...)]
AwsSdk(DynamoDBError),		AwsSdk(DynamoDBError),
#[display(...)]		#[display(...)]
Show All 35 Lines

#[derive(Debug, derive_more::Display, derive_more::Error)]		#[derive(Debug, derive_more::Display, derive_more::Error)]
pub enum DBItemAttributeError {		pub enum DBItemAttributeError {
#[display(...)]		#[display(...)]
Missing,		Missing,
#[display(...)]		#[display(...)]
IncorrectType,		IncorrectType,
#[display(...)]		#[display(...)]
		TimestampOutOfRange,
		#[display(...)]
InvalidTimestamp(chrono::ParseError),		InvalidTimestamp(chrono::ParseError),
#[display(...)]		#[display(...)]
InvalidNumberFormat(ParseIntError),		InvalidNumberFormat(ParseIntError),
}		}

pub fn parse_string_attribute(		pub fn parse_string_attribute(
attribute_name: &'static str,		attribute_name: &'static str,
attribute_value: Option<AttributeValue>,		attribute_value: Option<AttributeValue>,
Show All 27 Lines	match attribute_value {
None => Err(DBItemError::new(		None => Err(DBItemError::new(
attribute_name,		attribute_name,
Value::AttributeValue(attribute_value),		Value::AttributeValue(attribute_value),
DBItemAttributeError::Missing,		DBItemAttributeError::Missing,
)),		)),
}		}
}		}

		pub fn parse_int_attribute<T>(
		attribute_name: &'static str,
		attribute_value: Option<AttributeValue>,
		) -> Result<T, DBItemError>
		where
		T: FromStr<Err = ParseIntError>,
		{
		match &attribute_value {
		Some(AttributeValue::N(numeric_str)) => {
		parse_integer(attribute_name, &numeric_str)
		}
		Some(_) => Err(DBItemError::new(
		attribute_name,
		Value::AttributeValue(attribute_value),
		DBItemAttributeError::IncorrectType,
		)),
		None => Err(DBItemError::new(
		attribute_name,
		Value::AttributeValue(attribute_value),
		DBItemAttributeError::Missing,
		)),
		}
		}

pub fn parse_datetime_attribute(		pub fn parse_datetime_attribute(
attribute_name: &'static str,		attribute_name: &'static str,
attribute_value: Option<AttributeValue>,		attribute_value: Option<AttributeValue>,
) -> Result<DateTime<Utc>, DBItemError> {		) -> Result<DateTime<Utc>, DBItemError> {
if let Some(AttributeValue::S(datetime)) = &attribute_value {		if let Some(AttributeValue::S(datetime)) = &attribute_value {
// parse() accepts a relaxed RFC3339 string		// parse() accepts a relaxed RFC3339 string
datetime.parse().map_err(\|e\| {		datetime.parse().map_err(\|e\| {
DBItemError::new(		DBItemError::new(
attribute_name,		attribute_name,
Value::AttributeValue(attribute_value),		Value::AttributeValue(attribute_value),
DBItemAttributeError::InvalidTimestamp(e),		DBItemAttributeError::InvalidTimestamp(e),
)		)
})		})
} else {		} else {
Err(DBItemError::new(		Err(DBItemError::new(
attribute_name,		attribute_name,
Value::AttributeValue(attribute_value),		Value::AttributeValue(attribute_value),
DBItemAttributeError::Missing,		DBItemAttributeError::Missing,
))		))
}		}
}		}

		/// Parses the UTC timestamp in milliseconds from a DynamoDB numeric attribute
		pub fn parse_timestamp_attribute(
		attribute_name: &'static str,
		attribute_value: Option<AttributeValue>,
		) -> Result<DateTime<Utc>, DBItemError> {
		let timestamp =
		parse_int_attribute::<i64>(attribute_name, attribute_value.clone())?;
		let naive_datetime = chrono::NaiveDateTime::from_timestamp_millis(timestamp)
		.ok_or_else(\|\| {
		DBItemError::new(
		attribute_name,
		Value::AttributeValue(attribute_value),
		DBItemAttributeError::TimestampOutOfRange,
		)
		})?;
		Ok(DateTime::from_utc(naive_datetime, Utc))
		}

pub fn parse_map_attribute(		pub fn parse_map_attribute(
attribute_name: &'static str,		attribute_name: &'static str,
attribute_value: Option<AttributeValue>,		attribute_value: Option<AttributeValue>,
) -> Result<HashMap<String, AttributeValue>, DBItemError> {		) -> Result<HashMap<String, AttributeValue>, DBItemError> {
match attribute_value {		match attribute_value {
Some(AttributeValue::M(map)) => Ok(map),		Some(AttributeValue::M(map)) => Ok(map),
Some(_) => Err(DBItemError::new(		Some(_) => Err(DBItemError::new(
attribute_name,		attribute_name,
Value::AttributeValue(attribute_value),		Value::AttributeValue(attribute_value),
DBItemAttributeError::IncorrectType,		DBItemAttributeError::IncorrectType,
)),		)),
None => Err(DBItemError::new(		None => Err(DBItemError::new(
attribute_name,		attribute_name,
Value::AttributeValue(attribute_value),		Value::AttributeValue(attribute_value),
DBItemAttributeError::Missing,		DBItemAttributeError::Missing,
)),		)),
}		}
}		}

pub fn parse_number(		pub fn parse_integer<T>(
attribute_name: &'static str,		attribute_name: &'static str,
attribute_value: &str,		attribute_value: &str,
) -> Result<i32, DBItemError> {		) -> Result<T, DBItemError>
attribute_value.parse::<i32>().map_err(\|e\| {		where
		T: FromStr<Err = ParseIntError>,
		{
		attribute_value.parse::<T>().map_err(\|e\| {
DBItemError::new(		DBItemError::new(
attribute_name,		attribute_name,
Value::String(attribute_value.to_string()),		Value::String(attribute_value.to_string()),
DBItemAttributeError::InvalidNumberFormat(e),		DBItemAttributeError::InvalidNumberFormat(e),
)		)
})		})
}		}

		#[cfg(test)]

		mod tests {
		use super::*;

		#[test]
		fn test_parse_integer() {
		assert!(parse_integer::<i32>("some_attr", "123").is_ok());
		assert!(parse_integer::<i32>("negative", "-123").is_ok());

		assert!(parse_integer::<i32>("float", "3.14").is_err());
		assert!(parse_integer::<i32>("NaN", "foo").is_err());

		assert!(parse_integer::<u32>("negative_uint", "-123").is_err());
		assert!(parse_integer::<u8>("too_large", "65536").is_err());
		}

		#[test]
		fn test_parse_timestamp() {
		let timestamp = Utc::now().timestamp_millis();
		let attr = AttributeValue::N(timestamp.to_string());

		let parsed_timestamp = parse_timestamp_attribute("some_attr", Some(attr));

		assert!(parsed_timestamp.is_ok());
		assert_eq!(parsed_timestamp.unwrap().timestamp_millis(), timestamp);
		}

		#[test]
		fn test_parse_invalid_timestamp() {
		let attr = AttributeValue::N("foo".to_string());
		let parsed_timestamp = parse_timestamp_attribute("some_attr", Some(attr));
		assert!(parsed_timestamp.is_err());
		}

		#[test]
		fn test_parse_timestamp_out_of_range() {
		let attr = AttributeValue::N(i64::MAX.to_string());
		let parsed_timestamp = parse_timestamp_attribute("some_attr", Some(attr));
		assert!(parsed_timestamp.is_err());
		assert!(matches!(
		parsed_timestamp.unwrap_err().attribute_error,
		DBItemAttributeError::TimestampOutOfRange
		));
		}
		}

[services] Blob - Add server reactor implementations - put reactorClosedPublicActions

Details

Diff Detail

Event Timeline

Revision ContentsChangeset List

Diff 10720

services/blob/src/Reactors/server/PutReactor.h

[services] Blob - Add server reactor implementations - put reactor
ClosedPublic
Actions

Revision Contents
Changeset List