Repository: libsql/libsql
File: libsql-wal/src/storage/backend/s3.rs (Rust)

Code before:

    async fn fetch_segment_index(
        &self,
        config: &S3Config,
        folder_key: &FolderKey<'_>,
        segment_key: &SegmentKey,
    ) -> Result<fst::Map<Vec<u8>>> {
        let s3_index_key = s3_segment_index_key(folder_key, segment_key);
        let stream = self.s3_get(config, s3_index_key).await?;
        // TODO: parse header, check if too large to fit memory
        let bytes = stream
            .collect()
            .await
            .map_err(|e| Error::unhandled(e, ""))?
            .to_vec();
        let index = fst::Map::new(bytes).map_err(|_| Error::InvalidIndex)?;
        Ok(index)
    }

    /// Find the most recent, and biggest segment that may contain `frame_no`
    async fn find_segment(
        &self,
        config: &S3Config,
        folder_key: &FolderKey<'_>,
        frame_no: u64,
    ) -> Result<Option<SegmentKey>> {
        let lookup_key = s3_segment_index_lookup_key(&folder_key, frame_no);

        let objects = self
            .client
            .list_objects_v2()
            .bucket(&config.bucket)
            .start_after(lookup_key)
            .send()
            .await
            .map_err(|e| Error::unhandled(e, "failed to list bucket"))?;

        let Some(contents) = objects.contents().first() else {
            return Ok(None);
        };
        let key = contents.key().expect("misssing key?");
        let key_path: &Path = key.as_ref();
        let segment_key: SegmentKey = key_path
            .file_stem()
            .expect("invalid key")
            .to_str()
            .expect("invalid key")
            .parse()
            .expect("invalid key");

        Ok(Some(segment_key))

Review comment (suggestion):
maybe down the line I feel like we should make these two separate errors so we know which one triggered this.

Code after:
    }

    async fn s3_put(&self, config: &S3Config, key: String, body: ByteStream) -> Result<()> {
        self.client
            .put_object()
            .bucket(&config.bucket)
            .body(body)
            .key(key)
            .send()
            .await
            .map_err(|e| Error::unhandled(e, "error sending s3 PUT request"))?;
        Ok(())
    }

    async fn fetch_segment_index(
        &self,
        config: &S3Config,
        folder_key: &FolderKey<'_>,
        segment_key: &SegmentKey,
    ) -> Result<fst::Map<Vec<u8>>> {
        let s3_index_key = s3_segment_index_key(folder_key, segment_key);
        let mut stream = self.s3_get(config, s3_index_key).await?.into_async_read();
        let mut header: SegmentIndexHeader = SegmentIndexHeader::new_zeroed();
        stream.read_exact(header.as_bytes_mut()).await?;
        if header.magic.get() != LIBSQL_MAGIC && header.version.get() != 1 {
            return Err(Error::InvalidIndex("index header magic or version invalid"));
        }
        let mut data = Vec::with_capacity(header.len.get() as _);
        while stream.read_buf(&mut data).await? != 0 {}
        let checksum = crc32fast::hash(&data);
        if checksum != header.checksum.get() {
            return Err(Error::InvalidIndex("invalid index data checksum"));
        }
        let index = fst::Map::new(data).map_err(|_| Error::InvalidIndex("invalid index bytes"))?;
        Ok(index)
    }

    /// Find the most recent, and biggest segment that may contain `frame_no`
    async fn find_segment(
        &self,
        config: &S3Config,
        folder_key: &FolderKey<'_>,
        frame_no: u64,
    ) -> Result<Option<SegmentKey>> {
        let lookup_key = s3_segment_index_lookup_key(&folder_key, frame_no);

        let objects = self
            .client
            .list_objects_v2()
            .bucket(&config.bucket)
            .start_after(lookup_key)