diff options
| -rw-r--r-- | docs/openapi.yaml | 20 | ||||
| -rw-r--r-- | src/api/mod.rs | 3 | ||||
| -rw-r--r-- | src/api/store.rs | 123 | ||||
| -rw-r--r-- | src/document/store.rs | 249 | ||||
| -rw-r--r-- | src/document/typed.rs | 21 |
5 files changed, 317 insertions, 99 deletions
diff --git a/docs/openapi.yaml b/docs/openapi.yaml index 36c3ddc..6e106b9 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -118,6 +118,13 @@ paths: JSON maps and arrays become Automerge maps and lists; numbers become integers when they fit, floats otherwise; strings are plain strings (not text objects). + Types are sticky: plain JSON or text written over a value of a type JSON + doesn't have is stored as that type where it is its JSON form (a timestamp's + string, bytes' base64, a number over a float, uint or counter, a string over a + text; `api.md`, Types). A change of type is listed in `x-continuum-retyped`; + with `retype=false` it is `409` instead. Answers with what is there after the + write, read back. + Typed JSON (`application/vnd.continuum.typed+json`) writes values of every type, tagged (`{"$timestamp": …}`, `{"$bytes": …}`, `{"$uint": …}`, `{"$counter": …}`, `{"$float": …}`, `{"$text": …}`, `{"$map": …}`). @@ -126,6 +133,13 @@ paths: `.float`, `.boolean`, `.counter` (decimal or `true`/`false`), or `application/vnd.continuum.timestamp; format=rfc3339|unix|unix-ms|http-date`; a body that isn't one is `400`, saying why. + parameters: + - name: retype + in: query + description: Whether the write may change the type of what it writes over (default true); if not, a write that would is `409`. + schema: + type: boolean + default: true requestBody: required: true content: @@ -154,7 +168,7 @@ paths: '403': description: Protocol data (`_continuum`) and the user isn't in `protocol-writers`, or epoch metadata or the swarm ID (`_continuum/epochs`, `_continuum/swarm`, never writable). '409': - description: The path crosses an existing non-map value. + description: The path crosses an existing non-map value, or with `retype=false` the write would change a type (the message says where). '415': description: Content-Type is none the API takes (`application/json`, the default; `text/plain`; a type's own, `api.md`, Types). '500': @@ -1368,6 +1382,10 @@ components: description: SHA-256 of the body. schema: type: string + x-continuum-retyped: + description: 'On a write that changed the type of what it wrote over: where, and from what to what (`<path>: <from> -> <to>`, comma-separated, 16 at most, then `…`).' + schema: + type: string x-continuum-type: description: What the value is stored as (`map`, `list`, `text`, `string`, `bytes`, `int`, `uint`, `float`, `boolean`, `null`, `counter`, `timestamp`). schema: diff --git a/src/api/mod.rs b/src/api/mod.rs index 713f248..ede1cfa 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -211,7 +211,8 @@ pub(crate) fn store_status(error: StoreError, context: &str) -> ApiError { StoreError::Conflict | StoreError::NotAList | StoreError::PastTheEnd { .. } - | StoreError::NotACounter => StatusCode::CONFLICT, + | StoreError::NotACounter + | StoreError::Retyping(_) => StatusCode::CONFLICT, StoreError::NotFound | StoreError::UnknownHeads => StatusCode::NOT_FOUND, StoreError::Rejoining => StatusCode::SERVICE_UNAVAILABLE, other => { diff --git a/src/api/store.rs b/src/api/store.rs index 47aade1..e6921cd 100644 --- a/src/api/store.rs +++ b/src/api/store.rs @@ -291,11 +291,30 @@ fn mark_source( Ok(response) } -/// `PUT /store/{key}`. +#[derive(Debug, Deserialize)] +pub(crate) struct PutQuery { + /// Whether the write may change the type of what it writes over (`true`, the + /// default); if not, one that would is `409` (`store::Writing::retype`). + #[serde(default = "yes")] + retype: bool, +} + +fn yes() -> bool { + true +} + +/// Header listing the places whose type a write changed (`store::Retyped`), as +/// `<path>: <from> -> <to>`, comma-separated, `MAX_RETYPED_LISTED` at most (then `…`). +const RETYPED_HEADER: &str = "x-continuum-retyped"; +const MAX_RETYPED_LISTED: usize = 16; + +/// `PUT /store/{key}`: answered with what is there after the write, shown as `Accept` +/// asks (read back: with sticky types, a JSON value may be stored as another type). pub(crate) async fn put_store_entry( State(state): State<AppState>, Extension(user): Extension<AuthenticatedUser>, Path(key): Path<String>, + Query(query): Query<PutQuery>, RawQuery(raw_query): RawQuery, headers: HeaderMap, body: Bytes, @@ -308,19 +327,36 @@ pub(crate) async fn put_store_entry( if let Some(redirect) = to_positive(&state, &path, false, None, raw_query.as_deref()).await? { return Ok(redirect); } - let value = body_value(&headers, &body)?; - // Answered with the value written, shown as asked. - let shown = shown(headers.get(ACCEPT), value.kind())?; - let (held, json) = written(&value, shown); + let body = body_value(&headers, &body)?; + let accept = headers.get(ACCEPT).cloned(); + let writing = store::Writing::new(body.sticky, body.timestamps, query.retype); - let ((), heads) = state + // A value that can't be shown as asked is refused, and the write with it. + let ((retyped, read), heads) = state .mutate_doc_for(Some(user.0.clone()), move |doc| { - store::write_val_path(doc, &path, &value) + let retyped = store::write_val_path(doc, &path, &body.value, writing) + .map_err(|e| store_status(e, "Failed to write store value"))?; + let read = read_shown(&*doc, &path, accept.as_ref(), false)?; + Ok::<_, ApiError>((retyped, read)) }) - .await - .map_err(|e| store_status(e, "Failed to write store value"))?; + .await?; + let read = read.ok_or(ApiError(StatusCode::INTERNAL_SERVER_ERROR, None))?; - Ok(with_heads(show(&held, shown, json.as_ref())?, &heads)) + let mut response = with_heads(read.respond()?, &heads); + if !retyped.is_empty() { + let mut listed: Vec<String> = retyped + .iter() + .take(MAX_RETYPED_LISTED) + .map(ToString::to_string) + .collect(); + if retyped.len() > MAX_RETYPED_LISTED { + listed.push("…".to_owned()); + } + if let Ok(value) = HeaderValue::from_str(&listed.join(", ")) { + response.headers_mut().insert(RETYPED_HEADER, value); + } + } + Ok(response) } fn etag_for(bytes: &[u8]) -> String { @@ -530,9 +566,18 @@ fn respond(kind: Option<Kind>, content_type: &str, body: Vec<u8>) -> Result<Resp /// Header naming what a value read or written is stored as (`typed::Kind::name`). const TYPE_HEADER: &str = "x-continuum-type"; -/// A write's body as a value to write (`typed::Val`): JSON (the default), typed JSON, -/// text as a string, or a scalar of a type of its own (`typed::from_native`). -fn body_value(headers: &HeaderMap, body: &Bytes) -> Result<Val, ApiError> { +/// A write's body: the value to write (`typed::Val`), from JSON (the default), typed +/// JSON, text as a string, or a scalar of a type of its own (`typed::from_native`). +struct WriteBody { + value: Val, + /// Plain JSON or text, whose values keep the types of what they are written over + /// where they can (`store::Writing::sticky`). + sticky: bool, + /// How timestamps are written in it (`timestamp-format`). + timestamps: TimestampFormat, +} + +fn body_value(headers: &HeaderMap, body: &Bytes) -> Result<WriteBody, ApiError> { let content_type = headers .get(CONTENT_TYPE) .and_then(|value| value.to_str().ok()) @@ -543,38 +588,30 @@ fn body_value(headers: &HeaderMap, body: &Bytes) -> Result<Val, ApiError> { let json = || { serde_json::from_slice::<JsonValue>(body).map_err(|error| bad(format!("not JSON: {error}"))) }; - match media.as_str() { - "application/json" => Val::from_json(&json()?).map_err(bad), - typed::TYPED_JSON => { - let form = json_form_of(params)?; - Val::from_typed_json(&json()?, form.timestamps).map_err(bad) - } - typed::TEXT => String::from_utf8(body.to_vec()) - .map(|text| Val::Scalar(automerge::ScalarValue::Str(text.into()))) - .map_err(|_| bad("not UTF-8".to_owned())), + let timestamps = json_form_of(params)?.timestamps; + let (value, sticky) = match media.as_str() { + "application/json" => (Val::from_json(&json()?).map_err(bad)?, true), + typed::TYPED_JSON => ( + Val::from_typed_json(&json()?, timestamps).map_err(bad)?, + false, + ), + typed::TEXT => ( + String::from_utf8(body.to_vec()) + .map(|text| Val::Scalar(automerge::ScalarValue::Str(text.into()))) + .map_err(|_| bad("not UTF-8".to_owned()))?, + true, + ), media => match typed::from_native(media, params, body) { - Some(Ok(scalar)) => Ok(Val::Scalar(scalar)), - Some(Err(error)) => Err(bad(error)), - None => Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into()), - }, - } -} - -/// What `value` is stored as, and its JSON of `shown`'s form, to answer its write with. -fn written(value: &Val, shown: Shown) -> (store::Held, Option<JsonValue>) { - let held = store::Held { - kind: value.kind(), - value: match value { - Val::Scalar(scalar) => Some(scalar.clone()), - Val::Text(text) => Some(automerge::ScalarValue::Str(text.as_str().into())), - Val::Map(_) | Val::List(_) => None, + Some(Ok(scalar)) => (Val::Scalar(scalar), false), + Some(Err(error)) => return Err(bad(error)), + None => return Err(StatusCode::UNSUPPORTED_MEDIA_TYPE.into()), }, }; - let form = match shown { - Shown::Json(form) => form, - Shown::Text | Shown::Native(_) => JsonForm::default(), - }; - (held, value.to_json(form)) + Ok(WriteBody { + value, + sticky, + timestamps, + }) } /// `POST /store/{key}`: inserts the body into a list, `key` ending in the index it @@ -630,7 +667,7 @@ pub(crate) async fn post_store_entry( let value = body_value(&headers, &body)?; let inserted = state .mutate_doc_for(Some(user.0.clone()), move |doc| { - store::insert_val_path(doc, &path, &value) + store::insert_val_path(doc, &path, &value.value) }) .await; match inserted { diff --git a/src/document/store.rs b/src/document/store.rs index 4dd9d02..07f2e1a 100644 --- a/src/document/store.rs +++ b/src/document/store.rs @@ -9,7 +9,7 @@ use serde::Serialize; use serde_json::Value as JsonValue; use std::collections::BTreeSet; -use super::typed::{self, JsonForm, Kind, Val}; +use super::typed::{self, JsonForm, Kind, TimestampFormat, Val}; /// Store paths touched by one batch of document changes (`AppState::change_events`). #[derive(Debug)] @@ -103,6 +103,10 @@ pub(crate) enum StoreError { /// An increment of something that isn't a counter (`increment_path`). #[error("not a counter")] NotACounter, + /// A write that would change the type of what it writes over, where that isn't + /// wanted (`Writing::retype`). + #[error("would change {0}")] + Retyping(Retyped), /// A read at heads the document doesn't have (`at::At`). #[error("this node doesn't have these heads (yet)")] UnknownHeads, @@ -632,19 +636,27 @@ pub(crate) fn write_path( path: &[String], value: &JsonValue, ) -> Result<(), StoreError> { - write_val_path(doc, path, &from_json(value)?) + write_val_path( + doc, + path, + &from_json(value)?, + Writing::sticky(TimestampFormat::default()), + ) + .map(|_| ()) } /// [`write_path`] for a value of any type (`typed::Val`): from typed JSON, or a scalar -/// of a type JSON doesn't have. +/// of a type JSON doesn't have. Returns the places whose type the write changed. pub(crate) fn write_val_path( doc: &mut impl Transactable, path: &[String], value: &Val, -) -> Result<(), StoreError> { + mut writing: Writing, +) -> Result<Vec<Retyped>, StoreError> { let (leaf, parents) = path.split_last().ok_or(StoreError::NotFound)?; + writing.path = parents.to_vec(); match parent_for_write(doc, parents)? { - Parent::Maps(maps) => update_key(doc, &maps, leaf, value)?, + Parent::Maps(maps) => update_key(doc, &maps, leaf, value, &mut writing)?, Parent::List(list) => { let index = list_index(leaf).ok_or(StoreError::NotFound)?; let length = doc.length(&list); @@ -652,13 +664,139 @@ pub(crate) fn write_val_path( return Err(StoreError::NotFound); } if index < length { - update_item(doc, &list, index, value)?; + update_item(doc, &list, index, value, &mut writing)?; } else { write_value(doc, &list, Slot::Index(index), value)?; } } } - Ok(()) + Ok(writing.retyped) +} + +/// How a write treats the types of what it writes over. +pub(crate) struct Writing { + /// Types are sticky: the value comes as plain JSON, which has fewer types than + /// Automerge, and where it is the JSON form of a value of the type there (a string + /// that reads as a timestamp, as base64, a number without a fraction over a float, + /// an unsigned integer or a counter, a string over a text), it is written as that + /// type (`stuck`). Typed JSON and a type's own media type say the type themselves. + pub(crate) sticky: bool, + /// How sticky timestamps are written in JSON (`typed::TimestampFormat::json`). + pub(crate) timestamps: TimestampFormat, + /// Whether the write may change the type of what it writes over; if not, one that + /// would is refused (`Retyping`). + pub(crate) retype: bool, + path: Vec<String>, + retyped: Vec<Retyped>, +} + +impl Writing { + pub(crate) fn sticky(timestamps: TimestampFormat) -> Self { + Self::new(true, timestamps, true) + } + + pub(crate) fn new(sticky: bool, timestamps: TimestampFormat, retype: bool) -> Self { + Self { + sticky, + timestamps, + retype, + path: Vec::new(), + retyped: Vec::new(), + } + } + + /// `kind` is written over `held` at the place at `segment`, if `held` is another + /// kind: noted, or with `retype` off, refused. + fn retyping(&mut self, segment: &str, held: Kind, kind: Kind) -> Result<(), StoreError> { + if held == kind { + return Ok(()); + } + let mut path = self.path.clone(); + path.push(segment.to_owned()); + let retyped = Retyped { + path, + from: held, + to: kind, + }; + if !self.retype { + return Err(StoreError::Retyping(retyped)); + } + self.retyped.push(retyped); + Ok(()) + } +} + +/// A place whose type a write changed. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct Retyped { + pub(crate) path: Vec<String>, + pub(crate) from: Kind, + pub(crate) to: Kind, +} + +impl std::fmt::Display for Retyped { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "{}: {} -> {}", + self.path.join("/"), + self.from.name(), + self.to.name() + ) + } +} + +/// What a value held is stored as. +fn kind_of(held: &Value<'_>) -> Kind { + match held { + Value::Object(ObjType::Map | ObjType::Table) => Kind::Map, + Value::Object(ObjType::List) => Kind::List, + Value::Object(ObjType::Text) => Kind::Text, + Value::Scalar(scalar) => Kind::of(scalar), + } +} + +/// `value` written as the type of `held`, where it is a JSON form of a value of it +/// (`Writing::sticky`); None where it isn't (it is written as it is). +fn stuck(held: &Value<'_>, value: &Val, timestamps: TimestampFormat) -> Option<Val> { + let Val::Scalar(scalar) = value else { + return None; + }; + let held = match held { + Value::Object(ObjType::Text) => { + return match scalar { + ScalarValue::Str(text) => Some(Val::Text(text.to_string())), + _ => None, + }; + } + Value::Object(_) => return None, + Value::Scalar(held) => held.as_ref(), + }; + let stuck = match (held, scalar) { + (ScalarValue::Timestamp(_), ScalarValue::Str(text)) => { + ScalarValue::Timestamp(timestamps.read(text)?) + } + (ScalarValue::Timestamp(_), ScalarValue::Int(_) | ScalarValue::F64(_)) + if matches!(timestamps, TimestampFormat::Unix | TimestampFormat::UnixMs) => + { + let number = match scalar { + ScalarValue::Int(i) => i.to_string(), + ScalarValue::F64(f) => f.to_string(), + _ => return None, + }; + ScalarValue::Timestamp(timestamps.read(&number)?) + } + (ScalarValue::Bytes(_), ScalarValue::Str(text)) => { + use base64::{Engine, engine::general_purpose::STANDARD as BASE64}; + ScalarValue::Bytes(BASE64.decode(text.as_bytes()).ok()?) + } + (ScalarValue::Uint(_), ScalarValue::Int(i)) => ScalarValue::Uint(u64::try_from(*i).ok()?), + (ScalarValue::F64(_), ScalarValue::Int(i)) => ScalarValue::F64(*i as f64), + (ScalarValue::F64(_), ScalarValue::Uint(u)) => ScalarValue::F64(*u as f64), + (ScalarValue::Counter(_), ScalarValue::Int(i)) => ScalarValue::counter(*i), + _ => return None, + }; + Some(Val::Scalar(stuck)) } fn from_json(value: &JsonValue) -> Result<Val, StoreError> { @@ -689,17 +827,40 @@ enum Update { Anew, } -fn update_of(holds: &[(Value<'static>, ObjId)], value: &Val) -> Update { - let [(held, id)] = holds else { - return Update::Anew; +/// What updating the place at `segment`, holding `holds`, to `value` comes to, with the +/// value to write there: `value`, or as the type there (`stuck`). A change of type is +/// noted (`Writing::retyping`). +fn update_of<'v>( + holds: &[(Value<'static>, ObjId)], + value: &'v Val, + segment: &str, + writing: &mut Writing, +) -> Result<(Update, std::borrow::Cow<'v, Val>), StoreError> { + use std::borrow::Cow; + let Some((held, _)) = holds.first() else { + return Ok((Update::Anew, Cow::Borrowed(value))); }; - match (held, value) { - (Value::Object(ObjType::Map), Val::Map(_)) => Update::Map(id.clone()), - (Value::Object(ObjType::List), Val::List(_)) => Update::List(id.clone()), - (Value::Object(ObjType::Text), Val::Text(_)) => Update::Text(id.clone()), - (Value::Scalar(held), Val::Scalar(scalar)) if held.as_ref() == scalar => Update::Same, - _ => Update::Anew, - } + let value = match writing + .sticky + .then(|| stuck(held, value, writing.timestamps)) + .flatten() + { + Some(stuck) => Cow::Owned(stuck), + None => Cow::Borrowed(value), + }; + let update = match (holds, &*value) { + ([(Value::Object(ObjType::Map), id)], Val::Map(_)) => Update::Map(id.clone()), + ([(Value::Object(ObjType::List), id)], Val::List(_)) => Update::List(id.clone()), + ([(Value::Object(ObjType::Text), id)], Val::Text(_)) => Update::Text(id.clone()), + ([(Value::Scalar(held), _)], Val::Scalar(scalar)) if held.as_ref() == scalar => { + Update::Same + } + _ => { + writing.retyping(segment, kind_of(held), value.kind())?; + Update::Anew + } + }; + Ok((update, value)) } /// Updates `key` of the merged maps `maps` to `value`. @@ -708,18 +869,23 @@ fn update_key( maps: &[ObjId], key: &str, value: &Val, + writing: &mut Writing, ) -> Result<(), StoreError> { - match update_of(&candidates(doc, maps, key)?, value) { + let (update, value) = update_of(&candidates(doc, maps, key)?, value, key, writing)?; + writing.path.push(key.to_owned()); + let done = match update { Update::Same => Ok(()), - Update::Map(map) => update_map(doc, &map, value), - Update::List(list) => update_list(doc, &list, value), - Update::Text(text) => update_text(doc, &text, value), + Update::Map(map) => update_map(doc, &map, &value, writing), + Update::List(list) => update_list(doc, &list, &value, writing), + Update::Text(text) => update_text(doc, &text, &value), Update::Anew => { // Cleared in every merged map, or an old map would merge into the new value. delete_in_all(doc, maps, key)?; - write_value(doc, &maps[0], Slot::Key(key), value) + write_value(doc, &maps[0], Slot::Key(key), &value) } - } + }; + writing.path.pop(); + done } /// Updates the item at `index` of `list` (which has it) to `value`. @@ -728,17 +894,29 @@ fn update_item( list: &ObjId, index: usize, value: &Val, + writing: &mut Writing, ) -> Result<(), StoreError> { - match update_of(&candidates(doc, std::slice::from_ref(list), index)?, value) { + let segment = index.to_string(); + let holds = candidates(doc, std::slice::from_ref(list), index)?; + let (update, value) = update_of(&holds, value, &segment, writing)?; + writing.path.push(segment); + let done = match update { Update::Same => Ok(()), - Update::Map(map) => update_map(doc, &map, value), - Update::List(inner) => update_list(doc, &inner, value), - Update::Text(text) => update_text(doc, &text, value), - Update::Anew => write_value(doc, list, Slot::At(index), value), - } + Update::Map(map) => update_map(doc, &map, &value, writing), + Update::List(inner) => update_list(doc, &inner, &value, writing), + Update::Text(text) => update_text(doc, &text, &value), + Update::Anew => write_value(doc, list, Slot::At(index), &value), + }; + writing.path.pop(); + done } -fn update_map(doc: &mut impl Transactable, map: &ObjId, value: &Val) -> Result<(), StoreError> { +fn update_map( + doc: &mut impl Transactable, + map: &ObjId, + value: &Val, + writing: &mut Writing, +) -> Result<(), StoreError> { let Val::Map(entries) = value else { unreachable!("a map is updated to a map"); }; @@ -750,19 +928,24 @@ fn update_map(doc: &mut impl Transactable, map: &ObjId, value: &Val) -> Result<( } } for (key, value) in entries { - update_key(doc, maps, key, value)?; + update_key(doc, maps, key, value, writing)?; } Ok(()) } -fn update_list(doc: &mut impl Transactable, list: &ObjId, value: &Val) -> Result<(), StoreError> { +fn update_list( + doc: &mut impl Transactable, + list: &ObjId, + value: &Val, + writing: &mut Writing, +) -> Result<(), StoreError> { let Val::List(items) = value else { unreachable!("a list is updated to a list"); }; let length = doc.length(list); for (index, item) in items.iter().enumerate() { if index < length { - update_item(doc, list, index, item)?; + update_item(doc, list, index, item, writing)?; } else { write_value(doc, list, Slot::Index(index), item)?; } diff --git a/src/document/typed.rs b/src/document/typed.rs index a542578..cf7f49b 100644 --- a/src/document/typed.rs +++ b/src/document/typed.rs @@ -519,25 +519,4 @@ impl Val { Val::Scalar(scalar) => Kind::of(scalar), } } - - /// In JSON of `form`. - pub(crate) fn to_json(&self, form: JsonForm) -> Option<JsonValue> { - Some(match self { - Val::Map(entries) => map_to_json( - entries - .iter() - .map(|(key, value)| Some((key.clone(), value.to_json(form)?))) - .collect::<Option<_>>()?, - form, - ), - Val::List(items) => JsonValue::Array( - items - .iter() - .map(|item| item.to_json(form)) - .collect::<Option<_>>()?, - ), - Val::Text(text) => text_to_json(text.clone(), form), - Val::Scalar(scalar) => scalar_to_json(scalar, form)?, - }) - } } |