You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/05/07 12:34:21 UTC
(opendal) branch main updated: revert: fix gdrive behavior CI semantics (#7485)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new f98505bd0 revert: fix gdrive behavior CI semantics (#7485)
f98505bd0 is described below
commit f98505bd0970cc80cff70f299938dea0a1fe7cda
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu May 7 20:34:13 2026 +0800
revert: fix gdrive behavior CI semantics (#7485)
Revert "fix(services/gdrive): stabilize behavior CI semantics (#7390)"
This reverts commit d5f7dcf1097a96429b3684057c567ccc090a6b77.
---
.github/scripts/test_behavior/plan.py | 12 +
.github/scripts/test_behavior/test_plan.py | 18 +
core/services/gdrive/src/backend.rs | 243 +-----------
core/services/gdrive/src/builder.rs | 6 +-
core/services/gdrive/src/core.rs | 572 +++--------------------------
core/services/gdrive/src/deleter.rs | 65 +---
core/services/gdrive/src/docs.md | 6 -
core/services/gdrive/src/lib.rs | 1 -
core/services/gdrive/src/lister.rs | 530 ++++++--------------------
core/services/gdrive/src/path_index.rs | 451 -----------------------
core/services/gdrive/src/writer.rs | 54 +--
11 files changed, 244 insertions(+), 1714 deletions(-)
diff --git a/.github/scripts/test_behavior/plan.py b/.github/scripts/test_behavior/plan.py
index ddb21080b..6c5e8d087 100755
--- a/.github/scripts/test_behavior/plan.py
+++ b/.github/scripts/test_behavior/plan.py
@@ -35,6 +35,14 @@ LANGUAGE_BINDING = ["java", "python", "nodejs", "go", "c", "cpp", "dotnet"]
INTEGRATIONS = ["object_store"]
+TEMPORARILY_DISABLED_CORE_SERVICES = {
+ # FIXME: Re-enable gdrive behavior tests after #7384 is fixed.
+ # Gdrive is currently not reliable enough for CI: #6684 tracks missing
+ # freshly-created entries in listings, and existing behavior tests already
+ # document redirect loops caused by Google Drive abuse detection.
+ "gdrive",
+}
+
def provided_cases() -> list[dict[str, str]]:
root_dir = f"{GITHUB_DIR}/services"
@@ -59,6 +67,10 @@ def provided_cases() -> list[dict[str, str]]:
if not os.getenv("GITHUB_HAS_SECRETS") == "true":
cases[:] = [v for v in cases if "op://services" not in v["content"]]
+ cases[:] = [
+ v for v in cases if v["service"] not in TEMPORARILY_DISABLED_CORE_SERVICES
+ ]
+
# Remove content from cases.
cases = [
{
diff --git a/.github/scripts/test_behavior/test_plan.py b/.github/scripts/test_behavior/test_plan.py
index 76abd6511..fce380f37 100644
--- a/.github/scripts/test_behavior/test_plan.py
+++ b/.github/scripts/test_behavior/test_plan.py
@@ -75,5 +75,23 @@ class BehaviorTestPlan(unittest.TestCase):
# Should contain fs
self.assertTrue("fs" in cases)
+ @patch.dict("os.environ", {"GITHUB_HAS_SECRETS": "true"}, clear=False)
+ def test_gdrive_is_temporarily_disabled(self):
+ result = plan([".github/workflows/test_behavior.yml"])
+
+ self.assertTrue(result["components"]["core"])
+ core_cases = [v["service"] for target in result["core"] for v in target["cases"]]
+ self.assertFalse("gdrive" in core_cases)
+
+ for language in ["java", "python", "nodejs", "go", "c", "cpp", "dotnet"]:
+ self.assertTrue(result["components"][f"binding_{language}"])
+ binding_cases = [
+ v["service"]
+ for target in result[f"binding_{language}"]
+ for v in target["cases"]
+ ]
+ self.assertFalse("gdrive" in binding_cases)
+
+
if __name__ == "__main__":
unittest.main()
diff --git a/core/services/gdrive/src/backend.rs b/core/services/gdrive/src/backend.rs
index 082171fd6..cc03fe7d9 100644
--- a/core/services/gdrive/src/backend.rs
+++ b/core/services/gdrive/src/backend.rs
@@ -24,8 +24,6 @@ use http::StatusCode;
use super::core::GdriveCore;
use super::core::GdriveFile;
-use super::core::GdriveRecentPathState;
-use super::core::normalize_dir_path;
use super::deleter::GdriveDeleter;
use super::error::parse_error;
use super::lister::GdriveFlatLister;
@@ -54,58 +52,13 @@ impl Access for GdriveBackend {
async fn create_dir(&self, path: &str, _args: OpCreateDir) -> Result<RpCreateDir> {
let path = build_abs_path(&self.core.root, path);
- let dir_id = self.core.ensure_dir(&path).await?;
- let metadata = Metadata::new(EntryMode::DIR);
-
- self.core.cache_dir_id(&path, &dir_id).await;
- self.core.record_recent_upsert(&path, metadata).await;
+ let _ = self.core.path_cache.ensure_dir(&path).await?;
Ok(RpCreateDir::default())
}
async fn stat(&self, path: &str, _args: OpStat) -> Result<RpStat> {
- let path = build_abs_path(&self.core.root, path);
-
- match self.core.recent_entry_for_path(&path).await {
- GdriveRecentPathState::Present(metadata) => {
- if metadata.mode().is_dir() && path.ends_with('/') {
- return Ok(RpStat::new(*metadata));
- }
- }
- GdriveRecentPathState::Deleted => {
- return Err(Error::new(
- ErrorKind::NotFound,
- format!("path not found: {path}"),
- ));
- }
- GdriveRecentPathState::Missing => {}
- }
-
- let mut file_id = match self.core.resolve_path(&path).await? {
- Some(id) => id,
- None => match self.core.resolve_path_after_refresh(&path).await? {
- Some(id) => id,
- None => {
- return Err(Error::new(
- ErrorKind::NotFound,
- format!("path not found: {path}"),
- ));
- }
- },
- };
- let mut resp = self.core.gdrive_stat_by_id(&file_id).await?;
-
- if resp.status() == StatusCode::NOT_FOUND {
- file_id = self
- .core
- .resolve_path_after_refresh(&path)
- .await?
- .ok_or(Error::new(
- ErrorKind::NotFound,
- format!("path not found: {path}"),
- ))?;
- resp = self.core.gdrive_stat_by_id(&file_id).await?;
- }
+ let resp = self.core.gdrive_stat(path).await?;
if resp.status() != StatusCode::OK {
return Err(parse_error(resp));
@@ -135,34 +88,11 @@ impl Access for GdriveBackend {
}
async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
- let abs_path = build_abs_path(&self.core.root, path);
- let resp = match self.core.gdrive_get(path, args.range()).await {
- Ok(resp) => resp,
- Err(err) if err.kind() == ErrorKind::NotFound => {
- self.core.refresh_path(&abs_path).await;
- self.core.gdrive_get(path, args.range()).await?
- }
- Err(err) => return Err(err),
- };
+ let resp = self.core.gdrive_get(path, args.range()).await?;
let status = resp.status();
match status {
StatusCode::OK | StatusCode::PARTIAL_CONTENT => Ok((RpRead::new(), resp.into_body())),
- StatusCode::NOT_FOUND => {
- self.core.refresh_path(&abs_path).await;
- let resp = self.core.gdrive_get(path, args.range()).await?;
- let status = resp.status();
- match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
- Ok((RpRead::new(), resp.into_body()))
- }
- _ => {
- let (part, mut body) = resp.into_parts();
- let buf = body.to_buffer().await?;
- Err(parse_error(Response::from_parts(part, buf)))
- }
- }
- }
_ => {
let (part, mut body) = resp.into_parts();
let buf = body.to_buffer().await?;
@@ -176,10 +106,7 @@ impl Access for GdriveBackend {
// As Google Drive allows files have the same name, we need to check if the file exists.
// If the file exists, we will keep its ID and update it.
- let file_id = match self.core.resolve_path(&path).await? {
- Some(id) => Some(id),
- None => self.core.resolve_path_after_refresh(&path).await?,
- };
+ let file_id = self.core.path_cache.get(&path).await?;
Ok((
RpWrite::default(),
@@ -209,74 +136,10 @@ impl Access for GdriveBackend {
}
async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
- let source = build_abs_path(&self.core.root, from);
- let target = build_abs_path(&self.core.root, to);
let resp = self.core.gdrive_copy(from, to).await?;
match resp.status() {
- StatusCode::OK => {
- let body = resp.into_body();
- let meta: GdriveFile =
- serde_json::from_reader(body.reader()).map_err(new_json_deserialize_error)?;
-
- let to_path = build_abs_path(&self.core.root, to);
- let mut metadata = if meta.mime_type == "application/vnd.google-apps.folder" {
- Metadata::new(EntryMode::DIR)
- } else {
- Metadata::new(EntryMode::FILE)
- };
- if let Some(size) = meta.size {
- metadata = metadata.with_content_length(size.parse::<u64>().map_err(|e| {
- Error::new(ErrorKind::Unexpected, "parse content length").set_source(e)
- })?);
- }
-
- if metadata.mode().is_dir() {
- self.core.cache_dir_id(&to_path, &meta.id).await;
- } else {
- self.core.cache_file_id(&to_path, &meta.id).await;
- }
- self.core.record_recent_upsert(&to_path, metadata).await;
-
- Ok(RpCopy::default())
- }
- StatusCode::NOT_FOUND => {
- self.core.refresh_path(&source).await;
- self.core.refresh_path(&target).await;
- let resp = self.core.gdrive_copy(from, to).await?;
- match resp.status() {
- StatusCode::OK => {
- let body = resp.into_body();
- let meta: GdriveFile = serde_json::from_reader(body.reader())
- .map_err(new_json_deserialize_error)?;
-
- let to_path = build_abs_path(&self.core.root, to);
- let mut metadata = if meta.mime_type == "application/vnd.google-apps.folder"
- {
- Metadata::new(EntryMode::DIR)
- } else {
- Metadata::new(EntryMode::FILE)
- };
- if let Some(size) = meta.size {
- metadata =
- metadata.with_content_length(size.parse::<u64>().map_err(|e| {
- Error::new(ErrorKind::Unexpected, "parse content length")
- .set_source(e)
- })?);
- }
-
- if metadata.mode().is_dir() {
- self.core.cache_dir_id(&to_path, &meta.id).await;
- } else {
- self.core.cache_file_id(&to_path, &meta.id).await;
- }
- self.core.record_recent_upsert(&to_path, metadata).await;
-
- Ok(RpCopy::default())
- }
- _ => Err(parse_error(resp)),
- }
- }
+ StatusCode::OK => Ok(RpCopy::default()),
_ => Err(parse_error(resp)),
}
}
@@ -286,7 +149,15 @@ impl Access for GdriveBackend {
let target = build_abs_path(&self.core.root, to);
// rename will overwrite `to`, delete it if exist
- self.core.trash_path_if_exists(&target).await?;
+ if let Some(id) = self.core.path_cache.get(&target).await? {
+ let resp = self.core.gdrive_trash(&id).await?;
+ let status = resp.status();
+ if status != StatusCode::OK {
+ return Err(parse_error(resp));
+ }
+
+ self.core.path_cache.remove(&target).await;
+ }
let resp = self
.core
@@ -301,90 +172,12 @@ impl Access for GdriveBackend {
let meta: GdriveFile =
serde_json::from_reader(body.reader()).map_err(new_json_deserialize_error)?;
- let source_path = if meta.mime_type == "application/vnd.google-apps.folder" {
- normalize_dir_path(&build_abs_path(&self.core.root, from))
- } else {
- build_abs_path(&self.core.root, from)
- };
- let target_path = if meta.mime_type == "application/vnd.google-apps.folder" {
- normalize_dir_path(&build_abs_path(&self.core.root, to))
- } else {
- build_abs_path(&self.core.root, to)
- };
- let mut metadata = if meta.mime_type == "application/vnd.google-apps.folder" {
- Metadata::new(EntryMode::DIR)
- } else {
- Metadata::new(EntryMode::FILE)
- };
- if let Some(size) = meta.size {
- metadata = metadata.with_content_length(size.parse::<u64>().map_err(|e| {
- Error::new(ErrorKind::Unexpected, "parse content length").set_source(e)
- })?);
- }
-
- if metadata.mode().is_dir() {
- self.core.invalidate_dir_id(&source_path).await;
- self.core.cache_dir_id(&target_path, &meta.id).await;
- } else {
- self.core.invalidate_file_id(&source_path).await;
- self.core.cache_file_id(&target_path, &meta.id).await;
- }
- self.core
- .record_recent_delete(&source_path, metadata.mode())
- .await;
- self.core.record_recent_upsert(&target_path, metadata).await;
-
- Ok(RpRename::default())
- }
- StatusCode::NOT_FOUND => {
- self.core.refresh_path(&source).await;
- self.core.refresh_path(&target).await;
-
- let resp = self
- .core
- .gdrive_patch_metadata_request(&source, &target)
- .await?;
-
- if resp.status() != StatusCode::OK {
- return Err(parse_error(resp));
- }
-
- let body = resp.into_body();
- let meta: GdriveFile =
- serde_json::from_reader(body.reader()).map_err(new_json_deserialize_error)?;
+ let cache = &self.core.path_cache;
- let source_path = if meta.mime_type == "application/vnd.google-apps.folder" {
- normalize_dir_path(&build_abs_path(&self.core.root, from))
- } else {
- build_abs_path(&self.core.root, from)
- };
- let target_path = if meta.mime_type == "application/vnd.google-apps.folder" {
- normalize_dir_path(&build_abs_path(&self.core.root, to))
- } else {
- build_abs_path(&self.core.root, to)
- };
- let mut metadata = if meta.mime_type == "application/vnd.google-apps.folder" {
- Metadata::new(EntryMode::DIR)
- } else {
- Metadata::new(EntryMode::FILE)
- };
- if let Some(size) = meta.size {
- metadata = metadata.with_content_length(size.parse::<u64>().map_err(|e| {
- Error::new(ErrorKind::Unexpected, "parse content length").set_source(e)
- })?);
- }
-
- if metadata.mode().is_dir() {
- self.core.invalidate_dir_id(&source_path).await;
- self.core.cache_dir_id(&target_path, &meta.id).await;
- } else {
- self.core.invalidate_file_id(&source_path).await;
- self.core.cache_file_id(&target_path, &meta.id).await;
- }
- self.core
- .record_recent_delete(&source_path, metadata.mode())
+ cache.remove(&build_abs_path(&self.core.root, from)).await;
+ cache
+ .insert(&build_abs_path(&self.core.root, to), &meta.id)
.await;
- self.core.record_recent_upsert(&target_path, metadata).await;
Ok(RpRename::default())
}
diff --git a/core/services/gdrive/src/builder.rs b/core/services/gdrive/src/builder.rs
index 6226c36d3..b9a0d4a4d 100644
--- a/core/services/gdrive/src/builder.rs
+++ b/core/services/gdrive/src/builder.rs
@@ -27,7 +27,6 @@ use super::config::GdriveConfig;
use super::core::GdriveCore;
use super::core::GdrivePathQuery;
use super::core::GdriveSigner;
-use super::path_index::GdrivePathIndex;
use opendal_core::raw::*;
use opendal_core::*;
@@ -122,7 +121,6 @@ impl Builder for GdriveBuilder {
create_dir: true,
delete: true,
- delete_with_recursive: true,
rename: true,
copy: true,
@@ -182,8 +180,8 @@ impl Builder for GdriveBuilder {
info: accessor_info.clone(),
root,
signer: signer.clone(),
- path_index: GdrivePathIndex::new(GdrivePathQuery::new(accessor_info, signer)),
- recent_entries: Mutex::default(),
+ path_cache: PathCacher::new(GdrivePathQuery::new(accessor_info, signer))
+ .with_lock(),
}),
})
}
diff --git a/core/services/gdrive/src/core.rs b/core/services/gdrive/src/core.rs
index 0bbaa9c2b..b0245f08e 100644
--- a/core/services/gdrive/src/core.rs
+++ b/core/services/gdrive/src/core.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-use std::collections::HashMap;
use std::fmt::Debug;
use std::sync::Arc;
@@ -30,7 +29,6 @@ use serde::Deserialize;
use serde_json::json;
use super::error::parse_error;
-use super::path_index::GdrivePathIndex;
use opendal_core::raw::*;
use opendal_core::*;
@@ -41,13 +39,8 @@ pub struct GdriveCore {
pub signer: Arc<Mutex<GdriveSigner>>,
- /// Service-local path index for resolving path -> file id mappings.
- pub path_index: GdrivePathIndex<GdrivePathQuery>,
-
- /// Keep a short-lived local view of recent mutations so list and stat
- /// operations stay read-after-write consistent within the current operator
- /// session.
- pub recent_entries: Mutex<GdriveRecentState>,
+ /// Cache the mapping from path to file id
+ pub path_cache: PathCacher<GdrivePathQuery>,
}
impl Debug for GdriveCore {
@@ -59,177 +52,6 @@ impl Debug for GdriveCore {
}
impl GdriveCore {
- pub async fn cache_file_id(&self, path: &str, file_id: &str) {
- self.path_index.upsert_file(path, file_id).await;
- }
-
- pub async fn cache_dir_id(&self, path: &str, file_id: &str) {
- self.path_index.upsert_dir(path, file_id).await;
- }
-
- pub async fn invalidate_file_id(&self, path: &str) {
- self.path_index.invalidate_file(path).await;
- }
-
- pub async fn invalidate_dir_id(&self, path: &str) {
- self.path_index.invalidate_dir(path).await;
- }
-
- pub async fn refresh_path(&self, path: &str) {
- self.path_index.refresh_path(path).await;
- }
-
- pub async fn refresh_dir_path(&self, path: &str) {
- self.invalidate_dir_id(path).await;
- self.refresh_path(path).await;
- }
-
- pub async fn resolve_path(&self, path: &str) -> Result<Option<String>> {
- self.path_index.get(path).await
- }
-
- pub async fn resolve_path_after_refresh(&self, path: &str) -> Result<Option<String>> {
- self.refresh_path(path).await;
- self.resolve_path(path).await
- }
-
- pub async fn ensure_dir(&self, path: &str) -> Result<String> {
- self.path_index.ensure_dir(path).await
- }
-
- pub async fn trash_path_if_exists(&self, path: &str) -> Result<()> {
- let mut target_id = match self.resolve_path(path).await? {
- Some(id) => Some(id),
- None => self.resolve_path_after_refresh(path).await?,
- };
-
- if let Some(id) = target_id.take() {
- let mut resp = self.gdrive_trash(&id).await?;
- if resp.status() == StatusCode::NOT_FOUND {
- self.refresh_path(path).await;
- target_id = self.resolve_path(path).await?;
- if let Some(id) = target_id {
- resp = self.gdrive_trash(&id).await?;
- } else {
- return Ok(());
- }
- }
-
- if resp.status() == StatusCode::NOT_FOUND {
- return Ok(());
- }
- if resp.status() != StatusCode::OK {
- return Err(parse_error(resp));
- }
-
- self.invalidate_file_id(path).await;
- self.invalidate_dir_id(path).await;
- }
-
- Ok(())
- }
-
- pub async fn record_recent_upsert(&self, path: &str, metadata: Metadata) {
- let mut recent_entries = self.recent_entries.lock().await;
- prune_recent_entries(&mut recent_entries);
-
- let mode = metadata.mode();
- let path = canonical_recent_path(path, mode);
- let expires_at = Timestamp::now() + Duration::from_secs(15);
- insert_recent_entry(
- &mut recent_entries.entries,
- &path,
- mode,
- GdriveRecentEntry {
- metadata: Some(metadata),
- expires_at,
- },
- );
- revive_recent_parent_dirs(&mut recent_entries, &path, expires_at);
- }
-
- pub async fn record_recent_delete(&self, path: &str, mode: EntryMode) {
- let mut recent_entries = self.recent_entries.lock().await;
- prune_recent_entries(&mut recent_entries);
-
- let path = canonical_recent_path(path, mode);
- let expires_at = Timestamp::now() + Duration::from_secs(15);
- insert_recent_entry(
- &mut recent_entries.entries,
- &path,
- mode,
- GdriveRecentEntry {
- metadata: None,
- expires_at,
- },
- );
- if mode.is_dir() {
- recent_entries.tombstones.insert(path, expires_at);
- }
- }
-
- pub async fn recent_entry_for_path(&self, path: &str) -> GdriveRecentPathState {
- let mut recent_entries = self.recent_entries.lock().await;
- prune_recent_entries(&mut recent_entries);
-
- let recent_entry = lookup_recent_entry(&recent_entries.entries, path);
- let tombstone = lookup_recent_tombstone(&recent_entries.tombstones, path);
-
- match (recent_entry, tombstone) {
- (Some(entry), Some(tombstone_expires_at))
- if tombstone_expires_at > entry.expires_at =>
- {
- GdriveRecentPathState::Deleted
- }
- (Some(entry), _) => match entry.metadata.clone() {
- Some(metadata) => GdriveRecentPathState::Present(Box::new(metadata)),
- None => GdriveRecentPathState::Deleted,
- },
- (None, Some(_)) => GdriveRecentPathState::Deleted,
- (None, None) => GdriveRecentPathState::Missing,
- }
- }
-
- pub async fn recent_entries_for_list(
- &self,
- scope_path: &str,
- recursive: bool,
- ) -> Vec<(String, Metadata)> {
- let mut recent_entries = self.recent_entries.lock().await;
- prune_recent_entries(&mut recent_entries);
-
- let mut entries = recent_entries
- .entries
- .iter()
- .filter_map(|(path, entry)| {
- let metadata = entry.metadata.clone()?;
- if metadata.mode().is_dir() && path != &normalize_dir_path(path) {
- return None;
- }
- if let Some(tombstone_expires_at) =
- lookup_recent_tombstone(&recent_entries.tombstones, path)
- {
- if tombstone_expires_at > entry.expires_at {
- return None;
- }
- }
- if let Some(latest_entry) = lookup_recent_entry(&recent_entries.entries, path) {
- if latest_entry.expires_at > entry.expires_at {
- return None;
- }
- }
- if recent_entry_in_scope(scope_path, path, metadata.mode(), recursive) {
- Some((path.clone(), metadata))
- } else {
- None
- }
- })
- .collect::<Vec<_>>();
-
- entries.sort_by(|left, right| left.0.cmp(&right.0));
- entries
- }
-
pub async fn gdrive_stat_by_id(&self, file_id: &str) -> Result<Response<Buffer>> {
// The file metadata in the Google Drive API is very complex.
// For now, we only need the file id, name, mime type and modified time.
@@ -244,29 +66,22 @@ impl GdriveCore {
self.info.http_client().send(req).await
}
+ pub async fn gdrive_stat(&self, path: &str) -> Result<Response<Buffer>> {
+ let path = build_abs_path(&self.root, path);
+ let file_id = self.path_cache.get(&path).await?.ok_or(Error::new(
+ ErrorKind::NotFound,
+ format!("path not found: {path}"),
+ ))?;
+
+ self.gdrive_stat_by_id(&file_id).await
+ }
+
pub async fn gdrive_get(&self, path: &str, range: BytesRange) -> Result<Response<HttpBody>> {
let path = build_abs_path(&self.root, path);
- match self.recent_entry_for_path(&path).await {
- GdriveRecentPathState::Deleted => {
- return Err(Error::new(
- ErrorKind::NotFound,
- format!("path not found: {path}"),
- ));
- }
- GdriveRecentPathState::Present(_) | GdriveRecentPathState::Missing => {}
- }
- let path_id = match self.resolve_path(&path).await? {
- Some(id) => id,
- None => match self.resolve_path_after_refresh(&path).await? {
- Some(id) => id,
- None => {
- return Err(Error::new(
- ErrorKind::NotFound,
- format!("path not found: {path}"),
- ));
- }
- },
- };
+ let path_id = self.path_cache.get(&path).await?.ok_or(Error::new(
+ ErrorKind::NotFound,
+ format!("path not found: {path}"),
+ ))?;
let url: String = format!("https://www.googleapis.com/drive/v3/files/{path_id}?alt=media");
@@ -360,26 +175,18 @@ impl GdriveCore {
source: &str,
target: &str,
) -> Result<Response<Buffer>> {
- let source_file_id = match self.resolve_path(source).await? {
- Some(id) => id,
- None => self
- .resolve_path_after_refresh(source)
- .await?
- .ok_or(Error::new(
- ErrorKind::NotFound,
- format!("source path not found: {source}"),
- ))?,
- };
+ let source_file_id = self.path_cache.get(source).await?.ok_or(Error::new(
+ ErrorKind::NotFound,
+ format!("source path not found: {source}"),
+ ))?;
let source_parent = get_parent(source);
- let source_parent_id = match self.resolve_path(source_parent).await? {
- Some(id) => id,
- None => self
- .resolve_path_after_refresh(source_parent)
- .await?
- .expect("old parent must exist"),
- };
+ let source_parent_id = self
+ .path_cache
+ .get(source_parent)
+ .await?
+ .expect("old parent must exist");
- let target_parent_id = self.path_index.ensure_dir(get_parent(target)).await?;
+ let target_parent_id = self.path_cache.ensure_dir(get_parent(target)).await?;
let target_file_name = get_basename(target);
let metadata = &json!({
@@ -424,7 +231,7 @@ impl GdriveCore {
size: u64,
body: Buffer,
) -> Result<Response<Buffer>> {
- let parent = self.path_index.ensure_dir(get_parent(path)).await?;
+ let parent = self.path_cache.ensure_dir(get_parent(path)).await?;
let url = "https://www.googleapis.com/upload/drive/v3/files?uploadType=multipart";
@@ -500,24 +307,25 @@ impl GdriveCore {
pub async fn gdrive_copy(&self, from: &str, to: &str) -> Result<Response<Buffer>> {
let from = build_abs_path(&self.root, from);
- let from_file_id = match self.resolve_path(&from).await? {
- Some(id) => id,
- None => match self.resolve_path_after_refresh(&from).await? {
- Some(id) => id,
- None => {
- return Err(Error::new(
- ErrorKind::NotFound,
- "the file to copy does not exist",
- ));
- }
- },
- };
+ let from_file_id = self.path_cache.get(&from).await?.ok_or(Error::new(
+ ErrorKind::NotFound,
+ "the file to copy does not exist",
+ ))?;
let to_name = get_basename(to);
let to_path = build_abs_path(&self.root, to);
- let to_parent_id = self.path_index.ensure_dir(get_parent(&to_path)).await?;
+ let to_parent_id = self.path_cache.ensure_dir(get_parent(&to_path)).await?;
- self.trash_path_if_exists(&to_path).await?;
+ // copy will overwrite `to`, delete it if exist
+ if let Some(id) = self.path_cache.get(&to_path).await? {
+ let resp = self.gdrive_trash(&id).await?;
+ let status = resp.status();
+ if status != StatusCode::OK {
+ return Err(parse_error(resp));
+ }
+
+ self.path_cache.remove(&to_path).await;
+ }
let url = format!("https://www.googleapis.com/drive/v3/files/{from_file_id}/copy");
@@ -537,140 +345,6 @@ impl GdriveCore {
}
}
-#[derive(Clone)]
-pub struct GdriveRecentEntry {
- metadata: Option<Metadata>,
- expires_at: Timestamp,
-}
-
-#[derive(Default)]
-pub struct GdriveRecentState {
- entries: HashMap<String, GdriveRecentEntry>,
- tombstones: HashMap<String, Timestamp>,
-}
-
-#[derive(Clone, Debug)]
-pub enum GdriveRecentPathState {
- Missing,
- Deleted,
- Present(Box<Metadata>),
-}
-
-pub(crate) fn normalize_dir_path(path: &str) -> String {
- if path.is_empty() || path.ends_with('/') {
- path.to_string()
- } else {
- format!("{path}/")
- }
-}
-
-fn canonical_recent_path(path: &str, mode: EntryMode) -> String {
- if mode.is_dir() {
- normalize_dir_path(path)
- } else {
- path.to_string()
- }
-}
-
-fn prune_recent_entries(entries: &mut GdriveRecentState) {
- let now = Timestamp::now();
- entries.entries.retain(|_, entry| entry.expires_at > now);
- entries.tombstones.retain(|_, expires_at| *expires_at > now);
-}
-
-fn lookup_recent_entry<'a>(
- entries: &'a HashMap<String, GdriveRecentEntry>,
- path: &str,
-) -> Option<&'a GdriveRecentEntry> {
- entries.get(path)
-}
-
-fn lookup_recent_tombstone(
- tombstones: &HashMap<String, Timestamp>,
- path: &str,
-) -> Option<Timestamp> {
- let candidates = recent_path_candidates(path);
- tombstones
- .iter()
- .filter(|(tombstone, _)| {
- candidates
- .iter()
- .any(|candidate| path_under_tombstone(candidate, tombstone))
- })
- .map(|(_, expires_at)| *expires_at)
- .max()
-}
-
-fn revive_recent_parent_dirs(entries: &mut GdriveRecentState, path: &str, expires_at: Timestamp) {
- let mut parent = get_parent(path).to_string();
-
- while !parent.is_empty() && parent != "/" {
- if lookup_recent_tombstone(&entries.tombstones, &parent).is_some() {
- insert_recent_entry(
- &mut entries.entries,
- &parent,
- EntryMode::DIR,
- GdriveRecentEntry {
- metadata: Some(Metadata::new(EntryMode::DIR)),
- expires_at,
- },
- );
- }
-
- parent = get_parent(&parent).to_string();
- }
-}
-
-fn insert_recent_entry(
- entries: &mut HashMap<String, GdriveRecentEntry>,
- path: &str,
- mode: EntryMode,
- entry: GdriveRecentEntry,
-) {
- entries.insert(path.to_string(), entry.clone());
-
- if mode.is_dir() {
- let alias = path.trim_end_matches('/');
- if !alias.is_empty() {
- entries.insert(alias.to_string(), entry);
- }
- }
-}
-
-fn recent_path_candidates(path: &str) -> Vec<String> {
- let mut candidates = vec![path.to_string()];
- let dir_path = normalize_dir_path(path);
- if dir_path != path {
- candidates.push(dir_path);
- }
- candidates
-}
-
-fn path_under_tombstone(path: &str, tombstone: &str) -> bool {
- tombstone.is_empty() || path == tombstone || path.starts_with(tombstone)
-}
-
-fn recent_entry_in_scope(scope_path: &str, path: &str, mode: EntryMode, recursive: bool) -> bool {
- if recursive {
- if scope_path.is_empty() {
- return true;
- }
-
- if scope_path.ends_with('/') {
- return path.starts_with(scope_path) && path != scope_path;
- }
-
- return path == canonical_recent_path(scope_path, mode) || path.starts_with(scope_path);
- }
-
- let parent = get_parent(path);
- if scope_path.is_empty() {
- parent == "/"
- } else {
- parent == scope_path
- }
-}
-
#[derive(Clone)]
pub struct GdriveSigner {
pub info: Arc<AccessorInfo>,
@@ -777,12 +451,10 @@ impl PathQuery for GdrivePathQuery {
queries.push("mimeType = 'application/vnd.google-apps.folder'".to_string());
}
let query = queries.join(" and ");
- let order_by = "modifiedTime desc,createdTime desc";
let url = format!(
- "https://www.googleapis.com/drive/v3/files?q={}&orderBy={}&pageSize=1",
- percent_encode_path(query.as_str()),
- percent_encode_path(order_by)
+ "https://www.googleapis.com/drive/v3/files?q={}",
+ percent_encode_path(query.as_str())
);
let mut req = Request::get(&url)
@@ -877,159 +549,3 @@ pub(crate) struct GdriveFileList {
pub(crate) files: Vec<GdriveFile>,
pub(crate) next_page_token: Option<String>,
}
-
-#[cfg(test)]
-mod tests {
- use super::*;
-
- fn mock_gdrive_core() -> GdriveCore {
- let info = Arc::new(AccessorInfo::default());
- let signer = Arc::new(Mutex::new(GdriveSigner::new(info.clone())));
-
- GdriveCore {
- info: info.clone(),
- root: "/".to_string(),
- signer: signer.clone(),
- path_index: GdrivePathIndex::new(GdrivePathQuery::new(info, signer)),
- recent_entries: Mutex::new(GdriveRecentState::default()),
- }
- }
-
- #[tokio::test]
- async fn test_recent_entries_for_direct_list() {
- let core = mock_gdrive_core();
-
- core.record_recent_upsert(
- "parent/file.txt",
- Metadata::new(EntryMode::FILE).with_content_length(5),
- )
- .await;
- core.record_recent_upsert("parent/dir/", Metadata::new(EntryMode::DIR))
- .await;
- core.record_recent_upsert("parent/nested/file.txt", Metadata::new(EntryMode::FILE))
- .await;
- core.record_recent_delete("parent/deleted.txt", EntryMode::FILE)
- .await;
-
- let entries = core.recent_entries_for_list("parent/", false).await;
- let paths = entries
- .into_iter()
- .map(|(path, _)| path)
- .collect::<Vec<_>>();
-
- assert_eq!(
- paths,
- vec!["parent/dir/".to_string(), "parent/file.txt".to_string()]
- );
- }
-
- #[tokio::test]
- async fn test_recent_entries_for_recursive_prefix_list() {
- let core = mock_gdrive_core();
-
- core.record_recent_upsert("parent/file.txt", Metadata::new(EntryMode::FILE))
- .await;
- core.record_recent_upsert("parent/nested/file.txt", Metadata::new(EntryMode::FILE))
- .await;
- core.record_recent_upsert("prefix", Metadata::new(EntryMode::FILE))
- .await;
- core.record_recent_upsert("prefix-child", Metadata::new(EntryMode::FILE))
- .await;
-
- let parent_entries = core.recent_entries_for_list("parent/", true).await;
- let parent_paths = parent_entries
- .into_iter()
- .map(|(path, _)| path)
- .collect::<Vec<_>>();
- assert_eq!(
- parent_paths,
- vec![
- "parent/file.txt".to_string(),
- "parent/nested/file.txt".to_string()
- ]
- );
-
- let prefix_entries = core.recent_entries_for_list("prefix", true).await;
- let prefix_paths = prefix_entries
- .into_iter()
- .map(|(path, _)| path)
- .collect::<Vec<_>>();
- assert_eq!(
- prefix_paths,
- vec!["prefix".to_string(), "prefix-child".to_string()]
- );
- }
-
- #[tokio::test]
- async fn test_recent_entry_for_dir_alias() {
- let core = mock_gdrive_core();
-
- core.record_recent_upsert("parent/dir/", Metadata::new(EntryMode::DIR))
- .await;
-
- match core.recent_entry_for_path("parent/dir").await {
- GdriveRecentPathState::Present(_) => {}
- other => panic!("unexpected state for dir alias: {other:?}"),
- }
- }
-
- #[tokio::test]
- async fn test_recent_tombstone_hides_descendants_until_recreated() {
- let core = mock_gdrive_core();
-
- core.record_recent_upsert(
- "parent/dir/stale.txt",
- Metadata::new(EntryMode::FILE).with_content_length(1),
- )
- .await;
- core.record_recent_delete("parent/dir/", EntryMode::DIR)
- .await;
-
- match core.recent_entry_for_path("parent/dir").await {
- GdriveRecentPathState::Deleted => {}
- other => panic!("unexpected state for deleted dir: {other:?}"),
- }
-
- match core.recent_entry_for_path("parent/dir/file.txt").await {
- GdriveRecentPathState::Deleted => {}
- other => panic!("unexpected state for deleted child: {other:?}"),
- }
-
- match core.recent_entry_for_path("parent/dir/stale.txt").await {
- GdriveRecentPathState::Deleted => {}
- other => panic!("unexpected state for stale child: {other:?}"),
- }
-
- core.record_recent_upsert(
- "parent/dir/file.txt",
- Metadata::new(EntryMode::FILE).with_content_length(1),
- )
- .await;
-
- match core.recent_entry_for_path("parent/dir").await {
- GdriveRecentPathState::Present(_) => {}
- other => panic!("unexpected state for revived dir: {other:?}"),
- }
-
- match core.recent_entry_for_path("parent/dir/file.txt").await {
- GdriveRecentPathState::Present(_) => {}
- other => panic!("unexpected state for revived child: {other:?}"),
- }
-
- match core.recent_entry_for_path("parent/dir/stale.txt").await {
- GdriveRecentPathState::Deleted => {}
- other => panic!("unexpected state for stale sibling: {other:?}"),
- }
-
- let entries = core.recent_entries_for_list("parent/", true).await;
- let paths = entries
- .into_iter()
- .map(|(path, _)| path)
- .collect::<Vec<_>>();
-
- assert_eq!(
- paths,
- vec!["parent/dir/".to_string(), "parent/dir/file.txt".to_string()]
- );
- }
-}
diff --git a/core/services/gdrive/src/deleter.rs b/core/services/gdrive/src/deleter.rs
index 9ced50aeb..e58538fd4 100644
--- a/core/services/gdrive/src/deleter.rs
+++ b/core/services/gdrive/src/deleter.rs
@@ -37,69 +37,20 @@ impl GdriveDeleter {
impl oio::OneShotDelete for GdriveDeleter {
async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
let path = build_abs_path(&self.core.root, &path);
- let mut file_id = match self.core.resolve_path(&path).await? {
- Some(id) => id,
- None => match self.core.resolve_path_after_refresh(&path).await? {
- Some(id) => id,
- None => return Ok(()),
- },
- };
-
- let is_dir = if path.ends_with('/') {
- true
+ let file_id = self.core.path_cache.get(&path).await?;
+ let file_id = if let Some(id) = file_id {
+ id
} else {
- let mut resp = self.core.gdrive_stat_by_id(&file_id).await?;
- if resp.status() == StatusCode::NOT_FOUND {
- file_id = match self.core.resolve_path_after_refresh(&path).await? {
- Some(id) => id,
- None => return Ok(()),
- };
- resp = self.core.gdrive_stat_by_id(&file_id).await?;
- }
-
- if resp.status() == StatusCode::NOT_FOUND {
- return Ok(());
- }
- if resp.status() != StatusCode::OK {
- return Err(parse_error(resp));
- }
-
- let bs = resp.into_body().to_bytes();
- let file: GdriveFile =
- serde_json::from_slice(&bs).map_err(new_json_deserialize_error)?;
- file.mime_type == "application/vnd.google-apps.folder"
+ return Ok(());
};
- let mut resp = self.core.gdrive_trash(&file_id).await?;
- if resp.status() == StatusCode::NOT_FOUND {
- if is_dir {
- self.core.refresh_dir_path(&path).await;
- } else {
- self.core.refresh_path(&path).await;
- }
-
- file_id = match self.core.resolve_path(&path).await? {
- Some(id) => id,
- None => return Ok(()),
- };
-
- resp = self.core.gdrive_trash(&file_id).await?;
- }
-
- if resp.status() == StatusCode::NOT_FOUND {
- return Ok(());
- }
- if resp.status() != StatusCode::OK {
+ let resp = self.core.gdrive_trash(&file_id).await?;
+ let status = resp.status();
+ if status != StatusCode::OK {
return Err(parse_error(resp));
}
- if is_dir {
- self.core.invalidate_dir_id(&path).await;
- self.core.record_recent_delete(&path, EntryMode::DIR).await;
- } else {
- self.core.invalidate_file_id(&path).await;
- self.core.record_recent_delete(&path, EntryMode::FILE).await;
- }
+ self.core.path_cache.remove(&path).await;
Ok(())
}
diff --git a/core/services/gdrive/src/docs.md b/core/services/gdrive/src/docs.md
index 420573d3c..bdca4fe22 100644
--- a/core/services/gdrive/src/docs.md
+++ b/core/services/gdrive/src/docs.md
@@ -12,12 +12,6 @@ This service can be used to:
- [x] rename
- [ ] presign
-## Behavior Notes
-
-Google Drive allows duplicate file or directory names under the same parent.
-When multiple entries match the same path, OpenDAL resolves the most recently
-modified match, falling back to the newer creation time if needed.
-
# Configuration
- `root`: Set the work directory for backend
diff --git a/core/services/gdrive/src/lib.rs b/core/services/gdrive/src/lib.rs
index 4b358ea34..05baafb78 100644
--- a/core/services/gdrive/src/lib.rs
+++ b/core/services/gdrive/src/lib.rs
@@ -30,7 +30,6 @@ mod core;
mod deleter;
mod error;
mod lister;
-mod path_index;
mod writer;
pub use builder::GdriveBuilder as Gdrive;
diff --git a/core/services/gdrive/src/lister.rs b/core/services/gdrive/src/lister.rs
index 3b7d80e03..0fd07f95d 100644
--- a/core/services/gdrive/src/lister.rs
+++ b/core/services/gdrive/src/lister.rs
@@ -16,17 +16,14 @@
// under the License.
use std::collections::HashMap;
-use std::collections::HashSet;
use std::collections::VecDeque;
use std::sync::Arc;
use http::StatusCode;
-use mea::mutex::Mutex;
use super::core::GdriveCore;
use super::core::GdriveFile;
use super::core::GdriveFileList;
-use super::core::GdriveRecentPathState;
use super::error::parse_error;
use opendal_core::raw::*;
use opendal_core::*;
@@ -34,104 +31,17 @@ use opendal_core::*;
pub struct GdriveLister {
path: String,
core: Arc<GdriveCore>,
- emitted_paths: Mutex<HashSet<String>>,
- recent_entries_loaded: Mutex<bool>,
}
impl GdriveLister {
pub fn new(path: String, core: Arc<GdriveCore>) -> Self {
- Self {
- path,
- core,
- emitted_paths: Mutex::default(),
- recent_entries_loaded: Mutex::default(),
- }
- }
-
- async fn push_entry(
- &self,
- ctx: &mut oio::PageContext,
- path: String,
- metadata: Metadata,
- ) -> Result<()> {
- let mut emitted_paths = self.emitted_paths.lock().await;
- if emitted_paths.insert(path.clone()) {
- ctx.entries.push_back(oio::Entry::new(&path, metadata));
- }
- Ok(())
- }
-
- async fn apply_recent_entry(
- &self,
- ctx: &mut oio::PageContext,
- abs_path: String,
- metadata: Metadata,
- ) -> Result<()> {
- match self.core.recent_entry_for_path(&abs_path).await {
- GdriveRecentPathState::Present(recent_metadata) => {
- let path = build_rel_path(&self.core.root, &abs_path);
- self.push_entry(ctx, path, *recent_metadata).await
- }
- GdriveRecentPathState::Deleted => Ok(()),
- GdriveRecentPathState::Missing => {
- let path = build_rel_path(&self.core.root, &abs_path);
- self.push_entry(ctx, path, metadata).await
- }
- }
- }
-
- async fn inject_recent_entries(&self, ctx: &mut oio::PageContext) -> Result<()> {
- let mut recent_entries_loaded = self.recent_entries_loaded.lock().await;
- if *recent_entries_loaded {
- return Ok(());
- }
- *recent_entries_loaded = true;
-
- for (abs_path, metadata) in self.core.recent_entries_for_list(&self.path, false).await {
- let path = build_rel_path(&self.core.root, &abs_path);
- self.push_entry(ctx, path, metadata).await?;
- }
-
- Ok(())
+ Self { path, core }
}
}
-fn metadata_from_gdrive_file(file: &GdriveFile) -> Result<Metadata> {
- let mut metadata = Metadata::new(
- if file.mime_type.as_str() == "application/vnd.google-apps.folder" {
- EntryMode::DIR
- } else {
- EntryMode::FILE
- },
- )
- .with_content_type(file.mime_type.clone());
-
- if let Some(size) = &file.size {
- metadata = metadata.with_content_length(size.parse::<u64>().map_err(|e| {
- Error::new(ErrorKind::Unexpected, "parse content length").set_source(e)
- })?);
- }
- if let Some(modified_time) = &file.modified_time {
- metadata =
- metadata.with_last_modified(modified_time.parse::<Timestamp>().map_err(|e| {
- Error::new(ErrorKind::Unexpected, "parse last modified time").set_source(e)
- })?);
- }
-
- Ok(metadata)
-}
-
impl oio::PageList for GdriveLister {
async fn next_page(&self, ctx: &mut oio::PageContext) -> Result<()> {
- if let GdriveRecentPathState::Deleted = self.core.recent_entry_for_path(&self.path).await {
- ctx.done = true;
- return Ok(());
- }
-
- let file_id = match self.core.resolve_path(&self.path).await? {
- Some(file_id) => Some(file_id),
- None => self.core.resolve_path_after_refresh(&self.path).await?,
- };
+ let file_id = self.core.path_cache.get(&self.path).await?;
let file_id = match file_id {
Some(file_id) => file_id,
@@ -150,18 +60,7 @@ impl oio::PageList for GdriveLister {
// - `list("dir")` returns `dir/` (but does NOT list its children)
// - list children requires a trailing slash, like `list("dir/")`
if !is_dir_path {
- let mut resp = self.core.gdrive_stat_by_id(&file_id).await?;
- if resp.status() == StatusCode::NOT_FOUND {
- let file_id = match self.core.resolve_path_after_refresh(&self.path).await? {
- Some(file_id) => file_id,
- None => {
- ctx.done = true;
- return Ok(());
- }
- };
-
- resp = self.core.gdrive_stat_by_id(&file_id).await?;
- }
+ let resp = self.core.gdrive_stat_by_id(&file_id).await?;
if resp.status() != StatusCode::OK {
return Err(parse_error(resp));
}
@@ -176,33 +75,36 @@ impl oio::PageList for GdriveLister {
} else {
self.path.clone()
};
- let metadata = metadata_from_gdrive_file(&file)?;
+ let path = build_rel_path(&self.core.root, &abs_path);
- self.apply_recent_entry(ctx, abs_path, metadata).await?;
+ let mut meta = Metadata::new(if is_dir {
+ EntryMode::DIR
+ } else {
+ EntryMode::FILE
+ })
+ .with_content_type(file.mime_type);
+ if let Some(size) = file.size {
+ meta = meta.with_content_length(size.parse::<u64>().map_err(|e| {
+ Error::new(ErrorKind::Unexpected, "parse content length").set_source(e)
+ })?);
+ }
+ if let Some(modified_time) = file.modified_time {
+ meta =
+ meta.with_last_modified(modified_time.parse::<Timestamp>().map_err(|e| {
+ Error::new(ErrorKind::Unexpected, "parse last modified time").set_source(e)
+ })?);
+ }
+
+ ctx.entries.push_back(oio::Entry::new(&path, meta));
ctx.done = true;
return Ok(());
}
- let mut resp = self
+ let resp = self
.core
.gdrive_list(file_id.as_str(), 1000, &ctx.token)
.await?;
- if resp.status() == StatusCode::NOT_FOUND {
- self.core.refresh_dir_path(&self.path).await;
- let file_id = match self.core.resolve_path(&self.path).await? {
- Some(file_id) => file_id,
- None => {
- ctx.done = true;
- return Ok(());
- }
- };
- resp = self
- .core
- .gdrive_list(file_id.as_str(), 1000, &ctx.token)
- .await?;
- }
-
let bytes = match resp.status() {
StatusCode::OK => resp.into_body().to_bytes(),
_ => return Err(parse_error(resp)),
@@ -217,9 +119,8 @@ impl oio::PageList for GdriveLister {
// Include the current directory itself when handling the first page of the listing.
if ctx.token.is_empty() && !ctx.done {
let path = build_rel_path(&self.core.root, &self.path);
- self.push_entry(ctx, path, Metadata::new(EntryMode::DIR))
- .await?;
- self.inject_recent_entries(ctx).await?;
+ let e = oio::Entry::new(&path, Metadata::new(EntryMode::DIR));
+ ctx.entries.push_back(e);
}
let decoded_response =
@@ -232,15 +133,35 @@ impl oio::PageList for GdriveLister {
}
for mut file in decoded_response.files {
- if file.mime_type.as_str() == "application/vnd.google-apps.folder"
- && !file.name.ends_with('/')
- {
- file.name += "/";
- }
+ let file_type = if file.mime_type.as_str() == "application/vnd.google-apps.folder" {
+ if !file.name.ends_with('/') {
+ file.name += "/";
+ }
+ EntryMode::DIR
+ } else {
+ EntryMode::FILE
+ };
+ let root = &self.core.root;
let path = format!("{}{}", &self.path, file.name);
- let metadata = metadata_from_gdrive_file(&file)?;
- self.apply_recent_entry(ctx, path, metadata).await?;
+ let normalized_path = build_rel_path(root, &path);
+
+ let mut metadata = Metadata::new(file_type).with_content_type(file.mime_type.clone());
+ if let Some(size) = file.size {
+ metadata = metadata.with_content_length(size.parse::<u64>().map_err(|e| {
+ Error::new(ErrorKind::Unexpected, "parse content length").set_source(e)
+ })?);
+ }
+ if let Some(modified_time) = file.modified_time {
+ metadata = metadata.with_last_modified(
+ modified_time.parse::<Timestamp>().map_err(|e| {
+ Error::new(ErrorKind::Unexpected, "parse last modified time").set_source(e)
+ })?,
+ );
+ }
+
+ let entry = oio::Entry::new(&normalized_path, metadata);
+ ctx.entries.push_back(entry);
}
Ok(())
@@ -287,13 +208,6 @@ pub struct GdriveFlatLister {
/// Pagination token for current batch query
page_token: String,
- /// Track emitted paths so recent local entries and later remote pages don't
- /// produce duplicates.
- emitted_paths: HashSet<String>,
-
- /// Whether we've already merged recent local entries for this listing.
- recent_entries_loaded: bool,
-
/// Whether we've started processing
started: bool,
@@ -318,68 +232,11 @@ impl GdriveFlatLister {
current_batch: Vec::new(),
dir_id_to_path: HashMap::new(),
page_token: String::new(),
- emitted_paths: HashSet::new(),
- recent_entries_loaded: false,
started: false,
done: false,
}
}
- fn push_entry(&mut self, path: String, metadata: Metadata) {
- if self.emitted_paths.insert(path.clone()) {
- self.entry_buffer
- .push_back(oio::Entry::new(&path, metadata));
- }
- }
-
- fn apply_refreshed_batch(&mut self, refreshed_dirs: Vec<(String, String)>) -> bool {
- self.page_token.clear();
- self.current_batch.clear();
-
- for (dir_id, path) in refreshed_dirs {
- self.dir_id_to_path.insert(dir_id.clone(), path);
- self.current_batch.push(dir_id);
- }
-
- !self.current_batch.is_empty()
- }
-
- async fn apply_recent_entry(
- &mut self,
- abs_path: String,
- metadata: Metadata,
- recent_state: GdriveRecentPathState,
- ) -> Result<()> {
- match recent_state {
- GdriveRecentPathState::Present(recent_metadata) => {
- let rel_path = build_rel_path(&self.core.root, &abs_path);
- self.push_entry(rel_path, *recent_metadata);
- }
- GdriveRecentPathState::Deleted => {}
- GdriveRecentPathState::Missing => {
- let rel_path = build_rel_path(&self.core.root, &abs_path);
- self.push_entry(rel_path, metadata);
- }
- }
-
- Ok(())
- }
-
- async fn inject_recent_entries(&mut self) -> Result<()> {
- if self.recent_entries_loaded {
- return Ok(());
- }
- self.recent_entries_loaded = true;
-
- let scope_path = self.prefix.as_deref().unwrap_or(&self.root_path);
- for (abs_path, metadata) in self.core.recent_entries_for_list(scope_path, true).await {
- let rel_path = build_rel_path(&self.core.root, &abs_path);
- self.push_entry(rel_path, metadata);
- }
-
- Ok(())
- }
-
/// Initialize the lister by resolving the root directory ID
async fn initialize(&mut self) -> Result<()> {
log::debug!(
@@ -387,122 +244,58 @@ impl GdriveFlatLister {
&self.root_path
);
- if let GdriveRecentPathState::Deleted =
- self.core.recent_entry_for_path(&self.root_path).await
- {
- self.done = true;
- return Ok(());
- }
+ let root_id = self.core.path_cache.get(&self.root_path).await?;
- let root_id = match self.core.resolve_path(&self.root_path).await? {
+ let root_id = match root_id {
Some(id) => {
log::debug!("GdriveFlatLister: root path resolved to ID: {:?}", &id);
id
}
- None => match self
- .core
- .resolve_path_after_refresh(&self.root_path)
- .await?
- {
- Some(id) => {
- log::debug!("GdriveFlatLister: root path resolved to ID: {:?}", &id);
- id
+ None => {
+ log::debug!(
+ "GdriveFlatLister: root path not found: {:?}",
+ &self.root_path
+ );
+ // Directory listing on a non-existent directory should return empty.
+ if self.root_path.ends_with('/') {
+ self.done = true;
+ return Ok(());
}
- None => {
- log::debug!(
- "GdriveFlatLister: root path not found: {:?}",
- &self.root_path
- );
- if self.root_path.ends_with('/') {
+
+ // Prefix listing can still return matches even if the exact path doesn't exist.
+ let prefix = self.root_path.clone();
+ self.prefix = Some(prefix.clone());
+
+ let mut parent_path = get_parent(&prefix).to_string();
+ if parent_path == "/" {
+ parent_path.clear();
+ }
+
+ let parent_id = self.core.path_cache.get(&parent_path).await?;
+ let parent_id = match parent_id {
+ Some(id) => id,
+ None => {
self.done = true;
return Ok(());
}
+ };
- let prefix = self.root_path.clone();
- self.prefix = Some(prefix.clone());
-
- let mut parent_path = get_parent(&prefix).to_string();
- if parent_path == "/" {
- parent_path.clear();
- }
+ self.root_path = parent_path.clone();
+ self.pending_dirs.push_back(PendingDir {
+ id: parent_id,
+ path: parent_path,
+ });
- let parent_id = match self.core.resolve_path(&parent_path).await? {
- Some(id) => id,
- None => match self.core.resolve_path_after_refresh(&parent_path).await? {
- Some(id) => id,
- None => {
- self.done = true;
- return Ok(());
- }
- },
- };
-
- self.root_path = parent_path.clone();
- self.pending_dirs.push_back(PendingDir {
- id: parent_id,
- path: parent_path,
- });
-
- self.inject_recent_entries().await?;
- self.started = true;
- return Ok(());
- }
- },
+ self.started = true;
+ return Ok(());
+ }
};
// Resolve the entry type for root path so we can handle:
//
// - `list("dir", recursive=true)` where `dir` is a folder but without trailing slash.
// - `list("prefix", recursive=true)` where `prefix` points to a file or a file prefix.
- let mut resp = self.core.gdrive_stat_by_id(&root_id).await?;
- if resp.status() == StatusCode::NOT_FOUND {
- if self.root_path.ends_with('/') {
- self.core.refresh_dir_path(&self.root_path).await;
- } else {
- self.core.refresh_path(&self.root_path).await;
- }
-
- let root_id = match self.core.resolve_path(&self.root_path).await? {
- Some(id) => id,
- None => {
- if self.root_path.ends_with('/') {
- self.done = true;
- return Ok(());
- }
-
- let prefix = self.root_path.clone();
- self.prefix = Some(prefix.clone());
-
- let mut parent_path = get_parent(&prefix).to_string();
- if parent_path == "/" {
- parent_path.clear();
- }
-
- let parent_id = match self.core.resolve_path(&parent_path).await? {
- Some(id) => id,
- None => match self.core.resolve_path_after_refresh(&parent_path).await? {
- Some(id) => id,
- None => {
- self.done = true;
- return Ok(());
- }
- },
- };
-
- self.root_path = parent_path.clone();
- self.pending_dirs.push_back(PendingDir {
- id: parent_id,
- path: parent_path,
- });
-
- self.inject_recent_entries().await?;
- self.started = true;
- return Ok(());
- }
- };
-
- resp = self.core.gdrive_stat_by_id(&root_id).await?;
- }
+ let resp = self.core.gdrive_stat_by_id(&root_id).await?;
if resp.status() != StatusCode::OK {
return Err(parse_error(resp));
}
@@ -516,7 +309,7 @@ impl GdriveFlatLister {
// Directory paths must end with `/` (except the root which is represented by empty path).
if !self.root_path.is_empty() && !self.root_path.ends_with('/') {
self.root_path.push('/');
- self.core.cache_dir_id(&self.root_path, &root_id).await;
+ self.core.path_cache.insert(&self.root_path, &root_id).await;
}
// Add the root directory entry first.
@@ -524,7 +317,8 @@ impl GdriveFlatLister {
if !rel_path.is_empty() && !rel_path.ends_with('/') {
rel_path.push('/');
}
- self.push_entry(rel_path, Metadata::new(EntryMode::DIR));
+ let entry = oio::Entry::new(&rel_path, Metadata::new(EntryMode::DIR));
+ self.entry_buffer.push_back(entry);
// Queue the root directory for listing.
self.pending_dirs.push_back(PendingDir {
@@ -542,15 +336,13 @@ impl GdriveFlatLister {
parent_path.clear();
}
- let parent_id = match self.core.resolve_path(&parent_path).await? {
+ let parent_id = self.core.path_cache.get(&parent_path).await?;
+ let parent_id = match parent_id {
Some(id) => id,
- None => match self.core.resolve_path_after_refresh(&parent_path).await? {
- Some(id) => id,
- None => {
- self.done = true;
- return Ok(());
- }
- },
+ None => {
+ self.done = true;
+ return Ok(());
+ }
};
self.root_path = parent_path.clone();
@@ -560,7 +352,6 @@ impl GdriveFlatLister {
});
}
- self.inject_recent_entries().await?;
self.started = true;
Ok(())
}
@@ -591,34 +382,11 @@ impl GdriveFlatLister {
&self.current_batch
);
- let mut resp = self
+ let resp = self
.core
.gdrive_list_batch(&self.current_batch, PAGE_SIZE, &self.page_token)
.await?;
- if resp.status() == StatusCode::NOT_FOUND {
- let current_batch = self.current_batch.clone();
- let mut refreshed_batch = Vec::new();
- for dir_id in current_batch {
- if let Some(path) = self.dir_id_to_path.get(&dir_id).cloned() {
- self.core.refresh_dir_path(&path).await;
- if let Some(new_id) = self.core.resolve_path(&path).await? {
- refreshed_batch.push((new_id, path));
- }
- }
- }
-
- if !self.apply_refreshed_batch(refreshed_batch) {
- self.done = true;
- return Ok(());
- }
-
- resp = self
- .core
- .gdrive_list_batch(&self.current_batch, PAGE_SIZE, &self.page_token)
- .await?;
- }
-
let bytes = match resp.status() {
StatusCode::OK => resp.into_body().to_bytes(),
_ => return Err(parse_error(resp)),
@@ -678,8 +446,28 @@ impl GdriveFlatLister {
.unwrap_or_else(|| self.root_path.clone());
let abs_path = format!("{}{}", parent_path, file.name);
- let metadata = metadata_from_gdrive_file(&file)?;
- let recent_state = self.core.recent_entry_for_path(&abs_path).await;
+ let rel_path = build_rel_path(&self.core.root, &abs_path);
+
+ // Build metadata
+ let entry_mode = if is_dir {
+ EntryMode::DIR
+ } else {
+ EntryMode::FILE
+ };
+
+ let mut metadata = Metadata::new(entry_mode).with_content_type(file.mime_type.clone());
+
+ if let Some(size) = file.size {
+ if let Ok(size) = size.parse::<u64>() {
+ metadata = metadata.with_content_length(size);
+ }
+ }
+
+ if let Some(modified_time) = file.modified_time {
+ if let Ok(ts) = modified_time.parse::<Timestamp>() {
+ metadata = metadata.with_last_modified(ts);
+ }
+ }
let matches_prefix = self
.prefix
@@ -688,13 +476,12 @@ impl GdriveFlatLister {
.unwrap_or(true);
if matches_prefix {
- self.apply_recent_entry(abs_path.clone(), metadata, recent_state.clone())
- .await?;
+ let entry = oio::Entry::new(&rel_path, metadata);
+ self.entry_buffer.push_back(entry);
}
// If it's a directory, queue it for recursive listing if it can contain matching entries.
let should_traverse = is_dir
- && !matches!(recent_state, GdriveRecentPathState::Deleted)
&& self
.prefix
.as_ref()
@@ -745,76 +532,3 @@ impl oio::List for GdriveFlatLister {
}
}
}
-
-#[cfg(test)]
-mod tests {
- use std::sync::Arc;
-
- use mea::mutex::Mutex;
-
- use super::*;
- use crate::core::GdrivePathQuery;
- use crate::core::GdriveRecentState;
- use crate::core::GdriveSigner;
- use crate::path_index::GdrivePathIndex;
-
- fn mock_gdrive_core() -> Arc<GdriveCore> {
- let info = Arc::new(AccessorInfo::default());
- let signer = Arc::new(Mutex::new(GdriveSigner::new(info.clone())));
-
- Arc::new(GdriveCore {
- info: info.clone(),
- root: "/".to_string(),
- signer: signer.clone(),
- path_index: GdrivePathIndex::new(GdrivePathQuery::new(info, signer)),
- recent_entries: Mutex::new(GdriveRecentState::default()),
- })
- }
-
- #[tokio::test]
- async fn test_process_file_skips_deleted_dirs_for_traversal() {
- let core = mock_gdrive_core();
- core.record_recent_delete("parent/dir/", EntryMode::DIR)
- .await;
-
- let mut lister = GdriveFlatLister::new("parent/".to_string(), core);
- lister
- .dir_id_to_path
- .insert("parent-id".to_string(), "parent/".to_string());
-
- lister
- .process_file(GdriveFile {
- mime_type: "application/vnd.google-apps.folder".to_string(),
- id: "dir-id".to_string(),
- name: "dir".to_string(),
- size: None,
- modified_time: None,
- parents: vec!["parent-id".to_string()],
- })
- .await
- .unwrap();
-
- assert!(lister.pending_dirs.is_empty());
- assert!(!lister.dir_id_to_path.contains_key("dir-id"));
- assert!(lister.entry_buffer.is_empty());
- }
-
- #[test]
- fn test_apply_refreshed_batch_clears_page_token() {
- let core = mock_gdrive_core();
- let mut lister = GdriveFlatLister::new("parent/".to_string(), core);
- lister.page_token = "stale-page-token".to_string();
- lister.current_batch = vec!["old-id".to_string()];
- lister
- .dir_id_to_path
- .insert("old-id".to_string(), "parent/".to_string());
-
- assert!(lister.apply_refreshed_batch(vec![("new-id".to_string(), "parent/".to_string(),)]));
- assert!(lister.page_token.is_empty());
- assert_eq!(lister.current_batch, vec!["new-id".to_string()]);
- assert_eq!(
- lister.dir_id_to_path.get("new-id").map(String::as_str),
- Some("parent/")
- );
- }
-}
diff --git a/core/services/gdrive/src/path_index.rs b/core/services/gdrive/src/path_index.rs
deleted file mode 100644
index f7fd8962f..000000000
--- a/core/services/gdrive/src/path_index.rs
+++ /dev/null
@@ -1,451 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-use std::collections::HashMap;
-use std::collections::VecDeque;
-
-use mea::mutex::Mutex;
-
-use super::core::normalize_dir_path;
-use opendal_core::raw::*;
-use opendal_core::*;
-
-#[derive(Default)]
-struct GdrivePathIndexState {
- entries: HashMap<String, String>,
-}
-
-/// A gdrive-local path index that owns all cache semantics for path -> id resolution.
-///
-/// The index keeps canonical directory paths (`dir/`) internally, supports exact upserts,
-/// can invalidate subtrees, and can refresh ancestor chains before re-resolving stale paths.
-pub struct GdrivePathIndex<Q: PathQuery> {
- query: Q,
- state: Mutex<GdrivePathIndexState>,
- ensure_dir_lock: Mutex<()>,
-}
-
-impl<Q: PathQuery> GdrivePathIndex<Q> {
- pub fn new(query: Q) -> Self {
- Self {
- query,
- state: Mutex::new(GdrivePathIndexState::default()),
- ensure_dir_lock: Mutex::new(()),
- }
- }
-
- async fn root_id(&self) -> Result<String> {
- self.query.root().await
- }
-
- fn canonical_dir_path(path: &str) -> String {
- normalize_dir_path(path)
- }
-
- fn remove_path_candidates(entries: &mut HashMap<String, String>, path: &str) {
- entries.remove(path);
-
- let dir_path = Self::canonical_dir_path(path);
- if dir_path != path {
- entries.remove(&dir_path);
- }
- }
-
- fn lookup_cached_id(entries: &HashMap<String, String>, path: &str) -> Option<String> {
- if let Some(id) = entries.get(path) {
- return Some(id.clone());
- }
-
- let dir_path = Self::canonical_dir_path(path);
- if dir_path != path {
- return entries.get(&dir_path).cloned();
- }
-
- None
- }
-
- fn resolution_chain(path: &str) -> Vec<String> {
- let mut chain = VecDeque::new();
- let mut current = path;
-
- while current != "/" && !current.is_empty() {
- chain.push_front(current.to_string());
- current = get_parent(current);
- }
-
- chain.into_iter().collect()
- }
-
- async fn resolve_from_chain(
- &self,
- chain: Vec<String>,
- start_index: usize,
- start_id: String,
- ) -> Result<Option<String>> {
- let mut current_id = start_id;
-
- for path in chain.into_iter().skip(start_index) {
- let name = get_basename(&path);
- current_id = match self.query.query(¤t_id, name).await? {
- Some(id) => {
- let cache_path = path;
- if cache_path.ends_with('/') {
- self.upsert_dir(&cache_path, &id).await;
- } else {
- self.upsert_file(&cache_path, &id).await;
- }
- id
- }
- None => return Ok(None),
- };
- }
-
- Ok(Some(current_id))
- }
-
- async fn lookup_resolve_plan(&self, path: &str) -> Result<(Option<String>, Vec<String>)> {
- if path.is_empty() || path == "/" {
- return Ok((Some(self.root_id().await?), Vec::new()));
- }
-
- let chain = Self::resolution_chain(path);
- if chain.is_empty() {
- return Ok((Some(self.root_id().await?), Vec::new()));
- }
-
- let (start_id, start_index) = {
- let state = self.state.lock().await;
- let mut found = None;
-
- for idx in (0..chain.len()).rev() {
- if let Some(id) = Self::lookup_cached_id(&state.entries, &chain[idx]) {
- found = Some((id, idx + 1));
- break;
- }
- }
-
- found.unwrap_or((String::new(), 0))
- };
-
- if start_index == 0 {
- Ok((None, chain))
- } else {
- Ok((
- Some(start_id),
- chain.into_iter().skip(start_index).collect(),
- ))
- }
- }
-
- /// Get the id for the given path.
- pub async fn get(&self, path: &str) -> Result<Option<String>> {
- if path.is_empty() || path == "/" {
- return self.root_id().await.map(Some);
- }
-
- if let Some(id) = {
- let state = self.state.lock().await;
- Self::lookup_cached_id(&state.entries, path)
- } {
- return Ok(Some(id));
- }
-
- let (start_id, missing_paths) = self.lookup_resolve_plan(path).await?;
- match start_id {
- Some(start_id) => self.resolve_from_chain(missing_paths, 0, start_id).await,
- None => {
- let root_id = self.root_id().await?;
- self.resolve_from_chain(missing_paths, 0, root_id).await
- }
- }
- }
-
- /// Insert or replace a file mapping.
- pub async fn upsert_file(&self, path: &str, id: &str) {
- let mut state = self.state.lock().await;
- Self::remove_path_candidates(&mut state.entries, path);
- state.entries.insert(path.to_string(), id.to_string());
- }
-
- /// Insert or replace a directory mapping using canonical `dir/` storage.
- pub async fn upsert_dir(&self, path: &str, id: &str) {
- let dir_path = Self::canonical_dir_path(path);
-
- let mut state = self.state.lock().await;
- state.entries.remove(dir_path.trim_end_matches('/'));
- Self::remove_path_candidates(&mut state.entries, &dir_path);
- state.entries.insert(dir_path, id.to_string());
- }
-
- /// Remove a file mapping.
- pub async fn invalidate_file(&self, path: &str) {
- let mut state = self.state.lock().await;
- Self::remove_path_candidates(&mut state.entries, path);
- }
-
- /// Remove a directory mapping and all known descendants.
- pub async fn invalidate_dir(&self, path: &str) {
- let dir_path = Self::canonical_dir_path(path);
-
- let mut state = self.state.lock().await;
- let file_path = dir_path.trim_end_matches('/').to_string();
- state.entries.remove(&file_path);
- state
- .entries
- .retain(|entry_path, _| !entry_path.starts_with(&dir_path));
- }
-
- /// Invalidate the given path plus its ancestor chain before resolving it again.
- pub async fn refresh_path(&self, path: &str) {
- if path.is_empty() || path == "/" {
- return;
- }
-
- let mut current = path.to_string();
- while current != "/" && !current.is_empty() {
- let dir_path = Self::canonical_dir_path(¤t);
- let mut state = self.state.lock().await;
- Self::remove_path_candidates(&mut state.entries, ¤t);
- Self::remove_path_candidates(&mut state.entries, &dir_path);
- drop(state);
- current = get_parent(¤t).to_string();
- }
- }
-
- /// Ensure input dir exists.
- pub async fn ensure_dir(&self, path: &str) -> Result<String> {
- let _guard = self.ensure_dir_lock.lock().await;
-
- let path = Self::canonical_dir_path(path);
- if path.is_empty() {
- return self.root_id().await;
- }
-
- let mut tmp = String::new();
- let mut parents = vec![];
- for component in path.split('/') {
- if component.is_empty() {
- continue;
- }
-
- tmp.push_str(component);
- tmp.push('/');
- parents.push(tmp.clone());
- }
-
- let mut refreshed = false;
- 'retry: loop {
- let mut parent_id = self.root_id().await?;
- for parent in &parents {
- if let Some(value) = {
- let state = self.state.lock().await;
- Self::lookup_cached_id(&state.entries, parent)
- } {
- parent_id = value;
- continue;
- }
-
- let name = get_basename(parent);
- let value = match self.query.query(&parent_id, name).await {
- Ok(Some(value)) => value,
- Ok(None) => match self.query.create_dir(&parent_id, name).await {
- Ok(value) => value,
- Err(err) if !refreshed && err.kind() == ErrorKind::NotFound => {
- refreshed = true;
- self.refresh_path(parent).await;
- continue 'retry;
- }
- Err(err) => return Err(err),
- },
- Err(err) if !refreshed && err.kind() == ErrorKind::NotFound => {
- refreshed = true;
- self.refresh_path(parent).await;
- continue 'retry;
- }
- Err(err) => return Err(err),
- };
-
- self.upsert_dir(parent, &value).await;
- parent_id = value;
- }
-
- return Ok(parent_id);
- }
- }
-}
-
-#[cfg(test)]
-mod tests {
- use std::collections::HashMap;
- use std::collections::HashSet;
- use std::sync::Arc;
- use std::sync::atomic::AtomicUsize;
- use std::sync::atomic::Ordering;
-
- use mea::mutex::Mutex;
-
- use super::*;
-
- #[derive(Clone)]
- struct TestQuery {
- root_id: String,
- entries: Arc<Mutex<HashMap<(String, String), String>>>,
- stale_parents: Arc<Mutex<HashSet<String>>>,
- create_count: Arc<AtomicUsize>,
- }
-
- impl TestQuery {
- fn new() -> Self {
- Self {
- root_id: "remote-root".to_string(),
- entries: Arc::new(Mutex::new(HashMap::new())),
- stale_parents: Arc::new(Mutex::new(HashSet::new())),
- create_count: Arc::new(AtomicUsize::new(0)),
- }
- }
-
- async fn insert(&self, parent_id: &str, name: &str, id: &str) {
- self.entries
- .lock()
- .await
- .insert((parent_id.to_string(), name.to_string()), id.to_string());
- }
-
- async fn mark_stale_parent(&self, parent_id: &str) {
- self.stale_parents
- .lock()
- .await
- .insert(parent_id.to_string());
- }
- }
-
- impl PathQuery for TestQuery {
- async fn root(&self) -> Result<String> {
- Ok(self.root_id.clone())
- }
-
- async fn query(&self, parent_id: &str, name: &str) -> Result<Option<String>> {
- if self.stale_parents.lock().await.contains(parent_id) {
- return Err(Error::new(ErrorKind::NotFound, "stale parent"));
- }
- Ok(self
- .entries
- .lock()
- .await
- .get(&(parent_id.to_string(), name.to_string()))
- .cloned())
- }
-
- async fn create_dir(&self, parent_id: &str, name: &str) -> Result<String> {
- if self.stale_parents.lock().await.contains(parent_id) {
- return Err(Error::new(ErrorKind::NotFound, "stale parent"));
- }
- self.create_count.fetch_add(1, Ordering::SeqCst);
- let id = format!("{parent_id}:{name}");
- self.insert(parent_id, name, &id).await;
- Ok(id)
- }
- }
-
- #[tokio::test]
- async fn test_upsert_replaces_existing_path() {
- let index = GdrivePathIndex::new(TestQuery::new());
- index.upsert_file("root/file", "old-id").await;
- index.upsert_file("root/file", "new-id").await;
-
- assert_eq!(
- index.get("root/file").await.unwrap().as_deref(),
- Some("new-id")
- );
- }
-
- #[tokio::test]
- async fn test_dir_upsert_replaces_file_mapping() {
- let index = GdrivePathIndex::new(TestQuery::new());
- index.upsert_file("root/dir", "file-id").await;
- index.upsert_dir("root/dir", "dir-id").await;
-
- assert_eq!(
- index.get("root/dir").await.unwrap().as_deref(),
- Some("dir-id")
- );
- assert_eq!(
- index.get("root/dir/").await.unwrap().as_deref(),
- Some("dir-id")
- );
- }
-
- #[tokio::test]
- async fn test_subtree_invalidation_re_resolves_descendants() {
- let query = TestQuery::new();
- let index = GdrivePathIndex::new(query.clone());
-
- query.insert("remote-root", "root/", "svc-root").await;
- query.insert("svc-root", "dir/", "dir-old").await;
- query.insert("dir-old", "file", "file-old").await;
-
- assert_eq!(
- index.get("root/dir/file").await.unwrap().as_deref(),
- Some("file-old")
- );
-
- query.insert("svc-root", "dir/", "dir-new").await;
- query.insert("dir-new", "file", "file-new").await;
-
- index.invalidate_dir("root/dir/").await;
-
- assert_eq!(
- index.get("root/dir/file").await.unwrap().as_deref(),
- Some("file-new")
- );
- }
-
- #[tokio::test]
- async fn test_ensure_dir_is_serialized() {
- let query = TestQuery::new();
- query.insert("remote-root", "root/", "svc-root").await;
-
- let index = GdrivePathIndex::new(query.clone());
-
- let (first, second) =
- tokio::join!(index.ensure_dir("root/dir"), index.ensure_dir("root/dir"));
-
- assert_eq!(first.unwrap(), second.unwrap());
- assert_eq!(query.create_count.load(Ordering::SeqCst), 1);
- }
-
- #[tokio::test]
- async fn test_ensure_dir_refreshes_stale_ancestor_chain() {
- let query = TestQuery::new();
- query.insert("remote-root", "root/", "root-old").await;
-
- let index = GdrivePathIndex::new(query.clone());
- index.upsert_dir("root/", "root-old").await;
-
- query.mark_stale_parent("root-old").await;
- query.insert("remote-root", "root/", "root-new").await;
-
- let dir_id = index.ensure_dir("root/dir").await.unwrap();
-
- assert_eq!(dir_id, "root-new:dir/");
- assert_eq!(query.create_count.load(Ordering::SeqCst), 1);
- assert_eq!(
- index.get("root/dir").await.unwrap().as_deref(),
- Some("root-new:dir/")
- );
- }
-}
diff --git a/core/services/gdrive/src/writer.rs b/core/services/gdrive/src/writer.rs
index 583c35a4c..6234316ee 100644
--- a/core/services/gdrive/src/writer.rs
+++ b/core/services/gdrive/src/writer.rs
@@ -49,43 +49,29 @@ impl oio::OneShotWrite for GdriveWriter {
async fn write_once(&self, bs: Buffer) -> Result<Metadata> {
let size = bs.len();
- let mut current_file_id = self.file_id.clone();
- let mut retried = false;
+ let resp = if let Some(file_id) = &self.file_id {
+ self.core
+ .gdrive_upload_overwrite_simple_request(file_id, size as u64, bs)
+ .await
+ } else {
+ self.core
+ .gdrive_upload_simple_request(&self.path, size as u64, bs)
+ .await
+ }?;
- loop {
- let resp = if let Some(file_id) = ¤t_file_id {
- self.core
- .gdrive_upload_overwrite_simple_request(file_id, size as u64, bs.clone())
- .await
- } else {
- self.core
- .gdrive_upload_simple_request(&self.path, size as u64, bs.clone())
- .await
- }?;
-
- match resp.status() {
- StatusCode::OK | StatusCode::CREATED => {
- let mut metadata =
- Metadata::new(EntryMode::FILE).with_content_length(size as u64);
-
- if current_file_id.is_none() {
- let bs = resp.into_body();
- let file: GdriveFile = serde_json::from_reader(bs.reader())
- .map_err(new_json_deserialize_error)?;
- metadata = metadata.with_content_type(file.mime_type);
- self.core.cache_file_id(&self.path, &file.id).await;
- }
- self.core.record_recent_upsert(&self.path, metadata).await;
- return Ok(Metadata::default());
- }
- StatusCode::NOT_FOUND if !retried && current_file_id.is_some() => {
- retried = true;
- self.core.refresh_path(&self.path).await;
- current_file_id = self.core.resolve_path(&self.path).await?;
- continue;
+ let status = resp.status();
+ match status {
+ StatusCode::OK | StatusCode::CREATED => {
+ // If we don't have the file id before, let's update the cache to avoid re-fetching.
+ if self.file_id.is_none() {
+ let bs = resp.into_body();
+ let file: GdriveFile =
+ serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
+ self.core.path_cache.insert(&self.path, &file.id).await;
}
- _ => return Err(parse_error(resp)),
+ Ok(Metadata::default())
}
+ _ => Err(parse_error(resp)),
}
}
}