aven_core/sync/persistence/

server.rs

use std::collections::HashSet;
use std::path::Path;
use std::time::Instant;

use anyhow::{Context, Result, bail};
use sqlx::SqliteConnection;

use super::{ServerSyncPage, ServerSyncResult};
use crate::change_log::op_type;
use crate::db::{Database, begin_immediate};
#[cfg(test)]
use crate::sync::wire::SyncRequest;
use crate::sync::wire::{ChangeRow, ChangeWire, PushAck};

impl Database {
    pub async fn persist_server_sync_page(&self, page: ServerSyncPage) -> Result<ServerSyncResult> {
        self.persist_server_sync_page_inner(page, None, crate::sync::wire::SYNC_PROTOCOL_VERSION)
            .await
    }

    pub async fn persist_server_sync_page_with_blobs(
        &self,
        page: ServerSyncPage,
        blob_dir: &Path,
    ) -> Result<ServerSyncResult> {
        self.persist_server_sync_page_inner(
            page,
            Some(blob_dir),
            crate::sync::wire::SYNC_PROTOCOL_VERSION,
        )
        .await
    }

    #[cfg(test)]
    pub(in crate::sync) async fn persist_test_protocol_page(
        &self,
        request: SyncRequest,
        active_protocol: u32,
    ) -> Result<ServerSyncResult> {
        self.persist_server_sync_page_inner(ServerSyncPage { request }, None, active_protocol)
            .await
    }

    async fn persist_server_sync_page_inner(
        &self,
        page: ServerSyncPage,
        blob_dir: Option<&Path>,
        active_protocol: u32,
    ) -> Result<ServerSyncResult> {
        let envelope =
            crate::sync::wire::validate_request_at_protocol(&page.request, active_protocol)?;
        for change in &page.request.changes {
            super::super::protocol::validate_change(active_protocol, change)?;
            crate::sync::wire::validate_local_change_shape(change)?;
        }
        if blob_dir.is_none()
            && page
                .request
                .changes
                .iter()
                .any(|change| change.op_type == op_type::ATTACHMENT_ADD)
        {
            bail!("error attachment-blob-storage-required");
        }
        let blob_prepare_started = Instant::now();
        if let Some(blob_dir) = blob_dir {
            let mut conn = self.acquire_reader().await?;
            super::prepare_server_blobs(&mut conn, blob_dir, &page.request.changes).await?;
        }
        let blob_prepare_ms = blob_prepare_started.elapsed().as_millis();
        let mut conn = self.acquire_writer().await?;
        let assign_started = Instant::now();
        let (accepted_count, push_acks) =
            super::assign_server_sequences(&mut conn, page.request.changes, blob_dir).await?;
        let assign_ms = assign_started.elapsed().as_millis();
        let pull_query_started = Instant::now();
        let (changes, has_more) =
            load_server_changes_after(&mut conn, envelope.after, envelope.pull_limit).await?;
        let pull_query_ms = pull_query_started.elapsed().as_millis();
        Ok(ServerSyncResult {
            accepted_count,
            push_acks,
            changes,
            has_more,
            blob_prepare_ms,
            assign_ms,
            pull_query_ms,
        })
    }
}

pub(super) async fn assign_server_sequences(
    conn: &mut SqliteConnection,
    changes: Vec<ChangeWire>,
    blob_dir: Option<&Path>,
) -> Result<(i64, Vec<PushAck>)> {
    if changes.is_empty() {
        return Ok((0, Vec::new()));
    }
    let mut tx = begin_immediate(conn).await?;
    let assigned_change_ids = super::load_assigned_change_ids(&mut tx, &changes).await?;
    let unassigned_changes = changes
        .iter()
        .filter(|change| !assigned_change_ids.contains(&change.change_id))
        .cloned()
        .collect::<Vec<_>>();
    if unassigned_changes
        .iter()
        .any(|change| change.op_type == op_type::ATTACHMENT_ADD)
        && blob_dir.is_none()
    {
        bail!("error attachment-blob-storage-required");
    }
    if let Some(blob_dir) = blob_dir {
        super::ensure_attachment_blobs_admitted(&mut tx, blob_dir, &unassigned_changes).await?;
    }
    let mut next_server_seq = next_available_server_seq(&mut tx).await?;
    let mut accepted_count = 0_i64;
    let mut push_acks = Vec::with_capacity(changes.len());
    let mut affected_attachment_hashes = HashSet::new();
    for change in changes {
        let existing_server_seq = sqlx::query_scalar::<_, Option<i64>>(
            "SELECT server_seq FROM changes WHERE change_id = ?",
        )
        .bind(&change.change_id)
        .fetch_optional(&mut *tx)
        .await?;
        let server_seq = if let Some(existing_server_seq) = existing_server_seq {
            super::verify_existing_change(&mut tx, &change).await?;
            existing_server_seq.context("existing server change missing sequence")?
        } else {
            let server_seq = next_server_seq;
            next_server_seq += 1;
            let payload = change.payload.to_string();
            sqlx::query(
                "INSERT INTO changes(change_id, client_id, local_seq, entity_type, entity_id, field,
                 op_type, payload, base_version, created_at, server_seq)
                 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            )
            .bind(&change.change_id)
            .bind(&change.client_id)
            .bind(change.local_seq)
            .bind(&change.entity_type)
            .bind(&change.entity_id)
            .bind(&change.field)
            .bind(&change.op_type)
            .bind(payload)
            .bind(&change.base_version)
            .bind(&change.created_at)
            .bind(server_seq)
            .execute(&mut *tx)
            .await?;
            super::apply_server_blob_reference(&mut tx, &change, &mut affected_attachment_hashes)
                .await?;
            accepted_count += 1;
            server_seq
        };
        push_acks.push(PushAck {
            change_id: change.change_id,
            server_seq,
        });
    }
    let affected_attachment_hashes = affected_attachment_hashes.into_iter().collect::<Vec<_>>();
    crate::attachments::lifecycle::reconcile_liveness_for_hashes_in_transaction(
        &mut tx,
        &affected_attachment_hashes,
        &crate::attachments::lifecycle::SystemClock,
    )
    .await?;
    tx.commit().await?;
    Ok((accepted_count, push_acks))
}

async fn next_available_server_seq(conn: &mut SqliteConnection) -> Result<i64> {
    Ok(sqlx::query_scalar!(
        r#"SELECT COALESCE(MAX(server_seq), 0) + 1 AS "seq!: i64" FROM changes"#
    )
    .fetch_one(&mut *conn)
    .await?)
}

async fn load_server_changes_after(
    conn: &mut SqliteConnection,
    after: i64,
    pull_limit: u32,
) -> Result<(Vec<ChangeWire>, bool)> {
    let fetch_limit = i64::from(pull_limit) + 1;
    let rows = sqlx::query_as!(
        ChangeRow,
        r#"SELECT change_id AS "change_id!: String", client_id AS "client_id!: String",
         local_seq AS "local_seq!: i64", entity_type AS "entity_type!: String",
         entity_id AS "entity_id!: String", field, op_type AS "op_type!: String",
         payload AS "payload!: String", base_version, created_at AS "created_at!: String",
         server_seq
         FROM changes WHERE server_seq > ? ORDER BY server_seq LIMIT ?"#,
        after,
        fetch_limit,
    )
    .fetch_all(&mut *conn)
    .await?;
    let has_more = rows.len() > pull_limit as usize;
    let changes = rows
        .into_iter()
        .take(pull_limit as usize)
        .map(ChangeRow::into_wire)
        .collect();
    Ok((changes, has_more))
}