Limit tag namespace names to not have subdirectories

Now that `spfs clean` only discovers and walks tags in tag namespaces
defined at the top level, it is unsafe to allow creating tag namespaces
in subdirectories.

It is assumed that this change will not affect any existing uses of tag
namespaces and that a deprecation process is not necessary.

Signed-off-by: J Robert Ray <[email protected]>

Author: jrray

crates/spfs/src/config.rs

@@ -266,7 +266,7 @@ impl RemoteConfig {
                     builder.when(tracking::TimeSpec::parse(v)?);
                 }
                 "tag_namespace" => {
-                    builder.tag_namespace(TagNamespaceBuf::new(RelativePath::new(&v)));
+                    builder.tag_namespace(TagNamespaceBuf::new(RelativePath::new(&v))?);
                 }
                 _ => (),
             }

crates/spfs/src/fixtures.rs

@@ -42,9 +42,10 @@ impl TempRepo {
                 let mut repo = spfs::storage::fs::FsRepository::open(tempdir.path().join("repo"))
                     .await
                     .unwrap();
-                repo.set_tag_namespace(Some(spfs::storage::TagNamespaceBuf::new(
-                    namespace.as_ref(),
-                )));
+                repo.set_tag_namespace(Some(
+                    spfs::storage::TagNamespaceBuf::new(namespace.as_ref())
+                        .expect("tag namespaces used in tests must be valid"),
+                ));
                 TempRepo::FS(Arc::new(repo.into()), Arc::clone(tempdir))
             }
             _ => panic!("only TempRepo::FS type supports setting tag namespaces"),

crates/spfs/src/server/tag.rs

@@ -19,7 +19,10 @@ fn string_to_namespace(namespace: &String) -> Option<&TagNamespace> {
     if namespace.is_empty() {
         None
     } else {
-        Some(TagNamespace::new(RelativePath::new(namespace)))
+        Some(
+            TagNamespace::new(RelativePath::new(namespace))
+                .expect("namespace was valid before being passed over rpc as a string"),
+        )
     }
 }
 

crates/spfs/src/storage/tag_namespace.rs

@@ -7,6 +7,8 @@ use std::ops::Deref;
 use relative_path::{RelativePath, RelativePathBuf};
 use serde::{Deserialize, Serialize};
 
+use crate::{Error, Result};
+
 /// A suffix on directory names that indicates that the directory is a tag namespace.
 pub const TAG_NAMESPACE_MARKER: &str = "#ns";
 
@@ -16,7 +18,22 @@ pub const TAG_NAMESPACE_MARKER: &str = "#ns";
 pub struct TagNamespace(RelativePath);
 
 impl TagNamespace {
-    pub fn new(path: &RelativePath) -> &Self {
+    /// Create a new TagNamespace instance.
+    ///
+    /// The path provided must only contain a single path component.
+    pub fn new(path: &RelativePath) -> Result<&Self> {
+        TagNamespaceBuf::validate_path(path)?;
+
+        // Safety: TagNamespace is repr(transparent) over RelativePath
+        Ok(unsafe { &*(path as *const RelativePath as *const TagNamespace) })
+    }
+
+    /// Create a new TagNamespace instance without validation.
+    ///
+    /// # Safety
+    ///
+    /// The path provided must only contain a single path component.
+    unsafe fn new_unchecked(path: &RelativePath) -> &Self {
         // Safety: TagNamespace is repr(transparent) over RelativePath
         unsafe { &*(path as *const RelativePath as *const TagNamespace) }
     }
@@ -52,8 +69,34 @@ impl TagNamespaceBuf {
     }
 
     /// Create a new tag namespace from the given path.
-    pub fn new<P: AsRef<RelativePath>>(path: P) -> Self {
-        Self(path.as_ref().to_owned())
+    pub fn new<P: AsRef<RelativePath>>(path: P) -> Result<Self> {
+        let path = path.as_ref();
+        TagNamespaceBuf::validate_path(path)?;
+        Ok(Self(path.to_owned()))
+    }
+
+    /// Validate that a RelativePath is a valid tag namespace name.
+    ///
+    /// Tag namespaces may not have any directory structure meaning the
+    /// directory created for the tag namespace will be created in the top level
+    /// of the tags directory in the spfs repository.
+    ///
+    /// When `spfs clean` walks tags to avoid data loss it must walk all tags is
+    /// all namespaces and with the current implementation only tag namespaces
+    /// defined in the top level of the tags directory will be discovered.
+    #[inline]
+    fn validate_path(path: &RelativePath) -> Result<()> {
+        let mut it = path.components();
+        if it.next().is_none() {
+            return Err(Error::String("tag namespace must not be empty".to_string()));
+        }
+        if it.next().is_some() {
+            return Err(Error::String(
+                "tag namespace must not contain any directory structure".to_string(),
+            ));
+        }
+
+        Ok(())
     }
 }
 
@@ -73,7 +116,8 @@ impl std::ops::Deref for TagNamespaceBuf {
     type Target = TagNamespace;
 
     fn deref(&self) -> &Self::Target {
-        TagNamespace::new(self.0.deref())
+        // Safety: A TagNamespaceBuf has already been checked for validity.
+        unsafe { TagNamespace::new_unchecked(self.0.deref()) }
     }
 }
 

crates/spfs/src/storage/tag_test.rs

@@ -383,13 +383,15 @@ async fn test_tag_in_namespace_name_collision(
     // | none      | foo/bar/baz |
     // | foo       | bar/baz     |
     // | foo/bar   | baz         |
+    //
+    // However the last one is not currently checked because "foo/bar" has
+    // become an invalid tag namespace name. Should that change in the future,
+    // the code to check it was removed in the commit that added this message.
 
     let repo_in_foo = tmprepo.with_tag_namespace("foo").await;
-    let repo_in_foo_bar = tmprepo.with_tag_namespace("foo/bar").await;
 
     let foo_bar_baz = random_digest();
     let bar_baz = random_digest();
-    let baz = random_digest();
 
     let spec_foo_bar_baz = tracking::TagSpec::parse("foo/bar/baz").unwrap();
     tmprepo
@@ -400,14 +402,8 @@ async fn test_tag_in_namespace_name_collision(
     let spec_bar_baz = tracking::TagSpec::parse("bar/baz").unwrap();
     repo_in_foo.push_tag(&spec_bar_baz, &bar_baz).await.unwrap();
 
-    let spec_baz = tracking::TagSpec::parse("baz").unwrap();
-    repo_in_foo_bar.push_tag(&spec_baz, &baz).await.unwrap();
-
     // Now confirm these can be read back.
 
-    let tag = repo_in_foo_bar.resolve_tag(&spec_baz).await.unwrap();
-    assert_eq!(tag.target, baz);
-
     let tag = repo_in_foo.resolve_tag(&spec_bar_baz).await.unwrap();
     assert_eq!(tag.target, bar_baz);