Repository: libsql/libsql
File: libsql/src/sync.rs (Rust)

Code before:
    #[tracing::instrument(skip(self, frame))]
    pub(crate) async fn push_one_frame(
        &mut self,
        frame: Bytes,
        generation: u32,
        frame_no: u32,
    ) -> Result<u32> {
        let uri = format!(
            "{}/sync/{}/{}/{}",
            self.sync_url,
            generation,
            frame_no,
            frame_no + 1
        );
        tracing::debug!("pushing frame");

        let durable_frame_num = self.push_with_retry(uri, frame, self.max_retries).await?;

        if durable_frame_num > frame_no {
            tracing::error!(
                "server returned durable_frame_num larger than what we sent: sent={}, got={}",
                frame_no,
                durable_frame_num
            );

            // TODO: do we want to return an error here?
        }

        if durable_frame_num < frame_no {
            // Update our knowledge of where the server is at frame wise.
            self.durable_frame_num = durable_frame_num;

            tracing::debug!(
                "server returned durable_frame_num lower than what we sent: sent={}, got={}",
                frame_no,
                durable_frame_num
            );

            // Return an error and expect the caller to re-call push with the updated state.
            return Err(SyncError::InvalidPushFrameNo(frame_no, durable_frame_num).into());
        }

        tracing::debug!(?durable_frame_num, "frame successfully pushed");

        // Update our last known max_frame_no from the server.
        self.durable_frame_num = durable_frame_num;

        Ok(durable_frame_num)
    }

    async fn push_with_retry(&self, uri: String, frame: Bytes, max_retries: usize) -> Result<u32> {

Review comment (suggestion):
Yes, let's error out saying that we're behind the server. This will be the place where we wire our _pull logic_ likely when the server supports it

Code after:
    }

    #[tracing::instrument(skip(self, frame))]
    pub(crate) async fn push_one_frame(
        &mut self,
        frame: Bytes,
        generation: u32,
        frame_no: u32,
    ) -> Result<u32> {
        let uri = format!(
            "{}/sync/{}/{}/{}",
            self.sync_url,
            generation,
            frame_no,
            frame_no + 1
        );
        tracing::debug!("pushing frame");

        let durable_frame_num = self.push_with_retry(uri, frame, self.max_retries).await?;

        if durable_frame_num > frame_no {
            tracing::error!(
                "server returned durable_frame_num larger than what we sent: sent={}, got={}",
                frame_no,
                durable_frame_num
            );

            return Err(SyncError::InvalidPushFrameNoHigh(frame_no, durable_frame_num).into());
        }

        if durable_frame_num < frame_no {
            // Update our knowledge of where the server is at frame wise.
            self.durable_frame_num = durable_frame_num;

            tracing::debug!(
                "server returned durable_frame_num lower than what we sent: sent={}, got={}",
                frame_no,
                durable_frame_num
            );

            // Return an error and expect the caller to re-call push with the updated state.
            return Err(SyncError::InvalidPushFrameNoLow(frame_no, durable_frame_num).into());
        }

        tracing::debug!(?durable_frame_num, "frame successfully pushed");

        // Update our last known max_frame_no from the server.
        self.durable_frame_num = durable_frame_num;

        Ok(durable_frame_num)
    }