aven_core/sync/
wire.rs

1use std::collections::HashSet;
2
3use anyhow::{Context, Result, bail};
4use serde::{Deserialize, Serialize};
5use serde_json::Value;
6
7use crate::attachments::validation::{
8    MAX_BLOB_BYTES, validate_alt_text, validate_blob_size, validate_dimensions, validate_filename,
9    validate_media_type,
10};
11use crate::change_log::op_type;
12use crate::ids::{BASE32, MetadataFieldId, ProjectId, WorkspaceId};
13use crate::task_fields::TaskField;
14
15mod changes;
16mod envelope;
17mod recurrence;
18#[cfg(test)]
19mod test_support;
20#[cfg(test)]
21mod tests;
22
23pub const SYNC_PROTOCOL_VERSION: u32 = 18;
24const MAX_CHANGE_PAYLOAD_BYTES: usize = 64 * 1024;
25
26pub(crate) fn serialize_change_payload(payload: &Value) -> Result<String> {
27    let serialized = serde_json::to_string(payload)?;
28    if serialized.len() > MAX_CHANGE_PAYLOAD_BYTES {
29        return Err(crate::error::CoreError::validation(format!(
30            "error invalid-sync-change payload-too-large limit={MAX_CHANGE_PAYLOAD_BYTES}"
31        ))
32        .into());
33    }
34    Ok(serialized)
35}
36
37pub fn sync_server_url_is_valid(server: &str) -> bool {
38    url::Url::parse(server)
39        .as_ref()
40        .is_ok_and(sync_server_url_is_valid_url)
41}
42
43pub(crate) fn sync_server_url_is_valid_url(url: &url::Url) -> bool {
44    matches!(url.scheme(), "http" | "https")
45        && url.host_str().is_some()
46        && url.username().is_empty()
47        && url.password().is_none()
48        && url.query().is_none()
49        && url.fragment().is_none()
50}
51pub const MAX_PUSH_BATCH: usize = 256;
52/// Full decoded JSON metadata request allowance, shared by page selection and HTTP extraction.
53/// Compatible with the 2 MiB default body allowance in Axum 0.8.
54pub const MAX_SYNC_REQUEST_BYTES: usize = 2 * 1024 * 1024;
55pub const MAX_PULL_BATCH: u32 = 512;
56pub const MAX_BLOB_TRANSFER_OBJECTS: usize = 16;
57pub const MAX_BLOB_TRANSFER_BYTES: u64 = 64 * 1024 * 1024;
58pub const DAEMON_SYNC_PAGE_BUDGET: usize = 8;
59pub const DAEMON_INCOMPLETE_RESCHEDULE_MS: u64 = 100;
60
61#[derive(Debug, Clone, Serialize, Deserialize)]
62pub struct ChangeWire {
63    pub change_id: String,
64    pub client_id: String,
65    pub local_seq: i64,
66    pub entity_type: String,
67    pub entity_id: String,
68    pub field: Option<String>,
69    pub op_type: String,
70    pub payload: Value,
71    pub base_version: Option<String>,
72    pub created_at: String,
73    pub server_seq: Option<i64>,
74}
75
76#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
77pub(crate) struct AttachmentAddPayload {
78    pub(crate) workspace_id: String,
79    pub(crate) workspace_key: String,
80    pub(crate) attachment_id: String,
81    pub(crate) sha256: String,
82    pub(crate) byte_size: i64,
83    pub(crate) media_type: String,
84    pub(crate) filename: Option<String>,
85    pub(crate) alt_text: Option<String>,
86    pub(crate) width: Option<i64>,
87    pub(crate) height: Option<i64>,
88    pub(crate) created_at: String,
89}
90
91impl AttachmentAddPayload {
92    pub(crate) fn from_change(change: &ChangeWire) -> Result<Self> {
93        validate_attachment_change_envelope(change, op_type::ATTACHMENT_ADD)?;
94        required_workspace_payload(&change.payload)?;
95        let attachment_id = required_string_payload("attachment_id", &change.payload)?;
96        ensure_sync_id("attachment_id", &attachment_id)?;
97        let sha256 = required_string_payload("sha256", &change.payload)?;
98        validate_sha256_for_sync(&sha256)?;
99        let byte_size = required_i64_payload("byte_size", &change.payload)?;
100        validate_blob_size_for_sync(byte_size)?;
101        let media_type = required_string_payload("media_type", &change.payload)?;
102        map_attachment_validation(validate_media_type(&media_type))?;
103        let filename = optional_string_payload("filename", &change.payload)?;
104        map_attachment_validation(validate_filename(filename.as_deref()))?;
105        let alt_text = optional_string_payload("alt_text", &change.payload)?;
106        map_attachment_validation(validate_alt_text(alt_text.as_deref()))?;
107        let width = optional_i64_payload("width", &change.payload)?;
108        let height = optional_i64_payload("height", &change.payload)?;
109        map_attachment_validation(validate_dimensions(width, height))?;
110        required_timestamp_payload("created_at", &change.payload)?;
111        deserialize_attachment_payload(&change.payload)
112    }
113}
114
115#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
116pub(crate) struct AttachmentDeletePayload {
117    pub(crate) workspace_id: String,
118    pub(crate) workspace_key: String,
119    pub(crate) attachment_id: String,
120    pub(crate) deleted_at: String,
121}
122
123impl AttachmentDeletePayload {
124    pub(crate) fn from_change(change: &ChangeWire) -> Result<Self> {
125        validate_attachment_change_envelope(change, op_type::ATTACHMENT_DELETE)?;
126        required_workspace_payload(&change.payload)?;
127        let attachment_id = required_string_payload("attachment_id", &change.payload)?;
128        ensure_sync_id("attachment_id", &attachment_id)?;
129        required_timestamp_payload("deleted_at", &change.payload)?;
130        deserialize_attachment_payload(&change.payload)
131    }
132}
133
134fn deserialize_attachment_payload<T>(payload: &Value) -> Result<T>
135where
136    T: serde::de::DeserializeOwned,
137{
138    serde_json::from_value(payload.clone())
139        .map_err(|error| anyhow::anyhow!("error invalid-sync-change payload {error}"))
140}
141
142fn validate_attachment_change_envelope(change: &ChangeWire, expected_op_type: &str) -> Result<()> {
143    if change.op_type != expected_op_type {
144        bail!("error invalid-sync-change op_type={}", change.op_type);
145    }
146    ensure_entity_type(change, "task")?;
147    ensure_sync_id("entity_id", &change.entity_id)?;
148    if change.field.as_deref() != Some("attachments") {
149        bail!("error invalid-sync-change field=attachments");
150    }
151    if !change.payload.is_object() {
152        bail!("error invalid-sync-change payload expected-object");
153    }
154    serialize_change_payload(&change.payload)?;
155    Ok(())
156}
157
158#[derive(Debug, Clone, Serialize, Deserialize)]
159pub struct PushAck {
160    pub change_id: String,
161    pub server_seq: i64,
162}
163
164pub fn validate_blob_contracts(blobs: &[BlobUploadContract]) -> Result<()> {
165    envelope::validate_blob_contracts(blobs)
166}
167
168pub fn validate_blob_hashes(hashes: &[String]) -> Result<()> {
169    envelope::validate_blob_hashes(hashes)
170}
171
172pub fn validate_sync_protocol_version(client: u32, server: u32) -> Result<()> {
173    envelope::validate_sync_protocol_version(client, server)
174}
175
176pub fn validate_sync_request_protocol_version(client: Option<u32>) -> Result<()> {
177    envelope::validate_sync_request_protocol_version(client)
178}
179
180pub fn request_pull_limit(requested: Option<u32>) -> Result<u32> {
181    envelope::request_pull_limit(requested)
182}
183
184#[derive(Debug, Clone, Copy)]
185pub struct ValidatedSyncRequestEnvelope {
186    pub after: i64,
187    pub pull_limit: u32,
188    pub push_count: usize,
189}
190
191pub fn validate_sync_request_envelope(
192    request: &SyncRequest,
193) -> Result<ValidatedSyncRequestEnvelope> {
194    envelope::validate_sync_request_envelope(request)
195}
196
197pub(crate) fn validate_request_at_protocol(
198    request: &SyncRequest,
199    protocol: u32,
200) -> Result<ValidatedSyncRequestEnvelope> {
201    envelope::validate_request_at_protocol(request, protocol)
202}
203
204pub fn validate_sync_response_for_request(
205    after: i64,
206    pull_limit: u32,
207    request_change_ids: &[String],
208    response: &SyncResponse,
209) -> Result<()> {
210    envelope::validate_sync_response_for_request(after, pull_limit, request_change_ids, response)
211}
212
213pub(crate) fn validate_response_at_protocol(
214    protocol: u32,
215    after: i64,
216    pull_limit: u32,
217    request_change_ids: &[String],
218    response: &SyncResponse,
219) -> Result<()> {
220    envelope::validate_response_at_protocol(
221        protocol,
222        after,
223        pull_limit,
224        request_change_ids,
225        response,
226    )
227}
228
229#[derive(Debug)]
230pub struct ChangeRow {
231    pub change_id: String,
232    pub client_id: String,
233    pub local_seq: i64,
234    pub entity_type: String,
235    pub entity_id: String,
236    pub field: Option<String>,
237    pub op_type: String,
238    pub payload: String,
239    pub base_version: Option<String>,
240    pub created_at: String,
241    pub server_seq: Option<i64>,
242}
243
244impl ChangeRow {
245    pub fn into_wire(self) -> ChangeWire {
246        ChangeWire {
247            change_id: self.change_id,
248            client_id: self.client_id,
249            local_seq: self.local_seq,
250            entity_type: self.entity_type,
251            entity_id: self.entity_id,
252            field: self.field,
253            op_type: self.op_type,
254            payload: serde_json::from_str(&self.payload).unwrap_or(Value::Null),
255            base_version: self.base_version,
256            created_at: self.created_at,
257            server_seq: self.server_seq,
258        }
259    }
260}
261
262#[derive(Debug, Clone, Serialize, Deserialize)]
263pub struct SyncRequest {
264    #[serde(default)]
265    pub protocol_version: Option<u32>,
266    pub client_id: String,
267    pub after: i64,
268    #[serde(default)]
269    pub pull_limit: Option<u32>,
270    pub changes: Vec<ChangeWire>,
271}
272
273#[derive(Debug, Serialize, Deserialize)]
274pub struct SyncResponse {
275    pub protocol_version: u32,
276    pub cursor: i64,
277    pub has_more: bool,
278    #[serde(default)]
279    pub push_acks: Vec<PushAck>,
280    pub changes: Vec<ChangeWire>,
281}
282
283#[derive(Debug, Clone, Serialize, Deserialize)]
284pub struct BlobUploadContract {
285    pub workspace_id: String,
286    pub sha256: String,
287    pub byte_size: i64,
288    pub media_type: String,
289    pub width: i64,
290    pub height: i64,
291}
292
293#[derive(Debug, Serialize, Deserialize)]
294pub struct MissingBlobsRequest {
295    pub blobs: Vec<BlobUploadContract>,
296}
297
298#[derive(Debug, Serialize, Deserialize)]
299pub struct MissingBlobsResponse {
300    pub missing: Vec<String>,
301}
302
303#[derive(Debug, Clone, Copy, PartialEq, Eq)]
304enum ChangeDirection {
305    Pushed,
306    Pulled,
307}
308
309pub fn validate_pushed_change(change: &ChangeWire) -> Result<()> {
310    validate_local_change_shape(change)?;
311    super::protocol::validate_change(SYNC_PROTOCOL_VERSION, change)
312}
313
314pub(crate) fn validate_local_change_shape(change: &ChangeWire) -> Result<()> {
315    validate_change_shape(change, ChangeDirection::Pushed)
316}
317
318fn validate_pulled_change(change: &ChangeWire) -> Result<()> {
319    validate_change_shape(change, ChangeDirection::Pulled)
320}
321
322fn validate_change_shape(change: &ChangeWire, direction: ChangeDirection) -> Result<()> {
323    ensure_non_empty("change_id", &change.change_id)?;
324    ensure_non_empty("client_id", &change.client_id)?;
325    ensure_non_empty("entity_id", &change.entity_id)?;
326    ensure_non_empty("op_type", &change.op_type)?;
327    ensure_non_empty("entity_type", &change.entity_type)?;
328    if direction == ChangeDirection::Pushed {
329        ensure_sync_id("change_id", &change.change_id)?;
330    }
331    validate_change_server_seq(change, direction)?;
332    if !change.payload.is_object() {
333        bail!("error invalid-sync-change payload expected-object");
334    }
335    serialize_change_payload(&change.payload)?;
336
337    match change.op_type.as_str() {
338        op_type::CREATE_WORKSPACE => changes::validate_create_workspace(change)?,
339        op_type::SET_WORKSPACE_FIELD => changes::validate_set_workspace_field(change)?,
340        op_type::CREATE_PROJECT => changes::validate_create_project(change)?,
341        op_type::SET_PROJECT_METADATA => changes::validate_set_project_metadata(change)?,
342        op_type::CREATE_LABEL => changes::validate_create_label(change)?,
343        op_type::CREATE_METADATA_FIELD => changes::validate_create_metadata_field(change)?,
344        op_type::SET_METADATA_FIELD => changes::validate_set_metadata_field(change)?,
345        op_type::SET_TASK_METADATA | op_type::REMOVE_TASK_METADATA => {
346            changes::validate_task_metadata(change)?
347        }
348        op_type::CREATE_TASK => changes::validate_create_task(change)?,
349        op_type::SET_FIELD | op_type::RESOLVE_FIELD => changes::validate_task_field(change)?,
350        op_type::LABEL_ADD | op_type::LABEL_REMOVE => changes::validate_label_change(change)?,
351        op_type::NOTE_ADD => changes::validate_note_add(change)?,
352        op_type::DEPENDENCY_ADD | op_type::DEPENDENCY_REMOVE => {
353            changes::validate_dependency_change(change)?
354        }
355        op_type::RELATED_ADD | op_type::RELATED_REMOVE => changes::validate_related_change(change)?,
356        op_type::EPIC_LINK_ADD | op_type::EPIC_LINK_REMOVE => {
357            changes::validate_epic_link_change(change)?
358        }
359        op_type::PROJECT_DELETE => changes::validate_project_delete(change)?,
360        op_type::SET_LABEL_NAME => changes::validate_set_label_name(change)?,
361        op_type::LABEL_DELETE => changes::validate_label_delete(change)?,
362        op_type::NOTE_EDIT => changes::validate_note_edit(change)?,
363        op_type::LABEL_RESTORE => changes::validate_label_restore(change)?,
364        op_type::NOTE_DELETE => changes::validate_note_delete(change)?,
365        op_type::SET_RECURRENCE_METADATA | op_type::REMOVE_RECURRENCE_METADATA => {
366            changes::validate_recurrence_metadata(change)?
367        }
368        op_type::CREATE_RECURRENCE_SERIES => recurrence::validate_recurrence_create(change)?,
369        op_type::UPDATE_RECURRENCE_TEMPLATE => recurrence::validate_recurrence_template(change)?,
370        op_type::PROJECT_RECURRENCE_OCCURRENCE => {
371            recurrence::validate_recurrence_projection(change)?
372        }
373        op_type::RESOLVE_RECURRENCE_OCCURRENCE => recurrence::validate_recurrence_outcome(change)?,
374        op_type::SET_RECURRENCE_STATE | op_type::STOP_RECURRENCE_SERIES => {
375            recurrence::validate_recurrence_state(change)?
376        }
377        op_type::OPEN_RECURRENCE_PAUSE => recurrence::validate_recurrence_pause(change, true)?,
378        op_type::CLOSE_RECURRENCE_PAUSE => recurrence::validate_recurrence_pause(change, false)?,
379        op_type::ATTACHMENT_ADD => {
380            AttachmentAddPayload::from_change(change)?;
381        }
382        op_type::ATTACHMENT_DELETE => {
383            AttachmentDeletePayload::from_change(change)?;
384        }
385        _ => bail!("error invalid-sync-change op_type={}", change.op_type),
386    }
387    Ok(())
388}
389
390fn validate_sha256_for_sync(value: &str) -> Result<()> {
391    if value.len() == 64
392        && value
393            .bytes()
394            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
395    {
396        Ok(())
397    } else {
398        bail!("error invalid-sync-change invalid-sha256");
399    }
400}
401
402fn validate_blob_size_for_sync(byte_size: i64) -> Result<()> {
403    let Ok(bytes) = usize::try_from(byte_size) else {
404        bail!("error invalid-sync-change error invalid-attachment-size bytes={byte_size}");
405    };
406    if bytes == 0 || bytes > MAX_BLOB_BYTES {
407        bail!("error invalid-sync-change error invalid-attachment-size bytes={byte_size}");
408    }
409    map_attachment_validation(validate_blob_size(bytes))
410}
411
412fn map_attachment_validation(result: Result<()>) -> Result<()> {
413    result.map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))
414}
415
416fn validate_change_server_seq(change: &ChangeWire, direction: ChangeDirection) -> Result<()> {
417    match direction {
418        ChangeDirection::Pushed if change.server_seq.is_some() => {
419            bail!("error invalid-sync-change server_seq client-supplied");
420        }
421        ChangeDirection::Pulled => match change.server_seq {
422            Some(server_seq) if server_seq > 0 => {}
423            Some(server_seq) => {
424                bail!("error invalid-sync-change server_seq={server_seq}");
425            }
426            None => bail!("error invalid-sync-change server_seq missing"),
427        },
428        ChangeDirection::Pushed => {}
429    }
430    Ok(())
431}
432
433fn validate_sync_task_field_value(field: TaskField, value: &str) -> Result<()> {
434    field
435        .validate_value(value)
436        .map_err(|err| anyhow::anyhow!("error invalid-sync-change {err}"))
437}
438
439fn ensure_entity_type(change: &ChangeWire, expected: &str) -> Result<()> {
440    if change.entity_type == expected {
441        Ok(())
442    } else {
443        bail!(
444            "error invalid-sync-change op_type={} entity_type={} expected={}",
445            change.op_type,
446            change.entity_type,
447            expected
448        )
449    }
450}
451
452fn ensure_non_empty(name: &str, value: &str) -> Result<()> {
453    if value.trim().is_empty() {
454        bail!("error invalid-sync-change {name} empty");
455    }
456    Ok(())
457}
458
459fn ensure_sync_id(name: &str, value: &str) -> Result<()> {
460    if value.len() == 16 && value.bytes().all(|byte| BASE32.contains(&byte)) {
461        Ok(())
462    } else {
463        bail!("error invalid-sync-change {name} invalid-id");
464    }
465}
466
467fn ensure_project_id(name: &str, value: &str) -> Result<()> {
468    if value.parse::<ProjectId>().is_ok() {
469        Ok(())
470    } else {
471        bail!("error invalid-sync-change {name} invalid-id");
472    }
473}
474
475fn required_string_payload(key: &str, payload: &Value) -> Result<String> {
476    payload
477        .get(key)
478        .and_then(Value::as_str)
479        .map(str::to_string)
480        .with_context(|| format!("error invalid-sync-change payload.{key} missing"))
481}
482
483fn required_i64_payload(key: &str, payload: &Value) -> Result<i64> {
484    match payload.get(key) {
485        Some(Value::Number(value)) => value
486            .as_i64()
487            .with_context(|| format!("error invalid-sync-change payload.{key} invalid")),
488        Some(Value::Null) | None => bail!("error invalid-sync-change payload.{key} missing"),
489        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
490    }
491}
492
493fn required_timestamp_payload(key: &str, payload: &Value) -> Result<String> {
494    let value = required_string_payload(key, payload)?;
495    validate_timestamp_value(&format!("payload.{key}"), &value)?;
496    Ok(value)
497}
498
499fn validate_timestamp_value(label: &str, value: &str) -> Result<()> {
500    if value.ends_with('Z') && chrono::DateTime::parse_from_rfc3339(value).is_ok() {
501        Ok(())
502    } else {
503        bail!("error invalid-sync-change {label} invalid-timestamp");
504    }
505}
506
507fn required_workspace_payload(payload: &Value) -> Result<()> {
508    let id = required_string_payload("workspace_id", payload)?;
509    if id.parse::<WorkspaceId>().is_err() {
510        bail!("error invalid-sync-change workspace_id invalid-id");
511    }
512    required_string_payload("workspace_key", payload)?;
513    Ok(())
514}
515
516fn optional_workspace_payload(payload: &Value) -> Result<()> {
517    if payload.get("workspace_id").is_none() && payload.get("workspace_key").is_none() {
518        return Ok(());
519    }
520    required_workspace_payload(payload)
521}
522
523fn optional_string_payload(key: &str, payload: &Value) -> Result<Option<String>> {
524    match payload.get(key) {
525        Some(Value::String(value)) => Ok(Some(value.clone())),
526        Some(Value::Null) | None => Ok(None),
527        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
528    }
529}
530
531fn optional_i64_payload(key: &str, payload: &Value) -> Result<Option<i64>> {
532    match payload.get(key) {
533        Some(Value::Number(value)) => value
534            .as_i64()
535            .map(Some)
536            .with_context(|| format!("error invalid-sync-change payload.{key} invalid")),
537        Some(Value::Null) | None => Ok(None),
538        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
539    }
540}
541
542fn optional_bool_payload(key: &str, payload: &Value) -> Result<Option<bool>> {
543    match payload.get(key) {
544        Some(Value::Bool(value)) => Ok(Some(*value)),
545        Some(Value::Null) | None => Ok(None),
546        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
547    }
548}
549
550fn string_array_payload(key: &str, payload: &Value) -> Result<Vec<String>> {
551    payload
552        .get(key)
553        .and_then(Value::as_array)
554        .with_context(|| format!("error invalid-sync-change payload.{key} invalid"))?
555        .iter()
556        .map(|value| {
557            value
558                .as_str()
559                .map(str::to_string)
560                .with_context(|| format!("error invalid-sync-change payload.{key} invalid"))
561        })
562        .collect()
563}
564
565fn optional_string_array_payload(key: &str, payload: &Value) -> Result<()> {
566    match payload.get(key) {
567        Some(Value::Array(values))
568            if values
569                .iter()
570                .all(|value| value.as_str().is_some_and(|value| !value.trim().is_empty())) =>
571        {
572            Ok(())
573        }
574        Some(Value::Null) | None => Ok(()),
575        Some(_) => bail!("error invalid-sync-change payload.{key} invalid"),
576    }
577}
578
579fn ensure_metadata_field_id(name: &str, value: &str) -> Result<()> {
580    value
581        .parse::<MetadataFieldId>()
582        .map(|_| ())
583        .map_err(|_| anyhow::anyhow!("error invalid-sync-change {name} invalid-id"))
584}
585
586fn validate_metadata_key_payload(key: &str, payload: &Value) -> Result<String> {
587    let value = required_string_payload(key, payload)?;
588    let normalized = crate::metadata::normalize_metadata_key(&value)
589        .map_err(|_| anyhow::anyhow!("error invalid-sync-change payload.{key} invalid"))?;
590    if normalized != value {
591        bail!("error invalid-sync-change payload.{key} noncanonical");
592    }
593    Ok(value)
594}
595
596fn validate_metadata_value_payload(key: &str, payload: &Value) -> Result<String> {
597    let value = required_string_payload(key, payload)?;
598    if value.len() > crate::metadata::MAX_METADATA_VALUE_BYTES {
599        bail!("error invalid-sync-change payload.{key} too-large");
600    }
601    Ok(value)
602}
603
604fn validate_metadata_array_payload(payload: &Value) -> Result<()> {
605    let Some(values) = payload.get("metadata") else {
606        return Ok(());
607    };
608    let values = values
609        .as_array()
610        .context("error invalid-sync-change payload.metadata invalid")?;
611    if values.len() > crate::metadata::MAX_METADATA_VALUES {
612        bail!("error invalid-sync-change payload.metadata too-many");
613    }
614    let mut ids = HashSet::new();
615    let mut keys = HashSet::new();
616    let mut total_bytes = 0usize;
617    for value in values {
618        let object = value
619            .as_object()
620            .context("error invalid-sync-change payload.metadata invalid")?;
621        let field_id = required_string_payload("field_id", value)?;
622        ensure_metadata_field_id("metadata.field_id", &field_id)?;
623        let key = validate_metadata_key_payload("key", value)?;
624        let metadata_value = validate_metadata_value_payload("value", value)?;
625        if !ids.insert(field_id) || !keys.insert(key) {
626            bail!("error invalid-sync-change payload.metadata duplicate");
627        }
628        if object.len() != 3 {
629            bail!("error invalid-sync-change payload.metadata invalid");
630        }
631        total_bytes = total_bytes.saturating_add(metadata_value.len());
632    }
633    if total_bytes > crate::metadata::MAX_METADATA_TOTAL_BYTES {
634        bail!("error invalid-sync-change payload.metadata too-large");
635    }
636    Ok(())
637}
aven_core/sync/wire.rs

aven_core/sync/
wire.rs
